From 517599708d06a523e702e8d0b487af04a7e6a6f2 Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Thu, 15 Oct 2026 20:29:26 +0800 Subject: [PATCH 1/2] feat(datafusion): support create_tag_from_watermark Add a SQL entry point for watermark-based tags, including snapshots retained by existing tags and optional retention. Preserve existing tags on errors. Cover candidate selection and validation through SQL integration tests, and document watermark semantics. Signed-off-by: QuakeWang --- .../integrations/datafusion/src/procedures.rs | 65 ++++ .../datafusion/tests/procedures.rs | 277 ++++++++++++++++++ docs/src/sql.md | 30 +- 3 files changed, 371 insertions(+), 1 deletion(-) diff --git a/crates/integrations/datafusion/src/procedures.rs b/crates/integrations/datafusion/src/procedures.rs index ce150229d..3b9eb5c30 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> { + 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"], "create_global_index" => &["table", "index_column", "index_type", "options"], "drop_global_index" => &[ @@ -290,6 +321,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, "create_global_index" => proc_create_global_index(ctx, catalog, catalog_name, &args).await, "drop_global_index" => proc_drop_global_index(ctx, catalog, catalog_name, &args).await, @@ -709,6 +743,37 @@ async fn proc_create_tag_from_timestamp( ok_result(ctx) } +async fn proc_create_tag_from_watermark( + ctx: &SessionContext, + catalog: &Arc, + catalog_name: &str, + args: &HashMap, +) -> DFResult { + 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); + + 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, diff --git a/crates/integrations/datafusion/tests/procedures.rs b/crates/integrations/datafusion/tests/procedures.rs index 21bfc73e7..0b29b43a2 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; 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; @@ -2476,6 +2497,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/docs/src/sql.md b/docs/src/sql.md index 120b8a333..279ef4204 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -1080,12 +1080,40 @@ 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 + +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: From a9c3a5e8e230d5dfa027d88c8808a8e758870153 Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Fri, 16 Oct 2026 13:08:11 +0800 Subject: [PATCH 2/2] fix(datafusion): reject watermark tags on REST tables Reject REST-managed tables before filesystem candidate selection because their authoritative snapshots may exist only in the catalog. Add a SQL/HTTP regression with real Parquet data and a catalog-only latest snapshot, and document this limitation. Signed-off-by: QuakeWang --- .../integrations/datafusion/src/procedures.rs | 9 ++ .../tests/rest_management_procedures.rs | 145 ++++++++++++++++++ docs/src/sql.md | 5 + 3 files changed, 159 insertions(+) diff --git a/crates/integrations/datafusion/src/procedures.rs b/crates/integrations/datafusion/src/procedures.rs index 3b9eb5c30..0d2173951 100644 --- a/crates/integrations/datafusion/src/procedures.rs +++ b/crates/integrations/datafusion/src/procedures.rs @@ -757,6 +757,15 @@ async fn proc_create_tag_from_watermark( .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!( diff --git a/crates/integrations/datafusion/tests/rest_management_procedures.rs b/crates/integrations/datafusion/tests/rest_management_procedures.rs index bc6a6b85e..df6a64e8a 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::() + .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_argument_validation_follows_java() { let (_tmp, server, context) = setup().await; diff --git a/docs/src/sql.md b/docs/src/sql.md index 279ef4204..01c6f62f2 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -1088,6 +1088,11 @@ CALL sys.create_tag_from_timestamp(table => 'paimon.my_db.my_table', tag => 'my_ ### 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