fix(datafusion): snapshot metadata for synchronous catalog callbacks

bindings/python/src/context.rs

@@ -122,7 +122,7 @@ impl PaimonCatalog {
     fn new(py: Python<'_>, catalog_options: HashMap<String, String>) -> PyResult<Self> {
         let catalog = py.detach(|| build_paimon_catalog(catalog_options))?;
         let provider = Arc::new(
-            PaimonCatalogProvider::new(
+            PaimonCatalogProvider::new_uninitialized(
                 None,
                 Arc::clone(&catalog),
                 Default::default(),
@@ -134,12 +134,22 @@ impl PaimonCatalog {
         Ok(Self { catalog, provider })
     }
 
+    /// Refresh the metadata snapshot used by synchronous DataFusion callbacks.
+    fn refresh_metadata(&self, py: Python<'_>) -> PyResult<()> {
+        let provider = Arc::clone(&self.provider);
+        py.detach(|| runtime().block_on(provider.refresh_metadata()))
+            .map_err(df_to_py_err)
+    }
+
     /// Export this catalog as a DataFusion catalog provider PyCapsule.
     fn __datafusion_catalog_provider__<'py>(
         &self,
         py: Python<'py>,
         session: Bound<'py, PyAny>,
     ) -> PyResult<Bound<'py, PyCapsule>> {
+        let provider = Arc::clone(&self.provider);
+        py.detach(|| runtime().block_on(provider.initialize_metadata()))
+            .map_err(df_to_py_err)?;
         let name = cr"datafusion_catalog_provider".into();
         let provider = Arc::clone(&self.provider) as Arc<dyn CatalogProvider + Send>;
         let codec = ffi_logical_codec_from_pycapsule(session)?;

bindings/python/tests/test_catalog_gil.py

@@ -19,6 +19,7 @@
 from http.server import BaseHTTPRequestHandler, HTTPServer
 
 import pytest
+from datafusion import SessionContext
 
 from pypaimon_rust.datafusion import PaimonCatalog
 
@@ -84,3 +85,23 @@ def test_rest_catalog_calls_release_gil(rest_server, monkeypatch):
     assert catalog.list_tables("db") == ["table"]
     with pytest.raises(ValueError, match="does not exist"):
         catalog.get_table("db.missing")
+
+
+def test_rest_catalog_without_views_registers_datafusion_provider(
+    rest_server, monkeypatch
+):
+    monkeypatch.setenv("NO_PROXY", "localhost,127.0.0.1")
+    monkeypatch.setenv("no_proxy", "localhost,127.0.0.1")
+    catalog = PaimonCatalog(
+        {
+            "metastore": "rest",
+            "uri": rest_server,
+            "warehouse": "warehouse",
+            "token.provider": "bear",
+            "token": "test-token",
+        }
+    )
+
+    SessionContext().register_catalog_provider("paimon", catalog)
+    with pytest.raises(ValueError, match="Resource not found"):
+        catalog.refresh_metadata()

bindings/python/tests/test_datafusion.py

@@ -520,6 +520,37 @@ def test_query_simple_table_via_catalog_provider():
     ]
 
 
+def test_catalog_provider_initializes_information_schema_snapshot():
+    with tempfile.TemporaryDirectory() as warehouse:
+        writer = SQLContext()
+        writer.register_catalog("paimon", {"warehouse": warehouse})
+        writer.sql("CREATE TABLE paimon.default.users (id INT, name STRING)")
+
+        catalog = PaimonCatalog({"warehouse": warehouse})
+        ctx = SessionContext()
+        ctx.register_catalog_provider("paimon", catalog)
+        batches = ctx.sql(
+            "SELECT column_name FROM paimon.information_schema.columns "
+            "WHERE table_schema = 'default' AND table_name = 'users'"
+        ).collect()
+
+        assert set(pa.Table.from_batches(batches)["column_name"].to_pylist()) == {
+            "id",
+            "name",
+        }
+
+        writer.sql("CREATE TABLE paimon.default.orders (order_id BIGINT)")
+        catalog.refresh_metadata()
+        batches = ctx.sql(
+            "SELECT table_name FROM paimon.information_schema.tables "
+            "WHERE table_schema = 'default'"
+        ).collect()
+        assert set(pa.Table.from_batches(batches)["table_name"].to_pylist()) == {
+            "orders",
+            "users",
+        }
+
+
 def test_catalog_provider_returns_pyarrow_compatible_strings():
     with tempfile.TemporaryDirectory() as warehouse:
         writer = SQLContext()

crates/integrations/datafusion/Cargo.toml

@@ -41,9 +41,10 @@ datafusion = { workspace = true }
 log = "0.4"
 paimon = { workspace = true }
 futures = "0.3"
+indexmap = "2"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-tokio = { workspace = true, features = ["rt", "time", "fs"] }
+tokio = { workspace = true, features = ["rt", "time", "fs", "sync"] }
 lexical-write-float = "1.0.6"
 uuid = { version = "1", features = ["v4"] }