feat(duckdb): report per-file column statistics from the vortex COPY writer

vortex-duckdb/cpp/copy_function.cpp

@@ -9,6 +9,7 @@
 #include "duckdb/main/capi/capi_internal.hpp"
 #include "duckdb/main/client_context.hpp"
 #include "duckdb/main/connection.hpp"
+#include "duckdb/parser/keyword_helper.hpp"
 #include "duckdb/parser/parsed_data/create_copy_function_info.hpp"
 
 using namespace duckdb;
@@ -17,13 +18,18 @@ struct CopyBindData final : TableFunctionData {
     CopyBindData(unique_ptr<CData> ffi_data) : ffi_data(std::move(ffi_data)) {
     }
     unique_ptr<CData> ffi_data;
+    // Column names in write order, used to key WRITTEN_FILE_STATISTICS.
+    vector<string> column_names;
 };
 
 struct CopyGlobalData final : GlobalFunctionData {
     CopyGlobalData(unique_ptr<CData> ffi_data) : ffi_data(std::move(ffi_data)) {
     }
 
     unique_ptr<CData> ffi_data;
+    // Non-owning; set in copy_to_get_written_statistics (before the write) and filled in
+    // copy_to_finalize (after the write). Null when the plan does not request statistics.
+    CopyFunctionFileStatistics *written_stats = nullptr;
 };
 
 unique_ptr<FunctionData> copy_to_bind(ClientContext &,
@@ -53,7 +59,9 @@ unique_ptr<FunctionData> copy_to_bind(ClientContext &,
         throw BinderException(IntoErrString(error_out));
     }
     auto cdata = unique_ptr<CData>(reinterpret_cast<CData *>(ffi_bind_data));
-    return make_uniq<CopyBindData>(std::move(cdata));
+    auto bind = make_uniq<CopyBindData>(std::move(cdata));
+    bind->column_names = column_names;
+    return bind;
 }
 
 unique_ptr<GlobalFunctionData>
@@ -86,13 +94,77 @@ void copy_to_sink(ExecutionContext &,
     }
 }
 
