feat(storage/reads): add cache to reuse tags when reading from storage (#16041)

This adds an lru cache for the columns that are produced as tags. When
producing the columns that are part of the group key, it will generate
the column and then keep it in an lru cache to reuse for future tables.
The start and stop column are effectively cached for every table because
they are special and will be the same for all of the tables.

For the tags, it retains the most recently used since they may be used
by a future table. That way most of the columns will get shared with
each other.

When the size differs, a slice is used so the underlying data is still
shared, but the size is different.

storage/reads/reader.go

@@ -36,6 +36,7 @@ func (r *storeReader) ReadFilter(ctx context.Context, spec influxdb.ReadFilterSp
 		ctx:   ctx,
 		s:     r.s,
 		spec:  spec,
+		cache: newTagsCache(0),
 		alloc: alloc,
 	}, nil
 }
@@ -45,6 +46,7 @@ func (r *storeReader) ReadGroup(ctx context.Context, spec influxdb.ReadGroupSpec
 		ctx:   ctx,
 		s:     r.s,
 		spec:  spec,
+		cache: newTagsCache(0),
 		alloc: alloc,
 	}, nil
 }
@@ -96,6 +98,7 @@ type filterIterator struct {
 	s     Store
 	spec  influxdb.ReadFilterSpec
 	stats cursors.CursorStats
+	cache *tagsCache
 	alloc *memory.Allocator
 }
 
@@ -155,6 +158,7 @@ func (fi *filterIterator) handleRead(f func(flux.Table) error, rs ResultSet) err
 			cur.Close()
 		}
 		rs.Close()
+		fi.cache.Release()
 	}()
 
 READ:
@@ -171,19 +175,19 @@ READ:
 		switch typedCur := cur.(type) {
 		case cursors.IntegerArrayCursor:
 			cols, defs := determineTableColsForSeries(rs.Tags(), flux.TInt)
-			table = newIntegerTable(done, typedCur, bnds, key, cols, rs.Tags(), defs, fi.alloc)
+			table = newIntegerTable(done, typedCur, bnds, key, cols, rs.Tags(), defs, fi.cache, fi.alloc)
 		case cursors.FloatArrayCursor:
 			cols, defs := determineTableColsForSeries(rs.Tags(), flux.TFloat)
-			table = newFloatTable(done, typedCur, bnds, key, cols, rs.Tags(), defs, fi.alloc)
+			table = newFloatTable(done, typedCur, bnds, key, cols, rs.Tags(), defs, fi.cache, fi.alloc)
 		case cursors.UnsignedArrayCursor:
 			cols, defs := determineTableColsForSeries(rs.Tags(), flux.TUInt)
-			table = newUnsignedTable(done, typedCur, bnds, key, cols, rs.Tags(), defs, fi.alloc)
+			table = newUnsignedTable(done, typedCur, bnds, key, cols, rs.Tags(), defs, fi.cache, fi.alloc)
 		case cursors.BooleanArrayCursor:
 			cols, defs := determineTableColsForSeries(rs.Tags(), flux.TBool)
-			table = newBooleanTable(done, typedCur, bnds, key, cols, rs.Tags(), defs, fi.alloc)
+			table = newBooleanTable(done, typedCur, bnds, key, cols, rs.Tags(), defs, fi.cache, fi.alloc)
 		case cursors.StringArrayCursor:
 			cols, defs := determineTableColsForSeries(rs.Tags(), flux.TString)
-			table = newStringTable(done, typedCur, bnds, key, cols, rs.Tags(), defs, fi.alloc)
+			table = newStringTable(done, typedCur, bnds, key, cols, rs.Tags(), defs, fi.cache, fi.alloc)
 		default:
 			panic(fmt.Sprintf("unreachable: %T", typedCur))
 		}
@@ -218,6 +222,7 @@ type groupIterator struct {
 	s     Store
 	spec  influxdb.ReadGroupSpec
 	stats cursors.CursorStats
+	cache *tagsCache
 	alloc *memory.Allocator
 }
 
@@ -289,6 +294,7 @@ func (gi *groupIterator) handleRead(f func(flux.Table) error, rs GroupResultSet)
 			gc.Close()
 		}
 		rs.Close()
+		gi.cache.Release()
 	}()
 
 	gc = rs.Next()
@@ -313,19 +319,19 @@ READ:
 		switch typedCur := cur.(type) {
 		case cursors.IntegerArrayCursor:
 			cols, defs := determineTableColsForGroup(gc.Keys(), flux.TInt)
-			table = newIntegerGroupTable(done, gc, typedCur, bnds, key, cols, gc.Tags(), defs, gi.alloc)
+			table = newIntegerGroupTable(done, gc, typedCur, bnds, key, cols, gc.Tags(), defs, gi.cache, gi.alloc)
 		case cursors.FloatArrayCursor:
 			cols, defs := determineTableColsForGroup(gc.Keys(), flux.TFloat)
-			table = newFloatGroupTable(done, gc, typedCur, bnds, key, cols, gc.Tags(), defs, gi.alloc)
+			table = newFloatGroupTable(done, gc, typedCur, bnds, key, cols, gc.Tags(), defs, gi.cache, gi.alloc)
 		case cursors.UnsignedArrayCursor:
 			cols, defs := determineTableColsForGroup(gc.Keys(), flux.TUInt)
-			table = newUnsignedGroupTable(done, gc, typedCur, bnds, key, cols, gc.Tags(), defs, gi.alloc)
+			table = newUnsignedGroupTable(done, gc, typedCur, bnds, key, cols, gc.Tags(), defs, gi.cache, gi.alloc)
 		case cursors.BooleanArrayCursor:
 			cols, defs := determineTableColsForGroup(gc.Keys(), flux.TBool)
-			table = newBooleanGroupTable(done, gc, typedCur, bnds, key, cols, gc.Tags(), defs, gi.alloc)
+			table = newBooleanGroupTable(done, gc, typedCur, bnds, key, cols, gc.Tags(), defs, gi.cache, gi.alloc)
 		case cursors.StringArrayCursor:
 			cols, defs := determineTableColsForGroup(gc.Keys(), flux.TString)
-			table = newStringGroupTable(done, gc, typedCur, bnds, key, cols, gc.Tags(), defs, gi.alloc)
+			table = newStringGroupTable(done, gc, typedCur, bnds, key, cols, gc.Tags(), defs, gi.cache, gi.alloc)
 		default:
 			panic(fmt.Sprintf("unreachable: %T", typedCur))
 		}

storage/reads/table.gen.go

