You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "shyjsarah (via GitHub)" <gi...@apache.org> on 2026/09/20 03:23:27 UTC
[PR] fix(datafusion): snapshot metadata for synchronous catalog callbacks [paimon-rust]
shyjsarah opened a new pull request, #881:
URL: https://github.com/apache/paimon-rust/pull/881
### Purpose
Closes #872. Supersedes #873 with a fix at the catalog boundary instead of changing Tokio blocking behavior.
DataFusion's metadata discovery APIs are synchronous, while Paimon catalog and external table-engine resolution are asynchronous. Calling those async paths from `schema_names`, `schema`, `table_names`, or `table_exist` can block a DataFusion worker that the query itself needs, which is exposed by filtered `information_schema` queries with repartitioning.
### Brief change log
- Build database, table, and view metadata snapshots asynchronously when catalog/schema providers are initialized or explicitly refreshed.
- Make synchronous catalog discovery callbacks read only the in-memory snapshot; concrete table loading remains asynchronous.
- Refresh stale metadata before `SHOW`/`information_schema` queries and unknown-table planning, and refresh best-effort after metadata-changing DDL.
- Preserve the last good snapshot when a refresh fails.
- Initialize and expose metadata refresh through the Python `PaimonCatalog` wrapper.
- Keep catalog-declared external tables discoverable without synchronously invoking their engine resolver.
### Tests
- `cargo fmt --all --check`
- `cargo clippy -p paimon-datafusion --lib --tests -- -D warnings`
- `cargo test -p paimon-datafusion --test sql_context_tests` (60 passed)
- `cargo test -p paimon-datafusion --test table_type_routing` (38 passed)
- Targeted Python catalog-provider regressions (2 passed)
- `cargo test -p paimon-datafusion --lib`: 352 passed; the remaining 7 require the absent shared `partitioned_log_table` / `multi_partitioned_log_table` fixtures
### API and Format
Adds asynchronous `try_new` and `refresh_metadata` APIs to `PaimonCatalogProvider` and `PaimonSchemaProvider`, plus `PaimonCatalog.refresh_metadata()` in Python. Existing synchronous constructors remain available and start with an empty snapshot, so direct callers should initialize or refresh before registration.
No storage format changes.
### Documentation
Updated provider API documentation to describe snapshot initialization and refresh behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(datafusion): snapshot metadata for synchronous catalog callbacks [paimon-rust]
Posted by "shyjsarah (via GitHub)" <gi...@apache.org>.
shyjsarah closed pull request #881: fix(datafusion): snapshot metadata for synchronous catalog callbacks
URL: https://github.com/apache/paimon-rust/pull/881
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(datafusion): snapshot metadata for synchronous catalog callbacks [paimon-rust]
Posted by "shyjsarah (via GitHub)" <gi...@apache.org>.
shyjsarah commented on PR #881:
URL: https://github.com/apache/paimon-rust/pull/881#issuecomment-5756616958
Closing this PR because #891 has landed and addresses the original Python/process-runtime information_schema deadlock.
The snapshot approach here adds a much broader metadata caching, freshness, DDL-coherence, and lifecycle subsystem that is no longer justified for the original incident. #891 does not claim to cover every caller-owned Tokio runtime topology; if that remaining boundary can be reproduced, it should be tracked separately with a focused regression test and a narrower fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(datafusion): snapshot metadata for synchronous catalog callbacks [paimon-rust]
Posted by "shyjsarah (via GitHub)" <gi...@apache.org>.
shyjsarah commented on code in PR #881:
URL: https://github.com/apache/paimon-rust/pull/881#discussion_r4056420887
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -423,15 +646,22 @@ impl CatalogProvider for PaimonCatalogProvider {
let catalog_name = self.catalog_name.clone();
let session_state = self.session_state.clone();
let schema_force_view_types = self.schema_force_view_types;
+ let table_engines = self.table_engines();
+ let metadata = Arc::clone(&self.metadata);
let name = name.to_string();
block_on_with_runtime(
Review Comment:
Agreed. I narrowed the PR scope instead of claiming the remaining bridges are fixed: the description now uses Refs #872, explicitly documents register_schema, deregister_schema, and deregister_table as follow-up work, and leaves #872 open.
##########
crates/integrations/datafusion/src/sql_context.rs:
##########
@@ -473,7 +489,18 @@ impl SQLContext {
));
}
- match &statements[0] {
+ let refresh_metadata_after = self.metadata_change_targets(&statements[0], false)?;
+ let metadata_mutation_targets = self.metadata_change_targets(&statements[0], true)?;
+ {
+ let _refresh_guard = self.metadata_refresh_gate.lock().await;
Review Comment:
Fixed in 4f68a94. Refresh targets are computed before locking, locks are keyed by catalog/database and weakly retained, and independent targets refresh concurrently. Metadata-free queries and other databases no longer wait behind a stalled refresh; same-target misses still share the TTL/single-flight behavior.
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -783,12 +1068,19 @@ impl SchemaProvider for PaimonSchemaProvider {
declared,
))
.await?;
- Ok(resolved.map(|inner| {
- Arc::new(ReadOnlyTableProvider {
+ Ok(Some(match resolved {
Review Comment:
Fixed in 4f68a94. A registered resolver returning Ok(None) is now preserved as None. The UnavailableEngineTableProvider fallback is only used when no resolver is registered. Added a regression test for the registered-resolver not-found case.
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -924,101 +1208,25 @@ impl SchemaProvider for PaimonSchemaProvider {
return false;
}
};
- if let Some(system_name) = object.system_table() {
- if !system_tables::is_registered(system_name) {
- return false;
- }
+ if object
+ .system_table()
+ .is_some_and(|system_name| !system_tables::is_registered(system_name))
+ {
+ return false;
+ }
+ if object.system_table().is_some() {
Review Comment:
Fixed in 4f68a94. table_exist now recognizes supported system-table suffixes and derives existence from the snapshotted base table; missing bases and unknown suffixes remain false. Added direct table_exist coverage.
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -49,6 +53,118 @@ pub(crate) type SessionStateProvider = Arc<dyn Fn() -> Option<SessionState> + Se
/// providers, so registrations stay visible to schemas obtained earlier.
type TableEngines = Arc<RwLock<HashMap<PaimonTableType, Arc<dyn TableEngineResolver>>>>;
+#[derive(Clone, Debug, Default)]
+struct CatalogMetadataSnapshot {
+ databases: IndexMap<String, Arc<DatabaseMetadata>>,
+}
+
+#[derive(Clone, Debug, Default)]
+struct DatabaseMetadata {
+ objects: IndexMap<String, TableType>,
+}
+
+#[derive(Debug)]
+struct CatalogMetadataState {
+ snapshot: RwLock<Arc<CatalogMetadataSnapshot>>,
+ next_generation: AtomicU64,
+ published_generation: AtomicU64,
+}
+
+impl Default for CatalogMetadataState {
+ fn default() -> Self {
+ Self {
+ snapshot: RwLock::new(Arc::new(CatalogMetadataSnapshot::default())),
+ next_generation: AtomicU64::new(0),
+ published_generation: AtomicU64::new(0),
+ }
+ }
+}
+
+impl Deref for CatalogMetadataState {
+ type Target = RwLock<Arc<CatalogMetadataSnapshot>>;
+
+ fn deref(&self) -> &Self::Target {
+ &self.snapshot
+ }
+}
+
+impl CatalogMetadataState {
+ fn begin_refresh(&self) -> u64 {
+ self.next_generation.fetch_add(1, Ordering::AcqRel) + 1
+ }
+
+ fn publish(&self, generation: u64, next: Arc<CatalogMetadataSnapshot>) {
+ let mut current = self.snapshot.write().unwrap_or_else(|e| e.into_inner());
+ if generation >= self.published_generation.load(Ordering::Acquire) {
Review Comment:
Fixed in 4f68a94. Generation tracking is now per database, and a full refresh merges each database independently. Newer scoped updates and tombstones win only for their own database, while non-conflicting results from the older full refresh are still published. Added concurrent merge and stale-resurrection regression tests.
##########
crates/integrations/datafusion/src/sql_context.rs:
##########
@@ -751,7 +807,204 @@ impl SQLContext {
self.ctx.sql(&expanded.to_string()).await
}
_ => self.ctx.sql(sql).await,
+ };
+
+ if result.is_ok() && !refresh_metadata_after.is_empty() {
+ if let Err(error) = self.refresh_metadata_targets(refresh_metadata_after).await {
Review Comment:
Fixed in 4f68a94. Successful DDL now applies the known create/drop/rename delta directly to the in-memory snapshot and does not await a post-commit listing. Added a blocked-listing regression test to verify the DDL result returns immediately.
##########
crates/integrations/datafusion/src/sql_context.rs:
##########
@@ -751,7 +807,204 @@ impl SQLContext {
self.ctx.sql(&expanded.to_string()).await
}
_ => self.ctx.sql(sql).await,
+ };
+
+ if result.is_ok() && !refresh_metadata_after.is_empty() {
+ if let Err(error) = self.refresh_metadata_targets(refresh_metadata_after).await {
+ log::warn!("catalog metadata refresh after DDL failed: {error}");
+ }
+ }
+ result
+ }
+
+ fn metadata_refresh_targets(
+ &self,
+ statement: &Statement,
+ ) -> DFResult<HashSet<MetadataRefreshTarget>> {
+ let mut targets = HashSet::new();
+ if matches!(statement, Statement::ShowTables { .. }) {
+ let state = self.ctx.state();
+ targets.insert(MetadataRefreshTarget::Database {
+ catalog: self.current_catalog_name(),
+ database: state.config_options().catalog.default_schema.clone(),
+ });
+ return Ok(targets);
+ }
+ if let Statement::ShowColumns { show_options, .. } = statement {
+ if let Some(show_in) = &show_options.show_in {
+ if show_in.parent_type.is_none() {
+ if let Some(name) = &show_in.parent_name {
+ let (_, catalog, identifier) = self.resolve_catalog_and_table(name)?;
+ targets.insert(MetadataRefreshTarget::Database {
+ catalog,
+ database: identifier.database().to_string(),
+ });
+ }
+ }
+ }
+ return Ok(targets);
+ }
+ if matches!(statement, Statement::ShowFunctions { .. }) {
+ return Ok(targets);
+ }
+
+ let statement = datafusion::sql::parser::Statement::Statement(Box::new(statement.clone()));
+ let state = self.ctx.state();
+ let default_catalog = state.config_options().catalog.default_catalog.clone();
+ let default_schema = state.config_options().catalog.default_schema.clone();
+ for reference in state.resolve_table_references(&statement)? {
+ let schema = reference.schema().unwrap_or(&default_schema);
+ let catalog_name = reference.catalog().unwrap_or(&default_catalog);
+ if !self.catalogs.contains_key(catalog_name) {
+ continue;
+ }
+ let provider = self.ctx.catalog(catalog_name).ok_or_else(|| {
+ DataFusionError::Plan(format!("Unknown catalog '{catalog_name}'"))
+ })?;
+ let provider = provider
+ .downcast_ref::<crate::catalog::PaimonCatalogProvider>()
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "Catalog '{catalog_name}' is not a Paimon catalog"
+ ))
+ })?;
+ if schema.eq_ignore_ascii_case("information_schema") {
+ targets.insert(MetadataRefreshTarget::Catalog(catalog_name.to_string()));
Review Comment:
Fixed in 4f68a94. An information_schema reference now refreshes every registered Paimon catalog concurrently. Those catalog-wide refreshes are best-effort so one unavailable catalog preserves its last good snapshot without hiding strict failures for ordinary missing-object targets. Added multi-catalog, failure, and concurrency tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(datafusion): snapshot metadata for synchronous catalog callbacks [paimon-rust]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #881:
URL: https://github.com/apache/paimon-rust/pull/881#discussion_r4056237723
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -783,12 +1068,19 @@ impl SchemaProvider for PaimonSchemaProvider {
declared,
))
.await?;
- Ok(resolved.map(|inner| {
- Arc::new(ReadOnlyTableProvider {
+ Ok(Some(match resolved {
Review Comment:
`TableEngineResolver` documents `Ok(None)` as not found, but this converts it into `Some(UnavailableEngineTableProvider)`. When an engine reports a deleted, absent, or unowned object, DataFusion now treats the table as existing, planning can succeed, and execution fails later. Please preserve `Ok(None)` for a registered resolver; the metadata-only fallback can remain for the no-resolver case.
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -924,101 +1208,25 @@ impl SchemaProvider for PaimonSchemaProvider {
return false;
}
};
- if let Some(system_name) = object.system_table() {
- if !system_tables::is_registered(system_name) {
- return false;
- }
+ if object
+ .system_table()
+ .is_some_and(|system_name| !system_tables::is_registered(system_name))
+ {
+ return false;
+ }
+ if object.system_table().is_some() {
Review Comment:
Returning false for every recognized system table makes `table_exist("t$snapshots")` disagree with `table("t$snapshots")` and with successful Paimon system-table queries. This regresses the previous behavior and violates the `SchemaProvider::table_exist` existence contract for direct callers. Please derive the answer from the snapshotted base table, or retain enough metadata to distinguish supported and unsupported derived tables.
##########
crates/integrations/datafusion/src/sql_context.rs:
##########
@@ -473,7 +489,18 @@ impl SQLContext {
));
}
- match &statements[0] {
+ let refresh_metadata_after = self.metadata_change_targets(&statements[0], false)?;
+ let metadata_mutation_targets = self.metadata_change_targets(&statements[0], true)?;
+ {
+ let _refresh_guard = self.metadata_refresh_gate.lock().await;
Review Comment:
Could the refresh targets be computed before taking a per-target singleflight lock? This context-wide gate is acquired by every SQL statement, even `SELECT 1`, and is held across remote catalog I/O. One stalled missing-table refresh therefore blocks unrelated existing-table queries and queries against other catalogs.
##########
crates/integrations/datafusion/src/sql_context.rs:
##########
@@ -751,7 +807,204 @@ impl SQLContext {
self.ctx.sql(&expanded.to_string()).await
}
_ => self.ctx.sql(sql).await,
+ };
+
+ if result.is_ok() && !refresh_metadata_after.is_empty() {
+ if let Err(error) = self.refresh_metadata_targets(refresh_metadata_after).await {
Review Comment:
This refresh runs after the catalog mutation has committed, but the successful DDL result is still withheld while an unbounded metadata listing is awaited. If the listing stalls, the client is left uncertain and may retry an already-applied DDL. Please update the known snapshot directly, or detach or bound this best-effort refresh.
##########
crates/integrations/datafusion/src/sql_context.rs:
##########
@@ -751,7 +807,204 @@ impl SQLContext {
self.ctx.sql(&expanded.to_string()).await
}
_ => self.ctx.sql(sql).await,
+ };
+
+ if result.is_ok() && !refresh_metadata_after.is_empty() {
+ if let Err(error) = self.refresh_metadata_targets(refresh_metadata_after).await {
+ log::warn!("catalog metadata refresh after DDL failed: {error}");
+ }
+ }
+ result
+ }
+
+ fn metadata_refresh_targets(
+ &self,
+ statement: &Statement,
+ ) -> DFResult<HashSet<MetadataRefreshTarget>> {
+ let mut targets = HashSet::new();
+ if matches!(statement, Statement::ShowTables { .. }) {
+ let state = self.ctx.state();
+ targets.insert(MetadataRefreshTarget::Database {
+ catalog: self.current_catalog_name(),
+ database: state.config_options().catalog.default_schema.clone(),
+ });
+ return Ok(targets);
+ }
+ if let Statement::ShowColumns { show_options, .. } = statement {
+ if let Some(show_in) = &show_options.show_in {
+ if show_in.parent_type.is_none() {
+ if let Some(name) = &show_in.parent_name {
+ let (_, catalog, identifier) = self.resolve_catalog_and_table(name)?;
+ targets.insert(MetadataRefreshTarget::Database {
+ catalog,
+ database: identifier.database().to_string(),
+ });
+ }
+ }
+ }
+ return Ok(targets);
+ }
+ if matches!(statement, Statement::ShowFunctions { .. }) {
+ return Ok(targets);
+ }
+
+ let statement = datafusion::sql::parser::Statement::Statement(Box::new(statement.clone()));
+ let state = self.ctx.state();
+ let default_catalog = state.config_options().catalog.default_catalog.clone();
+ let default_schema = state.config_options().catalog.default_schema.clone();
+ for reference in state.resolve_table_references(&statement)? {
+ let schema = reference.schema().unwrap_or(&default_schema);
+ let catalog_name = reference.catalog().unwrap_or(&default_catalog);
+ if !self.catalogs.contains_key(catalog_name) {
+ continue;
+ }
+ let provider = self.ctx.catalog(catalog_name).ok_or_else(|| {
+ DataFusionError::Plan(format!("Unknown catalog '{catalog_name}'"))
+ })?;
+ let provider = provider
+ .downcast_ref::<crate::catalog::PaimonCatalogProvider>()
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "Catalog '{catalog_name}' is not a Paimon catalog"
+ ))
+ })?;
+ if schema.eq_ignore_ascii_case("information_schema") {
+ targets.insert(MetadataRefreshTarget::Catalog(catalog_name.to_string()));
Review Comment:
DataFusion information schema enumerates every catalog in the session, but this refreshes only the catalog that qualifies the `information_schema` relation. In a multi-catalog session, filtering for rows from another catalog can therefore return stale or missing metadata. Please refresh all participating Paimon catalogs, or safely derive a literal `table_catalog` restriction and refresh that target.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(datafusion): snapshot metadata for synchronous catalog callbacks [paimon-rust]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #881:
URL: https://github.com/apache/paimon-rust/pull/881#discussion_r4056237324
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -423,15 +646,22 @@ impl CatalogProvider for PaimonCatalogProvider {
let catalog_name = self.catalog_name.clone();
let session_state = self.session_state.clone();
let schema_force_view_types = self.schema_force_view_types;
+ let table_engines = self.table_engines();
+ let metadata = Arc::clone(&self.metadata);
let name = name.to_string();
block_on_with_runtime(
Review Comment:
Could we avoid keeping the blocking bridge in the remaining mutation callbacks before closing #872? `register_schema`, `deregister_schema`, and `deregister_table` still move the future to the process runtime and synchronously join the OS thread. If that future waits for work scheduled on a single-worker caller runtime, the caller worker remains blocked and the same deadlock class as #872 persists. Moving these catalog mutations to explicit async paths, or narrowing the issue scope, would make the claimed supersession precise.
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -49,6 +53,118 @@ pub(crate) type SessionStateProvider = Arc<dyn Fn() -> Option<SessionState> + Se
/// providers, so registrations stay visible to schemas obtained earlier.
type TableEngines = Arc<RwLock<HashMap<PaimonTableType, Arc<dyn TableEngineResolver>>>>;
+#[derive(Clone, Debug, Default)]
+struct CatalogMetadataSnapshot {
+ databases: IndexMap<String, Arc<DatabaseMetadata>>,
+}
+
+#[derive(Clone, Debug, Default)]
+struct DatabaseMetadata {
+ objects: IndexMap<String, TableType>,
+}
+
+#[derive(Debug)]
+struct CatalogMetadataState {
+ snapshot: RwLock<Arc<CatalogMetadataSnapshot>>,
+ next_generation: AtomicU64,
+ published_generation: AtomicU64,
+}
+
+impl Default for CatalogMetadataState {
+ fn default() -> Self {
+ Self {
+ snapshot: RwLock::new(Arc::new(CatalogMetadataSnapshot::default())),
+ next_generation: AtomicU64::new(0),
+ published_generation: AtomicU64::new(0),
+ }
+ }
+}
+
+impl Deref for CatalogMetadataState {
+ type Target = RwLock<Arc<CatalogMetadataSnapshot>>;
+
+ fn deref(&self) -> &Self::Target {
+ &self.snapshot
+ }
+}
+
+impl CatalogMetadataState {
+ fn begin_refresh(&self) -> u64 {
+ self.next_generation.fetch_add(1, Ordering::AcqRel) + 1
+ }
+
+ fn publish(&self, generation: u64, next: Arc<CatalogMetadataSnapshot>) {
+ let mut current = self.snapshot.write().unwrap_or_else(|e| e.into_inner());
+ if generation >= self.published_generation.load(Ordering::Acquire) {
Review Comment:
Could we make generation tracking per database, or merge non-conflicting results? A database-scoped refresh with generation N+1 causes an already-running full refresh at N to discard its entire snapshot, including newly loaded metadata for unrelated databases, while still returning `Ok(())`. That makes a successful full refresh silently ineffective for unrelated entries.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(datafusion): snapshot metadata for synchronous catalog callbacks [paimon-rust]
Posted by "shyjsarah (via GitHub)" <gi...@apache.org>.
shyjsarah commented on PR #881:
URL: https://github.com/apache/paimon-rust/pull/881#issuecomment-5749046669
Follow-up review findings are addressed in 1b705e0:
- automatic `information_schema` refresh now has freshness caching, failure backoff, per-target single-flight, a global concurrency bound, and a timeout;
- rejected same-database full-refresh publications are retried instead of silently dropping unrelated discoveries;
- `ALTER TABLE IF EXISTS` no-op renames no longer create phantom tables;
- same-database remote DDL and local deltas share one gate, including database/table DDL interactions;
- the original synchronous provider constructor signatures are restored, with `try_new` as the initialized async path;
- routed/not-found existence semantics and database-generation tombstone retention are tightened.
Local verification: 92 SQL-context tests, 40 routing tests, Clippy/format checks, and 185 Python tests passed. The PR description now reflects the final API and test matrix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org