Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

10 changes: 0 additions & 10 deletions vortex-array/src/stats/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -33,16 +33,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
Expand Down
25 changes: 18 additions & 7 deletions vortex-datafusion/src/persistent/format.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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 =
Expand Down
20 changes: 5 additions & 15 deletions vortex-duckdb/cpp/copy_function.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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 &,
Expand Down Expand Up @@ -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>
Expand Down Expand Up @@ -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();

Expand All @@ -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;
Expand Down Expand Up @@ -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));
}
}

Expand Down
4 changes: 1 addition & 3 deletions vortex-duckdb/cpp/include/copy_function.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
2 changes: 2 additions & 0 deletions vortex-duckdb/cpp/include/table_function.h
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
50 changes: 25 additions & 25 deletions vortex-duckdb/src/column_statistics.rs
Original file line number Diff line number Diff line change
@@ -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;

Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading