diff --git a/crates/integrations/datafusion/src/system_tables/partitions.rs b/crates/integrations/datafusion/src/system_tables/partitions.rs index b87db4f81..b2313ffc0 100644 --- a/crates/integrations/datafusion/src/system_tables/partitions.rs +++ b/crates/integrations/datafusion/src/system_tables/partitions.rs @@ -22,7 +22,8 @@ use std::sync::{Arc, OnceLock}; use async_trait::async_trait; use datafusion::arrow::array::{ - BooleanArray, Int32Array, Int64Array, RecordBatch, StringArray, TimestampMillisecondArray, + ArrayRef, BooleanArray, Int32Array, Int64Array, RecordBatch, StringArray, + TimestampMillisecondArray, }; use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit}; use datafusion::catalog::Session; @@ -32,7 +33,7 @@ use datafusion::error::{DataFusionError, Result as DFResult}; use datafusion::logical_expr::Expr; use datafusion::physical_plan::ExecutionPlan; use paimon::catalog::{list_partitions_from_file_system, Catalog, Identifier}; -use paimon::spec::{CoreOptions, Partition}; +use paimon::spec::{CoreOptions, LocalZonedTimestampType, Partition, TimestampType}; use paimon::table::Table; use crate::error::to_datafusion_error; @@ -216,8 +217,8 @@ impl TableProvider for PartitionsTable { Arc::new(Int64Array::from(record_counts)), Arc::new(Int64Array::from(file_sizes)), Arc::new(Int64Array::from(file_counts)), - Arc::new(TimestampMillisecondArray::from(last_update_times)), - Arc::new(TimestampMillisecondArray::from(created_ats)), + local_timestamp_column(last_update_times)?, + local_timestamp_column(created_ats)?, Arc::new(StringArray::from(created_bys)), Arc::new(StringArray::from(updated_bys)), Arc::new(StringArray::from(options_jsons)), @@ -234,6 +235,19 @@ impl TableProvider for PartitionsTable { } } +fn local_timestamp_column(values: Vec>) -> DFResult { + // Partition metadata stores epoch instants. Java PartitionsTable.toTimestamp + // converts these to the system-local wall clock for its TIMESTAMP columns. + let source = paimon::spec::DataType::LocalZonedTimestamp( + LocalZonedTimestampType::new(3).map_err(to_datafusion_error)?, + ); + let target = + paimon::spec::DataType::Timestamp(TimestampType::new(3).map_err(to_datafusion_error)?); + let values: ArrayRef = Arc::new(TimestampMillisecondArray::from(values)); + paimon::arrow::cast_array_for_schema_evolution(&values, &source, &target) + .map_err(to_datafusion_error) +} + /// Format `spec` as `key1=val1/key2=val2` in `partition_keys` order. Empty /// string for non-partitioned tables. NULL spec values fall back to /// `default_partition_name`. diff --git a/crates/integrations/datafusion/tests/sql_context_tests.rs b/crates/integrations/datafusion/tests/sql_context_tests.rs index b4ddf07ce..325048a0a 100644 --- a/crates/integrations/datafusion/tests/sql_context_tests.rs +++ b/crates/integrations/datafusion/tests/sql_context_tests.rs @@ -629,7 +629,8 @@ async fn test_partitions_system_table_shows_unreported_statistics_as_null() { sql_context .sql( "CREATE TABLE paimon.default.unknown_stats_orders \ - (id INT, name STRING) PARTITIONED BY (id)", + (id INT, name STRING) PARTITIONED BY (id) \ + WITH ('bucket' = '4', 'bucket-key' = 'name')", ) .await .unwrap() @@ -658,6 +659,15 @@ async fn test_partitions_system_table_shows_unreported_statistics_as_null() { "column {column} should carry a measurement before the switch" ); } + assert_eq!( + measured + .column(3) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 4 + ); catalog.set_unknown_statistics(true); diff --git a/crates/integrations/datafusion/tests/system_tables.rs b/crates/integrations/datafusion/tests/system_tables.rs index 48f299bcf..717f9baa5 100644 --- a/crates/integrations/datafusion/tests/system_tables.rs +++ b/crates/integrations/datafusion/tests/system_tables.rs @@ -1409,6 +1409,63 @@ fn single_int_partition_stat(value: &str) -> i32 { .expect("partition stats should contain one int partition value") } +#[tokio::test] +async fn test_partition_timestamps_match_local_file_creation_times() { + let (ctx, catalog, _tmp) = create_context().await; + let table = catalog + .get_table(&Identifier::new("default", FIXTURE_TABLE)) + .await + .unwrap(); + let core = table + .new_system_read("partitions") + .unwrap() + .read() + .await + .unwrap(); + let sql = format!("SELECT last_update_time FROM paimon.default.{FIXTURE_TABLE}$partitions"); + let batches = run_sql(&ctx, &sql).await; + let actual = batches + .iter() + .flat_map(|batch| { + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .iter() + }) + .collect::>(); + let expected = core + .column_by_name("last_update_time") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .collect::>(); + assert!(!expected.is_empty()); + assert_eq!(actual, expected); + let files = run_sql( + &ctx, + &format!("SELECT creation_time FROM paimon.default.{FIXTURE_TABLE}$files"), + ) + .await; + let last_file = files + .iter() + .flat_map(|batch| { + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .flatten() + }) + .max() + .unwrap(); + assert_eq!(actual, vec![Some(last_file)]); +} + #[tokio::test] async fn test_partitions_system_table() { let (ctx, _catalog, _tmp) = create_context().await; @@ -1790,6 +1847,7 @@ async fn test_resolved_system_table_reader_preserves_guards_and_schema() { "consumers", "file_key_ranges", "table_indexes", + "partitions", ] { let stream = read_metadata(table.clone(), name).await.unwrap(); let schema = table.new_system_read(name).unwrap().schema(); @@ -1800,13 +1858,11 @@ async fn test_resolved_system_table_reader_preserves_guards_and_schema() { "{name}" ); } - for name in ["partitions", "unknown"] { - let error = match read_metadata(table.clone(), name).await { - Ok(_) => panic!("{name} must require a catalog or be unknown"), - Err(error) => error.to_string(), - }; - assert!(error.contains("requires a catalog or is unknown")); - } + let error = match read_metadata(table.clone(), "unknown").await { + Ok(_) => panic!("Unknown system table must be rejected"), + Err(error) => error.to_string(), + }; + assert!(error.contains("requires a catalog or is unknown")); let denied = table.copy_with_options(std::collections::HashMap::from([( "query-auth.enabled".to_string(), "true".to_string(), diff --git a/crates/paimon/src/arrow/cast_text.rs b/crates/paimon/src/arrow/cast_text.rs index 424ec7cb8..eeb45e48e 100644 --- a/crates/paimon/src/arrow/cast_text.rs +++ b/crates/paimon/src/arrow/cast_text.rs @@ -139,7 +139,7 @@ fn time_components(text: &str, space: bool) -> Option<[Option; 4]> { Some(time) } -pub(super) fn parse_timestamp( +pub(crate) fn parse_timestamp( text: &str, precision: u32, zone: Option<&jiff::tz::TimeZone>, diff --git a/crates/paimon/src/arrow/evolution_string.rs b/crates/paimon/src/arrow/evolution_string.rs index 533cd90cd..d000877b7 100644 --- a/crates/paimon/src/arrow/evolution_string.rs +++ b/crates/paimon/src/arrow/evolution_string.rs @@ -23,7 +23,7 @@ use arrow_array::{cast::AsArray, types::Int64Type, ArrayRef, StringArray, Time32 use arrow_schema::{DataType as ArrowType, TimeUnit}; use jiff::tz::TimeZone; -use crate::common::date_time::{format_epoch_timestamp, format_time, local_offset}; +use crate::common::date_time::{format_time, format_timestamp_millis}; use crate::spec::DataType; /// Return None when this is not a temporal-to-character conversion. @@ -112,17 +112,7 @@ fn render_timestamp( TimeUnit::Nanosecond => value.rem_euclid(1_000_000), _ => 0, }; - let millis = match zone { - Some(zone) => { - let offset = local_offset(millis.div_euclid(1000), zone); - // Timestamp.fromLocalDateTime uses signed long millisecond storage. - // Applying an offset can wrap and change the millisecond fraction. - millis.wrapping_add(i64::from(offset) * 1000) - } - None => millis, - }; - let nanos = (millis.rem_euclid(1000) * 1_000_000 + sub_milli) as u32; - format_epoch_timestamp(millis.div_euclid(1000), nanos, precision) + format_timestamp_millis(millis, sub_milli, precision, zone) } fn invalid(message: impl Into) -> crate::Error { diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs index a71198099..7a057c0af 100644 --- a/crates/paimon/src/arrow/mod.rs +++ b/crates/paimon/src/arrow/mod.rs @@ -24,6 +24,7 @@ pub(crate) use cast_executor::CastExecutor; #[cfg(test)] mod cast_completion_java_test; mod cast_text; +pub(crate) use cast_text::parse_timestamp as parse_java_timestamp; mod evolution_cast; pub(crate) use evolution_cast::{cast_numeric, supports_numeric_cast}; mod evolution_decimal; diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs index 20a13da00..5c48f454b 100644 --- a/crates/paimon/src/catalog/mod.rs +++ b/crates/paimon/src/catalog/mod.rs @@ -36,7 +36,10 @@ pub use database::*; pub use factory::*; pub use filesystem::*; pub use function::*; +#[cfg(test)] +pub(crate) use partition_listing::aggregate_partition_entries; pub use partition_listing::list_partitions_from_file_system; +pub(crate) use partition_listing::{list_partition_entries, FileSystemPartitionEntry}; pub use rest::*; use serde::{Deserialize, Serialize}; pub use view::*; diff --git a/crates/paimon/src/catalog/partition_listing.rs b/crates/paimon/src/catalog/partition_listing.rs index dec7702be..5557cf3bc 100644 --- a/crates/paimon/src/catalog/partition_listing.rs +++ b/crates/paimon/src/catalog/partition_listing.rs @@ -20,43 +20,18 @@ //! Used as the catalog-side fallback when a backend doesn't track partitions //! (e.g. `FileSystemCatalog`, or `RESTCatalog` against a non-metastore server). -use std::collections::{BTreeMap, HashMap}; +use std::collections::BTreeMap; use crate::spec::{ - merge_active_entries, BinaryRow, CoreOptions, Manifest, ManifestList, Partition, + merge_active_entries, BinaryRow, CoreOptions, Manifest, ManifestEntry, ManifestList, Partition, PartitionComputer, }; -use crate::table::{SnapshotManager, Table}; +use crate::table::Table; use crate::Result; /// Scan a table's manifest entries and aggregate them into [`Partition`] rows, /// matching the shape catalogs would otherwise return from a metastore. pub async fn list_partitions_from_file_system(table: &Table) -> Result> { - // Manifests carry partition values and per-column stats. - crate::spec::CoreOptions::new(table.schema().options()).ensure_read_authorized()?; - let file_io = table.file_io(); - let manifest_sm = SnapshotManager::new(file_io.clone(), table.location().to_string()); - let Some(snapshot) = table.resolve_read_snapshot().await? else { - return Ok(Vec::new()); - }; - - let base_path = manifest_sm.manifest_path(snapshot.base_manifest_list()); - let delta_path = manifest_sm.manifest_path(snapshot.delta_manifest_list()); - let (base_metas, delta_metas) = futures::try_join!( - ManifestList::read(file_io, &base_path), - ManifestList::read(file_io, &delta_path), - )?; - - let mut all_entries = Vec::new(); - for meta in base_metas.into_iter().chain(delta_metas) { - let manifest_path = manifest_sm.manifest_path(meta.file_name()); - let entries = Manifest::read(file_io, &manifest_path).await?; - all_entries.extend(entries); - } - // Mirror Java AbstractFileStoreScan.readAndMergeFileEntries: drop entries - // shadowed by a later DELETE so file_count/record_count reflect live files. - let all_entries = merge_active_entries(all_entries); - let schema = table.schema(); let core = CoreOptions::new(schema.options()); let computer = PartitionComputer::new( @@ -65,56 +40,175 @@ pub async fn list_partitions_from_file_system(table: &Table) -> Result, + pub total_buckets: i32, +} - #[derive(Default)] - struct Agg { - record_count: i64, - file_size: i64, - file_count: i64, - last_file_creation_time: i64, +impl Default for FileSystemPartitionEntry { + fn default() -> Self { + Self { + partition: BinaryRow::from_datums(&[]), + record_count: 0, + file_size_in_bytes: 0, + file_count: 0, + last_file_creation_time: None, + total_buckets: 0, + } } - let mut buckets: BTreeMap, Agg> = BTreeMap::new(); - for entry in &all_entries { +} + +pub(crate) async fn list_partition_entries(table: &Table) -> Result> { + CoreOptions::new(table.schema().options()).ensure_read_authorized()?; + let Some(snapshot) = table.resolve_read_snapshot().await? else { + return Ok(Vec::new()); + }; + let file_io = table.file_io(); + let manager = table.snapshot_manager(); + let base_path = manager.manifest_path(snapshot.base_manifest_list()); + let delta_path = manager.manifest_path(snapshot.delta_manifest_list()); + let (base, delta) = futures::try_join!( + ManifestList::read(file_io, &base_path), + ManifestList::read(file_io, &delta_path), + )?; + let mut entries = Vec::new(); + for meta in base.into_iter().chain(delta) { + entries.extend(Manifest::read(file_io, &manager.manifest_path(meta.file_name())).await?); + } + aggregate_partition_entries(entries) +} + +pub(crate) fn aggregate_partition_entries( + entries: Vec, +) -> Result> { + aggregate_partition_entries_in_zone(entries, &jiff::tz::TimeZone::system()) +} + +fn aggregate_partition_entries_in_zone( + entries: Vec, + zone: &jiff::tz::TimeZone, +) -> Result> { + let mut grouped: BTreeMap, FileSystemPartitionEntry> = BTreeMap::new(); + for entry in merge_active_entries(entries) { let file = entry.file(); - let agg = buckets.entry(entry.partition().to_vec()).or_default(); - agg.record_count += file.row_count; - agg.file_size += file.file_size; + let agg = grouped.entry(entry.partition().to_vec()).or_default(); + agg.record_count = agg.record_count.wrapping_add(file.row_count); + agg.file_size_in_bytes = agg.file_size_in_bytes.wrapping_add(file.file_size); agg.file_count += 1; - if let Some(ct) = file.creation_time { - agg.last_file_creation_time = agg.last_file_creation_time.max(ct.timestamp_millis()); + agg.total_buckets = entry.total_buckets(); + if let Some(time) = file.creation_time { + let millis = time.timestamp_millis(); + let time = + crate::common::date_time::local_to_utc_seconds(millis.div_euclid(1000), zone)? + * 1000 + + millis.rem_euclid(1000); + agg.last_file_creation_time = Some( + agg.last_file_creation_time + .map_or(time, |old| old.max(time)), + ); } } - - let mut result = Vec::with_capacity(buckets.len()); - for (bytes, agg) in buckets { - let spec = if bytes.is_empty() { - HashMap::new() - } else { - let row = BinaryRow::from_serialized_bytes(&bytes)?; - computer.generate_part_values(&row)?.into_iter().collect() - }; - result.push(Partition { - spec, - record_count: agg.record_count, - file_size_in_bytes: agg.file_size, - file_count: agg.file_count, - last_file_creation_time: agg.last_file_creation_time, - total_buckets: 0, - done: false, - created_at: None, - created_by: None, - updated_at: None, - updated_by: None, - options: None, - }); - } - Ok(result) + grouped + .into_iter() + .map(|(bytes, mut entry)| { + entry.partition = if bytes.is_empty() { + BinaryRow::from_datums(&[]) + } else { + BinaryRow::from_serialized_bytes(&bytes)? + }; + Ok(entry) + }) + .collect() } #[cfg(test)] mod tests { use super::*; + #[test] + fn creation_times_are_compared_as_instants_across_dst_gaps() { + use crate::spec::{stats::BinaryTableStats, DataFileMeta, FileKind}; + let entries = ["2024-03-10T02:55:00Z", "2024-03-10T03:10:00Z"] + .into_iter() + .map(|time| { + let file = DataFileMeta { + file_name: time.into(), + file_size: 10, + row_count: 1, + min_key: vec![], + max_key: vec![], + key_stats: BinaryTableStats::empty(), + value_stats: BinaryTableStats::empty(), + min_sequence_number: 0, + max_sequence_number: 0, + schema_id: 0, + level: 0, + extra_files: vec![], + creation_time: Some(time.parse().unwrap()), + delete_row_count: None, + embedded_index: None, + file_source: None, + value_stats_cols: None, + external_path: None, + first_row_id: None, + write_cols: None, + column_max_sequence_numbers: None, + }; + ManifestEntry::new( + FileKind::Add, + BinaryRow::from_datums(&[]).to_serialized_bytes(), + 0, + 8, + file, + 2, + ) + }) + .collect(); + let rows = aggregate_partition_entries_in_zone( + entries, + &jiff::tz::TimeZone::get("America/New_York").unwrap(), + ) + .unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].record_count, 2); + assert_eq!(rows[0].total_buckets, 8); + // 02:55 is shifted through the gap to 03:55 EDT (07:55 UTC), later than 03:10 EDT. + assert_eq!(rows[0].last_file_creation_time, Some(1_710_057_300_000)); + } + #[tokio::test] async fn test_query_auth_table_refuses_partition_listing() { let table = crate::table::query_auth_table(); diff --git a/crates/paimon/src/common/date_time.rs b/crates/paimon/src/common/date_time.rs index ecb9fd530..e80b1dfc5 100644 --- a/crates/paimon/src/common/date_time.rs +++ b/crates/paimon/src/common/date_time.rs @@ -82,6 +82,23 @@ pub(crate) fn format_epoch_timestamp(seconds: i64, nanos: u32, precision: u32) - ) } +/// SQL timestamp CAST from Paimon's full millisecond and sub-millisecond storage. +pub(crate) fn format_timestamp_millis( + millis: i64, + sub_milli: i64, + precision: u32, + zone: Option<&jiff::tz::TimeZone>, +) -> String { + let millis = match zone { + Some(zone) => { + millis.wrapping_add(i64::from(local_offset(millis.div_euclid(1000), zone)) * 1000) + } + None => millis, + }; + let nanos = (millis.rem_euclid(1000) * 1_000_000 + sub_milli) as u32; + format_epoch_timestamp(millis.div_euclid(1000), nanos, precision) +} + /// Java Timestamp.toString uses LocalDateTime's ISO form, not a SQL CAST form. pub fn format_iso_timestamp(seconds: i64, nanos: u32) -> String { let (year, month, day) = epoch_day_to_date(i64::from(seconds.div_euclid(86_400) as i32)); diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs index 02c50c9c9..ba45b0511 100644 --- a/crates/paimon/src/spec/mod.rs +++ b/crates/paimon/src/spec/mod.rs @@ -119,6 +119,7 @@ pub use types::*; mod partition; pub use partition::Partition; mod partition_utils; +pub(crate) use partition_utils::decode_java_utf8; pub(crate) use partition_utils::{ bucket_path, bucket_path_under, data_file_path, escape_path_name, relative_bucket_path, unescape_path_name, PartitionComputer, diff --git a/crates/paimon/src/spec/partition_utils.rs b/crates/paimon/src/spec/partition_utils.rs index f534c1f40..85c1bf3a4 100644 --- a/crates/paimon/src/spec/partition_utils.rs +++ b/crates/paimon/src/spec/partition_utils.rs @@ -495,7 +495,7 @@ pub fn format_java_float(value: f32) -> String { /// Decode as Java's UTF-8 decoder does for `BinaryString.toString()`. /// Rust's lossy decoder replaces each byte of a UTF-8 encoded surrogate, /// while Java replaces the complete malformed surrogate sequence once. -fn decode_java_utf8(mut bytes: &[u8]) -> String { +pub(crate) fn decode_java_utf8(mut bytes: &[u8]) -> String { let mut decoded = String::with_capacity(bytes.len()); loop { match std::str::from_utf8(bytes) { diff --git a/crates/paimon/src/table/partition_stat.rs b/crates/paimon/src/table/partition_stat.rs index c7663a8a9..c59c5737a 100644 --- a/crates/paimon/src/table/partition_stat.rs +++ b/crates/paimon/src/table/partition_stat.rs @@ -15,21 +15,20 @@ // specific language governing permissions and limitations // under the License. -//! Per-partition statistics computed by scanning the latest snapshot's manifest entries. +//! Per-partition statistics computed by scanning the selected snapshot's manifest entries. //! //! Mirrors what Java Paimon exposes via the `$partitions` system table for runtime introspection. -use std::collections::{BTreeMap, HashMap}; +use std::collections::HashMap; -use crate::io::FileIO; -use crate::spec::{ - avro::from_avro_bytes_fast, merge_active_entries, BinaryRow, CoreOptions, ManifestEntry, - ManifestFileMeta, PartitionComputer, Snapshot, -}; +#[cfg(test)] +use crate::catalog::aggregate_partition_entries; +use crate::catalog::{list_partition_entries, FileSystemPartitionEntry}; +#[cfg(test)] +use crate::spec::ManifestEntry; +use crate::spec::{CoreOptions, PartitionComputer}; use crate::table::Table; -const MANIFEST_DIR: &str = "manifest"; - /// Per-partition aggregated statistics. #[derive(Debug, Clone, PartialEq, Eq)] pub struct PartitionStat { @@ -43,36 +42,19 @@ pub struct PartitionStat { pub total_size_bytes: u64, } -#[derive(Default)] -struct Accum { - record_count: i64, - file_count: i64, - total_size_bytes: i64, -} - impl Table { - /// Compute per-partition statistics from the latest snapshot. + /// Compute per-partition statistics from the selected snapshot. /// /// Mirrors [`catalog::list_partitions_from_file_system`]: calls /// `merge_active_entries` first to deduplicate entries that appear in both /// the base and delta manifests, then aggregates the remaining live ADD /// files in a `BTreeMap` for a deterministic result order. /// - /// **Warning:** This method reads all manifest lists and entries from the latest snapshot. + /// **Warning:** This method reads all manifest lists and entries from the selected snapshot. /// For tables with a large number of manifests, this operation can be expensive. /// /// Returns an empty Vec when the table has no snapshots yet. pub async fn partition_stats(&self) -> crate::Result> { - // Manifests carry partition values and per-column stats. - CoreOptions::new(self.schema().options()).ensure_read_authorized()?; - let sm = self.snapshot_manager(); - let snapshot = match sm.get_latest_snapshot().await? { - Some(s) => s, - None => return Ok(Vec::new()), - }; - - let entries = read_all_manifest_entries(self.file_io(), self.location(), &snapshot).await?; - let schema = self.schema(); let core = CoreOptions::new(schema.options()); let computer = PartitionComputer::new( @@ -82,10 +64,10 @@ impl Table { core.legacy_partition_name(), )?; - aggregate_partition_stats(entries, &computer) + partition_stats_from_entries(list_partition_entries(self).await?, &computer) } - /// List all partition values present in the latest snapshot. + /// List all partition values present in the selected snapshot. /// /// **Warning:** This method computes partition statistics which reads all manifest lists /// and entries. For large tables, this operation can be expensive. @@ -101,87 +83,32 @@ impl Table { } } -async fn read_manifest_list( - file_io: &FileIO, - table_path: &str, - list_name: &str, -) -> crate::Result> { - if list_name.is_empty() { - return Ok(Vec::new()); - } - let path = format!( - "{}/{}/{}", - table_path.trim_end_matches('/'), - MANIFEST_DIR, - list_name - ); - let input = file_io.new_input(&path)?; - let bytes = input.read().await?; - from_avro_bytes_fast::(&bytes) -} - -async fn read_all_manifest_entries( - file_io: &FileIO, - table_path: &str, - snapshot: &Snapshot, -) -> crate::Result> { - let mut metas = read_manifest_list(file_io, table_path, snapshot.base_manifest_list()).await?; - metas.extend(read_manifest_list(file_io, table_path, snapshot.delta_manifest_list()).await?); - - let manifest_dir = format!("{}/{}", table_path.trim_end_matches('/'), MANIFEST_DIR); - let mut all_entries = Vec::new(); - for meta in metas { - let path = format!("{}/{}", manifest_dir, meta.file_name()); - let input = file_io.new_input(&path)?; - let bytes = input.read().await?; - let entries = from_avro_bytes_fast::(&bytes)?; - all_entries.extend(entries); - } - Ok(all_entries) +fn partition_stats_from_entries( + entries: Vec, + computer: &PartitionComputer, +) -> crate::Result> { + entries + .into_iter() + .map(|entry| { + Ok(PartitionStat { + partition: computer + .generate_part_values(&entry.partition)? + .into_iter() + .collect(), + record_count: entry.record_count, + file_count: entry.file_count as u64, + total_size_bytes: entry.file_size_in_bytes as u64, + }) + }) + .collect() } -/// Aggregate live manifest entries into per-partition statistics. -/// -/// Mirrors `catalog::list_partitions_from_file_system`: -/// 1. Call `merge_active_entries` to collapse ADD/DELETE pairs and remove -/// entries shadowed by a later DELETE, including duplicate ADD entries -/// that may appear in both base and delta manifests after compaction. -/// 2. Accumulate the remaining live ADD entries in a `BTreeMap` keyed by -/// raw partition bytes for a deterministic, sorted result order. +#[cfg(test)] fn aggregate_partition_stats( entries: Vec, computer: &PartitionComputer, ) -> crate::Result> { - // Step 1: deduplicate — collapse ADD/DELETE pairs and drop files that have - // been deleted. After this point every remaining entry is a live ADD. - let live_entries = merge_active_entries(entries); - - // Step 2: accumulate into a BTreeMap for deterministic partition ordering. - let mut grouped: BTreeMap, Accum> = BTreeMap::new(); - for entry in &live_entries { - let file = entry.file(); - let accum = grouped.entry(entry.partition().to_vec()).or_default(); - accum.record_count += file.row_count; - accum.file_count += 1; - accum.total_size_bytes += file.file_size; - } - - let mut out = Vec::with_capacity(grouped.len()); - for (partition_bytes, accum) in grouped { - let partition = if partition_bytes.is_empty() { - HashMap::new() - } else { - let row = BinaryRow::from_serialized_bytes(&partition_bytes)?; - computer.generate_part_values(&row)?.into_iter().collect() - }; - out.push(PartitionStat { - partition, - record_count: accum.record_count, - file_count: accum.file_count as u64, - total_size_bytes: accum.total_size_bytes as u64, - }); - } - Ok(out) + partition_stats_from_entries(aggregate_partition_entries(entries)?, computer) } #[cfg(test)] diff --git a/crates/paimon/src/table/system_table/mod.rs b/crates/paimon/src/table/system_table/mod.rs index b7893dbe6..d63b33d4d 100644 --- a/crates/paimon/src/table/system_table/mod.rs +++ b/crates/paimon/src/table/system_table/mod.rs @@ -30,6 +30,7 @@ mod consumers; mod files; mod manifests; mod options; +mod partitions; mod row_string_cast; mod schemas; mod snapshots; @@ -44,6 +45,7 @@ const NAMES: &[&str] = &[ "file_key_ranges", "manifests", "options", + "partitions", "schemas", "snapshots", "table_indexes", @@ -202,6 +204,7 @@ impl SystemRead { "options" => { options::read(&self.table, needed.as_deref(), self.predicate.as_ref()).await? } + "partitions" => partitions::read(&self.table, needed.as_deref()).await?, "schemas" => { schemas::read(&self.table, needed.as_deref(), self.predicate.as_ref()).await? } @@ -282,6 +285,7 @@ fn schema_for_name(name: &str) -> SchemaRef { "file_key_ranges" => files::key_ranges_schema(), "manifests" => manifests::manifests_schema(), "options" => options::options_schema(), + "partitions" => partitions::partitions_schema(), "schemas" => schemas::schemas_schema(), "snapshots" => snapshots::snapshots_schema(), "table_indexes" => table_indexes::table_indexes_schema(), @@ -294,10 +298,11 @@ fn record_batch(schema: SchemaRef, columns: Vec) -> Result) -> Result { +fn local_timestamp_column( + values: impl Into, +) -> Result { use crate::spec::{DataType, LocalZonedTimestampType, TimestampType}; - use arrow_array::TimestampMillisecondArray; - let values: ArrayRef = Arc::new(TimestampMillisecondArray::from(values)); + let values: ArrayRef = Arc::new(values.into()); crate::arrow::cast_array_for_schema_evolution( &values, &DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(3)?), diff --git a/crates/paimon/src/table/system_table/partitions.rs b/crates/paimon/src/table/system_table/partitions.rs new file mode 100644 index 000000000..dd793cbf4 --- /dev/null +++ b/crates/paimon/src/table/system_table/partitions.rs @@ -0,0 +1,626 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Filesystem partition metadata, matching Java PartitionsTable's manifest path. + +use std::collections::HashMap; +use std::sync::{Arc, OnceLock}; + +use arrow_array::{ + ArrayRef, BooleanArray, Int32Array, Int64Array, RecordBatch, StringArray, + TimestampMillisecondArray, +}; +use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit}; +use jiff::tz::TimeZone; + +use crate::catalog::{list_partition_entries, FileSystemPartitionEntry}; +use crate::spec::{ + BinaryRow, CoreOptions, DataField, Datum, FileKind, IndexManifest, IndexManifestEntry, +}; +use crate::table::Table; +use crate::Result; + +use super::row_string_cast::{format_field_in_zone, format_row_field_as_java_object_string}; + +const DELETED_RECORD_COUNT: usize = 11; + +pub(super) fn partitions_schema() -> SchemaRef { + static SCHEMA: OnceLock = OnceLock::new(); + SCHEMA + .get_or_init(|| { + Arc::new(Schema::new(vec![ + Field::new("partition", DataType::Utf8, true), + Field::new("record_count", DataType::Int64, false), + Field::new("file_size_in_bytes", DataType::Int64, false), + Field::new("file_count", DataType::Int64, false), + Field::new( + "last_update_time", + DataType::Timestamp(TimeUnit::Millisecond, None), + true, + ), + Field::new( + "created_at", + DataType::Timestamp(TimeUnit::Millisecond, None), + true, + ), + Field::new("created_by", DataType::Utf8, true), + Field::new("updated_by", DataType::Utf8, true), + Field::new("options", DataType::Utf8, true), + Field::new("total_buckets", DataType::Int32, false), + Field::new("done", DataType::Boolean, false), + Field::new("deleted_record_count", DataType::Int64, true), + ])) + }) + .clone() +} + +pub(super) async fn read(table: &Table, projection: Option<&[usize]>) -> Result { + // Resolve once: files and deletion-vector counts must describe the same commit. + let snapshot = table.resolve_read_snapshot().await?; + let pinned = table.copy_with_pinned_snapshot(snapshot.as_ref()); + let entries = list_partition_entries(&pinned).await?; + let needs_deletions = projection.is_none_or(|columns| columns.contains(&DELETED_RECORD_COUNT)); + let deletions = if needs_deletions + && CoreOptions::new(table.schema().options()).deletion_vectors_enabled() + { + deleted_record_counts(&pinned).await? + } else { + HashMap::new() + }; + let fields = table.schema().partition_fields(); + let options = CoreOptions::new(table.schema().options()); + partition_batch( + entries, + &fields, + options.partition_default_name(), + options.legacy_partition_name(), + &deletions, + ) +} + +async fn deleted_record_counts(table: &Table) -> Result, Option>> { + let Some(snapshot) = table.resolve_read_snapshot().await? else { + return Ok(HashMap::new()); + }; + let Some(name) = snapshot.index_manifest() else { + return Ok(HashMap::new()); + }; + let path = table.snapshot_manager().manifest_path(name); + let entries = IndexManifest::read(table.file_io(), &path).await?; + Ok(accumulate_deletions(entries)) +} + +fn accumulate_deletions(entries: Vec) -> HashMap, Option> { + let mut counts = HashMap::new(); + for entry in entries { + if entry.kind != FileKind::Add || entry.index_file.index_type != "DELETION_VECTORS" { + continue; + } + let Some(ranges) = entry.index_file.deletion_vectors_ranges else { + continue; + }; + if ranges.is_empty() { + continue; + } + let count = counts.entry(entry.partition).or_insert(Some(0_i64)); + for meta in ranges.into_values() { + *count = count + .zip(meta.cardinality) + .map(|(sum, value)| sum.wrapping_add(value)); + } + } + counts +} + +fn partition_string( + row: &BinaryRow, + fields: &[DataField], + default_name: &str, + zone: &TimeZone, +) -> Result { + fields + .iter() + .enumerate() + .map(|(index, field)| { + let value = if row.is_null_at(index) { + default_name.to_string() + } else { + format_field_in_zone(row, index, field.data_type(), zone)? + }; + let value = if crate::spec::is_java_whitespace_only(&value) { + default_name + } else { + &value + }; + Ok(format!("{}={value}", field.name())) + }) + .collect::>>() + .map(|parts| parts.join("/")) +} + +// Java generates a spec using partition.legacy-name, then reconstructs a typed +// row for BOTH display and deletion lookup. Keep this conversion independent of +// whether deletion metadata was requested (notably for LTZ and DST overlaps). +fn normalize_partition( + row: &BinaryRow, + fields: &[DataField], + default_name: &str, + legacy: bool, + zone: &TimeZone, +) -> Result { + let values = fields + .iter() + .enumerate() + .map(|(index, field)| { + let value = row.get_datum(index, field.data_type())?; + let Some(value) = value else { + return Ok(None); + }; + let text = if legacy { + format_row_field_as_java_object_string(row, index, field.data_type())? + .expect("non-null partition field") + } else { + format_field_in_zone(row, index, field.data_type(), zone)? + }; + if text == default_name || crate::spec::is_java_whitespace_only(&text) { + return Ok(None); + } + Ok(Some(match value { + Datum::Bytes(_) => Datum::Bytes(text.into_bytes()), + Datum::Time(_) => Datum::Time(parse_time_spec(&text)?), + Datum::Timestamp { .. } => { + let crate::spec::DataType::Timestamp(typ) = field.data_type() else { + unreachable!() + }; + let (millis, nanos) = + crate::arrow::parse_java_timestamp(&text, typ.precision(), None)?; + Datum::Timestamp { + millis, + nanos: nanos as i32, + } + } + Datum::LocalZonedTimestamp { .. } => { + let crate::spec::DataType::LocalZonedTimestamp(typ) = field.data_type() else { + unreachable!() + }; + let (millis, nanos) = + crate::arrow::parse_java_timestamp(&text, typ.precision(), Some(zone))?; + Datum::LocalZonedTimestamp { + millis, + nanos: nanos as i32, + } + } + // The remaining scalar formatters round-trip the typed value + // unchanged; reuse it without narrowing through Arrow storage. + value => value, + })) + }) + .collect::>>()?; + let refs: Vec<_> = values + .iter() + .zip(fields) + .map(|(value, field)| (value.as_ref(), field.data_type())) + .collect(); + Ok(BinaryRow::from_datums(&refs)) +} + +fn parse_time_spec(text: &str) -> Result { + // BinaryStringUtils.toTime preserves the integer legacy spec. Otherwise + // parse the formatter's HH:mm:ss[.fraction] without chrono's hour limit. + if let Ok(value) = text.parse::() { + return Ok(value); + } + let (time, fraction) = text.split_once('.').unwrap_or((text, "")); + let parts = time + .split(':') + .map(|part| { + part.parse::() + .map_err(|error| super::invalid(error.to_string())) + }) + .collect::>>()?; + let [hours, minutes, seconds] = parts.as_slice() else { + return Err(super::invalid(format!( + "Invalid TIME partition spec '{text}'" + ))); + }; + let fraction = if fraction.is_empty() { + 0 + } else { + let digits = &fraction[..fraction.len().min(3)]; + digits + .parse::() + .map_err(|error| super::invalid(error.to_string()))? + * 10_i32.pow(3 - digits.len() as u32) + }; + Ok((hours * 3600 + minutes * 60 + seconds) * 1000 + fraction) +} + +fn partition_batch( + entries: Vec, + fields: &[DataField], + default_name: &str, + legacy: bool, + deletions: &HashMap, Option>, +) -> Result { + let zone = TimeZone::system(); + let mut rows = entries + .into_iter() + .map(|mut entry| { + entry.partition = + normalize_partition(&entry.partition, fields, default_name, legacy, &zone)?; + Ok(( + partition_string(&entry.partition, fields, default_name, &zone)?, + entry, + )) + }) + .collect::>>()?; + rows.sort_by(|a, b| a.0.cmp(&b.0)); + let n = rows.len(); + let mut names = Vec::with_capacity(n); + let mut records = Vec::with_capacity(n); + let mut sizes = Vec::with_capacity(n); + let mut files = Vec::with_capacity(n); + let mut times = Vec::with_capacity(n); + let mut buckets = Vec::with_capacity(n); + let mut deleted = Vec::with_capacity(n); + for (name, entry) in rows { + names.push(name); + records.push(entry.record_count); + sizes.push(entry.file_size_in_bytes); + files.push(entry.file_count); + times.push(entry.last_file_creation_time); + buckets.push(entry.total_buckets); + deleted.push(if deletions.is_empty() { + Some(0) + } else { + deletions + .get(&entry.partition.to_serialized_bytes()) + .copied() + .unwrap_or(Some(0)) + }); + } + super::record_batch( + partitions_schema(), + vec![ + Arc::new(StringArray::from(names)), + Arc::new(Int64Array::from(records)), + Arc::new(Int64Array::from(sizes)), + Arc::new(Int64Array::from(files)), + super::local_timestamp_column(times)?, + Arc::new(TimestampMillisecondArray::from(vec![None::; n])), + Arc::new(StringArray::from(vec![None::<&str>; n])), + Arc::new(StringArray::from(vec![None::<&str>; n])), + Arc::new(StringArray::from(vec![None::<&str>; n])), + Arc::new(Int32Array::from(buckets)), + Arc::new(BooleanArray::from(vec![false; n])) as ArrayRef, + Arc::new(Int64Array::from(deleted)), + ], + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::spec::{ + DataType as PaimonType, Datum, DeletionVectorMeta, IndexFileMeta, VarCharType, + }; + + fn index(partition: &str, cardinalities: Option<&[Option]>) -> IndexManifestEntry { + let typ = PaimonType::VarChar(VarCharType::string_type()); + let value = Datum::String(partition.into()); + let row = BinaryRow::from_datums(&[(Some(&value), &typ)]); + IndexManifestEntry { + version: 1, + kind: FileKind::Add, + partition: row.to_serialized_bytes(), + bucket: 0, + index_file: IndexFileMeta { + index_type: "DELETION_VECTORS".into(), + file_name: "index".into(), + file_size: 1, + row_count: 1, + deletion_vectors_ranges: cardinalities.map(|values| { + values + .iter() + .enumerate() + .map(|(i, count)| { + ( + format!("file-{i}"), + DeletionVectorMeta { + offset: 0, + length: 1, + cardinality: *count, + }, + ) + }) + .collect() + }), + external_path: None, + global_index_meta: None, + }, + } + } + + #[test] + fn deletion_counts_sum_buckets_and_keep_unknown_cardinality() { + let known = index("known", Some(&[Some(2), Some(3)])); + let mut other_bucket = index("known", Some(&[Some(4)])); + other_bucket.bucket = 1; + let unknown = index("unknown", Some(&[Some(2), None])); + let later = index("unknown", Some(&[Some(7)])); + let absent = index("absent", None); + let empty = index("empty", Some(&[])); + let mut deleted = index("known", Some(&[Some(100)])); + deleted.kind = FileKind::Delete; + let mut unrelated = index("known", Some(&[Some(100)])); + unrelated.index_file.index_type = "HASH".into(); + let counts = accumulate_deletions(vec![ + known.clone(), + other_bucket, + unknown.clone(), + later, + absent.clone(), + empty, + deleted, + unrelated, + ]); + assert_eq!(counts[&known.partition], Some(9)); + assert_eq!(counts[&unknown.partition], None); + assert!(!counts.contains_key(&absent.partition)); + assert_eq!(counts.len(), 2); + } + + #[test] + fn deletion_lookup_uses_typed_partition_specs() { + let zone = TimeZone::UTC; + let typ = PaimonType::VarChar(VarCharType::string_type()); + let fields = [DataField::new(0, "p".into(), typ.clone())]; + let null = BinaryRow::from_datums(&[(None, &typ)]).to_serialized_bytes(); + for value in ["DEFAULT", "\u{1c}"] { + let datum = Datum::String(value.into()); + let row = BinaryRow::from_datums(&[(Some(&datum), &typ)]); + assert_eq!( + normalize_partition(&row, &fields, "DEFAULT", false, &zone) + .unwrap() + .to_serialized_bytes(), + null + ); + } + let typ = PaimonType::VarBinary(crate::spec::VarBinaryType::new(3).unwrap()); + let fields = [DataField::new(0, "p".into(), typ.clone())]; + let original = Datum::Bytes(vec![0xed, 0xa0, 0x80]); + let canonical = Datum::Bytes("�".as_bytes().to_vec()); + let row = BinaryRow::from_datums(&[(Some(&original), &typ)]); + assert_eq!( + normalize_partition(&row, &fields, "DEFAULT", false, &zone) + .unwrap() + .to_serialized_bytes(), + BinaryRow::from_datums(&[(Some(&canonical), &typ)]).to_serialized_bytes() + ); + } + + #[test] + fn time_spec_parser_preserves_java_integer_and_clock_forms() { + for (spec, millis) in [ + ("0", 0), + ("00:00:00", 0), + ("45296120", 45_296_120), + ("12:34:56.12", 45_296_120), + ("23:59:59.999", 86_399_999), + ("24:00:00", 86_400_000), + ] { + assert_eq!(parse_time_spec(spec).unwrap(), millis); + } + } + + #[test] + fn time_spec_round_trip_respects_legacy_names() { + use crate::spec::TimeType; + for precision in 0..=3 { + let typ = PaimonType::Time(TimeType::new(precision).unwrap()); + let fields = [DataField::new(0, "p".into(), typ.clone())]; + let datum = Datum::Time(45_296_123); + let row = BinaryRow::from_datums(&[(Some(&datum), &typ)]); + for legacy in [true, false] { + let normalized = + normalize_partition(&row, &fields, "DEFAULT", legacy, &TimeZone::UTC).unwrap(); + let quantum = if legacy { 1 } else { 10_i32.pow(3 - precision) }; + assert_eq!( + normalized.get_int(0).unwrap(), + 45_296_123 / quantum * quantum + ); + let counts = HashMap::from([(row.to_serialized_bytes(), Some(1))]); + let batch = partition_batch( + vec![FileSystemPartitionEntry { + partition: row.clone(), + ..Default::default() + }], + &fields, + "DEFAULT", + legacy, + &counts, + ) + .unwrap(); + let count = batch + .column(DELETED_RECORD_COUNT) + .as_any() + .downcast_ref::() + .unwrap() + .value(0); + assert_eq!(count, i64::from(legacy || precision == 3)); + } + } + } + + #[test] + fn ltz_spec_round_trip_matches_java_legacy_and_dst_rules() { + use crate::spec::LocalZonedTimestampType; + let zone = TimeZone::get("America/New_York").unwrap(); + let typ = PaimonType::LocalZonedTimestamp(LocalZonedTimestampType::new(3).unwrap()); + let fields = [DataField::new(0, "p".into(), typ.clone())]; + // Expected keys and names were obtained with Java's partition computer, + // convertSpecToInternalRow and CastExecutors.resolveToString. + for (instant, legacy, key, display) in [ + ( + 1_704_067_200_000, + true, + 1_704_085_200_000, + "2024-01-01 00:00:00.000", + ), + ( + 1_704_067_200_000, + false, + 1_704_067_200_000, + "2023-12-31 19:00:00.000", + ), + ( + 1_730_615_400_000, + true, + 1_730_633_400_000, + "2024-11-03 06:30:00.000", + ), + // Parsing 01:30 during the overlap chooses the earlier EDT offset. + ( + 1_730_615_400_000, + false, + 1_730_611_800_000, + "2024-11-03 01:30:00.000", + ), + // A legacy wall time inside the spring gap shifts forward an hour. + ( + 1_710_038_700_000, + true, + 1_710_056_700_000, + "2024-03-10 03:45:00.000", + ), + ] { + let datum = Datum::LocalZonedTimestamp { + millis: instant, + nanos: 0, + }; + let row = BinaryRow::from_datums(&[(Some(&datum), &typ)]); + let normalized = normalize_partition(&row, &fields, "DEFAULT", legacy, &zone).unwrap(); + assert_eq!(normalized.get_timestamp_raw(0, 3).unwrap(), (key, 0)); + assert_eq!( + partition_string(&normalized, &fields, "DEFAULT", &zone).unwrap(), + format!("p={display}") + ); + let counts = HashMap::from([(row.to_serialized_bytes(), Some(1))]); + assert_eq!( + counts + .get(&normalized.to_serialized_bytes()) + .copied() + .unwrap_or(Some(0)), + Some(i64::from(key == instant)) + ); + } + } + + #[test] + fn timestamp_spec_round_trip_uses_declared_precision_without_arrow_overflow() { + use crate::spec::TimestampType; + for (precision, millis, nanos) in [ + (1, 16_725_225_600_100, 0), + (3, 16_725_225_600_123, 0), + (6, 16_725_225_600_123, 456_000), + (7, 16_725_225_600_123, 456_700), + (9, 16_725_225_600_123, 456_789), + ] { + let typ = PaimonType::Timestamp(TimestampType::new(precision).unwrap()); + let fields = [DataField::new(0, "p".into(), typ.clone())]; + let datum = Datum::Timestamp { + millis: 16_725_225_600_123, + nanos: 456_789, + }; + let row = BinaryRow::from_datums(&[(Some(&datum), &typ)]); + for legacy in [true, false] { + let normalized = + normalize_partition(&row, &fields, "DEFAULT", legacy, &TimeZone::UTC).unwrap(); + assert_eq!( + normalized.get_timestamp_raw(0, precision).unwrap(), + (millis, nanos) + ); + } + } + } + + #[test] + fn partition_defaults_use_java_whitespace() { + let typ = PaimonType::VarChar(VarCharType::string_type()); + let fields = [DataField::new(0, "p".into(), typ.clone())]; + for (input, expected) in [ + ("\u{a0}", "\u{a0}"), + ("\u{85}", "\u{85}"), + ("\u{2007}", "\u{2007}"), + ("\u{202f}", "\u{202f}"), + ("\u{1c}", "DEFAULT"), + ("", "DEFAULT"), + ] { + let datum = Datum::String(input.into()); + let row = BinaryRow::from_datums(&[(Some(&datum), &typ)]); + assert_eq!( + partition_string(&row, &fields, "DEFAULT", &TimeZone::UTC).unwrap(), + format!("p={expected}") + ); + } + } + + #[test] + fn partition_names_use_casts_default_name_and_unescaped_values() { + use crate::spec::{BooleanType, DecimalType, TimestampType}; + let types = [ + PaimonType::Boolean(BooleanType::new()), + PaimonType::Decimal(DecimalType::new(10, 2).unwrap()), + PaimonType::Timestamp(TimestampType::new(9).unwrap()), + PaimonType::VarChar(VarCharType::string_type()), + PaimonType::VarChar(VarCharType::string_type()), + ]; + let fields: Vec<_> = types + .iter() + .enumerate() + .map(|(i, t)| DataField::new(i as i32, format!("p{i}"), t.clone())) + .collect(); + let values = [ + Some(Datum::Bool(true)), + Some(Datum::Decimal { + unscaled: 120, + precision: 10, + scale: 2, + }), + Some(Datum::Timestamp { + millis: 16_725_225_600_123, + nanos: 456_789, + }), + Some(Datum::String("a/b=c".into())), + None, + ]; + let refs: Vec<_> = values + .iter() + .zip(types.iter()) + .map(|(v, t)| (v.as_ref(), t)) + .collect(); + let row = BinaryRow::from_datums(&refs); + assert_eq!( + partition_string(&row, &fields, "DEFAULT", &TimeZone::UTC).unwrap(), + "p0=true/p1=1.20/p2=2500-01-01 00:00:00.123456789/p3=a/b=c/p4=DEFAULT" + ); + assert_eq!( + partition_string(&BinaryRow::from_datums(&[]), &[], "DEFAULT", &TimeZone::UTC).unwrap(), + "" + ); + } +} diff --git a/crates/paimon/src/table/system_table/row_string_cast.rs b/crates/paimon/src/table/system_table/row_string_cast.rs index 1101f606f..74a5d829f 100644 --- a/crates/paimon/src/table/system_table/row_string_cast.rs +++ b/crates/paimon/src/table/system_table/row_string_cast.rs @@ -15,20 +15,12 @@ // specific language governing permissions and limitations // under the License. -use std::num::NonZeroI32; - +use crate::common::number::format_java_float; use crate::spec::{BinaryRow, DataField, DataType, Datum}; use crate::{Error, Result}; use chrono::NaiveDate; -use lexical_write_float::{format::STANDARD, Options, ToLexicalWithOptions}; const MILLIS_PER_DAY: i64 = 86_400_000; -const JAVA_FLOAT_OPTIONS: Options = Options::builder() - .positive_exponent_break(NonZeroI32::new(6)) - .negative_exponent_break(NonZeroI32::new(-3)) - .exponent(b'E') - .inf_string(Some(b"Infinity")) - .build_strict(); pub(super) fn format_row_as_java_cast_string( row: &BinaryRow, @@ -150,7 +142,16 @@ fn validate_row_has_field(row: &BinaryRow, pos: usize) -> Result<()> { Ok(()) } -fn format_field(row: &BinaryRow, pos: usize, data_type: &DataType) -> Result { +pub(super) fn format_field(row: &BinaryRow, pos: usize, data_type: &DataType) -> Result { + format_field_in_zone(row, pos, data_type, &jiff::tz::TimeZone::system()) +} + +pub(super) fn format_field_in_zone( + row: &BinaryRow, + pos: usize, + data_type: &DataType, + zone: &jiff::tz::TimeZone, +) -> Result { let Some(datum) = row.get_datum(pos, data_type)? else { return Ok("null".to_string()); }; @@ -161,11 +162,11 @@ fn format_field(row: &BinaryRow, pos: usize, data_type: &DataType) -> Result Ok(v.to_string()), (Datum::Int(v), DataType::Int(_)) => Ok(v.to_string()), (Datum::Long(v), DataType::BigInt(_)) => Ok(v.to_string()), - (Datum::Float(v), DataType::Float(_)) => Ok(format_float(v)), - (Datum::Double(v), DataType::Double(_)) => Ok(format_double(v)), + (Datum::Float(v), DataType::Float(_)) => Ok(format_java_float(v)), + (Datum::Double(v), DataType::Double(_)) => Ok(format_java_float(v)), (Datum::String(v), DataType::Char(_) | DataType::VarChar(_)) => Ok(v), (Datum::Bytes(v), DataType::Binary(_) | DataType::VarBinary(_)) => { - Ok(String::from_utf8_lossy(&v).into_owned()) + Ok(crate::spec::decode_java_utf8(&v)) } (Datum::Date(v), DataType::Date(_)) => format_date(v), (Datum::Time(v), DataType::Time(t)) => format_time(v, t.precision()), @@ -175,18 +176,21 @@ fn format_field(row: &BinaryRow, pos: usize, data_type: &DataType) -> Result Ok(format_decimal_plain(unscaled, scale)), - (Datum::Timestamp { .. } | Datum::LocalZonedTimestamp { .. }, _) => { - let source = row.field_to_arrow(pos, data_type)?; - let target = DataType::VarChar(crate::spec::VarCharType::string_type()); - let strings = - crate::arrow::cast_array_for_schema_evolution(&source, data_type, &target)?; - let strings = strings - .as_any() - .downcast_ref::() - .ok_or_else(|| { - data_invalid("Timestamp string cast returned a non-string array".into()) - })?; - Ok(strings.value(0).to_string()) + (Datum::Timestamp { millis, nanos }, DataType::Timestamp(typ)) => { + Ok(crate::common::date_time::format_timestamp_millis( + millis, + i64::from(nanos), + typ.precision(), + None, + )) + } + (Datum::LocalZonedTimestamp { millis, nanos }, DataType::LocalZonedTimestamp(typ)) => { + Ok(crate::common::date_time::format_timestamp_millis( + millis, + i64::from(nanos), + typ.precision(), + Some(zone), + )) } (datum, _) => Err(data_invalid(format!( "Decoded row string cast datum {datum:?} does not match type {data_type:?}" @@ -194,24 +198,6 @@ fn format_field(row: &BinaryRow, pos: usize, data_type: &DataType) -> Result String { - const BUFFER_SIZE: usize = JAVA_FLOAT_OPTIONS.buffer_size_const::(); - let mut buffer = [0u8; BUFFER_SIZE]; - let bytes = value.to_lexical_with_options::(&mut buffer, &JAVA_FLOAT_OPTIONS); - std::str::from_utf8(bytes) - .expect("lexical float output is valid UTF-8") - .to_string() -} - -fn format_double(value: f64) -> String { - const BUFFER_SIZE: usize = JAVA_FLOAT_OPTIONS.buffer_size_const::(); - let mut buffer = [0u8; BUFFER_SIZE]; - let bytes = value.to_lexical_with_options::(&mut buffer, &JAVA_FLOAT_OPTIONS); - std::str::from_utf8(bytes) - .expect("lexical double output is valid UTF-8") - .to_string() -} - fn format_date(epoch_days: i32) -> Result { let ce_days = epoch_days.checked_add(719_163).ok_or_else(|| { data_invalid(format!( @@ -432,6 +418,24 @@ mod tests { ); } + #[test] + fn floating_partition_casts_use_java_significant_digits() { + assert_eq!( + format_value(&[ + ( + Some(Datum::Float(f32::from_bits(1))), + DataType::Float(FloatType::new()) + ), + ( + Some(Datum::Double(f64::from_bits(1))), + DataType::Double(DoubleType::new()) + ) + ]) + .unwrap(), + "{1.4E-45, 4.9E-324}" + ); + } + #[test] fn test_format_binary_invalid_utf8_uses_lossy_string() { let bytes = vec![0xff]; @@ -444,6 +448,15 @@ mod tests { ); } + #[test] + fn binary_surrogates_use_java_utf8_replacement() { + let typ = DataType::VarBinary(VarBinaryType::new(3).unwrap()); + assert_eq!( + format_value(&[(Some(Datum::Bytes(vec![0xed, 0xa0, 0x80])), typ)]).unwrap(), + "{�}" + ); + } + #[test] fn test_format_null_values() { let values = vec![ @@ -479,6 +492,54 @@ mod tests { ); } + #[test] + fn high_precision_timestamps_do_not_narrow_to_arrow_ticks() { + for precision in 7..=9 { + let typ = DataType::Timestamp(TimestampType::new(precision).unwrap()); + for (millis, expected) in [ + (16_725_225_600_123, "2500-01-01 00:00:00.123456789"), + (-8_520_335_999_877, "1700-01-01 00:00:00.123456789"), + ] { + let expected = &expected[..expected.len() - (9 - precision) as usize]; + let quantum = 10_i32.pow(9 - precision); + let nanos = 456_789 / quantum * quantum; + assert_eq!( + format_value(&[(Some(Datum::Timestamp { millis, nanos }), typ.clone())]) + .unwrap(), + format!("{{{expected}}}") + ); + } + } + } + + #[test] + fn high_precision_ltz_timestamps_keep_submillisecond_values() { + let millis = 16_725_225_600_123; + let local = chrono::Local.timestamp_millis_opt(millis).single().unwrap(); + for precision in 7..=9 { + let quantum = 10_i32.pow(9 - precision); + let nanos = 456_789 / quantum * quantum; + let expected = format!( + "{{{}.{:09}}}", + local.format("%Y-%m-%d %H:%M:%S"), + 123_000_000 + nanos + ); + let expected = format!( + "{}{}", + &expected[..expected.len() - 1 - (9 - precision) as usize], + "}" + ); + assert_eq!( + format_value(&[( + Some(Datum::LocalZonedTimestamp { millis, nanos }), + DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(precision).unwrap()) + )]) + .unwrap(), + expected + ); + } + } + #[test] fn test_format_local_zoned_timestamp() { let data_type = DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(3).unwrap()); diff --git a/crates/paimon/tests/partition_system_read_test.rs b/crates/paimon/tests/partition_system_read_test.rs new file mode 100644 index 000000000..ae1f7d0a6 --- /dev/null +++ b/crates/paimon/tests/partition_system_read_test.rs @@ -0,0 +1,221 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +mod common; + +use arrow_array::{Int32Array, Int64Array, RecordBatch, StringArray}; +use common::incremental_helpers::{memory_table, persist_table_schema, setup_dirs, write_batch}; +use paimon::spec::{DataType, IntType, Schema, TableSchema, VarCharType}; +use paimon::table::Table; +use std::sync::Arc; + +async fn table() -> Table { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("pt", DataType::VarChar(VarCharType::string_type())) + .partition_keys(["pt"]) + .option("bucket", "8") + .option("bucket-key", "id") + .option("deletion-vectors.enabled", "true") + .build() + .unwrap(); + let (io, table) = memory_table("memory:/partition_metadata", TableSchema::new(0, &schema)); + setup_dirs(&io, table.location()).await; + persist_table_schema(&io, table.location(), table.schema()).await; + table +} + +async fn append(table: &Table, ids: Vec, partitions: Vec>) { + let batch = RecordBatch::try_from_iter([ + ( + "id", + Arc::new(Int32Array::from(ids)) as arrow_array::ArrayRef, + ), + ( + "pt", + Arc::new(StringArray::from(partitions)) as arrow_array::ArrayRef, + ), + ]) + .unwrap(); + write_batch(table, &batch).await; +} + +fn strings(batch: &RecordBatch, column: &str) -> Vec { + batch + .column_by_name(column) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .map(|value| value.unwrap().to_string()) + .collect() +} + +fn longs(batch: &RecordBatch, column: &str) -> Vec> { + batch + .column_by_name(column) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .collect() +} + +#[tokio::test] +async fn partition_statistics_follow_captured_snapshots_and_declared_buckets() { + let table = table().await; + let read = table.new_system_read("partitions").unwrap(); + let empty = read.read().await.unwrap(); + assert_eq!(empty.num_rows(), 0); + assert_eq!(empty.schema(), read.schema()); + let fixed_empty = table.copy_with_pinned_snapshot(None); + append(&table, vec![1, 2, 3], vec![Some("z"), None, Some("z")]).await; + let first = table.resolve_read_snapshot().await.unwrap().unwrap(); + let historical = table.copy_with_pinned_snapshot(Some(&first)); + append(&table, vec![4], vec![Some("a")]).await; + let batch = read.read().await.unwrap(); + assert_eq!( + strings(&batch, "partition"), + ["pt=__DEFAULT_PARTITION__", "pt=a", "pt=z"] + ); + assert_eq!(longs(&batch, "record_count"), [Some(1), Some(1), Some(2)]); + let buckets = batch + .column_by_name("total_buckets") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + assert!(buckets.iter().all(|value| value == Some(8))); + assert_eq!(longs(&batch, "deleted_record_count"), [Some(0); 3]); + assert_eq!( + historical + .new_system_read("partitions") + .unwrap() + .read() + .await + .unwrap() + .num_rows(), + 2 + ); + assert_eq!(historical.partition_stats().await.unwrap().len(), 2); + assert!(fixed_empty.partition_stats().await.unwrap().is_empty()); + assert!( + paimon::catalog::list_partitions_from_file_system(&fixed_empty) + .await + .unwrap() + .is_empty() + ); + assert_eq!( + fixed_empty + .new_system_read("partitions") + .unwrap() + .read() + .await + .unwrap() + .num_rows(), + 0 + ); + for partition in paimon::catalog::list_partitions_from_file_system(&table) + .await + .unwrap() + { + assert_eq!(partition.total_buckets, 8); + } +} + +#[tokio::test] +async fn deletion_count_filter_reads_hidden_column_and_projection_skips_index_io() { + use paimon::spec::{ + DeletionVectorMeta, FileKind, IndexFileMeta, IndexManifest, IndexManifestEntry, Predicate, + PredicateOperator, + }; + let table = table().await; + append(&table, vec![1, 2], vec![Some("a"), Some("b")]).await; + let plan = table.new_read_builder().new_scan().plan().await.unwrap(); + let mut entries = Vec::new(); + for split in plan.splits() { + let cardinality = if split.partition().get_string(0).unwrap() == "a" { + Some(1) + } else { + None + }; + entries.push(IndexManifestEntry { + version: 1, + kind: FileKind::Add, + partition: split.partition().to_serialized_bytes(), + bucket: split.bucket(), + index_file: IndexFileMeta { + index_type: "DELETION_VECTORS".into(), + file_name: "unused-index-file".into(), + file_size: 1, + row_count: 1, + deletion_vectors_ranges: Some( + [( + split.data_files()[0].file_name.clone(), + DeletionVectorMeta { + offset: 0, + length: 1, + cardinality, + }, + )] + .into_iter() + .collect(), + ), + external_path: None, + global_index_meta: None, + }, + }); + } + let manager = table.snapshot_manager(); + let path = manager.manifest_path("test-index-manifest"); + IndexManifest::write(table.file_io(), &path, &entries) + .await + .unwrap(); + let snapshot = table.resolve_read_snapshot().await.unwrap().unwrap(); + let mut json = serde_json::to_value(&snapshot).unwrap(); + json["indexManifest"] = serde_json::json!("test-index-manifest"); + table + .file_io() + .new_output(&manager.snapshot_path(snapshot.id())) + .unwrap() + .write(serde_json::to_vec(&json).unwrap().into()) + .await + .unwrap(); + let mut read = table.new_system_read("partitions").unwrap(); + assert_eq!( + longs(&read.read().await.unwrap(), "deleted_record_count"), + [Some(1), None] + ); + read.with_projection(&["partition"]).unwrap(); + read.with_filter(Predicate::Leaf { + column: "deleted_record_count".into(), + index: 999, + data_type: DataType::Int(IntType::new()), + op: PredicateOperator::IsNull, + literals: vec![], + }) + .unwrap(); + read.with_limit(1); + assert_eq!(strings(&read.read().await.unwrap(), "partition"), ["pt=b"]); + table.file_io().delete_file(&path).await.unwrap(); + let mut projected = table.new_system_read("partitions").unwrap(); + projected.with_projection(&["partition"]).unwrap(); + assert_eq!(projected.read().await.unwrap().num_rows(), 2); + assert!(read.read().await.is_err()); +}