feat(datafusion): Add $options system table

crates/integrations/datafusion/src/catalog.rs

@@ -25,12 +25,45 @@ use async_trait::async_trait;
 use datafusion::catalog::{CatalogProvider, SchemaProvider};
 use datafusion::datasource::TableProvider;
 use datafusion::error::Result as DFResult;
-use paimon::catalog::{Catalog, Identifier};
+use paimon::catalog::{Catalog, Identifier, SYSTEM_BRANCH_PREFIX, SYSTEM_TABLE_SPLITTER};
 
 use crate::error::to_datafusion_error;
 use crate::runtime::{await_with_runtime, block_on_with_runtime};
+use crate::system_tables;
 use crate::table::PaimonTableProvider;
 
+/// Parse a Paimon object name into `(base_table, optional system_table_name)`.
+///
+/// Mirrors Java [Identifier.splitObjectName](https://github.com/apache/paimon/blob/release-1.3/paimon-api/src/main/java/org/apache/paimon/catalog/Identifier.java).
+///
+/// - `t` → `("t", None)`
+/// - `t$options` → `("t", Some("options"))`
+/// - `t$branch_main` → `("t", None)` (branch reference, not a system table)
+/// - `t$branch_main$options` → `("t", Some("options"))` (branch + system table)
+fn split_object_name(name: &str) -> (&str, Option<&str>) {
+    let mut parts = name.splitn(3, SYSTEM_TABLE_SPLITTER);
+    let base = parts.next().unwrap_or(name);
+    match (parts.next(), parts.next()) {
+        (None, _) => (base, None),
+        (Some(second), None) => {
+            if second.starts_with(SYSTEM_BRANCH_PREFIX) {
+                (base, None)
+            } else {
+                (base, Some(second))
+            }
+        }
+        (Some(second), Some(third)) => {
+            if second.starts_with(SYSTEM_BRANCH_PREFIX) {
+                (base, Some(third))
+            } else {
+                // `$` is legal in table names, so `t$foo$bar` falls through as
+                // plain `t` and errors later as "table not found".
+                (base, None)
+            }
+        }
+    }
+}
+
 /// Provides an interface to manage and access multiple schemas (databases)
 /// within a Paimon [`Catalog`].
 ///
@@ -112,6 +145,43 @@ impl PaimonSchemaProvider {
     pub fn new(catalog: Arc<dyn Catalog>, database: String) -> Self {
         PaimonSchemaProvider { catalog, database }
     }
+
+    /// Resolves `<base>$<system_name>` into a system table provider.
+    ///
+    /// Unknown system names return `Ok(None)` (DataFusion reports "table not
+    /// found"). When the system name is registered but the base table is
+    /// missing, an explicit error is returned so users can tell the two cases
+    /// apart in error messages.
+    async fn load_system_table(
+        &self,
+        base: &str,
+        system_name: &str,
+    ) -> DFResult<Option<Arc<dyn TableProvider>>> {
+        if !system_tables::is_registered(system_name) {
+            return Ok(None);
+        }
+
+        let catalog = Arc::clone(&self.catalog);
+        let database = self.database.clone();
+        let base_owned = base.to_string();
+        let system_name_owned = system_name.to_string();
+        await_with_runtime(async move {
+            let identifier = Identifier::new(database, base_owned.clone());
+            match catalog.get_table(&identifier).await {
+                Ok(table) => system_tables::build(&system_name_owned, table)
+                    .expect("is_registered guarantees a builder")
+                    .map(Some),
+                Err(paimon::Error::TableNotExist { .. }) => {
+                    Err(datafusion::error::DataFusionError::Plan(format!(
+                        "Cannot read system table `${system_name_owned}`: \
+                         base table `{base_owned}` does not exist"
+                    )))
+                }
+                Err(e) => Err(to_datafusion_error(e)),
+            }
+        })
+        .await
+    }
 }
 
 #[async_trait]
@@ -130,8 +200,13 @@ impl SchemaProvider for PaimonSchemaProvider {
     }
 
     async fn table(&self, name: &str) -> DFResult<Option<Arc<dyn TableProvider>>> {
+        let (base, system_name) = split_object_name(name);
+        if let Some(system_name) = system_name {
+            return self.load_system_table(base, system_name).await;
+        }
+
         let catalog = Arc::clone(&self.catalog);
-        let identifier = Identifier::new(self.database.clone(), name);
+        let identifier = Identifier::new(self.database.clone(), base);
         await_with_runtime(async move {
             match catalog.get_table(&identifier).await {
                 Ok(table) => {
@@ -146,8 +221,17 @@ impl SchemaProvider for PaimonSchemaProvider {
     }
 
     fn table_exist(&self, name: &str) -> bool {
+        // Malformed `t$foo$bar` (no `branch_` segment) falls through as plain `t`,
+        // matching `table()`.
+        let (base, system_name) = split_object_name(name);
+        if let Some(system_name) = system_name {
+            if !system_tables::is_registered(system_name) {
+                return false;
+            }
+        }
+
         let catalog = Arc::clone(&self.catalog);
-        let identifier = Identifier::new(self.database.clone(), name);
+        let identifier = Identifier::new(self.database.clone(), base.to_string());
         block_on_with_runtime(
             async move {
                 match catalog.get_table(&identifier).await {
@@ -160,3 +244,39 @@ impl SchemaProvider for PaimonSchemaProvider {
         )
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::split_object_name;
+
+    #[test]
+    fn plain_table_name() {
+        assert_eq!(split_object_name("orders"), ("orders", None));
+    }
+
+    #[test]
+    fn system_table_only() {
+        assert_eq!(
+            split_object_name("orders$options"),
+            ("orders", Some("options"))
+        );
+    }
+
+    #[test]
+    fn branch_reference_is_not_a_system_table() {
+        assert_eq!(split_object_name("orders$branch_main"), ("orders", None));
+    }
+
+    #[test]
+    fn branch_plus_system_table() {
+        assert_eq!(
+            split_object_name("orders$branch_main$options"),
+            ("orders", Some("options"))
+        );
+    }
+
+    #[test]
+    fn three_parts_without_branch_prefix_is_not_a_system_table() {
+        assert_eq!(split_object_name("orders$foo$bar"), ("orders", None));
+    }
+}

crates/integrations/datafusion/src/lib.rs

@@ -44,6 +44,7 @@ mod full_text_search;
 mod physical_plan;
 mod relation_planner;
 pub mod runtime;
+mod system_tables;
 mod table;
 
 pub use catalog::{PaimonCatalogProvider, PaimonSchemaProvider};

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

@@ -0,0 +1,48 @@
+// 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.
+
+//! Paimon system tables (`<table>$<name>`) as DataFusion table providers.
+//!
+//! Mirrors Java [SystemTableLoader](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java):
+//! a single table maps each system-table name to a builder function. Add a new
+//! system table by dropping a file under this module and appending one entry to
+//! `TABLES`.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use datafusion::error::Result as DFResult;
+use paimon::table::Table;
+
+mod options;
+
+type Builder = fn(Table) -> DFResult<Arc<dyn TableProvider>>;
+
+const TABLES: &[(&str, Builder)] = &[("options", options::build)];
+
+/// Returns true if `name` is a recognised Paimon system table suffix.
+pub(crate) fn is_registered(name: &str) -> bool {
+    TABLES.iter().any(|(n, _)| name.eq_ignore_ascii_case(n))
+}
+
+/// Builds a system table provider for `name`, or `None` if unrecognised.
+pub(crate) fn build(name: &str, table: Table) -> Option<DFResult<Arc<dyn TableProvider>>> {
+    TABLES
+        .iter()
+        .find(|(n, _)| name.eq_ignore_ascii_case(n))
+        .map(|(_, build)| build(table))
+}

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

@@ -0,0 +1,103 @@
+// 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 [OptionsTable](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/system/OptionsTable.java).
+
+use std::any::Any;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use datafusion::arrow::array::{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::{DataFusionError, Result as DFResult};
+use datafusion::logical_expr::Expr;
+use datafusion::physical_plan::ExecutionPlan;
+use paimon::table::Table;
+
+use crate::error::to_datafusion_error;
+
+pub(super) fn build(table: Table) -> DFResult<Arc<dyn TableProvider>> {
+    Ok(Arc::new(OptionsTable {
+        table,
+        schema: options_schema(),
+    }))
+}
+
+fn options_schema() -> SchemaRef {
+    Arc::new(Schema::new(vec![
+        Field::new("key", DataType::Utf8, false),
+        Field::new("value", DataType::Utf8, false),
+    ]))
+}
+
+#[derive(Debug)]
+struct OptionsTable {
+    table: Table,
+    schema: SchemaRef,
+}
+
+#[async_trait]
+impl TableProvider for OptionsTable {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn schema(&self) -> SchemaRef {
+        self.schema.clone()
+    }
+
+    fn table_type(&self) -> TableType {
+        // Java: `ReadonlyTable` — virtual, read-only.
+        TableType::View
+    }
+
+    async fn scan(
+        &self,
+        _state: &dyn Session,
+        projection: Option<&Vec<usize>>,
+        _filters: &[Expr],
+        _limit: Option<usize>,
+    ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        // Re-read latest schema each scan so in-place schema evolution is visible (Java parity).
+        let latest = self
+            .table
+            .schema_manager()
+            .latest()
+            .await
+            .map_err(to_datafusion_error)?
+            .ok_or_else(|| DataFusionError::Plan("Table not exists.".to_string()))?;
+
+        // Java uses LinkedHashMap insertion order; HashMap has none — sort for stable output.
+        let mut entries: Vec<(&String, &String)> = latest.options().iter().collect();
+        entries.sort_by(|a, b| a.0.cmp(b.0));
+
+        let keys = StringArray::from_iter_values(entries.iter().map(|(k, _)| k.as_str()));
+        let values = StringArray::from_iter_values(entries.iter().map(|(_, v)| v.as_str()));
+
+        let batch =
+            RecordBatch::try_new(self.schema.clone(), vec![Arc::new(keys), Arc::new(values)])?;
+
+        Ok(MemorySourceConfig::try_new_exec(
+            &[vec![batch]],
+            self.schema.clone(),
+            projection.cloned(),
+        )?)
+    }
+}