opt: enforce consistent input ordering for streaming set ops

[Alignyx]

Summary and scope

Fix the optimizer/executor ordering contract for streaming set operations when an
output functional dependency (FD) holds on only one input. In #144925, this lets
the two inputs use inconsistent sort keys: the vectorized engine returns an
incorrect intersection, while the row engine detects badly ordered input.

This is a localized correctness repair in pkg/sql/opt/ordering/set.go, with
ordering-invariant and SQL regressions. It preserves valid output FDs, existing
streaming exploration, unordered hash operations, and the special UNION ALL
ordered-synchronizer behavior. It does not implement the broader bottom-up
planner redesign proposed in the issue discussion.

Problem and expected result

The following reduced setup preserves the report's INT2 input and explicit
INT8 output casts:

SET CLUSTER SETTING sql.query_cache.enabled = false;
SET distsql = off;

CREATE TABLE t5 (
  a2 INT2 NULL,
  a4 INT8 NULL,
  INDEX i4 (a4 ASC, a2 ASC)
) WITH (sql_stats_automatic_collection_enabled = false);
INSERT INTO t5 VALUES (32, NULL), (-129, -129);

-- W: the original failing shape.
SELECT a2::INT8 AS c1, a2::INT8 AS c3
FROM t5 WHERE a2 IS NOT NULL
INTERSECT
SELECT DISTINCT a2::INT8, a4
FROM t5 WHERE a2 IS NOT NULL
ORDER BY 1;

The left relation contains (32,32) and (-129,-129); the right contains
(32,NULL) and (-129,-129). Their distinct intersection must therefore be
exactly (-129,-129). The baseline vectorized query instead returns no rows.
This expected result follows directly from the input tuples, not from assuming
that a particular execution plan is a trustworthy oracle.

The same ordering-contract failure can affect EXCEPT and ALL variants: a
merge algorithm cannot reliably compute membership or multiplicity when an input
does not satisfy the comparison order it actually uses.

Metamorphic relations used to localize the defect

The investigation kept data, predicates, and output types fixed, transformed the
query into equivalent forms, and compared both results and execution plans. Four
controls independently preserve the expected distinct intersection:

MR1: remove redundant input DISTINCT

For ordinary, duplicate-eliminating INTERSECT,
R INTERSECT DISTINCT(S) is equivalent to R INTERSECT S:

SELECT a2::INT8 AS c1, a2::INT8 AS c3
FROM t5 WHERE a2 IS NOT NULL
INTERSECT
SELECT a2::INT8, a4 FROM t5 WHERE a2 IS NOT NULL
ORDER BY 1;

Removing the right-side DISTINCT changes planning opportunities but cannot
change set membership. This relation must not be applied unconditionally to
INTERSECT ALL, where removing DISTINCT can change multiplicities.

MR2: change only the access path

In W, replace the right-side FROM t5 with FROM t5@primary, leaving its
DISTINCT, predicates, and casts unchanged. The index hint changes how the same
relation is read, not its logical contents. This control is particularly useful
for separating an ordering-dependent failure from incorrect set semantics.

MR3: express intersection as null-safe membership

SELECT DISTINCT l.a2::INT8 AS c1, l.a2::INT8 AS c3
FROM t5 AS l
WHERE l.a2 IS NOT NULL
  AND EXISTS (
    SELECT 1 FROM t5 AS r
    WHERE r.a2 IS NOT NULL
      AND l.a2::INT8 IS NOT DISTINCT FROM r.a2::INT8
      AND l.a2::INT8 IS NOT DISTINCT FROM r.a4
  )
ORDER BY 1;

Left-side DISTINCT preserves the output's set semantics. Each output component
uses IS NOT DISTINCT FROM, rather than ordinary equality, to preserve the
NULL-equality semantics of set operations. This is a structurally different
membership computation, not another set-op plan used as the sole oracle.

MR4: introduce a materialization boundary

WITH rhs AS MATERIALIZED (
  SELECT DISTINCT a2::INT8 AS c1, a4 AS c3
  FROM t5 WHERE a2 IS NOT NULL
)
SELECT a2::INT8 AS c1, a2::INT8 AS c3
FROM t5 WHERE a2 IS NOT NULL
INTERSECT
SELECT c1, c3 FROM rhs
ORDER BY 1;

For this pure query, materialization changes the optimization boundary while
preserving the relation and output types. This claim does not generalize without
qualification to volatile expressions or evaluation-order-dependent errors.

Observed results

Three complete baseline runs and three complete candidate runs agreed:

Observation	Baseline	Candidate
W, vectorized local execution	Empty result: incorrect	(-129,-129)
Each of the four equivalent controls	(-129,-129)	(-129,-129)
W with vectorize = off	Exact badly-ordered-input error	(-129,-129), no error
W with vectorize = on, distsql = always	Empty result: incorrect	(-129,-129)

