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 4cf34379 feat(datafusion): support clear_consumers (#1213)
4cf34379 is described below
commit 4cf34379c2f56527039f7e706488582d72a8e6ab
Author: QuakeWang <[email protected]>
AuthorDate: Sat Oct 24 10:25:35 2026 +0800
feat(datafusion): support clear_consumers (#1213)
---
crates/integrations/datafusion/src/procedures.rs | 31 ++
.../datafusion/tests/clear_consumers.rs | 325 ++++++++++++++++
.../datafusion/tests/rest_management_procedures.rs | 9 +-
crates/paimon/src/table/consumer_manager.rs | 418 +++++++++++++++++++++
docs/src/sql.md | 56 +++
5 files changed, 838 insertions(+), 1 deletion(-)
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index 6db19eda..0f743224 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -22,6 +22,7 @@
//! - `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.clear_consumers(table => '...'[, including_consumers => '...',
excluding_consumers => '...'])`
//! - `CALL sys.rollback_to(table => '...', snapshot_id => ... | tag => '...')`
//! - `CALL sys.rollback_to_timestamp(table => '...', timestamp => ...)`
//! - `CALL sys.create_tag_from_timestamp(table => '...', tag => '...',
timestamp => ...)`
@@ -206,6 +207,7 @@ fn declared_parameters(proc_name: &str) -> Option<&'static
[&'static str]> {
"delete_tag" => &["table", "tag"],
"rename_tag" => &["table", "tag", "target_tag"],
"reset_consumer" => &["table", "consumer_id", "next_snapshot_id"],
+ "clear_consumers" => &["table", "including_consumers",
"excluding_consumers"],
"rollback_to" => &["table", "snapshot_id", "tag"],
"rollback_to_timestamp" => &["table", "timestamp"],
"create_tag_from_timestamp" => &["table", "tag", "timestamp"],
@@ -318,6 +320,7 @@ pub async fn execute_call(
"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,
+ "clear_consumers" => proc_clear_consumers(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
@@ -518,6 +521,34 @@ async fn proc_reset_consumer(
ok_result(ctx)
}
+async fn proc_clear_consumers(
+ ctx: &SessionContext,
+ catalog: &Arc<dyn Catalog>,
+ catalog_name: &str,
+ args: &HashMap<String, String>,
+) -> DFResult<DataFrame> {
+ let table = get_table(catalog, catalog_name, args).await?;
+ if table.rest_env().is_some() {
+ return Err(DataFusionError::NotImplemented(
+ "clear_consumers is not supported for tables managed by a REST
catalog".to_string(),
+ ));
+ }
+ if table.schema().core_options().is_format_table() {
+ return Err(DataFusionError::NotImplemented(
+ "clear_consumers requires a FileStoreTable; format-table is not
supported".to_string(),
+ ));
+ }
+ table
+ .consumer_manager()
+ .clear(
+ args.get("including_consumers").map(String::as_str),
+ args.get("excluding_consumers").map(String::as_str),
+ )
+ .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/clear_consumers.rs
b/crates/integrations/datafusion/tests/clear_consumers.rs
new file mode 100644
index 00000000..16d7fd9d
--- /dev/null
+++ b/crates/integrations/datafusion/tests/clear_consumers.rs
@@ -0,0 +1,325 @@
+// 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 paimon::catalog::Identifier;
+use paimon::spec::{DataType, IntType, Schema};
+use paimon::table::{BranchManager, Table};
+use paimon::Catalog;
+use paimon_datafusion::SQLContext;
+
+async fn table_with_snapshot() -> (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;
+ exec(&context, "INSERT INTO paimon.test_db.events VALUES (1)").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
+}
+
+#[tokio::test]
+async fn clear_filters_ids_and_preserves_exceptions_other_files_and_branches()
{
+ let (tmp, context, table) = table_with_snapshot().await;
+ let manager = table.consumer_manager();
+ manager.reset("job-old", 1).await.unwrap();
+ manager.reset("job-keep", 2).await.unwrap();
+ let snapshot = table.snapshot_manager().get_snapshot(1).await.unwrap();
+ table.tag_manager().create("base", &snapshot).await.unwrap();
+ BranchManager::new(table.file_io().clone(), table.location().to_string())
+ .create_branch_from_tag("dev", "base")
+ .await
+ .unwrap();
+ let branch = table.copy_with_branch("dev").await.unwrap();
+ branch.consumer_manager().reset("job-old", 3).await.unwrap();
+
+ let root = tmp.path().join("test_db.db/events");
+ let directory = root.join("consumer");
+ for name in [
+ "consumer-job-old",
+ "consumer-job-keep-extra",
+ "consumer-other",
+ ] {
+ std::fs::write(directory.join(name), "{").unwrap();
+ }
+ std::fs::write(
+ directory.join("consumer-job-keep"),
+ "{ \"nextSnapshot\": 2 }\n",
+ )
+ .unwrap();
+ std::fs::write(directory.join("notes"), "keep").unwrap();
+ std::fs::write(directory.join("consumer-pre-job-old"), "keep").unwrap();
+ std::fs::create_dir(directory.join("consumer-job-directory")).unwrap();
+ std::fs::write(directory.join("consumer-job-directory/checkpoint"),
"keep").unwrap();
+ let mut expected = table_files(&root);
+ expected
+ .remove(&directory.join("consumer-job-old"))
+ .unwrap();
+ expected
+ .remove(&directory.join("consumer-job-keep-extra"))
+ .unwrap();
+
+ for _ in 0..2 {
+ assert_ok(&context, "CALL paimon.sys.clear_consumers(table =>
'paimon.test_db.events', including_consumers => 'job-.*', excluding_consumers
=> 'job-keep')").await;
+ assert_eq!(table_files(&root), expected);
+ assert_eq!(manager.get("job-keep").await.unwrap(), Some(2));
+ assert_eq!(
+ branch.consumer_manager().get("job-old").await.unwrap(),
+ Some(3)
+ );
+ }
+ assert_sql_error(
+ &context,
+ "CALL sys.clear_consumers(table => 'test_db.events$branch_dev')",
+ "does not exist",
+ )
+ .await;
+ assert_eq!(table_files(&root), expected);
+}
+
+#[tokio::test]
+async fn clear_accepts_defaults_and_does_not_create_a_missing_directory() {
+ 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;
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "events"))
+ .await
+ .unwrap();
+ let root = tmp.path().join("test_db.db/events");
+ let original = table_files(&root);
+ for _ in 0..2 {
+ assert_ok(
+ &context,
+ "CALL sys.clear_consumers(table => 'test_db.events')",
+ )
+ .await;
+ assert_eq!(table_files(&root), original);
+ assert!(!root.join("consumer").exists());
+ }
+ for args in [
+ "",
+ ", including_consumers => ''",
+ ", including_consumers => ' \t', excluding_consumers => ''",
+ ] {
+ table.consumer_manager().reset("job", 2).await.unwrap();
+ assert_ok(
+ &context,
+ &format!("CALL sys.clear_consumers(table =>
'test_db.events'{args})"),
+ )
+ .await;
+ assert_eq!(table.consumer_manager().get("job").await.unwrap(), None);
+ assert_eq!(table_files(&root), original);
+ }
+ table.consumer_manager().reset("job", 2).await.unwrap();
+ assert_ok(
+ &context,
+ "CALL sys.clear_consumers(table => 'test_db.events',
excluding_consumers => 'job')",
+ )
+ .await;
+ assert_eq!(table.consumer_manager().get("job").await.unwrap(), Some(2));
+}
+
+#[tokio::test]
+async fn clear_rejects_invalid_arguments_before_changing_files() {
+ let (tmp, context, table) = table_with_snapshot().await;
+ table.consumer_manager().reset("job", 1).await.unwrap();
+ let root = tmp.path().join("test_db.db/events");
+ let original = table_files(&root);
+ for (args, message) in [
+ ("", "Missing required argument: 'table'"),
+ ("'test_db.events'", "require named arguments"),
+ ("table => 'other.test_db.events'", "Catalog name mismatch"),
+ ("table => 'test_db.events', including_consumer => 'job'", "is
unknown"),
+ ("table => 'test_db.events', excluding_consumer => 'job'", "is
unknown"),
+ ("table => 'test_db.events', including_consumers => 'job',
INCLUDING_CONSUMERS => 'other'", "is duplicated"),
+ ("table => 'test_db.events', excluding_consumers => 'job',
excluding_consumers => 'other'", "is duplicated"),
+ ("table => 'test_db.events', including_consumers => NULL",
"Unsupported argument value"),
+ ("table => 'test_db.events', excluding_consumers => NULL",
"Unsupported argument value"),
+ ("table => 'test_db.events', including_consumers => '['", "Invalid
including_consumers regex"),
+ ("table => 'test_db.events', excluding_consumers => '['", "Invalid
excluding_consumers regex"),
+ ("table => 'test_db.events', including_consumers => 'no-match',
excluding_consumers => '(?=job)job'", "Invalid excluding_consumers regex"),
+ (r"table => 'test_db.events', including_consumers => '(job)\1'",
"Invalid including_consumers regex"),
+ ] {
+ assert_sql_error(&context, &format!("CALL
sys.clear_consumers({args})"), message).await;
+ assert_eq!(table_files(&root), original);
+ }
+}
+
+#[tokio::test]
+async fn clear_uses_whole_id_matching_with_rust_regex_flags() {
+ let (_tmp, context, table) = table_with_snapshot().await;
+ for (pattern, id, deleted) in [
+ ("job", "job-old", false),
+ ("a|ab", "ab", true),
+ ("(?m)^job$", "job\nother", false),
+ ("(?s)job.*", "job\nother", true),
+ ("(?i)job", "JOB", true),
+ ("(?x) job # trailing comment", "job", true),
+ (" job", "job", false),
+ ] {
+ table.consumer_manager().reset(id, 2).await.unwrap();
+ assert_ok(&context, &format!("CALL sys.clear_consumers(table =>
'test_db.events', including_consumers => '{pattern}')")).await;
+ assert_eq!(
+ table.consumer_manager().get(id).await.unwrap(),
+ if deleted { None } else { Some(2) },
+ "{pattern:?}: {id:?}"
+ );
+ table.consumer_manager().delete(id).await.unwrap();
+ }
+}
+
+#[tokio::test]
+async fn clear_rejects_format_tables_without_changing_files() {
+ 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();
+ catalog
+ .get_table(&identifier)
+ .await
+ .unwrap()
+ .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);
+ assert_sql_error(
+ &context,
+ "CALL sys.clear_consumers(table => 'test_db.events')",
+ "requires a FileStoreTable",
+ )
+ .await;
+ assert_eq!(table_files(&root), original);
+}
+
+#[tokio::test]
+async fn clear_rejects_engine_served_table_types_without_changing_files() {
+ for table_type in ["object-table", "iceberg-table", "lance-table"] {
+ let (tmp, catalog) = create_test_env();
+ catalog
+ .create_database("test_db", false, Default::default())
+ .await
+ .unwrap();
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .option("type", table_type)
+ .build()
+ .unwrap();
+ catalog
+ .create_table(&Identifier::new("test_db", "events"), schema, false)
+ .await
+ .unwrap();
+ let root = tmp.path().join("test_db.db/events");
+ std::fs::create_dir(root.join("consumer")).unwrap();
+ std::fs::write(root.join("consumer/consumer-job"), "{").unwrap();
+ let original = table_files(&root);
+ let context = create_sql_context(catalog).await;
+ assert_sql_error(
+ &context,
+ "CALL sys.clear_consumers(table => 'test_db.events')",
+ "cannot be read as a Paimon table",
+ )
+ .await;
+ assert_eq!(table_files(&root), original);
+ }
+}
+
+#[tokio::test]
+async fn clear_preserves_running_scan_progress_and_checkpoint_protocol() {
+ let (_tmp, context, table) = table_with_snapshot().await;
+ table.consumer_manager().reset("keep", 2).await.unwrap();
+ let mut running = table.new_stream_scan().unwrap();
+ running
+ .with_consumer_id("job")
+ .unwrap()
+ .restore(Some(1))
+ .unwrap();
+ running.plan().await.unwrap().unwrap();
+ running
+ .notify_checkpoint_complete(running.checkpoint())
+ .await
+ .unwrap();
+ assert_eq!(table.consumer_manager().get("job").await.unwrap(), Some(2));
+
+ assert_ok(
+ &context,
+ "CALL sys.clear_consumers(table => 'test_db.events',
excluding_consumers => 'keep')",
+ )
+ .await;
+ assert_eq!(table.consumer_manager().get("job").await.unwrap(), None);
+ assert_eq!(table.consumer_manager().get("keep").await.unwrap(), Some(2));
+ assert_eq!(running.checkpoint(), Some(2));
+ assert!(running.plan().await.unwrap().is_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(1));
+ assert!(plan.splits().iter().all(|split| !split.is_streaming()));
+
+ running
+ .notify_checkpoint_complete(running.checkpoint())
+ .await
+ .unwrap();
+ assert_eq!(table.consumer_manager().get("job").await.unwrap(), Some(2));
+ assert_eq!(table.consumer_manager().get("keep").await.unwrap(), Some(2));
+}
diff --git a/crates/integrations/datafusion/tests/rest_management_procedures.rs
b/crates/integrations/datafusion/tests/rest_management_procedures.rs
index 00ebc159..22568abd 100644
--- a/crates/integrations/datafusion/tests/rest_management_procedures.rs
+++ b/crates/integrations/datafusion/tests/rest_management_procedures.rs
@@ -652,7 +652,7 @@ 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() {
+async fn
test_consumer_procedures_rejected_on_rest_catalog_without_changing_progress() {
let (tmp, server, context) = setup().await;
let schema = Schema::builder()
.column("id", DataType::Int(IntType::new()))
@@ -675,6 +675,13 @@ async fn
test_reset_consumer_rejected_on_rest_catalog_without_changing_progress(
).await;
assert_eq!(std::fs::read(&consumer_path).unwrap(), original);
}
+ common::assert_sql_error(
+ &context,
+ &format!("CALL sys.clear_consumers(table => '{DATABASE}.{TABLE}')"),
+ "clear_consumers is not supported for tables managed by a REST
catalog",
+ )
+ .await;
+ assert_eq!(std::fs::read(&consumer_path).unwrap(), original);
}
#[tokio::test]
diff --git a/crates/paimon/src/table/consumer_manager.rs
b/crates/paimon/src/table/consumer_manager.rs
index f5febbd4..1ba94647 100644
--- a/crates/paimon/src/table/consumer_manager.rs
+++ b/crates/paimon/src/table/consumer_manager.rs
@@ -22,6 +22,7 @@ use std::time::Duration;
use crate::io::FileIO;
use futures::{stream, StreamExt, TryStreamExt};
+use regex::Regex;
use serde::{Deserialize, Serialize};
const CONSUMER_DIR: &str = "consumer";
@@ -111,6 +112,74 @@ impl ConsumerManager {
self.file_io.delete_file(&path).await
}
+ /// Delete consumer files whose ids fully match the inclusion regex but
not the exclusion.
+ ///
+ /// Patterns use Rust `regex` syntax, including Unicode classes and inline
flags, but
+ /// not look-around or backreferences. Omitted, empty, or
Java-whitespace-only inclusion
+ /// defaults to `.*`; the equivalent exclusion means no exclusions.
Nonblank patterns
+ /// are never trimmed. Both patterns are validated before listing or
deleting files.
+ /// Only filenames are inspected, so corrupt consumer JSON can also be
cleared.
+ /// Missing directories are a no-op; other storage errors are propagated.
Deletion is
+ /// not atomic, and a running consumer can recreate its progress at
checkpoint completion.
+ pub async fn clear(
+ &self,
+ including_consumers: Option<&str>,
+ excluding_consumers: Option<&str>,
+ ) -> crate::Result<()> {
+ let including = consumer_pattern(
+ including_consumers
+ .filter(|pattern| !is_java_blank(pattern))
+ .unwrap_or(".*"),
+ "including_consumers",
+ )?;
+ let excluding = excluding_consumers
+ .filter(|pattern| !is_java_blank(pattern))
+ .map(|pattern| consumer_pattern(pattern, "excluding_consumers"))
+ .transpose()?;
+
+ let directory = format!("{}/{CONSUMER_DIR}", self.table_path);
+ let statuses = match self.file_io.list_status(&directory).await {
+ Ok(statuses) => statuses,
+ Err(crate::Error::IoUnexpected { ref source, .. })
+ if source.kind() == opendal::ErrorKind::NotFound =>
+ {
+ return Ok(());
+ }
+ Err(error) => return Err(error),
+ };
+ let mut paths = Vec::new();
+ for status in statuses {
+ if status.is_dir {
+ continue;
+ }
+ let name = status.path.rsplit('/').next().unwrap_or(&status.path);
+ let Some(id) = name.strip_prefix(CONSUMER_PREFIX) else {
+ continue;
+ };
+ if !including.is_match(id)
+ || excluding
+ .as_ref()
+ .is_some_and(|pattern| pattern.is_match(id))
+ {
+ continue;
+ }
+ // OpenDAL would trim this path and could delete an excluded
consumer instead.
+ if status.path.trim_end() != status.path {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Cannot clear consumer '{id}': trailing whitespace in
filename"
+ ),
+ source: None,
+ });
+ }
+ paths.push(status.path);
+ }
+ for path in paths {
+ self.file_io.delete_file(&path).await?;
+ }
+ Ok(())
+ }
+
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 {
@@ -190,6 +259,36 @@ impl ConsumerManager {
}
}
+fn consumer_pattern(pattern: &str, argument: &str) -> crate::Result<Regex> {
+ let invalid = |error: regex::Error| crate::Error::DataInvalid {
+ message: format!("Invalid {argument} regex: {error}"),
+ source: Some(Box::new(error)),
+ };
+ // Validate independently so unmatched groups cannot escape the boundary
wrapper.
+ Regex::new(pattern).map_err(invalid)?;
+ Regex::new(&format!(r"\A(?:{pattern})\z"))
+ // A trailing verbose-mode comment can otherwise swallow the closing
wrapper.
+ .or_else(|_| Regex::new(&format!("\\A(?:{pattern}\n)\\z")))
+ .map_err(invalid)
+}
+
+fn is_java_blank(value: &str) -> bool {
+ // Character.isWhitespace excludes nonbreaking spaces, unlike Rust's
trim().
+ value.chars().all(|c| {
+ matches!(
+ c,
+ '\u{0009}'..='\u{000d}'
+ | '\u{001c}'..='\u{0020}'
+ | '\u{1680}'
+ | '\u{2000}'..='\u{2006}'
+ | '\u{2008}'..='\u{200a}'
+ | '\u{2028}'..='\u{2029}'
+ | '\u{205f}'
+ | '\u{3000}'
+ )
+ })
+}
+
#[derive(Deserialize, Serialize)]
struct Consumer {
#[serde(rename = "nextSnapshot")]
@@ -205,6 +304,325 @@ mod tests {
use super::*;
use crate::io::{FileIOBuilder, FileIOProvider};
+ #[tokio::test]
+ async fn
clear_matches_filenames_without_reading_json_and_preserves_other_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-clear".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");
+ for scoped in [&manager, &branch, &sibling] {
+ scoped.reset("job-old", 2).await.unwrap();
+ scoped.reset("job-keep", 3).await.unwrap();
+ }
+ let directory = format!("{table_path}/consumer");
+ for name in [
+ "consumer-job-old",
+ "consumer-job-keep",
+ "consumer-job-keep-extra",
+ ] {
+ file_io
+ .new_output(&format!("{directory}/{name}"))
+ .unwrap()
+ .write("{".into())
+ .await
+ .unwrap();
+ }
+ for name in ["notes", "consumer-job", "consumer-pre-job-old"] {
+ file_io
+ .new_output(&format!("{directory}/{name}"))
+ .unwrap()
+ .write("keep".into())
+ .await
+ .unwrap();
+ }
+ let nested = format!("{directory}/consumer-job-directory");
+ file_io.mkdirs(&nested).await.unwrap();
+ file_io
+ .new_output(&format!("{nested}/consumer-job-child"))
+ .unwrap()
+ .write("keep child".into())
+ .await
+ .unwrap();
+
+ branch
+ .clear(Some("job-.*"), Some("job-keep"))
+ .await
+ .unwrap();
+ assert_eq!(branch.get("job-old").await.unwrap(), None);
+ assert_eq!(branch.get("job-keep").await.unwrap(), Some(3));
+ assert!(file_io
+ .new_input(&format!("{directory}/consumer-job-old"))
+ .unwrap()
+ .exists()
+ .await
+ .unwrap());
+ for _ in 0..2 {
+ manager
+ .clear(Some("job-.*"), Some("job-keep"))
+ .await
+ .unwrap();
+ for name in ["consumer-job-old", "consumer-job-keep-extra"] {
+ assert!(!file_io
+ .new_input(&format!("{directory}/{name}"))
+ .unwrap()
+ .exists()
+ .await
+ .unwrap());
+ }
+ for (name, expected) in [
+ ("consumer-job-keep", "{"),
+ ("notes", "keep"),
+ ("consumer-job", "keep"),
+ ("consumer-pre-job-old", "keep"),
+ ("consumer-job-directory/consumer-job-child", "keep
child"),
+ ] {
+ assert_eq!(
+ file_io
+ .new_input(&format!("{directory}/{name}"))
+ .unwrap()
+ .read()
+ .await
+ .unwrap()
+ .as_ref(),
+ expected.as_bytes()
+ );
+ }
+ assert_eq!(branch.get("job-keep").await.unwrap(), Some(3));
+ assert_eq!(sibling.get("job-old").await.unwrap(), Some(2));
+ assert_eq!(sibling.get("job-keep").await.unwrap(), Some(3));
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn clear_defaults_and_empty_directories_are_idempotent() {
+ 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-clear-defaults".to_string()
+ };
+ let file_io = FileIOBuilder::new(scheme).build().unwrap();
+ let manager = ConsumerManager::new(file_io.clone(),
table_path.clone());
+ manager.clear(None, None).await.unwrap();
+ assert!(!file_io
+ .exists_dir(&format!("{table_path}/consumer"))
+ .await
+ .unwrap());
+ for blank in [None, Some(""), Some(" \t\r\n"),
Some("\u{001c}\u{3000}")] {
+ manager.reset("job", 1).await.unwrap();
+ manager.clear(blank, Some("job")).await.unwrap();
+ assert_eq!(manager.get("job").await.unwrap(), Some(1));
+ manager.clear(blank, blank).await.unwrap();
+ assert_eq!(manager.get("job").await.unwrap(), None);
+ manager.clear(blank, blank).await.unwrap();
+ }
+ manager.reset("job", 1).await.unwrap();
+ for nonblank in [
+ "\u{0085}", "\u{00a0}", "\u{2007}", "\u{202f}", " job", "job ",
+ ] {
+ manager.clear(Some(nonblank), None).await.unwrap();
+ assert_eq!(manager.get("job").await.unwrap(), Some(1));
+ }
+ manager.clear(None, Some("\u{00a0}")).await.unwrap();
+ assert_eq!(manager.get("job").await.unwrap(), None);
+ }
+ }
+
+ #[test]
+ fn consumer_patterns_match_entire_ids_using_rust_regex_syntax() {
+ for (pattern, id, expected) in [
+ ("job", "job-old", false),
+ ("job", "pre-job", false),
+ ("a|ab", "ab", true),
+ ("job|other", "job-extra", false),
+ ("(?m)^job$", "job\nother", false),
+ ("(?s)job.*", "job\nother", true),
+ ("(?i)job", "JOB", true),
+ ("(?x) job # trailing comment", "job", true),
+ ("(?x) job # trailing comment", "job\n", false),
+ (r"\w+", "\u{4efb}\u{52a1}", true),
+ (r"(?-u:\w+)", "\u{4efb}\u{52a1}", false),
+ ("job.*", "job\rold", true),
+ (".*", "job\nold", false),
+ (" job", " job", true),
+ ] {
+ let regex = consumer_pattern(pattern,
"including_consumers").unwrap();
+ assert_eq!(regex.is_match(id), expected, "{pattern:?}: {id:?}");
+ }
+ }
+
+ #[tokio::test]
+ async fn clear_validates_both_patterns_before_deleting_any_files() {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let manager = ConsumerManager::new(file_io,
"memory:/consumer-clear-invalid".into());
+ for pattern in ["[", r"(job)\1", "(?=job)job", ")|(", "\\"] {
+ for (including, excluding, argument) in [
+ (Some(pattern), None, "including_consumers"),
+ (None, Some(pattern), "excluding_consumers"),
+ (Some("no-match"), Some(pattern), "excluding_consumers"),
+ ] {
+ manager.reset("job", 1).await.unwrap();
+ let error = manager.clear(including,
excluding).await.unwrap_err();
+ assert!(error
+ .to_string()
+ .contains(&format!("Invalid {argument} regex")));
+ assert_eq!(manager.get("job").await.unwrap(), Some(1));
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn clear_propagates_listing_and_deletion_errors() {
+ #[derive(Debug)]
+ struct DeniedProvider {
+ operator: opendal::Operator,
+ deny_listing: bool,
+ }
+
+ #[async_trait::async_trait]
+ impl FileIOProvider for DeniedProvider {
+ async fn create(&self, path: &str) ->
crate::Result<(opendal::Operator, String)> {
+ if self.deny_listing || path.ends_with("/consumer-job") {
+ return Err(opendal::Error::new(
+ opendal::ErrorKind::PermissionDenied,
+ "access denied",
+ )
+ .into());
+ }
+ Ok((
+ self.operator.clone(),
+ path.strip_prefix("memory:/").unwrap().to_string(),
+ ))
+ }
+ }
+
+ let operator =
opendal::Operator::new(opendal::services::Memory::default()).unwrap();
+ operator
+ .write("table/consumer/consumer-job", "{")
+ .await
+ .unwrap();
+ for deny_listing in [true, false] {
+ let file_io = FileIOBuilder::new("memory")
+ .build()
+ .unwrap()
+ .with_provider(Arc::new(DeniedProvider {
+ operator: operator.clone(),
+ deny_listing,
+ }));
+ let manager = ConsumerManager::new(file_io,
"memory:/table".into());
+ let error = manager.clear(None, None).await.unwrap_err();
+ assert!(matches!(error, crate::Error::IoUnexpected { source, .. }
+ if source.kind() == opendal::ErrorKind::PermissionDenied));
+ assert!(operator
+ .exists("table/consumer/consumer-job")
+ .await
+ .unwrap());
+ }
+ }
+
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn clear_uses_listed_paths_without_aliasing_excluded_consumers() {
+ let tmp = tempfile::tempdir().unwrap();
+ let directory = tmp.path().join("consumer");
+ std::fs::create_dir(&directory).unwrap();
+ let manager = ConsumerManager::new(
+ FileIOBuilder::new("file").build().unwrap(),
+ format!("file://{}", tmp.path().display()),
+ );
+ std::fs::write(directory.join(r"consumer-job\part"), "{").unwrap();
+ manager.clear(Some(r"job\\part"), None).await.unwrap();
+ assert!(!directory.join(r"consumer-job\part").exists());
+
+ std::fs::write(directory.join("consumer-job"), "keep").unwrap();
+ for suffix in [" ", "\t", "\u{00a0}", "\u{3000}"] {
+ let name = format!("consumer-job{suffix}");
+ std::fs::write(directory.join(&name), "keep alias").unwrap();
+ // The filesystem lister normalizes these names to "consumer-job".
+ manager
+ .clear(Some(&format!("job{suffix}")), Some("job"))
+ .await
+ .unwrap();
+ assert_eq!(
+ std::fs::read(directory.join("consumer-job")).unwrap(),
+ b"keep"
+ );
+ assert_eq!(std::fs::read(directory.join(&name)).unwrap(), b"keep
alias");
+ }
+ manager.clear(None, Some("job")).await.unwrap();
+ assert_eq!(
+ std::fs::read(directory.join("consumer-job")).unwrap(),
+ b"keep"
+ );
+ }
+
+ #[tokio::test]
+ async fn clear_rejects_literal_whitespace_paths_before_any_deletion() {
+ use opendal::raw::{oio::Write, OpWrite, Service};
+
+ let operator =
opendal::Operator::new(opendal::services::Memory::default()).unwrap();
+ let file_io = FileIOBuilder::new("file")
+ .with_fs_operator(operator.clone())
+ .build()
+ .unwrap();
+ let manager = ConsumerManager::new(file_io, "file:///table".into());
+ operator
+ .write("table/consumer/consumer-job", "keep")
+ .await
+ .unwrap();
+ operator
+ .write("table/consumer/consumer-delete", "keep")
+ .await
+ .unwrap();
+ for suffix in [" ", "\t", "\u{00a0}", "\u{3000}"] {
+ let path = format!("table/consumer/consumer-job{suffix}");
+ // Seed a literal object key without Operator's input path
normalization.
+ let mut writer = operator
+ .service()
+ .write(operator.context(), &path, OpWrite::default())
+ .unwrap();
+ writer.write("keep alias".into()).await.unwrap();
+ writer.close().await.unwrap();
+ let error = manager
+ .clear(Some(&format!("job{suffix}")), Some("job"))
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("trailing whitespace"));
+ assert!(manager.clear(None, Some("job")).await.is_err());
+ assert_eq!(
+ operator
+ .read("table/consumer/consumer-job")
+ .await
+ .unwrap()
+ .to_bytes(),
+ "keep"
+ );
+ assert_eq!(
+ operator
+ .read("table/consumer/consumer-delete")
+ .await
+ .unwrap()
+ .to_bytes(),
+ "keep"
+ );
+ assert!(operator
+ .list("table/consumer/")
+ .await
+ .unwrap()
+ .iter()
+ .any(|entry| entry.path() == path));
+ }
+ }
+
#[tokio::test]
async fn delete_is_idempotent_and_preserves_other_consumers_and_branches()
{
for scheme in ["memory", "file"] {
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 13f79fb9..82553172 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1098,6 +1098,62 @@ have already loaded their progress do not switch
positions when the record chang
A running consumer's checkpoint completion can overwrite or recreate the
record,
so stop the consumer before adjusting its progress manually.
+### clear_consumers
+
+Delete a group of consumer records while retaining exceptions:
+
+```sql
+CALL sys.clear_consumers(
+ table => 'paimon.test_db.events',
+ including_consumers => 'pipeline_.*',
+ excluding_consumers => 'pipeline_critical|pipeline_backup'
+);
+```
+
+Both expressions match the **entire consumer ID**, excluding the `consumer-`
filename
+prefix. A record is deleted only when inclusion matches and exclusion does not.
+For example, `pipeline_a` does not match `pipeline_a_backup`; use
`pipeline_a.*`
+to match that suffix. Alternation and inline multiline flags do not bypass the
+whole-ID requirement.
+
+Patterns use [Rust `regex` syntax](https://docs.rs/regex/1/regex/#syntax),
including
+groups, alternation, character classes, repetition, Unicode classes, and inline
+flags such as `(?i)`, `(?m)`, `(?s)`, and `(?x)`. Look-around and
backreferences are
+not supported. This is not the Java Pattern dialect: for example, `\w`, `\d`,
and
+`\s` are Unicode-aware by default, and `.` excludes LF but matches CR. Use
explicit
+classes or flags when those distinctions matter. Invalid expressions fail
before
+any files are deleted, including an invalid exclusion when inclusion matches
nothing.
+
+Omitting `including_consumers`, or passing an empty or Java-whitespace-only
string,
+uses `.*`. The equivalent exclusion means no exclusions. Nonbreaking spaces
such
+as U+00A0 are literal patterns, not blank placeholders. Nonblank patterns are
never
+trimmed. The default `.*` does not match IDs containing LF; use `(?s).*` if
needed.
+To clear all ordinary consumer IDs:
+
+```sql
+CALL sys.clear_consumers(table => 'paimon.test_db.events');
+```
+
+The procedure uses the same table resolution and branch-scoped manager as
+`reset_consumer`; it does not add branch-qualified name resolution to catalogs.
+REST-managed tables and format tables are not supported. Arguments require
named
+`=>` syntax; explicit `NULL` and positional arguments are not supported.
Successful
+calls return `OK`, including when the consumer directory is absent or nothing
matches.
+
+Only filenames are inspected, so corrupt consumer JSON does not prevent
cleanup.
+Directories and unrelated files are skipped. A matching listed path with
trailing
+whitespace is rejected before deletion because OpenDAL could otherwise
normalize it
+to a different consumer's path. Filenames are taken from the storage listing;
the
+filesystem backend normalizes trailing whitespace before returning entries.
+Retained consumer records, table data, snapshots, and manifests are unchanged.
+Storage errors are propagated; deletion is not atomic and an error can leave
some
+matching records already removed.
+
+This operation does not determine whether a consumer has stopped. Stop the
target
+consumers before cleanup: running scans retain their loaded positions, and
checkpoint
+completion can recreate deleted records. A fresh scan without an explicit
checkpoint
+follows its existing initial scan behavior after its consumer record is
deleted.
+
### create_tag
Create a named tag from a snapshot:
