general-kroll-4-life · Oct 14, 2026 · Oct 14, 2026
diff --git a/go.mod b/go.mod
@@ -21,7 +21,7 @@ require (
 	github.com/spf13/pflag v1.0.10
 	github.com/spf13/viper v1.10.1
 	github.com/stackql-labs/omnisdk v0.1.4-alpha05
-	github.com/stackql/any-sdk v0.6.1-alpha02
+	github.com/stackql/any-sdk v0.6.1-alpha03
 	github.com/stackql/go-suffix-map v0.0.1-alpha01
 	github.com/stackql/psql-wire v0.1.3-beta01
 	github.com/stackql/stackql-parser v0.0.16-alpha02

diff --git a/go.sum b/go.sum
@@ -370,8 +370,8 @@ github.com/spiffe/go-spiffe/v2 v2.7.0 h1:uXe1MflJoHw58wAUvxVlcM7WpKtijWG7I1UidcG
 github.com/spiffe/go-spiffe/v2 v2.7.0/go.mod h1:47Q0Q9/AqGha8QLHp+kxpH4Wca7X7EnOtlIJy3mxZ3U=
 github.com/stackql-labs/omnisdk v0.1.4-alpha05 h1:Z22XGi99mjKPJEAAYv3qkW1J0iqqFRd9OANK+9cw4Xs=
 github.com/stackql-labs/omnisdk v0.1.4-alpha05/go.mod h1:2P2vFi+c/wEeQ3kFgIt0QuIlefpuzIoTm6bPJAiLAvQ=
-github.com/stackql/any-sdk v0.6.1-alpha02 h1:kXnL1SOORkwh47foKVX4GGDmD0gdOx2K4JGgvz+sxcw=
-github.com/stackql/any-sdk v0.6.1-alpha02/go.mod h1:ahkRgwHHEn7RTmSfKrshPVk6i4X1BdHknSeBflSyRac=
+github.com/stackql/any-sdk v0.6.1-alpha03 h1:8WPgE+MftTnLNsR/WbDkyujnuyaQbM1PvQTgEsuNbR4=
+github.com/stackql/any-sdk v0.6.1-alpha03/go.mod h1:ahkRgwHHEn7RTmSfKrshPVk6i4X1BdHknSeBflSyRac=
 github.com/stackql/go-suffix-map v0.0.1-alpha01 h1:TDUDS8bySu41Oo9p0eniUeCm43mnRM6zFEd6j6VUaz8=
 github.com/stackql/go-suffix-map v0.0.1-alpha01/go.mod h1:QAi+SKukOyf4dBtWy8UMy+hsXXV+yyEE4vmBkji2V7g=
 github.com/stackql/psql-wire v0.1.3-beta01 h1:xDpR3e09xt74GBSkxYKIhfTMPnKY/Utgu2rl1ctEuF8=

diff --git a/internal/stackql/cmd/root.go b/internal/stackql/cmd/root.go
@@ -181,6 +181,7 @@ func init() {
 	rootCmd.PersistentFlags().StringVar(&runtimeCtx.NamespaceCfgRaw, dto.NamespaceCfgRawKey, "{}", "JSON / YAML string representing namespaces for cacheing, views etc")
 	rootCmd.PersistentFlags().StringVar(&runtimeCtx.StoreTxnCfgRaw, dto.StoreTxnCfgRawKey, "{}", "JSON / YAML string representing Txn store config")
 	rootCmd.PersistentFlags().StringVar(&runtimeCtx.GCCfgRaw, dto.GCCfgRawKey, "{}", "JSON / YAML string representing GC config")
+	rootCmd.PersistentFlags().StringVar(&runtimeCtx.UpstreamErrorCfgRaw, dto.UpstreamErrorCfgRawKey, "{}", `JSON / YAML string representing upstream errors that reads ignore, eg {"http": {"ignoreStatusCodes": ["5xx"]}}`)
 	rootCmd.PersistentFlags().StringVar(&runtimeCtx.ACIDCfgRaw, dto.ACIDCfgRawKey, "{}", "JSON / YAML string representing ACID config")
 	rootCmd.PersistentFlags().StringVar(&previewCfgRaw, intrinsic.CfgRawKey, "{}", "JSON string configuring the "+intrinsic.ProviderName+
 		" provider backend; keys: batchSize, flushInterval, endpoint, unstable")

diff --git a/internal/stackql/dependencyplanner/dependencyplanner.go b/internal/stackql/dependencyplanner/dependencyplanner.go
@@ -28,6 +28,7 @@ import (
 	"github.com/stackql/stackql/internal/stackql/tableinsertioncontainer"
 	"github.com/stackql/stackql/internal/stackql/taxonomy"
 	"github.com/stackql/stackql/internal/stackql/util"
+	"github.com/stackql/stackql/pkg/astformat"
 )
 
 type DependencyPlanner interface {
@@ -811,58 +812,76 @@ func (dp *standardDependencyPlanner) isVectorParam(param interface{}) bool {
 	return false
 }
 
-//nolint:gocognit,nestif // live with it
-func (dp *standardDependencyPlanner) getStreamFromEdge(
+//nolint:nestif // live with it
+func (dp *standardDependencyPlanner) getSQLStreamFromEdge(
 	e dataflow.Edge,
 	toAc taxonomy.AnnotationCtx,
 	tcc internaldto.TxnControlCounters,
 ) (streaming.MapStream, error) {
-	if e.IsSQL() {
-		selectCtx, err := dp.generateSelectDML(e, tcc)
-		if err != nil {
-			return nil, err
-		}
-		ann := e.GetSource().GetAnnotation()
-		meta := ann.GetTableMeta()
-		insertContainer, err := tableinsertioncontainer.NewTableInsertionContainer(meta,
-			dp.handlerCtx.GetSQLEngine(),
-			dp.handlerCtx.GetTxnCounterMgr())
-		if err != nil {
-			return nil, err
-		}
-		toParams := toAc.GetParameters()
+	columnDescriptors, err := e.GetColumnDescriptors()
+	if err != nil {
+		return nil, err
+	}
+	selectCtx, err := dp.generateSelectDML(e.GetSource().GetAnnotation(), columnDescriptors, tcc, nil)
+	if err != nil {
+		return nil, err
+	}
+	ann := e.GetSource().GetAnnotation()
+	meta := ann.GetTableMeta()
+	insertContainer, err := tableinsertioncontainer.NewTableInsertionContainer(meta,
+		dp.handlerCtx.GetSQLEngine(),
+		dp.handlerCtx.GetTxnCounterMgr())
+	if err != nil {
+		return nil, err
+	}
+	toParams := toAc.GetParameters()
 
-		transformedToStaticParams, paramTransformErr := util.TransformSQLRawParameters(toParams, false)
-		if paramTransformErr != nil {
-			return nil, paramTransformErr
-		}
-		for k, v := range transformedToStaticParams {
-			hasSourceAttr := e.HasSourceAttribute(k)
-			if !hasSourceAttr {
-				vIsUnpopulated := k == v
-				if vIsUnpopulated {
-					delete(transformedToStaticParams, k)
-				}
+	transformedToStaticParams, paramTransformErr := util.TransformSQLRawParameters(toParams, false)
+	if paramTransformErr != nil {
+		return nil, paramTransformErr
+	}
+	for k, v := range transformedToStaticParams {
+		hasSourceAttr := e.HasSourceAttribute(k)
+		if !hasSourceAttr {
+			vIsUnpopulated := k == v
+			if vIsUnpopulated {
+				delete(transformedToStaticParams, k)
 			}
 		}
-		return sqlstream.NewSimpleSQLMapStream(
-			selectCtx,
-			insertContainer,
-			dp.handlerCtx.GetDrmConfig(),
-			dp.handlerCtx.GetSQLEngine(),
-			transformedToStaticParams,
-		), nil
+	}
+	return sqlstream.NewSimpleSQLMapStream(
+		selectCtx,
+		insertContainer,
+		dp.handlerCtx.GetDrmConfig(),
+		dp.handlerCtx.GetSQLEngine(),
+		transformedToStaticParams,
+	), nil
+}
+
+//nolint:gocognit // live with it
+func (dp *standardDependencyPlanner) getStreamFromEdge(
+	e dataflow.Edge,
+	toAc taxonomy.AnnotationCtx,
+	tcc internaldto.TxnControlCounters,
+) (streaming.MapStream, error) {
+	if e.IsSQL() {
+		return dp.getSQLStreamFromEdge(e, toAc, tcc)
 	}
 	projection, err := e.GetProjection()
 	if err != nil {
 		return nil, err
 	}
-	projection = sourceResponseProjection(projection, e.GetSource().GetAnnotation().GetSchema().Tabulate(false, ""))
 	method, err := toAc.GetTableMeta().GetMethod()
 	if err != nil {
 		return nil, err
 	}
 	requestParameters := method.GetParametersIncludingNativeCasing()
+	sourceFilters, err := dp.sourceOnlyFilters(e.GetSource().GetAnnotation())
+	if err != nil {
+		return nil, err
+	}
+	sqlProjection := destinationRequestProjection(projection, requestParameters)
+	projection = sourceResponseProjection(projection, e.GetSource().GetAnnotation().GetSchema().Tabulate(false, ""))
 	projection = destinationRequestProjection(projection, requestParameters)
 	incomingCols := make(map[string]struct{})
 	for _, v := range projection {
@@ -888,9 +907,141 @@ func (dp *standardDependencyPlanner) getStreamFromEdge(
 			return nil, err
 		}
 	}
+	if len(sourceFilters) > 0 {
+		return dp.filteredSourceStream(e.GetSource().GetAnnotation(), sqlProjection, staticParams, sourceFilters, tcc)
+	}
 	return streaming.NewSimpleProjectionMapStream(projection, staticParams), nil
 }
 
+// filteredSourceStream reads the dependent table's request parameters from
+// the source table's stored rows, restricted by the source only filters, so
+// that no request is made for a source row that the WHERE clause discards.
+func (dp *standardDependencyPlanner) filteredSourceStream(
+	sourceAnnotation taxonomy.AnnotationCtx,
+	projection map[string]string,
+	staticParams map[string]interface{},
+	filters []string,
+	tcc internaldto.TxnControlCounters,
+) (streaming.MapStream, error) {
+	alias := sourceAnnotation.GetTableMeta().GetAlias()
+	var columnDescriptors []formulation.ColumnDescriptor
+	for source, destination := range projection {
+		sourceColumn := &sqlparser.ColName{
+			Name:      sqlparser.NewColIdent(source),
+			Qualifier: sqlparser.TableName{Name: sqlparser.NewTableIdent(alias)},
+		}
+		decoratedColumn := fmt.Sprintf(`%s AS "%s"`, sqlparser.ColDelimitedString(sourceColumn), destination)
+		columnDescriptors = append(
+			columnDescriptors,
+			formulation.NewColumnDescriptor(destination, "", "", decoratedColumn, nil, nil, nil),
+		)
+	}
+	selectCtx, err := dp.generateSelectDML(sourceAnnotation, columnDescriptors, tcc, filters)
+	if err != nil {
+		return nil, err
+	}
+	insertContainer, err := tableinsertioncontainer.NewTableInsertionContainer(
+		sourceAnnotation.GetTableMeta(),
+		dp.handlerCtx.GetSQLEngine(),
+		dp.handlerCtx.GetTxnCounterMgr())
+	if err != nil {
+		return nil, err
+	}
+	return sqlstream.NewSimpleSQLMapStream(
+		selectCtx,
+		insertContainer,
+		dp.handlerCtx.GetDrmConfig(),
+		dp.handlerCtx.GetSQLEngine(),
+		staticParams,
+	), nil
+}
+
+// sourceOnlyFilters returns the top level WHERE conjuncts that refer only to
+// response columns of the source table, rendered for its stored rows.
+// It returns none when the source table may be null extended by an outer
+// join, since a WHERE clause may then retain rows the source table does not
+// match.  Conjuncts on request parameters are already enforced by the
+// source request itself, and unqualified columns are not attributed.
+func (dp *standardDependencyPlanner) sourceOnlyFilters(sourceAnnotation taxonomy.AnnotationCtx) ([]string, error) {
+	where := dp.sqlStatement.Where
+	if where == nil || where.Type != sqlparser.WhereStr || isAnnotationIndirection(sourceAnnotation) {
+		return nil, nil
+	}
+	alias := sourceAnnotation.GetTableMeta().GetAlias()
+	if alias == "" || isNullExtended(dp.sqlStatement.From, alias, false) {
+		return nil, nil
+	}
+	method, err := sourceAnnotation.GetTableMeta().GetMethod()
+	if err != nil {
+		return nil, err
+	}
+	requestParameters := make(map[string]struct{})
+	for name, parameter := range method.GetParametersIncludingNativeCasing() {
+		requestParameters[name] = struct{}{}
+		requestParameters[parameter.GetName()] = struct{}{}
+	}
+	for name := range sourceAnnotation.GetParameters() {
+		requestParameters[name] = struct{}{}
+	}
+	var filters []string
+	for _, conjunct := range sqlparser.SplitAndExpression(nil, where.Expr) {
+		if isSourceOnlyResponseExpr(conjunct, alias, requestParameters) {
+			filters = append(filters, astformat.String(conjunct, dp.handlerCtx.GetASTFormatter()))
+		}
+	}
+	return filters, nil
+}
+
+func isSourceOnlyResponseExpr(expr sqlparser.Expr, alias string, requestParameters map[string]struct{}) bool {
+	columnCount := 0
+	eligible := true
+	//nolint:errcheck // the visitor never returns an error
+	sqlparser.Walk(func(node sqlparser.SQLNode) (bool, error) {
+		switch node := node.(type) {
+		case *sqlparser.ColName:
+			_, isRequestParameter := requestParameters[node.Name.GetRawVal()]
+			if node.Qualifier.Name.GetRawVal() != alias || !node.Qualifier.Qualifier.IsEmpty() || isRequestParameter {
+				eligible = false
+			}
+			columnCount++
+		case *sqlparser.Subquery, *sqlparser.ExistsExpr, *sqlparser.ValuesFuncExpr, *sqlparser.Default:
+			eligible = false
+		}
+		return eligible, nil
+	}, expr)
+	return eligible && columnCount > 0
+}
+
+// isNullExtended reports whether the table aliased by alias sits on the
+// optional side of an outer join within the table expressions.
+func isNullExtended(tableExprs sqlparser.TableExprs, alias string, isOptionalSide bool) bool {
+	for _, tableExpr := range tableExprs {
+		switch tableExpr := tableExpr.(type) {
+		case *sqlparser.AliasedTableExpr:
+			if tableExpr.As.GetRawVal() == alias {
+				return isOptionalSide
+			}
+		case *sqlparser.ParenTableExpr:
+			if isNullExtended(tableExpr.Exprs, alias, isOptionalSide) {
+				return true
+			}
+		case *sqlparser.JoinTableExpr:
+			leftOptional, rightOptional := isOptionalSide, isOptionalSide
+			switch tableExpr.Join {
+			case sqlparser.LeftJoinStr, sqlparser.LeftOuterJoinStr, sqlparser.NaturalLeftJoinStr:
+				rightOptional = true
+			case sqlparser.RightJoinStr, sqlparser.RightOuterJoinStr, sqlparser.NaturalRightJoinStr:
+				leftOptional = true
+			}
+			if isNullExtended(sqlparser.TableExprs{tableExpr.LeftExpr}, alias, leftOptional) ||
+				isNullExtended(sqlparser.TableExprs{tableExpr.RightExpr}, alias, rightOptional) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 func sourceResponseProjection(projection map[string]string, tab formulation.Tabulation) map[string]string {
 	wireNames := make(map[string]string)
 	for _, column := range tab.GetColumns() {
@@ -940,14 +1091,17 @@ func (dp *standardDependencyPlanner) harvestFilter(sourceAnnotation taxonomy.Ann
 }
 
 func (dp *standardDependencyPlanner) generateSelectDML(
-	e dataflow.Edge,
+	ann taxonomy.AnnotationCtx,
+	columnDescriptors []formulation.ColumnDescriptor,
 	tcc internaldto.TxnControlCounters,
+	filters []string,
 ) (drm.PreparedStatementCtx, error) {
-	ann := e.GetSource().GetAnnotation()
 	whereStr, _ := dp.harvestFilter(ann)
-	columnDescriptors, err := e.GetColumnDescriptors()
-	if err != nil {
-		return nil, err
+	for _, filter := range filters {
+		if whereStr != "" {
+			whereStr += " AND "
+		}
+		whereStr += fmt.Sprintf("(%s)", filter)
 	}
 	discoID, discoIDErr := dp.handlerCtx.GetSQLEngine().GetCurrentDiscoveryGenerationID(ann.GetHIDs().GetProviderStr())
 	if discoIDErr != nil {

diff --git a/internal/stackql/dependencyplanner/dependencyplanner_test.go b/internal/stackql/dependencyplanner/dependencyplanner_test.go
@@ -7,6 +7,7 @@ import (
 
 	"github.com/stackql/any-sdk/pkg/streaming"
 	"github.com/stackql/any-sdk/public/formulation"
+	"github.com/stackql/stackql-parser/go/vt/sqlparser"
 )
 
 type projectionColumn struct {
@@ -84,3 +85,50 @@ func TestSourceResponseProjection(t *testing.T) {
 		t.Fatalf("projected rows = %v", rows)
 	}
 }
+
+func parseSelect(t *testing.T, query string) *sqlparser.Select {
+	t.Helper()
+	stmt, err := sqlparser.Parse(query)
+	if err != nil {
+		t.Fatalf("parse error: %v", err)
+	}
+	selectStmt, ok := stmt.(*sqlparser.Select)
+	if !ok {
+		t.Fatalf("not a select: %T", stmt)
+	}
+	return selectStmt
+}
+
+func TestIsSourceOnlyResponseExpr(t *testing.T) {
+	requestParameters := map[string]struct{}{"org": {}}
+	for where, expected := range map[string]bool{
+		"r.archived = 0":                 true,
+		"r.archived = 0 and r.size > 1":  true,
+		"r.org = 'stackql'":              false,
+		"archived = 0":                   false,
+		"b.type = 'x'":                   false,
+		"r.name = b.repo":                false,
+		"1 = 1":                          false,
+		"r.name in (select name from t)": false,
+	} {
+		stmt := parseSelect(t, "select 1 from t where "+where)
+		if got := isSourceOnlyResponseExpr(stmt.Where.Expr, "r", requestParameters); got != expected {
+			t.Errorf("isSourceOnlyResponseExpr(%q) = %t, want %t", where, got, expected)
+		}
+	}
+}
+
+func TestIsNullExtended(t *testing.T) {
+	for from, expected := range map[string]bool{
+		"a r inner join b on r.x = b.x":                            false,
+		"a r left outer join b on r.x = b.x":                       false,
+		"b left outer join a r on r.x = b.x":                       true,
+		"a r right join b on r.x = b.x":                            true,
+		"c inner join (b left join a r on r.x = b.x) on c.y = b.y": true,
+	} {
+		stmt := parseSelect(t, "select 1 from "+from)
+		if got := isNullExtended(stmt.From, "r", false); got != expected {
+			t.Errorf("isNullExtended(%q) = %t, want %t", from, got, expected)
+		}
+	}
+}
diff --git a/test/python/stackql_test_tooling/flask/github/app.py b/test/python/stackql_test_tooling/flask/github/app.py
@@ -174,6 +174,11 @@ def get_user(userId):
 def get_aware_collaborators(org, repositoryId):
     global repos_collab_call_count
     global repas_collab_specialcase_call_count
+    if org == 'joinfaultorg':
+        # The archived repository's collaborators fail with an empty bodied 500.
+        if repositoryId == 'archived-repo':
+            return '', 500
+        return jsonify([{"login": f"{repositoryId}-collaborator", "type": "User", "role_name": "admin"}]), 200
     if org == 'specialcaseorg':
         repas_collab_specialcase_call_count += 1
         if repas_collab_specialcase_call_count == 1:
@@ -420,6 +425,15 @@ def get_special_org_repos():
     return jsonify(response), 200
 
 
+@app.route('/orgs/joinfaultorg/repos', methods=['GET'])
+def get_join_fault_org_repos():
+    return jsonify([
+        {"id": 1, "name": "active-repo-a", "archived": False},
+        {"id": 2, "name": "archived-repo", "archived": True},
+        {"id": 3, "name": "active-repo-b", "archived": False},
+    ]), 200
+
+
 @app.route('/orgs/ratelimitedorg/repos', methods=['GET'])
 def get_rate_limited_org_repos():
     """Mimics GitHub's rate-limit refusal: a 403 with a JSON object body.