Only W was repeated under the last two settings; this is not a claim that every
control was run under every engine/distribution combination. distsql = always
also does not by itself establish remote multi-node execution.

Root cause and how it was identified

The left projection emits the same expression twice, so its two output columns
are equivalent. The intersection output is a subset of that input, so the
equivalence remains a valid output FD. The right input's two columns are not
equivalent: (32,NULL) is a counterexample.

The first invalid transition is passing a parent/output ordering choice to both
inputs as though its freedoms were valid on each input. In the captured optimizer
plan, output columns (17,18) are equivalent and the required ordering becomes
+(17|18): either output column is acceptable. The left input sorts its duplicated
expression, but the right input, with output columns (16,10), satisfies the
choice using +10, its second output column, via the (a4,a2) index.

Execution does not independently choose a different merge key for each side.
StreamingSetOpOrdering produces one concrete ordering; execbuilder passes it
to ConstructStreamingSetOp, and the physical planner assigns the same
mergeOrdering to both MergeJoinerSpec.LeftOrdering and RightOrdering.
For W, that concrete comparison starts with the first output column.

Consequently, the right input can arrive as:

(32, NULL)
(-129, -129)

That is valid for the chosen second-column index order, but invalid for the
executor's first-column ascending comparison. The row engine exposes the mismatch:

detected badly ordered input: [32 NULL] > [-129 -129], but expected '<'

EXPLAIN (OPT, VERBOSE) exposed the FD/order-choice mismatch;
EXPLAIN (VERBOSE) and EXPLAIN ANALYZE connected it to the physical path:

• Baseline W uses a streaming intersection, reads two rows on each input, and
  emits zero rows.
• The equivalent primary-index control uses a hash intersection followed by a
  final sort, reads the same two rows on each input, and emits one row.
• With the candidate, W itself selects a hash intersection plus final sort and
  emits the expected row.

The equal scan cardinalities, concrete tuple counterexample, row-engine diagnostic,
and source-level ordering contract point to an invalid ordering guarantee, rather
than omitted scan data or an incorrect FD derivation rule. The candidate plan for
this particular query is not evidence of an executed repaired streaming plan;
the ordering-invariant tests and unchanged positive streaming fixtures provide
additional coverage of that path.

How the repair works

The shared construction in setOpBuildRequired now does the following:

1. Intersect the external required ordering with the set operation's private
   ordering, as before.
2. For non-UNION ALL streaming paths, execute
   result.FromOrdering(result.ToOrdering()). This selects a concrete permitted
   ordering, preserving directions while replacing equivalence groups with
   singleton choices and removing output-only optional-column freedom.
3. Append any missing output columns in the existing deterministic order, so the
   streaming comparison has a full-column ordering.
4. Map that same common order to each input. The existing
   setOpBuildChildReqOrdering may simplify it using only that child's FDs.
   An equivalence or constant on one side cannot weaken the other side's contract.
5. Use the same construction for StreamingSetOpOrdering, keeping child
   requirements aligned with the concrete order delivered to execution.

The early return for required.Any() is also narrowed. An explicitly ordered
streaming variant must pass through this construction even when its parent has
no external ordering requirement. Truly unordered paths, where both orderings
are Any, remain unchanged, as does the UNION ALL special case.

The key invariant is: each child's required ordering, interpreted using that
child's own FDs, must imply the corresponding mapping of the executor's common
merge ordering. Choosing one representative is sufficient for correctness:
it satisfies the permitted output ordering, and both children must now actually
support it. Recovering the parent's original, unsimplified ORDER BY is not
necessary for this invariant, although preserving more ordering alternatives
may matter for plan quality.

No FD derivation, type coercion, set-operation multiplicity rule, or execution
operator is changed. Existing set-op execution retains NullEquality: true.

Relationship to the discussion and previous fixes

This addresses the discussion's output-FD/child-ordering diagnosis,
including the fact that the output FDs themselves are correct. The SQL regression
also includes the equality-filter-derived, no-right-side-DISTINCT shape from the
additional reproduction;
the fix is not conditional on the original DISTINCT syntax.

Relevant history guided both the implementation boundary and regression coverage:

• c11a5582
  assigned distinct output column IDs when a left-input column is projected more
  than once, while preserving valid output equivalences. This patch does not
  revert those properties.
• 4eef17286c, fixing #69497
  stopped simplifying the common streaming order with output FDs that need not
  hold on both inputs. This repair closes the related gap where equivalence
  groups and optional columns already embedded in an ordering choice still
  transmit that unsafe freedom.
• bd6ec0ca, fixing #73084
  restricted interesting orderings before remapping to avoid invalid streaming
  alternatives. That behavior and genuinely unordered hash paths are preserved.

The latest design discussion
proposes a merge-join-style bottom-up approach using interesting orderings.
GenerateStreamingSetOp already consults bottom-up interesting orderings; this
patch retains it, but also retains implicit streaming driven by parent ordering
requirements. It restores a concrete common-key contract at the existing boundary,
not the proposed planner representation/enumeration redesign. There is a real
scope difference to review, not a claim that the architectural discussion is
fully resolved.

