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 2bb2abb6 feat(rest): add partition registration, lookup and statistics
APIs (#812)
2bb2abb6 is described below
commit 2bb2abb69f2eeaca9ad3bdab95dadcb2cd8a6e79
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Fri Sep 11 14:18:31 2026 +0800
feat(rest): add partition registration, lookup and statistics APIs (#812)
---
crates/paimon/src/api/api_request.rs | 263 +++++++++++-
crates/paimon/src/api/mod.rs | 4 +-
crates/paimon/src/api/resource_paths.rs | 30 ++
crates/paimon/src/api/rest_api.rs | 220 +++++++++-
crates/paimon/src/catalog/mod.rs | 67 ++++
crates/paimon/src/catalog/rest/rest_catalog.rs | 167 +++++++-
crates/paimon/src/spec/partition.rs | 34 ++
crates/paimon/src/spec/partition_statistics.rs | 8 +-
crates/paimon/src/spec/predicate.rs | 206 ++++++++++
crates/paimon/src/table/table_commit.rs | 4 +-
crates/paimon/tests/mock_server.rs | 534 ++++++++++++++++++++++++-
crates/paimon/tests/rest_api_test.rs | 276 ++++++++++++-
crates/paimon/tests/rest_catalog_test.rs | 321 ++++++++++++++-
13 files changed, 2109 insertions(+), 25 deletions(-)
diff --git a/crates/paimon/src/api/api_request.rs
b/crates/paimon/src/api/api_request.rs
index 84ee7c5c..48646757 100644
--- a/crates/paimon/src/api/api_request.rs
+++ b/crates/paimon/src/api/api_request.rs
@@ -19,12 +19,12 @@
//!
//! This module contains all request structures used in REST API calls.
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Deserializer, Serialize};
use std::collections::HashMap;
use crate::{
catalog::{Function, FunctionDefinition, Identifier, ViewSchema},
- spec::{DataField, Schema, SchemaChange},
+ spec::{DataField, PartitionStatistics, Schema, SchemaChange},
};
/// Request to create a new database.
@@ -167,6 +167,135 @@ impl AlterTableRequest {
}
}
+/// Request to create table partitions.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CreatePartitionsRequest {
+ /// Partition specs to register.
+ pub partition_specs: Vec<HashMap<String, String>>,
+ /// Whether already registered partitions should be ignored.
+ #[serde(
+ default = "default_true",
+ deserialize_with = "deserialize_null_to_true"
+ )]
+ pub ignore_if_exists: bool,
+ /// Statistics reported for the partitions, matched to them by spec;
absent unless reported.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub partition_statistics: Option<Vec<PartitionStatistics>>,
+ /// Whether the reported statistics replace what the catalog holds rather
than add to it;
+ /// present only together with `partition_statistics`.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub replace_statistics: Option<bool>,
+}
+
+impl CreatePartitionsRequest {
+ /// Create a request to register partitions.
+ pub fn new(partition_specs: Vec<HashMap<String, String>>,
ignore_if_exists: bool) -> Self {
+ Self {
+ partition_specs,
+ ignore_if_exists,
+ partition_statistics: None,
+ replace_statistics: None,
+ }
+ }
+
+ /// Report statistics for the partitions in the same request.
+ pub fn with_statistics(mut self, statistics: Vec<PartitionStatistics>,
replace: bool) -> Self {
+ self.partition_statistics = Some(statistics);
+ self.replace_statistics = Some(replace);
+ self
+ }
+}
+
+/// Request to drop (unregister) table partitions.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct DropPartitionsRequest {
+ /// Partition specs to unregister.
+ pub partition_specs: Vec<HashMap<String, String>>,
+ /// Whether missing partitions should be ignored.
+ #[serde(
+ default = "default_true",
+ deserialize_with = "deserialize_null_to_true"
+ )]
+ pub ignore_if_not_exists: bool,
+}
+
+impl DropPartitionsRequest {
+ /// Create a request to unregister partitions.
+ pub fn new(partition_specs: Vec<HashMap<String, String>>,
ignore_if_not_exists: bool) -> Self {
+ Self {
+ partition_specs,
+ ignore_if_not_exists,
+ }
+ }
+}
+
+/// Request to look up registered partitions by their complete specs.
+///
+/// Wire-compatible with Java `ListPartitionsByNamesRequest`, whose field is
`specs` rather than
+/// the `partitionSpecs` the create and drop requests use.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ListPartitionsByNamesRequest {
+ /// Complete partition specs to look up.
+ pub specs: Vec<HashMap<String, String>>,
+}
+
+impl ListPartitionsByNamesRequest {
+ /// Create a request to look up partitions by their complete specs.
+ pub fn new(specs: Vec<HashMap<String, String>>) -> Self {
+ Self { specs }
+ }
+}
+
+/// Request to list partitions matching a partition predicate.
+///
+/// Wire-compatible with Java `ListPartitionsByFilterRequest`: the predicate
travels as its JSON
+/// text inside the request body, and absent fields are left out.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ListPartitionsByFilterRequest {
+ /// Partition predicate in the REST catalog predicate JSON format.
+ pub filter: String,
+ /// Partition-name pattern combined with the filter as a conjunction.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub partition_name_pattern: Option<String>,
+ /// Maximum number of partitions in one page.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub max_results: Option<u32>,
+ /// Token of the page to return.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub page_token: Option<String>,
+}
+
+impl ListPartitionsByFilterRequest {
+ /// Create a request for one page of partitions matching `filter`.
+ pub fn new(
+ filter: String,
+ partition_name_pattern: Option<String>,
+ max_results: Option<u32>,
+ page_token: Option<String>,
+ ) -> Self {
+ Self {
+ filter,
+ partition_name_pattern,
+ max_results,
+ page_token,
+ }
+ }
+}
+
+fn default_true() -> bool {
+ true
+}
+
+fn deserialize_null_to_true<'de, D>(deserializer: D) -> Result<bool, D::Error>
+where
+ D: Deserializer<'de>,
+{
+ Ok(Option::<bool>::deserialize(deserializer)?.unwrap_or(true))
+}
+
/// Request for auth table query: the projected columns of the query.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AuthTableQueryRequest {
@@ -220,6 +349,136 @@ mod tests {
assert_eq!(serde_json::to_string(&req).unwrap(), "{}");
}
+ #[test]
+ fn test_create_partitions_request_serialization() {
+ let req = CreatePartitionsRequest::new(
+ vec![HashMap::from([
+ ("dt".to_string(), "2026-07-22".to_string()),
+ ("hour".to_string(), "10".to_string()),
+ ])],
+ false,
+ );
+
+ assert_eq!(
+ serde_json::to_value(req).unwrap(),
+ serde_json::json!({
+ "partitionSpecs": [{
+ "dt": "2026-07-22",
+ "hour": "10"
+ }],
+ "ignoreIfExists": false
+ })
+ );
+ }
+
+ #[test]
+ fn test_create_partitions_request_sends_statistics_only_when_reported() {
+ let spec = HashMap::from([("dt".to_string(),
"2026-07-22".to_string())]);
+ let request = CreatePartitionsRequest::new(vec![spec.clone()],
true).with_statistics(
+ vec![PartitionStatistics {
+ spec: spec.clone(),
+ record_count: -1,
+ file_size_in_bytes: 1024,
+ file_count: 2,
+ last_file_creation_time: 1_700_000_000_000,
+ total_buckets: -1,
+ }],
+ true,
+ );
+
+ assert_eq!(
+ serde_json::to_value(request).unwrap(),
+ serde_json::json!({
+ "partitionSpecs": [{"dt": "2026-07-22"}],
+ "ignoreIfExists": true,
+ "partitionStatistics": [{
+ "spec": {"dt": "2026-07-22"},
+ "recordCount": -1,
+ "fileSizeInBytes": 1024,
+ "fileCount": 2,
+ "lastFileCreationTime": 1_700_000_000_000_i64,
+ "totalBuckets": -1
+ }],
+ "replaceStatistics": true
+ })
+ );
+ }
+
+ #[test]
+ fn test_drop_partitions_request_serialization() {
+ let req = DropPartitionsRequest::new(
+ vec![HashMap::from([(
+ "dt".to_string(),
+ "2026-07-22".to_string(),
+ )])],
+ false,
+ );
+
+ assert_eq!(
+ serde_json::to_value(req).unwrap(),
+ serde_json::json!({
+ "partitionSpecs": [{"dt": "2026-07-22"}],
+ "ignoreIfNotExists": false
+ })
+ );
+ }
+
+ #[test]
+ fn test_partition_request_flags_default_to_true() {
+ for json in [
+ serde_json::json!({"partitionSpecs": []}),
+ serde_json::json!({"partitionSpecs": [], "ignoreIfExists": null}),
+ ] {
+ let request: CreatePartitionsRequest =
serde_json::from_value(json).unwrap();
+ assert!(request.ignore_if_exists);
+ }
+ for json in [
+ serde_json::json!({"partitionSpecs": []}),
+ serde_json::json!({"partitionSpecs": [], "ignoreIfNotExists":
null}),
+ ] {
+ let request: DropPartitionsRequest =
serde_json::from_value(json).unwrap();
+ assert!(request.ignore_if_not_exists);
+ }
+ }
+
+ #[test]
+ fn test_list_partitions_by_names_request_uses_the_specs_field() {
+ let request = ListPartitionsByNamesRequest::new(vec![HashMap::from([(
+ "dt".to_string(),
+ "2026-07-22".to_string(),
+ )])]);
+
+ assert_eq!(
+ serde_json::to_value(request).unwrap(),
+ serde_json::json!({"specs": [{"dt": "2026-07-22"}]})
+ );
+ }
+
+ #[test]
+ fn test_list_partitions_by_filter_request_leaves_out_absent_fields() {
+ let request = ListPartitionsByFilterRequest::new("{}".to_string(),
None, None, None);
+ assert_eq!(
+ serde_json::to_value(request).unwrap(),
+ serde_json::json!({"filter": "{}"})
+ );
+
+ let request = ListPartitionsByFilterRequest::new(
+ "{}".to_string(),
+ Some("dt=a/%".to_string()),
+ Some(1000),
+ Some("next".to_string()),
+ );
+ assert_eq!(
+ serde_json::to_value(request).unwrap(),
+ serde_json::json!({
+ "filter": "{}",
+ "partitionNamePattern": "dt=a/%",
+ "maxResults": 1000,
+ "pageToken": "next"
+ })
+ );
+ }
+
#[test]
fn test_rename_table_request_serialization() {
let source = Identifier::new("db1".to_string(), "table1".to_string());
diff --git a/crates/paimon/src/api/mod.rs b/crates/paimon/src/api/mod.rs
index 809b9021..181d0fa2 100644
--- a/crates/paimon/src/api/mod.rs
+++ b/crates/paimon/src/api/mod.rs
@@ -32,7 +32,9 @@ mod api_response;
// Re-export request types
pub use api_request::{
AlterDatabaseRequest, AlterTableRequest, AuthTableQueryRequest,
CreateDatabaseRequest,
- CreateFunctionRequest, CreateTableRequest, CreateViewRequest,
RenameTableRequest,
+ CreateFunctionRequest, CreatePartitionsRequest, CreateTableRequest,
CreateViewRequest,
+ DropPartitionsRequest, ListPartitionsByFilterRequest,
ListPartitionsByNamesRequest,
+ RenameTableRequest,
};
// Re-export response types
diff --git a/crates/paimon/src/api/resource_paths.rs
b/crates/paimon/src/api/resource_paths.rs
index 38c39c9a..9acea094 100644
--- a/crates/paimon/src/api/resource_paths.rs
+++ b/crates/paimon/src/api/resource_paths.rs
@@ -216,6 +216,27 @@ impl ResourcePaths {
Self::PARTITIONS
)
}
+
+ /// Get the endpoint path for dropping table partitions.
+ pub fn drop_partitions(&self, database_name: &str, table_name: &str) ->
String {
+ format!("{}/drop", self.partitions(database_name, table_name))
+ }
+
+ /// Get the endpoint path for looking up table partitions by their specs.
+ pub fn list_partitions_by_names(&self, database_name: &str, table_name:
&str) -> String {
+ format!(
+ "{}/list-by-names",
+ self.partitions(database_name, table_name)
+ )
+ }
+
+ /// Get the endpoint path for listing table partitions matching a
predicate.
+ pub fn list_partitions_by_filter(&self, database_name: &str, table_name:
&str) -> String {
+ format!(
+ "{}/list-by-filter",
+ self.partitions(database_name, table_name)
+ )
+ }
}
#[cfg(test)]
@@ -296,4 +317,13 @@ mod tests {
"/v1/catalog/databases/analytics/functions/rectangle+area"
);
}
+
+ #[test]
+ fn test_drop_partitions_path_encodes_names() {
+ let paths = ResourcePaths::new("catalog");
+ assert_eq!(
+ paths.drop_partitions("analytics db", "user events"),
+
"/v1/catalog/databases/analytics+db/tables/user+events/partitions/drop"
+ );
+ }
}
diff --git a/crates/paimon/src/api/rest_api.rs
b/crates/paimon/src/api/rest_api.rs
index cdde2b23..50605895 100644
--- a/crates/paimon/src/api/rest_api.rs
+++ b/crates/paimon/src/api/rest_api.rs
@@ -20,7 +20,7 @@
//! This module provides a REST API client for interacting with
//! Paimon rest catalog services, supporting database operations.
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use crate::api::rest_client::HttpClient;
use crate::catalog::{Function, Identifier, ViewSchema};
@@ -30,7 +30,9 @@ use crate::Result;
use super::api_request::{
AlterDatabaseRequest, AlterTableRequest, AuthTableQueryRequest,
CreateDatabaseRequest,
- CreateFunctionRequest, CreateTableRequest, CreateViewRequest,
RenameTableRequest,
+ CreateFunctionRequest, CreatePartitionsRequest, CreateTableRequest,
CreateViewRequest,
+ DropPartitionsRequest, ListPartitionsByFilterRequest,
ListPartitionsByNamesRequest,
+ RenameTableRequest,
};
use super::api_response::{
AuthTableQueryResponse, ConfigResponse, GetDatabaseResponse,
GetFunctionResponse,
@@ -91,6 +93,9 @@ impl RESTApi {
pub const TABLE_NAME_PATTERN: &'static str = "tableNamePattern";
pub const VIEW_NAME_PATTERN: &'static str = "viewNamePattern";
pub const FUNCTION_NAME_PATTERN: &'static str = "functionNamePattern";
+ pub const PARTITION_NAME_PATTERN: &'static str = "partitionNamePattern";
+ /// Bounds one request: catalog services cap the partitions a single call
may carry.
+ const PARTITION_REQUEST_SIZE: u32 = 1000;
pub const TABLE_TYPE: &'static str = "tableType";
/// Create a new RESTApi from options.
@@ -556,36 +561,232 @@ impl RESTApi {
// ==================== Partition Operations ====================
+ /// Create table partitions in a single REST request.
+ pub async fn create_partitions(
+ &self,
+ identifier: &Identifier,
+ partition_specs: Vec<HashMap<String, String>>,
+ ignore_if_exists: bool,
+ ) -> Result<()> {
+ self.create_partitions_with_statistics(
+ identifier,
+ partition_specs,
+ ignore_if_exists,
+ None,
+ false,
+ )
+ .await
+ }
+
+ /// Create table partitions and report statistics for them in a single
REST request.
+ ///
+ /// Statistics are matched to the specs by spec and may cover only some of
them.
+ /// `replace_statistics` says whether they replace what the catalog holds
or add to it, and is
+ /// not sent when no statistics are.
+ pub async fn create_partitions_with_statistics(
+ &self,
+ identifier: &Identifier,
+ partition_specs: Vec<HashMap<String, String>>,
+ ignore_if_exists: bool,
+ statistics: Option<Vec<PartitionStatistics>>,
+ replace_statistics: bool,
+ ) -> Result<()> {
+ let database = identifier.database();
+ let table = identifier.object();
+ validate_non_empty_multi(&[(database, "database name"), (table, "table
name")])?;
+ let path = self.resource_paths.partitions(database, table);
+ let mut request = CreatePartitionsRequest::new(partition_specs,
ignore_if_exists);
+ if let Some(statistics) = statistics {
+ request = request.with_statistics(statistics, replace_statistics);
+ }
+ let _resp: serde_json::Value = self.client.post(&path,
&request).await?;
+ Ok(())
+ }
+
+ /// Unregister table partitions in a single REST request.
+ ///
+ /// The REST service removes metadata only; it does not delete partition
+ /// directories or data files.
+ pub async fn drop_partitions(
+ &self,
+ identifier: &Identifier,
+ partition_specs: Vec<HashMap<String, String>>,
+ ignore_if_not_exists: bool,
+ ) -> Result<()> {
+ let database = identifier.database();
+ let table = identifier.object();
+ validate_non_empty_multi(&[(database, "database name"), (table, "table
name")])?;
+ let path = self.resource_paths.drop_partitions(database, table);
+ let request = DropPartitionsRequest::new(partition_specs,
ignore_if_not_exists);
+ let _resp: serde_json::Value = self.client.post(&path,
&request).await?;
+ Ok(())
+ }
+
/// List all partitions of a table, paging internally.
pub async fn list_partitions(&self, identifier: &Identifier) ->
Result<Vec<Partition>> {
+ self.drain_partitions(identifier, None, None, None).await
+ }
+
+ /// List partitions, asking the catalog to return only those whose
partition name
+ /// matches `partition_name_pattern`.
+ ///
+ /// The pattern is a pushdown hint: a catalog may apply it partially or
not at all, so
+ /// the result is a superset of the matching partitions and never misses
one. Callers
+ /// keep applying their own filter to what comes back.
+ pub async fn list_partitions_by_name_pattern(
+ &self,
+ identifier: &Identifier,
+ partition_name_pattern: Option<&str>,
+ ) -> Result<Vec<Partition>> {
+ self.drain_partitions(
+ identifier,
+ Some(Self::PARTITION_REQUEST_SIZE),
+ partition_name_pattern,
+ None,
+ )
+ .await
+ }
+
+ /// List partitions, asking the catalog to return only those matching
`filter`, a partition
+ /// predicate in the REST catalog predicate JSON format, together with
+ /// `partition_name_pattern` when one is given.
+ ///
+ /// Like the pattern, the filter is a pushdown hint: a catalog may apply
it partially or not
+ /// at all, so callers keep applying their own filter to what comes back.
+ pub async fn list_partitions_by_filter(
+ &self,
+ identifier: &Identifier,
+ filter: &str,
+ partition_name_pattern: Option<&str>,
+ ) -> Result<Vec<Partition>> {
+ self.drain_partitions(
+ identifier,
+ Some(Self::PARTITION_REQUEST_SIZE),
+ partition_name_pattern,
+ Some(filter),
+ )
+ .await
+ }
+
+ /// List one page of partitions matching `filter`. See
[`Self::list_partitions_by_filter`].
+ pub async fn list_partitions_by_filter_paged(
+ &self,
+ identifier: &Identifier,
+ filter: &str,
+ max_results: Option<u32>,
+ page_token: Option<&str>,
+ partition_name_pattern: Option<&str>,
+ ) -> Result<PagedList<Partition>> {
+ let database = identifier.database();
+ let table = identifier.object();
+ validate_non_empty_multi(&[(database, "database name"), (table, "table
name")])?;
+ let path = self
+ .resource_paths
+ .list_partitions_by_filter(database, table);
+ let request = ListPartitionsByFilterRequest::new(
+ filter.to_string(),
+ partition_name_pattern
+ .filter(|pattern| !pattern.is_empty())
+ .map(str::to_string),
+ max_results,
+ page_token.map(str::to_string),
+ );
+ let response: ListPartitionsResponse = self.client.post(&path,
&request).await?;
+ Ok(PagedList::new(
+ response.partitions.unwrap_or_default(),
+ response.next_page_token,
+ ))
+ }
+
+ /// Return those of the given complete partition specs that are registered.
+ ///
+ /// The specs go out in one request, so callers bound how many they send
at once.
+ pub async fn list_partitions_by_names(
+ &self,
+ identifier: &Identifier,
+ partition_specs: Vec<HashMap<String, String>>,
+ ) -> Result<Vec<Partition>> {
+ let database = identifier.database();
+ let table = identifier.object();
+ validate_non_empty_multi(&[(database, "database name"), (table, "table
name")])?;
+ let path = self
+ .resource_paths
+ .list_partitions_by_names(database, table);
+ let request = ListPartitionsByNamesRequest::new(partition_specs);
+ let response: ListPartitionsResponse = self.client.post(&path,
&request).await?;
+ Ok(response.partitions.unwrap_or_default())
+ }
+
+ async fn drain_partitions(
+ &self,
+ identifier: &Identifier,
+ max_results: Option<u32>,
+ partition_name_pattern: Option<&str>,
+ filter: Option<&str>,
+ ) -> Result<Vec<Partition>> {
let database = identifier.database();
let table = identifier.object();
validate_non_empty_multi(&[(database, "database name"), (table, "table
name")])?;
let mut results = Vec::new();
let mut page_token: Option<String> = None;
+ let mut seen_page_tokens = HashSet::new();
loop {
- let paged = self
- .list_partitions_paged(identifier, None, page_token.as_deref())
- .await?;
- let is_empty = paged.elements.is_empty();
+ let paged = match filter {
+ Some(filter) => {
+ self.list_partitions_by_filter_paged(
+ identifier,
+ filter,
+ max_results,
+ page_token.as_deref(),
+ partition_name_pattern,
+ )
+ .await?
+ }
+ None => {
+ self.list_partitions_paged(
+ identifier,
+ max_results,
+ page_token.as_deref(),
+ partition_name_pattern,
+ )
+ .await?
+ }
+ };
results.extend(paged.elements);
- page_token = paged.next_page_token;
- if page_token.is_none() || is_empty {
+
+ let Some(next_page_token) = paged.next_page_token.filter(|token|
!token.is_empty())
+ else {
break;
+ };
+ if !seen_page_tokens.insert(next_page_token.clone()) {
+ return Err(crate::Error::UnexpectedError {
+ message: format!(
+ "REST catalog returned partition page token
'{next_page_token}' more than \
+ once for table {}",
+ identifier.full_name()
+ ),
+ source: None,
+ });
}
+ page_token = Some(next_page_token);
}
Ok(results)
}
/// List partitions with pagination.
+ ///
+ /// `partition_name_pattern` is a SQL LIKE pattern over partition names,
where `%` is
+ /// the only wildcard. Like [`Self::list_partitions_by_name_pattern`], it
is a hint the
+ /// catalog may ignore.
pub async fn list_partitions_paged(
&self,
identifier: &Identifier,
max_results: Option<u32>,
page_token: Option<&str>,
+ partition_name_pattern: Option<&str>,
) -> Result<PagedList<Partition>> {
let database = identifier.database();
let table = identifier.object();
@@ -599,6 +800,9 @@ impl RESTApi {
if let Some(token) = page_token {
params.push((Self::PAGE_TOKEN, token.to_string()));
}
+ if let Some(pattern) = partition_name_pattern.filter(|pattern|
!pattern.is_empty()) {
+ params.push((Self::PARTITION_NAME_PATTERN, pattern.to_string()));
+ }
let response: ListPartitionsResponse = if params.is_empty() {
self.client.get(&path, None::<&[(&str, &str)]>).await?
diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs
index f16fd25a..308b78dc 100644
--- a/crates/paimon/src/catalog/mod.rs
+++ b/crates/paimon/src/catalog/mod.rs
@@ -560,6 +560,73 @@ pub trait Catalog: Send + Sync {
})
}
+ // ======================= partition methods
===============================
+
+ /// Register table partition specs in the catalog.
+ ///
+ /// When `ignore_if_exists` is false, implementations must reject the
+ /// entire request if any supplied spec already exists. When true, existing
+ /// specs are ignored so callers can safely retry the request.
+ async fn create_partitions(
+ &self,
+ identifier: &Identifier,
+ partition_specs: Vec<HashMap<String, String>>,
+ ignore_if_exists: bool,
+ ) -> Result<()> {
+ self.create_partitions_with_statistics(
+ identifier,
+ partition_specs,
+ ignore_if_exists,
+ None,
+ false,
+ )
+ .await
+ }
+
+ /// Register partition specs and report statistics for them in the same
call, so a partition
+ /// is never registered by a request whose statistics failed on their own.
+ ///
+ /// Statistics are matched to `partition_specs` by spec and may cover only
some of them.
+ /// `replace_statistics` says whether they replace what the catalog holds
or add to it. A field
+ /// reported as [`Partition::UNKNOWN`] says nothing about itself and
leaves the stored value as
+ /// it was. Reporting never unregisters a partition.
+ ///
+ /// This is the method implementations provide, so that one forwarding only
+ /// [`Self::create_partitions`] cannot drop every report without anyone
noticing.
+ async fn create_partitions_with_statistics(
+ &self,
+ _identifier: &Identifier,
+ _partition_specs: Vec<HashMap<String, String>>,
+ _ignore_if_exists: bool,
+ _statistics: Option<Vec<crate::spec::PartitionStatistics>>,
+ _replace_statistics: bool,
+ ) -> Result<()> {
+ Err(Error::Unsupported {
+ message: "Catalog does not support creating
partitions".to_string(),
+ })
+ }
+
+ /// Return those of the given complete partition specs that are registered.
+ ///
+ /// Specs are compared with the registered values as they are, without
normalizing them.
+ /// The default impl filters [`Self::list_partitions`]; catalogs that can
look partitions up
+ /// by name (e.g. `RESTCatalog`) override it so a few specs never cost a
full listing.
+ async fn list_partitions_by_names(
+ &self,
+ identifier: &Identifier,
+ partition_specs: Vec<HashMap<String, String>>,
+ ) -> Result<Vec<Partition>> {
+ if partition_specs.is_empty() {
+ return Ok(Vec::new());
+ }
+ Ok(self
+ .list_partitions(identifier)
+ .await?
+ .into_iter()
+ .filter(|partition| partition_specs.contains(&partition.spec))
+ .collect())
+ }
+
/// List partitions for a table.
///
/// Default impl scans the table's manifest entries via
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index db167c3d..df068b2b 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -34,10 +34,12 @@ use crate::catalog::{
use crate::common::{CatalogOptions, Options};
use crate::error::Error;
use crate::io::cache::{create_local_cache_with_namespace, LocalCache};
-use crate::spec::{Partition, Schema, SchemaChange};
+use crate::spec::{Partition, PartitionStatistics, Schema, SchemaChange};
use crate::table::{RESTEnv, Table};
use crate::Result;
+const PARTITION_BATCH_SIZE: usize = 1000;
+
/// REST catalog implementation.
///
/// This catalog communicates with a Paimon REST catalog server
@@ -411,6 +413,79 @@ impl Catalog for RESTCatalog {
))
}
+ async fn create_partitions_with_statistics(
+ &self,
+ identifier: &Identifier,
+ partition_specs: Vec<HashMap<String, String>>,
+ ignore_if_exists: bool,
+ statistics: Option<Vec<PartitionStatistics>>,
+ replace_statistics: bool,
+ ) -> Result<()> {
+ let statistics = statistics
+ .map(|statistics| index_statistics_by_spec(identifier,
&partition_specs, statistics))
+ .transpose()?;
+ if partition_specs.is_empty() {
+ return Ok(());
+ }
+ // A strict create is rejected whole when any partition already
exists, so it is never
+ // split; an idempotent one is sent in bounded batches, each with its
own statistics.
+ let batch_size = if ignore_if_exists {
+ PARTITION_BATCH_SIZE
+ } else {
+ partition_specs.len()
+ };
+ for batch in partition_specs.chunks(batch_size) {
+ let batch_statistics = statistics.as_ref().map(|by_spec| {
+ batch
+ .iter()
+ .filter_map(|spec| by_spec.get(&spec_key(spec)).cloned())
+ .collect::<Vec<_>>()
+ });
+ self.api
+ .create_partitions_with_statistics(
+ identifier,
+ batch.to_vec(),
+ ignore_if_exists,
+ batch_statistics,
+ replace_statistics,
+ )
+ .await
+ .map_err(|error| map_rest_error_for_create_partitions(error,
identifier))?;
+ }
+ Ok(())
+ }
+
+ async fn list_partitions_by_names(
+ &self,
+ identifier: &Identifier,
+ partition_specs: Vec<HashMap<String, String>>,
+ ) -> Result<Vec<Partition>> {
+ let mut partitions = Vec::new();
+ for batch in partition_specs.chunks(PARTITION_BATCH_SIZE) {
+ match self
+ .api
+ .list_partitions_by_names(identifier, batch.to_vec())
+ .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.
+ Err(Error::RestApi {
+ source: RestError::NotImplemented { .. },
+ }) => {
+ return Ok(self
+ .list_partitions(identifier)
+ .await?
+ .into_iter()
+ .filter(|partition|
partition_specs.contains(&partition.spec))
+ .collect());
+ }
+ Err(error) => return Err(map_rest_error_for_table(error,
identifier)),
+ }
+ }
+ Ok(partitions)
+ }
+
async fn list_partitions(&self, identifier: &Identifier) ->
Result<Vec<Partition>> {
match self.api.list_partitions(identifier).await {
Ok(parts) => Ok(parts),
@@ -432,7 +507,7 @@ impl Catalog for RESTCatalog {
) -> Result<PagedList<Partition>> {
match self
.api
- .list_partitions_paged(identifier, max_results, page_token)
+ .list_partitions_paged(identifier, max_results, page_token, None)
.await
{
Ok(page) => Ok(page),
@@ -493,6 +568,94 @@ fn map_rest_error_for_table(err: Error, identifier:
&Identifier) -> Error {
}
}
+/// A partition spec in a form that can key a map.
+fn spec_key(spec: &HashMap<String, String>) -> Vec<(String, String)> {
+ let mut entries = spec
+ .iter()
+ .map(|(key, value)| (key.clone(), value.clone()))
+ .collect::<Vec<_>>();
+ entries.sort_unstable();
+ entries
+}
+
+/// Index reported statistics by the spec they describe, before anything is
sent.
+///
+/// A partition registered twice is rejected: every occurrence would carry the
same report, so a
+/// spec repeated across batches would have an additive report applied once
per batch. A report
+/// for a partition that is not being created and a partition reported twice
are rejected too: the
+/// catalog cannot tell which of two reports is meant.
+///
+/// Mirrors Java
`CatalogFormatTablePartitionManager.validateAndIndexStatistics`.
+fn index_statistics_by_spec(
+ identifier: &Identifier,
+ partition_specs: &[HashMap<String, String>],
+ statistics: Vec<PartitionStatistics>,
+) -> Result<HashMap<Vec<(String, String)>, PartitionStatistics>> {
+ let mut requested =
std::collections::HashSet::with_capacity(partition_specs.len());
+ for spec in partition_specs {
+ if !requested.insert(spec_key(spec)) {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Partition {spec:?} of table {} is registered twice in one
request that \
+ reports statistics; report each partition once",
+ identifier.full_name()
+ ),
+ source: None,
+ });
+ }
+ }
+ let mut by_spec = HashMap::with_capacity(statistics.len());
+ for statistic in statistics {
+ let key = spec_key(&statistic.spec);
+ if !requested.contains(&key) {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Partition statistics were reported for {:?} of table {},
which is not among \
+ the partitions being created",
+ statistic.spec,
+ identifier.full_name()
+ ),
+ source: None,
+ });
+ }
+ let spec = statistic.spec.clone();
+ if by_spec.insert(key, statistic).is_some() {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Partition statistics were reported twice for {spec:?} of
table {}",
+ identifier.full_name()
+ ),
+ source: None,
+ });
+ }
+ }
+ Ok(by_spec)
+}
+
+fn map_rest_error_for_create_partitions(err: Error, identifier: &Identifier)
-> Error {
+ match err {
+ Error::RestApi {
+ source: RestError::AlreadyExists { message, .. },
+ } => Error::DataInvalid {
+ message: format!(
+ "One or more partitions already exist for table {}: {message}",
+ identifier.full_name()
+ ),
+ source: None,
+ },
+ Error::RestApi {
+ source: RestError::BadRequest { message },
+ } => Error::DataInvalid {
+ message: format!(
+ "Invalid partition request for table {}: {message}",
+ identifier.full_name()
+ ),
+ source: None,
+ },
+ other => map_rest_error_for_table(other, identifier),
+ }
+}
+
/// Map a REST API error from creating a persistent view.
fn map_rest_error_for_create_view(err: Error, identifier: &Identifier) ->
Error {
match err {
diff --git a/crates/paimon/src/spec/partition.rs
b/crates/paimon/src/spec/partition.rs
index a78fe5b2..24372650 100644
--- a/crates/paimon/src/spec/partition.rs
+++ b/crates/paimon/src/spec/partition.rs
@@ -26,9 +26,15 @@ use std::collections::HashMap;
#[serde(rename_all = "camelCase")]
pub struct Partition {
pub spec: HashMap<String, String>,
+ /// Statistics are optional on the wire: one the catalog leaves out was
never reported, so it
+ /// reads as [`Self::UNKNOWN`] rather than as an exact zero.
+ #[serde(default = "unknown_statistic")]
pub record_count: i64,
+ #[serde(default = "unknown_statistic")]
pub file_size_in_bytes: i64,
+ #[serde(default = "unknown_statistic")]
pub file_count: i64,
+ #[serde(default = "unknown_statistic")]
pub last_file_creation_time: i64,
#[serde(default)]
pub total_buckets: i32,
@@ -67,6 +73,10 @@ impl Partition {
}
}
+fn unknown_statistic() -> i64 {
+ Partition::UNKNOWN
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -130,6 +140,30 @@ mod tests {
assert_eq!(decoded.spec.get("dt"), Some(&"2024-01-01".to_string()));
}
+ #[test]
+ fn test_absent_statistics_decode_as_unknown_and_zero_stays_exact() {
+ let absent: Partition =
+ serde_json::from_str(r#"{"spec": {"pt": "1"}, "done":
true}"#).unwrap();
+ for value in [
+ absent.record_count,
+ absent.file_size_in_bytes,
+ absent.file_count,
+ absent.last_file_creation_time,
+ ] {
+ assert_eq!(value, Partition::UNKNOWN);
+ }
+
+ let zero: Partition = serde_json::from_str(
+ r#"{"spec": {"pt": "1"}, "recordCount": 0, "fileSizeInBytes": 0,
+ "fileCount": 0, "lastFileCreationTime": 0}"#,
+ )
+ .unwrap();
+ assert_eq!(
+ (zero.record_count, zero.file_size_in_bytes, zero.file_count),
+ (0, 0, 0)
+ );
+ }
+
#[test]
fn test_unknown_is_negative_and_zero_is_a_measurement() {
assert_eq!(Partition::UNKNOWN, -1);
diff --git a/crates/paimon/src/spec/partition_statistics.rs
b/crates/paimon/src/spec/partition_statistics.rs
index 30f05793..ff82f314 100644
--- a/crates/paimon/src/spec/partition_statistics.rs
+++ b/crates/paimon/src/spec/partition_statistics.rs
@@ -22,13 +22,17 @@ use std::collections::HashMap;
///
/// Reference:
[org.apache.paimon.partition.PartitionStatistics](https://github.com/apache/paimon)
/// and [pypaimon snapshot_commit.py
PartitionStatistics](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/snapshot/snapshot_commit.py)
-#[derive(Debug, Clone, Serialize, Deserialize)]
+///
+/// The same shape reports what a partition holds when a Format Table's
partitions are measured.
+/// There a negative field is unknown rather than a decrement, which is why
+/// `last_file_creation_time` is signed like the other counts.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PartitionStatistics {
pub spec: HashMap<String, String>,
pub record_count: i64,
pub file_size_in_bytes: i64,
pub file_count: i64,
- pub last_file_creation_time: u64,
+ pub last_file_creation_time: i64,
pub total_buckets: i32,
}
diff --git a/crates/paimon/src/spec/predicate.rs
b/crates/paimon/src/spec/predicate.rs
index c168e769..82a93eac 100644
--- a/crates/paimon/src/spec/predicate.rs
+++ b/crates/paimon/src/spec/predicate.rs
@@ -672,6 +672,116 @@ impl Predicate {
Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
}
}
+
+ /// Serialize this predicate in the REST catalog wire format, the inverse
of
+ /// [`Self::from_rest_json`].
+ ///
+ /// Returns `None` when some part of the predicate has no wire form: the
format has no `NOT`,
+ /// and Java cannot read temporal, decimal or binary literals back from
JSON. Field references
+ /// carry each leaf's index as it is, so a predicate meant for a partition
row must already be
+ /// indexed by partition field.
+ pub fn to_rest_json(&self) -> Option<serde_json::Value> {
+ match self {
+ Predicate::AlwaysTrue => Some(rest_constant_leaf("TRUE")),
+ Predicate::AlwaysFalse => Some(rest_constant_leaf("FALSE")),
+ Predicate::And(children) | Predicate::Or(children) => {
+ if children.is_empty() {
+ return None;
+ }
+ let function = if matches!(self, Predicate::And(_)) {
+ "AND"
+ } else {
+ "OR"
+ };
+ let children = children
+ .iter()
+ .map(Predicate::to_rest_json)
+ .collect::<Option<Vec<_>>>()?;
+ Some(serde_json::json!({
+ "kind": "COMPOUND",
+ "function": function,
+ "children": children,
+ }))
+ }
+ Predicate::Not(_) => None,
+ Predicate::Leaf {
+ column,
+ index,
+ data_type,
+ op,
+ literals,
+ } => {
+ if matches!(op, PredicateOperator::Like) && literals.len() !=
1 {
+ return None;
+ }
+ let literals = literals
+ .iter()
+ .map(datum_to_rest_json)
+ .collect::<Option<Vec<_>>>()?;
+ Some(serde_json::json!({
+ "kind": "LEAF",
+ "transform": {
+ "name": "FIELD_REF",
+ "fieldRef": {
+ "index": index,
+ "name": column,
+ "type": serde_json::to_value(data_type).ok()?,
+ },
+ },
+ "function": rest_leaf_function(*op),
+ "literals": literals,
+ }))
+ }
+ }
+ }
+}
+
+fn rest_constant_leaf(function: &str) -> serde_json::Value {
+ serde_json::json!({
+ "kind": "LEAF",
+ "transform": {"name": "NULL"},
+ "function": function,
+ "literals": [],
+ })
+}
+
+fn rest_leaf_function(op: PredicateOperator) -> &'static str {
+ match op {
+ PredicateOperator::IsNull => "IS_NULL",
+ PredicateOperator::IsNotNull => "IS_NOT_NULL",
+ PredicateOperator::Eq => "EQUAL",
+ PredicateOperator::NotEq => "NOT_EQUAL",
+ PredicateOperator::Lt => "LESS_THAN",
+ PredicateOperator::LtEq => "LESS_OR_EQUAL",
+ PredicateOperator::Gt => "GREATER_THAN",
+ PredicateOperator::GtEq => "GREATER_OR_EQUAL",
+ PredicateOperator::In => "IN",
+ PredicateOperator::NotIn => "NOT_IN",
+ PredicateOperator::StartsWith => "STARTS_WITH",
+ PredicateOperator::EndsWith => "ENDS_WITH",
+ PredicateOperator::Contains => "CONTAINS",
+ PredicateOperator::ArrayContains => "ARRAY_CONTAINS",
+ PredicateOperator::ArraysOverlap => "ARRAYS_OVERLAP",
+ PredicateOperator::ArrayContainsAll => "ARRAY_CONTAINS_ALL",
+ PredicateOperator::Like => "LIKE",
+ PredicateOperator::Between => "BETWEEN",
+ PredicateOperator::NotBetween => "NOT_BETWEEN",
+ }
+}
+
+/// A literal the way Java writes it, for the types Java can also read back.
+fn datum_to_rest_json(datum: &Datum) -> Option<serde_json::Value> {
+ match datum {
+ Datum::Bool(value) => Some(serde_json::Value::Bool(*value)),
+ Datum::TinyInt(value) => Some(serde_json::Value::from(*value)),
+ Datum::SmallInt(value) => Some(serde_json::Value::from(*value)),
+ Datum::Int(value) => Some(serde_json::Value::from(*value)),
+ Datum::Long(value) => Some(serde_json::Value::from(*value)),
+ Datum::Float(value) =>
serde_json::Number::from_f64(f64::from(*value)).map(Into::into),
+ Datum::Double(value) =>
serde_json::Number::from_f64(*value).map(Into::into),
+ Datum::String(value) => Some(serde_json::Value::String(value.clone())),
+ _ => None,
+ }
}
fn rest_json_err(detail: impl fmt::Display) -> Error {
@@ -3390,6 +3500,102 @@ mod tests {
}
}
+ #[test]
+ fn test_to_rest_json_is_read_back_by_from_rest_json() {
+ let fields = test_fields();
+ let builder = PredicateBuilder::new(&fields);
+ let id_is_one = builder.equal("id", Datum::Int(1)).unwrap();
+ let predicates = vec![
+ builder.is_null("id").unwrap(),
+ builder.is_not_null("name").unwrap(),
+ id_is_one.clone(),
+ builder
+ .not_equal("name", Datum::String("x".to_string()))
+ .unwrap(),
+ builder.less_than("hr", Datum::Int(3)).unwrap(),
+ builder.less_or_equal("hr", Datum::Int(3)).unwrap(),
+ builder.greater_than("hr", Datum::Int(3)).unwrap(),
+ builder.greater_or_equal("hr", Datum::Int(3)).unwrap(),
+ builder
+ .is_in("id", vec![Datum::Int(1), Datum::Int(2)])
+ .unwrap(),
+ builder
+ .is_not_in("id", vec![Datum::Int(1), Datum::Int(2)])
+ .unwrap(),
+ builder
+ .starts_with("name", Datum::String("a".to_string()))
+ .unwrap(),
+ builder
+ .ends_with("name", Datum::String("a".to_string()))
+ .unwrap(),
+ builder
+ .contains("name", Datum::String("a".to_string()))
+ .unwrap(),
+ builder
+ .like("name", Datum::String("a%".to_string()), None)
+ .unwrap(),
+ builder.between("hr", Datum::Int(1), Datum::Int(5)).unwrap(),
+ builder
+ .not_between("hr", Datum::Int(1), Datum::Int(5))
+ .unwrap(),
+ Predicate::and(vec![
+ id_is_one.clone(),
+ builder
+ .equal("name", Datum::String("x".to_string()))
+ .unwrap(),
+ ]),
+ Predicate::or(vec![id_is_one, builder.equal("id",
Datum::Int(2)).unwrap()]),
+ Predicate::AlwaysTrue,
+ Predicate::AlwaysFalse,
+ ];
+ for predicate in predicates {
+ let json = predicate
+ .to_rest_json()
+ .unwrap_or_else(|| panic!("{predicate} should have a wire
form"));
+ let parsed = Predicate::from_rest_json(&json.to_string(),
&fields).unwrap();
+ assert_eq!(parsed.to_string(), predicate.to_string(), "{json}");
+ }
+ }
+
+ /// Wire strings below are taken verbatim from Java
`PredicateJsonSerdeTest`.
+ #[test]
+ fn test_to_rest_json_matches_java_wire_format() {
+ let fields = vec![DataField::new(
+ 0,
+ "f0".to_string(),
+ DataType::Int(IntType::new()),
+ )];
+ for java in [
+
r#"{"kind":"LEAF","transform":{"name":"FIELD_REF","fieldRef":{"index":0,"name":"f0","type":"INT"}},"function":"EQUAL","literals":[1]}"#,
+
r#"{"kind":"LEAF","transform":{"name":"FIELD_REF","fieldRef":{"index":0,"name":"f0","type":"INT"}},"function":"IS_NULL","literals":[]}"#,
+
r#"{"kind":"LEAF","transform":{"name":"NULL"},"function":"TRUE","literals":[]}"#,
+
r#"{"kind":"LEAF","transform":{"name":"FIELD_REF","fieldRef":{"index":0,"name":"f0","type":"INT"}},"function":"BETWEEN","literals":[3,7]}"#,
+ ] {
+ let predicate = Predicate::from_rest_json(java, &fields).unwrap();
+ assert_eq!(
+ predicate.to_rest_json().unwrap(),
+ serde_json::from_str::<serde_json::Value>(java).unwrap()
+ );
+ }
+ }
+
+ #[test]
+ fn test_to_rest_json_refuses_what_java_cannot_read_back() {
+ let fields = test_fields();
+ let builder = PredicateBuilder::new(&fields);
+ let on_date = builder.equal("dt", Datum::Date(20_656)).unwrap();
+ let on_id = builder.equal("id", Datum::Int(1)).unwrap();
+
+ assert!(on_date.to_rest_json().is_none());
+ assert!(Predicate::Not(Box::new(on_id.clone()))
+ .to_rest_json()
+ .is_none());
+ // One child without a wire form keeps the whole compound off the wire.
+ assert!(Predicate::and(vec![on_id, on_date])
+ .to_rest_json()
+ .is_none());
+ }
+
#[test]
fn test_from_rest_json_rejects_empty_compounds() {
let fields = test_fields();
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 9f3f707d..676d5358 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -2792,8 +2792,8 @@ impl TableCommit {
let file = entry.file();
let file_creation_time = file
.creation_time
- .map(|t| t.timestamp_millis() as u64)
- .unwrap_or_else(current_time_millis);
+ .map(|t| t.timestamp_millis())
+ .unwrap_or_else(|| current_time_millis() as i64);
let stats =
stats_map.entry(partition_bytes.clone()).or_insert_with(|| {
// Parse partition spec from BinaryRow
diff --git a/crates/paimon/tests/mock_server.rs
b/crates/paimon/tests/mock_server.rs
index 6c8a0048..934c55de 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -35,12 +35,17 @@ use tokio::task::JoinHandle;
use paimon::api::{
AlterDatabaseRequest, AlterTableRequest, AuditRESTResponse, ConfigResponse,
- CreateFunctionRequest, CreateViewRequest, ErrorResponse,
GetDatabaseResponse,
- GetFunctionResponse, GetTableResponse, GetViewResponse,
ListDatabasesResponse,
- ListFunctionsResponse, ListTablesResponse, ListViewsResponse,
RenameTableRequest,
- ResourcePaths,
+ CreateFunctionRequest, CreatePartitionsRequest, CreateViewRequest,
DropPartitionsRequest,
+ ErrorResponse, GetDatabaseResponse, GetFunctionResponse, GetTableResponse,
GetViewResponse,
+ ListDatabasesResponse, ListFunctionsResponse,
ListPartitionsByFilterRequest,
+ ListPartitionsByNamesRequest, ListPartitionsResponse, ListTablesResponse,
ListViewsResponse,
+ RenameTableRequest, ResourcePaths,
};
use paimon::catalog::{Function, Identifier};
+use paimon::spec::Partition;
+
+type PartitionPageResponse = (Vec<Partition>, Option<String>);
+type PartitionSpecPageResponse = (Vec<HashMap<String, String>>,
Option<String>);
#[derive(Clone, Debug, Default)]
struct MockState {
@@ -48,17 +53,63 @@ struct MockState {
tables: HashMap<String, GetTableResponse>,
views: HashMap<String, GetViewResponse>,
functions: HashMap<String, GetFunctionResponse>,
+ partitions: HashMap<String, Vec<Partition>>,
+ partition_page_responses: HashMap<String, Vec<PartitionPageResponse>>,
+ partition_list_call_counts: HashMap<String, usize>,
+ partition_list_name_patterns: HashMap<String, Vec<Option<String>>>,
+ partition_list_by_names_calls: HashMap<String, Vec<Vec<HashMap<String,
String>>>>,
+ partition_list_by_filter_requests: HashMap<String,
Vec<ListPartitionsByFilterRequest>>,
+ list_partitions_by_names_error_status: Option<StatusCode>,
+ list_partitions_by_filter_error_status: Option<StatusCode>,
view_function_endpoints_unsupported: bool,
drop_view_error_status: Option<StatusCode>,
list_page_size: Option<usize>,
no_permission_databases: HashSet<String>,
no_permission_tables: HashSet<String>,
+ create_partitions_calls: Vec<(String, String, CreatePartitionsRequest)>,
+ drop_partitions_calls: Vec<(String, String, DropPartitionsRequest)>,
+ create_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)
ecs_token: Option<serde_json::Value>,
}
+/// Match a partition spec against a partition-name pattern the way a catalog
would.
+///
+/// Only the pattern shape clients send is understood: `key=value` segments
joined by `/`,
+/// optionally ending in a `/%` wildcard. Values are compared unescaped, so a
test that
+/// needs escaping should assert the pattern the client sent instead of the
rows returned.
+fn partition_spec_matches_name_pattern(spec: &HashMap<String, String>,
pattern: &str) -> bool {
+ let prefix = pattern.strip_suffix("/%");
+ let segments = prefix.unwrap_or(pattern).split('/').collect::<Vec<_>>();
+ if prefix.is_none() && segments.len() != spec.len() {
+ return false;
+ }
+ segments.iter().all(|segment| {
+ segment
+ .split_once('=')
+ .is_some_and(|(key, value)| spec.get(key).map(String::as_str) ==
Some(value))
+ })
+}
+
+fn partition_from_spec(spec: HashMap<String, String>) -> Partition {
+ Partition {
+ spec,
+ record_count: Partition::UNKNOWN,
+ file_size_in_bytes: Partition::UNKNOWN,
+ file_count: Partition::UNKNOWN,
+ last_file_creation_time: Partition::UNKNOWN,
+ total_buckets: 0,
+ done: false,
+ created_at: None,
+ created_by: None,
+ updated_at: None,
+ updated_by: None,
+ options: None,
+ }
+}
+
fn paginate_names(
names: Vec<String>,
params: &HashMap<String, String>,
@@ -282,6 +333,11 @@ impl RESTServer {
// Also remove all tables in this database
let prefix = format!("{name}.");
s.tables.retain(|key, _| !key.starts_with(&prefix));
+ s.partitions.retain(|key, _| !key.starts_with(&prefix));
+ s.partition_page_responses
+ .retain(|key, _| !key.starts_with(&prefix));
+ s.partition_list_call_counts
+ .retain(|key, _| !key.starts_with(&prefix));
s.no_permission_tables
.retain(|key| !key.starts_with(&prefix));
(StatusCode::OK, Json(serde_json::json!(""))).into_response()
@@ -724,6 +780,9 @@ impl RESTServer {
}
if s.tables.remove(&key).is_some() {
+ s.partitions.remove(&key);
+ s.partition_page_responses.remove(&key);
+ s.partition_list_call_counts.remove(&key);
s.no_permission_tables.remove(&key);
(StatusCode::OK, Json(serde_json::json!(""))).into_response()
} else {
@@ -771,6 +830,322 @@ impl RESTServer {
}
}
+ /// Handle POST /databases/:db/tables/:table/partitions - create
partitions.
+ pub async fn create_partitions(
+ Path((db, table)): Path<(String, String)>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ Json(request): Json<CreatePartitionsRequest>,
+ ) -> impl IntoResponse {
+ let mut inner = state.inner.lock().unwrap();
+ inner
+ .create_partitions_calls
+ .push((db.clone(), table.clone(), request.clone()));
+ if let Some(status) = inner.create_partitions_error_status {
+ let message = if status == StatusCode::CONFLICT {
+ "Some partitions already exist"
+ } else {
+ "Invalid partition request"
+ };
+ let error = ErrorResponse::new(
+ Some("partition".to_string()),
+ Some(table.clone()),
+ Some(message.to_string()),
+ Some(status.as_u16() as i32),
+ );
+ return (status, Json(error)).into_response();
+ }
+
+ let key = format!("{db}.{table}");
+ if !inner.tables.contains_key(&key) {
+ let error = ErrorResponse::new(
+ Some("table".to_string()),
+ Some(table),
+ Some("Not Found".to_string()),
+ Some(404),
+ );
+ return (StatusCode::NOT_FOUND, Json(error)).into_response();
+ }
+
+ let registered_partitions = inner.partitions.entry(key).or_default();
+ let has_conflict = request
+ .partition_specs
+ .iter()
+ .enumerate()
+ .any(|(index, spec)| {
+ registered_partitions
+ .iter()
+ .any(|partition| partition.spec == *spec)
+ || request.partition_specs[..index].contains(spec)
+ });
+ if has_conflict && !request.ignore_if_exists {
+ let error = ErrorResponse::new(
+ Some("partition".to_string()),
+ Some(table),
+ Some("Some partitions already exist".to_string()),
+ Some(StatusCode::CONFLICT.as_u16() as i32),
+ );
+ return (StatusCode::CONFLICT, Json(error)).into_response();
+ }
+
+ for spec in request.partition_specs {
+ if !registered_partitions
+ .iter()
+ .any(|partition| partition.spec == spec)
+ {
+ registered_partitions.push(partition_from_spec(spec));
+ }
+ }
+ // As the catalog does: a negative field was never measured and leaves
the stored value
+ // alone; a replacing report overwrites what is stored, any other is
added to it.
+ let replace = request.replace_statistics.unwrap_or(false);
+ for statistic in request.partition_statistics.unwrap_or_default() {
+ let Some(partition) = registered_partitions
+ .iter_mut()
+ .find(|partition| partition.spec == statistic.spec)
+ else {
+ continue;
+ };
+ for (stored, reported) in [
+ (&mut partition.record_count, statistic.record_count),
+ (
+ &mut partition.file_size_in_bytes,
+ statistic.file_size_in_bytes,
+ ),
+ (&mut partition.file_count, statistic.file_count),
+ ] {
+ if reported >= 0 {
+ *stored = if replace {
+ reported
+ } else {
+ (*stored).max(0) + reported
+ };
+ }
+ }
+ if statistic.last_file_creation_time >= 0 {
+ partition.last_file_creation_time = if replace {
+ statistic.last_file_creation_time
+ } else {
+ partition
+ .last_file_creation_time
+ .max(statistic.last_file_creation_time)
+ };
+ }
+ }
+ let response = json!({"success": true});
+ (StatusCode::OK, Json(response)).into_response()
+ }
+
+ /// Handle GET /databases/:db/tables/:table/partitions - list partitions.
+ pub async fn list_partitions(
+ Path((db, table)): Path<(String, String)>,
+ Query(params): Query<HashMap<String, String>>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ ) -> impl IntoResponse {
+ let mut inner = state.inner.lock().unwrap();
+ let key = format!("{db}.{table}");
+ let name_pattern = params.get("partitionNamePattern").cloned();
+ inner
+ .partition_list_name_patterns
+ .entry(key.clone())
+ .or_default()
+ .push(name_pattern.clone());
+ if !inner.tables.contains_key(&key) {
+ let error = ErrorResponse::new(
+ Some("table".to_string()),
+ Some(table),
+ Some("Not Found".to_string()),
+ Some(404),
+ );
+ return (StatusCode::NOT_FOUND, Json(error)).into_response();
+ }
+ if inner.partition_page_responses.contains_key(&key) {
+ let request_index = {
+ let calls = inner
+ .partition_list_call_counts
+ .entry(key.clone())
+ .or_default();
+ let request_index = *calls;
+ *calls += 1;
+ request_index
+ };
+ let responses = &inner.partition_page_responses[&key];
+ let expected_token = request_index
+ .checked_sub(1)
+ .and_then(|index| responses.get(index))
+ .and_then(|(_, token)| token.clone());
+ let actual_token = params.get("pageToken").cloned();
+ if actual_token != expected_token || request_index >=
responses.len() {
+ let error = ErrorResponse::new(
+ Some("partition".to_string()),
+ Some(table),
+ Some(format!(
+ "Invalid page token: expected {expected_token:?}, got
{actual_token:?}"
+ )),
+ Some(StatusCode::BAD_REQUEST.as_u16() as i32),
+ );
+ return (StatusCode::BAD_REQUEST, Json(error)).into_response();
+ }
+ let (partitions, next_page_token) =
responses[request_index].clone();
+ return (
+ StatusCode::OK,
+ Json(ListPartitionsResponse::new(
+ Some(partitions),
+ next_page_token,
+ )),
+ )
+ .into_response();
+ }
+ let mut partitions =
inner.partitions.get(&key).cloned().unwrap_or_default();
+ if let Some(pattern) = name_pattern {
+ partitions
+ .retain(|partition|
partition_spec_matches_name_pattern(&partition.spec, &pattern));
+ }
+ (
+ StatusCode::OK,
+ Json(ListPartitionsResponse::new(Some(partitions), None)),
+ )
+ .into_response()
+ }
+
+ /// Handle POST /databases/:db/tables/:table/partitions/drop - drop
partitions.
+ pub async fn drop_partitions(
+ Path((db, table)): Path<(String, String)>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ Json(request): Json<DropPartitionsRequest>,
+ ) -> impl IntoResponse {
+ let mut inner = state.inner.lock().unwrap();
+ inner
+ .drop_partitions_calls
+ .push((db.clone(), table.clone(), request.clone()));
+
+ let key = format!("{db}.{table}");
+ if !inner.tables.contains_key(&key) {
+ let error = ErrorResponse::new(
+ Some("table".to_string()),
+ Some(table),
+ Some("Not Found".to_string()),
+ Some(404),
+ );
+ return (StatusCode::NOT_FOUND, Json(error)).into_response();
+ }
+
+ let registered_partitions = inner.partitions.entry(key).or_default();
+ let has_missing = request.partition_specs.iter().any(|spec| {
+ !registered_partitions
+ .iter()
+ .any(|partition| partition.spec == *spec)
+ });
+ if has_missing && !request.ignore_if_not_exists {
+ let error = ErrorResponse::new(
+ Some("partition".to_string()),
+ Some(table),
+ Some("Some partitions do not exist".to_string()),
+ Some(StatusCode::NOT_FOUND.as_u16() as i32),
+ );
+ return (StatusCode::NOT_FOUND, Json(error)).into_response();
+ }
+
+ registered_partitions
+ .retain(|partition|
!request.partition_specs.contains(&partition.spec));
+ let response = json!({"success": true});
+ (StatusCode::OK, Json(response)).into_response()
+ }
+
+ /// Handle POST /databases/:db/tables/:table/partitions/list-by-names -
look up partitions.
+ pub async fn list_partitions_by_names(
+ Path((db, table)): Path<(String, String)>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ Json(request): Json<ListPartitionsByNamesRequest>,
+ ) -> impl IntoResponse {
+ let mut inner = state.inner.lock().unwrap();
+ let key = format!("{db}.{table}");
+ inner
+ .partition_list_by_names_calls
+ .entry(key.clone())
+ .or_default()
+ .push(request.specs.clone());
+ if !inner.tables.contains_key(&key) {
+ let error = ErrorResponse::new(
+ Some("table".to_string()),
+ Some(table),
+ Some("Not Found".to_string()),
+ Some(404),
+ );
+ return (StatusCode::NOT_FOUND, Json(error)).into_response();
+ }
+ if let Some(status) = inner.list_partitions_by_names_error_status {
+ let error = ErrorResponse::new(
+ Some("partition".to_string()),
+ Some(table),
+ Some("Listing partitions by names is not
implemented".to_string()),
+ Some(status.as_u16() as i32),
+ );
+ return (status, Json(error)).into_response();
+ }
+ let partitions = inner
+ .partitions
+ .get(&key)
+ .map(|partitions| {
+ partitions
+ .iter()
+ .filter(|partition|
request.specs.contains(&partition.spec))
+ .cloned()
+ .collect()
+ })
+ .unwrap_or_default();
+ (
+ StatusCode::OK,
+ Json(ListPartitionsResponse::new(Some(partitions), None)),
+ )
+ .into_response()
+ }
+
+ /// Handle POST /databases/:db/tables/:table/partitions/list-by-filter -
list partitions.
+ ///
+ /// Like a catalog that does not evaluate predicates yet, this applies the
name pattern only
+ /// and returns every other registered partition, which the endpoint
contract allows.
+ pub async fn list_partitions_by_filter(
+ Path((db, table)): Path<(String, String)>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ Json(request): Json<ListPartitionsByFilterRequest>,
+ ) -> impl IntoResponse {
+ let mut inner = state.inner.lock().unwrap();
+ let key = format!("{db}.{table}");
+ inner
+ .partition_list_by_filter_requests
+ .entry(key.clone())
+ .or_default()
+ .push(request.clone());
+ if !inner.tables.contains_key(&key) {
+ let error = ErrorResponse::new(
+ Some("table".to_string()),
+ Some(table),
+ Some("Not Found".to_string()),
+ Some(404),
+ );
+ return (StatusCode::NOT_FOUND, Json(error)).into_response();
+ }
+ if let Some(status) = inner.list_partitions_by_filter_error_status {
+ let error = ErrorResponse::new(
+ Some("partition".to_string()),
+ Some(table),
+ Some("Listing partitions by filter is not
implemented".to_string()),
+ Some(status.as_u16() as i32),
+ );
+ return (status, Json(error)).into_response();
+ }
+ let mut partitions =
inner.partitions.get(&key).cloned().unwrap_or_default();
+ if let Some(pattern) = &request.partition_name_pattern {
+ partitions
+ .retain(|partition|
partition_spec_matches_name_pattern(&partition.spec, pattern));
+ }
+ (
+ StatusCode::OK,
+ Json(ListPartitionsResponse::new(Some(partitions), None)),
+ )
+ .into_response()
+ }
+
/// Handle POST /rename-table - rename a table.
pub async fn rename_table(
Extension(state): Extension<Arc<RESTServer>>,
@@ -827,7 +1202,6 @@ impl RESTServer {
if s.no_permission_tables.remove(&source_key) {
s.no_permission_tables.insert(dest_key.clone());
}
-
(StatusCode::OK, Json(serde_json::json!(""))).into_response()
} else {
let err = ErrorResponse::new(
@@ -930,6 +1304,57 @@ impl RESTServer {
self.inner.lock().unwrap().drop_view_error_status = status;
}
+ /// Make the create-partitions endpoint return the given status.
+ pub fn set_create_partitions_error_status(&self, status:
Option<StatusCode>) {
+ self.inner.lock().unwrap().create_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
+ .lock()
+ .unwrap()
+ .list_partitions_by_names_error_status = status;
+ }
+
+ /// Make the list-partitions-by-filter endpoint return the given status.
+ pub fn set_list_partitions_by_filter_error_status(&self, status:
Option<StatusCode>) {
+ self.inner
+ .lock()
+ .unwrap()
+ .list_partitions_by_filter_error_status = status;
+ }
+
+ /// Return the specs of every list-by-names request the table received, in
order.
+ pub fn table_partition_list_by_names_calls(
+ &self,
+ database: &str,
+ table: &str,
+ ) -> Vec<Vec<HashMap<String, String>>> {
+ self.inner
+ .lock()
+ .unwrap()
+ .partition_list_by_names_calls
+ .get(&format!("{database}.{table}"))
+ .cloned()
+ .unwrap_or_default()
+ }
+
+ /// Return every list-by-filter request the table received, in order.
+ pub fn table_partition_list_by_filter_requests(
+ &self,
+ database: &str,
+ table: &str,
+ ) -> Vec<ListPartitionsByFilterRequest> {
+ self.inner
+ .lock()
+ .unwrap()
+ .partition_list_by_filter_requests
+ .get(&format!("{database}.{table}"))
+ .cloned()
+ .unwrap_or_default()
+ }
+
/// Add a table with schema and path to the server state.
///
/// This is needed for `RESTCatalog::get_table` which requires
@@ -972,6 +1397,89 @@ impl RESTServer {
let mut s = self.inner.lock().unwrap();
s.no_permission_tables.insert(format!("{database}.{table}"));
}
+
+ /// Set the catalog-registered partitions for a table.
+ pub fn set_table_partitions(
+ &self,
+ database: &str,
+ table: &str,
+ partition_specs: Vec<HashMap<String, String>>,
+ ) {
+ let partitions = partition_specs
+ .into_iter()
+ .map(partition_from_spec)
+ .collect();
+ let key = format!("{database}.{table}");
+ let mut inner = self.inner.lock().unwrap();
+ assert!(
+ inner.tables.contains_key(&key),
+ "table {key} does not exist"
+ );
+ inner.partitions.insert(key.clone(), partitions);
+ inner.partition_page_responses.remove(&key);
+ inner.partition_list_call_counts.remove(&key);
+ }
+
+ /// Set explicit list-partitions pages and response tokens in request
order.
+ pub fn set_table_partition_page_responses(
+ &self,
+ database: &str,
+ table: &str,
+ responses: Vec<PartitionSpecPageResponse>,
+ ) {
+ let responses = responses
+ .into_iter()
+ .map(|(specs, token)|
(specs.into_iter().map(partition_from_spec).collect(), token))
+ .collect();
+ let key = format!("{database}.{table}");
+ let mut inner = self.inner.lock().unwrap();
+ assert!(
+ inner.tables.contains_key(&key),
+ "table {key} does not exist"
+ );
+ inner
+ .partition_page_responses
+ .insert(key.clone(), responses);
+ inner.partition_list_call_counts.remove(&key);
+ }
+
+ /// Return the `partitionNamePattern` of every partition-list request the
table
+ /// received, in order, with `None` where the client sent no pattern.
+ pub fn table_partition_list_name_patterns(
+ &self,
+ database: &str,
+ table: &str,
+ ) -> Vec<Option<String>> {
+ self.inner
+ .lock()
+ .unwrap()
+ .partition_list_name_patterns
+ .get(&format!("{database}.{table}"))
+ .cloned()
+ .unwrap_or_default()
+ }
+
+ /// Return how many paged partition-list requests the table received.
+ pub fn table_partition_list_call_count(&self, database: &str, table: &str)
-> usize {
+ self.inner
+ .lock()
+ .unwrap()
+ .partition_list_call_counts
+ .get(&format!("{database}.{table}"))
+ .copied()
+ .unwrap_or_default()
+ }
+
+ /// Return all create-partitions calls received by the server.
+ pub fn create_partitions_calls(&self) -> Vec<(String, String,
CreatePartitionsRequest)> {
+ self.inner.lock().unwrap().create_partitions_calls.clone()
+ }
+
+ /// Return all drop-partitions calls received by the server.
+ pub fn drop_partitions_calls(&self) -> Vec<(String, String,
DropPartitionsRequest)> {
+ self.inner.lock().unwrap().drop_partitions_calls.clone()
+ }
+
/// Get the server URL.
pub fn url(&self) -> Option<String> {
self.addr.map(|a| format!("http://{a}";))
@@ -1089,6 +1597,22 @@ pub async fn start_mock_server(
.post(RESTServer::alter_table)
.delete(RESTServer::drop_table),
)
+ .route(
+ &format!("{prefix}/databases/:db/tables/:table/partitions"),
+
get(RESTServer::list_partitions).post(RESTServer::create_partitions),
+ )
+ .route(
+ &format!("{prefix}/databases/:db/tables/:table/partitions/drop"),
+ post(RESTServer::drop_partitions),
+ )
+ .route(
+
&format!("{prefix}/databases/:db/tables/:table/partitions/list-by-names"),
+ post(RESTServer::list_partitions_by_names),
+ )
+ .route(
+
&format!("{prefix}/databases/:db/tables/:table/partitions/list-by-filter"),
+ post(RESTServer::list_partitions_by_filter),
+ )
.route(
&format!("{prefix}/databases/:db/views"),
get(RESTServer::list_views).post(RESTServer::create_view),
diff --git a/crates/paimon/tests/rest_api_test.rs
b/crates/paimon/tests/rest_api_test.rs
index c7dff44b..086c3301 100644
--- a/crates/paimon/tests/rest_api_test.rs
+++ b/crates/paimon/tests/rest_api_test.rs
@@ -24,7 +24,7 @@ use std::collections::HashMap;
use paimon::api::auth::{DLFECSTokenLoader, DLFToken, DLFTokenLoader};
use paimon::api::rest_api::RESTApi;
-use paimon::api::ConfigResponse;
+use paimon::api::{ConfigResponse, CreatePartitionsRequest,
DropPartitionsRequest};
use paimon::catalog::{Function, FunctionDefinition, Identifier, ViewSchema};
use paimon::common::Options;
use paimon::spec::DataField;
@@ -71,6 +71,13 @@ async fn setup_test_server(initial_dbs: Vec<&str>) ->
TestContext {
TestContext { server, api, url }
}
+async fn setup_partition_api() -> (TestContext, Identifier) {
+ let ctx = setup_test_server(vec!["default"]).await;
+ let identifier = Identifier::new("default", "managed_table");
+ ctx.server.add_table("default", "managed_table");
+ (ctx, identifier)
+}
+
// ==================== Database Tests ====================
#[tokio::test]
async fn test_list_databases() {
@@ -594,6 +601,273 @@ async fn test_drop_table_no_permission() {
assert!(result.is_err(), "dropping no-permission table should fail");
}
+// ==================== Partition Tests ====================
+
+#[tokio::test]
+async fn test_partition_mutations_post_expected_requests() {
+ let (ctx, identifier) = setup_partition_api().await;
+ let partition_specs = vec![HashMap::from([
+ ("dt".to_string(), "2026-07-22".to_string()),
+ ("hour".to_string(), "10".to_string()),
+ ])];
+
+ ctx.api
+ .create_partitions(&identifier, partition_specs.clone(), false)
+ .await
+ .unwrap();
+
+ assert_eq!(
+ ctx.server.create_partitions_calls(),
+ vec![(
+ "default".to_string(),
+ "managed_table".to_string(),
+ CreatePartitionsRequest::new(partition_specs.clone(), false),
+ )]
+ );
+
+ ctx.api
+ .drop_partitions(&identifier, partition_specs.clone(), false)
+ .await
+ .unwrap();
+
+ assert_eq!(
+ ctx.server.drop_partitions_calls(),
+ vec![(
+ "default".to_string(),
+ "managed_table".to_string(),
+ DropPartitionsRequest::new(partition_specs, false),
+ )]
+ );
+}
+
+#[tokio::test]
+async fn test_list_partitions_by_names_returns_the_registered_ones() {
+ let (ctx, identifier) = setup_partition_api().await;
+ let registered = HashMap::from([("dt".to_string(),
"2026-07-22".to_string())]);
+ let missing = HashMap::from([("dt".to_string(),
"2026-07-23".to_string())]);
+ ctx.server
+ .set_table_partitions("default", "managed_table",
vec![registered.clone()]);
+
+ let partitions = ctx
+ .api
+ .list_partitions_by_names(&identifier, vec![registered.clone(),
missing.clone()])
+ .await
+ .unwrap();
+
+ assert_eq!(
+ partitions
+ .into_iter()
+ .map(|partition| partition.spec)
+ .collect::<Vec<_>>(),
+ vec![registered.clone()]
+ );
+ assert_eq!(
+ ctx.server
+ .table_partition_list_by_names_calls("default", "managed_table"),
+ vec![vec![registered, missing]]
+ );
+}
+
+#[tokio::test]
+async fn test_list_partitions_by_filter_sends_filter_pattern_and_page_size() {
+ let (ctx, identifier) = setup_partition_api().await;
+ let wanted = HashMap::from([("dt".to_string(), "2026-07-22".to_string())]);
+ ctx.server.set_table_partitions(
+ "default",
+ "managed_table",
+ vec![
+ wanted.clone(),
+ HashMap::from([("dt".to_string(), "2026-07-23".to_string())]),
+ ],
+ );
+ let filter =
r#"{"kind":"LEAF","transform":{"name":"FIELD_REF","fieldRef":{"index":0,"name":"dt","type":"STRING"}},"function":"EQUAL","literals":["2026-07-22"]}"#;
+
+ let partitions = ctx
+ .api
+ .list_partitions_by_filter(&identifier, filter, Some("dt=2026-07-22"))
+ .await
+ .unwrap();
+
+ assert_eq!(
+ partitions
+ .into_iter()
+ .map(|partition| partition.spec)
+ .collect::<Vec<_>>(),
+ vec![wanted]
+ );
+ let requests = ctx
+ .server
+ .table_partition_list_by_filter_requests("default", "managed_table");
+ assert_eq!(requests.len(), 1);
+ assert_eq!(requests[0].filter, filter);
+ assert_eq!(
+ requests[0].partition_name_pattern.as_deref(),
+ Some("dt=2026-07-22")
+ );
+ assert_eq!(requests[0].max_results, Some(1000));
+ assert_eq!(requests[0].page_token, None);
+}
+
+#[tokio::test]
+async fn test_list_partitions_by_filter_surfaces_not_implemented() {
+ let (ctx, identifier) = setup_partition_api().await;
+ ctx.server
+
.set_list_partitions_by_filter_error_status(Some(axum::http::StatusCode::NOT_IMPLEMENTED));
+
+ let error = ctx
+ .api
+ .list_partitions_by_filter(&identifier, "{}", None)
+ .await
+ .unwrap_err();
+
+ assert!(
+ matches!(
+ error,
+ paimon::Error::RestApi {
+ source: paimon::api::RestError::NotImplemented { .. }
+ }
+ ),
+ "{error}"
+ );
+}
+
+#[tokio::test]
+async fn test_list_partitions_by_name_pattern_sends_the_pattern() {
+ let (ctx, identifier) = setup_partition_api().await;
+ let wanted = HashMap::from([
+ ("dt".to_string(), "2026-07-22".to_string()),
+ ("hour".to_string(), "10".to_string()),
+ ]);
+ ctx.server.set_table_partitions(
+ "default",
+ "managed_table",
+ vec![
+ wanted.clone(),
+ HashMap::from([
+ ("dt".to_string(), "2026-07-23".to_string()),
+ ("hour".to_string(), "10".to_string()),
+ ]),
+ ],
+ );
+
+ let partitions = ctx
+ .api
+ .list_partitions_by_name_pattern(&identifier, Some("dt=2026-07-22/%"))
+ .await
+ .unwrap();
+ ctx.api.list_partitions(&identifier).await.unwrap();
+
+ assert_eq!(
+ partitions
+ .into_iter()
+ .map(|partition| partition.spec)
+ .collect::<Vec<_>>(),
+ vec![wanted]
+ );
+ assert_eq!(
+ ctx.server
+ .table_partition_list_name_patterns("default", "managed_table"),
+ vec![Some("dt=2026-07-22/%".to_string()), None]
+ );
+}
+
+#[tokio::test]
+async fn test_list_partitions_follows_non_empty_token_after_empty_page() {
+ let (ctx, identifier) = setup_partition_api().await;
+ let expected = HashMap::from([("dt".to_string(),
"2026-07-22".to_string())]);
+ ctx.server.set_table_partition_page_responses(
+ "default",
+ "managed_table",
+ vec![
+ (Vec::new(), Some("1".to_string())),
+ (vec![expected.clone()], None),
+ ],
+ );
+
+ let partitions = ctx.api.list_partitions(&identifier).await.unwrap();
+
+ assert_eq!(partitions.len(), 1);
+ assert_eq!(partitions[0].spec, expected);
+}
+
+#[tokio::test]
+async fn test_list_partitions_stops_on_empty_next_page_token() {
+ let (ctx, identifier) = setup_partition_api().await;
+ let expected = HashMap::from([("dt".to_string(),
"2026-07-22".to_string())]);
+ let unexpected = HashMap::from([("dt".to_string(),
"2026-07-23".to_string())]);
+ ctx.server.set_table_partition_page_responses(
+ "default",
+ "managed_table",
+ vec![
+ (vec![expected.clone()], Some(String::new())),
+ (vec![unexpected], None),
+ ],
+ );
+
+ let partitions = ctx.api.list_partitions(&identifier).await.unwrap();
+
+ assert_eq!(
+ partitions
+ .into_iter()
+ .map(|partition| partition.spec)
+ .collect::<Vec<_>>(),
+ vec![expected]
+ );
+ assert_eq!(
+ ctx.server
+ .table_partition_list_call_count("default", "managed_table"),
+ 1
+ );
+}
+
+#[tokio::test]
+async fn test_list_partitions_rejects_repeated_page_token() {
+ let (ctx, identifier) = setup_partition_api().await;
+ ctx.server.set_table_partition_page_responses(
+ "default",
+ "managed_table",
+ vec![
+ (
+ vec![HashMap::from([(
+ "dt".to_string(),
+ "2026-07-20".to_string(),
+ )])],
+ Some("a".to_string()),
+ ),
+ (
+ vec![HashMap::from([(
+ "dt".to_string(),
+ "2026-07-21".to_string(),
+ )])],
+ Some("b".to_string()),
+ ),
+ (
+ vec![HashMap::from([(
+ "dt".to_string(),
+ "2026-07-22".to_string(),
+ )])],
+ Some("a".to_string()),
+ ),
+ ],
+ );
+
+ let error = ctx.api.list_partitions(&identifier).await.unwrap_err();
+
+ let paimon::Error::UnexpectedError { message, .. } = error else {
+ panic!("expected a repeated-page-token error, got: {error}");
+ };
+ assert_eq!(
+ message,
+ "REST catalog returned partition page token 'a' more than once for
table \
+ default.managed_table"
+ );
+ assert_eq!(
+ ctx.server
+ .table_partition_list_call_count("default", "managed_table"),
+ 3
+ );
+}
+
// ==================== Rename Table Tests ====================
#[tokio::test]
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 41052c45..972fab05 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -31,8 +31,8 @@ use paimon::api::ConfigResponse;
use paimon::catalog::{Catalog, Function, FunctionDefinition, Identifier,
RESTCatalog, ViewSchema};
use paimon::common::Options;
use paimon::spec::{
- BigIntType, BlobType, BlobViewStruct, DataField, DataType, Datum, IntType,
PredicateBuilder,
- Schema, SchemaChange, VarCharType,
+ BigIntType, BlobType, BlobViewStruct, DataField, DataType, Datum, IntType,
PartitionStatistics,
+ PredicateBuilder, Schema, SchemaChange, VarCharType,
};
use paimon::{CatalogOptions, FileSystemCatalog, Table};
@@ -167,6 +167,323 @@ fn collect_blob_rows(batches: &[RecordBatch]) ->
Vec<(i32, String, Option<Vec<u8
rows
}
+// ==================== Partition Tests ====================
+
+#[tokio::test]
+async fn
test_rest_catalog_skips_empty_and_batches_idempotent_create_partitions() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "managed_table");
+ ctx.server.add_table("default", "managed_table");
+ let partition_specs = (0..1001)
+ .map(|value| HashMap::from([("dt".to_string(), value.to_string())]))
+ .collect::<Vec<_>>();
+
+ ctx.catalog
+ .create_partitions(&identifier, Vec::new(), false)
+ .await
+ .unwrap();
+ ctx.catalog
+ .create_partitions(&identifier, partition_specs.clone(), true)
+ .await
+ .unwrap();
+
+ let calls = ctx.server.create_partitions_calls();
+ assert_eq!(calls.len(), 2);
+ assert_eq!(calls[0].2.partition_specs, partition_specs[..1000]);
+ assert_eq!(calls[1].2.partition_specs, partition_specs[1000..]);
+ assert!(calls.iter().all(|(_, _, request)| request.ignore_if_exists));
+}
+
+#[tokio::test]
+async fn test_rest_catalog_keeps_non_idempotent_create_in_one_request() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "managed_table");
+ ctx.server.add_table("default", "managed_table");
+ let partition_specs = (0..2500)
+ .map(|value| HashMap::from([("dt".to_string(), value.to_string())]))
+ .collect::<Vec<_>>();
+
+ ctx.catalog
+ .create_partitions(&identifier, partition_specs.clone(), false)
+ .await
+ .unwrap();
+
+ let calls = ctx.server.create_partitions_calls();
+ assert_eq!(calls.len(), 1);
+ assert_eq!(calls[0].2.partition_specs, partition_specs);
+ assert!(!calls[0].2.ignore_if_exists);
+}
+
+#[tokio::test]
+async fn test_rest_catalog_sends_partition_statistics_with_their_batch() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "managed_table");
+ ctx.server.add_table("default", "managed_table");
+ let specs = (0..1001)
+ .map(|value| HashMap::from([("dt".to_string(), value.to_string())]))
+ .collect::<Vec<_>>();
+ let statistic = |spec: &HashMap<String, String>| PartitionStatistics {
+ spec: spec.clone(),
+ record_count: 1,
+ file_size_in_bytes: 2,
+ file_count: 3,
+ last_file_creation_time: 4,
+ total_buckets: -1,
+ };
+
+ // Reported for the last partition and the first, in that order.
+ ctx.catalog
+ .create_partitions_with_statistics(
+ &identifier,
+ specs.clone(),
+ true,
+ Some(vec![statistic(&specs[1000]), statistic(&specs[0])]),
+ true,
+ )
+ .await
+ .unwrap();
+
+ let calls = ctx.server.create_partitions_calls();
+ assert_eq!(calls.len(), 2);
+ assert_eq!(
+ calls[0].2.partition_statistics,
+ Some(vec![statistic(&specs[0])])
+ );
+ assert_eq!(
+ calls[1].2.partition_statistics,
+ Some(vec![statistic(&specs[1000])])
+ );
+ assert!(calls
+ .iter()
+ .all(|(_, _, request)| request.replace_statistics == Some(true)));
+
+ // A report the catalog could not place is refused before anything is sent.
+ let unknown = HashMap::from([("dt".to_string(), "x".to_string())]);
+ for statistics in [
+ vec![statistic(&unknown)],
+ vec![statistic(&specs[0]), statistic(&specs[0])],
+ ] {
+ let error = ctx
+ .catalog
+ .create_partitions_with_statistics(
+ &identifier,
+ specs.clone(),
+ true,
+ Some(statistics),
+ true,
+ )
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(error, paimon::Error::DataInvalid { .. }),
+ "{error}"
+ );
+ }
+ assert_eq!(ctx.server.create_partitions_calls().len(), 2);
+}
+
+#[tokio::test]
+async fn
test_rest_catalog_refuses_a_partition_registered_twice_with_statistics() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "managed_table");
+ ctx.server.add_table("default", "managed_table");
+ // The first and the last spec name the same partition, so they land in
different batches.
+ let mut specs = (0..1000)
+ .map(|value| HashMap::from([("dt".to_string(), value.to_string())]))
+ .collect::<Vec<_>>();
+ specs.push(specs[0].clone());
+ let additive = PartitionStatistics {
+ spec: specs[0].clone(),
+ record_count: 7,
+ file_size_in_bytes: 70,
+ file_count: 1,
+ last_file_creation_time: 4,
+ total_buckets: -1,
+ };
+
+ // Both batches would carry the one report, and an additive catalog would
apply it twice.
+ let error = ctx
+ .catalog
+ .create_partitions_with_statistics(
+ &identifier,
+ specs.clone(),
+ true,
+ Some(vec![additive]),
+ false,
+ )
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(&error, paimon::Error::DataInvalid { message, .. } if
message.contains("registered twice")),
+ "{error}"
+ );
+ assert!(ctx.server.create_partitions_calls().is_empty());
+
+ // Without statistics a repeated spec carries nothing that could be
counted twice.
+ ctx.catalog
+ .create_partitions(&identifier, specs, true)
+ .await
+ .unwrap();
+ assert_eq!(ctx.server.create_partitions_calls().len(), 2);
+}
+
+#[tokio::test]
+async fn test_rest_catalog_maps_create_partition_conflict() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "managed_table");
+ ctx.server
+ .set_create_partitions_error_status(Some(StatusCode::CONFLICT));
+
+ let error = ctx
+ .catalog
+ .create_partitions(
+ &identifier,
+ vec![HashMap::from([(
+ "dt".to_string(),
+ "2026-07-22".to_string(),
+ )])],
+ false,
+ )
+ .await
+ .unwrap_err();
+
+ assert!(matches!(
+ error,
+ paimon::Error::DataInvalid { message, .. }
+ if message.contains("default.managed_table")
+ && message.contains("already exist")
+ ));
+}
+
+#[tokio::test]
+async fn test_rest_catalog_maps_invalid_partition_request() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "managed_table");
+ ctx.server
+ .set_create_partitions_error_status(Some(StatusCode::BAD_REQUEST));
+
+ let error = ctx
+ .catalog
+ .create_partitions(
+ &identifier,
+ vec![HashMap::from([(
+ "unknown".to_string(),
+ "value".to_string(),
+ )])],
+ false,
+ )
+ .await
+ .unwrap_err();
+
+ assert!(matches!(
+ error,
+ paimon::Error::DataInvalid { message, .. }
+ if message.contains("default.managed_table")
+ ));
+}
+
+#[tokio::test]
+async fn test_rest_catalog_create_partitions_maps_missing_table() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "missing");
+
+ let error = ctx
+ .catalog
+ .create_partitions(
+ &identifier,
+ vec![HashMap::from([(
+ "dt".to_string(),
+ "2026-07-22".to_string(),
+ )])],
+ false,
+ )
+ .await
+ .unwrap_err();
+
+ assert!(matches!(
+ error,
+ paimon::Error::TableNotExist { full_name } if full_name ==
"default.missing"
+ ));
+}
+
+#[tokio::test]
+async fn test_rest_catalog_looks_up_partitions_by_names_in_batches() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "managed_table");
+ ctx.server.add_table("default", "managed_table");
+ let spec = |value: usize| HashMap::from([("dt".to_string(),
value.to_string())]);
+ let registered = (0..2500).step_by(2).map(spec).collect::<Vec<_>>();
+ ctx.server
+ .set_table_partitions("default", "managed_table", registered.clone());
+ let requested = (0..2500).map(spec).collect::<Vec<_>>();
+
+ assert!(ctx
+ .catalog
+ .list_partitions_by_names(&identifier, Vec::new())
+ .await
+ .unwrap()
+ .is_empty());
+ let found = ctx
+ .catalog
+ .list_partitions_by_names(&identifier, requested.clone())
+ .await
+ .unwrap();
+
+ assert_eq!(
+ found
+ .into_iter()
+ .map(|partition| partition.spec)
+ .collect::<Vec<_>>(),
+ registered
+ );
+ let calls = ctx
+ .server
+ .table_partition_list_by_names_calls("default", "managed_table");
+ assert_eq!(
+ calls.iter().map(Vec::len).collect::<Vec<_>>(),
+ vec![1000, 1000, 500]
+ );
+ assert_eq!(calls.concat(), requested);
+}
+
+#[tokio::test]
+async fn
test_rest_catalog_looks_up_partitions_by_names_through_the_listing_when_unsupported()
{
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "managed_table");
+ ctx.server.add_table("default", "managed_table");
+ let registered = HashMap::from([("dt".to_string(),
"2026-07-22".to_string())]);
+ let missing = HashMap::from([("dt".to_string(),
"2026-07-23".to_string())]);
+ ctx.server
+ .set_table_partitions("default", "managed_table",
vec![registered.clone()]);
+ ctx.server
+
.set_list_partitions_by_names_error_status(Some(StatusCode::NOT_IMPLEMENTED));
+
+ let found = ctx
+ .catalog
+ .list_partitions_by_names(&identifier, vec![registered.clone(),
missing])
+ .await
+ .unwrap();
+
+ assert_eq!(
+ found
+ .into_iter()
+ .map(|partition| partition.spec)
+ .collect::<Vec<_>>(),
+ vec![registered]
+ );
+ assert_eq!(
+ ctx.server
+ .table_partition_list_by_names_calls("default", "managed_table")
+ .len(),
+ 1
+ );
+ assert_eq!(
+ ctx.server
+ .table_partition_list_name_patterns("default", "managed_table"),
+ vec![None]
+ );
+}
+
// ==================== Database Tests ====================
#[tokio::test]
