diff --git a/crates/integrations/datafusion/src/procedures.rs b/crates/integrations/datafusion/src/procedures.rs index 3be52319..d73e25e3 100644 --- a/crates/integrations/datafusion/src/procedures.rs +++ b/crates/integrations/datafusion/src/procedures.rs @@ -276,10 +276,7 @@ async fn get_table( ) -> DFResult { let table_str = require_arg(args, "table")?; let identifier = resolve_table_identifier(table_str, catalog_name)?; - catalog - .get_table(&identifier) - .await - .map_err(to_datafusion_error) + crate::table_loader::get_paimon_table(catalog, &identifier).await } fn managers(table: &Table) -> (SnapshotManager, TagManager) { diff --git a/crates/integrations/datafusion/src/sql_context.rs b/crates/integrations/datafusion/src/sql_context.rs index cf306f2c..74a0df45 100644 --- a/crates/integrations/datafusion/src/sql_context.rs +++ b/crates/integrations/datafusion/src/sql_context.rs @@ -1173,6 +1173,7 @@ impl SQLContext { Err(paimon::Error::TableNotExist { .. }) => return self.ctx.sql(sql).await, Err(e) => return Err(to_datafusion_error(e)), }; + crate::table_loader::ensure_paimon_served(&table, &identifier)?; let definition = crate::table::build_table_definition(&table)?; let schema = Arc::new(Schema::new(vec![ diff --git a/crates/integrations/datafusion/src/table/mod.rs b/crates/integrations/datafusion/src/table/mod.rs index 8a18d476..c55b97e7 100644 --- a/crates/integrations/datafusion/src/table/mod.rs +++ b/crates/integrations/datafusion/src/table/mod.rs @@ -121,6 +121,8 @@ impl PaimonTableProvider { table: Table, table_definition: Option, ) -> DFResult { + let identifier = table.identifier().clone(); + crate::table_loader::ensure_paimon_served(&table, &identifier)?; let fields = datafusion_read_fields(&table); let schema = datafusion_arrow_schema(&fields, true)?; Ok(Self { @@ -134,6 +136,8 @@ impl PaimonTableProvider { table: Table, blob_reader_registry: BlobReaderRegistry, ) -> DFResult { + let identifier = table.identifier().clone(); + crate::table_loader::ensure_paimon_served(&table, &identifier)?; blob_reader_registry .register_if_absent(table.location().to_string(), table.file_io().clone()); Self::try_new(table) @@ -144,6 +148,8 @@ impl PaimonTableProvider { blob_reader_registry: BlobReaderRegistry, table_definition: Option, ) -> DFResult { + let identifier = table.identifier().clone(); + crate::table_loader::ensure_paimon_served(&table, &identifier)?; blob_reader_registry .register_if_absent(table.location().to_string(), table.file_io().clone()); Self::try_new_with_table_definition(table, table_definition) diff --git a/crates/integrations/datafusion/src/table_loader.rs b/crates/integrations/datafusion/src/table_loader.rs index f1d61f9a..c72d0385 100644 --- a/crates/integrations/datafusion/src/table_loader.rs +++ b/crates/integrations/datafusion/src/table_loader.rs @@ -23,6 +23,38 @@ use paimon::table::Table; use crate::error::to_datafusion_error; +/// [`Catalog::get_table`] for paths that need a `Table` rather than a +/// [`paimon::catalog::LoadedTable`], rejecting an engine-served declared +/// type: `get_table` is a trait method, so a catalog outside this repository +/// can return a table for any type. +pub(crate) async fn get_paimon_table( + catalog: &Arc, + identifier: &Identifier, +) -> DFResult
{ + let table = catalog + .get_table(identifier) + .await + .map_err(to_datafusion_error)?; + ensure_paimon_served(&table, identifier)?; + Ok(table) +} + +/// The check from [`get_paimon_table`], for callers that already hold the +/// table or special-case the `get_table` error. +pub(crate) fn ensure_paimon_served(table: &Table, identifier: &Identifier) -> DFResult<()> { + let declared = paimon::spec::CoreOptions::new(table.schema().options()) + .table_type() + .map_err(to_datafusion_error)?; + if declared.requires_table_engine() { + return Err(DataFusionError::Plan(format!( + "table '{}' is declared '{}' and cannot be read as a Paimon table", + identifier.full_name(), + declared + ))); + } + Ok(()) +} + pub(crate) async fn load_table_for_read( catalog: &Arc, identifier: &Identifier, diff --git a/crates/integrations/datafusion/tests/table_type_routing.rs b/crates/integrations/datafusion/tests/table_type_routing.rs index f75ae8f3..3777ea97 100644 --- a/crates/integrations/datafusion/tests/table_type_routing.rs +++ b/crates/integrations/datafusion/tests/table_type_routing.rs @@ -150,6 +150,102 @@ impl Catalog for TypedTestCatalog { } } +#[derive(Debug)] +struct LegacyTestCatalog { + inner: Arc, +} + +#[async_trait] +impl Catalog for LegacyTestCatalog { + async fn list_databases(&self) -> PaimonResult> { + self.inner.list_databases().await + } + + async fn create_database( + &self, + name: &str, + ignore_if_exists: bool, + properties: HashMap, + ) -> PaimonResult<()> { + self.inner + .create_database(name, ignore_if_exists, properties) + .await + } + + async fn get_database(&self, name: &str) -> PaimonResult { + self.inner.get_database(name).await + } + + async fn drop_database( + &self, + name: &str, + ignore_if_not_exists: bool, + cascade: bool, + ) -> PaimonResult<()> { + self.inner + .drop_database(name, ignore_if_not_exists, cascade) + .await + } + + async fn get_table(&self, identifier: &Identifier) -> PaimonResult
{ + let (location, schema) = self.inner.fetch_table_schema(identifier).await?; + Ok(Table::new( + self.inner.file_io().clone(), + identifier.clone(), + location, + schema, + None, + )) + } + + async fn list_tables(&self, database_name: &str) -> PaimonResult> { + self.inner.list_tables(database_name).await + } + + async fn create_table( + &self, + identifier: &Identifier, + creation: PaimonSchema, + ignore_if_exists: bool, + ) -> PaimonResult<()> { + self.inner + .create_table(identifier, creation, ignore_if_exists) + .await + } + + async fn drop_table( + &self, + identifier: &Identifier, + ignore_if_not_exists: bool, + ) -> PaimonResult<()> { + self.inner + .drop_table(identifier, ignore_if_not_exists) + .await + } + + async fn rename_table( + &self, + from: &Identifier, + to: &Identifier, + ignore_if_not_exists: bool, + ) -> PaimonResult<()> { + self.inner + .rename_table(from, to, ignore_if_not_exists) + .await + } + + async fn alter_table( + &self, + identifier: &Identifier, + changes: Vec, + ignore_if_not_exists: bool, + ) -> PaimonResult<()> { + self.inner + .alter_table(identifier, changes, ignore_if_not_exists) + .await + } +} + #[derive(Debug)] struct FakeEngineResolver; @@ -778,3 +874,271 @@ async fn an_external_type_without_an_engine_says_so() { assert!(msg.contains("no table engine is registered"), "{msg}"); assert!(msg.contains("iceberg-table"), "{msg}"); } + +async fn legacy_catalog_with_iceberg_table() -> (TempDir, Arc) { + 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()); + fs_catalog + .create_database(DB, false, HashMap::new()) + .await + .unwrap(); + let schema = PaimonSchema::builder() + .column( + "id", + paimon::spec::DataType::Int(paimon::spec::IntType::new()), + ) + .column( + "pt", + paimon::spec::DataType::Int(paimon::spec::IntType::new()), + ) + .partition_keys(["pt"]) + .option("type", "iceberg-table") + .build() + .unwrap(); + fs_catalog + .create_table(&Identifier::new(DB, "it"), schema, false) + .await + .unwrap(); + ( + paimon_dir, + Arc::new(LegacyTestCatalog { inner: fs_catalog }), + ) +} + +#[tokio::test] +async fn the_default_load_table_classifies_for_a_catalog_that_only_has_get_table() { + let (_dir, catalog) = legacy_catalog_with_iceberg_table().await; + + let loaded = catalog + .load_table(&Identifier::new(DB, "it")) + .await + .unwrap(); + assert!( + matches!(loaded, LoadedTable::External(ref e) if e.declared() == TableType::IcebergTable), + "{loaded:?}" + ); +} + +#[tokio::test] +async fn a_legacy_catalog_cannot_serve_an_external_table_as_paimon() { + let (_dir, catalog) = legacy_catalog_with_iceberg_table().await; + let mut ctx = SQLContext::new(); + ctx.register_catalog(CATALOG, catalog).await.unwrap(); + + let Err(err) = ctx.sql(&format!("SELECT * FROM {CATALOG}.{DB}.it")).await else { + panic!("a legacy catalog must not serve an iceberg table as Paimon"); + }; + let msg = err.to_string(); + assert!(msg.contains("no table engine is registered"), "{msg}"); + assert!(msg.contains("iceberg-table"), "{msg}"); +} + +#[tokio::test] +async fn a_hand_built_external_table_is_rejected_by_the_provider() { + let (_dir, catalog) = legacy_catalog_with_iceberg_table().await; + + let table = catalog.get_table(&Identifier::new(DB, "it")).await.unwrap(); + let Err(err) = paimon_datafusion::PaimonTableProvider::try_new(table) else { + panic!("a table declared iceberg-table must not become a Paimon provider"); + }; + let msg = err.to_string(); + assert!(msg.contains("cannot be read as a Paimon table"), "{msg}"); + assert!(msg.contains("iceberg-table"), "{msg}"); +} + +async fn legacy_sql_context() -> (TempDir, SQLContext) { + let (dir, catalog) = legacy_catalog_with_iceberg_table().await; + let mut ctx = SQLContext::new(); + ctx.register_catalog(CATALOG, catalog).await.unwrap(); + (dir, ctx) +} + +#[tokio::test] +async fn a_legacy_catalog_refuses_every_destructive_statement() { + let (_dir, ctx) = legacy_sql_context().await; + + for sql in [ + format!("INSERT INTO {CATALOG}.{DB}.it VALUES (1)"), + format!("INSERT OVERWRITE {CATALOG}.{DB}.it PARTITION (pt = 1) VALUES (1)"), + format!("UPDATE {CATALOG}.{DB}.it SET id = 2"), + format!("DELETE FROM {CATALOG}.{DB}.it"), + format!("TRUNCATE TABLE {CATALOG}.{DB}.it"), + format!("CALL {CATALOG}.sys.create_tag(table => '{DB}.it', tag => 't1')"), + ] { + let outcome = match ctx.sql(&sql).await { + Err(err) => Err(err), + Ok(df) => df.collect().await.map(|_| ()), + }; + let Err(err) = outcome else { + panic!("must not run against an iceberg-table: {sql}"); + }; + let msg = err.to_string(); + assert!( + msg.contains("iceberg-table") || msg.contains("no table engine is registered"), + "{sql} -> {msg}" + ); + } +} + +#[tokio::test] +async fn a_legacy_catalog_refuses_system_tables() { + let (_dir, ctx) = legacy_sql_context().await; + + let outcome = match ctx + .sql(&format!("SELECT * FROM {CATALOG}.{DB}.\"it$snapshots\"")) + .await + { + Err(err) => Err(err), + Ok(df) => df.collect().await.map(|_| ()), + }; + let Err(err) = outcome else { + panic!("a system table on an iceberg-table must not resolve"); + }; + let msg = err.to_string(); + assert!(msg.contains("iceberg-table"), "{msg}"); +} + +#[tokio::test] +async fn a_legacy_catalog_refuses_paimon_reads_and_writes_in_core() { + let (_dir, catalog) = legacy_catalog_with_iceberg_table().await; + let table = catalog.get_table(&Identifier::new(DB, "it")).await.unwrap(); + + let read = table.new_read_builder().new_read(); + assert!(read.is_err(), "core read must be refused"); + + let write = paimon::table::WriteBuilder::new(&table).new_write(); + assert!(write.is_err(), "core write must be refused"); + + let commit = paimon::table::WriteBuilder::new(&table).new_commit(); + assert!( + commit.commit(Vec::new()).await.is_err(), + "core commit must be refused" + ); + assert!( + commit.truncate_table_with_identifier(1).await.is_err(), + "core truncate must be refused" + ); + assert!( + commit.abort(&[]).await.is_err(), + "core abort must be refused" + ); + + let incremental = table + .new_read_builder() + .new_incremental_scan(paimon::table::IncrementalScanMode::Delta, 0, 1) + .plan() + .await; + assert!( + incremental.is_err(), + "core incremental scan must be refused" + ); +} + +#[tokio::test] +async fn a_legacy_catalog_refuses_scan_planning_rather_than_reporting_empty() { + let (_dir, catalog) = legacy_catalog_with_iceberg_table().await; + let table = catalog.get_table(&Identifier::new(DB, "it")).await.unwrap(); + + let plan = table.new_read_builder().new_scan().plan().await; + assert!( + plan.is_err(), + "planning must be refused, not answered with an empty plan" + ); + + let stats = table.partition_stats().await; + assert!(stats.is_err(), "partition stats must be refused"); +} + +#[tokio::test] +async fn a_legacy_catalog_refuses_the_infallible_commit_path() { + let (_dir, catalog) = legacy_catalog_with_iceberg_table().await; + let table = catalog.get_table(&Identifier::new(DB, "it")).await.unwrap(); + + let commit = paimon::table::WriteBuilder::new(&table).new_commit(); + assert!( + commit.truncate_table().await.is_err(), + "truncate must not write Paimon metadata over foreign data" + ); +} + +#[tokio::test] +async fn a_dynamic_copy_cannot_launder_the_declared_type() { + let (_dir, catalog) = legacy_catalog_with_iceberg_table().await; + let table = catalog.get_table(&Identifier::new(DB, "it")).await.unwrap(); + + let copied = + table.copy_with_options(HashMap::from([("type".to_string(), "table".to_string())])); + assert!( + copied.new_read_builder().new_read().is_err(), + "an override of 'type' must not re-route foreign data through the Paimon reader" + ); + assert!( + paimon::table::WriteBuilder::new(&copied) + .new_write() + .is_err(), + "an override of 'type' must not open a Paimon write on foreign data" + ); +} + +#[tokio::test] +async fn a_legacy_catalog_refuses_show_create() { + let (_dir, ctx) = legacy_sql_context().await; + + let outcome = match ctx + .sql(&format!("SHOW CREATE TABLE {CATALOG}.{DB}.it")) + .await + { + Err(err) => Err(err), + Ok(df) => df.collect().await.map(|_| ()), + }; + let Err(err) = outcome else { + panic!("SHOW CREATE must not emit Paimon DDL for an iceberg-table"); + }; + let msg = err.to_string(); + assert!(msg.contains("iceberg-table"), "{msg}"); +} + +#[tokio::test] +async fn a_branch_copy_cannot_launder_the_declared_type() { + let (_dir, catalog) = legacy_catalog_with_iceberg_table().await; + let table = catalog.get_table(&Identifier::new(DB, "it")).await.unwrap(); + + assert!( + table.copy_with_branch("b1").await.is_err(), + "a branch copy must not shed the declared type" + ); + assert!( + table + .copy_with_time_travel(HashMap::from([( + "scan.snapshot-id".to_string(), + "1".to_string(), + )])) + .await + .is_err(), + "time travel must not read Paimon snapshot paths of foreign data" + ); +} + +#[tokio::test] +async fn a_rejected_external_table_does_not_pollute_the_blob_registry() { + let (_dir, catalog) = legacy_catalog_with_iceberg_table().await; + let table = catalog.get_table(&Identifier::new(DB, "it")).await.unwrap(); + let location = table.location().to_string(); + + let registry = paimon_datafusion::BlobReaderRegistry::default(); + let built = paimon_datafusion::PaimonTableProvider::try_new_with_blob_reader_registry( + table, + registry.clone(), + ); + assert!( + built.is_err(), + "an iceberg-table must not become a provider" + ); + assert!( + registry.resolve(&format!("{location}/blob/x")).is_none(), + "a rejected table must leave no registration behind" + ); +} diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs index b2a5cf23..2b91ff19 100644 --- a/crates/paimon/src/catalog/mod.rs +++ b/crates/paimon/src/catalog/mod.rs @@ -373,16 +373,26 @@ pub trait Catalog: Send + Sync { /// Load a table, or classify it as [`LoadedTable::External`] when this /// reader cannot construct it. One metadata round-trip either way, and the - /// outcome depends only on the table's own metadata. The default - /// implementation always constructs, for catalogs without a table-type - /// concept. + /// outcome depends only on the table's own metadata. + /// + /// The default implementation classifies from the constructed table, so a + /// catalog that only implements [`Catalog::get_table`] still fails closed. + /// Override it to classify before construction and skip the token and + /// FileIO work an external table does not need. /// /// # Errors - /// Same as [`Catalog::get_table`]. + /// Everything [`Catalog::get_table`] returns, plus + /// [`crate::Error::Unsupported`] for an unknown declared type, or for an + /// external table whose stored options no engine can honor (query + /// authorization, unsupported scan options, time travel). async fn load_table(&self, identifier: &Identifier) -> Result { - Ok(LoadedTable::Paimon(Box::new( - self.get_table(identifier).await?, - ))) + let table = self.get_table(identifier).await?; + let options = crate::spec::CoreOptions::new(table.schema().options()); + let declared = options.table_type()?; + if declared.requires_table_engine() { + return LoadedTable::external(declared, &options, &identifier.full_name()); + } + Ok(LoadedTable::Paimon(Box::new(table))) } /// List table names in a database. System tables are not listed. @@ -514,8 +524,16 @@ pub trait Catalog: Send + Sync { /// `AbstractCatalog.listPartitions`. Catalogs with metastore-tracked /// partitions (e.g. `RESTCatalog`) override to return audit fields too. async fn list_partitions(&self, identifier: &Identifier) -> Result> { - let table = self.get_table(identifier).await?; - list_partitions_from_file_system(&table).await + match self.load_table(identifier).await? { + LoadedTable::Paimon(table) => list_partitions_from_file_system(&table).await, + LoadedTable::External(external) => Err(Error::Unsupported { + message: format!( + "table '{}' is declared '{}', so it has no Paimon partitions to list", + identifier.full_name(), + external.declared() + ), + }), + } } /// Like [`Self::list_partitions`] but paged. Default impl ignores diff --git a/crates/paimon/src/spec/core_options.rs b/crates/paimon/src/spec/core_options.rs index cc57ccef..b17d3168 100644 --- a/crates/paimon/src/spec/core_options.rs +++ b/crates/paimon/src/spec/core_options.rs @@ -519,10 +519,39 @@ impl<'a> CoreOptions<'a> { .unwrap_or(false) } - /// Fail closed when `query-auth.enabled` is set: this client can't enforce the row - /// filter / column masking, so refuse to read. Call at every read boundary (build, - /// plan, materialize) so no binding fast-path can bypass it. + /// Fail closed at every storage boundary (build, plan, materialize): refuses a + /// `query-auth.enabled` table — this client can't enforce its row filter / column + /// masking — and a table whose declared type needs an engine of its own, which + /// this client would misread as Paimon. pub fn ensure_read_authorized(&self) -> crate::Result<()> { + self.ensure_query_auth_absent()?; + let declared = self.table_type()?; + if declared.requires_table_engine() { + return Err(crate::Error::Unsupported { + message: format!( + "a table declared '{declared}' cannot be served as a Paimon table" + ), + }); + } + Ok(()) + } + + /// Type-only half of [`Self::ensure_read_authorized`], for paths that must + /// not touch an engine-served table's storage but stay usable under + /// `query-auth` (e.g. best-effort cleanup). + pub(crate) fn ensure_type_paimon_served(&self, full_name: &str) -> crate::Result<()> { + let declared = self.table_type()?; + if declared.requires_table_engine() { + return Err(crate::Error::Unsupported { + message: format!( + "table '{full_name}' is declared '{declared}' and cannot be served as a Paimon table" + ), + }); + } + Ok(()) + } + + fn ensure_query_auth_absent(&self) -> crate::Result<()> { if self.query_auth_enabled() { return Err(crate::Error::Unsupported { message: "reading a table with 'query-auth.enabled' = true is not supported: \ @@ -933,7 +962,7 @@ impl<'a> CoreOptions<'a> { /// Both the table's stored options and a session's options go through /// here, so neither source can skip a check the other applies. pub fn ensure_engine_can_serve(&self, full_name: &str) -> crate::Result<()> { - self.ensure_read_authorized()?; + self.ensure_query_auth_absent()?; self.validate_scan_options()?; if self.has_time_travel_selector() { return Err(crate::Error::Unsupported { diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index 0b52571b..635e188e 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -19,7 +19,7 @@ use crate::spec::core_options::{ first_row_supports_changelog_producer, ChangelogProducer, CoreOptions, MergeEngine, BLOB_DESCRIPTOR_FIELD_OPTION, BLOB_FIELD_OPTION, BLOB_VIEW_FIELD_OPTION, BUCKET_KEY_OPTION, CHANGELOG_PRODUCER_OPTION, POSTPONE_BUCKET, QUERY_AUTH_ENABLED_OPTION, SEQUENCE_FIELD_OPTION, - TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION, + TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION, TABLE_TYPE_OPTION, }; use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType, VarCharType}; use crate::spec::{ @@ -144,11 +144,17 @@ impl TableSchema { /// Create a copy of this schema with extra options merged in. /// - /// A stored `query-auth.enabled = true` can't be turned off by a dynamic override. + /// A stored `query-auth.enabled = true` can't be turned off by a dynamic + /// override, and the declared `type` can't be changed by one: an override + /// could re-route foreign data through the Paimon reader. pub fn copy_with_options(&self, mut extra: HashMap) -> Self { if self.core_options().query_auth_enabled() { extra.insert(QUERY_AUTH_ENABLED_OPTION.to_string(), "true".to_string()); } + match self.options.get(TABLE_TYPE_OPTION) { + Some(declared) => extra.insert(TABLE_TYPE_OPTION.to_string(), declared.clone()), + None => extra.remove(TABLE_TYPE_OPTION), + }; let mut new_schema = self.clone(); new_schema.options.extend(extra); new_schema diff --git a/crates/paimon/src/table/incremental_scan.rs b/crates/paimon/src/table/incremental_scan.rs index cdfc15fa..74727883 100644 --- a/crates/paimon/src/table/incremental_scan.rs +++ b/crates/paimon/src/table/incremental_scan.rs @@ -244,6 +244,7 @@ impl<'a> IncrementalScan<'a> { } pub async fn plan(&self) -> crate::Result { + crate::spec::CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?; let mode = self.resolve_mode(); self.validate_snapshot_range(mode).await?; if self.start_exclusive == self.end_inclusive { diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index 163bd0b1..b415f0dd 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -452,6 +452,9 @@ impl Table { extra: HashMap, strict: bool, ) -> Result { + // Resolution reads Paimon snapshot paths, so refuse before any IO. + CoreOptions::new(self.schema.options()) + .ensure_type_paimon_served(&self.identifier.full_name())?; let mut table = self.copy_with_options(extra); // Reject unimplemented scan options on the merged view before any IO, so // both table-level and per-read options are covered. @@ -482,6 +485,10 @@ impl Table { } pub async fn copy_with_branch(&self, branch_name: &str) -> Result { + // The branch schema replaces this one wholesale and could drop the + // declared type, so refuse before any branch I/O. + CoreOptions::new(self.schema.options()) + .ensure_type_paimon_served(&self.identifier.full_name())?; let branch = if branch_name.trim().is_empty() { return Err(crate::Error::DataInvalid { message: "Branch name cannot be empty.".to_string(), diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 39757ced..f7641f97 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -700,6 +700,8 @@ impl TableCommit { /// files or storage errors are ignored so abort cleanup never masks the /// original write failure. pub async fn abort(&self, commit_messages: &[CommitMessage]) -> Result<()> { + CoreOptions::new(self.table.schema().options()) + .ensure_type_paimon_served(&self.table.identifier().full_name())?; self.table.ensure_not_branch_reference_for_write()?; for message in commit_messages { @@ -3135,6 +3137,13 @@ fn rand_f64() -> f64 { #[cfg(test)] mod tests { use super::*; + + #[tokio::test] + async fn abort_still_cleans_up_for_a_query_auth_table() { + let table = crate::table::query_auth_table(); + let commit = crate::table::WriteBuilder::new(&table).new_commit(); + commit.abort(&[]).await.unwrap(); + } use crate::catalog::Identifier; use crate::io::FileIOBuilder; use crate::spec::stats::BinaryTableStats; diff --git a/crates/paimon/src/table/write_builder.rs b/crates/paimon/src/table/write_builder.rs index a792e783..7f00059d 100644 --- a/crates/paimon/src/table/write_builder.rs +++ b/crates/paimon/src/table/write_builder.rs @@ -205,6 +205,16 @@ impl<'a> PaimonWriteBuilder<'a> { } pub(super) fn ensure_table_write_allowed(table: &Table) -> crate::Result<()> { + let options = crate::spec::CoreOptions::new(table.schema().options()); + let declared = options.table_type()?; + if declared.requires_table_engine() { + return Err(crate::Error::Unsupported { + message: format!( + "table '{}' is declared '{declared}' and cannot be written as a Paimon table", + table.identifier().full_name() + ), + }); + } table.ensure_not_branch_reference_for_write()?; // A time-travel table may carry a historical schema. let selector =