Skip to content

[fix](fe) Prevent unsafe CTE runtime filter pushdown #65247

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
BiteTheDDDDt wants to merge 1 commit into
base: master
Choose a base branch
from
+140 −3
Open

[fix](fe) Prevent unsafe CTE runtime filter pushdown #65247

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 41 additions & 2 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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<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);
Expand All @@ -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)) {
Expand All @@ -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()) {
Expand Down
46 changes: 46 additions & 0 deletions fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,33 +28,45 @@
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.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;
Expand Down Expand Up @@ -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<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")
Expand Down
54 changes: 53 additions & 1 deletion regression-test/suites/query_p0/runtime_filter/cte-runtime-filter.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,4 +68,56 @@ suite('cte-runtime-filter') {
from cte a
join cte_runtime_filter_table b on a.user_id=b.user_id ;
'''
}

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))

@github-actions github-actions Bot Jul 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This new wrong-result regression should be recorded as a named qt_/order_qt_ regression output instead of a direct assertEquals(sql(...)) assertion. The suite already has regression-test/data/query_p0/runtime_filter/cte-runtime-filter.out, but this PR does not update that file, so the new expected count is not part of the generated expected-output artifact. Please convert these checks to named qt_ blocks and include the regenerated .out output.


sql "set runtime_filter_type=''"
assertEquals([[1L]], sql(sharedCteRuntimeFilterSql))
sql "set runtime_filter_wait_infinitely=false"
}
Loading