diff --git a/src/optimizer/graph_manager.hpp b/src/optimizer/graph_manager.hpp index d0f4b5e..44f3394 100644 --- a/src/optimizer/graph_manager.hpp +++ b/src/optimizer/graph_manager.hpp @@ -56,4 +56,9 @@ class FilterOperation { bool is_forward_pass = false; }; +struct FilterOpPair { + FilterOperation create_op; + FilterOperation probe_op; +}; + } // namespace duckdb diff --git a/src/optimizer/robust_optimizer.cpp b/src/optimizer/robust_optimizer.cpp index 2fd6dc0..9b1baf6 100644 --- a/src/optimizer/robust_optimizer.cpp +++ b/src/optimizer/robust_optimizer.cpp @@ -3,6 +3,7 @@ #include "duckdb/planner/operator/logical_get.hpp" #include "duckdb/planner/operator/logical_aggregate.hpp" #include "duckdb/planner/operator/logical_cteref.hpp" +#include "duckdb/planner/operator/logical_filter.hpp" #include "duckdb/planner/expression/bound_columnref_expression.hpp" #include "duckdb/common/types.hpp" #include "table_manager.hpp" @@ -19,6 +20,12 @@ #include "../utils/dag_printer.hpp" #include #include +#include "duckdb/storage/statistics/base_statistics.hpp" +#include "duckdb/storage/statistics/numeric_stats.hpp" +#include "duckdb/catalog/catalog_entry/duck_table_entry.hpp" +#include "duckdb/storage/data_table.hpp" +#include "duckdb/storage/table/table_statistics.hpp" +#include "duckdb/common/types/hyperloglog.hpp" namespace duckdb { // class LogicalCreateFilter; @@ -75,7 +82,36 @@ void RobustOptimizerContextState::ExtractOperatorsRecursive(LogicalOperator &pla switch (op->type) { case LogicalOperatorType::LOGICAL_FILTER: { + auto &logical_filter = op->Cast(); LogicalOperator *child = op->children[0].get(); + + // record base tables that have LOGICAL_FILTERs + if (!logical_filter.expressions.empty()) { + auto *current = child; + while (true) { + if (current->children.size() > 2 || current->children.empty()) { + break; + } else if (current->children.size() == 2) { + if (current->type != LogicalOperatorType::LOGICAL_COMPARISON_JOIN) { + break; + } + auto &join = current->Cast(); + if (join.join_type != JoinType::MARK) { + break; + } + } else if (current->children.size() == 1) { + if (current->type == LogicalOperatorType::LOGICAL_GET) { + break; + } + } + current = current->children[0].get(); + } + if (current->type == LogicalOperatorType::LOGICAL_GET) { + auto table_idx = current->Cast().table_index; + table_with_filters.insert(table_idx); + } + } + if (child->type == LogicalOperatorType::LOGICAL_GET) { table_mgr.AddTableOperator(child); return; @@ -1099,7 +1135,8 @@ RobustOptimizerContextState::GenerateStageModifications(const vector & std::pair>, unordered_map>> RobustOptimizerContextState::GenerateStageModificationsFromDAG(vector &all_nodes, - map &uf_parent) { + map &uf_parent, + vector &filter_pairs) { unordered_map> forward_filter_ops; unordered_map> backward_filter_ops; @@ -1165,6 +1202,7 @@ RobustOptimizerContextState::GenerateStageModificationsFromDAG(vectortable_op].push_back(use_op); + filter_pairs.push_back({create_op, use_op}); } } } @@ -1228,6 +1266,7 @@ RobustOptimizerContextState::GenerateStageModificationsFromDAG(vectortable_op].push_back(use_op); + filter_pairs.push_back({create_op, use_op}); } else { // new equivalence class at this edge — create BF on parent, use on child FilterOperation create_op; @@ -1251,6 +1290,7 @@ RobustOptimizerContextState::GenerateStageModificationsFromDAG(vectortable_op].push_back(use_op); + filter_pairs.push_back({create_op, use_op}); // record this as the source for this equivalence class equiv_class_bf_source[equiv_root] = {parent_node->table_op, create_idx, parent_node->table_idx, @@ -1686,6 +1726,339 @@ void RobustOptimizerContextState::LiftCreateFilterAboveFilter(unique_ptr> &filter_ops, + idx_t sequence_number) { + for (auto &entry : filter_ops) { + auto &table_op = entry.first; + auto &ops = entry.second; + for (auto &op : ops) { + if (op.sequence_number == sequence_number) { + return &op; + } + } + } + + return nullptr; +} + +static void EraseOperation(unordered_map> &filter_ops, + idx_t sequence_number) { + for (auto map_it = filter_ops.begin(); map_it != filter_ops.end(); map_it++) { + auto &ops = map_it->second; + for (auto op_it = ops.begin(); op_it != ops.end(); op_it++) { + if (op_it->sequence_number != sequence_number) { + continue; + } + ops.erase(op_it); + if (ops.empty()) { + filter_ops.erase(map_it); + } + return; + } + } +} + +// remove selected pair +static void RemoveFilterPair(const FilterOpPair &pair, + unordered_map> &forward_filter_ops, + unordered_map> &backward_filter_ops) { + auto &filter_ops = pair.create_op.is_forward_pass ? forward_filter_ops : backward_filter_ops; + + if (pair.create_op.is_forward_pass) { + EraseOperation(filter_ops, pair.create_op.sequence_number); + EraseOperation(filter_ops, pair.probe_op.sequence_number); + return; + } + + // if backward pass, CREATE needs to be partially deleted + auto *create_op = FindOperation(filter_ops, pair.create_op.sequence_number); + for (const auto &target_probe : pair.probe_op.probe_columns) { + for (idx_t i = 0; i < create_op->probe_columns.size(); i++) { + const auto ¤t_probe = create_op->probe_columns[i]; + + if (current_probe.table_index != target_probe.table_index || + current_probe.column_index != target_probe.column_index) { + continue; + } + + create_op->build_columns.erase(create_op->build_columns.begin() + (int)i); + create_op->probe_columns.erase(create_op->probe_columns.begin() + (int)i); + break; + } + } + + EraseOperation(filter_ops, pair.probe_op.sequence_number); + + // if CREATE is empty, delete it + if (create_op->probe_columns.empty()) { + EraseOperation(filter_ops, pair.create_op.sequence_number); + } +} + +unique_ptr RobustOptimizerContextState::GetColumnStatistics(const ColumnBinding &binding) { + auto &table_info = table_mgr.table_lookup[binding.table_index]; + auto *get = TableManager::FindLogicalGet(table_info.table_op); + if (!get) { + D_PRINTF("No matching base table found"); + return nullptr; + } + auto &column_id = get->GetColumnIds()[binding.column_index]; + + if (!get->function.statistics && !get->function.statistics_extended) { + return nullptr; + } + unique_ptr column_statistics; + if (get->function.statistics_extended) { + TableFunctionGetStatisticsInput input(get->bind_data.get(), column_id); + column_statistics = get->function.statistics_extended(context, input); + } else { + D_ASSERT(get->function.statistics); + column_statistics = get->function.statistics(context, get->bind_data.get(), column_id.GetPrimaryIndex()); + } + return column_statistics; +} + +idx_t RobustOptimizerContextState::GetBaseTableRowCount(const ColumnBinding &binding) { + auto resolved_binding = ResolveColumnBinding(binding); + auto table_it = table_mgr.table_lookup.find(resolved_binding.table_index); + if (table_it == table_mgr.table_lookup.end()) { + return DConstants::INVALID_INDEX; + } + auto *get = TableManager::FindLogicalGet(table_it->second.table_op); + if (!get || !get->function.cardinality) { + return DConstants::INVALID_INDEX; + } + auto table_stats = get->function.cardinality(context, get->bind_data.get()); + if (!table_stats || !table_stats->has_estimated_cardinality) { + return DConstants::INVALID_INDEX; + } + return table_stats->estimated_cardinality; +} + +unique_ptr RobustOptimizerContextState::GetColumnHLL(const ColumnBinding &binding) { + auto resolved_binding = ResolveColumnBinding(binding); + auto table_it = table_mgr.table_lookup.find(resolved_binding.table_index); + if (table_it == table_mgr.table_lookup.end()) { + return nullptr; + } + auto *get = TableManager::FindLogicalGet(table_it->second.table_op); + if (!get) { + D_PRINTF("No matching base table found"); + return nullptr; + } + auto table = get->GetTable(); + if (!table || !table->IsDuckTable()) { + return nullptr; + } + const auto &column_id = get->GetColumnIds()[resolved_binding.column_index]; + StorageIndex storage_index; + if (!get->TryGetStorageIndex(column_id, storage_index)) { + return nullptr; + } + auto &duck_table = table->Cast(); + TableStatistics table_statistics; + duck_table.GetStorage().GetRowGroupCollection()->CopyStats(table_statistics); + auto stats_lock = table_statistics.GetLock(); + auto &column_statistics = table_statistics.GetStats(*stats_lock, storage_index.GetPrimaryIndex()); + if (!column_statistics.HasDistinctStats()) { + return nullptr; + } + auto &distinct_statistics = column_statistics.DistinctStats(); + if (!distinct_statistics.log || distinct_statistics.sample_count.load() == 0 || + distinct_statistics.total_count.load() == 0) { + return nullptr; + } + return distinct_statistics.log->Copy(); +} + +// check HLL dominance as approximate evidence as whether probe side key set is contained in build side key set +bool RobustOptimizerContextState::HLLDominates(const ColumnBinding &build_binding, const ColumnBinding &probe_binding) { + auto build_hll = GetColumnHLL(build_binding); + auto probe_hll = GetColumnHLL(probe_binding); + if (!build_hll || !probe_hll) { + return false; + } + + for (idx_t i = 0; i < HyperLogLog::M; i++) { + if (build_hll->GetRegister(i) < probe_hll->GetRegister(i)) { + return false; + } + } + return true; +} + +enum class IncomingFilterStatus { NONE, ONE, MULTIPLE }; + +struct TableFilterState { + IncomingFilterStatus status = IncomingFilterStatus::NONE; + idx_t source_idx = DConstants::INVALID_INDEX; +}; + +bool RobustOptimizerContextState::HasFilteringLocalPredicate(const FilterOpPair &pair) { + const auto &pair_op = pair.probe_op; + const auto &build_columns = pair_op.build_columns; + const auto &probe_columns = pair_op.probe_columns; + auto build_table_idx = pair_op.build_table_idx; + + if (table_with_filters.count(build_table_idx)) { + return true; + } + + auto &table_info = table_mgr.table_lookup[build_table_idx]; + auto *get = TableManager::FindLogicalGet(table_info.table_op); + if (!get) { + D_PRINTF("No matching base table found"); + return true; + } + auto &column_ids = get->GetColumnIds(); + + for (auto &entry : get->table_filters.filters) { + auto &filter = *entry.second; + + if (filter.filter_type == TableFilterType::OPTIONAL_FILTER) { + continue; + } + + idx_t filter_column_index = DConstants::INVALID_INDEX; + for (idx_t i = 0; i < column_ids.size(); i++) { + if (column_ids[i].GetPrimaryIndex() == entry.first) { + filter_column_index = i; + break; + } + } + if (filter_column_index == DConstants::INVALID_INDEX) { + return true; + } + + ColumnBinding filter_binding(build_table_idx, filter_column_index); + auto build_stats = GetColumnStatistics(filter_binding); + if (!build_stats) { + return true; + } + auto build_result = filter.CheckStatistics(*build_stats); + if (build_result == FilterPropagateResult::FILTER_ALWAYS_TRUE) { + continue; + } + + auto resolved_filter_binding = ResolveColumnBinding(filter_binding); + idx_t key_index = DConstants::INVALID_INDEX; + for (idx_t i = 0; i < build_columns.size(); i++) { + if (ResolveColumnBinding(build_columns[i]) == resolved_filter_binding) { + key_index = i; + break; + } + } + if (key_index == DConstants::INVALID_INDEX) { + return true; + } + auto probe_binding = ResolveColumnBinding(probe_columns[key_index]); + auto probe_stats = GetColumnStatistics(probe_binding); + if (!probe_stats) { + return true; + } + auto probe_result = filter.CheckStatistics(*probe_stats); + if (probe_result != FilterPropagateResult::FILTER_ALWAYS_TRUE) { + return true; + } + } + + return false; +} + +bool RobustOptimizerContextState::IsRedundant(const FilterOpPair &pair) { + const auto &build_columns = pair.probe_op.build_columns; + const auto &probe_columns = pair.probe_op.probe_columns; + + for (idx_t i = 0; i < build_columns.size(); ++i) { + auto build_stats = GetColumnStatistics(build_columns[i]); + auto probe_stats = GetColumnStatistics(probe_columns[i]); + if (!build_stats || !probe_stats) { + return false; + } + if (!build_stats->GetType().IsIntegral() || !probe_stats->GetType().IsIntegral()) { + return false; + } + if (!NumericStats::HasMinMax(*build_stats) || !NumericStats::HasMinMax(*probe_stats)) { + return false; + } + + auto build_min = NumericStats::Min(*build_stats).GetValue(); + auto build_max = NumericStats::Max(*build_stats).GetValue(); + auto probe_min = NumericStats::Min(*probe_stats).GetValue(); + auto probe_max = NumericStats::Max(*probe_stats).GetValue(); + + // if probe range is outside of build range, then not redundant + if (probe_min < build_min || probe_max > build_max) { + return false; + } + auto build_range = build_max - build_min + 1; + auto build_row_count = GetBaseTableRowCount(build_columns[i]); + bool redundant_by_row_count = + build_row_count != DConstants::INVALID_INDEX && build_row_count == static_cast(build_range); + if (redundant_by_row_count) { + return true; + } + bool redundant_by_hll = HLLDominates(build_columns[i], probe_columns[i]); + if (redundant_by_hll) { + return true; + } + } + return false; +} + +void RobustOptimizerContextState::RemoveRedundantPairs( + vector &filter_pairs, unordered_map> &forward_filter_ops, + unordered_map> &backward_filter_ops) { + map table_states; + + // initialzize status of tables using local predicates + for (const auto &entry : table_mgr.table_lookup) { + table_states[entry.first] = TableFilterState(); + } + vector remove_pair(filter_pairs.size(), false); + + // iterate all pairs in sequence to mark removable pairs + for (idx_t pair_idx = 0; pair_idx < filter_pairs.size(); pair_idx++) { + const auto &pair = filter_pairs[pair_idx]; + const auto &pair_op = pair.probe_op; + auto &build_state = table_states[pair_op.build_table_idx]; + auto &probe_state = table_states[pair_op.probe_table_idx]; + bool redundant = false; + + if (build_state.status == IncomingFilterStatus::NONE && !HasFilteringLocalPredicate(pair) && + IsRedundant(pair)) { + redundant = true; + } + + // peel from turnaround + if (build_state.status == IncomingFilterStatus::ONE && build_state.source_idx == pair_op.probe_table_idx && + !HasFilteringLocalPredicate(pair) && IsRedundant(pair)) { + redundant = true; + } + + // if a pair is marked as redundant, it will not be potentially filtering source for the next pair + if (redundant) { + remove_pair[pair_idx] = true; + continue; + } + + if (probe_state.status == IncomingFilterStatus::NONE) { + probe_state.status = IncomingFilterStatus::ONE; + } else if (probe_state.status == IncomingFilterStatus::ONE) { + probe_state.status = IncomingFilterStatus::MULTIPLE; + } + probe_state.source_idx = pair_op.build_table_idx; + } + + for (idx_t pair_idx = 0; pair_idx < filter_pairs.size(); pair_idx++) { + if (!remove_pair[pair_idx]) { + continue; + } + + auto &pair = filter_pairs[pair_idx]; + RemoveFilterPair(pair, forward_filter_ops, backward_filter_ops); + } +} unique_ptr RobustOptimizerContextState::PreOptimize(unique_ptr plan) { // step 1: extract join operators vector edges = ExtractOperators(*plan); @@ -1721,6 +2094,7 @@ unique_ptr RobustOptimizerContextState::Optimize(unique_ptr uf_parent; auto all_nodes = BuildPhysicalPlanDAG(plan.get(), uf_parent); + vector filter_pairs; // flip non-largest roots to leaves (default: on) Value flip_val; @@ -1738,9 +2112,12 @@ unique_ptr RobustOptimizerContextState::Optimize(unique_ptr rename_col_bindings; + unordered_set table_with_filters; + + bool exist_cycle = false; + public: // extract all the join edges from the plan // vector ExtractOperators(LogicalOperator &plan, vector &join_ops); @@ -101,7 +106,8 @@ class RobustOptimizerContextState : public ClientContextState { std::pair>, unordered_map>> - GenerateStageModificationsFromDAG(vector &all_nodes, map &uf_parent); + GenerateStageModificationsFromDAG(vector &all_nodes, map &uf_paren, + vector &filter_pairs); unique_ptr BuildStackedBFOperators(unique_ptr base_plan, const vector &filter_ops, @@ -124,6 +130,22 @@ class RobustOptimizerContextState : public ClientContextState { // pass 2: lift BF operator block above FILTER (handles all FILTER cases) void LiftCreateFilterAboveFilter(unique_ptr &plan); + unique_ptr GetColumnStatistics(const ColumnBinding &binding); + + idx_t GetBaseTableRowCount(const ColumnBinding &binding); + + unique_ptr GetColumnHLL(const ColumnBinding &binding); + + bool HLLDominates(const ColumnBinding &build_binding, const ColumnBinding &probe_binding); + + bool HasFilteringLocalPredicate(const FilterOpPair &pair); + + bool IsRedundant(const FilterOpPair &pair); + + void RemoveRedundantPairs(vector &filter_pairs, + unordered_map> &forward_filter_ops, + unordered_map> &backward_filter_ops); + // resolve column binding through rename chain to get base table binding ColumnBinding ResolveColumnBinding(const ColumnBinding &binding) const;