From d6b6b99ea0d6c4d4af87b6b8dadb48dfd534e169 Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Mon, 6 Jul 2026 14:56:42 +0800 Subject: [PATCH] [fix](fe) Prevent unsafe CTE runtime filter pushdown ### What problem does this PR solve? Issue Number: None Related PR: #62383 Problem Summary: Runtime filters pushed from multiple CTE consumers into a shared CTE producer only checked for the same build-side source expression. After runtime filter generation was unified in #62383, OR expansion can create multiple consumer-side probe expressions such as pk + 6 and pk - 1 for the same shared producer. Mapping those filters to only the producer slot loses the expression difference and can apply one branch-specific filter to the shared scan, producing wrong results for anti joins. This patch requires all producer-side target expressions to be identical before pushing filters into the shared producer, and preserves the full producer target expression when doing the pushdown. It also adds a regression case with two shared CTE consumers whose producer-side target expressions are cast(pk as bigint) + 6 and cast(pk as bigint) - 1. ### Release note Fix incorrect query results caused by unsafe runtime filter pushdown into shared CTE producers. ### Check List (For Author) - Test: Unit Test / Regression test - ./run-fe-ut.sh --run org.apache.doris.nereids.postprocess.RuntimeFilterTest#testPushSharedCteRuntimeFilterOnlyForSameProducerTargetExpression - Added regression-test/suites/query_p0/runtime_filter/cte-runtime-filter.groovy coverage for shared CTE runtime filters with different producer-side target expressions. Not run locally because no FE was listening on the configured regression port 9030. - Behavior changed: Yes. The optimizer no longer pushes branch-specific runtime filters into a shared CTE producer when consumer probe expressions differ, which restores correct query results. - Does this need documentation: No --- .../post/RuntimeFilterGenerator.java | 43 ++++++++++++++- .../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 96cc0b0e7a1d02..22e9c00b2aaef9 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.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> SPJ_PLAN = ImmutableSet.of( PhysicalRelation.class, PhysicalProject.class, @@ -158,6 +163,9 @@ public Plan processRoot(Plan plan, CascadesContext ctx) { 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 static Slot checkTargetChild(Expression leftChild) { 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 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 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 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 @@ private boolean doPushDownIntoCTEProducerInternal(RuntimeFilter rf, Expression t } // 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 @@ private boolean doPushDownIntoCTEProducerInternal(RuntimeFilter rf, Expression t // 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 7a34e0923c6e02..6a5b858255ae1a 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.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.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 void testIgnoredRuntimeFilterIdDoesNotDropGroupedLegacyFilter() { } } + @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 sameTargetFilters = ImmutableList.of( + newCteConsumerRuntimeFilter(src, consumerPk1, consumerPk1, producerPk, cteId), + newCteConsumerRuntimeFilter(src, consumerPk2, consumerPk2, producerPk, cteId)); + Assertions.assertTrue(RuntimeFilterGenerator.canPushDownRuntimeFiltersIntoCTEProducer( + sameTargetFilters, cteId)); + + List 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 6e9393e05d21b6..d7a6cb2ca93631 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" +}