feat: route tables by declared TableType to pluggable engines

[plusplusjiajia]

Purpose

REST catalogs can declare table types a Paimon reader cannot serve via the type table option (e.g. type=iceberg-table). paimon-datafusion tries to read every table as Paimon, so these tables fail with unrelated errors. This PR routes them to pluggable engines instead:

• paimon::spec::TableType mirrors Java's org.apache.paimon.TableType, and CoreOptions::table_type() returns it, so DataFusion decides from the enum rather than from type strings.
• Catalog::load_table. A defaulted method returning LoadedTable::{Paimon, External}, so existing implementations keep working. Classification is unconditional — the outcome depends only on the table's own metadata, never on what the caller registered. Both built-in catalogs override it to classify before construction, matching Java's shared dispatch in CatalogUtils.loadTable; the default classifies from the constructed table, so a catalog that only implements get_table still fails closed.
• object-table, lance-table and iceberg-table all require an engine: Java builds a dedicated table for each, this client has none, so reading one as Paimon would misread it and writing could put Paimon
  snapshots over foreign data. Both catalogs fail closed on them, covering every raw get_table path. Raw metadata stays describable, so paimon-rest-server can still answer GetTable and let its clients route.
• An unknown declared type is rejected before it is persisted, on create and alter alike, and the declared type is immutable afterwards — mirroring SchemaValidation and SchemaManager.
• ExternalTableMetadata can only be built through LoadedTable::external, which rejects a type the Paimon reader serves and runs the stored-metadata checks.
• Historical reads fail loudly rather than silently: every version clause, session scan selectors, and selectors stored in the table's own options are rejected instead of being dropped and answered from current data. register_catalog_table_engine(&ctx, ..) installs PaimonRelationPlanner, so that holds without SQLContext too.
• Routing lives inside PaimonCatalogProvider so downcast paths keep working; table_exist mirrors resolution, and engine-resolved providers are wrapped read-only.

EngineTableRequest is #[non_exhaustive], so a follow-up can carry a snapshot selector — with an opt-in on the resolver, so an engine that ignores it never answers from current data — without breaking existing resolvers. A follow-up also adds an Iceberg-backed resolver crate.

[JingsongLi]

Detailed review found three correctness regressions and one creation-validation gap. I verified the current head locally; details are inline.

[JingsongLi]

[P1] Route every table kind the Rust reader cannot serve

This classifies only Iceberg as engine-required, so register_table_engine rejects both ObjectTable and LanceTable as if the ordinary Paimon reader supported them. It does not: the Rust read stack has no Object/Lance provider and its normal scan uses Paimon snapshots (including an empty plan when none exists), while Java CatalogUtils constructs dedicated Object, Lance, and Iceberg table implementations. A real object-table or lance-table can therefore fail or silently appear empty, and callers cannot register the resolver introduced here. Please mark every currently non-native type (at least Object/Lance/Iceberg) as engine-required/fail-closed until native providers exist, and replace the Lance rejection test with routing coverage.

[plusplusjiajia]

@JingsongLi Good catch. Done: object-table and lance-table now require an engine too, and both catalogs fail closed.

[JingsongLi]

[P1] Reject or implement time travel for routed providers

Returning this wrapper breaks the existing raw SQLContext::ctx().sql time-travel path: PaimonRelationPlanner cannot downcast it to PaimonTableProvider, returns Original, and DataFusion 54's default relation planner discards TableFactor.version. I reproduced this on this head: after selecting the Databricks dialect, SELECT * FROM ...it VERSION AS OF 999999 succeeded and returned the current two engine rows instead of rejecting the nonexistent version. The high-level SQLContext::sql path fails closed, but the public raw context is intentionally supported and tested for time travel. Please make the relation planner recognize routed providers and explicitly reject VERSION/TIMESTAMP, or extend the resolver contract to carry those selectors.

[plusplusjiajia]

@JingsongLi The reproduction steps helped. Sweeping that surface turned up three more ways in: session selectors, selectors stored in the table's own options, and unsupported scan options. All rejected now, and registration installs the relation planner so the rejection holds without SQLContext.

[JingsongLi]

[P1] Keep type immutable before trusting it for routing

This makes the latest schema's type an authoritative engine switch, but TableSchema::apply_changes still accepts arbitrary SetOption/RemoveOption, and SQL exposes that through ALTER TABLE ... SET TBLPROPERTIES. A populated Paimon table can be changed to type=iceberg-table; with a resolver, later reads are redirected to a same-named engine table, and without one the existing snapshots become inaccessible. Java SchemaManager explicitly rejects semantic changes/removal of CoreOptions.TYPE, even before snapshots exist. Please enforce the same invariant before saving a schema (allowing only case-insensitive/default-equivalent no-ops) and cover filesystem plus REST ALTER paths.

[plusplusjiajia]

@JingsongLi Agreed, rejecting unknown values wasn't enough. Done: alter_table rejects any change or removal of type, allowing only case-insensitive no-ops, matching SchemaManager.

[JingsongLi]

[P2] Validate the declared type before persisting schema-0

This validation currently first runs while loading/routing. FileSystemCatalog::create_table still creates the directory and writes schema-0 without calling table_type(), and the bundled REST server delegates to it. Thus CREATE TABLE ... WITH ('type'='iceberg-tabel') returns success but every later get_table/routing call fails with unknown table type, leaving an unusable catalog entry. Please invoke this parsing from shared create/schema validation before any metadata write and add filesystem/REST tests that assert rejection leaves no table artifacts.

[plusplusjiajia]

@JingsongLi Done: the parse runs in Schema::validate_final_schema, shared by create and alter.

[JingsongLi]

[P2] Keep DataFusion engine registration out of the core Catalog contract

Could we avoid Catalog::load_table_routing(..., engine_types)? engine_types is DataFusion/session state, not catalog metadata. Passing it into paimon::Catalog makes loading the same table depend on which integration/resolvers are currently registered, and it forces third-party catalogs to implement a DataFusion-specific routing protocol (while the default implementation ignores the set).

I do not think simply calling get_table() and then inspecting TableType in DataFusion is sufficient: by then FS/REST has entered the concrete Paimon Table construction path; REST may acquire a data token and construct FileIO, and the external-type fail-closed guard must run before that construction. Calling get_table_type() and then get_table() would also fetch metadata twice for native tables.

Could core instead expose an unconditional classification, for example:

async fn load_table(&self, id: &Identifier) -> Result<LoadedTable>;

enum LoadedTable {
    Paimon(Box<Table>),
    External(ExternalTableMetadata),
}

The catalog would load metadata once and decide only whether the core reader can construct the table. Object/Lance/Iceberg would always return External. DataFusion would then look up its own resolver and report a clear error if none is registered. The existing get_table() path can remain fail-closed for external types. In other words, core owns classification and stored-metadata validation; DataFusion owns resolver selection, session semantics, and the read-only wrapper.

This would also remove a hidden invariant in the current public API: RoutedTableLoad::engine does not check declared.requires_table_engine(). A non-DataFusion caller can include a native type in engine_types and make the FS/REST implementations skip native table construction. Correctness currently relies on the private check in the DataFusion registration path.

This boundary also matches the Java direction more closely: core dispatches from table metadata without accepting caller-registered engine types.

[plusplusjiajia]

[P2] Keep DataFusion engine registration out of the core Catalog contract

Could we avoid Catalog::load_table_routing(..., engine_types)? engine_types is DataFusion/session state, not catalog metadata. Passing it into paimon::Catalog makes loading the same table depend on which integration/resolvers are currently registered, and it forces third-party catalogs to implement a DataFusion-specific routing protocol (while the default implementation ignores the set).

I do not think simply calling get_table() and then inspecting TableType in DataFusion is sufficient: by then FS/REST has entered the concrete Paimon Table construction path; REST may acquire a data token and construct FileIO, and the external-type fail-closed guard must run before that construction. Calling get_table_type() and then get_table() would also fetch metadata twice for native tables.

Could core instead expose an unconditional classification, for example:

async fn load_table(&self, id: &Identifier) -> Result<LoadedTable>;

enum LoadedTable {
    Paimon(Box<Table>),
    External(ExternalTableMetadata),
}

The catalog would load metadata once and decide only whether the core reader can construct the table. Object/Lance/Iceberg would always return External. DataFusion would then look up its own resolver and report a clear error if none is registered. The existing get_table() path can remain fail-closed for external types. In other words, core owns classification and stored-metadata validation; DataFusion owns resolver selection, session semantics, and the read-only wrapper.

This would also remove a hidden invariant in the current public API: RoutedTableLoad::engine does not check declared.requires_table_engine(). A non-DataFusion caller can include a native type in engine_types and make the FS/REST implementations skip native table construction. Correctness currently relies on the private check in the DataFusion registration path.

This boundary also matches the Java direction more closely: core dispatches from table metadata without accepting caller-registered engine types.

@JingsongLi Good call on the boundary — engine_types is gone from core. Catalog::load_table returns LoadedTable::{Paimon, External} and classifies unconditionally, so the outcome depends only on the table's own metadata. Core owns classification and the stored-metadata checks; DataFusion owns resolver selection, session semantics, and the read-only wrapper.
The hidden invariant is closed too: ExternalTableMetadata can only be built through LoadedTable::external, which rejects any type requires_table_engine does not accept.

[JingsongLi]

[P1] Classify the table in the default load_table implementation

This default is the compatibility path for existing third-party Catalog implementations, but it unconditionally wraps the Table returned by get_table as LoadedTable::Paimon. If such a catalog returns a table whose stored type is object-table, lance-table, or iceberg-table, PaimonSchemaProvider::table takes the Paimon arm and never consults the registered resolver. PaimonTableProvider has no later type guard, so reads can misread the foreign table (or silently appear empty), and INSERT can write Paimon metadata into the foreign location.

This also contradicts the PR description that the default "classifies from the constructed table" so a catalog implementing only get_table still fails closed. Please first load the table, parse CoreOptions::new(table.schema().options()).table_type()?, and call the checked LoadedTable::external constructor when requires_table_engine() is true; otherwise return the Paimon variant. The built-in overrides can remain to preserve pre-construction classification and avoid token/FileIO work. A regression test should use a catalog that deliberately does not override load_table.

[JingsongLi]

+1