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

Aggregation Nation (fb-1955, fb-1887) (#2252)

Browse source 
* make nodeid come from the correct table

* refactored aggregates; added ability to aggregate on expressions not just references

* addressed feedback

* now with the compiler errors fixed after rebase
This commit is contained in:
Pat Okeeffe 2023-02-16 10:56:27 -06:00 • committed by GitHub
parent 10b60f5d51
commit 41b8505d70
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 382 additions and 354 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

115
sql3/planner/expressionagg.go
View file

@ -103,18 +103,10 @@ func (n *countPlanExpression) NewBuffer() (types.AggregationBuffer, error) {
return NewAggCountBuffer(n), nil
}
func (n *countPlanExpression) AggType() types.AggregateFunctionType {
return types.AGGREGATE_COUNT
}
func (n *countPlanExpression) AggExpression() types.PlanExpression {
func (n *countPlanExpression) FirstChildExpr() types.PlanExpression {
return n.arg
}
func (n *countPlanExpression) AggAdditionalExpr() []types.PlanExpression {
return []types.PlanExpression{}
}
func (n *countPlanExpression) Type() parser.ExprDataType {
return n.returnDataType
}
@ -172,18 +164,10 @@ func (n *countDistinctPlanExpression) NewBuffer() (types.AggregationBuffer, erro
return NewAggCountDistinctBuffer(n), nil
}
func (n *countDistinctPlanExpression) AggType() types.AggregateFunctionType {
return types.AGGREGATE_COUNT_DISTINCT
}
func (n *countDistinctPlanExpression) AggExpression() types.PlanExpression {
func (n *countDistinctPlanExpression) FirstChildExpr() types.PlanExpression {
return n.arg
}
func (n *countDistinctPlanExpression) AggAdditionalExpr() []types.PlanExpression {
return []types.PlanExpression{}
}
func (n *countDistinctPlanExpression) Type() parser.ExprDataType {
return n.returnDataType
}
@ -320,29 +304,21 @@ func newSumPlanExpression(arg types.PlanExpression, returnDataType parser.ExprDa
}
func (n *sumPlanExpression) Evaluate(currentRow []interface{}) (interface{}, error) {
arg, ok := n.arg.(*qualifiedRefPlanExpression)
if !ok {
return nil, sql3.NewErrInternalf("unexpected aggregate function arg type '%T'", n.arg)
arg, err := n.arg.Evaluate(currentRow)
if err != nil {
return nil, err
}
return currentRow[arg.columnIndex], nil
return arg, nil
}
func (n *sumPlanExpression) NewBuffer() (types.AggregationBuffer, error) {
return NewAggSumBuffer(n), nil
}
func (n *sumPlanExpression) AggType() types.AggregateFunctionType {
return types.AGGREGATE_SUM
}
func (n *sumPlanExpression) AggExpression() types.PlanExpression {
func (n *sumPlanExpression) FirstChildExpr() types.PlanExpression {
return n.arg
}
func (n *sumPlanExpression) AggAdditionalExpr() []types.PlanExpression {
return []types.PlanExpression{}
}
func (n *sumPlanExpression) Type() parser.ExprDataType {
return n.returnDataType
}
@ -426,7 +402,7 @@ func (a *aggregateAvg) Update(ctx context.Context, row types.Row) error {
a.sum = pql.AddDecimal(thisVal, aggVal)
case *parser.DataTypeInt:
case *parser.DataTypeInt, *parser.DataTypeID:
thisIVal, ok := v.(int64)
if !ok {
return sql3.NewErrInternalf("unexpected type conversion '%T'", v)
@ -435,15 +411,6 @@ func (a *aggregateAvg) Update(ctx context.Context, row types.Row) error {
thisVal := pql.FromInt64(thisIVal, returnType.Scale)
a.sum = pql.AddDecimal(thisVal, aggVal)
case *parser.DataTypeID:
thisIVal, ok := v.(uint64)
if !ok {
return sql3.NewErrInternalf("unexpected type conversion '%T'", v)
}
thisVal := pql.FromInt64(int64(thisIVal), returnType.Scale)
a.sum = pql.AddDecimal(thisVal, aggVal)
default:
return sql3.NewErrInternalf("unhandled aggregate expression datatype '%T'", dataType)
}
@ -503,29 +470,21 @@ func newAvgPlanExpression(arg types.PlanExpression, returnDataType parser.ExprDa
}
func (n *avgPlanExpression) Evaluate(currentRow []interface{}) (interface{}, error) {
arg, ok := n.arg.(*qualifiedRefPlanExpression)
if !ok {
return nil, sql3.NewErrInternalf("unexpected aggregate function arg type '%T'", n.arg)
arg, err := n.arg.Evaluate(currentRow)
if err != nil {
return nil, err
}
return currentRow[arg.columnIndex], nil
return arg, nil
}
func (n *avgPlanExpression) NewBuffer() (types.AggregationBuffer, error) {
return NewAggAvgBuffer(n), nil
}
func (n *avgPlanExpression) AggType() types.AggregateFunctionType {
return types.AGGREGATE_AVG
}
func (n *avgPlanExpression) AggExpression() types.PlanExpression {
func (n *avgPlanExpression) FirstChildExpr() types.PlanExpression {
return n.arg
}
func (n *avgPlanExpression) AggAdditionalExpr() []types.PlanExpression {
return []types.PlanExpression{}
}
func (n *avgPlanExpression) Type() parser.ExprDataType {
return n.returnDataType
}
@ -660,29 +619,21 @@ func newMinPlanExpression(arg types.PlanExpression, returnDataType parser.ExprDa
}
func (n *minPlanExpression) Evaluate(currentRow []interface{}) (interface{}, error) {
arg, ok := n.arg.(*qualifiedRefPlanExpression)
if !ok {
return nil, sql3.NewErrInternalf("unexpected aggregate function arg type '%T'", n.arg)
arg, err := n.arg.Evaluate(currentRow)
if err != nil {
return nil, err
}
return currentRow[arg.columnIndex], nil
return arg, nil
}
func (n *minPlanExpression) NewBuffer() (types.AggregationBuffer, error) {
return NewAggMinBuffer(n), nil
}
func (n *minPlanExpression) AggType() types.AggregateFunctionType {
return types.AGGREGATE_MIN
}
func (n *minPlanExpression) AggExpression() types.PlanExpression {
func (n *minPlanExpression) FirstChildExpr() types.PlanExpression {
return n.arg
}
func (n *minPlanExpression) AggAdditionalExpr() []types.PlanExpression {
return []types.PlanExpression{}
}
func (n *minPlanExpression) Type() parser.ExprDataType {
return n.returnDataType
}
@ -818,29 +769,21 @@ func newMaxPlanExpression(arg types.PlanExpression, returnDataType parser.ExprDa
}
func (n *maxPlanExpression) Evaluate(currentRow []interface{}) (interface{}, error) {
arg, ok := n.arg.(*qualifiedRefPlanExpression)
if !ok {
return nil, sql3.NewErrInternalf("unexpected aggregate function arg type '%T'", n.arg)
arg, err := n.arg.Evaluate(currentRow)
if err != nil {
return nil, err
}
return currentRow[arg.columnIndex], nil
return arg, nil
}
func (n *maxPlanExpression) NewBuffer() (types.AggregationBuffer, error) {
return NewAggMaxBuffer(n), nil
}
func (n *maxPlanExpression) AggType() types.AggregateFunctionType {
return types.AGGREGATE_MAX
}
func (n *maxPlanExpression) AggExpression() types.PlanExpression {
func (n *maxPlanExpression) FirstChildExpr() types.PlanExpression {
return n.arg
}
func (n *maxPlanExpression) AggAdditionalExpr() []types.PlanExpression {
return []types.PlanExpression{}
}
func (n *maxPlanExpression) Type() parser.ExprDataType {
return n.returnDataType
}
@ -900,20 +843,10 @@ func (n *percentilePlanExpression) NewBuffer() (types.AggregationBuffer, error)
return NewAggCountBuffer(n), nil
}
func (n *percentilePlanExpression) AggType() types.AggregateFunctionType {
return types.AGGREGATE_PERCENTILE
}
func (n *percentilePlanExpression) AggExpression() types.PlanExpression {
func (n *percentilePlanExpression) FirstChildExpr() types.PlanExpression {
return n.arg
}
func (n *percentilePlanExpression) AggAdditionalExpr() []types.PlanExpression {
return []types.PlanExpression{
n.nthArg,
}
}
func (n *percentilePlanExpression) Type() parser.ExprDataType {
return n.returnDataType
}

43
sql3/planner/expressionanalyzercall.go
View file

@ -56,23 +56,18 @@ func (p *ExecutionPlanner) analyzeCallExpression(ctx context.Context, call *pars
return nil, sql3.NewErrCallParameterCountMismatch(call.Rparen.Line, call.Rparen.Column, call.Name.Name, 1, len(call.Args))
}
//make sure it's a qualified ref
// if it is a ref, we shouldn't do a sum on the _id
ref, ok := call.Args[0].(*parser.QualifiedRef)
if !ok {
return nil, sql3.NewErrExpectedColumnReference(call.Args[0].Pos().Line, call.Args[0].Pos().Column)
}
//can't do a sum on _id
if strings.EqualFold(ref.Column.Name, "_id") {
if ok && strings.EqualFold(ref.Column.Name, "_id") {
return nil, sql3.NewErrIdColumnNotValidForAggregateFunction(call.Args[0].Pos().Line, call.Args[0].Pos().Column, call.Name.Name)
}
//make sure the ref is sum-able
if !(typeIsInteger(ref.DataType()) || typeIsDecimal(ref.DataType())) {
return nil, sql3.NewErrIntOrDecimalExpressionExpected(ref.Table.NamePos.Line, ref.Table.NamePos.Column)
if !(typeIsInteger(call.Args[0].DataType()) || typeIsDecimal(call.Args[0].DataType())) {
return nil, sql3.NewErrIntOrDecimalExpressionExpected(call.Args[0].Pos().Line, call.Args[0].Pos().Column)
}
call.ResultDataType = ref.DataType()
call.ResultDataType = call.Args[0].DataType()
case "AVG":
// can't do an avg on a *
@ -85,19 +80,15 @@ func (p *ExecutionPlanner) analyzeCallExpression(ctx context.Context, call *pars
return nil, sql3.NewErrCallParameterCountMismatch(call.Rparen.Line, call.Rparen.Column, call.Name.Name, 1, len(call.Args))
}
// if it is a ref, we shouldn't do a avg on the _id
ref, ok := call.Args[0].(*parser.QualifiedRef)
if !ok {
return nil, sql3.NewErrExpectedColumnReference(call.Args[0].Pos().Line, call.Args[0].Pos().Column)
}
//can't do a avg on _id
if strings.EqualFold(ref.Column.Name, "_id") {
if ok && strings.EqualFold(ref.Column.Name, "_id") {
return nil, sql3.NewErrIdColumnNotValidForAggregateFunction(call.Args[0].Pos().Line, call.Args[0].Pos().Column, call.Name.Name)
}
//make sure the ref is avg-able
if !(typeIsInteger(ref.DataType()) || typeIsDecimal(ref.DataType())) {
return nil, sql3.NewErrIntOrDecimalExpressionExpected(ref.Table.NamePos.Line, ref.Table.NamePos.Column)
if !(typeIsInteger(call.Args[0].DataType()) || typeIsDecimal(call.Args[0].DataType())) {
return nil, sql3.NewErrIntOrDecimalExpressionExpected(call.Args[0].Pos().Line, call.Args[0].Pos().Column)
}
call.ResultDataType = parser.NewDataTypeDecimal(4)
@ -153,24 +144,19 @@ func (p *ExecutionPlanner) analyzeCallExpression(ctx context.Context, call *pars
return nil, sql3.NewErrCallParameterCountMismatch(call.Rparen.Line, call.Rparen.Column, call.Name.Name, 1, len(call.Args))
}
// first arg should be a qualified ref
// if it is a ref, we shouldn't do a min/max on the _id
ref, ok := call.Args[0].(*parser.QualifiedRef)
if !ok {
return nil, sql3.NewErrExpectedColumnReference(call.Args[0].Pos().Line, call.Args[0].Pos().Column)
}
// can't do a min/max on _id
if strings.EqualFold(ref.Column.Name, "_id") {
if ok && strings.EqualFold(ref.Column.Name, "_id") {
return nil, sql3.NewErrIdColumnNotValidForAggregateFunction(call.Args[0].Pos().Line, call.Args[0].Pos().Column, call.Name.Name)
}
// make sure the ref is min/max-able
if !(typeIsInteger(ref.DataType()) || typeIsDecimal(ref.DataType()) || typeIsTimestamp(ref.DataType()) || typeIsString(ref.DataType())) {
return nil, sql3.NewErrIntOrDecimalOrTimestampOrStringExpressionExpected(ref.Table.NamePos.Line, ref.Table.NamePos.Column)
if !(typeIsInteger(call.Args[0].DataType()) || typeIsDecimal(call.Args[0].DataType()) || typeIsTimestamp(call.Args[0].DataType()) || typeIsString(call.Args[0].DataType())) {
return nil, sql3.NewErrIntOrDecimalOrTimestampOrStringExpressionExpected(call.Args[0].Pos().Line, call.Args[0].Pos().Column)
}
// return the data type of the referenced column
call.ResultDataType = ref.DataType()
call.ResultDataType = call.Args[0].DataType()
case "SETCONTAINS":
// two arguments
@ -239,7 +225,6 @@ func (p *ExecutionPlanner) analyzeCallExpression(ctx context.Context, call *pars
case "DATEPART":
return p.analyzeFunctionDatePart(call, scope)
case "SUBTABLE":
return p.analyzeFunctionSubtable(call, scope)
case "REVERSE":

10
sql3/planner/expressiontypes.go
View file

@ -537,6 +537,16 @@ func typeIsDecimal(testType parser.ExprDataType) bool {
}
}
// returns true if the type is bit-sliced
func typeIsBSI(testType parser.ExprDataType) bool {
switch testType.(type) {
case *parser.DataTypeInt, *parser.DataTypeDecimal, *parser.DataTypeTimestamp:
return true
default:
return false
}
}
// returns true if the types can be compared
func typesAreComparable(testTypeL parser.ExprDataType, testTypeR parser.ExprDataType) bool {
switch testTypeL.(type) {

38
sql3/planner/oppqlaggregate.go
View file

@ -42,7 +42,7 @@ func (p *PlanOpPQLAggregate) Plan() map[string]interface{} {
if p.filter != nil {
result["filter"] = p.filter.Plan()
}
result["aggregate"] = p.aggregate.AggExpression().Plan()
result["aggregate"] = p.aggregate.FirstChildExpr().Plan()
return result
}
@ -64,7 +64,7 @@ func (p *PlanOpPQLAggregate) Schema() types.Schema {
s := &types.PlannerColumn{
ColumnName: "",
RelationName: "",
Type: p.aggregate.AggExpression().Type(),
Type: p.aggregate.FirstChildExpr().Type(),
}
result[0] = s
return result
@ -114,13 +114,13 @@ func (i *pqlAggregateRowIter) Next(ctx context.Context) (types.Row, error) {
return nil, err
}
expr, ok := i.aggregate.AggExpression().(*qualifiedRefPlanExpression)
expr, ok := i.aggregate.FirstChildExpr().(*qualifiedRefPlanExpression)
if !ok {
return nil, sql3.NewErrInternalf("unexpected aggregate expression type '%T'", i.aggregate.AggExpression())
return nil, sql3.NewErrInternalf("unexpected aggregate expression type '%T'", i.aggregate.FirstChildExpr())
}
switch i.aggregate.AggType() {
case types.AGGREGATE_COUNT_DISTINCT:
switch i.aggregate.(type) {
case *countDistinctPlanExpression:
//make a distinct call
distinctCond := &pql.Call{
Name: "Distinct",
@ -135,7 +135,7 @@ func (i *pqlAggregateRowIter) Next(ctx context.Context) (types.Row, error) {
call = &pql.Call{Name: "Count", Children: []*pql.Call{cond}}
case types.AGGREGATE_COUNT:
case *countPlanExpression:
if cond == nil {
// COUNT() should ignore null values
// if the data type of the expression supports an existence bitmap for
@ -154,7 +154,7 @@ func (i *pqlAggregateRowIter) Next(ctx context.Context) (types.Row, error) {
}
call = &pql.Call{Name: "Count", Children: []*pql.Call{cond}}
case types.AGGREGATE_AVG:
case *avgPlanExpression:
if cond == nil {
// COUNT() should ignore null values
// if the data type of the expression supports an existence bitmap for
@ -178,7 +178,7 @@ func (i *pqlAggregateRowIter) Next(ctx context.Context) (types.Row, error) {
Children: []*pql.Call{cond},
}
case types.AGGREGATE_SUM:
case *sumPlanExpression:
if cond == nil {
cond = &pql.Call{Name: "All"}
}
@ -188,7 +188,7 @@ func (i *pqlAggregateRowIter) Next(ctx context.Context) (types.Row, error) {
Children: []*pql.Call{cond},
}
case types.AGGREGATE_MAX:
case *maxPlanExpression:
if cond == nil {
cond = &pql.Call{Name: "All"}
}
@ -199,7 +199,7 @@ func (i *pqlAggregateRowIter) Next(ctx context.Context) (types.Row, error) {
Children: []*pql.Call{cond},
}
case types.AGGREGATE_MIN:
case *minPlanExpression:
if cond == nil {
cond = &pql.Call{Name: "All"}
}
@ -210,13 +210,12 @@ func (i *pqlAggregateRowIter) Next(ctx context.Context) (types.Row, error) {
Children: []*pql.Call{cond},
}
case types.AGGREGATE_PERCENTILE:
additionalExprs := i.aggregate.AggAdditionalExpr()
if len(additionalExprs) != 1 {
return nil, sql3.NewErrInternalf("unexpected AggAdditionalExpr() length (%d)", len(additionalExprs))
case *percentilePlanExpression:
additionalExprs := i.aggregate.Children()
if len(additionalExprs) != 2 {
return nil, sql3.NewErrInternalf("unexpected Children() length (%d)", len(additionalExprs))
}
nthExpr := additionalExprs[0]
nthExpr := additionalExprs[1]
nthValue, err := nthExpr.Evaluate(nil)
if err != nil {
@ -245,7 +244,7 @@ func (i *pqlAggregateRowIter) Next(ctx context.Context) (types.Row, error) {
}
default:
return nil, sql3.NewErrInternalf("unhandled aggregate type '%d'", i.aggregate.AggType())
return nil, sql3.NewErrInternalf("unhandled aggregate type '%T'", i.aggregate)
}
tbl, err := i.planner.schemaAPI.TableByName(ctx, dax.TableName(i.tableName))
@ -268,7 +267,8 @@ func (i *pqlAggregateRowIter) Next(ctx context.Context) (types.Row, error) {
i.resultValue = int64(actualResult.Val)
case *parser.DataTypeDecimal:
if i.aggregate.AggType() == types.AGGREGATE_AVG {
_, isAvg := i.aggregate.(*avgPlanExpression)
if isAvg {
if actualResult.DecimalVal == nil {
average := float64(actualResult.Val) / float64(actualResult.Count)
daverage, err := pql.FromFloat64WithScale(average, int(t.Scale))

34
sql3/planner/oppqlgroupby.go
View file

@ -44,7 +44,7 @@ func (p *PlanOpPQLGroupBy) Plan() map[string]interface{} {
if p.filter != nil {
result["filter"] = p.filter.Plan()
}
result["aggregate"] = p.aggregate.AggExpression().Plan()
result["aggregate"] = p.aggregate.FirstChildExpr().Plan()
ps := make([]interface{}, 0)
for _, e := range p.groupByExprs {
ps = append(ps, e.Plan())
@ -82,7 +82,7 @@ func (p *PlanOpPQLGroupBy) Schema() types.Schema {
s := &types.PlannerColumn{
ColumnName: p.aggregate.String(),
RelationName: "",
Type: p.aggregate.AggExpression().Type(),
Type: p.aggregate.FirstChildExpr().Type(),
}
result[len(p.groupByExprs)] = s
@ -159,16 +159,16 @@ func (i *pqlGroupByRowIter) Next(ctx context.Context) (types.Row, error) {
}
// Apply filter & aggregate, if set.
aggExpr, ok := i.aggregate.AggExpression().(*qualifiedRefPlanExpression)
aggExpr, ok := i.aggregate.FirstChildExpr().(*qualifiedRefPlanExpression)
if !ok {
return nil, sql3.NewErrInternalf("unexpected aggregate expression type '%T'", i.aggregate.AggExpression())
return nil, sql3.NewErrInternalf("unexpected aggregate expression type '%T'", i.aggregate.FirstChildExpr())
}
switch i.aggregate.AggType() {
case types.AGGREGATE_COUNT:
switch i.aggregate.(type) {
case *countPlanExpression:
//nop
case types.AGGREGATE_COUNT_DISTINCT:
case *countDistinctPlanExpression:
aggregate := &pql.Call{
Name: "Count",
Children: []*pql.Call{{
@ -178,24 +178,24 @@ func (i *pqlGroupByRowIter) Next(ctx context.Context) (types.Row, error) {
}
call.Args["aggregate"] = aggregate
case types.AGGREGATE_SUM, types.AGGREGATE_AVG:
case *sumPlanExpression, *avgPlanExpression:
aggregate := &pql.Call{
Name: "Sum",
Args: map[string]interface{}{"field": aggExpr.columnName},
}
call.Args["aggregate"] = aggregate
case types.AGGREGATE_PERCENTILE:
case *percentilePlanExpression:
return nil, sql3.NewErrAggregateNotAllowedInGroupBy(0, 0, "PERCENTILE()")
case types.AGGREGATE_MIN:
case *minPlanExpression:
return nil, sql3.NewErrAggregateNotAllowedInGroupBy(0, 0, "MIN()")
case types.AGGREGATE_MAX:
case *maxPlanExpression:
return nil, sql3.NewErrAggregateNotAllowedInGroupBy(0, 0, "MAX()")
default:
return nil, sql3.NewErrInternalf("unexpected agg function type: %d", i.aggregate.AggType())
return nil, sql3.NewErrInternalf("unexpected agg function type: '%T'", i.aggregate)
}
if cond != nil {
call.Args["filter"] = cond
@ -244,14 +244,14 @@ func (i *pqlGroupByRowIter) Next(ctx context.Context) (types.Row, error) {
}
//now populate the aggregate value
aggIdx := len(i.groupByColumns)
switch i.aggregate.AggType() {
case types.AGGREGATE_COUNT:
switch i.aggregate.(type) {
case *countPlanExpression:
row[aggIdx] = int64(group.Count)
case types.AGGREGATE_COUNT_DISTINCT, types.AGGREGATE_SUM:
case *countDistinctPlanExpression, *sumPlanExpression:
row[aggIdx] = int64(group.Agg)
case types.AGGREGATE_AVG:
case *avgPlanExpression:
if group.DecimalAgg == nil {
average := float64(group.Agg) / float64(group.Count)
row[aggIdx] = pql.NewDecimal(int64(average*10000), 4)
@ -260,7 +260,7 @@ func (i *pqlGroupByRowIter) Next(ctx context.Context) (types.Row, error) {
row[aggIdx] = pql.NewDecimal(int64(average*10000), 4)
}
default:
return nil, sql3.NewErrInternalf("unhandled aggregate function type '%v'", i.aggregate.AggType())
return nil, sql3.NewErrInternalf("unhandled aggregate function type '%T'", i.aggregate)
}
// Move to next result element.

2
sql3/planner/oppqlmultigroupby.go
View file

@ -78,7 +78,7 @@ func (p *PlanOpPQLMultiGroupBy) Schema() types.Schema {
s := &types.PlannerColumn{
ColumnName: aggOp.aggregate.String(),
RelationName: "",
Type: aggOp.aggregate.AggExpression().Type(),
Type: aggOp.aggregate.FirstChildExpr().Type(),
}
result[idx+offset] = s
}

2
sql3/planner/opsystemtable.go
View file

@ -119,7 +119,7 @@ var systemTables = map[string]*systemTable{
name: fbExecRequests,
schema: types.Schema{
&types.PlannerColumn{
RelationName: fbPerformanceCounters,
RelationName: fbExecRequests,
ColumnName: "nodeid",
Type: parser.NewDataTypeString(),
},

45
sql3/planner/planoptimizer.go
View file

@ -636,26 +636,27 @@ func tryToReplaceGroupByWithPQLAggregate(ctx context.Context, a *ExecutionPlanne
return thisNode, true, nil
}
// make sure all the aggregates are bsi types
// we can push down to pql if:
// 1. the expression we are aggregating on is a qualifiedRef
// 2. it is a bsi type
// we always push down to pql if it's a ref and it's the _id column
for _, agg := range thisNode.Aggregates {
aggregable, ok := agg.(types.Aggregable)
if !ok {
return n, false, sql3.NewErrInternalf("unexpected aggregate function arg type '%T'", agg)
}
switch aggregable.AggExpression().Type().(type) {
case *parser.DataTypeID, *parser.DataTypeString:
// if it is any other column other than _id bail
ref, ok := aggregable.AggExpression().(*qualifiedRefPlanExpression)
if ok {
if !strings.EqualFold(ref.columnName, "_id") {
return thisNode, true, nil
}
switch ref := aggregable.FirstChildExpr().(type) {
case *qualifiedRefPlanExpression:
if !strings.EqualFold(ref.columnName, "_id") && !typeIsBSI(ref.Type()) {
return thisNode, true, nil
}
default:
return thisNode, true, nil
}
}
// if we got to here we are good to go
ops := make([]*PlanOpPQLAggregate, 0)
for _, agg := range thisNode.Aggregates {
aggregable, ok := agg.(types.Aggregable)
if !ok {
@ -962,21 +963,26 @@ func fixProjectionReferences(ctx context.Context, a *ExecutionPlanner, n types.P
switch childOp := thisNode.ChildOp.(type) {
case *PlanOpGroupBy, *PlanOpHaving, *PlanOpPQLGroupBy, *PlanOpPQLMultiAggregate, *PlanOpPQLMultiGroupBy:
// get the child op schema
childSchema := childOp.Schema()
// for each of the projections...
for idx, pj := range thisNode.Projections {
// apply a transform
expr, _, err := TransformExpr(pj, func(e types.PlanExpression) (types.PlanExpression, bool, error) {
switch thisAggregate := e.(type) {
case types.Aggregable:
// if we have a Aggregable, the AggExpression() will be a qualified ref
// given we are in the context of a PlanOpProjection with a PlanOpGroupBy/Having
// we can use the ordinal position of the projection as the column index
// if we have a Aggregable we can use the ordinal position of the matching projection
// as the column index
for idx, sc := range childSchema {
if strings.EqualFold(thisAggregate.String(), sc.ColumnName) {
ae := newQualifiedRefPlanExpression("", "", idx, e.Type())
return ae, false, nil
}
}
// if we get to here not finding a match we likely have an error
return nil, true, sql3.NewErrColumnNotFound(0, 0, thisAggregate.String())
case *qualifiedRefPlanExpression:
@ -989,22 +995,13 @@ func fixProjectionReferences(ctx context.Context, a *ExecutionPlanner, n types.P
return thisAggregate, true, nil
}
}
return nil, true, sql3.NewErrColumnNotFound(0, 0, thisAggregate.String())
// we didn't find a match in the schema so we just bail unchanged
return e, true, nil
default:
return e, true, nil
}
}, func(parentExpr, childExpr types.PlanExpression) bool {
// if the parent is an aggregable, and the child is a qualified ref
// we will skip, because the qualified ref should have already been handled in
// fixFieldRefs
switch parentExpr.(type) {
case types.Aggregable:
switch childExpr.(type) {
case *qualifiedRefPlanExpression:
return false
}
}
return true
})
if err != nil {

30
sql3/planner/types/planexpression.go
View file

@ -7,22 +7,6 @@ import (
"github.com/featurebasedb/featurebase/v3/sql3/parser"
)
// TODO(pok) we can get rid of this - we have expression types for all of these now...
type AggregateFunctionType int
// The list of AggregateFunction.
const (
// Special tokens
AGGREGATE_ILLEGAL AggregateFunctionType = iota
AGGREGATE_COUNT
AGGREGATE_COUNT_DISTINCT
AGGREGATE_SUM
AGGREGATE_AVG
AGGREGATE_PERCENTILE
AGGREGATE_MIN
AGGREGATE_MAX
)
// PlanExpression is an expression node for an execution plan
type PlanExpression interface {
fmt.Stringer
@ -44,7 +28,7 @@ type PlanExpression interface {
Plan() map[string]interface{}
}
// Aggregattion buffer is an interface to something that maintains an aggregate during query
// Aggregation buffer is an interface to something that maintains an aggregate during query
// execution
type AggregationBuffer interface {
Eval(ctx context.Context) (interface{}, error)
@ -55,10 +39,16 @@ type AggregationBuffer interface {
type Aggregable interface {
fmt.Stringer
// creates a new aggregation buffer for this aggregate
NewBuffer() (AggregationBuffer, error)
AggType() AggregateFunctionType
AggExpression() PlanExpression
AggAdditionalExpr() []PlanExpression
// convenience to get the first argument of the aggregate
FirstChildExpr() PlanExpression
// returns all the child expressions for this aggregate
Children() []PlanExpression
// returns the type of the aggregate
Type() parser.ExprDataType
}

274
sql3/sql_complex_test.go
View file

File diff suppressed because it is too large Load diff

11
sql3/sql_test.go
View file

@ -28,13 +28,13 @@ func TestSQL_Execute(t *testing.T) {
// Create a table with all field types.
if test.HasTable() {
_, _, err := sql_test.MustQueryRows(t, svr, test.CreateTable())
_, _, _, err := sql_test.MustQueryRows(t, svr, test.CreateTable())
assert.NoError(t, err)
}
if test.HasTable() && test.HasData() {
// Populate fields with data.
_, _, err := sql_test.MustQueryRows(t, svr, test.InsertInto(t))
_, _, _, err := sql_test.MustQueryRows(t, svr, test.InsertInto(t))
assert.NoError(t, err)
}
@ -43,7 +43,7 @@ func TestSQL_Execute(t *testing.T) {
for _, sql := range sqltest.SQLs {
t.Run(fmt.Sprintf("sql-%s", sql), func(t *testing.T) {
log.Printf("SQL: %s", sql)
rows, headers, err := sql_test.MustQueryRows(t, svr, sql)
rows, headers, plan, err := sql_test.MustQueryRows(t, svr, sql)
// Check expected error instead of results.
if sqltest.ExpErr != "" {
@ -96,6 +96,11 @@ func TestSQL_Execute(t *testing.T) {
assert.Contains(t, exp, row)
}
}
if sqltest.PlanCheck != nil {
err := sqltest.PlanCheck(plan)
require.NoError(t, err)
}
})
}
})

37
sql3/test/defs/defs.go
View file

@ -2,7 +2,14 @@
package defs
import (
"context"
"encoding/json"
"fmt"
"strings"
"time"
"github.com/PaesslerAG/gval"
"github.com/PaesslerAG/jsonpath"
)
// TableTests is the list of tests which get run by TestSQL_Execute in
@ -202,3 +209,33 @@ func timestampFromString(s string) time.Time {
}
return tm
}
// operatorPresentAtPath() tests if an named operator exists in a plan
//
// operatorPresentAtPath() takes a FeatureBase query plan as a []byte
// (so we do not have to convert to and from string), a path (as a
// jsonpath expression) and an operator. The function returns nil if
// the result of the jsonpath expression evaluation contains the operator.
func operatorPresentAtPath(jplan []byte, path string, operator string) error {
// fmt.Printf("%s\n", string(jplan))
v := interface{}(nil)
err := json.Unmarshal(jplan, &v)
if err != nil {
return err
}
builder := gval.Full(jsonpath.PlaceholderExtension())
expr, err := builder.NewEvaluable(path)
if err != nil {
return err
}
eval, err := expr(context.Background(), v)
if err != nil {
return err
}
s, ok := eval.(string)
if ok && strings.EqualFold(s, operator) {
return nil
}
return fmt.Errorf("expected '%s' to be present", operator)
}

72
sql3/test/defs/defs_aggregate.go
View file

@ -50,6 +50,9 @@ var countTests = TableTest{
row(int64(6)),
),
Compare: CompareExactUnordered,
PlanCheck: func(jplan []byte) error {
return operatorPresentAtPath(jplan, "$.child.child.operators[0]._op", "*planner.PlanOpPQLAggregate")
},
},
{
SQLs: sqls(
@ -245,7 +248,13 @@ var sumTests = TableTest{
SQLs: sqls(
"SELECT sum(1) AS sum_rows FROM sum_test",
),
ExpErr: "column reference expected",
ExpHdrs: hdrs(
hdr("sum_rows", fldTypeInt),
),
ExpRows: rows(
row(int64(6)),
),
Compare: CompareExactUnordered,
},
{
SQLs: sqls(
@ -277,6 +286,18 @@ var sumTests = TableTest{
),
Compare: CompareExactUnordered,
},
{
SQLs: sqls(
"SELECT sum(d1 + 5) AS sum_rows FROM sum_test",
),
ExpHdrs: hdrs(
hdr("sum_rows", fldTypeDecimal2),
),
ExpRows: rows(
row(pql.NewDecimal(9800, 2)),
),
Compare: CompareExactUnordered,
},
},
}
@ -357,6 +378,22 @@ var avgTests = TableTest{
),
Compare: CompareExactUnordered,
},
{
SQLs: sqls(
"SELECT avg(len(s1)) AS avg_rows FROM avg_test",
),
ExpHdrs: hdrs(
hdr("avg_rows", featurebase.WireQueryField{
Type: dax.BaseTypeDecimal + "(4)",
BaseType: dax.BaseTypeDecimal,
TypeInfo: map[string]interface{}{"scale": int64(4)},
}),
),
ExpRows: rows(
row(pql.NewDecimal(30000, 4)),
),
Compare: CompareExactUnordered,
},
},
}
@ -475,8 +512,13 @@ var minmaxTests = TableTest{
"SELECT min(1) AS p_rows FROM minmax_test",
"SELECT max(1) AS p_rows FROM minmax_test",
),
ExpErr: "column reference expected",
},
ExpHdrs: hdrs(
hdr("p_rows", fldTypeInt),
),
ExpRows: rows(
row(int64(1)),
),
Compare: CompareExactUnordered},
{
SQLs: sqls(
"SELECT min(_id) AS p_rows FROM minmax_test",
@ -508,6 +550,30 @@ var minmaxTests = TableTest{
),
Compare: CompareExactUnordered,
},
{
SQLs: sqls(
"SELECT min(len(s1)) AS p_rows FROM minmax_test",
),
ExpHdrs: hdrs(
hdr("p_rows", fldTypeInt),
),
ExpRows: rows(
row(int64(4)),
),
Compare: CompareExactUnordered,
},
{
SQLs: sqls(
"SELECT max(len(s1)) AS p_rows FROM minmax_test",
),
ExpHdrs: hdrs(
hdr("p_rows", fldTypeInt),
),
ExpRows: rows(
row(int64(4)),
),
Compare: CompareExactUnordered,
},
{
SQLs: sqls(
"SELECT min(i1) AS p_rows FROM minmax_test",

3
sql3/test/defs/types.go
View file

@ -107,6 +107,8 @@ func (tt TableTest) InsertInto(t *testing.T, rowSets ...int) string {
return tt.Table.insertInto(t, rowSets)
}
type PlanCheckFunc func([]byte) error
type SQLTest struct {
name string
SQLs []string
@ -117,6 +119,7 @@ type SQLTest struct {
Compare compareMethod
SortStringKeys bool
ExpRowCount int
PlanCheck PlanCheckFunc
}
// Name returns a string name which can be used to distingish test runs. It

20
sql3/test/helpers.go
View file

@ -3,6 +3,7 @@ package test
import (
"context"
"encoding/json"
"testing"
featurebase "github.com/featurebasedb/featurebase/v3"
@ -12,35 +13,36 @@ import (
uuid "github.com/satori/go.uuid"
)
// MustQueryRows returns the row results as a slice of []interface{}, along with the columns.
func MustQueryRows(tb testing.TB, svr *featurebase.Server, q string) ([][]interface{}, []*featurebase.WireQueryField, error) {
// MustQueryRows returns the row results as a slice of []interface{}, along with the columns, the query plan as a []byte or an error.
func MustQueryRows(tb testing.TB, svr *featurebase.Server, q string) ([][]interface{}, []*featurebase.WireQueryField, []byte, error) {
tb.Helper()
requestId, err := uuid.NewV4()
if err != nil {
return nil, nil, err
return nil, nil, nil, err
}
ctx := fbcontext.WithRequestID(context.Background(), requestId.String())
stmt, err := svr.CompileExecutionPlan(ctx, q)
if err != nil {
return nil, nil, err
return nil, nil, nil, err
}
// get the plan so that code runs during testing
_ = stmt.Plan()
plan := stmt.Plan()
bplan, _ := json.MarshalIndent(plan, "", " ")
ocolumns := stmt.Schema()
rowIter, err := stmt.Iterator(ctx, nil)
if err != nil {
return nil, nil, err
return nil, nil, nil, err
}
results := make([][]interface{}, 0)
next, err := rowIter.Next(ctx)
if err != nil && err != plannertypes.ErrNoMoreRows {
return nil, nil, err
return nil, nil, nil, err
}
for err != plannertypes.ErrNoMoreRows {
result := make([]interface{}, len(ocolumns))
@ -50,7 +52,7 @@ func MustQueryRows(tb testing.TB, svr *featurebase.Server, q string) ([][]interf
results = append(results, result)
next, err = rowIter.Next(ctx)
if err != nil && err != plannertypes.ErrNoMoreRows {
return nil, nil, err
return nil, nil, nil, err
}
}
// temporarily transform to Columns()
@ -63,5 +65,5 @@ func MustQueryRows(tb testing.TB, svr *featurebase.Server, q string) ([][]interf
TypeInfo: oc.Type.TypeInfo(),
})
}
return results, cols, nil
return results, cols, bplan, nil
}