This is an automated email from the ASF dual-hosted git repository. damccorm pushed a commit to branch backport-3de6c449-disable-project-set-op-transpose in repository https://gitbox.apache.org/repos/asf/beam.git
commit 605f0f6d2aebaffb47587a52096e3abfbd68d136 Author: Danny McCormick <[email protected]> AuthorDate: Thu Jul 23 14:14:50 2026 +0000 Disable PROJECT_SET_OP_TRANSPOSE rule in BeamRuleSets Remove CoreRules.PROJECT_SET_OP_TRANSPOSE from the active rule set. This rule pushes projections through set operations (UNION/INTERSECT/ EXCEPT), but doing so collapses IEEE-754 FP surrogates wrapped around FLOAT/DOUBLE set-operation columns, reintroducing non-deterministic key coders that Beam's CoGroup-based set operators reject. Adds BeamRuleSetsTest verifying PROJECT_SET_OP_TRANSPOSE is disabled and FILTER_SET_OP_TRANSPOSE remains enabled. --- .../extensions/sql/impl/planner/BeamRuleSets.java | 10 ++- .../sql/impl/planner/BeamRuleSetsTest.java | 72 ++++++++++++++++++++++ 2 files changed, 80 insertions(+), 2 deletions(-) diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamRuleSets.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamRuleSets.java index 8382fcb6e38..e8d4cc36159 100644 --- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamRuleSets.java +++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamRuleSets.java @@ -79,8 +79,14 @@ public class BeamRuleSets { CoreRules.FILTER_AGGREGATE_TRANSPOSE, // push filter through set operation CoreRules.FILTER_SET_OP_TRANSPOSE, - // push project through set operation - CoreRules.PROJECT_SET_OP_TRANSPOSE, + // NOTE: CoreRules.PROJECT_SET_OP_TRANSPOSE is intentionally NOT enabled. Pushing a + // projection through a set operation (UNION/INTERSECT/EXCEPT) transposes the projection + // into each branch, where PROJECT_MERGE then fuses it with an adjacent branch projection. + // That fusion cancels the deterministic IEEE-754 FP surrogate wrapped around FLOAT/DOUBLE + // set-operation columns (it collapses from_bits(bits(x)) back to the raw DOUBLE), + // reintroducing the non-deterministic Double/Float key coder that Beam's CoGroup-based + // set operators reject ("the keyCoder of a GroupByKey must be deterministic"). The rule is + // a pure optimization, so omitting it is correctness-preserving. // aggregation and projection rules // BeamAggregateProjectMergeRule.INSTANCE, diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamRuleSetsTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamRuleSetsTest.java new file mode 100644 index 00000000000..9d8ea667fdb --- /dev/null +++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamRuleSetsTest.java @@ -0,0 +1,72 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.extensions.sql.impl.planner; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.util.Collection; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.plan.RelRule; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.tools.RuleSet; +import org.junit.Test; + +/** Tests for {@link BeamRuleSets}, verifying the composition of the active planner rule set. */ +public class BeamRuleSetsTest { + + @Test + public void testProjectSetOpTransposeIsDisabled() { + Collection<RuleSet> ruleSets = BeamRuleSets.getRuleSets(); + boolean foundProjectSetOpTranspose = false; + for (RuleSet ruleSet : ruleSets) { + for (org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.plan.RelOptRule rule : + ruleSet) { + String ruleName = rule.toString(); + if (ruleName.contains("PROJECT_SET_OP_TRANSPOSE") + || ruleName.contains("ProjectSetOpTranspose")) { + foundProjectSetOpTranspose = true; + break; + } + } + } + assertFalse( + "CoreRules.PROJECT_SET_OP_TRANSPOSE should NOT be enabled in the active rule set, " + + "as it collapses IEEE-754 FP surrogates around FLOAT/DOUBLE set-operation columns, " + + "reintroducing non-deterministic key coders.", + foundProjectSetOpTranspose); + } + + @Test + public void testFilterSetOpTransposeIsEnabled() { + Collection<RuleSet> ruleSets = BeamRuleSets.getRuleSets(); + boolean foundFilterSetOpTranspose = false; + for (RuleSet ruleSet : ruleSets) { + for (org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.plan.RelOptRule rule : + ruleSet) { + String ruleName = rule.toString(); + if (ruleName.contains("FILTER_SET_OP_TRANSPOSE") + || ruleName.contains("FilterSetOpTranspose")) { + foundFilterSetOpTranspose = true; + break; + } + } + } + assertTrue( + "CoreRules.FILTER_SET_OP_TRANSPOSE should be enabled in the active rule set", + foundFilterSetOpTranspose); + } +}
