fix(datafusion): surface tag create-time and retention in $tags

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,23 +84,28 @@ 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);
         let mut snapshot_ids = Vec::with_capacity(n);
         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");
+    }
+}

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;