@@ -18,16 +18,6 @@ use vortex_error::VortexExpect;
 
 use crate::expr::stats::Stat;
 
-/// Statistics that are used for pruning files (i.e., we want to ensure they are computed when compressing/writing).
-/// Sum is included for boolean arrays.
-pub const PRUNING_STATS: &[Stat] = &[
-    Stat::Min,
-    Stat::Max,
-    Stat::Sum,
-    Stat::NullCount,
-    Stat::NaNCount,
-];
-
 pub fn as_stat_bitset_bytes(stats: &[Stat]) -> Vec<u8> {
     let max_stat = u8::from(last::<Stat>().vortex_expect("last stat")) as usize + 1;
     // TODO(ngates): use vortex-buffer::BitBuffer
 
@@ -45,8 +45,12 @@ use futures::stream;
 use object_store::ObjectMeta;
 use object_store::ObjectStore;
 use vortex::VortexSessionDefault;
+use vortex::aggregate_fn::AggregateFnVTableExt;
+use vortex::aggregate_fn::EmptyOptions;
+use vortex::aggregate_fn::fns::is_constant::IsConstant;
 use vortex::array::memory::MemorySessionExt;
 use vortex::dtype::DType;
+use vortex::dtype::FieldPath;
 use vortex::dtype::PType;
 use vortex::error::VortexExpect;
 use vortex::error::VortexResult;
@@ -605,7 +609,7 @@ impl FileFormat for VortexFormat {
                         })
                 });
 
