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 d0ae928b feat(catalog): add tag lifecycle APIs (#824)
d0ae928b is described below
commit d0ae928b729bcc6fe677a491eee440c4786e1e92
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Sep 14 18:26:25 2026 +0800
feat(catalog): add tag lifecycle APIs (#824)
---
crates/paimon/src/api/api_request.rs | 32 +++
crates/paimon/src/api/api_response.rs | 12 +-
crates/paimon/src/api/mod.rs | 8 +-
crates/paimon/src/api/resource_paths.rs | 28 +++
crates/paimon/src/api/rest_api.rs | 59 +++++-
crates/paimon/src/catalog/filesystem.rs | 278 ++++++++++++++++++++++++-
crates/paimon/src/catalog/mod.rs | 37 +++-
crates/paimon/src/catalog/rest/rest_catalog.rs | 136 +++++++++++-
crates/paimon/src/error.rs | 4 +
crates/paimon/src/table/tag_manager.rs | 93 ++++++++-
crates/paimon/tests/mock_server.rs | 116 ++++++++++-
crates/paimon/tests/rest_catalog_test.rs | 51 +++++
12 files changed, 828 insertions(+), 26 deletions(-)
diff --git a/crates/paimon/src/api/api_request.rs
b/crates/paimon/src/api/api_request.rs
index 8c721fdf..42ac80ad 100644
--- a/crates/paimon/src/api/api_request.rs
+++ b/crates/paimon/src/api/api_request.rs
@@ -72,6 +72,25 @@ pub struct RenameTableRequest {
pub destination: Identifier,
}
+/// Request to create a table tag.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CreateTagRequest {
+ pub tag_name: String,
+ pub snapshot_id: Option<i64>,
+ pub time_retained: Option<String>,
+}
+
+impl CreateTagRequest {
+ pub fn new(tag_name: String, snapshot_id: Option<i64>) -> Self {
+ Self {
+ tag_name,
+ snapshot_id,
+ time_retained: None,
+ }
+ }
+}
+
impl RenameTableRequest {
/// Create a new RenameTableRequest.
pub fn new(source: Identifier, destination: Identifier) -> Self {
@@ -349,6 +368,19 @@ mod tests {
assert!(json.contains("\"options\""));
}
+ #[test]
+ fn test_create_tag_request_serialization() {
+ let request = CreateTagRequest::new("release-1".to_string(), Some(42));
+ assert_eq!(
+ serde_json::to_value(request).unwrap(),
+ serde_json::json!({
+ "tagName": "release-1",
+ "snapshotId": 42,
+ "timeRetained": null
+ })
+ );
+ }
+
#[test]
fn test_alter_database_request_serialization() {
let mut updates = HashMap::new();
diff --git a/crates/paimon/src/api/api_response.rs
b/crates/paimon/src/api/api_response.rs
index ce62a392..20d1b7b7 100644
--- a/crates/paimon/src/api/api_response.rs
+++ b/crates/paimon/src/api/api_response.rs
@@ -24,7 +24,7 @@ use std::collections::HashMap;
use crate::api::management::PermissionAssignment;
use crate::catalog::{Function, FunctionDefinition, ViewSchema};
-use crate::spec::{DataField, Schema};
+use crate::spec::{DataField, Schema, Snapshot};
/// Error response from REST API calls.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -132,6 +132,16 @@ pub struct GetTableResponse {
pub schema: Option<Schema>,
}
+/// Response for getting a table tag.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct GetTagResponse {
+ pub tag_name: String,
+ pub snapshot: Snapshot,
+ pub tag_create_time: Option<i64>,
+ pub tag_time_retained: Option<String>,
+}
+
/// Response for getting a persistent view.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
diff --git a/crates/paimon/src/api/mod.rs b/crates/paimon/src/api/mod.rs
index 7f765cf7..88856741 100644
--- a/crates/paimon/src/api/mod.rs
+++ b/crates/paimon/src/api/mod.rs
@@ -33,15 +33,15 @@ mod api_response;
// Re-export request types
pub use api_request::{
AlterDatabaseRequest, AlterTableRequest, AuthTableQueryRequest,
CreateDatabaseRequest,
- CreateFunctionRequest, CreatePartitionsRequest, CreateTableRequest,
CreateViewRequest,
- DropPartitionsRequest, ListPartitionsByFilterRequest,
ListPartitionsByNamesRequest,
- RenameTableRequest, RevokePermissionRequest,
+ CreateFunctionRequest, CreatePartitionsRequest, CreateTableRequest,
CreateTagRequest,
+ CreateViewRequest, DropPartitionsRequest, ListPartitionsByFilterRequest,
+ ListPartitionsByNamesRequest, RenameTableRequest, RevokePermissionRequest,
};
// Re-export response types
pub use api_response::{
AuditRESTResponse, AuthTableQueryResponse, ConfigResponse, ErrorResponse,
GetDatabaseResponse,
- GetFunctionResponse, GetTableResponse, GetTableTokenResponse,
GetViewResponse,
+ GetFunctionResponse, GetTableResponse, GetTableTokenResponse,
GetTagResponse, GetViewResponse,
ListDatabasesResponse, ListFunctionsResponse, ListPartitionsResponse,
ListPermissionsResponse,
ListTablesResponse, ListViewsResponse, PagedList,
};
diff --git a/crates/paimon/src/api/resource_paths.rs
b/crates/paimon/src/api/resource_paths.rs
index b1652172..bc911db2 100644
--- a/crates/paimon/src/api/resource_paths.rs
+++ b/crates/paimon/src/api/resource_paths.rs
@@ -33,6 +33,7 @@ impl ResourcePaths {
const TABLES: &'static str = "tables";
const TABLE_DETAILS: &'static str = "table-details";
const PARTITIONS: &'static str = "partitions";
+ const TAGS: &'static str = "tags";
const VIEWS: &'static str = "views";
const FUNCTIONS: &'static str = "functions";
const PERMISSIONS: &'static str = "permissions";
@@ -211,6 +212,20 @@ impl ResourcePaths {
)
}
+ /// Get the tags endpoint path for a table.
+ pub fn tags(&self, database_name: &str, table_name: &str) -> String {
+ format!("{}/{}", self.table(database_name, table_name), Self::TAGS)
+ }
+
+ /// Get the endpoint path for a table tag.
+ pub fn tag(&self, database_name: &str, table_name: &str, tag_name: &str)
-> String {
+ format!(
+ "{}/{}",
+ self.tags(database_name, table_name),
+ RESTUtil::encode_string(tag_name)
+ )
+ }
+
/// Get the partitions endpoint path for a table.
pub fn partitions(&self, database_name: &str, table_name: &str) -> String {
format!(
@@ -311,6 +326,19 @@ mod tests {
);
}
+ #[test]
+ fn test_tag_paths_encode_names() {
+ let paths = ResourcePaths::new("catalog");
+ assert_eq!(
+ paths.tags("analytics db", "events/table"),
+ "/v1/catalog/databases/analytics+db/tables/events%2Ftable/tags"
+ );
+ assert_eq!(
+ paths.tag("analytics db", "events/table", "release/1"),
+
"/v1/catalog/databases/analytics+db/tables/events%2Ftable/tags/release%2F1"
+ );
+ }
+
#[test]
fn test_config_path() {
assert_eq!(ResourcePaths::config(), "/v1/config");
diff --git a/crates/paimon/src/api/rest_api.rs
b/crates/paimon/src/api/rest_api.rs
index da1f72b7..cac1c33f 100644
--- a/crates/paimon/src/api/rest_api.rs
+++ b/crates/paimon/src/api/rest_api.rs
@@ -30,15 +30,15 @@ use crate::Result;
use super::api_request::{
AlterDatabaseRequest, AlterTableRequest, AuthTableQueryRequest,
CreateDatabaseRequest,
- CreateFunctionRequest, CreatePartitionsRequest, CreateTableRequest,
CreateViewRequest,
- DropPartitionsRequest, ListPartitionsByFilterRequest,
ListPartitionsByNamesRequest,
- RenameTableRequest, RevokePermissionRequest,
+ CreateFunctionRequest, CreatePartitionsRequest, CreateTableRequest,
CreateTagRequest,
+ CreateViewRequest, DropPartitionsRequest, ListPartitionsByFilterRequest,
+ ListPartitionsByNamesRequest, RenameTableRequest, RevokePermissionRequest,
};
use super::api_response::{
AuthTableQueryResponse, ConfigResponse, GetDatabaseResponse,
GetFunctionResponse,
- GetTableResponse, GetViewResponse, ListDatabasesResponse,
ListFunctionsResponse,
- ListPartitionsResponse, ListPermissionsResponse, ListTablesResponse,
ListViewsResponse,
- PagedList,
+ GetTableResponse, GetTagResponse, GetViewResponse, ListDatabasesResponse,
+ ListFunctionsResponse, ListPartitionsResponse, ListPermissionsResponse,
ListTablesResponse,
+ ListViewsResponse, PagedList,
};
use super::auth::{AuthProviderFactory, RESTAuthFunction};
use super::management::{ListPermissionsRequest, PermissionAssignment,
PermissionResource};
@@ -890,6 +890,53 @@ impl RESTApi {
Ok(())
}
+ // ==================== Tag Operations ====================
+
+ pub async fn create_tag(
+ &self,
+ identifier: &Identifier,
+ tag_name: &str,
+ snapshot_id: Option<i64>,
+ ) -> Result<()> {
+ let database = identifier.database();
+ let table = identifier.object();
+ validate_non_empty_multi(&[
+ (database, "database name"),
+ (table, "table name"),
+ (tag_name, "tag name"),
+ ])?;
+ let path = self.resource_paths.tags(database, table);
+ let request = CreateTagRequest::new(tag_name.to_string(), snapshot_id);
+ let _response: serde_json::Value = self.client.post(&path,
&request).await?;
+ Ok(())
+ }
+
+ pub async fn get_tag(&self, identifier: &Identifier, tag_name: &str) ->
Result<GetTagResponse> {
+ let database = identifier.database();
+ let table = identifier.object();
+ validate_non_empty_multi(&[
+ (database, "database name"),
+ (table, "table name"),
+ (tag_name, "tag name"),
+ ])?;
+ let path = self.resource_paths.tag(database, table, tag_name);
+ self.client.get(&path, None::<&[(&str, &str)]>).await
+ }
+
+ pub async fn delete_tag(&self, identifier: &Identifier, tag_name: &str) ->
Result<()> {
+ let database = identifier.database();
+ let table = identifier.object();
+ validate_non_empty_multi(&[
+ (database, "database name"),
+ (table, "table name"),
+ (tag_name, "tag name"),
+ ])?;
+ let path = self.resource_paths.tag(database, table, tag_name);
+ let _response: serde_json::Value =
+ self.client.delete(&path, None::<&[(&str, &str)]>).await?;
+ Ok(())
+ }
+
// ==================== Commit Operations ====================
/// Commit a snapshot for a table.
diff --git a/crates/paimon/src/catalog/filesystem.rs
b/crates/paimon/src/catalog/filesystem.rs
index 8caa96eb..743f0028 100644
--- a/crates/paimon/src/catalog/filesystem.rs
+++ b/crates/paimon/src/catalog/filesystem.rs
@@ -21,18 +21,20 @@
use std::collections::HashMap;
+use crate::api::GetTagResponse;
use crate::catalog::{Catalog, Database, Identifier, DB_LOCATION_PROP,
DB_SUFFIX};
use crate::common::{CatalogOptions, Options};
use crate::error::{ConfigInvalidSnafu, Error, Result};
use crate::io::cache::{create_local_cache, LocalCache};
use crate::io::FileIO;
use crate::spec::{
- CoreOptions, Schema, TableSchema, TableType,
INDEX_FILE_IN_DATA_FILE_DIR_OPTION,
+ CoreOptions, Schema, Snapshot, TableSchema, TableType,
INDEX_FILE_IN_DATA_FILE_DIR_OPTION,
TABLE_TYPE_OPTION,
};
use crate::table::{ObjectTable, SchemaManager, Table};
use async_trait::async_trait;
use bytes::Bytes;
+use chrono::TimeZone;
use opendal::raw::get_basename;
use snafu::OptionExt;
@@ -550,6 +552,138 @@ impl Catalog for FileSystemCatalog {
.map_err(|e| fill_table_name(e, identifier))?;
self.save_table_schema(&table_path, &new_schema).await
}
+
+ async fn create_tag(
+ &self,
+ identifier: &Identifier,
+ tag_name: &str,
+ snapshot_id: Option<i64>,
+ ignore_if_exists: bool,
+ ) -> Result<()> {
+ let table = self.get_table(identifier).await?;
+ let manager = table.tag_manager();
+ if manager.tag_exists(tag_name).await? {
+ return if ignore_if_exists {
+ Ok(())
+ } else {
+ Err(Error::TagAlreadyExist {
+ tag_name: tag_name.to_string(),
+ })
+ };
+ }
+
+ let snapshot = resolve_tag_snapshot(&table, snapshot_id).await?;
+ manager.create(tag_name, &snapshot).await
+ }
+
+ async fn get_tag(&self, identifier: &Identifier, tag_name: &str) ->
Result<GetTagResponse> {
+ let (snapshot, tag_create_time, tag_time_retained) = self
+ .get_table(identifier)
+ .await?
+ .tag_manager()
+ .get_with_raw_metadata(tag_name)
+ .await?
+ .ok_or_else(|| Error::TagNotExist {
+ tag_name: tag_name.to_string(),
+ })?;
+ Ok(GetTagResponse {
+ tag_name: tag_name.to_string(),
+ snapshot,
+ tag_create_time: tag_create_time
+ .and_then(|value| local_datetime_to_millis(&chrono::Local,
value)),
+ tag_time_retained:
tag_time_retained.and_then(format_tag_time_retained),
+ })
+ }
+
+ async fn delete_tag(
+ &self,
+ identifier: &Identifier,
+ tag_name: &str,
+ ignore_if_not_exists: bool,
+ ) -> Result<()> {
+ let table = self.get_table(identifier).await?;
+ let manager = table.tag_manager();
+ let Some(snapshot) = manager.get(tag_name).await? else {
+ return if ignore_if_not_exists {
+ Ok(())
+ } else {
+ Err(Error::TagNotExist {
+ tag_name: tag_name.to_string(),
+ })
+ };
+ };
+ match table.snapshot_manager().get_snapshot(snapshot.id()).await {
+ Ok(_) => manager.delete(tag_name).await,
+ Err(Error::SnapshotNotExist { .. }) => Err(Error::Unsupported {
+ message: "deleting a tag after its snapshot expired is not
supported by FileSystemCatalog"
+ .to_string(),
+ }),
+ Err(error) => Err(error),
+ }
+ }
+}
+
+async fn resolve_tag_snapshot(table: &Table, snapshot_id: Option<i64>) ->
Result<Snapshot> {
+ let snapshot_manager = table.snapshot_manager();
+ let Some(snapshot_id) = snapshot_id else {
+ return snapshot_manager
+ .get_latest_snapshot()
+ .await?
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Cannot create tag because latest snapshot does not
exist".to_string(),
+ source: None,
+ });
+ };
+
+ match snapshot_manager.get_snapshot(snapshot_id).await {
+ Ok(snapshot) => Ok(snapshot),
+ Err(Error::SnapshotNotExist { .. }) => table
+ .tag_manager()
+ .list_all()
+ .await?
+ .into_iter()
+ .map(|(_, snapshot)| snapshot)
+ .find(|snapshot| snapshot.id() == snapshot_id)
+ .ok_or(Error::SnapshotNotExist { snapshot_id }),
+ Err(error) => Err(error),
+ }
+}
+
+fn format_tag_time_retained(seconds: f64) -> Option<String> {
+ let duration = std::time::Duration::try_from_secs_f64(seconds).ok()?;
+ let total_seconds = duration.as_secs();
+ let hours = total_seconds / 3600;
+ let minutes = total_seconds % 3600 / 60;
+ let seconds = total_seconds % 60;
+ let nanos = duration.subsec_nanos();
+
+ let mut value = String::from("PT");
+ if hours != 0 {
+ value.push_str(&format!("{hours}H"));
+ }
+ if minutes != 0 {
+ value.push_str(&format!("{minutes}M"));
+ }
+ if seconds != 0 || nanos != 0 || value == "PT" {
+ value.push_str(&seconds.to_string());
+ if nanos != 0 {
+ let fraction = format!("{nanos:09}");
+ value.push('.');
+ value.push_str(fraction.trim_end_matches('0'));
+ }
+ value.push('S');
+ }
+ Some(value)
+}
+
+fn local_datetime_to_millis<Tz: TimeZone>(
+ timezone: &Tz,
+ value: chrono::NaiveDateTime,
+) -> Option<i64> {
+ timezone
+ .from_local_datetime(&value)
+ .earliest()
+ .map(|value| value.timestamp_millis())
}
/// Options whose value is baked into the on-disk layout, so changing one on a
@@ -907,6 +1041,148 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_tag_operations() {
+ let (_temp_dir, catalog) = create_test_catalog();
+ catalog
+ .create_database("db1", false, HashMap::new())
+ .await
+ .unwrap();
+ let identifier = Identifier::new("db1", "t");
+ catalog
+ .create_table(&identifier, testing_schema(), false)
+ .await
+ .unwrap();
+ give_the_table_a_snapshot(&catalog, &identifier).await;
+
+ catalog
+ .create_tag(&identifier, "release-1", Some(1), false)
+ .await
+ .unwrap();
+ assert_eq!(
+ catalog
+ .get_tag(&identifier, "release-1")
+ .await
+ .unwrap()
+ .snapshot
+ .id(),
+ 1
+ );
+ assert!(matches!(
+ catalog
+ .create_tag(&identifier, "release-1", Some(1), false)
+ .await,
+ Err(Error::TagAlreadyExist { .. })
+ ));
+ catalog
+ .create_tag(&identifier, "release-1", Some(1), true)
+ .await
+ .unwrap();
+
+ catalog
+ .create_tag(&identifier, "latest", None, false)
+ .await
+ .unwrap();
+ assert_eq!(
+ catalog
+ .get_tag(&identifier, "latest")
+ .await
+ .unwrap()
+ .snapshot
+ .id(),
+ 1
+ );
+ assert!(matches!(
+ catalog
+ .create_tag(&identifier, "missing", Some(2), false)
+ .await,
+ Err(Error::SnapshotNotExist { snapshot_id: 2 })
+ ));
+
+ catalog
+ .delete_tag(&identifier, "release-1", false)
+ .await
+ .unwrap();
+ assert!(matches!(
+ catalog.get_tag(&identifier, "release-1").await,
+ Err(Error::TagNotExist { .. })
+ ));
+ assert!(matches!(
+ catalog.delete_tag(&identifier, "release-1", false).await,
+ Err(Error::TagNotExist { .. })
+ ));
+ catalog
+ .delete_tag(&identifier, "release-1", true)
+ .await
+ .unwrap();
+ }
+
+ #[tokio::test]
+ async fn test_delete_expired_tag_is_unsupported() {
+ let (_temp_dir, catalog) = create_test_catalog();
+ catalog
+ .create_database("db1", false, HashMap::new())
+ .await
+ .unwrap();
+ let identifier = Identifier::new("db1", "t");
+ catalog
+ .create_table(&identifier, testing_schema(), false)
+ .await
+ .unwrap();
+ give_the_table_a_snapshot(&catalog, &identifier).await;
+ catalog
+ .create_tag(&identifier, "release-1", Some(1), false)
+ .await
+ .unwrap();
+
+ catalog
+ .get_table(&identifier)
+ .await
+ .unwrap()
+ .snapshot_manager()
+ .delete_snapshot(1)
+ .await
+ .unwrap();
+ assert!(matches!(
+ catalog.delete_tag(&identifier, "release-1", false).await,
+ Err(Error::Unsupported { .. })
+ ));
+ assert_eq!(
+ catalog
+ .get_tag(&identifier, "release-1")
+ .await
+ .unwrap()
+ .snapshot
+ .id(),
+ 1
+ );
+ }
+
+ #[test]
+ fn test_format_tag_time_retained() {
+ assert_eq!(format_tag_time_retained(0.0).as_deref(), Some("PT0S"));
+ assert_eq!(format_tag_time_retained(90.0).as_deref(), Some("PT1M30S"));
+ assert_eq!(
+ format_tag_time_retained(259_200.0).as_deref(),
+ Some("PT72H")
+ );
+ assert_eq!(format_tag_time_retained(1.5).as_deref(), Some("PT1.5S"));
+ }
+
+ #[test]
+ fn test_tag_create_time_uses_catalog_timezone() {
+ let value = chrono::NaiveDate::from_ymd_opt(2024, 1, 2)
+ .unwrap()
+ .and_hms_nano_opt(3, 4, 5, 123_000_000)
+ .unwrap();
+ let shanghai = chrono::FixedOffset::east_opt(8 * 60 * 60).unwrap();
+
+ assert_eq!(
+ local_datetime_to_millis(&shanghai, value),
+ Some(1_704_135_845_123)
+ );
+ }
+
async fn create_table_for_alter(
catalog: &FileSystemCatalog,
options: HashMap<String, String>,
diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs
index 38bd9f13..0c66f8be 100644
--- a/crates/paimon/src/catalog/mod.rs
+++ b/crates/paimon/src/catalog/mod.rs
@@ -261,7 +261,7 @@ impl fmt::Debug for Identifier {
use async_trait::async_trait;
-use crate::api::PagedList;
+use crate::api::{GetTagResponse, PagedList};
use crate::spec::{Partition, Schema, SchemaChange, TableType};
use crate::table::{ObjectTable, Table};
@@ -490,6 +490,41 @@ pub trait Catalog: Send + Sync {
ignore_if_not_exists: bool,
) -> Result<()>;
+ /// Create a tag for a snapshot, or for the latest snapshot when
`snapshot_id` is `None`.
+ async fn create_tag(
+ &self,
+ _identifier: &Identifier,
+ _tag_name: &str,
+ _snapshot_id: Option<i64>,
+ _ignore_if_exists: bool,
+ ) -> Result<()> {
+ Err(Error::Unsupported {
+ message: "tag management is not supported by this
catalog".to_string(),
+ })
+ }
+
+ /// Return a tag and its snapshot metadata.
+ async fn get_tag(&self, _identifier: &Identifier, _tag_name: &str) ->
Result<GetTagResponse> {
+ Err(Error::Unsupported {
+ message: "tag management is not supported by this
catalog".to_string(),
+ })
+ }
+
+ /// Delete a tag.
+ ///
+ /// `FileSystemCatalog` returns [`crate::Error::Unsupported`] when the
tagged snapshot has
+ /// expired because Rust does not yet implement Java's tag file cleanup.
+ async fn delete_tag(
+ &self,
+ _identifier: &Identifier,
+ _tag_name: &str,
+ _ignore_if_not_exists: bool,
+ ) -> Result<()> {
+ Err(Error::Unsupported {
+ message: "tag management is not supported by this
catalog".to_string(),
+ })
+ }
+
// ======================= view methods ===============================
/// Create a persistent view.
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index 41ec6759..fb4f47de 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -28,7 +28,7 @@ use async_trait::async_trait;
use crate::api::management::{ListPermissionsRequest, PermissionAssignment,
PermissionResource};
use crate::api::rest_api::RESTApi;
use crate::api::rest_error::RestError;
-use crate::api::PagedList;
+use crate::api::{GetTagResponse, PagedList};
use crate::catalog::{
list_partitions_from_file_system, Catalog, Database, Identifier,
DB_LOCATION_PROP,
};
@@ -356,6 +356,57 @@ impl Catalog for RESTCatalog {
})
}
+ async fn create_tag(
+ &self,
+ identifier: &Identifier,
+ tag_name: &str,
+ snapshot_id: Option<i64>,
+ ignore_if_exists: bool,
+ ) -> Result<()> {
+ let result = self
+ .api
+ .create_tag(identifier, tag_name, snapshot_id)
+ .await
+ .map_err(|error| map_rest_error_for_tag(error, identifier,
tag_name, snapshot_id));
+ ignore_error_if(result, |error| {
+ ignore_if_exists && matches!(error, Error::TagAlreadyExist { .. })
+ })
+ }
+
+ async fn get_tag(&self, identifier: &Identifier, tag_name: &str) ->
Result<GetTagResponse> {
+ let response = self
+ .api
+ .get_tag(identifier, tag_name)
+ .await
+ .map_err(|error| map_rest_error_for_tag(error, identifier,
tag_name, None))?;
+ if response.tag_name != tag_name {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "REST catalog returned tag '{}' for requested tag
'{tag_name}'",
+ response.tag_name
+ ),
+ source: None,
+ });
+ }
+ Ok(response)
+ }
+
+ async fn delete_tag(
+ &self,
+ identifier: &Identifier,
+ tag_name: &str,
+ ignore_if_not_exists: bool,
+ ) -> Result<()> {
+ let result = self
+ .api
+ .delete_tag(identifier, tag_name)
+ .await
+ .map_err(|error| map_rest_error_for_tag(error, identifier,
tag_name, None));
+ ignore_error_if(result, |error| {
+ ignore_if_not_exists && matches!(error, Error::TagNotExist { .. })
+ })
+ }
+
async fn create_view(
&self,
identifier: &Identifier,
@@ -638,6 +689,64 @@ fn map_rest_error_for_table(err: Error, identifier:
&Identifier) -> Error {
}
}
+fn map_rest_error_for_tag(
+ error: Error,
+ identifier: &Identifier,
+ tag_name: &str,
+ snapshot_id: Option<i64>,
+) -> Error {
+ match error {
+ Error::RestApi {
+ source: RestError::AlreadyExists { .. },
+ } => Error::TagAlreadyExist {
+ tag_name: tag_name.to_string(),
+ },
+ Error::RestApi {
+ source:
+ RestError::NoSuchResource {
+ resource_type,
+ resource_name,
+ ..
+ },
+ } if resource_type
+ .as_deref()
+ .is_some_and(|kind| kind.eq_ignore_ascii_case("snapshot")) =>
+ {
+ match snapshot_id.or_else(|| resource_name.and_then(|value|
value.parse().ok())) {
+ Some(snapshot_id) => Error::SnapshotNotExist { snapshot_id },
+ None => Error::DataInvalid {
+ message: format!(
+ "Cannot create tag '{tag_name}' because the latest
snapshot does not exist"
+ ),
+ source: None,
+ },
+ }
+ }
+ Error::RestApi {
+ source: RestError::NoSuchResource { resource_type, .. },
+ } if resource_type
+ .as_deref()
+ .is_some_and(|kind| kind.eq_ignore_ascii_case("tag")) =>
+ {
+ Error::TagNotExist {
+ tag_name: tag_name.to_string(),
+ }
+ }
+ Error::RestApi {
+ source: RestError::NoSuchResource { .. },
+ } => Error::TableNotExist {
+ full_name: identifier.full_name(),
+ },
+ Error::RestApi {
+ source: RestError::BadRequest { message },
+ } => Error::DataInvalid {
+ message,
+ source: None,
+ },
+ other => map_unsupported_endpoint(other, "tag"),
+ }
+}
+
/// 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
@@ -837,4 +946,29 @@ mod tests {
assert!(catalog.has_local_cache());
}
+
+ #[test]
+ fn test_tag_error_mapping() {
+ let identifier = Identifier::new("db", "table");
+ let error = |resource_type: &str| Error::RestApi {
+ source: RestError::NoSuchResource {
+ resource_type: Some(resource_type.to_string()),
+ resource_name: None,
+ message: "missing".to_string(),
+ },
+ };
+
+ assert!(matches!(
+ map_rest_error_for_tag(error("tag"), &identifier, "release", None),
+ Error::TagNotExist { .. }
+ ));
+ assert!(matches!(
+ map_rest_error_for_tag(error("snapshot"), &identifier, "release",
Some(7)),
+ Error::SnapshotNotExist { snapshot_id: 7 }
+ ));
+ assert!(matches!(
+ map_rest_error_for_tag(error("table"), &identifier, "release",
None),
+ Error::TableNotExist { .. }
+ ));
+ }
}
diff --git a/crates/paimon/src/error.rs b/crates/paimon/src/error.rs
index a8be68f1..18696cdd 100644
--- a/crates/paimon/src/error.rs
+++ b/crates/paimon/src/error.rs
@@ -104,6 +104,10 @@ pub enum Error {
TableNotExist { full_name: String },
#[snafu(display("Snapshot {} does not exist.", snapshot_id))]
SnapshotNotExist { snapshot_id: i64 },
+ #[snafu(display("Tag {} already exists.", tag_name))]
+ TagAlreadyExist { tag_name: String },
+ #[snafu(display("Tag {} does not exist.", tag_name))]
+ TagNotExist { tag_name: String },
#[snafu(display("View {} already exists.", full_name))]
ViewAlreadyExist { full_name: String },
#[snafu(display("View {} does not exist.", full_name))]
diff --git a/crates/paimon/src/table/tag_manager.rs
b/crates/paimon/src/table/tag_manager.rs
index 8bf1a384..668dc9b3 100644
--- a/crates/paimon/src/table/tag_manager.rs
+++ b/crates/paimon/src/table/tag_manager.rs
@@ -66,6 +66,7 @@ impl TagManager {
/// Check if a tag exists.
pub async fn tag_exists(&self, tag_name: &str) -> crate::Result<bool> {
+ validate_tag_name(tag_name)?;
let path = self.tag_path(tag_name);
let input = self.file_io.new_input(&path)?;
input.exists().await
@@ -76,6 +77,7 @@ impl TagManager {
/// Tag files are JSON with the same schema as Snapshot.
/// Reads directly and catches NotFound to avoid a separate exists() IO
round-trip.
pub async fn get(&self, tag_name: &str) -> crate::Result<Option<Snapshot>>
{
+ validate_tag_name(tag_name)?;
let path = self.tag_path(tag_name);
let input = self.file_io.new_input(&path)?;
let bytes = match input.read().await {
@@ -103,6 +105,22 @@ impl TagManager {
&self,
tag_name: &str,
) -> crate::Result<Option<(Snapshot, Option<i64>, Option<f64>)>> {
+ Ok(self.get_with_raw_metadata(tag_name).await?.map(
+ |(snapshot, create_time, time_retained)| {
+ (
+ snapshot,
+ create_time.map(|value|
value.and_utc().timestamp_millis()),
+ time_retained,
+ )
+ },
+ ))
+ }
+
+ pub(crate) async fn get_with_raw_metadata(
+ &self,
+ tag_name: &str,
+ ) -> crate::Result<Option<(Snapshot, Option<chrono::NaiveDateTime>,
Option<f64>)>> {
+ validate_tag_name(tag_name)?;
let path = self.tag_path(tag_name);
let input = self.file_io.new_input(&path)?;
let bytes = match input.read().await {
@@ -126,7 +144,7 @@ impl TagManager {
})?;
let create_time = value
.get(FIELD_TAG_CREATE_TIME)
- .and_then(parse_tag_create_time_millis);
+ .and_then(parse_tag_create_time);
let time_retained = value
.get(FIELD_TAG_TIME_RETAINED)
.and_then(serde_json::Value::as_f64);
@@ -181,17 +199,20 @@ impl TagManager {
/// Create a tag by writing the snapshot JSON to the tag path.
pub async fn create(&self, tag_name: &str, snapshot: &Snapshot) ->
crate::Result<()> {
+ validate_tag_name(tag_name)?;
let path = self.tag_path(tag_name);
let json = serde_json::to_string(snapshot).map_err(|e|
crate::Error::DataInvalid {
message: format!("failed to serialize snapshot for tag
'{tag_name}': {e}"),
source: Some(Box::new(e)),
})?;
+ self.file_io.mkdirs(&self.tag_directory()).await?;
let output = self.file_io.new_output(&path)?;
output.write(bytes::Bytes::from(json)).await
}
/// Delete a tag file.
pub async fn delete(&self, tag_name: &str) -> crate::Result<()> {
+ validate_tag_name(tag_name)?;
let path = self.tag_path(tag_name);
self.file_io.delete_file(&path).await
}
@@ -217,15 +238,28 @@ impl TagManager {
const FIELD_TAG_CREATE_TIME: &str = "tagCreateTime";
const FIELD_TAG_TIME_RETAINED: &str = "tagTimeRetained";
-/// Decode a Jackson-serialized `LocalDateTime` into epoch millis, treating the
-/// wall-clock value as UTC.
+fn validate_tag_name(tag_name: &str) -> crate::Result<()> {
+ let invalid = tag_name.trim().is_empty()
+ || tag_name.trim_end() != tag_name
+ || tag_name.contains('/')
+ || tag_name.contains('\\')
+ || tag_name.chars().any(char::is_control);
+ if invalid {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!("Invalid tag name: {tag_name:?}"),
+ });
+ }
+ Ok(())
+}
+
+/// Decode a Jackson-serialized `LocalDateTime`.
///
/// Jackson's `LocalDateTimeSerializer` emits
/// `[year, month, day, hour, minute, second, nanoOfSecond]` and omits trailing
/// zero components, so the array may hold as few as five items. Anything that
is
/// not such an array -- or that does not describe a real instant -- yields
/// `None` so one odd tag file cannot fail the whole listing.
-fn parse_tag_create_time_millis(value: &serde_json::Value) -> Option<i64> {
+fn parse_tag_create_time(value: &serde_json::Value) ->
Option<chrono::NaiveDateTime> {
let items = value.as_array()?;
if items.len() < 5 || items.len() > 7 {
return None;
@@ -247,7 +281,7 @@ fn parse_tag_create_time_millis(value: &serde_json::Value)
-> Option<i64> {
u32::try_from(second).ok()?,
u32::try_from(nano).ok()?,
)?;
- Some(date.and_time(time).and_utc().timestamp_millis())
+ Some(date.and_time(time))
}
#[cfg(test)]
@@ -290,6 +324,55 @@ mod tests {
assert!(tm.list_all().await.unwrap().is_empty());
}
+ #[tokio::test]
+ async fn test_tag_operations_reject_unsafe_names() {
+ let tm = TagManager::new(test_file_io(),
"memory:/warehouse/table".to_string());
+ let snapshot = test_snapshot(1);
+
+ for name in ["", " ", "tag ", "nested/tag", "nested\\tag",
"bad\nname"] {
+ assert!(matches!(
+ tm.create(name, &snapshot).await,
+ Err(crate::Error::ConfigInvalid { .. })
+ ));
+ assert!(matches!(
+ tm.tag_exists(name).await,
+ Err(crate::Error::ConfigInvalid { .. })
+ ));
+ assert!(matches!(
+ tm.get(name).await,
+ Err(crate::Error::ConfigInvalid { .. })
+ ));
+ assert!(matches!(
+ tm.delete(name).await,
+ Err(crate::Error::ConfigInvalid { .. })
+ ));
+ }
+ }
+
+ #[tokio::test]
+ async fn test_leading_whitespace_tag_name_compatibility() {
+ let tm = TagManager::new(test_file_io(),
"memory:/warehouse/table".to_string());
+ let snapshot = test_snapshot(1);
+
+ tm.create(" tag", &snapshot).await.unwrap();
+ assert_eq!(tm.get(" tag").await.unwrap(), Some(snapshot));
+ assert_eq!(tm.list_all().await.unwrap().len(), 1);
+ assert_eq!(tm.list_all_with_metadata().await.unwrap().len(), 1);
+ }
+
+ #[tokio::test]
+ async fn test_trailing_whitespace_does_not_alias_tag() {
+ let tm = TagManager::new(test_file_io(),
"memory:/warehouse/table".to_string());
+ let snapshot = test_snapshot(1);
+
+ tm.create("tag", &snapshot).await.unwrap();
+ assert!(matches!(
+ tm.delete("tag ").await,
+ Err(crate::Error::ConfigInvalid { .. })
+ ));
+ assert_eq!(tm.get("tag").await.unwrap(), Some(snapshot));
+ }
+
#[tokio::test]
async fn test_list_all_names_sorted() {
let file_io = test_file_io();
diff --git a/crates/paimon/tests/mock_server.rs
b/crates/paimon/tests/mock_server.rs
index 81572025..929e1650 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -35,15 +35,16 @@ use tokio::task::JoinHandle;
use paimon::api::{
AlterDatabaseRequest, AlterTableRequest, AuditRESTResponse, ConfigResponse,
- CreateFunctionRequest, CreatePartitionsRequest, CreateViewRequest,
DropPartitionsRequest,
- ErrorResponse, GetDatabaseResponse, GetFunctionResponse, GetTableResponse,
GetViewResponse,
- ListDatabasesResponse, ListFunctionsResponse,
ListPartitionsByFilterRequest,
- ListPartitionsByNamesRequest, ListPartitionsResponse,
ListPermissionsResponse,
- ListTablesResponse, ListViewsResponse, PermissionAssignment,
PermissionResource,
- RenameTableRequest, ResourcePaths, ResourceType, RevokePermissionRequest,
+ CreateFunctionRequest, CreatePartitionsRequest, CreateTagRequest,
CreateViewRequest,
+ DropPartitionsRequest, ErrorResponse, GetDatabaseResponse,
GetFunctionResponse,
+ GetTableResponse, GetTagResponse, GetViewResponse, ListDatabasesResponse,
+ ListFunctionsResponse, ListPartitionsByFilterRequest,
ListPartitionsByNamesRequest,
+ ListPartitionsResponse, ListPermissionsResponse, ListTablesResponse,
ListViewsResponse,
+ PermissionAssignment, PermissionResource, RenameTableRequest,
ResourcePaths, ResourceType,
+ RevokePermissionRequest,
};
use paimon::catalog::{Function, Identifier};
-use paimon::spec::Partition;
+use paimon::spec::{CommitKind, Partition, Snapshot};
type PartitionPageResponse = (Vec<Partition>, Option<String>);
type PartitionSpecPageResponse = (Vec<HashMap<String, String>>,
Option<String>);
@@ -54,6 +55,7 @@ struct MockState {
tables: HashMap<String, GetTableResponse>,
views: HashMap<String, GetViewResponse>,
functions: HashMap<String, GetFunctionResponse>,
+ tags: HashMap<String, GetTagResponse>,
partitions: HashMap<String, Vec<Partition>>,
partition_page_responses: HashMap<String, Vec<PartitionPageResponse>>,
partition_list_call_counts: HashMap<String, usize>,
@@ -117,6 +119,39 @@ fn partition_from_spec(spec: HashMap<String, String>) ->
Partition {
}
}
+fn tag_snapshot(id: i64) -> Snapshot {
+ Snapshot::builder()
+ .version(3)
+ .id(id)
+ .schema_id(0)
+ .base_manifest_list("base-list".to_string())
+ .delta_manifest_list("delta-list".to_string())
+ .commit_user("test-user".to_string())
+ .commit_identifier(0)
+ .commit_kind(CommitKind::APPEND)
+ .time_millis(1000)
+ .build()
+}
+
+fn resource_error(
+ status: StatusCode,
+ resource_type: &str,
+ resource_name: &str,
+) -> axum::response::Response {
+ let message = if status == StatusCode::CONFLICT {
+ "Already Exists"
+ } else {
+ "Not Found"
+ };
+ let error = ErrorResponse::new(
+ Some(resource_type.to_string()),
+ Some(resource_name.to_string()),
+ Some(message.to_string()),
+ Some(status.as_u16() as i32),
+ );
+ (status, Json(error)).into_response()
+}
+
fn paginate<T: Clone>(
items: Vec<T>,
params: &HashMap<String, String>,
@@ -837,6 +872,65 @@ impl RESTServer {
}
}
+ pub async fn create_tag(
+ Path((db, table)): Path<(String, String)>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ Json(request): Json<CreateTagRequest>,
+ ) -> impl IntoResponse {
+ let mut state = state.inner.lock().unwrap();
+ if !state.tables.contains_key(&format!("{db}.{table}")) {
+ return resource_error(StatusCode::NOT_FOUND, "table", &table);
+ }
+
+ let key = format!("{db}.{table}.{}", request.tag_name);
+ if state.tags.contains_key(&key) {
+ return resource_error(StatusCode::CONFLICT, "tag",
&request.tag_name);
+ }
+ let snapshot_id = request.snapshot_id.unwrap_or(1);
+ if snapshot_id != 1 {
+ return resource_error(StatusCode::NOT_FOUND, "snapshot",
&snapshot_id.to_string());
+ }
+ state.tags.insert(
+ key,
+ GetTagResponse {
+ tag_name: request.tag_name,
+ snapshot: tag_snapshot(snapshot_id),
+ tag_create_time: None,
+ tag_time_retained: request.time_retained,
+ },
+ );
+ (StatusCode::OK, Json(json!(""))).into_response()
+ }
+
+ pub async fn get_tag(
+ Path((db, table, tag)): Path<(String, String, String)>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ ) -> impl IntoResponse {
+ let state = state.inner.lock().unwrap();
+ if !state.tables.contains_key(&format!("{db}.{table}")) {
+ return resource_error(StatusCode::NOT_FOUND, "table", &table);
+ }
+ match state.tags.get(&format!("{db}.{table}.{tag}")) {
+ Some(response) => (StatusCode::OK,
Json(response.clone())).into_response(),
+ None => resource_error(StatusCode::NOT_FOUND, "tag", &tag),
+ }
+ }
+
+ pub async fn delete_tag(
+ Path((db, table, tag)): Path<(String, String, String)>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ ) -> impl IntoResponse {
+ let mut state = state.inner.lock().unwrap();
+ if !state.tables.contains_key(&format!("{db}.{table}")) {
+ return resource_error(StatusCode::NOT_FOUND, "table", &table);
+ }
+ if state.tags.remove(&format!("{db}.{table}.{tag}")).is_some() {
+ (StatusCode::OK, Json(json!(""))).into_response()
+ } else {
+ resource_error(StatusCode::NOT_FOUND, "tag", &tag)
+ }
+ }
+
/// Handle POST /databases/:db/tables/:table/partitions - create
partitions.
pub async fn create_partitions(
Path((db, table)): Path<(String, String)>,
@@ -1815,6 +1909,14 @@ pub async fn start_mock_server(
.post(RESTServer::alter_table)
.delete(RESTServer::drop_table),
)
+ .route(
+ &format!("{prefix}/databases/:db/tables/:table/tags"),
+ post(RESTServer::create_tag),
+ )
+ .route(
+ &format!("{prefix}/databases/:db/tables/:table/tags/:tag"),
+ get(RESTServer::get_tag).delete(RESTServer::delete_tag),
+ )
.route(
&format!("{prefix}/databases/:db/tables/:table/partitions"),
get(RESTServer::list_partitions).post(RESTServer::create_partitions),
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 691c1159..3d7805e0 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -1761,6 +1761,57 @@ async fn test_catalog_alter_table() {
.unwrap();
}
+#[tokio::test]
+async fn test_catalog_tag_lifecycle() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ ctx.server.add_table("default", "managed_table");
+ let identifier = Identifier::new("default", "managed_table");
+
+ ctx.catalog
+ .create_tag(&identifier, "release-1", Some(1), false)
+ .await
+ .unwrap();
+ let tag = ctx.catalog.get_tag(&identifier, "release-1").await.unwrap();
+ assert_eq!(tag.tag_name, "release-1");
+ assert_eq!(tag.snapshot.id(), 1);
+
+ assert!(matches!(
+ ctx.catalog
+ .create_tag(&identifier, "release-1", Some(1), false)
+ .await,
+ Err(paimon::Error::TagAlreadyExist { .. })
+ ));
+ ctx.catalog
+ .create_tag(&identifier, "release-1", Some(1), true)
+ .await
+ .unwrap();
+ assert!(matches!(
+ ctx.catalog
+ .create_tag(&identifier, "missing", Some(2), false)
+ .await,
+ Err(paimon::Error::SnapshotNotExist { snapshot_id: 2 })
+ ));
+
+ ctx.catalog
+ .delete_tag(&identifier, "release-1", false)
+ .await
+ .unwrap();
+ assert!(matches!(
+ ctx.catalog.get_tag(&identifier, "release-1").await,
+ Err(paimon::Error::TagNotExist { .. })
+ ));
+ assert!(matches!(
+ ctx.catalog
+ .delete_tag(&identifier, "release-1", false)
+ .await,
+ Err(paimon::Error::TagNotExist { .. })
+ ));
+ ctx.catalog
+ .delete_tag(&identifier, "release-1", true)
+ .await
+ .unwrap();
+}
+
// ==================== Multiple Databases Tests ====================
#[tokio::test]
