Repository: polarsignals/frostdb
Pull request #702 (2024-01-23 18:40:26)

Nested arithmetic projections
This allows us to perform things like sum(value)/count(value) or any arbitrary combination of aggregations and arithmetic expressions.

To get this to work there were a couple of changes necessary:

1) Implement #208 
2) Make query plans dumber, and query optimizations only optimizations and not required for correctness.
3) Make query builders (the dataframe and SQL APIs) smarter to insert logical query plans correctly for the desired behaviors. For example, when there are arithmetic expressions that use aggregations a projection query plan is inserted after the aggregation to perform that calculation. Likewise, if there is a column that is used in a group by or as input to an aggregation, then a projection to perform that calculation needs to be inserted before the aggregation layer. In the SQL layer this was implemented, in the dataframe API it is expected that the user builds the query layers appropriately.

I wanted to open this early so it's not just on my computer and get some early feedback while I finish it up. A couple of things are still missing:

- [x] Not all tests are passing yet and there is some non-correct behavior for expressions like `(timestamp/1000)*1000 as timestamp_bucket`
- [x] Average queries

Side note: What's nice about these changes is that we could now definitively determine the input and output schema of a logical plan query layer which we could use eg. for validations.

jicanghaixb: I'm really looking forward to this feature, sum(value) maybe is  float64, count(value)  is int64, so sum(value)/count(value) need to support mathematical calculations of different types. 

I have also made some attempts, and below are my test cases,the modified code is quite ugly, and last two test cases have leaked memory, can't pass require.Equal(t, 0, fdb.allocator.Allocated()). I haven't found the reason yet:
```
createtable schema=custom
----

insert cols=(name, duration,  labels.code, labels.instance, labels.method, labels.status, labels.upstream_code, timestamp, trace,  value.body_send, value.header_time,message, value.request)
cdn 1 200 127.0.0.1 GET HIT  - 1699871519098 legacyplatforms.info 1664782 0.057256 /Triple-buffered.htm 1
cdn 1 200 127.0.0.1 GET MISS  200  1699871520098 productinnovative.com 1769821  3.836798 /solution-oriented.php 1
cdn 1 200 127.0.0.1 HEAD MISS 200 1699871521098 regionalone-to-one.biz  1581608 1.655158 /intangible-local.jpg 1
cdn 1 200 127.0.0.1 GET MISS 200 1699871522098 legacysticky.info  1786570 4.581297 /monitoring-Virtual.jpg 1
cdn 1 200 127.0.0.1 GET HIT  - 1699871523098 legacyback-end.com 120018 0.156603 /projection-Versatile.js 1
cdn 1 200 127.0.0.1 GET HIT - 1699871524098 dynamicweb-enabled.com  621713 0.154511 /incremental/core/Cross-platform/conglomeration.css 1
cdn 1 200 127.0.0.1 GET HIT  - 1699871525098 futureembrace.org  2130660 0.286799 /full-range-4th%20generation/Customizable_interface%20Business-focused.htm 1
cdn 1 200 127.0.0.1 PUT HIT  - 1699871526098 districtenable.org  195706 0.171824 /Face%20to%20face.gif 1
cdn 1 200 127.0.0.1 GET MISS 200 1699871527098 globalreal-time.biz 1970713 4.200371 /Total_Implemented-budgetary%20management.js 1
cdn 1 200 127.0.0.1 HEAD MISS 200 1699871528098 internationalmission-critical.name  2836182 2.633382 /Re-contextualized.gif 1
cdn 1 200 127.0.0.1 GET MISS 200 1699871529098 futuremindshare.org  1379273 2.028894 /archive/Implemented.htm 1
cdn 1 200 127.0.0.1 GET HIT - 1699871530098 futurerobust.io  825589 0.384408 /methodology.jpg 1
cdn 1 200 127.0.0.1 PUT MISS 200 1699871531098 futuresystems.biz  2320748 4.660421 /initiative.css 1
cdn 1 200 127.0.0.1 GET MISS 200 1699871532098 futuresynergize.name  166355 2.627923 /4th%20generation-national/emulation.js 1
cdn 1 200 127.0.0.1 GET HIT  - 1699871533098 principalsynthesize.name  1775497 0.435607 /Versatile/4th%20generation/Devolved-policy/core.htm 1
cdn 1 200 127.0.0.1 POST HIT - 1699871534098 centralsticky.org  2669659 0.436545 /24%20hour/responsive-Graphical%20User%20Interface.svg 1
cdn 1 200 127.0.0.1 GET MISS 200 1699871535098 principalcommunities.com  2939218 3.630523 /Re-engineered-Multi-tiered-Operative.svg 1
cdn 1 200 127.0.0.1 GET HIT - 1699871536098 districtempower.name  1668660 0.258968 /Profound-bottom-line.png 1
cdn 1 200 127.0.0.1 DELETE MISS 200 1699871537098 humane-business.name  2926877 0.279401 /flexibility/website.php 1
cdn 1 200 127.0.0.1 GET MISS 200 1699871538098 dynamicimplement.com  1240638  1.376430 /context-sensitive-Re-contextualized.htm 1
----

exec
select sum(value), avg(value) group by labels.instance,labels.status
----
labels.instance  labels.status  sum(value.body_send)  sum(value.header_time)  sum(value.request)  avg(value.body_send)  avg(value.header_time)  avg(value.request)
127.0.0.1        MISS           20918003.00           31.51                   11.00               1901636.64            2.86                    1.00
127.0.0.1        HIT            11672284.00           2.34                    9.00                1296920.44            0.26                    1.00

exec
select sum(value.body_send) as body_send, sum(value.request) as request, body_send / request as body_send_rate  group by labels.instance,labels.status
----
labels.instance  labels.status  body_send    request  body_send_rate
127.0.0.1        MISS           20918003.00  11.00    1901636.64
127.0.0.1        HIT            11672284.00  9.00     1296920.44

exec
select sum(value.request), avg(value.body_send),max(value.header_time) group by labels.instance,labels.status
----
labels.instance  labels.status  sum(value.request)  avg(value.body_send)  max(value.header_time)
127.0.0.1        MISS           11.00               1901636.64            4.66
127.0.0.1        HIT            9.00                1296920.44            0.44

exec
select min(value.body_send), max(value.body_send), sum(value.body_send), count(value.body_send), avg(value.body_send) group by labels.instance,labels.status
----
labels.instance  labels.status  min(value.body_send)  max(value.body_send)  sum(value.body_send)  count(value.body_send)  avg(value.body_send)
127.0.0.1        MISS           166355.00             2939218.00            20918003.00           11.00                   1901636.64
127.0.0.1        HIT            120018.00             2669659.00            11672284.00           9.00                    1296920.44

exec
select labels.status, min(value.body_send), max(value.body_send), sum(value.body_send), count(value.body_send), avg(value.body_send) group by labels.instance,labels.status
----
labels.status  min(value.body_send)  max(value.body_send)  sum(value.body_send)  count(value.body_send)  avg(value.body_send)
MISS           166355.00             2939218.00            20918003.00           11.00                   1901636.64
HIT            120018.00             2669659.00            11672284.00           9.00                    1296920.44

exec
select labels.status, min(value.body_send), max(value.body_send), sum(value.body_send)/count(value.body_send) as avg, avg(value.body_send) group by labels.instance,labels.status
----
labels.status  min(value.body_send)  max(value.body_send)  avg         avg(value.body_send)
MISS           166355.00             2939218.00            1901636.64  1901636.64
HIT            120018.00             2669659.00            1296920.44  1296920.44

exec
select labels.status, min(value.body_send), max(value.body_send), sum(value.body_send)/count(value.body_send) as avg, avg(value.body_send)  having avg > 1296921 group by labels.instance,labels.status
----
labels.status  min(value.body_send)  max(value.body_send)  avg         avg(value.body_send)
MISS           166355.00             2939218.00            1901636.64  1901636.64

exec
select labels.status, min(value.body_send), max(value.body_send), sum(value.body_send), avg(value.body_send) * count(value.body_send) as sum group by labels.instance,labels.status
----
labels.status  min(value.body_send)  max(value.body_send)  sum(value.body_send)  sum
MISS           166355.00             2939218.00            20918003.00           20918003.00
HIT            120018.00             2669659.00            11672284.00           11672284.00

exec
select  avg(value.body_send) as avg_1x , avg(value.body_send) * 2 as avg_2x, avg(value.body_send) * 3 as avg_3x  group by labels.instance,labels.status
----
labels.instance  labels.status  avg_1x      avg_2x      avg_3x
127.0.0.1        MISS           1901636.64  3803273.27  5704909.91
127.0.0.1        HIT            1296920.44  2593840.89  3890761.33

exec
select sum(value) group by labels
----
labels.code  labels.instance  labels.method  labels.status  labels.upstream_code  sum(value.body_send)  sum(value.header_time)  sum(value.request)
200          127.0.0.1        DELETE         MISS           200                   2926877.00            0.28                    1.00
200          127.0.0.1        GET            HIT            -                     8806919.00            1.73                    7.00
200          127.0.0.1        GET            MISS           200                   11252588.00           22.28                   7.00
200          127.0.0.1        HEAD           MISS           200                   4417790.00            4.29                    2.00
200          127.0.0.1        POST           HIT            -                     2669659.00            0.44                    1.00
200          127.0.0.1        PUT            HIT            -                     195706.00             0.17                    1.00
200          127.0.0.1        PUT            MISS           200                   2320748.00            4.66                    1.00

exec
select sum(value) group by labels.method, second(5)
----
labels.method  timestamp      sum(value.body_send)  sum(value.header_time)  sum(value.request)
DELETE         1699871537098  2926877.00            0.28                    1.00
GET            1699871519098  1664782.00            0.06                    1.00
GET            1699871523098  4298122.00            8.73                    4.00
GET            1699871525098  5480646.00            6.52                    3.00
GET            1699871530098  2767441.00            3.45                    3.00
GET            1699871536098  5848516.00            5.27                    3.00
HEAD           1699871521098  1581608.00            1.66                    1.00
HEAD           1699871528098  2836182.00            2.63                    1.00
POST           1699871534098  2669659.00            0.44                    1.00
PUT            1699871526098  195706.00             0.17                    1.00
PUT            1699871531098  2320748.00            4.66                    1.00

exec
select topk(sum(value.body_send),3) group by labels.method, second(300)
----
labels.method  timestamp      topk(sum(value.body_send)_3)
GET            1699871519098  20059507.00
HEAD           1699871521098  4417790.00
DELETE         1699871537098  2926877.00

exec
select name,topk(value.body_send,3) group by labels
----
name    topk(value.body_send_3)
cdn     2939218.00
cdn     2926877.00
cdn     2836182.00

exec
select name,bottomk(value.body_send,3), topk(value.header_time,3) as top group by labels
----
name    bottomk(value.body_send_3)  name    top
cdn     120018.00                   cdn     4.66
cdn     166355.00                   cdn     4.58
cdn     195706.00                   cdn     4.20

exec
select name,labels.method,IF(labels.method = "GET",1,2) as result
----
name    labels.method  result
cdn     DELETE         2.00
cdn     GET            1.00
cdn     GET            1.00
cdn     GET            1.00
cdn     GET            1.00
cdn     GET            1.00
cdn     GET            1.00
cdn     GET            1.00
cdn     GET            1.00
cdn     GET            1.00
cdn     GET            1.00
cdn     GET            1.00
cdn     GET            1.00
cdn     GET            1.00
cdn     GET            1.00
cdn     HEAD           2.00
cdn     HEAD           2.00
cdn     POST           2.00
cdn     PUT            2.00
cdn     PUT            2.00

exec
select sum(value.body_send) > 20000000  as body_send,labels group by labels.instance,labels.status
----
body_send  labels.instance  labels.status
true       127.0.0.1        MISS
false      127.0.0.1        HIT

exec
select IF(sum(value.body_send) > 20000000, "warning", "normal") as body_send,labels group by labels.instance,labels.status
----
labels.instance  labels.status  body_send
127.0.0.1        MISS           warning
127.0.0.1        HIT            normal

``` 
jicanghaixb: The following are the test cases I implemented after modifying the code, hoping that frost db can support them natively:
```
createtable schema=default
----

insert cols=(labels.label1, timestamp, value)
value1 1 2
value1 3 4
value1 5 6
----

exec
select value * timestamp
----
2
12
30

exec
select value * timestamp limit 0,2
----
2
12

exec
select value * timestamp limit 1,2
----
2
12

exec
select value * timestamp limit 3,2
----

exec
select value * timestamp limit 2,2
----
12

exec
select value * 2
----
4
8
12

exec
select 2 * 3
----
6
6
6

exec
select value * (2 * 3)
----
12
24
36

exec
select value * (timestamp * 2)
----
4
24
60

exec
select value * timestamp * 2
----
4
24
60

exec
select value * timestamp + 2 - 1
----
3
13
31

exec
select value * timestamp * (2 - 1)
----
2
12
30

exec
select value, value * timestamp * (2 - 1)
----
2       2
4       12
6       30

exec
select timestamp, value * timestamp * (2 - 1) where value > 2
----
3       12
5       30

exec
select value, timestamp  * value * (2 - 1)  where value > 2
----
4       12
6       30

exec
select value , value  * timestamp * (2 - 1) as result where value > 2
----
4       12
6       30

exec
select value as result, value  * timestamp * (2 - 1)  where value > 2
----
4       12
6       30

exec
select value , value  * timestamp * (2 - 1) as result  having  value > 4
----
6       30

exec
select value as result, value  * timestamp * (2 - 1)  having result > 4
----
6       30

exec
select sum(value), count(value) group by labels.label1
----
value1  12      3

exec
select   max(value) - min(value) , sum(value) / count(value)  group by labels.label1
----
4       4

exec
select  labels.label1, max(value) - min(value) , sum(value) / count(value)  group by labels.label1
----
value1  4       4

exec
select  labels.label1, (max(value) - min(value)) /2  , sum(value) / count(value)  group by labels.label1
----
value1  2       4

exec
select  labels.label1, (max(value) - min(value)) /2  , sum(value) / count(value) , sum(value) / count(value) * 2 group by labels.label1
----
value1  2       4       8

``` 
brancz: This ended up being a way bigger PR than I thought initially, but it's now ready for review!

Patch:
diff --git a/dynparquet/schema.go b/dynparquet/schema.go
index 9d01a1f..f136228 100644
--- a/dynparquet/schema.go
+++ b/dynparquet/schema.go
@@ -124,9 +124,10 @@ type Schema struct {
 	parquetSchemas *sync.Map
 }
 
