From 605f0f6d2aebaffb47587a52096e3abfbd68d136 Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Thu, 23 Jul 2026 14:14:50 +0000 Subject: [PATCH 1/3] 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. --- .../sql/impl/planner/BeamRuleSets.java | 10 ++- .../sql/impl/planner/BeamRuleSetsTest.java | 72 +++++++++++++++++++ 2 files changed, 80 insertions(+), 2 deletions(-) create mode 100644 sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamRuleSetsTest.java 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 8382fcb6e382..e8d4cc36159c 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 000000000000..9d8ea667fdb4 --- /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 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 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); + } +} From 188872473d12836bbc6eb7d96bdc6f675709208c Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Thu, 23 Jul 2026 14:47:06 +0000 Subject: [PATCH 2/3] Fix checkstyle: remove unused import in BeamRuleSetsTest --- .../beam/sdk/extensions/sql/impl/planner/BeamRuleSetsTest.java | 1 - 1 file changed, 1 deletion(-) 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 index 9d8ea667fdb4..a6f5c04fd80d 100644 --- 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 @@ -21,7 +21,6 @@ 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; From cdb5db9cda7b863516da5dbcdbec35a7569cbb7a Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Thu, 23 Jul 2026 15:59:50 +0000 Subject: [PATCH 3/3] Fix spotless formatting in BeamRuleSets comment --- .../sql/impl/planner/BeamRuleSets.java | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 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 e8d4cc36159c..c29a9bc12489 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,14 +79,15 @@ public class BeamRuleSets { CoreRules.FILTER_AGGREGATE_TRANSPOSE, // push filter through set operation CoreRules.FILTER_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. + // NOTE: CoreRules.PROJECT_SET_OP_TRANSPOSE is intentionally NOT enabled. + // Pushing a projection through a set operation (UNION/INTERSECT/EXCEPT) transposes + // it into each branch, where PROJECT_MERGE fuses it with an adjacent projection. + // That fusion cancels the deterministic IEEE-754 FP surrogate wrapped around + // FLOAT/DOUBLE set-operation columns (collapsing 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,