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 d3971a7a feat(core): support Java scalar partition values in native
table IO (#1083)
d3971a7a is described below
commit d3971a7a087fa27dd526afb0cc5836c2f52883d7
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 9 16:00:32 2026 +0800
feat(core): support Java scalar partition values in native table IO (#1083)
---
Cargo.lock | 1 +
crates/paimon/Cargo.toml | 2 +
crates/paimon/src/spec/partition_utils.rs | 158 ++++++++++++++++++++++++----
crates/paimon/src/table/table_commit.rs | 164 +++++++++++++++++++++---------
4 files changed, 257 insertions(+), 68 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 30b369e4..69911e8b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4708,6 +4708,7 @@ dependencies = [
"regex",
"reqwest 0.12.28",
"roaring",
+ "ryu",
"serde",
"serde_bytes",
"serde_json",
diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml
index 72eee590..f9aaffa8 100644
--- a/crates/paimon/Cargo.toml
+++ b/crates/paimon/Cargo.toml
@@ -109,6 +109,8 @@ roaring = "0.11"
datasketches = "0.2"
num-bigint = "0.4"
num-traits = "0.2"
+# Already in the Arrow dependency tree; shortest decimals use ties-to-even.
+ryu = "1"
crc32fast = "1"
zstd = "0.13"
snap = "1"
diff --git a/crates/paimon/src/spec/partition_utils.rs
b/crates/paimon/src/spec/partition_utils.rs
index d219d928..b0466eaa 100644
--- a/crates/paimon/src/spec/partition_utils.rs
+++ b/crates/paimon/src/spec/partition_utils.rs
@@ -390,6 +390,8 @@ fn format_partition_value(
DataType::SmallInt(_) => row.get_short(pos)?.to_string(),
DataType::Int(_) => row.get_int(pos)?.to_string(),
DataType::BigInt(_) => row.get_long(pos)?.to_string(),
+ DataType::Float(_) => format_java_float(row.get_float(pos)?),
+ DataType::Double(_) => format_java_float(row.get_double(pos)?),
DataType::Char(_) | DataType::VarChar(_) => {
let s = row.get_string(pos)?;
@@ -463,12 +465,7 @@ fn format_partition_value(
}
}
- // Float/Double: Rust f32/f64 Display differs from Java
Float/Double.toString()
- // on edge-case values. This could silently produce wrong partition
paths.
- // Since float partition keys are extremely rare, reject them
explicitly.
- DataType::Float(_)
- | DataType::Double(_)
- | DataType::Binary(_)
+ DataType::Binary(_)
| DataType::VarBinary(_)
| DataType::Variant(_)
| DataType::Blob(_)
@@ -486,6 +483,90 @@ fn format_partition_value(
Ok(value)
}
+/// Java's shortest-decimal string contract (JDK 19+), including its notation
+/// thresholds and closest choice among one- and two-digit decimals. Rust's
+/// shortest Display chooses different digits at ties; Ryū uses ties-to-even.
+fn format_java_float<T: ryu::Float + std::fmt::LowerExp + Copy +
Into<f64>>(value: T) -> String {
+ let number = value.into();
+ if number.is_nan() {
+ return "NaN".into();
+ }
+ if number.is_infinite() {
+ return if number.is_sign_negative() {
+ "-Infinity"
+ } else {
+ "Infinity"
+ }
+ .into();
+ }
+ if number == 0.0 {
+ return if number.is_sign_negative() {
+ "-0.0"
+ } else {
+ "0.0"
+ }
+ .into();
+ }
+
+ let mut buffer = ryu::Buffer::new();
+ let (mut negative, mut digits, mut exponent) =
decimal_parts(buffer.format_finite(value));
+ if digits.len() == 1 {
+ // Padding the shortest digit with zero would miss the closer two-digit
+ // form of small subnormals, e.g. Double.MIN_VALUE -> 4.9E-324.
+ (negative, digits, exponent) = decimal_parts(&format!("{value:.1e}"));
+ }
+ let mut result = if negative {
+ "-".to_string()
+ } else {
+ String::new()
+ };
+ if (-3..7).contains(&exponent) {
+ let point = exponent + 1;
+ if point <= 0 {
+ result.push_str("0.");
+ result.push_str(&"0".repeat((-point) as usize));
+ result.push_str(&digits);
+ } else if point as usize >= digits.len() {
+ result.push_str(&digits);
+ result.push_str(&"0".repeat(point as usize - digits.len()));
+ result.push_str(".0");
+ } else {
+ result.push_str(&digits[..point as usize]);
+ result.push('.');
+ result.push_str(&digits[point as usize..]);
+ }
+ } else {
+ result.push_str(&digits[..1]);
+ result.push('.');
+ result.push_str(if digits.len() > 1 { &digits[1..] } else { "0" });
+ result.push('E');
+ result.push_str(&exponent.to_string());
+ }
+ result
+}
+
+fn decimal_parts(text: &str) -> (bool, String, i32) {
+ let (mantissa, exponent) = text
+ .split_once('e')
+ .map_or((text, 0), |(mantissa, exponent)| {
+ (mantissa, exponent.parse::<i32>().expect("decimal exponent"))
+ });
+ let negative = mantissa.starts_with('-');
+ let mantissa = mantissa.trim_start_matches('-');
+ let point = mantissa.find('.').unwrap_or(mantissa.len()) as i32;
+ let digits = mantissa.replace('.', "");
+ let first = digits
+ .bytes()
+ .position(|byte| byte != b'0')
+ .expect("nonzero decimal");
+ let exponent = exponent + point - first as i32 - 1;
+ let mut digits = digits[first..].to_string();
+ while digits.len() > 1 && digits.ends_with('0') {
+ digits.pop();
+ }
+ (negative, digits, exponent)
+}
+
/// 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.
@@ -1514,20 +1595,57 @@ mod tests {
|b| b.write_int(0, 0),
true,
);
- // Float
- assert_single_partition_err(
- "f",
- DataType::Float(FloatType::new()),
- |b| b.write_int(0, 0),
- true,
- );
- // Double
- assert_single_partition_err(
- "d",
- DataType::Double(DoubleType::new()),
- |b| b.write_int(0, 0),
- true,
- );
+ }
+
+ #[test]
+ fn test_floating_partition_names_match_java() {
+ // Float.toString / Double.toString on JDK 21. Include decimal /
exponent
+ // boundaries and shortest-decimal ties where Rust Display differs.
+ for legacy in [false, true] {
+ for (value, expected) in [
+ (1.5_f32, "1.5"),
+ (42.0, "42.0"),
+ (0.001, "0.001"),
+ (0.0001, "1.0E-4"),
+ (1e7, "1.0E7"),
+ (f32::from_bits(3388994202), "-2096723.2"),
+ (f32::from_bits(1208999304), "147318.12"),
+ (f32::MIN_POSITIVE, "1.1754944E-38"),
+ (f32::from_bits(1), "1.4E-45"),
+ (f32::MAX, "3.4028235E38"),
+ ] {
+ assert_single_partition(
+ "f",
+ DataType::Float(FloatType::new()),
+ |b| b.write_int(0, value.to_bits() as i32),
+ &format!("f={expected}/"),
+ legacy,
+ );
+ }
+ for (value, expected) in [
+ (2.25_f64, "2.25"),
+ (-42.0, "-42.0"),
+ (0.001, "0.001"),
+ (0.0001, "1.0E-4"),
+ (1e7, "1.0E7"),
+ (1e23, "1.0E23"),
+ (
+ f64::from_bits(14053136104817660578),
+ "-8.011083811354442E14",
+ ),
+ (f64::MIN_POSITIVE, "2.2250738585072014E-308"),
+ (f64::from_bits(1), "4.9E-324"),
+ (f64::MAX, "1.7976931348623157E308"),
+ ] {
+ assert_single_partition(
+ "d",
+ DataType::Double(DoubleType::new()),
+ |b| b.write_long(0, value.to_bits() as i64),
+ &format!("d={expected}/"),
+ legacy,
+ );
+ }
+ }
}
#[test]
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 2b65486b..c508694d 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -3170,23 +3170,18 @@ impl TableCommit {
/// Generate per-partition statistics from commit entries.
///
- /// Reference: [pypaimon
FileStoreCommit._generate_partition_statistics](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/write/file_store_commit.py)
+ /// Java PartitionEntry.toPartitionStatistics uses the same computer as
paths.
fn generate_partition_statistics(
&self,
entries: &[ManifestEntry],
) -> Result<Vec<PartitionStatistics>> {
- let partition_fields = self.table.schema().partition_fields();
- let data_types: Vec<_> = partition_fields
- .iter()
- .map(|f| f.data_type().clone())
- .collect();
- let partition_keys: Vec<_> = self
- .table
- .schema()
- .partition_keys()
- .iter()
- .map(|s| s.to_string())
- .collect();
+ let options = self.table.schema().core_options();
+ let computer = PartitionComputer::new(
+ self.table.schema().partition_keys(),
+ self.table.schema().fields(),
+ options.partition_default_name(),
+ options.legacy_partition_name(),
+ )?;
let mut stats_map: HashMap<Vec<u8>, PartitionStatistics> =
HashMap::new();
@@ -3201,20 +3196,27 @@ impl TableCommit {
.map(|t| t.timestamp_millis())
.unwrap_or_else(|| current_time_millis() as i64);
- let stats =
stats_map.entry(partition_bytes.clone()).or_insert_with(|| {
- // Parse partition spec from BinaryRow
- let spec = self
- .parse_partition_spec(&partition_bytes, &partition_keys,
&data_types)
- .unwrap_or_default();
- PartitionStatistics {
- spec,
- record_count: 0,
- file_size_in_bytes: 0,
- file_count: 0,
- last_file_creation_time: 0,
- total_buckets: entry.total_buckets(),
+ let stats = match stats_map.entry(partition_bytes) {
+ std::collections::hash_map::Entry::Occupied(entry) =>
entry.into_mut(),
+ std::collections::hash_map::Entry::Vacant(vacant) => {
+ let spec = if
self.table.schema().partition_keys().is_empty() {
+ HashMap::new()
+ } else {
+ computer
+
.generate_part_values(&BinaryRow::from_serialized_bytes(vacant.key())?)?
+ .into_iter()
+ .collect()
+ };
+ vacant.insert(PartitionStatistics {
+ spec,
+ record_count: 0,
+ file_size_in_bytes: 0,
+ file_count: 0,
+ last_file_creation_time: 0,
+ total_buckets: entry.total_buckets(),
+ })
}
- });
+ };
stats.record_count += sign * file.row_count;
stats.file_size_in_bytes += sign * file.file_size;
@@ -3227,28 +3229,6 @@ impl TableCommit {
Ok(stats_map.into_values().collect())
}
- /// Parse partition BinaryRow bytes into a HashMap<String, String>.
- fn parse_partition_spec(
- &self,
- partition_bytes: &[u8],
- partition_keys: &[String],
- data_types: &[crate::spec::DataType],
- ) -> Result<HashMap<String, String>> {
- let mut spec = HashMap::new();
- if partition_bytes.is_empty() || partition_keys.is_empty() {
- return Ok(spec);
- }
- let row = BinaryRow::from_serialized_bytes(partition_bytes)?;
- for (i, key) in partition_keys.iter().enumerate() {
- let value = match extract_datum(&row, i, &data_types[i])? {
- Some(datum) => datum.to_string(),
- None => self.partition_default_name.clone(),
- };
- spec.insert(key.clone(), value);
- }
- Ok(spec)
- }
-
/// Check conflicts from the earliest writer snapshot, validating row-id
baselines.
fn check_from_snapshot(messages: &[CommitMessage]) -> Result<Option<i64>> {
let mut check_from_snapshot: Option<i64> = None;
@@ -3973,6 +3953,94 @@ mod tests {
TableCommit::new(table, "test-user".to_string())
}
+ #[test]
+ fn partition_statistics_use_java_partition_values() {
+ use crate::spec::{DateType, DecimalType, Schema, VarBinaryType,
VarCharType};
+
+ for legacy in [false, true] {
+ let schema = Schema::builder()
+ .column("text", DataType::VarChar(VarCharType::string_type()))
+ .column("day", DataType::Date(DateType::new()))
+ .column("amount", DataType::Decimal(DecimalType::new(8,
2).unwrap()))
+ .partition_keys(["text", "day", "amount"])
+ .option("partition.legacy-name", legacy.to_string())
+ .build()
+ .unwrap();
+ let table = Table::new(
+ test_file_io(),
+ Identifier::new("default", "stats"),
+ "memory:/partition-stat-values".into(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let commit = TableCommit::new(table, "user".into());
+ let mut row = BinaryRowBuilder::new(3);
+ row.write_string(0, "a/b");
+ row.write_int(1, 1);
+ row.write_long(2, 1234);
+ let entries = vec![ManifestEntry::new(
+ FileKind::Add,
+ row.build().to_serialized_bytes(),
+ 0,
+ 1,
+ test_data_file("data.parquet", 5),
+ 2,
+ )];
+ let stats =
commit.generate_partition_statistics(&entries).unwrap();
+ assert_eq!(
+ stats[0].spec,
+ HashMap::from([
+ ("text".into(), "a/b".into()),
+ ("day".into(), if legacy { "1" } else { "1970-01-02"
}.into()),
+ ("amount".into(), "12.34".into()),
+ ])
+ );
+ assert_eq!(stats[0].record_count, 5);
+ }
+
+ let schema = Schema::builder()
+ .column(
+ "bytes",
+ DataType::VarBinary(VarBinaryType::new(20).unwrap()),
+ )
+ .partition_keys(["bytes"])
+ .option("partition.legacy-name", "false")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ test_file_io(),
+ Identifier::new("default", "binary"),
+ "memory:/binary-partition-stats".into(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let commit = TableCommit::new(table, "user".into());
+ for (value, expected) in [
+ (Some(b"a/b".as_slice()), "a/b"),
+ (Some(b" \t".as_slice()), "__DEFAULT_PARTITION__"),
+ (None, "__DEFAULT_PARTITION__"),
+ ] {
+ let mut row = BinaryRowBuilder::new(1);
+ if let Some(value) = value {
+ row.write_binary(0, value);
+ } else {
+ row.set_null_at(0);
+ }
+ let entries = vec![ManifestEntry::new(
+ FileKind::Add,
+ row.build().to_serialized_bytes(),
+ 0,
+ 1,
+ test_data_file("data.parquet", 5),
+ 2,
+ )];
+ assert_eq!(
+
commit.generate_partition_statistics(&entries).unwrap()[0].spec["bytes"],
+ expected
+ );
+ }
+ }
+
fn partition_filter_for(commit: &TableCommit, partitions: Vec<Vec<u8>>) ->
PartitionFilter {
PartitionFilter::from_partition_set(
partitions.into_iter().collect(),
