fix

Signed-off-by: Mikhail Kot <to@myrrc.dev>

Author: myrrc

vortex-duckdb/cpp/include/duckdb_vx/table_function.h

@@ -42,7 +42,7 @@ void duckdb_vx_string_map_insert(duckdb_vx_string_map map, const char *key, cons
 
 // Input data passed into the init_global and init_local callbacks.
 typedef struct {
-    const void *bind_data;
+    void *bind_data;
 
     /**
      * Projected columns that are requested to be read. These are not

vortex-duckdb/cpp/table_function.cpp

@@ -112,14 +112,27 @@ unique_ptr<BaseStatistics> numeric_stats(duckdb_column_statistics &stats, Logica
 
 unique_ptr<BaseStatistics> string_stats(duckdb_column_statistics &stats, LogicalType type) {
     BaseStatistics out = StringStats::CreateUnknown(type);
-    if (stats.min) {
+    if (stats.min && stats.max) {
+        const std::string &min = StringValue::Get(UnwrapValue(stats.min));
+        StringStats::SetMin(out, min);
+
+        const std::string &max = StringValue::Get(UnwrapValue(stats.max));
+        StringStats::SetMax(out, max);
+
+        if (min == max) {
+            out.SetDistinctCount(1);
+        }
+
+        duckdb_destroy_value(&stats.min);
+        duckdb_destroy_value(&stats.max);
+    } else if (stats.min) {
         StringStats::SetMin(out, StringValue::Get(UnwrapValue(stats.min)));
         duckdb_destroy_value(&stats.min);
-    }
-    if (stats.max) {
+    } else if (stats.max) {
         StringStats::SetMax(out, StringValue::Get(UnwrapValue(stats.max)));
         duckdb_destroy_value(&stats.max);
     }
+
     if (stats.max_string_length >> 63) {
         StringStats::SetMaxStringLength(out, uint32_t(stats.max_string_length));
     }

vortex-duckdb/src/datasource.rs

@@ -7,6 +7,7 @@
 //! to get a blanket [`TableFunction`] implementation covering init, scan, progress, filter
 //! pushdown, cardinality, and partitioning.
 
+use std::cmp::max;
 use std::fmt::Debug;
 use std::ops::Range;
 use std::sync::Arc;
@@ -74,6 +75,7 @@ use crate::duckdb::DuckdbStringMapRef;
 use crate::duckdb::ExpressionRef;
 use crate::duckdb::LogicalType;
 use crate::duckdb::PartitionData;
+use crate::duckdb::TableFilterClass;
 use crate::duckdb::TableFilterSetRef;
 use crate::duckdb::TableFunction;
 use crate::duckdb::TableInitInput;
@@ -118,6 +120,7 @@ pub struct DataSourceBindData {
     data_source: Arc<MultiLayoutDataSource>,
     filter_exprs: Vec<Expression>,
     column_fields: Vec<DuckdbField>,
+    has_non_optional_filter: bool,
 }
 
 impl Clone for DataSourceBindData {
@@ -127,6 +130,7 @@ impl Clone for DataSourceBindData {
             // filter_exprs are consumed once in `init_global`.
             filter_exprs: vec![],
             column_fields: self.column_fields.clone(),
+            has_non_optional_filter: self.has_non_optional_filter,
         }
     }
 }
@@ -252,6 +256,23 @@ impl ColumnStatisticsAggregate {
     }
 }
 
+// Duckdb requires post-filter cardinality estimates, otherwise join
+// planner may flip join sides which is a huge regression for some
+// queries i.e. 1000x for tpcds 85.
+//
+// See duckdb/src/optimizer/join_order/relation_statistics_helper.cpp
+const DEFAULT_SELECTIVITY: f64 = 0.2;
+fn postfilter_cardinality(initial_cardinality: u64, has_non_optional_filter: bool) -> u64 {
+    if has_non_optional_filter {
+        let post_cardinality = initial_cardinality as f64 * DEFAULT_SELECTIVITY;
+        // Clamp intentionally
+        let post_cardinality: u64 = post_cardinality.as_();
+        max(1, post_cardinality)
+    } else {
+        initial_cardinality
+    }
+}
+
 impl<T: DataSourceTableFunction> TableFunction for T {
     type BindData = DataSourceBindData;
     type GlobalState = DataSourceGlobal;
@@ -275,6 +296,7 @@ impl<T: DataSourceTableFunction> TableFunction for T {
             data_source: Arc::new(data_source),
             filter_exprs: vec![],
             column_fields,
+            has_non_optional_filter: false,
         })
     }
 
@@ -297,13 +319,15 @@ impl<T: DataSourceTableFunction> TableFunction for T {
             row_range,
             file_selection,
             file_range,
+            has_non_optional_filter,
         } = extract_table_filter_expr(
             init_input.table_filter_set(),
             column_ids,
             &bind_data.column_fields,
             &bind_data.filter_exprs,
             bind_data.data_source.dtype(),
         )?;
+        bind_data.has_non_optional_filter |= has_non_optional_filter;
 
         debug!(
             %projection,
@@ -502,22 +526,37 @@ impl<T: DataSourceTableFunction> TableFunction for T {
         expr: &ExpressionRef,
     ) -> VortexResult<bool> {
         debug!(%expr, "pushing down expression");
+
         let Some(expr) = try_from_bound_expression(expr)? else {
             debug!(%expr, "failed to push down expression");
             return Ok(false);
         };
-        debug!(%expr, "pushed down expression");
-        bind_data.filter_exprs.push(expr);
 
-        // NOTE(ngates): Vortex does indeed run exact filters, so in theory we should return `true`
-        //  here to tell DuckDB we've handled the filter. However, DuckDB applies some crude
-        //  cardinality estimation heuristics (e.g. an equality filter => 20% selectivity) that
-        //  means by returning false, DuckDB runs an additional filter (a little bit of overhead)
-        //  but tends to end up with a better query plan.
-        //  If we plumb row count estimation into the layout tree, perhaps we could use zone maps
-        //  etc. to return estimates. But this function is probably called too late anyway. Maybe
-        //  we need our own cardinality heuristics.
-        Ok(false)
+        // Duckdb calls pushdown_complex_filter during planning phase.
+        // If all filters are pushed down, duckdb enables a LEFT_DELIM_JOIN ->
+        // COMPARISON_JOIN (HASH_JOIN) optimization:
+        // duckdb/src/optimizer/deliminator.cpp: Deliminator::HasSelection,
+        // Deliminator::Optimize.
+        //
+        // This leads to a massive regression on tpch sf=10 q17 and other
+        // benchmarks.
+        //
+        // This bug is reported to Duckdb
+        // https://github.com/duckdb/duckdb/issues/22669
+        //
+        // As a hack, report first filter as not pushed.
+        // As pushdown_complex_filter is called during planning phase,
+        // no table filters are pushed yet (this will happen in init_global),
+        // so our first filter is a first filter indeed.
+        let report_pushed = bind_data.has_non_optional_filter;
+
+        // Only table filters may be optional, any complex filter is
+        // non-optional by definition.
+        bind_data.has_non_optional_filter = true;
+
+        debug!(%expr, report_pushed, "pushed down expression");
+        bind_data.filter_exprs.push(expr);
+        Ok(report_pushed)
     }
 
     /// Get column-wise statistics. Available only if we're reading a single
@@ -545,8 +584,10 @@ impl<T: DataSourceTableFunction> TableFunction for T {
 
     fn cardinality(bind_data: &Self::BindData) -> Cardinality {
         match bind_data.data_source.row_count() {
-            Some(Precision::Exact(v)) => Cardinality::Maximum(v),
-            Some(Precision::Inexact(v)) => Cardinality::Estimate(v),
+            Some(Precision::Exact(v) | Precision::Inexact(v)) => {
+                // Post-filter estimate is always a heuristic.
+                Cardinality::Estimate(postfilter_cardinality(v, bind_data.has_non_optional_filter))
+            }
             None => Cardinality::Unknown,
         }
     }
@@ -565,8 +606,8 @@ impl<T: DataSourceTableFunction> TableFunction for T {
     fn to_string(bind_data: &Self::BindData, map: &mut DuckdbStringMapRef) {
         map.push("Function", "Vortex Scan");
         if !bind_data.filter_exprs.is_empty() {
-            let mut filters = bind_data.filter_exprs.iter().map(|f| format!("{}", f));
-            map.push("Filters", &filters.join(" /\\\n"));
+            let mut filters = bind_data.filter_exprs.iter().map(|f| format!("{f}"));
+            map.push("Filters", &filters.join("\n"));
         }
     }
 }
@@ -687,6 +728,7 @@ struct FilterWithVirtualColumns {
     row_range: Option<Range<u64>>,
     file_selection: Selection,
     file_range: Option<Range<u64>>,
+    has_non_optional_filter: bool,
 }
 
 /// Creates a table filter expression, row selection, and row range from the table filter set,
@@ -698,6 +740,8 @@ fn extract_table_filter_expr(
     additional_filters: &[Expression],
     dtype: &DType,
 ) -> VortexResult<FilterWithVirtualColumns> {
+    let mut has_non_optional_filter = false;
+
     let mut table_filter_exprs: HashSet<Expression> = if let Some(filter) = table_filter_set {
         filter
             .into_iter()
@@ -706,6 +750,8 @@ fn extract_table_filter_expr(
                 !is_virtual_column(column_ids[idx_u])
             })
             .map(|(idx, ex)| {
+                has_non_optional_filter |= !matches!(ex.as_class(), TableFilterClass::Optional(_));
+
                 let idx_u: usize = idx.as_();
                 let col_idx: usize = column_ids[idx_u].as_();
                 let name = &column_fields.get(col_idx).vortex_expect("exists").name;
@@ -741,6 +787,7 @@ fn extract_table_filter_expr(
         row_range,
         file_selection,
         file_range,
+        has_non_optional_filter,
     };
     Ok(out)
 }

vortex-duckdb/src/duckdb/table_function/init.rs

@@ -74,9 +74,10 @@ impl<'a, T: TableFunction> TableInitInput<'a, T> {
         }
     }
 
+    #[allow(clippy::mut_from_ref)]
     /// Returns the bind data for the table function.
-    pub fn bind_data(&self) -> &T::BindData {
-        unsafe { &*self.input.bind_data.cast::<T::BindData>() }
+    pub fn bind_data(&self) -> &mut T::BindData {
+        unsafe { &mut *self.input.bind_data.cast::<T::BindData>() }
     }
 
     pub fn column_ids(&self) -> &[u64] {