diff --git a/crates/integrations/datafusion/src/catalog.rs b/crates/integrations/datafusion/src/catalog.rs index 83ef810fa..cbee1a6e7 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 Option + Send + Sync>; +/// Engine registry shared between the catalog provider and its schema +/// providers, so registrations stay visible to schemas obtained earlier. +type TableEngines = Arc>>>; + +/// 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>>; +} + +/// 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, + 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> { + 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>, + filters: &[Expr], + limit: Option, + ) -> DFResult> { + self.inner.scan(state, projection, filters, limit).await + } + + fn supports_filters_pushdown( + &self, + filters: &[&Expr], + ) -> DFResult> { + self.inner.supports_filters_pushdown(filters) + } + + fn statistics(&self) -> Option { + self.inner.statistics() + } + + async fn insert_into( + &self, + _state: &dyn datafusion::catalog::Session, + _input: Arc, + _insert_op: datafusion::logical_expr::dml::InsertOp, + ) -> DFResult> { + 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, +) -> 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::() + .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, /// Reference to the Paimon catalog. @@ -65,6 +210,9 @@ pub struct PaimonCatalogProvider { blob_reader_registry: BlobReaderRegistry, session_state: Option, 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 { - 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, + ) -> 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> { + fn table_engines(&self) -> TableEngines { + Arc::clone(&self.table_engines) + } + + fn paimon_schema(&self, name: &str) -> Option> { 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), 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) } else { None @@ -174,6 +340,25 @@ impl CatalogProvider for PaimonCatalogProvider { "paimon catalog access thread panicked", ) } +} + +impl CatalogProvider for PaimonCatalogProvider { + fn schema_names(&self) -> Vec { + 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> { + self.paimon_schema(name) + } fn register_schema( &self, @@ -343,6 +528,8 @@ pub struct PaimonSchemaProvider { blob_reader_registry: BlobReaderRegistry, session_state: Option, 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> = 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 + })) + } + 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> = 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 d9044fbc9..985ad31fb 100644 --- a/crates/integrations/datafusion/src/lib.rs +++ b/crates/integrations/datafusion/src/lib.rs @@ -74,7 +74,10 @@ pub(crate) type DynamicOptions = Arc>>; 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 e51bdcd10..60978e3dd 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::() { + 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::() 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 ae02f0a95..cf306f2c7 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, + ) -> 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 000000000..f75ae8f3d --- /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, + declared_types: HashMap, +} + +#[async_trait] +impl Catalog for TypedTestCatalog { + async fn list_databases(&self) -> PaimonResult> { + self.inner.list_databases().await + } + + async fn create_database( + &self, + name: &str, + ignore_if_exists: bool, + properties: HashMap, + ) -> PaimonResult<()> { + self.inner + .create_database(name, ignore_if_exists, properties) + .await + } + + async fn get_database(&self, name: &str) -> PaimonResult { + self.inner.get_database(name).await + } + + async fn drop_database( + &self, + name: &str, + ignore_if_not_exists: bool, + cascade: bool, + ) -> PaimonResult<()> { + self.inner + .drop_database(name, ignore_if_not_exists, cascade) + .await + } + + async fn get_table(&self, identifier: &Identifier) -> PaimonResult { + 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 { + 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> { + 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, + 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>> { + 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 { + batches + .iter() + .flat_map(|b| { + let col = b.column(0).as_any().downcast_ref::().unwrap(); + (0..col.len()).map(|i| col.value(i)).collect::>() + }) + .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 = 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>> { + 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 434f38de3..403b42203 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>) -> 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>) - // 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 6edf274b3..db27bb3eb 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 ae015e254..0ebd8ab84 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
{ + // 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> { 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
{ - 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 { + 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> { @@ -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, + 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 38a414b97..b2a5cf232 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
), + /// 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 { + 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
; + /// 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 { + 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 401545cb8..be41cfc87 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 { + 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> { 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 e9f34332d..cc57ccef8 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 { + 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>> { 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 3fb3802e9..a2613fecf 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 3238ccee0..0b52571b4 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, ) -> 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 000000000..e4a095d23 --- /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 . + LanceTable, + /// An iceberg table, see . + 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 { + [ + 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 b61370b27..7d716cb7a 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>, ) -> Result
{ - 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 { + 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, + options: Options, + data_token_enabled: bool, + local_cache: Option>, + ) -> Result
{ 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 2000f9e40..83da7c8a7 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:?}" + ); +}