Repository: polarsignals/frostdb
Pull request #524 (2023-09-07 16:22:39)

Pre hash columns
This adds the ability to specify a column should be prehashed before inserted into the table. 

This allows to aggregator to skip performing the HashArray function for already prehashed columns. This comes with a small alloc cost for aggregations on non pre-hashed data.

```
name                  old time/op    new time/op    delta
Aggregation/sum-10       867µs ± 1%     874µs ± 2%    ~     (p=0.075 n=10+10)
Aggregation/count-10     869µs ± 1%     872µs ± 1%    ~     (p=0.218 n=10+10)
Aggregation/max-10       871µs ± 1%     876µs ± 1%  +0.62%  (p=0.003 n=10+9)

name                  old alloc/op   new alloc/op   delta
Aggregation/sum-10      1.16MB ± 0%    1.16MB ± 0%  +0.22%  (p=0.000 n=10+10)
Aggregation/count-10    1.16MB ± 0%    1.16MB ± 0%  +0.22%  (p=0.000 n=8+10)
Aggregation/max-10      1.16MB ± 0%    1.16MB ± 0%  +0.23%  (p=0.000 n=10+10)

name                  old allocs/op  new allocs/op  delta
Aggregation/sum-10         825 ± 0%       844 ± 0%  +2.30%  (p=0.000 n=9+10)
Aggregation/count-10       824 ± 0%       843 ± 0%  +2.31%  (p=0.000 n=10+9)
Aggregation/max-10         824 ± 0%       844 ± 0%  +2.37%  (p=0.000 n=10+9)
```

thorfour: Comparison on large aggregations 
```
name              old time/op    new time/op    delta
_PreAggregate-10     300ms ± 2%     289ms ± 4%  -3.62%  (p=0.000 n=10+10)

name              old alloc/op   new alloc/op   delta
_PreAggregate-10     603MB ± 0%     612MB ± 0%  +1.37%  (p=0.000 n=10+8)

name              old allocs/op  new allocs/op  delta
_PreAggregate-10     5.80M ± 0%     5.28M ± 0%  -9.02%  (p=0.000 n=10+10)
```
thorfour: Comparison when the data is prehashd
```
name                  old time/op    new time/op    delta
Aggregation/sum-10       379µs ± 0%     299µs ± 3%  -21.04%  (p=0.000 n=9+10)
Aggregation/count-10     374µs ± 1%     293µs ± 2%  -21.81%  (p=0.000 n=9+10)
Aggregation/max-10       384µs ± 1%     301µs ± 0%  -21.69%  (p=0.000 n=9+9)

name                  old alloc/op   new alloc/op   delta
Aggregation/sum-10       400kB ± 0%     401kB ± 0%   +0.06%  (p=0.000 n=9+9)
Aggregation/count-10     401kB ± 0%     401kB ± 0%   +0.06%  (p=0.000 n=10+10)
Aggregation/max-10       400kB ± 0%     401kB ± 0%   +0.06%  (p=0.000 n=10+10)

name                  old allocs/op  new allocs/op  delta
Aggregation/sum-10         686 ± 0%       686 ± 0%     ~     (all equal)
Aggregation/count-10       685 ± 0%       685 ± 0%     ~     (p=0.087 n=10+10)
Aggregation/max-10         686 ± 0%       686 ± 0%     ~     (all equal)
```
thorfour: Latest benchmark
```
thor@thors-MacBook-Pro frostdb % benchstat BenchmarkAggregation BenchmarkAggregation-prehashed
name                  old time/op    new time/op    delta
Aggregation/sum-10       406µs ± 1%     338µs ± 1%  -16.55%  (p=0.000 n=10+10)
Aggregation/count-10     400µs ± 1%     332µs ± 0%  -16.97%  (p=0.000 n=9+9)
Aggregation/max-10       409µs ± 0%     342µs ± 0%  -16.27%  (p=0.000 n=10+10)

name                  old alloc/op   new alloc/op   delta
Aggregation/sum-10       426kB ± 0%     433kB ± 0%   +1.68%  (p=0.000 n=10+7)
Aggregation/count-10     424kB ± 0%     432kB ± 0%   +1.82%  (p=0.000 n=10+10)
Aggregation/max-10       426kB ± 0%     434kB ± 0%   +1.70%  (p=0.000 n=8+10)

name                  old allocs/op  new allocs/op  delta
Aggregation/sum-10         788 ± 0%       853 ± 0%   +8.25%  (p=0.000 n=10+10)
Aggregation/count-10       787 ± 0%       853 ± 0%   +8.39%  (p=0.000 n=10+10)
Aggregation/max-10         788 ± 0%       853 ± 0%   +8.25%  (p=0.000 n=10+10)
```

Got a small bit worse due to some bug fixes

Patch:
diff --git a/dynparquet/example.go b/dynparquet/example.go
index 0579233..c73903b 100644
--- a/dynparquet/example.go
+++ b/dynparquet/example.go
@@ -192,6 +192,64 @@ func ExtractLocationIDs(locs []uuid.UUID) []byte {
 	return b
 }
 
