You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ko...@apache.org on 2025/01/08 07:30:25 UTC
(ignite-3) branch main updated: IGNITE-23968 Sql. Improve row count estimation for joins (#4992)
This is an automated email from the ASF dual-hosted git repository.
korlov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new df480dab84 IGNITE-23968 Sql. Improve row count estimation for joins (#4992)
df480dab84 is described below
commit df480dab84b73556a7d52d3c247765cfdbd4b179
Author: korlov42 <ko...@apache.org>
AuthorDate: Wed Jan 8 09:30:20 2025 +0200
IGNITE-23968 Sql. Improve row count estimation for joins (#4992)
---
.../internal/sql/engine/ItAggregatesTest.java | 2 +-
.../sql/engine/statistic/ItStatisticTest.java | 8 +-
.../internal/sql/engine/exec/rel/HashJoinNode.java | 1 +
.../sql/engine/exec/rel/NestedLoopJoinNode.java | 6 +
.../sql/engine/metadata/IgniteMdColumnOrigins.java | 136 ++++++++
.../metadata/IgniteMdPercentageOriginalRows.java | 2 +-
.../sql/engine/metadata/IgniteMdRowCount.java | 363 ++++++++++++++++++---
.../sql/engine/metadata/IgniteMetadata.java | 5 +-
.../sql/engine/rel/AbstractIgniteJoin.java | 10 -
.../engine/schema/AbstractIgniteDataSource.java | 21 +-
.../sql/engine/framework/TestBuilders.java | 22 +-
.../internal/sql/engine/framework/TestCluster.java | 7 +
.../engine/planner/JoinRowCountEstimationTest.java | 207 ++++++++++++
.../resources/mapping/test_backup_mapping.test | 6 +-
.../internal/sql/engine/util/QueryChecker.java | 42 ++-
15 files changed, 768 insertions(+), 70 deletions(-)
diff --git a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItAggregatesTest.java b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItAggregatesTest.java
index 1ae5024e6a..7b6b377393 100644
--- a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItAggregatesTest.java
+++ b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItAggregatesTest.java
@@ -332,7 +332,7 @@ public class ItAggregatesTest extends BaseSqlIntegrationTest {
assertQuery(sql)
.disableRules("HashJoinConverter", "MergeJoinConverter")
- .matches(QueryChecker.matches(".*Exchange.*Join.*Colocated.*Aggregate.*"))
+ .matches(QueryChecker.matches(".*Join.*Exchange.*Scan.*Exchange.*Colocated.*Aggregate.*"))
.returns("val0", 50L)
.returns("val1", 50L)
.check();
diff --git a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/statistic/ItStatisticTest.java b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/statistic/ItStatisticTest.java
index 243fd9727c..21cd93daf3 100644
--- a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/statistic/ItStatisticTest.java
+++ b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/statistic/ItStatisticTest.java
@@ -17,7 +17,7 @@
package org.apache.ignite.internal.sql.engine.statistic;
-import static org.apache.ignite.internal.sql.engine.util.QueryChecker.containsRowCount;
+import static org.apache.ignite.internal.sql.engine.util.QueryChecker.scanRowCount;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -51,13 +51,13 @@ public class ItStatisticTest extends BaseSqlIntegrationTest {
insertAndUpdateRunQuery(500);
// Minimum row count is 1000, even we have less rows.
assertQuery(getUniqueQuery())
- .matches(containsRowCount("PUBLIC", "T", 1000))
+ .matches(scanRowCount("PUBLIC", "T", 1000))
.check();
insertAndUpdateRunQuery(600);
// Should return actual number of rows in the table.
assertQuery(getUniqueQuery())
- .matches(containsRowCount("PUBLIC", "T", 1100))
+ .matches(scanRowCount("PUBLIC", "T", 1100))
.check();
sqlStatisticManager.setThresholdTimeToPostponeUpdateMs(Long.MAX_VALUE);
@@ -65,7 +65,7 @@ public class ItStatisticTest extends BaseSqlIntegrationTest {
// Statistics shouldn't be updated despite we inserted new rows.
assertQuery(getUniqueQuery())
- .matches(containsRowCount("PUBLIC", "T", 1100))
+ .matches(scanRowCount("PUBLIC", "T", 1100))
.check();
} finally {
sqlStatisticManager.setThresholdTimeToPostponeUpdateMs(prevValueOfThreshold);
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinNode.java
index 652b737b5c..bf8c88b104 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinNode.java
@@ -767,6 +767,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
if (requested > 0 && waitingLeft == NOT_WAITING && waitingRight == NOT_WAITING && leftInBuf.isEmpty() && left == null
&& !rightIt.hasNext()) {
requested = 0;
+ hashStore.clear();
downstream().end();
}
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/NestedLoopJoinNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/NestedLoopJoinNode.java
index 1050c4d047..2e86b473f1 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/NestedLoopJoinNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/NestedLoopJoinNode.java
@@ -202,6 +202,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
if (requested > 0 && waitingLeft == NOT_WAITING && waitingRight == NOT_WAITING && left == null && leftInBuf.isEmpty()) {
requested = 0;
+ rightMaterialized.clear();
downstream().end();
}
}
@@ -301,6 +302,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
if (requested > 0 && waitingLeft == NOT_WAITING && waitingRight == NOT_WAITING && left == null && leftInBuf.isEmpty()) {
requested = 0;
+ rightMaterialized.clear();
downstream().end();
}
}
@@ -429,6 +431,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
if (requested > 0 && waitingLeft == NOT_WAITING && waitingRight == NOT_WAITING && left == null
&& leftInBuf.isEmpty() && rightNotMatchedIndexes.isEmpty()) {
requested = 0;
+ rightMaterialized.clear();
downstream().end();
}
}
@@ -581,6 +584,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
if (requested > 0 && waitingLeft == NOT_WAITING && waitingRight == NOT_WAITING && left == null
&& leftInBuf.isEmpty() && rightNotMatchedIndexes.isEmpty()) {
requested = 0;
+ rightMaterialized.clear();
downstream().end();
}
}
@@ -649,6 +653,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
if (requested > 0 && waitingLeft == NOT_WAITING && waitingRight == NOT_WAITING && left == null
&& leftInBuf.isEmpty()) {
+ rightMaterialized.clear();
downstream().end();
requested = 0;
}
@@ -720,6 +725,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
if (requested > 0 && waitingLeft == NOT_WAITING && waitingRight == NOT_WAITING && left == null && leftInBuf.isEmpty()) {
requested = 0;
+ rightMaterialized.clear();
downstream().end();
}
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdColumnOrigins.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdColumnOrigins.java
new file mode 100644
index 0000000000..f9632ddb8b
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdColumnOrigins.java
@@ -0,0 +1,136 @@
+/*
+ * 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.ignite.internal.sql.engine.metadata;
+
+import static org.apache.ignite.internal.util.ExceptionUtils.sneakyThrow;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.List;
+import java.util.Set;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.plan.volcano.RelSubset;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.metadata.BuiltInMetadata;
+import org.apache.calcite.rel.metadata.BuiltInMetadata.ColumnOrigin;
+import org.apache.calcite.rel.metadata.MetadataDef;
+import org.apache.calcite.rel.metadata.MetadataHandler;
+import org.apache.calcite.rel.metadata.ReflectiveRelMetadataProvider;
+import org.apache.calcite.rel.metadata.RelColumnOrigin;
+import org.apache.calcite.rel.metadata.RelMdColumnOrigins;
+import org.apache.calcite.rel.metadata.RelMetadataProvider;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.ReflectUtil;
+import org.apache.calcite.util.mapping.Mapping;
+import org.apache.ignite.internal.sql.engine.rel.ProjectableFilterableTableScan;
+import org.apache.ignite.internal.sql.engine.rel.agg.IgniteReduceAggregateBase;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Metadata about the origins of columns.
+ */
+@SuppressWarnings({"unused", "MethodMayBeStatic"}) // actually all methods are used by runtime generated classes
+public class IgniteMdColumnOrigins implements MetadataHandler<ColumnOrigin> {
+ public static final RelMetadataProvider SOURCE =
+ ReflectiveRelMetadataProvider.reflectiveSource(
+ new IgniteMdColumnOrigins(), BuiltInMetadata.ColumnOrigin.Handler.class);
+
+ private IgniteMdColumnOrigins() {
+ }
+
+ @Override
+ public MetadataDef<ColumnOrigin> getDef() {
+ return BuiltInMetadata.ColumnOrigin.DEF;
+ }
+
+ /** Provides column origin for Subset relation. */
+ public @Nullable Set<RelColumnOrigin> getColumnOrigins(RelSubset rel,
+ RelMetadataQuery mq, int outputColumn) {
+ return mq.getColumnOrigins(rel.stripped(), outputColumn);
+ }
+
+ /** Provides column origin for IgniteReduceAggregateBase relation. */
+ public @Nullable Set<RelColumnOrigin> getColumnOrigins(IgniteReduceAggregateBase rel,
+ RelMetadataQuery mq, int outputColumn) {
+ if (outputColumn < rel.getGroupSet().cardinality()) {
+ // get actual index of Group columns.
+ return mq.getColumnOrigins(rel.getInput(), rel.getGroupSet().asList().get(outputColumn));
+ }
+
+ // TODO: IGNITE-24151 support derived column origin
+ return null;
+ }
+
+ /** Provides column origin for ProjectableFilterableTableScan relation. */
+ public @Nullable Set<RelColumnOrigin> getColumnOrigins(ProjectableFilterableTableScan scan,
+ RelMetadataQuery mq, int outputColumn) {
+ RelOptTable table = scan.getTable();
+ List<RexNode> projects = scan.projects();
+
+ if (projects != null) {
+ RexNode node = projects.get(outputColumn);
+ if (node instanceof RexInputRef) {
+ RexInputRef inputRef = (RexInputRef) node;
+ outputColumn = inputRef.getIndex();
+ } else {
+ // TODO: IGNITE-24151 support derived column origin
+ return null;
+ }
+ }
+
+ ImmutableBitSet requiredColumns = scan.requiredColumns();
+ if (requiredColumns != null) {
+ Mapping trimming = Commons.trimmingMapping(table.getRowType().getFieldCount(), requiredColumns);
+
+ outputColumn = trimming.getSourceOpt(outputColumn);
+
+ if (outputColumn == -1) {
+ return null;
+ }
+ }
+
+ return Set.of(new RelColumnOrigin(table, outputColumn, false));
+ }
+
+ /** Catch all method delegates call to {@link RelMdColumnOrigins#SOURCE} .*/
+ public @Nullable Set<RelColumnOrigin> getColumnOrigins(RelNode rel,
+ RelMetadataQuery mq, int outputColumn) {
+ Method method = ReflectUtil.lookupVisitMethod(
+ RelMdColumnOrigins.SOURCE.getClass(),
+ rel.getClass(),
+ "getColumnOrigins",
+ List.of(RelMetadataQuery.class, int.class)
+ );
+
+ if (method == null) {
+ return null;
+ }
+
+ try {
+ return (Set<RelColumnOrigin>) method.invoke(RelMdColumnOrigins.SOURCE, rel, mq, outputColumn);
+ } catch (IllegalAccessException | InvocationTargetException e) {
+ sneakyThrow(e);
+ }
+
+ throw new AssertionError("Should not get here");
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdPercentageOriginalRows.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdPercentageOriginalRows.java
index 4c5115b4f0..629aa90589 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdPercentageOriginalRows.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdPercentageOriginalRows.java
@@ -158,7 +158,7 @@ public class IgniteMdPercentageOriginalRows implements MetadataHandler<BuiltInMe
* GetPercentageOriginalRows.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
- private static Double quotientForPercentage(
+ public static Double quotientForPercentage(
Double numerator,
Double denominator) {
if ((numerator == null) || (denominator == null)) {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdRowCount.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdRowCount.java
index e130a4f18f..872625df65 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdRowCount.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdRowCount.java
@@ -17,9 +17,15 @@
package org.apache.ignite.internal.sql.engine.metadata;
-import static org.apache.calcite.util.NumberUtil.multiply;
-import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
-
+import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
+import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
+import java.util.BitSet;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.Intersect;
import org.apache.calcite.rel.core.Join;
import org.apache.calcite.rel.core.JoinInfo;
@@ -27,18 +33,18 @@ import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.core.Minus;
import org.apache.calcite.rel.core.Sort;
import org.apache.calcite.rel.metadata.ReflectiveRelMetadataProvider;
+import org.apache.calcite.rel.metadata.RelColumnOrigin;
import org.apache.calcite.rel.metadata.RelMdRowCount;
import org.apache.calcite.rel.metadata.RelMdUtil;
import org.apache.calcite.rel.metadata.RelMetadataProvider;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
-import org.apache.calcite.rex.RexNode;
import org.apache.calcite.util.BuiltInMethod;
-import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.ImmutableIntList;
-import org.apache.calcite.util.Util;
+import org.apache.calcite.util.mapping.IntPair;
import org.apache.ignite.internal.sql.engine.rel.IgniteAggregate;
import org.apache.ignite.internal.sql.engine.rel.IgniteLimit;
import org.apache.ignite.internal.sql.engine.rel.IgniteSortedIndexSpool;
+import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.jetbrains.annotations.Nullable;
/**
@@ -100,68 +106,335 @@ public class IgniteMdRowCount extends RelMdRowCount {
}
/**
- * JoinRowCount.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Estimates the number of rows produced by a join operation.
+ *
+ * <p>This method calculates an estimated row count for a join by analyzing the join type,
+ * join keys, and the cardinality of the left and right inputs. It provides specialized
+ * handling for primary key and foreign key relationships (see {@link JoiningRelationType for details}).
+ * When certain metadata is unavailable or when specific conditions are not met, it falls back to
+ * Calcite's default implementation for estimating the row count.
+ *
+ * <p>Implementation details:</p>
+ * <ul>
+ * <li>If the join type is not {@link JoinRelType#INNER}, Calcite's default implementation is used.</li>
+ * <li>If the join is non-equi join, Calcite's default implementation is used.</li>
+ * <li>The row counts of the left and right inputs are retrieved using
+ * {@link RelMetadataQuery#getRowCount}. If either value is unavailable, the result is {@code null}.</li>
+ * <li>If the row counts are very small (≤ 1.0), the method uses the maximum row count as a fallback.</li>
+ * <li>Join key origins are resolved for the left and right inputs, and relationships between tables
+ * (e.g., primary key or foreign key associations) are identified and grouped into join contexts.</li>
+ * <li>If no valid join context is found, the method falls back to Calcite's implementation.</li>
+ * <li>The base row count is determined by the type of join relationship:
+ * <ul>
+ * <li>For primary key-to-primary key joins, the row count is based on the smaller table,
+ * adjusted by a percentage of the larger table's rows.</li>
+ * <li>For foreign key joins, the base table is determined based on which table is
+ * joined using non-primary key columns.</li>
+ * </ul>
+ * </li>
+ * <li>An additional adjustment factor is applied for post-filtration conditions, such as extra join keys
+ * or non-equi conditions.</li>
+ * <li>If metadata for the percentage of original rows is unavailable, the adjustment defaults to 1.0.</li>
+ * </ul>
+ *
+ * <p>If none of the above criteria are satisfied, the method defaults to
+ * {@link RelMdUtil#getJoinRowCount} for the estimation.</p>
+ *
+ * @param mq The {@link RelMetadataQuery} used to retrieve metadata about relational expressions.
+ * @param rel The {@link Join} relational expression representing the join operation.
+ * @return The estimated number of rows resulting from the join, or {@code null} if the estimation cannot be determined.
+ *
+ * @see RelMetadataQuery#getRowCount
+ * @see RelMdUtil#getJoinRowCount
+ * @see JoiningRelationType
*/
- @Nullable
- public static Double joinRowCount(RelMetadataQuery mq, Join rel) {
- if (!rel.getJoinType().projectsRight()) {
- // Create a RexNode representing the selectivity of the
- // semijoin filter and pass it to getSelectivity
- RexNode semiJoinSelectivity =
- RelMdUtil.makeSemiJoinSelectivityRexNode(mq, rel);
+ public static @Nullable Double joinRowCount(RelMetadataQuery mq, Join rel) {
+ if (rel.getJoinType() != JoinRelType.INNER) {
+ // Fall-back to calcite's implementation.
+ return RelMdUtil.getJoinRowCount(mq, rel, rel.getCondition());
+ }
- return multiply(mq.getSelectivity(rel.getLeft(), semiJoinSelectivity),
- mq.getRowCount(rel.getLeft()));
+ JoinInfo joinInfo = rel.analyzeCondition();
+
+ if (joinInfo.pairs().isEmpty()) {
+ // Fall-back to calcite's implementation.
+ return RelMdUtil.getJoinRowCount(mq, rel, rel.getCondition());
}
- // Row count estimates of 0 will be rounded up to 1.
- // So, use maxRowCount where the product is very small.
- final Double left = mq.getRowCount(rel.getLeft());
- final Double right = mq.getRowCount(rel.getRight());
+ Double leftRowCount = mq.getRowCount(rel.getLeft());
+ Double rightRowCount = mq.getRowCount(rel.getRight());
- if (left == null || right == null) {
+ if (leftRowCount == null || rightRowCount == null) {
return null;
}
- if (left <= 1D || right <= 1D) {
+ // Row count estimates of 0 will be rounded up to 1.
+ // So, use maxRowCount where the product is very small.
+ if (leftRowCount <= 1.0 || rightRowCount <= 1.0) {
Double max = mq.getMaxRowCount(rel);
- if (max != null && max <= 1D) {
+ if (max != null && max <= 1.0) {
return max;
}
}
- JoinInfo joinInfo = rel.analyzeCondition();
+ Int2ObjectMap<KeyColumnOrigin> columnsFromLeft = resolveOrigins(mq, rel.getLeft(), joinInfo.leftKeys);
+ Int2ObjectMap<KeyColumnOrigin> columnsFromRight = resolveOrigins(mq, rel.getRight(), joinInfo.rightKeys);
+
+ Map<TablesPair, JoinContext> joinContexts = new HashMap<>();
+ for (IntPair joinKeys : joinInfo.pairs()) {
+ KeyColumnOrigin leftKey = columnsFromLeft.get(joinKeys.source);
+ KeyColumnOrigin rightKey = columnsFromRight.get(joinKeys.target);
+
+ if (leftKey == null || rightKey == null) {
+ continue;
+ }
+
+ joinContexts.computeIfAbsent(
+ new TablesPair(
+ leftKey.origin.getOriginTable(),
+ rightKey.origin.getOriginTable()
+ ),
+ key -> {
+ IgniteTable leftTable = key.left.unwrap(IgniteTable.class);
+ IgniteTable rightTable = key.right.unwrap(IgniteTable.class);
+
+ assert leftTable != null && rightTable != null;
+
+ int leftPkSize = leftTable.keyColumns().size();
+ int rightPkSize = rightTable.keyColumns().size();
+
+ return new JoinContext(leftPkSize, rightPkSize);
+ }
+ ).countKeys(leftKey, rightKey);
+ }
+
+ if (joinContexts.isEmpty()) {
+ // Fall-back to calcite's implementation.
+ return RelMdUtil.getJoinRowCount(mq, rel, rel.getCondition());
+ }
+
+ Iterator<JoinContext> it = joinContexts.values().iterator();
+ JoinContext context = it.next();
+ while (it.hasNext()) {
+ JoinContext nextContext = it.next();
+ if (nextContext.joinType().strength > context.joinType().strength) {
+ context = nextContext;
+ }
+
+ if (context.joinType().strength == JoiningRelationType.PK_ON_PK.strength) {
+ break;
+ }
+ }
+
+ if (context.joinType() == JoiningRelationType.UNKNOWN) {
+ // Fall-back to calcite's implementation.
+ return RelMdUtil.getJoinRowCount(mq, rel, rel.getCondition());
+ }
+
+ double postFiltrationAdjustment = joinContexts.size() == 1 && joinInfo.isEqui() ? 1.0
+ // Extra join keys as well as non-equi conditions serves as post-filtration,
+ // therefore we need to adjust final result with a little factor.
+ : 0.7;
+
+ double baseRowCount;
+ Double percentageAdjustment;
+ if (context.joinType() == JoiningRelationType.PK_ON_PK) {
+ // Assume we have two fact tables SALES and RETURNS sharing the same primary key. Every item
+ // can be sold, but only items which were sold can be returned back, therefore
+ // size(SALES) > size(RETURNS). When joining SALES on RETURNS by primary key, the estimated
+ // result size will be the same as the size of the smallest table (RETURNS in this case),
+ // adjusted by the percentage of rows of the biggest table (SALES in this case; percentage
+ // adjustment is required to account for predicates pushed down to the table, e.g. we are
+ // interested in returns of items with certain category)
+ if (leftRowCount > rightRowCount) {
+ baseRowCount = rightRowCount;
+ percentageAdjustment = mq.getPercentageOriginalRows(rel.getLeft());
+ } else {
+ baseRowCount = leftRowCount;
+ percentageAdjustment = mq.getPercentageOriginalRows(rel.getRight());
+ }
+ } else {
+ // For foreign key joins the base table is the one which is joined by non-primary key columns.
+ if (context.joinType() == JoiningRelationType.FK_ON_PK) {
+ baseRowCount = leftRowCount;
+ percentageAdjustment = mq.getPercentageOriginalRows(rel.getRight());
+ } else {
+ assert context.joinType() == JoiningRelationType.PK_ON_FK : context.joinType();
+ baseRowCount = rightRowCount;
+ percentageAdjustment = mq.getPercentageOriginalRows(rel.getLeft());
+ }
+ }
+
+ if (percentageAdjustment == null) {
+ percentageAdjustment = 1.0; // No info, let's be conservative
+ }
+
+ return baseRowCount * percentageAdjustment * postFiltrationAdjustment;
+ }
+
+ private static Int2ObjectMap<KeyColumnOrigin> resolveOrigins(RelMetadataQuery mq, RelNode joinShoulder, ImmutableIntList keys) {
+ Int2ObjectMap<KeyColumnOrigin> origins = new Int2ObjectOpenHashMap<>();
+ for (int i : keys) {
+ if (origins.containsKey(i)) {
+ continue;
+ }
+
+ RelColumnOrigin origin = mq.getColumnOrigin(joinShoulder, i);
+ if (origin == null) {
+ continue;
+ }
+
+ IgniteTable table = origin.getOriginTable().unwrap(IgniteTable.class);
+ if (table == null) {
+ continue;
+ }
+
+ int positionInKey = table.keyColumns().indexOf(origin.getOriginColumnOrdinal());
+ origins.put(i, new KeyColumnOrigin(origin, positionInKey));
+ }
+
+ return origins;
+ }
+
+ private static class KeyColumnOrigin {
+ private final RelColumnOrigin origin;
+ private final int positionInKey;
+
+ KeyColumnOrigin(RelColumnOrigin origin, int positionInKey) {
+ this.origin = origin;
+ this.positionInKey = positionInKey;
+ }
+ }
+
+ private static class TablesPair {
+ private final RelOptTable left;
+ private final RelOptTable right;
+
+ TablesPair(RelOptTable left, RelOptTable right) {
+ this.left = left;
+ this.right = right;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TablesPair that = (TablesPair) o;
+ // Reference equality on purpose.
+ return left == that.left && right == that.right;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(left, right);
+ }
+ }
+
+ private static class JoinContext {
+ /** Used columns of primary key of table from left side. */
+ private final BitSet leftKeys;
+ /** Used columns of primary key of table from right side. */
+ private final BitSet rightKeys;
+
+ /**
+ * Used columns of primary key in both tables.
+ *
+ * <p>This bitset is initialized when PK of both tables has equal columns count, and
+ * bits are cleared when join pair contains columns with equal positions in PK of corresponding
+ * table. For example, having tables T1 and T2 with primary keys in both tables defined as
+ * CONSTRAINT PRIMARY KEY (a, b), in case of query
+ * {@code SELECT ... FROM t1 JOIN t2 ON t1.a = t2.a AND t1.b = t2.b} commonKeys will be initialized
+ * and cleared, but in case of query {@code SELECT ... FROM t1 JOIN t2 ON t1.a = t2.b AND t1.b = t2.a}
+ * (mind the join condition, where column A of one table compared with column B of another), will be
+ * only initialized (since size of the primary keys are equal), but not cleared.
+ */
+ private final @Nullable BitSet commonKeys;
+
+ JoinContext(int leftPkSize, int rightPkSize) {
+ this.leftKeys = new BitSet();
+ this.rightKeys = new BitSet();
+ this.commonKeys = leftPkSize == rightPkSize ? new BitSet() : null;
- ImmutableIntList leftKeys = joinInfo.leftKeys;
- ImmutableIntList rightKeys = joinInfo.rightKeys;
+ leftKeys.set(0, leftPkSize);
+ rightKeys.set(0, rightPkSize);
- double selectivity = mq.getSelectivity(rel, rel.getCondition());
+ if (commonKeys != null) {
+ assert leftPkSize == rightPkSize;
- if (nullOrEmpty(leftKeys) || nullOrEmpty(rightKeys)) {
- return left * right * selectivity;
+ commonKeys.set(0, leftPkSize);
+ }
}
- double leftDistinct = Util.first(
- mq.getDistinctRowCount(rel.getLeft(), ImmutableBitSet.of(leftKeys), null), left);
- double rightDistinct = Util.first(
- mq.getDistinctRowCount(rel.getRight(), ImmutableBitSet.of(rightKeys), null), right);
+ void countKeys(KeyColumnOrigin left, KeyColumnOrigin right) {
+ if (left.positionInKey >= 0) {
+ leftKeys.clear(left.positionInKey);
+ }
- double leftCardinality = leftDistinct / left;
- double rightCardinality = rightDistinct / right;
+ if (right.positionInKey >= 0) {
+ rightKeys.clear(right.positionInKey);
+ }
- double rowsCount = (Math.min(left, right) / (leftCardinality * rightCardinality)) * selectivity;
+ if (commonKeys != null && left.positionInKey == right.positionInKey && left.positionInKey >= 0) {
+ commonKeys.clear(left.positionInKey);
+ }
+ }
- JoinRelType type = rel.getJoinType();
+ JoiningRelationType joinType() {
+ if (commonKeys != null && commonKeys.isEmpty()) {
+ return JoiningRelationType.PK_ON_PK;
+ }
+
+ if (rightKeys.isEmpty()) {
+ return JoiningRelationType.FK_ON_PK;
+ }
+
+ if (leftKeys.isEmpty()) {
+ return JoiningRelationType.PK_ON_FK;
+ }
- if (type == JoinRelType.LEFT) {
- rowsCount += left;
- } else if (type == JoinRelType.RIGHT) {
- rowsCount += right;
- } else if (type == JoinRelType.FULL) {
- rowsCount += left + right;
+ return JoiningRelationType.UNKNOWN;
}
+ }
+
+ /** Enumeration of join types by their semantic. */
+ private enum JoiningRelationType {
+ /**
+ * Join by non-primary key columns.
+ *
+ * <p>Semantic is unknown.
+ */
+ UNKNOWN(0),
+ /**
+ * Join by primary keys on non-primary keys.
+ *
+ * <p>Currently we don't support Foreign Keys, thus we will assume such types of joins
+ * as joins by foreign key.
+ */
+ PK_ON_FK(UNKNOWN.strength + 1),
+ /**
+ * Join by non-primary keys on primary keys.
+ *
+ * <p>Currently we don't support Foreign Keys, thus we will assume such types of joins
+ * as joins by foreign key.
+ */
+ FK_ON_PK(PK_ON_FK.strength + 1),
+ /**
+ * Join of two tables which sharing the same primary key.
+ *
+ * <p>For example, join of tables CATALOG_SALES and CATALOG_RETURN from TPC-DS suite: both tables
+ * have the same primary key (ITEM_ID, ORDER_ID).
+ */
+ PK_ON_PK(FK_ON_PK.strength + 1);
- return rowsCount;
+ // The higher the better.
+ private final int strength;
+
+ JoiningRelationType(int strength) {
+ this.strength = strength;
+ }
}
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMetadata.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMetadata.java
index 056e5f1d9b..2a53879aa0 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMetadata.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMetadata.java
@@ -39,7 +39,10 @@ public class IgniteMetadata {
IgniteMdCollation.SOURCE,
IgniteMdSelectivity.SOURCE,
IgniteMdDistinctRowCount.SOURCE,
+ IgniteMdColumnOrigins.SOURCE,
// Basic providers
- DefaultRelMetadataProvider.INSTANCE));
+ DefaultRelMetadataProvider.INSTANCE
+ )
+ );
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/AbstractIgniteJoin.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/AbstractIgniteJoin.java
index 336558105b..6e123e42f0 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/AbstractIgniteJoin.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/AbstractIgniteJoin.java
@@ -38,13 +38,11 @@ import org.apache.calcite.rel.RelCollations;
import org.apache.calcite.rel.RelDistribution;
import org.apache.calcite.rel.RelFieldCollation;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelWriter;
import org.apache.calcite.rel.core.CorrelationId;
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.rex.RexNode;
-import org.apache.calcite.sql.SqlExplainLevel;
import org.apache.calcite.util.ImmutableIntList;
import org.apache.calcite.util.Pair;
import org.apache.calcite.util.Util;
@@ -65,14 +63,6 @@ public abstract class AbstractIgniteJoin extends Join implements TraitsAwareIgni
super(cluster, traitSet, List.of(), left, right, condition, variablesSet, joinType);
}
- /** {@inheritDoc} */
- @Override
- public RelWriter explainTerms(RelWriter pw) {
- return super.explainTerms(pw)
- .itemIf("variablesSet", Commons.transform(variablesSet.asList(), CorrelationId::getId),
- pw.getDetailLevel() == SqlExplainLevel.ALL_ATTRIBUTES);
- }
-
/** {@inheritDoc} */
@Override
public List<Pair<RelTraitSet, List<RelTraitSet>>> deriveCollation(RelTraitSet nodeTraits, List<RelTraitSet> inputTraits) {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/AbstractIgniteDataSource.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/AbstractIgniteDataSource.java
index 6e2c14aa2f..ce7db81b60 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/AbstractIgniteDataSource.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/AbstractIgniteDataSource.java
@@ -17,13 +17,18 @@
package org.apache.ignite.internal.sql.engine.schema;
+import static org.apache.ignite.internal.sql.engine.metadata.IgniteMdPercentageOriginalRows.quotientForPercentage;
+
import java.util.List;
import org.apache.calcite.config.CalciteConnectionConfig;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.TableScan;
import org.apache.calcite.rel.hint.RelHint;
+import org.apache.calcite.rel.metadata.BuiltInMetadata.PercentageOriginalRows;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.schema.Schema;
@@ -40,7 +45,8 @@ import org.jetbrains.annotations.Nullable;
/**
* Base implementation of {@link IgniteDataSource}.
*/
-public abstract class AbstractIgniteDataSource extends AbstractTable implements IgniteDataSource {
+public abstract class AbstractIgniteDataSource extends AbstractTable
+ implements IgniteDataSource, PercentageOriginalRows.Handler {
private final String name;
@@ -150,4 +156,17 @@ public abstract class AbstractIgniteDataSource extends AbstractTable implements
}
return super.unwrap(cls);
}
+
+ /** {@inheritDoc} */
+ @Override
+ public @Nullable Double getPercentageOriginalRows(RelNode r, RelMetadataQuery mq) {
+ Double tableRowCount = statistic.getRowCount();
+ Double relRowCount = mq.getRowCount(r);
+
+ if (tableRowCount == null || relRowCount == null) {
+ return null;
+ }
+
+ return quotientForPercentage(relRowCount, tableRowCount);
+ }
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
index 8ad5ead8a2..4645b0d9f4 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
@@ -75,6 +75,7 @@ import org.apache.ignite.internal.catalog.commands.TablePrimaryKey;
import org.apache.ignite.internal.catalog.descriptors.CatalogColumnCollation;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
+import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
import org.apache.ignite.internal.catalog.events.CatalogEvent;
import org.apache.ignite.internal.catalog.events.CreateIndexEventParameters;
import org.apache.ignite.internal.catalog.events.MakeIndexAvailableEventParameters;
@@ -674,9 +675,8 @@ public class TestBuilders {
var parserService = new ParserServiceImpl();
- SqlStatisticManager sqlStatisticManager = tableId -> 10_000L;
-
- var schemaManager = new SqlSchemaManagerImpl(catalogManager, sqlStatisticManager, CaffeineCacheFactory.INSTANCE, 0);
+ ConcurrentMap<String, Long> tablesSize = new ConcurrentHashMap<>();
+ var schemaManager = createSqlSchemaManager(catalogManager, tablesSize);
var prepareService = new PrepareServiceImpl(clusterName, 0, CaffeineCacheFactory.INSTANCE,
new DdlSqlToCommandConverter(), PLANNING_TIMEOUT, PLANNING_THREAD_COUNT,
new NoOpMetricManager(), schemaManager);
@@ -785,6 +785,7 @@ public class TestBuilders {
.collect(Collectors.toMap(TestNode::name, Function.identity()));
return new TestCluster(
+ tablesSize,
dataProvidersByTableName,
assignmentsProviderByTableName,
nodes,
@@ -873,6 +874,21 @@ public class TestBuilders {
}
}
+ private static SqlSchemaManagerImpl createSqlSchemaManager(CatalogManager catalogManager, ConcurrentMap<String, Long> tablesSize) {
+ SqlStatisticManager sqlStatisticManager = tableId -> {
+ CatalogTableDescriptor descriptor = catalogManager.table(tableId, Long.MAX_VALUE);
+ long fallbackSize = 10_000;
+
+ if (descriptor == null) {
+ return fallbackSize;
+ }
+
+ return tablesSize.getOrDefault(descriptor.name(), 10_000L);
+ };
+
+ return new SqlSchemaManagerImpl(catalogManager, sqlStatisticManager, CaffeineCacheFactory.INSTANCE, 0);
+ }
+
private static class TableBuilderImpl implements TableBuilder {
private final List<AbstractTableIndexBuilderImpl<?>> indexBuilders = new ArrayList<>();
private final List<ColumnDescriptor> columns = new ArrayList<>();
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestCluster.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestCluster.java
index 0965b591f6..5483994fc2 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestCluster.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestCluster.java
@@ -53,8 +53,10 @@ public class TestCluster implements LifecycleAware {
private final CatalogManager catalogManager;
private final ConcurrentMap<String, ScannableTable> dataProvidersByTableName;
private final ConcurrentMap<String, AssignmentsProvider> assignmentsProvidersByTableName;
+ private final ConcurrentMap<String, Long> tablesSize;
TestCluster(
+ ConcurrentMap<String, Long> tablesSize,
ConcurrentMap<String, ScannableTable> dataProvidersByTableName,
ConcurrentMap<String, AssignmentsProvider> assignmentsProvidersByTableName,
Map<String, TestNode> nodeByName,
@@ -64,6 +66,7 @@ public class TestCluster implements LifecycleAware {
Runnable initClosure,
RunnableX stopClosure
) {
+ this.tablesSize = tablesSize;
this.dataProvidersByTableName = dataProvidersByTableName;
this.assignmentsProvidersByTableName = assignmentsProvidersByTableName;
this.nodeByName = nodeByName;
@@ -131,6 +134,10 @@ public class TestCluster implements LifecycleAware {
dataProvidersByTableName.put(tableName, dataProvider);
}
+ public void setTableSize(String name, long size) {
+ tablesSize.put(name, size);
+ }
+
private static class ComponentToLifecycleAwareAdaptor implements LifecycleAware {
private final IgniteComponent component;
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/JoinRowCountEstimationTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/JoinRowCountEstimationTest.java
new file mode 100644
index 0000000000..ae9ab7153b
--- /dev/null
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/JoinRowCountEstimationTest.java
@@ -0,0 +1,207 @@
+/*
+ * 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.ignite.internal.sql.engine.planner;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.internal.sql.engine.framework.DataProvider.fromCollection;
+import static org.apache.ignite.internal.sql.engine.framework.TestBuilders.tableScan;
+import static org.apache.ignite.internal.sql.engine.util.QueryChecker.nodeRowCount;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.manager.ComponentContext;
+import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
+import org.apache.ignite.internal.sql.engine.InternalSqlRow;
+import org.apache.ignite.internal.sql.engine.QueryProcessor;
+import org.apache.ignite.internal.sql.engine.framework.TestBuilders;
+import org.apache.ignite.internal.sql.engine.framework.TestCluster;
+import org.apache.ignite.internal.sql.engine.framework.TestNode;
+import org.apache.ignite.internal.sql.engine.prepare.QueryMetadata;
+import org.apache.ignite.internal.sql.engine.property.SqlProperties;
+import org.apache.ignite.internal.sql.engine.util.InjectQueryCheckerFactory;
+import org.apache.ignite.internal.sql.engine.util.QueryChecker;
+import org.apache.ignite.internal.sql.engine.util.QueryCheckerExtension;
+import org.apache.ignite.internal.sql.engine.util.QueryCheckerFactory;
+import org.apache.ignite.internal.sql.engine.util.TpcTable;
+import org.apache.ignite.internal.sql.engine.util.tpcds.TpcdsTables;
+import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.apache.ignite.internal.tx.HybridTimestampTracker;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.lang.CancellationToken;
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests to check row count estimation for join relation.
+ */
+@SuppressWarnings("ConcatenationWithEmptyString")
+@ExtendWith(QueryCheckerExtension.class)
+public class JoinRowCountEstimationTest extends BaseIgniteAbstractTest {
+ private static final int CATALOG_SALES_SIZE = 1_441_548;
+ private static final int CATALOG_RETURNS_SIZE = 144_067;
+ private static final int DATE_DIM_SIZE = 73_049;
+
+ @InjectQueryCheckerFactory
+ private static QueryCheckerFactory queryCheckerFactory;
+
+ private static final TestCluster CLUSTER = TestBuilders.cluster()
+ .nodes("N1")
+ .defaultAssignmentsProvider(tableName -> (partNum, includeBackups) -> IntStream.range(0, partNum)
+ .mapToObj(part -> List.of("N1"))
+ .collect(Collectors.toList())
+ )
+ .defaultDataProvider(tableName -> tableScan(fromCollection(List.of())))
+ .build();
+
+ private static final TestNode NODE = CLUSTER.node("N1");
+
+ @BeforeAll
+ static void startCluster() {
+ CLUSTER.start();
+
+ for (TpcTable table : List.of(TpcdsTables.CATALOG_SALES, TpcdsTables.CATALOG_RETURNS, TpcdsTables.DATE_DIM)) {
+ NODE.initSchema(table.ddlScript());
+ }
+
+ CLUSTER.setTableSize("CATALOG_SALES", CATALOG_SALES_SIZE);
+ CLUSTER.setTableSize("CATALOG_RETURNS", CATALOG_RETURNS_SIZE);
+ CLUSTER.setTableSize("DATE_DIM", DATE_DIM_SIZE);
+ }
+
+ @AfterAll
+ static void stopCluster() throws Exception {
+ CLUSTER.stop();
+ }
+
+ @Test
+ void joinByPrimaryKeys() {
+ assertQuery(""
+ + "SELECT *"
+ + " FROM catalog_sales"
+ + " ,catalog_returns"
+ + " WHERE cs_item_sk = cr_item_sk"
+ + " AND cs_order_number = cr_order_number"
+ )
+ .matches(nodeRowCount("HashJoin", approximatelyEqual(CATALOG_RETURNS_SIZE)))
+ .check();
+
+ // Defined by RelMdUtil.guessSelectivity(RexNode, boolean).
+ double isNotNullPredicateFactor = 0.9;
+ assertQuery(""
+ + "SELECT *"
+ + " FROM catalog_sales"
+ + " ,catalog_returns"
+ + " WHERE cs_item_sk = cr_item_sk"
+ + " AND cs_order_number = cr_order_number"
+ + " AND cs_promo_sk IS NOT NULL"
+ )
+ .matches(nodeRowCount("HashJoin", approximatelyEqual(CATALOG_RETURNS_SIZE * isNotNullPredicateFactor)))
+ .check();
+ }
+
+ @Test
+ void joinByForeignKey() {
+ assertQuery(""
+ + "SELECT *"
+ + " FROM catalog_returns"
+ + " ,date_dim"
+ + " WHERE cr_returned_date_sk = d_date_sk"
+ )
+ .matches(nodeRowCount("HashJoin", approximatelyEqual(CATALOG_RETURNS_SIZE)))
+ .check();
+
+ // Defined by RelMdUtil.guessSelectivity(RexNode, boolean).
+ double greaterPredicateFactor = 0.5;
+ assertQuery(""
+ + "SELECT *"
+ + " FROM date_dim"
+ + " ,catalog_returns"
+ + " WHERE cr_returned_date_sk = d_date_sk"
+ + " AND d_moy > 6"
+ )
+ .matches(nodeRowCount("HashJoin", approximatelyEqual(CATALOG_RETURNS_SIZE * greaterPredicateFactor)))
+ .check();
+ }
+
+ private static QueryChecker assertQuery(String query) {
+ //noinspection DataFlowIssue
+ return queryCheckerFactory.create(
+ NODE.name(),
+ new QueryProcessor() {
+ @Override
+ public CompletableFuture<QueryMetadata> prepareSingleAsync(SqlProperties properties,
+ @Nullable InternalTransaction transaction, String qry, Object... params) {
+ throw new AssertionError("Should not be called");
+ }
+
+ @Override
+ public CompletableFuture<AsyncSqlCursor<InternalSqlRow>> queryAsync(
+ SqlProperties properties,
+ HybridTimestampTracker observableTime,
+ @Nullable InternalTransaction transaction,
+ @Nullable CancellationToken token,
+ String qry,
+ Object... params
+ ) {
+ return completedFuture(NODE.executeQuery(qry));
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ return nullCompletedFuture();
+ }
+
+ @Override
+ public CompletableFuture<Void> stopAsync(ComponentContext componentContext) {
+ return nullCompletedFuture();
+ }
+ },
+ null,
+ null,
+ query
+ );
+ }
+
+ private static Matcher<Double> approximatelyEqual(double expected) {
+ return new BaseMatcher<>() {
+ @Override
+ public boolean matches(Object o) {
+ if (!(o instanceof Double)) {
+ return false;
+ }
+
+ double value = (double) o;
+ return expected * 0.05 < value && value < expected * 1.05;
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText("equals to ").appendValue(expected).appendText("±5%");
+ }
+ };
+ }
+}
diff --git a/modules/sql-engine/src/test/resources/mapping/test_backup_mapping.test b/modules/sql-engine/src/test/resources/mapping/test_backup_mapping.test
index e3e833b50f..a714bae97d 100644
--- a/modules/sql-engine/src/test/resources/mapping/test_backup_mapping.test
+++ b/modules/sql-engine/src/test/resources/mapping/test_backup_mapping.test
@@ -60,7 +60,7 @@ Fragment#1
---
# although root is colocated with one of the partitions, algorithm prefers to colocate join stage
N0
-SELECT * FROM t2_n0n1n2 as t1, t3_n1n2 as t2 WHERE t1.id = t2.id
+SELECT * FROM t2_n0n1n2 as t1, t3_n1n2 as t2 WHERE t1.id = t2.id AND t1.c1 = 5
---
Fragment#0 root
executionNodes: [N0]
@@ -94,7 +94,7 @@ Fragment#3
---
# everything is colocated
N1
-SELECT * FROM t2_n0n1n2 as t1, t3_n1n2 as t2 WHERE t1.id = t2.id
+SELECT * FROM t2_n0n1n2 as t1, t3_n1n2 as t2 WHERE t1.id = t2.id AND t1.c1 = 5
---
Fragment#0 root
executionNodes: [N1]
@@ -128,7 +128,7 @@ Fragment#3
---
# everything is colocated, but from different root
N2
-SELECT * FROM t2_n0n1n2 as t1, t3_n1n2 as t2 WHERE t1.id = t2.id
+SELECT * FROM t2_n0n1n2 as t1, t3_n1n2 as t2 WHERE t1.id = t2.id AND t1.c1 = 5
---
Fragment#0 root
executionNodes: [N2]
diff --git a/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryChecker.java b/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryChecker.java
index edecf289d8..341e90bdb7 100644
--- a/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryChecker.java
+++ b/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryChecker.java
@@ -30,7 +30,9 @@ import java.util.Objects;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import org.apache.ignite.internal.lang.IgniteStringBuilder;
+import org.hamcrest.BaseMatcher;
import org.hamcrest.CoreMatchers;
+import org.hamcrest.Description;
import org.hamcrest.Matcher;
import org.hamcrest.core.SubstringMatcher;
@@ -67,11 +69,49 @@ public interface QueryChecker {
* @param rowCount expected row count.
* @return Matcher.
*/
- static Matcher<String> containsRowCount(String schema, String sourceName, long rowCount) {
+ static Matcher<String> scanRowCount(String schema, String sourceName, long rowCount) {
return matchesOnce(".*(Table|Index)Scan\\(table=\\[\\[" + schema + ", " + sourceName
+ "\\]\\].*\\]\\)\\:.*cumulative cost = IgniteCost \\[rowCount=" + rowCount + ".0");
}
+ /**
+ * Matcher which ensures that node matching the pattern has row count matching provided matcher.
+ *
+ * @param nodePattern A pattern describing a node of interest.
+ * @param rowCountMatcher Matcher for the estimated row count.
+ * @return Matcher.
+ */
+ static Matcher<String> nodeRowCount(String nodePattern, Matcher<Double> rowCountMatcher) {
+ Pattern pattern = Pattern.compile(".*" + nodePattern
+ + ".*?: rowcount = (?<rowcount>\\d+(?:\\.\\d+)?(?:E\\d+)?).*");
+
+ return new BaseMatcher<>() {
+ @Override
+ public boolean matches(Object o) {
+ if (!(o instanceof String)) {
+ return false;
+ }
+
+ String sanitized = ((String) o).replace("\n", "");
+ java.util.regex.Matcher matcher = pattern.matcher(sanitized);
+
+ if (!matcher.matches()) {
+ return false;
+ }
+
+ String rowCountString = matcher.group("rowcount");
+
+ return rowCountMatcher.matches(Double.parseDouble(rowCountString));
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText("node that matches pattern \"" + nodePattern + "\" should have \"rowcount\" that ");
+ description.appendDescriptionOf(rowCountMatcher);
+ }
+ };
+ }
+
/**
* Ignite index scan matcher.
*