From e24978c4a76713cb816d0810a3ace1a797443b6c Mon Sep 17 00:00:00 2001 From: pokeeffe-molecula <85502298+pokeeffe-molecula@users.noreply.github.com> Date: Tue, 27 Sep 2022 09:27:37 -0500 Subject: [PATCH] And now....INNER JOIN! (#2230) * ID sql3 internal type representation is int64; fixed a bug that assumed incorrectly that it wasn't * refactored some names for clarity * primary: get nested loop joins to work; secondary get brute force aggregations for SUM working * added tests; removed debug output * review feedback * Update sql3/planner/compileselect.go review feedback Co-authored-by: Travis Turner Co-authored-by: Travis Turner --- http_handler.go | 2 +- sql3/errors.go | 18 +++ sql3/planner/compileselect.go | 84 +++++++++- sql3/planner/executionplanner_test.go | 214 +++++++++++++------------- sql3/planner/expression.go | 2 +- sql3/planner/expressionagg.go | 46 +++--- sql3/planner/memoryobj.go | 68 ++++++-- sql3/planner/opbulkinsert.go | 2 +- sql3/planner/opcreatetable.go | 2 +- sql3/planner/opdroptable.go | 2 +- sql3/planner/opfeaturebasecolumns.go | 86 +++++------ sql3/planner/opfeaturebasetables.go | 32 ++-- sql3/planner/opfilter.go | 2 +- sql3/planner/opgroupby.go | 198 ++++++++++++++++++++---- sql3/planner/opinsert.go | 4 +- sql3/planner/opnestedloops.go | 32 ++-- sql3/planner/opnulltable.go | 2 +- sql3/planner/oporderby.go | 2 +- sql3/planner/oppqlaggregate.go | 8 +- sql3/planner/oppqlgroupby.go | 14 +- sql3/planner/oppqlmultiaggregate.go | 8 +- sql3/planner/oppqlmultigroupby.go | 30 ++-- sql3/planner/oppqltablescan.go | 123 +++++++-------- sql3/planner/opprojection.go | 8 +- sql3/planner/opquery.go | 2 +- sql3/planner/oprelalias.go | 8 +- sql3/planner/opsubquery.go | 2 +- sql3/planner/optablevaluedfunction.go | 8 +- sql3/planner/optop.go | 2 +- sql3/planner/planoptimizer.go | 81 +++++----- sql3/planner/planwalker.go | 4 +- sql3/planner/types/operator.go | 11 +- sql3/sql_definitions_test.go | 5 + sql3/sql_defs_join_test.go | 65 ++++++++ sql3/sql_test.go | 8 +- sql3/test/helpers.go | 4 +- 36 files changed, 770 insertions(+), 419 deletions(-) create mode 100644 sql3/sql_defs_join_test.go diff --git a/http_handler.go b/http_handler.go index 652dcf895..95e85f71e 100644 --- a/http_handler.go +++ b/http_handler.go @@ -1439,7 +1439,7 @@ func (h *Handler) handlePostSQL(w http.ResponseWriter, r *http.Request) { } for i, col := range columns { schema.Fields[i] = &SQLField{ - Name: col.Name, + Name: col.ColumnName, Type: col.Type.TypeName(), } } diff --git a/sql3/errors.go b/sql3/errors.go index 4e03873ee..e5c388bd9 100644 --- a/sql3/errors.go +++ b/sql3/errors.go @@ -11,6 +11,8 @@ import ( const ( ErrInternal errors.Code = "ErrInternal" + ErrCacheKeyNotFound errors.Code = "ErrCacheKeyNotFound" + ErrDuplicateColumn errors.Code = "ErrDuplicateColumn" ErrUnknownType errors.Code = "ErrUnknownType" @@ -47,6 +49,8 @@ const ( ErrTypeAssignmentIncompatible errors.Code = "ErrTypeAssignmentIncompatible" + ErrInvalidUngroupedColumnReference errors.Code = "ErrInvalidUngroupedColumnReference" + ErrInvalidTimeUnit errors.Code = "ErrInvalidTimeUnit" ErrInvalidTimeEpoch errors.Code = "ErrInvalidTimeEpoch" ErrInvalidTimeQuantum errors.Code = "ErrInvalidTimeQuantum" @@ -124,6 +128,13 @@ func NewErrInternalf(format string, a ...interface{}) error { ) } +func NewErrCacheKeyNotFound(key uint64) error { + return errors.New( + ErrCacheKeyNotFound, + fmt.Sprintf("key '%d' not found", key), + ) +} + func NewErrTypeAssignmentIncompatible(line, col int, type1, type2 string) error { return errors.New( ErrTypeAssignmentIncompatible, @@ -131,6 +142,13 @@ func NewErrTypeAssignmentIncompatible(line, col int, type1, type2 string) error ) } +func NewErrInvalidUngroupedColumnReference(line, col int, column string) error { + return errors.New( + ErrInvalidUngroupedColumnReference, + fmt.Sprintf("[%d:%d] column '%s' invalid in select list because it is not aggregated or grouped", line, col, column), + ) +} + func NewErrInvalidCast(line, col int, from, to string) error { return errors.New( ErrInvalidCast, diff --git a/sql3/planner/compileselect.go b/sql3/planner/compileselect.go index 8d80170ab..6b99cb5cb 100644 --- a/sql3/planner/compileselect.go +++ b/sql3/planner/compileselect.go @@ -72,6 +72,39 @@ func (p *ExecutionPlanner) compileSelectStatement(stmt *parser.SelectStatement, // do we have straight projection or a group by? var compiledOp types.PlanOperator if len(query.aggregates) > 0 { + //check that any projections that are not aggregates are in the group by list + var nonAggregateReferences []*qualifiedRefPlanExpression + for _, expr := range projections { + InspectExpression(expr, func(expr types.PlanExpression) bool { + switch ex := expr.(type) { + case *sumPlanExpression: + return false + case *qualifiedRefPlanExpression: + nonAggregateReferences = append(nonAggregateReferences, ex) + return false + } + return true + }) + } + + for _, nae := range nonAggregateReferences { + found := false + for _, pe := range groupByExprs { + gbe, ok := pe.(*qualifiedRefPlanExpression) + if !ok { + continue + } + if strings.EqualFold(nae.columnName, gbe.columnName) && + strings.EqualFold(nae.tableName, gbe.tableName) { + found = true + break + } + } + if !found { + return nil, sql3.NewErrInvalidUngroupedColumnReference(0, 0, nae.columnName) + } + } + compiledOp = NewPlanOpProjection(projections, NewPlanOpGroupBy(query.aggregates, groupByExprs, source)) } else { compiledOp = NewPlanOpProjection(projections, source) @@ -126,6 +159,25 @@ func (p *ExecutionPlanner) compileSelectSource(scope *PlanOpQuery, source parser switch sourceExpr := source.(type) { case *parser.JoinClause: + scope.AddWarning("🦖 here there be dragons! JOINS are experimental.") + + var joinCondition types.PlanExpression + if sourceExpr.Constraint == nil { + scope.AddWarning("⚠️ cartesian products are never a good idea - are you missing a join constraint?") + joinCondition = nil + } else { + switch join := sourceExpr.Constraint.(type) { + case *parser.OnConstraint: + expr, err := p.compileExpr(join.X) + if err != nil { + return nil, err + } + joinCondition = expr + default: + return nil, sql3.NewErrInternalf("unexecpted constraint type '%T'", join) + } + } + topOp, err := p.compileSelectSource(scope, sourceExpr.X) if err != nil { return nil, err @@ -134,19 +186,23 @@ func (p *ExecutionPlanner) compileSelectSource(scope *PlanOpQuery, source parser if err != nil { return nil, err } - scope.AddWarning("🦖 here there be dragons! JOINS are experimental.") - if sourceExpr.Constraint == nil { - scope.AddWarning("⚠️ cartesian products are never a good idea - are you missing a join constraint?") - } - return NewPlanOpNestedLoops(topOp, bottomOp), nil + return NewPlanOpNestedLoops(topOp, bottomOp, joinCondition), nil case *parser.QualifiedTableName: // get all the qualified refs that refer to this table - extractColumns := []types.PlanExpression{} - + extractColumns := make([]string, 0) for _, r := range scope.referenceList { if sourceExpr.MatchesTablenameOrAlias(r.tableName) { - extractColumns = append(extractColumns, r) + found := false + for _, c := range extractColumns { + if strings.EqualFold(c, r.columnName) { + found = true + break + } + } + if !found { + extractColumns = append(extractColumns, r.columnName) + } } } @@ -210,6 +266,18 @@ func (p *ExecutionPlanner) analyzeSource(source parser.Source, scope parser.Stat if err != nil { return err } + if source.Constraint != nil { + switch join := source.Constraint.(type) { + case *parser.OnConstraint: + ex, err := p.analyzeExpression(join.X, scope) + if err != nil { + return err + } + join.X = ex + default: + return sql3.NewErrInternalf("unexpected constraint type '%T'", join) + } + } return nil case *parser.ParenSource: diff --git a/sql3/planner/executionplanner_test.go b/sql3/planner/executionplanner_test.go index 1cfb719e7..a815d885e 100644 --- a/sql3/planner/executionplanner_test.go +++ b/sql3/planner/executionplanner_test.go @@ -54,11 +54,11 @@ func TestPlanner_Show(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "name", Type: parser.NewDataTypeString()}, - {Name: "created_at", Type: parser.NewDataTypeTimestamp()}, - {Name: "track_existence", Type: parser.NewDataTypeBool()}, - {Name: "keys", Type: parser.NewDataTypeBool()}, - {Name: "shard_width", Type: parser.NewDataTypeInt()}, + {ColumnName: "name", Type: parser.NewDataTypeString()}, + {ColumnName: "created_at", Type: parser.NewDataTypeTimestamp()}, + {ColumnName: "track_existence", Type: parser.NewDataTypeBool()}, + {ColumnName: "keys", Type: parser.NewDataTypeBool()}, + {ColumnName: "shard_width", Type: parser.NewDataTypeInt()}, }, columns); diff != "" { t.Fatal(diff) } @@ -74,20 +74,20 @@ func TestPlanner_Show(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "name", Type: parser.NewDataTypeString()}, - {Name: "type", Type: parser.NewDataTypeString()}, - {Name: "internal_type", Type: parser.NewDataTypeString()}, - {Name: "created_at", Type: parser.NewDataTypeTimestamp()}, - {Name: "keys", Type: parser.NewDataTypeBool()}, - {Name: "cache_type", Type: parser.NewDataTypeString()}, - {Name: "cache_size", Type: parser.NewDataTypeInt()}, - {Name: "scale", Type: parser.NewDataTypeInt()}, - {Name: "min", Type: parser.NewDataTypeInt()}, - {Name: "max", Type: parser.NewDataTypeInt()}, - {Name: "timeunit", Type: parser.NewDataTypeString()}, - {Name: "epoch", Type: parser.NewDataTypeInt()}, - {Name: "timequantum", Type: parser.NewDataTypeString()}, - {Name: "ttl", Type: parser.NewDataTypeString()}, + {ColumnName: "name", Type: parser.NewDataTypeString()}, + {ColumnName: "type", Type: parser.NewDataTypeString()}, + {ColumnName: "internal_type", Type: parser.NewDataTypeString()}, + {ColumnName: "created_at", Type: parser.NewDataTypeTimestamp()}, + {ColumnName: "keys", Type: parser.NewDataTypeBool()}, + {ColumnName: "cache_type", Type: parser.NewDataTypeString()}, + {ColumnName: "cache_size", Type: parser.NewDataTypeInt()}, + {ColumnName: "scale", Type: parser.NewDataTypeInt()}, + {ColumnName: "min", Type: parser.NewDataTypeInt()}, + {ColumnName: "max", Type: parser.NewDataTypeInt()}, + {ColumnName: "timeunit", Type: parser.NewDataTypeString()}, + {ColumnName: "epoch", Type: parser.NewDataTypeInt()}, + {ColumnName: "timequantum", Type: parser.NewDataTypeString()}, + {ColumnName: "ttl", Type: parser.NewDataTypeString()}, }, columns); diff != "" { t.Fatal(diff) } @@ -103,20 +103,20 @@ func TestPlanner_Show(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "name", Type: parser.NewDataTypeString()}, - {Name: "type", Type: parser.NewDataTypeString()}, - {Name: "internal_type", Type: parser.NewDataTypeString()}, - {Name: "created_at", Type: parser.NewDataTypeTimestamp()}, - {Name: "keys", Type: parser.NewDataTypeBool()}, - {Name: "cache_type", Type: parser.NewDataTypeString()}, - {Name: "cache_size", Type: parser.NewDataTypeInt()}, - {Name: "scale", Type: parser.NewDataTypeInt()}, - {Name: "min", Type: parser.NewDataTypeInt()}, - {Name: "max", Type: parser.NewDataTypeInt()}, - {Name: "timeunit", Type: parser.NewDataTypeString()}, - {Name: "epoch", Type: parser.NewDataTypeInt()}, - {Name: "timequantum", Type: parser.NewDataTypeString()}, - {Name: "ttl", Type: parser.NewDataTypeString()}, + {ColumnName: "name", Type: parser.NewDataTypeString()}, + {ColumnName: "type", Type: parser.NewDataTypeString()}, + {ColumnName: "internal_type", Type: parser.NewDataTypeString()}, + {ColumnName: "created_at", Type: parser.NewDataTypeTimestamp()}, + {ColumnName: "keys", Type: parser.NewDataTypeBool()}, + {ColumnName: "cache_type", Type: parser.NewDataTypeString()}, + {ColumnName: "cache_size", Type: parser.NewDataTypeInt()}, + {ColumnName: "scale", Type: parser.NewDataTypeInt()}, + {ColumnName: "min", Type: parser.NewDataTypeInt()}, + {ColumnName: "max", Type: parser.NewDataTypeInt()}, + {ColumnName: "timeunit", Type: parser.NewDataTypeString()}, + {ColumnName: "epoch", Type: parser.NewDataTypeInt()}, + {ColumnName: "timequantum", Type: parser.NewDataTypeString()}, + {ColumnName: "ttl", Type: parser.NewDataTypeString()}, }, columns); diff != "" { t.Fatal(diff) } @@ -483,20 +483,20 @@ func TestPlanner_CreateTable(t *testing.T) { t.Fatal(err) } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "name", Type: parser.NewDataTypeString()}, - {Name: "type", Type: parser.NewDataTypeString()}, - {Name: "internal_type", Type: parser.NewDataTypeString()}, - {Name: "created_at", Type: parser.NewDataTypeTimestamp()}, - {Name: "keys", Type: parser.NewDataTypeBool()}, - {Name: "cache_type", Type: parser.NewDataTypeString()}, - {Name: "cache_size", Type: parser.NewDataTypeInt()}, - {Name: "scale", Type: parser.NewDataTypeInt()}, - {Name: "min", Type: parser.NewDataTypeInt()}, - {Name: "max", Type: parser.NewDataTypeInt()}, - {Name: "timeunit", Type: parser.NewDataTypeString()}, - {Name: "epoch", Type: parser.NewDataTypeInt()}, - {Name: "timequantum", Type: parser.NewDataTypeString()}, - {Name: "ttl", Type: parser.NewDataTypeString()}, + {ColumnName: "name", Type: parser.NewDataTypeString()}, + {ColumnName: "type", Type: parser.NewDataTypeString()}, + {ColumnName: "internal_type", Type: parser.NewDataTypeString()}, + {ColumnName: "created_at", Type: parser.NewDataTypeTimestamp()}, + {ColumnName: "keys", Type: parser.NewDataTypeBool()}, + {ColumnName: "cache_type", Type: parser.NewDataTypeString()}, + {ColumnName: "cache_size", Type: parser.NewDataTypeInt()}, + {ColumnName: "scale", Type: parser.NewDataTypeInt()}, + {ColumnName: "min", Type: parser.NewDataTypeInt()}, + {ColumnName: "max", Type: parser.NewDataTypeInt()}, + {ColumnName: "timeunit", Type: parser.NewDataTypeString()}, + {ColumnName: "epoch", Type: parser.NewDataTypeInt()}, + {ColumnName: "timequantum", Type: parser.NewDataTypeString()}, + {ColumnName: "ttl", Type: parser.NewDataTypeString()}, }, columns); diff != "" { t.Fatal(diff) } @@ -665,8 +665,8 @@ func TestPlanner_ExpressionsInSelectListParen(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "", Type: parser.NewDataTypeBool()}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "", Type: parser.NewDataTypeBool()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -686,8 +686,8 @@ func TestPlanner_ExpressionsInSelectListParen(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "", Type: parser.NewDataTypeBool()}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "", Type: parser.NewDataTypeBool()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -744,8 +744,8 @@ func TestPlanner_ExpressionsInSelectListLiterals(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "", Type: parser.NewDataTypeBool()}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "", Type: parser.NewDataTypeBool()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -765,8 +765,8 @@ func TestPlanner_ExpressionsInSelectListLiterals(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "", Type: parser.NewDataTypeInt()}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -786,8 +786,8 @@ func TestPlanner_ExpressionsInSelectListLiterals(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "", Type: parser.NewDataTypeInt()}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -807,8 +807,8 @@ func TestPlanner_ExpressionsInSelectListLiterals(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "", Type: parser.NewDataTypeDecimal(2)}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "", Type: parser.NewDataTypeDecimal(2)}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -828,8 +828,8 @@ func TestPlanner_ExpressionsInSelectListLiterals(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "", Type: parser.NewDataTypeString()}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "", Type: parser.NewDataTypeString()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -886,9 +886,9 @@ func TestPlanner_ExpressionsInSelectListCase(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "b", Type: parser.NewDataTypeInt()}, - {Name: "", Type: parser.NewDataTypeInt()}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "b", Type: parser.NewDataTypeInt()}, + {ColumnName: "", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -908,9 +908,9 @@ func TestPlanner_ExpressionsInSelectListCase(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "b", Type: parser.NewDataTypeInt()}, - {Name: "", Type: parser.NewDataTypeInt()}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "b", Type: parser.NewDataTypeInt()}, + {ColumnName: "", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -969,9 +969,9 @@ func TestPlanner_Select(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "a", Type: parser.NewDataTypeInt()}, - {Name: "b", Type: parser.NewDataTypeInt()}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "a", Type: parser.NewDataTypeInt()}, + {ColumnName: "b", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -991,9 +991,9 @@ func TestPlanner_Select(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "a", Type: parser.NewDataTypeInt()}, - {Name: "b", Type: parser.NewDataTypeInt()}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "a", Type: parser.NewDataTypeInt()}, + {ColumnName: "b", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1013,9 +1013,9 @@ func TestPlanner_Select(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "foo", Type: parser.NewDataTypeInt()}, - {Name: "bar", Type: parser.NewDataTypeInt()}, - {Name: "baz", Type: parser.NewDataTypeID()}, + {ColumnName: "foo", Type: parser.NewDataTypeInt()}, + {ColumnName: "bar", Type: parser.NewDataTypeInt()}, + {ColumnName: "baz", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1035,9 +1035,9 @@ func TestPlanner_Select(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "_id", Type: parser.NewDataTypeID()}, - {Name: "a", Type: parser.NewDataTypeInt()}, - {Name: "b", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "a", Type: parser.NewDataTypeInt()}, + {ColumnName: "b", Type: parser.NewDataTypeInt()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1057,9 +1057,9 @@ func TestPlanner_Select(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "_id", Type: parser.NewDataTypeID()}, - {Name: "a", Type: parser.NewDataTypeInt()}, - {Name: "b", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "a", Type: parser.NewDataTypeInt()}, + {ColumnName: "b", Type: parser.NewDataTypeInt()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1079,9 +1079,9 @@ func TestPlanner_Select(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "_id", Type: parser.NewDataTypeID()}, - {Name: "a", Type: parser.NewDataTypeInt()}, - {Name: "b", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "a", Type: parser.NewDataTypeInt()}, + {ColumnName: "b", Type: parser.NewDataTypeInt()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1101,8 +1101,8 @@ func TestPlanner_Select(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "a", Type: parser.NewDataTypeInt()}, - {Name: "b", Type: parser.NewDataTypeInt()}, + {ColumnName: "a", Type: parser.NewDataTypeInt()}, + {ColumnName: "b", Type: parser.NewDataTypeInt()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1157,9 +1157,9 @@ func TestPlanner_SelectOrderBy(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "a", Type: parser.NewDataTypeInt()}, - {Name: "b", Type: parser.NewDataTypeInt()}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "a", Type: parser.NewDataTypeInt()}, + {ColumnName: "b", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1207,9 +1207,9 @@ func TestPlanner_SelectSelectSource(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "a", Type: parser.NewDataTypeInt()}, - {Name: "b", Type: parser.NewDataTypeInt()}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "a", Type: parser.NewDataTypeInt()}, + {ColumnName: "b", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1229,9 +1229,9 @@ func TestPlanner_SelectSelectSource(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "a", Type: parser.NewDataTypeInt()}, - {Name: "b", Type: parser.NewDataTypeInt()}, - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "a", Type: parser.NewDataTypeInt()}, + {ColumnName: "b", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1304,7 +1304,7 @@ func TestPlanner_In(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "count", Type: parser.NewDataTypeInt()}, + {ColumnName: "count", Type: parser.NewDataTypeInt()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1434,7 +1434,7 @@ func TestPlanner_Distinct(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1456,7 +1456,7 @@ func TestPlanner_Distinct(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "parentid", Type: parser.NewDataTypeInt()}, + {ColumnName: "parentid", Type: parser.NewDataTypeInt()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1477,8 +1477,8 @@ func TestPlanner_Distinct(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "_id", Type: parser.NewDataTypeID()}, - {Name: "parentid", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "parentid", Type: parser.NewDataTypeInt()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1529,9 +1529,9 @@ func TestPlanner_SelectTop(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "_id", Type: parser.NewDataTypeID()}, - {Name: "a", Type: parser.NewDataTypeInt()}, - {Name: "b", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "a", Type: parser.NewDataTypeInt()}, + {ColumnName: "b", Type: parser.NewDataTypeInt()}, }, columns); diff != "" { t.Fatal(diff) } @@ -1552,9 +1552,9 @@ func TestPlanner_SelectTop(t *testing.T) { } if diff := cmp.Diff([]*planner_types.PlannerColumn{ - {Name: "_id", Type: parser.NewDataTypeID()}, - {Name: "a", Type: parser.NewDataTypeInt()}, - {Name: "b", Type: parser.NewDataTypeInt()}, + {ColumnName: "_id", Type: parser.NewDataTypeID()}, + {ColumnName: "a", Type: parser.NewDataTypeInt()}, + {ColumnName: "b", Type: parser.NewDataTypeInt()}, }, columns); diff != "" { t.Fatal(diff) } diff --git a/sql3/planner/expression.go b/sql3/planner/expression.go index d78c2a8c0..cc839f841 100644 --- a/sql3/planner/expression.go +++ b/sql3/planner/expression.go @@ -1498,7 +1498,7 @@ func (n *qualifiedRefPlanExpression) Evaluate(currentRow []interface{}) (interfa return result, nil case *parser.DataTypeID: - //TODO(pok) why are we trying two underlying types here? + //this could be an int64 or a uint64 internally iv, iok := currentRow[n.columnIndex].(int64) if iok { return iv, nil diff --git a/sql3/planner/expressionagg.go b/sql3/planner/expressionagg.go index 1d74e041f..b6a1a0ecd 100644 --- a/sql3/planner/expressionagg.go +++ b/sql3/planner/expressionagg.go @@ -7,9 +7,10 @@ import ( "fmt" "reflect" - "github.com/featurebasedb/featurebase/v3/sql3" - "github.com/featurebasedb/featurebase/v3/sql3/parser" - "github.com/featurebasedb/featurebase/v3/sql3/planner/types" + "github.com/molecula/featurebase/v3/pql" + "github.com/molecula/featurebase/v3/sql3" + "github.com/molecula/featurebase/v3/sql3/parser" + "github.com/molecula/featurebase/v3/sql3/planner/types" ) // aggregator for the COUNT function @@ -196,13 +197,15 @@ func (n *countDistinctPlanExpression) WithChildren(children ...types.PlanExpress // aggregator for the SUM function type aggregateSum struct { - isnil bool - sum float64 - expr types.PlanExpression + sum float64 + expr types.PlanExpression } func NewAggSumBuffer(child types.PlanExpression) *aggregateSum { - return &aggregateSum{true, float64(0), child} + return &aggregateSum{ + sum: float64(0), + expr: child, + } } func (m *aggregateSum) Update(ctx context.Context, row types.Row) error { @@ -211,27 +214,30 @@ func (m *aggregateSum) Update(ctx context.Context, row types.Row) error { return err } + //if null, skip if v == nil { return nil } - var val interface{} = 0 - - if m.isnil { - m.sum = 0 - m.isnil = false + sumExpr, ok := m.expr.(*sumPlanExpression) + if !ok { + return sql3.NewErrInternalf("unexpected aggregate expression type '%T'", m.expr) } - m.sum += val.(float64) - - //return nil - return sql3.NewErrInternalf("implement me") + switch dataType := sumExpr.arg.Type().(type) { + case *parser.DataTypeDecimal: + val, ok := v.(pql.Decimal) + if !ok { + return sql3.NewErrInternalf("unexpected type conversion '%T'", v) + } + m.sum += val.Float64() + default: + return sql3.NewErrInternalf("unhandled aggregate expression datatype '%T'", dataType) + } + return nil } func (m *aggregateSum) Eval(ctx context.Context) (interface{}, error) { - if m.isnil { - return nil, nil - } return m.sum, nil } @@ -646,7 +652,7 @@ func (n *percentilePlanExpression) WithChildren(children ...types.PlanExpression return n, nil } -//aggregator for last +// aggregator for last type aggregateLast struct { val interface{} expr types.PlanExpression diff --git a/sql3/planner/memoryobj.go b/sql3/planner/memoryobj.go index a39117465..cdbfdcf35 100644 --- a/sql3/planner/memoryobj.go +++ b/sql3/planner/memoryobj.go @@ -2,7 +2,26 @@ package planner -import "github.com/featurebasedb/featurebase/v3/sql3/planner/types" +import ( + "hash/maphash" + + "github.com/molecula/featurebase/v3/sql3" + "github.com/molecula/featurebase/v3/sql3/planner/types" +) + +var prototypeHash maphash.Hash + +// ObjectCache is a cache of interface{} values +type ObjectCache interface { + // Put a new value in the cache + PutObject(uint64, interface{}) error + + // Get the value with the given key + GetObject(uint64) (interface{}, error) + + // Size returns the number of values in the cache + Size() int +} // RowCache is a cache of rows used during row iteration type RowCache interface { @@ -15,35 +34,35 @@ type RowCache interface { // KeyedRowCache is a cache of keyed rows used during row iteration type KeyedRowCache interface { // Put adds row to the cache at the given key. - Put(key string, row types.Row) error + Put(key uint64, row types.Row) error // Get returns the rows specified by key. - Get(key string) (types.Row, error) + Get(key uint64) (types.Row, error) // Size returns the number of rows in the cache. Size() int } -// Ensure type implements interface. +// Ensure type implements interface var _ KeyedRowCache = (*inMemoryKeyedRowCache)(nil) // default implementation of KeyedRowCache (in memory) type inMemoryKeyedRowCache struct { - store map[string][]interface{} + store map[uint64][]interface{} } func newinMemoryKeyedRowCache() *inMemoryKeyedRowCache { return &inMemoryKeyedRowCache{ - store: make(map[string][]interface{}), + store: make(map[uint64][]interface{}), } } -func (m inMemoryKeyedRowCache) Put(u string, i types.Row) error { +func (m inMemoryKeyedRowCache) Put(u uint64, i types.Row) error { m.store[u] = i return nil } -func (m inMemoryKeyedRowCache) Get(u string) (types.Row, error) { +func (m inMemoryKeyedRowCache) Get(u uint64) (types.Row, error) { return m.store[u], nil } @@ -51,7 +70,7 @@ func (m inMemoryKeyedRowCache) Size() int { return len(m.store) } -// Ensure type implements interface. +// Ensure type implements interface var _ RowCache = (*inMemoryRowCache)(nil) type inMemoryRowCache struct { @@ -70,3 +89,34 @@ func (c *inMemoryRowCache) Add(row types.Row) error { func (c *inMemoryRowCache) AllRows() []types.Row { return c.rows } + +// Ensure type implements interface +var _ ObjectCache = (*mapObjectCache)(nil) + +// mapObjectCache is a simple in-memory implementation of a cache +type mapObjectCache struct { + cache map[uint64]interface{} +} + +func (m mapObjectCache) PutObject(u uint64, i interface{}) error { + m.cache[u] = i + return nil +} + +func (m mapObjectCache) GetObject(u uint64) (interface{}, error) { + v, ok := m.cache[u] + if !ok { + return nil, sql3.NewErrCacheKeyNotFound(u) + } + return v, nil +} + +func (m mapObjectCache) Size() int { + return len(m.cache) +} + +func NewMapObjectCache() mapObjectCache { + return mapObjectCache{ + cache: make(map[uint64]interface{}), + } +} diff --git a/sql3/planner/opbulkinsert.go b/sql3/planner/opbulkinsert.go index 2809f76ad..4522bb25b 100644 --- a/sql3/planner/opbulkinsert.go +++ b/sql3/planner/opbulkinsert.go @@ -72,7 +72,7 @@ func (p *PlanOpBulkInsert) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) sc := make([]string, 0) for _, e := range p.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc result["tableName"] = p.tableName diff --git a/sql3/planner/opcreatetable.go b/sql3/planner/opcreatetable.go index 29ed595b9..cd946b71a 100644 --- a/sql3/planner/opcreatetable.go +++ b/sql3/planner/opcreatetable.go @@ -39,7 +39,7 @@ func (p *PlanOpCreateTable) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) ps := make([]string, 0) for _, e := range p.Schema() { - ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = ps result["name"] = p.tableName diff --git a/sql3/planner/opdroptable.go b/sql3/planner/opdroptable.go index 8a80d4ed8..4c8f6e273 100644 --- a/sql3/planner/opdroptable.go +++ b/sql3/planner/opdroptable.go @@ -30,7 +30,7 @@ func (p *PlanOpDropTable) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) ps := make([]string, 0) for _, e := range p.Schema() { - ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = ps result["tableName"] = p.index.Name diff --git a/sql3/planner/opfeaturebasecolumns.go b/sql3/planner/opfeaturebasecolumns.go index a4ab8deab..aacac1434 100644 --- a/sql3/planner/opfeaturebasecolumns.go +++ b/sql3/planner/opfeaturebasecolumns.go @@ -31,7 +31,7 @@ func (p *PlanOpFeatureBaseColumns) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) ps := make([]string, 0) for _, e := range p.Schema() { - ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = ps return result @@ -52,74 +52,74 @@ func (p *PlanOpFeatureBaseColumns) Warnings() []string { func (p *PlanOpFeatureBaseColumns) Schema() types.Schema { return types.Schema{ &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "name", - Type: parser.NewDataTypeString(), + RelationName: "fb$table_columns", + ColumnName: "name", + Type: parser.NewDataTypeString(), }, &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "type", - Type: parser.NewDataTypeString(), + RelationName: "fb$table_columns", + ColumnName: "type", + Type: parser.NewDataTypeString(), }, &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "internal_type", - Type: parser.NewDataTypeString(), + RelationName: "fb$table_columns", + ColumnName: "internal_type", + Type: parser.NewDataTypeString(), }, &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "created_at", - Type: parser.NewDataTypeTimestamp(), + RelationName: "fb$table_columns", + ColumnName: "created_at", + Type: parser.NewDataTypeTimestamp(), }, &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "keys", - Type: parser.NewDataTypeBool(), + RelationName: "fb$table_columns", + ColumnName: "keys", + Type: parser.NewDataTypeBool(), }, &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "cache_type", - Type: parser.NewDataTypeString(), + RelationName: "fb$table_columns", + ColumnName: "cache_type", + Type: parser.NewDataTypeString(), }, &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "cache_size", - Type: parser.NewDataTypeInt(), + RelationName: "fb$table_columns", + ColumnName: "cache_size", + Type: parser.NewDataTypeInt(), }, &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "scale", - Type: parser.NewDataTypeInt(), + RelationName: "fb$table_columns", + ColumnName: "scale", + Type: parser.NewDataTypeInt(), }, &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "min", - Type: parser.NewDataTypeInt(), + RelationName: "fb$table_columns", + ColumnName: "min", + Type: parser.NewDataTypeInt(), }, &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "max", - Type: parser.NewDataTypeInt(), + RelationName: "fb$table_columns", + ColumnName: "max", + Type: parser.NewDataTypeInt(), }, &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "timeunit", - Type: parser.NewDataTypeString(), + RelationName: "fb$table_columns", + ColumnName: "timeunit", + Type: parser.NewDataTypeString(), }, &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "epoch", - Type: parser.NewDataTypeInt(), + RelationName: "fb$table_columns", + ColumnName: "epoch", + Type: parser.NewDataTypeInt(), }, &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "timequantum", - Type: parser.NewDataTypeString(), + RelationName: "fb$table_columns", + ColumnName: "timequantum", + Type: parser.NewDataTypeString(), }, &types.PlannerColumn{ - Table: "fb$table_columns", - Name: "ttl", - Type: parser.NewDataTypeInt(), + RelationName: "fb$table_columns", + ColumnName: "ttl", + Type: parser.NewDataTypeInt(), }, } } diff --git a/sql3/planner/opfeaturebasetables.go b/sql3/planner/opfeaturebasetables.go index eb2080c43..af15181a3 100644 --- a/sql3/planner/opfeaturebasetables.go +++ b/sql3/planner/opfeaturebasetables.go @@ -31,7 +31,7 @@ func (p *PlanOpFeatureBaseTables) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) ps := make([]string, 0) for _, e := range p.Schema() { - ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = ps return result @@ -52,29 +52,29 @@ func (p *PlanOpFeatureBaseTables) Warnings() []string { func (p *PlanOpFeatureBaseTables) Schema() types.Schema { return types.Schema{ &types.PlannerColumn{ - Table: "fb$tables", - Name: "name", - Type: parser.NewDataTypeString(), + RelationName: "fb$tables", + ColumnName: "name", + Type: parser.NewDataTypeString(), }, &types.PlannerColumn{ - Table: "fb$tables", - Name: "created_at", - Type: parser.NewDataTypeTimestamp(), + RelationName: "fb$tables", + ColumnName: "created_at", + Type: parser.NewDataTypeTimestamp(), }, &types.PlannerColumn{ - Table: "fb$tables", - Name: "track_existence", - Type: parser.NewDataTypeBool(), + RelationName: "fb$tables", + ColumnName: "track_existence", + Type: parser.NewDataTypeBool(), }, &types.PlannerColumn{ - Table: "fb$tables", - Name: "keys", - Type: parser.NewDataTypeBool(), + RelationName: "fb$tables", + ColumnName: "keys", + Type: parser.NewDataTypeBool(), }, &types.PlannerColumn{ - Table: "fb$tables", - Name: "shard_width", - Type: parser.NewDataTypeInt(), + RelationName: "fb$tables", + ColumnName: "shard_width", + Type: parser.NewDataTypeInt(), }, } } diff --git a/sql3/planner/opfilter.go b/sql3/planner/opfilter.go index dff39e24a..1a5a8aacf 100644 --- a/sql3/planner/opfilter.go +++ b/sql3/planner/opfilter.go @@ -58,7 +58,7 @@ func (p *PlanOpFilter) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) ps := make([]string, 0) for _, e := range p.Schema() { - ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = ps result["child"] = p.ChildOp.Plan() diff --git a/sql3/planner/opgroupby.go b/sql3/planner/opgroupby.go index 3e74e54d1..65deb4f6c 100644 --- a/sql3/planner/opgroupby.go +++ b/sql3/planner/opgroupby.go @@ -5,9 +5,12 @@ package planner import ( "context" "fmt" + "hash/maphash" + "log" - "github.com/featurebasedb/featurebase/v3/sql3" - "github.com/featurebasedb/featurebase/v3/sql3/planner/types" + "github.com/molecula/featurebase/v3/errors" + "github.com/molecula/featurebase/v3/sql3" + "github.com/molecula/featurebase/v3/sql3/planner/types" ) // PlanOpGroupBy handles the GROUP BY clause @@ -38,18 +41,18 @@ func (p *PlanOpGroupBy) Schema() types.Schema { continue } s := &types.PlannerColumn{ - Name: ref.columnName, - Table: ref.tableName, - Type: expr.Type(), + ColumnName: ref.columnName, + RelationName: ref.tableName, + Type: expr.Type(), } result[idx] = s } offset := len(p.GroupByExprs) for idx, agg := range p.Aggregates { s := &types.PlannerColumn{ - Name: "", - Table: "", - Type: agg.Type(), + ColumnName: "", + RelationName: "", + Type: agg.Type(), } result[idx+offset] = s } @@ -58,15 +61,15 @@ func (p *PlanOpGroupBy) Schema() types.Schema { } func (p *PlanOpGroupBy) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) { - // TODO(pok) implement group by with group by expressions i, err := p.ChildOp.Iterator(ctx, row) if err != nil { return nil, err } - aggs := []types.PlanExpression{} - aggs = append(aggs, p.GroupByExprs...) - aggs = append(aggs, p.Aggregates...) - return newGroupByIter(ctx, aggs, i), nil + if len(p.GroupByExprs) == 0 { + return newGroupByIter(ctx, p.Aggregates, i), nil + } else { + return newGroupByGroupingIter(ctx, p.Aggregates, p.GroupByExprs, i), nil + } } func (p *PlanOpGroupBy) Children() []types.PlanOperator { @@ -82,12 +85,26 @@ func (p *PlanOpGroupBy) WithChildren(children ...types.PlanOperator) (types.Plan return NewPlanOpGroupBy(p.Aggregates, p.GroupByExprs, children[0]), nil } +func (p *PlanOpGroupBy) Expressions() []types.PlanExpression { + result := []types.PlanExpression{} + result = append(result, p.GroupByExprs...) + return result +} + +func (p *PlanOpGroupBy) WithUpdatedExpressions(exprs ...types.PlanExpression) (types.PlanOperator, error) { + if len(exprs) != 1 { + return nil, sql3.NewErrInternalf("unexpected number of exprs '%d'", len(exprs)) + } + p.GroupByExprs = exprs + return p, nil +} + func (p *PlanOpGroupBy) Plan() map[string]interface{} { result := make(map[string]interface{}) result["_op"] = fmt.Sprintf("%T", p) sc := make([]string, 0) for _, e := range p.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc result["child"] = p.ChildOp.Plan() @@ -120,11 +137,11 @@ func (p *PlanOpGroupBy) Warnings() []string { } type groupByIter struct { - aggregates []types.PlanExpression - child types.RowIterator - ctx context.Context - buf []types.AggregationBuffer - done bool + aggregates []types.PlanExpression + child types.RowIterator + ctx context.Context + aggregationBuffers *keysAndAggregations + done bool } func newGroupByIter(ctx context.Context, aggregates []types.PlanExpression, child types.RowIterator) *groupByIter { @@ -132,7 +149,9 @@ func newGroupByIter(ctx context.Context, aggregates []types.PlanExpression, chil aggregates: aggregates, child: child, ctx: ctx, - buf: make([]types.AggregationBuffer, len(aggregates)), + aggregationBuffers: &keysAndAggregations{ + buffers: make([]types.AggregationBuffer, len(aggregates)), + }, } } @@ -145,7 +164,7 @@ func (i *groupByIter) Next(ctx context.Context) (types.Row, error) { var err error for j, a := range i.aggregates { - i.buf[j], err = newAggregationBuffer(a) + i.aggregationBuffers.buffers[j], err = newAggregationBuffer(a) if err != nil { return nil, err } @@ -160,12 +179,115 @@ func (i *groupByIter) Next(ctx context.Context) (types.Row, error) { return nil, err } - if err := updateBuffers(ctx, i.buf, row); err != nil { + if err := updateBuffers(ctx, i.aggregationBuffers, row); err != nil { return nil, err } } - return evalBuffers(ctx, i.buf) + return evalBuffers(ctx, i.aggregationBuffers) +} + +type keysAndAggregations struct { + groupByKeys []interface{} + buffers []types.AggregationBuffer +} + +type groupByGroupingIter struct { + aggregates []types.PlanExpression + groupByExprs []types.PlanExpression + aggregations ObjectCache + keys []uint64 + child types.RowIterator +} + +func newGroupByGroupingIter(ctx context.Context, aggregates, groupByExprs []types.PlanExpression, child types.RowIterator) *groupByGroupingIter { + return &groupByGroupingIter{ + aggregates: aggregates, + groupByExprs: groupByExprs, + child: child, + } +} + +func (i *groupByGroupingIter) Next(ctx context.Context) (types.Row, error) { + if i.aggregations == nil { + i.aggregations = NewMapObjectCache() + if err := i.compute(ctx); err != nil { + return nil, err + } + } + + if len(i.keys) > 0 { + buffers, err := i.get(i.keys[0]) + if err != nil { + return nil, err + } + + i.keys = i.keys[1:] + + aggRow, err := evalBuffers(ctx, buffers) + if err != nil { + return nil, err + } + + var row = make(types.Row, len(i.groupByExprs)+len(aggRow)) + copy(row, buffers.groupByKeys) + copy(row[len(buffers.groupByKeys):], aggRow) + return row, nil + } + return nil, types.ErrNoMoreRows +} + +func (i *groupByGroupingIter) compute(ctx context.Context) error { + for { + row, err := i.child.Next(ctx) + if err != nil { + if err == types.ErrNoMoreRows { + break + } + return err + } + + key, keyValues, err := groupingKeyHash(ctx, i.groupByExprs, row) + if err != nil { + return err + } + + b, err := i.get(key) + if errors.Is(err, sql3.ErrCacheKeyNotFound) { + b = &keysAndAggregations{} + b.buffers = make([]types.AggregationBuffer, len(i.aggregates)) + for j, a := range i.aggregates { + b.buffers[j], err = newAggregationBuffer(a) + if err != nil { + return err + } + } + b.groupByKeys = keyValues + if err := i.aggregations.PutObject(key, b); err != nil { + return err + } + i.keys = append(i.keys, key) + } else if err != nil { + return err + } + + err = updateBuffers(ctx, b, row) + if err != nil { + return err + } + } + return nil +} + +func (i *groupByGroupingIter) get(key uint64) (*keysAndAggregations, error) { + v, err := i.aggregations.GetObject(key) + if err != nil { + return nil, err + } + if v == nil { + return nil, nil + } + return v.(*keysAndAggregations), err } func newAggregationBuffer(expr types.PlanExpression) (types.AggregationBuffer, error) { @@ -177,8 +299,8 @@ func newAggregationBuffer(expr types.PlanExpression) (types.AggregationBuffer, e } } -func updateBuffers(ctx context.Context, buffers []types.AggregationBuffer, row types.Row) error { - for _, b := range buffers { +func updateBuffers(ctx context.Context, buffers *keysAndAggregations, row types.Row) error { + for _, b := range buffers.buffers { if err := b.Update(ctx, row); err != nil { return err } @@ -186,10 +308,10 @@ func updateBuffers(ctx context.Context, buffers []types.AggregationBuffer, row t return nil } -func evalBuffers(ctx context.Context, buffers []types.AggregationBuffer) (types.Row, error) { - var row = make(types.Row, len(buffers)) +func evalBuffers(ctx context.Context, aggregationBuffers *keysAndAggregations) (types.Row, error) { + var row = make(types.Row, len(aggregationBuffers.buffers)) var err error - for i, b := range buffers { + for i, b := range aggregationBuffers.buffers { row[i], err = b.Eval(ctx) if err != nil { return nil, err @@ -197,3 +319,23 @@ func evalBuffers(ctx context.Context, buffers []types.AggregationBuffer) (types. } return row, nil } + +func groupingKeyHash(ctx context.Context, groupByExprs []types.PlanExpression, row types.Row) (uint64, types.Row, error) { + rowKeys := make([]interface{}, len(groupByExprs)) + var hash maphash.Hash + hash.SetSeed(prototypeHash.Seed()) + for i, expr := range groupByExprs { + v, err := expr.Evaluate(row) + if err != nil { + return 0, nil, err + } + _, err = hash.Write(([]byte)(fmt.Sprintf("%#v,", v))) + if err != nil { + return 0, nil, err + } + rowKeys[i] = v + } + result := hash.Sum64() + log.Printf("Hash %v, %v", result, rowKeys) + return result, rowKeys, nil +} diff --git a/sql3/planner/opinsert.go b/sql3/planner/opinsert.go index b981897aa..33aba1440 100644 --- a/sql3/planner/opinsert.go +++ b/sql3/planner/opinsert.go @@ -39,7 +39,7 @@ func (p *PlanOpInsert) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) sc := make([]string, 0) for _, e := range p.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc result["tableName"] = p.tableName @@ -242,7 +242,7 @@ func (i *insertRowIter) Next(ctx context.Context) (types.Row, error) { } vals := make([]uint64, 1) - vals[0] = coercedVal.(uint64) + vals[0] = uint64(coercedVal.(int64)) req := &pilosa.ImportRequest{ Index: i.tableName, diff --git a/sql3/planner/opnestedloops.go b/sql3/planner/opnestedloops.go index 2a415412e..b40f7098e 100644 --- a/sql3/planner/opnestedloops.go +++ b/sql3/planner/opnestedloops.go @@ -19,10 +19,11 @@ type PlanOpNestedLoops struct { warnings []string } -func NewPlanOpNestedLoops(top, bottom types.PlanOperator) *PlanOpNestedLoops { +func NewPlanOpNestedLoops(top, bottom types.PlanOperator, condition types.PlanExpression) *PlanOpNestedLoops { return &PlanOpNestedLoops{ top: top, bottom: bottom, + cond: condition, warnings: make([]string, 0), } } @@ -32,7 +33,7 @@ func (p *PlanOpNestedLoops) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) ps := make([]string, 0) for _, e := range p.Schema() { - ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = ps result["top"] = p.top.Plan() @@ -73,17 +74,26 @@ func (p *PlanOpNestedLoops) Iterator(ctx context.Context, row types.Row) (types. } rowWidth := len(row) + len(p.top.Schema()) + len(p.bottom.Schema()) - return newNestedLoopsIter(ctx, joinTypeInner, topIter, p.bottom, row, nil, rowWidth, row), nil + return newNestedLoopsIter(ctx, joinTypeInner, topIter, p.bottom, row, p.cond, rowWidth, row), nil } func (p *PlanOpNestedLoops) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) { if len(children) != 2 { return nil, sql3.NewErrInternalf("unexpected number of children '%d'", len(children)) } - return NewPlanOpNestedLoops(children[0], children[1]), nil + return NewPlanOpNestedLoops(children[0], children[1], p.cond), nil } -func (p *PlanOpNestedLoops) NewWithExpressions(exprs ...types.PlanExpression) (types.PlanOperator, error) { +func (p *PlanOpNestedLoops) Expressions() []types.PlanExpression { + if p.cond != nil { + return []types.PlanExpression{ + p.cond, + } + } + return []types.PlanExpression{} +} + +func (p *PlanOpNestedLoops) WithUpdatedExpressions(exprs ...types.PlanExpression) (types.PlanOperator, error) { if len(exprs) != 1 { return nil, sql3.NewErrInternalf("unexpected number of exprs '%d'", len(exprs)) } @@ -114,7 +124,6 @@ type nestedLoopsIter struct { rowSize int originalRow types.Row - scopeLen int bottomRows RowCache } @@ -174,10 +183,8 @@ func (i *nestedLoopsIter) loadBottom(ctx context.Context) (row types.Row, err er } func (i *nestedLoopsIter) buildRow(primary, secondary types.Row) types.Row { - toCut := len(i.originalRow) - i.scopeLen - row := make(types.Row, i.rowSize-toCut) + row := make(types.Row, i.rowSize) - scope := primary[:i.scopeLen] primary = primary[len(i.originalRow):] var first, second types.Row @@ -190,11 +197,10 @@ func (i *nestedLoopsIter) buildRow(primary, secondary types.Row) types.Row { default: first = primary second = secondary - secondOffset = i.scopeLen + len(first) + secondOffset = len(first) } - copy(row, scope) - copy(row[i.scopeLen:], first) + copy(row, first) copy(row[secondOffset:], second) return row } @@ -240,6 +246,8 @@ func (i *nestedLoopsIter) Next(ctx context.Context) (types.Row, error) { } i.foundMatch = true + + //DEBUG log.Printf("Join result %v", row) return row, nil } } diff --git a/sql3/planner/opnulltable.go b/sql3/planner/opnulltable.go index 3ab206afe..a808f6516 100644 --- a/sql3/planner/opnulltable.go +++ b/sql3/planner/opnulltable.go @@ -42,7 +42,7 @@ func (p *PlanOpNullTable) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) ps := make([]string, 0) for _, e := range p.Schema() { - ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = ps return result diff --git a/sql3/planner/oporderby.go b/sql3/planner/oporderby.go index 4f22e629c..e24d2eabc 100644 --- a/sql3/planner/oporderby.go +++ b/sql3/planner/oporderby.go @@ -86,7 +86,7 @@ func (n *PlanOpOrderBy) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", n) sc := make([]string, 0) for _, e := range n.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc diff --git a/sql3/planner/oppqlaggregate.go b/sql3/planner/oppqlaggregate.go index 319806e8d..abf9013ac 100644 --- a/sql3/planner/oppqlaggregate.go +++ b/sql3/planner/oppqlaggregate.go @@ -38,7 +38,7 @@ func (p *PlanOpPQLAggregate) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) ps := make([]string, 0) for _, e := range p.Schema() { - ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = ps result["tableName"] = p.tableName @@ -65,9 +65,9 @@ func (p *PlanOpPQLAggregate) Warnings() []string { func (p *PlanOpPQLAggregate) Schema() types.Schema { result := make(types.Schema, 1) s := &types.PlannerColumn{ - Name: "", - Table: "", - Type: p.aggregate.AggExpression().Type(), + ColumnName: "", + RelationName: "", + Type: p.aggregate.AggExpression().Type(), } result[0] = s return result diff --git a/sql3/planner/oppqlgroupby.go b/sql3/planner/oppqlgroupby.go index 010f71c99..dcd7ed708 100644 --- a/sql3/planner/oppqlgroupby.go +++ b/sql3/planner/oppqlgroupby.go @@ -40,7 +40,7 @@ func (p *PlanOpPQLGroupBy) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) sc := make([]string, 0) for _, e := range p.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc result["tableName"] = p.tableName @@ -77,16 +77,16 @@ func (p *PlanOpPQLGroupBy) Schema() types.Schema { continue } s := &types.PlannerColumn{ - Name: ref.columnName, - Table: ref.tableName, - Type: expr.Type(), + ColumnName: ref.columnName, + RelationName: ref.tableName, + Type: expr.Type(), } result[idx] = s } s := &types.PlannerColumn{ - Name: "", - Table: "", - Type: p.aggregate.AggExpression().Type(), + ColumnName: "", + RelationName: "", + Type: p.aggregate.AggExpression().Type(), } result[len(p.groupByExprs)] = s diff --git a/sql3/planner/oppqlmultiaggregate.go b/sql3/planner/oppqlmultiaggregate.go index bd681f826..f6a86e356 100644 --- a/sql3/planner/oppqlmultiaggregate.go +++ b/sql3/planner/oppqlmultiaggregate.go @@ -29,7 +29,7 @@ func (p *PlanOpPQLMultiAggregate) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) sc := make([]string, 0) for _, e := range p.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc @@ -57,9 +57,9 @@ func (p *PlanOpPQLMultiAggregate) Schema() types.Schema { result := make(types.Schema, len(p.operators)) for idx, aggOp := range p.operators { s := &types.PlannerColumn{ - Name: "", - Table: "", - Type: aggOp.aggregate.AggExpression().Type(), + ColumnName: "", + RelationName: "", + Type: aggOp.aggregate.AggExpression().Type(), } result[idx] = s } diff --git a/sql3/planner/oppqlmultigroupby.go b/sql3/planner/oppqlmultigroupby.go index 13bfc73b1..e3198cd2f 100644 --- a/sql3/planner/oppqlmultigroupby.go +++ b/sql3/planner/oppqlmultigroupby.go @@ -34,7 +34,7 @@ func (p *PlanOpPQLMultiGroupBy) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) sc := make([]string, 0) for _, e := range p.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc @@ -71,18 +71,18 @@ func (p *PlanOpPQLMultiGroupBy) Schema() types.Schema { continue } s := &types.PlannerColumn{ - Name: ref.columnName, - Table: ref.tableName, - Type: expr.Type(), + ColumnName: ref.columnName, + RelationName: ref.tableName, + Type: expr.Type(), } result[idx] = s } offset := len(p.groupByExprs) for idx, aggOp := range p.operators { s := &types.PlannerColumn{ - Name: "", - Table: "", - Type: aggOp.aggregate.AggExpression().Type(), + ColumnName: "", + RelationName: "", + Type: aggOp.aggregate.AggExpression().Type(), } result[idx+offset] = s } @@ -125,7 +125,7 @@ type pqlMultiGroupByRowIter struct { iterators []types.RowIterator groupCache KeyedRowCache - groupKeys []string + groupKeys []uint64 } var _ types.RowIterator = (*pqlMultiGroupByRowIter)(nil) @@ -168,7 +168,7 @@ func (i *pqlMultiGroupByRowIter) computeMultiGroupBy(ctx context.Context) error for { //build a key for the group by columns for this row - key, err := groupingKey(ctx, i.groupByColumns, irow) + key, _, err := groupingKeyHash(ctx, i.groupByColumns, irow) if err != nil { return err } @@ -216,15 +216,3 @@ func (i *pqlMultiGroupByRowIter) computeMultiGroupBy(ctx context.Context) error return nil } - -func groupingKey(ctx context.Context, exprs []types.PlanExpression, row types.Row) (string, error) { - key := "" - for _, expr := range exprs { - v, err := expr.Evaluate(row) - if err != nil { - return "", err - } - key += fmt.Sprintf(":%v", v) - } - return key, nil -} diff --git a/sql3/planner/oppqltablescan.go b/sql3/planner/oppqltablescan.go index 06602ae32..00e0b5920 100644 --- a/sql3/planner/oppqltablescan.go +++ b/sql3/planner/oppqltablescan.go @@ -5,6 +5,7 @@ package planner import ( "context" "fmt" + "strings" pilosa "github.com/featurebasedb/featurebase/v3" "github.com/featurebasedb/featurebase/v3/pql" @@ -18,13 +19,13 @@ import ( type PlanOpPQLTableScan struct { planner *ExecutionPlanner tableName string - columns []types.PlanExpression + columns []string filter types.PlanExpression topExpr types.PlanExpression warnings []string } -func NewPlanOpPQLTableScan(p *ExecutionPlanner, tableName string, columns []types.PlanExpression) *PlanOpPQLTableScan { +func NewPlanOpPQLTableScan(p *ExecutionPlanner, tableName string, columns []string) *PlanOpPQLTableScan { return &PlanOpPQLTableScan{ planner: p, tableName: tableName, @@ -38,7 +39,7 @@ func (p *PlanOpPQLTableScan) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) sc := make([]string, 0) for _, e := range p.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc @@ -51,11 +52,7 @@ func (p *PlanOpPQLTableScan) Plan() map[string]interface{} { result["filter"] = p.filter.Plan() } - ps := make([]interface{}, 0) - for _, c := range p.columns { - ps = append(ps, c.Plan()) - } - result["columns"] = ps + result["columns"] = p.columns return result } @@ -82,15 +79,18 @@ func (p *PlanOpPQLTableScan) UpdateFilters(filterCondition types.PlanExpression) func (p *PlanOpPQLTableScan) Schema() types.Schema { result := make(types.Schema, 0) - for _, col := range p.columns { - si, ok := col.(types.IdentifiableByName) - if ok { - result = append(result, &types.PlannerColumn{ - Name: si.Name(), - Table: p.tableName, - Type: col.Type(), - }) - } + + table, err := p.planner.schemaAPI.IndexInfo(context.Background(), p.tableName) + if err != nil { + return result + } + + for _, col := range table.Fields { + result = append(result, &types.PlannerColumn{ + ColumnName: col.Name, + RelationName: p.tableName, + Type: fieldSQLDataType(col), + }) } return result } @@ -113,18 +113,23 @@ func (p *PlanOpPQLTableScan) WithChildren(children ...types.PlanOperator) (types return nil, nil } -// TODO(pok) remove the name mapping here and do it by ordinal position +type targetColumn struct { + columnIdx int + srcColumnIdx int + columnName string + dataType parser.ExprDataType +} + type tableScanRowIter struct { planner *ExecutionPlanner tableName string - columns []types.PlanExpression + columns []string predicate types.PlanExpression topExpr types.PlanExpression - result []pilosa.ExtractedTableColumn - rowWidth int - sourceColumnMap map[string]int - targetColumnMap map[string]int + result []pilosa.ExtractedTableColumn + rowWidth int + columnMap map[string]*targetColumn } var _ types.RowIterator = (*tableScanRowIter)(nil) @@ -146,9 +151,14 @@ func (i *tableScanRowIter) Next(ctx context.Context) (types.Row, error) { } i.rowWidth = len(table.Fields) - i.targetColumnMap = make(map[string]int) + i.columnMap = make(map[string]*targetColumn) for idx, fld := range table.Fields { - i.targetColumnMap[fld.Name] = idx + i.columnMap[fld.Name] = &targetColumn{ + columnIdx: idx, + srcColumnIdx: -1, + columnName: fld.Name, + dataType: fieldSQLDataType(fld), + } } var cond *pql.Call @@ -180,23 +190,19 @@ func (i *tableScanRowIter) Next(ctx context.Context) (types.Row, error) { call := &pql.Call{Name: "Extract", Children: []*pql.Call{cond}} for _, c := range i.columns { - col, ok := c.(types.IdentifiableByName) - if !ok { - return nil, sql3.NewErrInternalf("unexpected column type '%T'", c) - } - // Skip the _id field. - if col.Name() == "_id" { + // skip the _id field + if strings.EqualFold(c, "_id") { continue } + call.Children = append(call.Children, &pql.Call{ Name: "Rows", - Args: map[string]interface{}{"field": col.Name()}, + Args: map[string]interface{}{"field": c}, }, ) } - queryResponse, err := i.planner.executor.Execute(ctx, i.tableName, &pql.Query{Calls: []*pql.Call{call}}, nil, nil) if err != nil { return nil, err @@ -206,9 +212,14 @@ func (i *tableScanRowIter) Next(ctx context.Context) (types.Row, error) { return nil, sql3.NewErrInternalf("unexpected Extract() result type: %T", queryResponse.Results[0]) } i.result = tbl.Columns - i.sourceColumnMap = make(map[string]int) + + //set the source index for idx, fld := range tbl.Fields { - i.sourceColumnMap[fld.Name] = idx + mappedColumn, ok := i.columnMap[fld.Name] + if !ok { + return nil, sql3.NewErrInternalf("mapped column not found for column named '%s'", fld.Name) + } + mappedColumn.srcColumnIdx = idx } } @@ -218,55 +229,47 @@ func (i *tableScanRowIter) Next(ctx context.Context) (types.Row, error) { for _, c := range i.columns { result := i.result[0] - col, ok := c.(types.IdentifiableByName) + mappedColumn, ok := i.columnMap[c] if !ok { - return nil, sql3.NewErrInternalf("unexpected column type '%T'", c) + return nil, sql3.NewErrInternalf("mapped column not found for column named '%s'", c) } + mappedColIdx := mappedColumn.columnIdx + mappedSrcColIdx := mappedColumn.srcColumnIdx - targetColIdx, ok := i.targetColumnMap[col.Name()] - if !ok { - return nil, sql3.NewErrInternalf("target index not found for column named %s", col.Name()) - } - - if col.Name() == "_id" { + if strings.EqualFold(c, "_id") { if result.Column.Keyed { - row[targetColIdx] = result.Column.Key + row[mappedColIdx] = result.Column.Key } else { - row[targetColIdx] = int64(result.Column.ID) + row[mappedColIdx] = int64(result.Column.ID) } } else { - - sourceColIdx, ok := i.sourceColumnMap[col.Name()] - if !ok { - return nil, sql3.NewErrInternalf("source index not found for column named %s", col.Name()) - } - switch c.Type().(type) { + switch mappedColumn.dataType.(type) { case *parser.DataTypeIDSet: //empty sets are null - val, ok := result.Rows[sourceColIdx].([]uint64) + val, ok := result.Rows[mappedSrcColIdx].([]uint64) if !ok { - return nil, sql3.NewErrInternalf("unexpected type for column value '%T'", result.Rows[sourceColIdx]) + return nil, sql3.NewErrInternalf("unexpected type for column value '%T'", result.Rows[mappedSrcColIdx]) } if len(val) == 0 { - row[targetColIdx] = nil + row[mappedColIdx] = nil } else { - row[targetColIdx] = val + row[mappedColIdx] = val } case *parser.DataTypeStringSet: //empty sets are null - val, ok := result.Rows[sourceColIdx].([]string) + val, ok := result.Rows[mappedSrcColIdx].([]string) if !ok { - return nil, sql3.NewErrInternalf("unexpected type for column value '%T'", result.Rows[sourceColIdx]) + return nil, sql3.NewErrInternalf("unexpected type for column value '%T'", result.Rows[mappedSrcColIdx]) } if len(val) == 0 { - row[targetColIdx] = nil + row[mappedColIdx] = nil } else { - row[targetColIdx] = val + row[mappedColIdx] = val } default: - row[targetColIdx] = result.Rows[sourceColIdx] + row[mappedColIdx] = result.Rows[mappedSrcColIdx] } } } diff --git a/sql3/planner/opprojection.go b/sql3/planner/opprojection.go index 3d0c3d256..cf0cfc347 100644 --- a/sql3/planner/opprojection.go +++ b/sql3/planner/opprojection.go @@ -63,7 +63,7 @@ func (p *PlanOpProjection) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) sc := make([]string, 0) for _, e := range p.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc @@ -110,9 +110,9 @@ func ExpressionToColumn(e types.PlanExpression) *types.PlannerColumn { } return &types.PlannerColumn{ - Name: name, - Type: e.Type(), - Table: table, + ColumnName: name, + Type: e.Type(), + RelationName: table, } } diff --git a/sql3/planner/opquery.go b/sql3/planner/opquery.go index 27a8e956a..89b6f091c 100644 --- a/sql3/planner/opquery.go +++ b/sql3/planner/opquery.go @@ -64,7 +64,7 @@ func (p *PlanOpQuery) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) sc := make([]string, 0) for _, e := range p.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc diff --git a/sql3/planner/oprelalias.go b/sql3/planner/oprelalias.go index bcf3b42ce..1b775b4f7 100644 --- a/sql3/planner/oprelalias.go +++ b/sql3/planner/oprelalias.go @@ -26,7 +26,11 @@ func NewPlanOpRelAlias(alias string, child types.PlanOperator) *PlanOpRelAlias { } func (p *PlanOpRelAlias) Schema() types.Schema { - return p.ChildOp.Schema() + schema := p.ChildOp.Schema() + for _, s := range schema { + s.RelationName = p.alias + } + return schema } func (p *PlanOpRelAlias) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) { @@ -51,7 +55,7 @@ func (p *PlanOpRelAlias) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) sc := make([]string, 0) for _, e := range p.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc diff --git a/sql3/planner/opsubquery.go b/sql3/planner/opsubquery.go index 5cf616be4..d282ae7bb 100644 --- a/sql3/planner/opsubquery.go +++ b/sql3/planner/opsubquery.go @@ -45,7 +45,7 @@ func (p *PlanOpSubquery) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) sc := make([]string, 0) for _, e := range p.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc diff --git a/sql3/planner/optablevaluedfunction.go b/sql3/planner/optablevaluedfunction.go index 23cee50b7..b603b3dff 100644 --- a/sql3/planner/optablevaluedfunction.go +++ b/sql3/planner/optablevaluedfunction.go @@ -34,9 +34,9 @@ func (p *PlanOpTableValuedFunction) Schema() types.Schema { } for _, member := range tvfResultType.Columns { result = append(result, &types.PlannerColumn{ - Name: member.Name, - Table: "", - Type: member.DataType, + ColumnName: member.Name, + RelationName: "", + Type: member.DataType, }) } return result @@ -59,7 +59,7 @@ func (p *PlanOpTableValuedFunction) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) sc := make([]string, 0) for _, e := range p.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc diff --git a/sql3/planner/optop.go b/sql3/planner/optop.go index cd5836956..95ea0badb 100644 --- a/sql3/planner/optop.go +++ b/sql3/planner/optop.go @@ -47,7 +47,7 @@ func (p *PlanOpTop) Plan() map[string]interface{} { result["_op"] = fmt.Sprintf("%T", p) sc := make([]string, 0) for _, e := range p.Schema() { - sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName())) + sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) } result["_schema"] = sc diff --git a/sql3/planner/planoptimizer.go b/sql3/planner/planoptimizer.go index 08cca55d1..065dc3484 100644 --- a/sql3/planner/planoptimizer.go +++ b/sql3/planner/planoptimizer.go @@ -9,6 +9,7 @@ import ( "reflect" "strings" + "github.com/molecula/featurebase/v3/errors" "github.com/molecula/featurebase/v3/sql3" "github.com/molecula/featurebase/v3/sql3/parser" "github.com/molecula/featurebase/v3/sql3/planner/types" @@ -45,9 +46,11 @@ var optimizerFunctions = []OptimizerFunc{ // based on the child operator for a projection fixGroupByProjections, - // update the columnIdx for all the references in the projections - // based on the child operator for a projection - fixJoinProjections, + // update the columnIdx for all the references in joins + fixJoinFieldRefs, + + // update the columnIdx for all the references in group bys + fixGroupByFieldRefs, // if the query has one TableScanOperator then push the top // expression down into that operator @@ -61,6 +64,14 @@ type OptimizerScope struct { // optimizePlan takes a plan from the compiler and executes a series of transforms on it to optimize it func (p *ExecutionPlanner) optimizePlan(ctx context.Context, plan types.PlanOperator) (types.PlanOperator, error) { + + //log.Println("================================================================================") + //log.Println("plan pre-optimzation") + //jplan := plan.Plan() + //a, _ := json.MarshalIndent(jplan, "", " ") + //log.Println(string(a)) + //log.Println("--------------------------------------------------------------------------------") + var err error var result = plan for _, ofunc := range optimizerFunctions { @@ -891,44 +902,28 @@ func fixGroupByProjections(ctx context.Context, a *ExecutionPlanner, n types.Pla }) } -func fixJoinProjections(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) (types.PlanOperator, bool, error) { +func fixJoinFieldRefs(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) (types.PlanOperator, bool, error) { return TransformPlanOp(n, func(node types.PlanOperator) (types.PlanOperator, bool, error) { switch n := node.(type) { - case *PlanOpProjection: - switch childOp := n.ChildOp.(type) { - case *PlanOpNestedLoops: - //PlanOpNestedLoops iterator returns columns from top iterator and then columns from bottom iterator + case *PlanOpNestedLoops: + _, _, err := fixFieldRefIndexesForOperator(ctx, a, n, scope) + if err != nil { + return nil, true, err + } + return n, true, nil + default: + return n, true, nil + } + }) +} - //make a map of names from the schema - schemaNameMap := make(map[string]int) - schema := childOp.Schema() - for idx, s := range schema { - key := fmt.Sprintf("%s.%s", s.Table, s.Name) - schemaNameMap[key] = idx - } - - for idx, pj := range n.Projections { - expr, _, err := TransformExpr(pj, func(e types.PlanExpression) (types.PlanExpression, bool, error) { - switch thisExpr := e.(type) { - case *qualifiedRefPlanExpression: - key := fmt.Sprintf("%s.%s", thisExpr.tableName, thisExpr.columnName) - colIdx, ok := schemaNameMap[key] - if ok { - ae := newQualifiedRefPlanExpression(fmt.Sprintf("$PlanOpNestedLoops.%s.%s:%d", thisExpr.tableName, thisExpr.columnName, colIdx), thisExpr.columnName, colIdx, e.Type()) - return ae, false, nil - } - return e, true, nil - - default: - return e, true, nil - } - }) - if err != nil { - return n, true, err - } - n.Projections[idx] = expr - } - return n, false, nil +func fixGroupByFieldRefs(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) (types.PlanOperator, bool, error) { + return TransformPlanOp(n, func(node types.PlanOperator) (types.PlanOperator, bool, error) { + switch n := node.(type) { + case *PlanOpGroupBy: + _, _, err := fixFieldRefIndexesForOperator(ctx, a, n, scope) + if err != nil { + return nil, true, err } return n, true, nil default: @@ -1025,17 +1020,16 @@ func fixFieldRefIndexes(ctx context.Context, scope *OptimizerScope, a *Execution case *qualifiedRefPlanExpression: for i, col := range schema { newIndex := i - if e.Name() == col.Name && e.tableName == col.Table { + if e.Name() == col.ColumnName && e.tableName == col.RelationName { if newIndex != e.columnIndex { // update the column index - e.columnIndex = newIndex + return newQualifiedRefPlanExpression(e.tableName, e.columnName, newIndex, e.dataType), false, nil } return e, true, nil } } return nil, true, sql3.NewErrColumnNotFound(0, 0, e.Name()) } - return e, true, nil }) } @@ -1086,10 +1080,9 @@ func fixFieldRefIndexesForOperator(ctx context.Context, a *ExecutionPlanner, nod return fixed, same, nil } - if strings.Contains(err.Error(), "unexpected!") { + if errors.Is(err, sql3.ErrColumnNotFound) { continue } - return nil, true, err } @@ -1109,7 +1102,7 @@ func fixFieldRefIndexesForOperator(ctx context.Context, a *ExecutionPlanner, nod return nil, true, err } if !sameJ { - n, err = j.NewWithExpressions(cond) + n, err = j.WithUpdatedExpressions(cond) if err != nil { return nil, true, err } diff --git a/sql3/planner/planwalker.go b/sql3/planner/planwalker.go index 84e0d6e68..2af498888 100644 --- a/sql3/planner/planwalker.go +++ b/sql3/planner/planwalker.go @@ -316,7 +316,7 @@ func TransformSinglePlanOpExprsInPlanOpContext(op types.PlanOperator, f ExprWith } if len(newExprs) > 0 { - op, err = ne.NewWithExpressions(newExprs...) + op, err = ne.WithUpdatedExpressions(newExprs...) if err != nil { return nil, true, err } @@ -353,7 +353,7 @@ func TransformSinglePlanOpExpressions(op types.PlanOperator, f ExprFunc) (types. } } if len(newExprs) > 0 { - n, err := e.NewWithExpressions(newExprs...) + n, err := e.WithUpdatedExpressions(newExprs...) if err != nil { return nil, true, err } diff --git a/sql3/planner/types/operator.go b/sql3/planner/types/operator.go index 14bf8c073..dac15002c 100644 --- a/sql3/planner/types/operator.go +++ b/sql3/planner/types/operator.go @@ -41,15 +41,16 @@ type ContainsExpressions interface { // returns the list of expressions contained by the plan operator Expressions() []PlanExpression - // NewWithExpressions returns a new operator with expressions replaced - NewWithExpressions(exprs ...PlanExpression) (PlanOperator, error) + // WithUpdatedExpressions returns a the operator with expressions updated + WithUpdatedExpressions(exprs ...PlanExpression) (PlanOperator, error) } // PlannerColumn is the definition of a column returned as a set from each operator type PlannerColumn struct { - Name string - Table string - Type parser.ExprDataType + ColumnName string + RelationName string + AliasName string + Type parser.ExprDataType } // Relation is an interface to something that can be treated as a relation diff --git a/sql3/sql_definitions_test.go b/sql3/sql_definitions_test.go index 0c780a686..2e99dc623 100644 --- a/sql3/sql_definitions_test.go +++ b/sql3/sql_definitions_test.go @@ -338,6 +338,11 @@ var tableTests []tableTest = []tableTest{ //create table tests createTable, + //joins + joinTestsUsers, + joinTestsOrders, + joinTests, + //time quantums // Skip for now - timeQuantumInsertTest, } diff --git a/sql3/sql_defs_join_test.go b/sql3/sql_defs_join_test.go new file mode 100644 index 000000000..226a3dffe --- /dev/null +++ b/sql3/sql_defs_join_test.go @@ -0,0 +1,65 @@ +package sql3_test + +// join tests +var joinTestsUsers = tableTest{ + name: "jointestusers", + table: tbl( + "users", + srcHdrs( + srcHdr("_id", fldTypeID), + srcHdr("name", fldTypeString), + srcHdr("age", fldTypeInt), + ), + srcRows( + srcRow(int64(0), string("a"), int64(21)), + srcRow(int64(1), string("b"), int64(18)), + srcRow(int64(2), string("c"), int64(28)), + srcRow(int64(3), string("d"), int64(34)), + ), + ), + sqlTests: nil, +} + +var joinTestsOrders = tableTest{ + name: "jointestorders", + table: tbl( + "orders", + srcHdrs( + srcHdr("_id", fldTypeID), + srcHdr("userid", fldTypeID), + srcHdr("price", fldTypeDecimal2), + ), + srcRows( + srcRow(int64(0), int64(1), float64(9.99)), + srcRow(int64(1), int64(0), float64(3.99)), + srcRow(int64(2), int64(2), float64(14.99)), + srcRow(int64(3), int64(3), float64(5.99)), + srcRow(int64(4), int64(1), float64(12.99)), + srcRow(int64(5), int64(2), float64(1.99)), + ), + ), + sqlTests: nil, +} + +var joinTests = tableTest{ + name: "innerjointest", + sqlTests: []sqlTest{ + { + name: "innerjoin-aggregate-groupby", + sqls: sqls( + "select u._id, sum(orders.price) from orders o inner join users u on o.userid = u._id group by u._id;", + ), + expHdrs: hdrs( + hdr("_id", fldTypeID), + hdr("", fldTypeDecimal2), + ), + expRows: rows( + row(int64(1), float64(22.98)), + row(int64(0), float64(3.99)), + row(int64(2), float64(16.98)), + row(int64(3), float64(5.99)), + ), + compare: compareExactOrdered, + }, + }, +} diff --git a/sql3/sql_test.go b/sql3/sql_test.go index a6f22c87a..6c72f91bc 100644 --- a/sql3/sql_test.go +++ b/sql3/sql_test.go @@ -305,7 +305,7 @@ func TestSQL_Execute(t *testing.T) { // make a map of column name to header index m := make(map[string]int) for i := range headers { - m[headers[i].Name] = i + m[headers[i].ColumnName] = i } // Put the expRows in the same column order as the headers returned @@ -314,7 +314,7 @@ func TestSQL_Execute(t *testing.T) { for i := range sqltest.expRows { exp[i] = make([]interface{}, len(headers)) for j := range sqltest.expHdrs { - targetIdx := m[sqltest.expHdrs[j].Name] + targetIdx := m[sqltest.expHdrs[j].ColumnName] if !assert.GreaterOrEqual(t, len(sqltest.expRows[i]), len(headers)) { t.Fatalf("expected row set has fewer columns than returned headers") } @@ -453,8 +453,8 @@ func hdrs(hdrs ...*planner_types.PlannerColumn) []*planner_types.PlannerColumn { // hdr is just a helper function to make the test definition look cleaner. func hdr(name string, typ fldType) *planner_types.PlannerColumn { return &planner_types.PlannerColumn{ - Name: name, - Type: typ, + ColumnName: name, + Type: typ, } } diff --git a/sql3/test/helpers.go b/sql3/test/helpers.go index 65c6fc4fb..237fc837a 100644 --- a/sql3/test/helpers.go +++ b/sql3/test/helpers.go @@ -47,8 +47,8 @@ func MustQueryRows(tb testing.TB, svr *pilosa.Server, q string) ([][]interface{} cols := make([]*planner_types.PlannerColumn, 0) for _, oc := range ocolumns { cols = append(cols, &planner_types.PlannerColumn{ - Name: oc.Name, - Type: oc.Type, + ColumnName: oc.ColumnName, + Type: oc.Type, }) } return results, cols, nil