feat(datafusion): Add $options system table

crates/integrations/datafusion/src/catalog.rs

@@ -29,6 +29,7 @@ use paimon::catalog::{Catalog, Identifier};
 
 use crate::error::to_datafusion_error;
 use crate::runtime::{await_with_runtime, block_on_with_runtime};
+use crate::system_tables;
 use crate::table::PaimonTableProvider;
 
 /// Provides an interface to manage and access multiple schemas (databases)
@@ -130,8 +131,19 @@ impl SchemaProvider for PaimonSchemaProvider {
     }
 
     async fn table(&self, name: &str) -> DFResult<Option<Arc<dyn TableProvider>>> {
+        let (base, system_name) = system_tables::split_object_name(name);
+        if let Some(system_name) = system_name {
+            return await_with_runtime(system_tables::load(
+                Arc::clone(&self.catalog),
+                self.database.clone(),
+                base.to_string(),
+                system_name.to_string(),
+            ))
+            .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 +158,15 @@ impl SchemaProvider for PaimonSchemaProvider {
     }
 
     fn table_exist(&self, name: &str) -> bool {
+        let (base, system_name) = system_tables::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 {

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,161 @@
+// 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):
+//! `TABLES` maps each system-table name to its builder function.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use datafusion::error::{DataFusionError, Result as DFResult};
+use paimon::catalog::{Catalog, Identifier, SYSTEM_BRANCH_PREFIX, SYSTEM_TABLE_SPLITTER};
+use paimon::table::Table;
+
+use crate::error::to_datafusion_error;
+
+mod options;
+
+type Builder = fn(Table) -> DFResult<Arc<dyn TableProvider>>;
+
+const TABLES: &[(&str, Builder)] = &[("options", options::build)];
+
+/// 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)
+pub(crate) 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)
+            }
+        }
+    }
+}
+
+/// 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))
+}
+
+/// Wraps an already-loaded base table as the system table `name`.
+fn wrap_to_system_table(name: &str, base_table: Table) -> Option<DFResult<Arc<dyn TableProvider>>> {
+    TABLES
+        .iter()
+        .find(|(n, _)| name.eq_ignore_ascii_case(n))
+        .map(|(_, build)| build(base_table))
+}
+
+/// Loads `<base>$<system_name>` from the catalog and wraps it as a system
+/// table provider.
+///
+/// - Unknown `system_name` → `Ok(None)` (DataFusion reports "table not found")
+/// - Base table missing    → `Err(Plan)` so users can distinguish it from an
+///   unknown system name
+pub(crate) async fn load(
+    catalog: Arc<dyn Catalog>,
+    database: String,
+    base: String,
+    system_name: String,
+) -> DFResult<Option<Arc<dyn TableProvider>>> {
+    if !is_registered(&system_name) {
+        return Ok(None);
+    }
+    let identifier = Identifier::new(database, base.clone());
+    match catalog.get_table(&identifier).await {
+        Ok(table) => wrap_to_system_table(&system_name, table)
+            .expect("is_registered guarantees a builder")
+            .map(Some),
+        Err(paimon::Error::TableNotExist { .. }) => Err(DataFusionError::Plan(format!(
+            "Cannot read system table `${system_name}`: \
+             base table `{base}` does not exist"
+        ))),
+        Err(e) => Err(to_datafusion_error(e)),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{is_registered, split_object_name};
+
+    #[test]
+    fn is_registered_is_case_insensitive() {
+        assert!(is_registered("options"));
+        assert!(is_registered("Options"));
+        assert!(is_registered("OPTIONS"));
+        assert!(!is_registered("nonsense"));
+    }
+
+    #[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));
+    }
+
+    #[test]
+    fn system_table_name_preserves_case() {
+        assert_eq!(
+            split_object_name("orders$Options"),
+            ("orders", Some("Options"))
+        );
+    }
+}

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

@@ -0,0 +1,92 @@
+// 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, OnceLock};
+
+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::Result as DFResult;
+use datafusion::logical_expr::Expr;
+use datafusion::physical_plan::ExecutionPlan;
+use paimon::table::Table;
+
+pub(super) fn build(table: Table) -> DFResult<Arc<dyn TableProvider>> {
+    Ok(Arc::new(OptionsTable { table }))
+}
+
+fn options_schema() -> SchemaRef {
+    static SCHEMA: OnceLock<SchemaRef> = OnceLock::new();
+    SCHEMA
+        .get_or_init(|| {
+            Arc::new(Schema::new(vec![
+                Field::new("key", DataType::Utf8, false),
+                Field::new("value", DataType::Utf8, false),
+            ]))
+        })
+        .clone()
+}
+
+#[derive(Debug)]
+struct OptionsTable {
+    table: Table,
+}
+
+#[async_trait]
+impl TableProvider for OptionsTable {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn schema(&self) -> SchemaRef {
+        options_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>> {
+        // Java uses LinkedHashMap insertion order; HashMap has none — sort for stable output.
+        let mut entries: Vec<(&String, &String)> = self.table.schema().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 schema = options_schema();
+        let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(keys), Arc::new(values)])?;
+
+        Ok(MemorySourceConfig::try_new_exec(
+            &[vec![batch]],
+            schema,
+            projection.cloned(),
+        )?)
+    }
+}