select-insert-logic (#620)

* select-insert-logic

Summary:

- Support for `INSERT.. SELECT.. [LEFT OUTER JOIN]`.
- Added robot test `Insert Select Negative Returning Star`.
- Added robot test `Insert Select Positive Returning Star`.
- Added robot test `Insert Left Outer Join Positive Returning Star`.
- Added robot test `Insert Left Outer Join Negative Returning Star`.

* - Fixed robot.

Author: general-kroll-4-life

.vscode/launch.json

@@ -193,6 +193,10 @@
         "select split_part(rings.name, '/', -1) from google.cloudkms.key_rings rings inner join google.cloudkms.crypto_keys keys on keys.keyRingsId = split_part(rings.name, '/', -1) and keys.projectsId = 'testing-project' where rings.projectsId = 'testing-project' and rings.locationsId = 'global' and keys.locationsId = 'global';",
         "select * from local_openssl.keys.x509 where cert_file = '${workspaceFolder}/test/assets/input/manual_cert.pem';",
         "select t1.id as t1_id, t2.id as t2_id from (select 'my-id' as id) t1 left outer join (select a.vpcId id from aws.ec2_native.vpcs a where region = 'ap-southeast-2') t2 on t1.id = t2.id;",
+        "insert into aws.ec2.vpcs(CidrBlock, region) select '10.0.0.0/16', 'ap-southeast-2' where 0 = 1 returning *;",
+        "insert into aws.ec2.vpcs(CidrBlock, region) select '10.0.0.0/16', 'ap-southeast-2' where 1 = 1 returning *;",
+        "select lhs.proj, lhs.bucket from (select 'testing-project' as proj, 'silly-bucket' as bucket) lhs LEFT OUTER join (select name from google.storage.buckets where project = 'testing-project') rhs on lhs.bucket = rhs.name where rhs.name;",
+        "insert into google.storage.buckets( project, data__name) select lhs.proj, lhs.bucket from (select 'testing-project' as proj, 'silly-bucket' as bucket) lhs LEFT OUTER join (select name from google.storage.buckets where project = 'testing-project') rhs on lhs.bucket = rhs.name where rhs.name is null returning *;",
       ], 
       "default": "show providers;"
     },

go.mod

@@ -18,14 +18,13 @@ 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.5.1-alpha01
+	github.com/stackql/any-sdk v0.5.1-alpha03
 	github.com/stackql/go-suffix-map v0.0.1-alpha01
 	github.com/stackql/psql-wire v0.1.2-alpha01
 	github.com/stackql/stackql-parser v0.0.16-alpha01
 	github.com/stretchr/testify v1.10.0
 	golang.org/x/mod v0.25.0
 	golang.org/x/sync v0.15.0
-	golang.org/x/tools v0.34.0
 	gonum.org/v1/gonum v0.15.1
 	gopkg.in/yaml.v2 v2.4.0
 	gotest.tools v2.2.0+incompatible
