feat(core): add SystemRead for Native metadata reads

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -283,6 +283,7 @@ class Table:
     def branch(self) -> str: ...
     def location(self) -> str: ...
     def schema(self) -> TableSchema: ...
+    def new_system_read(self, name: str) -> SystemRead: ...
     def new_read_builder(self, options: Optional[Dict[str, str]] = None) -> ReadBuilder: ...
     def new_stream_scan(self) -> StreamTableScan: ...
     def new_full_text_search_builder(self) -> FullTextSearchBuilder: ...
@@ -863,3 +864,13 @@ class ShardTableUpdator:
         """Require a finished read and exactly as many update rows; repeated calls reuse messages."""
         ...
     def close(self) -> None: ...
+
+
+class SystemRead:
+    def read_type(self) -> List[DataField]: ...
+    def with_projection(self, columns: List[str]) -> SystemRead: ...
+    def with_filter(self, predicate: Dict[str, Any]) -> SystemRead: ...
+    def with_limit(self, limit: int) -> SystemRead: ...
+    def read(self) -> pyarrow.RecordBatch:
+        """Materialize metadata once, then filter, project and limit the result."""
+        ...

bindings/python/src/context.rs

@@ -796,6 +796,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()>
     let this = PyModule::new(py, "datafusion")?;
     this.add_class::<PaimonCatalog>()?;
     this.add_class::<crate::table::PyTable>()?;
+    this.add_class::<crate::system_table::PySystemRead>()?;
     this.add_class::<crate::index_build::PyGlobalIndexBuildBuilder>()?;
     this.add_class::<crate::index_build::PyGlobalIndexDropBuilder>()?;
     this.add_class::<crate::full_text_search::PyFullTextSearchBuilder>()?;

bindings/python/src/lib.rs

@@ -33,6 +33,7 @@ mod predicate;
 mod read;
 mod schema;
 mod shard_update;
+mod system_table;
 mod table;
 mod udf;
 mod update_assignment;

bindings/python/src/system_table.rs

@@ -0,0 +1,76 @@
+// 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.
+
+//! Thin bindings for core metadata reads.
+
+use crate::error::to_py_err;
+use crate::schema::PyDataField;
+use arrow::pyarrow::ToPyArrow;
+use paimon::table::SystemRead;
+use paimon_datafusion::runtime::runtime;
+use pyo3::prelude::*;
+use pyo3::types::PyDict;
+
+#[pyclass(name = "SystemRead", module = "pypaimon_rust.datafusion")]
+pub struct PySystemRead {
+    pub(crate) inner: SystemRead,
+}
+
+#[pymethods]
+impl PySystemRead {
+    fn read_type(&self) -> Vec<PyDataField> {
+        self.inner
+            .read_type()
+            .into_iter()
+            .map(PyDataField::new)
+            .collect()
+    }
+
+    fn with_projection(
+        mut slf: PyRefMut<'_, Self>,
+        columns: Vec<String>,
+    ) -> PyResult<PyRefMut<'_, Self>> {
+        let columns: Vec<_> = columns.iter().map(String::as_str).collect();
+        slf.inner.with_projection(&columns).map_err(to_py_err)?;
+        Ok(slf)
+    }
+
+    fn with_filter<'py>(
+        mut slf: PyRefMut<'py, Self>,
+        predicate: &Bound<'_, PyDict>,
+    ) -> PyResult<PyRefMut<'py, Self>> {
+        let predicate = crate::predicate::dict_to_predicate(
+            predicate,
+            slf.inner.full_row_type().fields(),
+            true,
+        )?;
+        slf.inner.with_filter(predicate).map_err(to_py_err)?;
+        Ok(slf)
+    }
+
+    fn with_limit(mut slf: PyRefMut<'_, Self>, limit: usize) -> PyRefMut<'_, Self> {
+        slf.inner.with_limit(limit);
+        slf
+    }
+
+    fn read(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
+        let batch = py
+            .detach(|| runtime().block_on(self.inner.read()))
+            .map_err(to_py_err)?;
+        Ok(batch.to_pyarrow(py)?.unbind())
+    }
+}

bindings/python/src/table.rs

@@ -198,6 +198,13 @@ impl PyTable {
         }
     }
 
+    /// Read metadata through the core system reader.
+    fn new_system_read(&self, name: &str) -> PyResult<crate::system_table::PySystemRead> {
+        Ok(crate::system_table::PySystemRead {
+            inner: self.inner.new_system_read(name).map_err(to_py_err)?,
+        })
+    }
+
     fn new_full_text_search_builder(&self) -> crate::full_text_search::PyFullTextSearchBuilder {
         crate::full_text_search::PyFullTextSearchBuilder::new(Arc::clone(&self.inner))
     }

crates/integrations/datafusion/Cargo.toml

@@ -45,7 +45,6 @@ futures = "0.3"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 tokio = { workspace = true, features = ["rt", "rt-multi-thread", "time", "fs"] }
-lexical-write-float = "1.0.6"
 uuid = { version = "1", features = ["v4"] }
 
 [dev-dependencies]