scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

And now....INNER JOIN! (#2230)

Browse source 
* 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 <travis@pilosa.com>

Co-authored-by: Travis Turner <travis@pilosa.com>
This commit is contained in:
pokeeffe-molecula 2022-09-27 09:27:37 -05:00 • committed by Kasey Rodgers
parent f194cb216b
commit e24978c4a7
36 changed files with 770 additions and 419 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
http_handler.go
View file

@ -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(),
}
}

18
sql3/errors.go
View file

@ -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,

84
sql3/planner/compileselect.go
View file

@ -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:

214
sql3/planner/executionplanner_test.go
View file

@ -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)
}

2
sql3/planner/expression.go
View file

@ -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

46
sql3/planner/expressionagg.go
View file

@ -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

68
sql3/planner/memoryobj.go
View file

@ -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{}),
}
}

2
sql3/planner/opbulkinsert.go
View file

@ -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

2
sql3/planner/opcreatetable.go
View file

@ -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

2
sql3/planner/opdroptable.go
View file

@ -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

86
sql3/planner/opfeaturebasecolumns.go
View file

@ -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(),
},
}
}

32
sql3/planner/opfeaturebasetables.go
View file

@ -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(),
},
}
}

2
sql3/planner/opfilter.go
View file

@ -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()

198
sql3/planner/opgroupby.go
View file

@ -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
}

4
sql3/planner/opinsert.go
View file

@ -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,

32
sql3/planner/opnestedloops.go
View file

@ -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
}
}

2
sql3/planner/opnulltable.go
View file

@ -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

2
sql3/planner/oporderby.go
View file

@ -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

8
sql3/planner/oppqlaggregate.go
View file

@ -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

14
sql3/planner/oppqlgroupby.go
View file

@ -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

8
sql3/planner/oppqlmultiaggregate.go
View file

@ -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
}

30
sql3/planner/oppqlmultigroupby.go
View file

@ -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
}

123
sql3/planner/oppqltablescan.go
View file

@ -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]
}
}
}

8
sql3/planner/opprojection.go
View file

@ -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,
}
}

2
sql3/planner/opquery.go
View file

@ -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

8
sql3/planner/oprelalias.go
View file

@ -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

2
sql3/planner/opsubquery.go
View file

@ -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

8
sql3/planner/optablevaluedfunction.go
View file

@ -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

2
sql3/planner/optop.go
View file

@ -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

81
sql3/planner/planoptimizer.go
View file

@ -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
}

4
sql3/planner/planwalker.go
View file

@ -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
}

11
sql3/planner/types/operator.go
View file

@ -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

5
sql3/sql_definitions_test.go
View file

@ -338,6 +338,11 @@ var tableTests []tableTest = []tableTest{
//create table tests
createTable,
//joins
joinTestsUsers,
joinTestsOrders,
joinTests,
//time quantums
// Skip for now - timeQuantumInsertTest,
}

65
sql3/sql_defs_join_test.go Normal file
View file

@ -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,
},
},
}

8
sql3/sql_test.go
View file

@ -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,
}
}

4
sql3/test/helpers.go
View file

@ -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