[core] Support filesystem partitions in SystemRead

crates/integrations/datafusion/src/system_tables/partitions.rs

@@ -22,7 +22,8 @@ use std::sync::{Arc, OnceLock};
 
 use async_trait::async_trait;
 use datafusion::arrow::array::{
-    BooleanArray, Int32Array, Int64Array, RecordBatch, StringArray, TimestampMillisecondArray,
+    ArrayRef, BooleanArray, Int32Array, Int64Array, RecordBatch, StringArray,
+    TimestampMillisecondArray,
 };
 use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
 use datafusion::catalog::Session;
@@ -32,7 +33,7 @@ use datafusion::error::{DataFusionError, Result as DFResult};
 use datafusion::logical_expr::Expr;
 use datafusion::physical_plan::ExecutionPlan;
 use paimon::catalog::{list_partitions_from_file_system, Catalog, Identifier};
-use paimon::spec::{CoreOptions, Partition};
+use paimon::spec::{CoreOptions, LocalZonedTimestampType, Partition, TimestampType};
 use paimon::table::Table;
 
 use crate::error::to_datafusion_error;
@@ -216,8 +217,8 @@ impl TableProvider for PartitionsTable {
                 Arc::new(Int64Array::from(record_counts)),
                 Arc::new(Int64Array::from(file_sizes)),
                 Arc::new(Int64Array::from(file_counts)),
-                Arc::new(TimestampMillisecondArray::from(last_update_times)),
-                Arc::new(TimestampMillisecondArray::from(created_ats)),
+                local_timestamp_column(last_update_times)?,
+                local_timestamp_column(created_ats)?,
                 Arc::new(StringArray::from(created_bys)),
                 Arc::new(StringArray::from(updated_bys)),
                 Arc::new(StringArray::from(options_jsons)),
@@ -234,6 +235,19 @@ impl TableProvider for PartitionsTable {
     }
 }
 
+fn local_timestamp_column(values: Vec<Option<i64>>) -> DFResult<ArrayRef> {
+    // Partition metadata stores epoch instants. Java PartitionsTable.toTimestamp
+    // converts these to the system-local wall clock for its TIMESTAMP columns.
+    let source = paimon::spec::DataType::LocalZonedTimestamp(
+        LocalZonedTimestampType::new(3).map_err(to_datafusion_error)?,
+    );
+    let target =
+        paimon::spec::DataType::Timestamp(TimestampType::new(3).map_err(to_datafusion_error)?);
+    let values: ArrayRef = Arc::new(TimestampMillisecondArray::from(values));
+    paimon::arrow::cast_array_for_schema_evolution(&values, &source, &target)
+        .map_err(to_datafusion_error)
+}
+
 /// Format `spec` as `key1=val1/key2=val2` in `partition_keys` order. Empty
 /// string for non-partitioned tables. NULL spec values fall back to
 /// `default_partition_name`.

crates/integrations/datafusion/tests/sql_context_tests.rs

@@ -629,7 +629,8 @@ async fn test_partitions_system_table_shows_unreported_statistics_as_null() {
     sql_context
         .sql(
             "CREATE TABLE paimon.default.unknown_stats_orders \
-             (id INT, name STRING) PARTITIONED BY (id)",
+             (id INT, name STRING) PARTITIONED BY (id) \
+             WITH ('bucket' = '4', 'bucket-key' = 'name')",
         )
         .await
         .unwrap()
@@ -658,6 +659,15 @@ async fn test_partitions_system_table_shows_unreported_statistics_as_null() {
             "column {column} should carry a measurement before the switch"
         );
     }
+    assert_eq!(
+        measured
+            .column(3)
+            .as_any()
+            .downcast_ref::<datafusion::arrow::array::Int32Array>()
+            .unwrap()
+            .value(0),
+        4
+    );
 
     catalog.set_unknown_statistics(true);
 

crates/integrations/datafusion/tests/system_tables.rs

@@ -1409,6 +1409,63 @@ fn single_int_partition_stat(value: &str) -> i32 {
         .expect("partition stats should contain one int partition value")
 }
 
