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 1812e0de feat: route tables by declared TableType to pluggable engines
(#733)
1812e0de is described below
commit 1812e0de03238e61e24b1f106d8ce78fdea31411
Author: Jiajia Li <[email protected]>
AuthorDate: Sat Aug 22 10:43:49 2026 +0800
feat: route tables by declared TableType to pluggable engines (#733)
---
crates/integrations/datafusion/src/catalog.rs | 322 ++++++++-
crates/integrations/datafusion/src/lib.rs | 5 +-
.../datafusion/src/relation_planner.rs | 24 +-
crates/integrations/datafusion/src/sql_context.rs | 18 +
.../datafusion/tests/table_type_routing.rs | 780 +++++++++++++++++++++
crates/paimon-rest-server/src/lib.rs | 11 +-
crates/paimon-rest-server/tests/e2e.rs | 96 +++
crates/paimon/src/catalog/filesystem.rs | 359 +++++++++-
crates/paimon/src/catalog/mod.rs | 64 +-
crates/paimon/src/catalog/rest/rest_catalog.rs | 25 +
crates/paimon/src/spec/core_options.rs | 45 +-
crates/paimon/src/spec/mod.rs | 3 +
crates/paimon/src/spec/schema.rs | 2 +
crates/paimon/src/spec/table_type.rs | 143 ++++
crates/paimon/src/table/rest_env.rs | 46 +-
crates/paimon/tests/rest_catalog_test.rs | 137 ++++
16 files changed, 2015 insertions(+), 65 deletions(-)
diff --git a/crates/integrations/datafusion/src/catalog.rs
b/crates/integrations/datafusion/src/catalog.rs
index 83ef810f..cbee1a6e 100644
--- a/crates/integrations/datafusion/src/catalog.rs
+++ b/crates/integrations/datafusion/src/catalog.rs
@@ -29,11 +29,13 @@ use datafusion::datasource::{TableProvider, TableType};
use datafusion::error::Result as DFResult;
use datafusion::execution::SessionState;
use datafusion::logical_expr::{expr_fn::cast, Expr, LogicalPlan,
LogicalPlanBuilder};
+use datafusion::prelude::SessionContext;
use datafusion::sql::planner::IdentNormalizer;
use datafusion::sql::sqlparser::ast::{Ident, ObjectName, Query, Statement,
Visit, Visitor};
use datafusion::sql::sqlparser::dialect::GenericDialect;
use datafusion::sql::sqlparser::parser::Parser;
use paimon::catalog::{Catalog, Identifier, View};
+use paimon::spec::TableType as PaimonTableType;
use crate::error::to_datafusion_error;
use crate::runtime::{await_with_runtime, block_on_with_runtime};
@@ -43,11 +45,154 @@ use crate::{BlobReaderRegistry, DynamicOptions};
pub(crate) type SessionStateProvider = Arc<dyn Fn() -> Option<SessionState> +
Send + Sync>;
+/// Engine registry shared between the catalog provider and its schema
+/// providers, so registrations stay visible to schemas obtained earlier.
+type TableEngines = Arc<RwLock<HashMap<PaimonTableType, Arc<dyn
TableEngineResolver>>>>;
+
+/// What an engine is asked to resolve. Non-exhaustive so later releases can
+/// carry more of the request — a snapshot selector, say — without breaking
+/// existing resolvers.
+#[derive(Debug, Clone)]
+#[non_exhaustive]
+pub struct EngineTableRequest {
+ pub database: String,
+ pub table: String,
+ /// The type the table's metadata declares.
+ pub declared: PaimonTableType,
+}
+
+impl EngineTableRequest {
+ pub fn new(database: String, table: String, declared: PaimonTableType) ->
Self {
+ Self {
+ database,
+ table,
+ declared,
+ }
+ }
+}
+
+/// Resolves tables owned by another engine (see
+/// [`PaimonCatalogProvider::register_table_engine`]). `Ok(None)` means not
+/// found; errors propagate, so an engine failure never looks like a missing
+/// table.
+#[async_trait]
+pub trait TableEngineResolver: Debug + Send + Sync {
+ /// Resolve a request to the engine's table provider.
+ async fn resolve_table(
+ &self,
+ request: &EngineTableRequest,
+ ) -> DFResult<Option<Arc<dyn TableProvider>>>;
+}
+
+/// Read-only wrapper around an engine-resolved provider: reads delegate,
+/// DML is rejected even when the engine's own provider is writable.
+#[derive(Debug)]
+pub(crate) struct ReadOnlyTableProvider {
+ inner: Arc<dyn TableProvider>,
+ pub(crate) declared: PaimonTableType,
+ pub(crate) table_name: String,
+}
+
+#[async_trait]
+impl TableProvider for ReadOnlyTableProvider {
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ self.inner.schema()
+ }
+
+ fn constraints(&self) -> Option<&datafusion::common::Constraints> {
+ self.inner.constraints()
+ }
+
+ fn table_type(&self) -> TableType {
+ self.inner.table_type()
+ }
+
+ fn get_table_definition(&self) -> Option<&str> {
+ self.inner.get_table_definition()
+ }
+
+ fn get_logical_plan(&self) -> Option<std::borrow::Cow<'_, LogicalPlan>> {
+ self.inner.get_logical_plan()
+ }
+
+ fn get_column_default(&self, column: &str) -> Option<&Expr> {
+ self.inner.get_column_default(column)
+ }
+
+ async fn scan(
+ &self,
+ state: &dyn datafusion::catalog::Session,
+ projection: Option<&Vec<usize>>,
+ filters: &[Expr],
+ limit: Option<usize>,
+ ) -> DFResult<Arc<dyn datafusion::physical_plan::ExecutionPlan>> {
+ self.inner.scan(state, projection, filters, limit).await
+ }
+
+ fn supports_filters_pushdown(
+ &self,
+ filters: &[&Expr],
+ ) -> DFResult<Vec<datafusion::logical_expr::TableProviderFilterPushDown>> {
+ self.inner.supports_filters_pushdown(filters)
+ }
+
+ fn statistics(&self) -> Option<datafusion::common::Statistics> {
+ self.inner.statistics()
+ }
+
+ async fn insert_into(
+ &self,
+ _state: &dyn datafusion::catalog::Session,
+ _input: Arc<dyn datafusion::physical_plan::ExecutionPlan>,
+ _insert_op: datafusion::logical_expr::dml::InsertOp,
+ ) -> DFResult<Arc<dyn datafusion::physical_plan::ExecutionPlan>> {
+ Err(plan_datafusion_err!(
+ "write is not supported for routed '{}' tables ('{}')",
+ self.declared,
+ self.table_name
+ ))
+ }
+}
+
+/// Register `resolver` as the engine for `table_type` on the Paimon catalog
+/// named `catalog_name`.
+///
+/// Also installs [`PaimonRelationPlanner`](crate::PaimonRelationPlanner), so
+/// version clauses cannot slip past this crate (see
[`PaimonCatalogProvider`]).
+///
[`SQLContext::register_catalog_table_engine`](crate::SQLContext::register_catalog_table_engine)
+/// is equivalent.
+pub fn register_catalog_table_engine(
+ ctx: &SessionContext,
+ catalog_name: &str,
+ table_type: PaimonTableType,
+ resolver: Arc<dyn TableEngineResolver>,
+) -> DFResult<()> {
+
ctx.register_relation_planner(Arc::new(crate::PaimonRelationPlanner::new()))?;
+ let provider = ctx
+ .catalog(catalog_name)
+ .ok_or_else(|| plan_datafusion_err!("Unknown catalog
'{catalog_name}'"))?;
+ provider
+ .downcast_ref::<PaimonCatalogProvider>()
+ .ok_or_else(|| plan_datafusion_err!("Catalog '{catalog_name}' is not a
Paimon catalog"))?
+ .register_table_engine(table_type, resolver)
+}
+
/// Provides an interface to manage and access multiple schemas (databases)
/// within a Paimon [`Catalog`].
///
/// This provider uses lazy loading - databases and tables are fetched
/// on-demand from the catalog, ensuring data is always fresh.
+///
+/// # Table-version clauses
+///
+/// SQL queries using `VERSION`/`TIMESTAMP AS OF` need
+/// [`PaimonRelationPlanner`](crate::PaimonRelationPlanner) installed on the
+/// session; DataFusion's default planner drops the clause before this crate
+/// sees it, so the query would read current data.
[`SQLContext`](crate::SQLContext)
+/// installs it, as does
+/// [`register_catalog_table_engine`]. A plain `SessionContext` querying Paimon
+/// tables directly must install it with
+/// `ctx.register_relation_planner(Arc::new(PaimonRelationPlanner::new()))`.
pub struct PaimonCatalogProvider {
catalog_name: Option<String>,
/// Reference to the Paimon catalog.
@@ -65,6 +210,9 @@ pub struct PaimonCatalogProvider {
blob_reader_registry: BlobReaderRegistry,
session_state: Option<SessionStateProvider>,
schema_force_view_types: bool,
+ /// Engines for table types served elsewhere, keyed by declared
+ /// [`PaimonTableType`]. Same poison-recovery stance as `temp_tables`.
+ table_engines: TableEngines,
}
impl Debug for PaimonCatalogProvider {
@@ -90,6 +238,7 @@ impl PaimonCatalogProvider {
blob_reader_registry,
session_state,
schema_force_view_types: true,
+ table_engines: Arc::new(RwLock::new(HashMap::new())),
}
}
@@ -102,24 +251,39 @@ impl PaimonCatalogProvider {
self.schema_force_view_types = schema_force_view_types;
self
}
-}
-impl CatalogProvider for PaimonCatalogProvider {
- fn schema_names(&self) -> Vec<String> {
- let catalog = Arc::clone(&self.catalog);
- block_on_with_runtime(
- async move {
- catalog.list_databases().await.unwrap_or_else(|e| {
- log::error!("failed to list databases: {e}");
- vec![]
- })
- },
- "paimon catalog access thread panicked",
- )
+ /// Register an engine for a table type the Paimon reader cannot serve
+ /// (e.g. [`PaimonTableType::IcebergTable`]); everything else takes the
+ /// Paimon path unchanged. Kept inside the provider so the registered
+ /// catalog type never changes and downcast-based paths (temp tables,
+ /// time travel) keep working.
+ pub(crate) fn register_table_engine(
+ &self,
+ table_type: PaimonTableType,
+ resolver: Arc<dyn TableEngineResolver>,
+ ) -> DFResult<()> {
+ // Routing a Paimon-served type would split it between engines:
+ // reads via the resolver, raw get_table paths via Paimon.
+ if !table_type.requires_table_engine() {
+ return Err(plan_datafusion_err!(
+ "table type '{table_type}' is served by the Paimon reader and
cannot be \
+ routed to a table engine"
+ ));
+ }
+ self.table_engines
+ .write()
+ .unwrap_or_else(|e| e.into_inner())
+ .insert(table_type, resolver);
+ Ok(())
}
- fn schema(&self, name: &str) -> Option<Arc<dyn SchemaProvider>> {
+ fn table_engines(&self) -> TableEngines {
+ Arc::clone(&self.table_engines)
+ }
+
+ fn paimon_schema(&self, name: &str) -> Option<Arc<dyn SchemaProvider>> {
let catalog = Arc::clone(&self.catalog);
+ let table_engines = self.table_engines();
let dynamic_options = Arc::clone(&self.dynamic_options);
let blob_reader_registry = self.blob_reader_registry.clone();
let catalog_name = self.catalog_name.clone();
@@ -145,7 +309,8 @@ impl CatalogProvider for PaimonCatalogProvider {
blob_reader_registry,
session_state,
)
- .with_schema_force_view_types(schema_force_view_types),
+ .with_schema_force_view_types(schema_force_view_types)
+ .with_table_engines(Arc::clone(&table_engines)),
) as Arc<dyn SchemaProvider>),
Err(paimon::Error::DatabaseNotExist { .. }) => {
if temp_provider.is_some() {
@@ -159,7 +324,8 @@ impl CatalogProvider for PaimonCatalogProvider {
blob_reader_registry,
session_state,
)
-
.with_schema_force_view_types(schema_force_view_types),
+
.with_schema_force_view_types(schema_force_view_types)
+
.with_table_engines(Arc::clone(&table_engines)),
) as Arc<dyn SchemaProvider>)
} else {
None
@@ -174,6 +340,25 @@ impl CatalogProvider for PaimonCatalogProvider {
"paimon catalog access thread panicked",
)
}
+}
+
+impl CatalogProvider for PaimonCatalogProvider {
+ fn schema_names(&self) -> Vec<String> {
+ let catalog = Arc::clone(&self.catalog);
+ block_on_with_runtime(
+ async move {
+ catalog.list_databases().await.unwrap_or_else(|e| {
+ log::error!("failed to list databases: {e}");
+ vec![]
+ })
+ },
+ "paimon catalog access thread panicked",
+ )
+ }
+
+ fn schema(&self, name: &str) -> Option<Arc<dyn SchemaProvider>> {
+ self.paimon_schema(name)
+ }
fn register_schema(
&self,
@@ -343,6 +528,8 @@ pub struct PaimonSchemaProvider {
blob_reader_registry: BlobReaderRegistry,
session_state: Option<SessionStateProvider>,
schema_force_view_types: bool,
+ /// Engines for table types served elsewhere; empty without routing.
+ table_engines: TableEngines,
}
impl Debug for PaimonSchemaProvider {
@@ -375,6 +562,7 @@ impl PaimonSchemaProvider {
blob_reader_registry,
session_state,
schema_force_view_types: true,
+ table_engines: Arc::new(RwLock::new(HashMap::new())),
}
}
@@ -382,6 +570,11 @@ impl PaimonSchemaProvider {
self.schema_force_view_types = schema_force_view_types;
self
}
+
+ pub(crate) fn with_table_engines(mut self, table_engines: TableEngines) ->
Self {
+ self.table_engines = table_engines;
+ self
+ }
}
#[async_trait]
@@ -465,9 +658,55 @@ impl SchemaProvider for PaimonSchemaProvider {
let schema_force_view_types = self.schema_force_view_types;
let identifier = Identifier::new(self.database.clone(),
object.table().to_string());
let branch = object.branch().map(str::to_string);
+ let table_engines: HashMap<PaimonTableType, Arc<dyn
TableEngineResolver>> = self
+ .table_engines
+ .read()
+ .unwrap_or_else(|e| e.into_inner())
+ .clone();
await_with_runtime(async move {
- match catalog.get_table(&identifier).await {
- Ok(mut table) => {
+ match catalog.load_table(&identifier).await {
+ Ok(paimon::catalog::LoadedTable::External(external)) => {
+ let declared = external.declared();
+ if branch.is_some() {
+ return Err(plan_datafusion_err!(
+ "branches are not supported for '{}' tables
('{}')",
+ declared,
+ identifier.full_name()
+ ));
+ }
+ // The Paimon arm below applies these; an engine would
+ // ignore them and answer from current data.
+ 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())
+ .map_err(to_datafusion_error)?;
+ let resolver = table_engines.get(&declared).ok_or_else(|| {
+ plan_datafusion_err!(
+ "no table engine is registered for '{}' tables
('{}')",
+ declared,
+ identifier.full_name()
+ )
+ })?;
+ let resolved = resolver
+ .resolve_table(&EngineTableRequest::new(
+ identifier.database().to_string(),
+ identifier.object().to_string(),
+ declared,
+ ))
+ .await?;
+ Ok(resolved.map(|inner| {
+ Arc::new(ReadOnlyTableProvider {
+ inner,
+ declared,
+ table_name: identifier.full_name(),
+ }) as Arc<dyn TableProvider>
+ }))
+ }
+ Ok(paimon::catalog::LoadedTable::Paimon(table)) => {
+ let mut table = *table;
if let Some(branch) = branch.as_deref() {
table = table
.copy_with_branch(branch)
@@ -524,7 +763,8 @@ impl SchemaProvider for PaimonSchemaProvider {
identifier.full_name()
)
})?;
- validate_view_dependencies(&catalog, &catalog_name,
&view).await?;
+ validate_view_dependencies(&catalog, &catalog_name, &view)
+ .await?;
let mut state = session_state
.and_then(|provider| provider())
.ok_or_else(|| {
@@ -606,15 +846,50 @@ impl SchemaProvider for PaimonSchemaProvider {
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 engines: HashMap<PaimonTableType, Arc<dyn TableEngineResolver>> =
self
+ .table_engines
+ .read()
+ .unwrap_or_else(|e| e.into_inner())
+ .clone();
block_on_with_runtime(
async move {
- match catalog.get_table(&identifier).await {
- Ok(table) => {
+ match catalog.load_table(&identifier).await {
+ Ok(paimon::catalog::LoadedTable::External(external)) => {
+ let declared = external.declared();
+ // Paimon-only; `table()` rejects them here too.
+ if branch.is_some() || has_system_suffix {
+ return false;
+ }
+ match engines.get(&declared) {
+ Some(resolver) => match resolver
+ .resolve_table(&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
+ }
+ },
+ None => false,
+ }
+ }
+ Ok(paimon::catalog::LoadedTable::Paimon(table)) => {
if let Some(branch) = branch.as_deref() {
if is_branches_table {
return true;
}
- table.copy_with_branch(branch).await.is_ok()
+ (*table).copy_with_branch(branch).await.is_ok()
} else {
true
}
@@ -721,7 +996,8 @@ async fn validate_view_dependencies(
let candidates = view_relation_identifiers(&view, catalog_name)?;
let mut view_dependencies = Vec::new();
for identifier in candidates {
- match catalog.get_table(&identifier).await {
+ // Routed engine tables count as existing dependencies.
+ match catalog.load_table(&identifier).await {
Ok(_) => continue,
Err(paimon::Error::TableNotExist { .. })
| Err(paimon::Error::Unsupported { .. }) => {}
diff --git a/crates/integrations/datafusion/src/lib.rs
b/crates/integrations/datafusion/src/lib.rs
index d9044fbc..985ad31f 100644
--- a/crates/integrations/datafusion/src/lib.rs
+++ b/crates/integrations/datafusion/src/lib.rs
@@ -74,7 +74,10 @@ pub(crate) type DynamicOptions = Arc<RwLock<HashMap<String,
String>>>;
pub use blob_reader::BlobReaderRegistry;
pub use blob_view::register_blob_view;
-pub use catalog::{PaimonCatalogProvider, PaimonSchemaProvider};
+pub use catalog::{
+ register_catalog_table_engine, EngineTableRequest, PaimonCatalogProvider,
PaimonSchemaProvider,
+ TableEngineResolver,
+};
pub use error::to_datafusion_error;
#[cfg(feature = "fulltext")]
pub use full_text_search::{register_full_text_search, FullTextSearchFunction};
diff --git a/crates/integrations/datafusion/src/relation_planner.rs
b/crates/integrations/datafusion/src/relation_planner.rs
index e51bdcd1..60978e3d 100644
--- a/crates/integrations/datafusion/src/relation_planner.rs
+++ b/crates/integrations/datafusion/src/relation_planner.rs
@@ -22,6 +22,7 @@ use std::fmt::Debug;
use std::sync::Arc;
use datafusion::catalog::default_table_source::{provider_as_source,
source_as_provider};
+use datafusion::common::plan_datafusion_err;
use datafusion::common::TableReference;
use datafusion::error::Result as DFResult;
use datafusion::logical_expr::builder::LogicalPlanBuilder;
@@ -31,6 +32,7 @@ use datafusion::logical_expr::planner::{
use datafusion::sql::sqlparser::ast::{self, TableFactor, TableVersion};
use paimon::spec::{SCAN_TIMESTAMP_MILLIS_OPTION, SCAN_VERSION_OPTION};
+use crate::catalog::ReadOnlyTableProvider;
use crate::table::PaimonTableProvider;
/// A [`RelationPlanner`] that intercepts `VERSION AS OF` and `TIMESTAMP AS OF`
@@ -70,10 +72,8 @@ impl RelationPlanner for PaimonRelationPlanner {
return Ok(RelationPlanning::Original(Box::new(relation)));
};
- let extra_options = match version {
- Some(TableVersion::VersionAsOf(expr)) =>
resolve_version_as_of(expr)?,
- Some(TableVersion::TimestampAsOf(expr)) =>
resolve_timestamp_as_of(expr)?,
- _ => return Ok(RelationPlanning::Original(Box::new(relation))),
+ let Some(version) = version else {
+ return Ok(RelationPlanning::Original(Box::new(relation)));
};
// Resolve the table reference.
@@ -83,6 +83,22 @@ impl RelationPlanner for PaimonRelationPlanner {
.get_table_source(table_ref.clone())?;
let provider = source_as_provider(&source)?;
+ // Every version clause, not just the two resolved below: falling
+ // through drops the clause silently.
+ if let Some(routed) = provider.downcast_ref::<ReadOnlyTableProvider>()
{
+ return Err(plan_datafusion_err!(
+ "time travel is not supported for routed '{}' tables ('{}')",
+ routed.declared,
+ routed.table_name
+ ));
+ }
+
+ let extra_options = match version {
+ TableVersion::VersionAsOf(expr) => resolve_version_as_of(expr)?,
+ TableVersion::TimestampAsOf(expr) =>
resolve_timestamp_as_of(expr)?,
+ _ => return Ok(RelationPlanning::Original(Box::new(relation))),
+ };
+
// Check if this is a Paimon table.
let Some(paimon_provider) =
provider.downcast_ref::<PaimonTableProvider>() else {
return Ok(RelationPlanning::Original(Box::new(relation)));
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index ae02f0a9..cf306f2c 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -348,6 +348,24 @@ impl SQLContext {
paimon_provider.register_temp_table(&database, &table_name, table)
}
+ /// Whether `name` is a Paimon catalog, rather than one DataFusion holds.
+ pub fn is_paimon_catalog(&self, name: &str) -> bool {
+ self.catalogs.contains_key(name)
+ }
+
+ /// Register an engine for a table type served by another engine (e.g.
+ /// [`paimon::spec::TableType::IcebergTable`]) on a registered Paimon
+ /// catalog.
+ /// See [`crate::catalog::PaimonCatalogProvider::register_table_engine`].
+ pub fn register_catalog_table_engine(
+ &self,
+ catalog_name: &str,
+ table_type: paimon::spec::TableType,
+ resolver: Arc<dyn crate::catalog::TableEngineResolver>,
+ ) -> DFResult<()> {
+ crate::catalog::register_catalog_table_engine(&self.ctx, catalog_name,
table_type, resolver)
+ }
+
/// Deregisters a temporary table or view.
///
/// Accepts the same flexible name format as `register_temp_table`.
diff --git a/crates/integrations/datafusion/tests/table_type_routing.rs
b/crates/integrations/datafusion/tests/table_type_routing.rs
new file mode 100644
index 00000000..f75ae8f3
--- /dev/null
+++ b/crates/integrations/datafusion/tests/table_type_routing.rs
@@ -0,0 +1,780 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+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::datasource::{MemTable, TableProvider};
+use datafusion::error::{DataFusionError, Result as DFResult};
+use paimon::catalog::{Catalog, Database, Identifier, LoadedTable};
+use paimon::spec::{Schema as PaimonSchema, SchemaChange, TableType};
+use paimon::table::Table;
+use paimon::{CatalogOptions, FileSystemCatalog, Options, Result as
PaimonResult};
+use paimon_datafusion::{EngineTableRequest, SQLContext, TableEngineResolver};
+use tempfile::TempDir;
+
+const CATALOG: &str = "cat";
+const DB: &str = "shared_db";
+
+#[derive(Debug)]
+struct TypedTestCatalog {
+ inner: Arc<FileSystemCatalog>,
+ declared_types: HashMap<String, TableType>,
+}
+
+#[async_trait]
+impl Catalog for TypedTestCatalog {
+ 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> {
+ if let Some(declared) = self.declared_types.get(identifier.object()) {
+ return Err(paimon::Error::Unsupported {
+ message: format!(
+ "table '{}' is declared '{declared}' and cannot be read as
a Paimon table",
+ identifier.full_name()
+ ),
+ });
+ }
+ self.inner.get_table(identifier).await
+ }
+
+ async fn load_table(&self, identifier: &Identifier) ->
PaimonResult<LoadedTable> {
+ if let Some(declared) = self.declared_types.get(identifier.object()) {
+ if declared.requires_table_engine() {
+ let options = HashMap::new();
+ return LoadedTable::external(
+ *declared,
+ &paimon::spec::CoreOptions::new(&options),
+ &identifier.full_name(),
+ );
+ }
+ }
+ Ok(LoadedTable::Paimon(Box::new(
+ self.get_table(identifier).await?,
+ )))
+ }
+
+ async fn list_tables(&self, database_name: &str) ->
PaimonResult<Vec<String>> {
+ let mut names = self.inner.list_tables(database_name).await?;
+ names.extend(self.declared_types.keys().cloned());
+ Ok(names)
+ }
+
+ 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;
+
+#[async_trait]
+impl TableEngineResolver for FakeEngineResolver {
+ async fn resolve_table(
+ &self,
+ request: &EngineTableRequest,
+ ) -> DFResult<Option<Arc<dyn TableProvider>>> {
+ if request.table != "it" {
+ return Ok(None);
+ }
+ let schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("payload", DataType::Utf8, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 3])),
+ Arc::new(StringArray::from(vec!["x", "y"])),
+ ],
+ )
+ .map_err(DataFusionError::from)?;
+ let table = MemTable::try_new(schema, vec![vec![batch]])?;
+ Ok(Some(Arc::new(table)))
+ }
+}
+
+struct TestEnv {
+ _paimon_dir: TempDir,
+ ctx: SQLContext,
+}
+
+async fn setup() -> 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,
+ declared_types: HashMap::from([
+ ("it".to_string(), TableType::IcebergTable),
+ ("ghost".to_string(), TableType::IcebergTable),
+ ("ft".to_string(), TableType::IcebergTable),
+ ]),
+ });
+ let mut ctx = SQLContext::new();
+ ctx.register_catalog(CATALOG, typed_catalog).await.unwrap();
+ ctx.sql(&format!("CREATE SCHEMA {CATALOG}.{DB}"))
+ .await
+ .unwrap();
+ ctx.sql(&format!(
+ "CREATE TABLE {CATALOG}.{DB}.pt (id INT NOT NULL, name STRING)"
+ ))
+ .await
+ .unwrap();
+ for stmt in [
+ format!("INSERT INTO {CATALOG}.{DB}.pt VALUES (1, 'a')"),
+ format!("INSERT INTO {CATALOG}.{DB}.pt VALUES (2, 'b')"),
+ ] {
+ ctx.sql(&stmt).await.unwrap().collect().await.unwrap();
+ }
+
+ ctx.register_catalog_table_engine(
+ CATALOG,
+ TableType::IcebergTable,
+ Arc::new(FakeEngineResolver),
+ )
+ .unwrap();
+
+ TestEnv {
+ _paimon_dir: paimon_dir,
+ ctx,
+ }
+}
+
+fn column_i32(batches: &[RecordBatch]) -> Vec<i32> {
+ batches
+ .iter()
+ .flat_map(|b| {
+ let col =
b.column(0).as_any().downcast_ref::<Int32Array>().unwrap();
+ (0..col.len()).map(|i| col.value(i)).collect::<Vec<_>>()
+ })
+ .collect()
+}
+
+#[tokio::test]
+async fn paimon_path_still_serves_paimon_tables() {
+ let env = setup().await;
+ let batches = env
+ .ctx
+ .sql(&format!("SELECT id FROM {CATALOG}.{DB}.pt ORDER BY id"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(column_i32(&batches), vec![1, 2]);
+}
+
+#[tokio::test]
+async fn declared_engine_table_routes_to_engine() {
+ let env = setup().await;
+ let df = env
+ .ctx
+ .sql(&format!("SELECT id, payload FROM {CATALOG}.{DB}.it"))
+ .await
+ .unwrap();
+ let names: Vec<String> = df
+ .schema()
+ .fields()
+ .iter()
+ .map(|f| f.name().clone())
+ .collect();
+ assert_eq!(names, vec!["id".to_string(), "payload".to_string()]);
+ let batches = df.collect().await.unwrap();
+ let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
+ assert_eq!(rows, 2);
+}
+
+#[tokio::test]
+async fn cross_engine_join_plans_and_runs() {
+ let env = setup().await;
+ let batches = env
+ .ctx
+ .sql(&format!(
+ "SELECT p.id FROM {CATALOG}.{DB}.pt p JOIN {CATALOG}.{DB}.it i ON
p.id = i.id"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(column_i32(&batches), vec![1]);
+}
+
+#[tokio::test]
+async fn missing_table_still_errors() {
+ let env = setup().await;
+ let Err(err) = env
+ .ctx
+ .sql(&format!("SELECT * FROM {CATALOG}.{DB}.does_not_exist"))
+ .await
+ else {
+ panic!("query against a missing table must fail");
+ };
+ let msg = err.to_string().to_lowercase();
+ assert!(
+ msg.contains("does_not_exist") || msg.contains("not found"),
+ "{msg}"
+ );
+}
+
+#[tokio::test]
+async fn time_travel_still_works_with_engines_registered() {
+ let env = setup().await;
+ let batches = env
+ .ctx
+ .sql(&format!("SELECT id FROM {CATALOG}.{DB}.pt VERSION AS OF 1"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(column_i32(&batches), vec![1]);
+}
+
+#[tokio::test]
+async fn temp_tables_still_work_with_engines_registered() {
+ let env = setup().await;
+ let schema =
Arc::new(datafusion::arrow::datatypes::Schema::new(vec![Field::new(
+ "id",
+ DataType::Int32,
+ false,
+ )]));
+ let mem = MemTable::try_new(schema, vec![vec![]]).unwrap();
+ env.ctx
+ .register_temp_table(format!("{CATALOG}.{DB}.tmp_t"), Arc::new(mem))
+ .expect("temp table registration must survive engine registration");
+ assert!(env
+ .ctx
+ .temp_table_exist(format!("{CATALOG}.{DB}.tmp_t"))
+ .unwrap());
+}
+
+#[tokio::test]
+async fn schemas_and_ddl_behave_with_engines_registered() {
+ let env = setup().await;
+ let provider = env.ctx.ctx().catalog(CATALOG).unwrap();
+ assert!(provider.schema(DB).is_some());
+ assert!(provider.schema("no_such_db").is_none());
+ env.ctx
+ .sql(&format!("CREATE SCHEMA {CATALOG}.fresh_db"))
+ .await
+ .expect("CREATE SCHEMA must work with engines registered");
+ env.ctx
+ .sql(&format!(
+ "CREATE TABLE {CATALOG}.fresh_db.t (id INT NOT NULL)"
+ ))
+ .await
+ .expect("CREATE TABLE in the fresh schema must work");
+}
+
+#[tokio::test]
+async fn table_names_come_from_the_catalog_listing() {
+ let env = setup().await;
+ let provider = env.ctx.ctx().catalog(CATALOG).unwrap();
+ let schema = provider.schema(DB).unwrap();
+ let mut names = schema.table_names();
+ names.sort();
+ names.dedup();
+ assert!(names.contains(&"pt".to_string()), "{names:?}");
+ assert!(names.contains(&"it".to_string()), "{names:?}");
+}
+
+#[derive(Debug)]
+struct BrokenResolver;
+
+#[async_trait]
+impl TableEngineResolver for BrokenResolver {
+ async fn resolve_table(
+ &self,
+ _request: &EngineTableRequest,
+ ) -> DFResult<Option<Arc<dyn TableProvider>>> {
+ Err(DataFusionError::Execution(
+ "engine backend exploded".to_string(),
+ ))
+ }
+}
+
+#[tokio::test]
+async fn engine_errors_are_surfaced() {
+ let env = setup().await;
+ env.ctx
+ .register_catalog_table_engine(CATALOG, TableType::IcebergTable,
Arc::new(BrokenResolver))
+ .unwrap();
+ let Err(err) = env
+ .ctx
+ .sql(&format!("SELECT * FROM {CATALOG}.{DB}.it"))
+ .await
+ else {
+ panic!("broken engine must fail loudly");
+ };
+ let msg = err.to_string();
+ assert!(msg.contains("engine backend exploded"), "{msg}");
+}
+
+#[tokio::test]
+async fn paimon_served_type_takes_paimon_path() {
+ 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,
+ declared_types: HashMap::from([("mt".to_string(),
TableType::MaterializedTable)]),
+ });
+ let mut ctx = SQLContext::new();
+ ctx.register_catalog(CATALOG, typed_catalog).await.unwrap();
+ ctx.sql(&format!("CREATE SCHEMA {CATALOG}.{DB}"))
+ .await
+ .unwrap();
+ let Err(err) = ctx.sql(&format!("SELECT * FROM {CATALOG}.{DB}.mt")).await
else {
+ panic!("a Paimon-served type must fall through to Paimon");
+ };
+ let msg = err.to_string().to_lowercase();
+ assert!(msg.contains("mt") || msg.contains("not found"), "{msg}");
+}
+
+#[tokio::test]
+async fn writes_to_routed_tables_fail_closed() {
+ let env = setup().await;
+ let Err(err) = env
+ .ctx
+ .sql(&format!("UPDATE {CATALOG}.{DB}.ft SET id = 1"))
+ .await
+ else {
+ panic!("UPDATE on a routed table must fail");
+ };
+ let msg = err.to_string();
+ assert!(msg.contains("cannot be read as a Paimon table"), "{msg}");
+ assert!(msg.contains(TableType::IcebergTable.as_str()), "{msg}");
+}
+
+#[tokio::test]
+async fn system_tables_on_routed_tables_error() {
+ let env = setup().await;
+ let Err(err) = env
+ .ctx
+ .sql(&format!("SELECT * FROM {CATALOG}.{DB}.\"it$snapshots\""))
+ .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}");
+}
+
+#[tokio::test]
+async fn table_exist_mirrors_the_resolver() {
+ let env = setup().await;
+ let provider = env.ctx.ctx().catalog(CATALOG).unwrap();
+ let schema = provider.schema(DB).unwrap();
+ assert!(schema.table_exist("it"));
+ assert!(!schema.table_exist("ghost"));
+ assert!(!schema.table_exist("it$snapshots"));
+}
+
+#[tokio::test]
+async fn paimon_managed_types_cannot_be_routed() {
+ let env = setup().await;
+ let Err(err) = env.ctx.register_catalog_table_engine(
+ CATALOG,
+ TableType::FormatTable,
+ Arc::new(FakeEngineResolver),
+ ) else {
+ panic!("registering an engine for a Paimon-managed type must fail");
+ };
+ let msg = err.to_string();
+ assert!(msg.contains("served by the Paimon reader"), "{msg}");
+}
+
+#[tokio::test]
+async fn insert_into_routed_table_is_rejected() {
+ let env = setup().await;
+ let err = async {
+ env.ctx
+ .sql(&format!("INSERT INTO {CATALOG}.{DB}.it VALUES (9, 'z')"))
+ .await?
+ .collect()
+ .await
+ }
+ .await
+ .expect_err("insert into a routed table must fail");
+ assert!(
+ err.to_string()
+ .contains("write is not supported for routed 'iceberg-table'
tables"),
+ "{err}"
+ );
+}
+
+#[tokio::test]
+async fn object_and_lance_tables_route_to_engines() {
+ 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,
+ declared_types: HashMap::from([
+ ("it".to_string(), TableType::ObjectTable),
+ ("lt".to_string(), TableType::LanceTable),
+ ]),
+ });
+ let mut ctx = SQLContext::new();
+ ctx.register_catalog(CATALOG, typed_catalog).await.unwrap();
+ ctx.sql(&format!("CREATE SCHEMA {CATALOG}.{DB}"))
+ .await
+ .unwrap();
+ for declared in [TableType::ObjectTable, TableType::LanceTable] {
+ ctx.register_catalog_table_engine(CATALOG, declared,
Arc::new(FakeEngineResolver))
+ .unwrap();
+ }
+
+ let batches = ctx
+ .sql(&format!("SELECT id FROM {CATALOG}.{DB}.it ORDER BY id"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(column_i32(&batches), vec![1, 3]);
+}
+
+#[tokio::test]
+async fn time_travel_on_routed_tables_is_rejected() {
+ let env = setup().await;
+ env.ctx
+ .ctx()
+ .sql("SET datafusion.sql_parser.dialect = 'databricks'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let Err(err) = env
+ .ctx
+ .ctx()
+ .sql(&format!(
+ "SELECT * FROM {CATALOG}.{DB}.it VERSION AS OF 999999"
+ ))
+ .await
+ else {
+ panic!("time travel on a routed table must not silently read current
data");
+ };
+ let msg = err.to_string();
+ assert!(msg.contains("time travel is not supported"), "{msg}");
+
+ let Err(err) = env
+ .ctx
+ .ctx()
+ .sql(&format!(
+ "SELECT * FROM {CATALOG}.{DB}.it TIMESTAMP AS OF '2020-01-01
00:00:00'"
+ ))
+ .await
+ else {
+ panic!("timestamp travel on a routed table must be rejected too");
+ };
+ assert!(
+ err.to_string().contains("time travel is not supported"),
+ "{err}"
+ );
+}
+
+#[tokio::test]
+async fn session_time_travel_on_routed_tables_is_rejected() {
+ let env = setup().await;
+ env.ctx
+ .sql("SET 'paimon.scan.version' = '1'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let Err(err) = env
+ .ctx
+ .sql(&format!("SELECT * FROM {CATALOG}.{DB}.it"))
+ .await
+ else {
+ panic!("a session scan selector must not silently read current data");
+ };
+ assert!(
+ err.to_string().contains("time travel is not supported"),
+ "{err}"
+ );
+
+ env.ctx
+ .sql("RESET 'paimon.scan.version'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ env.ctx
+ .sql(&format!("SELECT * FROM {CATALOG}.{DB}.it"))
+ .await
+ .expect("routing works again once the selector is reset");
+
+ env.ctx
+ .sql("SET 'paimon.incremental-between' = '1,5'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let Err(err) = env
+ .ctx
+ .sql(&format!("SELECT * FROM {CATALOG}.{DB}.it"))
+ .await
+ else {
+ panic!("an unsupported scan option must not be silently ignored");
+ };
+ assert!(err.to_string().contains("incremental-between"), "{err}");
+}
+
+#[tokio::test]
+async fn every_version_clause_on_routed_tables_is_rejected() {
+ let env = setup().await;
+ env.ctx
+ .ctx()
+ .sql("SET datafusion.sql_parser.dialect = 'databricks'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ for clause in [
+ "VERSION AS OF 999999",
+ "TIMESTAMP AS OF '2020-01-01 00:00:00'",
+ "FOR SYSTEM_TIME AS OF '2020-01-01 00:00:00'",
+ ] {
+ let Err(err) = env
+ .ctx
+ .ctx()
+ .sql(&format!("SELECT * FROM {CATALOG}.{DB}.it {clause}"))
+ .await
+ else {
+ panic!("{clause} must not silently read current data");
+ };
+ assert!(
+ err.to_string().contains("time travel is not supported"),
+ "{clause}: {err}"
+ );
+ }
+}
+
+#[tokio::test]
+async fn show_create_on_routed_tables_reports_the_declared_type() {
+ let env = setup().await;
+ let Err(err) = env
+ .ctx
+ .sql(&format!("SHOW CREATE TABLE {CATALOG}.{DB}.it"))
+ .await
+ else {
+ panic!("Paimon DDL would misrepresent an engine-served table");
+ };
+ let msg = err.to_string();
+ assert!(msg.contains("iceberg-table"), "{msg}");
+ assert!(msg.contains("cannot be read as a Paimon table"), "{msg}");
+}
+
+#[tokio::test]
+async fn session_query_auth_blocks_routed_reads() {
+ let env = setup().await;
+ env.ctx
+ .sql("SET 'paimon.query-auth.enabled' = 'true'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let Err(err) = env
+ .ctx
+ .sql(&format!("SELECT * FROM {CATALOG}.{DB}.it"))
+ .await
+ else {
+ panic!("query-auth must block a routed read, not just a Paimon one");
+ };
+ assert!(err.to_string().contains("query-auth"), "{err}");
+
+ env.ctx
+ .sql("RESET 'paimon.query-auth.enabled'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ env.ctx
+ .sql(&format!("SELECT * FROM {CATALOG}.{DB}.it"))
+ .await
+ .expect("routing works again once the flag is reset");
+}
+
+#[tokio::test]
+async fn registering_on_a_raw_session_installs_the_planner() {
+ use datafusion::prelude::SessionContext;
+ use paimon_datafusion::{register_catalog_table_engine,
PaimonCatalogProvider};
+
+ 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 typed_catalog = Arc::new(TypedTestCatalog {
+ inner: fs_catalog,
+ declared_types: HashMap::from([("it".to_string(),
TableType::IcebergTable)]),
+ });
+
+ // No SQLContext: the raw path a caller might take.
+ let ctx = SessionContext::new();
+ ctx.register_catalog(
+ CATALOG,
+ Arc::new(PaimonCatalogProvider::new(
+ Some(CATALOG.to_string()),
+ typed_catalog,
+ Default::default(),
+ Default::default(),
+ None,
+ )),
+ );
+ register_catalog_table_engine(
+ &ctx,
+ CATALOG,
+ TableType::IcebergTable,
+ Arc::new(FakeEngineResolver),
+ )
+ .unwrap();
+ ctx.sql("SET datafusion.sql_parser.dialect = 'databricks'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let Err(err) = ctx
+ .sql(&format!("SELECT * FROM {CATALOG}.{DB}.it VERSION AS OF 1"))
+ .await
+ else {
+ panic!("registration must install the planner, so this cannot read
current data");
+ };
+ assert!(err.to_string().contains("not supported"), "{err}");
+}
+
+#[tokio::test]
+async fn an_external_type_without_an_engine_says_so() {
+ 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,
+ declared_types: HashMap::from([("it".to_string(),
TableType::IcebergTable)]),
+ });
+ let mut ctx = SQLContext::new();
+ ctx.register_catalog(CATALOG, typed_catalog).await.unwrap();
+ ctx.sql(&format!("CREATE SCHEMA {CATALOG}.{DB}"))
+ .await
+ .unwrap();
+
+ let Err(err) = ctx.sql(&format!("SELECT * FROM {CATALOG}.{DB}.it")).await
else {
+ panic!("an external table without an engine must not resolve");
+ };
+ let msg = err.to_string();
+ assert!(msg.contains("no table engine is registered"), "{msg}");
+ assert!(msg.contains("iceberg-table"), "{msg}");
+}
diff --git a/crates/paimon-rest-server/src/lib.rs
b/crates/paimon-rest-server/src/lib.rs
index 434f38de..403b4220 100644
--- a/crates/paimon-rest-server/src/lib.rs
+++ b/crates/paimon-rest-server/src/lib.rs
@@ -416,12 +416,13 @@ async fn create_table(
async fn get_table(path: RestPath, Extension(state): Extension<Arc<AppState>>)
-> Response {
let table = path.get("table");
let identifier = Identifier::new(path.get("db"), table.clone());
- let resolved = match state.catalog.get_table(&identifier).await {
- Ok(t) => t,
+ // Raw metadata, not a constructed table: engine-served types must stay
+ // describable so clients can route them.
+ let (location, loaded_schema) = match
state.catalog.fetch_table_schema(&identifier).await {
+ Ok(loaded) => loaded,
Err(e) => return error_response(e),
};
-
- let table_schema = resolved.schema();
+ let table_schema = &loaded_schema;
// Convert the stored `TableSchema` into the DDL `Schema` the response
// carries. `Schema` is a field subset of `TableSchema` (both camelCase),
// and serde ignores the extra keys, preserving field ids exactly.
@@ -441,7 +442,7 @@ async fn get_table(path: RestPath, Extension(state):
Extension<Arc<AppState>>) -
// that satisfies the client's RESTEnv requirement.
Some(identifier.full_name()),
Some(table),
- Some(resolved.location().to_string()),
+ Some(location),
Some(false),
Some(table_schema.id()),
Some(schema),
diff --git a/crates/paimon-rest-server/tests/e2e.rs
b/crates/paimon-rest-server/tests/e2e.rs
index 6edf274b..db27bb3e 100644
--- a/crates/paimon-rest-server/tests/e2e.rs
+++ b/crates/paimon-rest-server/tests/e2e.rs
@@ -429,3 +429,99 @@ async fn test_special_char_names() {
cat.drop_database(plus_db, false, false).await.unwrap();
assert!(cat.list_databases().await.unwrap().is_empty());
}
+
+#[tokio::test]
+async fn declared_engine_type_routes_through_the_rest_catalog() {
+ use paimon::catalog::LoadedTable;
+ use paimon::spec::TableType;
+
+ let ctx = setup().await;
+ ctx.catalog
+ .create_database("db", false, HashMap::new())
+ .await
+ .expect("create database");
+
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .option("type", "iceberg-table")
+ .build()
+ .expect("build schema");
+ let identifier = Identifier::new("db", "ice_t");
+ ctx.catalog
+ .create_table(&identifier, schema, false)
+ .await
+ .expect("create table");
+
+ let routed = ctx
+ .catalog
+ .load_table(&identifier)
+ .await
+ .expect("routing must reach the declared type");
+ assert!(
+ matches!(routed, LoadedTable::External(ref e) if e.declared() ==
TableType::IcebergTable),
+ "{routed:?}"
+ );
+
+ let err = ctx
+ .catalog
+ .get_table(&identifier)
+ .await
+ .expect_err("an engine-served table must not construct as Paimon");
+ assert!(
+ err.to_string().contains("cannot be read as a Paimon table"),
+ "{err}"
+ );
+}
+
+#[tokio::test]
+async fn creating_a_table_with_an_unknown_type_is_rejected() {
+ let ctx = setup().await;
+ ctx.catalog
+ .create_database("db", false, HashMap::new())
+ .await
+ .expect("create database");
+
+ let mut payload =
serde_json::to_value(append_only_schema()).expect("serialize schema");
+ payload["options"]["type"] = serde_json::json!("iceberg-tabel");
+ let schema: Schema = serde_json::from_value(payload).expect("deserialize
schema");
+ let identifier = Identifier::new("db", "typo_t");
+
+ ctx.catalog
+ .create_table(&identifier, schema, false)
+ .await
+ .expect_err("a misspelled type must not create a table");
+ let tables = ctx.catalog.list_tables("db").await.expect("list tables");
+ assert!(!tables.contains(&"typo_t".to_string()), "{tables:?}");
+}
+
+#[tokio::test]
+async fn altering_the_declared_type_is_rejected() {
+ use paimon::spec::SchemaChange;
+
+ let ctx = setup().await;
+ ctx.catalog
+ .create_database("db", false, HashMap::new())
+ .await
+ .expect("create database");
+ let identifier = Identifier::new("db", "t");
+ ctx.catalog
+ .create_table(&identifier, append_only_schema(), false)
+ .await
+ .expect("create table");
+
+ ctx.catalog
+ .alter_table(
+ &identifier,
+ vec![SchemaChange::set_option(
+ "type".to_string(),
+ "iceberg-table".to_string(),
+ )],
+ false,
+ )
+ .await
+ .expect_err("changing 'type' over REST must be rejected");
+ ctx.catalog
+ .get_table(&identifier)
+ .await
+ .expect("still readable");
+}
diff --git a/crates/paimon/src/catalog/filesystem.rs
b/crates/paimon/src/catalog/filesystem.rs
index ae015e25..0ebd8ab8 100644
--- a/crates/paimon/src/catalog/filesystem.rs
+++ b/crates/paimon/src/catalog/filesystem.rs
@@ -26,7 +26,7 @@ use crate::common::{CatalogOptions, Options};
use crate::error::{ConfigInvalidSnafu, Error, Result};
use crate::io::cache::create_local_cache;
use crate::io::FileIO;
-use crate::spec::{Schema, TableSchema};
+use crate::spec::{CoreOptions, Schema, TableSchema, TableType,
TABLE_TYPE_OPTION};
use crate::table::{SchemaManager, Table};
use async_trait::async_trait;
use bytes::Bytes;
@@ -170,6 +170,61 @@ impl FileSystemCatalog {
Ok(dirs)
}
+ /// Fetch the stored path and schema of an existing table, bypassing the
+ /// engine-type guard in [`Self::build_table`]: routing and catalog servers
+ /// need the declared type before deciding anything.
+ pub async fn fetch_table_schema(
+ &self,
+ identifier: &Identifier,
+ ) -> Result<(String, TableSchema)> {
+ identifier.validate()?;
+
+ let table_path = self.table_path(identifier);
+
+ if !self.table_exists(identifier).await? {
+ return Err(Error::TableNotExist {
+ full_name: identifier.full_name(),
+ });
+ }
+
+ let schema = self
+ .load_latest_table_schema(&table_path)
+ .await?
+ .ok_or_else(|| Error::TableNotExist {
+ full_name: identifier.full_name(),
+ })?;
+
+ Ok((table_path, schema))
+ }
+
+ fn build_table(
+ &self,
+ identifier: &Identifier,
+ table_path: String,
+ schema: TableSchema,
+ ) -> Result<Table> {
+ // Fail closed: constructed as Paimon, raw `get_table` paths (writes,
+ // procedures, time travel) would misread it.
+ let declared = CoreOptions::new(schema.options()).table_type()?;
+ if declared.requires_table_engine() {
+ return Err(Error::Unsupported {
+ message: format!(
+ "table '{}' is declared '{declared}' and cannot be read as
a Paimon \
+ table; only plain reads through a registered table engine
are supported",
+ identifier.full_name()
+ ),
+ });
+ }
+
+ Ok(Table::new(
+ self.file_io.clone(),
+ identifier.clone(),
+ table_path,
+ schema,
+ None,
+ ))
+ }
+
/// Load the latest schema for a table (highest schema-{version} file
under table_path/schema).
async fn load_latest_table_schema(&self, table_path: &str) ->
Result<Option<TableSchema>> {
let manager = SchemaManager::new(self.file_io.clone(),
table_path.to_string());
@@ -297,30 +352,23 @@ impl Catalog for FileSystemCatalog {
}
async fn get_table(&self, identifier: &Identifier) -> Result<Table> {
- identifier.validate()?;
-
- let table_path = self.table_path(identifier);
+ let (table_path, schema) = self.fetch_table_schema(identifier).await?;
+ self.build_table(identifier, table_path, schema)
+ }
- if !self.table_exists(identifier).await? {
- return Err(Error::TableNotExist {
- full_name: identifier.full_name(),
- });
+ async fn load_table(&self, identifier: &Identifier) ->
Result<crate::catalog::LoadedTable> {
+ let (table_path, schema) = self.fetch_table_schema(identifier).await?;
+ let options = CoreOptions::new(schema.options());
+ let declared = options.table_type()?;
+ if declared.requires_table_engine() {
+ return crate::catalog::LoadedTable::external(
+ declared,
+ &options,
+ &identifier.full_name(),
+ );
}
-
- let schema = self
- .load_latest_table_schema(&table_path)
- .await?
- .ok_or_else(|| Error::TableNotExist {
- full_name: identifier.full_name(),
- })?;
-
- Ok(Table::new(
- self.file_io.clone(),
- identifier.clone(),
- table_path,
- schema,
- None,
- ))
+ self.build_table(identifier, table_path, schema)
+ .map(|table| crate::catalog::LoadedTable::Paimon(Box::new(table)))
}
async fn list_tables(&self, database_name: &str) -> Result<Vec<String>> {
@@ -351,6 +399,8 @@ impl Catalog for FileSystemCatalog {
ignore_if_exists: bool,
) -> Result<()> {
identifier.validate()?;
+ // Never persist a type nothing can load.
+ CoreOptions::new(creation.options()).table_type()?;
let table_path = self.table_path(identifier);
@@ -458,6 +508,8 @@ impl Catalog for FileSystemCatalog {
full_name: identifier.full_name(),
})?;
+ reject_table_type_changes(current.options(), &changes)?;
+
let new_schema = current
.apply_changes(changes)
.map_err(|e| fill_table_name(e, identifier))?;
@@ -465,6 +517,39 @@ impl Catalog for FileSystemCatalog {
}
}
+/// The declared type picks the reader, so it is fixed at creation: flipping
+/// it strands a populated table behind a reader that cannot see its data.
+/// Only case-insensitive no-ops pass, matching Java `SchemaManager`.
+fn reject_table_type_changes(
+ current_options: &HashMap<String, String>,
+ changes: &[crate::spec::SchemaChange],
+) -> Result<()> {
+ let current_type = current_options
+ .get(TABLE_TYPE_OPTION)
+ .map(String::as_str)
+ .unwrap_or_else(|| TableType::default().as_str());
+ for change in changes {
+ match change {
+ crate::spec::SchemaChange::SetOption { key, value }
+ if key == TABLE_TYPE_OPTION &&
!value.eq_ignore_ascii_case(current_type) =>
+ {
+ return Err(Error::Unsupported {
+ message: format!(
+ "changing '{TABLE_TYPE_OPTION}' from '{current_type}'
to '{value}' is not supported"
+ ),
+ });
+ }
+ crate::spec::SchemaChange::RemoveOption { key } if key ==
TABLE_TYPE_OPTION => {
+ return Err(Error::Unsupported {
+ message: format!("removing '{TABLE_TYPE_OPTION}' is not
supported"),
+ });
+ }
+ _ => {}
+ }
+ }
+ Ok(())
+}
+
/// `TableSchema::apply_changes` returns column errors without a table name;
/// fill in the identifier's full name so the message identifies the table.
fn fill_table_name(err: Error, identifier: &Identifier) -> Error {
@@ -665,6 +750,234 @@ mod tests {
assert!(escaped_path.exists());
}
+ #[tokio::test]
+ async fn test_alter_table_cannot_change_the_declared_type() {
+ use crate::spec::SchemaChange;
+
+ let (_temp_dir, catalog) = create_test_catalog();
+ catalog
+ .create_database("db1", false, HashMap::new())
+ .await
+ .unwrap();
+ let schema = Schema::builder()
+ .column(
+ "id",
+ crate::spec::DataType::Int(crate::spec::IntType::new()),
+ )
+ .build()
+ .unwrap();
+ let identifier = Identifier::new("db1", "t");
+ catalog
+ .create_table(&identifier, schema, false)
+ .await
+ .unwrap();
+
+ for change in [
+ SchemaChange::set_option("type".to_string(),
"iceberg-table".to_string()),
+ SchemaChange::set_option("type".to_string(),
"delta-table".to_string()),
+ SchemaChange::remove_option("type".to_string()),
+ ] {
+ let err = catalog
+ .alter_table(&identifier, vec![change], false)
+ .await
+ .unwrap_err();
+ assert!(matches!(err, Error::Unsupported { .. }), "{err:?}");
+ }
+ assert!(catalog
+ .alter_table(
+ &identifier,
+ vec![SchemaChange::set_option(
+ "type".to_string(),
+ "TABLE".to_string(),
+ )],
+ false,
+ )
+ .await
+ .is_ok());
+ catalog.get_table(&identifier).await.unwrap();
+ }
+
+ #[tokio::test]
+ async fn test_create_table_rejects_an_unknown_type() {
+ let (_temp_dir, catalog) = create_test_catalog();
+ catalog
+ .create_database("db1", false, HashMap::new())
+ .await
+ .unwrap();
+ let valid = Schema::builder()
+ .column(
+ "id",
+ crate::spec::DataType::Int(crate::spec::IntType::new()),
+ )
+ .build()
+ .unwrap();
+ let mut payload = serde_json::to_value(&valid).unwrap();
+ payload["options"] = serde_json::json!({ "type": "delta-table" });
+ let schema: Schema = serde_json::from_value(payload).unwrap();
+ let identifier = Identifier::new("db1", "delta_t");
+
+ let err = catalog
+ .create_table(&identifier, schema, false)
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(err, Error::Unsupported { ref message } if
message.contains("delta-table")),
+ "{err:?}"
+ );
+ assert!(!catalog.table_exists(&identifier).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_stored_scan_selectors_block_routing() {
+ use crate::catalog::LoadedTable;
+
+ let (_temp_dir, catalog) = create_test_catalog();
+ catalog
+ .create_database("db1", false, HashMap::new())
+ .await
+ .unwrap();
+
+ for (name, key, value) in [
+ ("ice_v", "scan.version", "1"),
+ ("ice_s", "scan.snapshot-id", "1"),
+ ("ice_i", "incremental-between", "1,5"),
+ ] {
+ let schema = Schema::builder()
+ .column(
+ "id",
+ crate::spec::DataType::Int(crate::spec::IntType::new()),
+ )
+ .option("type", "iceberg-table")
+ .option(key, value)
+ .build()
+ .unwrap();
+ let identifier = Identifier::new("db1", name);
+ catalog
+ .create_table(&identifier, schema, false)
+ .await
+ .unwrap();
+
+ let err = catalog.load_table(&identifier).await.unwrap_err();
+ assert!(matches!(err, Error::Unsupported { .. }), "{key}:
{err:?}");
+ }
+
+ let schema = Schema::builder()
+ .column(
+ "id",
+ crate::spec::DataType::Int(crate::spec::IntType::new()),
+ )
+ .option("type", "iceberg-table")
+ .build()
+ .unwrap();
+ let identifier = Identifier::new("db1", "ice_plain");
+ catalog
+ .create_table(&identifier, schema, false)
+ .await
+ .unwrap();
+ let routed = catalog.load_table(&identifier).await.unwrap();
+ assert!(matches!(routed, LoadedTable::External(_)), "{routed:?}");
+ }
+
+ #[tokio::test]
+ async fn test_routing_cannot_skip_the_read_guards() {
+ use crate::catalog::LoadedTable;
+ use crate::spec::CoreOptions;
+
+ // The only constructor of the `Engine` variant, so a third-party
+ // catalog cannot route past these guards either.
+ for (key, value) in [
+ ("query-auth.enabled", "true"),
+ ("scan.version", "1"),
+ ("incremental-between", "1,5"),
+ ] {
+ let options = HashMap::from([(key.to_string(),
value.to_string())]);
+ let err = LoadedTable::external(
+ TableType::IcebergTable,
+ &CoreOptions::new(&options),
+ "db1.t",
+ )
+ .unwrap_err();
+ assert!(matches!(err, Error::Unsupported { .. }), "{key}:
{err:?}");
+ }
+
+ let options = HashMap::new();
+ assert!(LoadedTable::external(
+ TableType::IcebergTable,
+ &CoreOptions::new(&options),
+ "db1.t",
+ )
+ .is_ok());
+ }
+
+ #[tokio::test]
+ async fn test_types_without_a_paimon_reader_fail_closed() {
+ let (_temp_dir, catalog) = create_test_catalog();
+ catalog
+ .create_database("db1", false, HashMap::new())
+ .await
+ .unwrap();
+
+ for (name, declared) in [("obj_t", "object-table"), ("lance_t",
"lance-table")] {
+ let schema = Schema::builder()
+ .column(
+ "id",
+ crate::spec::DataType::Int(crate::spec::IntType::new()),
+ )
+ .option("type", declared)
+ .build()
+ .unwrap();
+ let identifier = Identifier::new("db1", name);
+ catalog
+ .create_table(&identifier, schema, false)
+ .await
+ .unwrap();
+
+ let err = catalog.get_table(&identifier).await.unwrap_err();
+ assert!(
+ matches!(err, Error::Unsupported { ref message }
+ if message.contains("cannot be read as a Paimon table")),
+ "{declared}: {err:?}"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn test_declared_engine_type_routes_and_fails_closed() {
+ use crate::catalog::LoadedTable;
+
+ let (_temp_dir, catalog) = create_test_catalog();
+ catalog
+ .create_database("db1", false, HashMap::new())
+ .await
+ .unwrap();
+ let schema = Schema::builder()
+ .column(
+ "id",
+ crate::spec::DataType::Int(crate::spec::IntType::new()),
+ )
+ .option("type", "iceberg-table")
+ .build()
+ .unwrap();
+ let identifier = Identifier::new("db1", "ice_t");
+ catalog
+ .create_table(&identifier, schema, false)
+ .await
+ .unwrap();
+
+ let routed = catalog.load_table(&identifier).await.unwrap();
+ assert!(
+ matches!(routed, LoadedTable::External(ref e) if e.declared() ==
TableType::IcebergTable),
+ "{routed:?}"
+ );
+
+ let err = catalog.get_table(&identifier).await.unwrap_err();
+ assert!(
+ matches!(err, Error::Unsupported { ref message }
+ if message.contains("cannot be read as a Paimon table")),
+ "{err:?}"
+ );
+ }
+
#[tokio::test]
async fn test_table_operations() {
let (_temp_dir, catalog) = create_test_catalog();
diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs
index 38a414b9..b2a5cf23 100644
--- a/crates/paimon/src/catalog/mod.rs
+++ b/crates/paimon/src/catalog/mod.rs
@@ -262,9 +262,57 @@ impl fmt::Debug for Identifier {
use async_trait::async_trait;
use crate::api::PagedList;
-use crate::spec::{Partition, Schema, SchemaChange};
+use crate::spec::{Partition, Schema, SchemaChange, TableType};
use crate::table::Table;
+/// Outcome of [`Catalog::load_table`].
+#[derive(Debug)]
+pub enum LoadedTable {
+ /// A constructed Paimon table (boxed: far larger than the other variant).
+ Paimon(Box<Table>),
+ /// A table this reader cannot construct.
+ External(ExternalTableMetadata),
+}
+
+/// What a caller needs to pick an engine for a table Paimon cannot construct.
+/// Only [`LoadedTable::external`] can build one, so the stored-metadata checks
+/// always run before a caller sees it.
+#[derive(Debug)]
+pub struct ExternalTableMetadata {
+ declared: TableType,
+}
+
+impl ExternalTableMetadata {
+ /// The type the table's metadata declares.
+ pub fn declared(&self) -> TableType {
+ self.declared
+ }
+}
+
+impl LoadedTable {
+ /// Classify `full_name` as external, refusing options no engine can honor
+ /// and reads this client cannot authorize.
+ ///
+ /// Errors if [`TableType::requires_table_engine`] rejects `declared`:
+ /// classifying a Paimon-served type as external would skip the reader that
+ /// can actually construct it.
+ pub fn external(
+ declared: TableType,
+ options: &crate::spec::CoreOptions<'_>,
+ full_name: &str,
+ ) -> Result<Self> {
+ if !declared.requires_table_engine() {
+ return Err(Error::Unsupported {
+ message: format!(
+ "table '{full_name}' is declared '{declared}', which the
Paimon reader serves"
+ ),
+ });
+ }
+ options.ensure_engine_can_serve(full_name)?;
+ Ok(Self::External(ExternalTableMetadata { declared }))
+ }
+}
+
/// Catalog API for reading and writing metadata (databases, tables) in Paimon.
///
/// Corresponds to
[org.apache.paimon.catalog.Catalog](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java).
@@ -323,6 +371,20 @@ pub trait Catalog: Send + Sync {
/// * [`crate::Error::TableNotExist`] - table does not exist.
async fn get_table(&self, identifier: &Identifier) -> Result<Table>;
+ /// 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.
+ ///
+ /// # Errors
+ /// Same as [`Catalog::get_table`].
+ async fn load_table(&self, identifier: &Identifier) -> Result<LoadedTable>
{
+ Ok(LoadedTable::Paimon(Box::new(
+ self.get_table(identifier).await?,
+ )))
+ }
+
/// List table names in a database. System tables are not listed.
///
/// # Errors
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index 401545cb..be41cfc8 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -215,6 +215,31 @@ impl Catalog for RESTCatalog {
.await
}
+ async fn load_table(&self, identifier: &Identifier) ->
Result<crate::catalog::LoadedTable> {
+ let response = RESTEnv::fetch_table_response(identifier,
&self.api).await?;
+ if let Some(schema) = response.schema.as_ref() {
+ let options = crate::spec::CoreOptions::new(schema.options());
+ let declared = options.table_type()?;
+ if declared.requires_table_engine() {
+ return crate::catalog::LoadedTable::external(
+ declared,
+ &options,
+ &identifier.full_name(),
+ );
+ }
+ }
+ RESTEnv::build_table(
+ identifier,
+ response,
+ self.api.clone(),
+ self.options.clone(),
+ self.data_token_enabled,
+ self.local_cache.clone(),
+ )
+ .await
+ .map(|table| crate::catalog::LoadedTable::Paimon(Box::new(table)))
+ }
+
async fn list_tables(&self, database_name: &str) -> Result<Vec<String>> {
self.api
.list_tables(database_name)
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index e9f34332..cc57ccef 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -17,6 +17,8 @@
use std::collections::{HashMap, HashSet};
+use crate::spec::TableType;
+
const DELETION_VECTORS_ENABLED_OPTION: &str = "deletion-vectors.enabled";
const DELETION_VECTORS_MERGE_ON_READ_OPTION: &str =
"deletion-vectors.merge-on-read";
pub(crate) const QUERY_AUTH_ENABLED_OPTION: &str = "query-auth.enabled";
@@ -73,7 +75,7 @@ const STATS_MODE_SUFFIX: &str = "stats-mode";
const ROW_TRACKING_ENABLED_OPTION: &str = "row-tracking.enabled";
const CLUSTERING_INCREMENTAL_OPTION: &str = "clustering.incremental";
pub(crate) const TABLE_TYPE_OPTION: &str = "type";
-pub(crate) const FORMAT_TABLE_TYPE: &str = "format-table";
+
pub(crate) const PATH_OPTION: &str = "path";
const MANIFEST_COMPRESSION_OPTION: &str = "manifest.compression";
const MANIFEST_TARGET_FILE_SIZE_OPTION: &str = "manifest.target-file-size";
@@ -624,11 +626,17 @@ impl<'a> CoreOptions<'a> {
.unwrap_or(false)
}
+ /// The declared [`TableType`], defaulting to [`TableType::Table`].
+ /// Fails on a value this client does not know.
+ pub fn table_type(&self) -> crate::Result<TableType> {
+ match self.options.get(TABLE_TYPE_OPTION) {
+ Some(value) => value.parse(),
+ None => Ok(TableType::default()),
+ }
+ }
+
pub fn is_format_table(&self) -> bool {
- self.options
- .get(TABLE_TYPE_OPTION)
- .map(|value| value.eq_ignore_ascii_case(FORMAT_TABLE_TYPE))
- .unwrap_or(false)
+ matches!(self.table_type(), Ok(TableType::FormatTable))
}
pub fn path(&self) -> Option<&str> {
@@ -917,6 +925,33 @@ impl<'a> CoreOptions<'a> {
///
/// This is the semantic owner for selector mutual exclusion and strict
/// numeric parsing.
+ /// Fails when these options forbid the read outright, or ask for
+ /// something a table engine cannot honor: a scan option this client does
+ /// not support, or a historical state. Dropping any of them would answer
+ /// with unfiltered or current data instead.
+ ///
+ /// 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.validate_scan_options()?;
+ if self.has_time_travel_selector() {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "time travel is not supported for engine-served table
'{full_name}'"
+ ),
+ });
+ }
+ Ok(())
+ }
+
+ /// Whether these options ask for a historical state of the table. A
+ /// malformed selector counts too, so callers that cannot honor time
+ /// travel reject rather than answer from the current state.
+ pub fn has_time_travel_selector(&self) -> bool {
+ !matches!(self.try_time_travel_selector(), Ok(None))
+ }
+
pub(crate) fn try_time_travel_selector(&self) ->
crate::Result<Option<TimeTravelSelector<'a>>> {
let selectors = self.configured_time_travel_selectors();
if selectors.len() > 1 {
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index 3fb3802e..a2613fec 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -31,6 +31,9 @@ pub use blob_view_struct::BlobViewStruct;
mod data_file;
pub use data_file::*;
+mod table_type;
+pub use table_type::*;
+
mod core_options;
pub(crate) use core_options::TimeTravelSelector;
pub use core_options::*;
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 3238ccee..0b52571b 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -1155,6 +1155,8 @@ impl Schema {
primary_keys: &[String],
options: &HashMap<String, String>,
) -> crate::Result<()> {
+ // Create and alter share this, so an unparsable type never lands.
+ CoreOptions::new(options).table_type()?;
validate_no_reserved_field_names(fields)?;
Self::validate_key_field_types(fields, primary_keys, options)?;
Self::validate_row_tracking(primary_keys, options)?;
diff --git a/crates/paimon/src/spec/table_type.rs
b/crates/paimon/src/spec/table_type.rs
new file mode 100644
index 00000000..e4a095d2
--- /dev/null
+++ b/crates/paimon/src/spec/table_type.rs
@@ -0,0 +1,143 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::{Display, Formatter};
+use std::str::FromStr;
+
+use crate::error::Error;
+
+/// Type of the table, declared by the `type` table option.
+///
+/// Mirrors `org.apache.paimon.TableType`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
+pub enum TableType {
+ /// Normal Paimon table.
+ #[default]
+ Table,
+ /// A directory containing multiple files of the same format.
+ FormatTable,
+ /// A normal Paimon table combined with materialized SQL.
+ MaterializedTable,
+ /// A normal Paimon table combined with an object location.
+ ObjectTable,
+ /// A lance table, see <https://lancedb.github.io/lance/>.
+ LanceTable,
+ /// An iceberg table, see <https://iceberg.apache.org/>.
+ IcebergTable,
+}
+
+impl TableType {
+ /// The `type` option value of this table type.
+ pub fn as_str(&self) -> &'static str {
+ match self {
+ TableType::Table => "table",
+ TableType::FormatTable => "format-table",
+ TableType::MaterializedTable => "materialized-table",
+ TableType::ObjectTable => "object-table",
+ TableType::LanceTable => "lance-table",
+ TableType::IcebergTable => "iceberg-table",
+ }
+ }
+
+ /// Whether this type needs an engine of its own (see
+ /// [`Catalog::load_table`](crate::catalog::Catalog::load_table)). Java
+ /// builds a dedicated table for each; this client has none, so reading one
+ /// as Paimon misreads it and writing could put Paimon snapshots over
+ /// foreign data.
+ pub fn requires_table_engine(&self) -> bool {
+ matches!(
+ self,
+ TableType::ObjectTable | TableType::LanceTable |
TableType::IcebergTable
+ )
+ }
+}
+
+impl Display for TableType {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.write_str(self.as_str())
+ }
+}
+
+impl FromStr for TableType {
+ type Err = Error;
+
+ fn from_str(value: &str) -> Result<Self, Self::Err> {
+ [
+ TableType::Table,
+ TableType::FormatTable,
+ TableType::MaterializedTable,
+ TableType::ObjectTable,
+ TableType::LanceTable,
+ TableType::IcebergTable,
+ ]
+ .into_iter()
+ .find(|table_type| value.eq_ignore_ascii_case(table_type.as_str()))
+ .ok_or_else(|| Error::Unsupported {
+ message: format!("unknown table type: {value}"),
+ })
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn parses_every_declared_value() {
+ for table_type in [
+ TableType::Table,
+ TableType::FormatTable,
+ TableType::MaterializedTable,
+ TableType::ObjectTable,
+ TableType::LanceTable,
+ TableType::IcebergTable,
+ ] {
+ assert_eq!(
+ TableType::from_str(table_type.as_str()).unwrap(),
+ table_type
+ );
+ assert_eq!(
+
TableType::from_str(&table_type.as_str().to_uppercase()).unwrap(),
+ table_type
+ );
+ }
+ }
+
+ #[test]
+ fn rejects_unknown_value() {
+ let err = TableType::from_str("delta-table").expect_err("unknown type
must not parse");
+ assert!(err.to_string().contains("delta-table"), "{err}");
+ }
+
+ #[test]
+ fn types_without_a_paimon_reader_require_an_engine() {
+ for table_type in [
+ TableType::ObjectTable,
+ TableType::LanceTable,
+ TableType::IcebergTable,
+ ] {
+ assert!(table_type.requires_table_engine(), "{table_type}");
+ }
+ for table_type in [
+ TableType::Table,
+ TableType::FormatTable,
+ TableType::MaterializedTable,
+ ] {
+ assert!(!table_type.requires_table_engine(), "{table_type}");
+ }
+ }
+}
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index b61370b2..7d716cb7 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -106,11 +106,38 @@ impl RESTEnv {
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
) -> Result<Table> {
- let response = api
- .get_table(identifier)
+ let response = Self::fetch_table_response(identifier, &api).await?;
+ Self::build_table(
+ identifier,
+ response,
+ api,
+ options,
+ data_token_enabled,
+ local_cache,
+ )
+ .await
+ }
+
+ /// Fetch the raw table metadata, mapping REST errors to catalog errors.
+ pub(crate) async fn fetch_table_response(
+ identifier: &Identifier,
+ api: &RESTApi,
+ ) -> Result<crate::api::GetTableResponse> {
+ api.get_table(identifier)
.await
- .map_err(|e| map_rest_error_for_table(e, identifier))?;
+ .map_err(|e| map_rest_error_for_table(e, identifier))
+ }
+ /// Build a Table from an already-fetched response, so routing can
+ /// inspect the declared type first.
+ pub(crate) async fn build_table(
+ identifier: &Identifier,
+ response: crate::api::GetTableResponse,
+ api: Arc<RESTApi>,
+ options: Options,
+ data_token_enabled: bool,
+ local_cache: Option<Arc<LocalCache>>,
+ ) -> Result<Table> {
let schema = response.schema.ok_or_else(|| Error::DataInvalid {
message: format!("Table {} response missing schema",
identifier.full_name()),
source: None,
@@ -128,6 +155,19 @@ impl RESTEnv {
),
source: None,
})?;
+ // Fail closed: constructed as Paimon, raw `get_table` paths (writes,
+ // procedures, time travel) would misread it.
+ let declared = CoreOptions::new(schema.options()).table_type()?;
+ if declared.requires_table_engine() {
+ return Err(Error::Unsupported {
+ message: format!(
+ "table '{}' is declared '{declared}' and cannot be read as
a Paimon \
+ table; only plain reads through a registered table engine
are supported",
+ identifier.full_name()
+ ),
+ });
+ }
+
let mut table_schema = TableSchema::new(schema_id, &schema);
if CoreOptions::new(table_schema.options()).is_format_table() {
table_schema =
table_schema.copy_with_options(std::collections::HashMap::from([(
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 2000f9e4..83da7c8a 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -1501,3 +1501,140 @@ async fn
test_catalog_maps_unsupported_view_and_function_endpoints() {
paimon::Error::Unsupported { .. }
));
}
+
+#[tokio::test]
+async fn test_load_table_returns_external_for_declared_type() {
+ use paimon::catalog::LoadedTable;
+ use paimon::spec::TableType;
+
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = Schema::builder()
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .option("type", "iceberg-table")
+ .build()
+ .unwrap();
+ ctx.server
+ .add_table_with_schema("default", "ice_t", schema, "file:///unused");
+
+ let identifier = Identifier::new("default", "ice_t");
+ let routed = ctx.catalog.load_table(&identifier).await.unwrap();
+ assert!(
+ matches!(routed, LoadedTable::External(ref e) if e.declared() ==
TableType::IcebergTable),
+ "{routed:?}"
+ );
+}
+
+#[tokio::test]
+async fn test_load_table_fails_closed_on_query_auth() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = Schema::builder()
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .option("type", "iceberg-table")
+ .option("query-auth.enabled", "true")
+ .build()
+ .unwrap();
+ ctx.server
+ .add_table_with_schema("default", "authed_ice_t", schema,
"file:///unused");
+
+ let identifier = Identifier::new("default", "authed_ice_t");
+ let err = ctx.catalog.load_table(&identifier).await.unwrap_err();
+ assert!(
+ matches!(err, paimon::Error::Unsupported { ref message } if
message.contains("query-auth")),
+ "{err:?}"
+ );
+}
+
+#[tokio::test]
+async fn test_load_table_constructs_paimon_for_undeclared_type() {
+ use paimon::catalog::LoadedTable;
+
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = Schema::builder()
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .build()
+ .unwrap();
+ ctx.server
+ .add_table_with_schema("default", "plain_t", schema, "file:///unused");
+
+ let identifier = Identifier::new("default", "plain_t");
+ let routed = ctx.catalog.load_table(&identifier).await.unwrap();
+ match routed {
+ LoadedTable::Paimon(table) => {
+ assert_eq!(table.identifier().full_name(), "default.plain_t");
+ }
+ other => panic!("expected a constructed Paimon table, got {other:?}"),
+ }
+}
+
+#[tokio::test]
+async fn test_get_table_fails_closed_on_iceberg_table() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = Schema::builder()
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .option("type", "iceberg-table")
+ .build()
+ .unwrap();
+ ctx.server
+ .add_table_with_schema("default", "raw_ice_t", schema,
"file:///unused");
+
+ let err = ctx
+ .catalog
+ .get_table(&Identifier::new("default", "raw_ice_t"))
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(err, paimon::Error::Unsupported { ref message }
+ if message.contains("cannot be read as a Paimon table")),
+ "{err:?}"
+ );
+}
+
+#[tokio::test]
+async fn test_load_table_parses_declared_type_case_insensitively() {
+ use paimon::catalog::LoadedTable;
+ use paimon::spec::TableType;
+
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = Schema::builder()
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .option("type", "ICEBERG-TABLE")
+ .build()
+ .unwrap();
+ ctx.server
+ .add_table_with_schema("default", "upper_ice_t", schema,
"file:///unused");
+
+ let routed = ctx
+ .catalog
+ .load_table(&Identifier::new("default", "upper_ice_t"))
+ .await
+ .unwrap();
+ assert!(
+ matches!(routed, LoadedTable::External(ref e) if e.declared() ==
TableType::IcebergTable),
+ "{routed:?}"
+ );
+}
+
+#[tokio::test]
+async fn test_load_table_rejects_unknown_declared_type() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let valid = Schema::builder()
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .build()
+ .unwrap();
+ let mut payload = serde_json::to_value(&valid).unwrap();
+ payload["options"] = serde_json::json!({ "type": "delta-table" });
+ let schema: Schema = serde_json::from_value(payload).unwrap();
+ ctx.server
+ .add_table_with_schema("default", "delta_t", schema, "file:///unused");
+
+ let err = ctx
+ .catalog
+ .load_table(&Identifier::new("default", "delta_t"))
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(err, paimon::Error::Unsupported { ref message }
+ if message.contains("unknown table type")),
+ "{err:?}"
+ );
+}
