feat(datafusion): support branch_manager and system table

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

@@ -0,0 +1,123 @@
+// 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 [BranchesTable](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/system/BranchesTable.java).
+
+use std::any::Any;
+use std::sync::{Arc, OnceLock};
+
+use async_trait::async_trait;
+use datafusion::arrow::array::{RecordBatch, StringArray, TimestampMillisecondArray};
+use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
+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::table::{BranchManager, Table};
+
+use crate::error::to_datafusion_error;
+
+pub(super) fn build(table: Table) -> DFResult<Arc<dyn TableProvider>> {
+    Ok(Arc::new(BranchesTable { table }))
+}
+
+fn branches_schema() -> SchemaRef {
+    static SCHEMA: OnceLock<SchemaRef> = OnceLock::new();
+    SCHEMA
+        .get_or_init(|| {
+            Arc::new(Schema::new(vec![
+                Field::new("branch_name", DataType::Utf8, false),
+                Field::new(
+                    "create_time",
+                    DataType::Timestamp(TimeUnit::Millisecond, None),
+                    false,
+                ),
+            ]))
+        })
+        .clone()
+}
+
+#[derive(Debug)]
+struct BranchesTable {
+    table: Table,
+}
+
+#[async_trait]
+impl TableProvider for BranchesTable {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn schema(&self) -> SchemaRef {
+        branches_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 bm = BranchManager::new(
+            self.table.file_io().clone(),
+            self.table.location().to_string(),
+        );
+        let branch_names = bm.list_all().await.map_err(to_datafusion_error)?;
+
+        let n = branch_names.len();
+        let mut names: Vec<String> = Vec::with_capacity(n);
+        let mut create_times = Vec::with_capacity(n);
+
+        for name in branch_names {
+            let branch_path = bm.branch_path(&name);
+            let status = self
+                .table
+                .file_io()
+                .get_status(&branch_path)
+                .await
+                .map_err(to_datafusion_error)?;
+            let ts_millis = status
+                .last_modified
+                .map(|dt| dt.timestamp_millis())
+                .unwrap_or(0);
+            names.push(name);
+            create_times.push(ts_millis);
+        }
+
+        let schema = branches_schema();
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(StringArray::from(names)),
+                Arc::new(TimestampMillisecondArray::from(create_times)),
+            ],
+        )?;
+
+        Ok(MemorySourceConfig::try_new_exec(
+            &[vec![batch]],
+            schema,
+            projection.cloned(),
+        )?)
+    }
+}

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

@@ -29,6 +29,7 @@ use paimon::table::Table;
 
 use crate::error::to_datafusion_error;
 
+mod branches;
 mod manifests;
 mod options;
 mod schemas;
@@ -38,6 +39,7 @@ mod tags;
 type Builder = fn(Table) -> DFResult<Arc<dyn TableProvider>>;
 
 const TABLES: &[(&str, Builder)] = &[
+    ("branches", branches::build),
     ("manifests", manifests::build),
     ("options", options::build),
     ("schemas", schemas::build),
@@ -130,6 +132,9 @@ mod tests {
         assert!(is_registered("schemas"));
         assert!(is_registered("Schemas"));
         assert!(is_registered("SCHEMAS"));
+        assert!(is_registered("branches"));
+        assert!(is_registered("Branches"));
+        assert!(is_registered("BRANCHES"));
         assert!(is_registered("tags"));
         assert!(is_registered("Tags"));
         assert!(is_registered("TAGS"));

crates/integrations/datafusion/tests/system_tables.rs

@@ -329,6 +329,63 @@ async fn test_snapshots_system_table() {
     );
 }
 
+#[tokio::test]
+async fn test_branches_system_table_empty_when_no_branch_dir() {
+    let (ctx, _catalog, _tmp) = create_context().await;
+    let sql = format!("SELECT * FROM paimon.default.{FIXTURE_TABLE}$branches");
+    let batches = run_sql(&ctx, &sql).await;
+
+    // Schema must be present even with zero rows.
+    assert!(!batches.is_empty(), "$branches should return ≥1 batch");
+    let arrow_schema = batches[0].schema();
+    let expected_columns = [
+        ("branch_name", DataType::Utf8),
+        (
+            "create_time",
+            DataType::Timestamp(TimeUnit::Millisecond, None),
+        ),
+    ];
+    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_eq!(total_rows, 0, "fixture has no branch dir, expected 0 rows");
+}
+
+#[tokio::test]
+async fn test_branches_system_table_with_seeded_branches() {
+    let (ctx, _catalog, tmp) = create_context().await;
+
+    let table_dir = tmp.path().join("default.db").join(FIXTURE_TABLE);
+    let branch_dir = table_dir.join("branch");
+    std::fs::create_dir_all(&branch_dir).expect("create branch dir");
+    std::fs::create_dir_all(branch_dir.join("branch-b1")).unwrap();
+    std::fs::create_dir_all(branch_dir.join("branch-b2")).unwrap();
+
+    let sql = format!("SELECT branch_name FROM paimon.default.{FIXTURE_TABLE}$branches");
+    let batches = run_sql(&ctx, &sql).await;
+    let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
+    assert_eq!(total_rows, 2, "expected two seeded branches");
+
+    let mut names: Vec<String> = Vec::new();
+    for batch in &batches {
+        let names_col = batch
+            .column(0)
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .expect("branch_name is Utf8");
+        for i in 0..batch.num_rows() {
+            names.push(names_col.value(i).to_string());
+        }
+    }
+    let mut sorted_names = names.clone();
+    sorted_names.sort();
+    assert_eq!(names, sorted_names, "branch_name should be ascending");
+    assert_eq!(names, vec!["b1".to_string(), "b2".to_string()]);
+}
+
 #[tokio::test]
 async fn test_tags_system_table_empty_when_no_tag_dir() {
     let (ctx, _catalog, _tmp) = create_context().await;

crates/paimon/src/io/file_io.rs

@@ -213,6 +213,17 @@ impl FileIO {
         Ok(())
     }
 
+    /// Copy a file from src to dst.
+    ///
+    /// Overwrites dst if it already exists.
+    pub async fn copy_file(&self, src: &str, dst: &str) -> Result<()> {
+        let input = self.new_input(src)?;
+        let bytes = input.read().await?;
+        let output = self.new_output(dst)?;
+        output.write(bytes).await?;
+        Ok(())
+    }
+
     /// Renames the file/directory src to dst.
     ///
     /// Reference: <https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L159>