@@ -131,6 +130,7 @@ require (
 	golang.org/x/sys v0.33.0 // indirect
 	golang.org/x/term v0.32.0 // indirect
 	golang.org/x/text v0.26.0 // indirect
+	golang.org/x/tools v0.34.0 // indirect
 	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
 	google.golang.org/genproto v0.0.0-20211208223120-3a66f561d7aa // indirect
 	google.golang.org/grpc v1.67.1 // indirect

go.sum

@@ -467,8 +467,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.5.1-alpha01 h1:HHzdjSsR981GaqoZ1CAmRuVAfZZeMEunsLhcrVtCw/M=
-github.com/stackql/any-sdk v0.5.1-alpha01/go.mod h1:ZRTGrcDKFLaC+5yWo4OqXVs1HTNxgYM9nQsQDlq0Fe0=
+github.com/stackql/any-sdk v0.5.1-alpha03 h1:zYVr/ysJ26Tb1JCl/49MymMsHH+5dC2iGircAJhGbIg=
+github.com/stackql/any-sdk v0.5.1-alpha03/go.mod h1:ZRTGrcDKFLaC+5yWo4OqXVs1HTNxgYM9nQsQDlq0Fe0=
 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.2-alpha01 h1:RMBRURGspmSNqm2/sgoEc+D6Sri2y/3drjl4nKlOOi4=

internal/stackql/astanalysis/earlyanalysis/ast_expand.go

@@ -334,6 +334,10 @@ func (v *indirectExpandAstVisitor) Visit(node sqlparser.SQLNode) error {
 		if err != nil {
 			return err
 		}
+		err = node.Rows.Accept(v)
+		if err != nil {
+			return err
+		}
 
 	case *sqlparser.Update:
 		v.mutateCount++

internal/stackql/astvisit/placeholder_param_extract.go

@@ -141,6 +141,10 @@ func (v *standardParserPlaceholderParamAstVisitor) Visit(node sqlparser.SQLNode)
 		return node.Table.Accept(v)
 
 	case *sqlparser.Insert:
+		err := v.Visit(node.Rows)
+		if err != nil {
+			return err
+		}
 		if len(node.Columns) > 0 {
 			err := node.Columns.Accept(v)
 			if err != nil {

internal/stackql/astvisit/table_alias_pairing.go

@@ -175,6 +175,7 @@ func (v *standardParserTableAliasPairingAstVisitor) Visit(node sqlparser.SQLNode
 		return node.Table.Accept(v)
 
 	case *sqlparser.Insert:
+		return v.Visit(node.Rows)
 
 	case *sqlparser.Update:
 

internal/stackql/astvisit/table_extract.go

@@ -138,6 +138,7 @@ func (v *standardParserTableExtractAstVisitor) Visit(node sqlparser.SQLNode) err
 		return node.Table.Accept(v)
 
 	case *sqlparser.Insert:
+		return v.Visit(node.Rows)
 
 	case *sqlparser.Update:
 

internal/stackql/dependencyplanner/dependencyplanner.go

@@ -17,6 +17,7 @@ import (
 	"github.com/stackql/stackql/internal/stackql/docparser"
 	"github.com/stackql/stackql/internal/stackql/drm"
 	"github.com/stackql/stackql/internal/stackql/handler"
+	"github.com/stackql/stackql/internal/stackql/internal_data_transfer/builder_input"
 	"github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto"
 	"github.com/stackql/stackql/internal/stackql/parserutil"
 	"github.com/stackql/stackql/internal/stackql/primitivebuilder"
@@ -514,17 +515,25 @@ func (dp *standardDependencyPlanner) orchestrate(
 			nil,
 			insPsc,
 			nil,
+			// this is the connector
 			outStream,
 		)
 	} else {
+		bldrInput := builder_input.NewBuilderInput(
+			dp.primitiveComposer.GetGraphHolder(),
+			dp.handlerCtx,
+			annotationCtx.GetTableMeta(),
+		)
+		bldrInput.SetIsAwait(false) // returning hardcoded to false for now
 		builder = primitivebuilder.NewSingleSelectAcquire(
 			dp.primitiveComposer.GetGraphHolder(),
 			dp.handlerCtx,
 			rc,
 			insPsc,
 			nil,
+			// this is the connector
 			outStream,
-			false, // returning hardcoded to false for now
+			bldrInput,
 		)
 	}
 	dp.execSlice = append(dp.execSlice, builder)
@@ -557,6 +566,7 @@ func (dp *standardDependencyPlanner) processAcquire(
 			nil, annotationCtx.GetHIDs(),
 			inputTableName,
 			annotationCtx.GetTableMeta().GetAlias())
+		// TODO: repeat this for mutations
 		anTab.SetSQLDataSource(sqlDataSource)
 		return anTab, dp.tcc, nil
 	}

internal/stackql/execution/mono_valent_execution.go

@@ -23,6 +23,7 @@ import (
 	"github.com/stackql/stackql/internal/stackql/acid/binlog"
 	"github.com/stackql/stackql/internal/stackql/drm"
 	"github.com/stackql/stackql/internal/stackql/handler"
+	"github.com/stackql/stackql/internal/stackql/internal_data_transfer/builder_input"
 	"github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto"
 	"github.com/stackql/stackql/internal/stackql/primitive"
 	"github.com/stackql/stackql/internal/stackql/primitivegraph"
@@ -84,6 +85,7 @@ type monoValentExecution struct {
 	isMutation                 bool
 	isAwait                    bool
 	defaultHTTPClient          *http.Client // for testing purposes only
+	bldrInput                  builder_input.BuilderInput
 }
 
 func NewMonoValentExecutorFactory(
@@ -97,6 +99,7 @@ func NewMonoValentExecutorFactory(
 	isSkipResponse bool,
 	isMutation bool,
 	isAwait bool,
+	bldrInput builder_input.BuilderInput,
 ) MonoValentExecutorFactory {
 	var tcc internaldto.TxnControlCounters
 	if insertCtx != nil {
@@ -127,6 +130,7 @@ func NewMonoValentExecutorFactory(
 		isAwait:                    isAwait,
 		defaultHTTPClient:          defaultHTTPClient,
 		invoker:                    anysdkhttp.New(),
+		bldrInput:                  bldrInput,
 	}
 }
 
@@ -1257,6 +1261,59 @@ func (sp *standardProcessor) Process() processorResponse {
 	return newHTTPProcessorResponse(nil, reversalStream, false, nil)
 }
 
+var (
+	_ formulation.BaseArmouryGenerator = (*reconsititutedarmouryGenerator)(nil)
+)
+
+type rawTransformer struct {
+	node              sqlparser.SQLNode
+	insertValOnlyRows map[int]map[int]interface{}
+}
+
+func (rt *rawTransformer) Transform() (map[int]map[string]interface{}, error) {
+	return util.ExtractSQLNodeParams(rt.node, rt.insertValOnlyRows)
+}
+
+func newRawTransformer(
+	node sqlparser.SQLNode,
+	insertValOnlyRows map[int]map[int]interface{},
+) *rawTransformer {
+	return &rawTransformer{
+		node:              node,
+		insertValOnlyRows: insertValOnlyRows,
+	}
+}
+
+type reconsititutedarmouryGenerator struct {
+	prior         formulation.BaseArmouryGenerator
+	updatedParams map[int]map[string]any
+}
+
+func newUpdatedArmouryGenerator(
+	prior formulation.BaseArmouryGenerator,
+	updatedParams map[int]map[string]any,
+) formulation.BaseArmouryGenerator {
+	return &reconsititutedarmouryGenerator{
+		prior:         prior,
+		updatedParams: updatedParams,
+	}
+}
+
+func (r *reconsititutedarmouryGenerator) GetHTTPArmoury() (formulation.HTTPArmoury, error) {
+	if len(r.updatedParams) == 0 {
+		return r.prior.GetHTTPArmoury()
+	}
+	armoury, err := r.prior.GetHTTPArmoury()
+	if err != nil {
+		return nil, err
+	}
+	updatedArmoury, err := armoury.MergeLateBindingMaps(r.updatedParams)
+	if err != nil {
+		return nil, err
+	}
+	return updatedArmoury, nil
+}
+
 //nolint:revive,nestif,funlen,gocognit // TODO: investigate
 func (mv *monoValentExecution) GetExecutor() (func(pc primitive.IPrimitiveCtx) internaldto.ExecutorOutput, error) {
 	prov, err := mv.tableMeta.GetProvider()
@@ -1280,6 +1337,62 @@ func (mv *monoValentExecution) GetExecutor() (func(pc primitive.IPrimitiveCtx) i
 		return nil, authCtxErr
 	}
 	ex := func(pc primitive.IPrimitiveCtx) internaldto.ExecutorOutput {
+		requiredDepedencyKey, requiredKeyExists := mv.bldrInput.GetRequiredDataRequestKey()
+		// lateBindingData := map[int]map[string]any{}
+
+		rawData := map[int]map[int]any{}
+		if requiredKeyExists {
+			dataRequest, isRequestExist := mv.graphHolder.GetExecutorOutput(requiredDepedencyKey)
+			if !isRequestExist {
+				return internaldto.NewErroneousExecutorOutput(fmt.Errorf("required data request not found"))
+			}
+			requiredData := dataRequest.GetRawResult()
+			if requiredData == nil {
+				return internaldto.NewErroneousExecutorOutput(fmt.Errorf("nil required data"))
+			}
+			rowCount := 0
+			for res, err := requiredData.Read(); res != nil; {
+				rows, rowErr := res.GetMap()
+				if rowErr != nil {
+					return internaldto.NewErroneousExecutorOutput(fmt.Errorf("error reading required data: %w", rowErr))
+				}
+				for k, r := range rows {
+					rawData[k] = r
+					rowCount++
+				}
+				logging.GetLogger().Infoln(fmt.Sprintf("read %d rows for required data with key '%s'", len(rows), requiredDepedencyKey))
+				if err != nil {
+					if err == io.EOF {
+						break
+					}
+					return internaldto.NewErroneousExecutorOutput(fmt.Errorf("error reading required data: %w", err))
+				}
+				break // yes weird, the api should change
+			}
+			if rowCount == 0 {
+				return internaldto.NewNopEmptyExecutorOutput([]string{"operation skipped due to nil data"})
+			}
+			logging.GetLogger().Infoln(fmt.Sprintf("required data for key '%s' = %v", requiredDepedencyKey, rawData))
+		}
+
+		lateBindingData := map[int]map[string]any{}
+		var lateBindingErr error
+		if len(rawData) > 0 {
+			parserNode, _ := mv.bldrInput.GetParserNode()
+			transformer := newRawTransformer(
+				parserNode,
+				rawData,
+			)
+			lateBindingData, lateBindingErr = transformer.Transform()
+			if lateBindingErr != nil {
+				return internaldto.NewErroneousExecutorOutput(fmt.Errorf("error transforming required data: %w", lateBindingErr))
+			}
+			logging.GetLogger().Infoln(fmt.Sprintf("late binding data = %v", lateBindingData))
+		}
+		armouryGenerator := newUpdatedArmouryGenerator(
+			mv.tableMeta,
+			lateBindingData,
+		)
 		currentTcc := mv.insertPreparedStatementCtx.GetGCCtrlCtrs().Clone()
 		mv.graphHolder.AddTxnControlCounters(currentTcc)
 		mr := prov.InferMaxResultsElement(m)
@@ -1302,7 +1415,7 @@ func (mv *monoValentExecution) GetExecutor() (func(pc primitive.IPrimitiveCtx) i
 				inlines[1:],
 				nil,
 			)
-			armoury, armouryErr := mv.tableMeta.GetHTTPArmoury()
+			armoury, armouryErr := armouryGenerator.GetHTTPArmoury()
 			if armouryErr != nil {
 				return internaldto.NewErroneousExecutorOutput(armouryErr)
 			}
@@ -1402,7 +1515,7 @@ func (mv *monoValentExecution) GetExecutor() (func(pc primitive.IPrimitiveCtx) i
 		case client.HTTP:
 			invRes, invErr := mv.invoker.Invoke(context.Background(), providerinvoker.Request{
 				Payload: formulation.NewPayload(
-					mv.tableMeta,
+					armouryGenerator,
 					provider,
 					m,
 					tableName,

internal/stackql/internal_data_transfer/builder_input/builder_input.go

@@ -41,6 +41,8 @@ type BuilderInput interface {
 	SetIsAwait(isAwait bool)
 	SetCommentDirectives(commentDirectives sqlparser.CommentDirectives)
 	SetIsUndo(isUndo bool)
+	SetRequiredDataRequestKey(key string)
+	GetRequiredDataRequestKey() (string, bool)
 	SetDependencyNode(dependencyNode primitivegraph.PrimitiveNode)
 	SetParserNode(node sqlparser.SQLNode)
 	SetParamMap(paramMap map[int]map[string]interface{})
@@ -83,6 +85,7 @@ type builderInput struct {
 	txnCtrlCtrs             internaldto.TxnControlCounters
 	tableInsertionContainer tableinsertioncontainer.TableInsertionContainer
 	insertCtx               drm.PreparedStatementCtx
+	requiredDataRequestKey  string
 }
 
 func NewBuilderInput(
@@ -99,6 +102,14 @@ func NewBuilderInput(
 	}
 }
 
+func (bi *builderInput) GetRequiredDataRequestKey() (string, bool) {
+	return bi.requiredDataRequestKey, bi.requiredDataRequestKey != ""
+}
+
+func (bi *builderInput) SetRequiredDataRequestKey(key string) {
+	bi.requiredDataRequestKey = key
+}
+
 func (bi *builderInput) SetInsertCtx(insertCtx drm.PreparedStatementCtx) {
 	bi.insertCtx = insertCtx
 }
@@ -272,21 +283,22 @@ func (bi *builderInput) SetIsUndo(isUndo bool) {
 
 func (bi *builderInput) Clone() BuilderInput {
 	return &builderInput{
-		graphHolder:       bi.graphHolder,
-		handlerCtx:        bi.handlerCtx,
-		paramMap:          bi.paramMap,
-		tbl:               bi.tbl,
-		node:              bi.node,
-		dependencyNode:    bi.dependencyNode,
-		commentDirectives: bi.commentDirectives,
-		isAwait:           bi.isAwait,
-		verb:              bi.verb,
-		inputAlias:        bi.inputAlias,
-		isUndo:            bi.isUndo,
-		isTargetPhysical:  bi.isTargetPhysical,
-		annotatedAst:      bi.annotatedAst,
-		txnCtrlCtrs:       bi.txnCtrlCtrs,
-		isReturning:       bi.isReturning,
-		insertCtx:         bi.insertCtx,
+		graphHolder:            bi.graphHolder,
+		handlerCtx:             bi.handlerCtx,
+		paramMap:               bi.paramMap,
+		tbl:                    bi.tbl,
+		node:                   bi.node,
+		dependencyNode:         bi.dependencyNode,
+		commentDirectives:      bi.commentDirectives,
+		isAwait:                bi.isAwait,
+		verb:                   bi.verb,
+		inputAlias:             bi.inputAlias,
+		isUndo:                 bi.isUndo,
+		isTargetPhysical:       bi.isTargetPhysical,
+		annotatedAst:           bi.annotatedAst,
+		txnCtrlCtrs:            bi.txnCtrlCtrs,
+		isReturning:            bi.isReturning,
+		insertCtx:              bi.insertCtx,
+		requiredDataRequestKey: bi.requiredDataRequestKey,
 	}
 }