- Added robot test Insert Async Returning Simple Projection.

Author: general-kroll-4-life

.vscode/launch.json

@@ -176,6 +176,7 @@
         "select  rings.projectsId as project,  rings.locationsId as locale,  split_part(rings.name, '/', -1) as key_ring_name,  split_part(keys.name, '/', -1) as key_name,  json_extract(keys.\"versionTemplate\", '$.algorithm') as key_algorithm,  json_extract(keys.\"versionTemplate\", '$.protectionLevel') as key_protection_level  from  google.cloudkms.key_rings rings  inner join google.cloudkms.crypto_keys keys  on  keys.keyRingsId = split_part(rings.name, '/', -1)  and keys.projectsId = rings.projectsId  and keys.locationsId = rings.locationsId  where  rings.projectsId in ('testing-project', 'testing-project-two', 'testing-project-three')  and rings.locationsId in ('global', 'australia-southeast1', 'australia-southeast2')  order by project, locale, key_name ;",
         "delete from aws.cloud_control.resources  where  region = 'ap-southeast-1'  and data__TypeName = 'AWS::Logs::LogGroup'  and data__Identifier = 'LogGroupResourceExampleThird' ;",
         "insert into google.storage.buckets( project, data__name) select  'testing-project', 'silly-bucket' returning projectNumber;",
+        "insert /*+ AWAIT */ into google.compute.networks(project, data__name, data__autoCreateSubnetworks) select 'mutable-project', 'auto-test-01', false returning creationTimestamp, name;",
       ],
       "default": "show providers;"
     },

go.mod

