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 68244878 feat(table): read the registered partitions of a
catalog-managed format table (#814)
68244878 is described below
commit 6824487813c69b1d4975e1add8c37343b3ead75b
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Fri Sep 11 23:12:58 2026 +0800
feat(table): read the registered partitions of a catalog-managed format
table (#814)
---
Cargo.lock | 1 +
crates/integrations/datafusion/Cargo.toml | 1 +
.../datafusion/tests/rest_format_table_scan.rs | 446 ++++++++++++++++++
crates/paimon/src/catalog/rest/rest_catalog.rs | 29 +-
crates/paimon/src/spec/core_options.rs | 40 ++
crates/paimon/src/spec/mod.rs | 4 +-
crates/paimon/src/spec/partition_utils.rs | 2 +-
crates/paimon/src/table/format_partition.rs | 352 ++++++++++++++
crates/paimon/src/table/format_table_scan.rs | 517 ++++++++++++++++-----
crates/paimon/src/table/mod.rs | 65 +++
crates/paimon/src/table/rest_env.rs | 37 ++
crates/paimon/tests/mock_server.rs | 47 ++
crates/paimon/tests/rest_catalog_test.rs | 367 +++++++++++++++
13 files changed, 1769 insertions(+), 139 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 11ec61ea..f8374af8 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4745,6 +4745,7 @@ dependencies = [
"arrow-schema",
"arrow-select",
"async-trait",
+ "axum",
"bytes",
"chrono",
"constant_time_eq",
diff --git a/crates/integrations/datafusion/Cargo.toml
b/crates/integrations/datafusion/Cargo.toml
index 17acaea0..f7072fc3 100644
--- a/crates/integrations/datafusion/Cargo.toml
+++ b/crates/integrations/datafusion/Cargo.toml
@@ -50,6 +50,7 @@ uuid = { version = "1", features = ["v4"] }
[dev-dependencies]
arrow-array = { workspace = true }
arrow-schema = { workspace = true }
+axum = { version = "0.7", features = ["macros", "tokio", "http1", "http2"] }
bytes = "1.7.1"
flate2 = "1"
paimon-ftindex-core = "0.1.0"
diff --git a/crates/integrations/datafusion/tests/rest_format_table_scan.rs
b/crates/integrations/datafusion/tests/rest_format_table_scan.rs
new file mode 100644
index 00000000..a446147f
--- /dev/null
+++ b/crates/integrations/datafusion/tests/rest_format_table_scan.rs
@@ -0,0 +1,446 @@
+// 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.
+
+//! SQL reads of a Format Table whose partitions a REST catalog manages.
+//!
+//! Partitions are registered through the catalog API, the way any writer
registers them, so
+//! these tests do not depend on how a partition came to be registered.
+
+#[path = "../../../paimon/tests/mock_server.rs"]
+mod mock_server;
+
+use std::collections::HashMap;
+use std::path::Path;
+use std::sync::Arc;
+
+use arrow_array::{Int64Array, RecordBatch};
+use arrow_schema::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
+use axum::http::StatusCode;
+use paimon::api::ConfigResponse;
+use paimon::catalog::{Catalog, Identifier, RESTCatalog};
+use paimon::spec::{BigIntType, BooleanType, DataType, IntType, Schema,
VarCharType};
+use paimon::{CatalogOptions, Options};
+use paimon_datafusion::SQLContext;
+use parquet::arrow::ArrowWriter;
+use tempfile::TempDir;
+
+use mock_server::{start_mock_server, RESTServer};
+
+const DATABASE: &str = "default";
+const TABLE: &str = "events";
+const WAREHOUSE: &str = "test_warehouse";
+
+/// A catalog-managed Format Table served by a mock REST catalog, reading its
data from a
+/// temporary directory.
+struct ManagedTable {
+ server: RESTServer,
+ catalog: Arc<RESTCatalog>,
+ context: SQLContext,
+}
+
+impl ManagedTable {
+ async fn new(temp_dir: &TempDir, partition_columns: &[(&str, DataType)])
-> Self {
+ let server = start_mock_server(
+ WAREHOUSE.to_string(),
+ temp_dir.path().to_string_lossy().into_owned(),
+ ConfigResponse::new(HashMap::from([(
+ CatalogOptions::PREFIX.to_string(),
+ "mock-test".to_string(),
+ )])),
+ vec![DATABASE.to_string()],
+ )
+ .await;
+ server.add_table_with_schema(
+ DATABASE,
+ TABLE,
+ format_table_schema(partition_columns),
+ &format!("file://{}", temp_dir.path().display()),
+ );
+ server.set_table_external(DATABASE, TABLE, false);
+
+ let mut options = Options::new();
+ options.set(CatalogOptions::URI, server.url().unwrap());
+ options.set(CatalogOptions::WAREHOUSE, WAREHOUSE);
+ options.set(CatalogOptions::TOKEN_PROVIDER, "bear");
+ options.set(CatalogOptions::TOKEN, "test-token");
+ let catalog = Arc::new(RESTCatalog::new(options, true).await.unwrap());
+ let mut context = SQLContext::new();
+ context
+ .register_catalog("paimon", catalog.clone())
+ .await
+ .unwrap();
+ Self {
+ server,
+ catalog,
+ context,
+ }
+ }
+
+ /// Register partitions through the catalog API.
+ async fn register(&self, partitions: &[&[(&str, &str)]]) {
+ self.catalog
+ .create_partitions(
+ &Identifier::new(DATABASE, TABLE),
+ partitions.iter().map(|values| spec(values)).collect(),
+ true,
+ )
+ .await
+ .unwrap();
+ }
+
+ /// The sorted ids of the rows a filtered SELECT returns.
+ async fn ids(&self, predicate: &str) -> Vec<i64> {
+ let sql = format!("SELECT id FROM paimon.{DATABASE}.{TABLE} WHERE
{predicate}");
+ let mut ids = Vec::new();
+ for batch in self
+ .context
+ .sql(&sql)
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap()
+ {
+ let values = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ ids.extend(values.iter().flatten());
+ }
+ ids.sort_unstable();
+ ids
+ }
+
+ /// The error a filtered SELECT fails with, whether planning or execution
reports it.
+ async fn error(&self, predicate: &str) -> String {
+ let sql = format!("SELECT id FROM paimon.{DATABASE}.{TABLE} WHERE
{predicate}");
+ match self.context.sql(&sql).await {
+ Ok(frame) => frame.collect().await.unwrap_err(),
+ Err(error) => error,
+ }
+ .to_string()
+ }
+
+ /// How many listings the table has received from each endpoint: plain,
then by filter.
+ fn listing_counts(&self) -> (usize, usize) {
+ (
+ self.server
+ .table_partition_list_name_patterns(DATABASE, TABLE)
+ .len(),
+ self.server
+ .table_partition_list_by_filter_requests(DATABASE, TABLE)
+ .len(),
+ )
+ }
+
+ /// The name patterns of the listings received since `seen`, whichever
endpoint served them.
+ fn name_patterns_since(&self, seen: (usize, usize)) -> Vec<Option<String>>
{
+ let mut patterns = self
+ .server
+ .table_partition_list_name_patterns(DATABASE, TABLE)
+ .split_off(seen.0);
+ patterns.extend(
+ self.server
+ .table_partition_list_by_filter_requests(DATABASE, TABLE)
+ .into_iter()
+ .skip(seen.1)
+ .map(|request| request.partition_name_pattern),
+ );
+ patterns
+ }
+}
+
+fn format_table_schema(partition_columns: &[(&str, DataType)]) -> Schema {
+ let partition_keys = partition_columns
+ .iter()
+ .map(|(name, _)| (*name).to_string())
+ .collect::<Vec<_>>();
+ partition_columns
+ .iter()
+ .fold(Schema::builder(), |builder, (name, data_type)| {
+ builder.column(*name, data_type.clone())
+ })
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .partition_keys(partition_keys)
+ .option("type", "format-table")
+ .option("file.format", "parquet")
+ .option("metastore.partitioned-table", "true")
+ .build()
+ .unwrap()
+}
+
+fn varchar() -> DataType {
+ DataType::VarChar(VarCharType::new(255).unwrap())
+}
+
+fn spec(values: &[(&str, &str)]) -> HashMap<String, String> {
+ values
+ .iter()
+ .map(|(key, value)| ((*key).to_string(), (*value).to_string()))
+ .collect()
+}
+
+fn write_ids(directory: &Path, ids: &[i64]) {
+ std::fs::create_dir_all(directory).unwrap();
+ let schema = Arc::new(ArrowSchema::new(vec![Field::new(
+ "id",
+ ArrowDataType::Int64,
+ true,
+ )]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int64Array::from(ids.to_vec()))],
+ )
+ .unwrap();
+ let file =
std::fs::File::create(directory.join("part-0.parquet")).unwrap();
+ let mut writer = ArrowWriter::try_new(file, schema, None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+}
+
+#[cfg(not(windows))]
+// Planning a SELECT resolves the table on a blocking catalog-access thread,
so the mock
+// server needs a runtime thread of its own to answer while that one waits.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_managed_scan_reads_only_registered_partitions() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let table = ManagedTable::new(&temp_dir, &[("dt", varchar())]).await;
+ for (dt, id) in [("a", 1), ("b", 2), ("c", 3)] {
+ write_ids(&temp_dir.path().join(format!("dt={dt}")), &[id]);
+ }
+ table.register(&[&[("dt", "a")], &[("dt", "c")]]).await;
+
+ // `dt=b` holds a file but nobody registered it, so it is not part of the
table.
+ assert_eq!(table.ids("TRUE").await, vec![1, 3]);
+ assert!(table.ids("dt = 'b'").await.is_empty());
+
+ table.register(&[&[("dt", "b")]]).await;
+ assert_eq!(table.ids("TRUE").await, vec![1, 2, 3]);
+}
+
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_managed_scan_pushes_a_partition_name_pattern() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let table = ManagedTable::new(&temp_dir, &[("dt", varchar()), ("hh",
varchar())]).await;
+ table
+ .register(&[
+ &[("dt", "20260722"), ("hh", "10")],
+ &[("dt", "20260722"), ("hh", "11")],
+ &[("dt", "20260723"), ("hh", "10")],
+ ])
+ .await;
+
+ for (predicate, expected) in [
+ ("dt = '20260722' AND hh = '10'", Some("dt=20260722/hh=10")),
+ ("dt = '20260722'", Some("dt=20260722/%")),
+ (
+ "dt = '20260722' AND hh IN ('10', '11')",
+ Some("dt=20260722/%"),
+ ),
+ // Only a leading run of equalities becomes a prefix pattern.
+ ("hh = '10'", None),
+ ("dt > '20260722'", None),
+ ] {
+ let seen = table.listing_counts();
+ table.ids(predicate).await;
+ let pushed = table.name_patterns_since(seen);
+ assert!(!pushed.is_empty(), "{predicate} listed no partitions");
+ assert!(
+ pushed.iter().all(|pattern| pattern.as_deref() == expected),
+ "{predicate} pushed {pushed:?}, expected {expected:?}"
+ );
+ }
+}
+
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_managed_scan_sends_its_partition_predicate_as_a_filter() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let table = ManagedTable::new(&temp_dir, &[("dt", varchar()), ("hh",
varchar())]).await;
+ for (dt, hh, id) in [
+ ("20260722", "10", 1),
+ ("20260722", "11", 2),
+ ("20260723", "10", 3),
+ ] {
+ write_ids(&temp_dir.path().join(format!("dt={dt}/hh={hh}")), &[id]);
+ }
+ table
+ .register(&[
+ &[("dt", "20260722"), ("hh", "10")],
+ &[("dt", "20260722"), ("hh", "11")],
+ &[("dt", "20260723"), ("hh", "10")],
+ ])
+ .await;
+
+ // No leading equality, so only the filter can narrow what the catalog
returns.
+ assert_eq!(table.ids("hh = '10'").await, vec![1, 3]);
+ let requests = table
+ .server
+ .table_partition_list_by_filter_requests(DATABASE, TABLE);
+ let request = requests.last().expect("the scan should list by filter");
+ assert_eq!(request.partition_name_pattern, None);
+ assert_eq!(request.max_results, Some(1000));
+ let filter: serde_json::Value =
serde_json::from_str(&request.filter).unwrap();
+ assert_eq!(filter["function"], "EQUAL");
+ assert_eq!(filter["transform"]["fieldRef"]["name"], "hh");
+ assert_eq!(filter["transform"]["fieldRef"]["index"], 1);
+ assert_eq!(filter["literals"], serde_json::json!(["10"]));
+
+ // A catalog that cannot list by filter is still asked, by pattern; the
partition set never
+ // comes from the directory tree.
+ table
+ .server
+
.set_list_partitions_by_filter_error_status(Some(StatusCode::NOT_IMPLEMENTED));
+ let listed = table
+ .server
+ .table_partition_list_name_patterns(DATABASE, TABLE)
+ .len();
+ assert_eq!(table.ids("dt = '20260722' AND hh > '10'").await, vec![2]);
+ assert_eq!(
+ table
+ .server
+ .table_partition_list_name_patterns(DATABASE, TABLE)[listed..],
+ [Some("dt=20260722/%".to_string())]
+ );
+}
+
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_managed_scan_keeps_registrations_spelled_unlike_the_filter() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let table = ManagedTable::new(
+ &temp_dir,
+ &[
+ ("month", DataType::Int(IntType::new())),
+ ("active", DataType::Boolean(BooleanType::new())),
+ ],
+ )
+ .await;
+ // Another engine, or a repair that keeps directory values as they are,
can register
+ // spellings a typed literal never formats to.
+ write_ids(&temp_dir.path().join("month=01/active=TRUE"), &[1]);
+ write_ids(&temp_dir.path().join("month=2/active=false"), &[2]);
+ table
+ .register(&[
+ &[("month", "01"), ("active", "TRUE")],
+ &[("month", "2"), ("active", "false")],
+ ])
+ .await;
+
+ for (predicate, expected) in [
+ ("month = 1", vec![1]),
+ ("month = 1 AND active = true", vec![1]),
+ ("month = 2 AND active = false", vec![2]),
+ ] {
+ assert_eq!(table.ids(predicate).await, expected, "{predicate}");
+ }
+ // A pattern built from `month = 1` would have dropped `month=01` on the
catalog side.
+ assert!(table
+ .name_patterns_since((0, 0))
+ .iter()
+ .all(Option::is_none));
+}
+
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn
test_managed_scan_filter_on_a_partition_column_reads_the_registered_value() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let table = ManagedTable::new(
+ &temp_dir,
+ &[
+ ("dt", varchar()),
+ ("active", DataType::Boolean(BooleanType::new())),
+ ],
+ )
+ .await;
+ for (dt, active, id) in [("a", "true", 1), ("b", "false", 2)] {
+ write_ids(
+ &temp_dir.path().join(format!("dt={dt}/active={active}")),
+ &[id],
+ );
+ }
+ table
+ .register(&[
+ &[("dt", "a"), ("active", "true")],
+ &[("dt", "b"), ("active", "false")],
+ ])
+ .await;
+
+ // The data files hold no partition columns. A filter the scan cannot turn
into a partition
+ // predicate still has to see the partition's value, not a missing column.
+ for (predicate, expected) in [
+ ("active", vec![1]),
+ ("NOT active", vec![2]),
+ ("upper(dt) = 'B'", vec![2]),
+ ("concat(dt, '-') = 'a-'", vec![1]),
+ ] {
+ assert_eq!(table.ids(predicate).await, expected, "{predicate}");
+ }
+}
+
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_managed_scan_refuses_a_partition_at_a_custom_location() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let external_dir = tempfile::tempdir().unwrap();
+ let table = ManagedTable::new(&temp_dir, &[("dt", varchar())]).await;
+ table.register(&[&[("dt", "a")], &[("dt", "b")]]).await;
+ write_ids(&temp_dir.path().join("dt=a"), &[1]);
+ // Another engine registered dt=b somewhere else; the table directory
still has a stale copy.
+ write_ids(&temp_dir.path().join("dt=b"), &[2]);
+ write_ids(external_dir.path(), &[3]);
+ table.server.set_table_partition_options(
+ DATABASE,
+ TABLE,
+ &spec(&[("dt", "b")]),
+ HashMap::from([(
+ "path".to_string(),
+ format!("file://{}", external_dir.path().display()),
+ )]),
+ );
+
+ // Reading the default directory would return the stale row, so a scan
that reaches the
+ // partition fails instead. One that does not reach it is unaffected.
+ for predicate in ["dt = 'b'", "TRUE"] {
+ let error = table.error(predicate).await;
+ assert!(error.contains("custom location"), "{predicate}: {error}");
+ }
+ assert_eq!(table.ids("dt = 'a'").await, vec![1]);
+}
+
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn
test_managed_scan_fails_rather_than_reading_directories_when_listing_fails() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let table = ManagedTable::new(&temp_dir, &[("dt", varchar())]).await;
+ write_ids(&temp_dir.path().join("dt=a"), &[1]);
+ table.register(&[&[("dt", "a")]]).await;
+ assert_eq!(table.ids("TRUE").await, vec![1]);
+
+ table
+ .server
+ .set_list_partitions_error_status(Some(StatusCode::NOT_IMPLEMENTED));
+
+ // The directory still holds the row, but only the catalog says which
partitions exist.
+ let error = table.error("TRUE").await;
+ assert!(
+ error.to_ascii_lowercase().contains("not implemented"),
+ "{error}"
+ );
+}
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index df068b2b..b4b9a510 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -468,8 +468,9 @@ impl Catalog for RESTCatalog {
.await
{
Ok(found) => partitions.extend(found),
- // A catalog without the lookup still answers the plain
listing, which
- // `list_partitions` falls back from the same way.
+ // A catalog without the lookup still answers the plain
listing. That listing
+ // never stands in directories for the registrations of a
table whose catalog
+ // manages its partitions, so neither does this lookup.
Err(Error::RestApi {
source: RestError::NotImplemented { .. },
}) => {
@@ -489,10 +490,17 @@ impl Catalog for RESTCatalog {
async fn list_partitions(&self, identifier: &Identifier) ->
Result<Vec<Partition>> {
match self.api.list_partitions(identifier).await {
Ok(parts) => Ok(parts),
- Err(Error::RestApi {
- source: RestError::NotImplemented { .. },
- }) => {
+ Err(
+ error @ Error::RestApi {
+ source: RestError::NotImplemented { .. },
+ },
+ ) => {
let table = self.get_table(identifier).await?;
+ // The registrations of a catalog-managed table are its
partitions; the
+ // directories under it are not a substitute for them.
+ if table.has_catalog_managed_partitions() {
+ return Err(error);
+ }
list_partitions_from_file_system(&table).await
}
Err(e) => Err(map_rest_error_for_table(e, identifier)),
@@ -511,10 +519,15 @@ impl Catalog for RESTCatalog {
.await
{
Ok(page) => Ok(page),
- Err(Error::RestApi {
- source: RestError::NotImplemented { .. },
- }) => {
+ Err(
+ error @ Error::RestApi {
+ source: RestError::NotImplemented { .. },
+ },
+ ) => {
let table = self.get_table(identifier).await?;
+ if table.has_catalog_managed_partitions() {
+ return Err(error);
+ }
let parts = list_partitions_from_file_system(&table).await?;
Ok(PagedList::new(parts, None))
}
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index 967441cd..d3114fd4 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -44,6 +44,8 @@ const PARTITION_DEFAULT_NAME_OPTION: &str =
"partition.default-name";
const PARTITION_LEGACY_NAME_OPTION: &str = "partition.legacy-name";
const FORMAT_TABLE_PARTITION_PATH_ONLY_VALUE_OPTION: &str =
"format-table.partition-path-only-value";
+const METASTORE_PARTITIONED_TABLE_OPTION: &str = "metastore.partitioned-table";
+const FORMAT_TABLE_IMPLEMENTATION_OPTION: &str = "format-table.implementation";
const FORMAT_TABLE_SCAN_LIST_PARALLELISM_OPTION: &str =
"format-table.scan.list-parallelism";
const DEFAULT_FORMAT_TABLE_SCAN_LIST_PARALLELISM: usize = 64;
const MAX_FORMAT_TABLE_SCAN_LIST_PARALLELISM: i64 = 1000;
@@ -704,6 +706,22 @@ impl<'a> CoreOptions<'a> {
.unwrap_or(false)
}
+ /// Whether the catalog manages this Format Table's partitions
(`metastore.partitioned-table`).
+ pub(crate) fn partitioned_table_in_metastore(&self) -> bool {
+ self.options
+ .get(METASTORE_PARTITIONED_TABLE_OPTION)
+ .map(|value| value.eq_ignore_ascii_case("true"))
+ .unwrap_or(false)
+ }
+
+ /// Whether the engine's own file source reads this Format Table instead
of Paimon's reader
+ /// (`format-table.implementation=engine`).
+ pub(crate) fn format_table_implementation_is_engine(&self) -> bool {
+ self.options
+ .get(FORMAT_TABLE_IMPLEMENTATION_OPTION)
+ .is_some_and(|value| value.eq_ignore_ascii_case("engine"))
+ }
+
pub fn global_index_enabled(&self) -> bool {
self.options
.get(GLOBAL_INDEX_ENABLED_OPTION)
@@ -2275,6 +2293,28 @@ mod tests {
assert!(core.format_table_partition_only_value_in_path());
}
+ #[test]
+ fn test_catalog_managed_format_table_options() {
+ let empty = HashMap::new();
+ let core = CoreOptions::new(&empty);
+ assert!(!core.partitioned_table_in_metastore());
+ assert!(!core.format_table_implementation_is_engine());
+
+ let options = HashMap::from([
+ (
+ METASTORE_PARTITIONED_TABLE_OPTION.to_string(),
+ "TrUe".to_string(),
+ ),
+ (
+ FORMAT_TABLE_IMPLEMENTATION_OPTION.to_string(),
+ "ENGINE".to_string(),
+ ),
+ ]);
+ let core = CoreOptions::new(&options);
+ assert!(core.partitioned_table_in_metastore());
+ assert!(core.format_table_implementation_is_engine());
+ }
+
#[test]
fn test_format_table_scan_list_parallelism() {
let parallelism = |value: Option<&str>| {
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index 8eac9957..efc0634e 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -98,7 +98,9 @@ pub use types::*;
mod partition;
pub use partition::Partition;
mod partition_utils;
-pub(crate) use partition_utils::{bucket_path, bucket_path_under,
PartitionComputer};
+pub(crate) use partition_utils::{
+ bucket_path, bucket_path_under, escape_path_name, PartitionComputer,
+};
mod predicate;
pub(crate) use predicate::datum_cmp;
pub(crate) use predicate::eval_row;
diff --git a/crates/paimon/src/spec/partition_utils.rs
b/crates/paimon/src/spec/partition_utils.rs
index e4245c31..463ba653 100644
--- a/crates/paimon/src/spec/partition_utils.rs
+++ b/crates/paimon/src/spec/partition_utils.rs
@@ -498,7 +498,7 @@ fn format_timestamp_non_legacy(dt: NaiveDateTime,
precision: u32) -> String {
/// Escape a path component following Java `PartitionPathUtils.escapePathName`.
///
/// Characters that need escaping are encoded as `%XX` (uppercase hex).
-fn escape_path_name(path: &str) -> String {
+pub(crate) fn escape_path_name(path: &str) -> String {
if !path.chars().any(needs_escaping) {
return path.to_string();
}
diff --git a/crates/paimon/src/table/format_partition.rs
b/crates/paimon/src/table/format_partition.rs
new file mode 100644
index 00000000..90133d3b
--- /dev/null
+++ b/crates/paimon/src/table/format_partition.rs
@@ -0,0 +1,352 @@
+// 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.
+
+//! Format Table partition names, paths and values, shared by the scan and the
catalog
+//! registrations it reads.
+
+use std::collections::HashMap;
+
+use chrono::NaiveDate;
+
+use crate::spec::{escape_path_name, DataType, Datum};
+
+const UNIX_EPOCH_DAYS_FROM_CE: i32 = 719_163;
+
+/// Generates canonical names and physical paths for Format Table partitions.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct FormatTablePartitionPaths {
+ partition_keys: Vec<String>,
+ only_value_in_path: bool,
+}
+
+impl FormatTablePartitionPaths {
+ /// Create a helper for the declared partition-key order and physical
layout.
+ pub(crate) fn new<I, S>(partition_keys: I, only_value_in_path: bool) ->
Self
+ where
+ I: IntoIterator<Item = S>,
+ S: Into<String>,
+ {
+ Self {
+ partition_keys:
partition_keys.into_iter().map(Into::into).collect(),
+ only_value_in_path,
+ }
+ }
+
+ /// Return the canonical logical partition name (`key=value/...`).
+ pub(crate) fn partition_name(&self, spec: &HashMap<String, String>) ->
crate::Result<String> {
+ let values = self.ordered_values(spec)?;
+ Ok(self
+ .partition_keys
+ .iter()
+ .zip(values)
+ .map(|(key, value)| format!("{}={}", escape_path_name(key),
escape_path_name(value)))
+ .collect::<Vec<_>>()
+ .join("/"))
+ }
+
+ /// Build the partition-name pattern that selects the partitions whose
leading values
+ /// are `leading_values`, to push down to a partition-managing catalog.
+ ///
+ /// Pattern contract, shared by every engine talking to the catalog:
partition names are
+ /// the escaped `key=value` form joined by `/`, `%` is the only wildcard
and has no
+ /// escape sequence, and `_` stays literal. Values covering every
partition key give the
+ /// one exact name; a shorter prefix is suffixed with `%`.
+ ///
+ /// `None` means pushdown has to be skipped and the caller must list every
partition:
+ /// there are no leading values, one of them is blank, or escaping
produced a literal
+ /// `%` that the contract cannot express.
+ ///
+ /// Mirrors Java `PartitionPathUtils.buildPartitionNamePrefixPattern`.
+ pub(crate) fn name_prefix_pattern(&self, leading_values: &[String]) ->
Option<String> {
+ if leading_values.is_empty() || leading_values.len() >
self.partition_keys.len() {
+ return None;
+ }
+ let mut segments = Vec::with_capacity(leading_values.len());
+ for (key, value) in self.partition_keys.iter().zip(leading_values) {
+ if value.trim().is_empty() {
+ return None;
+ }
+ segments.push(format!(
+ "{}={}",
+ escape_path_name(key),
+ escape_path_name(value)
+ ));
+ }
+ let pattern = segments.join("/");
+ if pattern.contains('%') {
+ return None;
+ }
+ if leading_values.len() == self.partition_keys.len() {
+ Some(pattern)
+ } else {
+ Some(format!("{pattern}/%"))
+ }
+ }
+
+ /// Return the physical partition path relative to the table location.
+ pub(crate) fn relative_path(&self, spec: &HashMap<String, String>) ->
crate::Result<String> {
+ if !self.only_value_in_path {
+ return self.partition_name(spec);
+ }
+ Ok(self
+ .ordered_values(spec)?
+ .into_iter()
+ .map(escape_path_name)
+ .collect::<Vec<_>>()
+ .join("/"))
+ }
+
+ fn ordered_values<'a>(&self, spec: &'a HashMap<String, String>) ->
crate::Result<Vec<&'a str>> {
+ if spec.len() != self.partition_keys.len() {
+ return Err(crate::Error::DataInvalid {
+ message: self.invalid_partition_keys_message(spec),
+ source: None,
+ });
+ }
+
+ let mut values = Vec::with_capacity(self.partition_keys.len());
+ for key in &self.partition_keys {
+ let Some(value) = spec.get(key).map(String::as_str) else {
+ return Err(crate::Error::DataInvalid {
+ message: self.invalid_partition_keys_message(spec),
+ source: None,
+ });
+ };
+ if value.is_empty() || (self.only_value_in_path && matches!(value,
"." | "..")) {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Partition value {value:?} cannot be used as a
partition path component"
+ ),
+ source: None,
+ });
+ }
+ values.push(value);
+ }
+ Ok(values)
+ }
+
+ fn invalid_partition_keys_message(&self, spec: &HashMap<String, String>)
-> String {
+ let mut actual_keys = spec.keys().collect::<Vec<_>>();
+ actual_keys.sort();
+ format!(
+ "Partition spec must contain exactly keys {:?}, but contains
{actual_keys:?}",
+ self.partition_keys
+ )
+ }
+}
+
+/// Parse a raw Format Table partition value from a path or catalog
registration.
+pub(crate) fn parse_format_partition_value(value: &str, data_type: &DataType)
-> Option<Datum> {
+ match data_type {
+ DataType::Boolean(_) => parse_partition_bool(value).map(Datum::Bool),
+ DataType::TinyInt(_) => value.parse::<i8>().ok().map(Datum::TinyInt),
+ DataType::SmallInt(_) =>
value.parse::<i16>().ok().map(Datum::SmallInt),
+ DataType::Int(_) => value.parse::<i32>().ok().map(Datum::Int),
+ DataType::BigInt(_) => value.parse::<i64>().ok().map(Datum::Long),
+ DataType::Char(_) | DataType::VarChar(_) =>
Some(Datum::String(value.to_string())),
+ DataType::Date(_) => parse_partition_date(value).map(Datum::Date),
+ DataType::Time(_) => value.parse::<i32>().ok().map(Datum::Time),
+ _ => None,
+ }
+}
+
+/// Format a typed value for Format Table partition metadata and paths.
+pub(crate) fn format_partition_value(
+ datum: &Datum,
+ data_type: &DataType,
+ default_partition_name: &str,
+ legacy_partition_name: bool,
+) -> Option<String> {
+ match (datum, data_type) {
+ (Datum::Bool(value), DataType::Boolean(_)) => Some(value.to_string()),
+ (Datum::TinyInt(value), DataType::TinyInt(_)) =>
Some(value.to_string()),
+ (Datum::SmallInt(value), DataType::SmallInt(_)) =>
Some(value.to_string()),
+ (Datum::Int(value), DataType::Int(_)) => Some(value.to_string()),
+ (Datum::Long(value), DataType::BigInt(_)) => Some(value.to_string()),
+ (Datum::String(value), DataType::Char(_) | DataType::VarChar(_)) => {
+ if value.trim().is_empty() {
+ Some(default_partition_name.to_string())
+ } else {
+ Some(value.clone())
+ }
+ }
+ (Datum::Date(value), DataType::Date(_)) => {
+ if legacy_partition_name {
+ Some(value.to_string())
+ } else {
+ format_partition_date(*value)
+ }
+ }
+ (Datum::Time(value), DataType::Time(_)) => Some(value.to_string()),
+ _ => None,
+ }
+}
+
+/// Accept the boolean spellings Java accepts, so a partition another engine
registered as
+/// `TRUE`, `t`, `yes` or `1` reads back here instead of failing as invalid
metadata.
+///
+/// Mirrors Java `BinaryStringUtils.toBoolean`.
+fn parse_partition_bool(value: &str) -> Option<bool> {
+ const TRUE_VALUES: [&str; 5] = ["t", "true", "y", "yes", "1"];
+ const FALSE_VALUES: [&str; 5] = ["f", "false", "n", "no", "0"];
+ if TRUE_VALUES
+ .iter()
+ .any(|candidate| value.eq_ignore_ascii_case(candidate))
+ {
+ return Some(true);
+ }
+ if FALSE_VALUES
+ .iter()
+ .any(|candidate| value.eq_ignore_ascii_case(candidate))
+ {
+ return Some(false);
+ }
+ None
+}
+
+fn parse_partition_date(value: &str) -> Option<i32> {
+ if let Ok(epoch_days) = value.parse::<i32>() {
+ return Some(epoch_days);
+ }
+ let date = NaiveDate::parse_from_str(value, "%Y-%m-%d").ok()?;
+ let epoch = NaiveDate::from_num_days_from_ce_opt(UNIX_EPOCH_DAYS_FROM_CE)?;
+ date.signed_duration_since(epoch).num_days().try_into().ok()
+}
+
+fn format_partition_date(epoch_days: i32) -> Option<String> {
+
NaiveDate::from_num_days_from_ce_opt(epoch_days.checked_add(UNIX_EPOCH_DAYS_FROM_CE)?)
+ .map(|date| date.format("%Y-%m-%d").to_string())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::{BooleanType, DateType};
+
+ #[test]
+ fn test_parse_format_partition_value() {
+ assert_eq!(
+ parse_format_partition_value("true",
&DataType::Boolean(BooleanType::new())),
+ Some(Datum::Bool(true))
+ );
+ assert_eq!(
+ parse_format_partition_value("2026-07-22",
&DataType::Date(DateType::new())),
+ Some(Datum::Date(20_656))
+ );
+ assert_eq!(
+ parse_format_partition_value("20656",
&DataType::Date(DateType::new())),
+ Some(Datum::Date(20_656))
+ );
+ }
+
+ #[test]
+ fn test_parse_format_partition_bool_accepts_every_java_spelling() {
+ let boolean = DataType::Boolean(BooleanType::new());
+ for value in ["t", "T", "true", "TRUE", "True", "y", "YES", "1"] {
+ assert_eq!(
+ parse_format_partition_value(value, &boolean),
+ Some(Datum::Bool(true)),
+ "{value} should read as true"
+ );
+ }
+ for value in ["f", "F", "false", "FALSE", "False", "n", "NO", "0"] {
+ assert_eq!(
+ parse_format_partition_value(value, &boolean),
+ Some(Datum::Bool(false)),
+ "{value} should read as false"
+ );
+ }
+ for value in ["", "2", "tru", "yes please", "null"] {
+ assert_eq!(
+ parse_format_partition_value(value, &boolean),
+ None,
+ "{value} is not a boolean"
+ );
+ }
+ }
+
+ #[test]
+ fn test_partition_paths_escape_values_and_honor_layout() {
+ let spec = HashMap::from([
+ ("dt".to_string(), "2026/07=22".to_string()),
+ ("hour".to_string(), "10".to_string()),
+ ]);
+
+ let keyed = FormatTablePartitionPaths::new(["dt", "hour"], false);
+ assert_eq!(
+ keyed.partition_name(&spec).unwrap(),
+ "dt=2026%2F07%3D22/hour=10"
+ );
+ assert_eq!(
+ keyed.relative_path(&spec).unwrap(),
+ "dt=2026%2F07%3D22/hour=10"
+ );
+
+ let value_only = FormatTablePartitionPaths::new(["dt", "hour"], true);
+ assert_eq!(
+ value_only.partition_name(&spec).unwrap(),
+ "dt=2026%2F07%3D22/hour=10"
+ );
+ assert_eq!(
+ value_only.relative_path(&spec).unwrap(),
+ "2026%2F07%3D22/10"
+ );
+
+ // A spec that does not name exactly the partition keys has no path.
+ let missing_hour = HashMap::from([("dt".to_string(),
"a".to_string())]);
+ assert!(keyed.relative_path(&missing_hour).is_err());
+ let traversal = HashMap::from([
+ ("dt".to_string(), "..".to_string()),
+ ("hour".to_string(), "10".to_string()),
+ ]);
+ assert!(value_only.relative_path(&traversal).is_err());
+ }
+
+ #[test]
+ fn test_name_prefix_pattern() {
+ let paths = FormatTablePartitionPaths::new(["dt".to_string(),
"hh".to_string()], false);
+
+ // A complete prefix names one partition, a shorter one takes the
wildcard.
+ assert_eq!(
+ paths.name_prefix_pattern(&["20260722".to_string(),
"10".to_string()]),
+ Some("dt=20260722/hh=10".to_string())
+ );
+ assert_eq!(
+ paths.name_prefix_pattern(&["20260722".to_string()]),
+ Some("dt=20260722/%".to_string())
+ );
+
+ // The physical layout never changes the pattern: catalog names are
always key=value.
+ let value_only = FormatTablePartitionPaths::new(["dt".to_string(),
"hh".to_string()], true);
+ assert_eq!(
+ value_only.name_prefix_pattern(&["20260722".to_string()]),
+ Some("dt=20260722/%".to_string())
+ );
+
+ // Nothing to push down.
+ assert_eq!(paths.name_prefix_pattern(&[]), None);
+ assert_eq!(paths.name_prefix_pattern(&[" ".to_string()]), None);
+ assert_eq!(
+ paths.name_prefix_pattern(&["a".to_string(), "b".to_string(),
"c".to_string()]),
+ None
+ );
+
+ // Escaping a value would inject the pattern's only wildcard, so
pushdown is skipped
+ // rather than silently widened.
+ assert_eq!(paths.name_prefix_pattern(&["2026/07".to_string()]), None);
+ }
+}
diff --git a/crates/paimon/src/table/format_table_scan.rs
b/crates/paimon/src/table/format_table_scan.rs
index 569ac071..aabadcec 100644
--- a/crates/paimon/src/table/format_table_scan.rs
+++ b/crates/paimon/src/table/format_table_scan.rs
@@ -17,15 +17,21 @@
//! Scan implementation for Java-compatible `type=format-table` metadata.
-use super::{Plan, ScanTrace, Table};
+use std::collections::{HashMap, HashSet};
+
+use super::format_partition::{
+ format_partition_value, parse_format_partition_value,
FormatTablePartitionPaths,
+};
+use super::{Plan, RESTEnv, ScanTrace, Table};
+use crate::api::RestError;
use crate::spec::stats::BinaryTableStats;
use crate::spec::{
- extract_datum, BinaryRow, BinaryRowBuilder, CoreOptions, DataField,
DataFileMeta, DataType,
- Datum, PartitionComputer, Predicate, PredicateOperator,
+ escape_path_name, extract_datum, BinaryRow, BinaryRowBuilder, CoreOptions,
DataField,
+ DataFileMeta, DataType, Datum, Partition, PartitionComputer, Predicate,
PredicateOperator,
+ PATH_OPTION,
};
use crate::table::partition_filter::PartitionFilter;
use crate::table::source::{DataSplitBuilder, RowRange};
-use chrono::NaiveDate;
use futures::{StreamExt, TryStreamExt};
#[derive(Debug, Clone)]
@@ -90,7 +96,7 @@ impl<'a> FormatTableScan<'a> {
let partition_fields = self.table.schema().partition_fields();
let table_depth = path_segments(&table_path).len();
- let scan_roots = self.scan_roots(&core_options, &table_path)?;
+ let scan_roots = self.scan_roots(&core_options, &table_path).await?;
// A table with many partitions pays one listing per partition, so
they run concurrently.
// `buffered` keeps the roots in order and stops at the first failure.
let table_path = table_path.as_str();
@@ -149,7 +155,7 @@ impl<'a> FormatTableScan<'a> {
Ok(Plan::new(splits))
}
- fn scan_roots(
+ async fn scan_roots(
&self,
core_options: &CoreOptions<'_>,
table_path: &str,
@@ -162,6 +168,15 @@ impl<'a> FormatTableScan<'a> {
partition: BinaryRow::new(0),
}]);
}
+ if let Some(rest_env) = self
+ .table
+ .rest_env()
+ .filter(|_| self.table.has_catalog_managed_partitions())
+ {
+ return self
+ .catalog_managed_scan_roots(rest_env, table_path,
partition_keys, &partition_fields)
+ .await;
+ }
let Some(PartitionFilter::PartitionSet { partitions, .. }) =
&self.partition_filter else {
if let Some(PartitionFilter::Predicate(predicate)) =
&self.partition_filter {
@@ -216,6 +231,138 @@ impl<'a> FormatTableScan<'a> {
Ok(roots)
}
+ /// One scan root per registered partition this scan reaches, never one
discovered from the
+ /// directory tree.
+ async fn catalog_managed_scan_roots(
+ &self,
+ rest_env: &RESTEnv,
+ table_path: &str,
+ partition_keys: &[String],
+ partition_fields: &[DataField],
+ ) -> crate::Result<Vec<ScanRoot>> {
+ let core_options = CoreOptions::new(self.table.schema().options());
+ let partition_paths = FormatTablePartitionPaths::new(
+ partition_keys.iter().cloned(),
+ core_options.format_table_partition_only_value_in_path(),
+ );
+ let default_partition_name = core_options.partition_default_name();
+ // Ask the catalog only for the partitions the filter can reach.
Downloading every
+ // registration of a table with many partitions is what dominates
planning time,
+ // and the local match below still decides what is actually scanned.
+ let pattern = match &self.partition_filter {
+ Some(filter) => {
+ let mut leading_values = leading_equality_partition_values(
+ filter,
+ partition_fields,
+ default_partition_name,
+ core_options.legacy_partition_name(),
+ )?;
+ // A name pattern compares spellings, so it can only stand in
for an equality on a
+ // column whose values have one. Another engine may register
`month=01` or
+ // `active=TRUE`, which would fall out of a pattern built from
`month = 1` or
+ // `active = true` before the typed match below ever saw them.
+ let single_spelling = partition_fields
+ .iter()
+ .take_while(|field| {
+ matches!(field.data_type(), DataType::Char(_) |
DataType::VarChar(_))
+ })
+ .count();
+ leading_values.truncate(single_spelling);
+ partition_paths.name_prefix_pattern(&leading_values)
+ }
+ None => None,
+ };
+ let filter = match &self.partition_filter {
+ Some(PartitionFilter::Predicate(predicate)) => {
+ partition_filter_json(predicate).map(|filter|
filter.to_string())
+ }
+ _ => None,
+ };
+ let partitions = self
+ .list_catalog_partitions(rest_env, pattern.as_deref(),
filter.as_deref())
+ .await?;
+ let mut seen_paths = HashSet::with_capacity(partitions.len());
+ let mut roots = Vec::with_capacity(partitions.len());
+ for partition in partitions {
+ let partition_path = partition_paths
+ .relative_path(&partition.spec)
+ .map_err(|error|
self.invalid_catalog_partition_metadata(error))?;
+ if !seen_paths.insert(partition_path.clone()) {
+ continue;
+ }
+ let path = join_path(table_path, &partition_path);
+ let row = partition_row_from_catalog_spec(
+ &partition.spec,
+ partition_fields,
+ partition_keys,
+ default_partition_name,
+ )
+ .map_err(|error| self.invalid_catalog_partition_metadata(error))?;
+ if !self.partition_matches(&row)? {
+ continue;
+ }
+ // The Rust reader cannot resolve a partition's own location yet,
and reading the
+ // default directory in its place would return whatever happens to
be there.
+ if partition
+ .options
+ .as_ref()
+ .is_some_and(|options| options.contains_key(PATH_OPTION))
+ {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Partition {:?} of Format Table {} is registered at a
custom location, \
+ which the Rust reader does not support yet",
+ partition.spec,
+ self.table.identifier().full_name()
+ ),
+ });
+ }
+ roots.push(ScanRoot {
+ path,
+ partition: row,
+ });
+ }
+ roots.sort_by(|left, right| left.path.cmp(&right.path));
+ Ok(roots)
+ }
+
+ /// The registered partitions the catalog returns for this scan's pushdown
hints.
+ ///
+ /// A catalog that cannot list by filter is still asked by pattern, so the
catalog stays the
+ /// source of the partition set; the typed match on the result decides
what is scanned.
+ async fn list_catalog_partitions(
+ &self,
+ rest_env: &RESTEnv,
+ pattern: Option<&str>,
+ filter: Option<&str>,
+ ) -> crate::Result<Vec<Partition>> {
+ let api = rest_env.api();
+ let identifier = rest_env.identifier();
+ if let Some(filter) = filter {
+ match api
+ .list_partitions_by_filter(identifier, filter, pattern)
+ .await
+ {
+ Err(crate::Error::RestApi {
+ source: RestError::NotImplemented { .. },
+ }) => {}
+ result => return result,
+ }
+ }
+ api.list_partitions_by_name_pattern(identifier, pattern)
+ .await
+ }
+
+ fn invalid_catalog_partition_metadata(&self, source: crate::Error) ->
crate::Error {
+ crate::Error::DataInvalid {
+ message: format!(
+ "Catalog returned invalid partition metadata for Format Table
{}",
+ self.table.identifier().full_name()
+ ),
+ source: Some(Box::new(source)),
+ }
+ }
+
async fn list_status_recursive_if_exists(
&self,
path: &str,
@@ -398,8 +545,122 @@ fn leading_equality_partition_path(
legacy_partition_name: bool,
only_value_in_path: bool,
) -> Option<String> {
+ let values = leading_equality_values_from_predicate(
+ predicate,
+ partition_fields,
+ default_partition_name,
+ legacy_partition_name,
+ );
+ if values.is_empty() {
+ return None;
+ }
+ let segments = partition_keys
+ .iter()
+ .zip(&values)
+ .map(|(key, value)| {
+ if only_value_in_path {
+ escape_path_name(value)
+ } else {
+ format!("{}={}", escape_path_name(key),
escape_path_name(value))
+ }
+ })
+ .collect::<Vec<_>>();
+ Some(join_path(table_path, &segments.join("/")))
+}
+
+/// The part of a partition predicate that can be sent to the catalog as a
filter.
+///
+/// An AND keeps the children that have a wire form, since leaving out a
conjunct only widens
+/// what the catalog may return; anything else is sent whole or not at all.
Nothing is sent for a
+/// predicate that selects everything.
+///
+/// Mirrors Java `FormatTableScan.extractPartitionPredicate`.
+fn partition_filter_json(predicate: &Predicate) -> Option<serde_json::Value> {
+ match predicate {
+ Predicate::AlwaysTrue => None,
+ Predicate::And(children) => {
+ let mut pushed = children
+ .iter()
+ .filter(|child| child.to_rest_json().is_some())
+ .cloned()
+ .collect::<Vec<_>>();
+ match pushed.len() {
+ 0 => None,
+ 1 => pushed.pop().and_then(|child| child.to_rest_json()),
+ _ => Predicate::And(pushed).to_rest_json(),
+ }
+ }
+ other => other.to_rest_json(),
+ }
+}
+
+/// The leading run of partition values this filter pins to a single value, in
+/// partition-key order, formatted the way the catalog and the partition path
spell them.
+///
+/// Only a leading run is useful: a partition path prefix and the
partition-name pattern a
+/// catalog prunes on can express nothing else. A null value has no such
spelling, so the
+/// run stops there.
+fn leading_equality_partition_values(
+ filter: &PartitionFilter,
+ partition_fields: &[DataField],
+ default_partition_name: &str,
+ legacy_partition_name: bool,
+) -> crate::Result<Vec<String>> {
+ match filter {
+ PartitionFilter::Predicate(predicate) =>
Ok(leading_equality_values_from_predicate(
+ predicate,
+ partition_fields,
+ default_partition_name,
+ legacy_partition_name,
+ )),
+ // An enumerated partition set still pins a prefix whenever its rows
agree on one,
+ // which is what `dt = 'a' AND hh IN ('10', '11')` collapses to.
+ PartitionFilter::PartitionSet { partitions, .. } => {
+ let mut common: Option<Vec<String>> = None;
+ for serialized in partitions {
+ let row = BinaryRow::from_serialized_bytes(serialized)?;
+ let mut values = Vec::with_capacity(partition_fields.len());
+ for (index, field) in partition_fields.iter().enumerate() {
+ let Some(datum) = extract_datum(&row, index,
field.data_type())? else {
+ break;
+ };
+ let Some(value) = format_partition_value(
+ &datum,
+ field.data_type(),
+ default_partition_name,
+ legacy_partition_name,
+ ) else {
+ break;
+ };
+ values.push(value);
+ }
+ common = Some(match common {
+ None => values,
+ Some(common) => common
+ .into_iter()
+ .zip(values)
+ .take_while(|(left, right)| left == right)
+ .map(|(left, _)| left)
+ .collect(),
+ });
+ if common.as_ref().is_some_and(Vec::is_empty) {
+ break;
+ }
+ }
+ Ok(common.unwrap_or_default())
+ }
+ }
+}
+
+/// Mirrors Java
`FormatTableScan.extractLeadingEqualityPartitionSpecWhenOnlyAnd`.
+fn leading_equality_values_from_predicate(
+ predicate: &Predicate,
+ partition_fields: &[DataField],
+ default_partition_name: &str,
+ legacy_partition_name: bool,
+) -> Vec<String> {
+ let mut pinned: Vec<Option<&Datum>> = vec![None; partition_fields.len()];
let predicates = predicate.clone().split_and();
- let mut values: Vec<Option<&Datum>> = vec![None; partition_keys.len()];
for predicate in &predicates {
let Predicate::Leaf {
index,
@@ -410,38 +671,27 @@ fn leading_equality_partition_path(
else {
continue;
};
- if *index < values.len() {
- values[*index] = literals.first();
+ if *index < pinned.len() {
+ pinned[*index] = literals.first();
}
}
- let mut segments = Vec::new();
- for (idx, key) in partition_keys.iter().enumerate() {
- let Some(datum) = values[idx] else {
+ let mut values = Vec::new();
+ for (field, datum) in partition_fields.iter().zip(pinned) {
+ let Some(datum) = datum else {
break;
};
- let value = partition_value_from_datum(
+ let Some(value) = format_partition_value(
datum,
- partition_fields[idx].data_type(),
+ field.data_type(),
default_partition_name,
legacy_partition_name,
- )?;
- if only_value_in_path {
- segments.push(escape_path_name(&value));
- } else {
- segments.push(format!(
- "{}={}",
- escape_path_name(key),
- escape_path_name(&value)
- ));
- }
- }
-
- if segments.is_empty() {
- None
- } else {
- Some(join_path(table_path, &segments.join("/")))
+ ) else {
+ break;
+ };
+ values.push(value);
}
+ values
}
fn partition_path_from_row(
@@ -455,7 +705,7 @@ fn partition_path_from_row(
for (idx, field) in partition_fields.iter().enumerate() {
let value = match extract_datum(row, idx, field.data_type())? {
None => default_partition_name.to_string(),
- Some(datum) => partition_value_from_datum(
+ Some(datum) => format_partition_value(
&datum,
field.data_type(),
default_partition_name,
@@ -481,78 +731,6 @@ fn partition_path_from_row(
Ok(segments.join("/"))
}
-fn partition_value_from_datum(
- datum: &Datum,
- data_type: &DataType,
- default_partition_name: &str,
- legacy_partition_name: bool,
-) -> Option<String> {
- match (datum, data_type) {
- (Datum::Bool(value), DataType::Boolean(_)) => Some(value.to_string()),
- (Datum::TinyInt(value), DataType::TinyInt(_)) =>
Some(value.to_string()),
- (Datum::SmallInt(value), DataType::SmallInt(_)) =>
Some(value.to_string()),
- (Datum::Int(value), DataType::Int(_)) => Some(value.to_string()),
- (Datum::Long(value), DataType::BigInt(_)) => Some(value.to_string()),
- (Datum::String(value), DataType::Char(_) | DataType::VarChar(_)) => {
- if value.trim().is_empty() {
- Some(default_partition_name.to_string())
- } else {
- Some(value.clone())
- }
- }
- (Datum::Date(value), DataType::Date(_)) => {
- if legacy_partition_name {
- Some(value.to_string())
- } else {
- Some(format_partition_date(*value))
- }
- }
- (Datum::Time(value), DataType::Time(_)) => Some(value.to_string()),
- _ => None,
- }
-}
-
-fn format_partition_date(epoch_days: i32) -> String {
- let date = NaiveDate::from_num_days_from_ce_opt(epoch_days + 719_163)
- .unwrap_or(NaiveDate::from_ymd_opt(1970, 1, 1).unwrap());
- date.format("%Y-%m-%d").to_string()
-}
-
-fn escape_path_name(path: &str) -> String {
- let mut result = String::with_capacity(path.len());
- for byte in path.bytes() {
- if should_escape(byte) {
- result.push('%');
- result.push_str(&format!("{byte:02X}"));
- } else {
- result.push(byte as char);
- }
- }
- result
-}
-
-fn should_escape(byte: u8) -> bool {
- byte <= 0x1F
- || byte >= 0x7F
- || matches!(
- byte,
- b'"' | b'#'
- | b'%'
- | b'\''
- | b'*'
- | b'/'
- | b':'
- | b'='
- | b'?'
- | b'\\'
- | b'\x7F'
- | b'{'
- | b'['
- | b']'
- | b'^'
- )
-}
-
fn partition_row_from_path(
table_path: &str,
file_parent: &str,
@@ -605,7 +783,8 @@ fn partition_row_from_path(
builder.set_null_at(idx);
continue;
}
- let Some(datum) = parse_partition_datum(value,
partition_fields[idx].data_type()) else {
+ let Some(datum) = parse_format_partition_value(value,
partition_fields[idx].data_type())
+ else {
return Ok(None);
};
builder.write_datum(idx, &datum, partition_fields[idx].data_type());
@@ -613,6 +792,36 @@ fn partition_row_from_path(
Ok(Some(builder.build()))
}
+fn partition_row_from_catalog_spec(
+ spec: &HashMap<String, String>,
+ partition_fields: &[DataField],
+ partition_keys: &[String],
+ default_partition_name: &str,
+) -> crate::Result<BinaryRow> {
+ let mut builder = BinaryRowBuilder::new(partition_fields.len() as i32);
+ for (index, (key, field)) in
partition_keys.iter().zip(partition_fields).enumerate() {
+ let value = spec.get(key).ok_or_else(|| crate::Error::DataInvalid {
+ message: format!("Catalog partition is missing column '{key}'"),
+ source: None,
+ })?;
+ if value == default_partition_name {
+ builder.set_null_at(index);
+ continue;
+ }
+ let datum = parse_format_partition_value(value,
field.data_type()).ok_or_else(|| {
+ crate::Error::DataInvalid {
+ message: format!(
+ "Invalid catalog partition value {value:?} for column
'{key}' with type {:?}",
+ field.data_type()
+ ),
+ source: None,
+ }
+ })?;
+ builder.write_datum(index, &datum, field.data_type());
+ }
+ Ok(builder.build())
+}
+
fn partition_segment_value(segment: &str, key: &str) -> Option<String> {
let (segment_key, segment_value) = segment.split_once('=')?;
if unescape_path_name(segment_key)? == key {
@@ -652,31 +861,6 @@ fn hex_value(byte: u8) -> Option<u8> {
}
}
-fn parse_partition_datum(value: &str, data_type: &DataType) -> Option<Datum> {
- match data_type {
- DataType::Boolean(_) => value.parse::<bool>().ok().map(Datum::Bool),
- DataType::TinyInt(_) => value.parse::<i8>().ok().map(Datum::TinyInt),
- DataType::SmallInt(_) =>
value.parse::<i16>().ok().map(Datum::SmallInt),
- DataType::Int(_) => value.parse::<i32>().ok().map(Datum::Int),
- DataType::BigInt(_) => value.parse::<i64>().ok().map(Datum::Long),
- DataType::Char(_) | DataType::VarChar(_) =>
Some(Datum::String(value.to_string())),
- DataType::Date(_) => parse_partition_date(value).map(Datum::Date),
- DataType::Time(_) => value.parse::<i32>().ok().map(Datum::Time),
- _ => None,
- }
-}
-
-fn parse_partition_date(value: &str) -> Option<i32> {
- if let Ok(epoch_days) = value.parse::<i32>() {
- return Some(epoch_days);
- }
- let date = NaiveDate::parse_from_str(value, "%Y-%m-%d").ok()?;
- date.signed_duration_since(NaiveDate::from_ymd_opt(1970, 1, 1).unwrap())
- .num_days()
- .try_into()
- .ok()
-}
-
fn supported_format_table_formats() -> Vec<&'static str> {
vec![
"parquet",
@@ -1011,4 +1195,79 @@ mod tests {
);
}
}
+
+ #[test]
+ fn test_partition_filter_json_sends_what_can_only_widen_the_result() {
+ use crate::spec::{DateType, PredicateBuilder};
+
+ let fields = vec![
+ DataField::new(
+ 0,
+ "dt".to_string(),
+ DataType::VarChar(VarCharType::default()),
+ ),
+ DataField::new(1, "hh".to_string(), DataType::Int(IntType::new())),
+ DataField::new(2, "day".to_string(),
DataType::Date(DateType::new())),
+ ];
+ let builder = PredicateBuilder::new(&fields);
+ let dt = builder.equal("dt", Datum::String("a".to_string())).unwrap();
+ let hh = builder.greater_than("hh", Datum::Int(10)).unwrap();
+ let day = builder.equal("day", Datum::Date(20_656)).unwrap();
+ let sent = |predicate: &Predicate| {
+ partition_filter_json(predicate).map(|json| {
+ Predicate::from_rest_json(&json.to_string(), &fields)
+ .unwrap()
+ .to_string()
+ })
+ };
+
+ // A DATE literal has no wire form, so the AND goes without it.
+ assert_eq!(
+ sent(&Predicate::and(vec![dt.clone(), hh.clone(), day.clone()])),
+ Some(Predicate::and(vec![dt.clone(), hh]).to_string())
+ );
+ assert_eq!(
+ sent(&Predicate::and(vec![dt.clone(), day.clone()])),
+ Some(dt.to_string())
+ );
+ // Leaving out a child of an OR would narrow it, so it is sent whole
or not at all.
+ assert_eq!(sent(&Predicate::or(vec![dt.clone(), day])), None);
+ assert_eq!(sent(&Predicate::Not(Box::new(dt))), None);
+ assert_eq!(sent(&Predicate::AlwaysTrue), None);
+ }
+
+ #[test]
+ fn test_leading_equality_path_escapes_values_as_partition_paths_do() {
+ use crate::spec::PredicateBuilder;
+
+ let fields = vec![DataField::new(
+ 0,
+ "dt".to_string(),
+ DataType::VarChar(VarCharType::default()),
+ )];
+ let keys = vec!["dt".to_string()];
+ let path = |value: &str, only_value_in_path: bool| {
+ let predicate = PredicateBuilder::new(&fields)
+ .equal("dt", Datum::String(value.to_string()))
+ .unwrap();
+ leading_equality_partition_path(
+ "memory:/t",
+ &keys,
+ &fields,
+ &predicate,
+ "__DEFAULT_PARTITION__",
+ false,
+ only_value_in_path,
+ )
+ };
+
+ // Only the characters Java escapes are escaped; a non-ASCII value is
written as it is,
+ // as the directory a writer creates for it.
+ assert_eq!(
+ path("2026/07", false).as_deref(),
+ Some("memory:/t/dt=2026%2F07")
+ );
+ assert_eq!(path("中文", false).as_deref(), Some("memory:/t/dt=中文"));
+ assert_eq!(path("中文", true).as_deref(), Some("memory:/t/中文"));
+ }
}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 32484924..708aa6b3 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -40,6 +40,7 @@ pub mod data_evolution_writer;
mod data_file_reader;
mod data_file_writer;
mod dedicated_format_file_writer;
+mod format_partition;
mod format_read_builder;
mod format_table_read;
mod format_table_scan;
@@ -342,6 +343,15 @@ impl Table {
CoreOptions::new(self.schema.options()).is_format_table()
}
+ /// Whether this table uses catalog-managed Format Table partitions: a
Format Table loaded
+ /// from a REST catalog with `metastore.partitioned-table=true`.
+ pub fn has_catalog_managed_partitions(&self) -> bool {
+ let options = CoreOptions::new(self.schema.options());
+ self.rest_env.is_some()
+ && options.is_format_table()
+ && options.partitioned_table_in_metastore()
+ }
+
/// Create a read builder for scan/read.
///
/// Reference: [pypaimon
FileStoreTable.new_read_builder](https://github.com/apache/paimon/blob/release-1.3/paimon-python/pypaimon/table/file_store_table.py).
@@ -513,6 +523,60 @@ impl Table {
self.copy_with_time_travel_mode(extra, true).await
}
+ /// Refuse dynamic options that would change where a Format Table loaded
from a REST catalog
+ /// takes its partitions from, or, when the catalog manages them, how they
are read.
+ ///
+ /// Mirrors Java `FormatTable.copy`. Java also fixes a Format Table's
type, location and
+ /// format when the table is loaded; this table reads them from its
options, so changing
+ /// them is refused here too.
+ fn ensure_format_table_partition_options_unchanged(
+ &self,
+ extra: &HashMap<String, String>,
+ ) -> Result<()> {
+ let current = CoreOptions::new(self.schema.options());
+ if self.rest_env.is_none() || !current.is_format_table() {
+ return Ok(());
+ }
+ let mut merged_options = self.schema.options().clone();
+ merged_options.extend(
+ extra
+ .iter()
+ .map(|(key, value)| (key.clone(), value.clone())),
+ );
+ let merged = CoreOptions::new(&merged_options);
+ let managed = current.partitioned_table_in_metastore();
+ let changed = if merged.partitioned_table_in_metastore() != managed {
+ Some("metastore.partitioned-table")
+ } else if !managed {
+ None
+ } else if !merged.is_format_table() {
+ Some("type")
+ } else if merged.format_table_partition_only_value_in_path()
+ != current.format_table_partition_only_value_in_path()
+ {
+ Some("format-table.partition-path-only-value")
+ } else if merged.format_table_implementation_is_engine() {
+ Some("format-table.implementation")
+ } else if merged.path() != current.path() {
+ Some("path")
+ } else if merged.file_format() != current.file_format() {
+ Some("file.format")
+ } else {
+ None
+ };
+ match changed {
+ Some(key) => Err(crate::Error::DataInvalid {
+ message: format!(
+ "Dynamic option '{key}' cannot change where Format Table
{} takes its \
+ partitions from, or how it reads them",
+ self.identifier.full_name()
+ ),
+ source: None,
+ }),
+ None => Ok(()),
+ }
+ }
+
async fn copy_with_time_travel_mode(
&self,
extra: HashMap<String, String>,
@@ -521,6 +585,7 @@ impl Table {
// Resolution reads Paimon snapshot paths, so refuse before any IO.
CoreOptions::new(self.schema.options())
.ensure_type_paimon_served(&self.identifier.full_name())?;
+ self.ensure_format_table_partition_options_unchanged(&extra)?;
let mut table = self.copy_with_options(extra);
// Reject unimplemented scan options on the merged view before any IO,
so
// both table-level and per-read options are covered.
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index 133acf69..4e8a1f7c 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -183,6 +183,7 @@ impl RESTEnv {
),
source: None,
})?;
+ validate_catalog_managed_format_table(identifier, &table_schema,
is_external)?;
let uuid = response.id.ok_or_else(|| Error::DataInvalid {
message: format!(
@@ -314,6 +315,42 @@ impl RESTEnv {
}
}
+/// Refuse a Format Table that asks for catalog-managed partitions it cannot
have: an engine
+/// implementation reads the table directory itself, and only an internal
table's partitions
+/// belong to the catalog.
+///
+/// Mirrors Java `CatalogUtils.validateCatalogManagedFormatTablePartitions`.
+fn validate_catalog_managed_format_table(
+ identifier: &Identifier,
+ table_schema: &TableSchema,
+ is_external: bool,
+) -> Result<()> {
+ let options = CoreOptions::new(table_schema.options());
+ if !options.is_format_table() || !options.partitioned_table_in_metastore()
{
+ return Ok(());
+ }
+ if options.format_table_implementation_is_engine() {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Format Table {} cannot set metastore.partitioned-table=true
when \
+ format-table.implementation=engine",
+ identifier.full_name()
+ ),
+ source: None,
+ });
+ }
+ if is_external {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Catalog-managed partitions require an internal Format Table,
but {} is external",
+ identifier.full_name()
+ ),
+ source: None,
+ });
+ }
+ Ok(())
+}
+
fn map_rest_error_for_table(err: Error, identifier: &Identifier) -> Error {
match err {
Error::RestApi {
diff --git a/crates/paimon/tests/mock_server.rs
b/crates/paimon/tests/mock_server.rs
index 934c55de..bd73993b 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -69,6 +69,7 @@ struct MockState {
create_partitions_calls: Vec<(String, String, CreatePartitionsRequest)>,
drop_partitions_calls: Vec<(String, String, DropPartitionsRequest)>,
create_partitions_error_status: Option<StatusCode>,
+ list_partitions_error_status: Option<StatusCode>,
/// ECS metadata role name (for token loader testing)
ecs_role_name: Option<String>,
/// ECS metadata token (for token loader testing)
@@ -958,6 +959,15 @@ impl RESTServer {
);
return (StatusCode::NOT_FOUND, Json(error)).into_response();
}
+ if let Some(status) = inner.list_partitions_error_status {
+ let error = ErrorResponse::new(
+ Some("partition".to_string()),
+ Some(table),
+ Some("Partition listing is not implemented".to_string()),
+ Some(status.as_u16() as i32),
+ );
+ return (status, Json(error)).into_response();
+ }
if inner.partition_page_responses.contains_key(&key) {
let request_index = {
let calls = inner
@@ -1309,6 +1319,11 @@ impl RESTServer {
self.inner.lock().unwrap().create_partitions_error_status = status;
}
+ /// Make the list-partitions endpoint return the given status.
+ pub fn set_list_partitions_error_status(&self, status: Option<StatusCode>)
{
+ self.inner.lock().unwrap().list_partitions_error_status = status;
+ }
+
/// Make the list-partitions-by-names endpoint return the given status.
pub fn set_list_partitions_by_names_error_status(&self, status:
Option<StatusCode>) {
self.inner
@@ -1420,6 +1435,38 @@ impl RESTServer {
inner.partition_list_call_counts.remove(&key);
}
+ /// Set whether a stored table is external.
+ pub fn set_table_external(&self, database: &str, table: &str, is_external:
bool) {
+ let key = format!("{database}.{table}");
+ let mut state = self.inner.lock().unwrap();
+ state
+ .tables
+ .get_mut(&key)
+ .unwrap_or_else(|| panic!("table {key} does not exist"))
+ .is_external = Some(is_external);
+ }
+
+ /// Attach catalog options, such as a custom `path`, to a registered
partition.
+ pub fn set_table_partition_options(
+ &self,
+ database: &str,
+ table: &str,
+ spec: &HashMap<String, String>,
+ options: HashMap<String, String>,
+ ) {
+ let mut inner = self.inner.lock().unwrap();
+ let partition = inner
+ .partitions
+ .get_mut(&format!("{database}.{table}"))
+ .and_then(|partitions| {
+ partitions
+ .iter_mut()
+ .find(|partition| &partition.spec == spec)
+ })
+ .unwrap_or_else(|| panic!("partition {spec:?} is not registered"));
+ partition.options = Some(options);
+ }
+
/// Set explicit list-partitions pages and response tokens in request
order.
pub fn set_table_partition_page_responses(
&self,
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 972fab05..b3f6cf53 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -84,6 +84,33 @@ fn test_schema() -> Schema {
.expect("Failed to build schema")
}
+/// A Parquet Format Table partitioned by `dt` whose partitions the catalog
manages, unless
+/// `options` says otherwise.
+fn format_table_schema(options: &[(&str, &str)]) -> Schema {
+ let mut builder = Schema::builder()
+ .column("dt", DataType::VarChar(VarCharType::new(255).unwrap()))
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .partition_keys(["dt"])
+ .option("type", "format-table")
+ .option("file.format", "parquet")
+ .option("metastore.partitioned-table", "true");
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+}
+
+/// Catalog-managed partitions require an internal table, and the mock serves
external ones.
+fn add_internal_table_with_schema(
+ server: &RESTServer,
+ table: &str,
+ schema: Schema,
+ location: &str,
+) {
+ server.add_table_with_schema("default", table, schema, location);
+ server.set_table_external("default", table, false);
+}
+
fn blob_schema(options: &[(&str, &str)]) -> Schema {
let mut builder = Schema::builder()
.column("id", DataType::Int(IntType::new()))
@@ -484,6 +511,68 @@ async fn
test_rest_catalog_looks_up_partitions_by_names_through_the_listing_when
);
}
+#[cfg(not(windows))]
+#[tokio::test]
+async fn
test_rest_catalog_keeps_managed_partitions_off_the_filesystem_when_listing_is_unsupported()
+{
+ let tmp = tempfile::tempdir().unwrap();
+ let partition_dir = tmp.path().join("dt=2026-07-22");
+ std::fs::create_dir_all(&partition_dir).unwrap();
+ std::fs::write(partition_dir.join("part-0.parquet"), b"listed, never
read").unwrap();
+ let location = format!("file://{}", tmp.path().display());
+ let ctx = setup_catalog(vec!["default"]).await;
+ add_internal_table_with_schema(
+ &ctx.server,
+ "managed_table",
+ format_table_schema(&[]),
+ &location,
+ );
+ add_internal_table_with_schema(
+ &ctx.server,
+ "unmanaged_table",
+ format_table_schema(&[("metastore.partitioned-table", "false")]),
+ &location,
+ );
+ ctx.server
+ .set_list_partitions_error_status(Some(StatusCode::NOT_IMPLEMENTED));
+ ctx.server
+
.set_list_partitions_by_names_error_status(Some(StatusCode::NOT_IMPLEMENTED));
+ let managed = Identifier::new("default", "managed_table");
+ let partition = HashMap::from([("dt".to_string(),
"2026-07-22".to_string())]);
+ let is_not_implemented = |error: paimon::Error| {
+ matches!(
+ error,
+ paimon::Error::RestApi {
+ source: paimon::api::RestError::NotImplemented { .. }
+ }
+ )
+ };
+
+ // The directory is there, but it is not a registration of a
catalog-managed table.
+ assert!(is_not_implemented(
+ ctx.catalog.list_partitions(&managed).await.unwrap_err()
+ ));
+ assert!(is_not_implemented(
+ ctx.catalog
+ .list_partitions_paged(&managed, None, None)
+ .await
+ .unwrap_err()
+ ));
+ assert!(is_not_implemented(
+ ctx.catalog
+ .list_partitions_by_names(&managed, vec![partition.clone()])
+ .await
+ .unwrap_err()
+ ));
+
+ // A table whose partitions the catalog does not manage still falls back
to the file system
+ // listing rather than failing.
+ ctx.catalog
+ .list_partitions(&Identifier::new("default", "unmanaged_table"))
+ .await
+ .unwrap();
+}
+
// ==================== Database Tests ====================
#[tokio::test]
@@ -896,6 +985,284 @@ async fn test_rest_catalog_reads_format_table() {
);
}
+#[tokio::test]
+async fn test_rest_catalog_validates_dynamic_managed_partition_options() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = format_table_schema(&[]);
+ let identifier = Identifier::new("default", "managed_format");
+ add_internal_table_with_schema(
+ &ctx.server,
+ "managed_format",
+ schema,
+ "memory:/managed_format",
+ );
+
+ let table = ctx.catalog.get_table(&identifier).await.unwrap();
+
+ assert!(table.has_catalog_managed_partitions());
+ for (key, value) in [
+ ("metastore.partitioned-table", "false"),
+ ("format-table.partition-path-only-value", "true"),
+ ("format-table.implementation", "engine"),
+ ("type", "table"),
+ ("path", "memory:/elsewhere"),
+ ("file.format", "orc"),
+ ] {
+ let error = table
+ .copy_with_time_travel(HashMap::from([(key.to_string(),
value.to_string())]))
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(error, paimon::Error::DataInvalid { .. }) &&
error.to_string().contains(key),
+ "expected {key} validation error, got: {error}"
+ );
+ }
+
+ let copied = table
+ .copy_with_time_travel(HashMap::from([
+ (
+ "metastore.partitioned-table".to_string(),
+ "TRUE".to_string(),
+ ),
+ (
+ "format-table.partition-path-only-value".to_string(),
+ "false".to_string(),
+ ),
+ (
+ "format-table.implementation".to_string(),
+ "PAIMON".to_string(),
+ ),
+ (
+ "format-table.scan.list-parallelism".to_string(),
+ "8".to_string(),
+ ),
+ ]))
+ .await
+ .unwrap();
+ assert!(copied.has_catalog_managed_partitions());
+ assert!(copied
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .unwrap()
+ .splits()
+ .is_empty());
+}
+
+#[tokio::test]
+async fn
test_rest_catalog_rejects_enabling_managed_partitions_in_dynamic_options() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = format_table_schema(&[("metastore.partitioned-table",
"false")]);
+ let identifier = Identifier::new("default", "unmanaged_format");
+ add_internal_table_with_schema(
+ &ctx.server,
+ "unmanaged_format",
+ schema,
+ "memory:/unmanaged_format",
+ );
+
+ let table = ctx.catalog.get_table(&identifier).await.unwrap();
+ assert!(!table.has_catalog_managed_partitions());
+
+ let error = table
+ .copy_with_time_travel(HashMap::from([(
+ "metastore.partitioned-table".to_string(),
+ "true".to_string(),
+ )]))
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(error, paimon::Error::DataInvalid { .. })
+ && error.to_string().contains("metastore.partitioned-table"),
+ "expected partition source validation error, got: {error}"
+ );
+}
+
+#[tokio::test]
+async fn test_rest_catalog_rejects_external_catalog_managed_format_table() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = format_table_schema(&[]);
+ let identifier = Identifier::new("default", "external_managed_format");
+ ctx.server.add_table_with_schema(
+ "default",
+ "external_managed_format",
+ schema,
+ "memory:/external_managed_format",
+ );
+
+ let error = ctx.catalog.get_table(&identifier).await.unwrap_err();
+
+ assert!(matches!(
+ error,
+ paimon::Error::DataInvalid { message, .. }
+ if message.contains("default.external_managed_format")
+ && message.contains("internal")
+ ));
+}
+
+#[tokio::test]
+async fn test_rest_catalog_rejects_engine_managed_format_table() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = format_table_schema(&[("format-table.implementation",
"engine")]);
+ let identifier = Identifier::new("default", "engine_managed_format");
+ add_internal_table_with_schema(
+ &ctx.server,
+ "engine_managed_format",
+ schema,
+ "memory:/engine_managed_format",
+ );
+
+ let error = ctx.catalog.get_table(&identifier).await.unwrap_err();
+
+ assert!(matches!(
+ error,
+ paimon::Error::DataInvalid { message, .. }
+ if message.contains("metastore.partitioned-table")
+ && message.contains("format-table.implementation=engine")
+ ));
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_managed_format_scan_uses_registered_partition_paths_once() {
+ let tmp = tempfile::tempdir().unwrap();
+ for dt in ["2026-07-21", "2026-07-22"] {
+ let partition_dir = tmp.path().join(format!("dt={dt}"));
+ std::fs::create_dir_all(&partition_dir).unwrap();
+ std::fs::write(partition_dir.join("part-0.parquet"), b"data").unwrap();
+ }
+ let table_path = format!("file://{}", tmp.path().display());
+
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = format_table_schema(&[]);
+ let identifier = Identifier::new("default", "managed_visibility");
+ add_internal_table_with_schema(&ctx.server, "managed_visibility", schema,
&table_path);
+ ctx.server.set_table_partitions(
+ "default",
+ "managed_visibility",
+ vec![
+ HashMap::from([("dt".to_string(), "2026-07-22".to_string())]),
+ HashMap::from([("dt".to_string(), "2026-07-22".to_string())]),
+ ],
+ );
+
+ let table = ctx.catalog.get_table(&identifier).await.unwrap();
+ let plan = table.new_read_builder().new_scan().plan().await.unwrap();
+
+ // `dt=2026-07-21` holds a file but is not registered; the repeated
registration is read once.
+ assert_eq!(plan.splits().len(), 1);
+ assert!(plan.splits()[0].bucket_path().ends_with("/dt=2026-07-22"));
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn
test_managed_format_scan_treats_registered_missing_directory_as_empty() {
+ let tmp = tempfile::tempdir().unwrap();
+ let table_path = format!("file://{}", tmp.path().display());
+
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = format_table_schema(&[]);
+ let identifier = Identifier::new("default", "managed_missing_directory");
+ add_internal_table_with_schema(
+ &ctx.server,
+ "managed_missing_directory",
+ schema,
+ &table_path,
+ );
+ ctx.server.set_table_partitions(
+ "default",
+ "managed_missing_directory",
+ vec![HashMap::from([(
+ "dt".to_string(),
+ "2026-07-22".to_string(),
+ )])],
+ );
+
+ let table = ctx.catalog.get_table(&identifier).await.unwrap();
+ let plan = table.new_read_builder().new_scan().plan().await.unwrap();
+
+ assert!(plan.splits().is_empty());
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_managed_format_scan_propagates_partition_listing_error() {
+ let tmp = tempfile::tempdir().unwrap();
+ std::fs::create_dir_all(tmp.path().join("dt=2026-07-22")).unwrap();
+ std::fs::write(
+ tmp.path().join("dt=2026-07-22").join("part-0.parquet"),
+ b"data",
+ )
+ .unwrap();
+ let table_path = format!("file://{}", tmp.path().display());
+
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = format_table_schema(&[]);
+ let identifier = Identifier::new("default", "managed_scan_error");
+ add_internal_table_with_schema(&ctx.server, "managed_scan_error", schema,
&table_path);
+ ctx.server
+ .set_list_partitions_error_status(Some(StatusCode::NOT_IMPLEMENTED));
+
+ let table = ctx.catalog.get_table(&identifier).await.unwrap();
+ let error = table
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .unwrap_err();
+
+ // The directory is not read in place of the registrations the catalog
could not list.
+ assert!(matches!(
+ error,
+ paimon::Error::RestApi {
+ source: paimon::api::RestError::NotImplemented { .. }
+ }
+ ));
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn
test_managed_format_scan_reports_table_for_malformed_partition_metadata() {
+ let tmp = tempfile::tempdir().unwrap();
+ let table_path = format!("file://{}", tmp.path().display());
+
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = format_table_schema(&[]);
+ let identifier = Identifier::new("default", "managed_corrupt_metadata");
+ add_internal_table_with_schema(&ctx.server, "managed_corrupt_metadata",
schema, &table_path);
+ ctx.server.set_table_partitions(
+ "default",
+ "managed_corrupt_metadata",
+ vec![HashMap::from([
+ ("dt".to_string(), "2026-07-22".to_string()),
+ ("unexpected".to_string(), "value".to_string()),
+ ])],
+ );
+
+ let table = ctx.catalog.get_table(&identifier).await.unwrap();
+ let error = table
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .unwrap_err();
+
+ match error {
+ paimon::Error::DataInvalid {
+ message,
+ source: Some(source),
+ } => {
+ assert!(message.contains("invalid partition metadata"));
+ assert!(message.contains("default.managed_corrupt_metadata"));
+ let cause = source.to_string();
+ assert!(cause.contains("unexpected"), "unexpected cause: {cause}");
+ assert!(cause.contains("dt"), "unexpected cause: {cause}");
+ }
+ other => panic!("expected invalid partition metadata, got: {other}"),
+ }
+}
+
#[cfg(not(windows))]
#[tokio::test]
async fn test_rest_catalog_prunes_format_table_partition_filter() {
