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..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,8 +79,15 @@ 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 + // 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, 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..a6f5c04fd80d --- /dev/null +++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamRuleSetsTest.java @@ -0,0 +1,71 @@ +/* + * 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.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); + } +}