Skip to content

Commit e15f531

Browse files
general-kroll-4-lifegeneral-kroll-4-life
committed
composition-of-indirects
Summary: - Support for composition of indirects, including `view`, `materialized view` and subqueries. - Added robot test `View Depth Limitation Error Message Shows Correct Max`. - Added robot test `View JOIN View Returns Results`. - Added robot test `View JOIN Provider Table Returns Results`. - Added robot test `Subquery JOIN Subquery Returns Results`. - Added robot test `CTE Within View Returns Results`.
1 parent a6379fe commit e15f531

3 files changed

Lines changed: 214 additions & 2 deletions

File tree

‎internal/stackql/astanalysis/earlyanalysis/ast_expand.go‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,6 @@ import (
44
"fmt"
55
"strings"
66

7-
"github.com/stackql/any-sdk/pkg/constants"
87
"github.com/stackql/any-sdk/pkg/logging"
98
"github.com/stackql/stackql/internal/stackql/astanalysis/annotatedast"
109
"github.com/stackql/stackql/internal/stackql/astindirect"
@@ -141,6 +140,14 @@ func (v *indirectExpandAstVisitor) processCTEReference(
141140
}
142141

143142
func (v *indirectExpandAstVisitor) processIndirect(node sqlparser.SQLNode, indirect astindirect.Indirect) error {
143+
// Eager depth check: fail before recursively analyzing an indirection that would exceed the limit.
144+
if v.indirectionDepth+1 > v.handlerCtx.GetRuntimeContext().IndirectDepthMax {
145+
return fmt.Errorf(
146+
"query error: indirection chain length %d > %d and is therefore disallowed; please do not cite views at too deep a level", //nolint:lll
147+
v.indirectionDepth+1,
148+
v.handlerCtx.GetRuntimeContext().IndirectDepthMax,
149+
)
150+
}
144151
err := indirect.Parse()
145152
if err != nil {
146153
return nil //nolint:nilerr //TODO: investigate
@@ -178,7 +185,7 @@ func (v *indirectExpandAstVisitor) processIndirect(node sqlparser.SQLNode, indir
178185
return fmt.Errorf(
179186
"query error: indirection chain length %d > %d and is therefore disallowed; please do not cite views at too deep a level", //nolint:lll
180187
maxIndirectCount,
181-
constants.LimitsIndirectMaxChainLength,
188+
v.handlerCtx.GetRuntimeContext().IndirectDepthMax,
182189
)
183190
}
184191
indirectPrimitiveGenerator.GetPrimitiveComposer().GetAst()

‎internal/stackql/dependencyplanner/dependencyplanner.go‎

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,40 @@ func NewStandardDependencyPlanner(
103103
}, nil
104104
}
105105

106+
func isAnnotationIndirection(annotationCtx taxonomy.AnnotationCtx) bool {
107+
_, isView := annotationCtx.GetView()
108+
_, isSubquery := annotationCtx.GetSubquery()
109+
return isView || isSubquery
110+
}
111+
112+
// orchestrateIndirection handles indirection vertices (views, subqueries, CTEs) in the dependency planner.
113+
// Unlike regular tables that need HTTP acquisition, indirections are already materialized by the
114+
// "create tail" builder and appear in the FROM clause as inline subqueries.
115+
// This method creates a NopBuilder as a placeholder in the execution DAG.
116+
func (dp *standardDependencyPlanner) orchestrateIndirection(
117+
annotationCtx taxonomy.AnnotationCtx,
118+
) (int, error) {
119+
rc, err := tableinsertioncontainer.NewTableInsertionContainer(
120+
annotationCtx.GetTableMeta(),
121+
dp.handlerCtx.GetSQLEngine(),
122+
dp.handlerCtx.GetTxnCounterMgr(),
123+
)
124+
if err != nil {
125+
return -1, err
126+
}
127+
builder := primitivebuilder.NewNopBuilder(
128+
dp.primitiveComposer.GetGraphHolder(),
129+
dp.primitiveComposer.GetTxnCtrlCtrs(),
130+
dp.handlerCtx,
131+
dp.handlerCtx.GetSQLEngine(),
132+
[]string{},
133+
)
134+
dp.execSlice = append(dp.execSlice, builder)
135+
idx := len(dp.execSlice) - 1
136+
dp.tableSlice = append(dp.tableSlice, rc)
137+
return idx, nil
138+
}
139+
106140
func (dp *standardDependencyPlanner) dataflowEdgeExists(from, to int) bool {
107141
edges, ok := dp.dataflowToEdges[to]
108142
if !ok {
@@ -204,6 +238,7 @@ func (dp *standardDependencyPlanner) Plan() error {
204238
edgeCount, dependencyMax)
205239
}
206240
idsVisited := make(map[int64]struct{})
241+
indirectionNodeIDs := make(map[int64]struct{})
207242
// first pass: set up AOT stuff
208243
// - stream per edge.
209244
edgeStreams := make(map[dataflow.Edge]streaming.MapStream)
@@ -218,6 +253,17 @@ func (dp *standardDependencyPlanner) Plan() error {
218253
tableExpr := n.GetTableExpr()
219254
annotation := n.GetAnnotation()
220255
dp.annMap[tableExpr] = annotation
256+
// Indirection nodes (views, subqueries, CTEs) are already materialized
257+
// by the create tail builder; register them and skip acquisition.
258+
if isAnnotationIndirection(annotation) {
259+
indirectionNodeIDs[n.ID()] = struct{}{}
260+
idx, indErr := dp.orchestrateIndirection(annotation)
261+
if indErr != nil {
262+
return indErr
263+
}
264+
dp.nodeIDIdxMap[n.ID()] = idx
265+
continue
266+
}
221267
for _, e := range edges {
222268
if e.From().ID() == n.ID() {
223269
insPsc, tcc, insErr := dp.processOrphan(tableExpr, annotation, n)
@@ -228,6 +274,20 @@ func (dp *standardDependencyPlanner) Plan() error {
228274
toNode := e.GetDest()
229275
toAnnotation := toNode.GetAnnotation().Clone() // this bodge protects split source vertices
230276
toTableExpr := toNode.GetTableExpr()
277+
// Handle indirection destination nodes.
278+
if isAnnotationIndirection(toAnnotation) {
279+
if _, alreadyHandled := indirectionNodeIDs[toNode.ID()]; !alreadyHandled {
280+
indirectionNodeIDs[toNode.ID()] = struct{}{}
281+
dp.annMap[toTableExpr] = toAnnotation
282+
toIdx, toIndErr := dp.orchestrateIndirection(toAnnotation)
283+
if toIndErr != nil {
284+
return toIndErr
285+
}
286+
dp.nodeIDIdxMap[toNode.ID()] = toIdx
287+
}
288+
orderedEdges = append(orderedEdges, e)
289+
continue
290+
}
231291
stream, streamErr := dp.getStreamFromEdge(e, toAnnotation, tcc)
232292
if streamErr != nil {
233293
return streamErr
@@ -251,6 +311,69 @@ func (dp *standardDependencyPlanner) Plan() error {
251311
fromAnnotation := fromNode.GetAnnotation()
252312
toAnnotation := toNode.GetAnnotation().Clone() // this bodge protects split source vertices
253313
toTableExpr := toNode.GetTableExpr()
314+
// For indirection nodes, builders are already created; just wire up edge dependencies.
315+
_, fromIsIndirection := indirectionNodeIDs[fromNode.ID()]
316+
_, toIsIndirection := indirectionNodeIDs[toNode.ID()]
317+
if fromIsIndirection && toIsIndirection {
318+
// Both sides are indirections; no acquisition or streaming needed.
319+
// Just ensure edge dependencies are registered.
320+
fromIdx := dp.nodeIDIdxMap[fromNode.ID()]
321+
toIdx := dp.nodeIDIdxMap[toNode.ID()]
322+
if !dp.dataflowEdgeExists(fromIdx, toIdx) {
323+
dp.dataflowToEdges[toIdx] = append(dp.dataflowToEdges[toIdx], fromIdx)
324+
}
325+
continue
326+
}
327+
if fromIsIndirection {
328+
// Source is indirection, destination is a regular table.
329+
fromIdx := dp.nodeIDIdxMap[fromNode.ID()]
330+
toIdx, toBuilderExists := dp.nodeIDIdxMap[toNode.ID()]
331+
if !toBuilderExists {
332+
toInsPsc, pscExists := insertPrepearedStatements[toNode.ID()]
333+
if !pscExists {
334+
return fmt.Errorf("unknown insert prepared statement")
335+
}
336+
dp.annMap[toTableExpr] = toAnnotation
337+
toAnnotation.SetDynamic()
338+
arrivingDestinationNodeStream := nodeStreamCollections.GetArriving(toNode.ID())
339+
departingDestinationNodeStream := nodeStreamCollections.GetDeparting(toNode.ID())
340+
var toErr error
341+
toIdx, toErr = dp.orchestrate(
342+
-1, toAnnotation, toInsPsc, arrivingDestinationNodeStream, departingDestinationNodeStream)
343+
if toErr != nil {
344+
return toErr
345+
}
346+
dp.nodeIDIdxMap[toNode.ID()] = toIdx
347+
}
348+
if !dp.dataflowEdgeExists(fromIdx, toIdx) {
349+
dp.dataflowToEdges[toIdx] = append(dp.dataflowToEdges[toIdx], fromIdx)
350+
}
351+
continue
352+
}
353+
if toIsIndirection {
354+
// Destination is indirection, source is a regular table.
355+
toIdx := dp.nodeIDIdxMap[toNode.ID()]
356+
fromIdx, fromBuilderExists := dp.nodeIDIdxMap[fromNode.ID()]
357+
if !fromBuilderExists {
358+
insPsc, pscExists := insertPrepearedStatements[fromNode.ID()]
359+
if !pscExists {
360+
return fmt.Errorf("unknown insert prepared statement")
361+
}
362+
arrivingSourceNodeStream := nodeStreamCollections.GetArriving(fromNode.ID())
363+
departingSourceNodeStream := nodeStreamCollections.GetDeparting(fromNode.ID())
364+
var fromErr error
365+
fromIdx, fromErr = dp.orchestrate(-1, fromAnnotation, insPsc, arrivingSourceNodeStream, departingSourceNodeStream)
366+
if fromErr != nil {
367+
return fromErr
368+
}
369+
dp.nodeIDIdxMap[fromNode.ID()] = fromIdx
370+
}
371+
if !dp.dataflowEdgeExists(fromIdx, toIdx) {
372+
dp.dataflowToEdges[toIdx] = append(dp.dataflowToEdges[toIdx], fromIdx)
373+
}
374+
continue
375+
}
376+
// Neither side is an indirection; original logic.
254377
departingSourceNodeStream := nodeStreamCollections.GetDeparting(fromNode.ID())
255378
arrivingDestinationNodeStream := nodeStreamCollections.GetArriving(toNode.ID())
256379
arrivingSourceNodeStream := nodeStreamCollections.GetArriving(fromNode.ID())

‎test/robot/functional/stackql_mocked_from_cmd_line.robot‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9478,3 +9478,85 @@ Left Outer Join Positive LHS Inline
94789478
... ${outputStr}
94799479
... stdout=${CURDIR}/tmp/Left-Outer-Join-Positive-LHS-Inline.tmp
94809480
... stderr=${CURDIR}/tmp/Left-Outer-Join-Positive-LHS-Inline-stderr.tmp
9481+
9482+
View Depth Limitation Error Message Shows Correct Max
9483+
Should Stackql Exec Inline Contain Stderr
9484+
... ${STACKQL_EXE}
9485+
... ${OKTA_SECRET_STR}
9486+
... ${GITHUB_SECRET_STR}
9487+
... ${K8S_SECRET_STR}
9488+
... ${REGISTRY_NO_VERIFY_CFG_STR}
9489+
... ${AUTH_CFG_STR}
9490+
... ${SQL_BACKEND_CFG_STR_CANONICAL}
9491+
... create view zz1 as select name from stackql_repositories; create view zz2 as select name from zz1; create view zz3 as select name from zz2; create view zz4 as select name from zz3; create view zz5 as select name from zz4; select * from zz5;
9492+
... indirection chain length 6 > 5
9493+
... stdout=${CURDIR}/tmp/View-Depth-Limitation-Error-Message-Shows-Correct-Max-stdout.tmp
9494+
... stderr=${CURDIR}/tmp/View-Depth-Limitation-Error-Message-Shows-Correct-Max-stderr.tmp
9495+
9496+
View JOIN View Returns Results
9497+
${inputStr} = Catenate
9498+
... create or replace view vw_repos_name as select name from stackql_repositories;
9499+
... create or replace view vw_repos_url as select name, url from stackql_repositories;
9500+
... select v1.name from vw_repos_name v1 inner join vw_repos_url v2 on v1.name = v2.name;
9501+
Should Stackql Exec Inline Contain
9502+
... ${STACKQL_EXE}
9503+
... ${OKTA_SECRET_STR}
9504+
... ${GITHUB_SECRET_STR}
9505+
... ${K8S_SECRET_STR}
9506+
... ${REGISTRY_NO_VERIFY_CFG_STR}
9507+
... ${AUTH_CFG_STR}
9508+
... ${SQL_BACKEND_CFG_STR_CANONICAL}
9509+
... ${inputStr}
9510+
... dummyapp.io
9511+
... stdout=${CURDIR}/tmp/View-JOIN-View-Returns-Results-stdout.tmp
9512+
... stderr=${CURDIR}/tmp/View-JOIN-View-Returns-Results-stderr.tmp
9513+
9514+
View JOIN Provider Table Returns Results
9515+
${inputStr} = Catenate
9516+
... create or replace view vw_repos as select name, url from stackql_repositories;
9517+
... select v1.name from vw_repos v1 inner join github.repos.repos r on v1.name = r.name where r.org = 'stackql';
9518+
Should Stackql Exec Inline Contain
9519+
... ${STACKQL_EXE}
9520+
... ${OKTA_SECRET_STR}
9521+
... ${GITHUB_SECRET_STR}
9522+
... ${K8S_SECRET_STR}
9523+
... ${REGISTRY_NO_VERIFY_CFG_STR}
9524+
... ${AUTH_CFG_STR}
9525+
... ${SQL_BACKEND_CFG_STR_CANONICAL}
9526+
... ${inputStr}
9527+
... dummyapp.io
9528+
... stdout=${CURDIR}/tmp/View-JOIN-Provider-Table-Returns-Results-stdout.tmp
9529+
... stderr=${CURDIR}/tmp/View-JOIN-Provider-Table-Returns-Results-stderr.tmp
9530+
9531+
Subquery JOIN Subquery Returns Results
9532+
${inputStr} = Catenate
9533+
... select a.name from (select name from stackql_repositories) a inner join (select name, url from stackql_repositories) b on a.name = b.name;
9534+
Should Stackql Exec Inline Contain
9535+
... ${STACKQL_EXE}
9536+
... ${OKTA_SECRET_STR}
9537+
... ${GITHUB_SECRET_STR}
9538+
... ${K8S_SECRET_STR}
9539+
... ${REGISTRY_NO_VERIFY_CFG_STR}
9540+
... ${AUTH_CFG_STR}
9541+
... ${SQL_BACKEND_CFG_STR_CANONICAL}
9542+
... ${inputStr}
9543+
... dummyapp.io
9544+
... stdout=${CURDIR}/tmp/Subquery-JOIN-Subquery-Returns-Results-stdout.tmp
9545+
... stderr=${CURDIR}/tmp/Subquery-JOIN-Subquery-Returns-Results-stderr.tmp
9546+
9547+
CTE Within View Returns Results
9548+
${inputStr} = Catenate
9549+
... create or replace view vw_cte as with sub as (select name from stackql_repositories) select name from sub;
9550+
... select name from vw_cte;
9551+
Should Stackql Exec Inline Contain
9552+
... ${STACKQL_EXE}
9553+
... ${OKTA_SECRET_STR}
9554+
... ${GITHUB_SECRET_STR}
9555+
... ${K8S_SECRET_STR}
9556+
... ${REGISTRY_NO_VERIFY_CFG_STR}
9557+
... ${AUTH_CFG_STR}
9558+
... ${SQL_BACKEND_CFG_STR_CANONICAL}
9559+
... ${inputStr}
9560+
... dummyapp.io
9561+
... stdout=${CURDIR}/tmp/CTE-Within-View-Returns-Results-stdout.tmp
9562+
... stderr=${CURDIR}/tmp/CTE-Within-View-Returns-Results-stderr.tmp

0 commit comments

Comments
 (0)