diff --git a/crates/integrations/datafusion/src/procedures.rs b/crates/integrations/datafusion/src/procedures.rs
index ce150229d..0d2173951 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"],
         "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,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 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<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;
@@ -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/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::<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_argument_validation_follows_java() {
     let (_tmp, server, context) = setup().await;
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 120b8a333..01c6f62f2 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: