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 59bc642a fix(datafusion): surface tag create-time and retention in
$tags (#728)
59bc642a is described below
commit 59bc642a5a870d3aadc216745a88c5cded4e8b16
Author: jackylee <[email protected]>
AuthorDate: Thu Aug 20 10:53:19 2026 +0800
fix(datafusion): surface tag create-time and retention in $tags (#728)
---
.../datafusion/src/system_tables/tags.rs | 81 ++++++-
.../integrations/datafusion/tests/system_tables.rs | 82 ++++++-
crates/paimon/src/table/tag_manager.rs | 245 +++++++++++++++++++++
docs/src/sql.md | 4 +-
4 files changed, 400 insertions(+), 12 deletions(-)
diff --git a/crates/integrations/datafusion/src/system_tables/tags.rs
b/crates/integrations/datafusion/src/system_tables/tags.rs
index 5b9a22b4..09af4231 100644
--- a/crates/integrations/datafusion/src/system_tables/tags.rs
+++ b/crates/integrations/datafusion/src/system_tables/tags.rs
@@ -20,9 +20,7 @@
use std::sync::{Arc, OnceLock};
use async_trait::async_trait;
-use datafusion::arrow::array::{
- new_null_array, Int64Array, RecordBatch, StringArray,
TimestampMillisecondArray,
-};
+use datafusion::arrow::array::{Int64Array, RecordBatch, StringArray,
TimestampMillisecondArray};
use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef,
TimeUnit};
use datafusion::catalog::Session;
use datafusion::datasource::memory::MemorySourceConfig;
@@ -86,9 +84,10 @@ impl TableProvider for TagsTable {
_limit: Option<usize>,
) -> DFResult<Arc<dyn ExecutionPlan>> {
let tm = self.table.tag_manager();
- let tags = crate::runtime::await_with_runtime(async move {
tm.list_all().await })
- .await
- .map_err(to_datafusion_error)?;
+ let tags =
+ crate::runtime::await_with_runtime(async move {
tm.list_all_with_metadata().await })
+ .await
+ .map_err(to_datafusion_error)?;
let n = tags.len();
let mut tag_names: Vec<String> = Vec::with_capacity(n);
@@ -96,13 +95,17 @@ impl TableProvider for TagsTable {
let mut schema_ids = Vec::with_capacity(n);
let mut commit_times = Vec::with_capacity(n);
let mut record_counts: Vec<Option<i64>> = Vec::with_capacity(n);
+ let mut create_times: Vec<Option<i64>> = Vec::with_capacity(n);
+ let mut time_retained: Vec<Option<String>> = Vec::with_capacity(n);
- for (name, snap) in tags {
+ for (name, snap, created, retained) in tags {
tag_names.push(name);
snapshot_ids.push(snap.id());
schema_ids.push(snap.schema_id());
commit_times.push(snap.time_millis() as i64);
record_counts.push(snap.total_record_count());
+ create_times.push(created);
+ time_retained.push(retained.map(format_duration_iso8601));
}
let schema = tags_schema();
@@ -114,8 +117,8 @@ impl TableProvider for TagsTable {
Arc::new(Int64Array::from(schema_ids)),
Arc::new(TimestampMillisecondArray::from(commit_times)),
Arc::new(Int64Array::from(record_counts)),
- new_null_array(&DataType::Timestamp(TimeUnit::Millisecond,
None), n),
- new_null_array(&DataType::Utf8, n),
+ Arc::new(TimestampMillisecondArray::from(create_times)),
+ Arc::new(StringArray::from(time_retained)),
],
)?;
@@ -126,3 +129,63 @@ impl TableProvider for TagsTable {
)?)
}
}
+
+/// Render a retention in seconds the way Java's `Duration.toString()` does, so
+/// the column reads the same across engines: `PT72H`, `PT1M30S`, `PT0.5S`.
+fn format_duration_iso8601(total_seconds: f64) -> String {
+ if !total_seconds.is_finite() {
+ return "PT0S".to_string();
+ }
+ let negative = total_seconds < 0.0;
+ let magnitude = total_seconds.abs();
+ let whole = magnitude.trunc() as i64;
+ let fraction = magnitude - whole as f64;
+
+ let hours = whole / 3600;
+ let minutes = (whole % 3600) / 60;
+ let seconds = whole % 60;
+
+ let mut out = String::from(if negative { "-PT" } else { "PT" });
+ if hours != 0 {
+ out.push_str(&format!("{hours}H"));
+ }
+ if minutes != 0 {
+ out.push_str(&format!("{minutes}M"));
+ }
+ if seconds != 0 || fraction != 0.0 || (hours == 0 && minutes == 0) {
+ if fraction == 0.0 {
+ out.push_str(&format!("{seconds}S"));
+ } else {
+ // Java prints up to nanosecond precision with trailing zeros
trimmed.
+ let rendered = format!("{:.9}", seconds as f64 + fraction);
+ out.push_str(rendered.trim_end_matches('0').trim_end_matches('.'));
+ out.push('S');
+ }
+ }
+ out
+}
+
+#[cfg(test)]
+mod tests {
+ use super::format_duration_iso8601;
+
+ #[test]
+ fn test_format_duration_iso8601_matches_java() {
+ // Whole units, mirroring java.time.Duration.toString().
+ assert_eq!(format_duration_iso8601(259_200.0), "PT72H");
+ assert_eq!(format_duration_iso8601(3600.0), "PT1H");
+ assert_eq!(format_duration_iso8601(90.0), "PT1M30S");
+ assert_eq!(format_duration_iso8601(60.0), "PT1M");
+ assert_eq!(format_duration_iso8601(1.0), "PT1S");
+ // Zero keeps the seconds component so the string is never bare "PT".
+ assert_eq!(format_duration_iso8601(0.0), "PT0S");
+ // Sub-second retention: trailing zeros trimmed.
+ assert_eq!(format_duration_iso8601(0.5), "PT0.5S");
+ assert_eq!(format_duration_iso8601(1.25), "PT1.25S");
+ // Mixed with larger units.
+ assert_eq!(format_duration_iso8601(3661.0), "PT1H1M1S");
+ // Not a number cannot panic or produce a bogus unit.
+ assert_eq!(format_duration_iso8601(f64::NAN), "PT0S");
+ assert_eq!(format_duration_iso8601(f64::INFINITY), "PT0S");
+ }
+}
diff --git a/crates/integrations/datafusion/tests/system_tables.rs
b/crates/integrations/datafusion/tests/system_tables.rs
index ca32da12..267f3c97 100644
--- a/crates/integrations/datafusion/tests/system_tables.rs
+++ b/crates/integrations/datafusion/tests/system_tables.rs
@@ -22,7 +22,7 @@ mod common;
use std::sync::Arc;
use datafusion::arrow::array::{
- Array, BooleanArray, Int32Array, Int64Array, ListArray, StringArray,
+ Array, BooleanArray, Int32Array, Int64Array, ListArray, StringArray,
TimestampMillisecondArray,
};
use datafusion::arrow::datatypes::{DataType, Field, TimeUnit};
use datafusion::arrow::record_batch::RecordBatch;
@@ -672,6 +672,86 @@ async fn test_tags_system_table_with_seeded_tags() {
assert_eq!(snap_ids, vec![earliest.id(), earliest.id()]);
}
+/// A tag file written by Java carries `tagCreateTime` and `tagTimeRetained`
+/// alongside the snapshot fields; both columns must surface those values
instead
+/// of NULL. The retention is rendered as ISO-8601, matching Java's
+/// `Duration.toString()`.
+#[tokio::test]
+async fn test_tags_system_table_surfaces_create_time_and_retention() {
+ let (ctx, catalog, tmp) = create_context().await;
+
+ let identifier = Identifier::new("default".to_string(),
FIXTURE_TABLE.to_string());
+ let table = catalog.get_table(&identifier).await.unwrap();
+ let sm =
+ paimon::table::SnapshotManager::new(table.file_io().clone(),
table.location().to_string());
+ let earliest = sm.list_all().await.unwrap().into_iter().next().unwrap();
+
+ let table_dir = tmp.path().join("default.db").join(FIXTURE_TABLE);
+ let tag_dir = table_dir.join("tag");
+ std::fs::create_dir_all(&tag_dir).expect("create tag dir");
+ let src = table_dir
+ .join("snapshot")
+ .join(format!("snapshot-{}", earliest.id()));
+ let snapshot_json = std::fs::read_to_string(&src).unwrap();
+
+ // Jackson emits LocalDateTime as an array and Duration as decimal seconds.
+ let mut with_meta: serde_json::Value =
serde_json::from_str(&snapshot_json).unwrap();
+ let map = with_meta.as_object_mut().unwrap();
+ map.insert(
+ "tagCreateTime".to_string(),
+ serde_json::json!([2024, 1, 2, 3, 4, 5]),
+ );
+ map.insert("tagTimeRetained".to_string(), serde_json::json!(259_200.0));
+ std::fs::write(
+ tag_dir.join("tag-with-meta"),
+ serde_json::to_string(&with_meta).unwrap(),
+ )
+ .unwrap();
+ // A tag without the fields keeps both columns NULL.
+ std::fs::copy(&src, tag_dir.join("tag-plain")).unwrap();
+
+ let sql = format!(
+ "SELECT tag_name, create_time, time_retained \
+ FROM paimon.default.{FIXTURE_TABLE}$tags ORDER BY tag_name"
+ );
+ let batches = run_sql(&ctx, &sql).await;
+
+ let mut rows: Vec<(String, Option<i64>, Option<String>)> = Vec::new();
+ for batch in &batches {
+ let names = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .expect("tag_name is Utf8");
+ let created = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<TimestampMillisecondArray>()
+ .expect("create_time is Timestamp(ms)");
+ let retained = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .expect("time_retained is Utf8");
+ for i in 0..batch.num_rows() {
+ rows.push((
+ names.value(i).to_string(),
+ (!created.is_null(i)).then(|| created.value(i)),
+ (!retained.is_null(i)).then(|| retained.value(i).to_string()),
+ ));
+ }
+ }
+
+ assert_eq!(rows.len(), 2);
+ assert_eq!(rows[0].0, "plain");
+ assert_eq!(rows[0].1, None, "a tag without the field stays NULL");
+ assert_eq!(rows[0].2, None);
+ assert_eq!(rows[1].0, "with-meta");
+ // 2024-01-02T03:04:05 UTC
+ assert_eq!(rows[1].1, Some(1_704_164_645_000));
+ assert_eq!(rows[1].2.as_deref(), Some("PT72H"));
+}
+
#[tokio::test]
async fn test_manifests_system_table() {
let (ctx, catalog, _tmp) = create_context().await;
diff --git a/crates/paimon/src/table/tag_manager.rs
b/crates/paimon/src/table/tag_manager.rs
index 113cf142..8bf1a384 100644
--- a/crates/paimon/src/table/tag_manager.rs
+++ b/crates/paimon/src/table/tag_manager.rs
@@ -95,6 +95,64 @@ impl TagManager {
Ok(Some(snapshot))
}
+ /// Get a tag's snapshot together with the two tag-only fields Java writes
+ /// alongside it: the creation time as epoch millis and the retention as
+ /// seconds. Returns `None` when the tag file does not exist; either
metadata
+ /// field is `None` when absent or unparsable.
+ pub async fn get_with_metadata(
+ &self,
+ tag_name: &str,
+ ) -> crate::Result<Option<(Snapshot, Option<i64>, Option<f64>)>> {
+ let path = self.tag_path(tag_name);
+ let input = self.file_io.new_input(&path)?;
+ let bytes = match input.read().await {
+ Ok(b) => b,
+ Err(crate::Error::IoUnexpected { ref source, .. })
+ if source.kind() == opendal::ErrorKind::NotFound =>
+ {
+ return Ok(None);
+ }
+ Err(e) => return Err(e),
+ };
+ let value: serde_json::Value =
+ serde_json::from_slice(&bytes).map_err(|e|
crate::Error::DataInvalid {
+ message: format!("tag '{tag_name}' JSON invalid: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ let snapshot: Snapshot =
+ serde_json::from_value(value.clone()).map_err(|e|
crate::Error::DataInvalid {
+ message: format!("tag '{tag_name}' JSON invalid: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ let create_time = value
+ .get(FIELD_TAG_CREATE_TIME)
+ .and_then(parse_tag_create_time_millis);
+ let time_retained = value
+ .get(FIELD_TAG_TIME_RETAINED)
+ .and_then(serde_json::Value::as_f64);
+ Ok(Some((snapshot, create_time, time_retained)))
+ }
+
+ /// Like [`Self::list_all`], but each row also carries the tag creation
time
+ /// in epoch millis and the retention in seconds.
+ #[allow(clippy::type_complexity)]
+ pub async fn list_all_with_metadata(
+ &self,
+ ) -> crate::Result<Vec<(String, Snapshot, Option<i64>, Option<f64>)>> {
+ let names = self.list_all_names().await?;
+ try_join_all(names.into_iter().map(|name| async move {
+ let (snap, create_time, retained) =
+ self.get_with_metadata(&name)
+ .await?
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!("tag '{name}' disappeared during
listing"),
+ source: None,
+ })?;
+ Ok::<_, crate::Error>((name, snap, create_time, retained))
+ }))
+ .await
+ }
+
/// List all tag names sorted ascending. Returns an empty vector when the
/// tag directory does not exist.
pub async fn list_all_names(&self) -> crate::Result<Vec<String>> {
@@ -155,6 +213,43 @@ impl TagManager {
}
}
+/// Java `Tag` adds these two fields on top of the snapshot schema.
+const FIELD_TAG_CREATE_TIME: &str = "tagCreateTime";
+const FIELD_TAG_TIME_RETAINED: &str = "tagTimeRetained";
+
+/// Decode a Jackson-serialized `LocalDateTime` into epoch millis, treating the
+/// wall-clock value as UTC.
+///
+/// Jackson's `LocalDateTimeSerializer` emits
+/// `[year, month, day, hour, minute, second, nanoOfSecond]` and omits trailing
+/// zero components, so the array may hold as few as five items. Anything that
is
+/// not such an array -- or that does not describe a real instant -- yields
+/// `None` so one odd tag file cannot fail the whole listing.
+fn parse_tag_create_time_millis(value: &serde_json::Value) -> Option<i64> {
+ let items = value.as_array()?;
+ if items.len() < 5 || items.len() > 7 {
+ return None;
+ }
+ let mut parts = [0i64; 7];
+ for (slot, item) in parts.iter_mut().zip(items) {
+ *slot = item.as_i64()?;
+ }
+ let [year, month, day, hour, minute, second, nano] = parts;
+
+ let date = chrono::NaiveDate::from_ymd_opt(
+ i32::try_from(year).ok()?,
+ u32::try_from(month).ok()?,
+ u32::try_from(day).ok()?,
+ )?;
+ let time = chrono::NaiveTime::from_hms_nano_opt(
+ u32::try_from(hour).ok()?,
+ u32::try_from(minute).ok()?,
+ u32::try_from(second).ok()?,
+ u32::try_from(nano).ok()?,
+ )?;
+ Some(date.and_time(time).and_utc().timestamp_millis())
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -207,6 +302,156 @@ mod tests {
assert_eq!(tm.list_all_names().await.unwrap(), vec!["v1", "v2", "v3"]);
}
+ /// Write a raw tag JSON built from a snapshot plus the two Java-only
+ /// fields, so the on-disk shape matches what Flink/Spark produce.
+ async fn write_tag_json(
+ file_io: &FileIO,
+ tm: &TagManager,
+ name: &str,
+ snapshot: &Snapshot,
+ extra: &[(&str, serde_json::Value)],
+ ) {
+ let mut value = serde_json::to_value(snapshot).unwrap();
+ let map = value.as_object_mut().unwrap();
+ for (key, v) in extra {
+ map.insert((*key).to_string(), v.clone());
+ }
+ let output = file_io.new_output(&tm.tag_path(name)).unwrap();
+ output
+ .write(Bytes::from(serde_json::to_vec(&value).unwrap()))
+ .await
+ .unwrap();
+ }
+
+ #[tokio::test]
+ async fn test_get_with_metadata_reads_java_fields() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_tag_meta".to_string();
+ file_io.mkdirs(&format!("{table_path}/tag/")).await.unwrap();
+ let tm = TagManager::new(file_io.clone(), table_path);
+
+ // Jackson writes LocalDateTime as [y, mo, d, h, mi, s, nano] and omits
+ // trailing zero components; Duration is decimal seconds.
+ write_tag_json(
+ &file_io,
+ &tm,
+ "full",
+ &test_snapshot(1),
+ &[
+ (
+ "tagCreateTime",
+ serde_json::json!([2024, 1, 2, 3, 4, 5, 123_000_000]),
+ ),
+ ("tagTimeRetained", serde_json::json!(259_200.0)),
+ ],
+ )
+ .await;
+
+ let (snap, create_time, retained) =
tm.get_with_metadata("full").await.unwrap().unwrap();
+ assert_eq!(snap.id(), 1);
+ // 2024-01-02T03:04:05.123 UTC
+ assert_eq!(create_time, Some(1_704_164_645_123));
+ assert_eq!(retained, Some(259_200.0));
+ }
+
+ #[tokio::test]
+ async fn test_get_with_metadata_pads_truncated_time_array() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_tag_meta_short".to_string();
+ file_io.mkdirs(&format!("{table_path}/tag/")).await.unwrap();
+ let tm = TagManager::new(file_io.clone(), table_path);
+
+ // Jackson drops the trailing zero second and nano, leaving five items.
+ write_tag_json(
+ &file_io,
+ &tm,
+ "short",
+ &test_snapshot(2),
+ &[("tagCreateTime", serde_json::json!([2024, 1, 2, 3, 4]))],
+ )
+ .await;
+
+ let (_, create_time, retained) =
tm.get_with_metadata("short").await.unwrap().unwrap();
+ // 2024-01-02T03:04:00 UTC
+ assert_eq!(create_time, Some(1_704_164_640_000));
+ assert_eq!(retained, None, "absent retention stays absent");
+ }
+
+ #[tokio::test]
+ async fn test_get_with_metadata_absent_fields_are_none() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_tag_meta_absent".to_string();
+ file_io.mkdirs(&format!("{table_path}/tag/")).await.unwrap();
+ let tm = TagManager::new(file_io.clone(), table_path);
+ write_tag(&file_io, &tm, "plain", &test_snapshot(3)).await;
+
+ let (snap, create_time, retained) =
tm.get_with_metadata("plain").await.unwrap().unwrap();
+ assert_eq!(snap.id(), 3);
+ assert_eq!(create_time, None);
+ assert_eq!(retained, None);
+ }
+
+ /// A malformed shape must not fail the whole read: the snapshot still
loads
+ /// and the unparsable field is reported as absent.
+ #[tokio::test]
+ async fn test_get_with_metadata_tolerates_bad_shapes() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_tag_meta_bad".to_string();
+ file_io.mkdirs(&format!("{table_path}/tag/")).await.unwrap();
+ let tm = TagManager::new(file_io.clone(), table_path);
+
+ for (name, extra) in [
+ (
+ "iso_string",
+ vec![("tagCreateTime",
serde_json::json!("2024-01-02T03:04:05"))],
+ ),
+ (
+ "too_short",
+ vec![("tagCreateTime", serde_json::json!([2024, 1]))],
+ ),
+ (
+ "impossible_date",
+ vec![("tagCreateTime", serde_json::json!([2024, 13, 40, 3,
4]))],
+ ),
+ (
+ "retained_string",
+ vec![("tagTimeRetained", serde_json::json!("PT72H"))],
+ ),
+ ] {
+ write_tag_json(&file_io, &tm, name, &test_snapshot(4),
&extra).await;
+ let (snap, create_time, retained) =
tm.get_with_metadata(name).await.unwrap().unwrap();
+ assert_eq!(snap.id(), 4, "{name}: snapshot must still load");
+ assert!(
+ create_time.is_none() && retained.is_none(),
+ "{name}: unparsable metadata must be reported as absent"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn test_list_all_with_metadata_keeps_order() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_tag_meta_list".to_string();
+ file_io.mkdirs(&format!("{table_path}/tag/")).await.unwrap();
+ let tm = TagManager::new(file_io.clone(), table_path);
+
+ write_tag_json(
+ &file_io,
+ &tm,
+ "a",
+ &test_snapshot(1),
+ &[("tagTimeRetained", serde_json::json!(60.0))],
+ )
+ .await;
+ write_tag(&file_io, &tm, "b", &test_snapshot(2)).await;
+
+ let rows = tm.list_all_with_metadata().await.unwrap();
+ let names: Vec<&str> = rows.iter().map(|(n, _, _, _)|
n.as_str()).collect();
+ assert_eq!(names, vec!["a", "b"]);
+ assert_eq!(rows[0].3, Some(60.0));
+ assert_eq!(rows[1].3, None);
+ }
+
#[tokio::test]
async fn test_list_all_loads_pairs() {
let file_io = test_file_io();
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 230cd067..34a033e9 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1870,8 +1870,8 @@ Columns:
| `schema_id` | BIGINT | Schema ID |
| `commit_time` | TIMESTAMP | Commit time |
| `record_count` | BIGINT | Record count |
-| `create_time` | TIMESTAMP | Always `NULL`: the Rust snapshot does not carry
a tag creation time |
-| `time_retained` | STRING | Always `NULL`: the Rust snapshot does not carry a
tag retention |
+| `create_time` | TIMESTAMP | Tag creation time; `NULL` for tags written
without one |
+| `time_retained` | STRING | Tag retention as an ISO-8601 duration (for
example `PT72H`); `NULL` for tags written without one |
### $branches
