Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,9 @@ public Rule build() {
LogicalCTEConsumer cteConsumer = filter.child();
Set<Expression> exprs = filter.getConjuncts();
for (Expression expr : exprs) {
if (expr.containsUniqueFunction()) {
continue;
}
Expression rewrittenExpr = expr.rewriteUp(e -> {
if (e instanceof Slot) {
return cteConsumer.getProducerSlot((Slot) e);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,8 @@ public Rule build() {
Set<Expression> remainPredicates = Sets.newHashSet();
filter.getConjuncts().forEach(conjunct -> {
Set<Slot> conjunctSlots = conjunct.getInputSlots();
if (!conjunctSlots.isEmpty() && childOutputs.containsAll(conjunctSlots)) {
if (!conjunctSlots.isEmpty() && childOutputs.containsAll(conjunctSlots)
&& !conjunct.containsUniqueFunction()) {
pushDownPredicates.add(conjunct);
} else {
remainPredicates.add(conjunct);
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
// 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.doris.nereids.rules.rewrite;

import org.apache.doris.nereids.CascadesContext;
import org.apache.doris.nereids.rules.Rule;
import org.apache.doris.nereids.trees.expressions.CTEId;
import org.apache.doris.nereids.trees.expressions.EqualTo;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.GreaterThan;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.functions.scalar.Random;
import org.apache.doris.nereids.trees.expressions.literal.DoubleLiteral;
import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
import org.apache.doris.nereids.trees.plans.RelationId;
import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
import org.apache.doris.nereids.util.MemoTestUtils;
import org.apache.doris.nereids.util.PlanConstructor;
import org.apache.doris.qe.ConnectContext;

import com.google.common.collect.ImmutableSet;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

import java.util.Map;
import java.util.Set;

/**
* Tests for {@link CollectFilterAboveConsumer}.
*/
class CollectFilterAboveConsumerTest {

/**
* Filter has two conjuncts: one deterministic, one containing a unique (non-idempotent) function.
* After applying the rule, only the deterministic conjunct should be collected into
* consumerIdToFilter. Collecting the unique-function conjunct would cause it to be pushed into
* the CTE producer while still remaining on the consumer side, resulting in double execution
* (bug DORIS-25202).
*/
@Test
void testDoNotCollectUniqueFunctionConjunct() {
ConnectContext connectContext = new ConnectContext();
LogicalOlapScan producerPlan = PlanConstructor.newLogicalOlapScan(0, "t1", 0);
LogicalCTEConsumer consumer = new LogicalCTEConsumer(
PlanConstructor.getNextRelationId(), new CTEId(1), "cte1", producerPlan);

Slot idSlot = consumer.getOutput().get(0);
Expression deterministic = new EqualTo(idSlot, new IntegerLiteral(1));
Expression uniqueFn = new GreaterThan(new Random(), new DoubleLiteral(0.5));
LogicalFilter<LogicalCTEConsumer> filter = new LogicalFilter<>(
ImmutableSet.of(deterministic, uniqueFn), consumer);

CascadesContext cascadesContext = MemoTestUtils.createCascadesContext(connectContext, filter);
Rule rule = new CollectFilterAboveConsumer().build();
rule.transform(filter, cascadesContext);

Map<RelationId, Set<Expression>> collected = cascadesContext.getStatementContext()
.getConsumerIdToFilters();
Set<Expression> filters = collected.get(consumer.getRelationId());
Assertions.assertNotNull(filters, "deterministic conjunct must be collected");
// Only the deterministic conjunct (after slot-rewriting to producer slot) should be present,
// with the unique-function conjunct skipped.
Assertions.assertEquals(1, filters.size(),
"exactly one conjunct (the deterministic one) should be collected, "
+ "unique-function conjunct must NOT be collected");
Expression onlyCollected = filters.iterator().next();
Assertions.assertFalse(onlyCollected.containsUniqueFunction(),
"collected conjunct must not contain a unique function");
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
// 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.doris.nereids.rules.rewrite;

import org.apache.doris.nereids.trees.expressions.EqualTo;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.functions.generator.Unnest;
import org.apache.doris.nereids.trees.expressions.functions.scalar.Random;
import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
import org.apache.doris.nereids.trees.plans.logical.LogicalGenerate;
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.types.IntegerType;
import org.apache.doris.nereids.util.MemoPatternMatchSupported;
import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.nereids.util.PlanConstructor;
import org.apache.doris.qe.ConnectContext;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import org.junit.jupiter.api.Test;

/**
* Tests for {@link PushDownFilterThroughGenerate}.
*/
class PushDownFilterThroughGenerateTest implements MemoPatternMatchSupported {

private final LogicalOlapScan scan = PlanConstructor.newLogicalOlapScan(0, "t1", 0);

/** Deterministic filter whose slots only reference generate's child output should be pushed down. */
@Test
void testPushDownDeterministicFilter() {
SlotReference idSlot = (SlotReference) scan.getOutput().get(0);
SlotReference genOut = new SlotReference("g1", IntegerType.INSTANCE);
Unnest gen = new Unnest(new IntegerLiteral(1));
LogicalGenerate<LogicalPlan> generate = new LogicalGenerate<>(
ImmutableList.of(gen), ImmutableList.of(genOut), scan);

Expression predicate = new EqualTo(idSlot, new IntegerLiteral(1));
LogicalFilter<LogicalPlan> filter = new LogicalFilter<>(ImmutableSet.of(predicate), generate);

PlanChecker.from(new ConnectContext(), filter)
.applyTopDown(new PushDownFilterThroughGenerate())
.matchesFromRoot(
logicalGenerate(
logicalFilter(logicalOlapScan())
.when(f -> f.getConjuncts().contains(predicate))
)
);
}

/** Filter containing a unique (non-idempotent) function must NOT be pushed through generate. */
@Test
void testDoNotPushUniqueFunctionThroughGenerate() {
SlotReference idSlot = (SlotReference) scan.getOutput().get(0);
SlotReference genOut = new SlotReference("g1", IntegerType.INSTANCE);
Unnest gen = new Unnest(new IntegerLiteral(1));
LogicalGenerate<LogicalPlan> generate = new LogicalGenerate<>(
ImmutableList.of(gen), ImmutableList.of(genOut), scan);

// random() depends only on idSlot-like pattern but contains a UniqueFunction.
// Use EqualTo(id, random()) so the conjunct's input slots are a subset of scan output
// (would otherwise satisfy the push-down condition).
Expression predicate = new EqualTo(idSlot, new Random());
LogicalFilter<LogicalPlan> filter = new LogicalFilter<>(ImmutableSet.of(predicate), generate);

PlanChecker.from(new ConnectContext(), filter)
.applyTopDown(new PushDownFilterThroughGenerate())
.matchesFromRoot(
logicalFilter(
logicalGenerate(logicalOlapScan())
).when(f -> f.getConjuncts().contains(predicate))
);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
-- This file is automatically generated. You should know what you did if you want to edit this
-- !collect_filter_above_consumer_unique_1 --
PhysicalCteAnchor ( cteId=CTEId#0 )
--PhysicalCteProducer ( cteId=CTEId#0 )
----PhysicalOlapScan[t1]
--PhysicalResultSink
----PhysicalUnion
------filter((random() > 0.1))
--------PhysicalCteConsumer ( cteId=CTEId#0 )
------filter((random() > 0.2))
--------PhysicalCteConsumer ( cteId=CTEId#0 )

-- !collect_filter_above_consumer_unique_2 --
PhysicalCteAnchor ( cteId=CTEId#0 )
--PhysicalCteProducer ( cteId=CTEId#0 )
----filter((cte1.id > 10))
------PhysicalOlapScan[t1]
--PhysicalResultSink
----PhysicalUnion
------filter((cte1.id > 10) and (random() > 0.1))
--------PhysicalCteConsumer ( cteId=CTEId#0 )
------filter((cte1.id > 100) and (random() > 0.2))
--------PhysicalCteConsumer ( cteId=CTEId#0 )

Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
-- This file is automatically generated. You should know what you did if you want to edit this
-- !filter_through_generate_unique_1 --
PhysicalResultSink
--PhysicalProject[tmp1.e1]
----filter((random() > 0.1))
------PhysicalGenerate
--------PhysicalProject[1 AS `1`]
----------PhysicalStorageLayerAggregate[t1]

-- !filter_through_generate_unique_2 --
PhysicalResultSink
--PhysicalProject[tmp1.e1]
----filter(((cast(id as BIGINT) + random(1, 100)) > 5))
------PhysicalGenerate
--------PhysicalProject[t1.id]
----------filter((t1.id > 10))
------------PhysicalOlapScan[t1]

-- !filter_through_generate_unique_3 --
PhysicalResultSink
--PhysicalProject[tmp1.e1]
----filter(((cast(id as BIGINT) + random(1, 100)) > 5))
------PhysicalGenerate
--------PhysicalProject[t1.id]
----------PhysicalOlapScan[t1]

Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
// 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.

suite('collect_filter_above_consumer_with_unique_function') {
sql 'SET enable_nereids_planner=true'
sql 'SET runtime_filter_mode=OFF'
sql 'SET enable_fallback_to_original_planner=false'
// keep CTE materialized so CollectFilterAboveConsumer actually runs.
sql 'SET inline_cte_referenced_threshold=0'
sql "SET ignore_shape_nodes='PhysicalDistribute'"
sql "SET detail_shape_nodes='PhysicalProject'"
sql 'SET disable_nereids_rules=PRUNE_EMPTY_PARTITION'

// filter with unique function above a CTE consumer must NOT be collected and
// pushed into the CTE producer; otherwise the producer would be re-filtered
// and other consumers would see inconsistent rows.
qt_collect_filter_above_consumer_unique_1 '''
explain shape plan
with cte1 as (select id, msg from t1)
select * from cte1 where rand() > 0.1
union all
select * from cte1 where rand() > 0.2
'''

// mixed conjuncts: deterministic parts can be collected and pushed into the CTE producer,
// unique parts must stay above the consumer only. Use predicates that survive simplification
// so the collected-into-producer OR shape is visible.
qt_collect_filter_above_consumer_unique_2 '''
explain shape plan
with cte1 as (select id, msg from t1)
select * from cte1 where id > 10 and rand() > 0.1
union all
select * from cte1 where id > 100 and rand() > 0.2
'''
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
// 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.

suite('push_down_filter_through_generate_with_unique_function') {
sql 'SET enable_nereids_planner=true'
sql 'SET runtime_filter_mode=OFF'
sql 'SET enable_fallback_to_original_planner=false'
sql "SET ignore_shape_nodes='PhysicalDistribute'"
sql "SET detail_shape_nodes='PhysicalProject'"
sql 'SET disable_nereids_rules=PRUNE_EMPTY_PARTITION'

// unique function in filter must NOT be pushed below LogicalGenerate,
// otherwise call count / state of the unique function changes semantically.
qt_filter_through_generate_unique_1 '''
explain shape plan
select e1 from t1 lateral view explode_numbers(3) tmp1 as e1
where rand() > 0.1
'''

// mixed: deterministic conjunct pushable, unique conjunct in same WHERE must stay above generate.
// Here `t1.id > 10` should be pushed below generate while `t1.id + rand(1,100) > 5` (a conjunct
// that does reference a base slot, so old code would have pushed it) must stay above.
qt_filter_through_generate_unique_2 '''
explain shape plan
select e1 from t1 lateral view explode_numbers(3) tmp1 as e1
where t1.id > 10 and t1.id + rand(1,100) > 5
'''

// unique function combined with base slot in the same conjunct.
qt_filter_through_generate_unique_3 '''
explain shape plan
select e1 from t1 lateral view explode_numbers(3) tmp1 as e1
where t1.id + rand(1,100) > 5
'''
}
Loading