+func PrehashedSampleDefinition() *schemapb.Schema {
+	return &schemapb.Schema{
+		Name: "test",
+		Columns: []*schemapb.Column{{
+			Name: "example_type",
+			StorageLayout: &schemapb.StorageLayout{
+				Type:     schemapb.StorageLayout_TYPE_STRING,
+				Encoding: schemapb.StorageLayout_ENCODING_RLE_DICTIONARY,
+			},
+			Dynamic: false,
+		}, {
+			Name: "labels",
+			StorageLayout: &schemapb.StorageLayout{
+				Type:     schemapb.StorageLayout_TYPE_STRING,
+				Nullable: true,
+				Encoding: schemapb.StorageLayout_ENCODING_RLE_DICTIONARY,
+			},
+			Dynamic: true,
+			Prehash: true,
+		}, {
+			Name: "stacktrace",
+			StorageLayout: &schemapb.StorageLayout{
+				Type:     schemapb.StorageLayout_TYPE_STRING,
+				Encoding: schemapb.StorageLayout_ENCODING_RLE_DICTIONARY,
+			},
+			Dynamic: false,
+			Prehash: true,
+		}, {
+			Name: "timestamp",
+			StorageLayout: &schemapb.StorageLayout{
+				Type: schemapb.StorageLayout_TYPE_INT64,
+			},
+			Dynamic: false,
+		}, {
+			Name: "value",
+			StorageLayout: &schemapb.StorageLayout{
+				Type: schemapb.StorageLayout_TYPE_INT64,
+			},
+			Dynamic: false,
+		}},
+		SortingColumns: []*schemapb.SortingColumn{{
+			Name:      "example_type",
+			Direction: schemapb.SortingColumn_DIRECTION_ASCENDING,
+		}, {
+			Name:       "labels",
+			Direction:  schemapb.SortingColumn_DIRECTION_ASCENDING,
+			NullsFirst: true,
+		}, {
+			Name:      "timestamp",
+			Direction: schemapb.SortingColumn_DIRECTION_ASCENDING,
+		}, {
+			Name:       "stacktrace",
+			Direction:  schemapb.SortingColumn_DIRECTION_ASCENDING,
+			NullsFirst: true,
+		}},
+	}
+}
+
 func SampleDefinition() *schemapb.Schema {
 	return &schemapb.Schema{
 		Name: "test",
diff --git a/dynparquet/hashed.go b/dynparquet/hashed.go
new file mode 100644
index 0000000..61797ea
--- /dev/null
+++ b/dynparquet/hashed.go
@@ -0,0 +1,200 @@
+package dynparquet
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/apache/arrow/go/v14/arrow"
+	"github.com/apache/arrow/go/v14/arrow/array"
+	"github.com/apache/arrow/go/v14/arrow/memory"
+	"github.com/cespare/xxhash/v2"
+	"github.com/dgryski/go-metro"
+
+	"github.com/polarsignals/frostdb/pqarrow/arrowutils"
+)
+
+const prehashedPrefix = "hashed"
+
+func HashedColumnName(col string) string {
+	return prehashedPrefix + "." + col
+}
+
+func IsHashedColumn(col string) bool {
+	return strings.HasPrefix(col, prehashedPrefix)
+}
+
+// findHashedColumn finds the index of the column in the given fields that have been prehashed.
+func FindHashedColumn(col string, fields []arrow.Field) int {
+	for j, f := range fields {
+		if HashedColumnName(col) == f.Name {
+			return j
+		}
+	}
+
+	return -1
+}
+
+// prehashColumns prehashes the columns in the given record that have been marked as prehashed in the given schema.
+func PrehashColumns(schema *Schema, r arrow.Record) arrow.Record {
+	bldr := array.NewInt64Builder(memory.DefaultAllocator) // TODO pass in allocator
+	defer bldr.Release()
+
+	fields := r.Schema().Fields()
+	additionalFields := make([]arrow.Field, 0, len(fields))
+	additionalColumns := make([]arrow.Array, 0, len(fields))
+	defer func() {
+		for _, col := range additionalColumns {
+			col.Release()
+		}
+	}()
+
+	for _, col := range schema.Columns() {
+		if !col.PreHash {
+			continue
+		}
+
+		for i, f := range fields {
+			if col.Name == f.Name || (col.Dynamic && strings.HasPrefix(f.Name, col.Name)) {
+				additionalFields = append(additionalFields, arrow.Field{
+					Name: HashedColumnName(f.Name),
+					Type: arrow.PrimitiveTypes.Int64,
+				})
+
+				// Hash the column
+				hashed := HashArray(r.Column(i))
+
+				// Build the new column
+				bldr.Reserve(len(hashed))
+				for _, v := range hashed {
+					bldr.UnsafeAppend(int64(v))
+				}
+
+				additionalColumns = append(additionalColumns, bldr.NewArray())
+			}
+		}
+	}
+
+	if len(additionalColumns) == 0 {
+		r.Retain() // NOTE: we retain here because we expect the caller to release the record that we're returning
+		return r
+	}
+
+	sch := arrow.NewSchema(append(fields, additionalFields...), nil)
+	return array.NewRecord(sch, append(r.Columns(), additionalColumns...), r.NumRows())
+}
+
+func HashArray(arr arrow.Array) []uint64 {
+	switch ar := arr.(type) {
+	case *array.String:
+		return hashStringArray(ar)
+	case *array.Binary:
+		return hashBinaryArray(ar)
+	case *array.Int64:
+		return hashInt64Array(ar)
+	case *array.Boolean:
+		return hashBooleanArray(ar)
+	case *array.Dictionary:
+		return hashDictionaryArray(ar)
+	case *array.List:
+		return hashListArray(ar)
+	default:
+		panic("unsupported array type " + fmt.Sprintf("%T", arr))
+	}
+}
+
+func hashListArray(arr *array.List) []uint64 {
+	res := make([]uint64, arr.Len())
+	digest := xxhash.New()
+	for i := 0; i < arr.Len(); i++ {
+		if err := arrowutils.ForEachValueInList(i, arr, func(_ int, v any) {
+			switch val := v.(type) {
+			case []byte:
+				// NOTE: We can ignore the return here because from the xxhash Write function is says: It always returns len(b), nil.
+				_, _ = digest.Write(val)
+			}
+		}); err != nil {
+			panic(err)
+		}
+
+		res[i] = digest.Sum64()
+		digest.Reset()
+	}
+	return res
+}
+
+func hashDictionaryArray(arr *array.Dictionary) []uint64 {
+	res := make([]uint64, arr.Len())
+	for i := 0; i < arr.Len(); i++ {
+		if !arr.IsNull(i) {
+			switch dict := arr.Dictionary().(type) {
+			case *array.Binary:
+				res[i] = metro.Hash64(dict.Value(arr.GetValueIndex(i)), 0)
+			case *array.String:
+				res[i] = metro.Hash64([]byte(dict.Value(arr.GetValueIndex(i))), 0)
+			default:
+				panic("unsupported dictionary type " + fmt.Sprintf("%T", dict))
+			}
+		}
+	}
+	return res
+}
+
+func hashBinaryArray(arr *array.Binary) []uint64 {
+	res := make([]uint64, arr.Len())
+	for i := 0; i < arr.Len(); i++ {
+		if !arr.IsNull(i) {
+			res[i] = metro.Hash64(arr.Value(i), 0)
+		}
+	}
+	return res
+}
+
+func hashBooleanArray(arr *array.Boolean) []uint64 {
+	res := make([]uint64, arr.Len())
+	for i := 0; i < arr.Len(); i++ {
+		if arr.IsNull(i) {
+			res[i] = 0
+			continue
+		}
+		if arr.Value(i) {
+			res[i] = 2
+		} else {
+			res[i] = 1
+		}
+	}
+	return res
+}
+
+func hashStringArray(arr *array.String) []uint64 {
+	res := make([]uint64, arr.Len())
+	for i := 0; i < arr.Len(); i++ {
+		if !arr.IsNull(i) {
+			res[i] = metro.Hash64([]byte(arr.Value(i)), 0)
+		}
+	}
+	return res
+}
+
+func hashInt64Array(arr *array.Int64) []uint64 {
+	res := make([]uint64, arr.Len())
+	for i := 0; i < arr.Len(); i++ {
+		if !arr.IsNull(i) {
+			res[i] = uint64(arr.Value(i))
+		}
+	}
+	return res
+}
+
+// RemoveHashedColumns removes the hashed columns from the record.
+func RemoveHashedColumns(r arrow.Record) arrow.Record {
+	cols := make([]arrow.Array, 0, r.Schema().NumFields())
+	fields := make([]arrow.Field, 0, r.Schema().NumFields())
+	for i := 0; i < r.Schema().NumFields(); i++ {
+		if !IsHashedColumn(r.Schema().Field(i).Name) {
+			cols = append(cols, r.Column(i))
+			fields = append(fields, r.Schema().Field(i))
+		}
+	}
+
+	return array.NewRecord(arrow.NewSchema(fields, nil), cols, r.NumRows())
+}
diff --git a/dynparquet/schema.go b/dynparquet/schema.go
index 9e38336..eb7832c 100644
--- a/dynparquet/schema.go
+++ b/dynparquet/schema.go
@@ -30,6 +30,7 @@ type ColumnDefinition struct {
 	Name          string
 	StorageLayout parquet.Node
 	Dynamic       bool
+	PreHash       bool
 }
 
 // SortingColumn describes a column to sort by in a dynamic parquet schema.
@@ -215,6 +216,7 @@ func SchemaFromDefinition(msg proto.Message) (*Schema, error) {
 				Name:          col.Name,
 				StorageLayout: layout,
 				Dynamic:       col.Dynamic,
+				PreHash:       col.Prehash,
 			})
 		}
 
