This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 685d735e978 branch-4.1: [fix](fe) Prevent unsafe CTE runtime filter
pushdown #65247 (#66730)
685d735e978 is described below
commit 685d735e9789665940eac3829f863d0cb9f30e61
Author: yujun <[email protected]>
AuthorDate: Fri Aug 14 11:31:39 2026 +0800
branch-4.1: [fix](fe) Prevent unsafe CTE runtime filter pushdown #65247
(#66730)
cherry-pick: #65247
---
.../processor/post/RuntimeFilterGenerator.java | 45 ++++++++++++++++++
.../nereids/postprocess/RuntimeFilterTest.java | 49 ++++++++++++++++++++
.../runtime_filter/cte-runtime-filter.groovy | 54 +++++++++++++++++++++-
3 files changed, 147 insertions(+), 1 deletion(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java
index bbc926bbd11..478f4fab801 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java
@@ -60,11 +60,14 @@ import org.apache.doris.statistics.ColumnStatistic;
import org.apache.doris.thrift.TMinMaxRuntimeFilterType;
import org.apache.doris.thrift.TRuntimeFilterType;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
import java.util.HashMap;
@@ -88,6 +91,8 @@ public class RuntimeFilterGenerator extends PlanPostProcessor
{
JoinType.NULL_AWARE_LEFT_ANTI_JOIN
);
+ private static final Logger LOG =
LogManager.getLogger(RuntimeFilterGenerator.class);
+
private static final Set<Class<? extends PhysicalPlan>> SPJ_PLAN =
ImmutableSet.of(
PhysicalRelation.class,
PhysicalProject.class,
@@ -159,6 +164,9 @@ public class RuntimeFilterGenerator extends
PlanPostProcessor {
if (rfsToPushDown.isEmpty()) {
break;
}
+ if
(!canPushDownRuntimeFiltersIntoCTEProducer(rfsToPushDown, cteId)) {
+ continue;
+ }
// the most right deep buildNode from rfsToPushDown is
used as buildNode for pushDown rf
// since the srcExpr are the same, all buildNodes of
rfToPushDown are in the same tree path
@@ -509,6 +517,43 @@ public class RuntimeFilterGenerator extends
PlanPostProcessor {
return expression instanceof Slot ? ((Slot) expression) : null;
}
+ /**
+ * Check whether runtime filters on CTE consumers can be pushed into their
shared CTE producer.
+ */
+ @VisibleForTesting
+ public static boolean canPushDownRuntimeFiltersIntoCTEProducer(
+ List<RuntimeFilter> rfsToPushDown, CTEId cteId) {
+ if (rfsToPushDown.isEmpty()) {
+ LOG.warn("Skip pushing runtime filters into CTE producer because
no runtime filters exist for cteId: {}",
+ cteId);
+ return false;
+ }
+ Set<Expression> producerTargetExpressions = rfsToPushDown.stream()
+ .map(rf -> getProducerTargetExpression(rf, cteId))
+ .collect(Collectors.toSet());
+ return producerTargetExpressions.size() == 1;
+ }
+
+ private static Expression getProducerTargetExpression(RuntimeFilter rf,
CTEId cteId) {
+ List<PhysicalRelation> targetScans = rf.getTargetScans();
+ List<Expression> targetExpressions = rf.getTargetExpressions();
+ Preconditions.checkArgument(targetScans.size() ==
targetExpressions.size());
+ for (int i = 0; i < targetScans.size(); i++) {
+ PhysicalRelation rel = targetScans.get(i);
+ if (rel instanceof PhysicalCTEConsumer
+ && ((PhysicalCTEConsumer) rel).getCteId().equals(cteId)) {
+ PhysicalCTEConsumer consumer = (PhysicalCTEConsumer) rel;
+ Expression targetExpression = targetExpressions.get(i);
+ Map<Expression, Expression> replaceMap = Maps.newHashMap();
+ for (Slot slot : targetExpression.getInputSlots()) {
+ replaceMap.put(slot, consumer.getProducerSlot(slot));
+ }
+ return ExpressionUtils.replace(targetExpression, replaceMap);
+ }
+ }
+ throw new IllegalStateException("runtime filter does not target cteId:
" + cteId);
+ }
+
private boolean doPushDownIntoCTEProducerInternal(RuntimeFilter rf,
Expression targetExpression,
RuntimeFilterContext ctx,
PhysicalCTEProducer cteProducer) {
PhysicalPlan inputPlanNode = (PhysicalPlan) cteProducer.child(0);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java
index 59538f98e22..4122ceaabbe 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java
@@ -28,32 +28,46 @@ import org.apache.doris.nereids.hint.DistributeHint;
import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.processor.post.PlanPostProcessors;
import org.apache.doris.nereids.processor.post.RuntimeFilterContext;
+import org.apache.doris.nereids.processor.post.RuntimeFilterGenerator;
import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Add;
import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.CTEId;
import org.apache.doris.nereids.trees.expressions.EqualTo;
import org.apache.doris.nereids.trees.expressions.ExprId;
+import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.Subtract;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
import org.apache.doris.nereids.trees.plans.DistributeType;
import org.apache.doris.nereids.trees.plans.JoinType;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.commands.ExplainCommand;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalJoin;
import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalPlan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEConsumer;
import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
import org.apache.doris.nereids.trees.plans.physical.RuntimeFilter;
+import org.apache.doris.nereids.types.IntegerType;
import org.apache.doris.nereids.util.MemoTestUtils;
import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.planner.RuntimeFilterId;
import org.apache.doris.qe.OriginStatement;
+import org.apache.doris.thrift.TMinMaxRuntimeFilterType;
+import org.apache.doris.thrift.TRuntimeFilterType;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Sets;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
import java.util.ArrayList;
import java.util.List;
@@ -449,6 +463,41 @@ public class RuntimeFilterTest extends SSBTestBase {
.getAppliedRuntimeFilters().size());
}
+ @Test
+ public void
testPushSharedCteRuntimeFilterOnlyForSameProducerTargetExpression() {
+ CTEId cteId = new CTEId(1);
+ SlotReference src = new SlotReference("src", IntegerType.INSTANCE);
+ SlotReference producerPk = new SlotReference("pk",
IntegerType.INSTANCE);
+ SlotReference consumerPk1 = new SlotReference("pk",
IntegerType.INSTANCE);
+ SlotReference consumerPk2 = new SlotReference("pk",
IntegerType.INSTANCE);
+
+ List<RuntimeFilter> sameTargetFilters = ImmutableList.of(
+ newCteConsumerRuntimeFilter(src, consumerPk1, consumerPk1,
producerPk, cteId),
+ newCteConsumerRuntimeFilter(src, consumerPk2, consumerPk2,
producerPk, cteId));
+
Assertions.assertTrue(RuntimeFilterGenerator.canPushDownRuntimeFiltersIntoCTEProducer(
+ sameTargetFilters, cteId));
+
+ List<RuntimeFilter> differentTargetFilters = ImmutableList.of(
+ newCteConsumerRuntimeFilter(src, consumerPk1,
+ new Add(consumerPk1, new IntegerLiteral(6)),
producerPk, cteId),
+ newCteConsumerRuntimeFilter(src, consumerPk2,
+ new Subtract(consumerPk2, new IntegerLiteral(1)),
producerPk, cteId));
+
Assertions.assertFalse(RuntimeFilterGenerator.canPushDownRuntimeFiltersIntoCTEProducer(
+ differentTargetFilters, cteId));
+ }
+
+ private RuntimeFilter newCteConsumerRuntimeFilter(Expression src, Slot
targetSlot,
+ Expression targetExpression, Slot producerSlot, CTEId cteId) {
+ PhysicalCTEConsumer consumer = Mockito.mock(PhysicalCTEConsumer.class);
+ Mockito.when(consumer.getCteId()).thenReturn(cteId);
+
Mockito.when(consumer.getProducerSlot(targetSlot)).thenReturn(producerSlot);
+ AbstractPhysicalJoin builder =
Mockito.mock(AbstractPhysicalJoin.class);
+ return new
RuntimeFilter(RuntimeFilterId.createGenerator().getNextId(), src,
+ ImmutableList.of(targetSlot),
ImmutableList.of(targetExpression),
+ TRuntimeFilterType.IN_OR_BLOOM, 0, builder, -1L, true,
+ TMinMaxRuntimeFilterType.MIN_MAX, consumer);
+ }
+
@Test
public void testRuntimeFilterBlockByRecCte() {
String sql = new StringBuilder().append("with recursive xx as
(\n").append(" select\n")
diff --git
a/regression-test/suites/nereids_p0/runtime_filter/cte-runtime-filter.groovy
b/regression-test/suites/nereids_p0/runtime_filter/cte-runtime-filter.groovy
index 6e9393e05d2..d7a6cb2ca93 100644
--- a/regression-test/suites/nereids_p0/runtime_filter/cte-runtime-filter.groovy
+++ b/regression-test/suites/nereids_p0/runtime_filter/cte-runtime-filter.groovy
@@ -68,4 +68,56 @@ suite('cte-runtime-filter') {
from cte a
join cte_runtime_filter_table b on a.user_id=b.user_id ;
'''
-}
\ No newline at end of file
+
+ sql '''
+ drop table if exists cte_runtime_filter_shared_probe;
+ create table cte_runtime_filter_shared_probe (
+ pk int not null
+ ) ENGINE=OLAP
+ DUPLICATE KEY(pk)
+ DISTRIBUTED BY HASH(pk) BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1"
+ );
+
+ insert into cte_runtime_filter_shared_probe values (4), (11);
+
+ drop table if exists cte_runtime_filter_shared_build;
+ create table cte_runtime_filter_shared_build (
+ pk bigint not null
+ ) ENGINE=OLAP
+ DUPLICATE KEY(pk)
+ DISTRIBUTED BY HASH(pk) BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1"
+ );
+
+ insert into cte_runtime_filter_shared_build values (10);
+
+ set enable_nereids_planner=true;
+ set enable_fallback_to_original_planner=false;
+ set inline_cte_referenced_threshold=0;
+ set disable_join_reorder=true;
+ set enable_runtime_filter_prune=false;
+ set runtime_filter_mode=global;
+ set runtime_filter_wait_infinitely=true;
+ set runtime_filter_type=2;
+ '''
+
+ def sharedCteRuntimeFilterSql = '''
+ with probe as (
+ select pk from cte_runtime_filter_shared_probe
+ )
+ select count(*)
+ from probe p1
+ cross join probe p2
+ join cte_runtime_filter_shared_build b
+ on cast(p1.pk as bigint) + 6 = b.pk
+ and cast(p2.pk as bigint) - 1 = b.pk
+ '''
+ assertEquals([[1L]], sql(sharedCteRuntimeFilterSql))
+
+ sql "set runtime_filter_type=''"
+ assertEquals([[1L]], sql(sharedCteRuntimeFilterSql))
+ sql "set runtime_filter_wait_infinitely=false"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