@@ -19,7 +19,7 @@ require (
 	github.com/spf13/cobra v1.4.0
 	github.com/spf13/pflag v1.0.5
 	github.com/spf13/viper v1.10.1
-	github.com/stackql/any-sdk v0.1.3-beta02
+	github.com/stackql/any-sdk v0.1.4-alpha05
 	github.com/stackql/go-suffix-map v0.0.1-alpha01
 	github.com/stackql/psql-wire v0.1.1-beta23
 	github.com/stackql/stackql-parser v0.0.15-alpha06

go.sum

@@ -484,8 +484,8 @@ github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spf13/viper v1.10.1 h1:nuJZuYpG7gTj/XqiUwg8bA0cp1+M2mC3J4g5luUYBKk=
 github.com/spf13/viper v1.10.1/go.mod h1:IGlFPqhNAPKRxohIzWpI5QEy4kuI7tcl5WvR+8qy1rU=
-github.com/stackql/any-sdk v0.1.3-beta02 h1:0jSwyYFddjAN++U+yNNLF7SMLPIIaRhv522UzeWDf2E=
-github.com/stackql/any-sdk v0.1.3-beta02/go.mod h1:AKS/g28y7m4SWL/YW8veE9MCNy8XJgaicVibemVE9e8=
+github.com/stackql/any-sdk v0.1.4-alpha05 h1:QWMTCU4OHJdsRI3zT4pmM8L+Opgf1RUmg2MPXDjAtzQ=
+github.com/stackql/any-sdk v0.1.4-alpha05/go.mod h1:AKS/g28y7m4SWL/YW8veE9MCNy8XJgaicVibemVE9e8=
 github.com/stackql/go-suffix-map v0.0.1-alpha01 h1:TDUDS8bySu41Oo9p0eniUeCm43mnRM6zFEd6j6VUaz8=
 github.com/stackql/go-suffix-map v0.0.1-alpha01/go.mod h1:QAi+SKukOyf4dBtWy8UMy+hsXXV+yyEE4vmBkji2V7g=
 github.com/stackql/psql-wire v0.1.1-beta23 h1:1ayYMjZArfDcIMyEOKnm+Bp1zRCISw8pguvTFuUhhVQ=

…tackql/primitivebuilder/async_compose.go …al/stackql/asynccompose/async_compose.gointernal/stackql/primitivebuilder/async_compose.go renamed to internal/stackql/asynccompose/async_compose.go internal/stackql/primitivebuilder/async_compose.go renamed to internal/stackql/asynccompose/async_compose.go

@@ -1,21 +1,24 @@
-package primitivebuilder
+package asynccompose
 
 import (
 	"github.com/stackql/any-sdk/anysdk"
 	"github.com/stackql/stackql-parser/go/vt/sqlparser"
+	"github.com/stackql/stackql/internal/stackql/drm"
 	"github.com/stackql/stackql/internal/stackql/handler"
 	"github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto"
 	"github.com/stackql/stackql/internal/stackql/primitive"
 	"github.com/stackql/stackql/internal/stackql/provider"
 )
 
-func composeAsyncMonitor(
+func ComposeAsyncMonitor(
 	handlerCtx handler.HandlerContext,
 	precursor primitive.IPrimitive,
 	prov provider.IProvider,
 	method anysdk.OperationStore,
 	commentDirectives sqlparser.CommentDirectives,
 	isReturning bool,
+	insertCtx drm.PreparedStatementCtx,
+	drmCfg drm.Config,
 ) (primitive.IPrimitive, error) {
 	asm, err := NewAsyncMonitor(handlerCtx, prov, method, isReturning)
 	if err != nil {
@@ -32,7 +35,8 @@ func composeAsyncMonitor(
 		handlerCtx.GetOutfile(),
 		handlerCtx.GetOutErrFile(),
 	)
-	primitive, err := asm.GetMonitorPrimitive(prov, method, precursor, pl, commentDirectives, isReturning)
+	primitive, err := asm.GetMonitorPrimitive(
+		prov, method, precursor, pl, commentDirectives, isReturning, insertCtx, drmCfg)
 	if err != nil {
 		return nil, err
 	}

…stackql/primitivebuilder/asyncmonitor.go …nal/stackql/asynccompose/asyncmonitor.gointernal/stackql/primitivebuilder/asyncmonitor.go renamed to internal/stackql/asynccompose/asyncmonitor.go internal/stackql/primitivebuilder/asyncmonitor.go renamed to internal/stackql/asynccompose/asyncmonitor.go

@@ -1,11 +1,12 @@
-package primitivebuilder
+package asynccompose
 
 import (
 	"fmt"
 	"strings"
 
 	"github.com/stackql/any-sdk/anysdk"
 	"github.com/stackql/stackql/internal/stackql/acid/binlog"
+	"github.com/stackql/stackql/internal/stackql/drm"
 	"github.com/stackql/stackql/internal/stackql/execution"
 	"github.com/stackql/stackql/internal/stackql/handler"
 	"github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto"
@@ -23,6 +24,8 @@ type IAsyncMonitor interface {
 		initialCtx primitive.IPrimitiveCtx,
 		comments sqlparser.CommentDirectives,
 		isReturning bool,
+		insertCtx drm.PreparedStatementCtx,
+		drmCfg drm.Config,
 	) (primitive.IPrimitive, error)
 }
 
@@ -164,11 +167,13 @@ func (gm *DefaultGoogleAsyncMonitor) GetMonitorPrimitive(
 	initialCtx primitive.IPrimitiveCtx,
 	comments sqlparser.CommentDirectives,
 	isReturning bool,
+	insertCtx drm.PreparedStatementCtx,
+	drmCfg drm.Config,
 ) (primitive.IPrimitive, error) {
 	//nolint:gocritic,staticcheck //TODO: refactor
 	switch strings.ToLower(prov.GetVersion()) {
 	default:
-		return gm.getV1Monitor(prov, op, precursor, initialCtx, comments, isReturning)
+		return gm.getV1Monitor(prov, op, precursor, initialCtx, comments, isReturning, insertCtx, drmCfg)
 	}
 }
 
@@ -179,6 +184,8 @@ func (gm *DefaultGoogleAsyncMonitor) getV1Monitor(
 	initialCtx primitive.IPrimitiveCtx,
 	comments sqlparser.CommentDirectives,
 	isReturning bool,
+	insertCtx drm.PreparedStatementCtx,
+	drmCfg drm.Config,
 ) (primitive.IPrimitive, error) {
 	provider, providerErr := prov.GetProvider()
 	if providerErr != nil {
@@ -192,6 +199,8 @@ func (gm *DefaultGoogleAsyncMonitor) getV1Monitor(
 		initialCtx,
 		comments,
 		isReturning,
+		insertCtx,
+		drmCfg,
 	)
 	if exPrepErr != nil {
 		return nil, exPrepErr

internal/stackql/dependencyplanner/dependencyplanner.go

@@ -485,6 +485,7 @@ func (dp *standardDependencyPlanner) orchestrate(
 			insPsc,
 			nil,
 			outStream,
+			false,
 		)
 	}
 	dp.execSlice = append(dp.execSlice, builder)

internal/stackql/execution/mono_valent_execution.go

@@ -1130,6 +1130,7 @@ func (sp *standardProcessor) Process() ProcessorResponse {
 		if httpResponseErr != nil {
 			return newHTTPProcessorResponse(nil, reversalStream, false, httpResponseErr)
 		}
+		// TODO: add async monitor here
 		processed, resErr := method.ProcessResponse(httpResponse)
 		if resErr != nil {
 			if isSkipResponse && isMutation && httpResponse.StatusCode < 300 {
@@ -1479,6 +1480,8 @@ func GetMonitorExecutor(
 	initialCtx primitive.IPrimitiveCtx,
 	comments sqlparser.CommentDirectives,
 	isReturning bool,
+	insertCtx drm.PreparedStatementCtx,
+	drmCfg drm.Config,
 ) (primitive.IPrimitive, error) {
 	m := op
 	// tableName, err := mv.tableMeta.GetTableName()
@@ -1498,6 +1501,8 @@ func GetMonitorExecutor(
 		elapsedSeconds:      0,
 		pollIntervalSeconds: MonitorPollIntervalSeconds,
 		comments:            comments,
+		insertCtx:           insertCtx,
+		drmCfg:              drmCfg,
 	}
 	if comments != nil {
 		asyncPrim.noStatus = comments.IsSet("NOSTATUS")
@@ -1567,6 +1572,29 @@ func GetMonitorExecutor(
 				if targetErr != nil {
 					return internaldto.NewExecutorOutput(nil, nil, nil, nil, targetErr)
 				}
+				// TODO: insert into table here
+				if isReturning {
+					if asyncPrim.insertCtx != nil {
+						_, rErr := asyncPrim.drmCfg.ExecuteInsertDML(
+							handlerCtx.GetSQLEngine(),
+							asyncPrim.insertCtx,
+							target,
+							"", // TODO: figure out how on earth to compute this encoding
+						)
+						if rErr != nil {
+							return internaldto.NewExecutorOutput(nil, nil, nil, nil, rErr)
+						}
+					}
+					if asyncPrim.selectCtx != nil {
+						_, rErr := asyncPrim.drmCfg.QueryDML(
+							handlerCtx.GetSQLEngine(),
+							drm.NewPreparedStatementParameterized(asyncPrim.selectCtx, nil, true),
+						)
+						if rErr != nil {
+							return internaldto.NewExecutorOutput(nil, nil, nil, nil, rErr)
+						}
+					}
+				}
 				return prepareResultSet(&asyncPrim, pc, target, operationDescriptor)
 			}
 			return prepareResultSet(&asyncPrim, pc, body, operationDescriptor)
@@ -1718,6 +1746,9 @@ type asyncHTTPMonitorPrimitive struct {
 	noStatus            bool
 	id                  int64
 	comments            sqlparser.CommentDirectives
+	insertCtx           drm.PreparedStatementCtx
+	selectCtx           drm.PreparedStatementCtx
+	drmCfg              drm.Config
 }
 
 func (pr *asyncHTTPMonitorPrimitive) SetTxnID(_ int) {

internal/stackql/internal_data_transfer/builder_input/builder_input.go

@@ -5,6 +5,7 @@ import (
 	"github.com/stackql/any-sdk/pkg/streaming"
 	"github.com/stackql/stackql-parser/go/vt/sqlparser"
 	"github.com/stackql/stackql/internal/stackql/astanalysis/annotatedast"
+	"github.com/stackql/stackql/internal/stackql/drm"
 	"github.com/stackql/stackql/internal/stackql/handler"
 	"github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto"
 	"github.com/stackql/stackql/internal/stackql/primitivegraph"
@@ -56,6 +57,8 @@ type BuilderInput interface {
 	GetTxnCtrlCtrs() (internaldto.TxnControlCounters, bool)
 	GetTableInsertionContainer() (tableinsertioncontainer.TableInsertionContainer, bool)
 	SetTableInsertionContainer(tableinsertioncontainer.TableInsertionContainer)
+	SetInsertCtx(insertCtx drm.PreparedStatementCtx)
+	GetInsertCtx() (drm.PreparedStatementCtx, bool)
 }
 
 type builderInput struct {
@@ -79,6 +82,7 @@ type builderInput struct {
 	isTargetPhysical        bool
 	txnCtrlCtrs             internaldto.TxnControlCounters
 	tableInsertionContainer tableinsertioncontainer.TableInsertionContainer
+	insertCtx               drm.PreparedStatementCtx
 }
 
 func NewBuilderInput(
@@ -95,6 +99,17 @@ func NewBuilderInput(
 	}
 }
 
+func (bi *builderInput) SetInsertCtx(insertCtx drm.PreparedStatementCtx) {
+	bi.insertCtx = insertCtx
+}
+
+func (bi *builderInput) GetInsertCtx() (drm.PreparedStatementCtx, bool) {
+	if bi.insertCtx == nil {
+		return nil, false
+	}
+	return bi.insertCtx, true
+}
+
 func (bi *builderInput) IsReturning() bool {
 	return bi.isReturning
 }
@@ -271,5 +286,7 @@ func (bi *builderInput) Clone() BuilderInput {
 		isTargetPhysical:  bi.isTargetPhysical,
 		annotatedAst:      bi.annotatedAst,
 		txnCtrlCtrs:       bi.txnCtrlCtrs,
+		isReturning:       bi.isReturning,
+		insertCtx:         bi.insertCtx,
 	}
 }

internal/stackql/planbuilder/plan_builder.go

@@ -10,6 +10,7 @@ import (
 	"github.com/stackql/any-sdk/anysdk"
 
 	"github.com/stackql/any-sdk/pkg/logging"
+	"github.com/stackql/any-sdk/pkg/streaming"
 	"github.com/stackql/stackql/internal/stackql/acid/txn_context"
 	"github.com/stackql/stackql/internal/stackql/astanalysis/routeanalysis"
 	"github.com/stackql/stackql/internal/stackql/handler"
@@ -921,10 +922,7 @@ func (pgb *standardPlanGraphBuilder) handleInsert(pbi planbuilderinput.PlanBuild
 		if isPhysicalTable {
 			bldrInput.SetIsTargetPhysicalTable(true)
 		}
-		//nolint:stylecheck // not in the mood
-		var bldr primitivebuilder.Builder = primitivebuilder.NewInsertOrUpdate(
-			bldrInput,
-		)
+		var bldr primitivebuilder.Builder
 		if len(node.SelectExprs) > 0 {
 			// Two cases:
 			//   1. Synchronous.  Equivalent to select.
@@ -943,25 +941,57 @@ func (pgb *standardPlanGraphBuilder) handleInsert(pbi planbuilderinput.PlanBuild
 				return rcErr
 			}
 			bldrInput.SetTableInsertionContainer(rc)
-			//nolint:stylecheck // not in the mood
-			var returningBldr primitivebuilder.Builder = primitivebuilder.NewSingleAcquireAndSelect(
-				bldrInput,
-				primitiveGenerator.GetPrimitiveComposer().GetInsertPreparedStatementCtx(),
-				primitiveGenerator.GetPrimitiveComposer().GetSelectPreparedStatementCtx(),
-				nil,
-			)
+			bldrInput.SetIsReturning(true)
 			if !isAwait {
-				bldr = returningBldr
-			} else {
-				rhsBldr := primitivebuilder.NewSingleAcquireAndSelect(
+				bldr = primitivebuilder.NewSingleAcquireAndSelect(
 					bldrInput,
 					primitiveGenerator.GetPrimitiveComposer().GetInsertPreparedStatementCtx(),
 					primitiveGenerator.GetPrimitiveComposer().GetSelectPreparedStatementCtx(),
 					nil,
 				)
+			} else {
+				// bldr = returningBldr
+				// TODO: stuff the async output into an appropriate table
+				// rhsBldr := primitivebuilder.NewSingleSelect(
+				// 	pgb.planGraphHolder,
+				// 	handlerCtx,
+				// 	primitiveGenerator.GetPrimitiveComposer().GetSelectPreparedStatementCtx(),
+				// 	[]tableinsertioncontainer.TableInsertionContainer{rc},
+				// 	nil,
+				// 	streaming.NewNopMapStream(),
+				// )
+				bldrInput.SetIsAwait(true)
+				bldrInput.SetIsReturning(true)
+				bldrInput.SetInsertCtx(primitiveGenerator.GetPrimitiveComposer().GetInsertPreparedStatementCtx())
+				lhsBldr := primitivebuilder.NewInsertOrUpdate(
+					bldrInput,
+				)
+				newBldrInput := builder_input.NewBuilderInput(
+					pgb.planGraphHolder,
+					handlerCtx,
+					tbl,
+				)
+				newBldrInput.SetParserNode(node)
+				newBldrInput.SetAnnotatedAST(pbi.GetAnnotatedAST())
+				newBldrInput.SetTxnCtrlCtrs(pbi.GetTxnCtrlCtrs())
+				newBldrInput.SetTableInsertionContainer(rc)
+				newBldrInput.SetDependencyNode(selectPrimitiveNode)
+				newBldrInput.SetIsAwait(isAwait)
+				// rhsBldr := primitivebuilder.NewSingleAcquireAndSelect(
+				// 	newBldrInput,
+				// 	primitiveGenerator.GetPrimitiveComposer().GetInsertPreparedStatementCtx(),
+				// 	primitiveGenerator.GetPrimitiveComposer().GetSelectPreparedStatementCtx(),
+				// 	nil,
+				// )
+				rhsBldr := primitivebuilder.NewSingleSelect(
+					pgb.planGraphHolder, handlerCtx, primitiveGenerator.GetPrimitiveComposer().GetSelectPreparedStatementCtx(),
+					[]tableinsertioncontainer.TableInsertionContainer{rc},
+					nil,
+					streaming.NewNopMapStream(),
+				)
 				bldr = primitivebuilder.NewDependencySubDAGBuilder(
 					pgb.planGraphHolder,
-					[]primitivebuilder.Builder{bldr},
+					[]primitivebuilder.Builder{lhsBldr},
 					rhsBldr,
 				)
 			}

internal/stackql/primitivebuilder/delete.go

@@ -3,6 +3,7 @@ package primitivebuilder
 import (
 	"github.com/stackql/any-sdk/pkg/streaming"
 	"github.com/stackql/stackql-parser/go/vt/sqlparser"
+	"github.com/stackql/stackql/internal/stackql/asynccompose"
 	"github.com/stackql/stackql/internal/stackql/drm"
 	"github.com/stackql/stackql/internal/stackql/execution"
 	"github.com/stackql/stackql/internal/stackql/handler"
@@ -93,8 +94,8 @@ func (ss *Delete) Build() error {
 		primitive_context.NewPrimitiveContext(),
 	)
 	if ss.isAwait {
-		deletePrimitive, err = composeAsyncMonitor(
-			handlerCtx, deletePrimitive, prov, method, nil, false) // isReturning hardcoded to false for now
+		deletePrimitive, err = asynccompose.ComposeAsyncMonitor(
+			handlerCtx, deletePrimitive, prov, method, nil, false, nil, nil) // isReturning hardcoded to false for now
 	}
 	if err != nil {
 		return err