@@ -37,10 +37,11 @@ func newFloatTable(
 	cols []flux.ColMeta,
 	tags models.Tags,
 	defs [][]byte,
+	cache *tagsCache,
 	alloc *memory.Allocator,
 ) *floatTable {
 	t := &floatTable{
-		table: newTable(done, bounds, key, cols, defs, alloc),
+		table: newTable(done, bounds, key, cols, defs, cache, alloc),
 		cur:   cur,
 	}
 	t.readTags(tags)
@@ -113,10 +114,11 @@ func newFloatGroupTable(
 	cols []flux.ColMeta,
 	tags models.Tags,
 	defs [][]byte,
+	cache *tagsCache,
 	alloc *memory.Allocator,
 ) *floatGroupTable {
 	t := &floatGroupTable{
-		table: newTable(done, bounds, key, cols, defs, alloc),
+		table: newTable(done, bounds, key, cols, defs, cache, alloc),
 		gc:    gc,
 		cur:   cur,
 	}
@@ -220,10 +222,11 @@ func newIntegerTable(
 	cols []flux.ColMeta,
 	tags models.Tags,
 	defs [][]byte,
+	cache *tagsCache,
 	alloc *memory.Allocator,
 ) *integerTable {
 	t := &integerTable{
-		table: newTable(done, bounds, key, cols, defs, alloc),
+		table: newTable(done, bounds, key, cols, defs, cache, alloc),
 		cur:   cur,
 	}
 	t.readTags(tags)
@@ -296,10 +299,11 @@ func newIntegerGroupTable(
 	cols []flux.ColMeta,
 	tags models.Tags,
 	defs [][]byte,
+	cache *tagsCache,
 	alloc *memory.Allocator,
 ) *integerGroupTable {
 	t := &integerGroupTable{
-		table: newTable(done, bounds, key, cols, defs, alloc),
+		table: newTable(done, bounds, key, cols, defs, cache, alloc),
 		gc:    gc,
 		cur:   cur,
 	}
@@ -403,10 +407,11 @@ func newUnsignedTable(
 	cols []flux.ColMeta,
 	tags models.Tags,
 	defs [][]byte,
+	cache *tagsCache,
 	alloc *memory.Allocator,
 ) *unsignedTable {
 	t := &unsignedTable{
-		table: newTable(done, bounds, key, cols, defs, alloc),
+		table: newTable(done, bounds, key, cols, defs, cache, alloc),
 		cur:   cur,
 	}
 	t.readTags(tags)
@@ -479,10 +484,11 @@ func newUnsignedGroupTable(
 	cols []flux.ColMeta,
 	tags models.Tags,
 	defs [][]byte,
+	cache *tagsCache,
 	alloc *memory.Allocator,
 ) *unsignedGroupTable {
 	t := &unsignedGroupTable{
-		table: newTable(done, bounds, key, cols, defs, alloc),
+		table: newTable(done, bounds, key, cols, defs, cache, alloc),
 		gc:    gc,
 		cur:   cur,
 	}
@@ -586,10 +592,11 @@ func newStringTable(
 	cols []flux.ColMeta,
 	tags models.Tags,
 	defs [][]byte,
+	cache *tagsCache,
 	alloc *memory.Allocator,
 ) *stringTable {
 	t := &stringTable{
-		table: newTable(done, bounds, key, cols, defs, alloc),
+		table: newTable(done, bounds, key, cols, defs, cache, alloc),
 		cur:   cur,
 	}
 	t.readTags(tags)
@@ -662,10 +669,11 @@ func newStringGroupTable(
 	cols []flux.ColMeta,
 	tags models.Tags,
 	defs [][]byte,
+	cache *tagsCache,
 	alloc *memory.Allocator,
 ) *stringGroupTable {
 	t := &stringGroupTable{
-		table: newTable(done, bounds, key, cols, defs, alloc),
+		table: newTable(done, bounds, key, cols, defs, cache, alloc),
 		gc:    gc,
 		cur:   cur,
 	}
@@ -769,10 +777,11 @@ func newBooleanTable(
 	cols []flux.ColMeta,
 	tags models.Tags,
 	defs [][]byte,
+	cache *tagsCache,
 	alloc *memory.Allocator,
 ) *booleanTable {
 	t := &booleanTable{
-		table: newTable(done, bounds, key, cols, defs, alloc),
+		table: newTable(done, bounds, key, cols, defs, cache, alloc),
 		cur:   cur,
 	}
 	t.readTags(tags)
@@ -845,10 +854,11 @@ func newBooleanGroupTable(
 	cols []flux.ColMeta,
 	tags models.Tags,
 	defs [][]byte,
+	cache *tagsCache,
 	alloc *memory.Allocator,
 ) *booleanGroupTable {
 	t := &booleanGroupTable{
-		table: newTable(done, bounds, key, cols, defs, alloc),
+		table: newTable(done, bounds, key, cols, defs, cache, alloc),
 		gc:    gc,
 		cur:   cur,
 	}

storage/reads/table.gen.go.tmpl

@@ -31,10 +31,11 @@ func new{{.Name}}Table(
 	cols []flux.ColMeta,
 	tags models.Tags,
 	defs [][]byte,
+	cache *tagsCache,
 	alloc *memory.Allocator,
 ) *{{.name}}Table {
 	t := &{{.name}}Table{
-		table: newTable(done, bounds, key, cols, defs, alloc),
+		table: newTable(done, bounds, key, cols, defs, cache, alloc),
 		cur:   cur,
 	}
 	t.readTags(tags)
@@ -107,10 +108,11 @@ func new{{.Name}}GroupTable(
 	cols []flux.ColMeta,
 	tags models.Tags,
 	defs [][]byte,
+	cache *tagsCache,
 	alloc *memory.Allocator,
 ) *{{.name}}GroupTable {
 	t := &{{.name}}GroupTable{
-		table: newTable(done, bounds, key, cols, defs, alloc),
+		table: newTable(done, bounds, key, cols, defs, cache, alloc),
 		gc:    gc,
 		cur:   cur,
 	}

storage/reads/table.go

@@ -31,6 +31,7 @@ type table struct {
 	err error
 
 	cancelled, used int32
+	cache           *tagsCache
 	alloc           *memory.Allocator
 }
 
@@ -40,6 +41,7 @@ func newTable(
 	key flux.GroupKey,
 	cols []flux.ColMeta,
 	defs [][]byte,
+	cache *tagsCache,
 	alloc *memory.Allocator,
 ) table {
 	return table{
@@ -49,6 +51,7 @@ func newTable(
 		tags:   make([][]byte, len(cols)),
 		defs:   defs,
 		cols:   cols,
+		cache:  cache,
 		alloc:  alloc,
 	}
 }
@@ -197,30 +200,15 @@ func (t *table) appendTags(cr *colReader) {
 	for j := range t.cols {
 		v := t.tags[j]
 		if v != nil {
-			b := arrow.NewStringBuilder(t.alloc)
+			cr.cols[j] = t.cache.GetTag(string(v), cr.l, t.alloc)
-			b.Reserve(cr.l)
-			b.ReserveData(cr.l * len(v))
-			for i := 0; i < cr.l; i++ {
-				b.Append(v)
-			}
-			cr.cols[j] = b.NewArray()
-			b.Release()
 		}
 	}
 }
 
 // appendBounds fills the colBufs for the time bounds
 func (t *table) appendBounds(cr *colReader) {
-	bounds := []execute.Time{t.bounds.Start, t.bounds.Stop}
+	start, stop := t.cache.GetBounds(t.bounds, cr.l, t.alloc)
-	for j := range []int{startColIdx, stopColIdx} {
+	cr.cols[startColIdx], cr.cols[stopColIdx] = start, stop
-		b := arrow.NewIntBuilder(t.alloc)
-		b.Reserve(cr.l)
-		for i := 0; i < cr.l; i++ {
-			b.UnsafeAppend(int64(bounds[j]))
-		}
-		cr.cols[j] = b.NewArray()
-		b.Release()
-	}
 }
 
 func (t *table) closeDone() {

storage/reads/table_test.go

@@ -0,0 +1,174 @@
+package reads_test
+
+import (
+	"context"
+	"io/ioutil"
+	"math"
+	"math/rand"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/influxdata/flux"
+	"github.com/influxdata/flux/execute"
+	"github.com/influxdata/flux/memory"
+	"github.com/influxdata/flux/values"
+	"github.com/influxdata/influxdb/cmd/influxd/generate"
+	"github.com/influxdata/influxdb/mock"
+	"github.com/influxdata/influxdb/models"
+	"github.com/influxdata/influxdb/pkg/data/gen"
+	"github.com/influxdata/influxdb/query/stdlib/influxdata/influxdb"
+	"github.com/influxdata/influxdb/storage"
+	"github.com/influxdata/influxdb/storage/reads"
+	"github.com/influxdata/influxdb/storage/readservice"
+	"go.uber.org/zap/zaptest"
+)
+
+func BenchmarkReadFilter(b *testing.B) {
+	idgen := mock.NewMockIDGenerator()
+	tagsSpec := &gen.TagsSpec{
+		Tags: []*gen.TagValuesSpec{
+			{
+				TagKey: "t0",
+				Values: func() gen.CountableSequence {
+					return gen.NewCounterByteSequence("a-%d", 0, 5)
+				},
+			},
+			{
+				TagKey: "t1",
+				Values: func() gen.CountableSequence {
+					return gen.NewCounterByteSequence("b-%d", 0, 1000)
+				},
+			},
+		},
+	}
+	spec := gen.Spec{
+		OrgID:    idgen.ID(),
+		BucketID: idgen.ID(),
+		Measurements: []gen.MeasurementSpec{
+			{
+				Name:     "m0",
+				TagsSpec: tagsSpec,
+				FieldValuesSpec: &gen.FieldValuesSpec{
+					Name: "f0",
+					TimeSequenceSpec: gen.TimeSequenceSpec{
+						Count: math.MaxInt32,
+						Delta: time.Minute,
+					},
+					DataType: models.Float,
+					Values: func(spec gen.TimeSequenceSpec) gen.TimeValuesSequence {
+						r := rand.New(rand.NewSource(10))
+						return gen.NewTimeFloatValuesSequence(
+							spec.Count,
+							gen.NewTimestampSequenceFromSpec(spec),
+							gen.NewFloatRandomValuesSequence(0, 90, r),
+						)
+					},
+				},
+			},
+			{
+				Name:     "m0",
+				TagsSpec: tagsSpec,
+				FieldValuesSpec: &gen.FieldValuesSpec{
+					Name: "f1",
+					TimeSequenceSpec: gen.TimeSequenceSpec{
+						Count: math.MaxInt32,
+						Delta: time.Minute,
+					},
+					DataType: models.Float,
+					Values: func(spec gen.TimeSequenceSpec) gen.TimeValuesSequence {
+						r := rand.New(rand.NewSource(11))
+						return gen.NewTimeFloatValuesSequence(
+							spec.Count,
+							gen.NewTimestampSequenceFromSpec(spec),
+							gen.NewFloatRandomValuesSequence(0, 180, r),
+						)
+					},
+				},
+			},
+			{
+				Name:     "m0",
+				TagsSpec: tagsSpec,
+				FieldValuesSpec: &gen.FieldValuesSpec{
+					Name: "f1",
+					TimeSequenceSpec: gen.TimeSequenceSpec{
+						Count: math.MaxInt32,
+						Delta: time.Minute,
+					},
+					DataType: models.Float,
+					Values: func(spec gen.TimeSequenceSpec) gen.TimeValuesSequence {
+						r := rand.New(rand.NewSource(12))
+						return gen.NewTimeFloatValuesSequence(
+							spec.Count,
+							gen.NewTimestampSequenceFromSpec(spec),
+							gen.NewFloatRandomValuesSequence(10, 10000, r),
+						)
+					},
+				},
+			},
+		},
+	}
+	tr := gen.TimeRange{
+		Start: mustParseTime("2019-11-25T00:00:00Z"),
+		End:   mustParseTime("2019-11-26T00:00:00Z"),
+	}
+	sg := gen.NewSeriesGeneratorFromSpec(&spec, tr)
+	benchmarkRead(b, sg, func(r influxdb.Reader) error {
+		mem := &memory.Allocator{}
+		tables, err := r.ReadFilter(context.Background(), influxdb.ReadFilterSpec{
+			OrganizationID: spec.OrgID,
+			BucketID:       spec.BucketID,
+			Bounds: execute.Bounds{
+				Start: values.ConvertTime(tr.Start),
+				Stop:  values.ConvertTime(tr.End),
+			},
+		}, mem)
+		if err != nil {
+			return err
+		}
+		return tables.Do(func(table flux.Table) error {
+			table.Done()
+			return nil
+		})
+	})
+}
+
+func benchmarkRead(b *testing.B, sg gen.SeriesGenerator, f func(r influxdb.Reader) error) {
+	logger := zaptest.NewLogger(b)
+	rootDir, err := ioutil.TempDir("", "storage-reads-test")
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer func() { _ = os.RemoveAll(rootDir) }()
+
+	generator := generate.Generator{}
+	if _, err := generator.Run(context.Background(), rootDir, sg); err != nil {
+		b.Fatal(err)
+	}
+
+	enginePath := filepath.Join(rootDir, "engine")
+	engine := storage.NewEngine(enginePath, storage.NewConfig())
+	engine.WithLogger(logger)
+
+	if err := engine.Open(context.Background()); err != nil {
+		b.Fatal(err)
+	}
+	reader := reads.NewReader(readservice.NewStore(engine))
+
+	b.ResetTimer()
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		if err := f(reader); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func mustParseTime(s string) time.Time {
+	ts, err := time.Parse(time.RFC3339, s)
+	if err != nil {
+		panic(err)
+	}
+	return ts
+}

storage/reads/tags_cache.go

@@ -0,0 +1,214 @@
+package reads
+
+import (
+	"container/list"
+
+	"github.com/apache/arrow/go/arrow"
+	"github.com/apache/arrow/go/arrow/array"
+	"github.com/apache/arrow/go/arrow/memory"
+	"github.com/influxdata/flux/execute"
+)
+
+// defaultMaxLengthForTagsCache is the default maximum number of
+// tags that will be memoized when retrieving tags from the tags
+// cache.
+const defaultMaxLengthForTagsCache = 100
+
+type tagsCache struct {
+	// startColumn is a special slot for holding the start column.
+	startColumn *array.Int64
+
+	// stopColumn is a special slot for holding the stop column.
+	stopColumn *array.Int64
+
+	// tags holds cached arrays for various tag values.
+	// An lru is used to keep track of the least recently used
+	// item in the cache so that it can be ejected. An lru is used
+	// here because we cannot be certain if tag values are going to
+	// be used again and we do not want to retain a reference
+	// that may have already been released. This makes an lru a good
+	// fit since it will be more likely to eject a value that is not
+	// going to be used again than another data structure.
+	//
+	// The increase in performance by reusing arrays for tag values
+	// is dependent on the order of the tags coming out of storage.
+	// It is possible that a value will be reused but could get
+	// ejected from the cache before it would be reused.
+	//
+	// The map contains the tag **values** and not the tag keys.
+	// An array can get shared among two different tag keys that
+	// have the same value.
+	tags      map[string]*list.Element
+	lru       *list.List
+	maxLength int
+}
+
+// newTagsCache will create a tags cache that will retain
+// the last sz entries. If zero, the default will be used.
+func newTagsCache(sz int) *tagsCache {
+	return &tagsCache{maxLength: sz}
+}
+
+// GetBounds will return arrays that match with the bounds.
+// If an array that is within the cache works with the bounds
+// and can be sliced to the length, a reference to it will be
+// returned.
+func (c *tagsCache) GetBounds(b execute.Bounds, l int, mem memory.Allocator) (start *array.Int64, stop *array.Int64) {
+	if c == nil {
+		start = c.createBounds(b.Start, l, mem)
+		stop = c.createBounds(b.Stop, l, mem)
+		return start, stop
+	}
+
+	if c.startColumn != nil {
+		start = c.getOrReplaceBounds(&c.startColumn, b.Start, l, mem)
+	} else {
+		start = c.createBounds(b.Start, l, mem)
+		start.Retain()
+		c.startColumn = start
+	}
+
+	if c.stopColumn != nil {
+		stop = c.getOrReplaceBounds(&c.stopColumn, b.Stop, l, mem)
+	} else {
+		stop = c.createBounds(b.Stop, l, mem)
+		stop.Retain()
+		c.stopColumn = stop
+	}
+
+	return start, stop
+}
+
+// getOrReplaceBounds will get or replace an array of timestamps
+// and return a new reference to it.
+func (c *tagsCache) getOrReplaceBounds(arr **array.Int64, ts execute.Time, l int, mem memory.Allocator) *array.Int64 {
+	if (*arr).Len() < l {
+		(*arr).Release()
+		*arr = c.createBounds(ts, l, mem)
+		(*arr).Retain()
+		return *arr
+	} else if (*arr).Len() == l {
+		(*arr).Retain()
+		return *arr
+	}
+
+	// If the lengths do not match, but the cached array is less
+	// than the desired array, then we can use slice.
+	// NewSlice will automatically create a new reference to the
+	// passed in array so we do not need to manually retain.
+	data := array.NewSliceData((*arr).Data(), 0, int64(l))
+	vs := array.NewInt64Data(data)
+	data.Release()
+	return vs
+}
+
+func (c *tagsCache) createBounds(ts execute.Time, l int, mem memory.Allocator) *array.Int64 {
+	b := array.NewInt64Builder(mem)
+	b.Resize(l)
+	for i := 0; i < l; i++ {
+		b.Append(int64(ts))
+	}
+	return b.NewInt64Array()
+}
+
+// GetTag returns a binary arrow array that contains the value
+// repeated l times. If an array with a length greater than or
+// equal to the length and with the same value exists in the cache,
+// a reference to the data will be retained and returned.
+// Otherwise, the allocator will be used to construct a new column.
+func (c *tagsCache) GetTag(value string, l int, mem memory.Allocator) *array.Binary {
+	if l == 0 {
+		return c.createTag(value, l, mem)
+	}
+
+	if elem, ok := c.tags[value]; ok {
+		return c.getOrReplaceTag(elem, value, l, mem)
+	}
+
+	arr := c.createTag(value, l, mem)
+	if c.lru == nil {
+		c.lru = list.New()
+	}
+	if c.tags == nil {
+		c.tags = make(map[string]*list.Element)
+	}
+	c.tags[value] = c.lru.PushFront(arr)
+	c.maintainLRU()
+	arr.Retain()
+	return arr
+}
+
+func (c *tagsCache) getOrReplaceTag(elem *list.Element, value string, l int, mem memory.Allocator) *array.Binary {
+	// Move this element to the front of the lru.
+	c.lru.MoveBefore(elem, c.lru.Front())
+
+	// Determine if the array can be reused.
+	arr := elem.Value.(*array.Binary)
+	if arr.Len() < l {
+		// Create a new array with the appropriate length since
+		// this one cannot be reused here.
+		arr.Release()
+		arr = c.createTag(value, l, mem)
+		elem.Value = arr
+		arr.Retain()
+		return arr
+	} else if arr.Len() == l {
+		arr.Retain()
+		return arr
+	}
+
+	// If the lengths do not match, but the cached array is less
+	// than the desired array, then we can use slice.
+	// Slice will automatically create a new reference to the
+	// passed in array so we do not need to manually retain.
+	data := array.NewSliceData(arr.Data(), 0, int64(l))
+	vs := array.NewBinaryData(data)
+	data.Release()
+	return vs
+}
+
+// maintainLRU will ensure the lru cache maintains the appropriate
+// length by ejecting the least recently used value from the cache
+// until the cache is the appropriate size.
+func (c *tagsCache) maintainLRU() {
+	max := c.maxLength
+	if max == 0 {
+		max = defaultMaxLengthForTagsCache
+	}
+	if c.lru.Len() <= max {
+		return
+	}
+	arr := c.lru.Remove(c.lru.Back()).(*array.Binary)
+	value := arr.ValueString(0)
+	delete(c.tags, value)
+	arr.Release()
+}
+
+func (c *tagsCache) createTag(value string, l int, mem memory.Allocator) *array.Binary {
+	b := array.NewBinaryBuilder(mem, arrow.BinaryTypes.String)
+	b.Resize(l)
+	b.ReserveData(l * len(value))
+	for i := 0; i < l; i++ {
+		b.AppendString(value)
+	}
+	return b.NewBinaryArray()
+}
+
+// Release will release all references to cached tag columns.
+func (c *tagsCache) Release() {
+	if c.startColumn != nil {
+		c.startColumn.Release()
+		c.startColumn = nil
+	}
+
+	if c.stopColumn != nil {
+		c.stopColumn.Release()
+		c.stopColumn = nil
+	}
+
+	for _, elem := range c.tags {
+		elem.Value.(*array.Binary).Release()
+	}
+	c.tags = nil
+	c.lru = nil
+}