This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 20cdfe9 feat: write value stats for append data files (#477)
20cdfe9 is described below
commit 20cdfe9a8a37ce4fd8f91eaf8ab746c712f66602
Author: QuakeWang <[email protected]>
AuthorDate: Tue Jul 14 11:25:40 2026 +0800
feat: write value stats for append data files (#477)
---
crates/integrations/datafusion/src/merge_into.rs | 125 ++++-
crates/paimon/src/arrow/format/blob.rs | 6 +-
crates/paimon/src/arrow/format/mod.rs | 40 +-
crates/paimon/src/arrow/format/parquet.rs | 431 +++++++++++++++--
crates/paimon/src/arrow/format/row.rs | 6 +-
crates/paimon/src/arrow/format/shredding.rs | 6 +-
crates/paimon/src/arrow/format/vortex.rs | 30 +-
crates/paimon/src/spec/binary_row.rs | 332 +++++++------
crates/paimon/src/spec/core_options.rs | 174 +++++++
crates/paimon/src/table/data_evolution_reader.rs | 2 +-
crates/paimon/src/table/data_file_reader.rs | 21 +-
crates/paimon/src/table/data_file_writer.rs | 35 +-
.../src/table/dedicated_format_file_writer.rs | 6 +-
crates/paimon/src/table/kv_file_writer.rs | 2 +-
crates/paimon/src/table/postpone_file_writer.rs | 4 +-
crates/paimon/src/table/table_write.rs | 530 ++++++++++++++++++++-
16 files changed, 1493 insertions(+), 257 deletions(-)
diff --git a/crates/integrations/datafusion/src/merge_into.rs
b/crates/integrations/datafusion/src/merge_into.rs
index b73b155..2620d9d 100644
--- a/crates/integrations/datafusion/src/merge_into.rs
+++ b/crates/integrations/datafusion/src/merge_into.rs
@@ -26,7 +26,9 @@ use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
-use datafusion::arrow::array::{Array, Int32Array, RecordBatch, UInt32Array,
UInt64Array};
+use datafusion::arrow::array::{
+ Array, ArrayRef, Int32Array, RecordBatch, UInt32Array, UInt64Array,
+};
use datafusion::arrow::compute;
use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, Schema};
use datafusion::error::{DataFusionError, Result as DFResult};
@@ -37,7 +39,7 @@ use datafusion::sql::sqlparser::ast::{
};
use futures::TryStreamExt;
-use paimon::spec::{datums_to_binary_row, extract_datum_from_arrow,
CoreOptions};
+use paimon::spec::{datums_to_binary_row, extract_datum_from_arrow,
CoreOptions, DataField};
use paimon::table::{CopyOnWriteMergeWriter, DataSplitBuilder, Table,
WriteBuilder};
use crate::error::to_datafusion_error;
@@ -569,13 +571,6 @@ async fn execute_cow_merge_inner(
// Handle NOT MATCHED → INSERT
if !clauses.inserts.is_empty() {
- let table_fields: Vec<String> = table
- .schema()
- .fields()
- .iter()
- .map(|f| f.name().to_string())
- .collect();
-
let insert_sql = if has_target_data {
format!(
"SELECT {s_alias}.* FROM {source_ref} AS {s_alias} \
@@ -595,7 +590,7 @@ async fn execute_cow_merge_inner(
&clauses.inserts,
s_alias,
&[],
- &table_fields,
+ table.schema().fields(),
temp_tracker,
)
.await?;
@@ -734,13 +729,6 @@ async fn execute_merge_into_once(
injected_columns.push(format!("__upd_{col}"));
}
}
- // Table schema field names for reordering INSERT columns
- let table_fields: Vec<String> = table
- .schema()
- .fields()
- .iter()
- .map(|f| f.name().to_string())
- .collect();
let mut temp_tracker = TempTableTracker::new(ctx);
let insert_batches = build_insert_batches(
ctx,
@@ -748,7 +736,7 @@ async fn execute_merge_into_once(
&parsed.inserts,
s_alias,
&injected_columns,
- &table_fields,
+ table.schema().fields(),
&mut temp_tracker,
)
.await?;
@@ -854,7 +842,7 @@ async fn build_insert_batches(
inserts: &[MergeInsertClause],
s_alias: &str,
injected_columns: &[String],
- table_fields: &[String],
+ table_fields: &[DataField],
temp_tracker: &mut TempTableTracker<'_>,
) -> DFResult<Vec<RecordBatch>> {
if not_matched_batches.is_empty() || not_matched_batches.iter().all(|b|
b.num_rows() == 0) {
@@ -883,7 +871,7 @@ async fn build_insert_batches_inner(
inserts: &[MergeInsertClause],
s_alias: &str,
tmp_name: &str,
- table_fields: &[String],
+ table_fields: &[DataField],
) -> DFResult<Vec<RecordBatch>> {
let mut all_batches = Vec::new();
let mut consumed_predicates: Vec<String> = Vec::new();
@@ -910,12 +898,61 @@ async fn build_insert_batches_inner(
let sql = format!("SELECT {select_clause} FROM {tmp_name} AS
{s_alias}{where_clause}");
let batches = ctx.ctx().sql(&sql).await?.collect().await?;
- all_batches.extend(batches);
+ for batch in batches {
+ all_batches.push(normalize_insert_batch_to_table_schema(
+ &batch,
+ table_fields,
+ )?);
+ }
}
Ok(all_batches)
}
+fn normalize_insert_batch_to_table_schema(
+ batch: &RecordBatch,
+ table_fields: &[DataField],
+) -> DFResult<RecordBatch> {
+ if batch.num_columns() != table_fields.len() {
+ return Err(DataFusionError::Plan(format!(
+ "MERGE INSERT output has {} columns but target table has {}",
+ batch.num_columns(),
+ table_fields.len()
+ )));
+ }
+
+ let target_schema =
+
paimon::arrow::build_target_arrow_schema(table_fields).map_err(to_datafusion_error)?;
+ let mut columns = Vec::with_capacity(table_fields.len());
+
+ for (target_idx, field) in table_fields.iter().enumerate() {
+ let column = batch.column(target_idx).clone();
+ let target_type = target_schema.field(target_idx).data_type();
+ let column = cast_insert_column(field.name(), column, target_type)?;
+ columns.push(column);
+ }
+
+ RecordBatch::try_new(target_schema, columns).map_err(DataFusionError::from)
+}
+
+fn cast_insert_column(
+ name: &str,
+ column: ArrayRef,
+ target_type: &ArrowDataType,
+) -> DFResult<ArrayRef> {
+ if column.data_type() == target_type {
+ return Ok(column);
+ }
+
+ compute::cast(column.as_ref(), target_type).map_err(|e| {
+ DataFusionError::Plan(format!(
+ "Cannot cast MERGE INSERT column '{name}' from {:?} to {:?}: {e}",
+ column.data_type(),
+ target_type
+ ))
+ })
+}
+
/// Remove injected columns from batches, keeping only source columns.
fn strip_non_source_columns(
batches: &[RecordBatch],
@@ -946,7 +983,7 @@ fn strip_non_source_columns(
/// When the INSERT specifies explicit columns (`INSERT (col2, col1) VALUES
(expr2, expr1)`),
/// the output must be reordered to match the table schema so that
`write_arrow_batch`
/// (which reads columns by positional index) maps them correctly.
-fn insert_select_clause(ins: &MergeInsertClause, table_fields: &[String]) ->
String {
+fn insert_select_clause(ins: &MergeInsertClause, table_fields: &[DataField])
-> String {
if ins.columns.is_empty() && ins.value_exprs.is_empty() {
"*".to_string()
} else {
@@ -962,11 +999,11 @@ fn insert_select_clause(ins: &MergeInsertClause,
table_fields: &[String]) -> Str
table_fields
.iter()
.map(|field| {
- let key = field.to_lowercase();
+ let key = field.name().to_lowercase();
match col_expr_map.get(&key) {
- Some(expr) => format!("{expr} AS {}",
quote_identifier(field)),
+ Some(expr) => format!("{expr} AS {}",
quote_identifier(field.name())),
// Column not in INSERT list — fill with NULL
- None => format!("NULL AS {}", quote_identifier(field)),
+ None => format!("NULL AS {}",
quote_identifier(field.name())),
}
})
.collect::<Vec<_>>()
@@ -1546,7 +1583,7 @@ mod tests {
use datafusion::sql::sqlparser::parser::Parser;
use paimon::catalog::{Catalog, Identifier};
use paimon::io::FileIOBuilder;
- use paimon::spec::{DataType, IntType, Schema as PaimonSchema, TableSchema};
+ use paimon::spec::{DataField, DataType, IntType, Schema as PaimonSchema,
TableSchema};
use paimon::{CatalogOptions, FileSystemCatalog, Options};
use tempfile::TempDir;
@@ -1613,6 +1650,42 @@ mod tests {
}
}
+ #[test]
+ fn test_normalize_merge_insert_batch_uses_position() {
+ let table_fields = vec![
+ DataField::new(0, "a".to_string(), DataType::Int(IntType::new())),
+ DataField::new(1, "b".to_string(), DataType::Int(IntType::new())),
+ ];
+ let batch = RecordBatch::try_new(
+ Arc::new(Schema::new(vec![
+ Field::new("b", ArrowDataType::Int32, false),
+ Field::new("x", ArrowDataType::Int32, false),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(vec![100])),
+ Arc::new(Int32Array::from(vec![7])),
+ ],
+ )
+ .unwrap();
+
+ let normalized = normalize_insert_batch_to_table_schema(&batch,
&table_fields).unwrap();
+ let first = normalized
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let second = normalized
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+
+ assert_eq!(normalized.schema().field(0).name(), "a");
+ assert_eq!(normalized.schema().field(1).name(), "b");
+ assert_eq!(first.value(0), 100);
+ assert_eq!(second.value(0), 7);
+ }
+
#[test]
fn test_source_partition_pruning_requires_partition_equality() {
let merge = parse_merge(
diff --git a/crates/paimon/src/arrow/format/blob.rs
b/crates/paimon/src/arrow/format/blob.rs
index 65bd2f3..136170e 100644
--- a/crates/paimon/src/arrow/format/blob.rs
+++ b/crates/paimon/src/arrow/format/blob.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use super::{FilePredicates, FormatFileReader, FormatFileWriter};
+use super::{FilePredicates, FormatFileReader, FormatFileWriter,
FormatWriteResult};
use crate::arrow::build_target_arrow_schema;
use crate::io::{FileRead, FileWrite};
use crate::spec::{BlobDescriptor, DataField, DataType};
@@ -725,7 +725,7 @@ impl FormatFileWriter for BlobFormatWriter {
Ok(())
}
- async fn close(mut self: Box<Self>) -> crate::Result<u64> {
+ async fn close(mut self: Box<Self>) -> crate::Result<FormatWriteResult> {
let index_bytes = encode_delta_varints_write(&self.lengths);
let index_length = index_bytes.len() as i32;
@@ -739,7 +739,7 @@ impl FormatFileWriter for BlobFormatWriter {
let total = self.bytes_written + index_length as u64 +
BLOB_FOOTER_SIZE;
self.writer.close().await?;
- Ok(total)
+ Ok(FormatWriteResult::new(total))
}
}
diff --git a/crates/paimon/src/arrow/format/mod.rs
b/crates/paimon/src/arrow/format/mod.rs
index 2a2fcbd..cd981b8 100644
--- a/crates/paimon/src/arrow/format/mod.rs
+++ b/crates/paimon/src/arrow/format/mod.rs
@@ -19,7 +19,7 @@ mod avro;
pub(crate) mod blob;
mod mosaic;
mod orc;
-mod parquet;
+pub(crate) mod parquet;
mod row;
mod shredding;
#[cfg(feature = "vortex")]
@@ -29,6 +29,7 @@ mod vortex;
pub(crate) use parquet::ParquetFormatWriter;
use crate::io::{FileRead, OutputFile};
+use crate::spec::stats::BinaryTableStats;
use crate::spec::{DataField, Predicate};
use crate::table::{ArrowRecordBatchStream, RowRange};
use crate::Error;
@@ -102,8 +103,40 @@ pub(crate) trait FormatFileWriter: Send {
async fn flush(&mut self) -> crate::Result<()>;
/// Flush and close the writer, finalizing the file on storage.
- /// Returns the total number of bytes written.
- async fn close(self: Box<Self>) -> crate::Result<u64>;
+ async fn close(self: Box<Self>) -> crate::Result<FormatWriteResult>;
+}
+
+pub(crate) struct FormatWriteResult {
+ pub(crate) file_size: u64,
+ pub(crate) value_stats: Option<FormatValueStats>,
+}
+
+pub(crate) struct FormatValueStats {
+ pub(crate) stats: BinaryTableStats,
+ pub(crate) columns: Option<Vec<String>>,
+}
+
+impl FormatWriteResult {
+ pub(crate) fn new(file_size: u64) -> Self {
+ Self {
+ file_size,
+ value_stats: None,
+ }
+ }
+
+ pub(crate) fn with_value_stats(
+ file_size: u64,
+ value_stats: BinaryTableStats,
+ columns: Option<Vec<String>>,
+ ) -> Self {
+ Self {
+ file_size,
+ value_stats: Some(FormatValueStats {
+ stats: value_stats,
+ columns,
+ }),
+ }
+ }
}
/// Create a format reader based on the file extension.
@@ -180,6 +213,7 @@ pub(crate) async fn create_format_writer(
output,
compression,
zstd_level,
+ format_options.cloned().unwrap_or_default(),
));
shredding::ShreddingFormatWriter::create(
writer_factory,
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index 5da5eee..cf55e0b 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -16,10 +16,14 @@
// under the License.
use super::shredding::PhysicalFormatWriterFactory;
-use super::{FilePredicates, FormatFileReader, FormatFileWriter};
+use super::{FilePredicates, FormatFileReader, FormatFileWriter,
FormatWriteResult};
use crate::arrow::filtering::{predicates_may_match_with_schema, StatsAccessor};
use crate::io::{FileRead, OutputFile};
-use crate::spec::{DataField, DataType, Datum, Predicate, PredicateOperator};
+use crate::spec::stats::BinaryTableStats;
+use crate::spec::{
+ BinaryRowBuilder, CoreOptions, DataField, DataType, Datum,
MetadataStatsMode, Predicate,
+ PredicateOperator,
+};
use crate::table::{ArrowRecordBatchStream, RowRange};
use crate::Error;
use arrow_array::{BooleanArray, RecordBatch};
@@ -39,6 +43,7 @@ use parquet::file::metadata::{
use parquet::file::page_index::column_index::ColumnIndexMetaData;
use parquet::file::properties::WriterProperties;
use parquet::file::statistics::Statistics as ParquetStatistics;
+use std::cmp::Ordering;
use std::collections::HashMap;
use std::ops::Range;
use std::sync::Arc;
@@ -49,20 +54,30 @@ pub(crate) struct ParquetFormatReader;
/// Streams data directly to storage via `AsyncArrowWriter` + opendal.
pub(crate) struct ParquetFormatWriter {
inner: AsyncArrowWriter<Box<dyn crate::io::AsyncFileWrite>>,
+ write_fields: Option<Vec<DataField>>,
+ stats_modes: Option<Vec<MetadataStatsMode>>,
+ stats_dense_store: bool,
}
pub(crate) struct ParquetPhysicalWriterFactory {
output: OutputFile,
compression: String,
zstd_level: i32,
+ format_options: HashMap<String, String>,
}
impl ParquetPhysicalWriterFactory {
- pub(crate) fn new(output: &OutputFile, compression: &str, zstd_level: i32)
-> Self {
+ pub(crate) fn new(
+ output: &OutputFile,
+ compression: &str,
+ zstd_level: i32,
+ format_options: HashMap<String, String>,
+ ) -> Self {
Self {
output: output.clone(),
compression: compression.to_string(),
zstd_level,
+ format_options,
}
}
}
@@ -72,11 +87,18 @@ impl PhysicalFormatWriterFactory for
ParquetPhysicalWriterFactory {
async fn create_writer(
&mut self,
schema: arrow_schema::SchemaRef,
- _write_fields: Option<&[DataField]>,
+ write_fields: Option<&[DataField]>,
) -> crate::Result<Box<dyn FormatFileWriter>> {
Ok(Box::new(
- ParquetFormatWriter::new(&self.output, schema, &self.compression,
self.zstd_level)
- .await?,
+ ParquetFormatWriter::new(
+ &self.output,
+ schema,
+ &self.compression,
+ self.zstd_level,
+ write_fields,
+ &self.format_options,
+ )
+ .await?,
))
}
}
@@ -87,11 +109,22 @@ impl ParquetFormatWriter {
schema: arrow_schema::SchemaRef,
compression: &str,
zstd_level: i32,
+ write_fields: Option<&[DataField]>,
+ format_options: &HashMap<String, String>,
) -> crate::Result<Self> {
let async_write = output.async_writer().await?;
let codec = parse_compression(compression, zstd_level);
let inner = create_parquet_arrow_writer(async_write, schema, codec)?;
- Ok(Self { inner })
+ let core_options = CoreOptions::new(format_options);
+ let stats_modes = write_fields
+ .map(|fields|
core_options.metadata_stats_modes(fields.iter().map(DataField::name)))
+ .transpose()?;
+ Ok(Self {
+ inner,
+ write_fields: write_fields.map(|fields| fields.to_vec()),
+ stats_modes,
+ stats_dense_store: core_options.metadata_stats_dense_store(),
+ })
}
}
@@ -154,15 +187,27 @@ impl FormatFileWriter for ParquetFormatWriter {
})
}
- async fn close(mut self: Box<Self>) -> crate::Result<u64> {
- self.inner
+ async fn close(mut self: Box<Self>) -> crate::Result<FormatWriteResult> {
+ let metadata = self
+ .inner
.finish()
.await
.map_err(|e| crate::Error::DataInvalid {
message: format!("Failed to close parquet writer: {e}"),
source: None,
})?;
- Ok(self.inner.bytes_written() as u64)
+ let file_size = self.inner.bytes_written() as u64;
+ if let (Some(write_fields), Some(stats_modes)) = (&self.write_fields,
&self.stats_modes) {
+ let (value_stats, value_stats_cols) =
+ extract_value_stats(&metadata, write_fields, stats_modes,
self.stats_dense_store);
+ Ok(FormatWriteResult::with_value_stats(
+ file_size,
+ value_stats,
+ value_stats_cols,
+ ))
+ } else {
+ Ok(FormatWriteResult::new(file_size))
+ }
}
}
@@ -625,6 +670,272 @@ fn build_row_group_column_indices(
.collect()
}
+fn extract_value_stats(
+ metadata: &ParquetMetaData,
+ write_fields: &[DataField],
+ stats_modes: &[MetadataStatsMode],
+ stats_dense_store: bool,
+) -> (BinaryTableStats, Option<Vec<String>>) {
+ debug_assert_eq!(write_fields.len(), stats_modes.len());
+ let row_groups = metadata.row_groups();
+ let column_indices = row_groups
+ .first()
+ .map(|row_group| build_row_group_column_indices(row_group.columns(),
write_fields))
+ .unwrap_or_else(|| vec![None; write_fields.len()]);
+ let mut column_names = Vec::new();
+ let mut column_types = Vec::new();
+ let mut min_datums = Vec::new();
+ let mut max_datums = Vec::new();
+ let mut null_counts = Vec::new();
+
+ for (field_idx, field) in write_fields.iter().enumerate() {
+ let mode = stats_modes
+ .get(field_idx)
+ .copied()
+ .unwrap_or(MetadataStatsMode::None);
+ let column_stats = if mode == MetadataStatsMode::None {
+ None
+ } else {
+ column_indices
+ .get(field_idx)
+ .copied()
+ .flatten()
+ .and_then(|column_idx| {
+ extract_column_value_stats(row_groups, column_idx,
field.data_type(), mode)
+ })
+ };
+
+ // Non-dense stats stay aligned with write_fields, including
unavailable stats.
+ match column_stats {
+ Some((min_datum, max_datum, null_count)) => {
+ if stats_dense_store {
+ column_names.push(field.name().to_string());
+ }
+ column_types.push(field.data_type().clone());
+ min_datums.push(min_datum);
+ max_datums.push(max_datum);
+ null_counts.push(null_count);
+ }
+ None if !stats_dense_store => {
+ column_types.push(field.data_type().clone());
+ min_datums.push(None);
+ max_datums.push(None);
+ null_counts.push(None);
+ }
+ None => {}
+ }
+ }
+
+ let stats = if column_types.is_empty() {
+ BinaryTableStats::empty()
+ } else {
+ binary_table_stats_from_datums(&column_types, &min_datums,
&max_datums, null_counts)
+ };
+ // Java omits the dense mapping when stats already cover every write field.
+ let value_stats_cols = if stats_dense_store && column_types.len() !=
write_fields.len() {
+ Some(column_names)
+ } else {
+ None
+ };
+ (stats, value_stats_cols)
+}
+
+fn extract_column_value_stats(
+ row_groups: &[RowGroupMetaData],
+ column_idx: usize,
+ data_type: &DataType,
+ mode: MetadataStatsMode,
+) -> Option<(Option<Datum>, Option<Datum>, Option<i64>)> {
+ let collect_min_max = matches!(
+ mode,
+ MetadataStatsMode::Full | MetadataStatsMode::Truncate(_)
+ ) && supports_manifest_min_max(data_type);
+ let mut min_datum: Option<Datum> = None;
+ let mut max_datum: Option<Datum> = None;
+ let mut min_complete = true;
+ let mut max_complete = true;
+ let mut null_count = Some(0_i64);
+ let mut has_stats = false;
+
+ for row_group in row_groups {
+ let Some(stats) = row_group.column(column_idx).statistics() else {
+ min_complete = false;
+ max_complete = false;
+ null_count = None;
+ continue;
+ };
+ has_stats = true;
+
+ match stats
+ .null_count_opt()
+ .and_then(|count| i64::try_from(count).ok())
+ {
+ Some(count) => {
+ if let Some(total) = null_count.as_mut() {
+ *total += count;
+ }
+ }
+ None => null_count = None,
+ }
+
+ let row_group_all_null = stats.null_count_opt() ==
Some(row_group.num_rows().max(0) as u64);
+ if !collect_min_max || row_group_all_null {
+ continue;
+ }
+
+ match parquet_stats_to_datum(stats, data_type, true) {
+ Some(candidate) => {
+ if let Some(current) = &min_datum {
+ match candidate.partial_cmp(current) {
+ Some(Ordering::Less) => min_datum = Some(candidate),
+ Some(Ordering::Equal | Ordering::Greater) => {}
+ None => min_complete = false,
+ }
+ } else {
+ min_datum = Some(candidate);
+ }
+ }
+ None => min_complete = false,
+ }
+
+ match parquet_stats_to_datum(stats, data_type, false) {
+ Some(candidate) => {
+ if let Some(current) = &max_datum {
+ match candidate.partial_cmp(current) {
+ Some(Ordering::Greater) => max_datum = Some(candidate),
+ Some(Ordering::Less | Ordering::Equal) => {}
+ None => max_complete = false,
+ }
+ } else {
+ max_datum = Some(candidate);
+ }
+ }
+ None => max_complete = false,
+ }
+ }
+
+ if !has_stats {
+ return None;
+ }
+ if !min_complete {
+ min_datum = None;
+ }
+ if !max_complete {
+ max_datum = None;
+ }
+ let (min_datum, max_datum) = apply_stats_mode(data_type, mode, min_datum,
max_datum);
+ if min_datum.is_none() && max_datum.is_none() && null_count.is_none() {
+ None
+ } else {
+ Some((min_datum, max_datum, null_count))
+ }
+}
+
+fn supports_manifest_min_max(data_type: &DataType) -> bool {
+ matches!(
+ data_type,
+ DataType::Boolean(_)
+ | DataType::TinyInt(_)
+ | DataType::SmallInt(_)
+ | DataType::Int(_)
+ | DataType::BigInt(_)
+ | DataType::Char(_)
+ | DataType::VarChar(_)
+ | DataType::Decimal(_)
+ | DataType::Double(_)
+ | DataType::Float(_)
+ | DataType::Date(_)
+ | DataType::Time(_)
+ | DataType::LocalZonedTimestamp(_)
+ | DataType::Timestamp(_)
+ )
+}
+
+fn apply_stats_mode(
+ data_type: &DataType,
+ mode: MetadataStatsMode,
+ min_datum: Option<Datum>,
+ max_datum: Option<Datum>,
+) -> (Option<Datum>, Option<Datum>) {
+ let MetadataStatsMode::Truncate(length) = mode else {
+ return (min_datum, max_datum);
+ };
+ match data_type {
+ DataType::Char(_) | DataType::VarChar(_) => {
+ let min = min_datum.map(|datum| truncate_string_min_datum(datum,
length));
+ let max = match max_datum {
+ Some(datum) => match truncate_string_max_datum(datum, length) {
+ Some(max) => Some(max),
+ None => return (None, None),
+ },
+ None => None,
+ };
+ (min, max)
+ }
+ _ => (min_datum, max_datum),
+ }
+}
+
+fn truncate_string_min_datum(datum: Datum, length: usize) -> Datum {
+ match datum {
+ Datum::String(value) => Datum::String(truncate_string_min(&value,
length)),
+ other => other,
+ }
+}
+
+fn truncate_string_max_datum(datum: Datum, length: usize) -> Option<Datum> {
+ match datum {
+ Datum::String(value) => truncate_string_max(&value,
length).map(Datum::String),
+ other => Some(other),
+ }
+}
+
+fn truncate_string_min(value: &str, length: usize) -> String {
+ value.chars().take(length).collect()
+}
+
+fn truncate_string_max(value: &str, length: usize) -> Option<String> {
+ let char_count = value.chars().count();
+ if char_count <= length {
+ return Some(value.to_string());
+ }
+
+ let mut chars: Vec<char> = value.chars().take(length).collect();
+ for idx in (0..chars.len()).rev() {
+ if let Some(next) = char::from_u32(chars[idx] as u32 + 1) {
+ chars.truncate(idx);
+ chars.push(next);
+ return Some(chars.into_iter().collect());
+ }
+ }
+ None
+}
+
+fn binary_table_stats_from_datums(
+ column_types: &[DataType],
+ min_datums: &[Option<Datum>],
+ max_datums: &[Option<Datum>],
+ null_counts: Vec<Option<i64>>,
+) -> BinaryTableStats {
+ let mut min_builder = BinaryRowBuilder::new(column_types.len() as i32);
+ let mut max_builder = BinaryRowBuilder::new(column_types.len() as i32);
+ for (pos, data_type) in column_types.iter().enumerate() {
+ match &min_datums[pos] {
+ Some(datum) => min_builder.write_datum(pos, datum, data_type),
+ None => min_builder.set_null_at(pos),
+ }
+ match &max_datums[pos] {
+ Some(datum) => max_builder.write_datum(pos, datum, data_type),
+ None => max_builder.set_null_at(pos),
+ }
+ }
+ BinaryTableStats::new(
+ min_builder.build_serialized(),
+ max_builder.build_serialized(),
+ null_counts,
+ )
+}
+
// ---------------------------------------------------------------------------
// Page-index (ColumnIndex / OffsetIndex) pruning
// ---------------------------------------------------------------------------
@@ -984,17 +1295,33 @@ fn parquet_stats_to_datum(
.copied()
.map(Datum::Long)
}
- (ParquetStatistics::Int64(stats), DataType::Timestamp(ts)) if
ts.precision() <= 3 => {
+ (ParquetStatistics::Int32(stats), DataType::Decimal(d)) => {
exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
.copied()
- .map(|millis| Datum::Timestamp { millis, nanos: 0 })
+ .map(|unscaled| Datum::Decimal {
+ unscaled: unscaled as i128,
+ precision: d.precision(),
+ scale: d.scale(),
+ })
}
- (ParquetStatistics::Int64(stats), DataType::LocalZonedTimestamp(ts))
- if ts.precision() <= 3 =>
- {
+ (ParquetStatistics::Int64(stats), DataType::Decimal(d)) => {
exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
.copied()
- .map(|millis| Datum::LocalZonedTimestamp { millis, nanos: 0 })
+ .map(|unscaled| Datum::Decimal {
+ unscaled: unscaled as i128,
+ precision: d.precision(),
+ scale: d.scale(),
+ })
+ }
+ (ParquetStatistics::Int64(stats), DataType::Timestamp(ts)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .copied()
+ .and_then(|value| timestamp_datum_from_parquet_i64(value,
ts.precision(), false))
+ }
+ (ParquetStatistics::Int64(stats), DataType::LocalZonedTimestamp(ts))
=> {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .copied()
+ .and_then(|value| timestamp_datum_from_parquet_i64(value,
ts.precision(), true))
}
(ParquetStatistics::Float(stats), DataType::Float(_)) => {
exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
@@ -1022,10 +1349,58 @@ fn parquet_stats_to_datum(
exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
.map(|value| Datum::Bytes(value.data().to_vec()))
}
+ (ParquetStatistics::ByteArray(stats), DataType::Decimal(d)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .and_then(|value| signed_be_bytes_to_i128(value.data()))
+ .map(|unscaled| Datum::Decimal {
+ unscaled,
+ precision: d.precision(),
+ scale: d.scale(),
+ })
+ }
+ (ParquetStatistics::FixedLenByteArray(stats), DataType::Decimal(d)) =>
{
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .and_then(|value| signed_be_bytes_to_i128(value.data()))
+ .map(|unscaled| Datum::Decimal {
+ unscaled,
+ precision: d.precision(),
+ scale: d.scale(),
+ })
+ }
_ => None,
}
}
+fn timestamp_datum_from_parquet_i64(
+ value: i64,
+ precision: u32,
+ local_zoned: bool,
+) -> Option<Datum> {
+ let (millis, nanos) = match precision {
+ 0..=3 => (value, 0),
+ 4..=6 => (
+ value.div_euclid(1_000),
+ (value.rem_euclid(1_000) * 1_000) as i32,
+ ),
+ _ => return None,
+ };
+ if local_zoned {
+ Some(Datum::LocalZonedTimestamp { millis, nanos })
+ } else {
+ Some(Datum::Timestamp { millis, nanos })
+ }
+}
+
+fn signed_be_bytes_to_i128(bytes: &[u8]) -> Option<i128> {
+ if bytes.is_empty() || bytes.len() > 16 {
+ return None;
+ }
+ let sign_extend = if bytes[0] & 0x80 == 0 { 0 } else { 0xff };
+ let mut padded = [sign_extend; 16];
+ padded[16 - bytes.len()..].copy_from_slice(bytes);
+ Some(i128::from_be_bytes(padded))
+}
+
fn exact_parquet_value<'a, T>(
is_min: bool,
min: Option<&'a T>,
@@ -1766,14 +2141,14 @@ mod tests {
let schema = writer_arrow_schema();
let mut writer: Box<dyn FormatFileWriter> = Box::new(
- ParquetFormatWriter::new(&output, schema.clone(), "zstd", 1)
+ ParquetFormatWriter::new(&output, schema.clone(), "zstd", 1, None,
&HashMap::new())
.await
.unwrap(),
);
let batch = writer_test_batch(&schema, vec![1, 2, 3], vec![10, 20,
30]);
writer.write(&batch).await.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
// Verify valid parquet by reading back
let bytes = file_io.new_input(path).unwrap().read().await.unwrap();
@@ -1791,7 +2166,7 @@ mod tests {
let schema = writer_arrow_schema();
let mut writer: Box<dyn FormatFileWriter> = Box::new(
- ParquetFormatWriter::new(&output, schema.clone(), "zstd", 1)
+ ParquetFormatWriter::new(&output, schema.clone(), "zstd", 1, None,
&HashMap::new())
.await
.unwrap(),
);
@@ -1804,7 +2179,7 @@ mod tests {
.write(&writer_test_batch(&schema, vec![3, 4, 5], vec![30, 40,
50]))
.await
.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
let bytes = file_io.new_input(path).unwrap().read().await.unwrap();
let reader =
@@ -1844,12 +2219,12 @@ mod tests {
let path = "memory:/test_parquet_inline_vector.parquet";
let output = file_io.new_output(path).unwrap();
let mut writer: Box<dyn FormatFileWriter> = Box::new(
- ParquetFormatWriter::new(&output, schema.clone(), "zstd", 1)
+ ParquetFormatWriter::new(&output, schema.clone(), "zstd", 1, None,
&HashMap::new())
.await
.unwrap(),
);
writer.write(&batch).await.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
let bytes = file_io.new_input(path).unwrap().read().await.unwrap();
let reader =
@@ -1936,7 +2311,7 @@ mod tests {
.await
.unwrap();
writer.write(&batch).await.unwrap();
- let file_size = writer.close().await.unwrap();
+ let file_size = writer.close().await.unwrap().file_size;
let raw_bytes =
file_io.new_input(&path).unwrap().read().await.unwrap();
let raw_batches =
@@ -2069,7 +2444,7 @@ mod tests {
.write(&make_batch(vec![3], &late_variants))
.await
.unwrap();
- let file_size = writer.close().await.unwrap();
+ let file_size = writer.close().await.unwrap().file_size;
let raw_bytes =
file_io.new_input(&path).unwrap().read().await.unwrap();
let raw_batches =
@@ -2153,7 +2528,7 @@ mod tests {
.write(&writer_test_batch(&schema, ids, values))
.await
.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
}
buf
}
@@ -2296,7 +2671,7 @@ mod tests {
.write(&writer_test_batch(&schema, ids, values))
.await
.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
}
buf
}
@@ -2361,7 +2736,7 @@ mod tests {
.write(&writer_test_batch(&schema, ids, values))
.await
.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
}
let metadata = load_metadata_with_page_index(&buf, true);
let fields = vec![int_field("id"), int_field("value")];
@@ -2417,7 +2792,7 @@ mod tests {
let mut writer =
AsyncArrowWriter::try_new(&mut buf, schema.clone(),
Some(props)).unwrap();
writer.write(&batch).await.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
}
buf
}
diff --git a/crates/paimon/src/arrow/format/row.rs
b/crates/paimon/src/arrow/format/row.rs
index 0f86993..94b740f 100644
--- a/crates/paimon/src/arrow/format/row.rs
+++ b/crates/paimon/src/arrow/format/row.rs
@@ -20,7 +20,7 @@
//! Layout reference:
//! `org.apache.paimon.format.row.RowFormatWriter` in paimon-java.
-use super::{FilePredicates, FormatFileReader, FormatFileWriter};
+use super::{FilePredicates, FormatFileReader, FormatFileWriter,
FormatWriteResult};
use crate::arrow::{
arrow_to_paimon_type, build_target_arrow_schema, is_variant_arrow_fields,
paimon_type_to_arrow,
variant_arrow_type,
@@ -172,7 +172,7 @@ impl FormatFileWriter for RowFormatWriter {
self.flush_block().await
}
- async fn close(mut self: Box<Self>) -> crate::Result<u64> {
+ async fn close(mut self: Box<Self>) -> crate::Result<FormatWriteResult> {
self.flush_block().await?;
let index_offset = self.bytes_written;
@@ -206,7 +206,7 @@ impl FormatFileWriter for RowFormatWriter {
self.writer.write(Bytes::from(footer_bytes)).await?;
self.bytes_written += FOOTER_SIZE;
self.writer.close().await?;
- Ok(self.bytes_written)
+ Ok(FormatWriteResult::new(self.bytes_written))
}
}
diff --git a/crates/paimon/src/arrow/format/shredding.rs
b/crates/paimon/src/arrow/format/shredding.rs
index d33e98a..df283a9 100644
--- a/crates/paimon/src/arrow/format/shredding.rs
+++ b/crates/paimon/src/arrow/format/shredding.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use super::{FilePredicates, FormatFileReader, FormatFileWriter};
+use super::{FilePredicates, FormatFileReader, FormatFileWriter,
FormatWriteResult};
use crate::arrow::build_target_arrow_schema;
use crate::arrow::shredding::{
assemble_shredded_variant_batch, batch_to_shredded_physical,
@@ -289,12 +289,12 @@ impl FormatFileWriter for ShreddingFormatWriter {
}
}
- async fn close(mut self: Box<Self>) -> crate::Result<u64> {
+ async fn close(mut self: Box<Self>) -> crate::Result<FormatWriteResult> {
self.finalize_inferred_writer().await?;
match std::mem::replace(&mut self.state, ShreddingWriterState::Closed)
{
ShreddingWriterState::Ready { inner, .. } => inner.close().await,
ShreddingWriterState::Infer { .. } => unreachable!("infer writer
finalized above"),
- ShreddingWriterState::Closed => Ok(0),
+ ShreddingWriterState::Closed => Ok(FormatWriteResult::new(0)),
}
}
}
diff --git a/crates/paimon/src/arrow/format/vortex.rs
b/crates/paimon/src/arrow/format/vortex.rs
index 12250a7..631f8ad 100644
--- a/crates/paimon/src/arrow/format/vortex.rs
+++ b/crates/paimon/src/arrow/format/vortex.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use super::{FilePredicates, FormatFileReader, FormatFileWriter};
+use super::{FilePredicates, FormatFileReader, FormatFileWriter,
FormatWriteResult};
use crate::arrow::residual::{
filter_record_batch_by_predicates, same_data_field, widen_scan_fields,
};
@@ -480,7 +480,7 @@ impl FormatFileWriter for VortexFormatWriter {
Ok(())
}
- async fn close(self: Box<Self>) -> crate::Result<u64> {
+ async fn close(self: Box<Self>) -> crate::Result<FormatWriteResult> {
let this = *self;
let VortexFormatWriter {
dtype,
@@ -502,7 +502,7 @@ impl FormatFileWriter for VortexFormatWriter {
output.write(bytes::Bytes::from(buffer)).await?;
bytes_written.store(size, Ordering::Relaxed);
- Ok(size)
+ Ok(FormatWriteResult::new(size))
}
}
@@ -628,10 +628,10 @@ mod tests {
.enable_all()
.build()
.unwrap();
- let bytes = verifier_runtime
+ let result = verifier_runtime
.block_on(async { Box::new(writer).close().await })
.unwrap();
- assert!(bytes > 0);
+ assert!(result.file_size > 0);
}
#[test]
@@ -701,8 +701,8 @@ mod tests {
let batch = test_batch(&schema, vec![1, 2, 3], vec![10, 20, 30]);
writer.write(&batch).await.unwrap();
- let bytes = writer.close().await.unwrap();
- assert!(bytes > 0);
+ let result = writer.close().await.unwrap();
+ assert!(result.file_size > 0);
// Read back using VortexFormatReader.
let input = file_io.new_input(path).unwrap();
@@ -767,7 +767,7 @@ mod tests {
)
.unwrap();
writer.write(&batch).await.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
let input = file_io.new_input(path).unwrap();
let file_reader = input.reader().await.unwrap();
@@ -836,7 +836,7 @@ mod tests {
.write(&test_batch(&schema, vec![3, 4, 5], vec![30, 40, 50]))
.await
.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
let input = file_io.new_input(path).unwrap();
let file_reader = input.reader().await.unwrap();
@@ -897,7 +897,7 @@ mod tests {
))
.await
.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
let input = file_io.new_input(path).unwrap();
let file_reader = input.reader().await.unwrap();
@@ -965,7 +965,7 @@ mod tests {
))
.await
.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
let input = file_io.new_input(path).unwrap();
let file_reader = input.reader().await.unwrap();
@@ -1022,7 +1022,7 @@ mod tests {
)
.unwrap();
writer.write(&batch).await.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
let input = file_io.new_input(path).unwrap();
let file_bytes = input.read().await.unwrap();
@@ -1098,7 +1098,7 @@ mod tests {
))
.await
.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
let input = file_io.new_input(path).unwrap();
let file_reader = input.reader().await.unwrap();
@@ -1229,7 +1229,7 @@ mod tests {
))
.await
.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
let fields = test_file_fields();
let builder = PredicateBuilder::new(&fields);
@@ -1349,7 +1349,7 @@ mod tests {
))
.await
.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
let input = file_io.new_input(path).unwrap();
let file_reader = input.reader().await.unwrap();
diff --git a/crates/paimon/src/spec/binary_row.rs
b/crates/paimon/src/spec/binary_row.rs
index d96fa95..4a4017f 100644
--- a/crates/paimon/src/spec/binary_row.rs
+++ b/crates/paimon/src/spec/binary_row.rs
@@ -859,48 +859,24 @@ pub fn extract_datum_from_arrow(
}
DataType::Variant(_) => extract_variant_datum_from_arrow(col, row_idx,
col_idx)?,
DataType::Timestamp(ts) => {
- if ts.precision() <= 3 {
- let arr = col
- .as_any()
- .downcast_ref::<arrow_array::TimestampMillisecondArray>()
- .ok_or_else(|| type_mismatch_err("Timestamp(ms)",
col_idx))?;
- Datum::Timestamp {
- millis: arr.value(row_idx),
- nanos: 0,
- }
- } else {
- let arr = col
- .as_any()
- .downcast_ref::<arrow_array::TimestampMicrosecondArray>()
- .ok_or_else(|| type_mismatch_err("Timestamp(us)",
col_idx))?;
- let micros = arr.value(row_idx);
- Datum::Timestamp {
- millis: micros.div_euclid(1_000),
- nanos: (micros.rem_euclid(1_000) * 1_000) as i32,
- }
- }
+ let (millis, nanos) = extract_timestamp_parts_from_arrow(
+ col,
+ row_idx,
+ col_idx,
+ ts.precision(),
+ "Timestamp",
+ )?;
+ Datum::Timestamp { millis, nanos }
}
DataType::LocalZonedTimestamp(ts) => {
- if ts.precision() <= 3 {
- let arr = col
- .as_any()
- .downcast_ref::<arrow_array::TimestampMillisecondArray>()
- .ok_or_else(||
type_mismatch_err("LocalZonedTimestamp(ms)", col_idx))?;
- Datum::LocalZonedTimestamp {
- millis: arr.value(row_idx),
- nanos: 0,
- }
- } else {
- let arr = col
- .as_any()
- .downcast_ref::<arrow_array::TimestampMicrosecondArray>()
- .ok_or_else(||
type_mismatch_err("LocalZonedTimestamp(us)", col_idx))?;
- let micros = arr.value(row_idx);
- Datum::LocalZonedTimestamp {
- millis: micros.div_euclid(1_000),
- nanos: (micros.rem_euclid(1_000) * 1_000) as i32,
- }
- }
+ let (millis, nanos) = extract_timestamp_parts_from_arrow(
+ col,
+ row_idx,
+ col_idx,
+ ts.precision(),
+ "LocalZonedTimestamp",
+ )?;
+ Datum::LocalZonedTimestamp { millis, nanos }
}
_ => {
return Err(crate::Error::Unsupported {
@@ -915,6 +891,55 @@ pub fn extract_datum_from_arrow(
Ok(Some(datum))
}
+fn extract_timestamp_parts_from_arrow(
+ col: &std::sync::Arc<dyn arrow_array::Array>,
+ row_idx: usize,
+ col_idx: usize,
+ precision: u32,
+ expected: &str,
+) -> crate::Result<(i64, i32)> {
+ match precision {
+ 0..=3 => {
+ let arr = col
+ .as_any()
+ .downcast_ref::<arrow_array::TimestampMillisecondArray>()
+ .ok_or_else(|| type_mismatch_err(&format!("{expected}(ms)"),
col_idx))?;
+ Ok((arr.value(row_idx), 0))
+ }
+ 4..=6 => {
+ let arr = col
+ .as_any()
+ .downcast_ref::<arrow_array::TimestampMicrosecondArray>()
+ .ok_or_else(|| type_mismatch_err(&format!("{expected}(us)"),
col_idx))?;
+ Ok(timestamp_parts_from_micros(arr.value(row_idx)))
+ }
+ 7..=9 => {
+ let arr = col
+ .as_any()
+ .downcast_ref::<arrow_array::TimestampNanosecondArray>()
+ .ok_or_else(|| type_mismatch_err(&format!("{expected}(ns)"),
col_idx))?;
+ Ok(timestamp_parts_from_nanos(arr.value(row_idx)))
+ }
+ _ => Err(crate::Error::Unsupported {
+ message: format!("Unsupported {expected} precision {precision}"),
+ }),
+ }
+}
+
+fn timestamp_parts_from_micros(micros: i64) -> (i64, i32) {
+ (
+ micros.div_euclid(1_000),
+ (micros.rem_euclid(1_000) * 1_000) as i32,
+ )
+}
+
+fn timestamp_parts_from_nanos(nanos: i64) -> (i64, i32) {
+ (
+ nanos.div_euclid(1_000_000),
+ nanos.rem_euclid(1_000_000) as i32,
+ )
+}
+
fn encode_variant_bytes(value: &[u8], metadata: &[u8]) ->
crate::Result<Vec<u8>> {
VariantType::validate_payload(value, metadata)?;
let mut bytes = Vec::with_capacity(4 + value.len() + metadata.len());
@@ -1045,6 +1070,7 @@ enum TypedColumn<'a> {
Variant(&'a arrow_array::StructArray),
TimestampMs(&'a arrow_array::TimestampMillisecondArray),
TimestampUs(&'a arrow_array::TimestampMicrosecondArray),
+ TimestampNs(&'a arrow_array::TimestampNanosecondArray),
}
/// Downcast Arrow columns once, returning typed references paired with their
DataType.
@@ -1059,118 +1085,134 @@ fn downcast_columns<'a>(
.map(|&col_idx| {
let field = &fields[col_idx];
let col = batch.column(col_idx);
- let typed =
- match field.data_type() {
- DataType::Boolean(_) => TypedColumn::Boolean(
- col.as_any()
- .downcast_ref()
- .ok_or_else(|| type_mismatch_err("Boolean",
col_idx))?,
- ),
- DataType::TinyInt(_) => TypedColumn::Int8(
- col.as_any()
- .downcast_ref()
- .ok_or_else(|| type_mismatch_err("TinyInt",
col_idx))?,
- ),
- DataType::SmallInt(_) => TypedColumn::Int16(
- col.as_any()
- .downcast_ref()
- .ok_or_else(|| type_mismatch_err("SmallInt",
col_idx))?,
- ),
- DataType::Int(_) => TypedColumn::Int32(
- col.as_any()
- .downcast_ref()
- .ok_or_else(|| type_mismatch_err("Int", col_idx))?,
- ),
- DataType::BigInt(_) => TypedColumn::Int64(
- col.as_any()
- .downcast_ref()
- .ok_or_else(|| type_mismatch_err("BigInt",
col_idx))?,
- ),
- DataType::Float(_) => TypedColumn::Float32(
- col.as_any()
- .downcast_ref()
- .ok_or_else(|| type_mismatch_err("Float",
col_idx))?,
- ),
- DataType::Double(_) => TypedColumn::Float64(
- col.as_any()
- .downcast_ref()
- .ok_or_else(|| type_mismatch_err("Double",
col_idx))?,
- ),
- DataType::Char(_) | DataType::VarChar(_) => {
- if let Some(arr) =
col.as_any().downcast_ref::<arrow_array::StringArray>() {
- TypedColumn::Utf8(arr)
- } else if let Some(arr) =
-
col.as_any().downcast_ref::<arrow_array::StringViewArray>()
- {
- TypedColumn::Utf8View(arr)
- } else if let Some(arr) =
-
col.as_any().downcast_ref::<arrow_array::LargeStringArray>()
- {
- TypedColumn::LargeUtf8(arr)
- } else {
- return Err(type_mismatch_err("String", col_idx));
- }
+ let typed = match field.data_type() {
+ DataType::Boolean(_) => TypedColumn::Boolean(
+ col.as_any()
+ .downcast_ref()
+ .ok_or_else(|| type_mismatch_err("Boolean", col_idx))?,
+ ),
+ DataType::TinyInt(_) => TypedColumn::Int8(
+ col.as_any()
+ .downcast_ref()
+ .ok_or_else(|| type_mismatch_err("TinyInt", col_idx))?,
+ ),
+ DataType::SmallInt(_) => TypedColumn::Int16(
+ col.as_any()
+ .downcast_ref()
+ .ok_or_else(|| type_mismatch_err("SmallInt",
col_idx))?,
+ ),
+ DataType::Int(_) => TypedColumn::Int32(
+ col.as_any()
+ .downcast_ref()
+ .ok_or_else(|| type_mismatch_err("Int", col_idx))?,
+ ),
+ DataType::BigInt(_) => TypedColumn::Int64(
+ col.as_any()
+ .downcast_ref()
+ .ok_or_else(|| type_mismatch_err("BigInt", col_idx))?,
+ ),
+ DataType::Float(_) => TypedColumn::Float32(
+ col.as_any()
+ .downcast_ref()
+ .ok_or_else(|| type_mismatch_err("Float", col_idx))?,
+ ),
+ DataType::Double(_) => TypedColumn::Float64(
+ col.as_any()
+ .downcast_ref()
+ .ok_or_else(|| type_mismatch_err("Double", col_idx))?,
+ ),
+ DataType::Char(_) | DataType::VarChar(_) => {
+ if let Some(arr) =
col.as_any().downcast_ref::<arrow_array::StringArray>() {
+ TypedColumn::Utf8(arr)
+ } else if let Some(arr) =
+
col.as_any().downcast_ref::<arrow_array::StringViewArray>()
+ {
+ TypedColumn::Utf8View(arr)
+ } else if let Some(arr) =
+
col.as_any().downcast_ref::<arrow_array::LargeStringArray>()
+ {
+ TypedColumn::LargeUtf8(arr)
+ } else {
+ return Err(type_mismatch_err("String", col_idx));
}
- DataType::Date(_) => TypedColumn::Date32(
+ }
+ DataType::Date(_) => TypedColumn::Date32(
+ col.as_any()
+ .downcast_ref()
+ .ok_or_else(|| type_mismatch_err("Date", col_idx))?,
+ ),
+ DataType::Decimal(d) => TypedColumn::Decimal128(
+ col.as_any()
+ .downcast_ref()
+ .ok_or_else(|| type_mismatch_err("Decimal", col_idx))?,
+ d.precision(),
+ d.scale(),
+ ),
+ DataType::Binary(_) | DataType::VarBinary(_) =>
TypedColumn::Binary(
+ col.as_any()
+ .downcast_ref()
+ .ok_or_else(|| type_mismatch_err("Binary", col_idx))?,
+ ),
+ DataType::Variant(_) => {
+ let arr = col
+ .as_any()
+ .downcast_ref()
+ .ok_or_else(|| type_mismatch_err("Variant", col_idx))?;
+ validate_variant_struct_array(arr, col_idx)?;
+ TypedColumn::Variant(arr)
+ }
+ DataType::Timestamp(ts) => match ts.precision() {
+ 0..=3 => TypedColumn::TimestampMs(
col.as_any()
.downcast_ref()
- .ok_or_else(|| type_mismatch_err("Date",
col_idx))?,
+ .ok_or_else(|| type_mismatch_err("Timestamp(ms)",
col_idx))?,
),
- DataType::Decimal(d) => TypedColumn::Decimal128(
+ 4..=6 => TypedColumn::TimestampUs(
col.as_any()
.downcast_ref()
- .ok_or_else(|| type_mismatch_err("Decimal",
col_idx))?,
- d.precision(),
- d.scale(),
+ .ok_or_else(|| type_mismatch_err("Timestamp(us)",
col_idx))?,
),
- DataType::Binary(_) | DataType::VarBinary(_) =>
TypedColumn::Binary(
+ 7..=9 => TypedColumn::TimestampNs(
col.as_any()
.downcast_ref()
- .ok_or_else(|| type_mismatch_err("Binary",
col_idx))?,
+ .ok_or_else(|| type_mismatch_err("Timestamp(ns)",
col_idx))?,
),
- DataType::Variant(_) => {
- let arr = col
- .as_any()
- .downcast_ref()
- .ok_or_else(|| type_mismatch_err("Variant",
col_idx))?;
- validate_variant_struct_array(arr, col_idx)?;
- TypedColumn::Variant(arr)
- }
- DataType::Timestamp(ts) => {
- if ts.precision() <= 3 {
- TypedColumn::TimestampMs(
- col.as_any()
- .downcast_ref()
- .ok_or_else(||
type_mismatch_err("Timestamp(ms)", col_idx))?,
- )
- } else {
- TypedColumn::TimestampUs(
- col.as_any()
- .downcast_ref()
- .ok_or_else(||
type_mismatch_err("Timestamp(us)", col_idx))?,
- )
- }
- }
- DataType::LocalZonedTimestamp(ts) => {
- if ts.precision() <= 3 {
-
TypedColumn::TimestampMs(col.as_any().downcast_ref().ok_or_else(
- ||
type_mismatch_err("LocalZonedTimestamp(ms)", col_idx),
- )?)
- } else {
-
TypedColumn::TimestampUs(col.as_any().downcast_ref().ok_or_else(
- ||
type_mismatch_err("LocalZonedTimestamp(us)", col_idx),
- )?)
- }
- }
- other => {
+ _ => {
return Err(crate::Error::Unsupported {
- message: format!(
- "Unsupported data type {:?} for batch column
downcast at column {}",
- other, col_idx
- ),
+ message: format!("Unsupported Timestamp precision
{}", ts.precision()),
});
}
- };
+ },
+ DataType::LocalZonedTimestamp(ts) => {
+ match ts.precision() {
+ 0..=3 =>
TypedColumn::TimestampMs(col.as_any().downcast_ref().ok_or_else(
+ || type_mismatch_err("LocalZonedTimestamp(ms)",
col_idx),
+ )?),
+ 4..=6 =>
TypedColumn::TimestampUs(col.as_any().downcast_ref().ok_or_else(
+ || type_mismatch_err("LocalZonedTimestamp(us)",
col_idx),
+ )?),
+ 7..=9 =>
TypedColumn::TimestampNs(col.as_any().downcast_ref().ok_or_else(
+ || type_mismatch_err("LocalZonedTimestamp(ns)",
col_idx),
+ )?),
+ _ => {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Unsupported LocalZonedTimestamp precision
{}",
+ ts.precision()
+ ),
+ });
+ }
+ }
+ }
+ other => {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Unsupported data type {:?} for batch column
downcast at column {}",
+ other, col_idx
+ ),
+ });
+ }
+ };
Ok((typed, field))
})
.collect()
@@ -1339,9 +1381,15 @@ fn write_typed_value(
if arr.is_null(row_idx) {
builder.set_null_at(pos);
} else {
- let micros = arr.value(row_idx);
- let millis = micros.div_euclid(1_000);
- let nanos = (micros.rem_euclid(1_000) * 1_000) as i32;
+ let (millis, nanos) =
timestamp_parts_from_micros(arr.value(row_idx));
+ builder.write_timestamp_non_compact(pos, millis, nanos);
+ }
+ }
+ TypedColumn::TimestampNs(arr) => {
+ if arr.is_null(row_idx) {
+ builder.set_null_at(pos);
+ } else {
+ let (millis, nanos) =
timestamp_parts_from_nanos(arr.value(row_idx));
builder.write_timestamp_non_compact(pos, millis, nanos);
}
}
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index 1e1df3e..60d33ff 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -55,6 +55,14 @@ const CHANGELOG_FILE_PREFIX_OPTION: &str =
"changelog-file.prefix";
const CHANGELOG_FILE_FORMAT_OPTION: &str = "changelog-file.format";
const CHANGELOG_FILE_COMPRESSION_OPTION: &str = "changelog-file.compression";
const CHANGELOG_FILE_STATS_MODE_OPTION: &str = "changelog-file.stats-mode";
+const METADATA_STATS_MODE_OPTION: &str = "metadata.stats-mode";
+const METADATA_STATS_DENSE_STORE_OPTION: &str = "metadata.stats-dense-store";
+const METADATA_STATS_KEEP_FIRST_N_COLUMNS_OPTION: &str =
"metadata.stats-keep-first-n-columns";
+const DEFAULT_METADATA_STATS_MODE: &str = "truncate(16)";
+const DEFAULT_METADATA_STATS_DENSE_STORE: bool = true;
+const DEFAULT_METADATA_STATS_KEEP_FIRST_N_COLUMNS: i32 = -1;
+const FIELDS_PREFIX: &str = "fields";
+const STATS_MODE_SUFFIX: &str = "stats-mode";
const ROW_TRACKING_ENABLED_OPTION: &str = "row-tracking.enabled";
pub(crate) const TABLE_TYPE_OPTION: &str = "type";
pub(crate) const FORMAT_TABLE_TYPE: &str = "format-table";
@@ -160,6 +168,56 @@ pub enum GlobalIndexSearchMode {
Detail,
}
+/// Metadata stats collection mode.
+///
+/// Reference: Java `SimpleColStatsCollector.from`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum MetadataStatsMode {
+ None,
+ Counts,
+ Full,
+ Truncate(usize),
+}
+
+impl MetadataStatsMode {
+ pub(crate) fn parse(option_name: &str, value: &str) -> crate::Result<Self>
{
+ let value = value.trim();
+ let upper = value.to_ascii_uppercase();
+ match upper.as_str() {
+ "NONE" => Ok(Self::None),
+ "COUNTS" => Ok(Self::Counts),
+ "FULL" => Ok(Self::Full),
+ _ => {
+ let Some(length) = upper
+ .strip_prefix("TRUNCATE(")
+ .and_then(|value| value.strip_suffix(')'))
+ else {
+ return Err(crate::Error::Unsupported {
+ message: format!("Unsupported {option_name}:
'{value}'"),
+ });
+ };
+ let length = length
+ .parse::<usize>()
+ .map_err(|e| crate::Error::DataInvalid {
+ message: format!(
+ "Option '{option_name}' must use truncate(N) with
a positive integer, got: {value}"
+ ),
+ source: Some(Box::new(e)),
+ })?;
+ if length == 0 {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Option '{option_name}' must use truncate(N) with
N > 0, got: {value}"
+ ),
+ source: None,
+ });
+ }
+ Ok(Self::Truncate(length))
+ }
+ }
+ }
+}
+
/// Bucket function used to map bucket keys to fixed bucket ids.
///
/// Reference: Java `CoreOptions.BucketFunctionType`.
@@ -846,6 +904,79 @@ impl<'a> CoreOptions<'a> {
.map(String::as_str)
}
+ /// Whether metadata stats should omit columns without collected stats.
+ pub(crate) fn metadata_stats_dense_store(&self) -> bool {
+ self.options
+ .get(METADATA_STATS_DENSE_STORE_OPTION)
+ .map(|value| value.eq_ignore_ascii_case("true"))
+ .unwrap_or(DEFAULT_METADATA_STATS_DENSE_STORE)
+ }
+
+ /// Table-wide metadata stats mode.
+ pub(crate) fn metadata_stats_mode(&self) ->
crate::Result<MetadataStatsMode> {
+ let value = self
+ .options
+ .get(METADATA_STATS_MODE_OPTION)
+ .map(String::as_str)
+ .unwrap_or(DEFAULT_METADATA_STATS_MODE);
+ MetadataStatsMode::parse(METADATA_STATS_MODE_OPTION, value)
+ }
+
+ /// Number of leading columns whose stats should be kept.
+ ///
+ /// A negative value means the option is ignored, matching Java Paimon.
+ pub(crate) fn metadata_stats_keep_first_n_columns(&self) ->
crate::Result<i32> {
+ self.options
+ .get(METADATA_STATS_KEEP_FIRST_N_COLUMNS_OPTION)
+ .map(|value| {
+ value.parse::<i32>().map_err(|e| crate::Error::DataInvalid {
+ message: format!(
+ "Invalid value for
{METADATA_STATS_KEEP_FIRST_N_COLUMNS_OPTION}: '{value}'"
+ ),
+ source: Some(Box::new(e)),
+ })
+ })
+ .transpose()
+ .map(|value|
value.unwrap_or(DEFAULT_METADATA_STATS_KEEP_FIRST_N_COLUMNS))
+ }
+
+ /// Per-field metadata stats mode.
+ pub(crate) fn field_metadata_stats_mode(
+ &self,
+ field_name: &str,
+ ) -> crate::Result<Option<MetadataStatsMode>> {
+ let option_name =
format!("{FIELDS_PREFIX}.{field_name}.{STATS_MODE_SUFFIX}");
+ self.options
+ .get(&option_name)
+ .map(|value| MetadataStatsMode::parse(&option_name, value))
+ .transpose()
+ }
+
+ /// Resolve metadata stats modes for fields using Java's priority:
+ /// field override > keep-first-n > table-wide mode.
+ pub(crate) fn metadata_stats_modes<'b, I>(
+ &self,
+ field_names: I,
+ ) -> crate::Result<Vec<MetadataStatsMode>>
+ where
+ I: IntoIterator<Item = &'b str>,
+ {
+ let table_mode = self.metadata_stats_mode()?;
+ let keep_first_n = self.metadata_stats_keep_first_n_columns()?;
+ let mut modes = Vec::new();
+ for (column_count, field_name) in field_names.into_iter().enumerate() {
+ let mode = if let Some(field_mode) =
self.field_metadata_stats_mode(field_name)? {
+ field_mode
+ } else if keep_first_n >= 0 && column_count >= keep_first_n as
usize {
+ MetadataStatsMode::None
+ } else {
+ table_mode
+ };
+ modes.push(mode);
+ }
+ Ok(modes)
+ }
+
/// Avro compression codec for manifest, manifest-list and index-manifest
files.
/// Default is `"zstd"`, matching Java Paimon
`CoreOptions.MANIFEST_COMPRESSION`.
pub fn manifest_compression(&self) -> &str {
@@ -1351,6 +1482,49 @@ mod tests {
assert_eq!(custom_core.changelog_file_stats_mode(), Some("counts"));
}
+ #[test]
+ fn test_metadata_stats_modes_follow_java_priority() {
+ let options = HashMap::from([
+ (METADATA_STATS_MODE_OPTION.to_string(), "counts".to_string()),
+ (
+ METADATA_STATS_KEEP_FIRST_N_COLUMNS_OPTION.to_string(),
+ "2".to_string(),
+ ),
+ ("fields.name.stats-mode".to_string(), "full".to_string()),
+ (
+ "fields.payload.stats-mode".to_string(),
+ "truncate(8)".to_string(),
+ ),
+ ]);
+ let core = CoreOptions::new(&options);
+
+ assert_eq!(
+ core.metadata_stats_modes(["id", "name", "payload", "extra"])
+ .unwrap(),
+ vec![
+ MetadataStatsMode::Counts,
+ MetadataStatsMode::Full,
+ MetadataStatsMode::Truncate(8),
+ MetadataStatsMode::None,
+ ]
+ );
+ }
+
+ #[test]
+ fn test_metadata_stats_mode_rejects_invalid_values() {
+ let options = HashMap::from([(
+ METADATA_STATS_MODE_OPTION.to_string(),
+ "truncate(0)".to_string(),
+ )]);
+ let core = CoreOptions::new(&options);
+
+ let err = core
+ .metadata_stats_mode()
+ .expect_err("zero truncate length should fail");
+ assert!(matches!(err, crate::Error::DataInvalid { message, .. }
+ if message.contains(METADATA_STATS_MODE_OPTION)));
+ }
+
#[test]
fn test_vector_file_options_defaults_and_overrides() {
let default_options =
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 1c5e9a8..513d4db 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -3281,7 +3281,7 @@ mod tests {
.await
.unwrap();
writer.write(&batch).await.unwrap();
- writer.close().await.unwrap();
+ let _ = writer.close().await.unwrap();
}
let table_schema = TableSchema::new(
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 8410992..05df8b3 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -752,7 +752,7 @@ mod row_tests {
.await
.unwrap();
writer.write(&batch).await.unwrap();
- let file_size = writer.close().await.unwrap() as i64;
+ let file_size = writer.close().await.unwrap().file_size as i64;
let schema_id = 1;
let split = DataSplitBuilder::new()
@@ -823,7 +823,7 @@ mod row_tests {
.await
.unwrap();
writer.write(&batch).await.unwrap();
- let file_size = writer.close().await.unwrap() as i64;
+ let file_size = writer.close().await.unwrap().file_size as i64;
let schema_id = 1;
let split = DataSplitBuilder::new()
@@ -904,7 +904,7 @@ mod row_tests {
.await
.unwrap();
writer.write(&batch).await.unwrap();
- let file_size = writer.close().await.unwrap() as i64;
+ let file_size = writer.close().await.unwrap().file_size as i64;
let split = DataSplitBuilder::new()
.with_snapshot(1)
@@ -1545,12 +1545,19 @@ mod vector_parquet_tests {
let file_path = format!("{bucket_path}/{file_name}");
let output = file_io.new_output(&file_path).unwrap();
let mut writer: Box<dyn FormatFileWriter> = Box::new(
- ParquetFormatWriter::new(&output, arrow_schema.clone(), "zstd", 1)
- .await
- .unwrap(),
+ ParquetFormatWriter::new(
+ &output,
+ arrow_schema.clone(),
+ "zstd",
+ 1,
+ None,
+ &std::collections::HashMap::new(),
+ )
+ .await
+ .unwrap(),
);
writer.write(&batch).await.unwrap();
- let file_size = writer.close().await.unwrap();
+ let file_size = writer.close().await.unwrap().file_size;
// Build a split whose data file's schema_id matches the table
schema_id, so the
// read path uses `read_type` directly (no SchemaManager lookup
needed).
diff --git a/crates/paimon/src/table/data_file_writer.rs
b/crates/paimon/src/table/data_file_writer.rs
index 58824ed..75f17c6 100644
--- a/crates/paimon/src/table/data_file_writer.rs
+++ b/crates/paimon/src/table/data_file_writer.rs
@@ -22,7 +22,7 @@
//! handles file rolling when `target_file_size` is reached, and collects
//! [`DataFileMeta`] for the commit path.
-use crate::arrow::format::{create_format_writer, FormatFileWriter};
+use crate::arrow::format::{create_format_writer, FormatFileWriter,
FormatValueStats};
use crate::io::FileIO;
use crate::spec::stats::BinaryTableStats;
use crate::spec::{bucket_dir_name, DataField, DataFileMeta,
EMPTY_SERIALIZED_ROW};
@@ -117,8 +117,8 @@ impl DataFileWriter {
self.open_new_file(batch.schema()).await?;
}
- self.current_row_count += batch.num_rows() as i64;
self.current_writer.as_mut().unwrap().write(batch).await?;
+ self.current_row_count += batch.num_rows() as i64;
// Roll to a new file if target size is reached — close in background
if self.current_writer.as_ref().unwrap().num_bytes() as i64 >=
self.target_file_size {
@@ -142,7 +142,6 @@ impl DataFileWriter {
self.written_files.len(),
self.file_format,
);
-
let bucket_dir = if self.partition_path.is_empty() {
format!("{}/{}", self.table_location, bucket_dir_name(self.bucket))
} else {
@@ -183,16 +182,17 @@ impl DataFileWriter {
let row_count = self.current_row_count;
self.current_row_count = 0;
- let file_size = writer.close().await? as i64;
+ let write_result = writer.close().await?;
let meta = Self::build_meta(
file_name,
- file_size,
+ write_result.file_size as i64,
row_count,
self.schema_id,
self.file_source,
self.first_row_id,
self.write_cols.clone(),
+ write_result.value_stats,
);
self.written_files.push(meta);
Ok(())
@@ -213,15 +213,16 @@ impl DataFileWriter {
let write_cols = self.write_cols.clone();
self.in_flight_closes.spawn(async move {
- let file_size = writer.close().await? as i64;
+ let write_result = writer.close().await?;
Ok(Self::build_meta(
file_name,
- file_size,
+ write_result.file_size as i64,
row_count,
schema_id,
file_source,
first_row_id,
write_cols,
+ write_result.value_stats,
))
});
}
@@ -239,6 +240,7 @@ impl DataFileWriter {
Ok(std::mem::take(&mut self.written_files))
}
+ #[allow(clippy::too_many_arguments)]
fn build_meta(
file_name: String,
file_size: i64,
@@ -247,23 +249,20 @@ impl DataFileWriter {
file_source: Option<i32>,
first_row_id: Option<i64>,
write_cols: Option<Vec<String>>,
+ format_value_stats: Option<FormatValueStats>,
) -> DataFileMeta {
+ let (value_stats, value_stats_cols) = match format_value_stats {
+ Some(stats) => (stats.stats, stats.columns),
+ None => (BinaryTableStats::empty(), Some(Vec::new())),
+ };
DataFileMeta {
file_name,
file_size,
row_count,
min_key: EMPTY_SERIALIZED_ROW.clone(),
max_key: EMPTY_SERIALIZED_ROW.clone(),
- key_stats: BinaryTableStats::new(
- EMPTY_SERIALIZED_ROW.clone(),
- EMPTY_SERIALIZED_ROW.clone(),
- vec![],
- ),
- value_stats: BinaryTableStats::new(
- EMPTY_SERIALIZED_ROW.clone(),
- EMPTY_SERIALIZED_ROW.clone(),
- vec![],
- ),
+ key_stats: BinaryTableStats::empty(),
+ value_stats,
min_sequence_number: 0,
max_sequence_number: 0,
schema_id,
@@ -273,7 +272,7 @@ impl DataFileWriter {
delete_row_count: Some(0),
embedded_index: None,
file_source,
- value_stats_cols: Some(vec![]),
+ value_stats_cols,
external_path: None,
first_row_id,
write_cols,
diff --git a/crates/paimon/src/table/dedicated_format_file_writer.rs
b/crates/paimon/src/table/dedicated_format_file_writer.rs
index 15f8803..6417b9c 100644
--- a/crates/paimon/src/table/dedicated_format_file_writer.rs
+++ b/crates/paimon/src/table/dedicated_format_file_writer.rs
@@ -131,6 +131,10 @@ impl AppendDedicatedFormatFileWriter {
}
let normal_schema =
Arc::new(arrow_schema::Schema::new(normal_arrow_fields));
+ let normal_field_names = normal_table_fields
+ .iter()
+ .map(|field| field.name().to_string())
+ .collect();
let vector_writer = if let Some(vector_file_format) =
vector_file_format {
if vector_table_fields.is_empty() {
None
@@ -178,7 +182,7 @@ impl AppendDedicatedFormatFileWriter {
format_options.clone(),
Some(0),
None,
- None,
+ Some(normal_field_names),
);
Self {
diff --git a/crates/paimon/src/table/kv_file_writer.rs
b/crates/paimon/src/table/kv_file_writer.rs
index dd5ac7d..54ffa46 100644
--- a/crates/paimon/src/table/kv_file_writer.rs
+++ b/crates/paimon/src/table/kv_file_writer.rs
@@ -423,7 +423,7 @@ impl KeyValueFileWriter {
writer.write(&chunk_batch).await?;
}
- let file_size = writer.close().await? as i64;
+ let file_size = writer.close().await?.file_size as i64;
let key_columns: Vec<Arc<dyn Array>> = self
.config
diff --git a/crates/paimon/src/table/postpone_file_writer.rs
b/crates/paimon/src/table/postpone_file_writer.rs
index afe318e..8ee83cb 100644
--- a/crates/paimon/src/table/postpone_file_writer.rs
+++ b/crates/paimon/src/table/postpone_file_writer.rs
@@ -184,7 +184,7 @@ impl PostponeFileWriter {
let creation_time = self.current_file_creation_time;
self.in_flight_closes.spawn(async move {
- let file_size = writer.close().await? as i64;
+ let file_size = writer.close().await?.file_size as i64;
Ok(build_meta(
file_name,
file_size,
@@ -249,7 +249,7 @@ impl PostponeFileWriter {
let file_name = self.current_file_name.take().unwrap();
let row_count = self.current_row_count;
self.current_row_count = 0;
- let file_size = writer.close().await? as i64;
+ let file_size = writer.close().await?.file_size as i64;
let min_seq = self.current_file_start_seq;
let max_seq = self.next_sequence_number - 1;
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 375dfa3..86d8085 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -881,15 +881,16 @@ mod tests {
use crate::catalog::Identifier;
use crate::io::{FileIO, FileIOBuilder};
use crate::spec::{
- bucket_dir_name, BigIntType, BinaryRowBuilder, BlobType, DataField,
DataType, DecimalType,
- FileKind, FloatType, IndexManifest, IntType, LocalZonedTimestampType,
Manifest,
- ManifestList, Schema, TableSchema, TimestampType, TinyIntType,
VarCharType, VectorType,
+ bucket_dir_name, BigIntType, BinaryRow, BinaryRowBuilder, BinaryType,
BlobType, DataField,
+ DataType, Datum, DecimalType, FileKind, FloatType, IndexManifest,
IntType,
+ LocalZonedTimestampType, Manifest, ManifestList, PredicateBuilder,
Schema, TableSchema,
+ TimeType, TimestampType, TinyIntType, VarBinaryType, VarCharType,
VectorType,
SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME,
VALUE_KIND_FIELD_ID,
VALUE_KIND_FIELD_NAME,
};
use crate::table::{SnapshotManager, TableCommit};
use arrow_array::RecordBatchReader as _;
- use arrow_array::{Int32Array, Int64Array, Int8Array, StringArray};
+ use arrow_array::{Int32Array, Int64Array, Int8Array, StringArray,
Time32MillisecondArray};
use arrow_schema::{
DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema,
TimeUnit,
};
@@ -989,6 +990,21 @@ mod tests {
.unwrap()
}
+ fn make_nullable_batch(ids: Vec<Option<i32>>, values: Vec<Option<i32>>) ->
RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, true),
+ ArrowField::new("value", ArrowDataType::Int32, true),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(Int32Array::from(values)),
+ ],
+ )
+ .unwrap()
+ }
+
fn make_vector_batch(ids: Vec<i32>, vectors: Vec<Vec<f32>>) -> RecordBatch
{
use arrow_array::builder::{FixedSizeListBuilder, Float32Builder};
@@ -1203,6 +1219,445 @@ mod tests {
assert_eq!(snapshot.total_record_count(), Some(3));
}
+ #[tokio::test]
+ async fn test_append_write_populates_value_stats() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_table_write_value_stats";
+ setup_dirs(&file_io, table_path).await;
+
+ let table = test_table(&file_io, table_path);
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ table_write
+ .write_arrow_batch(&make_nullable_batch(
+ vec![Some(3), None],
+ vec![Some(30), Some(10)],
+ ))
+ .await
+ .unwrap();
+ table_write
+ .write_arrow_batch(&make_nullable_batch(vec![Some(1)], vec![None]))
+ .await
+ .unwrap();
+
+ let messages = table_write.prepare_commit().await.unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].new_files.len(), 1);
+ let file = &messages[0].new_files[0];
+ assert_eq!(file.row_count, 3);
+ assert_eq!(file.value_stats_cols, None);
+ assert_eq!(file.value_stats.null_counts(), &vec![Some(1), Some(1)]);
+
+ let min_values =
BinaryRow::from_serialized_bytes(file.value_stats.min_values()).unwrap();
+ let max_values =
BinaryRow::from_serialized_bytes(file.value_stats.max_values()).unwrap();
+ assert_eq!(min_values.get_int(0).unwrap(), 1);
+ assert_eq!(max_values.get_int(0).unwrap(), 3);
+ assert_eq!(min_values.get_int(1).unwrap(), 10);
+ assert_eq!(max_values.get_int(1).unwrap(), 30);
+ }
+
+ #[tokio::test]
+ async fn test_append_write_populates_microsecond_timestamp_value_stats() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_table_write_timestamp6_value_stats";
+ setup_dirs(&file_io, table_path).await;
+
+ let schema = Schema::builder()
+ .column("ts", DataType::Timestamp(TimestampType::new(6).unwrap()))
+ .column(
+ "lzts",
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(6).unwrap()),
+ )
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test_timestamp9_table"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new(
+ "ts",
+ ArrowDataType::Timestamp(TimeUnit::Microsecond, None),
+ true,
+ ),
+ ArrowField::new(
+ "lzts",
+ ArrowDataType::Timestamp(TimeUnit::Microsecond,
Some("UTC".into())),
+ true,
+ ),
+ ]));
+ let batch = RecordBatch::try_new(
+ arrow_schema,
+ vec![
+ Arc::new(arrow_array::TimestampMicrosecondArray::from(vec![
+ Some(1_704_067_200_987_654_i64),
+ Some(1_704_067_200_123_456_i64),
+ ])),
+ Arc::new(
+ arrow_array::TimestampMicrosecondArray::from(vec![
+ Some(1_704_067_201_999_999_i64),
+ Some(1_704_067_201_000_001_i64),
+ ])
+ .with_timezone("UTC"),
+ ),
+ ],
+ )
+ .unwrap();
+
+ table_write.write_arrow_batch(&batch).await.unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ let file = &messages[0].new_files[0];
+ assert_eq!(file.value_stats_cols, None);
+
+ let min_values =
BinaryRow::from_serialized_bytes(file.value_stats.min_values()).unwrap();
+ let max_values =
BinaryRow::from_serialized_bytes(file.value_stats.max_values()).unwrap();
+ assert_eq!(
+ min_values.get_timestamp_raw(0, 6).unwrap(),
+ (1_704_067_200_123, 456_000)
+ );
+ assert_eq!(
+ max_values.get_timestamp_raw(0, 6).unwrap(),
+ (1_704_067_200_987, 654_000)
+ );
+ assert_eq!(
+ min_values.get_timestamp_raw(1, 6).unwrap(),
+ (1_704_067_201_000, 1_000)
+ );
+ assert_eq!(
+ max_values.get_timestamp_raw(1, 6).unwrap(),
+ (1_704_067_201_999, 999_000)
+ );
+ }
+
+ #[tokio::test]
+ async fn
test_append_write_nanosecond_timestamp_value_stats_keeps_counts_only() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_table_write_timestamp9_value_stats";
+ setup_dirs(&file_io, table_path).await;
+
+ let schema = Schema::builder()
+ .column("ts", DataType::Timestamp(TimestampType::new(9).unwrap()))
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test_timestamp9_table"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![ArrowField::new(
+ "ts",
+ ArrowDataType::Timestamp(TimeUnit::Nanosecond, None),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(
+ arrow_schema,
+ vec![Arc::new(arrow_array::TimestampNanosecondArray::from(vec![
+ Some(1_704_067_200_987_654_321_i64),
+ None,
+ ]))],
+ )
+ .unwrap();
+
+ table_write.write_arrow_batch(&batch).await.unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ let file = &messages[0].new_files[0];
+ assert_eq!(file.value_stats_cols, None);
+ assert_eq!(file.value_stats.null_counts(), &vec![Some(1)]);
+
+ let min_values =
BinaryRow::from_serialized_bytes(file.value_stats.min_values()).unwrap();
+ let max_values =
BinaryRow::from_serialized_bytes(file.value_stats.max_values()).unwrap();
+ assert!(min_values.is_null_at(0));
+ assert!(max_values.is_null_at(0));
+ }
+
+ #[tokio::test]
+ async fn
test_append_write_truncates_string_value_stats_and_keeps_binary_counts() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_table_write_skip_variable_length_stats";
+ setup_dirs(&file_io, table_path).await;
+
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::string_type()))
+ .column(
+ "payload",
+ DataType::VarBinary(VarBinaryType::new(1024).unwrap()),
+ )
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test_variable_length_stats_table"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("name", ArrowDataType::Utf8, true),
+ ArrowField::new("payload", ArrowDataType::Binary, true),
+ ]));
+ let batch = RecordBatch::try_new(
+ arrow_schema,
+ vec![
+ Arc::new(Int32Array::from(vec![2, 1])),
+ Arc::new(StringArray::from(vec![
+ Some("a long string value"),
+ Some("another long string value"),
+ ])),
+ Arc::new(arrow_array::BinaryArray::from(vec![
+ Some(b"large-binary-value" as &[u8]),
+ Some(b"another-large-binary-value" as &[u8]),
+ ])),
+ ],
+ )
+ .unwrap();
+
+ table_write.write_arrow_batch(&batch).await.unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ let file = &messages[0].new_files[0];
+ assert_eq!(file.value_stats_cols, None);
+ assert_eq!(
+ file.value_stats.null_counts(),
+ &vec![Some(0), Some(0), Some(0)]
+ );
+
+ let min_values =
BinaryRow::from_serialized_bytes(file.value_stats.min_values()).unwrap();
+ let max_values =
BinaryRow::from_serialized_bytes(file.value_stats.max_values()).unwrap();
+ assert_eq!(min_values.get_int(0).unwrap(), 1);
+ assert_eq!(max_values.get_int(0).unwrap(), 2);
+ assert_eq!(min_values.get_string(1).unwrap(), "a long string va");
+ assert_eq!(max_values.get_string(1).unwrap(), "another long sts");
+ assert!(min_values.is_null_at(2));
+ assert!(max_values.is_null_at(2));
+ }
+
+ #[tokio::test]
+ async fn test_append_write_applies_configured_metadata_stats_modes() {
+ let file_io = test_file_io();
+ let table_path =
"memory:/test_table_write_configured_metadata_stats_modes";
+ setup_dirs(&file_io, table_path).await;
+
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::string_type()))
+ .column("payload",
DataType::Binary(BinaryType::new(1024).unwrap()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("metadata.stats-mode", "counts")
+ .option("metadata.stats-keep-first-n-columns", "2")
+ .option("fields.name.stats-mode", "full")
+ .option("fields.payload.stats-mode", "full")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test_configured_stats_modes_table"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, true),
+ ArrowField::new("name", ArrowDataType::Utf8, true),
+ ArrowField::new("payload", ArrowDataType::Binary, true),
+ ArrowField::new("value", ArrowDataType::Int32, true),
+ ]));
+ let batch = RecordBatch::try_new(
+ arrow_schema,
+ vec![
+ Arc::new(Int32Array::from(vec![Some(2), None])),
+ Arc::new(StringArray::from(vec![
+ Some("alpha-long-value-12345"),
+ Some("zeta-long-value-99999"),
+ ])),
+ Arc::new(arrow_array::BinaryArray::from(vec![
+ Some(b"first-binary-value" as &[u8]),
+ None,
+ ])),
+ Arc::new(Int32Array::from(vec![Some(10), Some(20)])),
+ ],
+ )
+ .unwrap();
+
+ table_write.write_arrow_batch(&batch).await.unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ let file = &messages[0].new_files[0];
+ assert_eq!(
+ file.value_stats_cols,
+ Some(vec![
+ "id".to_string(),
+ "name".to_string(),
+ "payload".to_string()
+ ])
+ );
+ assert_eq!(
+ file.value_stats.null_counts(),
+ &vec![Some(1), Some(0), Some(1)]
+ );
+
+ let min_values =
BinaryRow::from_serialized_bytes(file.value_stats.min_values()).unwrap();
+ let max_values =
BinaryRow::from_serialized_bytes(file.value_stats.max_values()).unwrap();
+ assert!(min_values.is_null_at(0));
+ assert!(max_values.is_null_at(0));
+ assert_eq!(min_values.get_string(1).unwrap(),
"alpha-long-value-12345");
+ assert_eq!(max_values.get_string(1).unwrap(), "zeta-long-value-99999");
+ assert!(min_values.is_null_at(2));
+ assert!(max_values.is_null_at(2));
+ }
+
+ #[tokio::test]
+ async fn test_append_write_respects_non_dense_metadata_stats_storage() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_table_write_non_dense_metadata_stats";
+ setup_dirs(&file_io, table_path).await;
+
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("ignored", DataType::Int(IntType::new()))
+ .option("metadata.stats-mode", "none")
+ .option("metadata.stats-dense-store", "false")
+ .option("fields.id.stats-mode", "counts")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test_non_dense_stats_table"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, true),
+ ArrowField::new("ignored", ArrowDataType::Int32, true),
+ ]));
+ let batch = RecordBatch::try_new(
+ arrow_schema,
+ vec![
+ Arc::new(Int32Array::from(vec![Some(1), None])),
+ Arc::new(Int32Array::from(vec![Some(10), Some(20)])),
+ ],
+ )
+ .unwrap();
+
+ table_write.write_arrow_batch(&batch).await.unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ let file = &messages[0].new_files[0];
+ assert_eq!(file.value_stats_cols, None);
+ assert_eq!(file.value_stats.null_counts(), &vec![Some(1), None]);
+
+ let min_values =
BinaryRow::from_serialized_bytes(file.value_stats.min_values()).unwrap();
+ let max_values =
BinaryRow::from_serialized_bytes(file.value_stats.max_values()).unwrap();
+ assert_eq!(min_values.arity(), 2);
+ assert_eq!(max_values.arity(), 2);
+ assert!(min_values.is_null_at(0));
+ assert!(max_values.is_null_at(0));
+ assert!(min_values.is_null_at(1));
+ assert!(max_values.is_null_at(1));
+ }
+
+ #[tokio::test]
+ async fn test_append_write_populates_time_value_stats() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_table_write_time_value_stats";
+ setup_dirs(&file_io, table_path).await;
+
+ let schema = Schema::builder()
+ .column("tm", DataType::Time(TimeType::new(3).unwrap()))
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test_time_value_stats_table"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![ArrowField::new(
+ "tm",
+ ArrowDataType::Time32(TimeUnit::Millisecond),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(
+ arrow_schema,
+ vec![Arc::new(Time32MillisecondArray::from(vec![
+ Some(3_000),
+ None,
+ Some(1_000),
+ ]))],
+ )
+ .unwrap();
+
+ table_write.write_arrow_batch(&batch).await.unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ let file = &messages[0].new_files[0];
+ assert_eq!(file.value_stats_cols, None);
+ assert_eq!(file.value_stats.null_counts(), &vec![Some(1)]);
+
+ let min_values =
BinaryRow::from_serialized_bytes(file.value_stats.min_values()).unwrap();
+ let max_values =
BinaryRow::from_serialized_bytes(file.value_stats.max_values()).unwrap();
+ assert_eq!(min_values.get_int(0).unwrap(), 1_000);
+ assert_eq!(max_values.get_int(0).unwrap(), 3_000);
+ }
+
+ #[tokio::test]
+ async fn test_scan_prunes_real_append_files_using_written_value_stats() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_table_write_value_stats_scan";
+ setup_dirs(&file_io, table_path).await;
+
+ let table = test_table(&file_io, table_path);
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ table_write
+ .write_arrow_batch(&make_batch(vec![1, 2], vec![10, 20]))
+ .await
+ .unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ table_write
+ .write_arrow_batch(&make_batch(vec![100, 101], vec![1000, 1010]))
+ .await
+ .unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .greater_than("id", Datum::Int(10))
+ .unwrap();
+ let mut reader = table.new_read_builder();
+ reader.with_filter(predicate);
+ let (_plan, trace) =
reader.new_scan().plan_with_trace().await.unwrap();
+
+ assert_eq!(trace.final_files, 1, "scan trace: {trace:?}");
+ assert!(
+ trace.manifest_entries_pruned_by_data_stats >= 1,
+ "scan trace: {trace:?}"
+ );
+ }
+
#[tokio::test]
async fn test_table_write_row_format_roundtrip() {
let file_io = test_file_io();
@@ -1328,6 +1783,7 @@ mod tests {
assert_eq!(parquet_files[0].row_count, 3);
assert_eq!(blob_files[0].row_count, 3);
+ assert_eq!(parquet_files[0].write_cols, Some(vec!["id".to_string()]));
assert_eq!(blob_files[0].write_cols,
Some(vec!["payload".to_string()]));
// Commit and verify snapshot
@@ -1339,6 +1795,71 @@ mod tests {
assert_eq!(snapshot.id(), 1);
}
+ #[tokio::test]
+ async fn test_dedicated_blob_stats_use_normal_write_columns() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_dedicated_blob_stats_mapping";
+ setup_dirs(&file_io, table_path).await;
+
+ let schema = Schema::builder()
+ .column("payload", DataType::Blob(BlobType::new()))
+ .column("a", DataType::Int(IntType::new()))
+ .column("b", DataType::Int(IntType::new()))
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test_dedicated_blob_stats_mapping"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("payload", ArrowDataType::Binary, true),
+ ArrowField::new("a", ArrowDataType::Int32, false),
+ ArrowField::new("b", ArrowDataType::Int32, false),
+ ])),
+ vec![
+ Arc::new(arrow_array::BinaryArray::from(vec![Some(
+ b"payload" as &[u8],
+ )])),
+ Arc::new(Int32Array::from(vec![100])),
+ Arc::new(Int32Array::from(vec![0])),
+ ],
+ )
+ .unwrap();
+
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ table_write.write_arrow_batch(&batch).await.unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ let normal_write_cols = messages[0]
+ .new_files
+ .iter()
+ .find(|file| file.file_name.ends_with(".parquet"))
+ .unwrap()
+ .write_cols
+ .clone();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .greater_than("a", Datum::Int(50))
+ .unwrap();
+ let mut reader = table.new_read_builder();
+ reader.with_filter(predicate);
+ let (_plan, trace) =
reader.new_scan().plan_with_trace().await.unwrap();
+
+ assert!(trace.final_files > 0, "scan trace: {trace:?}");
+ assert_eq!(
+ normal_write_cols,
+ Some(vec!["a".to_string(), "b".to_string()])
+ );
+ }
+
async fn assert_vector_write_uses_dedicated_file(table_path: &str,
vector_file_format: &str) {
let file_io = test_file_io();
setup_dirs(&file_io, table_path).await;
@@ -1379,6 +1900,7 @@ mod tests {
assert_eq!(vector_files.len(), 1);
assert_eq!(normal_files[0].row_count, 3);
assert_eq!(vector_files[0].row_count, 3);
+ assert_eq!(normal_files[0].write_cols, Some(vec!["id".to_string()]));
assert_eq!(
vector_files[0].write_cols,
Some(vec!["embedding".to_string()])
