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 52f82d33 fix(core): align timestamp partition paths with Java (#833)
52f82d33 is described below
commit 52f82d3364cd8ab0efc2261a1e7218a1f7e6879c
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 15 11:44:52 2026 +0800
fix(core): align timestamp partition paths with Java (#833)
---
bindings/python/tests/test_write.py | 33 +++++++
crates/paimon/src/spec/partition_utils.rs | 143 ++++++++++++++++++++++++-----
crates/paimon/src/table/index_file_path.rs | 88 ++++++++++++++++++
3 files changed, 240 insertions(+), 24 deletions(-)
diff --git a/bindings/python/tests/test_write.py
b/bindings/python/tests/test_write.py
index aeac37d5..8d524354 100644
--- a/bindings/python/tests/test_write.py
+++ b/bindings/python/tests/test_write.py
@@ -16,6 +16,7 @@
# under the License.
import tempfile
+from datetime import datetime
import pyarrow as pa
import pytest
@@ -248,3 +249,35 @@ def test_abort_different_builder_same_table_raises():
messages = write.prepare_commit()
with pytest.raises(ValueError):
table.new_write_builder().new_commit().abort(messages)
+
+
[email protected]("legacy,precision,unit,value,directory", [
+ (False, 3, "ms", datetime(2026, 9, 15, 12), "ts=2026-09-15
12%3A00%3A00.000"),
+ (False, 6, "us", datetime(2026, 9, 15, 12), "ts=2026-09-15
12%3A00%3A00.000000"),
+ (False, 3, "ms", datetime(2026, 9, 15, 12, 0, 0, 120000), "ts=2026-09-15
12%3A00%3A00.120"),
+ (True, 3, "ms", datetime(2026, 9, 15, 12, 0, 0, 100000),
"ts=2026-09-15T12%3A00%3A00.100"),
+ (True, 6, "us", datetime(2026, 9, 15, 12, 0, 0, 120100),
"ts=2026-09-15T12%3A00%3A00.120100"),
+])
+def test_timestamp_partition_writes_and_reads_use_java_paths(tmp_path, legacy,
precision, unit, value, directory):
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": str(tmp_path)})
+ ctx.sql("CREATE SCHEMA paimon.wdb")
+ ctx.sql(
+ "CREATE TABLE paimon.wdb.t (id INT, ts TIMESTAMP({})) PARTITIONED BY
(ts) "
+ "WITH ('partition.legacy-name' = '{}')".format(precision,
str(legacy).lower())
+ )
+ schema = pa.schema([("id", pa.int32()), ("ts", pa.timestamp(unit))])
+ for row_id in [1, 2]:
+ table = _get_table(str(tmp_path))
+ builder = table.new_write_builder()
+ write = builder.new_write()
+ write.write_arrow(pa.record_batch([[row_id], [value]], schema=schema))
+ builder.new_commit().commit(write.prepare_commit())
+ files = list(tmp_path.rglob("data-*.parquet"))
+ assert len(files) == 2
+ assert all(file.parent.parent.name == directory for file in files)
+ table = _get_table(str(tmp_path))
+ plan = table.new_read_builder().new_scan().plan()
+ assert sum(split.row_count() for split in plan.splits()) == 2
+ rows = pa.Table.from_batches(ctx.sql("SELECT id, ts FROM
paimon.wdb.t")).sort_by("id")
+ assert rows.to_pydict() == {"id": [1, 2], "ts": [value, value]}
diff --git a/crates/paimon/src/spec/partition_utils.rs
b/crates/paimon/src/spec/partition_utils.rs
index 69e1941d..60b83eb2 100644
--- a/crates/paimon/src/spec/partition_utils.rs
+++ b/crates/paimon/src/spec/partition_utils.rs
@@ -270,16 +270,17 @@ fn format_partition_value(
DataType::LocalZonedTimestamp(t) => {
let (millis, nano_of_milli) = row.get_timestamp_raw(pos,
t.precision())?;
+ let dt = if legacy {
+ // Legacy Timestamp.toString() does not apply timezone
conversion.
+ millis_to_naive_datetime(millis, nano_of_milli)
+ } else {
+ // Non-legacy TimestampToStringCastRule applies
TimeZone.getDefault().
+ epoch_millis_to_local_datetime(millis, nano_of_milli)
+ };
if legacy {
- // Legacy: Timestamp.toString() → toLocalDateTime().toString(),
- // which does NOT apply timezone conversion.
- let dt = millis_to_naive_datetime(millis, nano_of_milli);
format_timestamp_legacy(dt)
} else {
- // Non-legacy: convert to local timezone, mirroring Java
- // TimestampToStringCastRule which applies
TimeZone.getDefault().
- let local_dt = epoch_millis_to_local_datetime(millis,
nano_of_milli);
- format_timestamp_non_legacy(local_dt, t.precision())
+ format_timestamp_non_legacy(dt, t.precision())
}
}
@@ -446,7 +447,7 @@ fn epoch_millis_to_local_datetime(millis: i64,
nano_of_milli: i32) -> NaiveDateT
///
/// - Omits seconds if seconds == 0 and nanos == 0: `2024-01-01T12:34`
/// - Omits fractional part if nanos == 0: `2024-01-01T12:34:56`
-/// - Appends minimal fractional digits (strips trailing zeros):
`2024-01-01T12:34:56.123`
+/// - Uses 3, 6, or 9 fractional digits: `2024-01-01T12:34:56.100`
fn format_timestamp_legacy(dt: NaiveDateTime) -> String {
let nano = dt.nanosecond();
let sec = dt.second();
@@ -460,9 +461,15 @@ fn format_timestamp_legacy(dt: NaiveDateTime) -> String {
let mut result = format!("{date_hour_min}:{sec:02}");
if nano > 0 {
let frac = format!("{nano:09}");
- let trimmed = frac.trim_end_matches('0');
+ let digits = if nano.is_multiple_of(1_000_000) {
+ 3
+ } else if nano.is_multiple_of(1_000) {
+ 6
+ } else {
+ 9
+ };
result.push('.');
- result.push_str(trimmed);
+ result.push_str(&frac[..digits]);
}
result
}
@@ -470,16 +477,12 @@ fn format_timestamp_legacy(dt: NaiveDateTime) -> String {
/// Format a timestamp using non-legacy `DateTimeUtils.formatTimestamp()`
semantics.
///
/// Always uses space separator: `yyyy-MM-dd HH:mm:ss[.fraction]`.
-/// Fraction: pad nano to 9 digits, strip trailing zeros down to at most
`precision` digits.
+/// Fraction: pad nano to 9 digits, strip trailing zeros while keeping at
least `precision` digits.
fn format_timestamp_non_legacy(dt: NaiveDateTime, precision: u32) -> String {
let nano = dt.nanosecond();
let ymdhms = dt.format("%Y-%m-%d %H:%M:%S").to_string();
- if precision == 0 || nano == 0 {
- return ymdhms;
- }
-
- // Pad nano to 9 digits, then strip trailing zeros but keep at least up to
`precision` digits.
+ // Whole seconds keep the declared precision; nonzero fractions are never
truncated.
let nano_str = format!("{nano:09}");
let mut fraction = &nano_str[..];
@@ -930,12 +933,22 @@ mod tests {
.and_hms_opt(12, 34, 56)
.unwrap();
assert_eq!(format_timestamp_legacy(dt2), "2024-01-01T12:34:56");
- // 3. nano>0: include fraction with trailing zero stripping
- let dt3 = NaiveDate::from_ymd_opt(2024, 1, 1)
- .unwrap()
- .and_hms_nano_opt(12, 34, 56, 123_000_000)
- .unwrap();
- assert_eq!(format_timestamp_legacy(dt3), "2024-01-01T12:34:56.123");
+ // Java LocalDateTime.toString() emits fractions in groups of three
digits.
+ for (nanos, expected) in [
+ (100_000_000, "2024-01-01T12:34:56.100"),
+ (120_000_000, "2024-01-01T12:34:56.120"),
+ (123_000_000, "2024-01-01T12:34:56.123"),
+ (123_400_000, "2024-01-01T12:34:56.123400"),
+ (123_456_000, "2024-01-01T12:34:56.123456"),
+ (123_456_700, "2024-01-01T12:34:56.123456700"),
+ (1, "2024-01-01T12:34:56.000000001"),
+ ] {
+ let dt = NaiveDate::from_ymd_opt(2024, 1, 1)
+ .unwrap()
+ .and_hms_nano_opt(12, 34, 56, nanos)
+ .unwrap();
+ assert_eq!(format_timestamp_legacy(dt), expected);
+ }
}
#[test]
@@ -944,8 +957,11 @@ mod tests {
.unwrap()
.and_hms_nano_opt(12, 0, 0, 123_000_000)
.unwrap();
- // precision=0 → no fraction
- assert_eq!(format_timestamp_non_legacy(dt, 0), "2024-01-01 12:00:00");
+ // Precision is a minimum fraction width, not a truncation boundary.
+ assert_eq!(
+ format_timestamp_non_legacy(dt, 0),
+ "2024-01-01 12:00:00.123"
+ );
// precision=3 → strip trailing zeros down to 3 digits
assert_eq!(
format_timestamp_non_legacy(dt, 3),
@@ -958,6 +974,85 @@ mod tests {
);
}
+ #[test]
+ fn test_timestamp_non_legacy_whole_seconds_keep_declared_precision() {
+ let dt = NaiveDate::from_ymd_opt(2024, 1, 1)
+ .unwrap()
+ .and_hms_opt(12, 0, 0)
+ .unwrap();
+ for (precision, expected) in [
+ (0, "2024-01-01 12:00:00"),
+ (1, "2024-01-01 12:00:00.0"),
+ (3, "2024-01-01 12:00:00.000"),
+ (6, "2024-01-01 12:00:00.000000"),
+ (9, "2024-01-01 12:00:00.000000000"),
+ ] {
+ assert_eq!(format_timestamp_non_legacy(dt, precision), expected);
+ }
+ }
+
+ #[test]
+ fn test_timestamp_non_legacy_preserves_significant_fraction_digits() {
+ for (nanos, precision, expected) in [
+ (120_000_000, 0, "2024-01-01 12:00:00.12"),
+ (120_000_000, 3, "2024-01-01 12:00:00.120"),
+ (123_456_700, 0, "2024-01-01 12:00:00.1234567"),
+ (123_456_700, 3, "2024-01-01 12:00:00.1234567"),
+ (123_456_700, 9, "2024-01-01 12:00:00.123456700"),
+ ] {
+ let dt = NaiveDate::from_ymd_opt(2024, 1, 1)
+ .unwrap()
+ .and_hms_nano_opt(12, 0, 0, nanos)
+ .unwrap();
+ assert_eq!(format_timestamp_non_legacy(dt, precision), expected);
+ }
+ }
+
+ #[test]
+ fn test_timestamp_partition_paths_match_java() {
+ for (nanos, precision, legacy, expected) in [
+ (0, 0, false, "ts=2024-01-01 12%3A34%3A00/"),
+ (0, 3, false, "ts=2024-01-01 12%3A34%3A00.000/"),
+ (120_000_000, 0, false, "ts=2024-01-01 12%3A34%3A00.12/"),
+ (100_000_000, 3, true, "ts=2024-01-01T12%3A34%3A00.100/"),
+ (123_400_000, 6, true, "ts=2024-01-01T12%3A34%3A00.123400/"),
+ (
+ 123_456_700,
+ 9,
+ true,
+ "ts=2024-01-01T12%3A34%3A00.123456700/",
+ ),
+ ] {
+ let dt = NaiveDate::from_ymd_opt(2024, 1, 1)
+ .unwrap()
+ .and_hms_nano_opt(12, 34, 0, nanos)
+ .unwrap();
+ let fields = vec![make_field(
+ "ts",
+ DataType::Timestamp(TimestampType::new(precision).unwrap()),
+ )];
+ let computer = PartitionComputer::new(
+ &["ts".to_string()],
+ &fields,
+ TEST_DEFAULT_PARTITION_NAME,
+ legacy,
+ )
+ .unwrap();
+ let mut builder = crate::spec::BinaryRowBuilder::new(1);
+ if precision <= 3 {
+ builder.write_timestamp_compact(0,
dt.and_utc().timestamp_millis());
+ } else {
+ builder.write_timestamp_non_compact(
+ 0,
+ dt.and_utc().timestamp_millis(),
+ (nanos % 1_000_000) as i32,
+ );
+ }
+ let row = builder.build();
+ assert_eq!(computer.generate_partition_path(&row).unwrap(),
expected);
+ }
+ }
+
#[test]
fn test_timestamp_partition_legacy() {
// 2024-01-01 12:34:00 UTC = epoch millis 1704110040000
diff --git a/crates/paimon/src/table/index_file_path.rs
b/crates/paimon/src/table/index_file_path.rs
index 2b540332..d96743fc 100644
--- a/crates/paimon/src/table/index_file_path.rs
+++ b/crates/paimon/src/table/index_file_path.rs
@@ -312,6 +312,94 @@ mod tests {
}
}
+ #[tokio::test]
+ async fn timestamp_dvs_use_canonical_paths_and_explicit_precedence() {
+ use crate::catalog::Identifier;
+ use crate::io::FileIOBuilder;
+ use crate::spec::{
+ bucket_path_under, BinaryRowBuilder, DataType, FileKind, Schema,
TableSchema,
+ TimestampType,
+ };
+ for (legacy, millis, java_partition) in [
+ (false, 0, "ts=1970-01-01 00%3A00%3A00.000/"),
+ (true, 100, "ts=1970-01-01T00%3A00%3A00.100/"),
+ ] {
+ let dir = tempfile::tempdir().unwrap();
+ let table_path = dir.path().to_str().unwrap();
+ let schema = Schema::builder()
+ .column("ts",
DataType::Timestamp(TimestampType::new(3).unwrap()))
+ .partition_keys(vec!["ts".to_string()])
+ .option("partition.legacy-name", legacy.to_string())
+ .option("index-file-in-data-file-dir", "true")
+ .build()
+ .unwrap();
+ let schema = TableSchema::new(0, &schema);
+ let file_io = FileIOBuilder::new("file").build().unwrap();
+ let table = Table::new(
+ file_io,
+ Identifier::new("db", "t"),
+ table_path.to_string(),
+ schema,
+ None,
+ );
+ let java_bucket = bucket_path_under(table_path, java_partition, 7);
+ let table_index = format!("{table_path}/index");
+ for path in [&java_bucket, &table_index] {
+ std::fs::create_dir_all(path).unwrap();
+ }
+ for (parent, name, bytes) in [
+ (&java_bucket, "idx-java", b"Java".as_slice()),
+ (&table_index, "idx-java", b"wrong table copy".as_slice()),
+ (&table_index, "idx-table", b"table".as_slice()),
+ (&java_bucket, "idx-explicit", b"wrong Java copy".as_slice()),
+ (&table_index, "idx-explicit", b"wrong table copy".as_slice()),
+ ] {
+ std::fs::write(format!("{parent}/{name}"), bytes).unwrap();
+ }
+ let mut row = BinaryRowBuilder::new(1);
+ row.write_timestamp_compact(0, millis);
+ let partition = row.build_serialized();
+ let mut entries: Vec<_> = ["idx-java", "idx-table",
"idx-explicit", "idx-missing"]
+ .into_iter()
+ .map(|name| {
+ let mut file = committed_file(None);
+ file.file_name = name.to_string();
+ file.index_type = "DELETION_VECTORS".to_string();
+ IndexManifestEntry {
+ version: 1,
+ kind: FileKind::Add,
+ partition: partition.clone(),
+ bucket: 7,
+ index_file: file,
+ }
+ })
+ .collect();
+ let explicit =
format!("{table_path}/missing-explicit/idx-explicit");
+ entries[2].index_file.external_path = Some(explicit.clone());
+ resolve_legacy_deletion_vector_entries(&table, &mut entries)
+ .await
+ .unwrap();
+ assert_eq!(entries[0].index_file.external_path, None);
+ assert_eq!(
+ entries[1].index_file.external_path,
+ Some(format!("{table_index}/idx-table"))
+ );
+ assert_eq!(entries[2].index_file.external_path,
Some(explicit.clone()));
+ assert_eq!(entries[3].index_file.external_path, None);
+ let location = IndexFileLocation::BucketLocal {
+ table_path,
+ bucket_path: &java_bucket,
+ index_file_in_data_file_dir: true,
+ };
+ for (entry, expected) in
entries.iter().take(2).zip([b"Java".as_slice(), b"table"]) {
+ let file = &entry.index_file;
+ let path = location.resolve(&file.file_name,
file.external_path.as_deref());
+ assert_eq!(std::fs::read(path).unwrap(), expected);
+ }
+ assert!(!table.file_io().exists(&explicit).await.unwrap());
+ }
+ }
+
/// A committed index file with the given `_GLOBAL_INDEX` and
`_SOURCE_META`.
fn committed_file(global_index_meta: Option<Option<Vec<u8>>>) ->
IndexFileMeta {
IndexFileMeta {