Completed local validation

All results below refer to the existing repair commit
1f04f4c498931d10074acd00b15f706b42798124, based on
8812064a015d2faf99d3fc7e15880f94042954b0. They are completed local validation,
not a claim that upstream PR CI has passed.

• Native ordering regression: TestSetOpOrdering144925 covers 12 scenarios
  and 38 operator variants, including one-sided equivalences/constants,
  nontrivial output/input column mapping, internal-only ordering, descending
  prefixes, hash paths, UNION, UNION ALL, and non-mutation. Beyond expected
  strings, it asserts childReq.Implies(mergeReq) after mapping the actual
  StreamingSetOpOrdering and simplifying with the relevant child's FDs.
  The unchanged tests fail on baseline production code with actual implication
  failures and pass on the candidate.
• Native SQL regression: six queries in the union logic test cover the
  original DISTINCT shape, equality-derived equivalence without that DISTINCT,
  EXCEPT, INTERSECT ALL, EXCEPT ALL, NULL-containing inputs, and ordering
  through the other equivalent output column in descending direction. The frozen
  regression fails on baseline and passes on the candidate. Baseline execution
  stops at the first failing original query, so later queries are not claimed
  to have separate observed native-red runs. These queries use rowsort;
  their result checks do not establish the returned row order.
• Repeated metamorphic replay: all six complete baseline/candidate matrices
  agree with the observations above. The replay verifier checks results, the
  exact baseline-only row error, plan changes, and actual scan cardinalities;
  it also rejects four deliberately corrupted in-memory evidence variants.
• Independent held-out checks: 18 cases over 12 explicit integer/NULL rows,
  with expected multisets computed independently from those rows: minimum counts
  for INTERSECT ALL, nonnegative count differences for EXCEPT ALL, and count
  sums for UNION ALL. Candidate results match all 18 cases with zero errors;
  the four baseline wrong-result bags and one row-engine ordering error are
  repaired. These checks also validate actual ASC/DESC/NULL output ordering
  where specified, unlike the native rowsort checks.
• Broader regression coverage: full ordering, props, norm, xform, and
  execbuilder unit targets; the separate complete TestExecBuild_union plan
  fixture; complete union/distsql_union logic suites under local,
  local-vec-off, fakedist, fakedist-vec-off, and fakedist-disk; and the
  full SQL test target across 16 shards all passed locally. Existing opt: support streaming set ops when ordering not required #64062
  streaming-positive, opt: invalid stream ordering results in incorrect results for set ops #69497 single-side-key, and opt: internal error in set operations due to stream ordering not including all columns #68702 duplicate-mapping
  fixtures passed without changing their expected plans.

The committed changes are limited to set.go, the new set_test.go, its Bazel
registration, and the SQL union regression: four files, +239/-1. Supplemental
metamorphic and held-out logs/verifiers are retained in the local investigation
record; they are not additional files in this PR.

Current limitations and follow-up questions

• Architecture: this is not the full bottom-up/merge-join-style redesign.
  Implicit top-down streaming remains available, subject to the strengthened
  common-order contract.
• Plan quality: selecting a deterministic concrete representative can
  strengthen child requirements, require a sort, or change hash-versus-streaming
  selection. No performance benchmark or search-space completeness proof is
  supplied. The observed hash plan is disclosed above; the unchanged streaming
  fixtures do not establish an absence of performance regressions.
• Scope of correctness coverage: this is not a framework-wide audit of FD
  and ordering propagation in every relational operator, nor an exhaustive
  proof for every future FD/ordering combination. A different operator with a
  similar unsafe boundary would need its own analysis.
• Version/distribution coverage: validation is against the stated baseline
  and candidate, not a backport certification for every affected release or a
  production multi-node workload/performance qualification.

Would maintainers prefer this localized ordering-contract fix as an initial
correctness repair, with the bottom-up redesign tracked separately, or want that
redesign included before landing? If follow-up is desired, a possible direction
is to bind explicit concrete common comparison keys to streaming alternatives
derived from child interesting orderings, then derive each input requirement
using only its own FDs, as merge joins do. Retaining the new implication tests
and benchmarking representative sort/hash/streaming choices would help evaluate
that work. This is a proposed follow-up direction, not an implemented or
performance-validated part of this PR.

Resolves: #144925
Epic: none

Release note (bug fix): Fixed incorrect results or badly ordered input
errors from INTERSECT and EXCEPT when column equivalences allowed their
inputs to use inconsistent orderings for a streaming set operation.

[blathers-crl]

Thank you for contributing to CockroachDB. Please ensure you have followed the guidelines for creating a PR.

My owl senses detect your PR is good for review. Please keep an eye out for any test failures in CI.

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is dev-inf.}