+#[tokio::test]
+async fn test_partition_timestamps_match_local_file_creation_times() {
+    let (ctx, catalog, _tmp) = create_context().await;
+    let table = catalog
+        .get_table(&Identifier::new("default", FIXTURE_TABLE))
+        .await
+        .unwrap();
+    let core = table
+        .new_system_read("partitions")
+        .unwrap()
+        .read()
+        .await
+        .unwrap();
+    let sql = format!("SELECT last_update_time FROM paimon.default.{FIXTURE_TABLE}$partitions");
+    let batches = run_sql(&ctx, &sql).await;
+    let actual = batches
+        .iter()
+        .flat_map(|batch| {
+            batch
+                .column(0)
+                .as_any()
+                .downcast_ref::<TimestampMillisecondArray>()
+                .unwrap()
+                .iter()
+        })
+        .collect::<Vec<_>>();
+    let expected = core
+        .column_by_name("last_update_time")
+        .unwrap()
+        .as_any()
+        .downcast_ref::<TimestampMillisecondArray>()
+        .unwrap()
+        .iter()
+        .collect::<Vec<_>>();
+    assert!(!expected.is_empty());
+    assert_eq!(actual, expected);
+    let files = run_sql(
+        &ctx,
+        &format!("SELECT creation_time FROM paimon.default.{FIXTURE_TABLE}$files"),
+    )
+    .await;
+    let last_file = files
+        .iter()
+        .flat_map(|batch| {
+            batch
+                .column(0)
+                .as_any()
+                .downcast_ref::<TimestampMillisecondArray>()
+                .unwrap()
+                .iter()
+                .flatten()
+        })
+        .max()
+        .unwrap();
+    assert_eq!(actual, vec![Some(last_file)]);
+}
+
 #[tokio::test]
 async fn test_partitions_system_table() {
     let (ctx, _catalog, _tmp) = create_context().await;
@@ -1790,6 +1847,7 @@ async fn test_resolved_system_table_reader_preserves_guards_and_schema() {
         "consumers",
         "file_key_ranges",
         "table_indexes",
+        "partitions",
     ] {
         let stream = read_metadata(table.clone(), name).await.unwrap();
         let schema = table.new_system_read(name).unwrap().schema();
@@ -1800,13 +1858,11 @@ async fn test_resolved_system_table_reader_preserves_guards_and_schema() {
             "{name}"
         );
     }
-    for name in ["partitions", "unknown"] {
-        let error = match read_metadata(table.clone(), name).await {
-            Ok(_) => panic!("{name} must require a catalog or be unknown"),
-            Err(error) => error.to_string(),
-        };
-        assert!(error.contains("requires a catalog or is unknown"));
-    }
+    let error = match read_metadata(table.clone(), "unknown").await {
+        Ok(_) => panic!("Unknown system table must be rejected"),
+        Err(error) => error.to_string(),
+    };
+    assert!(error.contains("requires a catalog or is unknown"));
     let denied = table.copy_with_options(std::collections::HashMap::from([(
         "query-auth.enabled".to_string(),
         "true".to_string(),

crates/paimon/src/arrow/cast_text.rs

@@ -139,7 +139,7 @@ fn time_components(text: &str, space: bool) -> Option<[Option<i32>; 4]> {
     Some(time)
 }
 
-pub(super) fn parse_timestamp(
+pub(crate) fn parse_timestamp(
     text: &str,
     precision: u32,
     zone: Option<&jiff::tz::TimeZone>,

crates/paimon/src/arrow/evolution_string.rs

@@ -23,7 +23,7 @@ use arrow_array::{cast::AsArray, types::Int64Type, ArrayRef, StringArray, Time32
 use arrow_schema::{DataType as ArrowType, TimeUnit};
 use jiff::tz::TimeZone;
 
-use crate::common::date_time::{format_epoch_timestamp, format_time, local_offset};
+use crate::common::date_time::{format_time, format_timestamp_millis};
 use crate::spec::DataType;
 
 /// Return None when this is not a temporal-to-character conversion.
@@ -112,17 +112,7 @@ fn render_timestamp(
         TimeUnit::Nanosecond => value.rem_euclid(1_000_000),
         _ => 0,
     };
-    let millis = match zone {
-        Some(zone) => {
-            let offset = local_offset(millis.div_euclid(1000), zone);
-            // Timestamp.fromLocalDateTime uses signed long millisecond storage.
-            // Applying an offset can wrap and change the millisecond fraction.
-            millis.wrapping_add(i64::from(offset) * 1000)
-        }
-        None => millis,
-    };
-    let nanos = (millis.rem_euclid(1000) * 1_000_000 + sub_milli) as u32;
-    format_epoch_timestamp(millis.div_euclid(1000), nanos, precision)
+    format_timestamp_millis(millis, sub_milli, precision, zone)
 }
 
 fn invalid(message: impl Into<String>) -> crate::Error {

crates/paimon/src/arrow/mod.rs

@@ -24,6 +24,7 @@ pub(crate) use cast_executor::CastExecutor;
 #[cfg(test)]
 mod cast_completion_java_test;
 mod cast_text;
+pub(crate) use cast_text::parse_timestamp as parse_java_timestamp;
 mod evolution_cast;
 pub(crate) use evolution_cast::{cast_numeric, supports_numeric_cast};
 mod evolution_decimal;

crates/paimon/src/catalog/mod.rs

@@ -36,7 +36,10 @@ pub use database::*;
 pub use factory::*;
 pub use filesystem::*;
 pub use function::*;
+#[cfg(test)]
+pub(crate) use partition_listing::aggregate_partition_entries;
 pub use partition_listing::list_partitions_from_file_system;
+pub(crate) use partition_listing::{list_partition_entries, FileSystemPartitionEntry};
 pub use rest::*;
 use serde::{Deserialize, Serialize};
 pub use view::*;