-void copy_to_finalize(ClientContext &, FunctionData &, GlobalFunctionData &gstate) {
-    void *const ffi_global = gstate.Cast<CopyGlobalData>().ffi_data->DataPtr();
+// Called before the write (right after global init) when the plan requests
+// WRITTEN_FILE_STATISTICS. We stash the target struct and fill it at finalize, once the
+// footer statistics exist - mirroring the parquet writer's store-pointer-then-fill pattern.
+void copy_to_get_written_statistics(ClientContext &,
+                                    FunctionData &,
+                                    GlobalFunctionData &gstate,
+                                    CopyFunctionFileStatistics &statistics) {
+    gstate.Cast<CopyGlobalData>().written_stats = &statistics;
+}
+
+void copy_to_finalize(ClientContext &, FunctionData &bind_data, GlobalFunctionData &gstate) {
+    auto &global = gstate.Cast<CopyGlobalData>();
+    void *const ffi_global = global.ffi_data->DataPtr();
     duckdb_vx_error error_out = nullptr;
     duckdb_copy_function_copy_to_finalize(ffi_global, &error_out);
     if (error_out) {
         throw ExecutorException(IntoErrString(error_out));
     }
+
+    if (!global.written_stats) {
+        return;
+    }
+    // Fill the statistics captured in copy_to_get_written_statistics from the finished write's
+    // footer. min/max come back as owned duckdb values; DuckLake expects their string form.
+    auto &names = bind_data.Cast<CopyBindData>().column_names;
+    duckdb_vx_written_file_statistics file_stats;
+    if (!duckdb_copy_function_get_written_file_statistics(ffi_global, &file_stats)) {
+        // Statistics were requested (written_stats is set) but the finished write produced none;
+        // that is an internal inconsistency, not a silently empty result.
+        throw InternalException("vortex COPY: written statistics were requested but not produced");
+    }
+    global.written_stats->row_count = file_stats.row_count;
+    global.written_stats->file_size_bytes = file_stats.file_size_bytes;
+    global.written_stats->footer_size_bytes = Value::UBIGINT(file_stats.footer_size_bytes);
+    // Keyed by top-level column name only. The vortex footer reports one statistics set per
+    // top-level field, so nested struct/list leaf columns get no statistics here (unlike parquet,
+    // which recurses to leaf paths). Flat tables are fully covered.
+    for (idx_t i = 0; i < file_stats.num_columns && i < names.size(); i++) {
+        duckdb_vx_written_column_statistics col_stats {};
+        duckdb_vx_error col_error = nullptr;
+        if (!duckdb_copy_function_get_written_column_statistics(ffi_global, i, &col_stats, &col_error)) {
+            if (col_error) {
+                throw ExecutorException(IntoErrString(col_error));
+            }
+            // No statistics for this column (e.g. a type without min/max); skip it.
+            continue;
+        }
+        case_insensitive_map_t<Value> column;
+        column["num_values"] = Value::UBIGINT(col_stats.num_values);
+        if (col_stats.has_column_size) {
+            column["column_size_bytes"] = Value::UBIGINT(col_stats.column_size_bytes);
+        }
+        if (col_stats.has_null_count) {
+            column["null_count"] = Value::UBIGINT(col_stats.null_count);
+        }
+        if (col_stats.min) {
+            column["min"] = Value(reinterpret_cast<Value *>(col_stats.min)->ToString());
+            duckdb_destroy_value(&col_stats.min);
+        }
+        if (col_stats.max) {
+            column["max"] = Value(reinterpret_cast<Value *>(col_stats.max)->ToString());
+            duckdb_destroy_value(&col_stats.max);
+        }
+        if (col_stats.has_nan_stat) {
+            column["has_nan"] = Value::BOOLEAN(col_stats.contains_nan);
+        }
+        // DuckLake keys column statistics by a quoted, dot-separated path (see
+        // DuckLakeUtil::ParseQuotedList); match the parquet writer, which quotes each name.
+        global.written_stats->column_statistics.emplace(KeywordHelper::WriteQuoted(names[i], '"'),
+                                                        std::move(column));
+    }
 }
 
 extern "C" duckdb_state duckdb_vx_register_copy_function(duckdb_database ffi_db) {
@@ -108,6 +180,7 @@ extern "C" duckdb_state duckdb_vx_register_copy_function(duckdb_database ffi_db)
     };
     fn.copy_to_sink = copy_to_sink;
     fn.copy_to_finalize = copy_to_finalize;
+    fn.copy_to_get_written_statistics = copy_to_get_written_statistics;
     fn.extension = "vortex";
 
     // TODO(joe): expose this via c our api

vortex-duckdb/cpp/include/table_function.h

@@ -76,6 +76,30 @@ typedef struct {
     bool has_null;
 } duckdb_column_statistics;
 
+// File-level statistics of a written Vortex file, for the DuckLake
+// WRITTEN_FILE_STATISTICS return path. Filled by Rust from the WriteSummary.
+typedef struct {
+    uint64_t row_count;
+    uint64_t file_size_bytes;
+    uint64_t footer_size_bytes;
+    uint64_t num_columns;
+} duckdb_vx_written_file_statistics;
+
+// Per-column statistics of a written Vortex file. `min`/`max` are owned
+// duckdb_value handles (null if absent) that the caller must destroy.
+typedef struct {
+    duckdb_value min;
+    duckdb_value max;
+    bool has_null_count;
+    uint64_t null_count;
+    uint64_t num_values;
+    bool has_column_size;
+    uint64_t column_size_bytes;
+    // Whether a NaN-count statistic was available (float columns), and whether it saw any NaN.
+    bool has_nan_stat;
+    bool contains_nan;
+} duckdb_vx_written_column_statistics;
+
 const idx_t INVALID_IDX = UINT64_MAX;
 
 typedef struct {

vortex-duckdb/include/vortex.h

@@ -94,6 +94,16 @@ void duckdb_copy_function_copy_to_sink(const void *bind_data,
 
 extern void duckdb_copy_function_copy_to_finalize(void *global_data, duckdb_vx_error *error_out);
 
+extern
+bool duckdb_copy_function_get_written_file_statistics(const void *global_data,
+                                                      duckdb_vx_written_file_statistics *out);
+
+extern
+bool duckdb_copy_function_get_written_column_statistics(const void *global_data,
+                                                        size_t column_index,
+                                                        duckdb_vx_written_column_statistics *out,
+                                                        duckdb_vx_error *error_out);
+
 #ifdef __cplusplus
 }  // extern "C"
 #endif  // __cplusplus

vortex-duckdb/src/copy.rs

@@ -22,6 +22,8 @@ use vortex::dtype::StructFields;
 use vortex::error::VortexExpect;
 use vortex::error::VortexResult;
 use vortex::error::vortex_err;
+use vortex::expr::stats::Precision;
+use vortex::expr::stats::Stat;
 use vortex::file::WriteOptionsSessionExt;
 use vortex::file::WriteSummary;
 use vortex::file::multi::parse_uri_or_path;
@@ -32,14 +34,18 @@ use vortex::io::runtime::BlockingRuntime;
 use vortex::io::runtime::Task;
 use vortex::io::runtime::current::CurrentThreadWorkerPool;
 use vortex::io::session::RuntimeSessionExt;
+use vortex::scalar::Scalar;
+use vortex::scalar::ScalarValue;
 
 use crate::REGISTRY;
 use crate::RUNTIME;
 use crate::SESSION;
 use crate::convert::FromLogicalType;
+use crate::convert::ToDuckDBScalar;
 use crate::convert::data_chunk_to_vortex;
 use crate::duckdb::DataChunkRef;
 use crate::duckdb::LogicalTypeRef;
+use crate::duckdb::Value;
 
 #[derive(Clone)]
 pub struct CopyFunctionBind {
@@ -48,13 +54,22 @@ pub struct CopyFunctionBind {
 }
 assert_impl_all!(CopyFunctionBind: Send, Clone);
 
+/// Captured in `copy_to_finalize` and read back by the WRITTEN_FILE_STATISTICS path. The per-column
+/// compressed sizes are computed once here rather than per column, since DuckDB queries statistics
+/// one column at a time.
+struct FinishedWrite {
+    summary: WriteSummary,
+    column_sizes: Vec<u64>,
+}
+
 /// Write to a file has two phases, writing data chunks and then closing the file.
 /// We use a spawned tokio task to actually compress arrays and write it to disk.
 /// Each chunk is pushed into the sink and read from the task.
 /// Once finished we can close all sinks and then the task can be awaited and the file
 /// flushed to disk.
 pub struct CopyFunctionGlobal {
     write_task: Mutex<Option<Task<VortexResult<WriteSummary>>>>,
+    finished: Mutex<Option<FinishedWrite>>,
     sink: Option<Sender<VortexResult<ArrayRef>>>,
     // Pool of background workers helping to drive the write task.
     // Note that this is optional and without it, we would only drive the task when DuckDB calls
@@ -113,11 +128,125 @@ pub fn copy_to_finalize(init_global: &mut CopyFunctionGlobal) -> VortexResult<()
             .lock()
             .take()
             .vortex_expect("no file to close");
-        task.await?;
+        // Keep the write summary (footer + size) so DuckLake can read per-file statistics back
+        // without re-opening the file. Compute the per-column compressed sizes once, up front.
+        let summary = task.await?;
+        let column_sizes = summary.compressed_column_sizes().unwrap_or_default();
+        *init_global.finished.lock() = Some(FinishedWrite {
+            summary,
+            column_sizes,
+        });
         Ok(())
     })
 }
 
+/// File-level statistics of the written Vortex file, for the WRITTEN_FILE_STATISTICS return path.
+pub(crate) struct WrittenFileStats {
+    pub row_count: u64,
+    pub file_size_bytes: u64,
+    pub footer_size_bytes: u64,
+    pub num_columns: usize,
+}
+
+/// Per-column statistics of the written Vortex file. `min`/`max` are DuckDB values converted from
+/// the Vortex scalar; every field is optional and omitted when the statistic is not available.
+pub(crate) struct WrittenColumnStats {
+    pub min: Option<Value>,
+    pub max: Option<Value>,
+    pub null_count: Option<u64>,
+    pub num_values: u64,
+    pub column_size_bytes: Option<u64>,
+    pub has_nan: Option<bool>,
+}
+
+/// Read file-level statistics back from the finished write. `None` before finalize.
+pub(crate) fn written_file_stats(global: &CopyFunctionGlobal) -> Option<WrittenFileStats> {
+    let guard = global.finished.lock();
+    Some(file_stats_from_summary(&guard.as_ref()?.summary))
+}
+
+/// Read per-column statistics for `column_index` from the finished write. `Ok(None)` if the file is
+/// not finalized or the column has no statistics; `Err` if a scalar could not be converted.
+pub(crate) fn written_column_stats(
+    global: &CopyFunctionGlobal,
+    column_index: usize,
+) -> VortexResult<Option<WrittenColumnStats>> {
+    let guard = global.finished.lock();
+    let Some(finished) = guard.as_ref() else {
+        return Ok(None);
+    };
+    column_stats_from_summary(&finished.summary, column_index, &finished.column_sizes)
+}
+
+fn file_stats_from_summary(summary: &WriteSummary) -> WrittenFileStats {
+    let num_columns = summary
+        .footer()
+        .statistics()
+        .map_or(0, |s| s.stats_sets().len());
+    WrittenFileStats {
+        row_count: summary.row_count(),
+        file_size_bytes: summary.size(),
+        // Vortex has no separate footer-size hint; 0 means "read the footer normally".
+        footer_size_bytes: 0,
+        num_columns,
+    }
+}
+
+/// Per-column statistics from a finished write's summary and its precomputed compressed sizes
+/// (`column_sizes`, indexed the same as the footer's stats sets).
+///
+/// Only top-level columns are covered: the footer exposes one statistics set per top-level field,
+/// so nested struct/list leaf columns are not reported (parquet, by contrast, recurses to leaf
+/// paths). Flat tables - the common DuckLake case - are fully covered.
+fn column_stats_from_summary(
+    summary: &WriteSummary,
+    column_index: usize,
+    column_sizes: &[u64],
+) -> VortexResult<Option<WrittenColumnStats>> {
+    let Some(file_stats) = summary.footer().statistics() else {
+        return Ok(None);
+    };
+    let stats_sets = file_stats.stats_sets();
+    if column_index >= stats_sets.len() {
+        return Ok(None);
+    }
+    let stats = &stats_sets[column_index];
+    let dtype = &file_stats.dtypes()[column_index];
+
+    Ok(Some(WrittenColumnStats {
+        min: exact_scalar_to_duckdb(stats.get(Stat::Min), dtype)?,
+        max: exact_scalar_to_duckdb(stats.get(Stat::Max), dtype)?,
+        null_count: exact_u64(stats.get(Stat::NullCount)),
+        // NaNCount is exact only for float columns, so this is emitted just for them (as in parquet).
+        has_nan: exact_u64(stats.get(Stat::NaNCount)).map(|count| count > 0),
+        num_values: summary.row_count(),
+        // On-disk compressed size; excludes bytes not attributable to a column (e.g. struct validity).
+        column_size_bytes: column_sizes.get(column_index).copied(),
+    }))
+}
+
+/// Convert an exact scalar statistic to a DuckDB value, propagating a conversion failure rather than
+/// dropping it. `Ok(None)` when the statistic is not exactly known.
+fn exact_scalar_to_duckdb(
+    stat: Precision<ScalarValue>,
+    dtype: &DType,
+) -> VortexResult<Option<Value>> {
+    match stat {
+        Precision::Exact(value) => Ok(Some(
+            Scalar::try_new(dtype.clone(), Some(value))?.try_to_duckdb_scalar()?,
+        )),
+        _ => Ok(None),
+    }
+}
+
+/// Extract an exact `u64` statistic (e.g. a count), or `None` if not exactly known.
+fn exact_u64(stat: Precision<ScalarValue>) -> Option<u64> {
+    match stat {
+        Precision::Exact(value) => value.as_primitive().as_u64(),
+        _ => None,
+    }
+}
+
 pub fn copy_to_initialize_global(
     bind_data: &CopyFunctionBind,
     file_path: String,
@@ -163,6 +292,7 @@ pub fn copy_to_initialize_global(
     Ok(CopyFunctionGlobal {
         worker_pool,
         write_task: Mutex::new(Some(write_task)),
+        finished: Mutex::new(None),
         sink: Some(sink),
     })
 }