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

vortex-duckdb/cpp/copy_function.cpp

@@ -11,6 +11,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"
 
 unique_ptr<FunctionData> copy_to_bind(ClientContext &,
@@ -39,7 +40,7 @@ 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<VortexCopyBindData>(std::move(cdata));
+    return make_uniq<VortexCopyBindData>(std::move(cdata), names);
 }
 
 unique_ptr<GlobalFunctionData>
@@ -77,15 +78,78 @@ void copy_to_sink(ExecutionContext &,
     }
 }
 
-void copy_to_finalize(ClientContext &, FunctionData &, GlobalFunctionData &gstate) {
-    const VortexCopyGlobalState &global = gstate.Cast<VortexCopyGlobalState>();
+// CopyToFileInfo::file_stats is owned by the operator's sink state, which outlives gstate.
+void copy_to_get_written_statistics(ClientContext &,
+                                    FunctionData &,
+                                    GlobalFunctionData &gstate,
+                                    CopyFunctionFileStatistics &statistics) {
+    gstate.Cast<VortexCopyGlobalState>().written_stats = &statistics;
+}
 
+void copy_to_finalize(ClientContext &, FunctionData &bind_data, GlobalFunctionData &gstate) {
+    auto &global = gstate.Cast<VortexCopyGlobalState>();
     void *const ffi_global = global.ffi_global->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;
+    }
+    auto &names = bind_data.Cast<VortexCopyBindData>().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");
+    }
+    if (file_stats.num_columns != names.size()) {
+        throw InternalException("vortex COPY: %llu statistics columns for %llu written columns",
+                                file_stats.num_columns,
+                                names.size());
+    }
+    D_ASSERT(global.written_stats != nullptr);
+    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++) {
+        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));
+            }
+            throw InternalException("vortex COPY: no statistics for column %llu after finalize", i);
+        }
+        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));
+    }
 }
 
 unique_ptr<PreparedBatchData> copy_to_prepare_batch(ClientContext &,
@@ -143,6 +207,7 @@ extern "C" duckdb_state duckdb_vx_register_copy_function(duckdb_database ffi_db)
     fn.copy_to_finalize = copy_to_finalize;
     fn.prepare_batch = copy_to_prepare_batch;
     fn.flush_batch = copy_to_flush_batch;
+    fn.copy_to_get_written_statistics = copy_to_get_written_statistics;
     fn.extension = "vortex";
 
     fn.execution_mode = [](bool preserve_insertion_order, bool supports_batch_index) {

vortex-duckdb/cpp/include/copy_function.hpp

@@ -8,15 +8,20 @@
 using namespace duckdb;
 
 struct VortexCopyBindData final : TableFunctionData {
-    VortexCopyBindData(unique_ptr<CData> ffi_bind) : ffi_bind(std::move(ffi_bind)) {
+    VortexCopyBindData(unique_ptr<CData> ffi_bind, vector<string> column_names)
+        : ffi_bind(std::move(ffi_bind)), column_names(std::move(column_names)) {
     }
     unique_ptr<CData> ffi_bind;
+    // Column names in write order, used to key WRITTEN_FILE_STATISTICS.
+    vector<string> column_names;
 };
 
 struct VortexCopyGlobalState final : GlobalFunctionData {
     VortexCopyGlobalState(unique_ptr<CData> ffi_global) : ffi_global(std::move(ffi_global)) {
     }
     unique_ptr<CData> ffi_global;
+    // Non-owning; null when the plan does not request statistics.
+    CopyFunctionFileStatistics *written_stats = nullptr;
 };
 
 struct VortexCopyPreparedBatchData final : PreparedBatchData {

vortex-duckdb/cpp/include/table_function.h

@@ -51,6 +51,30 @@ typedef struct {
     duckdb_logical_type type;
 } 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.
+typedef struct {
+    // Owned values, null if absent; the caller must destroy them.
+    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;
+
 duckdb_state duckdb_vx_register_table_functions(duckdb_database ffi_db);
 
 typedef struct duckdb_vx_agg_input_ *duckdb_vx_agg_input;

vortex-duckdb/include/vortex.h

@@ -124,6 +124,16 @@ void duckdb_copy_function_flush_batch(const void *global,
                                       const void *batch,
                                       duckdb_vx_error *error);
 
+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