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 88091664 feat(datafusion): support reset_consumer (#1167)
88091664 is described below
commit 88091664a02f1ed22cb5cae71cf15f53979c1492
Author: QuakeWang <[email protected]>
AuthorDate: Mon Oct 19 20:15:46 2026 +0800
feat(datafusion): support reset_consumer (#1167)
---
crates/integrations/datafusion/src/procedures.rs | 61 +++-
.../datafusion/tests/reset_consumer.rs | 385 +++++++++++++++++++++
.../datafusion/tests/rest_management_procedures.rs | 26 ++
crates/paimon/src/table/consumer_manager.rs | 165 ++++++++-
docs/src/sql.md | 41 +++
5 files changed, 673 insertions(+), 5 deletions(-)
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index 9377e2e8..6db19eda 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -21,6 +21,7 @@
//! - `CALL sys.create_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 => ...])`
//! - `CALL sys.rollback_to(table => '...', snapshot_id => ... | tag => '...')`
//! - `CALL sys.rollback_to_timestamp(table => '...', timestamp => ...)`
//! - `CALL sys.create_tag_from_timestamp(table => '...', tag => '...',
timestamp => ...)`
@@ -74,8 +75,8 @@ use paimon::catalog::{Catalog, Identifier, RESTCatalog};
use paimon::lumina::{is_lumina_index_type, LUMINA_IDENTIFIER};
use paimon::spec::{CoreOptions, DataType, Datum, Predicate, PredicateBuilder,
Snapshot};
use paimon::table::{
- normalize_global_index_type, BranchManager, SnapshotManager, Table,
TagManager,
- SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP,
+ normalize_global_index_type, BranchManager, ConsumerManager,
SnapshotManager, Table,
+ TagManager, SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP,
};
use crate::error::to_datafusion_error;
@@ -204,6 +205,7 @@ fn declared_parameters(proc_name: &str) -> Option<&'static
[&'static str]> {
"create_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"],
"rollback_to" => &["table", "snapshot_id", "tag"],
"rollback_to_timestamp" => &["table", "timestamp"],
"create_tag_from_timestamp" => &["table", "tag", "timestamp"],
@@ -315,6 +317,7 @@ pub async fn execute_call(
"create_tag" => proc_create_tag(ctx, catalog, catalog_name,
&args).await,
"delete_tag" => proc_delete_tag(ctx, catalog, catalog_name,
&args).await,
"rename_tag" => proc_rename_tag(ctx, catalog, catalog_name,
&args).await,
+ "reset_consumer" => proc_reset_consumer(ctx, catalog, catalog_name,
&args).await,
"rollback_to" => proc_rollback_to(ctx, catalog, catalog_name,
&args).await,
"rollback_to_timestamp" => {
proc_rollback_to_timestamp(ctx, catalog, catalog_name, &args).await
@@ -461,6 +464,60 @@ fn managers(table: &Table) -> (SnapshotManager,
TagManager) {
(sm, tm)
}
+async fn proc_reset_consumer(
+ ctx: &SessionContext,
+ catalog: &Arc<dyn Catalog>,
+ catalog_name: &str,
+ args: &HashMap<String, String>,
+) -> DFResult<DataFrame> {
+ let consumer_id = require_arg(args, "consumer_id")?;
+
ConsumerManager::validate_consumer_id(consumer_id).map_err(to_datafusion_error)?;
+ let next_snapshot_id = args
+ .get("next_snapshot_id")
+ .map(|value| {
+ value
+ .parse::<i64>()
+ .ok()
+ .filter(|id| *id > 0)
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "Invalid next_snapshot_id: '{value}'. Expected a
positive BIGINT"
+ ))
+ })
+ })
+ .transpose()?;
+ let table = get_table(catalog, catalog_name, args).await?;
+ if table.rest_env().is_some() {
+ return Err(DataFusionError::NotImplemented(
+ "reset_consumer is not supported for tables managed by a REST
catalog".to_string(),
+ ));
+ }
+ if table.schema().core_options().is_format_table() {
+ return Err(DataFusionError::NotImplemented(
+ "reset_consumer requires a FileStoreTable; format-table is not
supported".to_string(),
+ ));
+ }
+ let manager = table.consumer_manager();
+ if let Some(id) = next_snapshot_id {
+ // Stream recovery reads live snapshot files, not tag-retained
metadata.
+ table
+ .snapshot_manager()
+ .get_snapshot(id)
+ .await
+ .map_err(to_datafusion_error)?;
+ manager
+ .reset(consumer_id, id)
+ .await
+ .map_err(to_datafusion_error)?;
+ } else {
+ manager
+ .delete(consumer_id)
+ .await
+ .map_err(to_datafusion_error)?;
+ }
+ ok_result(ctx)
+}
+
async fn proc_create_tag(
ctx: &SessionContext,
catalog: &Arc<dyn Catalog>,
diff --git a/crates/integrations/datafusion/tests/reset_consumer.rs
b/crates/integrations/datafusion/tests/reset_consumer.rs
new file mode 100644
index 00000000..6323ecce
--- /dev/null
+++ b/crates/integrations/datafusion/tests/reset_consumer.rs
@@ -0,0 +1,385 @@
+// 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;
+
+use std::collections::BTreeMap;
+use std::path::{Path, PathBuf};
+
+use common::{assert_sql_error, create_sql_context, create_test_env, exec,
string_value};
+use futures::TryStreamExt;
+use paimon::catalog::Identifier;
+use paimon::spec::{DataType, IntType, Schema};
+use paimon::table::{Plan, Table};
+use paimon::Catalog;
+use paimon_datafusion::SQLContext;
+
+async fn table_with_snapshots() -> (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)").await;
+ for id in 1..=3 {
+ exec(
+ &context,
+ &format!("INSERT INTO paimon.test_db.events VALUES ({id})"),
+ )
+ .await;
+ }
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "events"))
+ .await
+ .unwrap();
+ (tmp, context, table)
+}
+
+async fn assert_ok(context: &SQLContext, sql: &str) {
+ let batches = context.sql(sql).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
+}
+
+async fn scan_ids(table: &Table, plan: Plan) -> Vec<i32> {
+ let batches: Vec<_> = table
+ .new_read_builder()
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut ids: Vec<_> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<arrow_array::Int32Array>()
+ .unwrap()
+ .values()
+ .iter()
+ .copied()
+ })
+ .collect();
+ ids.sort_unstable();
+ ids
+}
+
+#[tokio::test]
+async fn reset_creates_overwrites_and_deletes_only_the_requested_consumer() {
+ let (tmp, context, table) = table_with_snapshots().await;
+ let manager = table.consumer_manager();
+ manager.reset("other", 3).await.unwrap();
+ let root = tmp.path().join("test_db.db/events");
+ let consumer_path = root.join("consumer/consumer-job");
+ let original_files = table_files(&root);
+ assert!(original_files
+ .keys()
+ .any(|path| path.starts_with(root.join("snapshot"))));
+ assert!(original_files
+ .keys()
+ .any(|path| path.starts_with(root.join("manifest"))));
+
+ for id in [2, 1] {
+ assert_ok(
+ &context,
+ &format!("CALL sys.reset_consumer(table =>
'paimon.test_db.events', consumer_id => 'job', next_snapshot_id => {id})"),
+ ).await;
+ assert_eq!(manager.get("job").await.unwrap(), Some(id));
+ let mut files = table_files(&root);
+ let progress: serde_json::Value =
+
serde_json::from_slice(&files.remove(&consumer_path).unwrap()).unwrap();
+ assert_eq!(progress, serde_json::json!({"nextSnapshot": id}));
+ assert_eq!(files, original_files);
+ }
+ for _ in 0..2 {
+ assert_ok(
+ &context,
+ "CALL sys.reset_consumer(table => 'test_db.events', consumer_id =>
'job')",
+ )
+ .await;
+ assert_eq!(manager.get("job").await.unwrap(), None);
+ assert_eq!(manager.get("other").await.unwrap(), Some(3));
+ assert_eq!(table_files(&root), original_files);
+ }
+}
+
+#[tokio::test]
+async fn
deleting_a_missing_consumer_on_an_empty_table_does_not_create_metadata() {
+ let (tmp, catalog) = create_test_env();
+ let context = create_sql_context(catalog).await;
+ exec(&context, "CREATE SCHEMA paimon.test_db").await;
+ exec(&context, "CREATE TABLE paimon.test_db.events (id INT)").await;
+ let root = tmp.path().join("test_db.db/events");
+ let original = table_files(&root);
+ assert_ok(
+ &context,
+ "CALL sys.reset_consumer(table => 'test_db.events', consumer_id =>
'job')",
+ )
+ .await;
+ assert_eq!(table_files(&root), original);
+ assert!(!root.join("consumer").exists());
+ assert_sql_error(
+ &context,
+ "CALL sys.reset_consumer(table => 'test_db.events', consumer_id =>
'job', next_snapshot_id => 1)",
+ "does not exist",
+ )
+ .await;
+ assert_eq!(table_files(&root), original);
+ assert!(!root.join("consumer").exists());
+}
+
+#[tokio::test]
+async fn trailing_whitespace_never_targets_another_consumers_progress() {
+ let (tmp, context, table) = table_with_snapshots().await;
+ let manager = table.consumer_manager();
+ manager.reset("job", 2).await.unwrap();
+ let root = tmp.path().join("test_db.db/events");
+ let original = table_files(&root);
+ for id in ["job ", "job\t", "job\n", "job\u{a0}", "job\u{3000}"] {
+ assert_eq!(manager.get(id).await.unwrap(), None);
+ for snapshot in ["", ", next_snapshot_id => 1"] {
+ assert_sql_error(
+ &context,
+ &format!("CALL sys.reset_consumer(table => 'test_db.events',
consumer_id => '{id}'{snapshot})"),
+ "trailing whitespace",
+ ).await;
+ assert_eq!(manager.get("job").await.unwrap(), Some(2));
+ assert_eq!(table_files(&root), original);
+ }
+ }
+}
+
+#[tokio::test]
+async fn reset_rejects_invalid_arguments_without_changing_files() {
+ let (tmp, context, table) = table_with_snapshots().await;
+ table.consumer_manager().reset("job", 2).await.unwrap();
+ let root = tmp.path().join("test_db.db/events");
+ let original = table_files(&root);
+ for (args, message) in [
+ ("consumer_id => 'job', next_snapshot_id => 1", "Missing required
argument: 'table'"),
+ ("table => 'test_db.events'", "Missing required argument:
'consumer_id'"),
+ ("table => 'test_db.events', consumer_id => 'job', next_snaphot_id =>
1", "is unknown"),
+ ("table => 'test_db.events', consumer_id => 'job', consumer_id =>
'other'", "is duplicated"),
+ ("table => 'test_db.events', consumer_id => 'job', next_snapshot_id =>
1, NEXT_SNAPSHOT_ID => 2", "is duplicated"),
+ ("'test_db.events', 'job', 1", "require named arguments"),
+ ("table => 'test_db.events', consumer_id => 'job', next_snapshot_id =>
NULL", "Unsupported argument value"),
+ ("table => 'other.test_db.events', consumer_id => 'job'", "Catalog
name mismatch"),
+ ] {
+ assert_sql_error(&context, &format!("CALL
sys.reset_consumer({args})"), message).await;
+ assert_eq!(table_files(&root), original);
+ }
+ for id in [
+ "0",
+ "-1",
+ "1.5",
+ "9223372036854775808",
+ "'invalid'",
+ "''",
+ "true",
+ ] {
+ assert_sql_error(
+ &context,
+ &format!("CALL sys.reset_consumer(table => 'test_db.events',
consumer_id => 'job', next_snapshot_id => {id})"),
+ "Invalid next_snapshot_id",
+ ).await;
+ assert_eq!(table_files(&root), original);
+ }
+ for id in ["", ".", "..", "../job", r"job\part"] {
+ for snapshot in ["", ", next_snapshot_id => 1"] {
+ assert_sql_error(
+ &context,
+ &format!("CALL sys.reset_consumer(table => 'test_db.events',
consumer_id => '{id}'{snapshot})"),
+ "consumer id must be",
+ ).await;
+ assert_eq!(table_files(&root), original);
+ }
+ }
+ for id in [4, 99] {
+ assert_sql_error(
+ &context,
+ &format!("CALL sys.reset_consumer(table => 'test_db.events',
consumer_id => 'job', next_snapshot_id => {id})"),
+ "does not exist",
+ ).await;
+ assert_eq!(table_files(&root), original);
+ }
+}
+
+#[tokio::test]
+async fn
reset_rejects_expired_or_corrupt_snapshots_even_when_a_tag_retains_them() {
+ for corrupt in [false, true] {
+ let (tmp, context, table) = table_with_snapshots().await;
+ exec(
+ &context,
+ "CALL sys.create_tag(table => 'test_db.events', tag => 'keep',
snapshot_id => 1)",
+ )
+ .await;
+ table.consumer_manager().reset("job", 2).await.unwrap();
+ let snapshot_path = table.snapshot_manager().snapshot_path(1);
+ if corrupt {
+ table
+ .file_io()
+ .new_output(&snapshot_path)
+ .unwrap()
+ .write("{".into())
+ .await
+ .unwrap();
+ } else {
+ table.file_io().delete_file(&snapshot_path).await.unwrap();
+ }
+ assert_eq!(
+ table.tag_manager().get("keep").await.unwrap().unwrap().id(),
+ 1
+ );
+ let root = tmp.path().join("test_db.db/events");
+ let original = table_files(&root);
+ assert_sql_error(
+ &context,
+ "CALL sys.reset_consumer(table => 'test_db.events', consumer_id =>
'job', next_snapshot_id => 1)",
+ if corrupt { "snapshot JSON invalid" } else { "does not exist" },
+ ).await;
+ assert_eq!(table.consumer_manager().get("job").await.unwrap(),
Some(2));
+ assert_eq!(table_files(&root), original);
+ }
+}
+
+#[tokio::test]
+async fn reset_rejects_format_tables_before_accessing_consumer_metadata() {
+ let (tmp, catalog) = create_test_env();
+ catalog
+ .create_database("test_db", false, Default::default())
+ .await
+ .unwrap();
+ let identifier = Identifier::new("test_db", "events");
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .option("type", "format-table")
+ .option("file.format", "parquet")
+ .build()
+ .unwrap();
+ catalog
+ .create_table(&identifier, schema, false)
+ .await
+ .unwrap();
+ let table = catalog.get_table(&identifier).await.unwrap();
+ table.consumer_manager().reset("job", 2).await.unwrap();
+ let context = create_sql_context(catalog).await;
+ let root = tmp.path().join("test_db.db/events");
+ let original = table_files(&root);
+ for snapshot in ["", ", next_snapshot_id => 1"] {
+ assert_sql_error(
+ &context,
+ &format!("CALL sys.reset_consumer(table => 'test_db.events',
consumer_id => 'job'{snapshot})"),
+ "requires a FileStoreTable",
+ ).await;
+ assert_eq!(table_files(&root), original);
+ }
+}
+
+#[tokio::test]
+async fn
sql_reset_is_used_by_new_scans_and_explicit_restore_takes_precedence() {
+ let (_tmp, context, table) = table_with_snapshots().await;
+ assert_ok(&context, "CALL sys.reset_consumer(table => 'test_db.events',
consumer_id => 'job', next_snapshot_id => 1)").await;
+ let mut running = table.new_stream_scan().unwrap();
+ running.with_consumer_id("job").unwrap();
+ assert_eq!(
+ scan_ids(&table, running.plan().await.unwrap().unwrap()).await,
+ vec![1]
+ );
+ assert_ok(&context, "CALL sys.reset_consumer(table => 'test_db.events',
consumer_id => 'job', next_snapshot_id => 3)").await;
+ assert_eq!(
+ scan_ids(&table, running.plan().await.unwrap().unwrap()).await,
+ vec![2]
+ );
+ let mut fresh = table.new_stream_scan().unwrap();
+ fresh.with_consumer_id("job").unwrap();
+ let plan = fresh.plan().await.unwrap().unwrap();
+ assert_eq!(plan.snapshot_id(), Some(3));
+ assert!(plan.splits().iter().all(|split| split.is_streaming()));
+ assert_eq!(scan_ids(&table, plan).await, vec![3]);
+ let mut restored = table.new_stream_scan().unwrap();
+ restored
+ .with_consumer_id("job")
+ .unwrap()
+ .restore(Some(1))
+ .unwrap();
+ assert_eq!(
+ scan_ids(&table, restored.plan().await.unwrap().unwrap()).await,
+ vec![1]
+ );
+ assert_ok(&context, "CALL sys.reset_consumer(table => 'test_db.events',
consumer_id => 'job', next_snapshot_id => 1)").await;
+ running
+ .notify_checkpoint_complete(running.checkpoint())
+ .await
+ .unwrap();
+ assert_eq!(table.consumer_manager().get("job").await.unwrap(), Some(3));
+}
+
+#[tokio::test]
+async fn
sql_delete_restores_initial_scan_behavior_and_checkpoint_can_recreate_progress()
{
+ let (_tmp, context, table) = table_with_snapshots().await;
+ let mut waiting = table.new_stream_scan().unwrap();
+ waiting
+ .with_consumer_id("job")
+ .unwrap()
+ .restore(Some(3))
+ .unwrap();
+ waiting.plan().await.unwrap().unwrap();
+ waiting
+ .notify_checkpoint_complete(waiting.checkpoint())
+ .await
+ .unwrap();
+ assert_eq!(table.consumer_manager().get("job").await.unwrap(), Some(4));
+ assert!(table.snapshot_manager().get_snapshot(4).await.is_err());
+ assert_ok(
+ &context,
+ "CALL sys.reset_consumer(table => 'test_db.events', consumer_id =>
'job')",
+ )
+ .await;
+ assert_eq!(table.consumer_manager().get("job").await.unwrap(), None);
+ let mut fresh = table.new_stream_scan().unwrap();
+ fresh.with_consumer_id("job").unwrap();
+ let plan = fresh.plan().await.unwrap().unwrap();
+ assert_eq!(plan.snapshot_id(), Some(3));
+ assert!(!plan.splits().is_empty());
+ assert!(plan.splits().iter().all(|split| !split.is_streaming()));
+ assert_eq!(scan_ids(&table, plan).await, vec![1, 2, 3]);
+ assert!(waiting.plan().await.unwrap().is_none());
+ assert_eq!(waiting.checkpoint(), Some(4));
+ waiting
+ .notify_checkpoint_complete(waiting.checkpoint())
+ .await
+ .unwrap();
+ assert_eq!(table.consumer_manager().get("job").await.unwrap(), Some(4));
+}
diff --git a/crates/integrations/datafusion/tests/rest_management_procedures.rs
b/crates/integrations/datafusion/tests/rest_management_procedures.rs
index 039c2bfc..00ebc159 100644
--- a/crates/integrations/datafusion/tests/rest_management_procedures.rs
+++ b/crates/integrations/datafusion/tests/rest_management_procedures.rs
@@ -651,6 +651,32 @@ async fn
test_non_rest_catalog_is_rejected_by_every_procedure() {
}
}
+#[tokio::test]
+async fn
test_reset_consumer_rejected_on_rest_catalog_without_changing_progress() {
+ 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 consumer_path = tmp
+ .path()
+ .join(format!("{DATABASE}.db/{TABLE}/consumer/consumer-job"));
+ std::fs::create_dir_all(consumer_path.parent().unwrap()).unwrap();
+ let original = br#"{"nextSnapshot":2}"#;
+ std::fs::write(&consumer_path, original).unwrap();
+
+ for snapshot in ["", ", next_snapshot_id => 1"] {
+ common::assert_sql_error(
+ &context,
+ &format!("CALL sys.reset_consumer(table => '{DATABASE}.{TABLE}',
consumer_id => 'job'{snapshot})"),
+ "reset_consumer is not supported for tables managed by a REST
catalog",
+ ).await;
+ assert_eq!(std::fs::read(&consumer_path).unwrap(), original);
+ }
+}
+
#[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/consumer_manager.rs
b/crates/paimon/src/table/consumer_manager.rs
index 20e1a084..f5febbd4 100644
--- a/crates/paimon/src/table/consumer_manager.rs
+++ b/crates/paimon/src/table/consumer_manager.rs
@@ -53,13 +53,16 @@ impl ConsumerManager {
)
}
- pub(super) fn validate_consumer_id(consumer_id: &str) -> crate::Result<()>
{
+ /// Validate a consumer name before reading or mutating its metadata.
+ pub fn validate_consumer_id(consumer_id: &str) -> crate::Result<()> {
if consumer_id.is_empty()
|| matches!(consumer_id, "." | "..")
|| consumer_id.contains(['/', '\\'])
+ // OpenDAL trims path boundaries, so trailing whitespace would
alias another ID.
+ || consumer_id.trim_end() != consumer_id
{
return Err(crate::Error::DataInvalid {
- message: "consumer id must be a non-empty name without path
separators".into(),
+ message: "consumer id must be a non-empty name without path
separators or trailing whitespace".into(),
source: None,
});
}
@@ -98,6 +101,16 @@ impl ConsumerManager {
self.file_io.new_output(&path)?.write(bytes.into()).await
}
+ /// Delete one consumer's progress. A missing file is a successful no-op.
+ pub async fn delete(&self, consumer_id: &str) -> crate::Result<()> {
+ Self::validate_consumer_id(consumer_id)?;
+ let path = format!(
+ "{}/{CONSUMER_DIR}/{CONSUMER_PREFIX}{consumer_id}",
+ self.table_path
+ );
+ self.file_io.delete_file(&path).await
+ }
+
async fn read_path(&self, consumer_id: &str, path: &str) ->
crate::Result<Option<i64>> {
let input = self.file_io.new_input(path)?;
for attempt in 0..READ_RETRIES {
@@ -185,10 +198,156 @@ struct Consumer {
#[cfg(test)]
mod tests {
+ use std::sync::Arc;
+
use bytes::Bytes;
use super::*;
- use crate::io::FileIOBuilder;
+ use crate::io::{FileIOBuilder, FileIOProvider};
+
+ #[tokio::test]
+ async fn delete_is_idempotent_and_preserves_other_consumers_and_branches()
{
+ for scheme in ["memory", "file"] {
+ let tmp = tempfile::tempdir().unwrap();
+ let table_path = if scheme == "file" {
+ format!("file://{}/table", tmp.path().display())
+ } else {
+ "memory:/consumer-delete".to_string()
+ };
+ let file_io = FileIOBuilder::new(scheme).build().unwrap();
+ let manager = ConsumerManager::new(file_io.clone(),
table_path.clone());
+ let branch = manager.with_branch("dev");
+ let sibling = manager.with_branch("release");
+ manager.delete("missing").await.unwrap();
+ assert!(!file_io
+ .exists_dir(&format!("{table_path}/consumer"))
+ .await
+ .unwrap());
+ manager.reset("job", 1).await.unwrap();
+ manager.reset("other", 2).await.unwrap();
+ branch.reset("job", 3).await.unwrap();
+ sibling.reset("job", 4).await.unwrap();
+
+ branch.delete("job").await.unwrap();
+ branch.delete("job").await.unwrap();
+ assert_eq!(branch.get("job").await.unwrap(), None);
+ assert_eq!(manager.get("job").await.unwrap(), Some(1));
+ assert_eq!(sibling.get("job").await.unwrap(), Some(4));
+ assert!(file_io
+
.exists_dir(&format!("{table_path}/branch/branch-dev/consumer"))
+ .await
+ .unwrap());
+
+ branch.reset("job", 5).await.unwrap();
+ manager.delete("job").await.unwrap();
+ manager.delete("job").await.unwrap();
+ assert_eq!(
+ manager.list_all().await.unwrap(),
+ vec![("other".to_string(), 2)]
+ );
+ assert_eq!(branch.get("job").await.unwrap(), Some(5));
+ assert_eq!(sibling.get("job").await.unwrap(), Some(4));
+ }
+ }
+
+ #[tokio::test]
+ async fn trailing_whitespace_cannot_alias_another_consumers_progress() {
+ for scheme in ["memory", "file"] {
+ let tmp = tempfile::tempdir().unwrap();
+ let table_path = if scheme == "file" {
+ format!("file://{}/table", tmp.path().display())
+ } else {
+ "memory:/consumer-whitespace".to_string()
+ };
+ let file_io = FileIOBuilder::new(scheme).build().unwrap();
+ let manager = ConsumerManager::new(file_io.clone(), table_path);
+ let branch = manager.with_branch("dev");
+ for scoped in [&manager, &branch] {
+ scoped.reset("job", 1).await.unwrap();
+ let path = format!("{}/consumer/consumer-job",
scoped.table_path);
+ let original =
file_io.new_input(&path).unwrap().read().await.unwrap();
+ for id in [
+ "job ",
+ "job\t",
+ "job\n",
+ "job\r\n",
+ "job\u{a0}",
+ "job\u{3000}",
+ ] {
+ assert_eq!(scoped.get(id).await.unwrap(), None);
+ assert!(scoped.delete(id).await.is_err(), "{scheme}:
{id:?}");
+ assert_eq!(
+
file_io.new_input(&path).unwrap().read().await.unwrap(),
+ original
+ );
+ assert!(scoped.reset(id, 3).await.is_err(), "{scheme}:
{id:?}");
+ assert_eq!(
+
file_io.new_input(&path).unwrap().read().await.unwrap(),
+ original
+ );
+ }
+ scoped.reset(" job", 2).await.unwrap();
+ assert_eq!(scoped.get(" job").await.unwrap(), Some(2));
+ scoped.delete(" job").await.unwrap();
+ assert_eq!(scoped.get("job").await.unwrap(), Some(1));
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn delete_rejects_invalid_ids_without_changing_files() {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let manager =
+ ConsumerManager::new(file_io.clone(),
"memory:/consumer-delete-invalid".into());
+ manager.reset("job", 2).await.unwrap();
+ for invalid in ["", ".", "..", "x/../../snapshot/snapshot-1", "x\\y"] {
+ assert!(manager.delete(invalid).await.is_err());
+ }
+ assert_eq!(
+ manager.list_all().await.unwrap(),
+ vec![("job".to_string(), 2)]
+ );
+ }
+
+ #[tokio::test]
+ async fn delete_propagates_storage_errors() {
+ #[derive(Debug)]
+ struct DeniedProvider;
+
+ #[async_trait::async_trait]
+ impl FileIOProvider for DeniedProvider {
+ async fn create(&self, _path: &str) ->
crate::Result<(opendal::Operator, String)> {
+ Err(
+ opendal::Error::new(opendal::ErrorKind::PermissionDenied,
"access denied")
+ .into(),
+ )
+ }
+ }
+
+ let file_io = FileIOBuilder::new("memory")
+ .build()
+ .unwrap()
+ .with_provider(Arc::new(DeniedProvider));
+ let manager = ConsumerManager::new(file_io,
"memory:/consumer-delete-denied".into());
+ let error = manager.delete("job").await.unwrap_err();
+ assert!(matches!(error, crate::Error::IoUnexpected { source, .. }
+ if source.kind() == opendal::ErrorKind::PermissionDenied));
+ }
+
+ #[tokio::test]
+ async fn delete_does_not_recursively_remove_a_directory() {
+ let tmp = tempfile::tempdir().unwrap();
+ let directory = tmp.path().join("consumer/consumer-job");
+ std::fs::create_dir_all(&directory).unwrap();
+ let child = directory.join("keep");
+ std::fs::write(&child, b"keep").unwrap();
+ let manager = ConsumerManager::new(
+ FileIOBuilder::new("file").build().unwrap(),
+ format!("file://{}", tmp.path().display()),
+ );
+ assert!(manager.delete("job").await.is_err());
+ assert_eq!(std::fs::read(child).unwrap(), b"keep");
+ }
#[tokio::test]
async fn reset_is_explicit_branch_scoped_and_rejects_path_traversal() {
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 4eff1e96..b00cdfaf 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1057,6 +1057,47 @@ anything is deleted.
Use `CALL` to invoke built-in procedures. All procedures are under the `sys`
namespace.
+### reset_consumer
+
+Create or overwrite a consumer's next snapshot to consume:
+
+```sql
+CALL sys.reset_consumer(
+ table => 'paimon.test_db.events',
+ consumer_id => 'pipeline_a',
+ next_snapshot_id => 12
+);
+```
+
+`next_snapshot_id` must be a positive BIGINT identifying an existing live
snapshot.
+Missing, expired, or invalid snapshots are rejected before changing consumer
progress;
+tag-retained metadata alone is not sufficient. The ID is the next snapshot to
process
+using the existing incremental or changelog scan rules, rather than a
full-state read.
+
+Omit `next_snapshot_id` to delete the consumer record. Deleting a missing
consumer succeeds:
+
+```sql
+CALL sys.reset_consumer(
+ table => 'paimon.test_db.events',
+ consumer_id => 'pipeline_a'
+);
+```
+
+Only filesystem-managed FileStore tables are supported. REST-managed tables and
+format tables are not supported. Consumer IDs must be non-empty names without
path
+separators or trailing whitespace; IDs are never trimmed. Arguments use the
existing
+named `=>` syntax; explicit `NULL` and positional arguments are not supported.
+Successful calls return `OK`.
+These operations change only consumer metadata, without publishing snapshots or
+changing table data, snapshot files, or manifests.
+
+A new scan with the same consumer ID resumes the stored progress unless the
caller
+explicitly restores a checkpoint, which takes precedence. After deletion, a
new scan
+without an explicit checkpoint follows its existing initial scan behavior.
Scans that
+have already loaded their progress do not switch positions when the record
changes.
+A running consumer's checkpoint completion can overwrite or recreate the
record,
+so stop the consumer before adjusting its progress manually.
+
### create_tag
Create a named tag from a snapshot:
