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 9133b996 feat(datafusion): support create_tag_from_watermark (#1134)
9133b996 is described below
commit 9133b996d7ff1b4514a31e04f1f6b3110ab144b3
Author: QuakeWang <[email protected]>
AuthorDate: Sun Oct 18 16:00:55 2026 +0800
feat(datafusion): support create_tag_from_watermark (#1134)
---
crates/integrations/datafusion/src/procedures.rs | 74 ++++++
crates/integrations/datafusion/tests/procedures.rs | 277 +++++++++++++++++++++
.../datafusion/tests/rest_management_procedures.rs | 145 +++++++++++
docs/src/sql.md | 35 ++-
4 files changed, 530 insertions(+), 1 deletion(-)
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index 3124a863..9377e2e8 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -24,6 +24,7 @@
//! - `CALL sys.rollback_to(table => '...', snapshot_id => ... | tag => '...')`
//! - `CALL sys.rollback_to_timestamp(table => '...', timestamp => ...)`
//! - `CALL sys.create_tag_from_timestamp(table => '...', tag => '...',
timestamp => ...)`
+//! - `CALL sys.create_tag_from_watermark(table => '...', tag => '...',
watermark => ..., time_retained => '...')`
//! - `CALL sys.create_global_index(table => '...', index_column => '...',
index_type => 'btree')`
//! - `CALL sys.create_global_index(table => '...', index_column => '...',
index_type => 'bitmap')`
//! - `CALL sys.create_global_index(table => '...', index_column => '...',
index_type => 'ivf-pq')`
@@ -134,6 +135,35 @@ async fn later_or_equal_from_all(
}
}
+/// Find a snapshot with watermark >= the requested value, including
tag-retained
+/// snapshots. Matches Java's Spark procedure: compare the first eligible live
and
+/// tagged snapshots by watermark, then by snapshot ID when watermarks are
equal.
+async fn later_or_equal_watermark_from_all(
+ sm: &SnapshotManager,
+ tm: &TagManager,
+ watermark: i64,
+) -> DFResult<Option<Snapshot>> {
+ let live = sm
+ .later_or_equal_watermark(watermark)
+ .await
+ .map_err(to_datafusion_error)?;
+ let tags = tm.list_all().await.map_err(to_datafusion_error)?;
+ let tag_candidate = tags
+ .into_iter()
+ .map(|(_, snapshot)| snapshot)
+ .filter(|snapshot| {
+ snapshot
+ .watermark()
+ .is_some_and(|value| value != i64::MIN && value >= watermark)
+ })
+ // Rust lists tags by name; Java's tag candidates are ordered by
snapshot ID.
+ .min_by_key(Snapshot::id);
+ Ok(live
+ .into_iter()
+ .chain(tag_candidate)
+ .min_by_key(|snapshot| (snapshot.watermark(), snapshot.id())))
+}
+
/// Find the latest snapshot with commit time <= timestamp_millis,
/// considering both live snapshots and tag-retained snapshots.
async fn earlier_or_equal_from_all(
@@ -177,6 +207,7 @@ fn declared_parameters(proc_name: &str) -> Option<&'static
[&'static str]> {
"rollback_to" => &["table", "snapshot_id", "tag"],
"rollback_to_timestamp" => &["table", "timestamp"],
"create_tag_from_timestamp" => &["table", "tag", "timestamp"],
+ "create_tag_from_watermark" => &["table", "tag", "watermark",
"time_retained"],
"rename_branch" => &["table", "from_branch", "to_branch"],
"delete_branch" => &["table", "branch"],
"create_global_index" => &["table", "index_column", "index_type",
"options"],
@@ -291,6 +322,9 @@ pub async fn execute_call(
"create_tag_from_timestamp" => {
proc_create_tag_from_timestamp(ctx, catalog, catalog_name,
&args).await
}
+ "create_tag_from_watermark" => {
+ proc_create_tag_from_watermark(ctx, catalog, catalog_name,
&args).await
+ }
"rename_branch" => proc_rename_branch(ctx, catalog, catalog_name,
&args).await,
"delete_branch" => proc_delete_branch(ctx, catalog, catalog_name,
&args).await,
"create_global_index" => proc_create_global_index(ctx, catalog,
catalog_name, &args).await,
@@ -778,6 +812,46 @@ async fn proc_create_tag_from_timestamp(
ok_result(ctx)
}
+async fn proc_create_tag_from_watermark(
+ 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 watermark_str = require_arg(args, "watermark")?;
+ let watermark: i64 = watermark_str
+ .parse()
+ .map_err(|_| DataFusionError::Plan(format!("Invalid watermark:
'{watermark_str}'")))?;
+ let time_retained = args.get("time_retained").map(String::as_str);
+
+ // REST snapshots may exist only in the catalog; storage listings cannot
+ // enumerate the authoritative history needed for watermark selection.
+ if table.rest_env().is_some() {
+ return Err(DataFusionError::NotImplemented(
+ "create_tag_from_watermark is not supported for tables managed by
a REST catalog"
+ .to_string(),
+ ));
+ }
+
+ let (sm, tm) = managers(&table);
+ if tm.tag_exists(tag_name).await.map_err(to_datafusion_error)? {
+ return Err(DataFusionError::Plan(format!(
+ "Tag '{tag_name}' already exists"
+ )));
+ }
+ let snapshot = later_or_equal_watermark_from_all(&sm, &tm, watermark)
+ .await?
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!("No snapshot found with watermark >=
{watermark}"))
+ })?;
+ tm.create_with_retention(tag_name, &snapshot, time_retained)
+ .await
+ .map_err(to_datafusion_error)?;
+ ok_result(ctx)
+}
+
async fn proc_create_lumina_index(
ctx: &SessionContext,
catalog: &Arc<dyn Catalog>,
diff --git a/crates/integrations/datafusion/tests/procedures.rs
b/crates/integrations/datafusion/tests/procedures.rs
index f4463685..5279155a 100644
--- a/crates/integrations/datafusion/tests/procedures.rs
+++ b/crates/integrations/datafusion/tests/procedures.rs
@@ -56,6 +56,27 @@ async fn setup_table_with_snapshots() -> (tempfile::TempDir,
paimon_datafusion::
(tmp, sql_context)
}
+async fn setup_table_with_watermarks(
+ watermarks: [Option<i64>; 3],
+) -> (tempfile::TempDir, paimon_datafusion::SQLContext) {
+ let (tmp, sql_context) = setup_table_with_snapshots().await;
+ for (index, watermark) in watermarks.into_iter().enumerate() {
+ let path = tmp
+ .path()
+ .join(format!("test_db.db/t1/snapshot/snapshot-{}", index + 1));
+ let mut snapshot: serde_json::Value =
+ serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
+ snapshot["watermark"] = serde_json::json!(watermark);
+ std::fs::write(path, serde_json::to_vec(&snapshot).unwrap()).unwrap();
+ }
+ (tmp, sql_context)
+}
+
+fn read_tag(tmp: &Path, name: &str) -> serde_json::Value {
+ let path = tmp.join(format!("test_db.db/t1/tag/tag-{name}"));
+ serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap()
+}
+
#[tokio::test]
async fn test_create_tag() {
let (_tmp, sql_context) = setup_table_with_snapshots().await;
@@ -2759,6 +2780,262 @@ async fn test_create_tag_from_timestamp() {
assert_eq!(count, 1);
}
+#[tokio::test]
+async fn test_create_tag_from_watermark_selects_first_live_snapshot() {
+ for (watermark, expected_id) in [(500, 1), (1000, 1), (1500, 2), (2000,
2)] {
+ let (tmp, sql_context) =
+ setup_table_with_watermarks([Some(1000), Some(2000),
Some(2000)]).await;
+ let tag_name = format!("wm_{watermark}");
+ exec(
+ &sql_context,
+ &format!(
+ "CALL sys.create_tag_from_watermark(table =>
'paimon.test_db.t1', \
+ tag => '{tag_name}', watermark => {watermark})"
+ ),
+ )
+ .await;
+ let tag = read_tag(tmp.path(), &tag_name);
+ assert_eq!(tag["id"], expected_id);
+ assert!(tag.get("tagTimeRetained").is_none());
+ assert!(tag.get("tagCreateTime").is_none());
+ assert_eq!(
+ row_count(
+ &sql_context,
+ &format!("SELECT * FROM paimon.test_db.t1 VERSION AS OF
'{tag_name}'"),
+ )
+ .await,
+ expected_id as usize
+ );
+ }
+}
+
+#[tokio::test]
+async fn test_create_tag_from_watermark_uses_retained_snapshots() {
+ let (tmp, sql_context) =
+ setup_table_with_watermarks([Some(1000), Some(2000),
Some(3000)]).await;
+ let snapshot_dir = tmp.path().join("test_db.db/t1/snapshot");
+ // Tag names deliberately put the later snapshot first in Rust's tag
listing.
+ for (name, id) in [("z_earlier", 1), ("a_later", 2)] {
+ exec(
+ &sql_context,
+ &format!(
+ "CALL sys.create_tag(table => 'test_db.t1', tag => '{name}',
snapshot_id => {id})"
+ ),
+ )
+ .await;
+
std::fs::remove_file(snapshot_dir.join(format!("snapshot-{id}"))).unwrap();
+ }
+
+ for (watermark, expected_id) in [(999, 1), (1000, 1), (1500, 2), (2000,
2), (2500, 3)] {
+ let tag_name = format!("wm_{watermark}");
+ exec(
+ &sql_context,
+ &format!(
+ "CALL sys.create_tag_from_watermark(table => 'test_db.t1', \
+ tag => '{tag_name}', watermark => {watermark}, time_retained
=> '7d')"
+ ),
+ )
+ .await;
+ let tag = read_tag(tmp.path(), &tag_name);
+ assert_eq!(tag["id"], expected_id);
+ assert_eq!(
+ tag["tagTimeRetained"].as_number().unwrap().to_string(),
+ "604800.000000000"
+ );
+ assert_eq!(tag["tagCreateTime"].as_array().unwrap().len(), 7);
+ assert_eq!(
+ row_count(
+ &sql_context,
+ &format!("SELECT * FROM paimon.test_db.t1 VERSION AS OF
'{tag_name}'"),
+ )
+ .await,
+ expected_id as usize
+ );
+ }
+
+ // All snapshot files can be gone while tag metadata still supplies a
candidate.
+ std::fs::remove_file(snapshot_dir.join("snapshot-3")).unwrap();
+ exec(
+ &sql_context,
+ "CALL sys.create_tag_from_watermark(table => 'test_db.t1', tag =>
'tags_only', watermark => 1500)",
+ )
+ .await;
+ assert_eq!(read_tag(tmp.path(), "tags_only")["id"], 2);
+}
+
+#[tokio::test]
+async fn
test_create_tag_from_watermark_equal_watermarks_prefer_earlier_snapshot() {
+ for (retained_id, expire_first) in [(1, true), (2, false)] {
+ let (tmp, sql_context) =
+ setup_table_with_watermarks([Some(1000), Some(1000),
Some(2000)]).await;
+ exec(
+ &sql_context,
+ &format!(
+ "CALL sys.create_tag(table => 'test_db.t1', tag => 'keep',
snapshot_id => {retained_id})"
+ ),
+ )
+ .await;
+ if expire_first {
+
std::fs::remove_file(tmp.path().join("test_db.db/t1/snapshot/snapshot-1")).unwrap();
+ }
+ exec(
+ &sql_context,
+ "CALL sys.create_tag_from_watermark(table => 'test_db.t1', tag =>
'equal', watermark => 1000)",
+ )
+ .await;
+ assert_eq!(read_tag(tmp.path(), "equal")["id"], 1);
+ }
+}
+
+#[tokio::test]
+async fn
test_create_tag_from_watermark_skips_missing_live_and_tag_watermarks() {
+ let (tmp, sql_context) = setup_table_with_watermarks([None,
Some(i64::MIN), Some(2000)]).await;
+ for id in [1, 2] {
+ exec(
+ &sql_context,
+ &format!(
+ "CALL sys.create_tag(table => 'test_db.t1', tag =>
'missing_{id}', snapshot_id => {id})"
+ ),
+ )
+ .await;
+ }
+ exec(
+ &sql_context,
+ "CALL sys.create_tag_from_watermark(table => 'test_db.t1', tag =>
'live', watermark => 0)",
+ )
+ .await;
+ assert_eq!(read_tag(tmp.path(), "live")["id"], 3);
+ for id in [1, 2] {
+ std::fs::remove_file(
+ tmp.path()
+ .join(format!("test_db.db/t1/snapshot/snapshot-{id}")),
+ )
+ .unwrap();
+ }
+ for (index, watermark) in [i64::MIN, 0, 2000].into_iter().enumerate() {
+ let name = format!("valid_{index}");
+ exec(
+ &sql_context,
+ &format!(
+ "CALL sys.create_tag_from_watermark(table => 'test_db.t1', tag
=> '{name}', watermark => {watermark})"
+ ),
+ )
+ .await;
+ assert_eq!(read_tag(tmp.path(), &name)["id"], 3);
+ }
+}
+
+#[tokio::test]
+async fn test_create_tag_from_watermark_without_candidate_does_not_write() {
+ for (watermarks, target) in [
+ ([None, None, None], 0),
+ ([None, Some(i64::MIN), None], i64::MIN),
+ ([Some(1000), Some(2000), Some(3000)], 3001),
+ ] {
+ let (tmp, sql_context) = setup_table_with_watermarks(watermarks).await;
+ exec(
+ &sql_context,
+ "CALL sys.create_tag(table => 'test_db.t1', tag => 'keep',
snapshot_id => 2)",
+ )
+ .await;
+ let keep_path = tmp.path().join("test_db.db/t1/tag/tag-keep");
+ let original = std::fs::read(&keep_path).unwrap();
+ assert_sql_error(
+ &sql_context,
+ &format!(
+ "CALL sys.create_tag_from_watermark(table => 'test_db.t1', tag
=> 'missing', watermark => {target})"
+ ),
+ &format!("No snapshot found with watermark >= {target}"),
+ )
+ .await;
+ assert!(!tmp.path().join("test_db.db/t1/tag/tag-missing").exists());
+ assert_eq!(std::fs::read(keep_path).unwrap(), original);
+ assert_eq!(
+ row_count(&sql_context, "SELECT * FROM
paimon.test_db.`t1$tags`").await,
+ 1
+ );
+ }
+
+ let (tmp, sql_context) = setup_sql_context().await;
+ exec(&sql_context, "CREATE TABLE paimon.test_db.t1 (id INT)").await;
+ assert_sql_error(
+ &sql_context,
+ "CALL sys.create_tag_from_watermark(table => 'test_db.t1', tag =>
'empty', watermark => 0)",
+ "No snapshot found with watermark >= 0",
+ )
+ .await;
+ assert!(!tmp.path().join("test_db.db/t1/tag").exists());
+}
+
+#[tokio::test]
+async fn
test_create_tag_from_watermark_rejects_invalid_arguments_without_writes() {
+ let (tmp, sql_context) =
+ setup_table_with_watermarks([Some(1000), Some(2000),
Some(3000)]).await;
+ exec(
+ &sql_context,
+ "CALL sys.create_tag_from_watermark(table => 'test_db.t1', tag =>
'keep', watermark => 1000)",
+ )
+ .await;
+ let keep_path = tmp.path().join("test_db.db/t1/tag/tag-keep");
+ let original = std::fs::read(&keep_path).unwrap();
+
+ for (args, expected_error) in [
+ ("tag => 'invalid', watermark => 1000", "Missing required argument:
'table'"),
+ ("table => 'test_db.t1', watermark => 1000", "Missing required
argument: 'tag'"),
+ ("table => 'test_db.t1', tag => 'invalid'", "Missing required
argument: 'watermark'"),
+ ("table => 'test_db.t1', tag => 'invalid', watermark => 'abc'",
"Invalid watermark"),
+ ("table => 'test_db.t1', tag => 'invalid', watermark => 1.5", "Invalid
watermark"),
+ ("table => 'test_db.t1', tag => 'invalid', watermark =>
9223372036854775808", "Invalid watermark"),
+ ("table => 'test_db.t1', tag => 'invalid', watermark =>
-9223372036854775809", "Invalid watermark"),
+ ("table => 'test_db.t1', tag => 'invalid', watermak => 1000",
"Argument watermak is unknown"),
+ ("table => 'test_db.t1', tag => 'invalid', watermark => 1000,
time_retain => '1d'", "Argument time_retain is unknown"),
+ ("table => 'test_db.t1', tag => 'invalid', watermark => 1000,
watermark => 2000", "Procedure argument watermark is duplicated"),
+ ("table => 'test_db.t1', tag => '', watermark => 1000", "Invalid tag
name"),
+ ("table => 'test_db.t1', tag => '../invalid', watermark => 1000",
"Invalid tag name"),
+ ("table => 'test_db.t1', tag => 'invalid', watermark => 1000,
time_retained => ''", "Invalid tag retention"),
+ ("table => 'test_db.t1', tag => 'invalid', watermark => 1000,
time_retained => 'invalid'", "Invalid tag retention"),
+ ("table => 'test_db.t1', tag => 'invalid', watermark => 1000,
time_retained => '-1d'", "Invalid tag retention"),
+ ] {
+ assert_sql_error(
+ &sql_context,
+ &format!("CALL sys.create_tag_from_watermark({args})"),
+ expected_error,
+ )
+ .await;
+ assert_eq!(std::fs::read(&keep_path).unwrap(), original);
+ }
+ assert!(!tmp.path().join("test_db.db/t1/tag/tag-invalid").exists());
+ assert_eq!(
+ row_count(&sql_context, "SELECT * FROM
paimon.test_db.`t1$tags`").await,
+ 1
+ );
+}
+
+#[tokio::test]
+async fn test_create_tag_from_watermark_preserves_existing_tag() {
+ let (tmp, sql_context) =
+ setup_table_with_watermarks([Some(1000), Some(2000),
Some(3000)]).await;
+ exec(
+ &sql_context,
+ "CALL sys.create_tag_from_watermark(table => 'test_db.t1', tag =>
'keep', watermark => 1000, time_retained => '1d')",
+ )
+ .await;
+ let keep_path = tmp.path().join("test_db.db/t1/tag/tag-keep");
+ let original = std::fs::read(&keep_path).unwrap();
+ for (watermark, retention) in [(2000, "2d"), (2000, "invalid"), (4000,
"1d")] {
+ assert_sql_error(
+ &sql_context,
+ &format!(
+ "CALL sys.create_tag_from_watermark(table => 'test_db.t1', tag
=> 'keep', \
+ watermark => {watermark}, time_retained => '{retention}')"
+ ),
+ "Tag 'keep' already exists",
+ )
+ .await;
+ assert_eq!(std::fs::read(&keep_path).unwrap(), original);
+ }
+}
+
#[tokio::test]
async fn test_rollback_cleans_newer_tags() {
let (_tmp, sql_context) = setup_table_with_snapshots().await;
diff --git a/crates/integrations/datafusion/tests/rest_management_procedures.rs
b/crates/integrations/datafusion/tests/rest_management_procedures.rs
index d38efa9d..039c2bfc 100644
--- a/crates/integrations/datafusion/tests/rest_management_procedures.rs
+++ b/crates/integrations/datafusion/tests/rest_management_procedures.rs
@@ -680,6 +680,151 @@ async fn test_rename_branch_rejected_on_rest_catalog() {
.await;
}
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_create_tag_from_watermark_rejects_readable_rest_catalog_table() {
+ use arrow_array::Int32Array;
+ use axum::http::{Method, StatusCode, Uri};
+ use axum::{Json, Router};
+ use paimon::catalog::Identifier;
+ use paimon::Catalog;
+ use std::sync::Mutex;
+
+ let (tmp, catalog) = common::create_test_env();
+ let writer = common::create_sql_context(catalog.clone()).await;
+ common::exec(&writer, "CREATE SCHEMA paimon.test_db").await;
+ common::exec(&writer, "CREATE TABLE paimon.test_db.t1 (id INT)").await;
+ let table_dir = tmp.path().join("test_db.db/t1");
+ for id in [1, 2] {
+ common::exec(
+ &writer,
+ &format!("INSERT INTO paimon.test_db.t1 VALUES ({id})"),
+ )
+ .await;
+ let path = table_dir.join(format!("snapshot/snapshot-{id}"));
+ let mut snapshot: serde_json::Value =
+ serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
+ snapshot["watermark"] = serde_json::json!(id * 1000);
+ std::fs::write(path, serde_json::to_vec(&snapshot).unwrap()).unwrap();
+ }
+ common::exec(
+ &writer,
+ "CALL sys.create_tag(table => 'test_db.t1', tag => 'keep', snapshot_id
=> 1, time_retained => '1d')",
+ )
+ .await;
+ let tag_path = table_dir.join("tag/tag-keep");
+ let original_tag = std::fs::read(&tag_path).unwrap();
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "t1"))
+ .await
+ .unwrap();
+ assert_eq!(table.schema().core_options().file_format(), "parquet");
+ let snapshot = table.snapshot_manager().get_snapshot(2).await.unwrap();
+ let snapshot_response = serde_json::json!({"snapshot": {"snapshot":
snapshot}});
+ let table_response = serde_json::json!({
+ "id": "table-uuid", "database": "test_db", "name": "t1",
+ "path": table.location(), "isExternal": false,
+ "schemaId": table.schema().id(), "schema": table.schema(),
+ });
+ // Only the catalog serves snapshot 2; the physical snapshot directory is
stale.
+ std::fs::remove_file(table_dir.join("snapshot/snapshot-2")).unwrap();
+ assert_eq!(table.snapshot_manager().list_all_ids().await.unwrap(), [1]);
+
+ let requests = Arc::new(Mutex::new(Vec::new()));
+ let handler_requests = requests.clone();
+ let table_path = "/v1/test/databases/test_db/tables/t1";
+ let snapshot_path = format!("{table_path}/snapshot");
+ let app = Router::new().fallback(move |method: Method, uri: Uri| {
+ handler_requests
+ .lock()
+ .unwrap()
+ .push((method, uri.path().to_string()));
+ let (status, body) = match uri.path() {
+ "/v1/config" => (
+ StatusCode::OK,
+ serde_json::json!({"defaults": {"prefix": "test"}}),
+ ),
+ "/v1/test/databases/test_db" => (
+ StatusCode::OK,
+ serde_json::json!({"name": "test_db", "options": {}}),
+ ),
+ path if path == table_path => (StatusCode::OK,
table_response.clone()),
+ path if path == snapshot_path => (StatusCode::OK,
snapshot_response.clone()),
+ _ => (
+ StatusCode::NOT_FOUND,
+ serde_json::json!({"code": 404, "message": "Unexpected
request"}),
+ ),
+ };
+ async move { (status, Json(body)) }
+ });
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let mut options = Options::new();
+ options.set(
+ CatalogOptions::URI,
+ format!("http://{}";, listener.local_addr().unwrap()),
+ );
+ options.set(CatalogOptions::WAREHOUSE, "test_warehouse");
+ options.set(CatalogOptions::TOKEN_PROVIDER, "bear");
+ options.set(CatalogOptions::TOKEN, "test-token");
+ let server = tokio::spawn(async move { axum::serve(listener,
app).await.unwrap() });
+ let catalog = Arc::new(RESTCatalog::new(options, true).await.unwrap());
+ let mut context = SQLContext::new();
+ context
+ .register_catalog_with_default_db("paimon", catalog, None)
+ .await
+ .unwrap();
+
+ let batches = context
+ .sql("SELECT id FROM paimon.test_db.t1 ORDER BY id")
+ .await
+ .unwrap_or_else(|error| panic!("{error}; requests: {:?}",
requests.lock().unwrap()))
+ .collect()
+ .await
+ .unwrap();
+ let ids: Vec<_> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect();
+ assert_eq!(ids, [1, 2]);
+ assert!(requests
+ .lock()
+ .unwrap()
+ .iter()
+ .any(|(_, path)| path.ends_with("/snapshot")));
+ requests.lock().unwrap().clear();
+
+ for (watermark, tag) in [(2000, "checkpoint"), (1000, "older"), (1000,
"keep")] {
+ let error = context.sql(&format!(
+ "CALL sys.create_tag_from_watermark(table => 'test_db.t1', tag =>
'{tag}', watermark => {watermark})"
+ )).await.unwrap_err();
+ assert!(
+ matches!(error,
datafusion::error::DataFusionError::NotImplemented(_)),
+ "{error}"
+ );
+ assert!(
+ error.to_string().contains(
+ "create_tag_from_watermark is not supported for tables managed
by a REST catalog"
+ ),
+ "{error}"
+ );
+ }
+ assert_eq!(std::fs::read(tag_path).unwrap(), original_tag);
+ assert_eq!(std::fs::read_dir(table_dir.join("tag")).unwrap().count(), 1);
+ let requests = requests.lock().unwrap();
+ assert!(!requests.is_empty());
+ assert!(requests
+ .iter()
+ .all(|(method, path)| method == Method::GET && path == table_path));
+ server.abort();
+}
+
#[tokio::test]
async fn test_delete_branch_rejected_on_rest_catalog() {
// A REST catalog owns branch metadata and the Rust REST catalog has no
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 120b8a33..01c6f62f 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1080,12 +1080,45 @@ This procedure does not automatically delete expired
tags.
### create_tag_from_timestamp
-Create a named tag from a timestamp (finds the latest snapshot at or before
the given time):
+Create a named tag from a commit timestamp (finds the earliest retained
snapshot at or after the given time):
```sql
CALL sys.create_tag_from_timestamp(table => 'paimon.my_db.my_table', tag =>
'my_tag', timestamp => 1234567890000);
```
+### create_tag_from_watermark
+
+This procedure is not supported for tables managed by a REST catalog. Their
+snapshots may exist only in the catalog, while watermark selection currently
+relies on stored snapshot files. REST-managed tables are rejected before
+candidate selection or tag creation.
+
+Create a named tag from the earliest retained snapshot whose watermark is
greater
+than or equal to the requested value. Candidates include live snapshots and
+historical snapshots preserved by existing tags. Equal watermarks select the
+earlier snapshot ID; snapshots without a watermark, including the no-watermark
+sentinel `-9223372036854775808`, are skipped.
+
+```sql
+CALL sys.create_tag_from_watermark(
+ table => 'paimon.my_db.my_table',
+ tag => 'business_checkpoint',
+ watermark => 1704067200000,
+ time_retained => '7d'
+);
+```
+
+`table`, `tag`, and `watermark` are required. `watermark` is a signed 64-bit
integer
+in milliseconds. It represents the input processing progress recorded by the
+writer, whereas `timestamp` in `create_tag_from_timestamp` refers to snapshot
+commit time. Selecting by watermark fixes an entire snapshot; it does not
filter
+rows by event time.
+
+The optional `time_retained` uses the same duration format and metadata as
+`create_tag`. This procedure does not automatically delete expired tags. An
+existing tag name, an invalid argument, or no matching snapshot causes an error
+without creating or replacing a tag. The call does not wait for a future
snapshot.
+
### delete_tag
Delete a named tag:
