feat(datafusion): Add $manifests system table (#287)

plusplusjiajia · web-flow · commit a5224ea410b9 · 2026-04-26T11:45:14.000+08:00
diff --git a/crates/integrations/datafusion/src/system_tables/manifests.rs b/crates/integrations/datafusion/src/system_tables/manifests.rs
@@ -0,0 +1,159 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Mirrors Java [ManifestsTable](https://github.com/apache/paimon/blob/release-1.4/paimon-core/src/main/java/org/apache/paimon/table/system/ManifestsTable.java).
+
+use std::any::Any;
+use std::sync::{Arc, OnceLock};
+
+use async_trait::async_trait;
+use datafusion::arrow::array::{new_null_array, Int64Array, RecordBatch, StringArray};
+use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use datafusion::catalog::Session;
+use datafusion::datasource::memory::MemorySourceConfig;
+use datafusion::datasource::{TableProvider, TableType};
+use datafusion::error::Result as DFResult;
+use datafusion::logical_expr::Expr;
+use datafusion::physical_plan::ExecutionPlan;
+use paimon::spec::{ManifestFileMeta, ManifestList};
+use paimon::table::{SnapshotManager, Table};
+
+use crate::error::to_datafusion_error;
+
+pub(super) fn build(table: Table) -> DFResult<Arc<dyn TableProvider>> {
+    Ok(Arc::new(ManifestsTable { table }))
+}
+
+fn manifests_schema() -> SchemaRef {
+    static SCHEMA: OnceLock<SchemaRef> = OnceLock::new();
+    SCHEMA
+        .get_or_init(|| {
+            Arc::new(Schema::new(vec![
+                Field::new("file_name", DataType::Utf8, false),
+                Field::new("file_size", DataType::Int64, false),
+                Field::new("num_added_files", DataType::Int64, false),
+                Field::new("num_deleted_files", DataType::Int64, false),
+                Field::new("schema_id", DataType::Int64, false),
+                Field::new("min_partition_stats", DataType::Utf8, true),
+                Field::new("max_partition_stats", DataType::Utf8, true),
+                Field::new("min_row_id", DataType::Int64, true),
+                Field::new("max_row_id", DataType::Int64, true),
+            ]))
+        })
+        .clone()
+}
+
+#[derive(Debug)]
+struct ManifestsTable {
+    table: Table,
+}
+
+#[async_trait]
+impl TableProvider for ManifestsTable {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn schema(&self) -> SchemaRef {
+        manifests_schema()
+    }
+
+    fn table_type(&self) -> TableType {
+        TableType::View
+    }
+
+    async fn scan(
+        &self,
+        _state: &dyn Session,
+        projection: Option<&Vec<usize>>,
+        _filters: &[Expr],
+        _limit: Option<usize>,
+    ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        let metas = collect_manifests(&self.table)
+            .await
+            .map_err(to_datafusion_error)?;
+
+        let n = metas.len();
+        let mut file_names: Vec<String> = Vec::with_capacity(n);
+        let mut file_sizes = Vec::with_capacity(n);
+        let mut num_added = Vec::with_capacity(n);
+        let mut num_deleted = Vec::with_capacity(n);
+        let mut schema_ids = Vec::with_capacity(n);
+        let mut min_row_ids: Vec<Option<i64>> = Vec::with_capacity(n);
+        let mut max_row_ids: Vec<Option<i64>> = Vec::with_capacity(n);
+
+        for meta in metas {
+            file_names.push(meta.file_name().to_string());
+            file_sizes.push(meta.file_size());
+            num_added.push(meta.num_added_files());
+            num_deleted.push(meta.num_deleted_files());
+            schema_ids.push(meta.schema_id());
+            min_row_ids.push(meta.min_row_id());
+            max_row_ids.push(meta.max_row_id());
+        }
+
+        let schema = manifests_schema();
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(StringArray::from(file_names)),
+                Arc::new(Int64Array::from(file_sizes)),
+                Arc::new(Int64Array::from(num_added)),
+                Arc::new(Int64Array::from(num_deleted)),
+                Arc::new(Int64Array::from(schema_ids)),
+                new_null_array(&DataType::Utf8, n),
+                new_null_array(&DataType::Utf8, n),
+                Arc::new(Int64Array::from(min_row_ids)),
+                Arc::new(Int64Array::from(max_row_ids)),
+            ],
+        )?;
+
+        Ok(MemorySourceConfig::try_new_exec(
+            &[vec![batch]],
+            schema,
+            projection.cloned(),
+        )?)
+    }
+}
+
+async fn collect_manifests(table: &Table) -> paimon::Result<Vec<ManifestFileMeta>> {
+    let file_io = table.file_io();
+    let sm = SnapshotManager::new(file_io.clone(), table.location().to_string());
+    let snapshot = match sm.get_latest_snapshot().await? {
+        Some(s) => s,
+        None => return Ok(Vec::new()),
+    };
+
+    let base_path = sm.manifest_path(snapshot.base_manifest_list());
+    let delta_path = sm.manifest_path(snapshot.delta_manifest_list());
+    let changelog_path = snapshot
+        .changelog_manifest_list()
+        .map(|c| sm.manifest_path(c));
+    let base_fut = ManifestList::read(file_io, &base_path);
+    let delta_fut = ManifestList::read(file_io, &delta_path);
+    let changelog_fut = async {
+        match &changelog_path {
+            Some(p) => ManifestList::read(file_io, p).await,
+            None => Ok(Vec::new()),
+        }
+    };
+    let (base, delta, changelog) = futures::try_join!(base_fut, delta_fut, changelog_fut)?;
+    let mut metas = base;
+    metas.extend(delta);
+    metas.extend(changelog);
+    Ok(metas)
+}
diff --git a/crates/integrations/datafusion/src/system_tables/mod.rs b/crates/integrations/datafusion/src/system_tables/mod.rs
@@ -29,6 +29,7 @@ use paimon::table::Table;
 
 use crate::error::to_datafusion_error;
 
