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 6342c2a6 feat(datafusion): add CALL sys.rename_tag (#1023)
6342c2a6 is described below
commit 6342c2a6ec43a6a481ab331f24548453a4ac99cc
Author: jackylee <[email protected]>
AuthorDate: Sun Oct 4 22:21:00 2026 +0800
feat(datafusion): add CALL sys.rename_tag (#1023)
---
crates/integrations/datafusion/src/procedures.rs | 20 +++++++++
crates/paimon/src/table/tag_manager.rs | 55 ++++++++++++++++++++++++
2 files changed, 75 insertions(+)
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index 369c81d9..49eced08 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -20,6 +20,7 @@
//! Supported procedures:
//! - `CALL sys.create_tag(table => '...', tag => '...', snapshot_id => ...)`
//! - `CALL sys.delete_tag(table => '...', tag => '...')`
+//! - `CALL sys.rename_tag(table => '...', tag => '...', target_tag => '...')`
//! - `CALL sys.rollback_to(table => '...', snapshot_id => ... | tag => '...')`
//! - `CALL sys.rollback_to_timestamp(table => '...', timestamp => ...)`
//! - `CALL sys.create_tag_from_timestamp(table => '...', tag => '...',
timestamp => ...)`
@@ -167,6 +168,7 @@ fn declared_parameters(proc_name: &str) -> Option<&'static
[&'static str]> {
Some(match proc_name {
"create_tag" => &["table", "tag", "snapshot_id"],
"delete_tag" => &["table", "tag"],
+ "rename_tag" => &["table", "tag", "target_tag"],
"rollback_to" => &["table", "snapshot_id", "tag"],
"rollback_to_timestamp" => &["table", "timestamp"],
"create_tag_from_timestamp" => &["table", "tag", "timestamp"],
@@ -276,6 +278,7 @@ pub async fn execute_call(
match proc_name.as_str() {
"create_tag" => proc_create_tag(ctx, catalog, catalog_name,
&args).await,
"delete_tag" => proc_delete_tag(ctx, catalog, catalog_name,
&args).await,
+ "rename_tag" => proc_rename_tag(ctx, catalog, catalog_name,
&args).await,
"rollback_to" => proc_rollback_to(ctx, catalog, catalog_name,
&args).await,
"rollback_to_timestamp" => {
proc_rollback_to_timestamp(ctx, catalog, catalog_name, &args).await
@@ -476,6 +479,23 @@ async fn proc_delete_tag(
ok_result(ctx)
}
+async fn proc_rename_tag(
+ ctx: &SessionContext,
+ catalog: &Arc<dyn Catalog>,
+ catalog_name: &str,
+ args: &HashMap<String, String>,
+) -> DFResult<DataFrame> {
+ let table = get_table(catalog, catalog_name, args).await?;
+ let tag_name = require_arg(args, "tag")?;
+ let target_tag_name = require_arg(args, "target_tag")?;
+
+ let (_, tm) = managers(&table);
+ tm.rename(tag_name, target_tag_name)
+ .await
+ .map_err(to_datafusion_error)?;
+ ok_result(ctx)
+}
+
async fn clean_larger_than(
sm: &SnapshotManager,
tm: &TagManager,
diff --git a/crates/paimon/src/table/tag_manager.rs
b/crates/paimon/src/table/tag_manager.rs
index b3a1cb2a..19baefed 100644
--- a/crates/paimon/src/table/tag_manager.rs
+++ b/crates/paimon/src/table/tag_manager.rs
@@ -269,6 +269,33 @@ impl TagManager {
self.file_io.delete_file(&path).await
}
+ /// Rename a tag, preserving its snapshot pointer and the `tagCreateTime` /
+ /// `tagTimeRetained` metadata by moving the raw tag file. Errors if the
+ /// source tag is missing or the target name already exists.
+ ///
+ /// Uses copy-then-delete rather than a native rename: object stores and
the
+ /// in-memory backend do not support rename.
+ pub async fn rename(&self, tag_name: &str, target_tag_name: &str) ->
crate::Result<()> {
+ validate_tag_name(tag_name)?;
+ validate_tag_name(target_tag_name)?;
+ let src = self.tag_path(tag_name);
+ let dst = self.tag_path(target_tag_name);
+ if !self.file_io.new_input(&src)?.exists().await? {
+ return Err(crate::Error::DataInvalid {
+ message: format!("tag '{tag_name}' does not exist"),
+ source: None,
+ });
+ }
+ if self.file_io.new_input(&dst)?.exists().await? {
+ return Err(crate::Error::DataInvalid {
+ message: format!("tag '{target_tag_name}' already exists"),
+ source: None,
+ });
+ }
+ self.file_io.copy_file(&src, &dst).await?;
+ self.file_io.delete_file(&src).await
+ }
+
/// List all tags as `(name, snapshot)` pairs sorted by name ascending.
pub async fn list_all(&self) -> crate::Result<Vec<(String, Snapshot)>> {
let names = self.list_all_names().await?;
@@ -425,6 +452,34 @@ mod tests {
}
}
+ #[tokio::test]
+ async fn test_rename_tag_preserves_metadata() {
+ let tm = TagManager::new(test_file_io(),
"memory:/test_tag_rename".to_string());
+ let snapshot = test_snapshot(7);
+ tm.create_with_retention("v1", &snapshot, Some("1d"))
+ .await
+ .unwrap();
+ let (_, before_create, before_retained) =
+ tm.get_with_metadata("v1").await.unwrap().unwrap();
+ assert!(before_create.is_some());
+ assert!(before_retained.is_some());
+
+ tm.rename("v1", "release").await.unwrap();
+
+ // Source gone; target keeps the original snapshot and metadata
verbatim.
+ assert!(tm.get("v1").await.unwrap().is_none());
+ let (snap, after_create, after_retained) =
+ tm.get_with_metadata("release").await.unwrap().unwrap();
+ assert_eq!(snap.id(), 7);
+ assert_eq!(after_create, before_create);
+ assert_eq!(after_retained, before_retained);
+
+ // A missing source or an existing target is rejected.
+ assert!(tm.rename("missing", "whatever").await.is_err());
+ tm.create("other", &test_snapshot(8)).await.unwrap();
+ assert!(tm.rename("release", "other").await.is_err());
+ }
+
#[tokio::test]
async fn test_create_tag_with_retention_writes_java_metadata() {
let file_io = test_file_io();
