omnii-gap-reduction

go.mod

@@ -20,7 +20,7 @@ require (
 	github.com/spf13/cobra v1.10.2
 	github.com/spf13/pflag v1.0.10
 	github.com/spf13/viper v1.10.1
-	github.com/stackql-labs/omnisdk v0.1.6-alpha05
+	github.com/stackql-labs/omnisdk v0.1.6-alpha17
 	github.com/stackql/any-sdk v0.6.1-beta01
 	github.com/stackql/go-suffix-map v0.0.1-alpha02
 	github.com/stackql/psql-wire v0.1.2-beta02

go.sum

@@ -367,8 +367,8 @@ 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/spiffe/go-spiffe/v2 v2.7.0 h1:uXe1MflJoHw58wAUvxVlcM7WpKtijWG7I1UidcGh6g4=
 github.com/spiffe/go-spiffe/v2 v2.7.0/go.mod h1:47Q0Q9/AqGha8QLHp+kxpH4Wca7X7EnOtlIJy3mxZ3U=
-github.com/stackql-labs/omnisdk v0.1.6-alpha05 h1:o4dcS21CFLRvBH4cNtTWyVroTrRELz/42/tgKGrIs4E=
-github.com/stackql-labs/omnisdk v0.1.6-alpha05/go.mod h1:bMRymDVvjy9j7v7nm01LugE5mF2I85ClmsjiUzURZ+I=
+github.com/stackql-labs/omnisdk v0.1.6-alpha17 h1:uiUbXqfxqCejmW8n1OaGUsnBC1PK88CgBizQfSRY/8M=
+github.com/stackql-labs/omnisdk v0.1.6-alpha17/go.mod h1:bMRymDVvjy9j7v7nm01LugE5mF2I85ClmsjiUzURZ+I=
 github.com/stackql/any-sdk v0.6.1-beta01 h1:2Jqru76BDS87Jd95n5P56ruUoB0NWQZEbVQIXa3wlOs=
 github.com/stackql/any-sdk v0.6.1-beta01/go.mod h1:xUR2t/dy8d01Ut4n3WdMe1lfxG5ipq/VNwa9dXDXI/o=
 github.com/stackql/go-suffix-map v0.0.1-alpha02 h1:JI4aiq/OfHW4a/BiNqK31NbyaR94zlUOhGZof3G/OUo=

internal/stackql/datasource/sql_datasource/generic.go

@@ -1,6 +1,7 @@
 package sql_datasource //nolint:revive,stylecheck // package name is helpful
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
 
@@ -86,3 +87,7 @@ func (ds *genericSQLDataSource) GetTableMetadata(args ...string) (sqltable.SQLTa
 // 	`
 // 	return nil, fmt.Errorf("could not obtain sql data source table metadata for table = '%s'", tableName)
 // }
+
+func (ds *genericSQLDataSource) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) {
+	return ds.db.QueryContext(ctx, query, args...)
+}

internal/stackql/datasource/sql_datasource/sql_datasource.go

@@ -1,6 +1,7 @@
 package sql_datasource //nolint:stylecheck,revive // package name is helpful
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
 
@@ -9,6 +10,12 @@ import (
 	"github.com/stackql/stackql/internal/stackql/datasource/sqltable"
 )
 
+// ContextQuerier is the optional cancellation-capable query handle used by read adapters.
+// Existing SQLDataSource implementations remain source compatible.
+type ContextQuerier interface {
+	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
+}
+
 type SQLDataSource interface {
 	Begin() (*sql.Tx, error)
 	Exec(string, ...interface{}) (sql.Result, error)

internal/stackql/intrinsic/completion_schema.go

@@ -0,0 +1,74 @@
+package intrinsic
+
+import (
+	"maps"
+	"slices"
+
+	"github.com/stackql-labs/omnisdk/pkg/omnisdk"
+	"github.com/stackql-labs/omnisdk/pkg/query"
+)
+
+type completionSchemaTable struct {
+	omnisdk.Table
+	columns []string
+	types   map[string]omnisdk.ColumnType
+}
+
+func withCompletionSchema(target, resource omnisdk.Table) omnisdk.Table {
+	columns := map[string]bool{}
+	types := map[string]omnisdk.ColumnType{}
+	for _, method := range resource.Methods() {
+		for _, name := range method.Columns() {
+			columns[name] = true
+		}
+		maps.Copy(types, method.ColumnTypes())
+	}
+	return &completionSchemaTable{Table: target, columns: slices.Sorted(maps.Keys(columns)), types: types}
+}
+func (t *completionSchemaTable) Methods() []omnisdk.MethodSignature {
+	methods := t.Table.Methods()
+	out := make([]omnisdk.MethodSignature, len(methods))
+	for i, method := range methods {
+		out[i] = newCompletedMethod(method, t.columns, t.types)
+	}
+	return out
+}
+
+type completedMethod struct {
+	omnisdk.MethodSignature
+	columns []string
+	types   map[string]omnisdk.ColumnType
+}
+
+func newCompletedMethod(
+	base omnisdk.MethodSignature, columns []string, types map[string]omnisdk.ColumnType,
+) omnisdk.MethodSignature {
+	return &completedMethod{MethodSignature: base, columns: slices.Clone(columns), types: maps.Clone(types)}
+}
+func (m *completedMethod) Columns() []string                          { return slices.Clone(m.columns) }
+func (m *completedMethod) ColumnTypes() map[string]omnisdk.ColumnType { return maps.Clone(m.types) }
+func (m *completedMethod) ServerEnvironmentBindings() map[string]string {
+	if metadata, ok := m.MethodSignature.(omnisdk.ServerEnvironmentBindings); ok {
+		return metadata.ServerEnvironmentBindings()
+	}
+	return nil
+}
+
+func describeMutationTable(registry string, target query.Target, await bool) (omnisdk.Table, error) {
+	table, err := omnisdk.DescribeMutation(registry, target.Resource().Handle(), target.Verb().String())
+	if err != nil || !await {
+		return table, err
+	}
+	resource, err := omnisdk.DescribeTable(registry, target.Resource().Handle())
+	if err != nil {
+		return nil, err
+	}
+	return withCompletionSchema(table, resource), nil
+}
+
+func (m *completedMethod) ServerTransportDefaults() map[string]string {
+	if metadata, ok := m.MethodSignature.(omnisdk.ServerTransportDefaults); ok {
+		return metadata.ServerTransportDefaults()
+	}
+	return nil
+}

internal/stackql/intrinsic/doc.go

@@ -9,9 +9,11 @@ import (
 
 	"github.com/stackql-labs/omnisdk/pkg/docparse/aot"
 	"github.com/stackql-labs/omnisdk/pkg/omnisdk"
+	"github.com/stackql-labs/omnisdk/pkg/query"
 	"github.com/stackql/any-sdk/pkg/dto"
 	"github.com/stackql/any-sdk/public/formulation"
 	"github.com/stackql/psql-wire/pkg/sqldata"
+	"github.com/stackql/stackql/internal/stackql/acid/binlog"
 	"github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto"
 	"github.com/stackql/stackql/internal/stackql/util"
 
@@ -136,7 +138,7 @@ func docMethods(ctx queryContext, bundle, service, resource string) ([]relationM
 	out := make([]relationMethod, 0, len(methods))
 	for _, method := range methods {
 		out = append(out, relationMethod{
-			requiredParams: method.RequiredParams,
+			requiredParams: unresolvedServerParams(method),
 			name:           method.Name,
 			description:    method.OperationID,
 			sqlVerb:        docSQLVerb(method.SQLVerb),
@@ -162,12 +164,13 @@ func docSelectFunc(
 	ctx queryContext,
 	node *sqlparser.Select,
 	currentProvider string,
+	initial ...map[string]omnisdk.Relation,
 ) (func() internaldto.ExecutorOutput, bool) {
 	expanded, expandErr := expandDocumentViews(ctx, node, currentProvider)
 	if expandErr != nil {
 		return refuse(expandErr), true
 	}
-	node, relations, relationErr := resolveDerivedRelations(ctx, expanded, currentProvider, 0)
+	node, relations, relationErr := resolveDerivedRelations(ctx, expanded, currentProvider, 0, initial...)
 	if relationErr != nil {
 		return refuse(relationErr), true
 	}
@@ -207,10 +210,18 @@ func docMutationFunc(
 	if err != nil {
 		return refuse(err), true
 	}
+	stmt, relations, err := prepareMutationRelations(ctx, stmt, currentProvider)
+	if err != nil {
+		return refuse(err), true
+	}
 	translated, err := newDocTranslator(currentProvider, dialect).mutation(stmt)
 	if err != nil {
 		return refuse(err), true
 	}
+	translated = withRelations(translated, relations)
+	if mutationAwait(stmt) {
+		translated = awaitedDocQuery{docQuery: translated}
+	}
 	return func() internaldto.ExecutorOutput { return runDocQuery(ctx, translated) }, true
 }
 
@@ -228,6 +239,9 @@ func openDocQueryIn(ctx queryContext, translated docQuery, runCtx context.Contex
 	relations := queryRelations(translated)
 	tables := make(map[string]omnisdk.Table, len(q.From())+1)
 	for _, join := range q.From() {
+		if _, function := join.Resource().(query.TableFunction); function {
+			continue
+		}
 		var tbl omnisdk.Table
 		var describeErr error
 		if relation, ok := relations[join.Resource().Handle()]; ok {
@@ -242,8 +256,7 @@ func openDocQueryIn(ctx queryContext, translated docQuery, runCtx context.Contex
 	}
 	var relation string
 	if target := q.Target(); target != nil {
-		tbl, describeErr := omnisdk.DescribeMutation(
-			registry, target.Resource().Handle(), target.Verb().String())
+		tbl, describeErr := describeMutationTable(registry, target, queryAwaitsMutation(translated))
 		if describeErr != nil {
 			return nil, "", describeErr
 		}
@@ -256,24 +269,31 @@ func openDocQueryIn(ctx queryContext, translated docQuery, runCtx context.Contex
 	if dialectErr != nil {
 		return nil, "", dialectErr
 	}
+	q, environmentErr := bindServerEnvironment(q, tables)
+	if environmentErr != nil {
+		return nil, "", environmentErr
+	}
 	res, resolveErr := omnisdk.ResolveIn(q, tables, dialect.catalogue())
 	if resolveErr != nil {
 		return nil, "", resolveErr
 	}
-	// omnisdk takes one credential per run: the first relation's cloud - a
-	// mutation's target - leaving the rest to the canonical environment
-	// variables.
+	// Preserve the host's configured credential for every provider, including
+	// providers following an opaque derived relation in the FROM order.
 	args, argsErr := previewArgs(ctx, translated.getBundles()[0], res.Params())
 	if argsErr != nil {
 		return nil, "", argsErr
 	}
+	args.AuthByProvider, argsErr = queryProviderAuth(ctx, translated.getBundles())
+	if argsErr != nil {
+		return nil, "", argsErr
+	}
+	if completionErr := configureMutationCompletion(ctx, translated, &args); completionErr != nil {
+		return nil, "", completionErr
+	}
+	configureMutationObserver(ctx, q.Target() != nil, &args)
 	args.Relations = relations
 	args.Tuning.Limit = translated.getLimit()
-	if hinted, ok := translated.(interface {
-		getReadHints() map[string]omnisdk.ReadHints
-	}); ok {
-		args.ReadHintsByAlias = hinted.getReadHints()
-	}
+	configureReadHints(translated, &args)
 	plan, planErr := omnisdk.NewGraphQuery(registry, res.Graph(), args)
 	if planErr != nil {
 		return nil, "", planErr
@@ -285,6 +305,23 @@ func openDocQueryIn(ctx queryContext, translated docQuery, runCtx context.Contex
 	return rows, relation, nil
 }
 
+// queryProviderAuth preserves configured credentials independently for every provider.
+func queryProviderAuth(ctx queryContext, bundles []string) (map[string]*omnisdk.Auth, error) {
+	credentials := map[string]*omnisdk.Auth{}
+	for _, bundle := range bundles {
+		auth := providerAuthContext(ctx, bundle)
+		if auth != nil && (auth.KeyFilePath != "" || auth.KeyFilePathEnvVar != "" || auth.KeyEnvVar != "") {
+			if _, err := auth.GetCredentialsBytes(); err != nil {
+				return nil, fmt.Errorf("service account credentials error: %w", err)
+			}
+		}
+		for provider, credential := range providerAuthByProvider(ctx, bundle) {
+			credentials[provider] = credential
+		}
+	}
+	return credentials, nil
+}
+
 // runDocQuery runs the query and streams its rows back.
 func runDocQuery(ctx queryContext, translated docQuery) internaldto.ExecutorOutput {
 	rows, relation, err := openDocQuery(ctx, translated)
@@ -293,16 +330,31 @@ func runDocQuery(ctx queryContext, translated docQuery) internaldto.ExecutorOutp
 	}
 	q := translated.getQuery()
 	if q.Target() != nil && len(translated.getOutputs()) == 0 {
-		return drainMutation(rows)
+		result := drainMutation(rows)
+		if mutationErr := result.GetError(); mutationErr != nil {
+			return internaldto.NewErroneousExecutorOutput(mutationHTTPFailure(ctx, q.Target().Verb().String(), mutationErr))
+		}
+		address := strings.TrimPrefix(q.Target().Resource().Handle(), UnstablePrefix)
+		message := "Undo the " + q.Target().Verb().String() + " on " + address
+		result.SetUndoLog(binlog.NewSimpleLogEntry(nil, []string{message}))
+		if queryAwaitsMutation(translated) && result.GetError() == nil {
+			return internaldto.NewExecutorOutput(nil, nil, nil, nil, nil)
+		}
+		if result.GetError() == nil && inlineMutation(ctx, q.Target().Resource().Handle(), q.Target().Verb().String()) {
+			return internaldto.NewExecutorOutput(nil, nil, nil, internaldto.NewBackendMessages([]string{"OK"}), nil)
+		}
+		return result
 	}
 	input := previewCfg
 	stream := &rowStream{
-		rows:          rows,
-		batchSize:     input.getBatchSize(),
-		flushInterval: input.getFlushInterval(),
-		table:         sqldata.NewSQLTable(0, relation),
-		typCfg:        ctx.GetTypingConfig(),
-		outputs:       translated.getOutputs(),
+		suppressFailedHeaders: true,
+		rows:                  rows,
+		batchSize:             input.getBatchSize(),
+		flushInterval:         input.getFlushInterval(),
+		table:                 sqldata.NewSQLTable(0, relation),
+		typCfg:                ctx.GetTypingConfig(),
+		outputs:               translated.getOutputs(),
+		emptyColumns:          func() []column { return declaredEmptyColumns(ctx, translated) },
 	}
 	primed, readErr := newPrimedStream(stream)
 	if readErr != nil {
@@ -311,6 +363,33 @@ func runDocQuery(ctx queryContext, translated docQuery) internaldto.ExecutorOutp
 	return internaldto.NewExecutorOutput(primed, nil, nil, nil, nil)
 }
 
+// Empty star results still have the schema declared by their read relations.
+// Actual rows continue to determine nonempty star results, including ragged fields.
+func declaredEmptyColumns(ctx queryContext, translated docQuery) []column {
+	q := translated.getQuery()
+	resources := q.From()
+	if target := q.Target(); target != nil {
+		resources = []query.Join{query.NewJoin(target.Resource(), query.Base)}
+	}
+	fields := omnisdk.Row{}
+	for _, join := range resources {
+		if _, function := join.Resource().(query.TableFunction); function {
+			continue
+		}
+		names, err := declaredRelationColumns(ctx, join.Resource().Handle(), queryRelations(translated))
+		if err != nil {
+			continue
+		}
+		for _, name := range names {
+			fields[name] = nil
+		}
+	}
+	if len(fields) == 0 {
+		return nil
+	}
+	return outputColumns(translated.getOutputs(), []omnisdk.Row{fields})
+}
+
 // drainMutation pulls a mutation's cursor to the end, which is what sends its
 // effects, and reports the outcome.
 func drainMutation(rows omnisdk.Rows) internaldto.ExecutorOutput {
@@ -406,3 +485,24 @@ func showDocMethods(
 	}
 	return prepare(ctx, columnOrder, rows, util.DefaultRowSort)
 }
+
+func inlineMutation(ctx queryContext, address, verb string) bool {
+	table, err := omnisdk.DescribeMutation(registryRoot(ctx), address, verb)
+	if err != nil {
+		return false
+	}
+	for _, method := range table.Methods() {
+		if local, ok := method.(omnisdk.InlineCommand); ok && len(local.InlineCommand()) > 0 {
+			return true
+		}
+	}
+	return false
+}
+
+func configureReadHints(translated docQuery, args *omnisdk.Args) {
+	if hinted, ok := translated.(interface {
+		getReadHints() map[string]omnisdk.ReadHints
+	}); ok {
+		args.ReadHintsByAlias = hinted.getReadHints()
+	}
+}