+mod manifests;
 mod options;
 mod schemas;
 mod snapshots;
@@ -37,6 +38,7 @@ mod tags;
 type Builder = fn(Table) -> DFResult<Arc<dyn TableProvider>>;
 
 const TABLES: &[(&str, Builder)] = &[
+    ("manifests", manifests::build),
     ("options", options::build),
     ("schemas", schemas::build),
     ("snapshots", snapshots::build),
@@ -131,6 +133,9 @@ mod tests {
         assert!(is_registered("tags"));
         assert!(is_registered("Tags"));
         assert!(is_registered("TAGS"));
+        assert!(is_registered("manifests"));
+        assert!(is_registered("Manifests"));
+        assert!(is_registered("MANIFESTS"));
         assert!(!is_registered("nonsense"));
     }
 
diff --git a/crates/integrations/datafusion/tests/system_tables.rs b/crates/integrations/datafusion/tests/system_tables.rs
@@ -410,3 +410,83 @@ async fn test_tags_system_table_with_seeded_tags() {
     assert_eq!(names, vec!["v1".to_string(), "v2".to_string()]);
     assert_eq!(snap_ids, vec![earliest.id(), earliest.id()]);
 }
+
+#[tokio::test]
+async fn test_manifests_system_table() {
+    let (ctx, catalog, _tmp) = create_context().await;
+    let sql = format!("SELECT * FROM paimon.default.{FIXTURE_TABLE}$manifests");
+    let batches = run_sql(&ctx, &sql).await;
+
+    assert!(!batches.is_empty(), "$manifests should return ≥1 batch");
+    let arrow_schema = batches[0].schema();
+    let expected_columns = [
+        ("file_name", DataType::Utf8),
+        ("file_size", DataType::Int64),
+        ("num_added_files", DataType::Int64),
+        ("num_deleted_files", DataType::Int64),
+        ("schema_id", DataType::Int64),
+        ("min_partition_stats", DataType::Utf8),
+        ("max_partition_stats", DataType::Utf8),
+        ("min_row_id", DataType::Int64),
+        ("max_row_id", DataType::Int64),
+    ];
+    for (i, (name, dtype)) in expected_columns.iter().enumerate() {
+        let field = arrow_schema.field(i);
+        assert_eq!(field.name(), name, "column {i} name");
+        assert_eq!(field.data_type(), dtype, "column {i} type");
+    }
+
+    let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
+    assert!(total_rows > 0, "fixture should have at least one manifest");
+
+    for batch in &batches {
+        let names = batch
+            .column(0)
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .expect("file_name is Utf8");
+        let sizes = batch
+            .column(1)
+            .as_any()
+            .downcast_ref::<Int64Array>()
+            .expect("file_size is Int64");
+        for i in 0..batch.num_rows() {
+            assert!(!names.value(i).is_empty(), "file_name must be non-empty");
+            assert!(sizes.value(i) >= 0, "file_size must be non-negative");
+        }
+    }
+
+    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 latest = sm
+        .get_latest_snapshot()
+        .await
+        .unwrap()
+        .expect("fixture has snapshots");
+    let mut expected = paimon::spec::ManifestList::read(
+        table.file_io(),
+        &sm.manifest_path(latest.base_manifest_list()),
+    )
+    .await
+    .unwrap()
+    .len();
+    expected += paimon::spec::ManifestList::read(
+        table.file_io(),
+        &sm.manifest_path(latest.delta_manifest_list()),
+    )
+    .await
+    .unwrap()
+    .len();
+    if let Some(changelog) = latest.changelog_manifest_list() {
+        expected += paimon::spec::ManifestList::read(table.file_io(), &sm.manifest_path(changelog))
+            .await
+            .unwrap()
+            .len();
+    }
+    assert_eq!(
+        total_rows, expected,
+        "$manifests rows should match base + delta + changelog manifest entries of the latest snapshot"
+    );
+}
diff --git a/crates/paimon/src/spec/avro/manifest_file_meta_decode.rs b/crates/paimon/src/spec/avro/manifest_file_meta_decode.rs
@@ -33,6 +33,8 @@ impl AvroRecordDecode for ManifestFileMeta {
         let mut num_deleted_files: Option<i64> = None;
         let mut partition_stats: Option<BinaryTableStats> = None;
         let mut schema_id: Option<i64> = None;
+        let mut min_row_id: Option<i64> = None;
+        let mut max_row_id: Option<i64> = None;
 
         for field in &writer_schema.fields {
             match field.name.as_str() {
@@ -50,6 +52,8 @@ impl AvroRecordDecode for ManifestFileMeta {
                         decode_nullable_binary_table_stats(cursor, &field.schema, field.nullable)?;
                 }
                 "_SCHEMA_ID" => schema_id = Some(read_long_field(cursor, field.nullable)?),
+                "_MIN_ROW_ID" => min_row_id = read_optional_long(cursor, field.nullable)?,
+                "_MAX_ROW_ID" => max_row_id = read_optional_long(cursor, field.nullable)?,
                 _ => skip_nullable_field(cursor, &field.schema, field.nullable)?,
             }
         }
@@ -62,10 +66,22 @@ impl AvroRecordDecode for ManifestFileMeta {
             num_deleted_files.unwrap_or(0),
             partition_stats.unwrap_or_else(|| BinaryTableStats::new(vec![], vec![], vec![])),
             schema_id.unwrap_or(0),
+            min_row_id,
+            max_row_id,
         ))
     }
 }
 
