diff --git a/go/Makefile b/go/Makefile
index 12feed1..398314d 100644
--- a/go/Makefile
+++ b/go/Makefile
@@ -26,4 +26,4 @@
 mocks:
 	go get github.com/vektra/mockery/...
 
-	go generate ./...
\ No newline at end of file
+	go generate ./...
diff --git a/go/gitstore/bt_gitstore/bt_gitstore.go b/go/gitstore/bt_gitstore/bt_gitstore.go
index 77bb840..fa1d4d3 100644
--- a/go/gitstore/bt_gitstore/bt_gitstore.go
+++ b/go/gitstore/bt_gitstore/bt_gitstore.go
@@ -454,12 +454,12 @@
 		egroup.Go(func() error {
 			rowNames := rowNames[chunkStart:chunkEnd]
 			mutations := mutations[chunkStart:chunkEnd]
-			errs, err := b.table.ApplyBulk(context.TODO(), rowNames, mutations)
+			errs, err := b.table.ApplyBulk(ctx, rowNames, mutations)
 			if err != nil {
-				return skerr.Fmt("Error writing batch: %s", err)
+				return skerr.Fmt("Error writing batch [%d:%d]: %s", chunkStart, chunkEnd, err)
 			}
 			if errs != nil {
-				return skerr.Fmt("Error writing some portions of batch: %s", errs)
+				return skerr.Fmt("Error writing some portions of batch [%d:%d]: %s", chunkStart, chunkEnd, errs)
 			}
 			return nil
 		})
diff --git a/go/paramtools/params.go b/go/paramtools/params.go
index ddf32bf..7e33bb2 100644
--- a/go/paramtools/params.go
+++ b/go/paramtools/params.go
@@ -10,6 +10,7 @@
 	"strings"
 	"sync"
 
+	"go.skia.org/infra/go/skerr"
 	"go.skia.org/infra/go/util"
 )
 
@@ -378,15 +379,15 @@
 		}
 		parts := strings.Split(pair, "=")
 		if len(parts) != 2 {
-			return nil, fmt.Errorf("Failed to parse: %s", pair)
+			return nil, skerr.Fmt("failed to parse: %s", pair)
 		}
 		key, ok := p.keys[parts[0]]
 		if !ok {
-			return nil, fmt.Errorf("Failed to find key: %q", parts[0])
+			return nil, skerr.Fmt("failed to find key: %q", parts[0])
 		}
 		value, ok := p.values[parts[0]][parts[1]]
 		if !ok {
-			return nil, fmt.Errorf("Failed to find value: %q", parts[1])
+			return nil, skerr.Fmt("failed to find value %q in %q (%q)", parts[1], p.values[parts[0]], parts[0])
 		}
 		ret[key] = value
 	}
@@ -450,7 +451,7 @@
 	return o.paramsEncoder
 }
 
-// EncodeParamsAsString encodes the Params as a string.
+// EncodeParamsAsString encodes the Params as a string containing indices.
 func (o *OrderedParamSet) EncodeParamsAsString(p Params) (string, error) {
 	return o.getParamsEncoder().encodeAsString(p)
 }
diff --git a/go/util/util.go b/go/util/util.go
index 1f71be7..6f28a0d 100644
--- a/go/util/util.go
+++ b/go/util/util.go
@@ -185,6 +185,23 @@
 	return r
 }
 
+// ReverseString reverses a string. It may not handle well for UTF
+// combining characters.
+// https://groups.google.com/forum/#!topic/golang-nuts/oPuBaYJ17t4
+func ReverseString(input string) string {
+	// Get Unicode code points.
+	runes := []rune(input)
+	n := len(runes)
+
+	// Reverse
+	for i := 0; i < n/2; i++ {
+		runes[i], runes[n-1-i] = runes[n-1-i], runes[i]
+	}
+
+	// Convert back to UTF-8.
+	return string(runes)
+}
+
 // InsertString inserts the given string into the slice at the given index.
 func InsertString(strs []string, idx int, s string) []string {
 	oldLen := len(strs)
diff --git a/go/vcsinfo/mocks/VCS.go b/go/vcsinfo/mocks/VCS.go
new file mode 100644
index 0000000..176048c
--- /dev/null
+++ b/go/vcsinfo/mocks/VCS.go
@@ -0,0 +1,221 @@
+// Code generated by mockery v1.0.0. DO NOT EDIT.
+
+package mocks
+
+import context "context"
+import mock "github.com/stretchr/testify/mock"
+import time "time"
+import vcsinfo "go.skia.org/infra/go/vcsinfo"
+
+// VCS is an autogenerated mock type for the VCS type
+type VCS struct {
+	mock.Mock
+}
+
+// ByIndex provides a mock function with given fields: ctx, N
+func (_m *VCS) ByIndex(ctx context.Context, N int) (*vcsinfo.LongCommit, error) {
+	ret := _m.Called(ctx, N)
+
+	var r0 *vcsinfo.LongCommit
+	if rf, ok := ret.Get(0).(func(context.Context, int) *vcsinfo.LongCommit); ok {
+		r0 = rf(ctx, N)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*vcsinfo.LongCommit)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int) error); ok {
+		r1 = rf(ctx, N)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Details provides a mock function with given fields: ctx, hash, includeBranchInfo
+func (_m *VCS) Details(ctx context.Context, hash string, includeBranchInfo bool) (*vcsinfo.LongCommit, error) {
+	ret := _m.Called(ctx, hash, includeBranchInfo)
+
+	var r0 *vcsinfo.LongCommit
+	if rf, ok := ret.Get(0).(func(context.Context, string, bool) *vcsinfo.LongCommit); ok {
+		r0 = rf(ctx, hash, includeBranchInfo)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*vcsinfo.LongCommit)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string, bool) error); ok {
+		r1 = rf(ctx, hash, includeBranchInfo)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// DetailsMulti provides a mock function with given fields: ctx, hashes, includeBranchInfo
+func (_m *VCS) DetailsMulti(ctx context.Context, hashes []string, includeBranchInfo bool) ([]*vcsinfo.LongCommit, error) {
+	ret := _m.Called(ctx, hashes, includeBranchInfo)
+
+	var r0 []*vcsinfo.LongCommit
+	if rf, ok := ret.Get(0).(func(context.Context, []string, bool) []*vcsinfo.LongCommit); ok {
+		r0 = rf(ctx, hashes, includeBranchInfo)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*vcsinfo.LongCommit)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, []string, bool) error); ok {
+		r1 = rf(ctx, hashes, includeBranchInfo)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// From provides a mock function with given fields: start
+func (_m *VCS) From(start time.Time) []string {
+	ret := _m.Called(start)
+
+	var r0 []string
+	if rf, ok := ret.Get(0).(func(time.Time) []string); ok {
+		r0 = rf(start)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]string)
+		}
+	}
+
+	return r0
+}
+
+// GetBranch provides a mock function with given fields:
+func (_m *VCS) GetBranch() string {
+	ret := _m.Called()
+
+	var r0 string
+	if rf, ok := ret.Get(0).(func() string); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(string)
+	}
+
+	return r0
+}
+
+// GetFile provides a mock function with given fields: ctx, fileName, commitHash
+func (_m *VCS) GetFile(ctx context.Context, fileName string, commitHash string) (string, error) {
+	ret := _m.Called(ctx, fileName, commitHash)
+
+	var r0 string
+	if rf, ok := ret.Get(0).(func(context.Context, string, string) string); ok {
+		r0 = rf(ctx, fileName, commitHash)
+	} else {
+		r0 = ret.Get(0).(string)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string, string) error); ok {
+		r1 = rf(ctx, fileName, commitHash)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// IndexOf provides a mock function with given fields: ctx, hash
+func (_m *VCS) IndexOf(ctx context.Context, hash string) (int, error) {
+	ret := _m.Called(ctx, hash)
+
+	var r0 int
+	if rf, ok := ret.Get(0).(func(context.Context, string) int); ok {
+		r0 = rf(ctx, hash)
+	} else {
+		r0 = ret.Get(0).(int)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
+		r1 = rf(ctx, hash)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// LastNIndex provides a mock function with given fields: N
+func (_m *VCS) LastNIndex(N int) []*vcsinfo.IndexCommit {
+	ret := _m.Called(N)
+
+	var r0 []*vcsinfo.IndexCommit
+	if rf, ok := ret.Get(0).(func(int) []*vcsinfo.IndexCommit); ok {
+		r0 = rf(N)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*vcsinfo.IndexCommit)
+		}
+	}
+
+	return r0
+}
+
+// Range provides a mock function with given fields: begin, end
+func (_m *VCS) Range(begin time.Time, end time.Time) []*vcsinfo.IndexCommit {
+	ret := _m.Called(begin, end)
+
+	var r0 []*vcsinfo.IndexCommit
+	if rf, ok := ret.Get(0).(func(time.Time, time.Time) []*vcsinfo.IndexCommit); ok {
+		r0 = rf(begin, end)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*vcsinfo.IndexCommit)
+		}
+	}
+
+	return r0
+}
+
+// ResolveCommit provides a mock function with given fields: ctx, commitHash
+func (_m *VCS) ResolveCommit(ctx context.Context, commitHash string) (string, error) {
+	ret := _m.Called(ctx, commitHash)
+
+	var r0 string
+	if rf, ok := ret.Get(0).(func(context.Context, string) string); ok {
+		r0 = rf(ctx, commitHash)
+	} else {
+		r0 = ret.Get(0).(string)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
+		r1 = rf(ctx, commitHash)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Update provides a mock function with given fields: ctx, pull, allBranches
+func (_m *VCS) Update(ctx context.Context, pull bool, allBranches bool) error {
+	ret := _m.Called(ctx, pull, allBranches)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, bool, bool) error); ok {
+		r0 = rf(ctx, pull, allBranches)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
diff --git a/go/vcsinfo/mocks/generate.go b/go/vcsinfo/mocks/generate.go
new file mode 100644
index 0000000..f926ff3
--- /dev/null
+++ b/go/vcsinfo/mocks/generate.go
@@ -0,0 +1,3 @@
+package mocks
+
+//go:generate mockery -name VCS -dir .. -output .
diff --git a/go/vcsinfo/types.go b/go/vcsinfo/types.go
index e6a980c..593ff23 100644
--- a/go/vcsinfo/types.go
+++ b/go/vcsinfo/types.go
@@ -83,7 +83,7 @@
 	// includes 'begin' and excludes 'end'.
 	Range(begin, end time.Time) []*IndexCommit
 
-	// IndexOf returns the index of the commit hash, where 0 is the index of the first hash.
+	// IndexOf returns the index of the commit hash, where 0 is the index of the first commit.
 	IndexOf(ctx context.Context, hash string) (int, error)
 
 	// ByIndex returns a LongCommit describing the commit
diff --git a/golden/go/digest_counter/digest_counter_test.go b/golden/go/digest_counter/digest_counter_test.go
index 0656410..58a2782 100644
--- a/golden/go/digest_counter/digest_counter_test.go
+++ b/golden/go/digest_counter/digest_counter_test.go
@@ -120,19 +120,17 @@
 	AlphaTest = types.TestName("test_alpha")
 	BetaTest  = types.TestName("test_beta")
 
-	x86TestAlphaTraceID = tiling.TraceId("x86:test_alpha:gm")
-	x64TestAlphaTraceID = tiling.TraceId("x86_64:test_alpha:image")
+	// TraceIDs are created like tracestore.TraceIDFromParams
+	x86TestAlphaTraceID = tiling.TraceId(",config=x86,source_type=test_alpha,name=gm")
+	x64TestAlphaTraceID = tiling.TraceId(",config=x86_64,source_type=test_alpha,name=image")
 
-	x64TestBetaTraceID = tiling.TraceId("x86_64:test_beta:image")
+	x64TestBetaTraceID = tiling.TraceId(",config=x86_64,source_type=test_beta,name=image")
 )
 
 func makePartialTileOne() *tiling.Tile {
 	return &tiling.Tile{
 		// Commits, Scale and Tile Index omitted (should not affect things)
-
 		Traces: map[tiling.TraceId]tiling.Trace{
-			// Reminder that the ids for the traces are created by concatenating
-			// all the values in alphabetical order of the keys.
 			x86TestAlphaTraceID: &types.GoldenTrace{
 				Digests: types.DigestSlice{FirstDigest, FirstDigest, SecondDigest},
 				Keys: map[string]string{
diff --git a/golden/go/goldingestion/goldingestion_test.go b/golden/go/goldingestion/goldingestion_test.go
index 137f093..e3d0bdf 100644
--- a/golden/go/goldingestion/goldingestion_test.go
+++ b/golden/go/goldingestion/goldingestion_test.go
@@ -41,6 +41,8 @@
 
 var (
 	// trace ids and values that are contained in the test file.
+	// These trace ids are the old format, new ones are like:
+	//   ,key1=value1,key2=value2
 	TEST_ENTRIES = []struct {
 		key   tiling.TraceId
 		value types.Digest
diff --git a/golden/go/indexer/indexer_test.go b/golden/go/indexer/indexer_test.go
index 094fc4e..9ee68e2 100644
--- a/golden/go/indexer/indexer_test.go
+++ b/golden/go/indexer/indexer_test.go
@@ -247,8 +247,8 @@
 func makeComplexTileWithCrosshatchIgnores() (types.ComplexTile, *tiling.Tile, *tiling.Tile) {
 	fullTile := data.MakeTestTile()
 	partialTile := data.MakeTestTile()
-	delete(partialTile.Traces, "crosshatch:test_alpha:gm")
-	delete(partialTile.Traces, "crosshatch:test_beta:gm")
+	delete(partialTile.Traces, ",device=crosshatch,name=test_alpha,source_type=gm,")
+	delete(partialTile.Traces, ",device=crosshatch,name=test_beta,source_type=gm,")
 
 	ct := types.NewComplexTile(fullTile)
 	ct.SetIgnoreRules(partialTile, []paramtools.ParamSet{
diff --git a/golden/go/paramsets/paramsets_test.go b/golden/go/paramsets/paramsets_test.go
index 2aa9930..03ad627 100644
--- a/golden/go/paramsets/paramsets_test.go
+++ b/golden/go/paramsets/paramsets_test.go
@@ -158,7 +158,7 @@
 		// Commits, Scale and Tile Index omitted (should not affect things)
 		Traces: map[tiling.TraceId]tiling.Trace{
 			// These trace ids have been shortened for test terseness.
-			// A real trace id would be like "8888:gm:foo"
+			// A real trace id would be like ",config=8888,source_type=gm,name=foo,"
 			"a": &types.GoldenTrace{
 				Digests: types.DigestSlice{DigestA, DigestB},
 				Keys: map[string]string{
diff --git a/golden/go/summary/summary_test.go b/golden/go/summary/summary_test.go
index 4b64d0c..1ef9692 100644
--- a/golden/go/summary/summary_test.go
+++ b/golden/go/summary/summary_test.go
@@ -86,7 +86,7 @@
 	tile := &tiling.Tile{
 		Traces: map[tiling.TraceId]tiling.Trace{
 			// These trace ids have been shortened for test terseness.
-			// A real trace id would be like "8888:gm:test_first"
+			// A real trace id would be like ",config=8888,source_type=gm,name=foo,"
 			"a": &types.GoldenTrace{
 				Digests: types.DigestSlice{"aaa", "bbb"},
 				Keys: map[string]string{
diff --git a/golden/go/testutils/data_three_devices/three_devices.go b/golden/go/testutils/data_three_devices/three_devices.go
index fa88386..1c281ebf 100644
--- a/golden/go/testutils/data_three_devices/three_devices.go
+++ b/golden/go/testutils/data_three_devices/three_devices.go
@@ -44,6 +44,10 @@
 
 	AlphaTest = types.TestName("test_alpha")
 	BetaTest  = types.TestName("test_beta")
+
+	AnglerDevice     = "angler"
+	BullheadDevice   = "bullhead"
+	CrosshatchDevice = "crosshatch"
 )
 
 func MakeTestBaseline() *baseline.Baseline {
@@ -95,63 +99,71 @@
 func MakeTestTile() *tiling.Tile {
 	return &tiling.Tile{
 		Commits:   MakeTestCommits(),
-		Scale:     1,
+		Scale:     0, // tile contains every data point.
 		TileIndex: 0,
 
 		Traces: map[tiling.TraceId]tiling.Trace{
-			// Reminder that the ids for the traces are created by concatenating
-			// all the values in alphabetical order of the keys.
-			"angler:test_alpha:gm": &types.GoldenTrace{
+			// Reminder that the ids for the traces are created using the
+			// logic in query.MakeKeyFast
+			",device=angler,name=test_alpha,source_type=gm,": &types.GoldenTrace{
 				Digests: types.DigestSlice{AlphaBad1Digest, AlphaBad1Digest, AlphaGood1Digest},
 				Keys: map[string]string{
-					"device":                "angler",
+					"device":                AnglerDevice,
 					types.PRIMARY_KEY_FIELD: string(AlphaTest),
 					types.CORPUS_FIELD:      "gm",
 				},
 			},
-			"angler:test_beta:gm": &types.GoldenTrace{
+			",device=angler,name=test_beta,source_type=gm,": &types.GoldenTrace{
 				Digests: types.DigestSlice{BetaGood1Digest, BetaGood1Digest, BetaGood1Digest},
 				Keys: map[string]string{
-					"device":                "angler",
+					"device":                AnglerDevice,
 					types.PRIMARY_KEY_FIELD: string(BetaTest),
 					types.CORPUS_FIELD:      "gm",
 				},
 			},
 
-			"bullhead:test_alpha:gm": &types.GoldenTrace{
+			",device=bullhead,name=test_alpha,source_type=gm,": &types.GoldenTrace{
 				Digests: types.DigestSlice{AlphaBad1Digest, AlphaBad1Digest, AlphaUntriaged1Digest},
 				Keys: map[string]string{
-					"device":                "bullhead",
+					"device":                BullheadDevice,
 					types.PRIMARY_KEY_FIELD: string(AlphaTest),
 					types.CORPUS_FIELD:      "gm",
 				},
 			},
-			"bullhead:test_beta:gm": &types.GoldenTrace{
+			",device=bullhead,name=test_beta,source_type=gm,": &types.GoldenTrace{
 				Digests: types.DigestSlice{BetaGood1Digest, BetaGood1Digest, BetaGood1Digest},
 				Keys: map[string]string{
-					"device":                "bullhead",
+					"device":                BullheadDevice,
 					types.PRIMARY_KEY_FIELD: string(BetaTest),
 					types.CORPUS_FIELD:      "gm",
 				},
 			},
 
-			"crosshatch:test_alpha:gm": &types.GoldenTrace{
+			",device=crosshatch,name=test_alpha,source_type=gm,": &types.GoldenTrace{
 				Digests: types.DigestSlice{AlphaBad1Digest, AlphaBad1Digest, AlphaGood1Digest},
 				Keys: map[string]string{
-					"device":                "crosshatch",
+					"device":                CrosshatchDevice,
 					types.PRIMARY_KEY_FIELD: string(AlphaTest),
 					types.CORPUS_FIELD:      "gm",
 				},
 			},
-			"crosshatch:test_beta:gm": &types.GoldenTrace{
+			",device=crosshatch,name=test_beta,source_type=gm,": &types.GoldenTrace{
 				Digests: types.DigestSlice{BetaUntriaged1Digest, types.MISSING_DIGEST, types.MISSING_DIGEST},
 				Keys: map[string]string{
-					"device":                "crosshatch",
+					"device":                CrosshatchDevice,
 					types.PRIMARY_KEY_FIELD: string(BetaTest),
 					types.CORPUS_FIELD:      "gm",
 				},
 			},
 		},
+
+		// Summarizes all the keys and values seen in this tile
+		// The values should be in alphabetical order (see paramset.Normalize())
+		ParamSet: map[string][]string{
+			"device":                {AnglerDevice, BullheadDevice, CrosshatchDevice},
+			types.PRIMARY_KEY_FIELD: {string(AlphaTest), string(BetaTest)},
+			types.CORPUS_FIELD:      {"gm"},
+		},
 	}
 }
 
diff --git a/golden/go/tracestore/bt_tracestore/.gitignore b/golden/go/tracestore/bt_tracestore/.gitignore
new file mode 100644
index 0000000..d383c56
--- /dev/null
+++ b/golden/go/tracestore/bt_tracestore/.gitignore
@@ -0,0 +1 @@
+testdata
diff --git a/golden/go/tracestore/bt_tracestore/BIGTABLE.md b/golden/go/tracestore/bt_tracestore/BIGTABLE.md
new file mode 100644
index 0000000..ac6cec6
--- /dev/null
+++ b/golden/go/tracestore/bt_tracestore/BIGTABLE.md
@@ -0,0 +1,209 @@
+Storing Gold traces in BigTable
+===============================
+
+This implementation is based on the original btts.go made for Perf.
+See perf/BIGTABLE.md for an overview of that schema.
+
+Note that we assume our Big Table instance is strongly consistent, which is
+only valid if there is no replication (which we shouldn't need for now).
+
+Data we want to store
+---------------------
+
+We need to store traces, which are arranged in a table.
+Traces have commits as their columns, a tiling.traceID as their rows, and
+a digest as their cells. These traceIDs are derived from the key:value pairs
+associated with the digest, e.g. {"os":"Android", "gpu": "NVidia1080"}
+
+Gold has an idea of a tile, which is a collection of N of these trace columns (i.e. commits).
+
+
+Mapping the data to BigTable
+----------------------------
+
+BigTable (BT) stores its data as one big table. Concretely, we have
+
+```
+     | col_1  | col_2  | col_3  | ...
+==================================
+row_1| [data] | [data] | [data] | ...
+row_2| [data] | [data] | [data] | ...
+...
+```
+
+For performance, BT provides [Column Families](https://cloud.google.com/bigtable/docs/schema-design#column_families_and_column_qualifiers)
+which group columns together so you can query some cells from a row belonging
+to that family w/o getting everything. As an analogy, if the whole table is a Sheets
+workbook, a column family can be thought of as a single spreadsheet in that workbook (at least,
+for our purposes).
+
+BT uses the term "Column Qualifier", which essentially just means "column name", because they
+can be arbitrary strings.
+
+It's tempting to just put the commit hashes as columns, the trace ids as the row names and store the
+digests in the cells, but this isn't ideal for these reasons:
+
+  1. traceIDs get really long with lots of params - BT has a limit of 4kb per
+     row name (aka a row key).
+  2. digests are long strings that are repeated a lot - if we stored them as int64s, it would save
+     a lot of data when we fetch a row (32 bytes per cell -> 8 bytes per cell).
+  3. BT has a soft limit of 100Mb of data per row.
+  4. BT will perform better if each row contains only the columns we are interested in
+    reading. Since we only really want to fetch the last few hundred commits, we would
+    like to limit each row to contain enough data for a single tile.
+    See also https://cloud.google.com/bigtable/docs/performance#slower-perf
+  5. We want to split our rows up further into shards, so we can execute multiple queries at once
+     (one query per shard).
+
+To address these performance issues, we need to store our traces and some auxiliary data.
+
+  1. An OrderedParamSet that can convert tiling.TraceId (long string) <->
+     EncodedTraceId (short string).
+  2. A DigestMap that can convert types.Digest (string) <-> DigestID (int64).
+  3. A monotonically increasing counter to derive more DigestIDs.
+
+These 3 "tables" along with the traces will be stored using 4 Column Families and can
+logically thought of being stored as independent "tables" or "spreadsheets" even
+though they are all stored in the "one big table".
+
+Row naming scheme
+-----------------
+There is a row naming scheme (for all 4 tables) as follows:
+
+    [shard]:[namespace]:[type]:[tile]:[subkey]
+
+Where shard and subkey are optional (can be ""). Some tables have tile with a constant value.
+"namespace" is constant for all tracestore data: "ts". Reminder that there's one table per Gold
+instance, so if we store other data to BT (e.g. expectations, tryjobs, etc) we need to have
+several unique namespaces.
+
+tile is a 0 padded 32 bit number (2^32-1) - [tile number].
+For example, tile 0 (the oldest commits) is number `2147483646`.
+Note that this reverses the order of the tiles, i.e. new tiles have
+smaller numbers, so that we can do a simple query to find the newest tile.
+
+BigTable can easily fetch rows starting with a given prefix, so this naming schema
+is set up to request things of a type for one or more tiles, with optional sharding.
+
+Note that sharding is a thing we have enabled by our choice of row names, not something
+given out for free by BigTable.
+
+Gold (and Perf) shards its data based on the subkey (conceptually subkey % 32)
+This makes the workload be spread more evenly, even when fetching only one tile.
+The shards come first in the naming convention to try to spread the rows across multiple
+tablets for better performance (rows on BT are stored on tablets sorted by the entire row name).
+
+Storing the OrderedParamSet for traceIDs
+----------------------------------------
+As mentioned above, traceIDs are derived from the paramtools.Params map of key-value pairs.
+We compress these maps using a paramstool.OrderedParamSet which concretely look like:
+
+    ,0=1,4=2,3=1,
+
+To do this compression/decompression, we need to store the OrderedParamSet (OPS).
+There is one OPS per tile. Conceptually, an OPS is stored like:
+```
+           |   OPS   |    H   |
+==============================
+ops_tile0 | [bytes] | [hash] |
+ops_tile1 | [bytes] | [hash] |
+...
+```
+
+The bytes stored in under the "OPS" column are just a gob encoded OrderedParamSet and
+the hash stored under the "H" column is the hash of the OPS, used to query the row when updating.
+
+As mentioned before, ops_tile0 is a conceptual simplification of the actual
+row name. Given the row naming schema, we define "type" for OPS to be "o" and
+let "shard" and "subkey" both be "".
+Thus, the actual row for tile 0 would be
+
+    :ts:o:2147483646:
+
+
+Storing the Digest Map
+----------------------
+We need to store a (potentially large) map of types.Digest (string) <-> DigestID (int64).
+This map is global across tiles. Conceptually, it is stored like:
+```
+        | [digest1] | [digest2] | [digest3] | ...
+====================================================
+map_000 |  [blank]  |  [int64]  |  [blank]  |
+map_001 |  [int64]  |  [blank]  |  [blank]  |
+...
+map_ffe |  [blank]  |  [blank]  |  [blank]  |
+map_fff |  [blank]  |  [blank]  |  [int64]  |
+...
+```
+
+Basically, we take a digest, chop off the first three characters, use those as
+the "subkey" in the row so we can make sure our rows don't exceed the maximum size.
+Given the soft row limit of 100Mb, a digest column + a cell is at most 40 bytes,
+which means a soft limit of 2.5 million digests per row. Splitting on 3 hex
+characters means we have 4096-way splitting, so a soft limit of 10 billion digests.
+
+Why not just have rows be the digests and a single column with the id? BT can only fetch
+so many rows per second (see https://cloud.google.com/bigtable/docs/performance#typical-workloads)
+If we had a million digests in the single column schema, one SSD node would take 100 seconds
+to request those million rows, where for the three character schema, it would take about .4 seconds.
+
+Given that we define "type" for the digest map to be "d", an
+example row for digest "92eb5ffee6ae2fec3ad71c777531578f"
+(assume this subkey is tied to shard 19) would be:
+
+    19:ts:d:0000000000:92e
+
+The ids of digests out start at 0 (for MISSING_DIGEST aka "") and increase monotonically.
+We manage these ids ourselves because using the autogenerated ids can lead to issues when
+migrating data from one table to another.
+
+Storing the Digest ID counter
+-----------------------------
+We assign newly seen digests an id that is simply an increasing int64. To store
+this int64 in BT, we essentially have a single cell dedicated to this:
+
+```
+               |    idc   |
+====================================================
+digest_counter |  [int64] |
+```
+We have one global id counter to go with the one global digestMap.
+
+The interfacing code will take care not to constantly update this value (BT frowns upon
+having very "hot" rows/cells) by requesting new ids in batches and saving them locally.
+
+There's only one row for the counter, which is:
+
+    :ts:i:0000000000:
+
+Storing the traces
+------------------
+
+With all the auxiliary data set, we can look at how the traces themselves are stored.
+Going with the default tile size of 256, the data would be like:
+
+```
+             | offset0 | offset1 | ... | offset255
+====================================================
+tile0_trace0 | [dig_1] | [dig_1] | ... | [dig_2] |
+tile0_trace1 | [dig_1] | [dig_2] | ... | [dig_2] |
+...
+tile0_traceN | [dig_8] | [blank] | ... | [dig_6] |
+tile1_trace0 | [blank] | [dig_2] | ... | [blank] |
+...
+```
+
+The columns are the offset into a tile of a commit. For example, the third commit in a repo would
+end up in tile 0, offset 3. The 1000th commit with a tile size of 256 would be in
+tile 3 (1000 / 256), offset 232 (1000 % 256). This has the effect of wrapping a given
+trace across many tiles.
+
+The rows follow the standard naming scheme, using "t" as "type", and making use of the shards
+(32 by default). The value for "subkey" is the encoded ParamSet (and from this subkey a shard
+is derived). An example row for encoded ParamSet ",0=1,1=3,3=0," on tile 0 (assume shard
+calculates to be 7) would be:
+
+    07:ts:t:2147483646:,0=1,1=3,3=0,
+
+The cells are the int64 ids of the digest that were drawn according to that ParamSet.
+Blank cells will be read as id 0, which is hard-coded to belong to MISSING_DIGEST ("").
diff --git a/golden/go/tracestore/bt_tracestore/bt_tracestore.go b/golden/go/tracestore/bt_tracestore/bt_tracestore.go
new file mode 100644
index 0000000..22f0069
--- /dev/null
+++ b/golden/go/tracestore/bt_tracestore/bt_tracestore.go
@@ -0,0 +1,992 @@
+// Package bt_tracestore implements a tracestore backed by BigTable
+// See BIGTABLE.md for an overview of the schema and design.
+package bt_tracestore
+
+import (
+	"context"
+	"encoding/binary"
+	"hash/crc32"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"cloud.google.com/go/bigtable"
+	"go.skia.org/infra/go/bt"
+	"go.skia.org/infra/go/metrics2"
+	"go.skia.org/infra/go/paramtools"
+	"go.skia.org/infra/go/skerr"
+	"go.skia.org/infra/go/sklog"
+	"go.skia.org/infra/go/tiling"
+	"go.skia.org/infra/go/util"
+	"go.skia.org/infra/go/vcsinfo"
+	"go.skia.org/infra/golden/go/tracestore"
+	"go.skia.org/infra/golden/go/types"
+	"golang.org/x/sync/errgroup"
+)
+
+// InitBT initializes the BT instance for the given configuration. It uses the default way
+// to get auth information from the environment and must be called with an account that has
+// admin rights.
+func InitBT(conf BTConfig) error {
+	return bt.InitBigtable(conf.ProjectID, conf.InstanceID, conf.TableID, btColumnFamilies)
+}
+
+// BTConfig contains the configuration information for the BigTable-based implementation of
+// TraceStore.
+type BTConfig struct {
+	ProjectID  string
+	InstanceID string
+	TableID    string
+	VCS        vcsinfo.VCS
+}
+
+// BTTraceStore implements the TraceStore interface.
+type BTTraceStore struct {
+	vcs    vcsinfo.VCS
+	client *bigtable.Client
+	table  *bigtable.Table
+
+	tileSize int32
+	shards   int32
+
+	// if cacheOps is true, then cache the OrderedParamSets between calls
+	// where possible.
+	cacheOps bool
+	// maps rowName (string) -> *OpsCacheEntry
+	opsCache sync.Map
+
+	availIDsMutex sync.Mutex
+	availIDs      []digestID
+}
+
+// New implements the TraceStore interface backed by BigTable. If cache is true,
+// the OrderedParamSets will be cached based on the row name.
+func New(ctx context.Context, conf BTConfig, cache bool) (*BTTraceStore, error) {
+	client, err := bigtable.NewClient(ctx, conf.ProjectID, conf.InstanceID)
+	if err != nil {
+		return nil, skerr.Fmt("could not instantiate client: %s", err)
+	}
+
+	ret := &BTTraceStore{
+		vcs:      conf.VCS,
+		client:   client,
+		tileSize: DefaultTileSize,
+		shards:   DefaultShards,
+		table:    client.Open(conf.TableID),
+		cacheOps: cache,
+		availIDs: []digestID{},
+	}
+	return ret, nil
+}
+
+// Put implements the TraceStore interface.
+func (b *BTTraceStore) Put(ctx context.Context, commitHash string, entries []*tracestore.Entry, ts time.Time) error {
+	defer metrics2.FuncTimer().Stop()
+	// if there are no entries this becomes a no-op.
+	if len(entries) == 0 {
+		return nil
+	}
+
+	// Accumulate all parameters into a paramset and collect all the digests.
+	paramSet := make(paramtools.ParamSet, len(entries[0].Params))
+	digestSet := make(types.DigestSet, len(entries))
+	for _, entry := range entries {
+		paramSet.AddParams(entry.Params)
+		digestSet[entry.Digest] = true
+	}
+
+	repoIndex, err := b.vcs.IndexOf(ctx, commitHash)
+	if err != nil {
+		return skerr.Fmt("could not look up commit %s: %s", commitHash, err)
+	}
+
+	// Find out what tile we need to fetch and what index into that tile we need.
+	// Reminder that tileKeys start at 2^32-1 and decrease in value.
+	tileKey, commitIndex := b.getTileKey(repoIndex)
+
+	// If these entries have any params we haven't seen before, we need to store those in BigTable.
+	ops, err := b.updateOrderedParamSet(ctx, tileKey, paramSet)
+	if err != nil {
+		sklog.Warningf("Bad paramset: %#v", paramSet)
+		return skerr.Fmt("cannot update paramset: %s", err)
+	}
+
+	// Similarly, if we have some new digests (almost certainly), we need to update
+	// the digestMap with them in there. Of note, we store this
+	// map of string (types.Digest) -> int64(DigestId) in big table, then refer to
+	// the DigestID elsewhere in the table. DigestIds are essentially a monotonically
+	// increasing arbitrary number.
+	digestMap, err := b.updateDigestMap(ctx, digestSet)
+	if err != nil {
+		sklog.Warningf("Bad digestSet: %#v", digestSet)
+		return skerr.Fmt("cannot update digest map: %s", err)
+	}
+
+	metrics2.GetInt64Metric("gold_digest_map_size").Update(int64(digestMap.Len()))
+
+	if len(digestMap.Delta(digestSet)) != 0 {
+		// Should never happen
+		return skerr.Fmt("delta should be empty at this point: %v", digestMap.Delta(digestSet))
+	}
+
+	// These are two parallel arrays. mutations[i] should be applied to rowNames[i] for all i.
+	rowNames, mutations, err := b.createPutMutations(entries, ts, tileKey, commitIndex, ops, digestMap)
+	if err != nil {
+		return skerr.Fmt("could not create mutations to put data: %s", err)
+	}
+
+	// Write the trace data. We pick a batchsize based on the assumption
+	// that the whole batch should be 2MB large and each entry is ~200 Bytes of data.
+	// 2MB / 200B = 10000. This is extremely conservative but should not be a problem
+	// since the batches are written in parallel.
+	return b.applyBulkBatched(ctx, rowNames, mutations, 10000)
+}
+
+// createPutMutations is a helper function that returns two parallel arrays of
+// the rows that need updating and the mutations to apply to those rows.
+// Specifically, the mutations will add the given entries to BT, clearing out
+// anything that was there previously.
+func (b *BTTraceStore) createPutMutations(entries []*tracestore.Entry, ts time.Time, tk tileKey, commitIndex int, ops *paramtools.OrderedParamSet, dm *digestMap) ([]string, []*bigtable.Mutation, error) {
+	// These mutations...
+	mutations := make([]*bigtable.Mutation, 0, len(entries))
+	// .. should be applied to these rows.
+	rowNames := make([]string, 0, len(entries))
+	btTS := bigtable.Time(ts)
+	before := bigtable.Time(ts.Add(-1 * time.Millisecond))
+
+	for _, entry := range entries {
+		// To save space, traceID isn't the long form tiling.TraceId
+		// (e.g. ,foo=bar,baz=gm,), it's a string of key-value numbers
+		// that refer to the params.(e.g. ,0=3,2=18,)
+		// See params.paramsEncoder
+		sTrace, err := ops.EncodeParamsAsString(entry.Params)
+		if err != nil {
+			return nil, nil, skerr.Fmt("invalid params: %s", err)
+		}
+		traceID := encodedTraceID(sTrace)
+
+		rowName := b.calcShardedRowName(tk, typeTrace, string(traceID))
+		rowNames = append(rowNames, rowName)
+
+		dID, err := dm.ID(entry.Digest)
+		if err != nil {
+			// this should never happen, the digest map should know about every digest already.
+			return nil, nil, skerr.Fmt("could not fetch id for digest %s: %s", entry.Digest, err)
+		}
+
+		// Create a mutation that puts the given digest at the given row
+		// (i.e. the trace combined with the tile), at the given column
+		// (i.e. the commit offset into this tile).
+		mut := bigtable.NewMutation()
+		column := strconv.Itoa(commitIndex)
+		dBytes, err := dID.MarshalBinary()
+		if err != nil {
+			// this should never happen, we are just marshalling an int to binary
+			return nil, nil, skerr.Fmt("could not encode digest id %d to bytes: %s", dID, err)
+		}
+		mut.Set(traceFamily, column, btTS, dBytes)
+		// Delete anything that existed at this cell before now.
+		mut.DeleteTimestampRange(traceFamily, column, 0, before)
+		mutations = append(mutations, mut)
+	}
+	return rowNames, mutations, nil
+}
+
+// GetTile implements the TraceStore interface.
+// Of note, due to this request possibly spanning over multiple tiles, the ParamsSet may have a
+// set of params that does not actually correspond to a trace (this shouldn't be a problem, but is
+// worth calling out). For example, suppose a trace with param " device=alpha" abruptly ends on
+// tile 4, commit 7 (where the device was removed from testing). If we are on tile 5 and need to
+// query both tile 4 starting at commit 10 and tile 5 (the whole thing), we'll just merge the
+// paramsets from both tiles, which includes the "device=alpha" params, but they don't exist in
+// any traces seen in the tile (since it ended prior to our cutoff point).
+func (b *BTTraceStore) GetTile(ctx context.Context, nCommits int) (*tiling.Tile, []*tiling.Commit, error) {
+	defer metrics2.FuncTimer().Stop()
+	// Look up the commits we need to query from BT
+	idxCommits := b.vcs.LastNIndex(nCommits)
+	if len(idxCommits) == 0 {
+		return nil, nil, skerr.Fmt("No commits found.")
+	}
+
+	// These commits could span across multiple tiles, so derive the tiles we need to query.
+	c := idxCommits[0]
+	startTileKey, startCommitIndex := b.getTileKey(c.Index)
+
+	c = idxCommits[len(idxCommits)-1]
+	endTileKey, endCommitIndex := b.getTileKey(c.Index)
+
+	var egroup errgroup.Group
+
+	var commits []*tiling.Commit
+	egroup.Go(func() error {
+		hashes := make([]string, 0, len(idxCommits))
+		for _, ic := range idxCommits {
+			hashes = append(hashes, ic.Hash)
+		}
+		var err error
+		commits, err = b.makeTileCommits(ctx, hashes)
+		if err != nil {
+			return skerr.Fmt("could not load tile commits: %s", err)
+		}
+		return nil
+	})
+
+	var traces traceMap
+	var params paramtools.ParamSet
+	egroup.Go(func() error {
+		var err error
+		traces, params, err = b.getTracesInRange(ctx, startTileKey, endTileKey, startCommitIndex, endCommitIndex)
+
+		if err != nil {
+			return skerr.Fmt("could not load tile commits: %s", err)
+		}
+		return nil
+	})
+
+	if err := egroup.Wait(); err != nil {
+		return nil, nil, skerr.Fmt("could not load last %d commits into tile: %s", nCommits, err)
+	}
+
+	ret := &tiling.Tile{
+		Traces:   traces,
+		ParamSet: params,
+		Commits:  commits,
+		Scale:    0,
+	}
+
+	return ret, commits, nil
+}
+
+// getTracesInRange returns a traceMap with data from the given start and stop points (tile and index).
+// It also includes the ParamSet for that range.
+func (b *BTTraceStore) getTracesInRange(ctx context.Context, startTileKey, endTileKey tileKey, startCommitIndex, endCommitIndex int) (traceMap, paramtools.ParamSet, error) {
+	// Query those tiles.
+	nTiles := int(startTileKey - endTileKey + 1)
+	nCommits := int(startTileKey-endTileKey)*int(b.tileSize) + (endCommitIndex - startCommitIndex) + 1
+	encTiles := make([]*encTile, nTiles)
+	var egroup errgroup.Group
+	tk := startTileKey
+	for idx := 0; idx < nTiles; idx++ {
+		func(idx int, tk tileKey) {
+			egroup.Go(func() error {
+				var err error
+				encTiles[idx], err = b.loadTile(ctx, tk)
+				if err != nil {
+					return skerr.Fmt("could not load tile with key %d to index %d: %s", tk, idx, err)
+				}
+				return nil
+			})
+		}(idx, tk)
+		tk--
+	}
+
+	var digestMap *digestMap
+	egroup.Go(func() error {
+		var err error
+		digestMap, err = b.getDigestMap(ctx)
+		if err != nil {
+			return skerr.Fmt("could not load digestMap: %s", err)
+		}
+		return nil
+	})
+
+	if err := egroup.Wait(); err != nil {
+		return nil, nil, skerr.Fmt("could not load %d tiles: %s", nTiles, err)
+	}
+
+	// This is the full tile we are going to return.
+	tileTraces := make(traceMap, len(encTiles[0].traces))
+	paramSet := paramtools.ParamSet{}
+
+	commitIDX := 0
+	for idx, encTile := range encTiles {
+		// Determine the offset within the tile that we should consider.
+		endOffset := int(b.tileSize - 1)
+		if idx == (len(encTiles) - 1) {
+			// If we are on the last tile, stop early (that is, at endCommitIndex)
+			endOffset = endCommitIndex
+		}
+		segLen := endOffset - startCommitIndex + 1
+
+		for encodedKey, encValues := range encTile.traces {
+			// at this point, the encodedKey looks like ,0=1,1=3,3=0,
+			// See params.paramsEncoder
+			params, err := encTile.ops.DecodeParamsFromString(string(encodedKey))
+			if err != nil {
+				sklog.Warningf("Incomplete OPS: %#v\n", encTile.ops)
+				return nil, nil, skerr.Fmt("corrupted trace key - could not decode %s: %s", encodedKey, err)
+			}
+
+			// Turn the params into the tiling.TraceId we expect elsewhere.
+			traceKey := tracestore.TraceIDFromParams(params)
+			if _, ok := tileTraces[traceKey]; !ok {
+				tileTraces[traceKey] = types.NewGoldenTraceN(nCommits)
+			}
+			gt := tileTraces[traceKey].(*types.GoldenTrace)
+			gt.Keys = params
+			// Build up the total set of params
+			paramSet.AddParams(params)
+
+			// Convert the digests from integer IDs to strings.
+			digestIDs := encValues[startCommitIndex : startCommitIndex+segLen]
+			digests, err := digestMap.DecodeIDs(digestIDs)
+			if err != nil {
+				return nil, nil, skerr.Fmt("corrupted digest id - could not decode: %s", err)
+			}
+			copy(gt.Digests[commitIDX:commitIDX+segLen], digests)
+		}
+
+		// After the first tile we always start at the first entry and advance the
+		// overall commit index by the segment length.
+		commitIDX += segLen
+		startCommitIndex = 0
+	}
+
+	// Sort the params for determinism.
+	paramSet.Normalize()
+
+	return tileTraces, paramSet, nil
+}
+
+// GetDenseTile implements the TraceStore interface. It fetches the most recent tile and sees if
+// there is enough non-empty data, then queries the next oldest tile until it has nCommits
+// non-empty commits.
+func (b *BTTraceStore) GetDenseTile(ctx context.Context, nCommits int) (*tiling.Tile, []*tiling.Commit, error) {
+	defer metrics2.FuncTimer().Stop()
+	// Figure out what index we are on.
+	idxCommits := b.vcs.LastNIndex(1)
+	if len(idxCommits) == 0 {
+		return nil, nil, skerr.Fmt("No commits found.")
+	}
+
+	c := idxCommits[0]
+	tKey, endIdx := b.getTileKey(c.Index)
+	tileStartCommitIdx := c.Index - endIdx
+
+	// commitsWithData is a slice of indexes of commits that have data. These indexes are
+	// relative to the repo itself, with index 0 being the first (oldest) commit in the repo.
+	commitsWithData := make([]int, 0, nCommits)
+	paramSet := paramtools.ParamSet{}
+	allTraces := traceMap{}
+
+	// Start at the most recent tile and step backwards until we have enough commits with data.
+	for {
+		traces, params, err := b.getTracesInRange(ctx, tKey, tKey, 0, endIdx)
+
+		if err != nil {
+			return nil, nil, skerr.Fmt("could not load commits from tile %d: %s", tKey, err)
+		}
+
+		paramSet.AddParamSet(params)
+		// filledCommits are the indexes in the traces that have data.
+		// That is, they are the indexes of commits in this tile.
+		// It will be sorted from low indexes to high indexes
+		filledCommits := traces.CommitIndicesWithData()
+
+		if len(filledCommits)+len(commitsWithData) > nCommits {
+			targetLength := nCommits - len(commitsWithData)
+			// trim filledCommits so we get to exactly nCommits
+			filledCommits = filledCommits[len(filledCommits)-targetLength:]
+		}
+
+		for _, tileIdx := range filledCommits {
+			commitsWithData = append(commitsWithData, tileStartCommitIdx+tileIdx)
+		}
+		cTraces := traces.MakeFromCommitIndexes(filledCommits)
+		allTraces.PrependTraces(cTraces)
+
+		if len(commitsWithData) >= nCommits || tKey == tileKeyFromIndex(0) {
+			break
+		}
+
+		tKey++                       // go backwards in time one tile
+		endIdx = int(b.tileSize - 1) // fetch the whole previous tile
+		tileStartCommitIdx -= int(b.tileSize)
+	}
+
+	if len(commitsWithData) == 0 {
+		return &tiling.Tile{}, nil, nil
+	}
+	// put them in oldest to newest order
+	sort.Ints(commitsWithData)
+
+	oldestIdx := commitsWithData[0]
+	oldestCommit, err := b.vcs.ByIndex(ctx, oldestIdx)
+	if err != nil {
+		return nil, nil, skerr.Fmt("invalid oldest index %d: %s", oldestIdx, err)
+	}
+	hashes := b.vcs.From(oldestCommit.Timestamp.Add(-1 * time.Millisecond))
+
+	// There's no guarantee that hashes[0] == oldestCommit[0] (e.g. two commits at same timestamp)
+	// So we trim hashes down if necessary
+	for i := 0; i < len(hashes); i++ {
+		if hashes[i] == oldestCommit.Hash {
+			hashes = hashes[i:]
+			break
+		}
+	}
+
+	allCommits, err := b.makeTileCommits(ctx, hashes)
+	if err != nil {
+		return nil, nil, skerr.Fmt("could not make tile commits: %s", err)
+	}
+
+	denseCommits := make([]*tiling.Commit, len(commitsWithData))
+	for i, idx := range commitsWithData {
+		denseCommits[i] = allCommits[idx-oldestIdx]
+	}
+
+	ret := &tiling.Tile{
+		Traces:   allTraces,
+		ParamSet: paramSet,
+		Commits:  denseCommits,
+		Scale:    0,
+	}
+	return ret, allCommits, nil
+}
+
+// getTileKey retrieves the tile key and the index of the commit in the given tile (commitIndex)
+// given the index of a commit in the repo (repoIndex).
+// commitIndex starts at 0 for the oldest commit in the tile.
+func (b *BTTraceStore) getTileKey(repoIndex int) (tileKey, int) {
+	tileIndex := int32(repoIndex) / b.tileSize
+	commitIndex := repoIndex % int(b.tileSize)
+	return tileKeyFromIndex(tileIndex), commitIndex
+}
+
+// loadTile returns an *encTile corresponding to the tileKey.
+func (b *BTTraceStore) loadTile(ctx context.Context, tileKey tileKey) (*encTile, error) {
+	defer metrics2.FuncTimer().Stop()
+	var egroup errgroup.Group
+
+	// Load the OrderedParamSet so the caller can decode the data from the tile.
+	var ops *paramtools.OrderedParamSet
+	egroup.Go(func() error {
+		opsEntry, _, err := b.getOPS(ctx, tileKey)
+		if err != nil {
+			return skerr.Fmt("could not load OPS: %s", err)
+		}
+		ops = opsEntry.ops
+		return nil
+	})
+
+	var traces map[encodedTraceID][]digestID
+	egroup.Go(func() error {
+		var err error
+		traces, err = b.loadEncodedTraces(ctx, tileKey)
+		if err != nil {
+			return skerr.Fmt("could not load traces: %s", err)
+		}
+		return nil
+	})
+
+	if err := egroup.Wait(); err != nil {
+		return nil, err
+	}
+
+	return &encTile{
+		ops:    ops,
+		traces: traces,
+	}, nil
+}
+
+// loadEncodedTraces returns all traces belonging to the given tileKey.
+// As outlined in BIGTABLE.md, the trace ids and the digest ids they
+// map to are in an encoded form and will need to be expanded prior to use.
+func (b *BTTraceStore) loadEncodedTraces(ctx context.Context, tileKey tileKey) (map[encodedTraceID][]digestID, error) {
+	defer metrics2.FuncTimer().Stop()
+	var egroup errgroup.Group
+	shardResults := make([]map[encodedTraceID][]digestID, b.shards)
+	traceCount := int64(0)
+
+	// Query all shards in parallel.
+	for shard := int32(0); shard < b.shards; shard++ {
+		func(shard int32) {
+			egroup.Go(func() error {
+				// This prefix will match all traces belonging to the
+				// current shard in the current tile.
+				prefixRange := bigtable.PrefixRange(shardedRowName(shard, typeTrace, tileKey, ""))
+				target := map[encodedTraceID][]digestID{}
+				shardResults[shard] = target
+				var parseErr error
+				err := b.table.ReadRows(ctx, prefixRange, func(row bigtable.Row) bool {
+					// The encoded trace id is the "subkey" part of the row name.
+					traceKey := encodedTraceID(extractSubkey(row.Key()))
+					// If this is the first time we've seen the trace, initialize the
+					// slice of digest ids for it.
+					if _, ok := target[traceKey]; !ok {
+						target[traceKey] = make([]digestID, b.tileSize)
+						atomic.AddInt64(&traceCount, 1)
+					}
+
+					for _, col := range row[traceFamily] {
+						// The columns are something like T:35 where the part
+						// after the colon is the commitIndex i.e. the index
+						// of this commit in the current tile.
+						idx, err := strconv.Atoi(strings.TrimPrefix(col.Column, traceFamilyPrefix))
+						if err != nil {
+							// Should never happen
+							parseErr = err
+							return false
+						}
+						var dID digestID
+						if err := dID.UnmarshalBinary(col.Value); err != nil {
+							// This should never happen
+							parseErr = err
+							return false
+						}
+						if idx < 0 || idx >= int(b.tileSize) {
+							// This would happen if the tile size changed from a past
+							// value. It shouldn't be changed, even if the Gold tile size
+							// (n_commits) changes.
+							parseErr = skerr.Fmt("got index %d that is outside of the target slice of length %d", idx, len(target))
+							return false
+						}
+						target[traceKey][idx] = dID
+					}
+					return true
+				}, bigtable.RowFilter(bigtable.LatestNFilter(1)))
+				if err != nil {
+					return skerr.Fmt("could not read rows: %s", err)
+				}
+				return parseErr
+			})
+		}(shard)
+	}
+
+	if err := egroup.Wait(); err != nil {
+		return nil, err
+	}
+
+	// Merge all the results together
+	ret := make(map[encodedTraceID][]digestID, traceCount)
+	for _, r := range shardResults {
+		for traceKey, digestIDs := range r {
+			// different shards should never share results for a tracekey
+			// since a trace always maps to the same shard.
+			ret[traceKey] = digestIDs
+		}
+	}
+
+	return ret, nil
+}
+
+// applyBulkBatched writes the given rowNames/mutation pairs to BigTable in batches that are
+// maximally of size 'batchSize'. The batches are written in parallel.
+func (b *BTTraceStore) applyBulkBatched(ctx context.Context, rowNames []string, mutations []*bigtable.Mutation, batchSize int) error {
+
+	var egroup errgroup.Group
+	err := util.ChunkIter(len(rowNames), batchSize, func(chunkStart, chunkEnd int) error {
+		egroup.Go(func() error {
+			tctx, cancel := context.WithTimeout(ctx, writeTimeout)
+			defer cancel()
+			rowNames := rowNames[chunkStart:chunkEnd]
+			mutations := mutations[chunkStart:chunkEnd]
+			errs, err := b.table.ApplyBulk(tctx, rowNames, mutations)
+			if err != nil {
+				return skerr.Fmt("error writing batch [%d:%d]: %s", chunkStart, chunkEnd, err)
+			}
+			if errs != nil {
+				return skerr.Fmt("error writing some portions of batch [%d:%d]: %s", chunkStart, chunkEnd, errs)
+			}
+			return nil
+		})
+		return nil
+	})
+	if err != nil {
+		return skerr.Fmt("error running ChunkIter: %s", err)
+	}
+	return egroup.Wait()
+}
+
+// calcShardedRowName deterministically assigns a shard for the given subkey (e.g. traceID)
+// Once this is done, the shard, rowtype, tileKey and the subkey are combined into a
+// single string to be used as a row name in BT.
+func (b *BTTraceStore) calcShardedRowName(tileKey tileKey, rowType, subkey string) string {
+	shard := int32(crc32.ChecksumIEEE([]byte(subkey)) % uint32(b.shards))
+	return shardedRowName(shard, rowType, tileKey, subkey)
+}
+
+// To avoid having one monolithic row, we take the first three characters of the digest
+// and use it as a subkey in the row. Then, what remains is used as the column name.
+// In practice this means our digests will be split using three hexadecimal characters, so
+// we will have 16^3 = 4096 rows for our digest map.
+func (b *BTTraceStore) rowAndColNameFromDigest(digest types.Digest) (string, string) {
+	subkey := string(digest[:3])
+	colName := string(digest[3:])
+	return b.calcShardedRowName(digestMapTile, typeDigestMap, subkey), colName
+}
+
+// getDigestMap gets the global (i.e. same for all tiles) digestMap.
+func (b *BTTraceStore) getDigestMap(ctx context.Context) (*digestMap, error) {
+	defer metrics2.FuncTimer().Stop()
+	// Query all shards in parallel.
+	var egroup errgroup.Group
+	shardResults := make([]map[types.Digest]digestID, b.shards)
+	total := int64(0)
+	for shard := int32(0); shard < b.shards; shard++ {
+		func(shard int32) {
+			egroup.Go(func() error {
+				prefRange := bigtable.PrefixRange(shardedRowName(shard, typeDigestMap, digestMapTile, ""))
+				var idx int64
+				var parseErr error = nil
+				ret := map[types.Digest]digestID{}
+				err := b.table.ReadRows(ctx, prefRange, func(row bigtable.Row) bool {
+					digestPrefix := extractSubkey(row.Key())
+					for _, col := range row[digestMapFamily] {
+						idx, parseErr = strconv.ParseInt(string(col.Value), 10, 64)
+						if parseErr != nil {
+							// Should never happen
+							return false
+						}
+						digest := types.Digest(digestPrefix + strings.TrimPrefix(col.Column, digestMapFamilyPrefix))
+						ret[digest] = digestID(idx)
+					}
+					return true
+				}, bigtable.RowFilter(bigtable.LatestNFilter(1)))
+
+				if err != nil {
+					return skerr.Fmt("problem fetching shard %d of digestmap: %s", shard, err)
+				}
+				if parseErr != nil {
+					return parseErr
+				}
+
+				shardResults[shard] = ret
+				atomic.AddInt64(&total, int64(len(ret)))
+				return nil
+			})
+		}(shard)
+	}
+	if err := egroup.Wait(); err != nil {
+		return nil, skerr.Fmt("problem fetching digestmap: %s", err)
+	}
+
+	ret := newDigestMap(int(total))
+	for _, dm := range shardResults {
+		if err := ret.Add(dm); err != nil {
+			// put the digest map latter in case it gets truncated
+			return nil, skerr.Fmt("could not build DigestMap: %s \nresults %#v", err, dm)
+		}
+	}
+	return ret, nil
+}
+
+// getIDs returns a []DigestID of length n where each of the
+// digestIDs are unique (even between processes).
+func (b *BTTraceStore) getIDs(ctx context.Context, n int) ([]digestID, error) {
+	defer metrics2.FuncTimer().Stop()
+	// Extract up to n ids from those we have already cached.
+	b.availIDsMutex.Lock()
+	defer b.availIDsMutex.Unlock()
+	toExtract := util.MinInt(len(b.availIDs), n)
+
+	ids := make([]digestID, 0, n)
+	ids = append(ids, b.availIDs[:toExtract]...)
+	b.availIDs = b.availIDs[toExtract:]
+
+	// missing is how many ids we are short
+	missing := int64(n - len(ids))
+	if missing == 0 {
+		return ids, nil
+	}
+	// For performance reasons, make a few big requests for ids instead of many small ones.
+	// That is, always request numReservedIds extra.
+	toRequest := missing + numReservedIds
+	// Reserve new IDs via the ID counter
+	rmw := bigtable.NewReadModifyWrite()
+	rmw.Increment(idCounterFamily, idCounterColumn, toRequest)
+	row, err := b.table.ApplyReadModifyWrite(ctx, idCounterRow, rmw)
+	if err != nil {
+		return nil, skerr.Fmt("could not fetch counter from BT: %s", err)
+	}
+
+	// ri are the cells in Row of the given counter family
+	// This should be 1 cell belonging to 1 column.
+	ri, ok := row[idCounterFamily]
+	if !ok {
+		// should never happen
+		return nil, skerr.Fmt("malformed response - no id counter family: %#v", ri)
+	}
+	if len(ri) != 1 {
+		// should never happen
+		return nil, skerr.Fmt("malformed response - expected 1 cell: %#v", ri)
+	}
+
+	maxID := digestID(binary.BigEndian.Uint64(ri[0].Value))
+
+	lastID := maxID - digestID(toRequest)
+	// ID of 0 is a special case - it's already assigned to MISSING_DIGEST, so skip it.
+	if lastID == missingDigestID {
+		lastID++
+	}
+	for i := lastID; i < maxID; i++ {
+		// Give the first ids to the current allocation request...
+		if missing > 0 {
+			ids = append(ids, i)
+		} else {
+			// ... and put the remainder in the store for later.
+			b.availIDs = append(b.availIDs, i)
+		}
+		missing--
+	}
+
+	return ids, nil
+}
+
+// returnIDs can be called with a []DigestID of ids that were not actually
+// assigned to digests. This allows them to be used by future requests to
+// getIDs.
+func (b *BTTraceStore) returnIDs(unusedIDs []digestID) {
+	b.availIDsMutex.Lock()
+	defer b.availIDsMutex.Unlock()
+	b.availIDs = append(b.availIDs, unusedIDs...)
+}
+
+// getOrAddDigests fills the given digestMap with the given digests
+// assigned to a DigestID if they don't already have an assignment.
+// This is a helper function for updateDigestMap
+// TODO(kjlubick): This currently makes a lot of requests to BT -
+// Should there be some caching done here to prevent that?
+func (b *BTTraceStore) getOrAddDigests(ctx context.Context, digests []types.Digest, digestMap *digestMap) (*digestMap, error) {
+	defer metrics2.FuncTimer().Stop()
+	availIDs, err := b.getIDs(ctx, len(digests))
+	if err != nil {
+		return nil, err
+	}
+
+	now := bigtable.Time(time.Now())
+	newIDMapping := make(map[types.Digest]digestID, len(digests))
+	unusedIDs := make([]digestID, 0, len(availIDs))
+	for idx, digest := range digests {
+		idVal := availIDs[idx]
+		if _, err := digestMap.ID(digest); err == nil {
+			// digestMap already has a mapping for this digest, no need to check
+			// if BT has seen it yet (because it has).
+			// Should never happen because we we've already done this check in updateDigestMap.
+			unusedIDs = append(unusedIDs, idVal)
+			continue
+		}
+		rowName, colName := b.rowAndColNameFromDigest(digest)
+		// This mutation says "Add an entry to the map for digest -> idVal iff
+		// the digest doesn't already have a mapping".
+		addMut := bigtable.NewMutation()
+		addMut.Set(digestMapFamily, colName, now, []byte(strconv.FormatInt(int64(idVal), 10)))
+		filter := bigtable.ColumnFilter(colName)
+		// Note that we only add the value if filter is false, i.e. the column does not
+		// already exist.
+		condMut := bigtable.NewCondMutation(filter, nil, addMut)
+		var digestAlreadyHadId bool
+		if err := b.table.Apply(ctx, rowName, condMut, bigtable.GetCondMutationResult(&digestAlreadyHadId)); err != nil {
+			return nil, skerr.Fmt("could not check if row %s col %s already had a DigestID: %s", rowName, colName, err)
+		}
+
+		// We didn't need this ID so let's re-use it later.
+		if digestAlreadyHadId {
+			unusedIDs = append(unusedIDs, idVal)
+		} else {
+			newIDMapping[digest] = idVal
+		}
+	}
+
+	// If all ids were added to BT, then we know our newIDMapping can simply be added
+	// to what we already have, since there were no collisions between digests and what
+	// was in the table already.
+	if len(unusedIDs) == 0 {
+		if err := digestMap.Add(newIDMapping); err != nil {
+			return nil, err
+		}
+		return digestMap, nil
+	}
+	// At this point, some of the digests already had ids, so we should reload
+	// the entire digestMap to make sure we have the full picture.
+	// TODO(kjlubick): Can we not just add what new ones we saw to what we already have?
+
+	// Return the unused IDs for later use.
+	b.returnIDs(unusedIDs)
+	return b.getDigestMap(ctx)
+}
+
+// updateDigestMap returns the current global DigestMap after making sure the given
+// digests are a part of it.
+func (b *BTTraceStore) updateDigestMap(ctx context.Context, digests types.DigestSet) (*digestMap, error) {
+	defer metrics2.FuncTimer().Stop()
+	// Load the digest map from BT.
+	// TODO(kjlubick): should we cache this map and first check to see if the digests
+	// are all in there?
+	digestMap, err := b.getDigestMap(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	delta := digestMap.Delta(digests)
+	if len(delta) == 0 {
+		return digestMap, nil
+	}
+
+	return b.getOrAddDigests(ctx, delta, digestMap)
+}
+
+// Copied from btts.go in infra/perf
+
+// UpdateOrderedParamSet will add all params from 'p' to the OrderedParamSet
+// for 'tileKey' and write it back to BigTable.
+func (b *BTTraceStore) updateOrderedParamSet(ctx context.Context, tileKey tileKey, p paramtools.ParamSet) (*paramtools.OrderedParamSet, error) {
+	defer metrics2.FuncTimer().Stop()
+
+	tctx, cancel := context.WithTimeout(ctx, writeTimeout)
+	defer cancel()
+	var newEntry *opsCacheEntry
+	for {
+		// Get OPS.
+		entry, existsInBT, err := b.getOPS(ctx, tileKey)
+		if err != nil {
+			return nil, skerr.Fmt("failed to get OPS: %s", err)
+		}
+
+		// If the OPS contains our paramset then we're done.
+		if delta := entry.ops.Delta(p); len(delta) == 0 {
+			return entry.ops, nil
+		}
+
+		// Create a new updated ops.
+		ops := entry.ops.Copy()
+		ops.Update(p)
+		newEntry, err = opsCacheEntryFromOPS(ops)
+		if err != nil {
+			return nil, skerr.Fmt("failed to create cache entry: %s", err)
+		}
+		encodedOps, err := newEntry.ops.Encode()
+		if err != nil {
+			return nil, skerr.Fmt("failed to encode new ops: %s", err)
+		}
+
+		now := bigtable.Time(time.Now())
+		condTrue := false
+		if existsInBT {
+			// Create an update that avoids the lost update problem.
+			cond := bigtable.ChainFilters(
+				bigtable.LatestNFilter(1),
+				bigtable.FamilyFilter(opsFamily),
+				bigtable.ColumnFilter(opsHashColumn),
+				bigtable.ValueFilter(string(entry.hash)),
+			)
+			updateMutation := bigtable.NewMutation()
+			updateMutation.Set(opsFamily, opsHashColumn, now, []byte(newEntry.hash))
+			updateMutation.Set(opsFamily, opsOpsColumn, now, encodedOps)
+
+			// Add a mutation that cleans up old versions.
+			before := bigtable.Time(now.Time().Add(-1 * time.Second))
+			updateMutation.DeleteTimestampRange(opsFamily, opsHashColumn, 0, before)
+			updateMutation.DeleteTimestampRange(opsFamily, opsOpsColumn, 0, before)
+			condUpdate := bigtable.NewCondMutation(cond, updateMutation, nil)
+
+			if err := b.table.Apply(tctx, tileKey.OpsRowName(), condUpdate, bigtable.GetCondMutationResult(&condTrue)); err != nil {
+				sklog.Warningf("Failed to apply: %s", err)
+				return nil, err
+			}
+
+			// If !condTrue then we need to try again,
+			// and clear our local cache.
+			if !condTrue {
+				sklog.Warningf("Exists !condTrue - clearing cache and trying again.")
+				b.opsCache.Delete(tileKey.OpsRowName())
+				continue
+			}
+		} else {
+			// Create an update that only works if the ops entry doesn't exist yet.
+			// I.e. only apply the mutation if the HASH column doesn't exist for this row.
+			cond := bigtable.ChainFilters(
+				bigtable.FamilyFilter(opsFamily),
+				bigtable.ColumnFilter(opsHashColumn),
+			)
+			updateMutation := bigtable.NewMutation()
+			updateMutation.Set(opsFamily, opsHashColumn, now, []byte(newEntry.hash))
+			updateMutation.Set(opsFamily, opsOpsColumn, now, encodedOps)
+
+			condUpdate := bigtable.NewCondMutation(cond, nil, updateMutation)
+			if err := b.table.Apply(tctx, tileKey.OpsRowName(), condUpdate, bigtable.GetCondMutationResult(&condTrue)); err != nil {
+				sklog.Warningf("Failed to apply: %s", err)
+				// clear cache and try again
+				b.opsCache.Delete(tileKey.OpsRowName())
+				continue
+			}
+
+			// If condTrue then we need to try again,
+			// and clear our local cache.
+			if condTrue {
+				sklog.Warningf("First Write condTrue - clearing cache and trying again.")
+				b.opsCache.Delete(tileKey.OpsRowName())
+				continue
+			}
+		}
+
+		// Successfully wrote OPS, so update the cache.
+		if b.cacheOps {
+			b.opsCache.Store(tileKey.OpsRowName(), newEntry)
+		}
+		break
+	}
+	return newEntry.ops, nil
+}
+
+// getOps returns the OpsCacheEntry for a given tile.
+//
+// Note that it will create a new OpsCacheEntry if none exists.
+//
+// getOps returns false if the OPS in BT was empty, true otherwise (even if cached).
+func (b *BTTraceStore) getOPS(ctx context.Context, tileKey tileKey) (*opsCacheEntry, bool, error) {
+	defer metrics2.FuncTimer().Stop()
+	if b.cacheOps {
+		entry, ok := b.opsCache.Load(tileKey.OpsRowName())
+		if ok {
+			return entry.(*opsCacheEntry), true, nil
+		}
+	}
+	tctx, cancel := context.WithTimeout(ctx, readTimeout)
+	defer cancel()
+	row, err := b.table.ReadRow(tctx, tileKey.OpsRowName(), bigtable.RowFilter(bigtable.LatestNFilter(1)))
+	if err != nil {
+		return nil, false, skerr.Fmt("failed to read OPS from BigTable for %s: %s", tileKey.OpsRowName(), err)
+	}
+	// If there is no entry in BigTable then return an empty OPS.
+	if len(row) == 0 {
+		sklog.Warningf("Failed to read OPS from BT for %s.", tileKey.OpsRowName())
+		entry, err := newOpsCacheEntry()
+		return entry, false, err
+	}
+	entry, err := newOpsCacheEntryFromRow(row)
+	if err == nil && b.cacheOps {
+		b.opsCache.Store(tileKey.OpsRowName(), entry)
+	}
+	return entry, true, err
+}
+
+// makeTileCommits creates a slice of tiling.Commit from the given git hashes.
+// Specifically, we need to look up the details to get the author information.
+func (b *BTTraceStore) makeTileCommits(ctx context.Context, hashes []string) ([]*tiling.Commit, error) {
+	longCommits, err := b.vcs.DetailsMulti(ctx, hashes, false)
+	if err != nil {
+		// put hashes second in case they get truncated for being quite long.
+		return nil, skerr.Fmt("could not fetch commit data for commits %s (hashes: %q)", err, hashes)
+	}
+
+	commits := make([]*tiling.Commit, len(hashes))
+	for i, lc := range longCommits {
+		if lc == nil {
+			return nil, skerr.Fmt("commit %s not found from VCS", hashes[i])
+		}
+		commits[i] = &tiling.Commit{
+			Hash:       lc.Hash,
+			Author:     lc.Author,
+			CommitTime: lc.Timestamp.Unix(),
+		}
+	}
+	return commits, nil
+}
+
+// Make sure BTTraceStore fulfills the TraceStore Interface
+var _ tracestore.TraceStore = (*BTTraceStore)(nil)
diff --git a/golden/go/tracestore/bt_tracestore/bt_tracestore_test.go b/golden/go/tracestore/bt_tracestore/bt_tracestore_test.go
new file mode 100644
index 0000000..56f37cd
--- /dev/null
+++ b/golden/go/tracestore/bt_tracestore/bt_tracestore_test.go
@@ -0,0 +1,990 @@
+package bt_tracestore
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/mock"
+	assert "github.com/stretchr/testify/require"
+	"go.skia.org/infra/go/bt"
+	"go.skia.org/infra/go/deepequal"
+	"go.skia.org/infra/go/fileutil"
+	"go.skia.org/infra/go/gcs/gcs_testutils"
+	"go.skia.org/infra/go/sktest"
+	"go.skia.org/infra/go/testutils/unittest"
+	"go.skia.org/infra/go/tiling"
+	"go.skia.org/infra/go/util"
+	"go.skia.org/infra/go/vcsinfo"
+	mock_vcs "go.skia.org/infra/go/vcsinfo/mocks"
+	"go.skia.org/infra/golden/go/serialize"
+	data "go.skia.org/infra/golden/go/testutils/data_three_devices"
+	"go.skia.org/infra/golden/go/tracestore"
+	"go.skia.org/infra/golden/go/types"
+)
+
+// TestBTTraceStorePutGet adds a bunch of entries one at a time and
+// then retrieves the full tile.
+func TestBTTraceStorePutGet(t *testing.T) {
+	unittest.LargeTest(t)
+	unittest.RequiresBigTableEmulator(t)
+
+	commits := data.MakeTestCommits()
+	mvcs := MockVCSWithCommits(commits, 0)
+	defer mvcs.AssertExpectations(t)
+
+	btConf := BTConfig{
+		ProjectID:  "should-use-the-emulator",
+		InstanceID: "testinstance",
+		TableID:    "three_devices_test",
+		VCS:        mvcs,
+	}
+
+	assert.NoError(t, bt.DeleteTables(btConf.ProjectID, btConf.InstanceID, btConf.TableID))
+	assert.NoError(t, InitBT(btConf))
+
+	ctx := context.Background()
+	traceStore, err := New(ctx, btConf, true)
+	assert.NoError(t, err)
+
+	// With no data, we should get an empty tile
+	actualTile, _, err := traceStore.GetTile(ctx, len(commits))
+	assert.NoError(t, err)
+	assert.NotNil(t, actualTile)
+	assert.Empty(t, actualTile.Traces)
+
+	// This time is an arbitrary point in time
+	now := time.Date(2019, time.May, 5, 1, 3, 4, 0, time.UTC)
+
+	// Build a tile up from the individual data points, one at a time
+	traces := data.MakeTestTile().Traces
+	for _, trace := range traces {
+		gTrace, ok := trace.(*types.GoldenTrace)
+		assert.True(t, ok)
+
+		// Put them in backwards, just to test that order doesn't matter
+		for i := len(gTrace.Digests) - 1; i >= 0; i-- {
+			e := tracestore.Entry{
+				Digest: gTrace.Digests[i],
+				Params: gTrace.Keys,
+			}
+			err := traceStore.Put(ctx, commits[i].Hash, []*tracestore.Entry{&e}, now)
+			assert.NoError(t, err)
+			// roll forward the clock by an arbitrary amount of time
+			now = now.Add(7 * time.Second)
+		}
+	}
+
+	// Get the tile back and make sure it exactly matches the tile
+	// we hand-crafted for the test data.
+	actualTile, actualCommits, err := traceStore.GetTile(ctx, len(commits))
+	assert.NoError(t, err)
+
+	assert.Equal(t, data.MakeTestTile(), actualTile)
+	assert.Equal(t, commits, actualCommits)
+}
+
+// TestBTTraceStorePutGetSpanTile is like TestBTTraceStorePutGet except the 3 commits
+// are lined up to go across two tiles.
+func TestBTTraceStorePutGetSpanTile(t *testing.T) {
+	unittest.LargeTest(t)
+	unittest.RequiresBigTableEmulator(t)
+
+	commits := data.MakeTestCommits()
+	mvcs := MockVCSWithCommits(commits, DefaultTileSize-2)
+	defer mvcs.AssertExpectations(t)
+
+	btConf := BTConfig{
+		ProjectID:  "should-use-the-emulator",
+		InstanceID: "testinstance",
+		TableID:    "three_devices_test_span",
+		VCS:        mvcs,
+	}
+
+	assert.NoError(t, bt.DeleteTables(btConf.ProjectID, btConf.InstanceID, btConf.TableID))
+	assert.NoError(t, InitBT(btConf))
+
+	ctx := context.Background()
+	traceStore, err := New(ctx, btConf, true)
+	assert.NoError(t, err)
+
+	// With no data, we should get an empty tile
+	actualTile, _, err := traceStore.GetTile(ctx, len(commits))
+	assert.NoError(t, err)
+	assert.NotNil(t, actualTile)
+	assert.Empty(t, actualTile.Traces)
+
+	// This time is an arbitrary point in time
+	now := time.Date(2019, time.May, 5, 1, 3, 4, 0, time.UTC)
+
+	// Build a tile up from the individual data points, one at a time
+	traces := data.MakeTestTile().Traces
+	for _, trace := range traces {
+		gTrace, ok := trace.(*types.GoldenTrace)
+		assert.True(t, ok)
+
+		// Put them in backwards, just to test that order doesn't matter
+		for i := len(gTrace.Digests) - 1; i >= 0; i-- {
+			e := tracestore.Entry{
+				Digest: gTrace.Digests[i],
+				Params: gTrace.Keys,
+			}
+			err := traceStore.Put(ctx, commits[i].Hash, []*tracestore.Entry{&e}, now)
+			assert.NoError(t, err)
+			// roll forward the clock by an arbitrary amount of time
+			now = now.Add(7 * time.Second)
+		}
+	}
+
+	// Get the tile back and make sure it exactly matches the tile
+	// we hand-crafted for the test data.
+	actualTile, actualCommits, err := traceStore.GetTile(ctx, len(commits))
+	assert.NoError(t, err)
+
+	assert.Equal(t, data.MakeTestTile(), actualTile)
+	assert.Equal(t, commits, actualCommits)
+}
+
+// TestBTTraceStorePutGetGrouped adds a bunch of entries batched by device and
+// then retrieves the full Tile.
+func TestBTTraceStorePutGetGrouped(t *testing.T) {
+	unittest.LargeTest(t)
+	unittest.RequiresBigTableEmulator(t)
+
+	commits := data.MakeTestCommits()
+	mvcs := MockVCSWithCommits(commits, 0)
+	defer mvcs.AssertExpectations(t)
+
+	btConf := BTConfig{
+		ProjectID:  "should-use-the-emulator",
+		InstanceID: "testinstance",
+		TableID:    "three_devices_test_grouped",
+		VCS:        mvcs,
+	}
+
+	assert.NoError(t, bt.DeleteTables(btConf.ProjectID, btConf.InstanceID, btConf.TableID))
+	assert.NoError(t, InitBT(btConf))
+
+	ctx := context.Background()
+	traceStore, err := New(ctx, btConf, true)
+	assert.NoError(t, err)
+
+	// With no data, we should get an empty tile
+	actualTile, _, err := traceStore.GetTile(ctx, len(commits))
+	assert.NoError(t, err)
+	assert.NotNil(t, actualTile)
+	assert.Empty(t, actualTile.Traces)
+
+	// Build a tile up from the individual data points, one at a time
+	now := time.Date(2019, time.May, 5, 1, 3, 4, 0, time.UTC)
+
+	// Group the traces by device, so we should have 3 groups of 2 traces.
+	traces := data.MakeTestTile().Traces
+	byDevice := map[string][]*types.GoldenTrace{
+		data.AnglerDevice:     nil,
+		data.BullheadDevice:   nil,
+		data.CrosshatchDevice: nil,
+	}
+	for _, trace := range traces {
+		gTrace, ok := trace.(*types.GoldenTrace)
+		assert.True(t, ok)
+		assert.Len(t, gTrace.Digests, len(commits), "test data should have one digest per commit")
+		dev := gTrace.Keys["device"]
+		byDevice[dev] = append(byDevice[dev], gTrace)
+	}
+	assert.Len(t, byDevice, 3, "test data should have exactly 3 devices")
+
+	// for each trace, report a group of two digests for each commit.
+	for dev, gTraces := range byDevice {
+		assert.Len(t, gTraces, 2, "test data for %s should have exactly 2 traces", dev)
+
+		for i := 0; i < len(commits); i++ {
+			var entries []*tracestore.Entry
+			for _, gTrace := range gTraces {
+				entries = append(entries, &tracestore.Entry{
+					Digest: gTrace.Digests[i],
+					Params: gTrace.Keys,
+				})
+			}
+
+			err = traceStore.Put(ctx, commits[i].Hash, entries, now)
+			assert.NoError(t, err)
+			// roll forward the clock by an arbitrary amount of time
+			now = now.Add(3 * time.Minute)
+		}
+	}
+
+	// Get the tile back and make sure it exactly matches the tile
+	// we hand-crafted for the test data.
+	actualTile, actualCommits, err := traceStore.GetTile(ctx, len(commits))
+	assert.NoError(t, err)
+
+	assert.Equal(t, data.MakeTestTile(), actualTile)
+	assert.Equal(t, commits, actualCommits)
+}
+
+// TestBTTraceStorePutGetThreaded is like TestBTTraceStorePutGet, just
+// with a bunch of reads/writes done in simultaneous go routines in
+// an effort to catch any race conditions.
+func TestBTTraceStorePutGetThreaded(t *testing.T) {
+	unittest.LargeTest(t)
+	unittest.RequiresBigTableEmulator(t)
+
+	commits := data.MakeTestCommits()
+	mvcs := MockVCSWithCommits(commits, 0)
+	defer mvcs.AssertExpectations(t)
+
+	btConf := BTConfig{
+		ProjectID:  "should-use-the-emulator",
+		InstanceID: "testinstance",
+		TableID:    "three_devices_test_threaded",
+		VCS:        mvcs,
+	}
+
+	assert.NoError(t, bt.DeleteTables(btConf.ProjectID, btConf.InstanceID, btConf.TableID))
+	assert.NoError(t, InitBT(btConf))
+
+	ctx := context.Background()
+	traceStore, err := New(ctx, btConf, true)
+	assert.NoError(t, err)
+
+	wg := sync.WaitGroup{}
+	wg.Add(2)
+
+	now := time.Date(2019, time.May, 5, 1, 3, 4, 0, time.UTC)
+
+	readTile := func() {
+		defer wg.Done()
+		_, _, err := traceStore.GetTile(ctx, len(commits))
+		assert.NoError(t, err)
+	}
+	go readTile()
+
+	// Build a tile up from the individual data points, one at a time
+	traces := data.MakeTestTile().Traces
+	for _, trace := range traces {
+		gTrace, ok := trace.(*types.GoldenTrace)
+		assert.True(t, ok)
+
+		// Put them in backwards, just to test that order doesn't matter
+		for i := len(gTrace.Digests) - 1; i >= 0; i-- {
+			wg.Add(1)
+			go func(now time.Time, i int) {
+				defer wg.Done()
+				e := tracestore.Entry{
+					Digest: gTrace.Digests[i],
+					Params: gTrace.Keys,
+				}
+				err := traceStore.Put(ctx, commits[i].Hash, []*tracestore.Entry{&e}, now)
+				assert.NoError(t, err)
+			}(now, i)
+			now = now.Add(7 * time.Second)
+		}
+	}
+	go readTile()
+
+	wg.Wait()
+
+	// Get the tile back and make sure it exactly matches the tile
+	// we hand-crafted for the test data.
+	actualTile, actualCommits, err := traceStore.GetTile(ctx, len(commits))
+	assert.NoError(t, err)
+
+	assert.Equal(t, data.MakeTestTile(), actualTile)
+	assert.Equal(t, commits, actualCommits)
+}
+
+// TestBTTraceStoreGetDenseTile makes sure we get an empty tile
+func TestBTTraceStoreGetDenseTileEmpty(t *testing.T) {
+	unittest.LargeTest(t)
+	unittest.RequiresBigTableEmulator(t)
+
+	commits := data.MakeTestCommits()
+	realCommitIndices := []int{300, 501, 557}
+	totalCommits := 1101
+	mvcs, _ := MockSparseVCSWithCommits(commits, realCommitIndices, totalCommits)
+	defer mvcs.AssertExpectations(t)
+
+	btConf := BTConfig{
+		ProjectID:  "should-use-the-emulator",
+		InstanceID: "testinstance",
+		TableID:    "three_devices_test_dense_empty",
+		VCS:        mvcs,
+	}
+
+	assert.NoError(t, bt.DeleteTables(btConf.ProjectID, btConf.InstanceID, btConf.TableID))
+	assert.NoError(t, InitBT(btConf))
+
+	ctx := context.Background()
+	traceStore, err := New(ctx, btConf, true)
+	assert.NoError(t, err)
+
+	// With no data, we should get an empty tile
+	actualTile, actualCommits, err := traceStore.GetDenseTile(ctx, len(commits))
+	assert.NoError(t, err)
+	assert.NotNil(t, actualTile)
+	assert.Empty(t, actualCommits)
+	assert.Empty(t, actualTile.Traces)
+
+}
+
+// TestBTTraceStoreGetDenseTile puts in a few data points sparsely spaced throughout
+// time and makes sure we can call GetDenseTile to get them condensed together
+// (i.e. with all the empty commits tossed out). It puts them in a variety of conditions
+// to try to identify any edge cases.
+func TestBTTraceStoreGetDenseTile(t *testing.T) {
+	unittest.LargeTest(t)
+	unittest.RequiresBigTableEmulator(t)
+
+	// 3 commits, arbitrarily spaced out across the last tile
+	commits := data.MakeTestCommits()
+	realCommitIndices := []int{795, 987, 1001}
+	totalCommits := (256 * 4) - 1
+	mvcs, lCommits := MockSparseVCSWithCommits(commits, realCommitIndices, totalCommits)
+	expectedTile := data.MakeTestTile()
+	testDenseTile(t, expectedTile, mvcs, commits, lCommits, realCommitIndices)
+
+	// 3 commits, arbitrarily spaced out across 3 tiles, with no data
+	// in the most recent tile
+	commits = data.MakeTestCommits()
+	realCommitIndices = []int{300, 501, 557}
+	totalCommits = 1101
+	mvcs, lCommits = MockSparseVCSWithCommits(commits, realCommitIndices, totalCommits)
+	expectedTile = data.MakeTestTile()
+	testDenseTile(t, expectedTile, mvcs, commits, lCommits, realCommitIndices)
+
+	// As above, just 2 commits
+	commits = data.MakeTestCommits()[1:]
+	realCommitIndices = []int{501, 557}
+	totalCommits = 1101
+	mvcs, lCommits = MockSparseVCSWithCommits(commits, realCommitIndices, totalCommits)
+	expectedTile = data.MakeTestTile()
+	expectedTile, err := expectedTile.Trim(1, 3)
+	assert.NoError(t, err)
+	testDenseTile(t, expectedTile, mvcs, commits, lCommits, realCommitIndices)
+
+	// All commits are on the first commit of their tile
+	commits = data.MakeTestCommits()
+	realCommitIndices = []int{0, 256, 512}
+	totalCommits = 1101
+	mvcs, lCommits = MockSparseVCSWithCommits(commits, realCommitIndices, totalCommits)
+	expectedTile = data.MakeTestTile()
+	testDenseTile(t, expectedTile, mvcs, commits, lCommits, realCommitIndices)
+
+	// All commits are on the last commit of their tile
+	commits = data.MakeTestCommits()
+	realCommitIndices = []int{255, 511, 767}
+	totalCommits = 1101
+	mvcs, lCommits = MockSparseVCSWithCommits(commits, realCommitIndices, totalCommits)
+	expectedTile = data.MakeTestTile()
+	testDenseTile(t, expectedTile, mvcs, commits, lCommits, realCommitIndices)
+
+	// Empty tiles between commits
+	commits = data.MakeTestCommits()
+	realCommitIndices = []int{50, 800, 1100}
+	totalCommits = 1101
+	mvcs, lCommits = MockSparseVCSWithCommits(commits, realCommitIndices, totalCommits)
+	expectedTile = data.MakeTestTile()
+	testDenseTile(t, expectedTile, mvcs, commits, lCommits, realCommitIndices)
+}
+
+// testDenseTile takes the data from tile, Puts it into BT, then pulls the tile given
+// the commit layout in VCS and returns it.
+func testDenseTile(t *testing.T, tile *tiling.Tile, mvcs *mock_vcs.VCS, commits []*tiling.Commit, lCommits []*vcsinfo.LongCommit, realCommitIndices []int) {
+	defer mvcs.AssertExpectations(t)
+
+	btConf := BTConfig{
+		ProjectID:  "should-use-the-emulator",
+		InstanceID: "testinstance",
+		TableID:    "three_devices_test_dense",
+		VCS:        mvcs,
+	}
+
+	assert.NoError(t, bt.DeleteTables(btConf.ProjectID, btConf.InstanceID, btConf.TableID))
+	assert.NoError(t, InitBT(btConf))
+
+	ctx := context.Background()
+	traceStore, err := New(ctx, btConf, true)
+	assert.NoError(t, err)
+
+	// This time is an arbitrary point in time
+	now := time.Date(2019, time.May, 5, 1, 3, 4, 0, time.UTC)
+
+	// Build a tile up from the individual data points, one at a time
+	traces := tile.Traces
+	for _, trace := range traces {
+		gTrace, ok := trace.(*types.GoldenTrace)
+		assert.True(t, ok)
+
+		// Put them in backwards, just to test that order doesn't matter
+		for i := len(gTrace.Digests) - 1; i >= 0; i-- {
+			e := tracestore.Entry{
+				Digest: gTrace.Digests[i],
+				Params: gTrace.Keys,
+			}
+			err := traceStore.Put(ctx, commits[i].Hash, []*tracestore.Entry{&e}, now)
+			assert.NoError(t, err)
+			// roll forward the clock by an arbitrary amount of time
+			now = now.Add(7 * time.Second)
+		}
+	}
+
+	// Get the tile back and make sure it exactly matches the tile
+	// we hand-crafted for the test data.
+	actualTile, allCommits, err := traceStore.GetDenseTile(ctx, len(commits))
+	assert.NoError(t, err)
+	assert.Len(t, allCommits, len(lCommits)-realCommitIndices[0])
+
+	// In MockSparseVCSWithCommits, we change the time of the commits, so we need
+	// to update the expected times to match.
+	for i, c := range commits {
+		c.CommitTime = lCommits[realCommitIndices[i]].Timestamp.Unix()
+	}
+	tile.Commits = commits
+
+	assert.Equal(t, tile, actualTile)
+}
+
+// TestBTDigestMap tests the internal workings of storing the
+// DigestMap. See BIGTABLE.md for more about the schemas for
+// the DigestMap family and the id counter family.
+func TestBTDigestMap(t *testing.T) {
+	unittest.LargeTest(t)
+	unittest.RequiresBigTableEmulator(t)
+
+	btConf := BTConfig{
+		ProjectID:  "should-use-the-emulator",
+		InstanceID: "testinstance",
+		TableID:    "digest_map_test",
+		VCS:        nil,
+	}
+
+	assert.NoError(t, bt.DeleteTables(btConf.ProjectID, btConf.InstanceID, btConf.TableID))
+	assert.NoError(t, InitBT(btConf))
+
+	ctx := context.Background()
+	traceStore, err := New(ctx, btConf, true)
+	assert.NoError(t, err)
+
+	dm, err := traceStore.getDigestMap(ctx)
+	assert.NoError(t, err)
+	assert.NotNil(t, dm)
+	// should be empty, except for the initial mapping.
+	assert.Equal(t, 1, dm.Len())
+	i, err := dm.ID(types.MISSING_DIGEST)
+	assert.NoError(t, err)
+	assert.Equal(t, missingDigestID, i)
+
+	digests := makeTestDigests(0, 100)
+
+	// add 90 of the 100 digests using update
+	ninety := make(types.DigestSet, 90)
+	for _, d := range digests[0:90] {
+		ninety[d] = true
+	}
+	dm, err = traceStore.updateDigestMap(ctx, ninety)
+	assert.NoError(t, err)
+	assert.NotNil(t, dm)
+	assert.Equal(t, 91, dm.Len())
+	// We can't check to see if our known digests map to
+	// a specific id because the digest map could present
+	// the digests in a non-deterministic order.
+	// We can spot check one of the ids though
+	_, err = dm.Digest(88)
+	assert.NoError(t, err)
+
+	ids, err := traceStore.getIDs(ctx, 3)
+	// The next 3 numbers should be 91, 92, 93 because they are
+	// monotonically increasing
+	assert.NoError(t, err)
+	assert.Equal(t, []digestID{91, 92, 93}, ids)
+	func() {
+		traceStore.availIDsMutex.Lock()
+		defer traceStore.availIDsMutex.Unlock()
+		assert.NotContains(t, traceStore.availIDs, digestID(92))
+		assert.NotContains(t, traceStore.availIDs, digestID(93))
+		assert.Contains(t, traceStore.availIDs, digestID(94))
+	}()
+
+	// give two ids back (pretend we used id 92)
+	traceStore.returnIDs([]digestID{91, 93})
+
+	func() {
+		traceStore.availIDsMutex.Lock()
+		defer traceStore.availIDsMutex.Unlock()
+		assert.NotContains(t, traceStore.availIDs, digestID(92))
+		assert.Contains(t, traceStore.availIDs, digestID(93))
+		assert.Contains(t, traceStore.availIDs, digestID(94))
+	}()
+
+	// call update with an overlap of new and old
+	twenty := make(types.DigestSet, 90)
+	for _, d := range digests[80:] {
+		twenty[d] = true
+	}
+	dm, err = traceStore.updateDigestMap(ctx, twenty)
+	assert.NoError(t, err)
+	assert.NotNil(t, dm)
+	assert.Equal(t, 101, dm.Len())
+
+	// Get it again and make sure it matches the last update phase.
+	dm2, err := traceStore.getDigestMap(ctx)
+	assert.NoError(t, err)
+	assert.NotNil(t, dm2)
+	assert.Equal(t, dm, dm2)
+
+	// Add a lot more digests to make sure the bulk requesting works
+	for i := 1; i < 10; i++ {
+		// 113 is an arbitrary prime number that does not divide batchIdRequest.
+		ds := make(types.DigestSet, 113)
+		ds.AddLists(makeTestDigests(113*i, 113))
+
+		dm, err := traceStore.updateDigestMap(ctx, ds)
+		assert.NoError(t, err)
+		assert.NotNil(t, dm)
+	}
+}
+
+// makeTestDigests returns n valid digests. These digests are easy
+// for humans to understand, as they are just the hex values [0, 99]
+// reversed and 0-padded to 32 chars long (a valid md5 hash).
+func makeTestDigests(start, n int) []types.Digest {
+	xd := make([]types.Digest, n)
+	for i := 0; i < n; i++ {
+		// Reverse them to exercise the prefixing of the digestMap.
+		s := util.ReverseString(fmt.Sprintf("%032x", start+i))
+		xd[i] = types.Digest(s)
+	}
+	return xd
+}
+
+// TestGetTileKey tests the internal workings of deriving a
+// tileKey from the commit index. See BIGTABLE.md for more.
+func TestGetTileKey(t *testing.T) {
+	unittest.LargeTest(t)
+	unittest.RequiresBigTableEmulator(t)
+
+	btConf := BTConfig{
+		// Leaving other things blank because we won't actually hit BT or use VCS.
+	}
+
+	ctx := context.Background()
+	traceStore, err := New(ctx, btConf, true)
+	assert.NoError(t, err)
+
+	type testStruct struct {
+		InputRepoIndex int
+
+		ExpectedKey   tileKey
+		ExpectedIndex int
+	}
+	// test data is valid, but arbitrary.
+	tests := []testStruct{
+		{
+			InputRepoIndex: 0,
+			ExpectedKey:    tileKey(2147483647),
+			ExpectedIndex:  0,
+		},
+		{
+			InputRepoIndex: 10,
+			ExpectedKey:    tileKey(2147483647),
+			ExpectedIndex:  10,
+		},
+		{
+			InputRepoIndex: 300,
+			ExpectedKey:    tileKey(2147483646),
+			ExpectedIndex:  44,
+		},
+		{
+			InputRepoIndex: 123456,
+			ExpectedKey:    tileKey(2147483165),
+			ExpectedIndex:  64,
+		},
+	}
+
+	for _, test := range tests {
+		key, index := traceStore.getTileKey(test.InputRepoIndex)
+		assert.Equal(t, test.ExpectedKey, key)
+		assert.Equal(t, test.ExpectedIndex, index)
+	}
+}
+
+// TestCalcShardedRowName tests the internal workings of sharding
+// a given subkey.
+func TestCalcShardedRowName(t *testing.T) {
+	unittest.LargeTest(t)
+	unittest.RequiresBigTableEmulator(t)
+
+	btConf := BTConfig{
+		// Leaving other things blank because we won't actually hit BT
+		// or use the VCS.
+	}
+
+	ctx := context.Background()
+	traceStore, err := New(ctx, btConf, true)
+	assert.NoError(t, err)
+
+	type testStruct struct {
+		InputKey     tileKey
+		InputRowType string
+		InputSubKey  string
+
+		ExpectedRowName string
+	}
+	// test data is valid, but arbitrary.
+	tests := []testStruct{
+		{
+			InputKey:     tileKey(2147483647),
+			InputRowType: typeTrace,
+			InputSubKey:  ",0=1,1=3,3=0,",
+
+			ExpectedRowName: "09:ts:t:2147483647:,0=1,1=3,3=0,",
+		},
+		{
+			InputKey:     tileKey(2147483647),
+			InputRowType: typeTrace,
+			InputSubKey:  ",0=1,1=3,9=0,",
+
+			ExpectedRowName: "13:ts:t:2147483647:,0=1,1=3,9=0,",
+		},
+		{
+			InputKey:     tileKey(2147483540),
+			InputRowType: typeDigestMap,
+			InputSubKey:  "abc",
+
+			ExpectedRowName: "02:ts:d:2147483540:abc",
+		},
+		{
+			InputKey:     tileKey(2147483540),
+			InputRowType: typeDigestMap,
+			InputSubKey:  "bcd",
+
+			ExpectedRowName: "25:ts:d:2147483540:bcd",
+		},
+	}
+
+	for _, test := range tests {
+		row := traceStore.calcShardedRowName(test.InputKey, test.InputRowType, test.InputSubKey)
+		assert.Equal(t, test.ExpectedRowName, row)
+	}
+}
+
+// TestRowAndColNameFromDigest tests the internal workings of sharding
+// a digest for use in the digest map
+func TestRowAndColNameFromDigest(t *testing.T) {
+	unittest.LargeTest(t)
+	unittest.RequiresBigTableEmulator(t)
+
+	btConf := BTConfig{
+		// Leaving other things blank because we won't actually hit BT
+		// or use the VCS.
+	}
+
+	ctx := context.Background()
+	traceStore, err := New(ctx, btConf, true)
+	assert.NoError(t, err)
+
+	type testStruct struct {
+		InputDigest types.Digest
+
+		ExpectedRowName string
+		ExpectedColName string
+	}
+	// test data is valid, but arbitrary.
+	tests := []testStruct{
+		{
+			InputDigest:     types.Digest("9e1d402515193304cafbdf02b8fd751b"),
+			ExpectedRowName: "21:ts:d:0000000000:9e1",
+			ExpectedColName: "d402515193304cafbdf02b8fd751b",
+		},
+		{
+			InputDigest:     types.Digest("e30a49351a7e45b9591a989073e755b2"),
+			ExpectedRowName: "05:ts:d:0000000000:e30",
+			ExpectedColName: "a49351a7e45b9591a989073e755b2",
+		},
+		{
+			InputDigest:     types.Digest("60b5e978d116cccc0ef12278d724245b"),
+			ExpectedRowName: "23:ts:d:0000000000:60b",
+			ExpectedColName: "5e978d116cccc0ef12278d724245b",
+		},
+	}
+
+	for _, test := range tests {
+		row, col := traceStore.rowAndColNameFromDigest(test.InputDigest)
+		assert.Equal(t, test.ExpectedRowName, row)
+		assert.Equal(t, test.ExpectedColName, col)
+	}
+}
+
+const (
+	// Directory with testdata.
+	TEST_DATA_DIR = "./testdata"
+
+	// Local file location of the test data.
+	TEST_DATA_PATH = TEST_DATA_DIR + "/10-test-sample-4bytes.tile"
+
+	// Folder in the testdata bucket. See go/testutils for details.
+	TEST_DATA_STORAGE_PATH = "gold-testdata/10-test-sample-4bytes.tile"
+
+	TILE_LENGTH = 50
+)
+
+// TestBTTraceStoreLargeTile stores a large amount of data into the tracestore
+// and retrieves it.
+func TestBTTraceStoreLargeTile(t *testing.T) {
+	unittest.LargeTest(t)
+	unittest.RequiresBigTableEmulator(t)
+
+	btConf, mvcs, tile := setupLargeTile(t)
+	defer mvcs.AssertExpectations(t)
+
+	ctx := context.Background()
+
+	traceStore, err := New(ctx, btConf, true)
+	assert.NoError(t, err)
+
+	// For each value in tile get the traceIDs that are not empty.
+	traceIDsPerCommit := make([]tiling.TraceIdSlice, TILE_LENGTH)
+	for traceID, trace := range tile.Traces {
+		gTrace := trace.(*types.GoldenTrace)
+		for i := 0; i < TILE_LENGTH; i++ {
+			if gTrace.Digests[i] != types.MISSING_DIGEST {
+				traceIDsPerCommit[i] = append(traceIDsPerCommit[i], traceID)
+			}
+		}
+	}
+	indices := make([]int, TILE_LENGTH)
+	maxIndex := 0
+	maxLen := len(traceIDsPerCommit[0])
+	for idx := range indices {
+		if len(traceIDsPerCommit[idx]) > maxLen {
+			maxLen = len(traceIDsPerCommit[idx])
+			maxIndex = idx
+		}
+		indices[idx] = idx
+	}
+
+	// Ingest the biggest tile.
+	entries := []*tracestore.Entry{}
+	allDigests := map[types.Digest]bool{"": true}
+	for _, traceID := range traceIDsPerCommit[maxIndex] {
+		t := tile.Traces[traceID].(*types.GoldenTrace)
+		digest := t.Digests[maxIndex]
+		allDigests[digest] = true
+		entries = append(entries, &tracestore.Entry{Digest: digest, Params: t.Params()})
+	}
+	assert.NoError(t, traceStore.Put(ctx, tile.Commits[maxIndex].Hash, entries, time.Now()))
+
+	foundDigestMap, err := traceStore.getDigestMap(ctx)
+	assert.NoError(t, err)
+	assert.Equal(t, len(allDigests), foundDigestMap.Len())
+
+	for digest := range allDigests {
+		id, err := foundDigestMap.ID(digest)
+		assert.NoError(t, err)
+		if digest == "" {
+			assert.Equal(t, missingDigestID, id)
+		} else {
+			assert.NotEqual(t, missingDigestID, id)
+		}
+	}
+
+	traceIDsPerCommit[maxIndex] = []tiling.TraceId{}
+
+	// Randomly add samples from the tile to that
+	for len(indices) > 0 {
+		idx := indices[0]
+		indices = indices[1:]
+		if len(traceIDsPerCommit[idx]) == 0 {
+			continue
+		}
+
+		entries := []*tracestore.Entry{}
+		for _, traceID := range traceIDsPerCommit[idx] {
+			t := tile.Traces[traceID].(*types.GoldenTrace)
+			digest := t.Digests[idx]
+			allDigests[digest] = true
+			entries = append(entries, &tracestore.Entry{Digest: digest, Params: t.Params()})
+		}
+		assert.NoError(t, traceStore.Put(ctx, tile.Commits[idx].Hash, entries, time.Now()))
+	}
+
+	// Load the tile and verify it's identical.
+	foundTile, commits, err := traceStore.GetTile(ctx, TILE_LENGTH)
+	assert.NoError(t, err)
+	assert.NotNil(t, commits)
+	assert.Equal(t, tile.Commits[len(tile.Commits)-TILE_LENGTH:], commits)
+
+	assert.Equal(t, len(tile.Traces), len(foundTile.Traces))
+	for traceID, trace := range tile.Traces {
+		gt := trace.(*types.GoldenTrace)
+		params := gt.Params()
+		found := false
+
+		foundCount := 0
+		for _, foundTrace := range foundTile.Traces {
+			if deepequal.DeepEqual(params, foundTrace.Params()) {
+				foundCount++
+			}
+		}
+		assert.Equal(t, 1, foundCount)
+
+		for foundID, foundTrace := range foundTile.Traces {
+			if deepequal.DeepEqual(params, foundTrace.Params()) {
+				expDigests := gt.Digests[len(gt.Digests)-TILE_LENGTH:]
+				found = true
+				fgt := foundTrace.(*types.GoldenTrace)
+				assert.Equal(t, len(expDigests), len(fgt.Digests))
+
+				var diff []string
+				diffStr := ""
+				for idx, digest := range expDigests {
+					isDiff := digest != fgt.Digests[idx]
+					if isDiff {
+						diff = append(diff, fmt.Sprintf("%d", idx))
+						diffStr += fmt.Sprintf("    %q  !=  %q   \n", digest, fgt.Digests[idx])
+					}
+				}
+				// Nothing should be different
+				assert.Nil(t, diff)
+				assert.Equal(t, "", diffStr)
+
+				delete(foundTile.Traces, foundID)
+				break
+			}
+		}
+		assert.True(t, found)
+		delete(tile.Traces, traceID)
+	}
+	assert.Equal(t, 0, len(foundTile.Traces))
+	assert.Equal(t, 0, len(tile.Traces))
+}
+
+func setupLargeTile(t sktest.TestingT) (BTConfig, *mock_vcs.VCS, *tiling.Tile) {
+	if !fileutil.FileExists(TEST_DATA_PATH) {
+		err := gcs_testutils.DownloadTestDataFile(t, gcs_testutils.TEST_DATA_BUCKET, TEST_DATA_STORAGE_PATH, TEST_DATA_PATH)
+		assert.NoError(t, err, "Unable to download testdata.")
+	}
+
+	tile := makeSampleTile(t, TEST_DATA_PATH)
+	assert.Len(t, tile.Commits, TILE_LENGTH)
+
+	mvcs := MockVCSWithCommits(tile.Commits, 0)
+
+	btConf := BTConfig{
+		ProjectID:  "should-use-the-emulator",
+		InstanceID: "testinstance",
+		TableID:    "large_tile_test",
+		VCS:        mvcs,
+	}
+
+	assert.NoError(t, bt.DeleteTables(btConf.ProjectID, btConf.InstanceID, btConf.TableID))
+	assert.NoError(t, InitBT(btConf))
+	fmt.Println("BT emulator set up")
+	return btConf, mvcs, tile
+}
+
+func makeSampleTile(t sktest.TestingT, fileName string) *tiling.Tile {
+	file, err := os.Open(fileName)
+	assert.NoError(t, err)
+
+	sample, err := serialize.DeserializeSample(file)
+	assert.NoError(t, err)
+
+	return sample.Tile
+}
+
+func MockVCSWithCommits(commits []*tiling.Commit, offset int) *mock_vcs.VCS {
+	mvcs := &mock_vcs.VCS{}
+
+	indexCommits := make([]*vcsinfo.IndexCommit, 0, len(commits))
+	hashes := make([]string, 0, len(commits))
+	longCommits := make([]*vcsinfo.LongCommit, 0, len(commits))
+	for i, c := range commits {
+		mvcs.On("IndexOf", ctx, c.Hash).Return(i+offset, nil).Maybe()
+
+		indexCommits = append(indexCommits, &vcsinfo.IndexCommit{
+			Hash:      c.Hash,
+			Index:     i + offset,
+			Timestamp: time.Unix(c.CommitTime, 0),
+		})
+		hashes = append(hashes, c.Hash)
+		longCommits = append(longCommits, &vcsinfo.LongCommit{
+			ShortCommit: &vcsinfo.ShortCommit{
+				Hash:    c.Hash,
+				Author:  c.Author,
+				Subject: fmt.Sprintf("Commit #%d in test", i),
+			},
+			Timestamp: time.Unix(c.CommitTime, 0),
+		})
+	}
+
+	mvcs.On("LastNIndex", len(commits)).Return(indexCommits)
+	mvcs.On("DetailsMulti", ctx, hashes, false).Return(longCommits, nil)
+
+	return mvcs
+}
+
+func MockSparseVCSWithCommits(commits []*tiling.Commit, realCommitIndices []int, totalCommits int) (*mock_vcs.VCS, []*vcsinfo.LongCommit) {
+	mvcs := &mock_vcs.VCS{}
+	if len(commits) != len(realCommitIndices) {
+		panic("commits should be same length as realCommitIndices")
+	}
+
+	// Create many synthetic commits.
+	indexCommits := make([]*vcsinfo.IndexCommit, totalCommits)
+	longCommits := make([]*vcsinfo.LongCommit, totalCommits)
+	hashes := []string{}
+	for i := 0; i < totalCommits; i++ {
+		h := fmt.Sprintf("%040d", i)
+		indexCommits[i] = &vcsinfo.IndexCommit{
+			Hash:  h,
+			Index: i,
+			// space the commits 1700 seconds apart, starting at the epoch
+			// This is an arbitrary amount of space.
+			Timestamp: time.Unix(int64(i*1700), 0),
+		}
+
+		longCommits[i] = &vcsinfo.LongCommit{
+			ShortCommit: &vcsinfo.ShortCommit{
+				Hash:   h,
+				Author: "nobody@example.com",
+			},
+			Timestamp: time.Unix(int64(i*1700), 0),
+		}
+		hashes = append(hashes, h)
+
+	}
+
+	for i, c := range commits {
+		index := realCommitIndices[i]
+		mvcs.On("IndexOf", ctx, c.Hash).Return(index, nil).Maybe()
+		indexCommits[index] = &vcsinfo.IndexCommit{
+			Hash:      c.Hash,
+			Index:     index,
+			Timestamp: time.Unix(int64(index*1700), 0),
+		}
+		hashes[index] = c.Hash
+		longCommits[index] = &vcsinfo.LongCommit{
+			ShortCommit: &vcsinfo.ShortCommit{
+				Hash:    c.Hash,
+				Author:  c.Author,
+				Subject: fmt.Sprintf("Real commit #%d in test", i),
+			},
+			Timestamp: time.Unix(int64(index*1700), 0),
+		}
+	}
+
+	firstRealCommitIdx := realCommitIndices[0]
+	mvcs.On("ByIndex", ctx, firstRealCommitIdx).Return(longCommits[firstRealCommitIdx], nil).Maybe()
+	mvcs.On("From", mock.Anything).Return(hashes[firstRealCommitIdx:], nil).Maybe()
+	mvcs.On("LastNIndex", 1).Return(indexCommits[totalCommits-1:]).Maybe()
+	mvcs.On("DetailsMulti", ctx, hashes[firstRealCommitIdx:], false).Return(longCommits[firstRealCommitIdx:], nil).Maybe()
+
+	return mvcs, longCommits
+}
+
+var ctx = mock.AnythingOfType("*context.emptyCtx")
diff --git a/golden/go/tracestore/bt_tracestore/types.go b/golden/go/tracestore/bt_tracestore/types.go
new file mode 100644
index 0000000..20d5d56
--- /dev/null
+++ b/golden/go/tracestore/bt_tracestore/types.go
@@ -0,0 +1,386 @@
+package bt_tracestore
+
+import (
+	"crypto/md5"
+	"encoding"
+	"encoding/binary"
+	"fmt"
+	"time"
+
+	"cloud.google.com/go/bigtable"
+	"go.skia.org/infra/go/paramtools"
+	"go.skia.org/infra/go/skerr"
+	"go.skia.org/infra/go/tiling"
+	"go.skia.org/infra/golden/go/types"
+)
+
+// Constants adapted from btts.go
+// See BIGTABLE.md for an overview of how the data is stored in BT.
+const (
+	// Namespace for this package's data. Due to the fact that there is one table per instance,
+	// This makes sure we don't have collisions between traces and something else
+	// in BT (i.e. git info)
+	traceStoreNameSpace = "ts"
+
+	// Column Families.
+	// https://cloud.google.com/bigtable/docs/schema-design#column_families_and_column_qualifiers
+	digestMapFamily = "D" // Store a mapping of Digest->DigestID
+	idCounterFamily = "I" // Keeps a monotonically increasing number to generate DigestIDs
+	opsFamily       = "O" // ops short for Ordered Param Set
+	traceFamily     = "T" // Holds "0"..."tilesize-1" columns with a DigestID at each cell
+
+	// The columns (and rows) in the digest map family are derived from the digest, which are md5
+	// hashes. The row will be the first three characters of the hash and the column will
+	// be the remaining characters (see b.rowAndColNameFromDigest)
+	// All entries will have the following prefix
+	digestMapFamilyPrefix = digestMapFamily + ":"
+	// All entries are part of a global map (set tile to 0)
+	digestMapTile = tileKey(0)
+
+	// Columns in the ID counter family. There is only one row and one column.
+	idCounterColumn = "idc"
+
+	// Columns in the OrderedParamSet column family.
+	opsHashColumn   = "H"
+	opsOpsColumn    = "OPS"
+	hashFullColName = opsFamily + ":" + opsHashColumn
+	opsFullColName  = opsFamily + ":" + opsOpsColumn
+
+	// The columns in the trace family are "0", "1", "2"..."N" where N is
+	// the BT tile size (default below). These values correspond to the commitOffset,
+	// where 0 is the first (most recent) commit in the tile and N is the last (oldest)
+	// commit in the tile.
+	// They will all have the following prefix.
+	traceFamilyPrefix = traceFamily + ":"
+
+	// Define the row types.
+	typeDigestMap = "d"
+	typeIdCounter = "i"
+	typeOPS       = "o"
+	typeTrace     = "t"
+
+	// This is the size of the tile in Big Table. That is, how many commits do we store in one tile.
+	// We can have up to 2^32 tiles in big table, so this would let us store 1 trillion
+	// commits worth of data. This tile size does not need to be related to the tile size that
+	// Gold operates on (although when tuning, it should be greater than, or an even divisor
+	// of the Gold tile size). The first commit in the repo belongs to tile 2^32-1 and tile numbers
+	// decrease for newer commits.
+	DefaultTileSize = 256
+
+	// Default number of shards used. A shard splits the traces up on a tile.
+	// If a trace exists on shard N in tile A, it will be on shard N for all tiles.
+	// Having traces on shards lets BT split up the work more evenly.
+	DefaultShards = 32
+
+	// To avoid many successive increment calls to the id counter cell, we request a number of
+	// ids at once. This number is arbitrarily picked and can be increased if need be.
+	numReservedIds = 256
+
+	readTimeout  = 4 * time.Minute
+	writeTimeout = 10 * time.Minute
+
+	// BadTileKey is returned in error conditions.
+	badTileKey = tileKey(-1)
+
+	// missingDigestID is the id for types.MISSING_DIGEST
+	missingDigestID = digestID(0)
+)
+
+// List of families (conceptually similar to tables) we are creating in BT.
+var btColumnFamilies = []string{
+	traceFamily,
+	opsFamily,
+	idCounterFamily,
+	digestMapFamily,
+}
+
+// We have one global digest id counter, so just hard-code it to tile 0.
+var idCounterRow = unshardedRowName(typeIdCounter, 0)
+
+// tileKey is the identifier for each tile held in BigTable.
+//
+// Note that tile keys are in the opposite order of tile offset, that is, the first commit
+// in a repo goes in the first tile, which has key 2^32-1. We do this so more recent
+// tiles come first in sort order.
+type tileKey int32
+
+// digestID is an arbitrary number for referring to a types.Digest (string)
+// that is stored in a digestMap.
+type digestID uint64
+
+// MarshalBinary implements the encoding.BinaryMarshaler interface, allowing for
+// us to compactly store these to BT (about 50% space savings over string representation).
+func (d digestID) MarshalBinary() ([]byte, error) {
+	rv := make([]byte, binary.MaxVarintLen64)
+	n := binary.PutUvarint(rv, uint64(d))
+	return rv[:n], nil
+}
+
+var _ encoding.BinaryMarshaler = digestID(0)
+
+// UnmarshalBinary implements the encoding.BinaryUnmarshaler interface, allowing for
+// us to compactly store these to BT (about 50% space savings over string representation).
+func (d *digestID) UnmarshalBinary(data []byte) error {
+	val, n := binary.Uvarint(data)
+	if n != len(data) {
+		return skerr.Fmt("Error decoding digestID from %x; Uvarint consumed %d bytes instead of %d", data, n, len(data))
+	}
+	*d = digestID(val)
+	return nil
+}
+
+var _ encoding.BinaryUnmarshaler = (*digestID)(nil)
+
+// encodedTraceID is a shortened form of a tiling.TraceId, e.g. 0=1,1=3,3=0,
+// Those indices are references to the OrderedParamSet stored in encTile.
+// See params.paramsEncoder
+type encodedTraceID string
+
+// encTile contains an encoded tile.
+type encTile struct {
+	// maps a trace id to the list of digest ids. The list corresponds to the commits, with index
+	// 0 being the oldest commit and the last commit being the most recent.
+	traces map[encodedTraceID][]digestID
+	ops    *paramtools.OrderedParamSet
+}
+
+// When ingesting we keep a cache of the OrderedParamSets we have seen per-tile.
+type opsCacheEntry struct {
+	ops  *paramtools.OrderedParamSet
+	hash string // md5 has of the serialized ops - used for deterministic querying.
+}
+
+// opsCacheEntryFromOPS creates and fills in an OpsCacheEntry from the given
+// OrderedParamSet and sets the hash appropriately.
+func opsCacheEntryFromOPS(ops *paramtools.OrderedParamSet) (*opsCacheEntry, error) {
+	buf, err := ops.Encode()
+	if err != nil {
+		return nil, skerr.Fmt("could not encode the given ops to bytes: %s", err)
+	}
+	hash := fmt.Sprintf("%x", md5.Sum(buf))
+	return &opsCacheEntry{
+		ops:  ops,
+		hash: hash,
+	}, nil
+}
+
+// newOpsCacheEntry returns an empty OpsCacheEntry.
+func newOpsCacheEntry() (*opsCacheEntry, error) {
+	return opsCacheEntryFromOPS(paramtools.NewOrderedParamSet())
+}
+
+// newOpsCacheEntryFromRow loads the appropriate data from the given BT row
+// and returns a OpsCacheEntry with that data.
+func newOpsCacheEntryFromRow(row bigtable.Row) (*opsCacheEntry, error) {
+	family := row[opsFamily]
+	if len(family) != 2 {
+		// This should never happen
+		return nil, skerr.Fmt("incorrect number of of OPS columns in BT for key %s, %d != 2", row.Key(), len(family))
+	}
+	ops := &paramtools.OrderedParamSet{}
+	hash := ""
+	for _, col := range family {
+		if col.Column == opsFullColName {
+			var err error
+			ops, err = paramtools.NewOrderedParamSetFromBytes(col.Value)
+			if err != nil {
+				// should never happen
+				return nil, skerr.Fmt("corrupted paramset in BT for key %s: %s", row.Key(), err)
+			}
+		} else if col.Column == hashFullColName {
+			hash = string(col.Value)
+		}
+	}
+	if hash == "" {
+		return nil, skerr.Fmt("missing hash for OPS for key %s: %#v", row.Key(), ops)
+	}
+	// You might be tempted to use opsCacheEntryFromOps and
+	// check that entry.hash == hash here, but that will fail
+	// because GoB encoding of maps is not deterministic.
+	entry := opsCacheEntry{
+		ops:  ops,
+		hash: hash,
+	}
+	return &entry, nil
+}
+
+// A digestMap keeps track of the mapping between encoded digestID and their corresponding
+// types.Digest.
+type digestMap struct {
+	intMap map[digestID]types.Digest
+	strMap map[types.Digest]digestID
+}
+
+// newDigestMap creates an empty digestMap with the given capacity.
+func newDigestMap(cap int) *digestMap {
+	ret := &digestMap{
+		intMap: make(map[digestID]types.Digest, cap),
+		strMap: make(map[types.Digest]digestID, cap),
+	}
+	ret.intMap[missingDigestID] = types.MISSING_DIGEST
+	ret.strMap[types.MISSING_DIGEST] = missingDigestID
+	return ret
+}
+
+// Delta returns a []types.Digest of those passed in digests that are
+// not in this mapping currently.
+func (d *digestMap) Delta(digests map[types.Digest]bool) []types.Digest {
+	ret := make([]types.Digest, 0, len(digests))
+	for digest := range digests {
+		if _, ok := d.strMap[digest]; !ok {
+			ret = append(ret, digest)
+		}
+	}
+	return ret
+}
+
+// Add expands the map with the given entries. It fails if any key or any value
+// is already in the mapping.
+func (d *digestMap) Add(newEntries map[types.Digest]digestID) error {
+	for digest, id := range newEntries {
+		if digest == types.MISSING_DIGEST || id == 0 {
+			return skerr.Fmt("invalid input id or digest: (%q -> %d)", digest, id)
+		}
+
+		foundID, strExists := d.strMap[digest]
+		foundDigest, intExists := d.intMap[id]
+		if strExists && intExists {
+			if (foundID != id) || (foundDigest != digest) {
+				return skerr.Fmt("inconsistent data - got (%q -> %d) when (%q -> %d) was expected", digest, id, foundDigest, foundID)
+			}
+			// Already contained so this is a no-op.
+			return nil
+		}
+
+		if strExists || intExists {
+			return skerr.Fmt("internal inconsistency - expected forward mapping (%q -> %d) and reverse mapping (%d -> %q) to both be present", digest, foundID, id, foundDigest)
+		}
+
+		// New mapping. Add it.
+		d.intMap[id] = digest
+		d.strMap[digest] = id
+	}
+	return nil
+}
+
+// ID returns the DigestID for a given types.Digest.
+func (d *digestMap) ID(digest types.Digest) (digestID, error) {
+	ret, ok := d.strMap[digest]
+	if !ok {
+		return 0, skerr.Fmt("unable to find id for %q", digest)
+	}
+	return ret, nil
+}
+
+// DecodeIDs is like Digest but in bulk.
+func (d *digestMap) DecodeIDs(ids []digestID) ([]types.Digest, error) {
+	ret := make([]types.Digest, len(ids))
+	var ok bool
+	for idx, id := range ids {
+		ret[idx], ok = d.intMap[id]
+		if !ok {
+			return nil, skerr.Fmt("unable to find id %d in intMap", id)
+		}
+	}
+	return ret, nil
+}
+
+// Digest returns the types.Digest for a given DigestID.
+func (d *digestMap) Digest(id digestID) (types.Digest, error) {
+	ret, ok := d.intMap[id]
+	if !ok {
+		return "", skerr.Fmt("unable to find digest for %d", id)
+	}
+	return ret, nil
+}
+
+// Len returns how many map entries are in this map.
+func (d *digestMap) Len() int {
+	return len(d.strMap)
+}
+
+// Define this as a type so we can define some helper functions.
+type traceMap map[tiling.TraceId]tiling.Trace
+
+// CommitIndicesWithData returns the indexes of the commits with at least one non-missing
+// digest in at least one trace.
+func (t traceMap) CommitIndicesWithData() []int {
+	if len(t) == 0 {
+		return nil
+	}
+	numCommits := 0
+	for _, trace := range t {
+		gt := trace.(*types.GoldenTrace)
+		numCommits = len(gt.Digests)
+		break
+	}
+	var haveData []int
+	for i := 0; i < numCommits; i++ {
+		for _, trace := range t {
+			gt := trace.(*types.GoldenTrace)
+			if !gt.IsMissing(i) {
+				haveData = append(haveData, i)
+				break
+			}
+		}
+	}
+	return haveData
+}
+
+// MakeFromCommitIndexes creates a new traceMap from the data in this one that
+// only has the digests belonging to the given commit indices. Conceptually,
+// this grabs a subset of the commit columns from the tile.
+func (t traceMap) MakeFromCommitIndexes(indices []int) traceMap {
+	if len(indices) == 0 {
+		return traceMap{}
+	}
+	r := make(traceMap, len(t))
+	for id, trace := range t {
+		gt := trace.(*types.GoldenTrace)
+
+		newDigests := make([]types.Digest, len(indices))
+		for i, idx := range indices {
+			newDigests[i] = gt.Digests[idx]
+		}
+
+		r[id] = &types.GoldenTrace{
+			Keys:    gt.Keys,
+			Digests: newDigests,
+		}
+	}
+	return r
+}
+
+// PrependTraces augments this traceMap with the data from the given one.
+// Specifically, it prepends that data, assuming the "other" data came
+// before the data in this map.
+// TODO(kjlubick): Deduplicate this with tiling.Merge
+func (t traceMap) PrependTraces(other traceMap) {
+	numCommits := 0
+	for _, trace := range t {
+		gt := trace.(*types.GoldenTrace)
+		numCommits = len(gt.Digests)
+		break
+	}
+
+	numOtherCommits := 0
+	for id, trace := range other {
+		numOtherCommits = trace.Len()
+		original, ok := t[id]
+		if ok {
+			// Keys are constant and are what the id is derived from
+			t[id] = trace.Merge(original)
+		} else {
+			// if we stopped seeing the trace in t, we need to pad the end with MISSING_DIGEST
+			trace.Grow(numOtherCommits+numCommits, tiling.FILL_AFTER) // Assumes we can modify other
+			t[id] = trace
+		}
+	}
+
+	// if we saw a trace in t, but not in other, we need to pad the beginning with MISSING_DIGEST
+	for id, trace := range t {
+		if _, ok := other[id]; !ok {
+			trace.Grow(numOtherCommits+numCommits, tiling.FILL_BEFORE)
+		}
+	}
+}
diff --git a/golden/go/tracestore/bt_tracestore/types_test.go b/golden/go/tracestore/bt_tracestore/types_test.go
new file mode 100644
index 0000000..6aefc3d
--- /dev/null
+++ b/golden/go/tracestore/bt_tracestore/types_test.go
@@ -0,0 +1,267 @@
+package bt_tracestore
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.skia.org/infra/go/testutils/unittest"
+	"go.skia.org/infra/golden/go/types"
+)
+
+// TestDigestMapMissingDigest verifies a newly created digest map
+// starts off with the mapping for a missing digest and nothing else.
+func TestDigestMapMissingDigest(t *testing.T) {
+	unittest.SmallTest(t)
+
+	digestMap := newDigestMap(1000)
+	assert.Equal(t, 1, digestMap.Len())
+	id, err := digestMap.ID(types.MISSING_DIGEST)
+	assert.NoError(t, err)
+	assert.Equal(t, id, missingDigestID)
+
+	_, err = digestMap.ID(AlphaDigest)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "unable to find")
+}
+
+// TestDigestMapAddGet tests the three ways to get data out of the map.
+func TestDigestMapAddGet(t *testing.T) {
+	unittest.SmallTest(t)
+
+	digestMap := newDigestMap(1)
+	err := digestMap.Add(map[types.Digest]digestID{
+		AlphaDigest: AlphaID,
+		BetaDigest:  BetaID,
+		GammaDigest: GammaID,
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, 4, digestMap.Len())
+
+	digests, err := digestMap.DecodeIDs([]digestID{GammaID, AlphaID})
+	assert.NoError(t, err)
+	assert.Equal(t, digests, []types.Digest{GammaDigest, AlphaDigest})
+
+	d, err := digestMap.Digest(BetaID)
+	assert.NoError(t, err)
+	assert.Equal(t, BetaDigest, d)
+
+	id, err := digestMap.ID(BetaDigest)
+	assert.NoError(t, err)
+	assert.Equal(t, BetaID, id)
+}
+
+// TestDigestMapAddBadGet tests getting things out of the map that don't exist.
+func TestDigestMapAddBadGet(t *testing.T) {
+	unittest.SmallTest(t)
+
+	notExistID := digestID(99)
+	notExistDigest := types.Digest("fffe39544765f38baab53350aef79966")
+
+	digestMap := newDigestMap(1)
+	err := digestMap.Add(map[types.Digest]digestID{
+		AlphaDigest: AlphaID,
+		BetaDigest:  BetaID,
+		GammaDigest: GammaID,
+	})
+	assert.NoError(t, err)
+	_, err = digestMap.DecodeIDs([]digestID{AlphaID, notExistID})
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "unable to find id")
+
+	_, err = digestMap.Digest(notExistID)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "unable to find digest")
+
+	_, err = digestMap.ID(notExistDigest)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "unable to find id")
+}
+
+// TestDigestMapBadAdd tests some corner cases with adding things to the map.
+func TestDigestMapBadAdd(t *testing.T) {
+	unittest.SmallTest(t)
+
+	digestMap := newDigestMap(1)
+	err := digestMap.Add(map[types.Digest]digestID{AlphaDigest: AlphaID})
+	assert.NoError(t, err)
+	err = digestMap.Add(map[types.Digest]digestID{BetaDigest: BetaID})
+	assert.NoError(t, err)
+
+	// Adding something multiple times is no error
+	err = digestMap.Add(map[types.Digest]digestID{BetaDigest: BetaID})
+	assert.NoError(t, err)
+	err = digestMap.Add(map[types.Digest]digestID{BetaDigest: BetaID})
+	assert.NoError(t, err)
+
+	// Can't add something with MISSING_DIGEST as a key ...
+	err = digestMap.Add(map[types.Digest]digestID{types.MISSING_DIGEST: 5})
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "invalid input id")
+	// ... or as a value.
+	err = digestMap.Add(map[types.Digest]digestID{BetaDigest: missingDigestID})
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "invalid input id")
+
+	// Can't add something that is already in the map as a key
+	err = digestMap.Add(map[types.Digest]digestID{BetaDigest: GammaID})
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "internal inconsistency")
+	// ... or as a value.
+	err = digestMap.Add(map[types.Digest]digestID{GammaDigest: BetaID})
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "internal inconsistency")
+
+	// Can't mix up data that has already been seen before
+	err = digestMap.Add(map[types.Digest]digestID{AlphaDigest: BetaID})
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "inconsistent data")
+}
+
+func TestTraceMapCommitIndicesWithData(t *testing.T) {
+	unittest.SmallTest(t)
+
+	tm := traceMap{
+		",key=first,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				types.MISSING_DIGEST, types.MISSING_DIGEST, AlphaDigest,
+				AlphaDigest, types.MISSING_DIGEST, BetaDigest,
+			},
+		},
+		",key=second,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				GammaDigest, types.MISSING_DIGEST, types.MISSING_DIGEST,
+				GammaDigest, types.MISSING_DIGEST, GammaDigest,
+			},
+		},
+	}
+	assert.Equal(t, []int{0, 2, 3, 5}, tm.CommitIndicesWithData())
+
+	empty := traceMap{
+		",key=first,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				types.MISSING_DIGEST, types.MISSING_DIGEST, types.MISSING_DIGEST,
+			},
+		},
+		",key=second,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				types.MISSING_DIGEST, types.MISSING_DIGEST, types.MISSING_DIGEST,
+			},
+		},
+	}
+	assert.Nil(t, empty.CommitIndicesWithData())
+
+	assert.Nil(t, traceMap{}.CommitIndicesWithData())
+}
+
+func TestTraceMapMakeFromCommitIndexes(t *testing.T) {
+	unittest.SmallTest(t)
+
+	tm := traceMap{
+		",key=first,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				types.MISSING_DIGEST, types.MISSING_DIGEST, AlphaDigest,
+				AlphaDigest, types.MISSING_DIGEST, BetaDigest,
+			},
+		},
+		",key=second,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				GammaDigest, types.MISSING_DIGEST, types.MISSING_DIGEST,
+				GammaDigest, types.MISSING_DIGEST, GammaDigest,
+			},
+		},
+	}
+
+	assert.Equal(t, traceMap{
+		",key=first,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				types.MISSING_DIGEST, AlphaDigest,
+				AlphaDigest, BetaDigest,
+			},
+		},
+		",key=second,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				GammaDigest, types.MISSING_DIGEST,
+				GammaDigest, GammaDigest,
+			},
+		},
+	}, tm.MakeFromCommitIndexes([]int{0, 2, 3, 5}))
+
+	assert.Equal(t, traceMap{
+		",key=first,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				types.MISSING_DIGEST, types.MISSING_DIGEST, types.MISSING_DIGEST,
+			},
+		},
+		",key=second,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				GammaDigest, types.MISSING_DIGEST, types.MISSING_DIGEST,
+			},
+		},
+	}, tm.MakeFromCommitIndexes([]int{0, 1, 4}))
+
+	assert.Equal(t, traceMap{}, tm.MakeFromCommitIndexes([]int{}))
+	assert.Equal(t, traceMap{}, tm.MakeFromCommitIndexes(nil))
+}
+
+func TestTraceMapPrependTraces(t *testing.T) {
+	unittest.SmallTest(t)
+
+	tm1 := traceMap{
+		",key=first,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				types.MISSING_DIGEST, types.MISSING_DIGEST, AlphaDigest,
+			},
+		},
+		",key=second,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				GammaDigest, types.MISSING_DIGEST, types.MISSING_DIGEST,
+			},
+		},
+	}
+
+	tm2 := traceMap{
+		",key=first,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				types.MISSING_DIGEST, GammaDigest,
+			},
+		},
+		",key=third,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				GammaDigest, BetaDigest,
+			},
+		},
+	}
+
+	tm1.PrependTraces(tm2)
+
+	assert.Equal(t, traceMap{
+		",key=first,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				types.MISSING_DIGEST, GammaDigest,
+				types.MISSING_DIGEST, types.MISSING_DIGEST, AlphaDigest,
+			},
+		},
+		",key=second,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				types.MISSING_DIGEST, types.MISSING_DIGEST,
+				GammaDigest, types.MISSING_DIGEST, types.MISSING_DIGEST,
+			},
+		},
+		",key=third,": &types.GoldenTrace{
+			Digests: []types.Digest{
+				GammaDigest, BetaDigest,
+				types.MISSING_DIGEST, types.MISSING_DIGEST, types.MISSING_DIGEST,
+			},
+		},
+	}, tm1)
+}
+
+const (
+	AlphaDigest = types.Digest("aaa6fc936d06e6569788366f1e3fda4e")
+	BetaDigest  = types.Digest("bbb15c047d150d961573062854f35a55")
+	GammaDigest = types.Digest("cccd42f3ee0b02687f63963adb36a580")
+
+	AlphaID = digestID(1)
+	BetaID  = digestID(2)
+	GammaID = digestID(3)
+)
diff --git a/golden/go/tracestore/bt_tracestore/util.go b/golden/go/tracestore/bt_tracestore/util.go
new file mode 100644
index 0000000..9bb8c87
--- /dev/null
+++ b/golden/go/tracestore/bt_tracestore/util.go
@@ -0,0 +1,45 @@
+package bt_tracestore
+
+import (
+	"fmt"
+	"math"
+	"strings"
+)
+
+// tileKeyFromIndex converts the tile index to the tileKey.
+// See BIGTABLE.md for more on this conversion.
+func tileKeyFromIndex(tileIndex int32) tileKey {
+	if tileIndex < 0 {
+		return badTileKey
+	}
+	return tileKey(math.MaxInt32 - tileIndex)
+}
+
+// OpsRowName returns the name of the BigTable row which stores the OrderedParamSet
+// for this tile.
+func (t tileKey) OpsRowName() string {
+	return unshardedRowName(typeOPS, t)
+}
+
+// unshardedRowName calculates the row for the given data which all has the same format:
+// :[namespace]:[type]:[tile]:
+func unshardedRowName(rowType string, tileKey tileKey) string {
+	return fmt.Sprintf(":%s:%s:%010d:", traceStoreNameSpace, rowType, tileKey)
+}
+
+// shardedRowName calculates the row for the given data which all has the same format:
+// [shard]:[namespace]:[type]:[tile]:[subkey]
+// For some data types, where there is only one row,  or when doing a prefix-match,
+// subkey may be "".
+func shardedRowName(shard int32, rowType string, tileKey tileKey, subkey string) string {
+	return fmt.Sprintf("%02d:%s:%s:%010d:%s", shard, traceStoreNameSpace, rowType, tileKey, subkey)
+}
+
+// extractKey returns the subkey from the given row name. This could be "".
+func extractSubkey(rowName string) string {
+	parts := strings.Split(rowName, ":")
+	if len(parts) == 0 {
+		return ""
+	}
+	return parts[len(parts)-1]
+}
diff --git a/golden/go/tracestore/bt_tracestore/util_test.go b/golden/go/tracestore/bt_tracestore/util_test.go
new file mode 100644
index 0000000..586a593
--- /dev/null
+++ b/golden/go/tracestore/bt_tracestore/util_test.go
@@ -0,0 +1,57 @@
+package bt_tracestore
+
+import (
+	"math"
+	"testing"
+
+	assert "github.com/stretchr/testify/require"
+	"go.skia.org/infra/go/testutils/unittest"
+)
+
+// The tests in this package are mainly to make sure changes that
+// are not backward-compatible are detected.
+
+func TestTileKeyFromIndex(t *testing.T) {
+	unittest.SmallTest(t)
+
+	// spot-check some arbitrary values
+	assert.Equal(t, tileKey(2147483647), tileKeyFromIndex(0))
+	assert.Equal(t, tileKey(2147483451), tileKeyFromIndex(196))
+	assert.Equal(t, tileKey(908536335), tileKeyFromIndex(1238947312))
+}
+
+func TestOpsRowName(t *testing.T) {
+	unittest.SmallTest(t)
+
+	// spot-check some arbitrary values
+	assert.Equal(t, ":ts:o:2147483647:", tileKeyFromIndex(0).OpsRowName())
+	assert.Equal(t, ":ts:o:2147483451:", tileKeyFromIndex(196).OpsRowName())
+	assert.Equal(t, ":ts:o:0908536335:", tileKeyFromIndex(1238947312).OpsRowName())
+}
+
+func TestShardedRowName(t *testing.T) {
+	unittest.SmallTest(t)
+
+	shard := int32(3) // arbitrarily picked
+	tileZeroKey := tileKey(math.MaxInt32 - 1)
+	veryNewTileKey := tileKey(57)
+
+	// Example RowName for a trace
+	encodedTrace := ",0=1,1=3,3=0,"
+	assert.Equal(t, "03:ts:t:2147483646:,0=1,1=3,3=0,", shardedRowName(shard, typeTrace, tileZeroKey, encodedTrace))
+	assert.Equal(t, "03:ts:t:0000000057:,0=1,1=3,3=0,", shardedRowName(shard, typeTrace, veryNewTileKey, encodedTrace))
+
+	// Example RowName for a digest
+	// digests are stored in a row based on the first three characters and a
+	// column with the remaining characters.
+	digestPrefix := string(AlphaDigest[:3])
+	assert.Equal(t, "03:ts:d:2147483646:aaa", shardedRowName(shard, typeDigestMap, tileZeroKey, digestPrefix))
+}
+
+func TestExtractKeyFromRowName(t *testing.T) {
+	unittest.SmallTest(t)
+
+	assert.Equal(t, "ae3", extractSubkey("07:ts:d:2147483646:ae3"))
+	assert.Equal(t, "", extractSubkey(":ts:o:2147483646:"))
+	assert.Equal(t, ",0=1,1=3,3=0,", extractSubkey("03:ts:t:2147483646:,0=1,1=3,3=0,"))
+}
diff --git a/golden/go/tracestore/types.go b/golden/go/tracestore/types.go
new file mode 100644
index 0000000..7b70785
--- /dev/null
+++ b/golden/go/tracestore/types.go
@@ -0,0 +1,75 @@
+package tracestore
+
+import (
+	"context"
+	"regexp"
+	"time"
+
+	"go.skia.org/infra/go/paramtools"
+	"go.skia.org/infra/go/query"
+	"go.skia.org/infra/go/sklog"
+	"go.skia.org/infra/go/tiling"
+	"go.skia.org/infra/golden/go/types"
+)
+
+// Entry is one digests and related params to be added to the TraceStore.
+// TODO(kjlubick): Make sure this does not include the keys included in
+// options.
+type Entry struct {
+	// Params describe the configuration that produced the digest/image.
+	Params map[string]string
+
+	// Digest references the image that was generated by the test.
+	Digest types.Digest
+}
+
+// TraceStore is the interface to store trace data.
+type TraceStore interface {
+	// Put writes the given entries to the TraceStore at the given commit hash. The timestamp is
+	// assumed to be the time when the entries were generated.
+	// It is undefined behavior to have multiple entries with the exact same Params.
+	Put(ctx context.Context, commitHash string, entries []*Entry, ts time.Time) error
+
+	// GetTile reads the last n commits and returns them as a tile.
+	// The second return value is all commits that are in the tile.
+	GetTile(ctx context.Context, nCommits int) (*tiling.Tile, []*tiling.Commit, error)
+
+	// GetDenseTile constructs a tile containing only commits that have data for at least one trace.
+	// The returned tile will always have length exactly nCommits unless there are fewer than
+	// nCommits commits with data. The second return value contains all commits starting with the
+	// first commit of the tile and ending with the most recent commit, in order; i.e. it includes
+	// all commits in the tile as well as the omitted commits.
+	GetDenseTile(ctx context.Context, nCommits int) (*tiling.Tile, []*tiling.Commit, error)
+}
+
+// TraceIDFromParams deterministically returns a TraceId that uniquely encodes
+// the given params. It follows the same convention as perf's trace ids, that
+// is something like ",key1=value1,key2=value2,...," where the keys
+// are in alphabetical order.
+func TraceIDFromParams(params paramtools.Params) tiling.TraceId {
+	// Clean up any params with , or =
+	params = forceValid(params)
+	s, err := query.MakeKeyFast(params)
+	if err != nil {
+		sklog.Warningf("Invalid params passed in for trace id %#v: %s", params, err)
+	}
+	return tiling.TraceId(s)
+}
+
+var (
+	invalidChar = regexp.MustCompile("([,=])")
+)
+
+func clean(s string) string {
+	return invalidChar.ReplaceAllLiteralString(s, "_")
+}
+
+// forceValid ensures that the resulting map will make a valid structured key.
+func forceValid(m map[string]string) map[string]string {
+	ret := make(map[string]string, len(m))
+	for key, value := range m {
+		ret[clean(key)] = clean(value)
+	}
+
+	return ret
+}
diff --git a/golden/go/tracestore/types_test.go b/golden/go/tracestore/types_test.go
new file mode 100644
index 0000000..d535146
--- /dev/null
+++ b/golden/go/tracestore/types_test.go
@@ -0,0 +1,42 @@
+package tracestore
+
+import (
+	"testing"
+
+	assert "github.com/stretchr/testify/require"
+	"go.skia.org/infra/go/paramtools"
+	"go.skia.org/infra/go/testutils/unittest"
+	"go.skia.org/infra/go/tiling"
+	"go.skia.org/infra/golden/go/types"
+)
+
+func TestTraceIDFromParams(t *testing.T) {
+	unittest.SmallTest(t)
+
+	input := paramtools.Params{
+		"cpu":                   "x86",
+		"gpu":                   "nVidia",
+		types.PRIMARY_KEY_FIELD: "test_alpha",
+		types.CORPUS_FIELD:      "dm",
+	}
+
+	expected := tiling.TraceId(",cpu=x86,gpu=nVidia,name=test_alpha,source_type=dm,")
+
+	assert.Equal(t, expected, TraceIDFromParams(input))
+}
+
+// TestTraceIDFromParamsMalicious adds some values with invalid chars.
+func TestTraceIDFromParamsMalicious(t *testing.T) {
+	unittest.SmallTest(t)
+
+	input := paramtools.Params{
+		"c=p,u":                 `"x86"`,
+		"gpu":                   "nVi,,=dia",
+		types.PRIMARY_KEY_FIELD: "test=alpha",
+		types.CORPUS_FIELD:      "dm!",
+	}
+
+	expected := tiling.TraceId(`,c_p_u="x86",gpu=nVi___dia,name=test_alpha,source_type=dm!,`)
+
+	assert.Equal(t, expected, TraceIDFromParams(input))
+}
diff --git a/golden/go/types/types.go b/golden/go/types/types.go
index 62b8dc3..10ca302 100644
--- a/golden/go/types/types.go
+++ b/golden/go/types/types.go
@@ -387,6 +387,11 @@
 	return -1
 }
 
+// String prints a human friendly version of this trace.
+func (g *GoldenTrace) String() string {
+	return fmt.Sprintf("Keys: %#v, Digests: %q", g.Keys, g.Digests)
+}
+
 // NewGoldenTrace allocates a new Trace set up for the given number of samples.
 //
 // The Trace Digests are pre-filled in with the missing data sentinel since not
diff --git a/perf/BIGTABLE.md b/perf/BIGTABLE.md
index 6990c05..f219726 100644
--- a/perf/BIGTABLE.md
+++ b/perf/BIGTABLE.md
@@ -94,8 +94,8 @@
 
 Values used in row names:
 
-    TileKey = 2^22 - (tile number)
-       - With 256 values per tile this let's us store 1 billion values per trace.
+    TileKey = (2^32-1) - (tile number)
+       - With 256 values per tile this lets us store 1 trillion values per trace.
        - Formatted as %07d
        - Note that this reverses the order of the tiles, i.e. new tiles have
          smaller numbers, so that we can do a simple query to find the newest tile.