-// IsDynamicColumn returns true if the passed in column name of the type
-// "labels.label1" is a dynamic column the schema recognizes.
-func (s *Schema) IsDynamicColumn(column string) bool {
+// FindDynamicColumnForConcreteColumn returns a column definition for the
+// column passed. So "labels.label1" would return the column definition for the
+// dynamic column "labels" if it exists.
+func (s *Schema) FindDynamicColumnForConcreteColumn(column string) (ColumnDefinition, bool) {
 	periodPosition := 0
 	foundPeriod := false
 	for i, c := range column {
@@ -135,15 +136,48 @@ func (s *Schema) IsDynamicColumn(column string) bool {
 		}
 		if foundPeriod {
 			// Can't have more than one period.
-			return false
+			return ColumnDefinition{}, false
 		}
 		foundPeriod = true
 		periodPosition = i
 	}
 	if !foundPeriod {
-		return false
+		return ColumnDefinition{}, false
+	}
+
+	return s.FindDynamicColumn(column[:periodPosition])
+}
+
+// FindDynamicColumn returns a dynamic column definition for the column passed.
+func (s *Schema) FindDynamicColumn(dynamicColumnName string) (ColumnDefinition, bool) {
+	idx, ok := s.columnIndexes[dynamicColumnName]
+	if !ok {
+		return ColumnDefinition{}, false
+	}
+
+	colDef := s.columns[idx]
+	// Note: This is different from the FindColumn function.
+	if !colDef.Dynamic {
+		return ColumnDefinition{}, false
 	}
-	return s.columns[s.columnIndexes[column[0:periodPosition]]].Dynamic
+
+	return colDef, true
+}
+
+// FindColumn returns a column definition for the column passed.
+func (s *Schema) FindColumn(column string) (ColumnDefinition, bool) {
+	idx, ok := s.columnIndexes[column]
+	if !ok {
+		return ColumnDefinition{}, false
+	}
+
+	colDef := s.columns[idx]
+	// Note: This is different from the FindDynamicColumn function.
+	if colDef.Dynamic {
+		return ColumnDefinition{}, false
+	}
+
+	return colDef, true
 }
 
 func findLeavesFromNode(node *schemav2pb.Node) []ColumnDefinition {
diff --git a/examples/aggregations/aggregations.go b/examples/aggregations/aggregations.go
index 4f73337..aef3c41 100644
--- a/examples/aggregations/aggregations.go
+++ b/examples/aggregations/aggregations.go
@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"fmt"
+	"log"
 
 	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/apache/arrow/go/v14/arrow/memory"
@@ -71,9 +72,9 @@ func main() {
 	engine := query.NewEngine(memory.DefaultAllocator, database.TableProvider())
 
 	// snowfall statistics by city:
-	_ = engine.ScanTable("snowfall_table").
+	err := engine.ScanTable("snowfall_table").
 		Aggregate(
-			[]logicalplan.Expr{
+			[]*logicalplan.AggregationFunction{
 				logicalplan.Max(logicalplan.Col("snowfall")),
 				logicalplan.Min(logicalplan.Col("snowfall")),
 				logicalplan.Avg(logicalplan.Col("snowfall")),
@@ -85,11 +86,14 @@ func main() {
 			fmt.Println(r)
 			return nil
 		})
+	if err != nil {
+		log.Fatal("snowfall statistics by city:", err)
+	}
 
 	// Total snowfall on each day of week:
-	_ = engine.ScanTable("snowfall_table").
+	err = engine.ScanTable("snowfall_table").
 		Aggregate(
-			[]logicalplan.Expr{
+			[]*logicalplan.AggregationFunction{
 				logicalplan.Sum(logicalplan.Col("snowfall")),
 			},
 			[]logicalplan.Expr{logicalplan.Col("day")},
@@ -99,4 +103,7 @@ func main() {
 			fmt.Println(r)
 			return nil
 		})
+	if err != nil {
+		log.Fatal("total snowfall on each day of week:", err)
+	}
 }
diff --git a/index/compaction.go b/index/compaction.go
index 61ff0f6..6895c5c 100644
--- a/index/compaction.go
+++ b/index/compaction.go
@@ -248,17 +248,14 @@ func (f *FileCompaction) recover(options ...parts.Option) ([]parts.Part, error)
 					return err
 				}
 
-				var tx int
 				txstr, ok := buf.ParquetFile().Lookup(ParquetCompactionTXKey)
 				if !ok {
-					level.Warn(f.logger).Log("msg", "failed to find compaction_tx metadata", "file", file.Name())
-					tx = 0 // Downgrade the compaction tx so that all future reads will be able to read this part.
-				} else {
-					tx, err = strconv.Atoi(txstr)
-					if err != nil {
-						level.Warn(f.logger).Log("msg", "failed to parse compaction_tx metadata", "file", file.Name(), "err", err)
-						tx = 0 // Downgrade the compaction tx so that all future reads will be able to read this part.
-					}
+					return fmt.Errorf("failed to find compaction_tx metadata: %s", file.Name())
+				}
+
+				tx, err := strconv.Atoi(txstr)
+				if err != nil {
+					return fmt.Errorf("failed to parse compaction_tx metadata: %v", err)
 				}
 
 				f.parts.Add(1)
diff --git a/index/lsm.go b/index/lsm.go
index 7d8aa00..5684905 100644
--- a/index/lsm.go
+++ b/index/lsm.go
@@ -195,7 +195,7 @@ func (l *LSM) Close() error {
 
 	// Remove the index directory
 	if err := os.RemoveAll(l.dir); err != nil {
-		return err
+		return fmt.Errorf("remove lsm dir: %w", err)
 	}
 
 	return nil
diff --git a/pqarrow/arrow.go b/pqarrow/arrow.go
index b2006e1..21af248 100644
--- a/pqarrow/arrow.go
+++ b/pqarrow/arrow.go
@@ -20,11 +20,11 @@ import (
 )
 
 // ParquetRowGroupToArrowSchema converts a parquet row group to an arrow schema.
-func ParquetRowGroupToArrowSchema(ctx context.Context, rg parquet.RowGroup, options logicalplan.IterOptions) (*arrow.Schema, error) {
-	return ParquetSchemaToArrowSchema(ctx, rg.Schema(), options)
+func ParquetRowGroupToArrowSchema(ctx context.Context, rg parquet.RowGroup, s *dynparquet.Schema, options logicalplan.IterOptions) (*arrow.Schema, error) {
+	return ParquetSchemaToArrowSchema(ctx, rg.Schema(), s, options)
 }
 
-func ParquetSchemaToArrowSchema(ctx context.Context, schema *parquet.Schema, options logicalplan.IterOptions) (*arrow.Schema, error) {
+func ParquetSchemaToArrowSchema(ctx context.Context, schema *parquet.Schema, s *dynparquet.Schema, options logicalplan.IterOptions) (*arrow.Schema, error) {
 	parquetFields := schema.Fields()
 
 	if len(options.DistinctColumns) == 1 && options.Filter == nil {
@@ -61,7 +61,15 @@ func ParquetSchemaToArrowSchema(ctx context.Context, schema *parquet.Schema, opt
 
 	for _, distinctExpr := range options.DistinctColumns {
 		if distinctExpr.Computed() {
-			dataType, err := distinctExpr.DataType(schema)
+			// Usually we would pass the logical query plan as the data type
+			// finder, but we're here because of an intended layering
+			// violation, which is pushing distinct queries down to the scan
+			// layer. In this case there are no other possible physical types
+			// other than the actual schema, so we can just implement a
+			// simplified version of the type finder that doesn't need to
+			// traverse the logical plan, since this is already the physical
+			// scan layer execution.
+			dataType, err := distinctExpr.DataType(&exprTypeFinder{s: s})
 			if err != nil {
 				return nil, err
 			}
@@ -76,6 +84,14 @@ func ParquetSchemaToArrowSchema(ctx context.Context, schema *parquet.Schema, opt
 	return arrow.NewSchema(fields, nil), nil
 }
 
+type exprTypeFinder struct {
+	s *dynparquet.Schema
+}
+
+func (e *exprTypeFinder) DataTypeForExpr(expr logicalplan.Expr) (arrow.DataType, error) {
+	return logicalplan.DataTypeForExprWithSchema(expr, e.s)
+}
+
 func parquetFieldToArrowField(prefix string, field parquet.Field, physicalProjections []logicalplan.Expr) (arrow.Field, error) {
 	if includedProjection(physicalProjections, fullPath(prefix, field)) {
 		af, err := convert.ParquetFieldToArrowField(field)
@@ -234,8 +250,8 @@ func NewParquetConverter(
 	return c
 }
 
-func (c *ParquetConverter) Convert(ctx context.Context, rg parquet.RowGroup) error {
-	schema, err := ParquetRowGroupToArrowSchema(ctx, rg, c.iterOpts)
+func (c *ParquetConverter) Convert(ctx context.Context, rg parquet.RowGroup, s *dynparquet.Schema) error {
+	schema, err := ParquetRowGroupToArrowSchema(ctx, rg, s, c.iterOpts)
 	if err != nil {
 		return err
 	}
diff --git a/pqarrow/parquet.go b/pqarrow/parquet.go
index 286f06e..ec02a6b 100644
--- a/pqarrow/parquet.go
+++ b/pqarrow/parquet.go
@@ -71,13 +71,13 @@ func (w *singlePassThroughWriter) Reset(_ io.Writer) {}
 // RecordToRow converts an arrow record with dynamic columns into a row using a dynamic parquet schema.
 func RecordToRow(schema *dynparquet.Schema, final *parquet.Schema, record arrow.Record, index int) (parquet.Row, error) {
 	w := &singlePassThroughWriter{}
-	if err := recordToRows(w, schema.IsDynamicColumn, record, index, index+1, final.Fields()); err != nil {
+	if err := recordToRows(w, schema.FindDynamicColumnForConcreteColumn, record, index, index+1, final.Fields()); err != nil {
 		return nil, err
 	}
 	return w.rows[0], nil
 }
 
-type dynamicColumnVerifier func(string) bool
+type dynamicColumnVerifier func(string) (dynparquet.ColumnDefinition, bool)
 
 // recordToRows converts a full arrow record to parquet rows which are written
 // to the parquet writer.
@@ -93,7 +93,7 @@ func recordToRows(w dynparquet.ParquetWriter, dcv dynamicColumnVerifier, record
 		f := finalFields[i]
 		name := f.Name()
 		def := 0
-		if dcv(name) {
+		if _, ok := dcv(name); ok {
 			def = 1
 		}
 		idx := schema.FieldIndices(name)
@@ -380,7 +380,7 @@ func RecordsToFile(schema *dynparquet.Schema, w dynparquet.ParquetWriter, recs [
 	finalFields := ps.Schema.Fields()
 
 	for _, r := range recs {
-		if err := recordToRows(w, schema.IsDynamicColumn, r, 0, int(r.NumRows()), finalFields); err != nil {
+		if err := recordToRows(w, schema.FindDynamicColumnForConcreteColumn, r, 0, int(r.NumRows()), finalFields); err != nil {
 			return err
 		}
 	}
diff --git a/query/engine.go b/query/engine.go
index 749b1fe..030e538 100644
--- a/query/engine.go
+++ b/query/engine.go
@@ -12,7 +12,7 @@ import (
 )
 
 type Builder interface {
-	Aggregate(aggExpr, groupExprs []logicalplan.Expr) Builder
+	Aggregate(aggExpr []*logicalplan.AggregationFunction, groupExprs []logicalplan.Expr) Builder
 	Filter(expr logicalplan.Expr) Builder
 	Distinct(expr ...logicalplan.Expr) Builder
 	Project(projections ...logicalplan.Expr) Builder
@@ -85,7 +85,7 @@ func (e *LocalEngine) ScanSchema(name string) Builder {
 }
 
 func (b LocalQueryBuilder) Aggregate(
-	aggExpr []logicalplan.Expr,
+	aggExpr []*logicalplan.AggregationFunction,
 	groupExprs []logicalplan.Expr,
 ) Builder {
 	return LocalQueryBuilder{
diff --git a/query/logicalplan/builder.go b/query/logicalplan/builder.go
index 52e8cfd..6807cd9 100644
--- a/query/logicalplan/builder.go
+++ b/query/logicalplan/builder.go
@@ -1,14 +1,15 @@
 package logicalplan
 
 import (
+	"errors"
 	"fmt"
 
 	"github.com/apache/arrow/go/v14/arrow"
-	"github.com/parquet-go/parquet-go"
 )
 
 type Builder struct {
 	plan *LogicalPlan
+	err  error
 }
 
 func (b Builder) Scan(
@@ -16,6 +17,7 @@ func (b Builder) Scan(
 	tableName string,
 ) Builder {
 	return Builder{
+		err: b.err,
 		plan: &LogicalPlan{
 			TableScan: &TableScan{
 				TableProvider: provider,
@@ -30,6 +32,7 @@ func (b Builder) ScanSchema(
 	tableName string,
 ) Builder {
 	return Builder{
+		err: b.err,
 		plan: &LogicalPlan{
 			SchemaScan: &SchemaScan{
 				TableProvider: provider,
@@ -43,6 +46,7 @@ func (b Builder) Project(
 	exprs ...Expr,
 ) Builder {
 	return Builder{
+		err: b.err,
 		plan: &LogicalPlan{
 			Input: b.plan,
 			Projection: &Projection{
@@ -58,10 +62,15 @@ type Visitor interface {
 	PostVisit(expr Expr) bool
 }
 
+type ExprTypeFinder interface {
+	DataTypeForExpr(expr Expr) (arrow.DataType, error)
+}
+
 type Expr interface {
-	DataType(*parquet.Schema) (arrow.DataType, error)
+	DataType(ExprTypeFinder) (arrow.DataType, error)
 	Accept(Visitor) bool
 	Name() string
+	Equal(Expr) bool
 	fmt.Stringer
 
 	// ColumnsUsedExprs extracts all the expressions that are used that cause
@@ -95,6 +104,7 @@ func (b Builder) Filter(expr Expr) Builder {
 	}
 
 	return Builder{
+		err: b.err,
 		plan: &LogicalPlan{
 			Input: b.plan,
 			Filter: &Filter{
@@ -108,31 +118,114 @@ func (b Builder) Distinct(
 	exprs ...Expr,
 ) Builder {
 	return Builder{
+		err: b.err,
 		plan: &LogicalPlan{
-			Input: b.plan,
 			Distinct: &Distinct{
 				Exprs: exprs,
 			},
+			Input: &LogicalPlan{
+				Projection: &Projection{
+					Exprs: exprs,
+				},
+				Input: b.plan,
+			},
 		},
 	}
 }
 
 func (b Builder) Aggregate(
-	aggExpr []Expr,
+	aggExpr []*AggregationFunction,
 	groupExprs []Expr,
 ) Builder {
+	resolvedAggExpr := make([]*AggregationFunction, 0, len(aggExpr))
+	projectExprs := make([]Expr, 0, len(aggExpr))
+	needsPostProcessing := false
+
+	var err error
+	for _, agg := range aggExpr {
+		resolvedAggregations, projections, changed, rerr := resolveAggregation(b.plan, agg)
+		if err != nil {
+			err = errors.Join(err, rerr)
+		}
+
+		if changed {
+			needsPostProcessing = true
+		}
+
+		resolvedAggExpr = append(resolvedAggExpr, resolvedAggregations...)
+		projectExprs = append(projectExprs, projections...)
+	}
+
+	if !needsPostProcessing {
+		return Builder{
+			err: err,
+			plan: &LogicalPlan{
+				Aggregation: &Aggregation{
+					GroupExprs: groupExprs,
+					AggExprs:   aggExpr,
+				},
+				Input: b.plan,
+			},
+		}
+	}
+
 	return Builder{
+		err: err,
 		plan: &LogicalPlan{
-			Input: b.plan,
-			Aggregation: &Aggregation{
-				GroupExprs: groupExprs,
-				AggExprs:   aggExpr,
+			Projection: &Projection{
+				Exprs: append(groupExprs, projectExprs...),
+			},
+			Input: &LogicalPlan{
+				Aggregation: &Aggregation{
+					GroupExprs: groupExprs,
+					AggExprs:   resolvedAggExpr,
+				},
+				Input: b.plan,
 			},
 		},
 	}
 }
 
+func resolveAggregation(plan *LogicalPlan, agg *AggregationFunction) ([]*AggregationFunction, []Expr, bool, error) {
+	switch agg.Func {
+	case AggFuncAvg:
+		sum := &AggregationFunction{
+			Func: AggFuncSum,
+			Expr: agg.Expr,
+		}
+		count := &AggregationFunction{
+			Func: AggFuncCount,
+			Expr: agg.Expr,
+		}
+
+		var (
+			countExpr Expr = count
+			aggType   arrow.DataType
+		)
+		aggType, err := agg.Expr.DataType(plan)
+		// intentionally not handling the error here, as it will be handled
+		// in the build function.
+		if !arrow.TypeEqual(aggType, arrow.PrimitiveTypes.Int64) {
+			countExpr = Convert(countExpr, aggType)
+		}
+
+		div := (&BinaryExpr{
+			Left:  sum,
+			Op:    OpDiv,
+			Right: countExpr,
+		}).Alias(agg.String())
+
+		return []*AggregationFunction{sum, count}, []Expr{div}, true, err
+	default:
+		return []*AggregationFunction{agg}, []Expr{agg}, false, nil
+	}
+}
+
 func (b Builder) Build() (*LogicalPlan, error) {
+	if b.err != nil {
+		return nil, b.err
+	}
+
 	if err := Validate(b.plan); err != nil {
 		return nil, err
 	}
diff --git a/query/logicalplan/expr.go b/query/logicalplan/expr.go
index 8f73391..7ea34cd 100644
--- a/query/logicalplan/expr.go
+++ b/query/logicalplan/expr.go
@@ -3,15 +3,11 @@ package logicalplan
 import (
 	"errors"
 	"fmt"
-	"regexp"
 	"strings"
 	"time"
 
 	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/apache/arrow/go/v14/arrow/scalar"
-	"github.com/parquet-go/parquet-go"
-
-	"github.com/polarsignals/frostdb/pqarrow/convert"
 )
 
 type Op uint32
@@ -106,6 +102,19 @@ type BinaryExpr struct {
 	Right Expr
 }
 
+func (e *BinaryExpr) Equal(other Expr) bool {
+	if other == nil {
+		// if both are nil, they are equal
+		return e == nil
+	}
+
+	if b, ok := other.(*BinaryExpr); ok {
+		return e.Op == b.Op && e.Left.Equal(b.Left) && e.Right.Equal(b.Right)
+	}
+
+	return false
+}
+
 func (e *BinaryExpr) Clone() Expr {
 	return &BinaryExpr{
 		Left:  e.Left.Clone(),
@@ -138,26 +147,26 @@ func (e *BinaryExpr) Accept(visitor Visitor) bool {
 	return visitor.PostVisit(e)
 }
 
-func (e *BinaryExpr) DataType(s *parquet.Schema) (arrow.DataType, error) {
+func (e *BinaryExpr) DataType(l ExprTypeFinder) (arrow.DataType, error) {
+	leftType, err := e.Left.DataType(l)
+	if err != nil {
+		return nil, fmt.Errorf("left operand: %w", err)
+	}
+
+	rightType, err := e.Right.DataType(l)
+	if err != nil {
+		return nil, fmt.Errorf("right operand: %w", err)
+	}
+
+	if !arrow.TypeEqual(leftType, rightType) {
+		return nil, fmt.Errorf("left and right operands must be of the same type, got %s and %s", leftType, rightType)
+	}
+
 	switch e.Op {
 	case OpEq, OpNotEq, OpLt, OpLtEq, OpGt, OpGtEq, OpAnd, OpOr:
 		return arrow.FixedWidthTypes.Boolean, nil
 	case OpAdd, OpSub, OpMul, OpDiv:
-		leftType, err := e.Left.DataType(s)
-		if err != nil {
-			return nil, fmt.Errorf("left operand: %w", err)
-		}
-
-		rightType, err := e.Right.DataType(s)
-		if err != nil {
-			return nil, fmt.Errorf("right operand: %w", err)
-		}
-
-		if arrow.TypeEqual(leftType, rightType, arrow.CheckMetadata()) {
-			return leftType, nil
-		}
-
-		return nil, fmt.Errorf("left and right operands must be of the same type, got %s and %s", leftType, rightType)
+		return leftType, nil
 	default:
 		return nil, errors.New("unknown operator")
 	}
@@ -189,10 +198,108 @@ func (e *BinaryExpr) Alias(alias string) *AliasExpr {
 	return &AliasExpr{Expr: e, Alias: alias}
 }
 
+func Convert(e Expr, t arrow.DataType) *ConvertExpr {
+	return &ConvertExpr{Expr: e, Type: t}
+}
+
+type ConvertExpr struct {
+	Expr Expr
+	Type arrow.DataType
+}
+
+func (e *ConvertExpr) Equal(other Expr) bool {
+	if other == nil {
+		// if both are nil, they are equal
+		return e == nil
+	}
+
+	if c, ok := other.(*ConvertExpr); ok {
+		return arrow.TypeEqual(e.Type, c.Type) && e.Expr.Equal(c.Expr)
+	}
+
+	return false
+}
+
+func (e *ConvertExpr) Clone() Expr {
+	return &ConvertExpr{
+		Expr: e.Expr.Clone(),
+		Type: e.Type,
+	}
+}
+
+func (e *ConvertExpr) Accept(visitor Visitor) bool {
+	continu := visitor.PreVisit(e)
+	if !continu {
+		return false
+	}
+
+	continu = e.Expr.Accept(visitor)
+	if !continu {
+		return false
+	}
+
+	continu = visitor.Visit(e)
+	if !continu {
+		return false
+	}
+
+	return visitor.PostVisit(e)
+}
+
+func (e *ConvertExpr) DataType(l ExprTypeFinder) (arrow.DataType, error) {
+	// We don't care about the result, but we want to make sure the expression
+	// tree typing is correct.
+	_, err := e.Expr.DataType(l)
+	if err != nil {
+		return nil, fmt.Errorf("convert type: %w", err)
+	}
+
+	return e.Type, nil
+}
+
+func (e *ConvertExpr) Name() string {
+	return "convert(" + e.Expr.Name() + ", " + e.Type.String() + ")"
+}
+
+func (e *ConvertExpr) String() string { return e.Name() }
+
+func (e *ConvertExpr) ColumnsUsedExprs() []Expr {
+	return e.Expr.ColumnsUsedExprs()
+}
+
+func (e *ConvertExpr) MatchPath(path string) bool {
+	return strings.HasPrefix(e.Name(), path)
+}
+
+func (e *ConvertExpr) MatchColumn(columnName string) bool {
+	return e.Name() == columnName
+}
+
+func (e *ConvertExpr) Computed() bool {
+	return true
+}
+
+func (e *ConvertExpr) Alias(alias string) *AliasExpr {
+	return &AliasExpr{Expr: e, Alias: alias}
+}
+
 type Column struct {
 	ColumnName string
 }
 
+func (c *Column) Equal(other Expr) bool {
+	if other == nil {
+		// if both are nil, they are equal
+		return c == nil
+	}
+
+	if col, ok := other.(*Column); ok {
+		return c.ColumnName == col.ColumnName
+	}
+
+	return false
+}
+
 func (c *Column) Clone() Expr {
 	return &Column{ColumnName: c.ColumnName}
 }
@@ -216,52 +323,13 @@ func (c *Column) Name() string {
 
 func (c *Column) String() string { return c.Name() }
 
-func (c *Column) DataType(s *parquet.Schema) (arrow.DataType, error) {
-	for _, field := range s.Fields() {
-		af, err := c.findField("", field)
-		if err != nil {
-			return nil, err
-		}
-		if af.Name != "" {
-			return af.Type, nil
-		}
+func (c *Column) DataType(l ExprTypeFinder) (arrow.DataType, error) {
+	t, err := l.DataTypeForExpr(c)
+	if err != nil {
+		return nil, fmt.Errorf("column %q type: %w", c.ColumnName, err)
 	}
 
-	return nil, errors.New("column not found")
-}
-
-func fullPath(prefix string, parquetField parquet.Field) string {
-	if prefix == "" {
-		return parquetField.Name()
-	}
-	return strings.Join([]string{prefix, parquetField.Name()}, ".")
-}
-
-func (c *Column) findField(prefix string, field parquet.Field) (arrow.Field, error) {
-	if c.ColumnName == fullPath(prefix, field) {
-		return convert.ParquetFieldToArrowField(field)
-	}
-
-	if !field.Leaf() && strings.HasPrefix(c.ColumnName, fullPath(prefix, field)) {
-		group := []arrow.Field{}
-		for _, f := range field.Fields() {
-			af, err := c.findField(fullPath(prefix, field), f)
-			if err != nil {
-				return arrow.Field{}, err
-			}
-			if af.Name != "" {
-				group = append(group, af)
-			}
-		}
-		if len(group) > 0 {
-			return arrow.Field{
-				Name:     field.Name(),
-				Type:     arrow.StructOf(group...),
-				Nullable: field.Optional(),
-			}, nil
-		}
-	}
-	return arrow.Field{}, nil
+	return t, nil
 }
 
 func (c *Column) Alias(alias string) *AliasExpr {
@@ -429,6 +497,19 @@ type DynamicColumn struct {
 	ColumnName string
 }
 
+func (c *DynamicColumn) Equal(other Expr) bool {
+	if other == nil {
+		// if both are nil, they are equal
+		return c == nil
+	}
+
+	if col, ok := other.(*DynamicColumn); ok {
+		return c.ColumnName == col.ColumnName
+	}
+
+	return false
+}
+
 func (c *DynamicColumn) Clone() Expr {
 	return &DynamicColumn{ColumnName: c.ColumnName}
 }
@@ -441,16 +522,13 @@ func DynCol(name string) *DynamicColumn {
 	return &DynamicColumn{ColumnName: name}
 }
 
-func (c *DynamicColumn) DataType(s *parquet.Schema) (arrow.DataType, error) {
-	for _, field := range s.Fields() {
-		if names := strings.Split(field.Name(), "."); len(names) == 2 {
-			if names[0] == c.ColumnName {
-				return convert.ParquetNodeToType(field)
-			}
-		}
+func (c *DynamicColumn) DataType(l ExprTypeFinder) (arrow.DataType, error) {
+	t, err := l.DataTypeForExpr(c)
+	if err != nil {
+		return nil, fmt.Errorf("dynamic column %q type: %w", c.ColumnName, err)
 	}
 
-	return nil, errors.New("column not found")
+	return t, nil
 }
 
 func (c *DynamicColumn) ColumnsUsedExprs() []Expr {
@@ -487,6 +565,19 @@ type LiteralExpr struct {
 	Value scalar.Scalar
 }
 
+func (e *LiteralExpr) Equal(other Expr) bool {
+	if other == nil {
+		// if both are nil, they are equal
+		return e == nil
+	}
+
+	if lit, ok := other.(*LiteralExpr); ok {
+		return scalar.Equals(e.Value, lit.Value)
+	}
+
+	return false
+}
+
 func (e *LiteralExpr) Clone() Expr {
 	return &LiteralExpr{
 		Value: e.Value,
@@ -503,7 +594,7 @@ func Literal(v interface{}) *LiteralExpr {
 	}
 }
 
-func (e *LiteralExpr) DataType(_ *parquet.Schema) (arrow.DataType, error) {
+func (e *LiteralExpr) DataType(_ ExprTypeFinder) (arrow.DataType, error) {
 	return e.Value.DataType(), nil
 }
 
@@ -537,6 +628,19 @@ type AggregationFunction struct {
 	Expr Expr
 }
 
+func (f *AggregationFunction) Equal(other Expr) bool {
+	if other == nil {
+		// if both are nil, they are equal
+		return f == nil
+	}
+
+	if agg, ok := other.(*AggregationFunction); ok {
+		return f.Func == agg.Func && f.Expr.Equal(agg.Expr)
+	}
+
+	return false
+}
+
 func (f *AggregationFunction) Clone() Expr {
 	return &AggregationFunction{
 		Func: f.Func,
@@ -544,8 +648,8 @@ func (f *AggregationFunction) Clone() Expr {
 	}
 }
 
-func (f *AggregationFunction) DataType(s *parquet.Schema) (arrow.DataType, error) {
-	return f.Expr.DataType(s)
+func (f *AggregationFunction) DataType(l ExprTypeFinder) (arrow.DataType, error) {
+	return f.Expr.DataType(l)
 }
 
 func (f *AggregationFunction) Accept(visitor Visitor) bool {
@@ -559,6 +663,11 @@ func (f *AggregationFunction) Accept(visitor Visitor) bool {
 		return false
 	}
 
+	continu = visitor.Visit(f)
+	if !continu {
+		return false
+	}
+
 	return visitor.PostVisit(f)
 }
 
@@ -652,6 +761,19 @@ type AliasExpr struct {
 	Alias string
 }
 
+func (e *AliasExpr) Equal(other Expr) bool {
+	if other == nil {
+		// if both are nil, they are equal
+		return e == nil
+	}
+
+	if alias, ok := other.(*AliasExpr); ok {
+		return e.Alias == alias.Alias && e.Expr.Equal(alias.Expr)
+	}
+
+	return false
+}
+
 func (e *AliasExpr) Clone() Expr {
 	return &AliasExpr{
 		Expr:  e.Expr.Clone(),
@@ -659,15 +781,15 @@ func (e *AliasExpr) Clone() Expr {
 	}
 }
 
-func (e *AliasExpr) DataType(s *parquet.Schema) (arrow.DataType, error) {
-	return e.Expr.DataType(s)
+func (e *AliasExpr) DataType(l ExprTypeFinder) (arrow.DataType, error) {
+	return e.Expr.DataType(l)
 }
 
 func (e *AliasExpr) Name() string {
 	return e.Alias
 }
 
-func (e *AliasExpr) String() string { return e.Name() }
+func (e *AliasExpr) String() string { return fmt.Sprintf("%s as %s", e.Expr.String(), e.Alias) }
 
 func (e *AliasExpr) Computed() bool {
 	return e.Expr.Computed()
@@ -714,13 +836,26 @@ type DurationExpr struct {
 	duration time.Duration
 }
 
+func (d *DurationExpr) Equal(other Expr) bool {
+	if other == nil {
+		// if both are nil, they are equal
+		return d == nil
+	}
+
+	if dur, ok := other.(*DurationExpr); ok {
+		return d.duration == dur.duration
+	}
+
+	return false
+}
+
 func (d *DurationExpr) Clone() Expr {
 	return &DurationExpr{
 		duration: d.duration,
 	}
 }
 
-func (d *DurationExpr) DataType(_ *parquet.Schema) (arrow.DataType, error) {
+func (d *DurationExpr) DataType(_ ExprTypeFinder) (arrow.DataType, error) {
 	return &arrow.DurationType{}, nil
 }
 
@@ -738,7 +873,7 @@ func (d *DurationExpr) Accept(visitor Visitor) bool {
 }
 
 func (d *DurationExpr) Name() string {
-	return ""
+	return fmt.Sprintf("second(%d)", int(d.duration.Seconds()))
 }
 
 func (d *DurationExpr) String() string { return d.Name() }
@@ -760,117 +895,23 @@ func (d *DurationExpr) Value() time.Duration {
 	return d.duration
 }
 
-type AverageExpr struct {
-	Expr Expr
-}
-
-func (a *AverageExpr) Clone() Expr {
-	return &AverageExpr{
-		Expr: a.Expr.Clone(),
-	}
-}
-
-func (a *AverageExpr) DataType(s *parquet.Schema) (arrow.DataType, error) {
-	return a.Expr.DataType(s)
-}
-
-func (a *AverageExpr) Name() string {
-	return a.Expr.Name()
-}
-
-func (a *AverageExpr) String() string { return a.Name() }
-
-func (a *AverageExpr) ColumnsUsedExprs() []Expr {
-	return a.Expr.ColumnsUsedExprs()
-}
-
-func (a *AverageExpr) MatchPath(path string) bool {
-	return a.Expr.MatchPath(path)
-}
-
-func (a *AverageExpr) MatchColumn(name string) bool {
-	return a.Expr.MatchColumn(name)
-}
-
-func (a *AverageExpr) Computed() bool {
-	return true
-}
-
-func (a *AverageExpr) Accept(visitor Visitor) bool {
-	continu := visitor.PreVisit(a)
-	if !continu {
-		return false
-	}
-
-	return visitor.PostVisit(a)
-}
-
-func RegExpColumnMatch(match *regexp.Regexp) *RegexpColumnMatch {
-	return &RegexpColumnMatch{
-		match: match,
-	}
-}
-
-func RegExpNotColumnMatch(match *regexp.Regexp) *RegexpColumnMatch {
-	return &RegexpColumnMatch{
-		inverse: true,
-		match:   match,
-	}
-}
-
-type RegexpColumnMatch struct {
-	inverse bool
-	match   *regexp.Regexp
-}
-
-func (a *RegexpColumnMatch) Clone() Expr {
-	return &RegexpColumnMatch{
-		match: a.match,
-	}
-}
-
-func (a *RegexpColumnMatch) DataType(_ *parquet.Schema) (arrow.DataType, error) {
-	return nil, nil
-}
-
-func (a *RegexpColumnMatch) Name() string {
-	return a.match.String()
-}
-
-func (a *RegexpColumnMatch) String() string { return a.Name() }
-
-func (a *RegexpColumnMatch) ColumnsUsedExprs() []Expr {
-	return []Expr{a}
-}
-
-func (a *RegexpColumnMatch) MatchPath(path string) bool {
-	return a.match.MatchString(path) != a.inverse
-}
-
-func (a *RegexpColumnMatch) MatchColumn(name string) bool {
-	return a.match.MatchString(name) != a.inverse
-}
+type AllExpr struct{}
 
-func (a *RegexpColumnMatch) Computed() bool {
-	return true
+func All() *AllExpr {
+	return &AllExpr{}
 }
 
-func (a *RegexpColumnMatch) Accept(visitor Visitor) bool {
-	continu := visitor.PreVisit(a)
-	if !continu {
-		return false
+func (a *AllExpr) Equal(other Expr) bool {
+	if other == nil {
+		// if both are nil, they are equal
+		return a == nil
 	}
 
-	return visitor.PostVisit(a)
+	_, ok := other.(*AllExpr)
+	return ok
 }
 
-type AllExpr struct{}
-
-func All() *AllExpr {
-	return &AllExpr{}
-}
-
-func (a *AllExpr) DataType(*parquet.Schema) (arrow.DataType, error) { return nil, nil }
+func (a *AllExpr) DataType(ExprTypeFinder) (arrow.DataType, error) { return nil, nil }
 func (a *AllExpr) Accept(visitor Visitor) bool {
 	continu := visitor.PreVisit(a)
 	if !continu {
@@ -899,7 +940,20 @@ func Not(expr Expr) *NotExpr {
 	}
 }
 
-func (n *NotExpr) DataType(*parquet.Schema) (arrow.DataType, error) { return nil, nil }
+func (n *NotExpr) Equal(other Expr) bool {
+	if other == nil {
+		// if both are nil, they are equal
+		return n == nil
+	}
+
+	if not, ok := other.(*NotExpr); ok {
+		return n.Expr.Equal(not.Expr)
+	}
+
+	return false
+}
+
+func (n *NotExpr) DataType(ExprTypeFinder) (arrow.DataType, error) { return nil, nil }
 func (n *NotExpr) Accept(visitor Visitor) bool {
 	continu := visitor.PreVisit(n)
 	if !continu {
diff --git a/query/logicalplan/logicalplan.go b/query/logicalplan/logicalplan.go
index d3c7ba1..a0fc4dc 100644
--- a/query/logicalplan/logicalplan.go
+++ b/query/logicalplan/logicalplan.go
@@ -9,6 +9,7 @@ import (
 	"github.com/apache/arrow/go/v14/arrow/memory"
 
 	"github.com/polarsignals/frostdb/dynparquet"
+	"github.com/polarsignals/frostdb/pqarrow/convert"
 )
 
 // LogicalPlan is a logical representation of a query. Each LogicalPlan is a
@@ -100,6 +101,63 @@ func (plan *LogicalPlan) string(indent int) string {
 	return res
 }
 
+func (plan *LogicalPlan) DataTypeForExpr(expr Expr) (arrow.DataType, error) {
+	switch {
+	case plan.SchemaScan != nil:
+		t, err := plan.SchemaScan.DataTypeForExpr(expr)
+		if err != nil {
+			return nil, fmt.Errorf("data type for expr %v within SchemaScan: %w", expr, err)
+		}
+
+		return t, nil
+	case plan.TableScan != nil:
+		t, err := plan.TableScan.DataTypeForExpr(expr)
+		if err != nil {
+			return nil, fmt.Errorf("data type for expr %v within TableScan: %w", expr, err)
+		}
+
+		return t, nil
+	case plan.Filter != nil:
+		t, err := plan.Input.DataTypeForExpr(expr)
+		if err != nil {
+			return nil, fmt.Errorf("data type for expr %v within Filter: %w", expr, err)
+		}
+
+		return t, nil
+	case plan.Projection != nil:
+		t, err := expr.DataType(plan.Input)
+		if err != nil {
+			return nil, fmt.Errorf("data type for expr %v within Projection: %w", expr, err)
+		}
+
+		return t, nil
+	case plan.Aggregation != nil:
+		if agg, ok := expr.(*AggregationFunction); ok {
+			if agg.Func == AggFuncCount {
+				return arrow.PrimitiveTypes.Int64, nil
+			}
+
+			return agg.Expr.DataType(plan.Input)
+		}
+
+		t, err := expr.DataType(plan.Input)
+		if err != nil {
+			return nil, fmt.Errorf("data type for expr %v within Aggregation: %w", expr, err)
+		}
+
+		return t, nil
+	case plan.Distinct != nil:
+		t, err := expr.DataType(plan.Input)
+		if err != nil {
+			return nil, fmt.Errorf("data type for expr %v within Distinct: %w", expr, err)
+		}
+
+		return t, nil
+	default:
+		return nil, fmt.Errorf("unknown logical plan")
+	}
+}
+
 // TableReader returns the table reader.
 func (plan *LogicalPlan) TableReader() (TableReader, error) {
 	if plan.TableScan != nil {
@@ -172,7 +230,8 @@ type TableScan struct {
 	TableName     string
 
 	// PhysicalProjection describes the columns that are to be physically read
-	// by the table scan.
+	// by the table scan. This is an Expr so it can be either a column or
+	// dynamic column.
 	PhysicalProjection []Expr
 
 	// Filter is the predicate that is to be applied by the table scan to rule
@@ -189,6 +248,63 @@ type TableScan struct {
 	SkipSources bool
 }
 
+func (scan *TableScan) DataTypeForExpr(expr Expr) (arrow.DataType, error) {
+	tp, err := scan.TableProvider.GetTable(scan.TableName)
+	if err != nil {
+		return nil, fmt.Errorf("get table %q: %w", scan.TableName, err)
+	}
+
+	s := tp.Schema()
+
+	t, err := DataTypeForExprWithSchema(expr, s)
+	if err != nil {
+		return nil, fmt.Errorf("type for expr %q in table %q", expr, scan.TableName)
+	}
+
+	return t, nil
+}
+
+func DataTypeForExprWithSchema(expr Expr, s *dynparquet.Schema) (arrow.DataType, error) {
+	switch expr := expr.(type) {
+	case *Column:
+		colDef, found := s.FindDynamicColumnForConcreteColumn(expr.ColumnName)
+		if found {
+			t, err := convert.ParquetNodeToType(colDef.StorageLayout)
+			if err != nil {
+				return nil, fmt.Errorf("convert parquet node to type: %w", err)
+			}
+
+			return t, nil
+		}
+
+		colDef, found = s.FindColumn(expr.ColumnName)
+		if found {
+			t, err := convert.ParquetNodeToType(colDef.StorageLayout)
+			if err != nil {
+				return nil, fmt.Errorf("convert parquet node to type: %w", err)
+			}
+
+			return t, nil
+		}
+
+		return nil, fmt.Errorf("column %q not found", expr.ColumnName)
+	case *DynamicColumn:
+		colDef, found := s.FindDynamicColumn(expr.ColumnName)
+		if found {
+			t, err := convert.ParquetNodeToType(colDef.StorageLayout)
+			if err != nil {
+				return nil, fmt.Errorf("convert parquet node to type: %w", err)
+			}
+
+			return t, nil
+		}
+
+		return nil, fmt.Errorf("dynamic column %q not found", expr.ColumnName)
+	default:
+		return nil, fmt.Errorf("unhandled expr type %T", expr)
+	}
+}
+
 func (scan *TableScan) String() string {
 	return "TableScan" +
 		" Table: " + scan.TableName +
@@ -223,6 +339,19 @@ func (s *SchemaScan) String() string {
 	return "SchemaScan"
 }
 
+func (s *SchemaScan) DataTypeForExpr(expr Expr) (arrow.DataType, error) {
+	switch expr := expr.(type) {
+	case *Column:
+		if expr.ColumnName == "name" {
+			return arrow.BinaryTypes.String, nil
+		}
+
+		return nil, fmt.Errorf("unknown column %s", expr.ColumnName)
+	default:
+		return nil, fmt.Errorf("unhandled expr %T", expr)
+	}
+}
+
 type Filter struct {
 	Expr Expr
 }
@@ -248,7 +377,7 @@ func (p *Projection) String() string {
 }
 
 type Aggregation struct {
-	AggExprs   []Expr
+	AggExprs   []*AggregationFunction
 	GroupExprs []Expr
 }
 
diff --git a/query/logicalplan/optimize.go b/query/logicalplan/optimize.go
index 74cc3c7..f0a08c9 100644
--- a/query/logicalplan/optimize.go
+++ b/query/logicalplan/optimize.go
@@ -1,7 +1,5 @@
 package logicalplan
 
-import "slices"
-
 var hashedMatch = "hashed"
 
 type Optimizer interface {
@@ -10,7 +8,6 @@ type Optimizer interface {
 
 func DefaultOptimizers() []Optimizer {
 	return []Optimizer{
-		&AverageAggregationPushDown{},
 		&PhysicalProjectionPushDown{
 			defaultProjections: []Expr{
 				Not(DynCol(hashedMatch)),
@@ -18,74 +15,14 @@ func DefaultOptimizers() []Optimizer {
 		},
 		&FilterPushDown{},
 		&DistinctPushDown{},
-		&ProjectionPushDown{},
 	}
 }
 
-type AverageAggregationPushDown struct{}
-
-func (p *AverageAggregationPushDown) Optimize(plan *LogicalPlan) *LogicalPlan {
-	if plan.Aggregation == nil {
-		return plan
-	}
-
-	for i, aggExpr := range plan.Aggregation.AggExprs {
-		var aggFunc *AggregationFunction
-		var alias *AliasExpr
-		var column Expr
-
-		// In case the aggregation contains an alias
-		if aliasExpr, ok := aggExpr.(*AliasExpr); ok {
-			alias = aliasExpr
-			if af, ok := aliasExpr.Expr.(*AggregationFunction); ok {
-				if af.Func == AggFuncAvg {
-					column = af.Expr
-					aggFunc = af
-				}
-			}
-		}
-		if ae, ok := aggExpr.(*AggregationFunction); ok {
-			if ae.Func == AggFuncAvg {
-				column = ae.Expr
-				aggFunc = ae
-			}
-		}
-
-		if aggFunc == nil {
-			// no aggregation func found, skipping
-			continue
-		}
-
-		// Delete this average aggregation from the logicalplan.
-		plan.Aggregation.AggExprs = slices.Delete(plan.Aggregation.AggExprs, i, i+1)
-		// Add sum and count aggregation for the column to the logicalplan.
-		plan.Aggregation.AggExprs = append(plan.Aggregation.AggExprs,
-			Sum(aggFunc.Expr),
-			Count(aggFunc.Expr),
-		)
-
-		projection := &AverageExpr{Expr: column}
-		if alias != nil {
-			alias.Expr = column
-			projection.Expr = alias
-		}
-
-		// Wrap the aggregations with the average projection to always call it after aggregating.
-		plan = &LogicalPlan{
-			Input: plan,
-			Projection: &Projection{
-				Exprs: []Expr{projection},
-			},
-		}
-	}
-
-	return plan
-}
-
-// PhysicalProjectionPushDown optimizer tries to push down the actual
-// 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.
+// PhysicalProjectionPushDown finds the first projecting logical plan and
+// collects all columns it needs, it is concatenated with all other columns
+// used until it, for example a filter layer. Because the tree has the scan
+// layer as the inner most layer, the logic actually works by resetting the
+// list every time a projecting layer is found.
 type PhysicalProjectionPushDown struct {
 	defaultProjections []Expr
 }
@@ -105,16 +42,20 @@ func (p *PhysicalProjectionPushDown) optimize(plan *LogicalPlan, columnsUsedExpr
 		p.defaultProjections = []Expr{}
 		columnsUsedExprs = append(columnsUsedExprs, plan.Filter.Expr.ColumnsUsedExprs()...)
 	case plan.Distinct != nil:
-		p.defaultProjections = []Expr{}
+		// distinct is projecting so we need to reset
+		columnsUsedExprs = []Expr{}
 		for _, expr := range plan.Distinct.Exprs {
 			columnsUsedExprs = append(columnsUsedExprs, expr.ColumnsUsedExprs()...)
 		}
 	case plan.Projection != nil:
-		p.defaultProjections = []Expr{}
+		// projections are is projecting so we need to reset
+		columnsUsedExprs = []Expr{}
 		for _, expr := range plan.Projection.Exprs {
 			columnsUsedExprs = append(columnsUsedExprs, expr.ColumnsUsedExprs()...)
 		}
 	case plan.Aggregation != nil:
+		// aggregations are projecting so we need to reset
+		columnsUsedExprs = []Expr{}
 		for _, expr := range plan.Aggregation.GroupExprs {
 			columnsUsedExprs = append(columnsUsedExprs, expr.ColumnsUsedExprs()...)
 		}
@@ -130,154 +71,6 @@ func (p *PhysicalProjectionPushDown) optimize(plan *LogicalPlan, columnsUsedExpr
 	}
 }
 
-// ProjectionPushDown finds the projection expressions that can be pushed
-// down. If there is no projection expression, but there is an implicit
-// projection such as a `Distinct` query plan, then it will insert a new
-// projection plan and push it down. It functions in three steps, first it will
-// find the projection expressions in the plan, then remove explicit projection
-// plans from the overall plan if it exists, and will then synthesize one if it
-// doesn't exist, and insert it in the deepest possible position in the plan.
-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)
-	aggColumns := aggregationColumns(plan)
-	for _, m := range projectColumns {
-		projectMap[m.Name()] = true
-	}
-	for _, m := range filterColumns {
-		if !projectMap[m.Name()] {
-			return plan
-		}
-	}
-	for _, m := range aggColumns {
-		if !projectMap[m.Name()] {
-			return plan
-		}
-	}
-
-	c := &projectionCollector{}
-	c.collect(plan)
-
-	if len(c.projections) == 0 {
-		// If there are no projection expressions, then we don't need to do
-		// anything.
-		return plan
-	}
-
-	plan = removeProjection(plan)
-	return insertProjection(plan, &Projection{Exprs: c.projections})
-}
-
-type projectionCollector struct {
-	projections []Expr
-}
-
-func (p *projectionCollector) collect(plan *LogicalPlan) {
-	switch {
-	case plan.Distinct != nil:
-		p.projections = append(p.projections, plan.Distinct.Exprs...)
-	case plan.Projection != nil:
-		p.projections = append(p.projections, plan.Projection.Exprs...)
-	}
-
-	if plan.Input != nil {
-		p.collect(plan.Input)
-	}
-}
-
-// filterColumns returns all the column matchers for filters in a given plan.
-func filterColumns(plan *LogicalPlan) []Expr {
-	if plan == nil {
-		return nil
-	}
-
-	columnsUsedExprs := []Expr{}
-	switch {
-	case plan.Filter != nil:
-		columnsUsedExprs = append(columnsUsedExprs, plan.Filter.Expr.ColumnsUsedExprs()...)
-	}
-
-	return append(columnsUsedExprs, filterColumns(plan.Input)...)
-}
-
-func aggregationColumns(plan *LogicalPlan) []Expr {
-	if plan == nil {
-		return nil
-	}
-
-	columnsUsedExprs := []Expr{}
-	switch {
-	case plan.Aggregation != nil:
-		for _, expr := range plan.Aggregation.GroupExprs {
-			columnsUsedExprs = append(columnsUsedExprs, expr.ColumnsUsedExprs()...)
-		}
-		for _, expr := range plan.Aggregation.AggExprs {
-			columnsUsedExprs = append(columnsUsedExprs, expr.ColumnsUsedExprs()...)
-		}
-	}
-
-	return append(columnsUsedExprs, aggregationColumns(plan.Input)...)
-}
-
-// projectionColumns returns all the column matchers for projections in a given plan.
-func projectionColumns(plan *LogicalPlan) []Expr {
-	if plan == nil {
-		return nil
-	}
-
-	columnsUsedExprs := []Expr{}
-	switch {
-	case plan.Projection != nil:
-		for _, expr := range plan.Projection.Exprs {
-			columnsUsedExprs = append(columnsUsedExprs, expr.ColumnsUsedExprs()...)
-		}
-	}
-
-	return append(columnsUsedExprs, projectionColumns(plan.Input)...)
-}
-
-func removeProjection(plan *LogicalPlan) *LogicalPlan {
-	if plan == nil {
-		return nil
-	}
-
-	switch {
-	case plan.Projection != nil:
-		return plan.Input
-	}
-
-	plan.Input = removeProjection(plan.Input)
-	return plan
-}
-
-func insertProjection(cur *LogicalPlan, projection *Projection) *LogicalPlan {
-	if cur == nil {
-		return nil
-	}
-
-	switch {
-	case cur.TableScan != nil:
-		return &LogicalPlan{
-			Input:      cur,
-			Projection: projection,
-		}
-	case cur.SchemaScan != nil:
-		return &LogicalPlan{
-			Input:      cur,
-			Projection: projection,
-		}
-	}
-
-	cur.Input = insertProjection(cur.Input, projection)
-	return cur
-}
-
 // FilterPushDown optimizer tries to push down the filters of a query down
 // to the actual physical table scan. This allows the table provider to make
 // smarter decisions about which pieces of data to load in the first place or
@@ -323,6 +116,20 @@ func (p *DistinctPushDown) Optimize(plan *LogicalPlan) *LogicalPlan {
 	return plan
 }
 
+func exprsEqual(a, b []Expr) bool {
+	if len(a) != len(b) {
+		return false
+	}
+
+	for i, expr := range a {
+		if !expr.Equal(b[i]) {
+			return false
+		}
+	}
+
+	return true
+}
+
 func (p *DistinctPushDown) optimize(plan *LogicalPlan, distinctColumns []Expr) {
 	switch {
 	case plan.TableScan != nil:
@@ -331,6 +138,16 @@ func (p *DistinctPushDown) optimize(plan *LogicalPlan, distinctColumns []Expr) {
 		}
 	case plan.Distinct != nil:
 		distinctColumns = append(distinctColumns, plan.Distinct.Exprs...)
+	case plan.Projection != nil:
+		if !exprsEqual(distinctColumns, plan.Projection.Exprs) {
+			// if and only if the distinct columns are identical to the
+			// projection columns we can perform the optimization, so we need
+			// to reset it in this case.
+			distinctColumns = []Expr{}
+		}
+	default:
+		// reset distinct columns
+		distinctColumns = []Expr{}
 	}
 
 	if plan.Input != nil {
diff --git a/query/logicalplan/validate.go b/query/logicalplan/validate.go
index 9e04dd8..11193db 100644
--- a/query/logicalplan/validate.go
+++ b/query/logicalplan/validate.go
@@ -171,8 +171,6 @@ type Named interface {
 }
 
 func ValidateAggregationExpr(plan *LogicalPlan) *ExprValidationError {
-	aliases := map[string]struct{}{}
-
 	for _, expr := range plan.Aggregation.AggExprs {
 		// check that the aggregation expression has the required structure
 		colFinder := newTypeFinder((*Column)(nil))
@@ -212,16 +210,6 @@ func ValidateAggregationExpr(plan *LogicalPlan) *ExprValidationError {
 			}
 		}
 
-		if alias, ok := expr.(*AliasExpr); ok {
-			if _, found := aliases[alias.Alias]; found {
-				return &ExprValidationError{
-					message: fmt.Sprintf("alias used twice: %s", alias.Alias),
-					expr:    expr,
-				}
-			}
-			aliases[alias.Alias] = struct{}{}
-		}
-
 		// check that the column type can be aggregated by the function type
 		columnType := column.StorageLayout.Type()
 		aggFuncExpr := aggFuncFinder.result.(*AggregationFunction)
diff --git a/query/physicalplan/aggregate.go b/query/physicalplan/aggregate.go
index 67f3876..91ad247 100644
--- a/query/physicalplan/aggregate.go
+++ b/query/physicalplan/aggregate.go
@@ -30,40 +30,23 @@ func Aggregate(
 ) (PhysicalPlan, error) {
 	aggregations := make([]Aggregation, 0, len(agg.AggExprs))
 
+	// TODO(brancz): This is not correct, it doesn't handle aggregations
+	// correctly of previously projected columns like `sum(value + timestamp)`.
+	// Need to understand why we need to handle dynamic columns here
+	// differently and not just use the aggregation funciton's expression.
 	for _, expr := range agg.AggExprs {
-		var (
-			aggregation    Aggregation
-			aggFunc        logicalplan.AggFunc
-			aggFuncFound   bool
-			aggColumnFound bool
-		)
+		aggregation := Aggregation{}
 		expr.Accept(PreExprVisitorFunc(func(expr logicalplan.Expr) bool {
-			switch e := expr.(type) {
-			case *logicalplan.AggregationFunction:
-				aggFunc = e.Func
-				aggFuncFound = true
-			case *logicalplan.Column:
-				aggregation.expr = e
-				aggColumnFound = true
-			case *logicalplan.DynamicColumn:
-				aggregation.expr = e
+			if _, ok := expr.(*logicalplan.DynamicColumn); ok {
 				aggregation.dynamic = true
-				aggColumnFound = true
 			}
 
 			return true
 		}))
 
-		if !aggFuncFound {
-			return nil, errors.New("aggregation function not found")
-		}
-
-		if !aggColumnFound {
-			return nil, errors.New("aggregation column not found")
-		}
-
 		aggregation.resultName = expr.Name()
-		aggregation.function = aggFunc
+		aggregation.function = expr.Func
+		aggregation.expr = expr.Expr
 
 		aggregations = append(aggregations, aggregation)
 	}
@@ -247,7 +230,7 @@ func (a *HashAggregate) Draw() *Diagram {
 
 	var groupings []string
 	for _, grouping := range a.groupByColumnMatchers {
-		groupings = append(groupings, grouping.Name())
+		groupings = append(groupings, grouping.String())
 	}
 	details := fmt.Sprintf("HashAggregate (%s by %s)", strings.Join(names, ","), strings.Join(groupings, ","))
 	return &Diagram{Details: details, Child: child}
@@ -273,23 +256,6 @@ func (u *uint64HashCombine) hashCombine(rhs uint64) uint64 {
 	return hashCombine(u.value, rhs)
 }
 
-// durationHashCombine hashes a given timestamp by dividing it through a given duration.
-// timestamp | duration | hash
-// 0 		 | 2		| 0
-// 1 		 | 2		| 0
-// 2 		 | 2		| 1
-// 3 		 | 2		| 1
-// 4 		 | 2		| 2
-// 5 		 | 2		| 2
-// Essentially hashing timestamps into buckets of durations.
-type durationHashCombine struct {
-	milliseconds uint64
-}
-
-func (d *durationHashCombine) hashCombine(rhs uint64) uint64 {
-	return rhs / d.milliseconds // floors by default
-}
-
 func (a *HashAggregate) Callback(_ 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")
@@ -327,17 +293,9 @@ func (a *HashAggregate) Callback(_ context.Context, r arrow.Record) error {
 					continue
 				}
 
-				switch v := matcher.(type) {
-				case *logicalplan.DurationExpr:
-					duration := v.Value()
-					groupByFieldHashes = append(groupByFieldHashes,
-						&durationHashCombine{milliseconds: uint64(duration.Milliseconds())},
-					)
-				default:
-					groupByFieldHashes = append(groupByFieldHashes,
-						&uint64HashCombine{value: scalar.Hash(a.hashSeed, scalar.NewStringScalar(field.Name))},
-					)
-				}
+				groupByFieldHashes = append(groupByFieldHashes,
+					&uint64HashCombine{value: scalar.Hash(a.hashSeed, scalar.NewStringScalar(field.Name))},
+				)
 			}
 		}
 
@@ -398,11 +356,23 @@ func (a *HashAggregate) Callback(_ context.Context, r arrow.Record) error {
 		}
 	}
 
-	if concreteAggregateFieldsFound != aggregate.concreteAggregations ||
+	// It's ok for the same aggregation to be found multiple times, optimizers
+	// should remove them but for correctness in the case where they don't we
+	// need to handle it, so concrete aggregates are allowed to be different
+	// from concrete aggregations.
+	if ((concreteAggregateFieldsFound == 0 || aggregate.concreteAggregations == 0) && (len(aggregate.dynamicAggregations) == 0)) ||
 		(len(aggregate.dynamicAggregations) > 0) && dynamicAggregateFieldsFound == 0 {
 		// To perform an aggregation ALL concrete columns must have been matched
 		// or at least one dynamic column if performing dynamic aggregations.
-		return errors.New("aggregate field not found, aggregations are not possible without it")
+		exprs := make([]string, len(aggregate.aggregations))
+		for i, col := range aggregate.aggregations {
+			exprs[i] = col.expr.String()
+		}
+
+		if a.finalStage {
+			return fmt.Errorf("aggregate field(s) not found %#v, final aggregations are not possible without it (%d concrete aggregation fields found; %d concrete aggregations)", exprs, concreteAggregateFieldsFound, aggregate.concreteAggregations)
+		}
+		return fmt.Errorf("aggregate field(s) not found %#v, aggregations are not possible without it (%d concrete aggregation fields found; %d concrete aggregations)", exprs, concreteAggregateFieldsFound, aggregate.concreteAggregations)
 	}
 
 	numRows := int(r.NumRows())
diff --git a/query/physicalplan/filter.go b/query/physicalplan/filter.go
index 0107c61..838aae4 100644
--- a/query/physicalplan/filter.go
+++ b/query/physicalplan/filter.go
@@ -63,7 +63,7 @@ func (f PreExprVisitorFunc) PostVisit(_ logicalplan.Expr) bool {
 
 func binaryBooleanExpr(expr *logicalplan.BinaryExpr) (BooleanExpression, error) {
 	switch expr.Op {
-	case logicalplan.OpEq, logicalplan.OpNotEq, logicalplan.OpLt, logicalplan.OpLtEq, logicalplan.OpGt, logicalplan.OpGtEq, logicalplan.OpRegexMatch, logicalplan.OpRegexNotMatch:
+	case logicalplan.OpEq, logicalplan.OpNotEq, logicalplan.OpLt, logicalplan.OpLtEq, logicalplan.OpGt, logicalplan.OpGtEq, logicalplan.OpRegexMatch, logicalplan.OpRegexNotMatch, logicalplan.OpAdd, logicalplan.OpSub, logicalplan.OpMul, logicalplan.OpDiv:
 		var leftColumnRef *ArrayRef
 		expr.Left.Accept(PreExprVisitorFunc(func(expr logicalplan.Expr) bool {
 			switch e := expr.(type) {
@@ -119,12 +119,12 @@ func binaryBooleanExpr(expr *logicalplan.BinaryExpr) (BooleanExpression, error)
 	case logicalplan.OpAnd:
 		left, err := booleanExpr(expr.Left)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("left bool expr: %w", err)
 		}
 
 		right, err := booleanExpr(expr.Right)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("right bool expr: %w", err)
 		}
 
 		return &AndExpr{
@@ -134,12 +134,12 @@ func binaryBooleanExpr(expr *logicalplan.BinaryExpr) (BooleanExpression, error)
 	case logicalplan.OpOr:
 		left, err := booleanExpr(expr.Left)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("left bool expr: %w", err)
 		}
 
 		right, err := booleanExpr(expr.Right)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("right bool expr: %w", err)
 		}
 
 		return &OrExpr{
@@ -147,7 +147,7 @@ func binaryBooleanExpr(expr *logicalplan.BinaryExpr) (BooleanExpression, error)
 			Right: right,
 		}, nil
 	default:
-		panic("unsupported binary boolean expression")
+		return nil, fmt.Errorf("binary expr %s: %w", expr.Op.String(), ErrUnsupportedBooleanExpression)
 	}
 }
 
@@ -213,7 +213,7 @@ func booleanExpr(expr logicalplan.Expr) (BooleanExpression, error) {
 func Filter(pool memory.Allocator, tracer trace.Tracer, filterExpr logicalplan.Expr) (*PredicateFilter, error) {
 	expr, err := booleanExpr(filterExpr)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("create bool expr: %w", err)
 	}
 
 	return newFilter(pool, tracer, expr), nil
diff --git a/query/physicalplan/project.go b/query/physicalplan/project.go
index 18427f0..9b6064f 100644
--- a/query/physicalplan/project.go
+++ b/query/physicalplan/project.go
@@ -8,6 +8,7 @@ import (
 	"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/apache/arrow/go/v14/arrow/scalar"
 	"go.opentelemetry.io/otel/trace"
 
 	"github.com/polarsignals/frostdb/query/logicalplan"
@@ -16,11 +17,14 @@ import (
 type columnProjection interface {
 	// Name returns the name of the column that this projection will return.
 	Name() string
+	// String returns the string representation as it may have been used to create the projection.
+	String() string
 	// Projects one or more columns. Each element in the field list corresponds to an element in the list of arrays.
 	Project(mem memory.Allocator, ar arrow.Record) ([]arrow.Field, []arrow.Array, error)
 }
 
 type aliasProjection struct {
+	p    columnProjection
 	expr *logicalplan.AliasExpr
 	name string
 }
@@ -29,37 +33,26 @@ func (a aliasProjection) Name() string {
 	return a.name
 }
 
+func (a aliasProjection) String() string {
+	return a.expr.String()
+}
+
 func (a aliasProjection) Project(mem memory.Allocator, ar arrow.Record) ([]arrow.Field, []arrow.Array, error) {
-	switch e := a.expr.Expr.(type) {
-	case *logicalplan.BinaryExpr:
-		boolExpr, err := binaryBooleanExpr(e)
-		if err != nil {
-			return nil, nil, err
-		}
-		fields, array, err := boolExprProjection{boolExpr: boolExpr}.Project(mem, ar)
-		if err != nil {
-			return nil, nil, err
-		}
-		for i, field := range fields {
-			if a.expr.Expr.MatchColumn(field.Name) {
-				fields[i].Name = a.name
-			}
-		}
-		return fields, array, nil
-	case *logicalplan.Column:
-		for i := 0; i < ar.Schema().NumFields(); i++ {
-			field := ar.Schema().Field(i)
-			if a.expr.MatchColumn(field.Name) {
-				field.Name = a.name
-				ar.Column(i).Retain() // Retain the column since we're keeping it.
-				return []arrow.Field{field}, []arrow.Array{ar.Column(i)}, nil
-			}
-		}
-	default:
-		return nil, nil, fmt.Errorf("unsupported alias expression type: %T", e)
+	fields, arrays, err := a.p.Project(mem, ar)
+	if err != nil {
+		return nil, nil, err
 	}
 
-	return nil, nil, nil
+	if len(fields) == 0 || len(fields) != len(arrays) {
+		return nil, nil, fmt.Errorf("invalid projection for alias: fields and arrays must be non-empty and of equal length")
+	}
+
+	return []arrow.Field{{
+		Name:     a.name,
+		Type:     fields[0].Type,
+		Nullable: fields[0].Nullable,
+		Metadata: fields[0].Metadata,
+	}}, arrays, nil
 }
 
 type binaryExprProjection struct {
@@ -73,61 +66,85 @@ func (b binaryExprProjection) Name() string {
 	return b.expr.String()
 }
 
+func (b binaryExprProjection) String() string {
+	return b.expr.String()
+}
+
 func (b binaryExprProjection) Project(mem memory.Allocator, ar arrow.Record) ([]arrow.Field, []arrow.Array, error) {
 	leftFields, leftArrays, err := b.left.Project(mem, ar)
 	if err != nil {
 		return nil, nil, fmt.Errorf("project left side of binary expression: %w", err)
 	}
+	defer func() {
+		for _, arr := range leftArrays {
+			arr.Release()
+		}
+	}()
 
 	rightFields, rightArrays, err := b.right.Project(mem, ar)
 	if err != nil {
 		return nil, nil, fmt.Errorf("project right side of binary expression: %w", err)
 	}
+	defer func() {
+		for _, arr := range rightArrays {
+			arr.Release()
+		}
+	}()
+
+	if len(leftFields) != 1 || len(leftArrays) != 1 {
+		return nil, nil, fmt.Errorf("binary expression projection expected one field and one array for each side, got %d fields and %d arrays on left", len(leftFields), len(leftArrays))
+	}
 
-	if len(leftFields) != 1 || len(rightFields) != 1 || len(leftArrays) != 1 || len(rightArrays) != 1 {
-		return nil, nil, fmt.Errorf("binary expression projection expected one field and one array for each side")
+	if len(rightFields) != 1 || len(rightArrays) != 1 {
+		return nil, nil, fmt.Errorf("binary expression projection expected one field and one array for each side, got %d fields and %d arrays on right", len(rightFields), len(rightArrays))
 	}
 
 	leftArray := leftArrays[0]
 	rightArray := rightArrays[0]
 
+	resultFields := []arrow.Field{{
+		Name:     b.expr.Name(),
+		Type:     leftFields[0].Type,
+		Nullable: leftFields[0].Nullable,
+		Metadata: leftFields[0].Metadata,
+	}}
 	switch leftArray := leftArray.(type) {
 	case *array.Int64:
 		switch b.expr.Op {
 		case logicalplan.OpAdd:
-			return []arrow.Field{leftFields[0]}, []arrow.Array{AddInt64s(mem, leftArray, rightArray.(*array.Int64))}, nil
+			return resultFields, []arrow.Array{AddInt64s(mem, leftArray, rightArray.(*array.Int64))}, nil
 		case logicalplan.OpSub:
-			return []arrow.Field{leftFields[0]}, []arrow.Array{SubInt64s(mem, leftArray, rightArray.(*array.Int64))}, nil
+			return resultFields, []arrow.Array{SubInt64s(mem, leftArray, rightArray.(*array.Int64))}, nil
 		case logicalplan.OpMul:
-			return []arrow.Field{leftFields[0]}, []arrow.Array{MulInt64s(mem, leftArray, rightArray.(*array.Int64))}, nil
+			return resultFields, []arrow.Array{MulInt64s(mem, leftArray, rightArray.(*array.Int64))}, nil
 		case logicalplan.OpDiv:
-			return []arrow.Field{leftFields[0]}, []arrow.Array{DivInt64s(mem, leftArray, rightArray.(*array.Int64))}, nil
+			return resultFields, []arrow.Array{DivInt64s(mem, leftArray, rightArray.(*array.Int64))}, nil
 		default:
 			return nil, nil, fmt.Errorf("unsupported binary expression: %s", b.expr.String())
 		}
 	case *array.Float64:
 		switch b.expr.Op {
 		case logicalplan.OpAdd:
-			return []arrow.Field{leftFields[0]}, []arrow.Array{AddFloat64s(mem, leftArray, rightArray.(*array.Float64))}, nil
+			return resultFields, []arrow.Array{AddFloat64s(mem, leftArray, rightArray.(*array.Float64))}, nil
 		case logicalplan.OpSub:
-			return []arrow.Field{leftFields[0]}, []arrow.Array{SubFloat64s(mem, leftArray, rightArray.(*array.Float64))}, nil
+			return resultFields, []arrow.Array{SubFloat64s(mem, leftArray, rightArray.(*array.Float64))}, nil
 		case logicalplan.OpMul:
-			return []arrow.Field{leftFields[0]}, []arrow.Array{MulFloat64s(mem, leftArray, rightArray.(*array.Float64))}, nil
+			return resultFields, []arrow.Array{MulFloat64s(mem, leftArray, rightArray.(*array.Float64))}, nil
 		case logicalplan.OpDiv:
-			return []arrow.Field{leftFields[0]}, []arrow.Array{DivFloat64s(mem, leftArray, rightArray.(*array.Float64))}, nil
+			return resultFields, []arrow.Array{DivFloat64s(mem, leftArray, rightArray.(*array.Float64))}, nil
 		default:
 			return nil, nil, fmt.Errorf("unsupported binary expression: %s", b.expr.String())
 		}
 	case *array.Int32:
 		switch b.expr.Op {
 		case logicalplan.OpAdd:
-			return []arrow.Field{leftFields[0]}, []arrow.Array{AddInt32s(mem, leftArray, rightArray.(*array.Int32))}, nil
+			return resultFields, []arrow.Array{AddInt32s(mem, leftArray, rightArray.(*array.Int32))}, nil
 		case logicalplan.OpSub:
-			return []arrow.Field{leftFields[0]}, []arrow.Array{SubInt32s(mem, leftArray, rightArray.(*array.Int32))}, nil
+			return resultFields, []arrow.Array{SubInt32s(mem, leftArray, rightArray.(*array.Int32))}, nil
 		case logicalplan.OpMul:
-			return []arrow.Field{leftFields[0]}, []arrow.Array{MulInt32s(mem, leftArray, rightArray.(*array.Int32))}, nil
+			return resultFields, []arrow.Array{MulInt32s(mem, leftArray, rightArray.(*array.Int32))}, nil
 		case logicalplan.OpDiv:
-			return []arrow.Field{leftFields[0]}, []arrow.Array{DivInt32s(mem, leftArray, rightArray.(*array.Int32))}, nil
+			return resultFields, []arrow.Array{DivInt32s(mem, leftArray, rightArray.(*array.Int32))}, nil
 		default:
 			return nil, nil, fmt.Errorf("unsupported binary expression: %s", b.expr.String())
 		}
@@ -182,7 +199,12 @@ func DivInt64s(mem memory.Allocator, left, right *array.Int64) *array.Int64 {
 	res.Resize(left.Len())
 
 	for i := 0; i < left.Len(); i++ {
-		res.Append(left.Value(i) / right.Value(i))
+		rightValue := right.Value(i)
+		if right.Value(i) == 0 {
+			res.AppendNull()
+		} else {
+			res.Append(left.Value(i) / rightValue)
+		}
 	}
 
 	return res.NewInt64Array()
@@ -234,7 +256,12 @@ func DivFloat64s(mem memory.Allocator, left, right *array.Float64) *array.Float6
 	res.Resize(left.Len())
 
 	for i := 0; i < left.Len(); i++ {
-		res.Append(left.Value(i) / right.Value(i))
+		rightValue := right.Value(i)
+		if right.Value(i) == 0 {
+			res.AppendNull()
+		} else {
+			res.Append(left.Value(i) / rightValue)
+		}
 	}
 
 	return res.NewFloat64Array()
@@ -286,7 +313,12 @@ func DivInt32s(mem memory.Allocator, left, right *array.Int32) *array.Int32 {
 	res.Resize(left.Len())
 
 	for i := 0; i < left.Len(); i++ {
-		res.Append(left.Value(i) / right.Value(i))
+		rightValue := right.Value(i)
+		if right.Value(i) == 0 {
+			res.AppendNull()
+		} else {
+			res.Append(left.Value(i) / rightValue)
+		}
 	}
 
 	return res.NewInt32Array()
@@ -300,6 +332,10 @@ func (b boolExprProjection) Name() string {
 	return b.boolExpr.String()
 }
 
+func (b boolExprProjection) String() string {
+	return b.boolExpr.String()
+}
+
 func (b boolExprProjection) Project(mem memory.Allocator, ar arrow.Record) ([]arrow.Field, []arrow.Array, error) {
 	var partiallyComputedExprRes arrow.Array
 	if ar.Schema().HasField(b.boolExpr.String()) {
@@ -368,6 +404,10 @@ func (p plainProjection) Name() string {
 	return p.expr.ColumnName
 }
 
+func (p plainProjection) String() string {
+	return p.expr.ColumnName
+}
+
 func (p plainProjection) Project(_ memory.Allocator, ar arrow.Record) ([]arrow.Field, []arrow.Array, error) {
 	for i := 0; i < ar.Schema().NumFields(); i++ {
 		field := ar.Schema().Field(i)
@@ -380,6 +420,97 @@ func (p plainProjection) Project(_ memory.Allocator, ar arrow.Record) ([]arrow.F
 	return nil, nil, nil
 }
 
+type convertProjection struct {
+	p    columnProjection
+	expr *logicalplan.ConvertExpr
+}
+
+func (p convertProjection) Name() string {
+	return p.expr.Name()
+}
+
+func (p convertProjection) String() string {
+	return p.expr.Name()
+}
+
+func (p convertProjection) convert(mem memory.Allocator, c arrow.Array) (arrow.Array, error) {
+	defer c.Release()
+
+	switch c := c.(type) {
+	case *array.Int64:
+		switch p.expr.Type {
+		case arrow.PrimitiveTypes.Float64:
+			return convertInt64ToFloat64(mem, c), nil
+		default:
+			return nil, fmt.Errorf("unsupported conversion from %s to %s", c.DataType(), p.expr.Type)
+		}
+	default:
+		return nil, fmt.Errorf("unsupported conversion from %s to %s", c.DataType(), p.expr.Type)
+	}
+}
+
+func convertInt64ToFloat64(mem memory.Allocator, c *array.Int64) *array.Float64 {
+	res := array.NewFloat64Builder(mem)
+	defer res.Release()
+
+	res.Resize(c.Len())
+
+	for i := 0; i < c.Len(); i++ {
+		res.Append(float64(c.Value(i)))
+	}
+
+	return res.NewFloat64Array()
+}
+
+func (p convertProjection) Project(mem memory.Allocator, ar arrow.Record) ([]arrow.Field, []arrow.Array, error) {
+	fields, cols, err := p.p.Project(mem, ar)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	if len(fields) == 0 || len(fields) != len(cols) {
+		return nil, nil, fmt.Errorf("invalid projection for convert: fields and arrays must be non-empty and of equal length")
+	}
+
+	c, err := p.convert(mem, cols[0])
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return []arrow.Field{{
+		Name:     p.expr.Name(),
+		Type:     p.expr.Type,
+		Nullable: fields[0].Nullable,
+		Metadata: fields[0].Metadata,
+	}}, []arrow.Array{c}, nil
+}
+
+type literalProjection struct {
+	value scalar.Scalar
+}
+
+func (p literalProjection) Name() string {
+	return p.value.String()
+}
+
+func (p literalProjection) String() string {
+	return p.value.String()
+}
+
+func (p literalProjection) Project(mem memory.Allocator, ar arrow.Record) ([]arrow.Field, []arrow.Array, error) {
+	arr, err := scalar.MakeArrayFromScalar(p.value, int(ar.NumRows()), mem)
+	if err != nil {
+		return nil, nil, fmt.Errorf("make array from literal value: %w", err)
+	}
+
+	return []arrow.Field{{
+			Name: p.value.String(),
+			Type: p.value.DataType(),
+		}}, []arrow.Array{
+			arr,
+		}, nil
+}
+
 type dynamicProjection struct {
 	expr *logicalplan.DynamicColumn
 }
@@ -388,6 +519,10 @@ func (p dynamicProjection) Name() string {
 	return p.expr.ColumnName
 }
 
+func (p dynamicProjection) String() string {
+	return p.expr.ColumnName
+}
+
 func (p dynamicProjection) Project(_ memory.Allocator, ar arrow.Record) ([]arrow.Field, []arrow.Array, error) {
 	fields := []arrow.Field{}
 	arrays := []arrow.Array{}
@@ -411,21 +546,45 @@ func projectionFromExpr(expr logicalplan.Expr) (columnProjection, error) {
 		return plainProjection{
 			expr: e,
 		}, nil
+	case *logicalplan.ConvertExpr:
+		p, err := projectionFromExpr(e.Expr)
+		if err != nil {
+			return nil, fmt.Errorf("projection to convert: %w", err)
+		}
+
+		return convertProjection{
+			p:    p,
+			expr: e,
+		}, nil
+	case *logicalplan.AggregationFunction:
+		return plainProjection{
+			expr: logicalplan.Col(e.Name()),
+		}, nil
 	case *logicalplan.DynamicColumn:
 		return dynamicProjection{
 			expr: e,
 		}, nil
+	case *logicalplan.LiteralExpr:
+		return literalProjection{
+			value: e.Value,
+		}, nil
 	case *logicalplan.AliasExpr:
+		p, err := projectionFromExpr(e.Expr)
+		if err != nil {
+			return nil, fmt.Errorf("projection to convert: %w", err)
+		}
+
 		return aliasProjection{
+			p:    p,
 			expr: e,
-			name: e.Name(),
+			name: e.Alias,
 		}, nil
 	case *logicalplan.BinaryExpr:
 		switch e.Op {
 		case logicalplan.OpEq, logicalplan.OpNotEq, logicalplan.OpGt, logicalplan.OpGtEq, logicalplan.OpLt, logicalplan.OpLtEq, logicalplan.OpRegexMatch, logicalplan.OpRegexNotMatch, logicalplan.OpAnd, logicalplan.OpOr:
 			boolExpr, err := binaryBooleanExpr(e)
 			if err != nil {
-				return nil, err
+				return nil, fmt.Errorf("boolean projection from expr: %w", err)
 			}
 			return boolExprProjection{
 				boolExpr: boolExpr,
@@ -433,12 +592,12 @@ func projectionFromExpr(expr logicalplan.Expr) (columnProjection, error) {
 		case logicalplan.OpAdd, logicalplan.OpSub, logicalplan.OpMul, logicalplan.OpDiv:
 			left, err := projectionFromExpr(e.Left)
 			if err != nil {
-				return nil, err
+				return nil, fmt.Errorf("left projection for arithmetic projection: %w", err)
 			}
 
 			right, err := projectionFromExpr(e.Right)
 			if err != nil {
-				return nil, err
+				return nil, fmt.Errorf("right projection for arithmetic projection: %w", err)
 			}
 
 			return binaryExprProjection{
@@ -450,8 +609,6 @@ func projectionFromExpr(expr logicalplan.Expr) (columnProjection, error) {
 		default:
 			return nil, fmt.Errorf("unknown binary expression: %s", e.String())
 		}
-	case *logicalplan.AverageExpr:
-		return &averageProjection{expr: e}, nil
 	default:
 		return nil, fmt.Errorf("unsupported expression type for projection: %T", expr)
 	}
@@ -542,111 +699,18 @@ func (p *Projection) Draw() *Diagram {
 
 	var columns []string
 	for _, p := range p.colProjections {
-		columns = append(columns, p.Name())
+		columns = append(columns, p.String())
 	}
-	details := fmt.Sprintf("Projection (%s)", strings.Join(columns, ","))
+	details := fmt.Sprintf("Projection (%s)", strings.Join(columns, ", "))
 	return &Diagram{Details: details, Child: child}
 }
 
-type averageProjection struct {
-	expr logicalplan.Expr
-}
-
-func (a *averageProjection) Name() string {
-	return a.expr.Name()
-}
-
-func (a *averageProjection) Project(mem memory.Allocator, r arrow.Record) ([]arrow.Field, []arrow.Array, error) {
-	columnName := a.expr.Name()
-	resultName := "avg(" + columnName + ")"
-	if avgExpr, ok := a.expr.(*logicalplan.AverageExpr); ok {
-		if ae, ok := avgExpr.Expr.(*logicalplan.AliasExpr); ok {
-			columnName = ae.Expr.Name()
-			resultName = ae.Alias
-		}
-	}
-
-	columnSum := "sum(" + columnName + ")"
-	columnCount := "count(" + columnName + ")"
-
-	schema := r.Schema()
-
-	sumIndex := schema.FieldIndices(columnSum)
-	if len(sumIndex) != 1 {
-		return nil, nil, fmt.Errorf("sum column for average projection for column %s not found", columnName)
-	}
-	countIndex := schema.FieldIndices(columnCount)
-	if len(countIndex) != 1 {
-		return nil, nil, fmt.Errorf("count column for average projection for column %s not found", columnName)
-	}
-
-	sums := r.Column(sumIndex[0])
-	counts := r.Column(countIndex[0])
-
-	fields := make([]arrow.Field, 0, schema.NumFields()-1)
-	columns := make([]arrow.Array, 0, schema.NumFields()-1)
-
-	// Only add the fields and columns that aren't the average's underlying sum and count columns.
-	for i := 0; i < schema.NumFields(); i++ {
-		field := schema.Field(i)
-		if i != sumIndex[0] && i != countIndex[0] {
-			fields = append(fields, field)
-			columns = append(columns, r.Column(i))
-			r.Column(i).Retain() // Retain the column since we're keeping it.
-		}
-	}
-
-	// Add the field and column for the projected average aggregation.
-	switch sums.DataType().ID() {
-	case arrow.INT64:
-		fields = append(fields, arrow.Field{
-			Name: resultName,
-			Type: &arrow.Int64Type{},
-		})
-		columns = append(columns, avgInt64arrays(mem, sums, counts))
-	case arrow.FLOAT64:
-		fields = append(fields, arrow.Field{
-			Name: resultName,
-			Type: &arrow.Float64Type{},
-		})
-		columns = append(columns, avgFloat64arrays(mem, sums, counts))
-	default:
-		return nil, nil, fmt.Errorf("Datatype %s is not supported for average projection", sums.DataType().ID())
-	}
-
-	return fields, columns, nil
-}
-
-func avgInt64arrays(pool memory.Allocator, sums, counts arrow.Array) arrow.Array {
-	sumsInts := sums.(*array.Int64)
-	countsInts := counts.(*array.Int64)
-
-	res := array.NewInt64Builder(pool)
-	defer res.Release()
-	for i := 0; i < sumsInts.Len(); i++ {
-		res.Append(sumsInts.Value(i) / countsInts.Value(i))
-	}
-
-	return res.NewArray()
-}
-
-func avgFloat64arrays(pool memory.Allocator, sums, counts arrow.Array) arrow.Array {
-	sumsFloats := sums.(*array.Float64)
-	countsInts := counts.(*array.Int64)
-
-	res := array.NewFloat64Builder(pool)
-	defer res.Release()
-	for i := 0; i < sumsFloats.Len(); i++ {
-		res.Append(sumsFloats.Value(i) / float64(countsInts.Value(i)))
-	}
-
-	return res.NewArray()
-}
-
 type allProjection struct{}
 
 func (a allProjection) Name() string { return "all" }
 
+func (a allProjection) String() string { return "*" }
+
 func (a allProjection) Project(_ memory.Allocator, ar arrow.Record) ([]arrow.Field, []arrow.Array, error) {
 	return ar.Schema().Fields(), ar.Columns(), nil
 }
diff --git a/snapshot.go b/snapshot.go
index b697cfe..ad5a565 100644
--- a/snapshot.go
+++ b/snapshot.go
@@ -774,7 +774,7 @@ func restoreIndexFilesFromSnapshot(db *DB, table, snapshotDir, blockID string) e
 
 		// Hard link the file back into the index directory.
 		if err := os.Link(path, filepath.Join(db.indexDir(), table, blockID, lvl, filename)); err != nil {
-			return err
+			return fmt.Errorf("hard link file: %w", err)
 		}
 
 		return nil
diff --git a/sqlparse/parser.go b/sqlparse/parser.go
index a0ecd20..5e9af31 100644
--- a/sqlparse/parser.go
+++ b/sqlparse/parser.go
@@ -27,7 +27,11 @@ type ParseResult struct {
 // queryEngine.ScanTable is provided as a starting point and no table needs to
 // be specified in the SQL statement. Additionally, the idea is to change to
 // creating logical plans directly (rather than through a builder).
-func (p *Parser) ExperimentalParse(builder query.Builder, dynColNames []string, sql string) (ParseResult, error) {
+func (p *Parser) ExperimentalParse(
+	builder query.Builder,
+	dynColNames []string,
+	sql string,
+) (ParseResult, error) {
 	asts, _, err := p.p.Parse(sql, "", "")
 	if err != nil {
 		return ParseResult{}, err
diff --git a/sqlparse/visitor.go b/sqlparse/visitor.go
index 52fc687..2f5d638 100644
--- a/sqlparse/visitor.go
+++ b/sqlparse/visitor.go
@@ -5,8 +5,10 @@ import (
 	"strings"
 	"time"
 
+	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/apache/arrow/go/v14/arrow/scalar"
 	"github.com/pingcap/tidb/parser/ast"
+	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/parser/opcode"
 	"github.com/pingcap/tidb/parser/test_driver"
 
@@ -25,7 +27,10 @@ type astVisitor struct {
 
 var _ ast.Visitor = &astVisitor{}
 
-func newASTVisitor(builder query.Builder, dynColNames []string) *astVisitor {
+func newASTVisitor(
+	builder query.Builder,
+	dynColNames []string,
+) *astVisitor {
 	dynMap := make(map[string]struct{})
 	for _, n := range dynColNames {
 		dynMap[n] = struct{}{}
@@ -51,20 +56,86 @@ func (v *astVisitor) Enter(n ast.Node) (nRes ast.Node, skipChildren bool) {
 		expr.Fields.Accept(v)
 		switch {
 		case expr.GroupBy != nil:
+			// This represents everything before the "group by" clause.
+			beforeGroupBy := v.exprStack
 			expr.GroupBy.Accept(v)
-			var agg []logicalplan.Expr
-			var groups []logicalplan.Expr
-
-			for _, expr := range v.exprStack {
-				switch expr.(type) {
-				case *logicalplan.AliasExpr, *logicalplan.AggregationFunction:
-					agg = append(agg, expr)
-				default:
-					groups = append(groups, expr)
+			// This represents everything after the "group by" clause.
+			afterGroupBy := v.exprStack[len(beforeGroupBy):]
+			groups := afterGroupBy
+
+			includedPreprojections := make(map[string]struct{})
+			preProjections := []logicalplan.Expr{}
+			postProjections := []logicalplan.Expr{}
+			aggregations := []*logicalplan.AggregationFunction{}
+
+			for _, expr := range beforeGroupBy {
+				// Walk the expression tree and separate out what projections
+				// need to be done before aggregations, and which can be done
+				// after. We don't support nested aggregations, so we can just
+				// find aggregation functions and put their expressions into
+				// the pre-projections. If we wanted to support nested
+				// aggregations, this would need to be more sophisticated.
+				aggCollector := &aggregationCollector{}
+				expr.Accept(aggCollector)
+
+				// If we have any aggregations nested in the AST like
+				// "sum(value) as value_sum_or_anything_else" then the actual
+				// query plan nesting looks something like:
+				// alias(value_sum_or_anything_else, sum(value))
+
+				if len(aggCollector.aggregations) > 0 {
+					// This is the expression that will be aggregated so we
+					// need to ensure that it is in the pre-projection.
+					for _, agg := range aggCollector.aggregations {
+						if _, ok := includedPreprojections[agg.Expr.Name()]; !ok {
+							preProjections = append(preProjections, agg.Expr)
+							// The same expression can be used in multiple
+							// aggregations, but we only need to project it
+							// once.
+							includedPreprojections[agg.Expr.Name()] = struct{}{}
+						}
+
+						aggregations = append(aggregations, agg)
+					}
+					postProjections = append(postProjections, expr)
+				} else {
+					preProjections = append(preProjections, expr)
+					if _, ok := expr.(*logicalplan.DynamicColumn); ok {
+						postProjections = append(postProjections, expr)
+					} else {
+						postProjections = append(postProjections, logicalplan.Col(expr.Name()))
+					}
+				}
+			}
+
+			// We need to ensure that anything we group by is in the pre-projection.
+			for _, expr := range afterGroupBy {
+				found := false
+				for _, preExpr := range preProjections {
+					if expr.Name() == preExpr.Name() {
+						found = true
+						break
+					}
+				}
+				if !found {
+					preProjections = append(preProjections, expr)
 				}
 			}
-			v.builder = v.builder.Aggregate(agg, groups)
+
+			// Insert a projection for any groups that need to be computed
+			// before they can be used for an aggregation, for example:
+			//
+			// SELECT sum(value), (timestamp/1000) as timestamp_bucket group by timestamp_bucket
+			v.builder = v.builder.Project(preProjections...)
+			v.builder = v.builder.Aggregate(aggregations, groups)
+
+			// Insert a projection for any groups that need to be computed
+			// before they can be used for an aggregation, for example:
+			//
+			// SELECT sum(value)/count(value) value_avg, (timestamp/1000) as timestamp_bucket group by timestamp_bucket
+			v.builder = v.builder.Project(postProjections...)
 		case expr.Distinct:
+			v.builder = v.builder.Project(v.exprStack...)
 			v.builder = v.builder.Distinct(v.exprStack...)
 		default:
 			v.builder = v.builder.Project(v.exprStack...)
@@ -74,6 +145,25 @@ func (v *astVisitor) Enter(n ast.Node) (nRes ast.Node, skipChildren bool) {
 	return n, false
 }
 
+type aggregationCollector struct {
+	aggregations []*logicalplan.AggregationFunction
+}
+
+func (a *aggregationCollector) PreVisit(_ logicalplan.Expr) bool {
+	return true
+}
+
+func (a *aggregationCollector) Visit(e logicalplan.Expr) bool {
+	if agg, ok := e.(*logicalplan.AggregationFunction); ok {
+		a.aggregations = append(a.aggregations, agg)
+	}
+	return true
+}
+
+func (a *aggregationCollector) PostVisit(_ logicalplan.Expr) bool {
+	return true
+}
+
 func (v *astVisitor) Leave(n ast.Node) (nRes ast.Node, ok bool) {
 	if err := v.leaveImpl(n); err != nil {
 		v.err = err
@@ -144,8 +234,9 @@ func (v *astVisitor) leaveImpl(n ast.Node) error {
 			v.exprStack = append(v.exprStack, logicalplan.Or(leftExpr, rightExpr))
 			return nil
 		}
+
 		v.exprStack = append(v.exprStack, &logicalplan.BinaryExpr{
-			Left:  logicalplan.Col(leftExpr.Name()),
+			Left:  leftExpr,
 			Op:    frostDBOp,
 			Right: rightExpr,
 		})
@@ -170,7 +261,16 @@ func (v *astVisitor) leaveImpl(n ast.Node) error {
 	case *ast.SelectField:
 		if as := expr.AsName.String(); as != "" {
 			lastExpr := len(v.exprStack) - 1
-			v.exprStack[lastExpr] = v.exprStack[lastExpr].(*logicalplan.AggregationFunction).Alias(as) // TODO should probably just be an alias expr and not from an aggregate function
+			switch e := v.exprStack[lastExpr].(type) {
+			case *logicalplan.AggregationFunction:
+				v.exprStack[lastExpr] = e.Alias(as)
+			case *logicalplan.BinaryExpr:
+				v.exprStack[lastExpr] = e.Alias(as)
+			case *logicalplan.Column:
+				v.exprStack[lastExpr] = e.Alias(as)
+			default:
+				return fmt.Errorf("unhandled select field %s", as)
+			}
 		}
 	case *ast.PatternRegexpExpr:
 		rightExpr, newExprs := pop(v.exprStack)
@@ -186,7 +286,8 @@ func (v *astVisitor) leaveImpl(n ast.Node) error {
 			e.Op = logicalplan.OpRegexNotMatch
 		}
 		v.exprStack = append(v.exprStack, e)
-	case *ast.FieldList, *ast.ColumnNameExpr, *ast.GroupByClause, *ast.ByItem, *ast.RowExpr,
+	case *ast.GroupByClause:
+	case *ast.FieldList, *ast.ColumnNameExpr, *ast.ByItem, *ast.RowExpr,
 		*ast.ParenthesesExpr:
 		// Deliberate pass-through nodes.
 	case *ast.FuncCallExpr:
@@ -206,6 +307,17 @@ func (v *astVisitor) leaveImpl(n ast.Node) error {
 		default:
 			return fmt.Errorf("unhandled func call: %s", expr.FnName.String())
 		}
+	case *ast.FuncCastExpr:
+		var t arrow.DataType
+		switch expr.Tp.GetType() {
+		case mysql.TypeFloat:
+			t = arrow.PrimitiveTypes.Float64
+		default:
+			return fmt.Errorf("unhandled cast type: %s", expr.Tp)
+		}
+		e, newExprs := pop(v.exprStack)
+		v.exprStack = newExprs
+		v.exprStack = append(v.exprStack, logicalplan.Convert(e, t))
 	default:
 		return fmt.Errorf("unhandled ast node %T", expr)
 	}
diff --git a/table.go b/table.go
index c972bab..a34add1 100644
--- a/table.go
+++ b/table.go
@@ -818,18 +818,18 @@ func (t *Table) Iterator(
 						return callback(ctx, r)
 					}
 
-					switch t := rg.(type) {
+					switch rg := rg.(type) {
 					case arrow.Record:
-						defer t.Release()
-						r := pqarrow.Project(t, iterOpts.PhysicalProjection)
+						defer rg.Release()
+						r := pqarrow.Project(rg, iterOpts.PhysicalProjection)
 						defer r.Release()
 						err := callback(ctx, r)
 						if err != nil {
 							return err
 						}
 					case index.ReleaseableRowGroup:
-						defer t.Release()
-						if err := converter.Convert(ctx, t); err != nil {
+						defer rg.Release()
+						if err := converter.Convert(ctx, rg, t.schema); err != nil {
 							return fmt.Errorf("failed to convert row group to arrow record: %v", err)
 						}
 						// This RowGroup had no relevant data. Ignore it.
@@ -848,7 +848,7 @@ func (t *Table) Iterator(
 							}
 						}
 					case dynparquet.DynamicRowGroup:
-						if err := converter.Convert(ctx, t); err != nil {
+						if err := converter.Convert(ctx, rg, t.schema); err != nil {
 							return fmt.Errorf("failed to convert row group to arrow record: %v", err)
 						}
 						// This RowGroup had no relevant data. Ignore it.
@@ -1300,7 +1300,7 @@ func (t *Table) compactParts(w io.Writer, compact []parts.Part, options ...parqu
 		}
 	}
 
-	bufs, err := t.buffersForCompaction(w, compact, options...)
+	bufs, err := t.buffersForCompaction(w, compact)
 	if err != nil {
 		return 0, err
 	}
@@ -1362,7 +1362,7 @@ func (t *Table) compactParts(w io.Writer, compact []parts.Part, options ...parqu
 // the minimum slice of dynamic row groups to be merged together for compaction.
 // If nil, nil is returned, the resulting serialized buffer is written directly
 // to w as an optimization.
-func (t *Table) buffersForCompaction(w io.Writer, inputParts []parts.Part, options ...parquet.WriterOption) ([]dynparquet.DynamicRowGroup, error) {
+func (t *Table) buffersForCompaction(w io.Writer, inputParts []parts.Part) ([]dynparquet.DynamicRowGroup, error) {
 	nonOverlappingParts, overlappingParts, err := parts.FindMaximumNonOverlappingSet(t.schema, inputParts)
 	if err != nil {
 		return nil, err
@@ -1424,7 +1424,7 @@ func (t *Table) buffersForCompaction(w io.Writer, inputParts []parts.Part, optio
 		records = append(records, p.Record())
 	}
 
-	if err := t.writeRecordsToParquet(w, records, false, options...); err != nil {
+	if err := t.writeRecordsToParquet(w, records, false); err != nil {
 		return nil, err
 	}
 


Test patch:
diff --git a/aggregate_test.go b/aggregate_test.go
index c2f8f01..9716751 100644
--- a/aggregate_test.go
+++ b/aggregate_test.go
@@ -117,19 +117,23 @@ func TestAggregateInconsistentSchema(t *testing.T) {
 			var res arrow.Record
 			err = engine.ScanTable("test").
 				Aggregate(
-					[]logicalplan.Expr{testCase.fn(logicalplan.Col("value")).Alias(testCase.alias)},
+					[]*logicalplan.AggregationFunction{
+						testCase.fn(logicalplan.Col("value")),
+					},
 					[]logicalplan.Expr{logicalplan.Col("labels.label2")},
-				).Execute(context.Background(), func(ctx context.Context, r arrow.Record) error {
-				r.Retain()
-				res = r
-				return nil
-			})
+				).
+				Project(testCase.fn(logicalplan.Col("value")).Alias(testCase.alias)).
+				Execute(context.Background(), func(ctx context.Context, r arrow.Record) error {
+					r.Retain()
+					res = r
+					return nil
+				})
 			require.NoError(t, err)
 			require.NotNil(t, res)
 			defer res.Release()
 
 			cols := res.Columns()
-			require.Equal(t, 2, len(cols))
+			require.Equal(t, 1, len(cols))
 			for i, col := range cols {
 				require.Equal(t, 2, col.Len(), "unexpected number of values in column %s", res.Schema().Field(i).Name)
 			}
@@ -212,8 +216,13 @@ func TestAggregationProjection(t *testing.T) {
 
 	records := []arrow.Record{}
 	err = engine.ScanTable("test").
+		Project(
+			logicalplan.DynCol("labels"),
+			logicalplan.Col("timestamp"),
+			logicalplan.Col("value"),
+		).
 		Aggregate(
-			[]logicalplan.Expr{
+			[]*logicalplan.AggregationFunction{
 				logicalplan.Sum(logicalplan.Col("value")),
 				logicalplan.Max(logicalplan.Col("value")),
 			},
@@ -307,7 +316,7 @@ func TestDurationAggregation(t *testing.T) {
 
 	_ = engine.ScanTable("test").
 		Aggregate(
-			[]logicalplan.Expr{
+			[]*logicalplan.AggregationFunction{
 				logicalplan.Sum(logicalplan.Col("value")),
 			},
 			[]logicalplan.Expr{
@@ -384,7 +393,7 @@ func BenchmarkAggregation(b *testing.B) {
 		name: "sum",
 		builder: engine.ScanTable("test").
 			Aggregate(
-				[]logicalplan.Expr{
+				[]*logicalplan.AggregationFunction{
 					logicalplan.Sum(logicalplan.Col("value")),
 				},
 				[]logicalplan.Expr{
@@ -395,7 +404,7 @@ func BenchmarkAggregation(b *testing.B) {
 		name: "count",
 		builder: engine.ScanTable("test").
 			Aggregate(
-				[]logicalplan.Expr{
+				[]*logicalplan.AggregationFunction{
 					logicalplan.Count(logicalplan.Col("value")),
 				},
 				[]logicalplan.Expr{
@@ -406,7 +415,7 @@ func BenchmarkAggregation(b *testing.B) {
 		name: "max",
 		builder: engine.ScanTable("test").
 			Aggregate(
-				[]logicalplan.Expr{
+				[]*logicalplan.AggregationFunction{
 					logicalplan.Max(logicalplan.Col("value")),
 				},
 				[]logicalplan.Expr{
@@ -498,7 +507,7 @@ func Test_Aggregation_DynCol(t *testing.T) {
 
 	err = engine.ScanTable("test").
 		Aggregate(
-			[]logicalplan.Expr{logicalplan.Max(logicalplan.DynCol("foo"))},
+			[]*logicalplan.AggregationFunction{logicalplan.Max(logicalplan.DynCol("foo"))},
 			nil,
 		).
 		Execute(context.Background(), func(ctx context.Context, ar arrow.Record) error {
diff --git a/bench_test.go b/bench_test.go
index c2082fd..81b3934 100644
--- a/bench_test.go
+++ b/bench_test.go
@@ -74,7 +74,7 @@ func getLatest15MinInterval(ctx context.Context, b testing.TB, engine *query.Loc
 	var result arrow.Record
 	require.NoError(b, engine.ScanTable(tableName).
 		Aggregate(
-			[]logicalplan.Expr{
+			[]*logicalplan.AggregationFunction{
 				logicalplan.Max(logicalplan.Col("timestamp")),
 			},
 			nil,
@@ -294,7 +294,7 @@ func BenchmarkQuery(b *testing.B) {
 					logicalplan.And(fullFilter...),
 				).
 				Aggregate(
-					[]logicalplan.Expr{
+					[]*logicalplan.AggregationFunction{
 						logicalplan.Sum(logicalplan.Col("value")),
 					},
 					[]logicalplan.Expr{
@@ -320,7 +320,7 @@ func BenchmarkQuery(b *testing.B) {
 					logicalplan.And(fullFilter...),
 				).
 				Aggregate(
-					[]logicalplan.Expr{
+					[]*logicalplan.AggregationFunction{
 						logicalplan.Sum(logicalplan.Col("value")),
 					},
 					[]logicalplan.Expr{
diff --git a/db_test.go b/db_test.go
index 4974b79..a334488 100644
--- a/db_test.go
+++ b/db_test.go
@@ -254,7 +254,7 @@ func TestDBWithWAL(t *testing.T) {
 	engine := query.NewEngine(pool, db.TableProvider())
 	err = engine.ScanTable("test").
 		Aggregate(
-			[]logicalplan.Expr{logicalplan.Sum(logicalplan.Col("value"))},
+			[]*logicalplan.AggregationFunction{logicalplan.Sum(logicalplan.Col("value"))},
 			[]logicalplan.Expr{logicalplan.Col("labels.label2")},
 		).
 		Execute(context.Background(), func(ctx context.Context, r arrow.Record) error {
@@ -1364,7 +1364,7 @@ func Test_DB_ReadOnlyQuery(t *testing.T) {
 	engine := query.NewEngine(pool, db.TableProvider())
 	err = engine.ScanTable("test").
 		Aggregate(
-			[]logicalplan.Expr{logicalplan.Sum(logicalplan.Col("value"))},
+			[]*logicalplan.AggregationFunction{logicalplan.Sum(logicalplan.Col("value"))},
 			[]logicalplan.Expr{logicalplan.Col("labels.label2")},
 		).
 		Execute(context.Background(), func(ctx context.Context, r arrow.Record) error {
@@ -1815,7 +1815,7 @@ func Test_DB_Limiter(t *testing.T) {
 					),
 				).
 				Aggregate(
-					[]logicalplan.Expr{logicalplan.Sum(logicalplan.Col("value"))},
+					[]*logicalplan.AggregationFunction{logicalplan.Sum(logicalplan.Col("value"))},
 					[]logicalplan.Expr{logicalplan.Col("labels.namespace")},
 				).
 				Execute(context.Background(), func(ctx context.Context, r arrow.Record) error {
@@ -1950,7 +1950,7 @@ func Test_DB_EngineInMemory(t *testing.T) {
 	engine := query.NewEngine(pool, db.TableProvider(), query.WithPhysicalplanOptions(physicalplan.WithInMemoryOnly()))
 	err = engine.ScanTable("test").
 		Aggregate(
-			[]logicalplan.Expr{logicalplan.Sum(logicalplan.Col("value"))},
+			[]*logicalplan.AggregationFunction{logicalplan.Sum(logicalplan.Col("value"))},
 			[]logicalplan.Expr{logicalplan.Col("labels.namespace")},
 		).
 		Execute(context.Background(), func(ctx context.Context, r arrow.Record) error {
@@ -1962,7 +1962,7 @@ func Test_DB_EngineInMemory(t *testing.T) {
 	engine = query.NewEngine(pool, db.TableProvider())
 	err = engine.ScanTable("test").
 		Aggregate(
-			[]logicalplan.Expr{logicalplan.Sum(logicalplan.Col("value"))},
+			[]*logicalplan.AggregationFunction{logicalplan.Sum(logicalplan.Col("value"))},
 			[]logicalplan.Expr{logicalplan.Col("labels.namespace")},
 		).
 		Execute(context.Background(), func(ctx context.Context, r arrow.Record) error {
@@ -2796,87 +2796,3 @@ func Test_DB_PersistentDiskCompaction_BlockRotation(t *testing.T) {
 
 	validateRows(2400)
 }
-
-func Test_DB_PersistentDiskCompaction_NonOverlappingCompaction(t *testing.T) {
-	t.Parallel()
-	config := NewTableConfig(
-		dynparquet.SampleDefinition(),
-	)
-	logger := newTestLogger(t)
-
-	dir := t.TempDir()
-	cfg := []*index.LevelConfig{
-		{Level: index.L0, MaxSize: 336, Type: index.CompactionTypeParquetDisk},
-		{Level: index.L1, MaxSize: 1024 * 1024 * 128, Type: index.CompactionTypeParquetDisk},
-		{Level: index.L2, MaxSize: 1024 * 1024 * 512},
-	}
-	c, err := New(
-		WithLogger(logger),
-		WithIndexConfig(cfg),
-		WithStoragePath(dir),
-		WithWAL(),
-		WithManualBlockRotation(),
-	)
-	t.Cleanup(func() {
-		require.NoError(t, c.Close())
-	})
-	require.NoError(t, err)
-	db, err := c.DB(context.Background(), "test")
-	require.NoError(t, err)
-	table, err := db.Table("test", config)
-	require.NoError(t, err)
-
-	ctx := context.Background()
-	for i := 0; i < 3; i++ { // Create non-overlapping samples for compaction
-		samples := dynparquet.GenerateTestSamples(1)
-		r, err := samples.ToRecord()
-		require.NoError(t, err)
-		_, err = table.InsertRecord(ctx, r)
-		require.NoError(t, err)
-	}
-	require.Eventually(t, func() bool {
-		return table.active.index.LevelSize(index.L1) != 0
-	}, time.Second, time.Millisecond*10)
-
-	// Snapshot the file
-	success := false
-	table.db.snapshot(context.Background(), false, func() {
-		success = true
-	})
-	require.True(t, success)
-
-	// Close the database
-	require.NoError(t, c.Close())
-
-	// Recover from the snapshot
-	c, err = New(
-		WithLogger(logger),
-		WithIndexConfig(cfg),
-		WithStoragePath(dir),
-		WithWAL(),
-	)
-	require.NoError(t, err)
-	t.Cleanup(func() {
-		require.NoError(t, c.Close())
-	})
-	db, err = c.DB(context.Background(), "test")
-	require.NoError(t, err)
-	table, err = db.Table("test", config)
-	require.NoError(t, err)
-
-	validateRows := func(expected int64) {
-		pool := memory.NewCheckedAllocator(memory.DefaultAllocator)
-		defer pool.AssertSize(t, 0)
-		rows := int64(0)
-		engine := query.NewEngine(pool, db.TableProvider())
-		err = engine.ScanTable("test").
-			Execute(context.Background(), func(ctx context.Context, r arrow.Record) error {
-				rows += r.NumRows()
-				return nil
-			})
-		require.NoError(t, err)
-		require.Equal(t, expected, rows)
-	}
-
-	validateRows(3)
-}
diff --git a/dynparquet/schema_test.go b/dynparquet/schema_test.go
index 63f0287..0e7af16 100644
--- a/dynparquet/schema_test.go
+++ b/dynparquet/schema_test.go
@@ -332,7 +332,7 @@ func TestIsDynamicColumn(t *testing.T) {
 		}
 		schema, err := SchemaFromDefinition(def)
 		require.NoError(t, err)
-		schema.IsDynamicColumn(tc.cName)
+		schema.FindDynamicColumnForConcreteColumn(tc.cName)
 	}
 }
 
@@ -351,7 +351,7 @@ func BenchmarkIsDynamicColumn(b *testing.B) {
 	schema, err := SchemaFromDefinition(def)
 	require.NoError(b, err)
 	for i := 0; i < b.N; i++ {
-		_ = schema.IsDynamicColumn("labels.label1")
+		schema.FindDynamicColumnForConcreteColumn("labels.label1")
 	}
 }
 
diff --git a/logictest/logic_test.go b/logictest/logic_test.go
index 5b45bb6..b4f5c06 100644
--- a/logictest/logic_test.go
+++ b/logictest/logic_test.go
@@ -20,7 +20,7 @@ const testdataDirectory = "testdata"
 
 type frostDB struct {
 	*frostdb.DB
-	allocator *query.LimitAllocator
+	allocator *memory.CheckedAllocator
 }
 
 func (db frostDB) CreateTable(name string, schema *schemapb.Schema) (Table, error) {
@@ -129,14 +129,14 @@ func TestLogic(t *testing.T) {
 		require.NoError(t, err)
 		fdb := frostDB{
 			DB:        db,
-			allocator: query.NewLimitAllocator(1024*1024*1024, memory.DefaultAllocator),
+			allocator: memory.NewCheckedAllocator(memory.DefaultAllocator),
 		}
 		r := NewRunner(fdb, schemas)
 		datadriven.RunTest(t, path, func(t *testing.T, c *datadriven.TestData) string {
 			return r.RunCmd(ctx, c)
 		})
 		if path != "testdata/exec/aggregate/ordered_aggregate" { // NOTE: skip checking the limit for the ordered aggregator as it still leaks memory.
-			require.Equal(t, 0, fdb.allocator.Allocated())
+			fdb.allocator.AssertSize(t, 0)
 		}
 	})
 }
diff --git a/logictest/runner.go b/logictest/runner.go
index 294acf3..ba90547 100644
--- a/logictest/runner.go
+++ b/logictest/runner.go
@@ -71,6 +71,7 @@ type Runner struct {
 	schemas                   map[string]*schemapb.Schema
 	activeTable               Table
 	activeTableName           string
+	activeTableParquetSchema  *parquet.Schema
 	activeTableDynamicColumns []string
 	sqlParser                 *sqlparse.Parser
 }
@@ -127,6 +128,7 @@ func (r *Runner) handleCreateTable(_ context.Context, c *datadriven.TestData) (s
 	}
 	r.activeTable = table
 	r.activeTableName = name
+	r.activeTableParquetSchema = table.Schema().ParquetSchema()
 	for _, c := range table.Schema().Columns() {
 		if c.Dynamic {
 			r.activeTableDynamicColumns = append(r.activeTableDynamicColumns, c.Name)
@@ -265,7 +267,7 @@ func (r *Runner) handleInsert(ctx context.Context, c *datadriven.TestData) (stri
 	converter := pqarrow.NewParquetConverter(memory.NewGoAllocator(), logicalplan.IterOptions{})
 	defer converter.Close()
 
-	if err := converter.Convert(ctx, buf); err != nil {
+	if err := converter.Convert(ctx, buf, schema); err != nil {
 		return "", err
 	}
 
@@ -384,9 +386,14 @@ func (r *Runner) handleExec(ctx context.Context, c *datadriven.TestData) (string
 	return b.String(), nil
 }
 
-func (r *Runner) parseSQL(dynColNames []string, sql string) (sqlparse.ParseResult, error) {
+func (r *Runner) parseSQL(
+	dynColNames []string,
+	sql string,
+) (sqlparse.ParseResult, error) {
 	res, err := r.sqlParser.ExperimentalParse(
-		r.db.ScanTable(r.activeTableName), dynColNames, sql,
+		r.db.ScanTable(r.activeTableName),
+		dynColNames,
+		sql,
 	)
 	if err != nil {
 		return sqlparse.ParseResult{}, err
diff --git a/logictest/testdata/exec/aggregate/aggregate b/logictest/testdata/exec/aggregate/aggregate
index 87c799d..6b4620d 100644
--- a/logictest/testdata/exec/aggregate/aggregate
+++ b/logictest/testdata/exec/aggregate/aggregate
@@ -14,78 +14,83 @@ value6  value2  null    value4  stack1  6   6   6.6
 ----
 
 exec
-select sum(value) as value_sum group by labels.label2
+select sum(value) as value_sum, labels.label2 group by labels.label2
 ----
-value2  21
+21      value2
 
 exec
-select sum(floatvalue) as float_value_sum group by labels.label2
+select labels.label2, sum(floatvalue) as float_value_sum group by labels.label2
 ----
 value2  23.100000
 
 exec
-select max(value) as value_max group by labels.label2
+select labels.label2, max(value) as value_max group by labels.label2
 ----
 value2  6
 
 exec
-select max(floatvalue) as value_max group by labels.label2
+select labels.label2, max(floatvalue) as value_max group by labels.label2
 ----
 value2  6.600000
 
 exec
-select min(floatvalue) as value_min group by labels.label2
+select labels.label2, min(floatvalue) as value_min group by labels.label2
 ----
 value2  1.100000
 
 exec
-select count(value) as value_count group by labels.label2
+select labels.label2, count(value) as value_count group by labels.label2
 ----
 value2  6
 
 exec
-select avg(value) as value_avg group by labels.label2
+select labels.label2, avg(value) group by labels.label2
 ----
 value2  3
 
 exec
-select avg(floatvalue) as value_avg group by labels.label2
+select labels.label2, avg(value) as value_avg group by labels.label2
+----
+value2  3
+
+exec
+select labels.label2, avg(floatvalue) as value_avg group by labels.label2
 ----
 value2  3.850000
 
 exec
-select avg(value) as value_avg group by labels.label4
+select labels.label4, avg(value) as value_avg group by labels.label4
 ----
 null    3
 value4  4
 
 exec
-select sum(value), count(value) group by stacktrace
+select stacktrace, sum(value), count(value) group by stacktrace
 ----
 stack1  21      6
 
 exec
-select sum(floatvalue), count(floatvalue) group by stacktrace
+select stacktrace, sum(floatvalue), count(floatvalue) group by stacktrace
 ----
 stack1  23.100000  6
 
 exec
-select sum(value) as value_sum, count(value) as value_count group by stacktrace
+select stacktrace, sum(value) as value_sum, count(value) as value_count group by stacktrace
 ----
 stack1  21      6
 
 exec
-select sum(value), count(value), min(value), max(value) group by labels.label2
+select labels.label2, sum(value), count(value), min(value), max(value) group by labels.label2
 ----
 value2  21      6       1       6
 
 exec
-select sum(floatvalue), count(floatvalue), min(floatvalue), max(floatvalue) group by labels.label2
+select labels.label2, sum(floatvalue), count(floatvalue), min(floatvalue), max(floatvalue) group by labels.label2
 ----
 value2  23.100000  6       1.100000  6.600000
 
 exec unordered
-select sum(value) as value_sum where timestamp >= 1 group by labels.label1
+select labels.label1, sum(value) as value_sum where timestamp >= 1 group by labels.label1
 ----
 value1  1
 value2  2
@@ -93,3 +98,13 @@ value3  3
 value4  4
 value5  5
 value6  6
+
+exec unordered
+select labels, sum(value) as value_sum group by labels
+----
+value1  value2  null    null    1
+value2  value2  value3  null    2
+value3  value2  null    value4  3
+value4  value2  null    null    4
+value5  value2  value3  null    5
+value6  value2  null    value4  6
diff --git a/logictest/testdata/exec/aggregate/aggregate_nulls b/logictest/testdata/exec/aggregate/aggregate_nulls
index ec0394f..4f0d81f 100644
--- a/logictest/testdata/exec/aggregate/aggregate_nulls
+++ b/logictest/testdata/exec/aggregate/aggregate_nulls
@@ -10,25 +10,25 @@ null    value2  stack1  3   3
 # TODO query results should explicitly return null values
 # https://github.com/polarsignals/frostdb/issues/259
 exec
-select sum(value) as value_sum group by labels.label2
+select labels.label2, sum(value) as value_sum group by labels.label2
 ----
 value2  5
 null    1
 
 exec
-select max(value) as value_max group by labels.label2
+select labels.label2, max(value) as value_max group by labels.label2
 ----
 value2  3
 null    1
 
 exec
-select count(value) as value_count group by labels.label2
+select labels.label2, count(value) as value_count group by labels.label2
 ----
 value2  2
 null    1
 
 exec
-select sum(value) as value_sum, count(value) as value_count group by labels.label2
+select labels.label2, sum(value) as value_sum, count(value) as value_count group by labels.label2
 ----
 value2  5       2
 null    1       1
diff --git a/logictest/testdata/exec/aggregate/math b/logictest/testdata/exec/aggregate/math
new file mode 100644
index 0000000..100e4f7
--- /dev/null
+++ b/logictest/testdata/exec/aggregate/math
@@ -0,0 +1,138 @@
+createtable schema=default
+----
+
+insert cols=(labels.label1, timestamp, value)
+value1 1 2
+value1 3 4
+value1 5 6
+value1 11 0
+----
+
+exec
+select value * timestamp
+----
+2
+12
+30
+0
+
+exec
+select value * 2
+----
+4
+8
+12
+0
+
+exec
+select 2 * 3
+----
+6
+6
+6
+6
+
+exec
+select value * (2 * 3)
+----
+12
+24
+36
+0
+
+exec
+select value * (timestamp * 2)
+----
+4
+24
+60
+0
+
+exec
+select value * timestamp * 2
+----
+4
+24
+60
+0
+
+exec
+select value * timestamp + 2 - 1
+----
+3
+13
+31
+1
+
+exec
+select value * timestamp * (2 - 1)
+----
+2
+12
+30
+0
+
+exec
+select value, value * timestamp * (2 - 1)
+----
+2       2
+4       12
+6       30
+0       0
+
+exec
+select timestamp, value * timestamp * (2 - 1) where value > 2
+----
+3       12
+5       30
+
+exec
+select value, timestamp  * value * (2 - 1)  where value > 2
+----
+4       12
+6       30
+
+exec
+select value , value  * timestamp * (2 - 1) as result where value > 2
+----
+4       12
+6       30
+
+exec
+select value as result, value  * timestamp * (2 - 1)  where value > 2
+----
+4       12
+6       30
+
+exec
+select labels.label1, sum(value), count(value) group by labels.label1
+----
+value1  12      4
+
+exec
+select   max(value) - min(value) , sum(value) / count(value)  group by labels.label1
+----
+6       3
+
+exec
+select  labels.label1, max(value) - min(value) , sum(value) / count(value)  group by labels.label1
+----
+value1  6       3
+
+exec
+select  labels.label1, (max(value) - min(value)) /2  , sum(value) / count(value)  group by labels.label1
+----
+value1  3       3
+
+exec
+select  labels.label1, (max(value) - min(value)) /2  , sum(value) / count(value) , (sum(value) / count(value)) * 2 group by labels.label1
+----
+value1  3       3       6
+
+exec
+select timestamp / value
+----
+0
+0
+0
+null
diff --git a/logictest/testdata/exec/aggregate/ordered_aggregate b/logictest/testdata/exec/aggregate/ordered_aggregate
deleted file mode 100644
index 4e8818c..0000000
--- a/logictest/testdata/exec/aggregate/ordered_aggregate
+++ /dev/null
@@ -1,53 +0,0 @@
-createtable schema=default
-----
-
-insert cols=(example_type, labels.label1, timestamp, value)
-type1   value1  1   1
-type1   value1  2   1
-type2   value1  1   1
-----
-
-insert cols=(example_type, labels.label1, timestamp, value)
-type1   value1  1   1
-type1   value1  2   1
-type2   value1  1   1
-----
-
-insert cols=(example_type, labels.label2, timestamp, value)
-type1   value2  1   1
-type1   value2  2   1
-type2   value2  1   1
-----
-
-insert cols=(example_type, labels.label3, timestamp, value)
-type1   value3  1   1
-type1   value3  2   1
-type2   value3  1   1
-----
-
-exec
-select sum(value) as value_sum group by (example_type, labels)
-----
-type1   null    null    value3  2
-type1   null    value2  null    2
-type1   value1  null    null    4
-type2   null    null    value3  1
-type2   null    value2  null    1
-type2   value1  null    null    2
-
-exec
-select sum(value) as value_sum where example_type = 'type1' group by (labels)
-----
-null    null    value3  2
-null    value2  null    2
-value1  null    null    4
-
-exec
-select sum(value) as value_sum where example_type = 'type1' group by (labels, timestamp)
-----
-null    null    value3  1       1
-null    null    value3  2       1
-null    value2  null    1       1
-null    value2  null    2       1
-value1  null    null    1       2
-value1  null    null    2       2
diff --git a/logictest/testdata/exec/aggregate/window b/logictest/testdata/exec/aggregate/window
index 7838bf4..c4d444f 100644
--- a/logictest/testdata/exec/aggregate/window
+++ b/logictest/testdata/exec/aggregate/window
@@ -11,52 +11,53 @@ value4  stack1  123000  4
 ----
 
 exec
-select sum(value) as value_sum group by second(1)
+select sum(value) as value_sum, (timestamp/1000)*1000 as timestamp_bucket group by timestamp_bucket
 ----
-120000  1
-121000  2
-122000  3
-123000  4
+1       120000
+2       121000
+3       122000
+4       123000
 
 exec
-select sum(value) as value_sum group by second(2)
+select sum(value) as value_sum, (timestamp/2000)*2000 as timestamp_bucket group by timestamp_bucket
 ----
-120000  3
-122000  7
+3       120000
+7       122000
 
 exec
-select sum(value) as value_sum group by second(3)
+select sum(value) as value_sum, (timestamp/3000)*3000 as timestamp_bucket group by timestamp_bucket
 ----
-120000  6
-123000  4
+6       120000
+4       123000
 
 exec
-select sum(value) as value_sum, count(value) as value_count group by second(3)
+select sum(value) as value_sum, count(value) as value_count, (timestamp/3000)*3000 as timestamp_bucket group by timestamp_bucket
 ----
-120000  6       3
-123000  4       1
+6       3       120000
+4       1       123000
 
 exec
-select sum(value) as value_sum group by second(4)
+select sum(value) as value_sum, (timestamp/4000)*4000 as timestamp_bucket group by timestamp_bucket
 ----
-120000  10
+10      120000
 
 exec
-select sum(value) as value_sum group by labels.label1, second(5)
+select labels.label1, (timestamp/5000)*5000 as timestamp_bucket, sum(value) as value_sum group by labels.label1, timestamp_bucket
+# TODO: Not sure if this is strictly correct. Previously with "group by second(5)" we preserved the timestamp if there was only a single timestamp.
 ----
 value1  120000  1
-value2  121000  2
-value3  122000  3
-value4  123000  4
+value2  120000  2
+value3  120000  3
+value4  120000  4
 
 exec
-select sum(value) as value_sum, count(timestamp) as timestamp_count group by second(2)
+select (timestamp/2000)*2000 as timestamp_bucket, sum(value) as value_sum, count(timestamp) as timestamp_count group by timestamp_bucket
 ----
 120000  3       2
 122000  7       2
 
 exec
-select count(timestamp) as timestamp_count group by second(3)
+select (timestamp/3000)*3000 as timestamp_bucket, count(timestamp) as timestamp_count group by timestamp_bucket
 ----
 120000  3
 123000  1
diff --git a/logictest/testdata/exec/projection/convert b/logictest/testdata/exec/projection/convert
new file mode 100644
index 0000000..12edf0c
--- /dev/null
+++ b/logictest/testdata/exec/projection/convert
@@ -0,0 +1,15 @@
+createtable schema=default
+----
+
+insert cols=(labels.label1, stacktrace, timestamp, value, floatvalue)
+value1 stack1 1 2 1.1
+value1 stack1 3 4 1.1
+value1 stack2 5 6 1.1
+----
+
+exec
+select convert(value, float) * floatvalue
+----
+2.200000
+4.400000
+6.600000
diff --git a/logictest/testdata/exec/projection/math_projection b/logictest/testdata/exec/projection/math_projection
index e6f3ad7..6238f30 100644
--- a/logictest/testdata/exec/projection/math_projection
+++ b/logictest/testdata/exec/projection/math_projection
@@ -1,10 +1,10 @@
 createtable schema=default
 ----
 
-insert cols=(labels.label1, timestamp, value)
-value1 1 2
-value1 3 4
-value1 5 6
+insert cols=(labels.label1, stacktrace, timestamp, value)
+value1 stack1 1 2
+value1 stack1 3 4
+value1 stack2 5 6
 ----
 
 exec
@@ -13,3 +13,9 @@ select value * timestamp
 2
 12
 30
+
+exec
+select stacktrace, sum(value * timestamp) group by stacktrace
+----
+stack1  14
+stack2  30
diff --git a/logictest/testdata/plan/aggregate/aggregate b/logictest/testdata/plan/aggregate/aggregate
index c18ec75..4be8a6a 100644
--- a/logictest/testdata/plan/aggregate/aggregate
+++ b/logictest/testdata/plan/aggregate/aggregate
@@ -7,48 +7,78 @@ createtable schema=default
 exec
 explain select sum(value) as value_sum group by (example_type, labels.label1)
 ----
-TableScan [concurrent] - HashAggregate (value_sum by example_type,labels.label1) - Synchronizer - HashAggregate (value_sum by example_type,labels.label1)
+TableScan [concurrent] - Projection (value, example_type, labels.label1) - HashAggregate (sum(value) by example_type,labels.label1) - Synchronizer - HashAggregate (sum(value) by example_type,labels.label1) - Projection (sum(value) as value_sum)
 
 # An ordered aggregation should be planned in this case because the full set of dynamic columns is specified.
 exec
 explain select sum(value) as value_sum group by (example_type, labels)
 ----
-TableScan [concurrent] - OrderedAggregate (value by example_type,labels) - OrderedSynchronizer - OrderedAggregate (value by example_type,labels)
+TableScan [concurrent] - Projection (value, example_type, labels) - HashAggregate (sum(value) by example_type,labels) - Synchronizer - HashAggregate (sum(value) by example_type,labels) - Projection (sum(value) as value_sum)
 
 # A hash aggregation is planned in the case that the group by columns are inverted. TODO(asubiotto): We could probably
 # plan an ordered aggregation in this case, but let's not do so unless necessary.
 exec
 explain select sum(value) as value_sum group by (labels, example_type)
 ----
-TableScan [concurrent] - HashAggregate (value_sum by labels,example_type) - Synchronizer - HashAggregate (value_sum by labels,example_type)
+TableScan [concurrent] - Projection (value, labels, example_type) - HashAggregate (sum(value) by labels,example_type) - Synchronizer - HashAggregate (sum(value) by labels,example_type) - Projection (sum(value) as value_sum)
 
 # A hash aggregation should be planned in this case because we are not grouping by example_type, so the group by columns
 # are not a prefix of the sorting columns.
 exec
 explain select sum(value) as value_sum group by (labels)
 ----
-TableScan [concurrent] - HashAggregate (value_sum by labels) - Synchronizer - HashAggregate (value_sum by labels)
+TableScan [concurrent] - Projection (value, labels) - HashAggregate (sum(value) by labels) - Synchronizer - HashAggregate (sum(value) by labels) - Projection (sum(value) as value_sum)
 
 # If however, the prefix that is not contained by the grouping columns is covered by an equality filter, an ordered
 # aggregation can be planned since we "fix" a prefix of the sorting columns to a given value.
 exec
 explain select sum(value) as value_sum where example_type = 'some_value' group by (labels)
 ----
-TableScan [concurrent] - PredicateFilter (example_type == some_value) - OrderedAggregate (value by labels) - OrderedSynchronizer - OrderedAggregate (value by labels)
+TableScan [concurrent] - PredicateFilter (example_type == some_value) - Projection (value, labels) - HashAggregate (sum(value) by labels) - Synchronizer - HashAggregate (sum(value) by labels) - Projection (sum(value) as value_sum)
 
 exec
 explain select sum(value) as value_sum where example_type = 'some_value' group by (labels, timestamp)
 ----
-TableScan [concurrent] - PredicateFilter (example_type == some_value) - OrderedAggregate (value by labels,timestamp) - OrderedSynchronizer - OrderedAggregate (value by labels,timestamp)
+TableScan [concurrent] - PredicateFilter (example_type == some_value) - Projection (value, labels, timestamp) - HashAggregate (sum(value) by labels,timestamp) - Synchronizer - HashAggregate (sum(value) by labels,timestamp) - Projection (sum(value) as value_sum)
 
 # The above only applies to equality filters of course.
 exec
 explain select sum(value) as value_sum where example_type > 'some_value' group by (labels)
 ----
-TableScan [concurrent] - PredicateFilter (example_type > some_value) - HashAggregate (value_sum by labels) - Synchronizer - HashAggregate (value_sum by labels)
+TableScan [concurrent] - PredicateFilter (example_type > some_value) - Projection (value, labels) - HashAggregate (sum(value) by labels) - Synchronizer - HashAggregate (sum(value) by labels) - Projection (sum(value) as value_sum)
 
 # And here's a hash aggregation with a filter on another column.
 exec
 explain select sum(value) as value_sum where timestamp >= 1 group by labels.label1
 ----
-TableScan [concurrent] - PredicateFilter (timestamp >= 1) - HashAggregate (value_sum by labels.label1) - Synchronizer - HashAggregate (value_sum by labels.label1)
+TableScan [concurrent] - PredicateFilter (timestamp >= 1) - Projection (value, labels.label1) - HashAggregate (sum(value) by labels.label1) - Synchronizer - HashAggregate (sum(value) by labels.label1) - Projection (sum(value) as value_sum)
+
+exec
+explain select labels.label2, sum(value) as value_sum, count(value) as value_count group by labels.label2
+----
+TableScan [concurrent] - Projection (labels.label2, value) - HashAggregate (sum(value),count(value) by labels.label2) - Synchronizer - HashAggregate (sum(value),count(value) by labels.label2) - Projection (labels.label2, sum(value) as value_sum, count(value) as value_count)
+
+exec
+explain select example_type, labels, sum(value) as value_sum group by (example_type, labels)
+----
+TableScan [concurrent] - Projection (example_type, labels, value) - HashAggregate (sum(value) by example_type,labels) - Synchronizer - HashAggregate (sum(value) by example_type,labels) - Projection (example_type, labels, sum(value) as value_sum)
+
+exec
+explain select sum(value * timestamp) group by stacktrace
+----
+TableScan [concurrent] - Projection (value * timestamp, stacktrace) - HashAggregate (sum(value * timestamp) by stacktrace) - Synchronizer - HashAggregate (sum(value * timestamp) by stacktrace) - Projection (sum(value * timestamp))
+
+exec
+explain select avg(value) group by stacktrace
+----
+TableScan [concurrent] - Projection (value, stacktrace) - HashAggregate (sum(value),count(value) by stacktrace) - Synchronizer - HashAggregate (sum(value),count(value) by stacktrace) - Projection (stacktrace, sum(value) / count(value) as avg(value)) - Projection (avg(value))
+
+exec
+explain select max(value) - min(value), sum(value) / count(value) group by labels.label1
+----
+TableScan [concurrent] - Projection (value, labels.label1) - HashAggregate (max(value),min(value),sum(value),count(value) by labels.label1) - Synchronizer - HashAggregate (max(value),min(value),sum(value),count(value) by labels.label1) - Projection (max(value) - min(value), sum(value) / count(value))
+
+exec
+explain select labels.label1, (max(value) - min(value)) / 2, sum(value) / count(value), (sum(value) / count(value)) * 2 group by labels.label1
+----
+TableScan [concurrent] - Projection (labels.label1, value) - HashAggregate (max(value),min(value),sum(value),count(value),sum(value),count(value) by labels.label1) - Synchronizer - HashAggregate (max(value),min(value),sum(value),count(value),sum(value),count(value) by labels.label1) - Projection (labels.label1, max(value) - min(value) / 2, sum(value) / count(value), sum(value) / count(value) * 2)
diff --git a/logictest/testdata/plan/aggregate/window b/logictest/testdata/plan/aggregate/window
new file mode 100644
index 0000000..b6310ef
--- /dev/null
+++ b/logictest/testdata/plan/aggregate/window
@@ -0,0 +1,9 @@
+createtable schema=default
+----
+
+# timestamps have to be in milliseconds for these tests
+
+exec
+explain select sum(value) as value_sum, ((timestamp / 1000) * 1000) as timestamp_bucket group by timestamp_bucket
+----
+TableScan [concurrent] - Projection (value, timestamp / 1000 * 1000 as timestamp_bucket) - HashAggregate (sum(value) by timestamp_bucket) - Synchronizer - HashAggregate (sum(value) by timestamp_bucket) - Projection (sum(value) as value_sum, timestamp_bucket)
diff --git a/pqarrow/arrow_test.go b/pqarrow/arrow_test.go
index ee6777c..4f299df 100644
--- a/pqarrow/arrow_test.go
+++ b/pqarrow/arrow_test.go
@@ -13,6 +13,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/polarsignals/frostdb/dynparquet"
+	schemapb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/schema/v1alpha1"
 	"github.com/polarsignals/frostdb/query/logicalplan"
 )
 
@@ -109,11 +110,11 @@ func TestDifferentSchemasToArrow(t *testing.T) {
 	c := NewParquetConverter(alloc, logicalplan.IterOptions{})
 	defer c.Close()
 
-	require.NoError(t, c.Convert(ctx, buf0))
-	require.NoError(t, c.Convert(ctx, buf1))
-	require.NoError(t, c.Convert(ctx, buf2))
-	require.NoError(t, c.Convert(ctx, buf3))
-	require.NoError(t, c.Convert(ctx, buf4))
+	require.NoError(t, c.Convert(ctx, buf0, dynSchema))
+	require.NoError(t, c.Convert(ctx, buf1, dynSchema))
+	require.NoError(t, c.Convert(ctx, buf2, dynSchema))
+	require.NoError(t, c.Convert(ctx, buf3, dynSchema))
+	require.NoError(t, c.Convert(ctx, buf4, dynSchema))
 
 	ar := c.NewRecord()
 	defer ar.Release()
@@ -225,7 +226,7 @@ func TestMergeToArrow(t *testing.T) {
 
 	ctx := context.Background()
 
-	as, err := ParquetRowGroupToArrowSchema(ctx, merge, logicalplan.IterOptions{})
+	as, err := ParquetRowGroupToArrowSchema(ctx, merge, dynSchema, logicalplan.IterOptions{})
 	require.NoError(t, err)
 	require.Len(t, as.Fields(), 8)
 	require.Equal(t, as.Field(0), arrow.Field{Name: "example_type", Type: &arrow.DictionaryType{
@@ -257,7 +258,7 @@ func TestMergeToArrow(t *testing.T) {
 
 	c := NewParquetConverter(memory.DefaultAllocator, logicalplan.IterOptions{})
 	defer c.Close()
-	require.NoError(t, c.Convert(ctx, merge))
+	require.NoError(t, c.Convert(ctx, merge, dynSchema))
 	ar := c.NewRecord()
 	require.Equal(t, int64(5), ar.NumRows())
 	require.Equal(t, int64(8), ar.NumCols())
@@ -292,7 +293,7 @@ func BenchmarkNestedParquetToArrow(b *testing.B) {
 	b.ResetTimer()
 
 	for i := 0; i < b.N; i++ {
-		require.NoError(b, c.Convert(ctx, pb))
+		require.NoError(b, c.Convert(ctx, pb, schema))
 		// Reset converter.
 		_ = c.NewRecord()
 	}
@@ -328,7 +329,7 @@ func BenchmarkParquetToArrow(b *testing.B) {
 	b.ResetTimer()
 
 	for i := 0; i < b.N; i++ {
-		require.NoError(b, c.Convert(ctx, buf))
+		require.NoError(b, c.Convert(ctx, buf, dynSchema))
 		// Reset converter.
 		_ = c.NewRecord()
 	}
@@ -446,6 +447,7 @@ func TestDistinctBinaryExprOptimization(t *testing.T) {
 	as, err := ParquetRowGroupToArrowSchema(
 		ctx,
 		buf,
+		dynSchema,
 		logicalplan.IterOptions{
 			PhysicalProjection: []logicalplan.Expr{
 				logicalplan.Col("example_type"),
@@ -474,7 +476,7 @@ func TestDistinctBinaryExprOptimization(t *testing.T) {
 			DistinctColumns: distinctColumns,
 		})
 	defer c.Close()
-	require.NoError(t, c.Convert(ctx, buf))
+	require.NoError(t, c.Convert(ctx, buf, dynSchema))
 	ar := c.NewRecord()
 	require.Equal(t, int64(1), ar.NumRows())
 	require.Equal(t, int64(3), ar.NumCols())
@@ -542,6 +544,7 @@ func TestDistinctBinaryExprOptimizationMixed(t *testing.T) {
 	as, err := ParquetRowGroupToArrowSchema(
 		ctx,
 		buf,
+		dynSchema,
 		logicalplan.IterOptions{
 			PhysicalProjection: []logicalplan.Expr{
 				logicalplan.Col("example_type"),
@@ -567,7 +570,7 @@ func TestDistinctBinaryExprOptimizationMixed(t *testing.T) {
 		DistinctColumns: distinctColumns,
 	})
 	defer c.Close()
-	require.NoError(t, c.Convert(ctx, buf))
+	require.NoError(t, c.Convert(ctx, buf, dynSchema))
 	ar := c.NewRecord()
 	require.Equal(t, int64(2), ar.NumRows())
 	require.Equal(t, int64(3), ar.NumCols())
@@ -587,7 +590,18 @@ func TestList(t *testing.T) {
 
 	c := NewParquetConverter(memory.DefaultAllocator, logicalplan.IterOptions{})
 	defer c.Close()
-	require.NoError(t, c.Convert(ctx, buf))
+	s, err := dynparquet.SchemaFromDefinition(&schemapb.Schema{
+		Name: "test",
+		Columns: []*schemapb.Column{{
+			Name: "data",
+			StorageLayout: &schemapb.StorageLayout{
+				Type:     schemapb.StorageLayout_TYPE_INT64,
+				Repeated: true,
+			},
+		}},
+	})
+	require.NoError(t, err)
+	require.NoError(t, c.Convert(ctx, buf, s))
 
 	record := c.NewRecord()
 	t.Log(record)
@@ -765,6 +779,7 @@ func Test_ParquetRowGroupToArrowSchema_Groups(t *testing.T) {
 			as, err := ParquetRowGroupToArrowSchema(
 				ctx,
 				buf,
+				schema,
 				logicalplan.IterOptions{
 					PhysicalProjection: test.physicalProjections,
 				},
@@ -804,7 +819,7 @@ func Test_ParquetToArrowV2(t *testing.T) {
 			},
 		})
 		require.NoError(t, err)
-		require.NoError(t, c.Convert(ctx, pb))
+		require.NoError(t, c.Convert(ctx, pb, schema))
 	}
 	r := c.NewRecord()
 	defer r.Release()
@@ -840,7 +855,7 @@ func Test_ParquetToArrow(t *testing.T) {
 	c := NewParquetConverter(alloc, logicalplan.IterOptions{})
 	defer c.Close()
 
-	require.NoError(t, c.Convert(ctx, buf))
+	require.NoError(t, c.Convert(ctx, buf, dynSchema))
 	r := c.NewRecord()
 	defer r.Release()
 	require.Equal(t, int64(1000), r.NumRows())
diff --git a/pqarrow/parquet_test.go b/pqarrow/parquet_test.go
index 804bbdc..d3b5a99 100644
--- a/pqarrow/parquet_test.go
+++ b/pqarrow/parquet_test.go
@@ -10,6 +10,8 @@ import (
 	"github.com/apache/arrow/go/v14/arrow/memory"
 	"github.com/parquet-go/parquet-go"
 	"github.com/stretchr/testify/require"
+
+	"github.com/polarsignals/frostdb/dynparquet"
 )
 
 type noopWriter struct{}
@@ -80,7 +82,7 @@ func BenchmarkRecordsToFile(b *testing.B) {
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		if err := recordToRows(
-			noopWriter{}, func(string) bool { return false }, record, 0, numRows, parquetFields.Fields(),
+			noopWriter{}, func(string) (dynparquet.ColumnDefinition, bool) { return dynparquet.ColumnDefinition{}, false }, record, 0, numRows, parquetFields.Fields(),
 		); err != nil {
 			b.Fatal(err)
 		}
@@ -167,7 +169,7 @@ func TestRecordToRows_list(t *testing.T) {
 	}
 	clone := &cloneWriter{}
 	if err := recordToRows(
-		clone, func(string) bool { return false }, r, 0, int(r.NumRows()), parquetFields.Fields(),
+		clone, func(string) (dynparquet.ColumnDefinition, bool) { return dynparquet.ColumnDefinition{}, false }, r, 0, int(r.NumRows()), parquetFields.Fields(),
 	); err != nil {
 		t.Fatal(err)
 	}
diff --git a/query/logicalplan/builder_test.go b/query/logicalplan/builder_test.go
index 5fd3c22..ca50ac1 100644
--- a/query/logicalplan/builder_test.go
+++ b/query/logicalplan/builder_test.go
@@ -15,10 +15,10 @@ func TestLogicalPlanBuilder(t *testing.T) {
 		Scan(tableProvider, "table1").
 		Filter(Col("labels.test").Eq(Literal("abc"))).
 		Aggregate(
-			[]Expr{Sum(Col("value")).Alias("value_sum")},
+			[]*AggregationFunction{Sum(Col("value"))},
 			[]Expr{Col("stacktrace")},
 		).
-		Project(Col("stacktrace")).
+		Project(Col("stacktrace"), Sum(Col("value")).Alias("value_sum")).
 		Build()
 
 	require.Nil(t, err)
@@ -27,15 +27,16 @@ func TestLogicalPlanBuilder(t *testing.T) {
 		Projection: &Projection{
 			Exprs: []Expr{
 				&Column{ColumnName: "stacktrace"},
+				&AliasExpr{
+					Expr:  &AggregationFunction{Func: AggFuncSum, Expr: &Column{ColumnName: "value"}},
+					Alias: "value_sum",
+				},
 			},
 		},
 		Input: &LogicalPlan{
 			Aggregation: &Aggregation{
 				GroupExprs: []Expr{&Column{ColumnName: "stacktrace"}},
-				AggExprs: []Expr{&AliasExpr{
-					Expr:  &AggregationFunction{Func: AggFuncSum, Expr: &Column{ColumnName: "value"}},
-					Alias: "value_sum",
-				}},
+				AggExprs:   []*AggregationFunction{{Func: AggFuncSum, Expr: &Column{ColumnName: "value"}}},
 			},
 			Input: &LogicalPlan{
 				Filter: &Filter{
@@ -68,9 +69,14 @@ func TestLogicalPlanBuilderWithoutProjection(t *testing.T) {
 			Exprs: []Expr{&Column{ColumnName: "labels.test"}},
 		},
 		Input: &LogicalPlan{
-			TableScan: &TableScan{
-				TableProvider: tableProvider,
-				TableName:     "table1",
+			Projection: &Projection{
+				Exprs: []Expr{&Column{ColumnName: "labels.test"}},
+			},
+			Input: &LogicalPlan{
+				TableScan: &TableScan{
+					TableProvider: tableProvider,
+					TableName:     "table1",
+				},
 			},
 		},
 	}, p)
diff --git a/query/logicalplan/logicalplan_test.go b/query/logicalplan/logicalplan_test.go
index bf86212..b151def 100644
--- a/query/logicalplan/logicalplan_test.go
+++ b/query/logicalplan/logicalplan_test.go
@@ -60,10 +60,10 @@ func TestInputSchemaGetter(t *testing.T) {
 		Scan(&mockTableProvider{schema}, "table1").
 		Filter(Col("labels.test").Eq(Literal("abc"))).
 		Aggregate(
-			[]Expr{Sum(Col("value")).Alias("value_sum")},
+			[]*AggregationFunction{Sum(Col("value"))},
 			[]Expr{Col("stacktrace")},
 		).
-		Project(Col("stacktrace")).
+		Project(Col("stacktrace"), Sum(Col("value")).Alias("value_sum")).
 		Build()
 	require.Equal(t, schema, plan.InputSchema())
 
@@ -72,10 +72,10 @@ func TestInputSchemaGetter(t *testing.T) {
 		ScanSchema(&mockTableProvider{schema}, "table1").
 		Filter(Col("labels.test").Eq(Literal("abc"))).
 		Aggregate(
-			[]Expr{Sum(Col("value")).Alias("value_sum")},
+			[]*AggregationFunction{Sum(Col("value"))},
 			[]Expr{Col("stacktrace")},
 		).
-		Project(Col("stacktrace")).
+		Project(Col("stacktrace"), Sum(Col("value")).Alias("value_sum")).
 		Build()
 	require.Equal(t, schema, plan.InputSchema())
 
@@ -84,10 +84,10 @@ func TestInputSchemaGetter(t *testing.T) {
 	plan, _ = (&Builder{}).
 		Filter(Col("labels.test").Eq(Literal("abc"))).
 		Aggregate(
-			[]Expr{Sum(Col("value")).Alias("value_sum")},
+			[]*AggregationFunction{Sum(Col("value"))},
 			[]Expr{Col("stacktrace")},
 		).
-		Project(Col("stacktrace")).
+		Project(Col("stacktrace"), Sum(Col("value")).Alias("value_sum")).
 		Build()
 	require.Nil(t, plan.InputSchema())
 }
diff --git a/query/logicalplan/optimize_test.go b/query/logicalplan/optimize_test.go
index b8937cf..b369676 100644
--- a/query/logicalplan/optimize_test.go
+++ b/query/logicalplan/optimize_test.go
@@ -15,10 +15,10 @@ func TestOptimizePhysicalProjectionPushDown(t *testing.T) {
 		Scan(tableProvider, "table1").
 		Filter(Col("labels.test").Eq(Literal("abc"))).
 		Aggregate(
-			[]Expr{Sum(Col("value")).Alias("value_sum")},
+			[]*AggregationFunction{Sum(Col("value"))},
 			[]Expr{Col("stacktrace")},
 		).
-		Project(Col("stacktrace")).
+		Project(Col("stacktrace"), Sum(Col("value")).Alias("value_sum")).
 		Build()
 
 	optimizer := &PhysicalProjectionPushDown{}
@@ -32,7 +32,6 @@ func TestOptimizePhysicalProjectionPushDown(t *testing.T) {
 		// columns they access. The optimizer could potentially deduplicate or
 		// use a more efficient datastructure in the future.
 		PhysicalProjection: []Expr{
-			&Column{ColumnName: "stacktrace"},
 			&Column{ColumnName: "stacktrace"},
 			&Column{ColumnName: "value"},
 			DynCol(hashedMatch),
@@ -60,7 +59,7 @@ func TestOptimizeDistinctPushDown(t *testing.T) {
 		},
 	},
 		// Distinct -> TableScan
-		p.Input.TableScan,
+		p.Input.Input.TableScan,
 	)
 }
 
@@ -70,10 +69,10 @@ func TestOptimizeFilterPushDown(t *testing.T) {
 		Scan(tableProvider, "table1").
 		Filter(Col("labels.test").Eq(Literal("abc"))).
 		Aggregate(
-			[]Expr{Sum(Col("value")).Alias("value_sum")},
+			[]*AggregationFunction{Sum(Col("value"))},
 			[]Expr{Col("stacktrace")},
 		).
-		Project(Col("stacktrace")).
+		Project(Col("stacktrace"), Sum(Col("value")).Alias("value_sum")).
 		Build()
 
 	optimizer := &FilterPushDown{}
@@ -96,68 +95,45 @@ func TestOptimizeFilterPushDown(t *testing.T) {
 	)
 }
 
-func TestRemoveProjectionAtRoot(t *testing.T) {
-	p, _ := (&Builder{}).
-		Scan(&mockTableProvider{schema: dynparquet.NewSampleSchema()}, "table1").
-		Filter(Col("labels.test").Eq(Literal("abc"))).
-		Aggregate(
-			[]Expr{Sum(Col("value")).Alias("value_sum")},
-			[]Expr{Col("stacktrace")},
-		).
-		Project(Col("stacktrace")).
-		Build()
-
-	p = removeProjection(p)
-
-	require.True(t, p.Projection == nil)
-}
-
-func TestRemoveMiddleProjection(t *testing.T) {
-	p, _ := (&Builder{}).
-		Scan(&mockTableProvider{schema: dynparquet.NewSampleSchema()}, "table1").
-		Filter(Col("labels.test").Eq(Literal("abc"))).
-		Project(Col("stacktrace")).
-		Aggregate(
-			[]Expr{Sum(Col("value")).Alias("value_sum")},
-			[]Expr{Col("stacktrace")},
-		).
-		Build()
-
-	p = removeProjection(p)
-
-	require.True(t, p.Input.Projection == nil)
-}
-
-func TestRemoveLowestProjection(t *testing.T) {
+func TestProjectionPushDown(t *testing.T) {
 	p, _ := (&Builder{}).
 		Scan(&mockTableProvider{schema: dynparquet.NewSampleSchema()}, "table1").
-		Project(Col("stacktrace")).
 		Filter(Col("labels.test").Eq(Literal("abc"))).
 		Aggregate(
-			[]Expr{Sum(Col("value")).Alias("value_sum")},
+			[]*AggregationFunction{Sum(Col("value"))},
 			[]Expr{Col("stacktrace")},
 		).
 		Build()
 
-	p = removeProjection(p)
+	p = (&PhysicalProjectionPushDown{}).Optimize(p)
 
-	require.True(t, p.Input.Input.Projection == nil)
+	require.Equal(t, []Expr{
+		Col("stacktrace"),
+		Col("value"),
+		DynCol("hashed"),
+		Col("labels.test"),
+	}, p.Input.Input.TableScan.PhysicalProjection)
 }
 
-func TestProjectionPushDown(t *testing.T) {
+func TestProjectionPushDownReset(t *testing.T) {
 	p, _ := (&Builder{}).
 		Scan(&mockTableProvider{schema: dynparquet.NewSampleSchema()}, "table1").
 		Filter(Col("labels.test").Eq(Literal("abc"))).
 		Aggregate(
-			[]Expr{Sum(Col("value")).Alias("value_sum")},
+			[]*AggregationFunction{Sum(Col("value"))},
 			[]Expr{Col("stacktrace")},
 		).
-		Project(Col("labels")).
+		Project(Col("test")).
 		Build()
 
-	p = (&ProjectionPushDown{}).Optimize(p)
+	p = (&PhysicalProjectionPushDown{}).Optimize(p)
 
-	require.True(t, p.Input.Input.Projection == nil)
+	require.Equal(t, []Expr{
+		Col("stacktrace"),
+		Col("value"),
+		DynCol("hashed"),
+		Col("labels.test"),
+	}, p.Input.Input.Input.TableScan.PhysicalProjection)
 }
 
 func TestProjectionPushDownOfDistinct(t *testing.T) {
@@ -166,9 +142,9 @@ func TestProjectionPushDownOfDistinct(t *testing.T) {
 		Distinct(DynCol("labels")).
 		Build()
 
-	p = (&ProjectionPushDown{}).Optimize(p)
+	p = (&PhysicalProjectionPushDown{}).Optimize(p)
 
-	require.True(t, p.Input.Projection != nil)
+	require.Equal(t, []Expr{DynCol("labels")}, p.Input.Input.TableScan.PhysicalProjection)
 }
 
 func TestAllOptimizers(t *testing.T) {
@@ -177,18 +153,13 @@ func TestAllOptimizers(t *testing.T) {
 		Scan(tableProvider, "table1").
 		Filter(Col("labels.test").Eq(Literal("abc"))).
 		Aggregate(
-			[]Expr{Sum(Col("value")).Alias("value_sum")},
+			[]*AggregationFunction{Sum(Col("value"))},
 			[]Expr{Col("stacktrace")},
 		).
 		Project(Col("stacktrace")).
 		Build()
 
-	optimizers := []Optimizer{
-		&PhysicalProjectionPushDown{},
-		&FilterPushDown{},
-		&DistinctPushDown{},
-		&ProjectionPushDown{},
-	}
+	optimizers := DefaultOptimizers()
 
 	for _, optimizer := range optimizers {
 		p = optimizer.Optimize(p)
@@ -202,7 +173,6 @@ func TestAllOptimizers(t *testing.T) {
 		// columns they access. The optimizer could potentially deduplicate or
 		// use a more efficient datastructure in the future.
 		PhysicalProjection: []Expr{
-			&Column{ColumnName: "stacktrace"},
 			&Column{ColumnName: "stacktrace"},
 			&Column{ColumnName: "value"},
 			DynCol(hashedMatch),
diff --git a/query/logicalplan/validate_test.go b/query/logicalplan/validate_test.go
index 3ceb678..46dd8d9 100644
--- a/query/logicalplan/validate_test.go
+++ b/query/logicalplan/validate_test.go
@@ -69,32 +69,10 @@ func TestAggregationMustHaveExpr(t *testing.T) {
 	require.True(t, strings.HasPrefix(planErr.message, "invalid aggregation: expression cannot be nil"))
 }
 
-func TestAggregationExprCannotHaveInvalidType(t *testing.T) {
-	invalidExprs := [][]Expr{
-		{Literal(4)},
-		{Col("Test")},
-	}
-
-	for _, expr := range invalidExprs {
-		_, err := (&Builder{}).
-			Aggregate(expr, nil).
-			Build()
-
-		require.NotNil(t, err)
-		require.NotNil(t, err)
-		planErr, ok := err.(*PlanValidationError)
-		require.True(t, ok)
-		require.True(t, strings.HasPrefix(planErr.message, "invalid aggregation"))
-		require.Len(t, planErr.children, 1)
-		exprErr := planErr.children[0]
-		require.True(t, strings.HasPrefix(exprErr.message, "aggregation expression is invalid"))
-	}
-}
-
 func TestAggregationExprColumnMustExistInSchema(t *testing.T) {
 	_, err := (&Builder{}).
 		Scan(&mockTableProvider{dynparquet.NewSampleSchema()}, "table1").
-		Aggregate([]Expr{Sum(Col("bad_column"))}, nil).
+		Aggregate([]*AggregationFunction{Sum(Col("bad_column"))}, nil).
 		Build()
 
 	require.NotNil(t, err)
@@ -123,7 +101,7 @@ func TestAggregationCannotSumOrMaxTextColumn(t *testing.T) {
 	} {
 		_, err := (&Builder{}).
 			Scan(&mockTableProvider{dynparquet.NewSampleSchema()}, "table1").
-			Aggregate([]Expr{testCase.fn(Col("example_type"))}, nil).
+			Aggregate([]*AggregationFunction{testCase.fn(Col("example_type"))}, nil).
 			Build()
 
 		require.NotNil(t, err)
@@ -137,24 +115,6 @@ func TestAggregationCannotSumOrMaxTextColumn(t *testing.T) {
 	}
 }
 
-func TestAggregationCannotUseAliasTwice(t *testing.T) {
-	_, err := (&Builder{}).
-		Scan(&mockTableProvider{dynparquet.NewSampleSchema()}, "table1").
-		Aggregate([]Expr{
-			Sum(Col("value")).Alias("value"), // should use e.g. sum_foo
-			Max(Col("value")).Alias("value"), // should use e.g. max_foo
-		}, nil).
-		Build()
-
-	require.NotNil(t, err)
-	planErr, ok := err.(*PlanValidationError)
-	require.True(t, ok)
-	require.True(t, strings.HasPrefix(planErr.message, "invalid aggregation"))
-	require.Len(t, planErr.children, 1)
-	exprErr := planErr.children[0]
-	require.True(t, strings.HasPrefix(exprErr.message, "alias used twice: value"))
-}
-
 func TestFilterBinaryExprLeftSideMustBeColumn(t *testing.T) {
 	_, err := (&Builder{}).
 		Scan(&mockTableProvider{dynparquet.NewSampleSchema()}, "table1").
diff --git a/query/physicalplan/physicalplan_test.go b/query/physicalplan/physicalplan_test.go
index 307c5b9..b7e84b9 100644
--- a/query/physicalplan/physicalplan_test.go
+++ b/query/physicalplan/physicalplan_test.go
@@ -60,10 +60,10 @@ func TestBuildPhysicalPlan(t *testing.T) {
 		Scan(&mockTableProvider{schema: dynparquet.NewSampleSchema()}, "table1").
 		Filter(logicalplan.Col("labels.test").Eq(logicalplan.Literal("abc"))).
 		Aggregate(
-			[]logicalplan.Expr{logicalplan.Sum(logicalplan.Col("value")).Alias("value_sum")},
+			[]*logicalplan.AggregationFunction{logicalplan.Sum(logicalplan.Col("value"))},
 			[]logicalplan.Expr{logicalplan.Col("stacktrace")},
 		).
-		Project(logicalplan.Col("stacktrace"), logicalplan.Col("value_sum")).
+		Project(logicalplan.Col("stacktrace"), logicalplan.Sum(logicalplan.Col("value")).Alias("value_sum")).
 		Build()
 
 	optimizers := []logicalplan.Optimizer{
diff --git a/snapshot_test.go b/snapshot_test.go
index 244e4b5..381369f 100644
--- a/snapshot_test.go
+++ b/snapshot_test.go
@@ -145,7 +145,7 @@ func TestSnapshot(t *testing.T) {
 			},
 		} {
 			if testCase.expMaxTimestamp != 0 {
-				max := []logicalplan.Expr{
+				max := []*logicalplan.AggregationFunction{
 					logicalplan.Max(logicalplan.Col("timestamp")),
 				}
 				require.NoError(
@@ -220,7 +220,7 @@ func TestSnapshot(t *testing.T) {
 			query.NewEngine(
 				memory.DefaultAllocator, snapshotDB.TableProvider(),
 			).ScanTable(tableName).Aggregate(
-				[]logicalplan.Expr{logicalplan.Max(logicalplan.Col("timestamp"))}, nil,
+				[]*logicalplan.AggregationFunction{logicalplan.Max(logicalplan.Col("timestamp"))}, nil,
 			).Execute(ctx, func(ctx context.Context, r arrow.Record) error {
 				require.Equal(
 					t, int(snapshotTx-highWatermarkAtStart), int(r.Column(0).(*array.Int64).Int64Values()[0]),
@@ -318,7 +318,7 @@ func TestSnapshotWithWAL(t *testing.T) {
 		t,
 		engine.ScanTable(dbAndTableName).
 			Aggregate(
-				[]logicalplan.Expr{logicalplan.Min(logicalplan.Col("timestamp"))},
+				[]*logicalplan.AggregationFunction{logicalplan.Min(logicalplan.Col("timestamp"))},
 				nil,
 			).Execute(ctx, func(_ context.Context, r arrow.Record) error {
 			// This check verifies that the snapshot data (i.e. the first
@@ -331,7 +331,7 @@ func TestSnapshotWithWAL(t *testing.T) {
 		t,
 		engine.ScanTable(dbAndTableName).
 			Aggregate(
-				[]logicalplan.Expr{logicalplan.Max(logicalplan.Col("timestamp"))},
+				[]*logicalplan.AggregationFunction{logicalplan.Max(logicalplan.Col("timestamp"))},
 				nil,
 			).Execute(ctx, func(_ context.Context, r arrow.Record) error {
 			// This check verifies that the write that is only represented in
diff --git a/table_test.go b/table_test.go
index 127700f..9923319 100644
--- a/table_test.go
+++ b/table_test.go
@@ -891,7 +891,7 @@ func Test_Insert_Repeated(t *testing.T) {
 			converter := pqarrow.NewParquetConverter(memory.NewGoAllocator(), logicalplan.IterOptions{})
 			defer converter.Close()
 
-			require.NoError(t, converter.Convert(ctx, buffer))
+			require.NoError(t, converter.Convert(ctx, buffer, table.Schema()))
 			record := converter.NewRecord()
 			defer record.Release()
 
@@ -916,7 +916,7 @@ func Test_Insert_Repeated(t *testing.T) {
 			engine := query.NewEngine(memory.NewGoAllocator(), db.TableProvider())
 			err = engine.ScanTable("test").
 				Aggregate(
-					[]logicalplan.Expr{logicalplan.Sum(logicalplan.Col("value"))},
+					[]*logicalplan.AggregationFunction{logicalplan.Sum(logicalplan.Col("value"))},
 					[]logicalplan.Expr{logicalplan.Col("values")},
 				).
 				Execute(context.Background(), func(ctx context.Context, r arrow.Record) error {
@@ -1004,7 +1004,7 @@ func Test_Compact_Repeated(t *testing.T) {
 	converter := pqarrow.NewParquetConverter(memory.NewGoAllocator(), logicalplan.IterOptions{})
 	defer converter.Close()
 
-	require.NoError(t, converter.Convert(ctx, buffer))
+	require.NoError(t, converter.Convert(ctx, buffer, table.Schema()))
 	before := converter.NewRecord()
 	defer before.Release()
 
@@ -1210,7 +1210,7 @@ func Test_Issue685(t *testing.T) {
 	engine := query.NewEngine(memory.NewGoAllocator(), db.TableProvider())
 	err = engine.ScanTable("test").
 		Aggregate(
-			[]logicalplan.Expr{
+			[]*logicalplan.AggregationFunction{
 				logicalplan.Sum(logicalplan.Col("value.age")),
 			},
 			nil,