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 4e6c25ad [core] Support filesystem partitions in SystemRead (#1138)
4e6c25ad is described below
commit 4e6c25ad330e259c4a2f578f2a30292df788acc3
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 16 20:28:06 2026 +0800
[core] Support filesystem partitions in SystemRead (#1138)
---
.../datafusion/src/system_tables/partitions.rs | 22 +-
.../datafusion/tests/sql_context_tests.rs | 12 +-
.../integrations/datafusion/tests/system_tables.rs | 70 ++-
crates/paimon/src/arrow/cast_text.rs | 2 +-
crates/paimon/src/arrow/evolution_string.rs | 14 +-
crates/paimon/src/arrow/mod.rs | 1 +
crates/paimon/src/catalog/mod.rs | 3 +
crates/paimon/src/catalog/partition_listing.rs | 226 +++++---
crates/paimon/src/common/date_time.rs | 17 +
crates/paimon/src/spec/mod.rs | 1 +
crates/paimon/src/spec/partition_utils.rs | 2 +-
crates/paimon/src/table/partition_stat.rs | 137 ++---
crates/paimon/src/table/system_table/mod.rs | 11 +-
crates/paimon/src/table/system_table/partitions.rs | 626 +++++++++++++++++++++
.../src/table/system_table/row_string_cast.rs | 147 +++--
crates/paimon/tests/partition_system_read_test.rs | 221 ++++++++
16 files changed, 1269 insertions(+), 243 deletions(-)
diff --git a/crates/integrations/datafusion/src/system_tables/partitions.rs
b/crates/integrations/datafusion/src/system_tables/partitions.rs
index b87db4f8..b2313ffc 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<Option<i64>>) -> DFResult<ArrayRef> {
+ // 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 b4ddf07c..325048a0 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::<datafusion::arrow::array::Int32Array>()
+ .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 48f299bc..717f9baa 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::<TimestampMillisecondArray>()
+ .unwrap()
+ .iter()
+ })
+ .collect::<Vec<_>>();
+ let expected = core
+ .column_by_name("last_update_time")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<TimestampMillisecondArray>()
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>();
+ 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::<TimestampMillisecondArray>()
+ .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 424ec7cb..eeb45e48 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<i32>; 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 533cd90c..d000877b 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<String>) -> crate::Error {
diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs
index a7119809..7a057c0a 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 20a13da0..5c48f454 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 dec7702b..5557cf3b 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<Vec<Partition>> {
- // 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<Vec<Parti
core.partition_default_name(),
core.legacy_partition_name(),
)?;
+ list_partition_entries(table)
+ .await?
+ .into_iter()
+ .map(|entry| {
+ let spec = computer
+ .generate_part_values(&entry.partition)?
+ .into_iter()
+ .collect();
+ Ok(Partition {
+ spec,
+ record_count: entry.record_count,
+ file_size_in_bytes: entry.file_size_in_bytes,
+ file_count: entry.file_count,
+ last_file_creation_time:
entry.last_file_creation_time.unwrap_or(0),
+ total_buckets: entry.total_buckets,
+ done: false,
+ created_at: None,
+ created_by: None,
+ updated_at: None,
+ updated_by: None,
+ options: None,
+ })
+ })
+ .collect()
+}
+
+/// Typed file statistics shared by filesystem catalog listings and metadata
reads.
+#[derive(Debug)]
+pub(crate) struct FileSystemPartitionEntry {
+ pub partition: BinaryRow,
+ pub record_count: i64,
+ pub file_size_in_bytes: i64,
+ pub file_count: i64,
+ // Epoch instant, matching Java DataFileMeta.creationTimeEpochMillis.
+ pub last_file_creation_time: Option<i64>,
+ 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<Vec<u8>, Agg> = BTreeMap::new();
- for entry in &all_entries {
+}
+
+pub(crate) async fn list_partition_entries(table: &Table) ->
Result<Vec<FileSystemPartitionEntry>> {
+ 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<ManifestEntry>,
+) -> Result<Vec<FileSystemPartitionEntry>> {
+ aggregate_partition_entries_in_zone(entries, &jiff::tz::TimeZone::system())
+}
+
+fn aggregate_partition_entries_in_zone(
+ entries: Vec<ManifestEntry>,
+ zone: &jiff::tz::TimeZone,
+) -> Result<Vec<FileSystemPartitionEntry>> {
+ let mut grouped: BTreeMap<Vec<u8>, 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 ecb9fd53..e80b1dfc 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 02c50c9c..ba45b051 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 f534c1f4..85c1bf3a 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 c7663a8a..c59c5737 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<Vec<PartitionStat>> {
- // 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<Vec<ManifestFileMeta>> {
- 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::<ManifestFileMeta>(&bytes)
-}
-
-async fn read_all_manifest_entries(
- file_io: &FileIO,
- table_path: &str,
- snapshot: &Snapshot,
-) -> crate::Result<Vec<ManifestEntry>> {
- 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::<ManifestEntry>(&bytes)?;
- all_entries.extend(entries);
- }
- Ok(all_entries)
+fn partition_stats_from_entries(
+ entries: Vec<FileSystemPartitionEntry>,
+ computer: &PartitionComputer,
+) -> crate::Result<Vec<PartitionStat>> {
+ 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<ManifestEntry>,
computer: &PartitionComputer,
) -> crate::Result<Vec<PartitionStat>> {
- // 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<Vec<u8>, 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 b7893dbe..d63b33d4 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<ArrayRef>) -> Result<RecordBatch
RecordBatch::try_new(schema, columns).map_err(arrow_error)
}
-fn local_timestamp_column(values: Vec<i64>) -> Result<ArrayRef> {
+fn local_timestamp_column(
+ values: impl Into<arrow_array::TimestampMillisecondArray>,
+) -> Result<ArrayRef> {
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 00000000..dd793cbf
--- /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<SchemaRef> = 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<RecordBatch> {
+ // 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<HashMap<Vec<u8>,
Option<i64>>> {
+ 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<IndexManifestEntry>) -> HashMap<Vec<u8>,
Option<i64>> {
+ 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<String> {
+ 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::<Result<Vec<_>>>()
+ .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<BinaryRow> {
+ 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::<Result<Vec<_>>>()?;
+ 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<i32> {
+ // 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::<i32>() {
+ return Ok(value);
+ }
+ let (time, fraction) = text.split_once('.').unwrap_or((text, ""));
+ let parts = time
+ .split(':')
+ .map(|part| {
+ part.parse::<i32>()
+ .map_err(|error| super::invalid(error.to_string()))
+ })
+ .collect::<Result<Vec<_>>>()?;
+ 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::<i32>()
+ .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<FileSystemPartitionEntry>,
+ fields: &[DataField],
+ default_name: &str,
+ legacy: bool,
+ deletions: &HashMap<Vec<u8>, Option<i64>>,
+) -> Result<RecordBatch> {
+ 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::<Result<Vec<_>>>()?;
+ 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::<i64>; 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<i64>]>) ->
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::<Int64Array>()
+ .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 1101f606..74a5d829 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<String> {
+pub(super) fn format_field(row: &BinaryRow, pos: usize, data_type: &DataType)
-> Result<String> {
+ 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<String> {
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<Str
(Datum::SmallInt(v), DataType::SmallInt(_)) => 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<Str
},
DataType::Decimal(_),
) => 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::<arrow_array::StringArray>()
- .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<Str
}
}
-fn format_float(value: f32) -> String {
- const BUFFER_SIZE: usize = JAVA_FLOAT_OPTIONS.buffer_size_const::<f32,
STANDARD>();
- let mut buffer = [0u8; BUFFER_SIZE];
- let bytes = value.to_lexical_with_options::<STANDARD>(&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::<f64,
STANDARD>();
- let mut buffer = [0u8; BUFFER_SIZE];
- let bytes = value.to_lexical_with_options::<STANDARD>(&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<String> {
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 00000000..ae1f7d0a
--- /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<i32>, partitions: Vec<Option<&str>>) {
+ 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<String> {
+ batch
+ .column_by_name(column)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .iter()
+ .map(|value| value.unwrap().to_string())
+ .collect()
+}
+
+fn longs(batch: &RecordBatch, column: &str) -> Vec<Option<i64>> {
+ batch
+ .column_by_name(column)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .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::<Int32Array>()
+ .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());
+}
