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 2ee305fa fix: fail closed for tables an engine must serve (#740)
2ee305fa is described below
commit 2ee305fac6403b50f9203b311d3ce3095467ad2c
Author: Jiajia Li <[email protected]>
AuthorDate: Wed Aug 26 10:56:25 2026 +0800
fix: fail closed for tables an engine must serve (#740)
---
crates/integrations/datafusion/src/procedures.rs | 5 +-
crates/integrations/datafusion/src/sql_context.rs | 1 +
crates/integrations/datafusion/src/table/mod.rs | 6 +
crates/integrations/datafusion/src/table_loader.rs | 32 ++
.../datafusion/tests/table_type_routing.rs | 364 +++++++++++++++++++++
crates/paimon/src/catalog/mod.rs | 36 +-
crates/paimon/src/spec/core_options.rs | 37 ++-
crates/paimon/src/spec/schema.rs | 10 +-
crates/paimon/src/table/incremental_scan.rs | 1 +
crates/paimon/src/table/mod.rs | 7 +
crates/paimon/src/table/table_commit.rs | 9 +
crates/paimon/src/table/write_builder.rs | 10 +
12 files changed, 499 insertions(+), 19 deletions(-)
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<Table> {
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<String>,
) -> DFResult<Self> {
+ 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<Self> {
+ 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<String>,
) -> DFResult<Self> {
+ 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<dyn Catalog>,
+ identifier: &Identifier,
+) -> DFResult<Table> {
+ 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<dyn Catalog>,
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<FileSystemCatalog>,
+}
+
+#[async_trait]
+impl Catalog for LegacyTestCatalog {
+ async fn list_databases(&self) -> PaimonResult<Vec<String>> {
+ self.inner.list_databases().await
+ }
+
+ async fn create_database(
+ &self,
+ name: &str,
+ ignore_if_exists: bool,
+ properties: HashMap<String, String>,
+ ) -> PaimonResult<()> {
+ self.inner
+ .create_database(name, ignore_if_exists, properties)
+ .await
+ }
+
+ async fn get_database(&self, name: &str) -> PaimonResult<Database> {
+ 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<Table> {
+ 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<Vec<String>> {
+ 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<SchemaChange>,
+ 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<LegacyTestCatalog>) {
+ 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<LoadedTable>
{
- 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<Vec<Partition>> {
- 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<String, String>) ->
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<IncrementalPlan> {
+
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<String, String>,
strict: bool,
) -> Result<Self> {
+ // 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<Self> {
+ // 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 7bf70936..96749512 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 =