+fn read_optional_long(cursor: &mut AvroCursor, nullable: bool) -> crate::Result<Option<i64>> {
+    if nullable {
+        let idx = cursor.read_union_index()?;
+        if idx == 0 {
+            return Ok(None);
+        }
+    }
+    Ok(Some(cursor.read_long()?))
+}
+
 /// Decode a nullable BinaryTableStats: union ["null", record] or direct record.
 pub(crate) fn decode_nullable_binary_table_stats(
     cursor: &mut AvroCursor,
diff --git a/crates/paimon/src/spec/manifest_file_meta.rs b/crates/paimon/src/spec/manifest_file_meta.rs
@@ -50,6 +50,22 @@ pub struct ManifestFileMeta {
     /// schema id when writing this manifest file.
     #[serde(rename = "_SCHEMA_ID")]
     schema_id: i64,
+
+    /// minimum row id covered by this manifest, when row tracking is enabled.
+    #[serde(
+        rename = "_MIN_ROW_ID",
+        default,
+        skip_serializing_if = "Option::is_none"
+    )]
+    min_row_id: Option<i64>,
+
+    /// maximum row id covered by this manifest, when row tracking is enabled.
+    #[serde(
+        rename = "_MAX_ROW_ID",
+        default,
+        skip_serializing_if = "Option::is_none"
+    )]
+    max_row_id: Option<i64>,
 }
 
 impl ManifestFileMeta {
@@ -94,6 +110,18 @@ impl ManifestFileMeta {
         self.version
     }
 
+    /// Get the minimum row id covered by this manifest (None when row tracking is disabled).
+    #[inline]
+    pub fn min_row_id(&self) -> Option<i64> {
+        self.min_row_id
+    }
+
+    /// Get the maximum row id covered by this manifest (None when row tracking is disabled).
+    #[inline]
+    pub fn max_row_id(&self) -> Option<i64> {
+        self.max_row_id
+    }
+
     #[inline]
     pub fn new(
         file_name: String,
@@ -111,10 +139,13 @@ impl ManifestFileMeta {
             num_deleted_files,
             partition_stats,
             schema_id,
+            min_row_id: None,
+            max_row_id: None,
         }
     }
 
     #[inline]
+    #[allow(clippy::too_many_arguments)]
     pub(crate) fn new_with_version(
         version: i32,
         file_name: String,
@@ -123,6 +154,8 @@ impl ManifestFileMeta {
         num_deleted_files: i64,
         partition_stats: BinaryTableStats,
         schema_id: i64,
+        min_row_id: Option<i64>,
+        max_row_id: Option<i64>,
     ) -> ManifestFileMeta {
         Self {
             version,
@@ -132,6 +165,8 @@ impl ManifestFileMeta {
             num_deleted_files,
             partition_stats,
             schema_id,
+            min_row_id,
+            max_row_id,
         }
     }
 }
@@ -156,7 +191,9 @@ pub const MANIFEST_FILE_META_SCHEMA: &str = r#"["null", {
                 {"name": "_NULL_COUNTS", "type": ["null", {"type": "array", "items": ["null", "long"]}], "default": null}
             ]
         }], "default": null},
-        {"name": "_SCHEMA_ID", "type": "long"}
+        {"name": "_SCHEMA_ID", "type": "long"},
+        {"name": "_MIN_ROW_ID", "type": ["null", "long"], "default": null},
+        {"name": "_MAX_ROW_ID", "type": ["null", "long"], "default": null}
     ]
 }]"#;
 
diff --git a/crates/paimon/src/spec/objects_file.rs b/crates/paimon/src/spec/objects_file.rs
diff --git a/docs/src/datafusion.md b/docs/src/datafusion.md
