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 370428b0 feat(datafusion): support replace_tag (#1214)
370428b0 is described below
commit 370428b053da5692fad3cf9db32f95e62d72ab98
Author: QuakeWang <[email protected]>
AuthorDate: Sun Oct 25 23:09:55 2026 +0800
feat(datafusion): support replace_tag (#1214)
---
Cargo.lock | 1 +
crates/integrations/datafusion/Cargo.toml | 1 +
crates/integrations/datafusion/src/procedures.rs | 72 +++
.../integrations/datafusion/tests/replace_tag.rs | 627 +++++++++++++++++++++
.../datafusion/tests/replace_tag/support.rs | 258 +++++++++
.../datafusion/tests/rest_management_procedures.rs | 30 +
crates/paimon/src/table/tag_manager.rs | 180 ++++++
docs/src/sql.md | 41 ++
8 files changed, 1210 insertions(+)
diff --git a/Cargo.lock b/Cargo.lock
index 3b57ab16..046a85f8 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4770,6 +4770,7 @@ dependencies = [
"flate2",
"futures",
"log",
+ "opendal-core",
"paimon",
"paimon-ftindex-core",
"parquet",
diff --git a/crates/integrations/datafusion/Cargo.toml
b/crates/integrations/datafusion/Cargo.toml
index b3b33a99..09df5bb3 100644
--- a/crates/integrations/datafusion/Cargo.toml
+++ b/crates/integrations/datafusion/Cargo.toml
@@ -53,6 +53,7 @@ arrow-schema = { workspace = true }
axum = { version = "0.7", features = ["macros", "tokio", "http1", "http2"] }
bytes = "1.7.1"
flate2 = "1"
+opendal = { package = "opendal-core", version = "0.58.0", features =
["services-memory"] }
paimon-ftindex-core = "0.1.0"
parquet = { workspace = true }
tar = "0.4"
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index 385fda0c..6b1d0b28 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -19,6 +19,7 @@
//!
//! Supported procedures:
//! - `CALL sys.create_tag(table => '...', tag => '...', snapshot_id => ...,
time_retained => '...')`
+//! - `CALL sys.replace_tag(table => '...', tag => '...', snapshot_id => ...,
time_retained => '...')`
//! - `CALL sys.delete_tag(table => '...', tag => '...')`
//! - `CALL sys.rename_tag(table => '...', tag => '...', target_tag => '...')`
//! - `CALL sys.reset_consumer(table => '...', consumer_id => '...'[,
next_snapshot_id => ...])`
@@ -204,6 +205,7 @@ async fn earlier_or_equal_from_all(
fn declared_parameters(proc_name: &str) -> Option<&'static [&'static str]> {
Some(match proc_name {
"create_tag" => &["table", "tag", "snapshot_id", "time_retained"],
+ "replace_tag" => &["table", "tag", "snapshot_id", "time_retained"],
"delete_tag" => &["table", "tag"],
"rename_tag" => &["table", "tag", "target_tag"],
"reset_consumer" => &["table", "consumer_id", "next_snapshot_id"],
@@ -326,6 +328,7 @@ pub async fn execute_procedure(
match proc_name {
"create_tag" => proc_create_tag(ctx, catalog, catalog_name,
&args).await,
+ "replace_tag" => proc_replace_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,
"reset_consumer" => proc_reset_consumer(ctx, catalog, catalog_name,
&args).await,
@@ -595,6 +598,75 @@ async fn proc_create_tag(
ok_result(ctx)
}
+async fn proc_replace_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 snapshot_id = args
+ .get("snapshot_id")
+ .map(|value| {
+ value
+ .parse::<i64>()
+ .ok()
+ .filter(|id| *id > 0)
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "Invalid snapshot_id: '{value}'. Expected a positive
BIGINT"
+ ))
+ })
+ })
+ .transpose()?;
+ let time_retained = args.get("time_retained").map(String::as_str);
+
+ if table.rest_env().is_some() {
+ return Err(DataFusionError::NotImplemented(
+ "replace_tag is not supported for tables managed by a REST
catalog".to_string(),
+ ));
+ }
+ if table.schema().core_options().is_format_table() {
+ return Err(DataFusionError::NotImplemented(
+ "replace_tag requires a FileStoreTable; format-table is not
supported".to_string(),
+ ));
+ }
+
+ let sm = table.snapshot_manager();
+ let tm = table.tag_manager();
+ if !tm.tag_exists(tag_name).await.map_err(to_datafusion_error)? {
+ return Err(to_datafusion_error(paimon::Error::TagNotExist {
+ tag_name: tag_name.to_string(),
+ }));
+ }
+ let snapshot = if let Some(id) = snapshot_id {
+ match sm.get_snapshot(id).await {
+ Ok(snapshot) => snapshot,
+ Err(paimon::Error::SnapshotNotExist { .. }) => tm
+ .list_all()
+ .await
+ .map_err(to_datafusion_error)?
+ .into_iter()
+ .map(|(_, snapshot)| snapshot)
+ .find(|snapshot| snapshot.id() == id)
+ .ok_or_else(|| {
+ to_datafusion_error(paimon::Error::SnapshotNotExist {
snapshot_id: id })
+ })?,
+ Err(error) => return Err(to_datafusion_error(error)),
+ }
+ } else {
+ sm.get_latest_snapshot()
+ .await
+ .map_err(to_datafusion_error)?
+ .ok_or_else(|| DataFusionError::Plan("No snapshots
exist".to_string()))?
+ };
+ tm.replace(tag_name, &snapshot, time_retained)
+ .await
+ .map_err(to_datafusion_error)?;
+ ok_result(ctx)
+}
+
async fn proc_rename_branch(
ctx: &SessionContext,
catalog: &Arc<dyn Catalog>,
diff --git a/crates/integrations/datafusion/tests/replace_tag.rs
b/crates/integrations/datafusion/tests/replace_tag.rs
new file mode 100644
index 00000000..0cf5de04
--- /dev/null
+++ b/crates/integrations/datafusion/tests/replace_tag.rs
@@ -0,0 +1,627 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+#[path = "replace_tag/support.rs"]
+mod support;
+
+use std::collections::BTreeMap;
+use std::path::{Path, PathBuf};
+use std::sync::atomic::Ordering;
+
+use common::{
+ assert_sql_error, collect_id_name, create_sql_context, create_test_env,
exec, string_value,
+};
+use paimon::catalog::Identifier;
+use paimon::io::FileIOBuilder;
+use paimon::spec::{DataType, IntType, Schema};
+use paimon::table::{BranchManager, Table};
+use paimon::Catalog;
+use paimon_datafusion::SQLContext;
+
+async fn setup() -> (tempfile::TempDir, SQLContext, Table) {
+ let (tmp, catalog) = create_test_env();
+ let context = create_sql_context(catalog.clone()).await;
+ exec(&context, "CREATE SCHEMA paimon.test_db").await;
+ exec(
+ &context,
+ "CREATE TABLE paimon.test_db.events (id INT, name STRING) \
+ WITH ('tag.default-time-retained' = '7d')",
+ )
+ .await;
+ for (id, name) in [(1, "alice"), (2, "bob")] {
+ exec(
+ &context,
+ &format!("INSERT INTO paimon.test_db.events VALUES ({id},
'{name}')"),
+ )
+ .await;
+ }
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "events"))
+ .await
+ .unwrap();
+ table
+ .tag_manager()
+ .create_with_retention(
+ "release",
+ &table.snapshot_manager().get_snapshot(1).await.unwrap(),
+ Some("1d"),
+ )
+ .await
+ .unwrap();
+ (tmp, context, table)
+}
+
+async fn assert_ok(context: &SQLContext, args: &str) {
+ let batches = context
+ .sql(&format!("CALL sys.replace_tag({args})"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+ assert_eq!(batches[0].num_rows(), 1);
+ assert_eq!(batches[0].num_columns(), 1);
+ assert_eq!(batches[0].schema().field(0).name(), "result");
+ assert_eq!(string_value(batches[0].column(0).as_ref(), 0), "OK");
+}
+
+fn table_files(root: &Path) -> BTreeMap<PathBuf, Vec<u8>> {
+ let mut files = BTreeMap::new();
+ for entry in std::fs::read_dir(root).unwrap() {
+ let path = entry.unwrap().path();
+ if path.is_dir() {
+ files.extend(table_files(&path));
+ } else {
+ files.insert(path.clone(), std::fs::read(path).unwrap());
+ }
+ }
+ files
+}
+
+#[tokio::test]
+async fn replace_tag_updates_snapshots_retention_and_time_travel() {
+ let (tmp, context, table) = setup().await;
+ let root = tmp.path().join("test_db.db/events");
+ let tag_path = root.join("tag/tag-release");
+ let mut original_files = table_files(&root);
+ original_files.remove(&tag_path).unwrap();
+ let first = vec![(1, "alice".to_string())];
+ let latest = vec![(1, "alice".to_string()), (2, "bob".to_string())];
+ let tagged_query = "SELECT id, name FROM paimon.test_db.events VERSION AS
OF 'release'";
+ assert_eq!(collect_id_name(&context, tagged_query).await, first);
+
+ for (args, snapshot_id, retention, expected_rows) in [
+ ("", 2, None, &latest),
+ (
+ ", snapshot_id => 1, time_retained => '2d'",
+ 1,
+ Some("PT48H"),
+ &first,
+ ),
+ (
+ ", snapshot_id => 1, time_retained => '12h'",
+ 1,
+ Some("PT12H"),
+ &first,
+ ),
+ (", snapshot_id => 1", 1, None, &first),
+ ] {
+ assert_ok(
+ &context,
+ &format!("table => 'paimon.test_db.events', tag =>
'release'{args}"),
+ )
+ .await;
+ let batches = context.sql("SELECT snapshot_id, create_time,
time_retained FROM paimon.test_db.`events$tags` WHERE tag_name = 'release'")
+ .await.unwrap().collect().await.unwrap();
+ assert_eq!(
+ batches.iter().map(|batch| batch.num_rows()).sum::<usize>(),
+ 1
+ );
+ let row = &batches[0];
+ let ids = row
+ .column(0)
+ .as_any()
+ .downcast_ref::<arrow_array::Int64Array>()
+ .unwrap();
+ assert_eq!(ids.value(0), snapshot_id);
+ assert_eq!(row.column(1).is_null(0), retention.is_none());
+ let ttl = row.column(2);
+ assert_eq!(
+ (!ttl.is_null(0)).then(|| string_value(ttl.as_ref(), 0)),
+ retention
+ );
+ let value: serde_json::Value =
+
serde_json::from_slice(&std::fs::read(&tag_path).unwrap()).unwrap();
+ assert_eq!(value.get("tagCreateTime").is_some(), retention.is_some());
+ assert_eq!(value.get("tagTimeRetained").is_some(),
retention.is_some());
+ assert_eq!(
+ &collect_id_name(&context, tagged_query).await,
+ expected_rows
+ );
+ assert_eq!(
+ collect_id_name(&context, "SELECT id, name FROM
paimon.test_db.events").await,
+ latest
+ );
+ let mut actual_files = table_files(&root);
+ actual_files.remove(&tag_path).unwrap();
+ assert_eq!(actual_files, original_files);
+ }
+ assert_eq!(
+ table
+ .tag_manager()
+ .get("release")
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ 1
+ );
+}
+
+#[tokio::test]
+async fn replace_tag_works_with_create_rename_and_delete() {
+ let (tmp, context, _table) = setup().await;
+ let root = tmp.path().join("test_db.db/events/tag");
+ let original = std::fs::read(root.join("tag-release")).unwrap();
+ exec(
+ &context,
+ "CALL sys.create_tag(table => 'test_db.events', tag => 'checkpoint',
snapshot_id => 1)",
+ )
+ .await;
+ assert_ok(
+ &context,
+ "table => 'test_db.events', tag => 'checkpoint', time_retained =>
'2d'",
+ )
+ .await;
+ let replaced = std::fs::read(root.join("tag-checkpoint")).unwrap();
+ exec(
+ &context,
+ "CALL sys.rename_tag(table => 'test_db.events', tag => 'checkpoint',
target_tag => 'checkpoint_v2')",
+ )
+ .await;
+ assert!(!root.join("tag-checkpoint").exists());
+ assert_eq!(
+ std::fs::read(root.join("tag-checkpoint_v2")).unwrap(),
+ replaced
+ );
+ assert_eq!(
+ collect_id_name(
+ &context,
+ "SELECT id, name FROM paimon.test_db.events VERSION AS OF
'checkpoint_v2'"
+ )
+ .await,
+ vec![(1, "alice".to_string()), (2, "bob".to_string())]
+ );
+ exec(
+ &context,
+ "CALL sys.delete_tag(table => 'test_db.events', tag =>
'checkpoint_v2')",
+ )
+ .await;
+ assert!(!root.join("tag-checkpoint_v2").exists());
+ assert_eq!(std::fs::read(root.join("tag-release")).unwrap(), original);
+ assert_eq!(std::fs::read_dir(root).unwrap().count(), 1);
+}
+
+#[tokio::test]
+async fn replace_tag_resolves_retained_snapshots_including_its_own() {
+ let (tmp, context, table) = setup().await;
+ let snapshot = table.snapshot_manager().get_snapshot(1).await.unwrap();
+ table.tag_manager().create("keep", &snapshot).await.unwrap();
+ assert_ok(&context, "table => 'test_db.events', tag => 'release'").await;
+
std::fs::remove_file(tmp.path().join("test_db.db/events/snapshot/snapshot-1")).unwrap();
+ assert_ok(
+ &context,
+ "table => 'test_db.events', tag => 'release', snapshot_id => 1",
+ )
+ .await;
+ table.tag_manager().delete("keep").await.unwrap();
+ assert_ok(
+ &context,
+ "table => 'test_db.events', tag => 'release', snapshot_id => 1,
time_retained => '2d'",
+ )
+ .await;
+ assert_eq!(
+ collect_id_name(
+ &context,
+ "SELECT id, name FROM paimon.test_db.events VERSION AS OF
'release'"
+ )
+ .await,
+ vec![(1, "alice".to_string())]
+ );
+}
+
+#[tokio::test]
+async fn replace_tag_rejects_invalid_arguments_without_writes() {
+ let (tmp, context, _table) = setup().await;
+ let root = tmp.path().join("test_db.db/events");
+ let original = table_files(&root);
+ for (args, message) in [
+ ("tag => 'release'", "Missing required argument: 'table'"),
+ (
+ "table => 'test_db.events'",
+ "Missing required argument: 'tag'",
+ ),
+ (
+ "table => 'test_db.events', tag => 'missing'",
+ "Tag missing does not exist",
+ ),
+ ("table => 'test_db.events', tag => ''", "Invalid tag name"),
+ (
+ "table => 'test_db.events', tag => '../release'",
+ "Invalid tag name",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release '",
+ "Invalid tag name",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', snapshot_id =>
'abc'",
+ "Invalid snapshot_id",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', snapshot_id => 1.5",
+ "Invalid snapshot_id",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', snapshot_id => 0",
+ "Invalid snapshot_id",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', snapshot_id => -1",
+ "Invalid snapshot_id",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', snapshot_id =>
9223372036854775808",
+ "Invalid snapshot_id",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', snapshot_id =>
-9223372036854775809",
+ "Invalid snapshot_id",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', snapshot_id => 42",
+ "Snapshot 42 does not exist",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', snapshot => 1",
+ "Argument snapshot is unknown",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', time_retain => '1d'",
+ "Argument time_retain is unknown",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', snapshot_id => 1,
snapshot_id => 2",
+ "Procedure argument snapshot_id is duplicated",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', tag => 'other'",
+ "Procedure argument tag is duplicated",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', time_retained => ''",
+ "Invalid tag retention",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', time_retained =>
'-1d'",
+ "Invalid tag retention",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', time_retained =>
'invalid'",
+ "Invalid tag retention",
+ ),
+ (
+ "table => 'test_db.events', tag => 'release', time_retained =>
'106751991167301d'",
+ "Invalid tag retention",
+ ),
+ ] {
+ assert_sql_error(&context, &format!("CALL sys.replace_tag({args})"),
message).await;
+ assert_eq!(table_files(&root), original, "{args}");
+ }
+}
+
+#[tokio::test]
+async fn replace_tag_requires_a_live_latest_snapshot_when_id_is_omitted() {
+ let (tmp, context, _table) = setup().await;
+ let root = tmp.path().join("test_db.db/events");
+ for id in [1, 2] {
+
std::fs::remove_file(root.join(format!("snapshot/snapshot-{id}"))).unwrap();
+ }
+ for (remove_hint, message) in [
+ (false, "Snapshot 2 does not exist"),
+ (true, "No snapshots exist"),
+ ] {
+ if remove_hint {
+ std::fs::remove_file(root.join("snapshot/LATEST")).unwrap();
+ }
+ let original = table_files(&root);
+ assert_sql_error(
+ &context,
+ "CALL sys.replace_tag(table => 'test_db.events', tag =>
'release')",
+ message,
+ )
+ .await;
+ assert_eq!(table_files(&root), original);
+ }
+ assert_ok(
+ &context,
+ "table => 'test_db.events', tag => 'release', snapshot_id => 1",
+ )
+ .await;
+}
+
+#[tokio::test]
+async fn replace_tag_propagates_invalid_live_snapshot_instead_of_using_a_tag()
{
+ let (tmp, context, table) = setup().await;
+ let root = tmp.path().join("test_db.db/events");
+ let snapshot = table.snapshot_manager().get_snapshot(1).await.unwrap();
+ for (content, message) in [
+ (b"{".to_vec(), "snapshot JSON invalid"),
+ (
+
serde_json::to_vec(&table.snapshot_manager().get_snapshot(2).await.unwrap()).unwrap(),
+ "snapshot file id mismatch",
+ ),
+ ] {
+ std::fs::write(root.join("snapshot/snapshot-1"), content).unwrap();
+ let original = table_files(&root);
+ assert_sql_error(
+ &context,
+ "CALL sys.replace_tag(table => 'test_db.events', tag => 'release',
snapshot_id => 1)",
+ message,
+ )
+ .await;
+ assert_eq!(table_files(&root), original);
+ assert_eq!(
+ table.tag_manager().get("release").await.unwrap().unwrap(),
+ snapshot
+ );
+ }
+}
+
+#[tokio::test]
+async fn replace_tag_preserves_other_branches_and_existing_name_resolution() {
+ let (tmp, context, table) = setup().await;
+ BranchManager::new(table.file_io().clone(), table.location().to_string())
+ .create_branch_from_tag("dev", "release")
+ .await
+ .unwrap();
+ let branch = table.copy_with_branch("dev").await.unwrap();
+ branch
+ .tag_manager()
+ .create(
+ "release",
+ &table.snapshot_manager().get_snapshot(1).await.unwrap(),
+ )
+ .await
+ .unwrap();
+ let root = tmp.path().join("test_db.db/events");
+ let original_branch = table_files(&root.join("branch"));
+ assert_ok(&context, "table => 'test_db.events', tag => 'release'").await;
+ assert_eq!(table_files(&root.join("branch")), original_branch);
+ let original = table_files(&root);
+ assert_sql_error(
+ &context,
+ "CALL sys.replace_tag(table => 'test_db.events$branch_dev', tag =>
'release')",
+ "does not exist",
+ )
+ .await;
+ assert_eq!(table_files(&root), original);
+}
+
+#[tokio::test]
+async fn replace_tag_uses_the_resolved_branch_for_snapshots_and_tags() {
+ let (tmp, _context, table) = setup().await;
+ BranchManager::new(table.file_io().clone(), table.location().to_string())
+ .create_branch_from_tag("dev", "release")
+ .await
+ .unwrap();
+ let branch = table.copy_with_branch("dev").await.unwrap();
+ assert_eq!(branch.branch(), "dev");
+
+ // Distinct metadata for the same ID also catches lookups in the wrong
namespace.
+ let mut value =
+
serde_json::to_value(table.snapshot_manager().get_snapshot(1).await.unwrap()).unwrap();
+ value["commitUser"] = serde_json::json!("dev-live");
+ let live: paimon::spec::Snapshot =
serde_json::from_value(value.clone()).unwrap();
+ branch
+ .file_io()
+ .new_output(&branch.snapshot_manager().snapshot_path(1))
+ .unwrap()
+ .write(serde_json::to_vec(&live).unwrap().into())
+ .await
+ .unwrap();
+ branch
+ .tag_manager()
+ .create_with_retention("release", &live, Some("1d"))
+ .await
+ .unwrap();
+ value["id"] = serde_json::json!(3);
+ value["commitUser"] = serde_json::json!("dev-retained");
+ let retained: paimon::spec::Snapshot =
serde_json::from_value(value.clone()).unwrap();
+ branch
+ .tag_manager()
+ .create("keep", &retained)
+ .await
+ .unwrap();
+ value["commitUser"] = serde_json::json!("main-retained");
+ table
+ .tag_manager()
+ .create("keep", &serde_json::from_value(value.clone()).unwrap())
+ .await
+ .unwrap();
+ value["id"] = serde_json::json!(4);
+ table
+ .tag_manager()
+ .create("main_only", &serde_json::from_value(value).unwrap())
+ .await
+ .unwrap();
+
+ let root = tmp.path().join("test_db.db/events");
+ let branch_root = root.join("branch/branch-dev");
+ let mut original_main = table_files(&root);
+ original_main.retain(|path, _| !path.starts_with(&branch_root));
+ let context = support::context_with_table(tmp.path(),
branch.clone()).await;
+
+ for (args, expected) in [
+ ("", &live),
+ (", snapshot_id => 1", &live),
+ (", snapshot_id => 3", &retained),
+ ] {
+ assert_ok(
+ &context,
+ &format!("table => 'test_db.events', tag => 'release'{args}"),
+ )
+ .await;
+ assert_eq!(
+ branch.tag_manager().get("release").await.unwrap().as_ref(),
+ Some(expected)
+ );
+ let mut current_main = table_files(&root);
+ current_main.retain(|path, _| !path.starts_with(&branch_root));
+ assert_eq!(current_main, original_main);
+ }
+
+ // Neither a main-only live snapshot nor a main-only retained snapshot may
resolve.
+ for id in [2, 4] {
+ let original = table_files(&root);
+ assert_sql_error(
+ &context,
+ &format!("CALL sys.replace_tag(table => 'test_db.events', tag =>
'release', snapshot_id => {id})"),
+ &format!("Snapshot {id} does not exist"),
+ ).await;
+ assert_eq!(table_files(&root), original);
+ }
+}
+
+#[tokio::test]
+async fn replace_tag_propagates_writer_creation_failure_through_core_and_sql()
{
+ let (tmp, _context, table) = setup().await;
+ let snapshot = table.snapshot_manager().get_snapshot(1).await.unwrap();
+ let operator =
opendal::Operator::new(opendal::services::Memory::default()).unwrap();
+ let file_io = FileIOBuilder::new("fs")
+ .with_fs_operator(operator.clone())
+ .build()
+ .unwrap();
+ let clean = Table::from_resolved_schema(
+ file_io,
+ table.identifier().clone(),
+ "file:/write_failure".to_string(),
+ table.schema().clone(),
+ "main",
+ )
+ .unwrap();
+ clean
+ .snapshot_manager()
+ .commit_snapshot(&snapshot)
+ .await
+ .unwrap();
+ clean
+ .tag_manager()
+ .create("release", &snapshot)
+ .await
+ .unwrap();
+
+ let (operator, attempts) = support::fail_tag_writes(operator);
+ let file_io = FileIOBuilder::new("fs")
+ .with_fs_operator(operator)
+ .build()
+ .unwrap();
+ let failing = Table::from_resolved_schema(
+ file_io,
+ clean.identifier().clone(),
+ clean.location().to_string(),
+ clean.schema().clone(),
+ "main",
+ )
+ .unwrap();
+ assert_eq!(
+ failing.tag_manager().get("release").await.unwrap(),
+ Some(snapshot.clone())
+ );
+ assert_eq!(
+ failing.snapshot_manager().get_snapshot(1).await.unwrap(),
+ snapshot
+ );
+ assert_eq!(attempts.load(Ordering::SeqCst), 0);
+
+ let error = failing
+ .tag_manager()
+ .replace("release", &snapshot, Some("2d"))
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(&error, paimon::Error::IoUnexpected { source, .. } if
source.kind() == opendal::ErrorKind::Unexpected)
+ );
+ assert!(
+ error.to_string().contains(support::WRITE_FAILURE),
+ "{error}"
+ );
+ assert_eq!(attempts.load(Ordering::SeqCst), 1);
+
+ let context = support::context_with_table(tmp.path(), failing).await;
+ assert_sql_error(
+ &context,
+ "CALL sys.replace_tag(table => 'test_db.events', tag => 'release',
snapshot_id => 1, time_retained => '2d')",
+ support::WRITE_FAILURE,
+ ).await;
+ assert_eq!(attempts.load(Ordering::SeqCst), 2);
+}
+
+#[tokio::test]
+async fn replace_tag_rejects_unsupported_table_types_before_reading_metadata()
{
+ for table_type in [
+ "format-table",
+ "object-table",
+ "iceberg-table",
+ "lance-table",
+ ] {
+ let (tmp, catalog) = create_test_env();
+ catalog
+ .create_database("test_db", false, Default::default())
+ .await
+ .unwrap();
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .option("type", table_type)
+ .option("file.format", "parquet")
+ .build()
+ .unwrap();
+ catalog
+ .create_table(&Identifier::new("test_db", "events"), schema, false)
+ .await
+ .unwrap();
+ let root = tmp.path().join("test_db.db/events");
+ std::fs::create_dir(root.join("tag")).unwrap();
+ std::fs::write(root.join("tag/tag-release"), "{").unwrap();
+ let original = table_files(&root);
+ let context = create_sql_context(catalog).await;
+ let message = if table_type == "format-table" {
+ "requires a FileStoreTable"
+ } else {
+ "cannot be read as a Paimon table"
+ };
+ assert_sql_error(
+ &context,
+ "CALL sys.replace_tag(table => 'test_db.events', tag => 'release',
snapshot_id => 1)",
+ message,
+ )
+ .await;
+ assert_eq!(table_files(&root), original);
+ }
+}
diff --git a/crates/integrations/datafusion/tests/replace_tag/support.rs
b/crates/integrations/datafusion/tests/replace_tag/support.rs
new file mode 100644
index 00000000..5691e9af
--- /dev/null
+++ b/crates/integrations/datafusion/tests/replace_tag/support.rs
@@ -0,0 +1,258 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+use std::path::Path;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use opendal::raw::*;
+use opendal::{Capability, Operator};
+use paimon::catalog::{Database, Identifier};
+use paimon::spec::{Schema, SchemaChange};
+use paimon::table::Table;
+use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options};
+use paimon_datafusion::SQLContext;
+
+pub async fn context_with_table(warehouse: &Path, table: Table) -> SQLContext {
+ let mut options = Options::new();
+ options.set(
+ CatalogOptions::WAREHOUSE,
+ format!("file://{}", warehouse.display()),
+ );
+ let catalog = ResolvedTableCatalog {
+ inner: FileSystemCatalog::new(options).unwrap(),
+ table,
+ };
+ let mut context = SQLContext::new();
+ context
+ .register_catalog_with_default_db("paimon", Arc::new(catalog), None)
+ .await
+ .unwrap();
+ context
+}
+
+struct ResolvedTableCatalog {
+ inner: FileSystemCatalog,
+ table: Table,
+}
+
+#[async_trait]
+impl Catalog for ResolvedTableCatalog {
+ async fn list_databases(&self) -> paimon::Result<Vec<String>> {
+ self.inner.list_databases().await
+ }
+
+ async fn create_database(
+ &self,
+ name: &str,
+ ignore_if_exists: bool,
+ properties: HashMap<String, String>,
+ ) -> paimon::Result<()> {
+ self.inner
+ .create_database(name, ignore_if_exists, properties)
+ .await
+ }
+
+ async fn get_database(&self, name: &str) -> paimon::Result<Database> {
+ self.inner.get_database(name).await
+ }
+
+ async fn drop_database(
+ &self,
+ name: &str,
+ ignore_if_not_exists: bool,
+ cascade: bool,
+ ) -> paimon::Result<()> {
+ self.inner
+ .drop_database(name, ignore_if_not_exists, cascade)
+ .await
+ }
+
+ async fn get_table(&self, identifier: &Identifier) ->
paimon::Result<Table> {
+ assert_eq!(identifier, self.table.identifier());
+ Ok(self.table.clone())
+ }
+
+ async fn list_tables(&self, database_name: &str) ->
paimon::Result<Vec<String>> {
+ self.inner.list_tables(database_name).await
+ }
+
+ async fn create_table(
+ &self,
+ identifier: &Identifier,
+ creation: Schema,
+ ignore_if_exists: bool,
+ ) -> paimon::Result<()> {
+ self.inner
+ .create_table(identifier, creation, ignore_if_exists)
+ .await
+ }
+
+ async fn drop_table(
+ &self,
+ identifier: &Identifier,
+ ignore_if_not_exists: bool,
+ ) -> paimon::Result<()> {
+ self.inner
+ .drop_table(identifier, ignore_if_not_exists)
+ .await
+ }
+
+ async fn rename_table(
+ &self,
+ from: &Identifier,
+ to: &Identifier,
+ ignore_if_not_exists: bool,
+ ) -> paimon::Result<()> {
+ self.inner
+ .rename_table(from, to, ignore_if_not_exists)
+ .await
+ }
+
+ async fn alter_table(
+ &self,
+ identifier: &Identifier,
+ changes: Vec<SchemaChange>,
+ ignore_if_not_exists: bool,
+ ) -> paimon::Result<()> {
+ self.inner
+ .alter_table(identifier, changes, ignore_if_not_exists)
+ .await
+ }
+}
+
+pub const WRITE_FAILURE: &str = "injected tag writer creation failure";
+
+pub fn fail_tag_writes(operator: Operator) -> (Operator, Arc<AtomicUsize>) {
+ let attempts = Arc::new(AtomicUsize::new(0));
+ let (context, inner) = operator.into_parts();
+ let service = FailTagWrites {
+ inner,
+ attempts: attempts.clone(),
+ };
+ (Operator::from_parts(context, Arc::new(service)), attempts)
+}
+
+#[derive(Debug)]
+struct FailTagWrites {
+ inner: Servicer,
+ attempts: Arc<AtomicUsize>,
+}
+
+impl Service for FailTagWrites {
+ type Reader = oio::Reader;
+ type Writer = oio::Writer;
+ type Lister = oio::Lister;
+ type Deleter = oio::Deleter;
+ type Copier = oio::Copier;
+
+ fn info(&self) -> ServiceInfo {
+ self.inner.info()
+ }
+
+ fn capability(&self) -> Capability {
+ self.inner.capability()
+ }
+
+ async fn create_dir(
+ &self,
+ ctx: &opendal::OperationContext,
+ path: &str,
+ args: OpCreateDir,
+ ) -> opendal::Result<RpCreateDir> {
+ self.inner.create_dir(ctx, path, args).await
+ }
+
+ async fn stat(
+ &self,
+ ctx: &opendal::OperationContext,
+ path: &str,
+ args: OpStat,
+ ) -> opendal::Result<RpStat> {
+ self.inner.stat(ctx, path, args).await
+ }
+
+ fn read(
+ &self,
+ ctx: &opendal::OperationContext,
+ path: &str,
+ args: OpRead,
+ ) -> opendal::Result<Self::Reader> {
+ self.inner.read(ctx, path, args)
+ }
+
+ fn write(
+ &self,
+ ctx: &opendal::OperationContext,
+ path: &str,
+ args: OpWrite,
+ ) -> opendal::Result<Self::Writer> {
+ if path.ends_with("/tag/tag-release") {
+ self.attempts.fetch_add(1, Ordering::SeqCst);
+ return Err(opendal::Error::new(
+ opendal::ErrorKind::Unexpected,
+ WRITE_FAILURE,
+ ));
+ }
+ self.inner.write(ctx, path, args)
+ }
+
+ fn delete(&self, ctx: &opendal::OperationContext) ->
opendal::Result<Self::Deleter> {
+ self.inner.delete(ctx)
+ }
+
+ fn list(
+ &self,
+ ctx: &opendal::OperationContext,
+ path: &str,
+ args: OpList,
+ ) -> opendal::Result<Self::Lister> {
+ self.inner.list(ctx, path, args)
+ }
+
+ fn copy(
+ &self,
+ ctx: &opendal::OperationContext,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> opendal::Result<Self::Copier> {
+ self.inner.copy(ctx, from, to, args, opts)
+ }
+
+ async fn rename(
+ &self,
+ ctx: &opendal::OperationContext,
+ from: &str,
+ to: &str,
+ args: OpRename,
+ ) -> opendal::Result<RpRename> {
+ self.inner.rename(ctx, from, to, args).await
+ }
+
+ async fn presign(
+ &self,
+ ctx: &opendal::OperationContext,
+ path: &str,
+ args: OpPresign,
+ ) -> opendal::Result<RpPresign> {
+ self.inner.presign(ctx, path, args).await
+ }
+}
diff --git a/crates/integrations/datafusion/tests/rest_management_procedures.rs
b/crates/integrations/datafusion/tests/rest_management_procedures.rs
index 22568abd..486e4c8b 100644
--- a/crates/integrations/datafusion/tests/rest_management_procedures.rs
+++ b/crates/integrations/datafusion/tests/rest_management_procedures.rs
@@ -684,6 +684,36 @@ async fn
test_consumer_procedures_rejected_on_rest_catalog_without_changing_prog
assert_eq!(std::fs::read(&consumer_path).unwrap(), original);
}
+#[tokio::test]
+async fn test_replace_tag_rejected_on_rest_catalog_without_changing_metadata()
{
+ let (tmp, server, context) = setup().await;
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .build()
+ .unwrap();
+ let path = format!("file://{}/{DATABASE}.db/{TABLE}",
tmp.path().display());
+ server.add_table_with_schema(DATABASE, TABLE, schema, &path);
+ let tag_dir = tmp.path().join(format!("{DATABASE}.db/{TABLE}/tag"));
+ std::fs::create_dir_all(&tag_dir).unwrap();
+ let tag_path = tag_dir.join("tag-release");
+ // Invalid metadata must remain unread: REST rejection precedes storage
access.
+ let original = b"{";
+ std::fs::write(&tag_path, original).unwrap();
+
+ for snapshot in ["", ", snapshot_id => 1"] {
+ common::assert_sql_error(
+ &context,
+ &format!(
+ "CALL sys.replace_tag(table => '{DATABASE}.{TABLE}', tag =>
'release'{snapshot})"
+ ),
+ "replace_tag is not supported for tables managed by a REST
catalog",
+ )
+ .await;
+ assert_eq!(std::fs::read(&tag_path).unwrap(), original);
+ assert_eq!(std::fs::read_dir(&tag_dir).unwrap().count(), 1);
+ }
+}
+
#[tokio::test]
async fn test_rename_branch_rejected_on_rest_catalog() {
// A REST catalog is authoritative for branch metadata and has no rename
diff --git a/crates/paimon/src/table/tag_manager.rs
b/crates/paimon/src/table/tag_manager.rs
index 19baefed..a9dce5db 100644
--- a/crates/paimon/src/table/tag_manager.rs
+++ b/crates/paimon/src/table/tag_manager.rs
@@ -262,6 +262,26 @@ impl TagManager {
output.write(bytes::Bytes::from(json)).await
}
+ /// Replace an existing tag's snapshot and optional retention.
+ ///
+ /// An explicit retention resets the creation time. `None` removes both
TTL fields.
+ /// Validation completes before writing. The existence check and overwrite
are not
+ /// atomic; concurrent tag changes and rollback after storage failures are
not supported.
+ pub async fn replace(
+ &self,
+ tag_name: &str,
+ snapshot: &Snapshot,
+ time_retained: Option<&str>,
+ ) -> crate::Result<()> {
+ if !self.tag_exists(tag_name).await? {
+ return Err(crate::Error::TagNotExist {
+ tag_name: tag_name.to_string(),
+ });
+ }
+ self.create_with_retention(tag_name, snapshot, time_retained)
+ .await
+ }
+
/// Delete a tag file.
pub async fn delete(&self, tag_name: &str) -> crate::Result<()> {
validate_tag_name(tag_name)?;
@@ -452,6 +472,166 @@ mod tests {
}
}
+ #[tokio::test]
+ async fn test_replace_tag_changes_snapshot_and_requires_existing_tag() {
+ let file_io = test_file_io();
+ let tm = TagManager::new(file_io.clone(),
"memory:/test_tag_replace".to_string());
+ assert!(matches!(
+ tm.replace("missing", &test_snapshot(2), None).await,
+ Err(crate::Error::TagNotExist { tag_name }) if tag_name ==
"missing"
+ ));
+ assert!(!file_io.exists(&tm.tag_directory()).await.unwrap());
+
+ tm.create("release", &test_snapshot(1)).await.unwrap();
+ tm.replace("release", &test_snapshot(2), None)
+ .await
+ .unwrap();
+ assert_eq!(tm.get("release").await.unwrap(), Some(test_snapshot(2)));
+ assert_eq!(tm.list_all_names().await.unwrap(), vec!["release"]);
+ }
+
+ #[tokio::test]
+ async fn
test_replace_tag_resets_retention_metadata_and_clears_omitted_ttl() {
+ let file_io = test_file_io();
+ let tm = TagManager::new(file_io.clone(),
"memory:/test_tag_replace_ttl".to_string());
+ let snapshot = test_snapshot(1);
+ let path = tm.tag_path("release");
+ tm.create("release", &snapshot).await.unwrap();
+ for (retention, seconds) in [("1d", "86400.000000000"), ("2d",
"172800.000000000")] {
+ if retention == "2d" {
+ // Seed a known old time so the test does not depend on clock
resolution.
+ let bytes =
file_io.new_input(&path).unwrap().read().await.unwrap();
+ let mut value: serde_json::Value =
serde_json::from_slice(&bytes).unwrap();
+ value[FIELD_TAG_CREATE_TIME] = serde_json::json!([2000, 1, 1,
0, 0, 0, 0]);
+ file_io
+ .new_output(&path)
+ .unwrap()
+ .write(Bytes::from(serde_json::to_vec(&value).unwrap()))
+ .await
+ .unwrap();
+ }
+ let before = chrono::Local::now().naive_local();
+ tm.replace("release", &snapshot, Some(retention))
+ .await
+ .unwrap();
+ let after = chrono::Local::now().naive_local();
+ let bytes =
file_io.new_input(&path).unwrap().read().await.unwrap();
+ let value: serde_json::Value =
serde_json::from_slice(&bytes).unwrap();
+ assert_eq!(
+ value[FIELD_TAG_TIME_RETAINED]
+ .as_number()
+ .unwrap()
+ .to_string(),
+ seconds
+ );
+ let created =
parse_tag_create_time(&value[FIELD_TAG_CREATE_TIME]).unwrap();
+ assert!(before <= created && created <= after);
+ assert_eq!(serde_json::from_value::<Snapshot>(value).unwrap(),
snapshot);
+ }
+ tm.replace("release", &snapshot, None).await.unwrap();
+ let bytes = file_io.new_input(&path).unwrap().read().await.unwrap();
+ assert_eq!(bytes.as_ref(), serde_json::to_vec(&snapshot).unwrap());
+ let value: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+ assert!(value.get(FIELD_TAG_CREATE_TIME).is_none());
+ assert!(value.get(FIELD_TAG_TIME_RETAINED).is_none());
+ }
+
+ #[tokio::test]
+ async fn test_replace_tag_validation_preserves_original_bytes() {
+ let file_io = test_file_io();
+ let tm = TagManager::new(
+ file_io.clone(),
+ "memory:/test_tag_replace_invalid".to_string(),
+ );
+ tm.create_with_retention("release", &test_snapshot(1), Some("1d"))
+ .await
+ .unwrap();
+ let path = tm.tag_path("release");
+ let original = file_io.new_input(&path).unwrap().read().await.unwrap();
+ for name in ["", " ", "release ", "../tag-release", "a/b", "a\\b",
"a\nb"] {
+ assert!(matches!(
+ tm.replace(name, &test_snapshot(2), None).await,
+ Err(crate::Error::ConfigInvalid { .. })
+ ));
+ assert_eq!(
+ file_io.new_input(&path).unwrap().read().await.unwrap(),
+ original
+ );
+ }
+ for retention in [
+ "",
+ "bad",
+ "-1d",
+ "1.5h",
+ "PT24H",
+ "9223372036854775808ns",
+ "106751991167301d",
+ ] {
+ assert!(matches!(
+ tm.replace("release", &test_snapshot(2), Some(retention))
+ .await,
+ Err(crate::Error::ConfigInvalid { .. })
+ ));
+ assert_eq!(
+ file_io.new_input(&path).unwrap().read().await.unwrap(),
+ original
+ );
+ }
+ assert_eq!(tm.list_all_names().await.unwrap(), vec!["release"]);
+ }
+
+ #[tokio::test]
+ async fn test_replace_tag_keeps_branch_managers_isolated() {
+ let file_io = test_file_io();
+ let main = TagManager::new(
+ file_io.clone(),
+ "memory:/test_tag_replace_branch".to_string(),
+ );
+ let branch = main.with_branch("dev");
+ main.create("release", &test_snapshot(1)).await.unwrap();
+ branch.create("release", &test_snapshot(10)).await.unwrap();
+ let main_path = main.tag_path("release");
+ let original_main =
file_io.new_input(&main_path).unwrap().read().await.unwrap();
+ branch
+ .replace("release", &test_snapshot(20), Some("2d"))
+ .await
+ .unwrap();
+ assert_eq!(
+ file_io.new_input(&main_path).unwrap().read().await.unwrap(),
+ original_main
+ );
+ assert_eq!(
+ branch.get("release").await.unwrap(),
+ Some(test_snapshot(20))
+ );
+ let branch_path = branch.tag_path("release");
+ let original_branch = file_io
+ .new_input(&branch_path)
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ main.replace("release", &test_snapshot(2), None)
+ .await
+ .unwrap();
+ assert_eq!(
+ file_io
+ .new_input(&branch_path)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ original_branch
+ );
+ assert_eq!(main.get("release").await.unwrap(), Some(test_snapshot(2)));
+ main.create("main_only", &test_snapshot(2)).await.unwrap();
+ assert!(matches!(
+ branch.replace("main_only", &test_snapshot(20), None).await,
+ Err(crate::Error::TagNotExist { .. })
+ ));
+ assert!(!branch.tag_exists("main_only").await.unwrap());
+ }
+
#[tokio::test]
async fn test_rename_tag_preserves_metadata() {
let tm = TagManager::new(test_file_io(),
"memory:/test_tag_rename".to_string());
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 82553172..0e1ebea7 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1180,6 +1180,47 @@ the tag file. For example, `'1d'` appears as `PT24H` in
the `time_retained` colu
of `$tags`. Omitting `time_retained` creates a tag without these metadata
fields.
This procedure does not automatically delete expired tags.
+### replace_tag
+
+Replace an existing manual tag's snapshot and retention without changing its
name:
+
+```sql
+CALL sys.replace_tag(
+ table => 'paimon.my_db.my_table',
+ tag => 'release',
+ snapshot_id => 1,
+ time_retained => '2d'
+);
+```
+
+`table` and `tag` are required. The tag must already exist. `snapshot_id` is an
+optional positive BIGINT; omitting it selects the latest live snapshot when the
+operation resolves its target. An explicit ID first resolves a live snapshot,
+then falls back to snapshots preserved by tags in the same branch if the live
+snapshot does not exist. Other snapshot read errors are returned. No suitable
+snapshot causes an error without changing the tag. `snapshot` is not an alias
+for `snapshot_id`. Success returns one row with `result = 'OK'`.
+
+`time_retained` uses the same duration format as `create_tag`. Providing it
+replaces the TTL and resets `tagCreateTime` to the replacement time, even when
+the snapshot ID stays the same. **Omitting it removes the previous TTL and
+creation time**: the tag is written as plain Snapshot JSON. It does not inherit
+the old TTL or apply a table's default tag retention. Argument, name,
retention,
+and snapshot resolution errors leave the original tag file unchanged.
+
+Only non-REST FileStore tables and ordinary manual tags are supported. Format
+tables and engine-served table types are rejected. This procedure does not add
+branch-qualified SQL name resolution; an already resolved branch Table uses its
+own snapshot and tag managers. Automatic tag management and external tag
+callbacks are outside its scope.
+
+Replacement overwrites the tag metadata directly. It does not create a snapshot
+or clean up old snapshots, manifests, or data files, and it does not run tag
+expiration. Storage write failures are returned, but atomic replacement and
+rollback are not guaranteed across storage backends. The existence check and
+write are separate operations; concurrent tag modifications are not protected
+by a conditional update or transaction.
+
### create_tag_from_timestamp
Create a named tag from a commit timestamp (finds the earliest retained
snapshot at or after the given time):
