This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 28102b09 feat(datafusion): route metadata tables to registered table
engines (#1100)
28102b09 is described below
commit 28102b093c4c3904534454bdf981d2e549c03815
Author: Jiajia Li <[email protected]>
AuthorDate: Mon Oct 12 13:48:39 2026 +0800
feat(datafusion): route metadata tables to registered table engines (#1100)
---
crates/integrations/datafusion/src/catalog.rs | 191 ++++++---
.../datafusion/src/system_tables/mod.rs | 53 +--
.../datafusion/tests/table_type_routing.rs | 437 ++++++++++++++++++++-
3 files changed, 574 insertions(+), 107 deletions(-)
diff --git a/crates/integrations/datafusion/src/catalog.rs
b/crates/integrations/datafusion/src/catalog.rs
index 06f34c3c..9ba17d95 100644
--- a/crates/integrations/datafusion/src/catalog.rs
+++ b/crates/integrations/datafusion/src/catalog.rs
@@ -82,6 +82,17 @@ pub trait TableEngineResolver: Debug + Send + Sync {
&self,
request: &EngineTableRequest,
) -> DFResult<Option<Arc<dyn TableProvider>>>;
+
+ /// Resolves `<table>$<metadata_table>`; returns `Ok(None)` if missing or
unsupported.
+ /// The request names the base table; `metadata_table` is ASCII-lowercase.
+ /// Session scan options are checked before this call.
+ async fn resolve_metadata_table(
+ &self,
+ _request: &EngineTableRequest,
+ _metadata_table: &str,
+ ) -> DFResult<Option<Arc<dyn TableProvider>>> {
+ Ok(None)
+ }
}
/// Read-only wrapper around an engine-resolved provider: reads delegate,
@@ -91,6 +102,7 @@ pub(crate) struct ReadOnlyTableProvider {
inner: Arc<dyn TableProvider>,
pub(crate) declared: PaimonTableType,
pub(crate) table_name: String,
+ metadata_table: bool,
}
#[async_trait]
@@ -104,7 +116,11 @@ impl TableProvider for ReadOnlyTableProvider {
}
fn table_type(&self) -> TableType {
- self.inner.table_type()
+ if self.metadata_table {
+ TableType::View
+ } else {
+ self.inner.table_type()
+ }
}
fn get_table_definition(&self) -> Option<&str> {
@@ -692,21 +708,7 @@ impl SchemaProvider for PaimonSchemaProvider {
}
let object = system_tables::parse_object_name_for_datafusion(name)?;
- if let Some(system_name) = object.system_table().map(str::to_string) {
- let dynamic_options = self
- .dynamic_options
- .read()
- .unwrap_or_else(|e| e.into_inner())
- .clone();
- return await_with_runtime(system_tables::load(
- Arc::clone(&self.catalog),
- self.database.clone(),
- object,
- system_name,
- dynamic_options,
- ))
- .await;
- }
+ let system_name = object.system_table().map(str::to_string);
let catalog = Arc::clone(&self.catalog);
let dynamic_options = Arc::clone(&self.dynamic_options);
@@ -724,6 +726,15 @@ impl SchemaProvider for PaimonSchemaProvider {
await_with_runtime(async move {
match catalog.load_table(&identifier).await {
Ok(paimon::catalog::LoadedTable::Object(table)) => {
+ if let Some(system_name) = &system_name {
+ if !system_tables::is_registered(system_name) {
+ return Ok(None);
+ }
+ return Err(plan_datafusion_err!(
+ "table '{}' is declared 'object-table' and cannot
be read as a Paimon table",
+ identifier.full_name()
+ ));
+ }
if branch.is_some() {
return Err(plan_datafusion_err!(
"branches are not supported for 'object-table'
tables ('{}')",
@@ -751,7 +762,16 @@ impl SchemaProvider for PaimonSchemaProvider {
identifier.full_name()
));
}
+ // Keep the metadata suffix in errors.
+ let table_name = format!("{}.{}", identifier.database(),
name);
let Some(resolver) = table_engines.get(&declared) else {
+ if system_name.is_some() {
+ return Err(plan_datafusion_err!(
+ "no table engine is registered for '{}' tables
('{}')",
+ declared,
+ table_name
+ ));
+ }
let schema = match external.fields() {
Some(fields) =>
crate::table::datafusion_arrow_schema(
fields,
@@ -765,33 +785,52 @@ impl SchemaProvider for PaimonSchemaProvider {
table_name: identifier.full_name(),
}) as Arc<dyn TableProvider>));
};
- // The Paimon arm below applies these; an engine would
- // ignore them and answer from current data. A missing
- // engine only exposes catalog metadata, so read-specific
- // session options do not apply to that fallback.
+ // Engines cannot apply Paimon scan options or query
authorization.
let session_options = dynamic_options
.read()
.unwrap_or_else(|e| e.into_inner())
.clone();
paimon::spec::CoreOptions::new(&session_options)
- .ensure_engine_can_serve(&identifier.full_name())
+ .ensure_engine_can_serve(&table_name)
.map_err(to_datafusion_error)?;
- let resolved = resolver
- .resolve_table(&EngineTableRequest::new(
- identifier.database().to_string(),
- identifier.object().to_string(),
- declared,
- ))
- .await?;
+ let request = EngineTableRequest::new(
+ identifier.database().to_string(),
+ identifier.object().to_string(),
+ declared,
+ );
+ let metadata_table = system_name.is_some();
+ let resolved = match system_name {
+ Some(system_name) => {
+ let system_name = system_name.to_ascii_lowercase();
+ resolver.resolve_metadata_table(&request,
&system_name).await?
+ }
+ None => resolver.resolve_table(&request).await?,
+ };
Ok(resolved.map(|inner| {
Arc::new(ReadOnlyTableProvider {
inner,
declared,
- table_name: identifier.full_name(),
+ table_name,
+ metadata_table,
}) as Arc<dyn TableProvider>
}))
}
Ok(paimon::catalog::LoadedTable::Paimon(table)) => {
+ if let Some(system_name) = system_name {
+ let opts = dynamic_options
+ .read()
+ .unwrap_or_else(|e| e.into_inner())
+ .clone();
+ return system_tables::load(
+ catalog,
+ identifier,
+ *table,
+ object,
+ system_name,
+ opts,
+ )
+ .await;
+ }
let mut table = *table;
if let Some(branch) = branch.as_deref() {
table = table
@@ -824,6 +863,16 @@ impl SchemaProvider for PaimonSchemaProvider {
Ok(Some(Arc::new(provider) as Arc<dyn TableProvider>))
}
Err(paimon::Error::TableNotExist { .. }) => {
+ if let Some(system_name) = system_name {
+ return if system_tables::is_registered(&system_name) {
+ Err(plan_datafusion_err!(
+ "Cannot read system table `${system_name}`:
base table `{}` does not exist",
+ object.table()
+ ))
+ } else {
+ Ok(None)
+ };
+ }
if branch.is_some() {
return Ok(None);
}
@@ -924,19 +973,16 @@ impl SchemaProvider for PaimonSchemaProvider {
return false;
}
};
- if let Some(system_name) = object.system_table() {
- if !system_tables::is_registered(system_name) {
- return false;
- }
- }
-
let catalog = Arc::clone(&self.catalog);
+ let dynamic_options = Arc::clone(&self.dynamic_options);
let identifier = Identifier::new(self.database.clone(),
object.table().to_string());
let branch = object.branch().map(str::to_string);
let is_branches_table = object
.system_table()
.is_some_and(|name| name.eq_ignore_ascii_case("branches"));
- let has_system_suffix = object.system_table().is_some();
+ let system_name = object.system_table().map(str::to_string);
+ let has_system_suffix = system_name.is_some();
+ let table_name = format!("{}.{}", self.database, name);
let engines: HashMap<PaimonTableType, Arc<dyn TableEngineResolver>> =
self
.table_engines
.read()
@@ -950,37 +996,57 @@ impl SchemaProvider for PaimonSchemaProvider {
}
Ok(paimon::catalog::LoadedTable::External(external)) => {
let declared = external.declared();
- // Paimon-only; `table()` rejects them here too.
- if branch.is_some() || has_system_suffix {
+ if branch.is_some() {
return false;
}
match engines.get(&declared) {
- Some(resolver) => match resolver
- .resolve_table(&EngineTableRequest::new(
+ Some(resolver) => {
+ let request = EngineTableRequest::new(
identifier.database().to_string(),
identifier.object().to_string(),
declared,
- ))
- .await
- {
- Ok(table) => table.is_some(),
- // Report failures as existing so `table()`
- // surfaces the real error.
- Err(err) => {
- log::warn!(
- "failed to probe engine table
existence for '{}': {err}",
- identifier.full_name()
- );
- true
+ );
+ let resolved = async {
+ if let Some(system_name) = system_name {
+ let opts = dynamic_options
+ .read()
+ .unwrap_or_else(|e| e.into_inner())
+ .clone();
+ paimon::spec::CoreOptions::new(&opts)
+
.ensure_engine_can_serve(&table_name)
+ .map_err(to_datafusion_error)?;
+ let system_name =
system_name.to_ascii_lowercase();
+ resolver
+ .resolve_metadata_table(&request,
&system_name)
+ .await
+ } else {
+ resolver.resolve_table(&request).await
+ }
}
- },
- // `table()` returns a metadata-only provider in
this
- // case, so the SchemaProvider existence contract
- // requires the same answer here.
- None => true,
+ .await;
+ match resolved {
+ Ok(table) => table.is_some(),
+ // Let `table()` report the engine error.
+ Err(err) => {
+ log::warn!(
+ "failed to probe engine table
existence for '{}': {err}",
+ table_name
+ );
+ true
+ }
+ }
+ }
+ // Only base tables have a metadata-only fallback.
+ None => !has_system_suffix,
}
}
Ok(paimon::catalog::LoadedTable::Paimon(table)) => {
+ if system_name
+ .as_deref()
+ .is_some_and(|name|
!system_tables::is_registered(name))
+ {
+ return false;
+ }
if let Some(branch) = branch.as_deref() {
if is_branches_table {
return true;
@@ -991,7 +1057,7 @@ impl SchemaProvider for PaimonSchemaProvider {
}
}
Err(paimon::Error::TableNotExist { .. }) => {
- if branch.is_some() {
+ if branch.is_some() || has_system_suffix {
return false;
}
match catalog.get_view(&identifier).await {
@@ -1032,6 +1098,15 @@ impl SchemaProvider for PaimonSchemaProvider {
}
fn deregister_table(&self, name: &str) -> DFResult<Option<Arc<dyn
TableProvider>>> {
+ // View DDL can reach this callback for metadata tables.
+ let object = system_tables::parse_object_name_for_datafusion(name)?;
+ if object.system_table().is_some() {
+ return Err(plan_datafusion_err!(
+ "Metadata table '{}.{}' is read-only",
+ self.database,
+ name
+ ));
+ }
let catalog = Arc::clone(&self.catalog);
let identifier = Identifier::new(self.database.clone(), name);
block_on_with_runtime(
diff --git a/crates/integrations/datafusion/src/system_tables/mod.rs
b/crates/integrations/datafusion/src/system_tables/mod.rs
index 7cbafb44..0a206fdf 100644
--- a/crates/integrations/datafusion/src/system_tables/mod.rs
+++ b/crates/integrations/datafusion/src/system_tables/mod.rs
@@ -158,15 +158,11 @@ pub(crate) fn provider_for_table(
.map(Some)
}
-/// 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
+/// Builds a system-table provider from a loaded Paimon table.
pub(crate) async fn load(
catalog: Arc<dyn Catalog>,
- database: String,
+ identifier: Identifier,
+ mut table: Table,
object: ParsedObjectName,
system_name: String,
dynamic_options: HashMap<String, String>,
@@ -185,36 +181,25 @@ pub(crate) async fn load(
paimon::spec::CoreOptions::new(&dynamic_options)
.ensure_read_authorized()
.map_err(to_datafusion_error)?;
- let identifier = Identifier::new(database, object.table().to_string());
- match catalog.get_table(&identifier).await {
- Ok(mut table) => {
- crate::table_loader::ensure_paimon_served(&table, &identifier)?;
- paimon::spec::CoreOptions::new(table.schema().options())
- .ensure_read_authorized()
+ crate::table_loader::ensure_paimon_served(&table, &identifier)?;
+ paimon::spec::CoreOptions::new(table.schema().options())
+ .ensure_read_authorized()
+ .map_err(to_datafusion_error)?;
+ if let Some(branch) = object.branch() {
+ if !system_name.eq_ignore_ascii_case("branches") {
+ table = table
+ .copy_with_branch(branch)
+ .await
.map_err(to_datafusion_error)?;
- if let Some(branch) = object.branch() {
- if !system_name.eq_ignore_ascii_case("branches") {
- table = table
- .copy_with_branch(branch)
- .await
- .map_err(to_datafusion_error)?;
- }
- }
- if system_name.eq_ignore_ascii_case("audit_log") &&
!dynamic_options.is_empty() {
- table = table
- .copy_with_time_travel(dynamic_options)
- .await
- .map_err(to_datafusion_error)?;
- }
- provider_for_table(catalog, identifier, table, &system_name)
}
- Err(paimon::Error::TableNotExist { .. }) =>
Err(DataFusionError::Plan(format!(
- "Cannot read system table `${system_name}`: \
- base table `{}` does not exist",
- object.table()
- ))),
- Err(e) => Err(to_datafusion_error(e)),
}
+ if system_name.eq_ignore_ascii_case("audit_log") &&
!dynamic_options.is_empty() {
+ table = table
+ .copy_with_time_travel(dynamic_options)
+ .await
+ .map_err(to_datafusion_error)?;
+ }
+ provider_for_table(catalog, identifier, table, &system_name)
}
#[cfg(test)]
diff --git a/crates/integrations/datafusion/tests/table_type_routing.rs
b/crates/integrations/datafusion/tests/table_type_routing.rs
index 3e979637..d29a3938 100644
--- a/crates/integrations/datafusion/tests/table_type_routing.rs
+++ b/crates/integrations/datafusion/tests/table_type_routing.rs
@@ -16,14 +16,17 @@
// under the License.
use std::collections::HashMap;
+use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use async_trait::async_trait;
use datafusion::arrow::array::{Array, Int32Array, StringArray};
use datafusion::arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
use datafusion::arrow::record_batch::RecordBatch;
+use datafusion::assert_batches_eq;
use datafusion::datasource::{MemTable, TableProvider};
use datafusion::error::{DataFusionError, Result as DFResult};
+use futures::TryStreamExt;
use paimon::catalog::{Catalog, Database, Identifier, LoadedTable};
use paimon::spec::{Schema as PaimonSchema, SchemaChange, TableType};
use paimon::table::Table;
@@ -298,19 +301,78 @@ impl TableEngineResolver for FakeEngineResolver {
}
}
+#[derive(Debug, Default)]
+struct MetadataEngineResolver {
+ calls: AtomicUsize,
+}
+
+#[async_trait]
+impl TableEngineResolver for MetadataEngineResolver {
+ async fn resolve_table(
+ &self,
+ request: &EngineTableRequest,
+ ) -> DFResult<Option<Arc<dyn TableProvider>>> {
+ FakeEngineResolver.resolve_table(request).await
+ }
+
+ async fn resolve_metadata_table(
+ &self,
+ request: &EngineTableRequest,
+ metadata_table: &str,
+ ) -> DFResult<Option<Arc<dyn TableProvider>>> {
+ self.calls.fetch_add(1, Ordering::Relaxed);
+ assert_eq!(request.database, DB);
+ assert_eq!(request.declared, TableType::IcebergTable);
+ if request.table != "it" {
+ return Ok(None);
+ }
+ match metadata_table {
+ "snapshots" | "manifests" | "history" => {}
+ "broken" => {
+ return Err(DataFusionError::Execution(
+ "metadata backend exploded".to_string(),
+ ));
+ }
+ _ => return Ok(None),
+ }
+ let schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("snapshot_id", DataType::Int32, false),
+ Field::new("operation", DataType::Utf8, false),
+ Field::new("metadata_table", DataType::Utf8, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from(vec![2, 4])),
+ Arc::new(StringArray::from(vec!["append", "overwrite"])),
+ Arc::new(StringArray::from(vec![metadata_table; 2])),
+ ],
+ )?;
+ Ok(Some(Arc::new(MemTable::try_new(
+ schema,
+ vec![vec![batch]],
+ )?)))
+ }
+}
+
struct TestEnv {
_paimon_dir: TempDir,
ctx: SQLContext,
+ catalog: Arc<FileSystemCatalog>,
}
async fn setup() -> TestEnv {
+ setup_with_engine(Some(Arc::new(FakeEngineResolver))).await
+}
+
+async fn setup_with_engine(resolver: Option<Arc<dyn TableEngineResolver>>) ->
TestEnv {
let paimon_dir = TempDir::new().unwrap();
let warehouse = format!("file://{}", paimon_dir.path().display());
let mut options = Options::new();
options.set(CatalogOptions::WAREHOUSE, warehouse);
let fs_catalog = Arc::new(FileSystemCatalog::new(options).unwrap());
let typed_catalog = Arc::new(TypedTestCatalog {
- inner: fs_catalog,
+ inner: Arc::clone(&fs_catalog),
declared_types: HashMap::from([
("it".to_string(), TableType::IcebergTable),
("ghost".to_string(), TableType::IcebergTable),
@@ -334,16 +396,15 @@ async fn setup() -> TestEnv {
ctx.sql(&stmt).await.unwrap().collect().await.unwrap();
}
- ctx.register_catalog_table_engine(
- CATALOG,
- TableType::IcebergTable,
- Arc::new(FakeEngineResolver),
- )
- .unwrap();
+ if let Some(resolver) = resolver {
+ ctx.register_catalog_table_engine(CATALOG, TableType::IcebergTable,
resolver)
+ .unwrap();
+ }
TestEnv {
_paimon_dir: paimon_dir,
ctx,
+ catalog: fs_catalog,
}
}
@@ -557,17 +618,363 @@ async fn writes_to_routed_tables_fail_closed() {
}
#[tokio::test]
-async fn system_tables_on_routed_tables_error() {
+async fn metadata_tables_without_engine_support_are_unavailable() {
let env = setup().await;
- let Err(err) = env
+ let schema = env.ctx.ctx().catalog(CATALOG).unwrap().schema(DB).unwrap();
+ for name in ["it$snapshots", "it$history"] {
+ assert!(schema.table(name).await.unwrap().is_none(), "{name}");
+ assert!(!schema.table_exist(name), "{name}");
+ let err = env
+ .ctx
+ .sql(&format!("SELECT * FROM {CATALOG}.{DB}.\"{name}\""))
+ .await
+ .unwrap_err();
+ assert!(err.to_string().contains("not found"), "{err}");
+ }
+}
+
+#[tokio::test]
+async fn
metadata_tables_without_a_registered_engine_report_the_declared_type() {
+ let env = setup_with_engine(None).await;
+ let schema = env.ctx.ctx().catalog(CATALOG).unwrap().schema(DB).unwrap();
+ for name in ["it$snapshots", "it$history"] {
+ assert!(!schema.table_exist(name), "{name}");
+ let err = schema.table(name).await.unwrap_err();
+ assert!(
+ err.to_string()
+ .contains("no table engine is registered for 'iceberg-table'"),
+ "{err}"
+ );
+ }
+}
+
+#[tokio::test]
+async fn engine_metadata_tables_support_projection_filters_and_aggregation() {
+ let env =
setup_with_engine(Some(Arc::new(MetadataEngineResolver::default()))).await;
+ for suffix in ["snapshots", "manifests", "History"] {
+ let batches = env
+ .ctx
+ .sql(&format!(
+ "SELECT operation, snapshot_id FROM
{CATALOG}.{DB}.\"it${suffix}\" \
+ WHERE snapshot_id > 2"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_batches_eq!(
+ [
+ "+-----------+-------------+",
+ "| operation | snapshot_id |",
+ "+-----------+-------------+",
+ "| overwrite | 4 |",
+ "+-----------+-------------+",
+ ],
+ &batches
+ );
+ let batches = env
+ .ctx
+ .sql(&format!(
+ "SELECT metadata_table FROM {CATALOG}.{DB}.\"it${suffix}\" \
+ WHERE snapshot_id > 2"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ string_column_values(&batches, "metadata_table"),
+ vec![suffix.to_ascii_lowercase()]
+ );
+ let batches = env
+ .ctx
+ .sql(&format!(
+ "SELECT count(*) AS n FROM {CATALOG}.{DB}.\"it${suffix}\""
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_batches_eq!(["+---+", "| n |", "+---+", "| 2 |", "+---+"],
&batches);
+ }
+}
+
+#[tokio::test]
+async fn metadata_table_existence_matches_the_resolver() {
+ let env =
setup_with_engine(Some(Arc::new(MetadataEngineResolver::default()))).await;
+ let schema = env.ctx.ctx().catalog(CATALOG).unwrap().schema(DB).unwrap();
+ for (name, exists) in [
+ ("it$snapshots", true),
+ ("it$history", true),
+ ("it$SNAPSHOTS", true),
+ ("it$History", true),
+ ("it$unknown", false),
+ ("ghost$history", false),
+ ("missing$history", false),
+ ("pt$history", false),
+ ] {
+ assert_eq!(
+ schema.table(name).await.unwrap().is_some(),
+ exists,
+ "{name}"
+ );
+ assert_eq!(schema.table_exist(name), exists, "{name}");
+ }
+}
+
+#[tokio::test]
+async fn metadata_engine_errors_are_surfaced() {
+ let env =
setup_with_engine(Some(Arc::new(MetadataEngineResolver::default()))).await;
+ let schema = env.ctx.ctx().catalog(CATALOG).unwrap().schema(DB).unwrap();
+ assert!(schema.table_exist("it$broken"));
+ let err = env
.ctx
- .sql(&format!("SELECT * FROM {CATALOG}.{DB}.\"it$snapshots\""))
+ .sql(&format!("SELECT * FROM {CATALOG}.{DB}.\"it$broken\""))
.await
- else {
- panic!("system table on a routed table must fail");
- };
- let msg = err.to_string();
- assert!(msg.contains("cannot be read as a Paimon table"), "{msg}");
+ .unwrap_err();
+ assert!(
+ err.to_string().contains("metadata backend exploded"),
+ "{err}"
+ );
+}
+
+#[tokio::test]
+async fn paimon_system_tables_keep_the_paimon_provider() {
+ let env =
setup_with_engine(Some(Arc::new(MetadataEngineResolver::default()))).await;
+ let schema = env.ctx.ctx().catalog(CATALOG).unwrap().schema(DB).unwrap();
+ assert!(schema.table_exist("pt$snapshots"));
+ let batches = env
+ .ctx
+ .sql(&format!(
+ "SELECT snapshot_id FROM {CATALOG}.{DB}.\"pt$snapshots\" ORDER BY
snapshot_id"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_batches_eq!(
+ [
+ "+-------------+",
+ "| snapshot_id |",
+ "+-------------+",
+ "| 1 |",
+ "| 2 |",
+ "+-------------+",
+ ],
+ &batches
+ );
+}
+
+#[tokio::test]
+async fn routed_metadata_tables_are_read_only() {
+ let env =
setup_with_engine(Some(Arc::new(MetadataEngineResolver::default()))).await;
+ let err = async {
+ env.ctx
+ .sql(&format!(
+ "INSERT INTO {CATALOG}.{DB}.\"it$history\" VALUES (5,
'append', 'history')"
+ ))
+ .await?
+ .collect()
+ .await
+ }
+ .await
+ .unwrap_err();
+ assert!(err.to_string().contains("write is not supported"), "{err}");
+}
+
+#[tokio::test]
+async fn routed_metadata_tables_are_views() {
+ use datafusion::datasource::TableType as DFTableType;
+
+ let env =
setup_with_engine(Some(Arc::new(MetadataEngineResolver::default()))).await;
+ let schema = env.ctx.ctx().catalog(CATALOG).unwrap().schema(DB).unwrap();
+ let metadata = schema.table("it$history").await.unwrap().unwrap();
+ assert_eq!(metadata.table_type(), DFTableType::View);
+ let base = schema.table("it").await.unwrap().unwrap();
+ assert_eq!(base.table_type(), DFTableType::Base);
+
+ // Native DROP TABLE must reject metadata views.
+ let result = async {
+ env.ctx
+ .ctx()
+ .sql(&format!("DROP TABLE {CATALOG}.{DB}.\"it$history\""))
+ .await?
+ .collect()
+ .await
+ }
+ .await;
+ assert!(result.is_err(), "{result:?}");
+ assert!(schema.table_exist("it$history"));
+}
+
+#[tokio::test]
+async fn metadata_drop_view_preserves_physical_name_collisions() {
+ for name in ["it$history", "it$History", "pt$snapshots"] {
+ assert_metadata_ddl_preserves_physical_table(
+ name,
+ &format!("DROP VIEW {CATALOG}.{DB}.\"{name}\""),
+ )
+ .await;
+ }
+}
+
+#[tokio::test]
+async fn metadata_replace_view_preserves_physical_name_collisions() {
+ for name in ["it$history", "it$History", "pt$snapshots"] {
+ assert_metadata_ddl_preserves_physical_table(
+ name,
+ &format!("CREATE OR REPLACE VIEW {CATALOG}.{DB}.\"{name}\" AS
SELECT 42 AS id"),
+ )
+ .await;
+ }
+}
+
+async fn assert_metadata_ddl_preserves_physical_table(name: &str, ddl: &str) {
+ let env =
setup_with_engine(Some(Arc::new(MetadataEngineResolver::default()))).await;
+ env.ctx
+ .sql(&format!(
+ "CREATE TABLE {CATALOG}.{DB}.\"{name}\" (id INT NOT NULL)"
+ ))
+ .await
+ .unwrap();
+ let identifier = Identifier::new(DB, name);
+ let table = env.catalog.get_table(&identifier).await.unwrap();
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![Field::new(
+ "id",
+ DataType::Int32,
+ false,
+ )])),
+ vec![Arc::new(Int32Array::from(vec![7, 9]))],
+ )
+ .unwrap();
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ write.write_arrow_batch(&batch).await.unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+
+ // Use native DDL to bypass SQLContext guards.
+ let result = async { env.ctx.ctx().sql(ddl).await?.collect().await }.await;
+ assert!(
+ table.file_io().exists_dir(table.location()).await.unwrap(),
+ "{ddl} deleted the physical table directory"
+ );
+ let remaining = env.catalog.get_table(&identifier).await.unwrap();
+ let read = remaining.new_read_builder();
+ let plan = read.new_scan().plan().await.unwrap();
+ let batches = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(column_i32(&batches), vec![7, 9], "{ddl}");
+ let error = result.unwrap_err().to_string();
+ assert!(error.contains(name), "{ddl}: {error}");
+}
+
+#[tokio::test]
+async fn metadata_deregistration_reports_read_only() {
+ let env =
setup_with_engine(Some(Arc::new(MetadataEngineResolver::default()))).await;
+ let schema = env.ctx.ctx().catalog(CATALOG).unwrap().schema(DB).unwrap();
+ for name in ["it$history", "it$History", "it$unknown", "pt$snapshots"] {
+ let error = schema.deregister_table(name).unwrap_err().to_string();
+ assert!(error.contains("read-only"), "{name}: {error}");
+ assert!(error.contains(&format!("{DB}.{name}")), "{name}: {error}");
+ }
+}
+
+#[tokio::test]
+async fn native_drop_table_still_removes_a_regular_table() {
+ let env = setup().await;
+ let identifier = Identifier::new(DB, "pt");
+ let table = env.catalog.get_table(&identifier).await.unwrap();
+ env.ctx
+ .ctx()
+ .sql(&format!("DROP TABLE {CATALOG}.{DB}.pt"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert!(!table.file_io().exists_dir(table.location()).await.unwrap());
+ assert!(matches!(
+ env.catalog.get_table(&identifier).await,
+ Err(paimon::Error::TableNotExist { .. })
+ ));
+}
+
+#[tokio::test]
+async fn routed_metadata_tables_reject_branches_and_time_travel() {
+ let env =
setup_with_engine(Some(Arc::new(MetadataEngineResolver::default()))).await;
+ let schema = env.ctx.ctx().catalog(CATALOG).unwrap().schema(DB).unwrap();
+ let err = schema.table("it$branch_b1$history").await.unwrap_err();
+ assert!(
+ err.to_string().contains("branches are not supported"),
+ "{err}"
+ );
+ assert!(!schema.table_exist("it$branch_b1$history"));
+
+ env.ctx
+ .ctx()
+ .sql("SET datafusion.sql_parser.dialect = 'databricks'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ for clause in [
+ "VERSION AS OF 1",
+ "TIMESTAMP AS OF '2020-01-01 00:00:00'",
+ "FOR SYSTEM_TIME AS OF '2020-01-01 00:00:00'",
+ ] {
+ let err = env
+ .ctx
+ .ctx()
+ .sql(&format!(
+ "SELECT * FROM {CATALOG}.{DB}.\"it$history\" {clause}"
+ ))
+ .await
+ .unwrap_err();
+ assert!(
+ err.to_string().contains("time travel is not supported"),
+ "{err}"
+ );
+ }
+}
+
+#[tokio::test]
+async fn
metadata_resolution_enforces_session_options_before_calling_the_engine() {
+ for (option, value, error) in [
+ ("query-auth.enabled", "true", "query-auth"),
+ ("scan.version", "1", "time travel"),
+ ("incremental-between", "1,5", "incremental-between"),
+ ] {
+ let resolver = Arc::new(MetadataEngineResolver::default());
+ let env = setup_with_engine(Some(resolver.clone())).await;
+ env.ctx
+ .sql(&format!("SET 'paimon.{option}' = '{value}'"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let schema =
env.ctx.ctx().catalog(CATALOG).unwrap().schema(DB).unwrap();
+ let err = schema.table("it$history").await.unwrap_err();
+ assert!(err.to_string().contains(error), "{err}");
+ if option == "scan.version" {
+ assert!(err.to_string().contains("it$history"), "{err}");
+ }
+ assert!(schema.table_exist("it$history"));
+ assert_eq!(resolver.calls.load(Ordering::Relaxed), 0);
+ }
}
#[tokio::test]