-            let (dtype, file_stats, row_count) = match cached_metadata {
+            let (_dtype, file_stats, row_count) = match cached_metadata {
                 Some(metadata) => metadata,
                 None => {
                     // Not entry - open the file
@@ -642,10 +646,6 @@ impl FileFormat for VortexFormat {
                 }
             };
 
-            let struct_dtype = dtype
-                .as_struct_fields_opt()
-                .vortex_expect("dtype is not a struct");
-
             // Evaluate the statistics for each column that we are able to return to DataFusion.
             let Some(file_stats) = file_stats else {
                 // If the file has no column stats, the best we can do is return a row count.
@@ -668,12 +668,23 @@ impl FileFormat for VortexFormat {
             for field in table_schema.fields().iter() {
                 // If the column does not exist, continue. This can happen if the schema has evolved
                 // but we have not yet updated the Vortex file.
-                let Some(col_idx) = struct_dtype.find(field.name()) else {
+                let Some((aggregates, stats_dtype)) =
+                    file_stats.get_by_path(&FieldPath::from_name(field.name().as_str()))
+                else {
                     // The default sets all statistics to `Precision<Absent>`.
                     column_statistics.push(ColumnStatistics::default());
                     continue;
                 };
-                let (stats_set, stats_dtype) = file_stats.get(col_idx);
+                let mut stats_set = aggregates.to_stats_set();
+                // `IsConstant` has no aggregate behind its stat slot, so `to_stats_set` omits it.
+                if let Some(is_constant) = aggregates
+                    .get(&IsConstant.bind(EmptyOptions))
+                    .as_exact()
+                    .and_then(|is_constant| bool::try_from(&is_constant).ok())
+                {
+                    stats_set.set(Stat::IsConstant, Precision::exact(is_constant));
+                }
+                let stats_set = &stats_set;
 
                 // Update the total size in bytes.
                 let column_size =
 
@@ -13,7 +13,6 @@
 #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 &,
@@ -42,7 +41,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), names);
+    return make_uniq<VortexCopyBindData>(std::move(cdata));
 }
 
 unique_ptr<GlobalFunctionData>
@@ -88,7 +87,7 @@ void copy_to_get_written_statistics(ClientContext &,
     gstate.Cast<VortexCopyGlobalState>().written_stats = &statistics;
 }
 
-void copy_to_finalize(ClientContext &, FunctionData &bind_data, GlobalFunctionData &gstate) {
+void copy_to_finalize(ClientContext &, FunctionData &, GlobalFunctionData &gstate) {
     auto &global = gstate.Cast<VortexCopyGlobalState>();
     void *const ffi_global = global.ffi_global->DataPtr();
 
@@ -101,23 +100,14 @@ void copy_to_finalize(ClientContext &, FunctionData &bind_data, GlobalFunctionDa
     if (!global.written_stats) {
         return;
     }
-    const 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)) {
         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;
@@ -146,9 +136,9 @@ void copy_to_finalize(ClientContext &, FunctionData &bind_data, GlobalFunctionDa
         if (col_stats.has_nan_stat) {
             column["has_nan"] = Value::BOOLEAN(col_stats.contains_nan);
         }
-
-        string quoted = SQLQuotedIdentifier::ToString(names[i]);
-        global.written_stats->column_statistics.emplace(std::move(quoted), std::move(column));
+        auto column_key = reinterpret_cast<Value *>(col_stats.column_key)->ToString();
+        duckdb_destroy_value(&col_stats.column_key);
+        global.written_stats->column_statistics.emplace(std::move(column_key), std::move(column));
     }
 }
 
 
@@ -8,11 +8,9 @@
 using namespace duckdb;
 
 struct VortexCopyBindData final : TableFunctionData {
-    VortexCopyBindData(unique_ptr<CData> ffi_bind, vector<Identifier> column_names)
-        : ffi_bind(std::move(ffi_bind)), column_names(std::move(column_names)) {
+    VortexCopyBindData(unique_ptr<CData> ffi_bind) : ffi_bind(std::move(ffi_bind)) {
     }
     unique_ptr<CData> ffi_bind;
-    vector<Identifier> column_names;
 };
 
 struct VortexCopyGlobalState final : GlobalFunctionData {
 
@@ -83,6 +83,8 @@ typedef struct {
 
 // Per-column statistics of a written Vortex file.
 typedef struct {
+    // Owned VARCHAR holding column's dot-separated path.
+    duckdb_value column_key;
     // Owned values, null if absent; the caller must destroy them.
     duckdb_value min;
     duckdb_value max;
 
@@ -1,14 +1,22 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+use vortex::aggregate_fn::AggregateFnRef;
+use vortex::aggregate_fn::AggregateFnVTableExt;
+use vortex::aggregate_fn::EmptyOptions;
+use vortex::aggregate_fn::NumericalAggregateOpts;
+use vortex::aggregate_fn::fns::max::Max;
+use vortex::aggregate_fn::fns::min::Min;
+use vortex::aggregate_fn::fns::nan_count::NanCount;
+use vortex::aggregate_fn::fns::null_count::NullCount;
+use vortex::aggregate_fn::fns::uncompressed_size_in_bytes::UncompressedSizeInBytes;
+use vortex::aggregate_fn::fns::uncompressed_size_in_bytes::UncompressedSizeOpts;
 use vortex::array::match_each_float_ptype;
-use vortex::array::stats::StatsSet;
 use vortex::dtype::DType;
-use vortex::dtype::PType;
 use vortex::error::VortexExpect as _;
 use vortex::error::VortexResult;
 use vortex::expr::stats::Precision;
-use vortex::expr::stats::Stat;
+use vortex::layout::layouts::file_stats::AggregateStats;
 use vortex::scalar::Scalar;
 use vortex::scalar::ScalarValue;
 
@@ -74,33 +82,25 @@ pub struct ColumnStatisticsAggregate {
 }
 
 impl ColumnStatisticsAggregate {
-    pub fn new(stats: &StatsSet) -> Self {
-        let min = match stats.get(Stat::Min) {
-            Precision::Exact(min) => Some(min),
-            _ => None,
-        };
-        let max = match stats.get(Stat::Max) {
-            Precision::Exact(max) => Some(max),
+    pub fn new(aggregates: &AggregateStats) -> Self {
+        let exact = |aggregate_fn: AggregateFnRef| match aggregates.get(&aggregate_fn) {
+            Precision::Exact(value) => value.into_value(),
             _ => None,
         };
 
-        let max_string_length =
-            if let Precision::Exact(value) = stats.get(Stat::UncompressedSizeInBytes) {
-                // DuckDB's string length is u32
-                #[allow(clippy::cast_possible_truncation)]
-                Some(value.as_primitive().as_u64().vortex_expect("not a u64") as u32)
-            } else {
-                None
-            };
+        let min = exact(Min.bind(NumericalAggregateOpts::skip_nans()));
+        let max = exact(Max.bind(NumericalAggregateOpts::skip_nans()));
 
-        let has_null = match stats.get(Stat::NullCount) {
-            Precision::Exact(cnt) => cnt.as_primitive().as_u64().vortex_expect("not a u64") > 0,
-            _ => true,
-        };
+        // DuckDB's string length is u32
+        #[allow(clippy::cast_possible_truncation)]
+        let max_string_length = exact(UncompressedSizeInBytes.bind(UncompressedSizeOpts::exact()))
+            .map(|value| value.as_primitive().as_u64().vortex_expect("not a u64") as u32);
+
+        let has_null = exact(NullCount.bind(EmptyOptions))
+            .is_none_or(|count| count.as_primitive().as_u64().vortex_expect("not a u64") > 0);
 
-        let nan_count = stats
-            .get_as::<u64>(Stat::NaNCount, &PType::U64.into())
-            .as_exact();
+        let nan_count = exact(NanCount.bind(EmptyOptions))
+            .map(|count| count.as_primitive().as_u64().vortex_expect("not a u64"));
 
         Self {
             min,