@@ -611,12 +613,7 @@ func (s *Schema) ParquetSchema() *parquet.Schema {
 
 // dynamicParquetSchema returns the parquet schema for the dynamic schema with the
 // concrete dynamic column names given in the argument.
-func (s Schema) dynamicParquetSchema(
-	dynamicColumns map[string][]string,
-) (
-	*parquet.Schema,
-	error,
-) {
+func (s Schema) dynamicParquetSchema(dynamicColumns map[string][]string) (*parquet.Schema, error) {
 	switch def := s.def.(type) {
 	case *schemav2pb.Schema:
 		return ParquetSchemaFromV2Definition(def), nil
@@ -627,10 +624,16 @@ func (s Schema) dynamicParquetSchema(
 				dyn := dynamicColumnsFor(col.Name, dynamicColumns)
 				for _, name := range dyn {
 					g[col.Name+"."+name] = col.StorageLayout
+					if col.PreHash {
+						g[HashedColumnName(col.Name+"."+name)] = parquet.Int(64) // TODO(thor): Do we need compression etc. here?
+					}
 				}
 				continue
 			}
 			g[col.Name] = col.StorageLayout
+			if col.PreHash {
+				g[HashedColumnName(col.Name)] = parquet.Int(64) // TODO(thor): Do we need compression etc. here?
+			}
 		}
 
 		return parquet.NewSchema(s.Name(), g), nil
diff --git a/gen/proto/go/frostdb/schema/v1alpha1/schema.pb.go b/gen/proto/go/frostdb/schema/v1alpha1/schema.pb.go
index c37b99c..da5b72f 100644
--- a/gen/proto/go/frostdb/schema/v1alpha1/schema.pb.go
+++ b/gen/proto/go/frostdb/schema/v1alpha1/schema.pb.go
@@ -339,6 +339,9 @@ type Column struct {
 	StorageLayout *StorageLayout `protobuf:"bytes,2,opt,name=storage_layout,json=storageLayout,proto3" json:"storage_layout,omitempty"`
 	// Whether the column can dynamically expand.
 	Dynamic bool `protobuf:"varint,3,opt,name=dynamic,proto3" json:"dynamic,omitempty"`
+	// Prehash the column before storing it. This is an optimization to speed up aggregation queries when this column is often aggregated.
+	// This will create a separate non-dynamic column with the same name and the prefix "hashed." that contains the prehashed values.
+	Prehash bool `protobuf:"varint,4,opt,name=prehash,proto3" json:"prehash,omitempty"`
 }
 
 func (x *Column) Reset() {
@@ -394,6 +397,13 @@ func (x *Column) GetDynamic() bool {
 	return false
 }
 
+func (x *Column) GetPrehash() bool {
+	if x != nil {
+		return x.Prehash
+	}
+	return false
+}
+
 // Storage layout describes the physical storage properties of a column.
 type StorageLayout struct {
 	state         protoimpl.MessageState
@@ -563,7 +573,7 @@ var file_frostdb_schema_v1alpha1_schema_proto_rawDesc = []byte{
 	0x28, 0x0b, 0x32, 0x26, 0x2e, 0x66, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x2e, 0x73, 0x63, 0x68,
 	0x65, 0x6d, 0x61, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x2e, 0x53, 0x6f, 0x72,
 	0x74, 0x69, 0x6e, 0x67, 0x43, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x52, 0x0e, 0x73, 0x6f, 0x72, 0x74,
-	0x69, 0x6e, 0x67, 0x43, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x73, 0x22, 0x85, 0x01, 0x0a, 0x06, 0x43,
+	0x69, 0x6e, 0x67, 0x43, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x73, 0x22, 0x9f, 0x01, 0x0a, 0x06, 0x43,
 	0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20,
 	0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x4d, 0x0a, 0x0e, 0x73, 0x74, 0x6f,
 	0x72, 0x61, 0x67, 0x65, 0x5f, 0x6c, 0x61, 0x79, 0x6f, 0x75, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28,
@@ -572,86 +582,87 @@ var file_frostdb_schema_v1alpha1_schema_proto_rawDesc = []byte{
 	0x61, 0x67, 0x65, 0x4c, 0x61, 0x79, 0x6f, 0x75, 0x74, 0x52, 0x0d, 0x73, 0x74, 0x6f, 0x72, 0x61,
 	0x67, 0x65, 0x4c, 0x61, 0x79, 0x6f, 0x75, 0x74, 0x12, 0x18, 0x0a, 0x07, 0x64, 0x79, 0x6e, 0x61,
 	0x6d, 0x69, 0x63, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x07, 0x64, 0x79, 0x6e, 0x61, 0x6d,
-	0x69, 0x63, 0x22, 0xea, 0x05, 0x0a, 0x0d, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x4c, 0x61,
-	0x79, 0x6f, 0x75, 0x74, 0x12, 0x3f, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01,
-	0x28, 0x0e, 0x32, 0x2b, 0x2e, 0x66, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x2e, 0x73, 0x63, 0x68,
-	0x65, 0x6d, 0x61, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x2e, 0x53, 0x74, 0x6f,
-	0x72, 0x61, 0x67, 0x65, 0x4c, 0x61, 0x79, 0x6f, 0x75, 0x74, 0x2e, 0x54, 0x79, 0x70, 0x65, 0x52,
-	0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x4b, 0x0a, 0x08, 0x65, 0x6e, 0x63, 0x6f, 0x64, 0x69, 0x6e,
-	0x67, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x2f, 0x2e, 0x66, 0x72, 0x6f, 0x73, 0x74, 0x64,
-	0x62, 0x2e, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61,
-	0x31, 0x2e, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x4c, 0x61, 0x79, 0x6f, 0x75, 0x74, 0x2e,
-	0x45, 0x6e, 0x63, 0x6f, 0x64, 0x69, 0x6e, 0x67, 0x52, 0x08, 0x65, 0x6e, 0x63, 0x6f, 0x64, 0x69,
-	0x6e, 0x67, 0x12, 0x54, 0x0a, 0x0b, 0x63, 0x6f, 0x6d, 0x70, 0x72, 0x65, 0x73, 0x73, 0x69, 0x6f,
-	0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x32, 0x2e, 0x66, 0x72, 0x6f, 0x73, 0x74, 0x64,
-	0x62, 0x2e, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61,
-	0x31, 0x2e, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x4c, 0x61, 0x79, 0x6f, 0x75, 0x74, 0x2e,
-	0x43, 0x6f, 0x6d, 0x70, 0x72, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x0b, 0x63, 0x6f, 0x6d,
-	0x70, 0x72, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x1a, 0x0a, 0x08, 0x6e, 0x75, 0x6c, 0x6c,
-	0x61, 0x62, 0x6c, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x08, 0x52, 0x08, 0x6e, 0x75, 0x6c, 0x6c,
-	0x61, 0x62, 0x6c, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x72, 0x65, 0x70, 0x65, 0x61, 0x74, 0x65, 0x64,
-	0x18, 0x05, 0x20, 0x01, 0x28, 0x08, 0x52, 0x08, 0x72, 0x65, 0x70, 0x65, 0x61, 0x74, 0x65, 0x64,
-	0x22, 0x65, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x12, 0x1c, 0x0a, 0x18, 0x54, 0x59, 0x50, 0x45,
-	0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49,
-	0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x0f, 0x0a, 0x0b, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x53,
-	0x54, 0x52, 0x49, 0x4e, 0x47, 0x10, 0x01, 0x12, 0x0e, 0x0a, 0x0a, 0x54, 0x59, 0x50, 0x45, 0x5f,
-	0x49, 0x4e, 0x54, 0x36, 0x34, 0x10, 0x02, 0x12, 0x0f, 0x0a, 0x0b, 0x54, 0x59, 0x50, 0x45, 0x5f,
-	0x44, 0x4f, 0x55, 0x42, 0x4c, 0x45, 0x10, 0x03, 0x12, 0x0d, 0x0a, 0x09, 0x54, 0x59, 0x50, 0x45,
-	0x5f, 0x42, 0x4f, 0x4f, 0x4c, 0x10, 0x04, 0x22, 0xae, 0x01, 0x0a, 0x08, 0x45, 0x6e, 0x63, 0x6f,
-	0x64, 0x69, 0x6e, 0x67, 0x12, 0x1e, 0x0a, 0x1a, 0x45, 0x4e, 0x43, 0x4f, 0x44, 0x49, 0x4e, 0x47,
-	0x5f, 0x50, 0x4c, 0x41, 0x49, 0x4e, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49,
-	0x45, 0x44, 0x10, 0x00, 0x12, 0x1b, 0x0a, 0x17, 0x45, 0x4e, 0x43, 0x4f, 0x44, 0x49, 0x4e, 0x47,
-	0x5f, 0x52, 0x4c, 0x45, 0x5f, 0x44, 0x49, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x41, 0x52, 0x59, 0x10,
-	0x01, 0x12, 0x20, 0x0a, 0x1c, 0x45, 0x4e, 0x43, 0x4f, 0x44, 0x49, 0x4e, 0x47, 0x5f, 0x44, 0x45,
-	0x4c, 0x54, 0x41, 0x5f, 0x42, 0x49, 0x4e, 0x41, 0x52, 0x59, 0x5f, 0x50, 0x41, 0x43, 0x4b, 0x45,
-	0x44, 0x10, 0x02, 0x12, 0x1d, 0x0a, 0x19, 0x45, 0x4e, 0x43, 0x4f, 0x44, 0x49, 0x4e, 0x47, 0x5f,
-	0x44, 0x45, 0x4c, 0x54, 0x41, 0x5f, 0x42, 0x59, 0x54, 0x45, 0x5f, 0x41, 0x52, 0x52, 0x41, 0x59,
-	0x10, 0x03, 0x12, 0x24, 0x0a, 0x20, 0x45, 0x4e, 0x43, 0x4f, 0x44, 0x49, 0x4e, 0x47, 0x5f, 0x44,
-	0x45, 0x4c, 0x54, 0x41, 0x5f, 0x4c, 0x45, 0x4e, 0x47, 0x54, 0x48, 0x5f, 0x42, 0x59, 0x54, 0x45,
-	0x5f, 0x41, 0x52, 0x52, 0x41, 0x59, 0x10, 0x04, 0x22, 0xa4, 0x01, 0x0a, 0x0b, 0x43, 0x6f, 0x6d,
-	0x70, 0x72, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x20, 0x0a, 0x1c, 0x43, 0x4f, 0x4d, 0x50,
-	0x52, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x5f, 0x4e, 0x4f, 0x4e, 0x45, 0x5f, 0x55, 0x4e, 0x53,
-	0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x16, 0x0a, 0x12, 0x43, 0x4f,
-	0x4d, 0x50, 0x52, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x5f, 0x53, 0x4e, 0x41, 0x50, 0x50, 0x59,
-	0x10, 0x01, 0x12, 0x14, 0x0a, 0x10, 0x43, 0x4f, 0x4d, 0x50, 0x52, 0x45, 0x53, 0x53, 0x49, 0x4f,
-	0x4e, 0x5f, 0x47, 0x5a, 0x49, 0x50, 0x10, 0x02, 0x12, 0x16, 0x0a, 0x12, 0x43, 0x4f, 0x4d, 0x50,
-	0x52, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x5f, 0x42, 0x52, 0x4f, 0x54, 0x4c, 0x49, 0x10, 0x03,
-	0x12, 0x17, 0x0a, 0x13, 0x43, 0x4f, 0x4d, 0x50, 0x52, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x5f,
-	0x4c, 0x5a, 0x34, 0x5f, 0x52, 0x41, 0x57, 0x10, 0x04, 0x12, 0x14, 0x0a, 0x10, 0x43, 0x4f, 0x4d,
-	0x50, 0x52, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x5f, 0x5a, 0x53, 0x54, 0x44, 0x10, 0x05, 0x22,
-	0xf7, 0x01, 0x0a, 0x0d, 0x53, 0x6f, 0x72, 0x74, 0x69, 0x6e, 0x67, 0x43, 0x6f, 0x6c, 0x75, 0x6d,
-	0x6e, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x4e, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69,
-	0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x30, 0x2e, 0x66, 0x72, 0x6f, 0x73, 0x74,
-	0x64, 0x62, 0x2e, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68,
-	0x61, 0x31, 0x2e, 0x53, 0x6f, 0x72, 0x74, 0x69, 0x6e, 0x67, 0x43, 0x6f, 0x6c, 0x75, 0x6d, 0x6e,
-	0x2e, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x64, 0x69, 0x72, 0x65,
-	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1f, 0x0a, 0x0b, 0x6e, 0x75, 0x6c, 0x6c, 0x73, 0x5f, 0x66,
-	0x69, 0x72, 0x73, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0a, 0x6e, 0x75, 0x6c, 0x6c,
-	0x73, 0x46, 0x69, 0x72, 0x73, 0x74, 0x22, 0x61, 0x0a, 0x09, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74,
-	0x69, 0x6f, 0x6e, 0x12, 0x21, 0x0a, 0x1d, 0x44, 0x49, 0x52, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e,
-	0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49,
-	0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x17, 0x0a, 0x13, 0x44, 0x49, 0x52, 0x45, 0x43, 0x54,
-	0x49, 0x4f, 0x4e, 0x5f, 0x41, 0x53, 0x43, 0x45, 0x4e, 0x44, 0x49, 0x4e, 0x47, 0x10, 0x01, 0x12,
-	0x18, 0x0a, 0x14, 0x44, 0x49, 0x52, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x5f, 0x44, 0x45, 0x53,
-	0x43, 0x45, 0x4e, 0x44, 0x49, 0x4e, 0x47, 0x10, 0x02, 0x42, 0xfd, 0x01, 0x0a, 0x1b, 0x63, 0x6f,
-	0x6d, 0x2e, 0x66, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x2e, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61,
-	0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x42, 0x0b, 0x53, 0x63, 0x68, 0x65, 0x6d,
-	0x61, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x53, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62,
-	0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x70, 0x6f, 0x6c, 0x61, 0x72, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c,
-	0x73, 0x2f, 0x66, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x2f, 0x67, 0x65, 0x6e, 0x2f, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x2f, 0x67, 0x6f, 0x2f, 0x66, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x2f, 0x73,
-	0x63, 0x68, 0x65, 0x6d, 0x61, 0x2f, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x3b, 0x73,
-	0x63, 0x68, 0x65, 0x6d, 0x61, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0xa2, 0x02, 0x03,
-	0x46, 0x53, 0x58, 0xaa, 0x02, 0x17, 0x46, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x2e, 0x53, 0x63,
-	0x68, 0x65, 0x6d, 0x61, 0x2e, 0x56, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0xca, 0x02, 0x17,
-	0x46, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x5c, 0x53, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x5c, 0x56,
-	0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0xe2, 0x02, 0x23, 0x46, 0x72, 0x6f, 0x73, 0x74, 0x64,
+	0x69, 0x63, 0x12, 0x18, 0x0a, 0x07, 0x70, 0x72, 0x65, 0x68, 0x61, 0x73, 0x68, 0x18, 0x04, 0x20,
+	0x01, 0x28, 0x08, 0x52, 0x07, 0x70, 0x72, 0x65, 0x68, 0x61, 0x73, 0x68, 0x22, 0xea, 0x05, 0x0a,
+	0x0d, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x4c, 0x61, 0x79, 0x6f, 0x75, 0x74, 0x12, 0x3f,
+	0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x2b, 0x2e, 0x66,
+	0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x2e, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x2e, 0x76, 0x31,
+	0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x2e, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x4c, 0x61,
+	0x79, 0x6f, 0x75, 0x74, 0x2e, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12,
+	0x4b, 0x0a, 0x08, 0x65, 0x6e, 0x63, 0x6f, 0x64, 0x69, 0x6e, 0x67, 0x18, 0x02, 0x20, 0x01, 0x28,
+	0x0e, 0x32, 0x2f, 0x2e, 0x66, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x2e, 0x73, 0x63, 0x68, 0x65,
+	0x6d, 0x61, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x2e, 0x53, 0x74, 0x6f, 0x72,
+	0x61, 0x67, 0x65, 0x4c, 0x61, 0x79, 0x6f, 0x75, 0x74, 0x2e, 0x45, 0x6e, 0x63, 0x6f, 0x64, 0x69,
+	0x6e, 0x67, 0x52, 0x08, 0x65, 0x6e, 0x63, 0x6f, 0x64, 0x69, 0x6e, 0x67, 0x12, 0x54, 0x0a, 0x0b,
+	0x63, 0x6f, 0x6d, 0x70, 0x72, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28,
+	0x0e, 0x32, 0x32, 0x2e, 0x66, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x2e, 0x73, 0x63, 0x68, 0x65,
+	0x6d, 0x61, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x2e, 0x53, 0x74, 0x6f, 0x72,
+	0x61, 0x67, 0x65, 0x4c, 0x61, 0x79, 0x6f, 0x75, 0x74, 0x2e, 0x43, 0x6f, 0x6d, 0x70, 0x72, 0x65,
+	0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x0b, 0x63, 0x6f, 0x6d, 0x70, 0x72, 0x65, 0x73, 0x73, 0x69,
+	0x6f, 0x6e, 0x12, 0x1a, 0x0a, 0x08, 0x6e, 0x75, 0x6c, 0x6c, 0x61, 0x62, 0x6c, 0x65, 0x18, 0x04,
+	0x20, 0x01, 0x28, 0x08, 0x52, 0x08, 0x6e, 0x75, 0x6c, 0x6c, 0x61, 0x62, 0x6c, 0x65, 0x12, 0x1a,
+	0x0a, 0x08, 0x72, 0x65, 0x70, 0x65, 0x61, 0x74, 0x65, 0x64, 0x18, 0x05, 0x20, 0x01, 0x28, 0x08,
+	0x52, 0x08, 0x72, 0x65, 0x70, 0x65, 0x61, 0x74, 0x65, 0x64, 0x22, 0x65, 0x0a, 0x04, 0x54, 0x79,
+	0x70, 0x65, 0x12, 0x1c, 0x0a, 0x18, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f,
+	0x57, 0x4e, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00,
+	0x12, 0x0f, 0x0a, 0x0b, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x53, 0x54, 0x52, 0x49, 0x4e, 0x47, 0x10,
+	0x01, 0x12, 0x0e, 0x0a, 0x0a, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x49, 0x4e, 0x54, 0x36, 0x34, 0x10,
+	0x02, 0x12, 0x0f, 0x0a, 0x0b, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x44, 0x4f, 0x55, 0x42, 0x4c, 0x45,
+	0x10, 0x03, 0x12, 0x0d, 0x0a, 0x09, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x42, 0x4f, 0x4f, 0x4c, 0x10,
+	0x04, 0x22, 0xae, 0x01, 0x0a, 0x08, 0x45, 0x6e, 0x63, 0x6f, 0x64, 0x69, 0x6e, 0x67, 0x12, 0x1e,
+	0x0a, 0x1a, 0x45, 0x4e, 0x43, 0x4f, 0x44, 0x49, 0x4e, 0x47, 0x5f, 0x50, 0x4c, 0x41, 0x49, 0x4e,
+	0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x1b,
+	0x0a, 0x17, 0x45, 0x4e, 0x43, 0x4f, 0x44, 0x49, 0x4e, 0x47, 0x5f, 0x52, 0x4c, 0x45, 0x5f, 0x44,
+	0x49, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x41, 0x52, 0x59, 0x10, 0x01, 0x12, 0x20, 0x0a, 0x1c, 0x45,
+	0x4e, 0x43, 0x4f, 0x44, 0x49, 0x4e, 0x47, 0x5f, 0x44, 0x45, 0x4c, 0x54, 0x41, 0x5f, 0x42, 0x49,
+	0x4e, 0x41, 0x52, 0x59, 0x5f, 0x50, 0x41, 0x43, 0x4b, 0x45, 0x44, 0x10, 0x02, 0x12, 0x1d, 0x0a,
+	0x19, 0x45, 0x4e, 0x43, 0x4f, 0x44, 0x49, 0x4e, 0x47, 0x5f, 0x44, 0x45, 0x4c, 0x54, 0x41, 0x5f,
+	0x42, 0x59, 0x54, 0x45, 0x5f, 0x41, 0x52, 0x52, 0x41, 0x59, 0x10, 0x03, 0x12, 0x24, 0x0a, 0x20,
+	0x45, 0x4e, 0x43, 0x4f, 0x44, 0x49, 0x4e, 0x47, 0x5f, 0x44, 0x45, 0x4c, 0x54, 0x41, 0x5f, 0x4c,
+	0x45, 0x4e, 0x47, 0x54, 0x48, 0x5f, 0x42, 0x59, 0x54, 0x45, 0x5f, 0x41, 0x52, 0x52, 0x41, 0x59,
+	0x10, 0x04, 0x22, 0xa4, 0x01, 0x0a, 0x0b, 0x43, 0x6f, 0x6d, 0x70, 0x72, 0x65, 0x73, 0x73, 0x69,
+	0x6f, 0x6e, 0x12, 0x20, 0x0a, 0x1c, 0x43, 0x4f, 0x4d, 0x50, 0x52, 0x45, 0x53, 0x53, 0x49, 0x4f,
+	0x4e, 0x5f, 0x4e, 0x4f, 0x4e, 0x45, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49,
+	0x45, 0x44, 0x10, 0x00, 0x12, 0x16, 0x0a, 0x12, 0x43, 0x4f, 0x4d, 0x50, 0x52, 0x45, 0x53, 0x53,
+	0x49, 0x4f, 0x4e, 0x5f, 0x53, 0x4e, 0x41, 0x50, 0x50, 0x59, 0x10, 0x01, 0x12, 0x14, 0x0a, 0x10,
+	0x43, 0x4f, 0x4d, 0x50, 0x52, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x5f, 0x47, 0x5a, 0x49, 0x50,
+	0x10, 0x02, 0x12, 0x16, 0x0a, 0x12, 0x43, 0x4f, 0x4d, 0x50, 0x52, 0x45, 0x53, 0x53, 0x49, 0x4f,
+	0x4e, 0x5f, 0x42, 0x52, 0x4f, 0x54, 0x4c, 0x49, 0x10, 0x03, 0x12, 0x17, 0x0a, 0x13, 0x43, 0x4f,
+	0x4d, 0x50, 0x52, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x5f, 0x4c, 0x5a, 0x34, 0x5f, 0x52, 0x41,
+	0x57, 0x10, 0x04, 0x12, 0x14, 0x0a, 0x10, 0x43, 0x4f, 0x4d, 0x50, 0x52, 0x45, 0x53, 0x53, 0x49,
+	0x4f, 0x4e, 0x5f, 0x5a, 0x53, 0x54, 0x44, 0x10, 0x05, 0x22, 0xf7, 0x01, 0x0a, 0x0d, 0x53, 0x6f,
+	0x72, 0x74, 0x69, 0x6e, 0x67, 0x43, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x12, 0x12, 0x0a, 0x04, 0x6e,
+	0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12,
+	0x4e, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01,
+	0x28, 0x0e, 0x32, 0x30, 0x2e, 0x66, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x2e, 0x73, 0x63, 0x68,
+	0x65, 0x6d, 0x61, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x2e, 0x53, 0x6f, 0x72,
+	0x74, 0x69, 0x6e, 0x67, 0x43, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x2e, 0x44, 0x69, 0x72, 0x65, 0x63,
+	0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12,
+	0x1f, 0x0a, 0x0b, 0x6e, 0x75, 0x6c, 0x6c, 0x73, 0x5f, 0x66, 0x69, 0x72, 0x73, 0x74, 0x18, 0x03,
+	0x20, 0x01, 0x28, 0x08, 0x52, 0x0a, 0x6e, 0x75, 0x6c, 0x6c, 0x73, 0x46, 0x69, 0x72, 0x73, 0x74,
+	0x22, 0x61, 0x0a, 0x09, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x21, 0x0a,
+	0x1d, 0x44, 0x49, 0x52, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f,
+	0x57, 0x4e, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00,
+	0x12, 0x17, 0x0a, 0x13, 0x44, 0x49, 0x52, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x5f, 0x41, 0x53,
+	0x43, 0x45, 0x4e, 0x44, 0x49, 0x4e, 0x47, 0x10, 0x01, 0x12, 0x18, 0x0a, 0x14, 0x44, 0x49, 0x52,
+	0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x5f, 0x44, 0x45, 0x53, 0x43, 0x45, 0x4e, 0x44, 0x49, 0x4e,
+	0x47, 0x10, 0x02, 0x42, 0xfd, 0x01, 0x0a, 0x1b, 0x63, 0x6f, 0x6d, 0x2e, 0x66, 0x72, 0x6f, 0x73,
+	0x74, 0x64, 0x62, 0x2e, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70,
+	0x68, 0x61, 0x31, 0x42, 0x0b, 0x53, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x50, 0x72, 0x6f, 0x74, 0x6f,
+	0x50, 0x01, 0x5a, 0x53, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x70,
+	0x6f, 0x6c, 0x61, 0x72, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c, 0x73, 0x2f, 0x66, 0x72, 0x6f, 0x73,
+	0x74, 0x64, 0x62, 0x2f, 0x67, 0x65, 0x6e, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x67, 0x6f,
+	0x2f, 0x66, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x2f, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x2f,
+	0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x3b, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x76,
+	0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0xa2, 0x02, 0x03, 0x46, 0x53, 0x58, 0xaa, 0x02, 0x17,
+	0x46, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x2e, 0x53, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x2e, 0x56,
+	0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0xca, 0x02, 0x17, 0x46, 0x72, 0x6f, 0x73, 0x74, 0x64,
 	0x62, 0x5c, 0x53, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x5c, 0x56, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61,
-	0x31, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x19,
-	0x46, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x3a, 0x3a, 0x53, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x3a,
-	0x3a, 0x56, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f,
-	0x33,
+	0x31, 0xe2, 0x02, 0x23, 0x46, 0x72, 0x6f, 0x73, 0x74, 0x64, 0x62, 0x5c, 0x53, 0x63, 0x68, 0x65,
+	0x6d, 0x61, 0x5c, 0x56, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x5c, 0x47, 0x50, 0x42, 0x4d,
+	0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x19, 0x46, 0x72, 0x6f, 0x73, 0x74, 0x64,
+	0x62, 0x3a, 0x3a, 0x53, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x3a, 0x3a, 0x56, 0x31, 0x61, 0x6c, 0x70,
+	0x68, 0x61, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
diff --git a/gen/proto/go/frostdb/schema/v1alpha1/schema_vtproto.pb.go b/gen/proto/go/frostdb/schema/v1alpha1/schema_vtproto.pb.go
index c001418..5b16213 100644
--- a/gen/proto/go/frostdb/schema/v1alpha1/schema_vtproto.pb.go
+++ b/gen/proto/go/frostdb/schema/v1alpha1/schema_vtproto.pb.go
@@ -112,6 +112,16 @@ func (m *Column) MarshalToSizedBufferVT(dAtA []byte) (int, error) {
 		i -= len(m.unknownFields)
 		copy(dAtA[i:], m.unknownFields)
 	}
+	if m.Prehash {
+		i--
+		if m.Prehash {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i--
+		dAtA[i] = 0x20
+	}
 	if m.Dynamic {
 		i--
 		if m.Dynamic {
@@ -319,6 +329,9 @@ func (m *Column) SizeVT() (n int) {
 	if m.Dynamic {
 		n += 2
 	}
+	if m.Prehash {
+		n += 2
+	}
 	n += len(m.unknownFields)
 	return n
 }
@@ -642,6 +655,26 @@ func (m *Column) UnmarshalVT(dAtA []byte) error {
 				}
 			}
 			m.Dynamic = bool(v != 0)
+		case 4:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Prehash", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflow
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.Prehash = bool(v != 0)
 		default:
 			iNdEx = preIndex
 			skippy, err := skip(dAtA[iNdEx:])
diff --git a/proto/frostdb/schema/v1alpha1/schema.proto b/proto/frostdb/schema/v1alpha1/schema.proto
index 66d6791..11ed6c1 100644
--- a/proto/frostdb/schema/v1alpha1/schema.proto
+++ b/proto/frostdb/schema/v1alpha1/schema.proto
@@ -20,6 +20,9 @@ message Column {
     StorageLayout storage_layout = 2;
     // Whether the column can dynamically expand.
     bool dynamic = 3;
+    // Prehash the column before storing it. This is an optimization to speed up aggregation queries when this column is often aggregated. 
+    // This will create a separate non-dynamic column with the same name and the prefix "hashed." that contains the prehashed values.
+    bool prehash = 4;
 }
 
 // Storage layout describes the physical storage properties of a column.
diff --git a/query/engine.go b/query/engine.go
index a972555..749b1fe 100644
--- a/query/engine.go
+++ b/query/engine.go
@@ -155,7 +155,7 @@ func (b LocalQueryBuilder) buildPhysical(ctx context.Context) (*physicalplan.Out
 		return nil, err
 	}
 
-	for _, optimizer := range logicalplan.DefaultOptimizers {
+	for _, optimizer := range logicalplan.DefaultOptimizers() {
 		logicalPlan = optimizer.Optimize(logicalPlan)
 	}
 
diff --git a/query/logicalplan/expr.go b/query/logicalplan/expr.go
index 6035e2e..2af76d8 100644
--- a/query/logicalplan/expr.go
+++ b/query/logicalplan/expr.go
@@ -2,6 +2,7 @@ package logicalplan
 
 import (
 	"errors"
+	"regexp"
 	"strings"
 	"time"
 
@@ -691,6 +692,65 @@ func (a *AverageExpr) Accept(visitor Visitor) bool {
 	return visitor.PostVisit(a)
 }
 
+func RegExpColumnMatch(regexp string) *RegexpColumnMatch {
+	return &RegexpColumnMatch{
+		regexp: regexp,
+	}
+}
+
+func RegExpNotColumnMatch(regexp string) *RegexpColumnMatch {
+	return &RegexpColumnMatch{
+		inverse: true,
+		regexp:  regexp,
+	}
+}
+
+type RegexpColumnMatch struct {
+	inverse bool
+	regexp  string
+}
+
+func (a *RegexpColumnMatch) Clone() Expr {
+	return &RegexpColumnMatch{
+		regexp: a.regexp,
+	}
+}
+
+func (a *RegexpColumnMatch) DataType(s *parquet.Schema) (arrow.DataType, error) {
+	return nil, nil
+}
+
+func (a *RegexpColumnMatch) Name() string {
+	return a.regexp
+}
+
+func (a *RegexpColumnMatch) ColumnsUsedExprs() []Expr {
+	return []Expr{a}
+}
+
+func (a *RegexpColumnMatch) MatchPath(path string) bool {
+	match, err := regexp.MatchString(a.regexp, path)
+	return err == nil && match != a.inverse
+}
+
+func (a *RegexpColumnMatch) MatchColumn(name string) bool {
+	match, err := regexp.MatchString(a.regexp, name)
+	return err == nil && match != a.inverse
+}
+
+func (a *RegexpColumnMatch) Computed() bool {
+	return true
+}
+
+func (a *RegexpColumnMatch) Accept(visitor Visitor) bool {
+	continu := visitor.PreVisit(a)
+	if !continu {
+		return false
+	}
+
+	return visitor.PostVisit(a)
+}
+
 type AllExpr struct{}
 
 func All() *AllExpr {
diff --git a/query/logicalplan/optimize.go b/query/logicalplan/optimize.go
index 9c1599d..494ec5d 100644
--- a/query/logicalplan/optimize.go
+++ b/query/logicalplan/optimize.go
@@ -8,12 +8,18 @@ type Optimizer interface {
 	Optimize(plan *LogicalPlan) *LogicalPlan
 }
 
-var DefaultOptimizers = []Optimizer{
-	&AverageAggregationPushDown{},
-	&PhysicalProjectionPushDown{},
-	&FilterPushDown{},
-	&DistinctPushDown{},
-	&ProjectionPushDown{},
+func DefaultOptimizers() []Optimizer {
+	return []Optimizer{
+		&AverageAggregationPushDown{},
+		&PhysicalProjectionPushDown{
+			defaultProjections: []Expr{
+				RegExpNotColumnMatch("^hashed."),
+			},
+		},
+		&FilterPushDown{},
+		&DistinctPushDown{},
+		&ProjectionPushDown{},
+	}
 }
 
 type AverageAggregationPushDown struct{}
@@ -80,7 +86,9 @@ func (p *AverageAggregationPushDown) Optimize(plan *LogicalPlan) *LogicalPlan {
 // physical columns used by the query to the table scan, so the table provider
 // can decide to only read the columns that are actually going to be used by
 // the query.
-type PhysicalProjectionPushDown struct{}
+type PhysicalProjectionPushDown struct {
+	defaultProjections []Expr
+}
 
 func (p *PhysicalProjectionPushDown) Optimize(plan *LogicalPlan) *LogicalPlan {
 	p.optimize(plan, nil)
@@ -90,16 +98,19 @@ func (p *PhysicalProjectionPushDown) Optimize(plan *LogicalPlan) *LogicalPlan {
 func (p *PhysicalProjectionPushDown) optimize(plan *LogicalPlan, columnsUsedExprs []Expr) {
 	switch {
 	case plan.SchemaScan != nil:
-		plan.SchemaScan.PhysicalProjection = columnsUsedExprs
+		plan.SchemaScan.PhysicalProjection = append(p.defaultProjections, columnsUsedExprs...)
 	case plan.TableScan != nil:
-		plan.TableScan.PhysicalProjection = columnsUsedExprs
+		plan.TableScan.PhysicalProjection = append(p.defaultProjections, columnsUsedExprs...)
 	case plan.Filter != nil:
+		p.defaultProjections = []Expr{}
 		columnsUsedExprs = append(columnsUsedExprs, plan.Filter.Expr.ColumnsUsedExprs()...)
 	case plan.Distinct != nil:
+		p.defaultProjections = []Expr{}
 		for _, expr := range plan.Distinct.Exprs {
 			columnsUsedExprs = append(columnsUsedExprs, expr.ColumnsUsedExprs()...)
 		}
 	case plan.Projection != nil:
+		p.defaultProjections = []Expr{}
 		for _, expr := range plan.Projection.Exprs {
 			columnsUsedExprs = append(columnsUsedExprs, expr.ColumnsUsedExprs()...)
 		}
@@ -110,6 +121,8 @@ func (p *PhysicalProjectionPushDown) optimize(plan *LogicalPlan, columnsUsedExpr
 		for _, expr := range plan.Aggregation.AggExprs {
 			columnsUsedExprs = append(columnsUsedExprs, expr.ColumnsUsedExprs()...)
 		}
+		p.defaultProjections = []Expr{}
+		columnsUsedExprs = append(columnsUsedExprs, RegExpColumnMatch("^hashed."))
 	}
 
 	if plan.Input != nil {
@@ -129,6 +142,7 @@ type ProjectionPushDown struct{}
 func (p *ProjectionPushDown) Optimize(plan *LogicalPlan) *LogicalPlan {
 	// Don't perform the optimization if filters or aggregations contain a column that projections do not.
 	// Otherwise we'll removed the columns we're filtering/aggregating.
+	// Also never remove prehashed columns if there is an aggregation being performed.
 	projectColumns := projectionColumns(plan)
 	projectMap := map[string]bool{}
 	filterColumns := filterColumns(plan)
diff --git a/query/physicalplan/aggregate.go b/query/physicalplan/aggregate.go
index 89849c3..e4f6060 100644
--- a/query/physicalplan/aggregate.go
+++ b/query/physicalplan/aggregate.go
@@ -12,12 +12,10 @@ import (
 	"github.com/apache/arrow/go/v14/arrow/math"
 	"github.com/apache/arrow/go/v14/arrow/memory"
 	"github.com/apache/arrow/go/v14/arrow/scalar"
-	"github.com/cespare/xxhash/v2"
-	"github.com/dgryski/go-metro"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
 
-	"github.com/polarsignals/frostdb/pqarrow/arrowutils"
+	"github.com/polarsignals/frostdb/dynparquet"
 	"github.com/polarsignals/frostdb/pqarrow/builder"
 	"github.com/polarsignals/frostdb/query/logicalplan"
 )
@@ -28,6 +26,7 @@ func Aggregate(
 	agg *logicalplan.Aggregation,
 	final bool,
 	ordered bool,
+	seed maphash.Seed,
 ) (PhysicalPlan, error) {
 	aggregations := make([]Aggregation, 0, len(agg.AggExprs))
 
@@ -91,6 +90,7 @@ func Aggregate(
 		tracer,
 		aggregations,
 		agg.GroupExprs,
+		seed,
 		final,
 	), nil
 }
@@ -182,6 +182,7 @@ func NewHashAggregate(
 	tracer trace.Tracer,
 	aggregations []Aggregation,
 	groupByColumnMatchers []logicalplan.Expr,
+	seed maphash.Seed,
 	finalStage bool,
 ) *HashAggregate {
 	dynamic := []Aggregation{}
@@ -199,7 +200,7 @@ func NewHashAggregate(
 		tracer: tracer,
 		// TODO: Matchers can be optimized to be something like a radix tree or just a fast-lookup datastructure for exact matches or prefix matches.
 		groupByColumnMatchers: groupByColumnMatchers,
-		hashSeed:              maphash.MakeSeed(),
+		hashSeed:              seed,
 		finalStage:            finalStage,
 
 		groupByFields:      make([]arrow.Field, 0, 10),
@@ -295,108 +296,6 @@ func (d *durationHashCombine) hashCombine(rhs uint64) uint64 {
 	return rhs / d.milliseconds // floors by default
 }
 
-func hashArray(arr arrow.Array) []uint64 {
-	switch ar := arr.(type) {
-	case *array.String:
-		return hashStringArray(ar)
-	case *array.Binary:
-		return hashBinaryArray(ar)
-	case *array.Int64:
-		return hashInt64Array(ar)
-	case *array.Boolean:
-		return hashBooleanArray(ar)
-	case *array.Dictionary:
-		return hashDictionaryArray(ar)
-	case *array.List:
-		return hashListArray(ar)
-	default:
-		panic("unsupported array type " + fmt.Sprintf("%T", arr))
-	}
-}
-
-func hashListArray(arr *array.List) []uint64 {
-	res := make([]uint64, arr.Len())
-	digest := xxhash.New()
-	for i := 0; i < arr.Len(); i++ {
-		if err := arrowutils.ForEachValueInList(i, arr, func(_ int, v any) {
-			switch val := v.(type) {
-			case []byte:
-				// NOTE: We can ignore the return here because from the xxhash Write function is says: It always returns len(b), nil.
-				_, _ = digest.Write(val)
-			}
-		}); err != nil {
-			panic(err)
-		}
-
-		res[i] = digest.Sum64()
-		digest.Reset()
-	}
-	return res
-}
-
-func hashDictionaryArray(arr *array.Dictionary) []uint64 {
-	res := make([]uint64, arr.Len())
-	for i := 0; i < arr.Len(); i++ {
-		if !arr.IsNull(i) {
-			switch dict := arr.Dictionary().(type) {
-			case *array.Binary:
-				res[i] = metro.Hash64(dict.Value(arr.GetValueIndex(i)), 0)
-			case *array.String:
-				res[i] = metro.Hash64([]byte(dict.Value(arr.GetValueIndex(i))), 0)
-			default:
-				panic("unsupported dictionary type " + fmt.Sprintf("%T", dict))
-			}
-		}
-	}
-	return res
-}
-
-func hashBinaryArray(arr *array.Binary) []uint64 {
-	res := make([]uint64, arr.Len())
-	for i := 0; i < arr.Len(); i++ {
-		if !arr.IsNull(i) {
-			res[i] = metro.Hash64(arr.Value(i), 0)
-		}
-	}
-	return res
-}
-
-func hashBooleanArray(arr *array.Boolean) []uint64 {
-	res := make([]uint64, arr.Len())
-	for i := 0; i < arr.Len(); i++ {
-		if arr.IsNull(i) {
-			res[i] = 0
-			continue
-		}
-		if arr.Value(i) {
-			res[i] = 2
-		} else {
-			res[i] = 1
-		}
-	}
-	return res
-}
-
-func hashStringArray(arr *array.String) []uint64 {
-	res := make([]uint64, arr.Len())
-	for i := 0; i < arr.Len(); i++ {
-		if !arr.IsNull(i) {
-			res[i] = metro.Hash64([]byte(arr.Value(i)), 0)
-		}
-	}
-	return res
-}
-
-func hashInt64Array(arr *array.Int64) []uint64 {
-	res := make([]uint64, arr.Len())
-	for i := 0; i < arr.Len(); i++ {
-		if !arr.IsNull(i) {
-			res[i] = uint64(arr.Value(i))
-		}
-	}
-	return res
-}
-
 func (a *HashAggregate) Callback(ctx context.Context, r arrow.Record) error {
 	// Generates high volume of spans. Comment out if needed during development.
 	// ctx, span := a.tracer.Start(ctx, "HashAggregate/Callback")
@@ -405,6 +304,7 @@ func (a *HashAggregate) Callback(ctx context.Context, r arrow.Record) error {
 	// aggregate is the current aggregation
 	aggregate := a.aggregates[len(a.aggregates)-1]
 
+	fields := r.Schema().Fields() // NOTE: call Fields() once to avoid creating a copy each time
 	groupByFields := a.groupByFields
 	groupByFieldHashes := a.groupByFieldHashes
 	groupByArrays := a.groupByArrays
@@ -424,6 +324,13 @@ func (a *HashAggregate) Callback(ctx context.Context, r arrow.Record) error {
 				groupByFields = append(groupByFields, field)
 				groupByArrays = append(groupByArrays, r.Column(i))
 
+				if a.finalStage { // in the final stage expect the hashes to already exist, so only need to combine them as normal hashes
+					groupByFieldHashes = append(groupByFieldHashes,
+						&uint64HashCombine{value: scalar.Hash(a.hashSeed, scalar.NewStringScalar(field.Name))},
+					)
+					continue
+				}
+
 				switch v := matcher.(type) {
 				case *logicalplan.DurationExpr:
 					duration := v.Value()
@@ -491,7 +398,16 @@ func (a *HashAggregate) Callback(ctx context.Context, r arrow.Record) error {
 
 	colHashes := make([][]uint64, len(groupByArrays))
 	for i, arr := range groupByArrays {
-		colHashes[i] = hashArray(arr)
+		col := dynparquet.FindHashedColumn(groupByFields[i].Name, fields)
+		if col != -1 {
+			vals := make([]uint64, 0, numRows)
+			for _, v := range r.Column(col).(*array.Int64).Int64Values() {
+				vals = append(vals, uint64(v))
+			}
+			colHashes[i] = vals
+		} else {
+			colHashes[i] = dynparquet.HashArray(arr)
+		}
 	}
 
 	for i := 0; i < numRows; i++ {
@@ -618,8 +534,8 @@ func (a *HashAggregate) Finish(ctx context.Context) error {
 	defer span.End()
 
 	totalRows := 0
-	for _, aggregate := range a.aggregates {
-		if err := a.finishAggregate(ctx, aggregate); err != nil {
+	for i, aggregate := range a.aggregates {
+		if err := a.finishAggregate(ctx, i, aggregate); err != nil {
 			return err
 		}
 		totalRows += aggregate.rowCount
@@ -628,7 +544,7 @@ func (a *HashAggregate) Finish(ctx context.Context) error {
 	return a.next.Finish(ctx)
 }
 
-func (a *HashAggregate) finishAggregate(ctx context.Context, aggregate *hashAggregate) error {
+func (a *HashAggregate) finishAggregate(ctx context.Context, aggIdx int, aggregate *hashAggregate) error {
 	numCols := len(aggregate.groupByCols) + len(aggregate.aggregations)
 	numRows := aggregate.rowCount
 
@@ -639,6 +555,9 @@ func (a *HashAggregate) finishAggregate(ctx context.Context, aggregate *hashAggr
 	groupByFields := make([]arrow.Field, 0, numCols)
 	groupByArrays := make([]arrow.Array, 0, numCols)
 	for _, fieldName := range aggregate.colOrdering {
+		if a.finalStage && dynparquet.IsHashedColumn(fieldName) {
+			continue
+		}
 		groupByCol, ok := aggregate.groupByCols[fieldName]
 		if !ok {
 			return fmt.Errorf("unknown field name: %s", fieldName)
@@ -654,6 +573,22 @@ func (a *HashAggregate) finishAggregate(ctx context.Context, aggregate *hashAggr
 		arr := groupByCol.NewArray()
 		groupByFields = append(groupByFields, arrow.Field{Name: fieldName, Type: arr.DataType()})
 		groupByArrays = append(groupByArrays, arr)
+		// Pass forward the hashings of the group-by columns
+		if !a.finalStage {
+			groupByFields = append(groupByFields, arrow.Field{Name: dynparquet.HashedColumnName(fieldName), Type: arrow.PrimitiveTypes.Int64})
+			func() {
+				bldr := array.NewInt64Builder(a.pool)
+				defer bldr.Release()
+				sortedHashes := make([]int64, arr.Len())
+				for hash, tuple := range a.hashToAggregate {
+					if tuple.aggregate == aggIdx { // only append the hash for the current aggregate
+						sortedHashes[tuple.array] = int64(hash)
+					}
+				}
+				bldr.AppendValues(sortedHashes, nil)
+				groupByArrays = append(groupByArrays, bldr.NewArray())
+			}()
+		}
 	}
 
 	// Rename to clarity upon appending aggregations later
@@ -679,6 +614,7 @@ func (a *HashAggregate) finishAggregate(ctx context.Context, aggregate *hashAggr
 			Name: aggregation.resultName, Type: aggregateArray.DataType(),
 		})
 	}
+
 	r := array.NewRecord(
 		arrow.NewSchema(aggregateFields, nil),
 		aggregateColumns,
diff --git a/query/physicalplan/distinct.go b/query/physicalplan/distinct.go
index 33f08cb..cd49942 100644
--- a/query/physicalplan/distinct.go
+++ b/query/physicalplan/distinct.go
@@ -13,6 +13,7 @@ import (
 	"github.com/apache/arrow/go/v14/arrow/scalar"
 	"go.opentelemetry.io/otel/trace"
 
+	"github.com/polarsignals/frostdb/dynparquet"
 	"github.com/polarsignals/frostdb/pqarrow/builder"
 	"github.com/polarsignals/frostdb/query/logicalplan"
 )
@@ -100,7 +101,7 @@ func (d *Distinction) Callback(ctx context.Context, r arrow.Record) error {
 
 	colHashes := make([][]uint64, len(distinctFields))
 	for i, arr := range distinctArrays {
-		colHashes[i] = hashArray(arr)
+		colHashes[i] = dynparquet.HashArray(arr)
 	}
 
 	for i := 0; i < numRows; i++ {
diff --git a/query/physicalplan/physicalplan.go b/query/physicalplan/physicalplan.go
index 5e2da04..aecff2a 100644
--- a/query/physicalplan/physicalplan.go
+++ b/query/physicalplan/physicalplan.go
@@ -3,6 +3,7 @@ package physicalplan
 import (
 	"context"
 	"fmt"
+	"hash/maphash"
 	"runtime"
 
 	"github.com/apache/arrow/go/v14/arrow"
@@ -418,8 +419,9 @@ func Build(
 					sync = Synchronize(len(prev))
 				}
 			}
+			seed := maphash.MakeSeed()
 			for i := 0; i < len(prev); i++ {
-				a, err := Aggregate(pool, tracer, plan.Aggregation, sync == nil, ordered)
+				a, err := Aggregate(pool, tracer, plan.Aggregation, sync == nil, ordered, seed)
 				if err != nil {
 					visitErr = err
 					return false
@@ -433,7 +435,7 @@ func Build(
 			if sync != nil {
 				// Plan an aggregate operator to run an aggregation on all the
 				// aggregations.
-				a, err := Aggregate(pool, tracer, plan.Aggregation, true, ordered)
+				a, err := Aggregate(pool, tracer, plan.Aggregation, true, ordered, seed)
 				if err != nil {
 					visitErr = err
 					return false
diff --git a/table.go b/table.go
index 184f8bd..4e33be3 100644
--- a/table.go
+++ b/table.go
@@ -1124,6 +1124,9 @@ func (t *TableBlock) InsertRecord(ctx context.Context, tx uint64, record arrow.R
 		return nil
 	}
 
+	record = dynparquet.PrehashColumns(t.table.schema, record)
+	defer record.Release()
+
 	if err := t.insertRecordToGranules(tx, record); err != nil {
 		return fmt.Errorf("failed to insert record into granules: %w", err)
 	}


Test patch:
diff --git a/aggregate_test.go b/aggregate_test.go
index 5c2b712..9fa3e61 100644
--- a/aggregate_test.go
+++ b/aggregate_test.go
@@ -280,9 +280,9 @@ func BenchmarkAggregation(b *testing.B) {
 			})
 		}
 
-		buf, err := samples.ToBuffer(table.Schema())
+		r, err := samples.ToRecord()
 		require.NoError(b, err)
-		_, err = table.InsertBuffer(ctx, buf)
+		_, err = table.InsertRecord(ctx, r)
 		require.NoError(b, err)
 	}
 
diff --git a/db_test.go b/db_test.go
index b01c63c..dec1568 100644
--- a/db_test.go
+++ b/db_test.go
@@ -20,6 +20,7 @@ import (
 	"github.com/parquet-go/parquet-go"
 	"github.com/stretchr/testify/require"
 	"github.com/thanos-io/objstore"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/polarsignals/frostdb/dynparquet"
 	schemapb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/schema/v1alpha1"
@@ -1259,113 +1260,118 @@ func Test_DB_TableNotExist(t *testing.T) {
 }
 
 func Test_DB_TableWrite_ArrowRecord(t *testing.T) {
-	ctx := context.Background()
-	config := NewTableConfig(
+	for _, schema := range []proto.Message{
 		dynparquet.SampleDefinition(),
-	)
+		dynparquet.PrehashedSampleDefinition(),
+	} {
+		ctx := context.Background()
+		config := NewTableConfig(
+			schema,
+		)
 
-	c, err := New(WithLogger(newTestLogger(t)))
-	require.NoError(t, err)
-	defer c.Close()
+		c, err := New(WithLogger(newTestLogger(t)))
+		require.NoError(t, err)
+		defer c.Close()
 
-	db, err := c.DB(ctx, "sampleschema")
-	require.NoError(t, err)
+		db, err := c.DB(ctx, "sampleschema")
+		require.NoError(t, err)
 
-	table, err := db.Table("test", config)
-	require.NoError(t, err)
+		table, err := db.Table("test", config)
+		require.NoError(t, err)
 
-	samples := dynparquet.Samples{
-		{
-			ExampleType: "test",
-			Labels: []dynparquet.Label{
-				{Name: "label1", Value: "value1"},
-				{Name: "label2", Value: "value2"},
-			},
-			Stacktrace: []uuid.UUID{
-				{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
-				{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
-			},
-			Timestamp: 10,
-			Value:     1,
-		},
-		{
-			ExampleType: "test",
-			Labels: []dynparquet.Label{
-				{Name: "label1", Value: "value1"},
-				{Name: "label2", Value: "value2"},
-				{Name: "label3", Value: "value3"},
-			},
-			Stacktrace: []uuid.UUID{
-				{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
-				{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
+		samples := dynparquet.Samples{
+			{
+				ExampleType: "test",
+				Labels: []dynparquet.Label{
+					{Name: "label1", Value: "value1"},
+					{Name: "label2", Value: "value2"},
+				},
+				Stacktrace: []uuid.UUID{
+					{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
+					{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
+				},
+				Timestamp: 10,
+				Value:     1,
 			},
-			Timestamp: 11,
-			Value:     2,
-		},
-		{
-			ExampleType: "test",
-			Labels: []dynparquet.Label{
-				{Name: "label1", Value: "value1"},
-				{Name: "label2", Value: "value2"},
+			{
+				ExampleType: "test",
+				Labels: []dynparquet.Label{
+					{Name: "label1", Value: "value1"},
+					{Name: "label2", Value: "value2"},
+					{Name: "label3", Value: "value3"},
+				},
+				Stacktrace: []uuid.UUID{
+					{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
+					{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
+				},
+				Timestamp: 11,
+				Value:     2,
 			},
-			Stacktrace: []uuid.UUID{
-				{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
-				{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
+			{
+				ExampleType: "test",
+				Labels: []dynparquet.Label{
+					{Name: "label1", Value: "value1"},
+					{Name: "label2", Value: "value2"},
+				},
+				Stacktrace: []uuid.UUID{
+					{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
+					{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
+				},
+				Timestamp: 12,
+				Value:     3,
 			},
-			Timestamp: 12,
-			Value:     3,
-		},
-	}
+		}
 
-	r, err := samples.ToRecord()
-	require.NoError(t, err)
+		r, err := samples.ToRecord()
+		require.NoError(t, err)
 
-	_, err = table.InsertRecord(ctx, r)
-	require.NoError(t, err)
+		_, err = table.InsertRecord(ctx, r)
+		require.NoError(t, err)
 
-	engine := query.NewEngine(
-		memory.NewGoAllocator(),
-		db.TableProvider(),
-	)
+		engine := query.NewEngine(
+			memory.NewGoAllocator(),
+			db.TableProvider(),
+		)
 
-	tests := map[string]struct {
-		filter   logicalplan.Expr
-		distinct logicalplan.Expr
-		rows     int64
-		cols     int64
-	}{
-		"none": {
-			rows: 3,
-			cols: 7,
-		},
-		"timestamp filter": {
-			filter: logicalplan.Col("timestamp").GtEq(logicalplan.Literal(12)),
-			rows:   1,
-			cols:   1,
-		},
-		"distinct": {
-			distinct: logicalplan.DynCol("labels"),
-			rows:     2,
-			cols:     3,
-		},
-	}
+		tests := map[string]struct {
+			filter   logicalplan.Expr
+			distinct logicalplan.Expr
+			rows     int64
+			cols     int64
+		}{
+			"none": {
+				rows: 3,
+				cols: 7,
+			},
+			"timestamp filter": {
+				filter: logicalplan.Col("timestamp").GtEq(logicalplan.Literal(12)),
+				rows:   1,
+				cols:   1,
+			},
+			"distinct": {
+				distinct: logicalplan.DynCol("labels"),
+				rows:     2,
+				cols:     3,
+			},
+		}
 
-	for name, test := range tests {
-		t.Run(name, func(t *testing.T) {
-			bldr := engine.ScanTable("test")
-			if test.filter != nil {
-				bldr = bldr.Filter(test.filter)
-			}
-			if test.distinct != nil {
-				bldr = bldr.Distinct(test.distinct)
-			}
-			err = bldr.Execute(ctx, func(ctx context.Context, ar arrow.Record) error {
-				require.Equal(t, test.rows, ar.NumRows())
-				require.Equal(t, test.cols, ar.NumCols())
-				return nil
+		for name, test := range tests {
+			t.Run(name, func(t *testing.T) {
+				bldr := engine.ScanTable("test")
+				if test.filter != nil {
+					bldr = bldr.Filter(test.filter)
+				}
+				if test.distinct != nil {
+					bldr = bldr.Distinct(test.distinct)
+				}
+				err = bldr.Execute(ctx, func(ctx context.Context, ar arrow.Record) error {
+					require.Equal(t, test.rows, ar.NumRows())
+					require.Equal(t, test.cols, ar.NumCols())
+					return nil
+				})
+				require.NoError(t, err)
 			})
-			require.NoError(t, err)
-		})
+		}
 	}
 }
 
@@ -2065,3 +2071,104 @@ func Test_DB_All(t *testing.T) {
 		})
 	require.NoError(t, err)
 }
+
+func Test_DB_PrehashedStorage(t *testing.T) {
+	config := NewTableConfig(
+		dynparquet.PrehashedSampleDefinition(),
+	)
+
+	bucket := objstore.NewInMemBucket()
+	sinksource := NewDefaultObjstoreBucket(bucket)
+	logger := newTestLogger(t)
+
+	c, err := New(
+		WithLogger(logger),
+		WithReadWriteStorage(sinksource),
+	)
+	require.NoError(t, err)
+
+	db, err := c.DB(context.Background(), t.Name())
+	require.NoError(t, err)
+	defer os.RemoveAll(t.Name())
+	table, err := db.Table(t.Name(), config)
+	require.NoError(t, err)
+
+	samples := dynparquet.Samples{{
+		ExampleType: "test",
+		Labels: []dynparquet.Label{
+			{Name: "label1", Value: "value1"},
+			{Name: "label2", Value: "value2"},
+		},
+		Stacktrace: []uuid.UUID{
+			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
+			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
+		},
+		Timestamp: 1,
+		Value:     1,
+	}, {
+		ExampleType: "test",
+		Labels: []dynparquet.Label{
+			{Name: "label1", Value: "value2"},
+			{Name: "label2", Value: "value2"},
+			{Name: "label3", Value: "value3"},
+		},
+		Stacktrace: []uuid.UUID{
+			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
+			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
+		},
+		Timestamp: 2,
+		Value:     2,
+	}, {
+		ExampleType: "test",
+		Labels: []dynparquet.Label{
+			{Name: "label1", Value: "value3"},
+			{Name: "label2", Value: "value2"},
+			{Name: "label4", Value: "value4"},
+		},
+		Stacktrace: []uuid.UUID{
+			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
+			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
+		},
+		Timestamp: 3,
+		Value:     3,
+	}}
+
+	r, err := samples.ToRecord()
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	_, err = table.InsertRecord(ctx, r)
+	require.NoError(t, err)
+
+	// Gracefully close the db to persist blocks
+	c.Close()
+
+	c, err = New(
+		WithLogger(logger),
+		WithReadWriteStorage(sinksource),
+	)
+	require.NoError(t, err)
+	defer c.Close()
+
+	db, err = c.DB(context.Background(), t.Name())
+	require.NoError(t, err)
+	table, err = db.Table(t.Name(), config)
+	require.NoError(t, err)
+
+	// Read the raw data back and expect prehashed columns to be returned
+	allocator := memory.NewCheckedAllocator(memory.NewGoAllocator())
+	defer allocator.AssertSize(t, 0)
+	err = table.View(ctx, func(ctx context.Context, tx uint64) error {
+		return table.Iterator(
+			ctx,
+			tx,
+			allocator,
+			[]logicalplan.Callback{func(ctx context.Context, ar arrow.Record) error {
+				require.Equal(t, int64(3), ar.NumRows())
+				require.Equal(t, int64(13), ar.NumCols())
+				return nil
+			}},
+		)
+	})
+	require.NoError(t, err)
+}
diff --git a/logictest/logic_test.go b/logictest/logic_test.go
index e0235bb..3f5caf7 100644
--- a/logictest/logic_test.go
+++ b/logictest/logic_test.go
@@ -59,6 +59,54 @@ var schemas = map[string]*schemapb.Schema{
 			Direction: schemapb.SortingColumn_DIRECTION_ASCENDING,
 		}},
 	},
+	"prehashed": {
+		Name: "test",
+		Columns: []*schemapb.Column{{
+			Name: "labels",
+			StorageLayout: &schemapb.StorageLayout{
+				Type:     schemapb.StorageLayout_TYPE_STRING,
+				Nullable: true,
+				Encoding: schemapb.StorageLayout_ENCODING_RLE_DICTIONARY,
+			},
+			Dynamic: true,
+			Prehash: true,
+		}, {
+			Name: "stacktrace",
+			StorageLayout: &schemapb.StorageLayout{
+				Type:     schemapb.StorageLayout_TYPE_STRING,
+				Encoding: schemapb.StorageLayout_ENCODING_RLE_DICTIONARY,
+			},
+			Dynamic: false,
+			Prehash: true,
+		}, {
+			Name: "timestamp",
+			StorageLayout: &schemapb.StorageLayout{
+				Type: schemapb.StorageLayout_TYPE_INT64,
+			},
+			Dynamic: false,
+		}, {
+			Name: "value",
+			StorageLayout: &schemapb.StorageLayout{
+				Type: schemapb.StorageLayout_TYPE_INT64,
+			},
+			Dynamic: false,
+		}},
+		SortingColumns: []*schemapb.SortingColumn{{
+			Name:      "example_type",
+			Direction: schemapb.SortingColumn_DIRECTION_ASCENDING,
+		}, {
+			Name:       "labels",
+			Direction:  schemapb.SortingColumn_DIRECTION_ASCENDING,
+			NullsFirst: true,
+		}, {
+			Name:      "timestamp",
+			Direction: schemapb.SortingColumn_DIRECTION_ASCENDING,
+		}, {
+			Name:       "stacktrace",
+			Direction:  schemapb.SortingColumn_DIRECTION_ASCENDING,
+			NullsFirst: true,
+		}},
+	},
 }
 
 // TestLogic runs all the datadriven tests in the testdata directory. Refer to
diff --git a/query/logicalplan/optimize_test.go b/query/logicalplan/optimize_test.go
index fd9c3bb..4cedd1f 100644
--- a/query/logicalplan/optimize_test.go
+++ b/query/logicalplan/optimize_test.go
@@ -35,6 +35,7 @@ func TestOptimizePhysicalProjectionPushDown(t *testing.T) {
 			&Column{ColumnName: "stacktrace"},
 			&Column{ColumnName: "stacktrace"},
 			&Column{ColumnName: "value"},
+			RegExpColumnMatch("^hashed."),
 			&Column{ColumnName: "labels.test"},
 		},
 	},
@@ -204,6 +205,7 @@ func TestAllOptimizers(t *testing.T) {
 			&Column{ColumnName: "stacktrace"},
 			&Column{ColumnName: "stacktrace"},
 			&Column{ColumnName: "value"},
+			RegExpColumnMatch("^hashed."),
 			&Column{ColumnName: "labels.test"},
 		},
 		Filter: &BinaryExpr{
diff --git a/query/physicalplan/aggregate_test.go b/query/physicalplan/aggregate_test.go
index 2e18fb8..851d930 100644
--- a/query/physicalplan/aggregate_test.go
+++ b/query/physicalplan/aggregate_test.go
@@ -3,6 +3,7 @@ package physicalplan
 import (
 	"context"
 	"encoding/binary"
+	"hash/maphash"
 	"math/rand"
 	"testing"
 
@@ -44,13 +45,14 @@ func Test_Aggregate_ArrayOverflow(t *testing.T) {
 			logicalplan.Col("stacktrace"),
 			logicalplan.Col("id"),
 		},
+		maphash.MakeSeed(),
 		false,
 	)
 
 	totalRows := int64(0)
 	agg.SetNext(&OutputPlan{
 		callback: func(ctx context.Context, r arrow.Record) error {
-			require.Equal(t, 3, len(r.Schema().Fields()))
+			require.Equal(t, 5, len(r.Schema().Fields()))
 			for i := 0; i < int(r.NumCols()); i++ {
 				require.Equal(t, r.NumRows(), int64(r.Column(i).Len()))
 			}