Merge pull request ClickHouse#80329 from ClickHouse/do-not-push-down-non-deterministic-function-for-steps-which-change-rows

Disable filter-push-down for the predicate with non-deterministic fun

Author: KochetovNicolai

src/Interpreters/ActionsDAG.cpp

@@ -2357,7 +2357,7 @@ struct ConjunctionNodes
 /// Assuming predicate is a conjunction (probably, trivial).
 /// Find separate conjunctions nodes. Split nodes into allowed and rejected sets.
 /// Allowed predicate is a predicate which can be calculated using only nodes from the allowed_nodes set.
-ConjunctionNodes getConjunctionNodes(ActionsDAG::Node * predicate, std::unordered_set<const ActionsDAG::Node *> allowed_nodes)
+ConjunctionNodes getConjunctionNodes(ActionsDAG::Node * predicate, std::unordered_set<const ActionsDAG::Node *> allowed_nodes, bool allow_non_deterministic_functions)
 {
     ConjunctionNodes conjunction;
     std::unordered_set<const ActionsDAG::Node *> allowed;
@@ -2428,7 +2428,13 @@ ConjunctionNodes getConjunctionNodes(ActionsDAG::Node * predicate, std::unordere
         {
             if (cur.num_allowed_children == cur.node->children.size())
             {
-                if (cur.node->type != ActionsDAG::ActionType::ARRAY_JOIN && cur.node->type != ActionsDAG::ActionType::INPUT)
+                bool is_deprecated_function = !allow_non_deterministic_functions
+                    && cur.node->type == ActionsDAG::ActionType::FUNCTION
+                    && !cur.node->function_base->isDeterministicInScopeOfQuery();
+
+                if (cur.node->type != ActionsDAG::ActionType::ARRAY_JOIN
+                    && cur.node->type != ActionsDAG::ActionType::INPUT
+                    && !is_deprecated_function)
                     allowed_nodes.emplace(cur.node);
             }
 
@@ -2578,7 +2584,8 @@ std::optional<ActionsDAG> ActionsDAG::splitActionsForFilterPushDown(
     const std::string & filter_name,
     bool removes_filter,
     const Names & available_inputs,
-    const ColumnsWithTypeAndName & all_inputs)
+    const ColumnsWithTypeAndName & all_inputs,
+    bool allow_non_deterministic_functions)
 {
     Node * predicate = const_cast<Node *>(tryFindInOutputs(filter_name));
     if (!predicate)
@@ -2611,7 +2618,7 @@ std::optional<ActionsDAG> ActionsDAG::splitActionsForFilterPushDown(
         }
     }
 
-    auto conjunction = getConjunctionNodes(predicate, allowed_nodes);
+    auto conjunction = getConjunctionNodes(predicate, allowed_nodes, allow_non_deterministic_functions);
 
     if (conjunction.allowed.empty())
         return {};
@@ -2688,9 +2695,9 @@ ActionsDAG::ActionsForJOINFilterPushDown ActionsDAG::splitActionsForJOINFilterPu
     auto right_stream_allowed_nodes = get_input_nodes(right_stream_available_columns_to_push_down);
     auto both_streams_allowed_nodes = get_input_nodes(equivalent_columns_to_push_down);
 
-    auto left_stream_push_down_conjunctions = getConjunctionNodes(predicate, left_stream_allowed_nodes);
-    auto right_stream_push_down_conjunctions = getConjunctionNodes(predicate, right_stream_allowed_nodes);
-    auto both_streams_push_down_conjunctions = getConjunctionNodes(predicate, both_streams_allowed_nodes);
+    auto left_stream_push_down_conjunctions = getConjunctionNodes(predicate, left_stream_allowed_nodes, false);
+    auto right_stream_push_down_conjunctions = getConjunctionNodes(predicate, right_stream_allowed_nodes, false);
+    auto both_streams_push_down_conjunctions = getConjunctionNodes(predicate, both_streams_allowed_nodes, false);
 
     NodeRawConstPtrs left_stream_allowed_conjunctions = std::move(left_stream_push_down_conjunctions.allowed);
     NodeRawConstPtrs right_stream_allowed_conjunctions = std::move(right_stream_push_down_conjunctions.allowed);

src/Interpreters/ActionsDAG.h

@@ -390,7 +390,8 @@ class ActionsDAG
         const std::string & filter_name,
         bool removes_filter,
         const Names & available_inputs,
-        const ColumnsWithTypeAndName & all_inputs);
+        const ColumnsWithTypeAndName & all_inputs,
+        bool allow_non_deterministic_functions);
 
     struct ActionsForJOINFilterPushDown;
 

