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, ) -> DFResult> { 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 = 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> = Vec::with_capacity(n); + let mut create_times: Vec> = Vec::with_capacity(n); + let mut time_retained: Vec> = 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, Option)> = Vec::new(); + for batch in &batches { + let names = batch + .column(0) + .as_any() + .downcast_ref::() + .expect("tag_name is Utf8"); + let created = batch + .column(1) + .as_any() + .downcast_ref::() + .expect("create_time is Timestamp(ms)"); + let retained = batch + .column(2) + .as_any() + .downcast_ref::() + .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)>> { + 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, Option)>> { + 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> { @@ -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 { + 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 eaf9758a..4c6e4a25 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -1756,8 +1756,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