diff --git a/pkg/sql/plan/build_test.go b/pkg/sql/plan/build_test.go index 14553b5c55b5c..897b2b1dc521c 100644 --- a/pkg/sql/plan/build_test.go +++ b/pkg/sql/plan/build_test.go @@ -5561,6 +5561,15 @@ func TestSubQuery(t *testing.T) { WHERE n3.N_NATIONKEY = n2.N_NATIONKEY AND n2.N_NATIONKEY < n1.N_NATIONKEY ) )`, // two-level correlated ALL subquery + `SELECT n1.N_NATIONKEY, + (SELECT MAX(n2.N_REGIONKEY) + FROM NATION n2 + WHERE n2.N_REGIONKEY = ( + SELECT MAX(n3.N_REGIONKEY) + FROM NATION n3 + WHERE n3.N_NATIONKEY = n1.N_NATIONKEY + )) + FROM NATION n1`, // two-level correlated scalar aggregate subquery } runTestShouldPass(mock, t, sqls, false, false) diff --git a/pkg/sql/plan/build_util_test.go b/pkg/sql/plan/build_util_test.go index eb7c78227f4db..d5af290729817 100644 --- a/pkg/sql/plan/build_util_test.go +++ b/pkg/sql/plan/build_util_test.go @@ -603,7 +603,16 @@ func TestAssignmentCastPreservesNestedExplicitTemporalCast(t *testing.T) { // (cast_strict): an over-length value is rejected, not silently truncated. func TestBuildGeneratedExprUsesStrictForCharVarchar(t *testing.T) { proc := testutil.NewProcess(t) - moruntime.ServiceRuntime(proc.GetService()).SetGlobalVariables( + rt := moruntime.ServiceRuntime(proc.GetService()) + original, hadOriginal := rt.GetGlobalVariables(moruntime.MOProtocolVersion) + t.Cleanup(func() { + if hadOriginal { + rt.SetGlobalVariables(moruntime.MOProtocolVersion, original) + } else { + rt.SetGlobalVariables(moruntime.MOProtocolVersion, defines.MORPCLatestVersion) + } + }) + rt.SetGlobalVariables( moruntime.MOProtocolVersion, defines.MORPCVersion5, ) diff --git a/pkg/sql/plan/flatten_subquery.go b/pkg/sql/plan/flatten_subquery.go index 6bee2fb18a07a..2d5454d4d8be0 100644 --- a/pkg/sql/plan/flatten_subquery.go +++ b/pkg/sql/plan/flatten_subquery.go @@ -57,25 +57,44 @@ var ( ) func (builder *QueryBuilder) flattenSubqueries(nodeID int32, expr *plan.Expr, ctx *BindContext) (int32, *plan.Expr, error) { + return builder.flattenSubqueriesWithContext(nodeID, expr, ctx, false) +} + +func (builder *QueryBuilder) flattenFilterSubqueries(nodeID int32, expr *plan.Expr, ctx *BindContext) (int32, *plan.Expr, error) { + return builder.flattenSubqueriesWithContext(nodeID, expr, ctx, true) +} + +func (builder *QueryBuilder) flattenSubqueriesWithContext( + nodeID int32, + expr *plan.Expr, + ctx *BindContext, + nullResultRejected bool, +) (int32, *plan.Expr, error) { var err error switch exprImpl := expr.Expr.(type) { case *plan.Expr_F: + childNullResultRejected := nullResultRejected && nullPropagatesThroughDeepScalarConsumer(exprImpl.F.Func) for i, arg := range exprImpl.F.Args { - nodeID, exprImpl.F.Args[i], err = builder.flattenSubqueries(nodeID, arg, ctx) + nodeID, exprImpl.F.Args[i], err = builder.flattenSubqueriesWithContext(nodeID, arg, ctx, childNullResultRejected) if err != nil { return 0, nil, err } } case *plan.Expr_Sub: - nodeID, expr, err = builder.flattenSubquery(nodeID, exprImpl.Sub, ctx) + nodeID, expr, err = builder.flattenSubquery(nodeID, exprImpl.Sub, ctx, nullResultRejected) } return nodeID, expr, err } -func (builder *QueryBuilder) flattenSubquery(nodeID int32, subquery *plan.SubqueryRef, ctx *BindContext) (int32, *plan.Expr, error) { +func (builder *QueryBuilder) flattenSubquery( + nodeID int32, + subquery *plan.SubqueryRef, + ctx *BindContext, + nullResultRejected bool, +) (int32, *plan.Expr, error) { if subquery.Child != nil && hasSubquery(subquery.Child) { return 0, nil, moerr.NewNotSupported(builder.GetContext(), "a quantified subquery's left operand can't contain subquery") } @@ -152,10 +171,19 @@ func (builder *QueryBuilder) flattenSubquery(nodeID int32, subquery *plan.Subque filterPreds, joinPreds := decreaseDepthAndDispatch(preds) if len(filterPreds) > 0 { - if !canPullupDeepCorrelatedPredicates(subquery.Typ) { + deepScalarAggregate := subquery.Typ == plan.SubqueryRef_SCALAR && + subCtx.hasSingleRow && len(subCtx.groups) == 0 && len(subCtx.aggregates) > 0 && + scalarAggregateResultReturnsNullOnEmpty(subCtx) && nullResultRejected && + builder.scalarAggregatePlanSupportsDeepCorrelation(subID, subCtx.aggregateTag) + if !deepScalarAggregate && !canPullupDeepCorrelatedPredicates(subquery.Typ) { return 0, nil, moerr.NewNYIf(builder.GetContext(), "correlated columns in %s subquery deeper than 1 level will be supported in future version", subquery.Typ.String()) } - if builder.hasInnerColumnInDeepCorrelatedFilters(subID, filterPreds) { + // MARK JOIN only exposes its marker, so a predicate that still refers + // to the inner relation cannot be moved above it. A scalar aggregate + // is different: pulling the predicate through the aggregate has already + // turned its inner columns into grouping keys, and LEFT JOIN preserves + // those keys so the enclosing subquery can pull them up again. + if !deepScalarAggregate && builder.hasInnerColumnInDeepCorrelatedFilters(subID, filterPreds) { return 0, nil, moerr.NewNYIf(builder.GetContext(), "deep correlated predicate containing inner columns cannot be pulled above mark join") } } @@ -805,6 +833,125 @@ func (builder *QueryBuilder) findAggrCount(aggrs []*plan.Expr) bool { return false } +// allAggregatesReturnNullOnEmpty is deliberately conservative. Pulling a deep +// correlated predicate through an aggregate turns its inner expression into a +// GROUP BY key. If that key has no input rows, the grouped plan has no row and +// the LEFT JOIN exposes NULL. Matching the aggregate's SQL result for empty +// input is necessary but not sufficient: the complete consuming expression +// must also reject that NULL. Unknown and newly added aggregates remain on the +// NYI path until their empty-input contract is verified here. +func allAggregatesReturnNullOnEmpty(aggrs []*plan.Expr) bool { + if len(aggrs) == 0 { + return false + } + + for _, aggr := range aggrs { + f := aggr.GetF() + if f == nil || f.Func == nil { + return false + } + + fid, _ := function.DecodeOverloadID(f.Func.Obj & function.DistinctMask) + switch fid { + case function.MIN, function.MAX, function.SUM, function.AVG, function.ANY_VALUE: + default: + return false + } + } + return true +} + +// scalarAggregateResultReturnsNullOnEmpty verifies the missing-group contract +// through the scalar subquery's own result projection. It is not enough for +// the underlying aggregates to return NULL: a projection such as +// COALESCE(MAX(...), 0) observes that NULL, while the grouped rewrite has no +// row on which to evaluate the projection at all. +func scalarAggregateResultReturnsNullOnEmpty(ctx *BindContext) bool { + return len(ctx.projects) > 0 && + allAggregatesReturnNullOnEmpty(ctx.aggregates) && + nullPropagatesFromAggregate(ctx.projects[0], ctx.aggregateTag) +} + +// scalarAggregatePlanSupportsDeepCorrelation verifies the complete path from +// the scalar root to its implicit aggregate. pullupThroughAgg adds the deep +// correlation key to GROUP BY, so row limiting that was once evaluated inside +// each scalar invocation would otherwise become global across all keys. +// +// PROJECT is the only wrapper proven to remain per-key after that rewrite. +// LIMIT, OFFSET, rank, and every other operator stay on the NYI path until they +// are explicitly rewritten or proven partition-local. +func (builder *QueryBuilder) scalarAggregatePlanSupportsDeepCorrelation(nodeID, aggregateTag int32) bool { + for range builder.qry.Nodes { + if nodeID < 0 || int(nodeID) >= len(builder.qry.Nodes) { + return false + } + + node := builder.qry.Nodes[nodeID] + if node == nil || node.Limit != nil || node.Offset != nil || node.RankOption != nil { + return false + } + + switch node.NodeType { + case plan.Node_PROJECT: + if len(node.Children) != 1 { + return false + } + nodeID = node.Children[0] + + case plan.Node_AGG: + return len(node.BindingTags) > 1 && node.BindingTags[1] == aggregateTag + + default: + return false + } + } + + return false +} + +func nullPropagatesFromAggregate(expr *plan.Expr, aggregateTag int32) bool { + switch exprImpl := expr.Expr.(type) { + case *plan.Expr_Col: + return exprImpl.Col.RelPos == aggregateTag + + case *plan.Expr_F: + if !nullPropagatesThroughDeepScalarConsumer(exprImpl.F.Func) { + return false + } + for _, arg := range exprImpl.F.Args { + if nullPropagatesFromAggregate(arg, aggregateTag) { + return true + } + } + } + + return false +} + +// nullPropagatesThroughDeepScalarConsumer identifies the deliberately narrow +// set of scalar functions through which a missing deep scalar result can be +// proven to remain NULL. Combined with a FILTER root, that means both the SQL +// expression and the decorrelated plan discard the enclosing input row. +// +// Keep this list conservative. In particular, COALESCE/CASE can observe NULL, +// and logical AND is not NULL-propagating for every input combination. +func nullPropagatesThroughDeepScalarConsumer(fn *plan.ObjectRef) bool { + if fn == nil { + return false + } + + fid, _ := function.DecodeOverloadID(fn.Obj) + switch fid { + case function.EQUAL, function.NOT_EQUAL, + function.GREAT_THAN, function.GREAT_EQUAL, + function.LESS_THAN, function.LESS_EQUAL, + function.NOT, function.CAST, function.CAST_STRICT: + return true + default: + return false + } +} + func (builder *QueryBuilder) findNonEqPred(preds []*plan.Expr) bool { for _, pred := range preds { if containsNonEqComparison(pred) { diff --git a/pkg/sql/plan/flatten_subquery_test.go b/pkg/sql/plan/flatten_subquery_test.go index 9c0644cbc3f76..1b91be26f73ea 100644 --- a/pkg/sql/plan/flatten_subquery_test.go +++ b/pkg/sql/plan/flatten_subquery_test.go @@ -72,6 +72,212 @@ func TestHasInnerColumnInDeepCorrelatedFilters(t *testing.T) { })) } +func TestScalarAggregatePlanSupportsDeepCorrelation(t *testing.T) { + const ( + groupTag int32 = 10 + aggregateTag int32 = 11 + ) + + tests := []struct { + name string + wrapper plan.Node_NodeType + configure func(*plan.Node) + directAgg bool + wrongTag bool + want bool + }{ + {name: "direct aggregate", directAgg: true, want: true}, + {name: "projection", wrapper: plan.Node_PROJECT, want: true}, + { + name: "limit", + wrapper: plan.Node_PROJECT, + configure: func(node *plan.Node) { + node.Limit = makePlan2Uint64ConstExprWithType(1) + }, + }, + { + name: "offset", + wrapper: plan.Node_PROJECT, + configure: func(node *plan.Node) { + node.Offset = makePlan2Uint64ConstExprWithType(1) + }, + }, + { + name: "rank", + wrapper: plan.Node_PROJECT, + configure: func(node *plan.Node) { + node.RankOption = &plan.RankOption{Mode: "rank"} + }, + }, + {name: "sort", wrapper: plan.Node_SORT}, + {name: "distinct", wrapper: plan.Node_DISTINCT}, + {name: "filter", wrapper: plan.Node_FILTER}, + {name: "wrong aggregate", directAgg: true, wrongTag: true}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + tag := aggregateTag + if test.wrongTag { + tag++ + } + nodes := []*plan.Node{{ + NodeType: plan.Node_AGG, + BindingTags: []int32{groupTag, tag}, + }} + rootID := int32(0) + if !test.directAgg { + wrapper := &plan.Node{ + NodeType: test.wrapper, + Children: []int32{0}, + } + if test.configure != nil { + test.configure(wrapper) + } + nodes = append(nodes, wrapper) + rootID = 1 + } + + builder := &QueryBuilder{qry: &plan.Query{Nodes: nodes}} + require.Equal(t, test.want, + builder.scalarAggregatePlanSupportsDeepCorrelation(rootID, aggregateTag)) + }) + } +} + +func TestNestedCorrelatedScalarAggregatePullsUpGroupingKey(t *testing.T) { + logicPlan, err := runOneStmt(NewMockOptimizer(true), t, ` + SELECT n1.N_NATIONKEY, + (SELECT MAX(n2.N_REGIONKEY) + FROM NATION n2 + WHERE n2.N_REGIONKEY = ( + SELECT MAX(n3.N_REGIONKEY) + FROM NATION n3 + WHERE n3.N_NATIONKEY = n1.N_NATIONKEY)) + FROM NATION n1`) + require.NoError(t, err) + + query := logicPlan.GetQuery() + require.NotNil(t, query) + require.NotEmpty(t, query.Steps) + + visited := make(map[int32]bool) + var visit func(int32) + visit = func(nodeID int32) { + require.GreaterOrEqual(t, nodeID, int32(0)) + require.Less(t, int(nodeID), len(query.Nodes)) + if visited[nodeID] { + return + } + visited[nodeID] = true + + node := query.Nodes[nodeID] + require.NotNil(t, node) + for _, exprs := range [][]*plan.Expr{ + node.ProjectList, + node.OnList, + node.FilterList, + node.GroupBy, + node.AggList, + } { + for _, expr := range exprs { + require.False(t, hasCorrCol(expr), "reachable %s node contains a correlated expression", node.NodeType) + } + } + for _, orderBy := range node.OrderBy { + require.False(t, hasCorrCol(orderBy.Expr), "reachable SORT contains a correlated expression") + } + for _, childID := range node.Children { + visit(childID) + } + } + + for _, rootID := range query.Steps { + visit(rootID) + } +} + +func TestNestedCorrelatedScalarStillRejectsUnsafeShapes(t *testing.T) { + for _, sql := range []string{ + `SELECT n1.N_NATIONKEY, + (SELECT MAX(n2.N_REGIONKEY) + FROM NATION n2 + WHERE n2.N_REGIONKEY = ( + SELECT n3.N_REGIONKEY + FROM NATION n3 + WHERE n3.N_NATIONKEY = n1.N_NATIONKEY)) + FROM NATION n1`, + `SELECT n1.N_NATIONKEY, + (SELECT MAX(n2.N_REGIONKEY) + FROM NATION n2 + WHERE n2.N_REGIONKEY = ( + SELECT COUNT(*) + FROM NATION n3 + WHERE n3.N_NATIONKEY = n1.N_NATIONKEY)) + FROM NATION n1`, + `SELECT n1.N_NATIONKEY, + (SELECT MAX(n2.N_REGIONKEY) + FROM NATION n2 + WHERE n2.N_REGIONKEY = ( + SELECT APPROX_COUNT(n3.N_REGIONKEY) + FROM NATION n3 + WHERE n3.N_NATIONKEY = n1.N_NATIONKEY)) + FROM NATION n1`, + `SELECT n1.N_NATIONKEY, + (SELECT MAX(n2.N_REGIONKEY) + FROM NATION n2 + WHERE n2.N_REGIONKEY = ( + SELECT APPROX_COUNT_DISTINCT(n3.N_REGIONKEY) + FROM NATION n3 + WHERE n3.N_NATIONKEY = n1.N_NATIONKEY)) + FROM NATION n1`, + `SELECT n1.N_NATIONKEY, + (SELECT COUNT(COALESCE(( + SELECT MAX(n3.N_REGIONKEY) + FROM NATION n3 + WHERE n3.N_NATIONKEY = n1.N_NATIONKEY), 0)) + FROM NATION n2) + FROM NATION n1`, + `SELECT n1.N_NATIONKEY, + (SELECT MAX(n2.N_REGIONKEY) + FROM NATION n2 + WHERE n2.N_REGIONKEY = COALESCE(( + SELECT MAX(n3.N_REGIONKEY) + FROM NATION n3 + WHERE n3.N_NATIONKEY = n1.N_NATIONKEY), 0)) + FROM NATION n1`, + `SELECT n1.N_NATIONKEY, + (SELECT MAX(n2.N_REGIONKEY) + FROM NATION n2 + WHERE n2.N_REGIONKEY = ( + SELECT COALESCE(MAX(n3.N_REGIONKEY), 0) + FROM NATION n3 + WHERE n3.N_NATIONKEY = n1.N_NATIONKEY)) + FROM NATION n1`, + `SELECT n1.N_NATIONKEY, + (SELECT MAX(n2.N_REGIONKEY) + FROM NATION n2 + WHERE n2.N_REGIONKEY = ( + SELECT MAX(n3.N_REGIONKEY) + FROM NATION n3 + WHERE n3.N_NATIONKEY = n1.N_NATIONKEY + LIMIT 1)) + FROM NATION n1`, + `SELECT n1.N_NATIONKEY, + (SELECT MAX(n2.N_REGIONKEY) + FROM NATION n2 + WHERE n2.N_REGIONKEY = ( + SELECT MAX(n3.N_REGIONKEY) + FROM NATION n3 + WHERE n3.N_NATIONKEY = n1.N_NATIONKEY + LIMIT 1 OFFSET 1)) + FROM NATION n1`, + } { + _, err := runOneStmt(NewMockOptimizer(true), t, sql) + require.ErrorContains(t, err, "correlated columns in SCALAR subquery deeper than 1 level") + } +} + func TestInSubqueryJoinShapePreservesThreeValuedSemantics(t *testing.T) { tests := []struct { name string diff --git a/pkg/sql/plan/query_builder.go b/pkg/sql/plan/query_builder.go index b7f70f09bf459..490394c6f8d26 100644 --- a/pkg/sql/plan/query_builder.go +++ b/pkg/sql/plan/query_builder.go @@ -6687,7 +6687,7 @@ func (builder *QueryBuilder) bindWhere( } var expr *plan.Expr for _, cond := range whereList { - if nodeID, expr, err = builder.flattenSubqueries(nodeID, cond, ctx); err != nil { + if nodeID, expr, err = builder.flattenFilterSubqueries(nodeID, cond, ctx); err != nil { return } boundFilterList = append(boundFilterList, expr) @@ -7884,7 +7884,7 @@ func (builder *QueryBuilder) appendSampleNode( var expr *plan.Expr for _, cond := range boundHavingList { - if nodeID, expr, err = builder.flattenSubqueries(nodeID, cond, ctx); err != nil { + if nodeID, expr, err = builder.flattenFilterSubqueries(nodeID, cond, ctx); err != nil { return } @@ -7944,7 +7944,7 @@ func (builder *QueryBuilder) appendAggNode( var expr *plan.Expr for _, cond := range preWindowHavingList { - if nodeID, expr, err = builder.flattenSubqueries(nodeID, cond, ctx); err != nil { + if nodeID, expr, err = builder.flattenFilterSubqueries(nodeID, cond, ctx); err != nil { return } @@ -8102,7 +8102,7 @@ func (builder *QueryBuilder) appendWindowNode( var expr *plan.Expr for _, cond := range postWindowHavingList { - if nodeID, expr, err = builder.flattenSubqueries(nodeID, cond, ctx); err != nil { + if nodeID, expr, err = builder.flattenFilterSubqueries(nodeID, cond, ctx); err != nil { return } @@ -9648,7 +9648,7 @@ func (builder *QueryBuilder) buildJoinTable(tbl *tree.JoinTableExpr, ctx *BindCo var onConds, filterConds []*plan.Expr for _, cond := range joinConds { if hasSubquery(cond) { - nodeID, cond, err = builder.flattenSubqueries(nodeID, cond, ctx) + nodeID, cond, err = builder.flattenFilterSubqueries(nodeID, cond, ctx) if err != nil { return 0, err } diff --git a/test/distributed/cases/subquery/nested-correlated-scalar.result b/test/distributed/cases/subquery/nested-correlated-scalar.result new file mode 100644 index 0000000000000..055f1d8e48a08 --- /dev/null +++ b/test/distributed/cases/subquery/nested-correlated-scalar.result @@ -0,0 +1,89 @@ +drop database if exists test_nested_correlated_scalar; +create database test_nested_correlated_scalar; +use test_nested_correlated_scalar; +create table j_dim (id int); +create table j_fact (id int, dim_id int, ts int, val double); +insert into j_dim values (1), (2), (3); +insert into j_fact values +(1, 1, 10, 100), +(2, 1, 20, 200), +(3, 2, 20, 220), +(4, 2, 30, 300), +(5, 99, 0, 900); +SELECT a.id, +(SELECT MAX(x.val) +FROM j_fact x +WHERE x.ts = (SELECT MAX(y.ts) +FROM j_fact y +WHERE y.dim_id = a.id) +) AS latest +FROM j_dim a +ORDER BY a.id; +id latest +1 220 +2 300 +3 null +SELECT a.id, +(SELECT MAX(x.val) +FROM j_fact x +WHERE x.ts = (SELECT APPROX_COUNT(y.ts) +FROM j_fact y +WHERE y.dim_id = a.id) +) AS latest +FROM j_dim a +WHERE a.id = 3; +(correlated columns in SCALAR subquery deeper than 1 level.*) +SELECT a.id, +(SELECT COUNT(COALESCE( +(SELECT MAX(y.ts) +FROM j_fact y +WHERE y.dim_id = a.id), +0)) +FROM j_fact x) AS actual +FROM j_dim a +WHERE a.id = 3; +(correlated columns in SCALAR subquery deeper than 1 level.*) +SELECT a.id, +(SELECT MAX(x.val) +FROM j_fact x +WHERE x.ts = COALESCE( +(SELECT MAX(y.ts) +FROM j_fact y +WHERE y.dim_id = a.id), +0)) AS latest +FROM j_dim a +WHERE a.id = 3; +(correlated columns in SCALAR subquery deeper than 1 level.*) +SELECT a.id, +(SELECT MAX(x.val) +FROM j_fact x +WHERE x.ts = ( +SELECT COALESCE(MAX(y.ts), 0) +FROM j_fact y +WHERE y.dim_id = a.id)) AS latest +FROM j_dim a +WHERE a.id = 3; +(correlated columns in SCALAR subquery deeper than 1 level.*) +SELECT a.id, +(SELECT MAX(x.val) +FROM j_fact x +WHERE x.ts = ( +SELECT MAX(y.ts) +FROM j_fact y +WHERE y.dim_id = a.id +LIMIT 1)) AS latest +FROM j_dim a +ORDER BY a.id; +(correlated columns in SCALAR subquery deeper than 1 level.*) +SELECT a.id, +(SELECT MAX(x.val) +FROM j_fact x +WHERE x.ts = ( +SELECT MAX(y.ts) +FROM j_fact y +WHERE y.dim_id = a.id +LIMIT 1 OFFSET 1)) AS latest +FROM j_dim a +ORDER BY a.id; +(correlated columns in SCALAR subquery deeper than 1 level.*) +drop database test_nested_correlated_scalar; diff --git a/test/distributed/cases/subquery/nested-correlated-scalar.sql b/test/distributed/cases/subquery/nested-correlated-scalar.sql new file mode 100644 index 0000000000000..4ad59aad9e17c --- /dev/null +++ b/test/distributed/cases/subquery/nested-correlated-scalar.sql @@ -0,0 +1,123 @@ +-- @suite +-- @setup +drop database if exists test_nested_correlated_scalar; +create database test_nested_correlated_scalar; +use test_nested_correlated_scalar; +create table j_dim (id int); +create table j_fact (id int, dim_id int, ts int, val double); +insert into j_dim values (1), (2), (3); +insert into j_fact values + (1, 1, 10, 100), + (2, 1, 20, 200), + (3, 2, 20, 220), + (4, 2, 30, 300), + (5, 99, 0, 900); + +-- @case +-- @desc: scalar aggregate nested inside another scalar aggregate may correlate two levels up +-- @label:bvt +SELECT a.id, + (SELECT MAX(x.val) + FROM j_fact x + WHERE x.ts = (SELECT MAX(y.ts) + FROM j_fact y + WHERE y.dim_id = a.id) + ) AS latest +FROM j_dim a +ORDER BY a.id; + +-- approx_count returns 0 for empty input. Until the deep decorrelation can +-- synthesize that missing aggregate row, keep this shape on the NYI path +-- instead of silently turning 0 into NULL. id=3 has no matching y rows while +-- x contains ts=0, so an unsafe rewrite would return NULL instead of 900. +-- @pattern +SELECT a.id, + (SELECT MAX(x.val) + FROM j_fact x + WHERE x.ts = (SELECT APPROX_COUNT(y.ts) + FROM j_fact y + WHERE y.dim_id = a.id) + ) AS latest +FROM j_dim a +WHERE a.id = 3; + +-- Even a NULL-on-empty inner aggregate is unsafe when its complete consuming +-- expression observes that NULL. For id=3, MAX(y.ts) is NULL and COALESCE +-- produces 0 for each of the five x rows, so SQL COUNT returns 5. The current +-- deep rewrite cannot synthesize the missing per-key aggregate row and must +-- keep this shape on the NYI path instead of returning 0. +-- @pattern +SELECT a.id, + (SELECT COUNT(COALESCE( + (SELECT MAX(y.ts) + FROM j_fact y + WHERE y.dim_id = a.id), + 0)) + FROM j_fact x) AS actual + FROM j_dim a + WHERE a.id = 3; + +-- The same missing-key case is unsafe when COALESCE feeds the enclosing +-- filter. SQL semantics match x.ts=0 and return 900; dropping the x row would +-- incorrectly expose NULL, so this shape also remains NYI. +-- @pattern +SELECT a.id, + (SELECT MAX(x.val) + FROM j_fact x + WHERE x.ts = COALESCE( + (SELECT MAX(y.ts) + FROM j_fact y + WHERE y.dim_id = a.id), + 0)) AS latest + FROM j_dim a + WHERE a.id = 3; + +-- A NULL-observing projection inside the deep scalar is unsafe too. For id=3, +-- SQL evaluates COALESCE(MAX(y.ts), 0) to 0 and matches the x.ts=0 row, yielding +-- 900. The grouped rewrite has no y.dim_id=3 row on which to run COALESCE, so +-- it would expose NULL and incorrectly drop that x row. Keep this shape NYI. +-- @pattern +SELECT a.id, + (SELECT MAX(x.val) + FROM j_fact x + WHERE x.ts = ( + SELECT COALESCE(MAX(y.ts), 0) + FROM j_fact y + WHERE y.dim_id = a.id)) AS latest + FROM j_dim a + WHERE a.id = 3; + +-- LIMIT 1 is redundant for an implicit scalar aggregate when evaluated once +-- per outer row. After decorrelation adds dim_id to GROUP BY, however, leaving +-- the limit on the grouped plan would keep only one correlation key globally. +-- Reject the shape until LIMIT can be rewritten per key. This query spans two +-- matching outer keys and one missing key so a global limit cannot hide. +-- @pattern +SELECT a.id, + (SELECT MAX(x.val) + FROM j_fact x + WHERE x.ts = ( + SELECT MAX(y.ts) + FROM j_fact y + WHERE y.dim_id = a.id + LIMIT 1)) AS latest + FROM j_dim a + ORDER BY a.id; + +-- OFFSET 1 removes the sole implicit aggregate row independently for every +-- outer key. A global offset after grouping would instead skip only one key +-- and expose another, so this topology must remain NYI too. +-- @pattern +SELECT a.id, + (SELECT MAX(x.val) + FROM j_fact x + WHERE x.ts = ( + SELECT MAX(y.ts) + FROM j_fact y + WHERE y.dim_id = a.id + LIMIT 1 OFFSET 1)) AS latest + FROM j_dim a + ORDER BY a.id; + +-- @teardown +drop database test_nested_correlated_scalar;