src/Processors/QueryPlan/Optimizations/filterPushDown.cpp

@@ -104,7 +104,7 @@ static NameSet findIdentifiersOfNode(const ActionsDAG::Node * node)
     return res;
 }
 
-static std::optional<ActionsDAG> splitFilter(QueryPlan::Node * parent_node, const Names & available_inputs, size_t child_idx = 0)
+static std::optional<ActionsDAG> splitFilter(QueryPlan::Node * parent_node, bool step_changes_the_number_of_rows, const Names & available_inputs, size_t child_idx = 0)
 {
     QueryPlan::Node * child_node = parent_node->children.front();
     checkChildrenSize(child_node, child_idx + 1);
@@ -118,7 +118,8 @@ static std::optional<ActionsDAG> splitFilter(QueryPlan::Node * parent_node, cons
     bool removes_filter = filter->removesFilterColumn();
 
     const auto & all_inputs = child->getInputHeaders()[child_idx].getColumnsWithTypeAndName();
-    return expression.splitActionsForFilterPushDown(filter_column_name, removes_filter, available_inputs, all_inputs);
+    bool allow_deterministic_functions = !step_changes_the_number_of_rows;
+    return expression.splitActionsForFilterPushDown(filter_column_name, removes_filter, available_inputs, all_inputs, allow_deterministic_functions);
 }
 
 static size_t
@@ -186,23 +187,23 @@ addNewFilterStepOrThrow(QueryPlan::Node * parent_node, QueryPlan::Nodes & nodes,
 }
 
 static size_t
-tryAddNewFilterStep(QueryPlan::Node * parent_node, QueryPlan::Nodes & nodes, const Names & allowed_inputs,
+tryAddNewFilterStep(QueryPlan::Node * parent_node, bool step_changes_the_number_of_rows, QueryPlan::Nodes & nodes, const Names & allowed_inputs,
                     bool can_remove_filter = true, size_t child_idx = 0)
 {
-    if (auto split_filter = splitFilter(parent_node, allowed_inputs, child_idx))
+    if (auto split_filter = splitFilter(parent_node, step_changes_the_number_of_rows, allowed_inputs, child_idx))
         return addNewFilterStepOrThrow(parent_node, nodes, std::move(*split_filter), can_remove_filter, child_idx);
     return 0;
 }
 
 
 /// Push down filter through specified type of step
 template <typename Step>
-static size_t simplePushDownOverStep(QueryPlan::Node * parent_node, QueryPlan::Nodes & nodes, QueryPlanStepPtr & child)
+static size_t simplePushDownOverStep(QueryPlan::Node * parent_node, bool step_changes_the_number_of_rows, QueryPlan::Nodes & nodes, QueryPlanStepPtr & child)
 {
     if (typeid_cast<Step *>(child.get()))
     {
         Names allowed_inputs = child->getOutputHeader().getNames();
-        if (auto updated_steps = tryAddNewFilterStep(parent_node, nodes, allowed_inputs))
+        if (auto updated_steps = tryAddNewFilterStep(parent_node, step_changes_the_number_of_rows, nodes, allowed_inputs))
             return updated_steps;
     }
     return 0;
@@ -507,7 +508,7 @@ size_t tryPushDownFilter(QueryPlan::Node * parent_node, QueryPlan::Nodes & nodes
         bool filter_is_not_among_aggregates_arguments = merging_aggregated || filterColumnIsNotAmongAggregatesArguments(params.aggregates, filter->getFilterColumnName());
         const bool can_remove_filter = filter_column_is_not_among_aggregation_keys && filter_is_not_among_aggregates_arguments;
 
-        if (auto updated_steps = tryAddNewFilterStep(parent_node, nodes, keys, can_remove_filter))
+        if (auto updated_steps = tryAddNewFilterStep(parent_node, true, nodes, keys, can_remove_filter))
             return updated_steps;
     }
 
@@ -568,7 +569,7 @@ size_t tryPushDownFilter(QueryPlan::Node * parent_node, QueryPlan::Nodes & nodes
         ///    ) group by y with totals) where y != 2`
         /// Optimization will replace totals row `y, sum(x)` from `(0, 45)` to `(0, 37)`.
         /// It is expected to ok, cause AST optimization `enable_optimize_predicate_expression = 1` also brakes it.
-        if (auto updated_steps = tryAddNewFilterStep(parent_node, nodes, keys))
+        if (auto updated_steps = tryAddNewFilterStep(parent_node, false, nodes, keys))
             return updated_steps;
     }
 
@@ -584,11 +585,11 @@ size_t tryPushDownFilter(QueryPlan::Node * parent_node, QueryPlan::Nodes & nodes
             if (!keys_set.contains(column.name))
                 allowed_inputs.push_back(column.name);
 
-        if (auto updated_steps = tryAddNewFilterStep(parent_node, nodes, allowed_inputs))
+        if (auto updated_steps = tryAddNewFilterStep(parent_node, true, nodes, allowed_inputs))
             return updated_steps;
     }
 
-    if (auto updated_steps = simplePushDownOverStep<DistinctStep>(parent_node, nodes, child))
+    if (auto updated_steps = simplePushDownOverStep<DistinctStep>(parent_node, true, nodes, child))
         return updated_steps;
 
     if (auto updated_steps = tryPushDownOverJoinStep(parent_node, nodes, child))
@@ -619,14 +620,14 @@ size_t tryPushDownFilter(QueryPlan::Node * parent_node, QueryPlan::Nodes & nodes
         bool can_remove_filter = sort_description_it == sort_description.end();
 
         Names allowed_inputs = child->getOutputHeader().getNames();
-        if (auto updated_steps = tryAddNewFilterStep(parent_node, nodes, allowed_inputs, can_remove_filter))
+        if (auto updated_steps = tryAddNewFilterStep(parent_node, false, nodes, allowed_inputs, can_remove_filter))
             return updated_steps;
     }
 
     if (typeid_cast<CustomMetricLogViewStep *>(child.get()))
     {
         Names allowed_inputs = {"event_date", "event_time", "hostname"};
-        if (auto updated_steps = tryAddNewFilterStep(parent_node, nodes, allowed_inputs, true))
+        if (auto updated_steps = tryAddNewFilterStep(parent_node, true, nodes, allowed_inputs, true))
             return updated_steps;
     }
 
@@ -636,7 +637,7 @@ size_t tryPushDownFilter(QueryPlan::Node * parent_node, QueryPlan::Nodes & nodes
         bool can_remove_filter = !join_filter_set_step->isColumnPartOfSetKey(filter_column_name);
 
         Names allowed_inputs = child->getOutputHeader().getNames();
-        if (auto updated_steps = tryAddNewFilterStep(parent_node, nodes, allowed_inputs, can_remove_filter))
+        if (auto updated_steps = tryAddNewFilterStep(parent_node, false, nodes, allowed_inputs, can_remove_filter))
             return updated_steps;
     }
 

tests/queries/0_stateless/03513_filter_push_down_rand_bug.reference

@@ -0,0 +1 @@
+0

tests/queries/0_stateless/03513_filter_push_down_rand_bug.sql

@@ -0,0 +1,21 @@
+drop table if exists users_items;
+CREATE TABLE users_items (user_id UInt64) ENGINE = Log;
+INSERT INTO users_items SELECT bitAnd(number, 15) from numbers(64);
+
+SELECT sum(in_sample)
+FROM
+(
+    WITH RandomUsers AS
+        (
+            SELECT
+                user_id,
+                rand() % 2 AS in_sample
+            FROM users_items
+            GROUP BY user_id
+        )
+    SELECT
+        user_id,
+        in_sample
+    FROM RandomUsers
+    WHERE in_sample = 0
+);