This is an automated email from the ASF dual-hosted git repository.
zzwqqq pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/calcite.git
The following commit(s) were added to refs/heads/main by this push:
new 152a4208c5 [CALCITE-7755] Support IEJoin for inequality joins
152a4208c5 is described below
commit 152a4208c54ecbcea74b9555644f804f764912eb
Author: zzwqqq <[email protected]>
AuthorDate: Sun Aug 30 14:49:00 2026 +0800
[CALCITE-7755] Support IEJoin for inequality joins
---
.../adapter/enumerable/EnumerableIEJoin.java | 287 +++++++++++++
.../adapter/enumerable/EnumerableIEJoinRule.java | 111 +++++
.../adapter/enumerable/EnumerableRules.java | 7 +
.../java/org/apache/calcite/tools/Programs.java | 1 +
.../org/apache/calcite/util/BuiltInMethod.java | 5 +
.../test/enumerable/EnumerableIEJoinTest.java | 451 +++++++++++++++++++++
.../apache/calcite/test/CombineRelOptRulesTest.xml | 4 +-
.../apache/calcite/linq4j/EnumerableDefaults.java | 45 ++
.../apache/calcite/linq4j/IEJoinEnumerator.java | 286 +++++++++++++
.../org/apache/calcite/linq4j/test/IEJoinTest.java | 245 +++++++++++
10 files changed, 1440 insertions(+), 2 deletions(-)
diff --git
a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableIEJoin.java
b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableIEJoin.java
new file mode 100644
index 0000000000..197c3ab3d9
--- /dev/null
+++
b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableIEJoin.java
@@ -0,0 +1,287 @@
+/*
+ * 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.calcite.adapter.enumerable;
+
+import org.apache.calcite.adapter.java.JavaTypeFactory;
+import org.apache.calcite.linq4j.function.Function1;
+import org.apache.calcite.linq4j.tree.BlockBuilder;
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.linq4j.tree.ExpressionType;
+import org.apache.calcite.linq4j.tree.Expressions;
+import org.apache.calcite.linq4j.tree.ParameterExpression;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptCost;
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelCollations;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelNodes;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.metadata.RelMdUtil;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.type.SqlTypeUtil;
+import org.apache.calcite.util.BuiltInMethod;
+import org.apache.calcite.util.Util;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/** Implementation of an inner IEJoin with two inequality predicates in
+ * {@link EnumerableConvention enumerable calling convention}. */
+public class EnumerableIEJoin extends Join implements EnumerableRel {
+ private final ImmutableList<Condition> conditions;
+
+ protected EnumerableIEJoin(RelOptCluster cluster, RelTraitSet traitSet,
+ RelNode left, RelNode right, RexNode condition) {
+ super(cluster, traitSet, ImmutableList.of(), left, right, condition,
+ ImmutableSet.of(), JoinRelType.INNER);
+ final List<RexNode> conjunctions = RelOptUtil.conjunctions(condition);
+ if (conjunctions.size() != 2) {
+ throw new IllegalArgumentException(
+ "condition must contain exactly two supported cross-input
inequalities");
+ }
+ final int leftFieldCount = left.getRowType().getFieldCount();
+ final Condition first =
+ analyzeConjunction(conjunctions.get(0), leftFieldCount);
+ final Condition second =
+ analyzeConjunction(conjunctions.get(1), leftFieldCount);
+ if (first == null || second == null) {
+ throw new IllegalArgumentException(
+ "condition must contain supported cross-input inequalities");
+ }
+ final ImmutableList<Condition> inequalities = ImmutableList.of(first,
second);
+ for (Condition inequality : inequalities) {
+ if (!supportsKeyTypes(left, right, inequality)) {
+ throw new IllegalArgumentException("unsupported IEJoin key types: left
"
+ +
left.getRowType().getFieldList().get(inequality.leftKey).getType()
+ + ", right "
+ +
right.getRowType().getFieldList().get(inequality.rightKey).getType());
+ }
+ }
+ this.conditions = inequalities;
+ }
+
+ /** Creates an EnumerableIEJoin. */
+ public static EnumerableIEJoin create(RelNode left, RelNode right,
+ RexNode condition) {
+ return new EnumerableIEJoin(left.getCluster(),
+ left.getCluster().traitSetOf(EnumerableConvention.INSTANCE),
+ left, right, condition);
+ }
+
+ @Override public EnumerableIEJoin copy(RelTraitSet traitSet,
+ RexNode condition, RelNode left, RelNode right, JoinRelType joinType,
+ boolean semiJoinDone) {
+ if (joinType != JoinRelType.INNER) {
+ throw new IllegalArgumentException("EnumerableIEJoin only supports inner
joins");
+ }
+ return new EnumerableIEJoin(getCluster(), traitSet, left, right,
+ condition);
+ }
+
+ @Override public @Nullable RelOptCost computeSelfCost(RelOptPlanner planner,
+ RelMetadataQuery mq) {
+ final Double leftRows = mq.getRowCount(left);
+ final Double rightRows = mq.getRowCount(right);
+ final Double joinRows = mq.getRowCount(this);
+ if (leftRows == null || rightRows == null || joinRows == null) {
+ return null;
+ }
+ double outputRows = joinRows;
+ if (RelNodes.COMPARATOR.compare(left, right) > 0) {
+ outputRows = RelMdUtil.addEpsilon(outputRows);
+ }
+ final double inputRows = leftRows + rightRows;
+ // Sort the combined inputs by each inequality key, then scan and emit
pairs.
+ final double cost =
+ 2D * Util.nLogN(inputRows) + inputRows + outputRows;
+ return planner.getCostFactory().makeCost(cost, 0, 0);
+ }
+
+ @Override public Result implement(EnumerableRelImplementor implementor,
+ Prefer pref) {
+ final BlockBuilder builder = new BlockBuilder();
+ final Result leftResult =
+ implementor.visitChild(this, 0, (EnumerableRel) left, pref);
+ final Expression leftExpression =
+ builder.append("left", leftResult.block);
+ final Result rightResult =
+ implementor.visitChild(this, 1, (EnumerableRel) right, pref);
+ final Expression rightExpression =
+ builder.append("right", rightResult.block);
+ final ParameterExpression leftParameter =
+ Expressions.parameter(leftResult.physType.getJavaRowType(), "leftRow");
+ final ParameterExpression rightParameter =
+ Expressions.parameter(rightResult.physType.getJavaRowType(),
"rightRow");
+ final JavaTypeFactory typeFactory = implementor.getTypeFactory();
+ final List<Expression> keySelectors = new ArrayList<>();
+ final List<Expression> comparators = new ArrayList<>();
+
+ for (Condition condition : conditions) {
+ final RelDataType leftType =
+ left.getRowType().getFieldList().get(condition.leftKey).getType();
+ final RelDataType rightType =
+ right.getRowType().getFieldList().get(condition.rightKey).getType();
+ // Use SQL storage types so timestamp comparisons use millisecond
precision.
+ final RelDataType keyType =
+ typeFactory.toSql(
+
requireNonNull(typeFactory.leastRestrictive(ImmutableList.of(leftType,
rightType))));
+ final Type keyClass = typeFactory.getJavaClass(keyType);
+ // For nullable INTEGER keys in array rows:
+ // leftRow -> (Integer) leftRow[leftKey]
+ // rightRow -> (Integer) rightRow[rightKey]
+ keySelectors.add(
+ Expressions.lambda(
+ Function1.class,
+ EnumUtils.convert(
+ leftResult.physType.fieldReference(
+ leftParameter, condition.leftKey), keyClass), leftParameter));
+ keySelectors.add(
+ Expressions.lambda(
+ Function1.class,
+ EnumUtils.convert(
+ rightResult.physType.fieldReference(
+ rightParameter, condition.rightKey), keyClass), rightParameter));
+ // PhysType generates comparators for row fields, so wrap the key in a
+ // scalar row type.
+ // For nullable INTEGER keys: (a, b) -> Utilities.compareNullsLast(a, b)
+ final RelDataType keyRowType =
+ typeFactory.builder().add("key", keyType).build();
+ final PhysType keyPhysType =
+ PhysTypeImpl.of(typeFactory, keyRowType, JavaRowFormat.SCALAR);
+ comparators.add(
+ keyPhysType.generateComparator(
+ RelCollations.of(
+ new RelFieldCollation(0,
+ RelFieldCollation.Direction.ASCENDING,
+ RelFieldCollation.NullDirection.LAST))));
+ }
+
+ final PhysType physType =
+ PhysTypeImpl.of(typeFactory, getRowType(), pref.preferArray());
+ final List<Expression> arguments = new ArrayList<>();
+ arguments.add(leftExpression);
+ arguments.add(rightExpression);
+ arguments.addAll(keySelectors);
+ arguments.addAll(comparators);
+ arguments.add(Expressions.constant(conditions.get(0).operator));
+ arguments.add(Expressions.constant(conditions.get(1).operator));
+ // For two-field array rows, the result selector is:
+ // (leftRow, rightRow) -> new Object[] {
+ // leftRow[0], leftRow[1], rightRow[0], rightRow[1] }
+ arguments.add(
+ EnumUtils.joinSelector(joinType, physType,
+ ImmutableList.of(leftResult.physType, rightResult.physType)));
+
+ // return EnumerableDefaults.ieJoin(left, right,
+ // leftKey1, rightKey1, leftKey2, rightKey2,
+ // comparator1, comparator2, operator1, operator2, resultSelector);
+ return implementor.result(physType,
+ builder.append(
+ Expressions.call(BuiltInMethod.IE_JOIN.method,
+ arguments)).toBlock());
+ }
+
+ static @Nullable Condition analyzeConjunction(RexNode node,
+ int leftFieldCount) {
+ if (!(node instanceof RexCall) || ((RexCall) node).operands.size() != 2) {
+ return null;
+ }
+ final RexCall call = (RexCall) node;
+ if (!(call.operands.get(0) instanceof RexInputRef)
+ || !(call.operands.get(1) instanceof RexInputRef)) {
+ return null;
+ }
+ final int first = ((RexInputRef) call.operands.get(0)).getIndex();
+ final int second = ((RexInputRef) call.operands.get(1)).getIndex();
+ final boolean firstIsLeft = first < leftFieldCount;
+ final boolean secondIsLeft = second < leftFieldCount;
+ if (firstIsLeft == secondIsLeft) {
+ return null;
+ }
+ final ExpressionType operator;
+ switch (firstIsLeft ? call.getKind() : call.getKind().reverse()) {
+ case LESS_THAN:
+ operator = ExpressionType.LessThan;
+ break;
+ case LESS_THAN_OR_EQUAL:
+ operator = ExpressionType.LessThanOrEqual;
+ break;
+ case GREATER_THAN:
+ operator = ExpressionType.GreaterThan;
+ break;
+ case GREATER_THAN_OR_EQUAL:
+ operator = ExpressionType.GreaterThanOrEqual;
+ break;
+ default:
+ return null;
+ }
+ return firstIsLeft
+ ? new Condition(first, second - leftFieldCount, operator)
+ : new Condition(second, first - leftFieldCount, operator);
+ }
+
+ static boolean supportsKeyTypes(RelNode left, RelNode right,
+ Condition condition) {
+ final RelDataType leftType =
+ left.getRowType().getFieldList().get(condition.leftKey).getType();
+ final RelDataType rightType =
+ right.getRowType().getFieldList().get(condition.rightKey).getType();
+ final SqlTypeName typeName = leftType.getSqlTypeName();
+ // Floating-point sort order disagrees with <, <=, > and >= for NaN
+ // and signed zero.
+ return SqlTypeUtil.equalSansNullability(
+ left.getCluster().getTypeFactory(), leftType, rightType)
+ && (SqlTypeUtil.isBoolean(leftType)
+ || (SqlTypeUtil.isExactNumeric(leftType)
+ && !SqlTypeName.UNSIGNED_TYPES.contains(typeName))
+ || SqlTypeUtil.isCharacter(leftType)
+ || SqlTypeUtil.isBinary(leftType)
+ || SqlTypeUtil.isDatetime(leftType)
+ || SqlTypeUtil.isInterval(leftType));
+ }
+
+ /** A normalized IEJoin condition. */
+ static final class Condition {
+ final int leftKey;
+ final int rightKey;
+ final ExpressionType operator;
+
+ private Condition(int leftKey, int rightKey,
+ ExpressionType operator) {
+ this.leftKey = leftKey;
+ this.rightKey = rightKey;
+ this.operator = operator;
+ }
+ }
+}
diff --git
a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableIEJoinRule.java
b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableIEJoinRule.java
new file mode 100644
index 0000000000..5023fc5e57
--- /dev/null
+++
b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableIEJoinRule.java
@@ -0,0 +1,111 @@
+/*
+ * 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.calcite.adapter.enumerable;
+
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.convert.ConverterRule;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.logical.LogicalJoin;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexUtil;
+
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/** Planner rule that converts an inner {@link LogicalJoin} whose condition
+ * consists of at least two cross-input field inequalities to an
+ * {@link EnumerableIEJoin}.
+ *
+ * <p>Based on Khayyat et al.,
+ * <a href="https://doi.org/10.14778/2831360.2831362";>"Lightning Fast and Space
+ * Efficient Inequality Joins," PVLDB 8(13), 2015</a>. The first two
+ * inequalities drive the join and additional inequalities are evaluated by an
+ * {@link EnumerableCalc}.
+ *
+ * @see EnumerableRules#ENUMERABLE_IE_JOIN_RULE
+ */
+class EnumerableIEJoinRule extends ConverterRule {
+ /** Default configuration. */
+ static final Config DEFAULT_CONFIG = Config.INSTANCE
+ .withConversion(LogicalJoin.class, Convention.NONE,
+ EnumerableConvention.INSTANCE, "EnumerableIEJoinRule")
+ .withRuleFactory(EnumerableIEJoinRule::new);
+
+ /** Called from the Config. */
+ protected EnumerableIEJoinRule(Config config) {
+ super(config);
+ }
+
+ @Override public @Nullable RelNode convert(RelNode rel) {
+ final Join join = (Join) rel;
+ if (join.getJoinType() != JoinRelType.INNER
+ || !join.getVariablesSet().isEmpty()
+ || !join.getSystemFieldList().isEmpty()) {
+ return null;
+ }
+
+ final int leftFieldCount = join.getLeft().getRowType().getFieldCount();
+ final List<RexNode> conjunctions =
+ RelOptUtil.conjunctions(join.getCondition());
+ if (conjunctions.size() < 2) {
+ return null;
+ }
+ for (int i = 0; i < conjunctions.size(); i++) {
+ final EnumerableIEJoin.Condition condition =
+ EnumerableIEJoin.analyzeConjunction(conjunctions.get(i),
leftFieldCount);
+ if (condition == null) {
+ return null;
+ }
+ if (i < 2
+ && !EnumerableIEJoin.supportsKeyTypes(
+ join.getLeft(), join.getRight(), condition)) {
+ return null;
+ }
+ }
+
+ final RelNode left = convert(join.getLeft(), join.getLeft().getTraitSet()
+ .replace(EnumerableConvention.INSTANCE));
+ final RelNode right = convert(join.getRight(),
join.getRight().getTraitSet()
+ .replace(EnumerableConvention.INSTANCE));
+ final RexBuilder rexBuilder = join.getCluster().getRexBuilder();
+ final RexNode ieCondition =
+ requireNonNull(RexUtil.composeConjunction(rexBuilder,
conjunctions.subList(0, 2)));
+ final EnumerableIEJoin ieJoin =
+ EnumerableIEJoin.create(left, right, ieCondition);
+ if (conjunctions.size() == 2) {
+ return ieJoin;
+ }
+
+ final RexNode residual =
+ requireNonNull(
+ RexUtil.composeConjunction(rexBuilder,
+ conjunctions.subList(2, conjunctions.size())));
+ final RexProgram program =
+ RexProgram.create(ieJoin.getRowType(),
+ rexBuilder.identityProjects(ieJoin.getRowType()), residual,
+ ieJoin.getRowType(), rexBuilder);
+ return EnumerableCalc.create(ieJoin, program);
+ }
+}
diff --git
a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRules.java
b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRules.java
index a470d3b7ce..5b20e665f4 100644
---
a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRules.java
+++
b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRules.java
@@ -61,6 +61,12 @@ private EnumerableRules() {
EnumerableMergeJoinRule.DEFAULT_CONFIG
.toRule(EnumerableMergeJoinRule.class);
+ /** Rule that converts a compatible
+ * {@link org.apache.calcite.rel.logical.LogicalJoin} to an
+ * {@link EnumerableIEJoin}. */
+ public static final RelOptRule ENUMERABLE_IE_JOIN_RULE =
+ EnumerableIEJoinRule.DEFAULT_CONFIG.toRule(EnumerableIEJoinRule.class);
+
public static final RelOptRule ENUMERABLE_CORRELATE_RULE =
EnumerableCorrelateRule.DEFAULT_CONFIG
.toRule(EnumerableCorrelateRule.class);
@@ -225,6 +231,7 @@ private EnumerableRules() {
ImmutableList.of(EnumerableRules.ENUMERABLE_JOIN_RULE,
EnumerableRules.ENUMERABLE_ASOFJOIN_RULE,
EnumerableRules.ENUMERABLE_MERGE_JOIN_RULE,
+ EnumerableRules.ENUMERABLE_IE_JOIN_RULE,
EnumerableRules.ENUMERABLE_CORRELATE_RULE,
EnumerableRules.ENUMERABLE_CONDITIONAL_CORRELATE_RULE,
EnumerableRules.ENUMERABLE_PROJECT_RULE,
diff --git a/core/src/main/java/org/apache/calcite/tools/Programs.java
b/core/src/main/java/org/apache/calcite/tools/Programs.java
index 1307e70100..27f6850be6 100644
--- a/core/src/main/java/org/apache/calcite/tools/Programs.java
+++ b/core/src/main/java/org/apache/calcite/tools/Programs.java
@@ -81,6 +81,7 @@ public class Programs {
EnumerableRules.ENUMERABLE_TABLE_SCAN_RULE,
EnumerableRules.ENUMERABLE_JOIN_RULE,
EnumerableRules.ENUMERABLE_MERGE_JOIN_RULE,
+ EnumerableRules.ENUMERABLE_IE_JOIN_RULE,
EnumerableRules.ENUMERABLE_CORRELATE_RULE,
EnumerableRules.ENUMERABLE_CONDITIONAL_CORRELATE_RULE,
EnumerableRules.ENUMERABLE_PROJECT_RULE,
diff --git a/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java
b/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java
index 731d420f34..a8c006d61e 100644
--- a/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java
+++ b/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java
@@ -51,6 +51,7 @@
import org.apache.calcite.linq4j.function.NullablePredicate2;
import org.apache.calcite.linq4j.function.Predicate1;
import org.apache.calcite.linq4j.function.Predicate2;
+import org.apache.calcite.linq4j.tree.ExpressionType;
import org.apache.calcite.linq4j.tree.FunctionExpression;
import org.apache.calcite.linq4j.tree.Primitive;
import org.apache.calcite.linq4j.tree.Types;
@@ -254,6 +255,10 @@ public enum BuiltInMethod {
MERGE_JOIN(EnumerableDefaults.class, "mergeJoin", Enumerable.class,
Enumerable.class, Function1.class, Function1.class, Predicate2.class,
Function2.class,
JoinType.class, Comparator.class, EqualityComparer.class),
+ IE_JOIN(EnumerableDefaults.class, "ieJoin", Enumerable.class,
+ Enumerable.class, Function1.class, Function1.class, Function1.class,
+ Function1.class, Comparator.class, Comparator.class,
+ ExpressionType.class, ExpressionType.class, Function2.class),
SLICE0(Enumerables.class, "slice0", Enumerable.class),
SEMI_JOIN(EnumerableDefaults.class, "semiJoin", Enumerable.class,
Enumerable.class, Function1.class, Function1.class,
diff --git
a/core/src/test/java/org/apache/calcite/test/enumerable/EnumerableIEJoinTest.java
b/core/src/test/java/org/apache/calcite/test/enumerable/EnumerableIEJoinTest.java
new file mode 100644
index 0000000000..42023ad051
--- /dev/null
+++
b/core/src/test/java/org/apache/calcite/test/enumerable/EnumerableIEJoinTest.java
@@ -0,0 +1,451 @@
+/*
+ * 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.calcite.test.enumerable;
+
+import org.apache.calcite.adapter.enumerable.EnumerableIEJoin;
+import org.apache.calcite.adapter.enumerable.EnumerableRules;
+import org.apache.calcite.adapter.java.ReflectiveSchema;
+import org.apache.calcite.config.CalciteConnectionProperty;
+import org.apache.calcite.config.Lex;
+import org.apache.calcite.jdbc.JavaCollation;
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelRoot;
+import org.apache.calcite.rel.RelVisitor;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.runtime.Hook;
+import org.apache.calcite.sql.SqlCollation;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.test.CalciteAssert;
+import org.apache.calcite.test.schemata.hr.Employee;
+import org.apache.calcite.test.schemata.hr.Event;
+import org.apache.calcite.test.schemata.hr.HrSchema;
+import org.apache.calcite.test.schemata.hr.HrSchemaBig;
+import org.apache.calcite.util.Holder;
+import org.apache.calcite.util.Util;
+
+import com.google.common.collect.ImmutableList;
+
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Timestamp;
+import java.text.Collator;
+import java.util.Locale;
+import java.util.function.Consumer;
+
+import static org.apache.calcite.linq4j.Nullness.castNonNull;
+
+import static org.hamcrest.CoreMatchers.allOf;
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.not;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import static java.util.Objects.requireNonNull;
+
+/** Unit tests for {@link EnumerableIEJoin}.
+ *
+ * <p>Disabling {@code ENUMERABLE_JOIN_RULE} forces execution through IEJoin
+ * for supported conditions, even on small inputs.
+ */
+class EnumerableIEJoinTest {
+ private static final SqlCollation PRIMARY_COLLATION =
+ new JavaCollation(SqlCollation.Coercibility.IMPLICIT, Locale.US,
+ Util.getDefaultCharset(), Collator.PRIMARY);
+
+ @Test void ieJoin() {
+ // The first comparison reverses the operands: rx < lx becomes lx > rx.
+ // Equal keys do not match these strict comparisons; null keys never match.
+ final Holder<@Nullable RelRoot> root = Holder.empty();
+ tester(new HrSchema())
+ .withRel(builder -> {
+ builder
+ .values(new String[]{"lx", "ly"},
+ 2, 8,
+ 5, 4,
+ null, 3)
+ .values(new String[]{"rx", "ry"},
+ 4, 3,
+ 7, 6,
+ 5, 4)
+ .join(JoinRelType.INNER,
+ builder.and(
+ builder.lessThan(
+ builder.field(2, 1, "rx"),
+ builder.field(2, 0, "lx")),
+ builder.greaterThan(
+ builder.field(2, 0, "ly"),
+ builder.field(2, 1, "ry"))));
+ return builder.build();
+ })
+ .withHook(Hook.PLANNER, (Consumer<RelOptPlanner>) planner ->
+ planner.removeRule(EnumerableRules.ENUMERABLE_JOIN_RULE))
+ .withHook(Hook.PLAN_BEFORE_IMPLEMENTATION,
+ (Consumer<RelRoot>) root::set)
+ .explainHookMatches(
+ "EnumerableIEJoin(condition=[AND(<($2, $0), >($1, $3))],
joinType=[inner])\n"
+ + " EnumerableValues(tuples=[[{ 2, 8 }, { 5, 4 }, { null, 3
}]])\n"
+ + " EnumerableValues(tuples=[[{ 4, 3 }, { 7, 6 }, { 5, 4
}]])\n")
+ .returnsUnordered("lx=5; ly=4; rx=4; ry=3");
+
+ final EnumerableIEJoin join =
+ (EnumerableIEJoin) requireNonNull(root.get()).rel;
+ assertThrows(IllegalArgumentException.class,
+ () -> join.copy(join.getTraitSet(), join.getCondition(),
+ join.getLeft(), join.getRight(), JoinRelType.LEFT, false));
+ assertThrows(IllegalArgumentException.class,
+ () -> join.copy(join.getTraitSet(),
+ join.getCluster().getRexBuilder().makeLiteral(true),
+ join.getLeft(), join.getRight(), JoinRelType.INNER, false));
+
+ // A missing row count on either input or the join leaves the cost unknown.
+ for (RelNode unknown : ImmutableList.of(join.getLeft(), join.getRight(),
join)) {
+ final RelMetadataQuery mq = new RelMetadataQuery() {
+ @Override public Double getRowCount(RelNode rel) {
+ if (rel == unknown) {
+ // The signature is not nullable yet; see CALCITE-4263.
+ return castNonNull(null);
+ }
+ return 10D;
+ }
+ };
+ assertNull(join.computeSelfCost(join.getCluster().getPlanner(), mq));
+ }
+ }
+
+ @Test void ieJoinDoesNotSupportApproximateNumbers() {
+ // Both signed zeros satisfy >= 0; NaN does not.
+ final Holder<@Nullable RelRoot> root = Holder.empty();
+ tester(new TestSchema())
+ .query("select l.name as left_name, r.name as right_name "
+ + "from lefts l join rights r "
+ + "on l.x >= r.x and l.y > r.y")
+ .withHook(Hook.PLAN_BEFORE_IMPLEMENTATION,
+ (Consumer<RelRoot>) root::set)
+ .explainHookMatches(
+ allOf(
+ containsString("EnumerableNestedLoopJoin"),
+ not(containsString("EnumerableIEJoin"))))
+ .returnsUnordered(
+ "left_name=minusZero; right_name=zero",
+ "left_name=plusZero; right_name=zero");
+ final Join join = findJoin(requireNonNull(root.get()).rel);
+ assertThat(
+ assertThrows(IllegalArgumentException.class,
+ () -> EnumerableIEJoin.create(
+ join.getLeft(), join.getRight(), join.getCondition()))
+ .getMessage(),
+ allOf(containsString("left JavaType(double)"),
+ containsString("right JavaType(double)")));
+ }
+
+ @Test void ieJoinDoesNotSupportAny() {
+ // ANY keys must be compared according to the values' runtime types.
+ final Holder<@Nullable RelNode> root = Holder.empty();
+ tester(new TestSchema())
+ .query("select l.name as left_name, r.name as right_name "
+ + "from anyLefts l join anyRights r "
+ + "on l.x < r.x and l.y > r.y")
+ .withHook(Hook.TRIMMED, (Consumer<RelNode>) root::set)
+ .explainContains("EnumerableNestedLoopJoin");
+ final Join join = findJoin(requireNonNull(root.get()));
+ assertThrows(IllegalArgumentException.class,
+ () -> EnumerableIEJoin.create(
+ join.getLeft(), join.getRight(), join.getCondition()));
+ }
+
+ @Test void ieJoinCostAndDefaultSelection() {
+ final HrSchemaBig schema = new HrSchemaBig();
+ // Count matching pairs independently of IEJoin.
+ int expectedCount = 0;
+ for (Employee left : schema.emps) {
+ for (Employee right : schema.emps) {
+ if (left.empid < right.empid && left.deptno >= right.deptno) {
+ expectedCount++;
+ }
+ }
+ }
+
+ // Keep all join rules enabled to test cost-based selection.
+ final Holder<@Nullable RelRoot> root = Holder.empty();
+ tester(schema)
+ .query("select count(*) from emps l join emps r "
+ + "on l.empid < r.empid and l.deptno >= r.deptno")
+ .withHook(Hook.PLAN_BEFORE_IMPLEMENTATION,
+ (Consumer<RelRoot>) root::set)
+ .explainHookMatches(containsString("EnumerableIEJoin"))
+ .returns("EXPR$0=" + expectedCount + "\n");
+
+ final Join join = findJoin(requireNonNull(root.get()).rel);
+ final EnumerableIEJoin selfJoin =
+ EnumerableIEJoin.create(join.getLeft(), join.getLeft(),
+ join.getCondition());
+ final RelMetadataQuery mq = new RelMetadataQuery() {
+ @Override public Double getRowCount(RelNode rel) {
+ return rel == selfJoin ? 3D : 10D;
+ }
+ };
+ assertThat(
+ requireNonNull(
+ selfJoin.computeSelfCost(selfJoin.getCluster().getPlanner(), mq))
+ .getRows(),
+ is(2D * Util.nLogN(20D) + 23D));
+ }
+
+ @Test void ieJoinWithBooleanAndDecimalKeys() {
+ // Boolean ordering is false < true.
+ tester(new HrSchema())
+ .query("select l.id as left_id, r.id as right_id "
+ + "from (values (1, true, cast(2 as decimal(5, 2))), "
+ + "(2, false, cast(4 as decimal(5, 2)))) as l(id, b, d) "
+ + "join (values (3, false, cast(3 as decimal(5, 2))), "
+ + "(4, true, cast(1 as decimal(5, 2)))) as r(id, b, d) "
+ + "on l.b > r.b and l.d < r.d")
+ .withHook(Hook.PLANNER, (Consumer<RelOptPlanner>) planner ->
+ planner.removeRule(EnumerableRules.ENUMERABLE_JOIN_RULE))
+ .explainHookMatches(containsString("EnumerableIEJoin"))
+ .returnsUnordered("left_id=1; right_id=3");
+ }
+
+ @Test void ieJoinWithBinaryAndNullableDateKeys() {
+ // A null date prevents a match even when the binary comparison succeeds.
+ tester(new HrSchema())
+ .query("select l.id as left_id, r.id as right_id "
+ + "from (values (1, cast(X'01' as varbinary(2)), date
'2020-01-02'), "
+ + "(2, cast(X'01' as varbinary(2)), cast(null as date))) as l(id,
b, d) "
+ + "join (values (3, cast(X'02' as varbinary(2)), date
'2020-01-01'), "
+ + "(4, cast(X'00' as varbinary(2)), date '2020-01-03')) "
+ + "as r(id, b, d) on l.b < r.b and l.d > r.d")
+ .withHook(Hook.PLANNER, (Consumer<RelOptPlanner>) planner ->
+ planner.removeRule(EnumerableRules.ENUMERABLE_JOIN_RULE))
+ .explainHookMatches(containsString("EnumerableIEJoin"))
+ .returnsUnordered("left_id=1; right_id=3");
+ }
+
+ @Test void ieJoinWithJavaTimestampKeys() {
+ // Java timestamps that differ only within a millisecond compare equal in
SQL.
+ tester(new TestSchema())
+ .query("select l.eventid from timestampLefts l join timestampRights r "
+ + "on l.ts > r.ts and l.eventid > r.eventid")
+ .withHook(Hook.PLANNER, (Consumer<RelOptPlanner>) planner ->
+ planner.removeRule(EnumerableRules.ENUMERABLE_JOIN_RULE))
+ .explainHookMatches(containsString("EnumerableIEJoin"))
+ .returnsUnordered("eventid=3");
+ }
+
+ @Test void ieJoinWithIntervalKeys() {
+ tester(new HrSchema())
+ .query("select l.id as left_id, r.id as right_id "
+ + "from (values (1, interval '2' day, interval '4' hour), "
+ + "(2, interval '4' day, interval '1' hour)) "
+ + "as l(id, d, h) "
+ + "join (values (3, interval '3' day, interval '2' hour), "
+ + "(4, interval '1' day, interval '5' hour)) "
+ + "as r(id, d, h) on l.d < r.d and l.h > r.h")
+ .withHook(Hook.PLANNER, (Consumer<RelOptPlanner>) planner ->
+ planner.removeRule(EnumerableRules.ENUMERABLE_JOIN_RULE))
+ .explainHookMatches(containsString("EnumerableIEJoin"))
+ .returnsUnordered("left_id=1; right_id=3");
+ }
+
+ @Test void ieJoinWithCollatedVarcharKeys() {
+ // PRIMARY collation ignores case and accents: "abc" and "ÀBC" compare
equal.
+ tester(new HrSchema())
+ .withRel(builder -> {
+ final RelDataType stringType =
+ builder.getTypeFactory().createTypeWithCharsetAndCollation(
+ builder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR),
+ builder.getTypeFactory().getDefaultCharset(),
PRIMARY_COLLATION);
+ final RelDataType rowType = builder.getTypeFactory().builder()
+ .add("s", stringType)
+ .add("n", SqlTypeName.INTEGER)
+ .build();
+ return builder
+ .values(rowType, "abc", 2, "z", 1).as("l")
+ .values(rowType, "ÀBC", 1, "b", 3).as("r")
+ .join(JoinRelType.INNER,
+ builder.and(
+ builder.greaterThanOrEqual(
+ builder.field(2, "l", "s"),
+ builder.field(2, "r", "s")),
+ builder.greaterThan(
+ builder.field(2, "l", "n"),
+ builder.field(2, "r", "n"))))
+ .project(builder.field("l", "s"), builder.field("r", "s"))
+ .build();
+ })
+ .withHook(Hook.PLANNER, (Consumer<RelOptPlanner>) planner ->
+ planner.removeRule(EnumerableRules.ENUMERABLE_JOIN_RULE))
+ .explainHookMatches(containsString("EnumerableIEJoin"))
+ .returnsUnordered("s=abc; s0=ÀBC");
+ }
+
+ @Test void ieJoinWithApproximateRemainingPredicate() {
+ // EnumerableCalc evaluates the remaining z >= 0 comparison:
+ // both signed zeros match, but NaN does not.
+ tester(new TestSchema())
+ .query("select l.name as left_name, r.name as right_name "
+ + "from residualLefts l join residualRights r "
+ + "on l.x < r.x and l.y > r.y and l.z >= r.z")
+ .withHook(Hook.PLANNER, (Consumer<RelOptPlanner>) planner ->
+ planner.removeRule(EnumerableRules.ENUMERABLE_JOIN_RULE))
+ .explainHookMatches(
+ allOf(
+ containsString("EnumerableIEJoin"),
+ containsString(">=($t3, $t7)")))
+ .returnsUnordered(
+ "left_name=minusZero; right_name=zero",
+ "left_name=plusZero; right_name=zero");
+ }
+
+ @Test void ieJoinUnsupportedConditionsUseExistingRules() {
+ for (String condition : ImmutableList.of(
+ "e.empid < d.deptno",
+ "e.empid < d.deptno and e.deptno > d.deptno "
+ + "and e.empid = d.deptno",
+ "e.empid + 1 < d.deptno and e.deptno > d.deptno",
+ "e.empid < d.deptno or e.deptno > d.deptno",
+ "e.empid < e.deptno and e.deptno > d.deptno")) {
+ tester(new HrSchema())
+ .query("select * from emps e join depts d on " + condition)
+ .explainHookMatches(not(containsString("EnumerableIEJoin")))
+ .runs();
+ }
+ tester(new HrSchema())
+ .query("select * from emps e left join depts d "
+ + "on e.empid < d.deptno and e.deptno > d.deptno")
+ .explainHookMatches(not(containsString("EnumerableIEJoin")))
+ .runs();
+ }
+
+ @Test void ieJoinWithSharedField() {
+ // x = 2 matches both inclusive intervals.
+ tester(new HrSchema())
+ .query("select l.x, r.lo, r.hi "
+ + "from (values (1), (2), (3)) as l(x) "
+ + "join (values (1, 2), (2, 3)) as r(lo, hi) "
+ + "on l.x >= r.lo and l.x <= r.hi")
+ .withHook(Hook.PLANNER, (Consumer<RelOptPlanner>) planner ->
+ planner.removeRule(EnumerableRules.ENUMERABLE_JOIN_RULE))
+ .explainHookMatches(containsString("EnumerableIEJoin"))
+ .returnsUnordered("x=1; lo=1; hi=2", "x=2; lo=1; hi=2",
+ "x=2; lo=2; hi=3", "x=3; lo=2; hi=3");
+ }
+
+ private CalciteAssert.AssertThat tester(Object schema) {
+ return CalciteAssert.that()
+ .with(CalciteConnectionProperty.LEX, Lex.JAVA)
+ .with(CalciteConnectionProperty.FORCE_DECORRELATE, false)
+ .withSchema("s", new ReflectiveSchema(schema));
+ }
+
+ private static Join findJoin(RelNode rel) {
+ final Holder<@Nullable Join> join = Holder.empty();
+ new RelVisitor() {
+ @Override public void visit(RelNode node, int ordinal,
+ @Nullable RelNode parent) {
+ if (node instanceof Join) {
+ join.set((Join) node);
+ } else {
+ super.visit(node, ordinal, parent);
+ }
+ }
+ }.go(rel);
+ return requireNonNull(join.get());
+ }
+
+ /** Test schema for unsupported and residual key types. */
+ public static class TestSchema {
+ public final ApproximatePoint[] lefts = {
+ new ApproximatePoint("nan", Double.NaN, 1),
+ new ApproximatePoint("minusZero", -0D, 1),
+ new ApproximatePoint("plusZero", 0D, 1)
+ };
+ public final ApproximatePoint[] rights = {
+ new ApproximatePoint("zero", 0D, 0)
+ };
+ public final AnyPoint[] anyLefts = {
+ new AnyPoint("one", 1, 2)
+ };
+ public final AnyPoint[] anyRights = {
+ new AnyPoint("two", 2, 1)
+ };
+ public final ResidualPoint[] residualLefts = {
+ new ResidualPoint("nan", 2, 8, Double.NaN),
+ new ResidualPoint("minusZero", 2, 8, -0D),
+ new ResidualPoint("plusZero", 2, 8, 0D)
+ };
+ public final ResidualPoint[] residualRights = {
+ new ResidualPoint("zero", 4, 3, 0D)
+ };
+ public final Event[] timestampLefts = {
+ new Event(2, Timestamp.valueOf("2020-01-01 00:00:00.000002")),
+ new Event(3, Timestamp.valueOf("2020-01-01 00:00:00.001")),
+ new Event(4, null)
+ };
+ public final Event[] timestampRights = {
+ new Event(1, Timestamp.valueOf("2020-01-01 00:00:00.000001"))
+ };
+ }
+
+ /** Row with an approximate key. */
+ public static class ApproximatePoint {
+ public final String name;
+ public final double x;
+ public final int y;
+
+ ApproximatePoint(String name, double x, int y) {
+ this.name = name;
+ this.x = x;
+ this.y = y;
+ }
+ }
+
+ /** Row with dynamically typed keys. */
+ public static class AnyPoint {
+ public final String name;
+ public final Object x;
+ public final Object y;
+
+ AnyPoint(String name, Object x, Object y) {
+ this.name = name;
+ this.x = x;
+ this.y = y;
+ }
+ }
+
+ /** Row with an approximate residual key. */
+ public static class ResidualPoint {
+ public final String name;
+ public final int x;
+ public final int y;
+ public final double z;
+
+ ResidualPoint(String name, int x, int y, double z) {
+ this.name = name;
+ this.x = x;
+ this.y = y;
+ this.z = z;
+ }
+ }
+}
diff --git
a/core/src/test/resources/org/apache/calcite/test/CombineRelOptRulesTest.xml
b/core/src/test/resources/org/apache/calcite/test/CombineRelOptRulesTest.xml
index 1ae0496621..fc1ab908f8 100644
--- a/core/src/test/resources/org/apache/calcite/test/CombineRelOptRulesTest.xml
+++ b/core/src/test/resources/org/apache/calcite/test/CombineRelOptRulesTest.xml
@@ -110,7 +110,7 @@ EnumerableCombine
EnumerableTableScan(table=[[scott, DEPT]])
EnumerableTableScan(table=[[scott, EMP]])
EnumerableProject(ENAME=[$1], GRADE=[$8])
- EnumerableNestedLoopJoin(condition=[AND(>=($5, $9), <=($5, $10))],
joinType=[inner])
+ EnumerableIEJoin(condition=[AND(>=($5, $9), <=($5, $10))],
joinType=[inner])
EnumerableTableScan(table=[[scott, EMP]])
EnumerableTableScan(table=[[scott, SALGRADE]])
]]>
@@ -180,7 +180,7 @@ Combine
EnumerableCombine
EnumerableAggregate(group=[{9, 11}], EMP_COUNT=[COUNT()])
EnumerableTableSpool(readType=[LAZY], writeType=[LAZY], table=[[TEMP,
spool_0]])
- EnumerableNestedLoopJoin(condition=[AND(>=($5, $12), <=($5, $13))],
joinType=[inner])
+ EnumerableIEJoin(condition=[AND(>=($5, $12), <=($5, $13))],
joinType=[inner])
EnumerableProject(EMPNO=[$3], ENAME=[$4], JOB=[$5], MGR=[$6],
HIREDATE=[$7], SAL=[$8], COMM=[$9], DEPTNO=[$10], DEPTNO0=[$0], DNAME=[$1],
LOC=[$2])
EnumerableHashJoin(condition=[=($0, $10)], joinType=[inner])
EnumerableTableScan(table=[[scott, DEPT]])
diff --git
a/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java
b/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java
index 0eca4b6d52..2687c15131 100644
--- a/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java
+++ b/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java
@@ -34,6 +34,7 @@
import org.apache.calcite.linq4j.function.NullablePredicate2;
import org.apache.calcite.linq4j.function.Predicate1;
import org.apache.calcite.linq4j.function.Predicate2;
+import org.apache.calcite.linq4j.tree.ExpressionType;
import com.google.common.base.Suppliers;
import com.google.common.collect.HashMultiset;
@@ -2533,6 +2534,50 @@ private static <TSource, TInner, TKey>
Enumerable<TSource> semiEquiJoin_(
};
}
+ /**
+ * Joins pairs that satisfy two inequality predicates.
+ *
+ * <p>Each operator compares a left key with a right key using the
+ * corresponding comparator. Null keys do not match. Both inputs are
+ * materialized and sorted before the first result is returned.
+ *
+ * @throws IllegalArgumentException if either operator is not
+ * {@link ExpressionType#LessThan}, {@link
ExpressionType#LessThanOrEqual},
+ * {@link ExpressionType#GreaterThan}, or
+ * {@link ExpressionType#GreaterThanOrEqual}
+ */
+ public static <TLeft, TRight, TKey1, TKey2, TResult>
+ Enumerable<TResult> ieJoin(Enumerable<TLeft> left,
+ Enumerable<TRight> right,
+ Function1<? super TLeft, TKey1> leftKeySelector1,
+ Function1<? super TRight, TKey1> rightKeySelector1,
+ Function1<? super TLeft, TKey2> leftKeySelector2,
+ Function1<? super TRight, TKey2> rightKeySelector2,
+ Comparator<? super TKey1> comparator1,
+ Comparator<? super TKey2> comparator2,
+ ExpressionType operator1, ExpressionType operator2,
+ Function2<? super TLeft, ? super TRight, TResult> resultSelector) {
+ for (ExpressionType operator : Arrays.asList(operator1, operator2)) {
+ switch (operator) {
+ case LessThan:
+ case LessThanOrEqual:
+ case GreaterThan:
+ case GreaterThanOrEqual:
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported IEJoin operator: " +
operator);
+ }
+ }
+ return new AbstractEnumerable<TResult>() {
+ @Override public Enumerator<TResult> enumerator() {
+ return new IEJoinEnumerator<>(left, right,
+ leftKeySelector1, rightKeySelector1,
+ leftKeySelector2, rightKeySelector2,
+ comparator1, comparator2, operator1, operator2, resultSelector);
+ }
+ };
+ }
+
/**
* Correlates the elements of two sequences based on a predicate.
*/
diff --git
a/linq4j/src/main/java/org/apache/calcite/linq4j/IEJoinEnumerator.java
b/linq4j/src/main/java/org/apache/calcite/linq4j/IEJoinEnumerator.java
new file mode 100644
index 0000000000..9450d5960e
--- /dev/null
+++ b/linq4j/src/main/java/org/apache/calcite/linq4j/IEJoinEnumerator.java
@@ -0,0 +1,286 @@
+/*
+ * 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.calcite.linq4j;
+
+import org.apache.calcite.linq4j.function.Function1;
+import org.apache.calcite.linq4j.function.Function2;
+import org.apache.calcite.linq4j.tree.ExpressionType;
+
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.Comparator;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+import static org.apache.calcite.linq4j.Nullness.castNonNull;
+
+/** Enumerator that implements IEJoin for two inequality predicates.
+ *
+ * <p>Both inputs are materialized; rows with a null key are discarded. The
+ * entries are sorted by each key. A right entry satisfies predicate 1 exactly
+ * when it follows a left entry in {@code firstOrder}, and predicate 2 exactly
+ * when it precedes that left entry in {@code secondOrder}. Sort directions and
+ * tie-breaking depend on the operators; see {@link #entryComparator}.
+ *
+ * <p>{@code permutation} maps positions in {@code secondOrder} to positions in
+ * {@code firstOrder}. Scanning in the second order, each right entry sets its
+ * first-order position in {@code activeRights}. For each left entry, set bits
+ * after its first-order position identify the right entries satisfying both
+ * predicates.
+ *
+ * <p>For example, for {@code l.x < r.x AND l.y > r.y}, let the left row be
+ * {@code l=(2,8)} and the right rows be {@code a=(4,3)}, {@code b=(1,6)}, and
+ * {@code c=(5,9)}:
+ *
+ * <pre>
+ * position: 0 1 2 3
+ * firstOrder (x asc): b l a c
+ * secondOrder (y asc): a b l c
+ * permutation: 2 0 1 3
+ * </pre>
+ *
+ * <p>When the scan reaches l, {@code activeRights} contains positions 0 (b)
+ * and 2 (a). Searching after l's position, 1, yields only a. Row b fails
+ * predicate 1, and row c has not yet been visited because it fails predicate
2.
+ *
+ * <p>Based on the union-array algorithm in Section 4.2 of Khayyat et al.,
+ * <a href="https://doi.org/10.14778/2831360.2831362";>"Lightning Fast and Space
+ * Efficient Inequality Joins," PVLDB 8(13), 2015</a>.
+ *
+ * @param <TLeft> Left row type
+ * @param <TRight> Right row type
+ * @param <TKey1> First key type
+ * @param <TKey2> Second key type
+ * @param <TResult> Result row type
+ */
+final class IEJoinEnumerator<TLeft, TRight, TKey1, TKey2, TResult>
+ implements Enumerator<TResult> {
+ private final List<TLeft> leftRows = new ArrayList<>();
+
+ private final List<TRight> rightRows = new ArrayList<>();
+
+ /** Entries from both inputs in first-key order. */
+ private final List<Entry<TKey1, TKey2>> firstOrder;
+
+ /** Maps second-order positions to {@link #firstOrder} positions. */
+ private int[] permutation;
+
+ /** First-order positions of right entries already visited in the
+ * second-order scan. */
+ private final BitSet activeRights = new BitSet();
+
+ private final Function2<? super TLeft, ? super TRight, TResult>
resultSelector;
+
+ /** Next index in {@link #permutation} to visit. */
+ private int secondPosition;
+
+ /** Inclusive first-order position from which to resume searching for
+ * matches for {@link #currentLeft}. */
+ private int nextBit;
+
+ /** Left entry whose matches are being emitted, or null between entries. */
+ private @Nullable Entry<TKey1, TKey2> currentLeft;
+
+ private @Nullable TResult current;
+
+ /** Whether {@link #current} is a valid result. Kept separately because the
+ * result selector may itself return null. */
+ private boolean hasCurrent;
+
+ /** Reads and closes both input enumerators, caches the rows and their keys,
+ * and prepares both sort orders. Rows with either key null are excluded.
+ * {@link #reset()} reuses this state without reading the inputs again.
+ *
+ * <p>Selectors with the same numeric suffix form a predicate of the form
+ * {@code leftKey operator rightKey}, using the corresponding comparator.
+ * Comparators define key order; this class derives sort directions and
+ * equal-key tie-breaks from the operators. The result selector is called
+ * only as {@link #moveNext()} emits matching pairs.
+ */
+ IEJoinEnumerator(Enumerable<TLeft> left, Enumerable<TRight> right,
+ Function1<? super TLeft, TKey1> leftKeySelector1,
+ Function1<? super TRight, TKey1> rightKeySelector1,
+ Function1<? super TLeft, TKey2> leftKeySelector2,
+ Function1<? super TRight, TKey2> rightKeySelector2,
+ Comparator<? super TKey1> comparator1,
+ Comparator<? super TKey2> comparator2,
+ ExpressionType operator1, ExpressionType operator2,
+ Function2<? super TLeft, ? super TRight, TResult> resultSelector) {
+ this.resultSelector = resultSelector;
+
+ final List<Entry<TKey1, TKey2>> entries = new ArrayList<>();
+ try (Enumerator<TLeft> enumerator = left.enumerator()) {
+ while (enumerator.moveNext()) {
+ final TLeft row = enumerator.current();
+ final @Nullable TKey1 key1 = leftKeySelector1.apply(row);
+ final @Nullable TKey2 key2 = leftKeySelector2.apply(row);
+ if (key1 != null && key2 != null) {
+ final int rowIndex = leftRows.size();
+ leftRows.add(row);
+ entries.add(new Entry<>(true, rowIndex, key1, key2));
+ }
+ }
+ }
+ try (Enumerator<TRight> enumerator = right.enumerator()) {
+ while (enumerator.moveNext()) {
+ final TRight row = enumerator.current();
+ final @Nullable TKey1 key1 = rightKeySelector1.apply(row);
+ final @Nullable TKey2 key2 = rightKeySelector2.apply(row);
+ if (key1 != null && key2 != null) {
+ final int rowIndex = rightRows.size();
+ rightRows.add(row);
+ entries.add(new Entry<>(false, rowIndex, key1, key2));
+ }
+ }
+ }
+
+ firstOrder = new ArrayList<>(entries);
+ firstOrder.sort(
+ entryComparator(comparator1, operator1, true));
+ for (int i = 0; i < firstOrder.size(); i++) {
+ firstOrder.get(i).firstPosition = i;
+ }
+
+ final List<Entry<TKey1, TKey2>> secondOrder = new ArrayList<>(entries);
+ secondOrder.sort(
+ entryComparator(comparator2, operator2, false));
+ permutation = new int[secondOrder.size()];
+ for (int i = 0; i < secondOrder.size(); i++) {
+ permutation[i] = secondOrder.get(i).firstPosition;
+ }
+ }
+
+ /** Orders entries by key 1 if {@code isFirstOrder}, or by key 2 otherwise.
+ *
+ * <p>For equal keys from different inputs, strict operators put the right
+ * entry first in the first order and last in the second order, excluding the
+ * pair. Non-strict operators reverse both tie-breaks to include the pair.
+ */
+ @SuppressWarnings("unchecked")
+ private static <TKey1, TKey2, TKey> Comparator<Entry<TKey1, TKey2>>
+ entryComparator(Comparator<? super TKey> comparator,
+ ExpressionType operator, boolean isFirstOrder) {
+ final boolean greaterThan =
+ operator == ExpressionType.GreaterThan
+ || operator == ExpressionType.GreaterThanOrEqual;
+ final boolean descending =
+ isFirstOrder ? greaterThan : !greaterThan;
+ final boolean strict =
+ operator == ExpressionType.LessThan
+ || operator == ExpressionType.GreaterThan;
+ final boolean leftSideFirst = isFirstOrder != strict;
+ return (entry1, entry2) -> {
+ final TKey key1 =
+ castNonNull((TKey) (isFirstOrder ? entry1.key1 : entry1.key2));
+ final TKey key2 =
+ castNonNull((TKey) (isFirstOrder ? entry2.key1 : entry2.key2));
+ final int c = descending
+ ? comparator.compare(key2, key1)
+ : comparator.compare(key1, key2);
+ if (c != 0 || entry1.isLeft == entry2.isLeft) {
+ return c;
+ }
+
+ return entry1.isLeft == leftSideFirst ? -1 : 1;
+ };
+ }
+
+ @Override public TResult current() {
+ if (!hasCurrent) {
+ throw new NoSuchElementException();
+ }
+ return castNonNull(current);
+ }
+
+ @Override public boolean moveNext() {
+ hasCurrent = false;
+ while (true) {
+ // Rights already seen in secondOrder satisfy predicate 2; active bits
+ // after currentLeft's position in firstOrder also satisfy predicate 1.
+ if (currentLeft != null) {
+ final int bit = activeRights.nextSetBit(nextBit);
+ if (bit >= 0) {
+ nextBit = bit + 1;
+ final Entry<TKey1, TKey2> right = firstOrder.get(bit);
+ current =
+ resultSelector.apply(leftRows.get(currentLeft.rowIndex),
+ rightRows.get(right.rowIndex));
+ hasCurrent = true;
+ return true;
+ }
+ currentLeft = null;
+ }
+
+ if (secondPosition >= permutation.length) {
+ current = null;
+ return false;
+ }
+
+ final int firstPosition = permutation[secondPosition++];
+ final Entry<TKey1, TKey2> entry = firstOrder.get(firstPosition);
+ if (entry.isLeft) {
+ currentLeft = entry;
+ nextBit = firstPosition + 1;
+ } else {
+ activeRights.set(firstPosition);
+ }
+ }
+ }
+
+ @Override public void reset() {
+ activeRights.clear();
+ secondPosition = 0;
+ nextBit = 0;
+ currentLeft = null;
+ current = null;
+ hasCurrent = false;
+ }
+
+ @Override public void close() {
+ reset();
+ leftRows.clear();
+ rightRows.clear();
+ firstOrder.clear();
+ permutation = new int[0];
+ }
+
+ /** Row entry shared by the two sorted orders.
+ *
+ * @param <TKey1> First key type
+ * @param <TKey2> Second key type
+ */
+ private static final class Entry<TKey1, TKey2> {
+ final boolean isLeft;
+
+ /** Index into {@code leftRows} or {@code rightRows}, according to
+ * {@link #isLeft}. */
+ final int rowIndex;
+ final TKey1 key1;
+ final TKey2 key2;
+ /** Index in {@code firstOrder}, assigned after sorting. */
+ int firstPosition;
+
+ private Entry(boolean isLeft, int rowIndex, TKey1 key1, TKey2 key2) {
+ this.isLeft = isLeft;
+ this.rowIndex = rowIndex;
+ this.key1 = key1;
+ this.key2 = key2;
+ }
+ }
+}
diff --git
a/linq4j/src/test/java/org/apache/calcite/linq4j/test/IEJoinTest.java
b/linq4j/src/test/java/org/apache/calcite/linq4j/test/IEJoinTest.java
new file mode 100644
index 0000000000..eaabc407ad
--- /dev/null
+++ b/linq4j/src/test/java/org/apache/calcite/linq4j/test/IEJoinTest.java
@@ -0,0 +1,245 @@
+/*
+ * 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.calcite.linq4j.test;
+
+import org.apache.calcite.linq4j.Enumerable;
+import org.apache.calcite.linq4j.EnumerableDefaults;
+import org.apache.calcite.linq4j.Enumerator;
+import org.apache.calcite.linq4j.Linq4j;
+import org.apache.calcite.linq4j.tree.ExpressionType;
+
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Random;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/** Tests for IEJoin. */
+class IEJoinTest {
+ private static final List<ExpressionType> OPERATORS =
+ Arrays.asList(ExpressionType.LessThan, ExpressionType.LessThanOrEqual,
+ ExpressionType.GreaterThan, ExpressionType.GreaterThanOrEqual);
+
+ private static final List<Point> LEFT =
+ Arrays.asList(new Point("l0", 2, 8),
+ new Point("l1", 5, 4),
+ new Point("l2", 5, 4),
+ new Point("lnx", null, 3),
+ new Point("lny", 3, null));
+
+ private static final List<Point> RIGHT =
+ Arrays.asList(new Point("r0", 4, 3),
+ new Point("r1", 7, 6),
+ new Point("r2", 5, 4),
+ new Point("rnx", null, 2),
+ new Point("rny", 6, null));
+
+ @Test void testAllOperatorCombinationsAgainstNestedLoop() {
+ for (ExpressionType operator1 : OPERATORS) {
+ for (ExpressionType operator2 : OPERATORS) {
+ assertMatches("fixed input", LEFT, RIGHT, operator1, operator2);
+ }
+ }
+ }
+
+ @Test void testRandomInputsAgainstNestedLoop() {
+ final Random random = new Random(0);
+ for (int trial = 0; trial < 200; trial++) {
+ final List<Point> left = new ArrayList<>();
+ final List<Point> right = new ArrayList<>();
+ final int leftCount = random.nextInt(9);
+ final int rightCount = random.nextInt(9);
+ for (int i = 0; i < leftCount; i++) {
+ left.add(
+ new Point("l" + i,
+ random.nextInt(5) == 0 ? null : random.nextInt(7) - 3,
+ random.nextInt(5) == 0 ? null : random.nextInt(7) - 3));
+ }
+ for (int i = 0; i < rightCount; i++) {
+ right.add(
+ new Point("r" + i,
+ random.nextInt(5) == 0 ? null : random.nextInt(7) - 3,
+ random.nextInt(5) == 0 ? null : random.nextInt(7) - 3));
+ }
+ for (ExpressionType operator1 : OPERATORS) {
+ for (ExpressionType operator2 : OPERATORS) {
+ assertMatches("trial " + trial, left, right, operator1, operator2);
+ }
+ }
+ }
+ }
+
+ @Test void testEmptyAndSameInputs() {
+ assertMatches("empty left", Collections.emptyList(), RIGHT,
+ ExpressionType.LessThan, ExpressionType.GreaterThan);
+ assertMatches("empty right", LEFT, Collections.emptyList(),
+ ExpressionType.LessThan, ExpressionType.GreaterThan);
+ assertMatches("same input", LEFT, LEFT,
+ ExpressionType.LessThanOrEqual,
+ ExpressionType.GreaterThanOrEqual);
+ }
+
+ @Test void testSameKeyForBothPredicates() {
+ for (ExpressionType operator1 : OPERATORS) {
+ for (ExpressionType operator2 : OPERATORS) {
+ final List<String> expected = new ArrayList<>();
+ for (Point left : LEFT) {
+ for (Point right : RIGHT) {
+ if (test(left.x, right.x, operator1)
+ && test(left.x, right.x, operator2)) {
+ expected.add(left.name + ":" + right.name);
+ }
+ }
+ }
+ final List<String> actual =
+ EnumerableDefaults.ieJoin(
+ Linq4j.asEnumerable(LEFT), Linq4j.asEnumerable(RIGHT),
+ point -> point.x, point -> point.x,
+ point -> point.x, point -> point.x,
+ Comparator.naturalOrder(), Comparator.naturalOrder(),
+ operator1, operator2,
+ (left, right) -> left.name + ":" + right.name).toList();
+ Collections.sort(expected);
+ Collections.sort(actual);
+ assertThat(operator1 + "/" + operator2, actual, is(expected));
+ }
+ }
+ }
+
+ @Test void testUnsupportedOperators() {
+ for (ExpressionType operator : ExpressionType.values()) {
+ if (OPERATORS.contains(operator)) {
+ continue;
+ }
+ assertThrows(IllegalArgumentException.class,
+ () -> ieJoin(Collections.emptyList(), Collections.emptyList(),
+ operator, ExpressionType.LessThan));
+ assertThrows(IllegalArgumentException.class,
+ () -> ieJoin(Collections.emptyList(), Collections.emptyList(),
+ ExpressionType.LessThan, operator));
+ }
+ }
+
+ @Test void testReset() {
+ final Enumerable<String> join =
+ ieJoin(LEFT, RIGHT, ExpressionType.LessThan,
+ ExpressionType.GreaterThan);
+ final List<String> first = new ArrayList<>();
+ final List<String> second = new ArrayList<>();
+ try (Enumerator<String> enumerator = join.enumerator()) {
+ assertThrows(NoSuchElementException.class, enumerator::current);
+ while (enumerator.moveNext()) {
+ first.add(enumerator.current());
+ }
+ assertThrows(NoSuchElementException.class, enumerator::current);
+ enumerator.reset();
+ assertThrows(NoSuchElementException.class, enumerator::current);
+ while (enumerator.moveNext()) {
+ second.add(enumerator.current());
+ }
+ }
+ assertThat(second, is(first));
+ assertThat(join.toList(), is(first));
+ }
+
+ private static void assertMatches(String context, List<Point> left,
+ List<Point> right, ExpressionType operator1,
+ ExpressionType operator2) {
+ final List<String> expected = nestedLoop(left, right, operator1,
operator2);
+ final List<String> actual = ieJoin(left, right, operator1,
operator2).toList();
+ Collections.sort(actual);
+ assertThat(context + ", " + operator1 + "/" + operator2
+ + ", left=" + left + ", right=" + right,
+ actual, is(expected));
+ }
+
+ /** Computes the actual row pairs using the IEJoin implementation under
test. */
+ private static Enumerable<String> ieJoin(List<Point> left, List<Point> right,
+ ExpressionType operator1, ExpressionType operator2) {
+ return EnumerableDefaults.ieJoin(
+ Linq4j.asEnumerable(left), Linq4j.asEnumerable(right),
+ point -> point.x, point -> point.x,
+ point -> point.y, point -> point.y,
+ Comparator.naturalOrder(), Comparator.naturalOrder(),
+ operator1, operator2,
+ (leftPoint, rightPoint) -> leftPoint.name + ":" + rightPoint.name);
+ }
+
+ /** Computes the expected row pairs independently using nested loops and
+ * direct comparisons. Null keys do not match. Sorts the result for
comparison
+ * with the IEJoin result. */
+ private static List<String> nestedLoop(List<Point> leftRows,
+ List<Point> rightRows, ExpressionType operator1,
+ ExpressionType operator2) {
+ final List<String> result = new ArrayList<>();
+ for (Point left : leftRows) {
+ for (Point right : rightRows) {
+ if (test(left.x, right.x, operator1)
+ && test(left.y, right.y, operator2)) {
+ result.add(left.name + ":" + right.name);
+ }
+ }
+ }
+ Collections.sort(result);
+ return result;
+ }
+
+ private static boolean test(@Nullable Integer left, @Nullable Integer right,
+ ExpressionType operator) {
+ if (left == null || right == null) {
+ return false;
+ }
+ switch (operator) {
+ case LessThan:
+ return left < right;
+ case LessThanOrEqual:
+ return left <= right;
+ case GreaterThan:
+ return left > right;
+ case GreaterThanOrEqual:
+ return left >= right;
+ default:
+ throw new AssertionError(operator);
+ }
+ }
+
+ /** Test row. */
+ private static final class Point {
+ final String name;
+ final @Nullable Integer x;
+ final @Nullable Integer y;
+
+ private Point(String name, @Nullable Integer x, @Nullable Integer y) {
+ this.name = name;
+ this.x = x;
+ this.y = y;
+ }
+
+ @Override public String toString() {
+ return name + "(" + x + ", " + y + ")";
+ }
+ }
+}
