Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
Original file line number Diff line number Diff line change
@@ -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<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);
}
}
Loading