diff --git a/go.mod b/go.mod index 0c37df5a..598aab88 100644 --- a/go.mod +++ b/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-alpha18 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 diff --git a/go.sum b/go.sum index 63f06939..31fe286a 100644 --- a/go.sum +++ b/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-alpha18 h1:xRiXUY79rs3Wt6K/z5f5/RXuYUWVfAD7mQX6zcNcV9k= +github.com/stackql-labs/omnisdk v0.1.6-alpha18/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= diff --git a/internal/stackql/datasource/sql_datasource/generic.go b/internal/stackql/datasource/sql_datasource/generic.go index 95fa677d..bb6e9801 100644 --- a/internal/stackql/datasource/sql_datasource/generic.go +++ b/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...) +} diff --git a/internal/stackql/datasource/sql_datasource/sql_datasource.go b/internal/stackql/datasource/sql_datasource/sql_datasource.go index dcc0a3b6..95cc6159 100644 --- a/internal/stackql/datasource/sql_datasource/sql_datasource.go +++ b/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) diff --git a/internal/stackql/intrinsic/completion_schema.go b/internal/stackql/intrinsic/completion_schema.go new file mode 100644 index 00000000..7b564f8a --- /dev/null +++ b/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 +} diff --git a/internal/stackql/intrinsic/doc.go b/internal/stackql/intrinsic/doc.go index 80838f6b..9b6a97e6 100644 --- a/internal/stackql/intrinsic/doc.go +++ b/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() + } +} diff --git a/internal/stackql/intrinsic/doc_views.go b/internal/stackql/intrinsic/doc_views.go index 99ce99c7..ced9aca3 100644 --- a/internal/stackql/intrinsic/doc_views.go +++ b/internal/stackql/intrinsic/doc_views.go @@ -38,6 +38,10 @@ func expandDocumentViews( if !isDoc { return node, nil } + // Analyzer-owned handles have no provider document or resource view. + if bundle == "omni_relations" { + return node, nil + } service, resource := name.Qualifier.GetRawVal(), name.Name.GetRawVal() root, err := docRoot(ctx, bundle) if err != nil { @@ -59,7 +63,7 @@ func expandDocumentViews( if err != nil { return nil, err } - ddl, err := viewSQL(view, dialect.catalogue()) + ddl, err := viewSQL(view, dialect.catalogue(), viewLiteralBindings(node.Where, table.As.GetRawVal(), resource)) if err != nil { return nil, fmt.Errorf("view %s: %w", key, err) } @@ -93,19 +97,15 @@ func expandDocumentViews( return nil, fmt.Errorf("document views exceed the expansion depth bound") } -func viewSQL(view aot.View, dialect sqlfn.Dialect) (string, error) { +func viewSQL(view aot.View, dialect sqlfn.Dialect, bindings map[string]*sqlparser.ComparisonExpr) (string, error) { target := "sqlite3" if dialect == sqlfn.Postgres { target = "postgres" } for depth := 0; depth < 16; depth++ { - condition := strings.Join(strings.Fields(view.Predicate()), "") - matches := condition == "" || condition == "true" - if strings.HasPrefix(condition, "sqlDialect==") { - value := strings.Trim(strings.TrimPrefix(condition, "sqlDialect=="), "\"'") - matches = value == target - } else if condition != "" && condition != "true" && condition != "false" { - return "", fmt.Errorf("unsupported view predicate %q", view.Predicate()) + matches, err := matchViewPredicate(view.Predicate(), target, bindings) + if err != nil { + return "", err } if matches { return view.DDL(), nil @@ -179,36 +179,26 @@ func inlineDocumentView(node, view *sqlparser.Select, alias string) (*sqlparser. node.SelectExprs = selections // Explicit scope/body bindings on the outer query override the view's // declared defaults; business predicates remain conjunctive. - bindings := map[string]bool{} + bindings := viewLiteralBindings(node.Where, alias, alias) if node.Where != nil { - _ = sqlparser.Walk(func(n sqlparser.SQLNode) (bool, error) { - if compare, ok := n.(*sqlparser.ComparisonExpr); ok && compare.Operator == sqlparser.EqualStr { - if col, ok := compare.Left.(*sqlparser.ColName); ok { - name := col.Name.GetRawVal() - if name == "region" || strings.HasPrefix(name, "data__") { - bindings[name] = true - } - } - } - return true, nil - }, node.Where.Expr) node.Where.Expr = rewrite(node.Where.Expr).(sqlparser.Expr) } var strip func(sqlparser.Expr) sqlparser.Expr strip = func(expr sqlparser.Expr) sqlparser.Expr { if and, ok := expr.(*sqlparser.AndExpr); ok { lhs, rhs := strip(and.Left), strip(and.Right) - if lhs == nil { - return rhs - } - if rhs == nil { - return lhs - } return &sqlparser.AndExpr{Left: lhs, Right: rhs} } - if compare, ok := expr.(*sqlparser.ComparisonExpr); ok && compare.Operator == sqlparser.EqualStr { - if col, ok := compare.Left.(*sqlparser.ColName); ok && bindings[col.Name.GetRawVal()] { - return nil + if compare, ok := expr.(*sqlparser.ComparisonExpr); ok && literalViewBinding(compare) { + if col, ok := compare.Left.(*sqlparser.ColName); ok { + name := col.Name.GetRawVal() + if replacement := bindings[name]; replacement != nil && (name == "region" || strings.HasPrefix(name, "data__")) { + // Replace only literal inbox bindings. Column-to-column beta + // dependencies and business predicates retain their meaning. + bound := *replacement + bound.Left = col + return &bound + } } } return expr diff --git a/internal/stackql/intrinsic/doc_views_test.go b/internal/stackql/intrinsic/doc_views_test.go index 76bb25e6..2902b2a3 100644 --- a/internal/stackql/intrinsic/doc_views_test.go +++ b/internal/stackql/intrinsic/doc_views_test.go @@ -21,7 +21,7 @@ func TestInlineDocumentViewPreservesBusinessFiltersAndCallerScope(t *testing.T) if err != nil { t.Fatal(err) } - result := sqlparser.String(expanded) + result := strings.ReplaceAll(sqlparser.String(expanded), "\"", "") for _, want := range []string{"json_extract", "as bucket_name", "aws.cloud_control.resources", "ap-southeast-2", "AWS::S3::Bucket", "prod%"} { if !strings.Contains(result, want) { t.Fatalf("expanded query lacks %q: %s", want, result) @@ -34,3 +34,40 @@ func TestInlineDocumentViewPreservesBusinessFiltersAndCallerScope(t *testing.T) t.Fatal("view expansion mutated the caller's AST") } } + +func TestViewVariantRequiresExplicitInboxInputs(t *testing.T) { + predicate := `sqlDialect == "sqlite3" && requiredParams == [ "data__Identifier", region ]` + for _, tc := range []struct { + query string + want bool + }{ + {"select * from v where data__Identifier = 'one' and region in ('east','west')", true}, + {"select * from v where region = 'east'", false}, + {"select * from v where data__Identifier = 'one' or region = 'east'", false}, + {"select * from v where other.data__Identifier = 'one' and region = 'east'", false}, + } { + node := parseSelect(t, tc.query) + got, err := matchViewPredicate(predicate, "sqlite3", viewLiteralBindings(node.Where, "v", "v")) + if err != nil || got != tc.want { + t.Fatalf("%s: match=%v err=%v", tc.query, got, err) + } + } +} + +func TestViewFanoutOverridesLiteralInboxButRetainsDependency(t *testing.T) { + outer := parseSelect(t, "select region, id from v where region in ('east','west')") + view := parseSelect(t, "select detail.region, detail.id from p.s.listing listing join p.s.detail detail on detail.id = listing.id and detail.region = listing.region where listing.region in ('old','older') and listing.region = detail.region and listing.data__TypeName = 'kind'") + expanded, err := inlineDocumentView(outer, view, "v") + if err != nil { + t.Fatal(err) + } + result := strings.ReplaceAll(sqlparser.String(expanded), "\"", "") + for _, want := range []string{"detail.region = listing.region", "listing.region = detail.region", "listing.region in ('east', 'west')", "listing.data__TypeName = 'kind'"} { + if !strings.Contains(result, want) { + t.Fatalf("missing %s: %s", want, result) + } + } + if strings.Contains(result, "older") { + t.Fatal(result) + } +} diff --git a/internal/stackql/intrinsic/external_relations.go b/internal/stackql/intrinsic/external_relations.go new file mode 100644 index 00000000..b38fc097 --- /dev/null +++ b/internal/stackql/intrinsic/external_relations.go @@ -0,0 +1,113 @@ +package intrinsic + +import ( + "context" + "fmt" + "strings" + + "github.com/stackql-labs/omnisdk/pkg/omnisdk" + "github.com/stackql/stackql-parser/go/vt/sqlparser" + "github.com/stackql/stackql/internal/stackql/datasource/sql_datasource" +) + +type sqlSourceProvider interface { + GetSQLDataSource(string) (sql_datasource.SQLDataSource, bool) +} + +type externalSQLRelation interface { + omnisdk.Relation + externalSQLSource() bool +} +type externalRelationHandle struct{ omnisdk.Relation } + +func (externalRelationHandle) externalSQLSource() bool { return true } +func newExternalRelationHandle(relation omnisdk.Relation) externalSQLRelation { + return externalRelationHandle{Relation: relation} +} + +func externalSource(ctx queryContext, name sqlparser.TableName) (sql_datasource.SQLDataSource, bool) { + host, ok := ctx.(sqlSourceProvider) + if !ok { + return nil, false + } + connection := name.QualifierSecond.GetRawVal() + if connection == "" { + connection = ctx.GetCurrentProvider() + } + if connection == "" { + return nil, false + } + return host.GetSQLDataSource(connection) +} + +func hasExternalRelations(relations map[string]omnisdk.Relation) bool { + for _, relation := range relations { + if external, ok := relation.(externalSQLRelation); ok && external.externalSQLSource() { + return true + } + } + return false +} + +func (r *storedResolver) registerExternal( + table *sqlparser.AliasedTableExpr, name sqlparser.TableName, source sql_datasource.SQLDataSource, +) error { + reader, ok := source.(sql_datasource.ContextQuerier) + if !ok { + return fmt.Errorf("external source %s does not support cancellable reads", name.QualifierSecond.GetRawVal()) + } + if name.Qualifier.IsEmpty() { + return fmt.Errorf("external relation %s requires an explicit schema", qualifiedName(name)) + } + // SQL identifiers are parsed names and escaped individually; connection names never enter SQL. + actual := quoteRelationIdentifier(name.Qualifier.GetRawVal()) + "." + quoteRelationIdentifier(name.Name.GetRawVal()) + discovery, cancel := context.WithTimeout(context.Background(), omnisdk.DefaultRequestTimeout) + defer cancel() + //nolint:gosec // Only individually quoted SQL identifiers enter this metadata query. + metadata, err := reader.QueryContext(discovery, "SELECT * FROM "+actual+" WHERE 1=0") + if err != nil { + return err + } + defer metadata.Close() + columnTypes, err := metadata.ColumnTypes() + if err != nil { + return err + } + if metadataErr := metadata.Err(); metadataErr != nil { + return metadataErr + } + if len(columnTypes) == 0 { + return fmt.Errorf("external relation %s has no declared columns", qualifiedName(name)) + } + names := make([]string, 0, len(columnTypes)) + quoted := make([]string, 0, len(columnTypes)) + columns := map[string]omnisdk.ColumnType{} + for _, column := range columnTypes { + names = append(names, column.Name()) + quoted = append(quoted, quoteRelationIdentifier(column.Name())) + columns[column.Name()] = storedColumnType(column.DatabaseTypeName()) + } + statement := "SELECT " + strings.Join(quoted, ",") + " FROM " + actual + open := func(runCtx context.Context, _ map[string]any) (omnisdk.Rows, error) { + rows, queryErr := reader.QueryContext(runCtx, statement) + if queryErr != nil { + return nil, queryErr + } + return materializeSQLRows(runCtx, rows, names, nil) + } + local := fmt.Sprintf("external%d", len(r.relations)) + handle := UnstablePrefix + "omni_relations.sources." + local + relation, err := omnisdk.NewRelation(handle, nil, columns, open) + if err != nil { + return err + } + r.relations[handle] = newExternalRelationHandle(relation) + table.Expr = sqlparser.TableName{ + QualifierSecond: sqlparser.NewTableIdent(UnstablePrefix + "omni_relations"), + Qualifier: sqlparser.NewTableIdent("sources"), Name: sqlparser.NewTableIdent(local), + } + if table.As.IsEmpty() { + table.As = name.Name + } + return nil +} diff --git a/internal/stackql/intrinsic/external_relations_test.go b/internal/stackql/intrinsic/external_relations_test.go new file mode 100644 index 00000000..c5d79ea8 --- /dev/null +++ b/internal/stackql/intrinsic/external_relations_test.go @@ -0,0 +1,90 @@ +package intrinsic //nolint:testpackage // Verifies host metadata and execution remain behind read handles. + +import ( + "context" + "database/sql" + "errors" + "strings" + "sync/atomic" + "testing" + + "github.com/stackql/stackql/internal/stackql/datasource/sql_datasource" +) + +type externalTestContext struct { + *stagingTestCtx + source sql_datasource.SQLDataSource +} + +func (c *externalTestContext) GetSQLDataSource(name string) (sql_datasource.SQLDataSource, bool) { + return c.source, name == "pgi" +} + +type externalTestSource struct { + sql_datasource.SQLDataSource + db *sql.DB + metadata atomic.Int32 + data atomic.Int32 +} + +func (s *externalTestSource) QueryContext(ctx context.Context, statement string, args ...any) (*sql.Rows, error) { + if strings.HasSuffix(statement, "WHERE 1=0") { + s.metadata.Add(1) + if _, bounded := ctx.Deadline(); !bounded { + return nil, errors.New("unbounded metadata read") + } + } else { + s.data.Add(1) + } + return s.db.QueryContext(ctx, statement, args...) +} + +func TestConfiguredExternalSourceBecomesLazyOmniRelation(t *testing.T) { + base := newStagingTestCtx(t, "") + db, err := base.GetSQLEngine().GetDB() + if err != nil { + t.Fatal(err) + } + for _, statement := range []string{"CREATE TABLE external_rows (id INTEGER, name TEXT)", "INSERT INTO external_rows VALUES (1,'a'),(2,'b')"} { + if _, err = db.Exec(statement); err != nil { + t.Fatal(err) + } + } + source := &externalTestSource{db: db} + ctx := &externalTestContext{stagingTestCtx: base, source: source} + node := parseSelect(t, "select name from pgi.main.external_rows where id = 2") + if execution, owned := storedSelectFunc(ctx, node, ""); !owned || execution == nil { + t.Fatal("configured SQL source fell through to provider discovery") + } + prepared, relations, err := prepareStoredRelations(ctx, node) + if err != nil { + t.Fatal(err) + } + if !hasExternalRelations(relations) || source.metadata.Load() == 0 || source.data.Load() != 0 { + t.Fatalf("relations=%d metadata=%d data=%d", len(relations), source.metadata.Load(), source.data.Load()) + } + translated, err := newDocTranslator("", sqliteDialect{}).selectQuery(prepared) + if err != nil { + t.Fatal(err) + } + translated = withRelations(translated, relations) + rows, _, err := openDocQueryIn(ctx, translated, context.Background()) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + if !rows.Next() || rows.Row()["name"] != "b" { + t.Fatalf("row=%v err=%v", rows.Row(), rows.Err()) + } + if rows.Next() || rows.Err() != nil || source.data.Load() != 1 { + t.Fatalf("continuation=%v data=%d", rows.Err(), source.data.Load()) + } + canceled, cancel := context.WithCancel(context.Background()) + cancel() + if _, _, err = openDocQueryIn(ctx, translated, canceled); !errors.Is(err, context.Canceled) { + t.Fatalf("canceled read=%v", err) + } + if source.data.Load() != 1 { + t.Fatal("canceled plan queried the data source") + } +} diff --git a/internal/stackql/intrinsic/mutation_relations.go b/internal/stackql/intrinsic/mutation_relations.go new file mode 100644 index 00000000..7e073268 --- /dev/null +++ b/internal/stackql/intrinsic/mutation_relations.go @@ -0,0 +1,83 @@ +package intrinsic + +import ( + "fmt" + + "github.com/stackql-labs/omnisdk/pkg/omnisdk" + "github.com/stackql/stackql-parser/go/vt/sqlparser" +) + +// INSERT sources use the same opaque relation handles as ordinary SELECTs. +// Clone the mutation so planning does not modify the caller's syntax tree. +func prepareMutationRelations( + ctx queryContext, stmt sqlparser.Statement, current string, +) (sqlparser.Statement, map[string]omnisdk.Relation, error) { + if deletion, ok := stmt.(*sqlparser.Delete); ok && len(deletion.SelectExprs) > 0 { + if sqlparser.ExtractCommentDirectives(deletion.Comments).IsSet("AWAIT") { + return nil, nil, fmt.Errorf("returning from asynchronous delete is disallowed") + } + } + insert, ok := stmt.(*sqlparser.Insert) + if !ok { + return stmt, nil, nil + } + source, ok := insert.Rows.(*sqlparser.Select) + if !ok || literalSource(source.From) { + return stmt, nil, nil + } + expanded, err := expandDocumentViews(ctx, source, current) + if err != nil { + return nil, nil, err + } + prepared, relations, err := resolveDerivedRelations(ctx, expanded, current, 0) + if err != nil { + return nil, nil, err + } + dialect, err := backendDialect(ctx) + if err != nil { + return nil, nil, err + } + translator := newDocTranslator(current, dialect) + if err = expandDerivedStars(ctx, prepared, translator, relations); err != nil { + return nil, nil, err + } + sourcePlan := *prepared + sourcePlan.SelectExprs = make(sqlparser.SelectExprs, len(prepared.SelectExprs)) + projected := make(sqlparser.SelectExprs, len(prepared.SelectExprs)) + for i, selection := range prepared.SelectExprs { + expression, isExpression := selection.(*sqlparser.AliasedExpr) + if !isExpression { + return nil, nil, fmt.Errorf("insert source must name its columns") + } + field := expression.As.GetRawVal() + if field == "" { + field = fmt.Sprintf("_omni_insert_%d", i) + } + cloned := *expression + cloned.As = sqlparser.NewColIdent(field) + sourcePlan.SelectExprs[i] = &cloned + projected[i] = &sqlparser.AliasedExpr{Expr: &sqlparser.ColName{Name: sqlparser.NewColIdent(field)}} + } + names, open, err := planDerivedRead(ctx, &sourcePlan, translator, relations) + if err != nil { + return nil, nil, err + } + columns, err := derivedColumns(names) + if err != nil { + return nil, nil, err + } + local := fmt.Sprintf("r%d", len(relations)) + handle := UnstablePrefix + "omni_relations.sources." + local + relation, err := omnisdk.NewRelation(handle, nil, columns, open) + if err != nil { + return nil, nil, err + } + relations[handle] = relation + clone := *insert + clone.Rows = &sqlparser.Select{SelectExprs: projected, + From: sqlparser.TableExprs{&sqlparser.AliasedTableExpr{Expr: sqlparser.TableName{ + QualifierSecond: sqlparser.NewTableIdent(UnstablePrefix + "omni_relations"), + Qualifier: sqlparser.NewTableIdent("sources"), Name: sqlparser.NewTableIdent(local), + }}}} + return &clone, relations, nil +} diff --git a/internal/stackql/intrinsic/mutation_relations_test.go b/internal/stackql/intrinsic/mutation_relations_test.go new file mode 100644 index 00000000..7090013d --- /dev/null +++ b/internal/stackql/intrinsic/mutation_relations_test.go @@ -0,0 +1,59 @@ +package intrinsic //nolint:testpackage // Exercises the host analyzer boundary. + +import ( + "context" + "testing" + + "github.com/stackql-labs/omnisdk/pkg/omnisdk" + + "github.com/stackql/stackql-parser/go/vt/sqlparser" +) + +func TestMutationDerivedSourceUsesStatementLocalHandle(t *testing.T) { + ctx := newStagingTestCtx(t, "") + original, err := sqlparser.Parse("insert into stackql_unstable_google.cloudkms.key_rings(projectsId, locationsId, keyRingId) select s.project, 'global', s.id from (select 'p' as project, 'r' as id) s returning name") + if err != nil { + t.Fatal(err) + } + before := sqlparser.String(original) + prepared, relations, err := prepareMutationRelations(ctx, original, "") + if err != nil { + t.Fatal(err) + } + if sqlparser.String(original) != before { + t.Fatal("planning mutated the caller's statement") + } + if len(relations) != 2 { + t.Fatalf("relations=%v", relations) + } + translated, err := newDocTranslator("", sqliteDialect{}).mutation(prepared) + if err != nil { + t.Fatal(err) + } + if len(translated.getQuery().From()) != 1 { + t.Fatal("lost mutation source") + } + for _, relation := range []omnisdk.Relation{relations[translated.getQuery().From()[0].Resource().Handle()]} { + rows, err := relation.Open(context.Background(), nil) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + if !rows.Next() || rows.Row()["_omni_insert_0"] != "p" || rows.Row()["_omni_insert_2"] != "r" { + t.Fatalf("source row=%v error=%v", rows.Row(), rows.Err()) + } + if rows.Next() || rows.Err() != nil { + t.Fatalf("source continuation=%v", rows.Err()) + } + } +} + +func TestAwaitDeleteReturningIsRejectedBeforeExecution(t *testing.T) { + statement, err := sqlparser.Parse("delete /*+ AWAIT */ from stackql_unstable_google.compute.firewalls where project = 'p' and firewall = 'f' returning kind") + if err != nil { + t.Fatal(err) + } + if _, _, err := prepareMutationRelations(nil, statement, ""); err == nil || err.Error() != "returning from asynchronous delete is disallowed" { + t.Fatalf("error=%v", err) + } +} diff --git a/internal/stackql/intrinsic/omnisdk.go b/internal/stackql/intrinsic/omnisdk.go index a22f83a3..061e200a 100644 --- a/internal/stackql/intrinsic/omnisdk.go +++ b/internal/stackql/intrinsic/omnisdk.go @@ -194,17 +194,19 @@ func openStream( } type rowStream struct { - rows omnisdk.Rows - batchSize int - flushInterval time.Duration - produced chan omnisdk.Row - producerOnce sync.Once - columns []column - table sqldata.ISQLTable - projection sqlparser.SelectExprs - outputs []string - typCfg columnFactory - done bool + suppressFailedHeaders bool + rows omnisdk.Rows + batchSize int + flushInterval time.Duration + produced chan omnisdk.Row + producerOnce sync.Once + columns []column + table sqldata.ISQLTable + projection sqlparser.SelectExprs + outputs []string + emptyColumns func() []column + typCfg columnFactory + done bool } type columnFactory interface { @@ -213,6 +215,9 @@ type columnFactory interface { func (rs *rowStream) Read() (sqldata.ISQLResult, error) { if rs.done { + if err := rs.rows.Err(); err != nil { + return rs.errorResult(), err + } return rs.result(nil), io.EOF } rs.startProducer() @@ -228,7 +233,7 @@ func (rs *rowStream) Read() (sqldata.ISQLResult, error) { if !ok { rs.done = true if err := rs.rows.Err(); err != nil { - return rs.result(nil), err + return rs.errorResult(), err } return rs.result(nil), io.EOF } @@ -240,7 +245,8 @@ func (rs *rowStream) Read() (sqldata.ISQLResult, error) { if !more { rs.done = true if err := rs.rows.Err(); err != nil { - return rs.result(batch), err + //nolint:nilerr // Deliver the buffered rows before reporting this error on the next read. + return rs.result(batch), nil } return rs.result(batch), io.EOF } @@ -252,6 +258,13 @@ func (rs *rowStream) Read() (sqldata.ISQLResult, error) { return rs.result(batch), nil } +func (rs *rowStream) errorResult() sqldata.ISQLResult { + if rs.suppressFailedHeaders { + return nil + } + return rs.result(nil) +} + func (rs *rowStream) flushIntervalOrDefault() time.Duration { if rs.flushInterval <= 0 { return defaultFlushInterval @@ -273,19 +286,32 @@ func (rs *rowStream) startProducer() { }) } -func (rs *rowStream) result(batch []omnisdk.Row) sqldata.ISQLResult { - if len(rs.columns) == 0 && len(rs.outputs) > 0 { +func (rs *rowStream) resolveColumns(batch []omnisdk.Row) { + if len(rs.columns) > 0 { + return + } + if len(rs.outputs) > 0 { rs.columns = outputColumns(rs.outputs, batch) - } else if len(rs.columns) == 0 && len(batch) > 0 { - for _, name := range sortedKeys(batch[0]) { - rs.columns = append(rs.columns, column{name: name}) + if len(batch) == 0 && len(rs.columns) == 0 && rs.emptyColumns != nil { + rs.columns = rs.emptyColumns() } - if len(rs.projection) > 0 { - if selected, err := projection(rs.projection, rs.columns); err == nil { - rs.columns = selected - } + return + } + if len(batch) == 0 { + return + } + for _, name := range sortedKeys(batch[0]) { + rs.columns = append(rs.columns, column{name: name}) + } + if len(rs.projection) > 0 { + if selected, err := projection(rs.projection, rs.columns); err == nil { + rs.columns = selected } } +} + +func (rs *rowStream) result(batch []omnisdk.Row) sqldata.ISQLResult { + rs.resolveColumns(batch) columns := rs.GetColumns() rows := make([]sqldata.ISQLRow, 0, len(batch)) for _, row := range batch { @@ -359,6 +385,9 @@ func selectFunc( node *sqlparser.Select, currentProvider string, ) (func() internaldto.ExecutorOutput, bool) { + if execute, handled := storedSelectFunc(ctx, node, currentProvider); handled { + return execute, true + } if isDoc, err := fromDocProviders(node.From, currentProvider); isDoc { if err != nil { return refuse(err), true @@ -676,7 +705,7 @@ type primedStream struct { func newPrimedStream(inner *rowStream) (sqldata.ISQLResultStream, error) { first, err := inner.Read() - if err != nil && !errors.Is(err, io.EOF) { + if err != nil && !errors.Is(err, io.EOF) && (first == nil || len(first.GetColumns()) == 0) { inner.Close() //nolint:errcheck // the read error is the one worth reporting return nil, err } @@ -686,8 +715,14 @@ func newPrimedStream(inner *rowStream) (sqldata.ISQLResultStream, error) { func (ps *primedStream) Read() (sqldata.ISQLResult, error) { if !ps.replayed { ps.replayed = true + if ps.firstErr != nil && !errors.Is(ps.firstErr, io.EOF) { + return ps.first, nil + } return ps.first, ps.firstErr } + if ps.firstErr != nil && !errors.Is(ps.firstErr, io.EOF) { + return ps.inner.result(nil), ps.firstErr + } return ps.inner.Read() } diff --git a/internal/stackql/intrinsic/omnisdk_test.go b/internal/stackql/intrinsic/omnisdk_test.go index c9f18c56..08dfbd07 100644 --- a/internal/stackql/intrinsic/omnisdk_test.go +++ b/internal/stackql/intrinsic/omnisdk_test.go @@ -2,6 +2,7 @@ package intrinsic //nolint:testpackage // tests unexported rowStream and pickMet import ( "bytes" + "context" "errors" "fmt" "io" @@ -359,3 +360,38 @@ func TestNullValuesRemainJSONNull(t *testing.T) { t.Fatalf("got %q, want %q", buf.String(), want) } } + +func TestPrimedStreamPreservesRowsAndHeadersBeforeTerminalError(t *testing.T) { + original := errors.New("upstream failed") + for _, rows := range [][]omnisdk.Row{nil, {{"name": "kept"}}} { + stream, cursor := newTestStream(rows, []string{"name"}) + cursor.err = original + primed, err := newPrimedStream(stream) + if err != nil { + t.Fatal(err) + } + first, err := primed.Read() + if err != nil || len(first.GetRows()) != len(rows) || len(first.GetColumns()) != 1 { + t.Fatalf("rows=%v result=%v error=%v", rows, first, err) + } + _, err = primed.Read() + if !errors.Is(err, original) { + t.Fatalf("lost original error: %v", err) + } + if err = primed.Close(); err != nil { + t.Fatal(err) + } + if !cursor.closed { + t.Fatal("cursor not closed") + } + } +} + +func TestEmptyStagedSourceRetainsTerminalErrorForPresentation(t *testing.T) { + original := errors.New("first page failed") + source := newOmniBatchSource(&fakeRows{err: original}, []string{"name"}, 10) + batch, err := source.Next(context.Background()) + if len(batch) != 0 || !errors.Is(err, io.EOF) || !errors.Is(source.terminalErr, original) { + t.Fatalf("batch=%v error=%v terminal=%v", batch, err, source.terminalErr) + } +} diff --git a/internal/stackql/intrinsic/operation_completion.go b/internal/stackql/intrinsic/operation_completion.go new file mode 100644 index 00000000..c855a7d8 --- /dev/null +++ b/internal/stackql/intrinsic/operation_completion.go @@ -0,0 +1,281 @@ +package intrinsic + +import ( + "context" + "fmt" + "io" + "maps" + "net/url" + "os" + "strings" + "sync" + "time" + + "github.com/stackql-labs/omnisdk/pkg/docparse/aot" + "github.com/stackql-labs/omnisdk/pkg/docparse/stackqldoc" + "github.com/stackql-labs/omnisdk/pkg/omnisdk" + "github.com/stackql/stackql-parser/go/vt/sqlparser" +) + +type awaitedDocQuery struct{ docQuery } + +func (awaitedDocQuery) awaitsMutation() bool { return true } + +func mutationAwait(stmt sqlparser.Statement) bool { + var comments sqlparser.Comments + switch node := stmt.(type) { + case *sqlparser.Insert: + comments = node.Comments + case *sqlparser.Update: + comments = node.Comments + case *sqlparser.Delete: + comments = node.Comments + } + return sqlparser.ExtractCommentDirectives(comments).IsSet("AWAIT") +} +func queryAwaitsMutation(q docQuery) bool { + a, ok := q.(interface{ awaitsMutation() bool }) + return ok && a.awaitsMutation() +} + +const ( + operationPollInterval = 10 * time.Second + operationWaitTimeout = 10 * time.Minute +) + +type completionRead func(context.Context, string, map[string]any) (omnisdk.Row, error) +type completionWait func(context.Context, time.Duration) error + +type linkedOperationCompletion struct { + mu sync.Mutex + read completionRead + wait completionWait + writer io.Writer +} + +func newLinkedOperationCompletion( + read completionRead, wait completionWait, writer io.Writer, +) omnisdk.MutationCompletion { + return &linkedOperationCompletion{read: read, wait: wait, writer: writer} +} + +func newOperationCompletion(ctx queryContext, address string) (omnisdk.MutationCompletion, error) { + parts := strings.Split(address, ".") + if len(parts) != 3 || strings.TrimPrefix(parts[0], UnstablePrefix) != "google" || parts[1] != "compute" { + return nil, fmt.Errorf("AWAIT completion is not supported for %s", address) + } + bundle := strings.TrimPrefix(parts[0], UnstablePrefix) + root, err := docRoot(ctx, bundle) + if err != nil { + return nil, err + } + catalogue, err := stackqldoc.Open(os.DirFS(root)) + if err != nil { + return nil, err + } + var writer io.Writer = io.Discard + if output, ok := ctx.(interface{ GetOutErrFile() io.Writer }); ok { + writer = output.GetOutErrFile() + } + read := func(runCtx context.Context, link string, original map[string]any) (omnisdk.Row, error) { + return readCompletionLink(ctx, runCtx, root, bundle, parts[1], catalogue, link, original) + } + wait := func(runCtx context.Context, duration time.Duration) error { + timer := time.NewTimer(duration) + defer timer.Stop() + select { + case <-runCtx.Done(): + return runCtx.Err() + case <-timer.C: + return nil + } + } + return newLinkedOperationCompletion(read, wait, writer), nil +} + +func (c *linkedOperationCompletion) Complete( + parent context.Context, _ string, verb string, inputs map[string]any, operation omnisdk.Row, +) (omnisdk.Row, error) { + ctx, cancel := context.WithTimeout(parent, operationWaitTimeout) + defer cancel() + if err := ctx.Err(); err != nil { + return nil, err + } + kind, _ := operation["kind"].(string) + if kind != "compute#operation" { + return nil, fmt.Errorf("AWAIT expected compute#operation, got %q", kind) + } + action, _ := operation["operationType"].(string) + if action == "" { + return nil, fmt.Errorf("AWAIT operation has no operationType") + } + status, _ := operation["status"].(string) + poll, _ := operation["selfLink"].(string) + target, _ := operation["targetLink"].(string) + elapsed := time.Duration(0) + for status != "DONE" { + if status != "PENDING" && status != "RUNNING" { + return nil, fmt.Errorf("AWAIT operation has unsupported status %q", status) + } + if poll == "" { + return nil, fmt.Errorf("AWAIT operation has no selfLink") + } + if err := c.wait(ctx, operationPollInterval); err != nil { + return nil, err + } + elapsed += operationPollInterval + c.diagnosticf("%s: %s in progress, %d seconds elapsed\n", kind, action, int(elapsed.Seconds())) + next, err := c.read(ctx, poll, inputs) + if err != nil { + return nil, err + } + operation = next + status, _ = operation["status"].(string) + if link, ok := operation["targetLink"].(string); ok && link != "" { + target = link + } + } + if failure, exists := operation["error"]; exists && failure != nil { + return nil, fmt.Errorf("AWAIT operation failed: %v", failure) + } + c.diagnosticf("%s: %s complete\n", kind, action) + if verb == "delete" { + return nil, nil //nolint:nilnil // Successful DELETE has no resource row. + } + if target == "" { + return nil, fmt.Errorf("AWAIT completed without targetLink") + } + return c.read(ctx, target, inputs) +} + +func (c *linkedOperationCompletion) diagnosticf(format string, values ...any) { + c.mu.Lock() + defer c.mu.Unlock() + fmt.Fprintf(c.writer, format, values...) //nolint:errcheck // Diagnostics do not determine completion. +} + +// Links select only declared GET operations within the original provider/service. +// The returned host is never used for dispatch; SDK catalog routing and auth remain authoritative. +func readCompletionLink( + ctx queryContext, runCtx context.Context, root, bundle, service string, + catalogue aot.Catalog, link string, original map[string]any, +) (omnisdk.Row, error) { + address, params, err := completionAddress(catalogue, service, link) + if err != nil { + return nil, err + } + if err = checkCompletionScope(params, original); err != nil { + return nil, err + } + args, err := previewArgs(ctx, bundle, params) + if err != nil { + return nil, err + } + args.HTTPReadErrors = nil // Completion absence remains an unknown outcome. + plan, err := omnisdk.NewSelectFromCatalog(root, address, args) + if err != nil { + return nil, err + } + rows, err := plan.Open(runCtx) + if err != nil { + return nil, err + } + return singleCompletionRow(rows) +} + +func checkCompletionScope(params map[string]string, original map[string]any) error { + for name, value := range params { + if supplied, present := original[name]; present && fmt.Sprint(supplied) != value { + return fmt.Errorf("AWAIT link changes explicit %s scope", name) + } + } + return nil +} + +func completionAddress(catalogue aot.Catalog, service, link string) (string, map[string]string, error) { + resources, err := catalogue.Resources(service) + if err != nil { + return "", nil, err + } + for _, resource := range resources { + address, ok := catalogue.Address(service, resource) + if !ok { + continue + } + operations, opErr := catalogue.Operations(address, "select") + if opErr != nil { + continue + } + for _, operation := range operations { + if operation.Request().Method() != "GET" { + continue + } + if params, matched := completionLinkParams(operation.Request().URL(), link); matched { + return address, params, nil + } + } + } + return "", nil, fmt.Errorf("AWAIT link does not match a declared GET operation in %s", service) +} + +func configureMutationCompletion(ctx queryContext, translated docQuery, args *omnisdk.Args) error { + if !queryAwaitsMutation(translated) { + return nil + } + completion, err := newOperationCompletion(ctx, translated.getQuery().Target().Resource().Handle()) + if err != nil { + return err + } + args.MutationCompletion = completion + return nil +} + +func singleCompletionRow(rows omnisdk.Rows) (omnisdk.Row, error) { + defer rows.Close() + if !rows.Next() { + if err := rows.Err(); err != nil { + return nil, err + } + return nil, fmt.Errorf("AWAIT read returned no row") + } + row := maps.Clone(rows.Row()) + if rows.Next() { + return nil, fmt.Errorf("AWAIT read returned multiple rows") + } + if err := rows.Err(); err != nil { + return nil, err + } + return row, nil +} + +func completionLinkParams(template, link string) (map[string]string, bool) { + declared, err := url.Parse(template) + if err != nil { + return nil, false + } + actual, err := url.Parse(link) + if err != nil || actual.Scheme == "" || actual.Host == "" || actual.RawQuery != "" { + return nil, false + } + expected := strings.Split(strings.Trim(declared.Path, "/"), "/") + received := strings.Split(strings.Trim(actual.Path, "/"), "/") + if len(expected) != len(received) { + return nil, false + } + params := map[string]string{} + for i, part := range expected { + if strings.HasPrefix(part, "{") && strings.HasSuffix(part, "}") { + name := strings.TrimSuffix(strings.TrimPrefix(part, "{"), "}") + if received[i] == "" { + return nil, false + } + if prior, exists := params[name]; exists && prior != received[i] { + return nil, false + } + params[name] = received[i] + } else if part != received[i] { + return nil, false + } + } + return params, true +} diff --git a/internal/stackql/intrinsic/operation_completion_test.go b/internal/stackql/intrinsic/operation_completion_test.go new file mode 100644 index 00000000..f3de6387 --- /dev/null +++ b/internal/stackql/intrinsic/operation_completion_test.go @@ -0,0 +1,86 @@ +package intrinsic //nolint:testpackage // Exercises completion policy independently of HTTP fixtures. + +import ( + "bytes" + "context" + "errors" + "reflect" + "testing" + "time" + + "github.com/stackql-labs/omnisdk/pkg/omnisdk" +) + +func TestOperationCompletionPollsBeforeReturningResource(t *testing.T) { + var diagnostics bytes.Buffer + links := []string{} + waits := 0 + read := func(_ context.Context, link string, inputs map[string]any) (omnisdk.Row, error) { + links = append(links, link) + if inputs["project"] != "p" { + t.Errorf("scope=%v", inputs) + } + if link == "https://original/operation" { + return omnisdk.Row{"status": "DONE", "targetLink": "https://original/resource"}, nil + } + return omnisdk.Row{"name": "completed", "direction": "INGRESS"}, nil + } + wait := func(_ context.Context, duration time.Duration) error { + waits++ + if duration != 10*time.Second { + t.Errorf("interval=%v", duration) + } + return nil + } + completion := newLinkedOperationCompletion(read, wait, &diagnostics) + row, err := completion.Complete(context.Background(), "google.compute.firewalls", "update", map[string]any{"project": "p"}, omnisdk.Row{"kind": "compute#operation", "operationType": "patch", "status": "PENDING", "selfLink": "https://original/operation"}) + if err != nil { + t.Fatal(err) + } + if row["name"] != "completed" || waits != 1 || !reflect.DeepEqual(links, []string{"https://original/operation", "https://original/resource"}) { + t.Fatalf("row=%v waits=%d links=%v", row, waits, links) + } + expected := "compute#operation: patch in progress, 10 seconds elapsed\ncompute#operation: patch complete\n" + if diagnostics.String() != expected { + t.Fatalf("diagnostics=%q", diagnostics.String()) + } +} + +func TestOperationCompletionCancellationAndDelete(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + completion := newLinkedOperationCompletion(func(context.Context, string, map[string]any) (omnisdk.Row, error) { + t.Fatal("canceled/deleted operation read its target") + return nil, context.Canceled + }, func(context.Context, time.Duration) error { t.Fatal("canceled operation waited"); return nil }, &bytes.Buffer{}) + operation := omnisdk.Row{"kind": "compute#operation", "operationType": "delete", "status": "DONE"} + if _, err := completion.Complete(ctx, "google.compute.firewalls", "delete", nil, operation); !errors.Is(err, context.Canceled) { + t.Fatalf("error=%v", err) + } + if row, err := completion.Complete(context.Background(), "google.compute.firewalls", "delete", nil, operation); row != nil || err != nil { + t.Fatalf("row=%v error=%v", row, err) + } +} + +func TestCompletionLinkMatchingUsesDeclaredPath(t *testing.T) { + template := "https://catalog.example/compute/v1/projects/{project}/global/operations/{operation}" + params, ok := completionLinkParams(template, "https://operation.example/compute/v1/projects/p/global/operations/o") + if !ok || !reflect.DeepEqual(params, map[string]string{"project": "p", "operation": "o"}) { + t.Fatalf("params=%v matched=%v", params, ok) + } + for _, link := range []string{"https://operation.example/unrelated", "https://operation.example/compute/v1/projects/p/global/operations/o?scope=other", "/compute/v1/projects/p/global/operations/o"} { + if _, matched := completionLinkParams(template, link); matched { + t.Fatalf("accepted undeclared link %s", link) + } + } +} + +func TestCompletionLinkCannotChangeExplicitScope(t *testing.T) { + original := map[string]any{"project": "explicit-project"} + if err := checkCompletionScope(map[string]string{"project": "other-project"}, original); err == nil { + t.Fatal("completion changed project") + } + if err := checkCompletionScope(map[string]string{"project": "explicit-project", "operation": "new-token"}, original); err != nil { + t.Fatal(err) + } +} diff --git a/internal/stackql/intrinsic/preview_services.go b/internal/stackql/intrinsic/preview_services.go index efd9f6e0..c5c99f6d 100644 --- a/internal/stackql/intrinsic/preview_services.go +++ b/internal/stackql/intrinsic/preview_services.go @@ -13,10 +13,14 @@ package intrinsic import ( "context" "fmt" + "io" + "net/http" "path/filepath" "strings" + "sync" "github.com/stackql-labs/omnisdk/pkg/omnisdk" + "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/internal_data_transfer/internaldto" @@ -158,13 +162,23 @@ func previewArgs(ctx queryContext, cloud string, params map[string]string) (omni return omnisdk.Args{}, err } input := previewCfg + upstream, err := dto.GetUpstreamErrorCfg(ctx.GetRuntimeContext().UpstreamErrorCfgRaw) + if err != nil { + return omnisdk.Args{}, err + } return omnisdk.Args{ - Params: params, - AuthByProvider: providerAuthByProvider(ctx, cloud), - Endpoint: input.getEndpoint(), - InsecureSkipTLSVerify: input.getInsecureSkipTLSVerify() || ctx.GetRuntimeContext().AllowInsecure, - Tuning: omnisdk.Tuning{RequestTimeout: apiRequestTimeout(ctx)}, - Dialect: dialect.catalogue(), + LegacyPagination: true, + HTTPReadErrors: newReadErrorPolicy(ctx, upstream.HTTP), + StrictResponseSelection: true, + Params: params, + AuthByProvider: providerAuthByProvider(ctx, cloud), + Endpoint: input.getEndpoint(), + InsecureSkipTLSVerify: input.getInsecureSkipTLSVerify() || ctx.GetRuntimeContext().AllowInsecure, + Tuning: omnisdk.Tuning{ + RequestTimeout: apiRequestTimeout(ctx), + PageLimit: ctx.GetRuntimeContext().HTTPPageLimit, + }, + Dialect: dialect.catalogue(), }, nil } @@ -329,3 +343,40 @@ func describePreviewTableFunc( "run SHOW RESOURCES IN %s.%s for what it accepts", relationAddress(service, resource), ProviderName, service)), true } + +// The host owns diagnostic wording and its output stream. +type readErrorPolicy struct { + dto.HTTPUpstreamErrorCfg + writer io.Writer + mu sync.Mutex +} + +func newReadErrorPolicy(ctx queryContext, config dto.HTTPUpstreamErrorCfg) omnisdk.HTTPReadErrorPolicy { + var writer io.Writer = io.Discard + if output, ok := ctx.(interface{ GetOutErrFile() io.Writer }); ok { + writer = output.GetOutErrFile() + } + return &readErrorPolicy{HTTPUpstreamErrorCfg: config, writer: writer} +} + +// StackQL presents absent provider resources as empty tables. This host policy +// applies only to resource reads; SDK mutation and auth exchanges remain strict. +func (p *readErrorPolicy) IsIgnored(status int) bool { + return status == http.StatusNotFound || p.HTTPUpstreamErrorCfg.IsIgnored(status) +} +func (p *readErrorPolicy) IgnoredStatus(status int, body string) { + p.mu.Lock() + defer p.mu.Unlock() + diagnostic := fmt.Sprintf("http response status code: %d, response body: %s", status, body) + if body == "" { + diagnostic = fmt.Sprintf("http response status code: %d, response body is nil", status) + } + if status == http.StatusNotFound && !p.HTTPUpstreamErrorCfg.IsIgnored(status) { + // Absence ends only this read, retaining the host's established diagnostic. + fmt.Fprintln(p.writer, diagnostic) //nolint:errcheck // Diagnostics do not change the read result. + return + } + //nolint:errcheck // Diagnostics do not change the read result. + fmt.Fprintf(p.writer, "%s\nignoring upstream response: %d %s\n", + diagnostic, status, strings.ToUpper(http.StatusText(status))) +} diff --git a/internal/stackql/intrinsic/projection_names.go b/internal/stackql/intrinsic/projection_names.go new file mode 100644 index 00000000..aa5297b2 --- /dev/null +++ b/internal/stackql/intrinsic/projection_names.go @@ -0,0 +1,28 @@ +package intrinsic + +import ( + "github.com/stackql/stackql-parser/go/vt/sqlparser" + "github.com/stackql/stackql/internal/stackql/parserutil" +) + +// projectionName retains the host's legacy labels without changing the SDK +// expression or the values transferred between exchanges. +func projectionName(node *sqlparser.AliasedExpr) (string, error) { + if name := node.As.GetRawVal(); name != "" { + return name, nil + } + if col, ok := node.Expr.(*sqlparser.ColName); ok { + return col.Name.GetRawVal(), nil + } + function, isFunction := node.Expr.(*sqlparser.FuncExpr) + if isFunction && len(function.Exprs) > 1 && !function.IsAggregate() && function.Over == nil { + handles, err := parserutil.ExtractSelectColumnNames(&sqlparser.Select{SelectExprs: sqlparser.SelectExprs{node}}, nil) + if err != nil { + return "", err + } + if len(handles) == 1 && handles[0].Name != "" { + return handles[0].Name, nil + } + } + return sqlparser.String(node.Expr), nil +} diff --git a/internal/stackql/intrinsic/projection_names_test.go b/internal/stackql/intrinsic/projection_names_test.go new file mode 100644 index 00000000..1489845a --- /dev/null +++ b/internal/stackql/intrinsic/projection_names_test.go @@ -0,0 +1,29 @@ +//nolint:testpackage // Verify host presentation metadata separately from SDK expression evaluation. +package intrinsic + +import ( + "testing" + + "github.com/stackql/stackql-parser/go/vt/sqlparser" +) + +func TestProjectionLabelsPreserveLegacyMultiArgumentNames(t *testing.T) { + for _, tc := range []struct{ sql, name string }{ + {"select substr(u.name, 1, instr(u.name, '@') - 1) from p.s.users u", "name"}, + {"select split_part(u.name, '/', -1) from p.s.users u", "name"}, + {"select split_part(u.name, '/', -1) as short from p.s.users u", "short"}, + {"select upper(name) from p.s.users", "upper(name)"}, + {"select sum(size) from p.s.users", "sum(size)"}, + } { + node := parseSelect(t, tc.sql) + selected := node.SelectExprs[0].(*sqlparser.AliasedExpr) + before := sqlparser.String(selected) + got, err := projectionName(selected) + if err != nil || got != tc.name { + t.Fatalf("%s: name=%q err=%v want=%q", tc.sql, got, err, tc.name) + } + if sqlparser.String(selected) != before { + t.Fatal("labeling changed the executable expression") + } + } +} diff --git a/internal/stackql/intrinsic/read_absence_test.go b/internal/stackql/intrinsic/read_absence_test.go new file mode 100644 index 00000000..f2e97e17 --- /dev/null +++ b/internal/stackql/intrinsic/read_absence_test.go @@ -0,0 +1,34 @@ +package intrinsic //nolint:testpackage // Verifies host read-absence policy and diagnostics. + +import ( + "bytes" + "strings" + "testing" + + "github.com/stackql/any-sdk/pkg/dto" +) + +func TestReadAbsencePreservesStrictOtherFailures(t *testing.T) { + var diagnostic bytes.Buffer + policy := &readErrorPolicy{writer: &diagnostic} + for _, status := range []int{400, 401, 403, 429, 500, 503} { + if policy.IsIgnored(status) { + t.Fatalf("status %d silently ignored", status) + } + } + if !policy.IsIgnored(404) { + t.Fatal("absent resource is not an empty read") + } + policy.IgnoredStatus(404, `{"msg":"Project Not Found"}`) + if diagnostic.String() != "http response status code: 404, response body: {\"msg\":\"Project Not Found\"}\n" { + t.Fatalf("absence diagnostics=%q", diagnostic.String()) + } + policy.HTTPUpstreamErrorCfg = dto.HTTPUpstreamErrorCfg{IgnoreStatusCodes: []string{"503"}} + if !policy.IsIgnored(503) { + t.Fatal("explicit policy lost") + } + policy.IgnoredStatus(503, "temporarily unavailable") + if !strings.Contains(diagnostic.String(), "ignoring upstream response: 503 SERVICE UNAVAILABLE") { + t.Fatalf("diagnostics=%q", diagnostic.String()) + } +} diff --git a/internal/stackql/intrinsic/relation_reads.go b/internal/stackql/intrinsic/relation_reads.go index f699a97c..14488391 100644 --- a/internal/stackql/intrinsic/relation_reads.go +++ b/internal/stackql/intrinsic/relation_reads.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "fmt" + "maps" "sort" "github.com/stackql-labs/omnisdk/pkg/omnisdk" @@ -48,6 +49,7 @@ const derivedRelationMaxDepth = 32 // Planning performs no reads; each child retains its own relational boundary. func resolveDerivedRelations( ctx queryContext, original *sqlparser.Select, current string, depth int, + inherited ...map[string]omnisdk.Relation, ) (*sqlparser.Select, map[string]omnisdk.Relation, error) { if depth >= derivedRelationMaxDepth { return nil, nil, fmt.Errorf("derived relations exceed nesting bound") @@ -55,6 +57,11 @@ func resolveDerivedRelations( clone := *original clone.From = cloneRelationTables(original.From) relations := map[string]omnisdk.Relation{} + if len(inherited) > 0 { + for handle, relation := range inherited[0] { + relations[handle] = relation + } + } var walk func(sqlparser.TableExpr) error walk = func(expr sqlparser.TableExpr) error { switch table := expr.(type) { @@ -101,7 +108,7 @@ func resolveDerivedTable( if err != nil { return err } - prepared, children, err := resolveDerivedRelations(ctx, expanded, current, depth+1) + prepared, children, err := resolveDerivedRelations(ctx, expanded, current, depth+1, maps.Clone(relations)) if err != nil { return err } @@ -345,7 +352,7 @@ func expandDerivedStars( expressions = append(expressions, expr) continue } - expanded, expandErr := derivedStarColumns(ctx, source, star, relations) + expanded, expandErr := resolvedStarColumns(ctx, source, star, relations) if expandErr != nil { return expandErr } @@ -355,32 +362,6 @@ func expandDerivedStars( return nil } -func derivedStarColumns( - ctx queryContext, source translatedSource, star *sqlparser.StarExpr, relations map[string]omnisdk.Relation, -) (sqlparser.SelectExprs, error) { - var expressions sqlparser.SelectExprs - for _, join := range source.joins() { - resource := join.Resource() - if qualifier := star.TableName.Name.GetRawVal(); qualifier != "" && qualifier != resource.Alias() { - continue - } - names, err := declaredRelationColumns(ctx, resource.Handle(), relations) - if err != nil { - return nil, err - } - for _, name := range names { - expressions = append(expressions, &sqlparser.AliasedExpr{Expr: &sqlparser.ColName{ - Name: sqlparser.NewColIdent(name), - Qualifier: sqlparser.TableName{Name: sqlparser.NewTableIdent(resource.Alias())}, - }}) - } - } - if len(expressions) == 0 { - return nil, fmt.Errorf("unknown derived star qualifier %s", star.TableName.Name.GetRawVal()) - } - return expressions, nil -} - func declaredRelationColumns( ctx queryContext, handle string, relations map[string]omnisdk.Relation, ) ([]string, error) { diff --git a/internal/stackql/intrinsic/server_environment.go b/internal/stackql/intrinsic/server_environment.go new file mode 100644 index 00000000..89a09723 --- /dev/null +++ b/internal/stackql/intrinsic/server_environment.go @@ -0,0 +1,86 @@ +package intrinsic + +import ( + "fmt" + "maps" + "os" + "slices" + + "github.com/stackql-labs/omnisdk/pkg/omnisdk" + "github.com/stackql-labs/omnisdk/pkg/query" +) + +// Environment resolution is host policy, limited to explicit document annotations. +// SQL parameters take precedence. The SDK receives ordinary explicit bindings. +func bindServerEnvironment(q query.Unresolved, tables map[string]omnisdk.Table) (query.Unresolved, error) { + where := slices.Clone(q.Where()) + for _, alias := range slices.Sorted(maps.Keys(tables)) { + declared := map[string]string{} + for _, method := range tables[alias].Methods() { + if metadata, ok := method.(omnisdk.ServerEnvironmentBindings); ok { + for name, environment := range metadata.ServerEnvironmentBindings() { + declared[name] = environment + } + } + } + for _, name := range slices.Sorted(maps.Keys(declared)) { + if explicitServerBinding(q, alias, name) { + continue + } + value, exists := os.LookupEnv(declared[name]) + if !exists || value == "" { + return nil, fmt.Errorf("cannot find any viable servers: %s requires %s", alias, name) + } + where = append(where, query.NewEq(query.NewColumn(alias, name), query.NewLiteral(value))) + } + } + bound, err := queryWithWhere(q, where) + if err != nil { + return nil, err + } + return bindServerTransport(bound, tables) +} + +func explicitServerBinding(q query.Unresolved, alias, name string) bool { + matches := func(e query.Expr) bool { + col, ok := e.(query.Column) + return ok && col.Name() == name && (col.Qualifier() == "" || col.Qualifier() == alias) + } + predicates := slices.Clone(q.Where()) + for _, join := range q.From() { + predicates = append(predicates, join.On()...) + } + for _, predicate := range predicates { + switch p := predicate.(type) { + case query.Compare: + if p.Op() == query.Eq && (matches(p.Left()) || matches(p.Right())) { + return true + } + case query.In: + if matches(p.Expr()) { + return true + } + } + } + target := q.Target() + if target == nil || target.Resource().Alias() != alias { + return false + } + return slices.ContainsFunc(target.Set(), func(assignment query.Assignment) bool { return assignment.Column() == name }) +} + +func unresolvedServerParams(method omnisdk.DocMethod) []string { + required := make([]string, 0, len(method.RequiredParams)) + for _, name := range method.RequiredParams { + if method.TransportDefaults[name] != "" { + continue + } + if environment := method.ServerEnvironment[name]; environment != "" { + if value, exists := os.LookupEnv(environment); exists && value != "" { + continue + } + } + required = append(required, name) + } + return required +} diff --git a/internal/stackql/intrinsic/server_environment_test.go b/internal/stackql/intrinsic/server_environment_test.go new file mode 100644 index 00000000..7671db16 --- /dev/null +++ b/internal/stackql/intrinsic/server_environment_test.go @@ -0,0 +1,92 @@ +package intrinsic //nolint:testpackage // Tests host environment policy before SDK resolution. + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stackql-labs/omnisdk/pkg/omnisdk" + "github.com/stackql-labs/omnisdk/pkg/query" +) + +//nolint:gocognit // End-to-end metadata, precedence, and missing-input assertions share one fixture. +func TestServerEnvironmentBindingsPreserveSQLPrecedence(t *testing.T) { + dir := t.TempDir() + document := `openapi: 3.0.0 +servers: + - url: 'https://{deployment}.example.test' + variables: + deployment: {default: ignored, x-stackQL-envVar: OMNI_EXPLICIT_TEST_DEPLOYMENT} +paths: + /items: + get: + responses: + '200': + content: + application/json: + schema: {type: array, items: {type: object, properties: {name: {type: string}}}} +components: + x-stackQL-resources: + items: + methods: + select: + operation: {$ref: '#/paths/~1items/get'} + response: {mediaType: application/json} +` + if err := os.WriteFile(filepath.Join(dir, "provider.yaml"), []byte("name: p\nproviderServices:\n s:\n service: {$ref: s.yaml}\n"), 0600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "s.yaml"), []byte(document), 0600); err != nil { + t.Fatal(err) + } + table, err := omnisdk.DescribeTable(dir, "stackql_unstable_p.s.items") + if err != nil { + t.Fatal(err) + } + tables := map[string]omnisdk.Table{"i": table} + methods, err := omnisdk.DocMethods(dir, "p", "s", "items") + if err != nil || len(methods) != 1 { + t.Fatalf("methods=%v err=%v", methods, err) + } + t.Setenv("OMNI_EXPLICIT_TEST_DEPLOYMENT", "from-env") + for _, explicit := range []bool{false, true} { + where := []query.Predicate{} + expected := "from-env" + if explicit { + expected = "from-sql" + where = append(where, query.NewEq(query.NewColumn("i", "deployment"), query.NewLiteral(expected))) + } + original, err := query.New([]query.Join{query.NewJoin(query.NewResource("i", table.Address()), query.Base)}, where, []query.Output{query.NewOutput("name", query.NewColumn("i", "name"))}) + if err != nil { + t.Fatal(err) + } + bound, err := bindServerEnvironment(original, tables) + if err != nil { + t.Fatal(err) + } + if len(original.Where()) != len(where) { + t.Fatal("host mutated original query") + } + if len(bound.Where()) != 1 { + t.Fatalf("bindings=%v", bound.Where()) + } + predicate := bound.Where()[0].(query.Compare) + if got := predicate.Right().(query.Literal).Value(); got != expected { + t.Fatalf("binding=%v want=%v", got, expected) + } + } + if got := unresolvedServerParams(methods[0]); len(got) != 0 { + t.Fatalf("env-resolved required=%v", got) + } + t.Setenv("OMNI_EXPLICIT_TEST_DEPLOYMENT", "") + original, err := query.New([]query.Join{query.NewJoin(query.NewResource("i", table.Address()), query.Base)}, nil, []query.Output{query.NewOutput("name", query.NewColumn("i", "name"))}) + if err != nil { + t.Fatal(err) + } + if _, err := bindServerEnvironment(original, tables); err == nil { + t.Fatal("missing environment silently used document default") + } + if got := unresolvedServerParams(methods[0]); len(got) != 1 || got[0] != "deployment" { + t.Fatalf("unresolved required=%v", got) + } +} diff --git a/internal/stackql/intrinsic/server_transport.go b/internal/stackql/intrinsic/server_transport.go new file mode 100644 index 00000000..5b96f9fe --- /dev/null +++ b/internal/stackql/intrinsic/server_transport.go @@ -0,0 +1,52 @@ +package intrinsic + +import ( + "fmt" + "maps" + "slices" + + "github.com/stackql-labs/omnisdk/pkg/omnisdk" + "github.com/stackql-labs/omnisdk/pkg/query" +) + +// Transport defaults are explicit document metadata for the URL scheme only. +// Neither the SDK nor this host substitutes default hosts, regions or accounts. +func bindServerTransport(q query.Unresolved, tables map[string]omnisdk.Table) (query.Unresolved, error) { + where := slices.Clone(q.Where()) + for _, alias := range slices.Sorted(maps.Keys(tables)) { + defaults := map[string]string{} + for _, method := range tables[alias].Methods() { + metadata, ok := method.(omnisdk.ServerTransportDefaults) + if !ok { + continue + } + if err := mergeTransportDefaults(defaults, metadata.ServerTransportDefaults(), q, alias); err != nil { + return nil, err + } + } + for _, name := range slices.Sorted(maps.Keys(defaults)) { + where = append(where, query.NewEq(query.NewColumn(alias, name), query.NewLiteral(defaults[name]))) + } + } + return queryWithWhere(q, where) +} + +func mergeTransportDefaults(target, declared map[string]string, q query.Unresolved, alias string) error { + for name, value := range declared { + if explicitServerBinding(q, alias, name) { + continue + } + if prior, exists := target[name]; exists && prior != value { + return fmt.Errorf("conflicting transport defaults for %s.%s; supply an explicit binding", alias, name) + } + target[name] = value + } + return nil +} + +func queryWithWhere(q query.Unresolved, where []query.Predicate) (query.Unresolved, error) { + if target := q.Target(); target != nil { + return query.NewMutation(target, q.From(), where, q.Select()) + } + return query.New(q.From(), where, q.Select()) +} diff --git a/internal/stackql/intrinsic/server_transport_test.go b/internal/stackql/intrinsic/server_transport_test.go new file mode 100644 index 00000000..cf8e7422 --- /dev/null +++ b/internal/stackql/intrinsic/server_transport_test.go @@ -0,0 +1,91 @@ +package intrinsic //nolint:testpackage // Verifies host-only transport defaults and explicit scope. + +import ( + "os" + "path/filepath" + "reflect" + "testing" + + "github.com/stackql-labs/omnisdk/pkg/omnisdk" + "github.com/stackql-labs/omnisdk/pkg/query" +) + +func TestTransportDefaultsPreserveExplicitInputsAndScope(t *testing.T) { + dir := t.TempDir() + files := map[string]string{ + "provider.yaml": "name: p\nproviderServices:\n s:\n service: {$ref: s.yaml}\n", + "s.yaml": `openapi: 3.0.0 +servers: + - url: '{scheme}://{host}/{region}' + variables: + scheme: {default: https} + host: {default: ignored.example.test} + region: {default: ignored-region} +paths: + /items: + get: + responses: + '200': + content: + application/json: + schema: {type: array, items: {type: object, properties: {id: {type: string}}}} +components: + x-stackQL-resources: + items: + methods: + select: {operation: {$ref: '#/paths/~1items/get'}, response: {mediaType: application/json}} +`, + } + for name, body := range files { + if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0600); err != nil { + t.Fatal(err) + } + } + table, err := omnisdk.DescribeTable(dir, "stackql_unstable_p.s.items") + if err != nil { + t.Fatal(err) + } + from := []query.Join{query.NewJoin(query.NewResource("i", table.Address()), query.Base)} + selected := []query.Output{query.NewOutput("id", query.NewColumn("i", "id"))} + for _, explicit := range []bool{false, true} { + where := []query.Predicate{query.NewEq(query.NewColumn("i", "host"), query.NewLiteral("caller.test")), query.NewEq(query.NewColumn("i", "region"), query.NewLiteral("caller-region"))} + expected := "https" + if explicit { + expected = "http" + where = append(where, query.NewEq(query.NewColumn("i", "scheme"), query.NewLiteral(expected))) + } + original, queryErr := query.New(from, where, selected) + if queryErr != nil { + t.Fatal(queryErr) + } + bound, bindingErr := bindServerEnvironment(original, map[string]omnisdk.Table{"i": table}) + if bindingErr != nil { + t.Fatal(bindingErr) + } + resolved, resolveErr := omnisdk.Resolve(bound, map[string]omnisdk.Table{"i": table}) + if resolveErr != nil { + t.Fatal(resolveErr) + } + if got := resolved.Graph().Nodes()[0].Params()["scheme"]; got != expected { + t.Fatalf("scheme=%s want=%s", got, expected) + } + } + original, err := query.New(from, nil, selected) + if err != nil { + t.Fatal(err) + } + bound, err := bindServerEnvironment(original, map[string]omnisdk.Table{"i": table}) + if err != nil { + t.Fatal(err) + } + if len(bound.Where()) != 1 { + t.Fatal("host substituted scope defaults") + } + methods, err := omnisdk.DocMethods(dir, "p", "s", "items") + if err != nil { + t.Fatal(err) + } + if got := unresolvedServerParams(methods[0]); !reflect.DeepEqual(got, []string{"host", "region"}) { + t.Fatalf("required=%v", got) + } +} diff --git a/internal/stackql/intrinsic/staged.go b/internal/stackql/intrinsic/staged.go index 2b549c03..6c3c7fd8 100644 --- a/internal/stackql/intrinsic/staged.go +++ b/internal/stackql/intrinsic/staged.go @@ -156,7 +156,11 @@ func stagedSelectWithRelations( if err != nil { return refuse(err) } - staged, err := planStagedSelect(node, newDocTranslator(currentProvider, dialect), ctx.GetASTFormatter()) + translator := newDocTranslator(currentProvider, dialect) + if err = expandDerivedStars(ctx, node, translator, relations); err != nil { + return refuse(err) + } + staged, err := planStagedSelect(node, translator, ctx.GetASTFormatter()) if err != nil { return refuse(err) } @@ -237,23 +241,24 @@ func (r *stagedRefs) collectSelectExprs( sqlparser.String(expr), UnstablePrefix+"*") } // Output names match the streamed path: an alias, else a bare - // column's own name, else the expression as written. The last is + // column's own name, else its legacy expression label. The last is // reported by position, so the outer statement aliases it safely. outerExpr := &sqlparser.AliasedExpr{Expr: aliased.Expr, As: aliased.As} - name := aliased.As.GetRawVal() - if name == "" { - if col, isCol := aliased.Expr.(*sqlparser.ColName); isCol { - name = col.Name.GetRawVal() + name, err := projectionName(aliased) + if err != nil { + return nil, nil, nil, err + } + if aliased.As.IsEmpty() { + if _, isColumn := aliased.Expr.(*sqlparser.ColName); isColumn { outerExpr.As = sqlparser.NewColIdent(name) } else { - name = sqlparser.String(aliased.Expr) outerExpr.As = sqlparser.NewColIdent(stagedColumnPrefix + "out_" + strconv.Itoa(i)) } } outputNames = append(outputNames, name) selectExprs = append(selectExprs, outerExpr) aliases[strings.ToLower(outerExpr.As.GetRawVal())] = true - if err := r.collect(aliased.Expr, nil); err != nil { + if err = r.collect(aliased.Expr, nil); err != nil { return nil, nil, nil, err } } @@ -294,10 +299,10 @@ func planStagedSelect( } } // A statement referencing no column, such as count(*), still needs one - // staged row per source row. omnisdk outputs only columns, so the rows - // are taken whole and staged as a single null placeholder column. + // staged row per source row. Read only the first source so joined sources + // with duplicate column names do not cause ambiguous wildcard expansion. if len(refs.outputs) == 0 { - refs.outputs = append(refs.outputs, query.NewOutput("", query.NewStar(""))) + refs.outputs = append(refs.outputs, query.NewOutput("", query.NewStar(src.joins()[0].Resource().Alias()))) refs.names = append(refs.names, stagedColumnPrefix+"0") } aliasOutputs, err := whereAliasOutputs(node) @@ -552,9 +557,6 @@ func (s *omniBatchSource) Next(context.Context) ([][]any, error) { return batch, nil } if err := s.rows.Err(); err != nil { - if s.readRows == 0 { - return nil, err - } // Preserve completed pages, then surface the failed continuation to the // client after the partial result. The upstream failure remains an error. s.terminalErr = err diff --git a/internal/stackql/intrinsic/staged_test.go b/internal/stackql/intrinsic/staged_test.go index 6d9a96f8..fc23fa28 100644 --- a/internal/stackql/intrinsic/staged_test.go +++ b/internal/stackql/intrinsic/staged_test.go @@ -10,10 +10,12 @@ import ( "path/filepath" "testing" + "github.com/stackql/any-sdk/pkg/db/sqlcontrol" "github.com/stackql/any-sdk/pkg/dto" "github.com/stackql/any-sdk/public/sqlengine" "github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto" "github.com/stackql/stackql/internal/stackql/output" + "github.com/stackql/stackql/internal/stackql/sql_system" "github.com/stackql/stackql/internal/stackql/typing" "github.com/stackql/stackql/pkg/astformat" @@ -24,8 +26,11 @@ type stagingTestCtx struct { engine sqlengine.SQLEngine runtime dto.RuntimeCtx typCfg typing.Config + system sql_system.SQLSystem } +func (c *stagingTestCtx) GetSQLSystem() sql_system.SQLSystem { return c.system } + func (c *stagingTestCtx) GetCurrentProvider() string { return "" } func (c *stagingTestCtx) SetCurrentProvider(string) {} func (c *stagingTestCtx) GetTypingConfig() typing.Config { return c.typCfg } @@ -161,6 +166,7 @@ func newStagingTestCtx(t *testing.T, endpoint string) *stagingTestCtx { return &stagingTestCtx{ engine: engine, runtime: dto.RuntimeCtx{ + IndirectDepthMax: 5, RegistryRaw: `{"url":"file:` + filepath.ToSlash(filepath.Join(dir, "registry")) + `"}`, SQLBackendCfgRaw: sqlBackendRaw, }, @@ -208,12 +214,111 @@ func TestStagedSelectEndToEnd(t *testing.T) { ctx := newStagingTestCtx(t, srv.URL) t.Setenv("FIXTURE_TOKEN", "test-token") + sqlCfg, sysErr := dto.GetSQLBackendCfg(ctx.runtime.SQLBackendCfgRaw) + if sysErr != nil { + t.Fatal(sysErr) + } + ctx.system, sysErr = sql_system.NewSQLSystem(ctx.engine, "", sqlcontrol.GetControlAttributes("standard"), sqlCfg, nil, ctx.typCfg, "") + if sysErr != nil { + t.Fatal(sysErr) + } + if sysErr = ctx.system.CreateView("scoped_members", "select login,id from stackql_unstable_fixture.orgs.members where org = 'dummyorg'", false, nil); sysErr != nil { + t.Fatal(sysErr) + } + columns := []typing.RelationalColumn{typing.NewRelationalColumn("id", "integer"), typing.NewRelationalColumn("note", "text")} + if sysErr = ctx.system.CreatePhysicalTable("local_people", columns, `CREATE TABLE "local_people" (id INTEGER,note TEXT)`, false); sysErr != nil { + t.Fatal(sysErr) + } + if _, sysErr = ctx.engine.Exec(`INSERT INTO "local_people" (id,note) VALUES (10,'chosen')`); sysErr != nil { + t.Fatal(sysErr) + } + if sysErr = ctx.system.CreateMaterializedView("local_cache", []typing.RelationalColumn{typing.NewRelationalColumn("id", "integer")}, "select 10 as id", false, "select 10 as id"); sysErr != nil { + t.Fatal(sysErr) + } + + if sysErr = ctx.system.CreateView("stored_people", "select id,note from local_people", false, nil); sysErr != nil { + t.Fatal(sysErr) + } + const from = " from stackql_unstable_fixture.orgs.members where org = 'dummyorg'" for _, tc := range []struct { sql string want string }{ + { + sql: "select m.login from stackql_unstable_fixture.orgs.members m join local_people p " + + "on p.id=e.value, json_each('[10]') e where m.org='dummyorg' and m.id=p.id", + want: "login\nb\n", + }, + + { + sql: "select login, case when id=10 then 'chosen' else json('invalid-json') end as choice" + from + " and id=10", + want: "login,choice\nb,chosen\n", + }, + { + sql: "select login from stackql_unstable_fixture.orgs.members m join local_people p " + + "on p.id=case m.id when 10 then m.id else -1 end where m.org='dummyorg'", + want: "login\nb\n", + }, + { + sql: "select substr(login,1,instr(login,'b')-1) as prefix" + from + " and id=10", + want: "prefix\n\n", + }, + + { + sql: "select 'member:' || login as label" + from + " and id=10", + want: "label\nmember:b\n", + }, + + { + sql: "with local_people as (select 2 as id) " + + "select m.login, v.note from stackql_unstable_fixture.orgs.members m " + + "join stored_people v on m.id=v.id where m.org='dummyorg'", + want: "login,note\nb,chosen\n", + }, + + { + sql: "with picked as (select login,id" + from + " order by id desc limit 2) " + + "select login from picked order by login", + want: "login\nb\nd\n", + }, + { + sql: "with picked as (select login,id" + from + "), " + + "tallied(user_count) as (select count(*) from picked) select user_count from tallied", + want: "user_count\n4\n", + }, + { + sql: "with picked as (select login" + from + " and id=10), " + + "nested as (with picked as (select login" + from + " and id=2) select login from picked) " + + "select p.login, n.login as nested from picked p cross join nested n", + want: "login,nested\nb,c\n", + }, + { + sql: "with local_people as (select id" + from + " and id=2) " + + "select m.login, p.id from stackql_unstable_fixture.orgs.members m " + + "join local_people p on m.id=p.id where m.org='dummyorg'", + want: "login,id\nc,2\n", + }, + + { + sql: "select count(*) as n from stackql_unstable_fixture.orgs.members m " + + "join stackql_unstable_fixture.orgs.members n on m.org=n.org and m.id=n.id " + + "where m.org='dummyorg' and n.org='dummyorg'", + want: "n\n4\n", + }, + + { + sql: "select m.login, e.value from stackql_unstable_fixture.orgs.members m, json_each('[1,2]') e " + + "where m.org='dummyorg' and m.id=10 order by e.value", + want: "login,value\nb,1\nb,2\n", + }, + { + sql: "select count(*) as n from stackql_unstable_fixture.orgs.members m " + + "join stackql_unstable_fixture.orgs.members n on m.id=n.id where m.org='dummyorg' and n.org='dummyorg'", + want: "n\n4\n", + }, + { sql: "select s.login from (select login, id" + from + ") s where s.id > 2 order by s.id desc", want: "login\nb\nd\n", @@ -250,6 +355,19 @@ func TestStagedSelectEndToEnd(t *testing.T) { sql: "select s.login from (select login, lag(id) over (order by id) as previous_id" + from + ") s where s.previous_id is null", want: "login\na\n", }, + + { + sql: "select v.login,p.note from scoped_members v join local_people p on v.id=p.id order by v.login", + want: "login,note\nb,chosen\n", + }, + { + sql: "select v.login from local_cache m join scoped_members v on m.id=v.id", + want: "login\nb\n", + }, + { + sql: "select v.login from (select * from scoped_members order by id desc limit 2) v join local_people p on v.id=p.id", + want: "login\nb\n", + }, { // Numeric ordering: as text, 10 would sort before 2. sql: "select login, id" + from + " order by id desc limit 2 offset 1", @@ -300,3 +418,13 @@ func TestStagedSelectEndToEnd(t *testing.T) { t.Fatalf("staging tables left behind: %d", remaining) } } + +func TestExpandedCTEDepthHonorsHostLimit(t *testing.T) { + query := parseSelect(t, "select x.id from (select y.id from (select 1 as id) y) x") + if err := validateRelationDepth(query.From, 1, 0); err == nil { + t.Fatal("nested relations exceeded the configured depth without an error") + } + if err := validateRelationDepth(query.From, 2, 0); err != nil { + t.Fatal(err) + } +} diff --git a/internal/stackql/intrinsic/star_schema.go b/internal/stackql/intrinsic/star_schema.go new file mode 100644 index 00000000..86afca65 --- /dev/null +++ b/internal/stackql/intrinsic/star_schema.go @@ -0,0 +1,98 @@ +package intrinsic + +import ( + "slices" + "strings" + + "github.com/stackql-labs/omnisdk/pkg/omnisdk" + "github.com/stackql-labs/omnisdk/pkg/query" + "github.com/stackql/stackql-parser/go/vt/sqlparser" +) + +// resolvedStarColumns shares operation selection with streamed execution. The +// host only turns the resulting projection metadata back into SQL AST columns. +func resolvedStarColumns(ctx queryContext, source translatedSource, star *sqlparser.StarExpr, + relations map[string]omnisdk.Relation) (sqlparser.SelectExprs, error) { + tables, err := starTables(ctx, source, relations) + if err != nil { + return nil, err + } + outputs := []query.Output{query.NewOutput("", query.NewStar(star.TableName.Name.GetRawVal()))} + q, err := query.New(source.joins(), source.where(), outputs) + if err != nil { + return nil, err + } + q, err = bindServerEnvironment(q, tables) + if err != nil { + return nil, err + } + dialect, err := backendDialect(ctx) + if err != nil { + return nil, err + } + resolved, err := omnisdk.ResolveIn(q, tables, dialect.catalogue()) + if err != nil { + return nil, err + } + var methods map[string]omnisdk.MethodSignature + if selected, ok := resolved.(omnisdk.SelectedMethods); ok { + methods = selected.SelectedMethods() + } + return starSQLColumns(resolved.Graph(), methods), nil +} + +func starTables( + ctx queryContext, source translatedSource, relations map[string]omnisdk.Relation, +) (map[string]omnisdk.Table, error) { + tables := map[string]omnisdk.Table{} + for _, join := range source.joins() { + resource := join.Resource() + if _, function := resource.(query.TableFunction); function { + continue + } + if relation, present := relations[resource.Handle()]; present { + tables[resource.Alias()] = relation + continue + } + table, err := omnisdk.DescribeTable(registryRoot(ctx), resource.Handle()) + if err != nil { + return nil, err + } + tables[resource.Alias()] = table + } + return tables, nil +} + +func starScopeNames(method omnisdk.MethodSignature) []string { + if declared, ok := method.(omnisdk.ServerBindings); ok { + return declared.ServerVariables() + } + return nil +} + +func starSQLColumns( + graph omnisdk.Graph, methods map[string]omnisdk.MethodSignature, +) sqlparser.SelectExprs { + var expressions sqlparser.SelectExprs + for _, projection := range graph.Projections() { + var names []string + for _, column := range projection.Columns() { + if !strings.HasPrefix(column.Out(), "\x00") { + names = append(names, column.Out()) + } + } + if len(names) == 0 { + continue + } + // Scope columns describe host output; they do not introduce value producers. + names = append(names, starScopeNames(methods[projection.Alias()])...) + slices.Sort(names) + for _, name := range slices.Compact(names) { + expressions = append(expressions, &sqlparser.AliasedExpr{Expr: &sqlparser.ColName{ + Name: sqlparser.NewColIdent(name), + Qualifier: sqlparser.TableName{Name: sqlparser.NewTableIdent(projection.Alias())}, + }}) + } + } + return expressions +} diff --git a/internal/stackql/intrinsic/stored_relations.go b/internal/stackql/intrinsic/stored_relations.go new file mode 100644 index 00000000..3231a2c1 --- /dev/null +++ b/internal/stackql/intrinsic/stored_relations.go @@ -0,0 +1,366 @@ +package intrinsic + +import ( + "context" + "fmt" + "maps" + "strings" + + "github.com/stackql-labs/omnisdk/pkg/omnisdk" + "github.com/stackql/stackql-parser/go/vt/sqlparser" + "github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto" + "github.com/stackql/stackql/internal/stackql/sql_system" +) + +// storedRelationResolver resolves metadata before provider discovery. Backend +// relations become ordinary read handles; views keep their own query boundary. +type storedRelationResolver interface { + Prepare(*sqlparser.Select) (*sqlparser.Select, map[string]omnisdk.Relation, error) +} +type storedResolver struct { + ctx queryContext + system sql_system.SQLSystem + relations map[string]omnisdk.Relation + active map[string]bool + ctes map[string]*sqlparser.Select +} + +func newStoredResolver(ctx queryContext, system sql_system.SQLSystem) storedRelationResolver { + return &storedResolver{ctx: ctx, system: system, relations: map[string]omnisdk.Relation{}, active: map[string]bool{}} +} +func prepareStoredRelations( + ctx queryContext, node *sqlparser.Select, +) (*sqlparser.Select, map[string]omnisdk.Relation, error) { + if !IsUnstableEnabled() { + return node, nil, nil + } + var system sql_system.SQLSystem + if host, ok := ctx.(interface{ GetSQLSystem() sql_system.SQLSystem }); ok { + system = host.GetSQLSystem() + } + if _, external := ctx.(sqlSourceProvider); system == nil && !external { + return node, nil, nil + } + return newStoredResolver(ctx, system).Prepare(node) +} +func (r *storedResolver) Prepare(node *sqlparser.Select) (*sqlparser.Select, map[string]omnisdk.Relation, error) { + prepared, err := r.selectNode(node, 0) + if err == nil { + err = validateRelationDepth(prepared.From, r.ctx.GetRuntimeContext().IndirectDepthMax, 0) + } + return prepared, r.relations, err +} + +// validateRelationDepth applies the host's configured limit to the expanded +// tree, including chains of earlier CTE references rather than only definitions. +func validateRelationDepth(tables sqlparser.TableExprs, maximum, depth int) error { + for _, expression := range tables { + switch table := expression.(type) { + case *sqlparser.JoinTableExpr: + if err := validateRelationDepth(sqlparser.TableExprs{table.LeftExpr, table.RightExpr}, maximum, depth); err != nil { + return err + } + case *sqlparser.ParenTableExpr: + if err := validateRelationDepth(table.Exprs, maximum, depth); err != nil { + return err + } + case *sqlparser.AliasedTableExpr: + if err := validateDerivedDepth(table, maximum, depth); err != nil { + return err + } + } + } + return nil +} + +func validateDerivedDepth(table *sqlparser.AliasedTableExpr, maximum, depth int) error { + sub, isSub := table.Expr.(*sqlparser.Subquery) + if !isSub { + return nil + } + if depth+1 > maximum { + return fmt.Errorf("indirection chain length %d > %d; please do not cite views at too deep a level", + depth+1, maximum) + } + if query, isSelect := sub.Select.(*sqlparser.Select); isSelect { + return validateRelationDepth(query.From, maximum, depth+1) + } + return nil +} + +func (r *storedResolver) selectNode(node *sqlparser.Select, depth int) (*sqlparser.Select, error) { + if depth >= derivedRelationMaxDepth { + return nil, fmt.Errorf("stored relation expansion exceeds nesting bound") + } + parent := r.ctes + r.ctes = maps.Clone(parent) + if r.ctes == nil { + r.ctes = map[string]*sqlparser.Select{} + } + defer func() { r.ctes = parent }() + if err := r.with(node.With, depth); err != nil { + return nil, err + } + clone := *node + clone.With = nil + clone.From = cloneRelationTables(node.From) + for _, table := range clone.From { + if err := r.table(table, depth); err != nil { + return nil, err + } + } + return &clone, nil +} + +// with resolves definitions in lexical order. A nested WITH owns a cloned +// scope, and a CTE name shadows a stored relation only in that scope. +func (r *storedResolver) with(clause *sqlparser.With, depth int) error { + if clause == nil { + return nil + } + if clause.Recursive { + return fmt.Errorf("recursive provider CTEs require an iterative execution boundary") + } + for _, definition := range clause.CTEs { + query, isSelect := definition.Select.(*sqlparser.Select) + if !isSelect { + return fmt.Errorf("CTE %s requires a SELECT boundary", definition.Name.GetRawVal()) + } + prepared, err := r.selectNode(query, depth+1) + if err != nil { + return err + } + if err = renameCTEColumns(prepared, definition.Columns); err != nil { + return err + } + r.ctes[strings.ToLower(definition.Name.GetRawVal())] = prepared + } + return nil +} + +func renameCTEColumns(node *sqlparser.Select, columns sqlparser.Columns) error { + if len(columns) == 0 { + return nil + } + if len(columns) != len(node.SelectExprs) { + return fmt.Errorf("CTE column count does not match its projection") + } + expressions := make(sqlparser.SelectExprs, len(columns)) + for i, name := range columns { + projected, named := node.SelectExprs[i].(*sqlparser.AliasedExpr) + if !named { + return fmt.Errorf("CTE with explicit column names must name its projected expressions") + } + clone := *projected + clone.As = name + expressions[i] = &clone + } + node.SelectExprs = expressions + return nil +} + +func (r *storedResolver) table(expr sqlparser.TableExpr, depth int) error { + switch table := expr.(type) { + case *sqlparser.JoinTableExpr: + if err := r.table(table.LeftExpr, depth); err != nil { + return err + } + return r.table(table.RightExpr, depth) + case *sqlparser.ParenTableExpr: + for _, inner := range table.Exprs { + if err := r.table(inner, depth); err != nil { + return err + } + } + case *sqlparser.AliasedTableExpr: + if sub, isSub := table.Expr.(*sqlparser.Subquery); isSub { + if inner, isSelect := sub.Select.(*sqlparser.Select); isSelect { + prepared, err := r.selectNode(inner, depth+1) + if err != nil { + return err + } + table.Expr = &sqlparser.Subquery{Select: prepared} + } + return nil + } + name, isName := table.Expr.(sqlparser.TableName) + if !isName { + return nil + } + return r.named(table, name, depth) + } + return nil +} +func (r *storedResolver) named(table *sqlparser.AliasedTableExpr, name sqlparser.TableName, depth int) error { + address := qualifiedName(name) + if name.Qualifier.IsEmpty() && name.QualifierSecond.IsEmpty() { + if cte, exists := r.ctes[strings.ToLower(name.Name.GetRawVal())]; exists { + table.Expr = &sqlparser.Subquery{Select: cte} + if table.As.IsEmpty() { + table.As = name.Name + } + return nil + } + } + if source, exists := externalSource(r.ctx, name); exists { + return r.registerExternal(table, name, source) + } + if r.system == nil { + return nil + } + if view, exists := r.system.GetViewByName(address); exists { + return r.expandView(table, name, view, depth) + } + relation, exists := r.system.GetMaterializedViewByName(address) + if !exists { + relation, exists = r.system.GetPhysicalTableByName(address) + } + if !exists { + return nil + } + return r.register(table, name, relation) +} +func (r *storedResolver) register( + table *sqlparser.AliasedTableExpr, name sqlparser.TableName, metadata internaldto.RelationDTO, +) error { + handleName := fmt.Sprintf("stored%d", len(r.relations)) + handle := UnstablePrefix + "omni_relations.sources." + handleName + columns := map[string]omnisdk.ColumnType{} + var names, quoted []string + for _, column := range metadata.GetColumns() { + names = append(names, column.GetName()) + quoted = append(quoted, quoteRelationIdentifier(column.GetName())) + columns[column.GetName()] = storedColumnType(column.GetType()) + } + if len(names) == 0 { + return fmt.Errorf("stored relation %s has no declared columns", metadata.GetName()) + } + actualName, err := r.system.GetFullyQualifiedTableName(metadata.GetName()) + if err != nil { + return err + } + // Identifiers come from registered metadata and are quoted; data is never interpolated. + //nolint:gosec // G202: metadata identifiers only. + statement := "SELECT " + strings.Join(quoted, ",") + " FROM " + actualName + open := func(runCtx context.Context, _ map[string]any) (omnisdk.Rows, error) { + db, dbErr := r.ctx.GetSQLEngine().GetDB() + if dbErr != nil { + return nil, dbErr + } + result, queryErr := db.QueryContext(runCtx, statement) + if queryErr != nil { + return nil, queryErr + } + return materializeSQLRows(runCtx, result, names, nil) + } + relation, err := omnisdk.NewRelation(handle, nil, columns, open) + if err != nil { + return err + } + r.relations[handle] = relation + table.Expr = sqlparser.TableName{ + QualifierSecond: sqlparser.NewTableIdent(UnstablePrefix + "omni_relations"), + Qualifier: sqlparser.NewTableIdent("sources"), Name: sqlparser.NewTableIdent(handleName), + } + if table.As.IsEmpty() { + table.As = name.Name + } + return nil +} +func quoteRelationIdentifier(name string) string { + return `"` + strings.ReplaceAll(name, `"`, `""`) + `"` +} +func storedColumnType(raw string) omnisdk.ColumnType { + typ := "string" + switch strings.ToLower(raw) { + case "integer", "int", "bigint", "smallint", "int4", "int8": + typ = "integer" + case "numeric", "decimal", "number", "double precision", "real", "float", "float8": + typ = "number" + case "bool", "boolean": + typ = "boolean" + } + return omnisdk.ColumnType{Type: typ} +} + +// containsProviderRead excludes handles already resolved as stored relations. +// Pure database queries retain their established backend execution path. +func containsProviderRead(node *sqlparser.Select, current string, relations map[string]omnisdk.Relation) bool { + found := false + _ = sqlparser.Walk(func(n sqlparser.SQLNode) (bool, error) { + table, isTable := n.(*sqlparser.AliasedTableExpr) + if !isTable { + return true, nil + } + name, isName := table.Expr.(sqlparser.TableName) + if !isName { + return true, nil + } + if _, stored := relations[qualifiedName(name)]; stored { + return false, nil + } + if _, provider := docProvider(resolveProvider(name.QualifierSecond.GetRawVal(), current)); provider { + found = true + } + return true, nil + }, node) + return found +} + +func (r *storedResolver) expandView( + table *sqlparser.AliasedTableExpr, name sqlparser.TableName, view internaldto.RelationDTO, depth int, +) error { + address := qualifiedName(name) + if r.active[qualifiedName(name)] { + return fmt.Errorf("stored view cycle at %s", address) + } + r.active[qualifiedName(name)] = true + defer delete(r.active, address) + // Stored view definitions have their own lexical scope; a caller's CTE + // must not capture a table name inside an independently defined view. + callerCTEs := r.ctes + r.ctes = nil + defer func() { r.ctes = callerCTEs }() + statement, err := sqlparser.Parse(view.GetRawQuery()) + if err != nil { + return err + } + query, isSelect := statement.(*sqlparser.Select) + if !isSelect { + return fmt.Errorf("stored view %s requires SELECT", address) + } + prepared, err := r.selectNode(query, depth+1) + if err != nil { + return err + } + table.Expr = &sqlparser.Subquery{Select: prepared} + if table.As.IsEmpty() { + table.As = name.Name + } + return nil +} + +func storedSelectFunc( + ctx queryContext, node *sqlparser.Select, currentProvider string, +) (func() internaldto.ExecutorOutput, bool) { + prepared, stored, err := prepareStoredRelations(ctx, node) + // Preserve established view expansion for queries without a direct provider + // or stored-table read. Those paths also resolve parameterized views and unions. + if !containsProviderRead(node, currentProvider, nil) && len(stored) == 0 { + return nil, false + } + if err != nil { + return refuse(err), true + } + if !containsProviderRead(prepared, currentProvider, stored) && !hasExternalRelations(stored) { + return nil, false + } + isDoc, err := fromDocProviders(prepared.From, currentProvider) + if !isDoc { + return nil, false + } + if err != nil { + return refuse(err), true + } + return docSelectFunc(ctx, prepared, currentProvider, stored) +} diff --git a/internal/stackql/intrinsic/transaction.go b/internal/stackql/intrinsic/transaction.go new file mode 100644 index 00000000..e331853d --- /dev/null +++ b/internal/stackql/intrinsic/transaction.go @@ -0,0 +1,219 @@ +package intrinsic + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "strings" + "sync" + + "github.com/stackql-labs/omnisdk/pkg/omnisdk" + "github.com/stackql/stackql-parser/go/vt/sqlparser" + "github.com/stackql/stackql/internal/stackql/acid/binlog" + "github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto" +) + +type completionChain struct{ first, second omnisdk.MutationCompletion } + +func (c completionChain) Complete( + ctx context.Context, address, verb string, inputs map[string]any, row omnisdk.Row, +) (omnisdk.Row, error) { + var err error + if c.first != nil { + row, err = c.first.Complete(ctx, address, verb, inputs, row) + if err != nil { + return nil, err + } + } + return c.second.Complete(ctx, address, verb, inputs, row) +} + +type observedQueryContext struct { + queryContext + observer *inverseCapture +} + +func (c observedQueryContext) mutationObserver() omnisdk.MutationCompletion { return c.observer } + +type inverseCapture struct { + mu sync.Mutex + rows []map[string]string + declaration omnisdk.DeclaredInverse +} + +func (c *inverseCapture) Complete( + _ context.Context, _ string, _ string, inputs map[string]any, row omnisdk.Row, +) (omnisdk.Row, error) { + params := map[string]string{} + for name, value := range inputs { + params[name] = fmt.Sprint(value) + } + for name, token := range c.declaration.Tokens() { + if token.Location != "body" || token.Algorithm != "toString" { + return nil, fmt.Errorf("unsupported inverse token %s: %s/%s", name, token.Location, token.Algorithm) + } + value, ok := row[token.Key] + if !ok || value == nil { + return nil, fmt.Errorf("inverse token %s missing response field %s", name, token.Key) + } + params[name] = fmt.Sprint(value) + } + c.mu.Lock() + c.rows = append(c.rows, params) + c.mu.Unlock() + return row, nil +} + +// GenerateTransactionStreamFunc attaches only document-declared reversals to the +// host transaction graph. Accepted results supply their inverse identity. +// +//nolint:gocognit // Planning handles optional inverse metadata before constructing independent executors. +func GenerateTransactionStreamFunc( + ctx queryContext, stmt sqlparser.Statement, +) (func() internaldto.ExecutorOutput, func() internaldto.ExecutorOutput, bool) { + if _, ok := stmt.(*sqlparser.Insert); !ok { + f, handled := GenerateStreamFunc(ctx, stmt) + return f, nil, handled + } + dialect, err := backendDialect(ctx) + if err != nil { + f, handled := GenerateStreamFunc(ctx, stmt) + return f, nil, handled + } + translated, err := newDocTranslator(ctx.GetCurrentProvider(), dialect).mutation(stmt) + if err != nil { + f, handled := GenerateStreamFunc(ctx, stmt) + return f, nil, handled + } + target := translated.getQuery().Target() + if target == nil { + f, handled := GenerateStreamFunc(ctx, stmt) + return f, nil, handled + } + address := target.Resource().Handle() + declaration, err := omnisdk.DocInverse(registryRoot(ctx), address, "insert") + if err != nil || declaration == nil { + f, handled := GenerateStreamFunc(ctx, stmt) + return f, nil, handled + } + capture := &inverseCapture{declaration: declaration} + forward, handled := GenerateStreamFunc(observedQueryContext{queryContext: ctx, observer: capture}, stmt) + if !handled { + return nil, nil, false + } + run := func() internaldto.ExecutorOutput { + result := forward() + if result.GetError() == nil { + message := "Undo the insert on " + strings.TrimPrefix(address, UnstablePrefix) + result.SetUndoLog(binlog.NewSimpleLogEntry(nil, []string{message})) + } + return result + } + inverse := func() internaldto.ExecutorOutput { return executeDeclaredInverse(ctx, address, declaration, capture) } + return run, inverse, true +} + +func configureMutationObserver(ctx queryContext, mutation bool, args *omnisdk.Args) { + if observed, ok := ctx.(interface { + mutationObserver() omnisdk.MutationCompletion + }); ok && mutation { + args.MutationCompletion = completionChain{first: args.MutationCompletion, second: observed.mutationObserver()} + } +} +func (c observedQueryContext) GetOutErrFile() io.Writer { + if host, ok := c.queryContext.(interface{ GetOutErrFile() io.Writer }); ok { + return host.GetOutErrFile() + } + return io.Discard +} + +func executeDeclaredInverse( + ctx queryContext, address string, declaration omnisdk.DeclaredInverse, capture *inverseCapture, +) internaldto.ExecutorOutput { + capture.mu.Lock() + rows := append([]map[string]string(nil), capture.rows...) + capture.mu.Unlock() + for i := len(rows) - 1; i >= 0; i-- { + params, err := inverseParameters(ctx, address, declaration.Method(), rows[i]) + if err != nil { + return internaldto.NewErroneousExecutorOutput(err) + } + node, e := omnisdk.NewInverseMutationNode("inverse", address, "delete", declaration.Method(), params) + if e != nil { + return internaldto.NewErroneousExecutorOutput(e) + } + graph, e := omnisdk.NewGraph([]omnisdk.Node{node}, nil) + if e != nil { + return internaldto.NewErroneousExecutorOutput(e) + } + args, e := previewArgs(ctx, address, params) + if e != nil { + return internaldto.NewErroneousExecutorOutput(e) + } + p, e := omnisdk.NewGraphQuery(registryRoot(ctx), graph, args) + if e != nil { + return internaldto.NewErroneousExecutorOutput(e) + } + cursor, e := p.Open(context.Background()) + if e != nil { + return internaldto.NewErroneousExecutorOutput(e) + } + output := drainMutation(cursor) + if e = output.GetError(); e != nil { + return internaldto.NewErroneousExecutorOutput(mutationHTTPFailure(ctx, "undo", e)) + } + } + return internaldto.NewExecutorOutput(nil, nil, nil, nil, nil) +} + +func inverseParameters( + ctx queryContext, address, method string, original map[string]string, +) (map[string]string, error) { + table, err := omnisdk.DescribeMutation(registryRoot(ctx), address, "delete") + if err != nil { + return nil, err + } + for _, signature := range table.Methods() { + if signature.Method() != method { + continue + } + params := map[string]string{} + for _, p := range signature.Params() { + if value, ok := original[p.Name()]; ok { + params[p.Name()] = value + } + } + return params, nil + } + return nil, fmt.Errorf("declared inverse method %s not found", method) +} + +type mutationPresentationError struct { + cause error + message string +} + +func (e mutationPresentationError) Error() string { return e.message } +func (e mutationPresentationError) Unwrap() error { return e.cause } +func mutationHTTPFailure(ctx queryContext, verb string, err error) error { + var response omnisdk.HTTPStatusError + if !errors.As(err, &response) { + return err + } + if output, ok := ctx.(interface{ GetOutErrFile() io.Writer }); ok { + diagnostic := fmt.Sprintf( + "http response status code: %d, response body: %s", response.StatusCode(), response.ResponseBody(), + ) + if response.ResponseBody() == "" { + diagnostic = fmt.Sprintf("http response status code: %d, response body is nil", response.StatusCode()) + } + //nolint:errcheck // Diagnostics do not determine the outcome. + fmt.Fprintln(output.GetOutErrFile(), diagnostic) + } + message := fmt.Sprintf( + "%s over HTTP error: %d %s", verb, response.StatusCode(), strings.ToUpper(http.StatusText(response.StatusCode())), + ) + return mutationPresentationError{cause: err, message: message} +} diff --git a/internal/stackql/intrinsic/transaction_test.go b/internal/stackql/intrinsic/transaction_test.go new file mode 100644 index 00000000..66db342b --- /dev/null +++ b/internal/stackql/intrinsic/transaction_test.go @@ -0,0 +1,55 @@ +package intrinsic //nolint:testpackage // Exercises private reversal binding and error contracts. + +import ( + "context" + "errors" + "testing" + + "github.com/stackql-labs/omnisdk/pkg/omnisdk" +) + +type inverseFixture struct{} + +func (inverseFixture) Method() string { return "delete" } +func (inverseFixture) Tokens() map[string]omnisdk.InverseToken { + return map[string]omnisdk.InverseToken{ + "userKey": {Key: "id", Location: "body", Algorithm: "toString"}, + } +} +func TestInverseRequiresAcceptedResponseIdentity(t *testing.T) { + c := &inverseCapture{declaration: inverseFixture{}} + if _, err := c.Complete(context.Background(), "p.s.users", "insert", nil, omnisdk.Row{}); err == nil { + t.Fatal("missing identity accepted") + } + if len(c.rows) != 0 { + t.Fatal("registered reversal without identity") + } + row := omnisdk.Row{"id": "created-user"} + if _, err := c.Complete(context.Background(), "p.s.users", "insert", map[string]any{"region": "east"}, row); err != nil { + t.Fatal(err) + } + if len(c.rows) != 1 || c.rows[0]["userKey"] != "created-user" || c.rows[0]["region"] != "east" { + t.Fatalf("bindings=%v", c.rows) + } +} + +type failedMutationError struct{} + +func (failedMutationError) Error() string { return "upstream failed" } +func (failedMutationError) StatusCode() int { return 500 } +func (failedMutationError) ResponseBody() string { return "server error" } +func (failedMutationError) RequestURL() string { return "https://example.test" } +func TestMutationPresentationPreservesUnknownOutcome(t *testing.T) { + cause := errors.Join(omnisdk.ErrOutcomeUnknown, failedMutationError{}) + err := mutationHTTPFailure(nil, "insert", cause) + if err.Error() != "insert over HTTP error: 500 INTERNAL SERVER ERROR" { + t.Fatal(err) + } + if !errors.Is(err, omnisdk.ErrOutcomeUnknown) { + t.Fatal("lost unknown outcome") + } + var status omnisdk.HTTPStatusError + if !errors.As(err, &status) || status.StatusCode() != 500 { + t.Fatal("lost HTTP status") + } +} diff --git a/internal/stackql/intrinsic/translate.go b/internal/stackql/intrinsic/translate.go index d5cbf3e3..6cc1f173 100644 --- a/internal/stackql/intrinsic/translate.go +++ b/internal/stackql/intrinsic/translate.go @@ -94,6 +94,9 @@ func fromDocProviders(from sqlparser.TableExprs, currentProvider string) (bool, return } others = append(others, qualifiedName(tableName)) + case *sqlparser.TableValuedFuncTableExpr: + // Function signatures and lateral inputs are resolved by Omni. + return case *sqlparser.JoinTableExpr: walk(node.LeftExpr) walk(node.RightExpr) @@ -161,14 +164,78 @@ func (d docTranslation) translateSource(node *sqlparser.Select) (*translator, [] return nil, nil, err } } + deferred := t.deferForwardInnerConditions() if node.Where == nil { - return t, nil, nil + return t, deferred, nil } where, err := t.conjuncts(node.Where.Expr) if err != nil { return nil, nil, err } - return t, where, nil + return t, append(deferred, where...), nil +} + +// SQLite accepts INNER JOIN conditions referring to a later comma relation. +// Move only those predicates to WHERE; outer-join conditions retain their boundary. +func (t *translator) deferForwardInnerConditions() []query.Predicate { + scope := map[string]bool{} + var deferred []query.Predicate + for i, join := range t.joins { + scope[join.Resource().Alias()] = true + if join.Form() != query.Inner { + continue + } + var retained []query.Predicate + for _, predicate := range join.On() { + if joinPredicateInScope(predicate, scope) { + retained = append(retained, predicate) + } else { + deferred = append(deferred, predicate) + } + } + t.joins[i] = query.NewJoin(join.Resource(), join.Form(), retained...) + } + return deferred +} + +func joinPredicateInScope(predicate query.Predicate, scope map[string]bool) bool { + switch p := predicate.(type) { + case query.Compare: + return joinExpressionInScope(p.Left(), scope) && joinExpressionInScope(p.Right(), scope) + case query.In: + return joinExpressionInScope(p.Expr(), scope) && joinExpressionInScope(p.Set(), scope) + case query.Test: + return joinExpressionInScope(p.Cond(), scope) + case query.Or: + for _, child := range p.Any() { + if !joinPredicateInScope(child, scope) { + return false + } + } + case query.Not: + return joinPredicateInScope(p.Negated(), scope) + } + return true +} + +func joinExpressionInScope(expression query.Expr, scope map[string]bool) bool { + switch e := expression.(type) { + case query.Column: + return e.Qualifier() == "" || scope[e.Qualifier()] + case query.Call: + for _, child := range e.Args() { + if !joinExpressionInScope(child, scope) { + return false + } + } + case query.Collection: + for _, child := range e.Items() { + if !joinExpressionInScope(child, scope) { + return false + } + } + } + return true } // unsupportedDocClauses names what omnisdk returns unapplied: its row stream is @@ -269,6 +336,8 @@ type translator struct { // the join's form and ON; everything on its left keeps its own. func (t *translator) from(expr sqlparser.TableExpr, form query.JoinForm, on []query.Predicate) error { switch node := expr.(type) { + case *sqlparser.TableValuedFuncTableExpr: + return t.tableFunction(node, form, on) case *sqlparser.AliasedTableExpr: tableName, isName := node.Expr.(sqlparser.TableName) if !isName { @@ -313,6 +382,25 @@ func (t *translator) from(expr sqlparser.TableExpr, form query.JoinForm, on []qu } } +func (t *translator) tableFunction( + node *sqlparser.TableValuedFuncTableExpr, form query.JoinForm, on []query.Predicate, +) error { + call, err := expression(node.FuncExpr) + if err != nil { + return err + } + function, ok := call.(query.Call) + if !ok { + return fmt.Errorf("table function requires a function call") + } + alias := node.As.GetRawVal() + if alias == "" { + alias = function.Func() + } + t.joins = append(t.joins, query.NewJoin(query.NewTableFunction(alias, function), form, on...)) + return nil +} + // resource addresses a relation by its registry handle, aliased as written or // else by its own name, and records the bundle whose credential it runs under. func (t *translator) resource(tableName sqlparser.TableName, as sqlparser.TableIdent) query.Resource { @@ -523,6 +611,21 @@ func expression(expr sqlparser.Expr) (query.Expr, error) { switch node := expr.(type) { case *sqlparser.ColName: return query.NewColumn(node.Qualifier.Name.GetRawVal(), node.Name.GetRawVal()), nil + case *sqlparser.OrExpr: + // This parser formats the node as ||. In value position, retain that + // backend expression; predicate() handles logical OR separately. + return binaryExpression("||", node.Left, node.Right) + case *sqlparser.BinaryExpr: + if node.Operator != "||" && node.Operator != "+" && node.Operator != "-" && + node.Operator != "*" && node.Operator != "/" && node.Operator != "%" { + return nil, fmt.Errorf("unsupported scalar operator %s", node.Operator) + } + return binaryExpression(node.Operator, node.Left, node.Right) + case *sqlparser.ComparisonExpr: + if _, known := compareOps[node.Operator]; !known { + return nil, fmt.Errorf("unsupported scalar comparison %s", node.Operator) + } + return binaryExpression(node.Operator, node.Left, node.Right) case *sqlparser.SQLVal: return literal(node) case sqlparser.BoolVal: @@ -539,31 +642,123 @@ func expression(expr sqlparser.Expr) (query.Expr, error) { items = append(items, translated) } return query.NewCollection(items...), nil + case *sqlparser.CaseExpr: + return caseExpression(node) case *sqlparser.FuncExpr: - if node.IsAggregate() || node.Distinct || node.Over != nil { - return nil, fmt.Errorf("'%s' cannot be applied to %s relations", - sqlparser.String(node), UnstablePrefix+"*") - } - args := make([]query.Expr, 0, len(node.Exprs)) - for _, arg := range node.Exprs { - aliased, isAliased := arg.(*sqlparser.AliasedExpr) - if !isAliased { - return nil, fmt.Errorf("argument '%s' cannot be applied to %s relations", - sqlparser.String(arg), UnstablePrefix+"*") - } - translated, err := expression(aliased.Expr) - if err != nil { - return nil, err - } - args = append(args, translated) - } - return query.NewCall(node.Name.Lowered(), args...), nil + return functionExpression(node) default: return nil, fmt.Errorf("'%s' cannot be applied to %s relations", sqlparser.String(expr), UnstablePrefix+"*") } } +func conditionExpression(node sqlparser.Expr) (query.Expr, error) { + switch condition := node.(type) { + case *sqlparser.AndExpr: + return logicalExpression("and", condition.Left, condition.Right) + case *sqlparser.OrExpr: + return logicalExpression("or", condition.Left, condition.Right) + case *sqlparser.NotExpr: + test, err := conditionExpression(condition.Expr) + return query.NewCall("not", test), err + case *sqlparser.IsExpr: + value, err := expression(condition.Expr) + if err != nil { + return nil, err + } + test := query.NewCall("is_null", value) + if condition.Operator == sqlparser.IsNullStr { + return test, nil + } + if condition.Operator == sqlparser.IsNotNullStr { + return query.NewCall("not", test), nil + } + return nil, fmt.Errorf("unsupported CASE test %s", condition.Operator) + default: + return expression(node) + } +} + +func logicalExpression(operator string, left, right sqlparser.Expr) (query.Expr, error) { + lhs, err := conditionExpression(left) + if err != nil { + return nil, err + } + rhs, err := conditionExpression(right) + return query.NewCall(operator, lhs, rhs), err +} + +func caseExpression(node *sqlparser.CaseExpr) (query.Expr, error) { + args := []query.Expr{} + name := "case_when" + if node.Expr != nil { + base, err := expression(node.Expr) + if err != nil { + return nil, err + } + args = append(args, base) + name = "case_value" + } + for _, branch := range node.Whens { + condition := branch.Cond + translate := conditionExpression + if node.Expr != nil { + translate = expression + } + test, err := translate(condition) + if err != nil { + return nil, err + } + value, err := expression(branch.Val) + if err != nil { + return nil, err + } + args = append(args, test, value) + } + var otherwise query.Expr = query.NewLiteral(nil) + if node.Else != nil { + var err error + otherwise, err = expression(node.Else) + if err != nil { + return nil, err + } + } + return query.NewCall(name, append(args, otherwise)...), nil +} + +func functionExpression(node *sqlparser.FuncExpr) (query.Expr, error) { + if node.IsAggregate() || node.Distinct || node.Over != nil { + return nil, fmt.Errorf("'%s' cannot be applied to %s relations", + sqlparser.String(node), UnstablePrefix+"*") + } + args := make([]query.Expr, 0, len(node.Exprs)) + for _, arg := range node.Exprs { + aliased, isAliased := arg.(*sqlparser.AliasedExpr) + if !isAliased { + return nil, fmt.Errorf("argument '%s' cannot be applied to %s relations", + sqlparser.String(arg), UnstablePrefix+"*") + } + translated, err := expression(aliased.Expr) + if err != nil { + return nil, err + } + args = append(args, translated) + } + return query.NewCall(node.Name.Lowered(), args...), nil +} + +func binaryExpression(operator string, left, right sqlparser.Expr) (query.Expr, error) { + lhs, err := expression(left) + if err != nil { + return nil, err + } + rhs, err := expression(right) + if err != nil { + return nil, err + } + return query.NewCall(operator, lhs, rhs), nil +} + func literal(val *sqlparser.SQLVal) (query.Expr, error) { switch val.Type { //nolint:exhaustive // hex, bit and bind values have no omnisdk literal case sqlparser.StrVal: @@ -587,7 +782,7 @@ func literal(val *sqlparser.SQLVal) (query.Expr, error) { } // selectOutputs names every output: an alias, else a bare column's own name, -// else the expression as written. A star is left unnamed for omnisdk to expand. +// else the host's legacy expression label. A star is left unnamed for omnisdk to expand. func selectOutputs(exprs sqlparser.SelectExprs) ([]query.Output, []string, error) { outputs := make([]query.Output, 0, len(exprs)) names := make([]string, 0, len(exprs)) @@ -601,13 +796,9 @@ func selectOutputs(exprs sqlparser.SelectExprs) ([]query.Output, []string, error if err != nil { return nil, nil, err } - name := node.As.GetRawVal() - if name == "" { - if col, isCol := node.Expr.(*sqlparser.ColName); isCol { - name = col.Name.GetRawVal() - } else { - name = sqlparser.String(node.Expr) - } + name, err := projectionName(node) + if err != nil { + return nil, nil, err } outputs = append(outputs, query.NewOutput(name, translated)) names = append(names, name) diff --git a/internal/stackql/intrinsic/view_bindings.go b/internal/stackql/intrinsic/view_bindings.go new file mode 100644 index 00000000..1ccf3c1f --- /dev/null +++ b/internal/stackql/intrinsic/view_bindings.go @@ -0,0 +1,89 @@ +package intrinsic + +import ( + "fmt" + "strings" + + "github.com/stackql/stackql-parser/go/vt/sqlparser" +) + +// viewLiteralBindings collects only unconditional caller inputs. An OR branch +// cannot establish an inbox value, and another alias cannot bind this view. +func viewLiteralBindings(where *sqlparser.Where, alias, resource string) map[string]*sqlparser.ComparisonExpr { + bindings := map[string]*sqlparser.ComparisonExpr{} + var visit func(sqlparser.Expr) + visit = func(expr sqlparser.Expr) { + if and, ok := expr.(*sqlparser.AndExpr); ok { + visit(and.Left) + visit(and.Right) + return + } + compare, ok := expr.(*sqlparser.ComparisonExpr) + if !ok || !literalViewBinding(compare) { + return + } + col, ok := compare.Left.(*sqlparser.ColName) + if !ok { + return + } + qualifier := col.Qualifier.Name.GetRawVal() + if qualifier != "" && qualifier != alias && qualifier != resource { + return + } + bindings[col.Name.GetRawVal()] = compare + } + if where != nil { + visit(where.Expr) + } + return bindings +} + +func literalViewBinding(compare *sqlparser.ComparisonExpr) bool { + switch compare.Operator { + case sqlparser.EqualStr: + _, literal := compare.Right.(*sqlparser.SQLVal) + return literal + case sqlparser.InStr: + tuple, ok := compare.Right.(sqlparser.ValTuple) + if !ok || len(tuple) == 0 { + return false + } + for _, value := range tuple { + if _, literal := value.(*sqlparser.SQLVal); !literal { + return false + } + } + return true + } + return false +} + +// matchViewPredicate selects declared variants from explicit inputs. It neither +// interprets provider code nor guesses omitted scope values. +func matchViewPredicate(predicate, dialect string, bindings map[string]*sqlparser.ComparisonExpr) (bool, error) { + condition := strings.Join(strings.Fields(predicate), "") + if condition == "" || condition == "true" { + return true, nil + } + matches := true + for _, term := range strings.Split(condition, "&&") { + switch { + case term == "false": + matches = false + case term == "true": + case strings.HasPrefix(term, "sqlDialect=="): + matches = matches && strings.Trim(strings.TrimPrefix(term, "sqlDialect=="), "\"'") == dialect + case strings.HasPrefix(term, "requiredParams==[") && strings.HasSuffix(term, "]"): + names := strings.TrimSuffix(strings.TrimPrefix(term, "requiredParams==["), "]") + if names == "" { + continue + } + for _, name := range strings.Split(names, ",") { + matches = matches && bindings[strings.Trim(name, "\"'")] != nil + } + default: + return false, fmt.Errorf("unsupported view predicate %q", predicate) + } + } + return matches, nil +} diff --git a/internal/stackql/planbuilder/entrypoint.go b/internal/stackql/planbuilder/entrypoint.go index 5dbb9a01..2436b9d3 100644 --- a/internal/stackql/planbuilder/entrypoint.go +++ b/internal/stackql/planbuilder/entrypoint.go @@ -107,7 +107,7 @@ func (pb *standardPlanBuilder) BuildPlanFromContext(handlerCtx handler.HandlerCo // An omnisdk data relation streams its rows straight to the output writer, // so it is not backed by a view and analysis would try, and fail, to resolve // it as a registry-backed relation. Plan it here, ahead of that analysis. - if executor, isStream := intrinsic.GenerateStreamFunc(handlerCtx, statement); isStream { + if executor, inverse, isStream := intrinsic.GenerateTransactionStreamFunc(handlerCtx, statement); isStream { stmtType := sqlparser.ASTToStatementType(statement) qPlan.SetType(stmtType) qPlan.SetReadOnly(stmtType == sqlparser.StmtSelect) @@ -120,6 +120,11 @@ func (pb *standardPlanBuilder) BuildPlanFromContext(handlerCtx handler.HandlerCo }, ), ) + if inverse != nil { + pGBuilder.getPlanGraphHolder().CreateInversePrimitiveNode(primitive.NewLocalPrimitive( + func(_ primitive.IPrimitiveCtx) internaldto.ExecutorOutput { return inverse() }, + )) + } qPlan.SetInstructions(pGBuilder.getPlanGraphHolder()) if optimiseErr := qPlan.GetInstructions().GetPrimitiveGraph().Optimise(); optimiseErr != nil { return createErroneousPlan(handlerCtx, qPlan, rowSort, optimiseErr) diff --git a/internal/stackql/responsehandler/program_error_test.go b/internal/stackql/responsehandler/program_error_test.go new file mode 100644 index 00000000..57d4a32c --- /dev/null +++ b/internal/stackql/responsehandler/program_error_test.go @@ -0,0 +1,17 @@ +package responsehandler //nolint:testpackage // Tests the CLI presentation boundary. + +import ( + "fmt" + "testing" +) + +type unsupportedTestProgramError struct{} + +func (unsupportedTestProgramError) Error() string { return "unsupported template type: unknown_v1" } +func (unsupportedTestProgramError) ProgramType() string { return "unknown_v1" } +func TestUnsupportedProgramPresentationKeepsLegacyDiagnostic(t *testing.T) { + err := fmt.Errorf("resource: %w", unsupportedTestProgramError{}) + if got := presentationError(err).Error(); got != "unsupported template type: unknown_v1" { + t.Fatalf("diagnostic=%q", got) + } +} diff --git a/internal/stackql/responsehandler/responsehandler.go b/internal/stackql/responsehandler/responsehandler.go index 7d835429..80ced3ca 100644 --- a/internal/stackql/responsehandler/responsehandler.go +++ b/internal/stackql/responsehandler/responsehandler.go @@ -85,6 +85,10 @@ func HandleResponse(handlerCtx handler.HandlerContext, response internaldto.Exec // Keep the established CLI diagnostic while retaining typed SDK errors for // programmatic consumers and transport/effect classification. func presentationError(err error) error { + var program omnisdk.UnsupportedProgramError + if errors.As(err, &program) { + return fmt.Errorf("unsupported template type: %s", program.ProgramType()) + } var response omnisdk.HTTPStatusError if !errors.As(err, &response) { return err diff --git a/test/robot/functional/stackql_mocked_from_cmd_line.robot b/test/robot/functional/stackql_mocked_from_cmd_line.robot index 8075da92..4c9ef636 100644 --- a/test/robot/functional/stackql_mocked_from_cmd_line.robot +++ b/test/robot/functional/stackql_mocked_from_cmd_line.robot @@ -12290,3 +12290,63 @@ Omni Nested Derived Relation Preserves Aggregate Boundary ... n\n2 ... \-o\=csv ... --preview\={"unstable":true,"staging":true,"omni":"all"} + +Omni Stored View Joins Provider Through Read Handles + ${query} = Catenate SEPARATOR=${SPACE} + ... create or replace view omni_members as select login,contributions from github.repos.contributors where owner='dummyorg' and repo='dummyapp.io'; + ... select count(*) as n from (select login from omni_members order by contributions desc limit 2) v + ... join github.repos.contributors c on v.login=c.login where c.owner='dummyorg' and c.repo='dummyapp.io'; + Should StackQL Exec Inline Equal + ... ${STACKQL_EXE} + ... ${OKTA_SECRET_STR} + ... ${GITHUB_SECRET_STR} + ... ${K8S_SECRET_STR} + ... ${REGISTRY_NO_VERIFY_CFG_STR} + ... ${AUTH_CFG_STR} + ... ${SQL_BACKEND_CFG_STR_CANONICAL} + ... ${query} + ... n\n2 + ... \-o\=csv + ... --preview\={"unstable":true,"staging":true,"omni":"all"} + + +Omni CTEs Preserve Ordered Boundaries And Column Names + ${inputStr} = Catenate + ... with picked as (select login,contributions from github.repos.contributors + ... where owner='dummyorg' and repo='dummyapp.io' order by contributions desc limit 2), + ... tallied(member_count) as (select count(*) from picked) + ... select member_count from tallied; + ${expectedStr} = Catenate SEPARATOR=\n + ... member_count + ... 2 + Should StackQL Exec Inline Equal + ... ${STACKQL_EXE} + ... ${OKTA_SECRET_STR} + ... ${GITHUB_SECRET_STR} + ... ${K8S_SECRET_STR} + ... ${REGISTRY_NO_VERIFY_CFG_STR} + ... ${AUTH_CFG_STR} + ... ${SQL_BACKEND_CFG_STR_CANONICAL} + ... ${inputStr} + ... ${expectedStr} + ... \-o\=csv + ... --preview\={"unstable":true,"staging":true,"omni":"all"} + + +Omni CASE Join Skips Unselected Error Branches + ${query} = Catenate + ... select count(*) as n from (select login from github.repos.contributors + ... where owner='dummyorg' and repo='dummyapp.io' order by login limit 1) c + ... join (select 1 as join_key) one on one.join_key=case when c.login is not null then 1 else json('invalid-json') end; + Should StackQL Exec Inline Equal + ... ${STACKQL_EXE} + ... ${OKTA_SECRET_STR} + ... ${GITHUB_SECRET_STR} + ... ${K8S_SECRET_STR} + ... ${REGISTRY_NO_VERIFY_CFG_STR} + ... ${AUTH_CFG_STR} + ... ${SQL_BACKEND_CFG_STR_CANONICAL} + ... ${query} + ... n\n1 + ... \-o\=csv + ... --preview\={"unstable":true,"staging":true,"omni":"all"}