This is an automated email from the ASF dual-hosted git repository.
BiteTheDDDDt pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 271d56a379d [fix](fe) Prevent unsafe CTE runtime filter pushdown
(#65247)
271d56a379d is described below
commit 271d56a379dbb043b66c20553c8c799d7d100fa5
Author: Pxl <[email protected]>
AuthorDate: Wed Jul 8 12:29:08 2026 +0800
[fix](fe) Prevent unsafe CTE runtime filter pushdown (#65247)
Related PR: #62383
Problem Summary:
RQG found a wrong-result case in the OR expansion + anti join + shared
scan shape. A representative query is:
```sql
SET enable_sql_cache = false;
SET enable_runtime_filter_prune = false;
SELECT COUNT(*)
FROM (
SELECT *
FROM table_5 AS t1
LEFT ANTI JOIN table_9 AS t2
ON t1.pk + 0 = t2.pk + 6
OR t1.pk + 1 = t2.pk
INNER JOIN table_20 AS t3
) q;
```
With runtime filters disabled (`SET runtime_filter_type = ''`), the RQG
case returns the expected count `0`. With runtime filters enabled before
this fix, it returned `100`.
The regression was introduced by #62383, which unified runtime filter
generation and changed runtime filters to the v2-style single-target
pushdown path. That made OR-expanded branches create separate runtime
filters on CTE consumers of the same shared producer. In this case the
consumers use different probe expressions after mapping back to the
producer, such as `pk + 6` and `pk - 1`.
The shared CTE producer pushdown only checked that every consumer had a
runtime filter with the same build-side source expression. It did not
check that the consumer probe expressions mapped to the same
producer-side target expression. It then pushed the filter using only
the producer slot, losing the `+ 6` / `- 1` expression difference. As a
result, one branch-specific runtime filter could be applied to the
shared scan and affect the other branch, causing the anti join to return
wrong results.
The added regression case uses a shared CTE with two consumers and a
build row `10`; one consumer needs `cast(pk as bigint) + 6 = 10`, the
other needs `cast(pk as bigint) - 1 = 10`. The correct count is `1` both
with runtime filters enabled and disabled.
This patch only pushes runtime filters into a shared CTE producer when
all candidate filters map to the same producer-side target expression,
and it preserves that full producer target expression during pushdown.
### Release note
Fix incorrect query results caused by unsafe runtime filter pushdown
into shared CTE producers.
---
.../processor/post/RuntimeFilterGenerator.java | 43 ++++++++++++++++-
.../nereids/postprocess/RuntimeFilterTest.java | 46 ++++++++++++++++++
.../runtime_filter/cte-runtime-filter.groovy | 54 +++++++++++++++++++++-
3 files changed, 140 insertions(+), 3 deletions(-)
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 96cc0b0e7a1..22e9c00b2aa 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,12 +60,15 @@ import org.apache.doris.statistics.Statistics;
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.ImmutableList;
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.HashMap;
import java.util.HashSet;
@@ -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,
@@ -158,6 +163,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
@@ -872,6 +880,35 @@ 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) {
+ PhysicalRelation rel = rf.getTargetScan();
+ Preconditions.checkArgument(rel instanceof PhysicalCTEConsumer
+ && ((PhysicalCTEConsumer) rel).getCteId().equals(cteId));
+ PhysicalCTEConsumer consumer = (PhysicalCTEConsumer) rel;
+ Map<Expression, Expression> replaceMap = Maps.newHashMap();
+ for (Slot slot : rf.getTargetExpression().getInputSlots()) {
+ replaceMap.put(slot, consumer.getProducerSlot(slot));
+ }
+ return ExpressionUtils.replace(rf.getTargetExpression(), replaceMap);
+ }
+
private boolean doPushDownIntoCTEProducerInternal(RuntimeFilter rf,
Expression targetExpression,
RuntimeFilterContext ctx,
PhysicalCTEProducer cteProducer) {
PhysicalPlan inputPlanNode = (PhysicalPlan) cteProducer.child(0);
@@ -895,7 +932,9 @@ public class RuntimeFilterGenerator extends
PlanPostProcessor {
}
// Map consumer slot to producer slot
Slot producerSlot = cteConsumer.getProducerSlot(consumerSlot);
- if (producerSlot == null) {
+ Expression producerTargetExpression = getProducerTargetExpression(rf,
cteProducer.getCteId());
+ Slot producerTargetSlot = checkTargetChild(producerTargetExpression);
+ if (!producerSlot.equals(producerTargetSlot)) {
return false;
}
if (!checkCanPushDownIntoBasicTable(inputPlanNode)) {
@@ -904,7 +943,7 @@ public class RuntimeFilterGenerator extends
PlanPostProcessor {
// Use the PushDownVisitor to push inside the CTE producer subtree
RuntimeFilterPushDownVisitor.PushDownContext pushDownContext =
RuntimeFilterPushDownVisitor.PushDownContext.createPushDownContext(
- ctx, rf.getBuilderNode(), rf.getSrcExpr(),
producerSlot,
+ ctx, rf.getBuilderNode(), rf.getSrcExpr(),
producerTargetExpression,
rf.getType(), rf.gettMinMaxType(),
!rf.isBloomFilterSizeCalculatedByNdv(),
rf.getBuildSideNdv(), rf.getExprOrder());
if (pushDownContext.isValid()) {
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 7a34e0923c6..6a5b858255a 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,12 +28,19 @@ 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;
@@ -41,20 +48,25 @@ 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.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.PhysicalSetOperation;
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 org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
import java.util.ArrayList;
import java.util.List;
@@ -616,6 +628,40 @@ public class RuntimeFilterTest extends SSBTestBase {
}
}
+ @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);
+ AbstractPhysicalPlan builder =
Mockito.mock(AbstractPhysicalPlan.class);
+ return new
RuntimeFilter(RuntimeFilterId.createGenerator().getNextId(), src, targetSlot,
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/query_p0/runtime_filter/cte-runtime-filter.groovy
b/regression-test/suites/query_p0/runtime_filter/cte-runtime-filter.groovy
index 6e9393e05d2..d7a6cb2ca93 100644
--- a/regression-test/suites/query_p0/runtime_filter/cte-runtime-filter.groovy
+++ b/regression-test/suites/query_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]
