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 cdab14d4 feat(datafusion): add MSCK REPAIR TABLE for catalog-managed
format tables (#817)
cdab14d4 is described below
commit cdab14d4902b4ee2c785e5c23792144efa0cbcdb
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Sun Sep 13 19:56:34 2026 +0800
feat(datafusion): add MSCK REPAIR TABLE for catalog-managed format tables
(#817)
---
.../datafusion/src/format_partition_repair.rs | 179 +++++++++++++++++++++
crates/integrations/datafusion/src/lib.rs | 1 +
crates/integrations/datafusion/src/sql_context.rs | 2 +
.../datafusion/tests/rest_format_partition_sql.rs | 109 +++++++++++++
crates/paimon/src/spec/mod.rs | 2 +-
crates/paimon/src/spec/partition_utils.rs | 39 +++++
crates/paimon/src/table/format_partition.rs | 121 +++++++++++++-
crates/paimon/src/table/format_table_scan.rs | 36 +----
crates/paimon/tests/format_partition_test.rs | 132 +++++++++++++++
docs/src/sql.md | 19 ++-
10 files changed, 604 insertions(+), 36 deletions(-)
diff --git a/crates/integrations/datafusion/src/format_partition_repair.rs
b/crates/integrations/datafusion/src/format_partition_repair.rs
new file mode 100644
index 00000000..7b097670
--- /dev/null
+++ b/crates/integrations/datafusion/src/format_partition_repair.rs
@@ -0,0 +1,179 @@
+// 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.
+
+//! MSCK REPAIR TABLE for Format Tables with catalog-managed partitions.
+
+use std::collections::{BTreeMap, HashMap, HashSet};
+
+use datafusion::error::{DataFusionError, Result as DFResult};
+use datafusion::prelude::DataFrame;
+use datafusion::sql::sqlparser::ast::{AddDropSync, Msck};
+use paimon::catalog::{Catalog, Identifier};
+use paimon::spec::CoreOptions;
+use paimon::table::{FormatTablePartitionPaths, Table};
+
+use crate::error::to_datafusion_error;
+use crate::format_partition_ddl::ensure_catalog_managed_format_table;
+use crate::sql_context::{ok_result, SQLContext};
+
+pub(crate) async fn execute_msck(ctx: &SQLContext, msck: &Msck) ->
DFResult<DataFrame> {
+ if !msck.repair {
+ return Err(DataFusionError::Plan(
+ "MSCK requires the REPAIR keyword".to_string(),
+ ));
+ }
+ SQLContext::ensure_partition_command_target(&msck.table_name, "MSCK REPAIR
TABLE")?;
+ let (catalog, _catalog_name, identifier) =
ctx.resolve_catalog_and_table(&msck.table_name)?;
+ let table = catalog
+ .get_table(&identifier)
+ .await
+ .map_err(to_datafusion_error)?;
+ ensure_catalog_managed_format_table(&table, "MSCK REPAIR TABLE")?;
+ let mode = match msck.partition_action {
+ None | Some(AddDropSync::ADD) => RepairMode::Add,
+ Some(AddDropSync::DROP) => RepairMode::Drop,
+ Some(AddDropSync::SYNC) => RepairMode::Sync,
+ };
+ repair(catalog.as_ref(), &identifier, &table, mode)
+ .await
+ .map_err(to_datafusion_error)?;
+ ok_result(ctx.ctx())
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum RepairMode {
+ Add,
+ Drop,
+ Sync,
+}
+
+async fn repair(
+ catalog: &dyn Catalog,
+ identifier: &Identifier,
+ table: &Table,
+ mode: RepairMode,
+) -> paimon::Result<()> {
+ let core_options = CoreOptions::new(table.schema().options());
+ let partition_paths = FormatTablePartitionPaths::new(
+ table.schema().partition_keys().iter().cloned(),
+ core_options.format_table_partition_only_value_in_path(),
+ );
+ let table_path = table.location();
+
+ if matches!(mode, RepairMode::Drop | RepairMode::Sync) {
+ // Discovery reads a missing root as empty, which would unregister
every partition.
+ ensure_table_root_is_directory(table).await?;
+ }
+
+ // Load both views before changing catalog metadata so a listing failure
leaves
+ // metadata unchanged. Discovery preserves raw directory values such as
month=01.
+ let discovered_specs = partition_paths
+ .discover(
+ table.file_io(),
+ table_path,
+ core_options.partition_default_name(),
+ )
+ .await?;
+ let registered_partitions = catalog.list_partitions(identifier).await?;
+ // A partition registered at a location of its own is never discovered
under the table
+ // directory, so repair leaves it registered.
+ let custom_located = registered_partitions
+ .iter()
+ .filter(|partition| {
+ partition
+ .options
+ .as_ref()
+ .is_some_and(|options| options.contains_key("path"))
+ })
+ .map(|partition| partition_paths.partition_name(&partition.spec))
+ .collect::<paimon::Result<HashSet<_>>>()?;
+
+ let discovered_by_name = index_specs_by_name(&partition_paths,
discovered_specs)?;
+ let registered_by_name = index_specs_by_name(
+ &partition_paths,
+ registered_partitions
+ .into_iter()
+ .map(|partition| partition.spec)
+ .collect(),
+ )?;
+
+ let to_register = if matches!(mode, RepairMode::Add | RepairMode::Sync) {
+ discovered_by_name
+ .iter()
+ .filter(|(name, _)| !registered_by_name.contains_key(*name))
+ .map(|(_, spec)| spec.clone())
+ .collect::<Vec<_>>()
+ } else {
+ Vec::new()
+ };
+ let to_unregister = if matches!(mode, RepairMode::Drop | RepairMode::Sync)
{
+ registered_by_name
+ .iter()
+ .filter(|(name, _)| {
+ !discovered_by_name.contains_key(*name) &&
!custom_located.contains(*name)
+ })
+ .map(|(_, spec)| spec.clone())
+ .collect::<Vec<_>>()
+ } else {
+ Vec::new()
+ };
+
+ if !to_register.is_empty() {
+ catalog
+ .create_partitions(identifier, to_register, true)
+ .await?;
+ }
+ if !to_unregister.is_empty() {
+ catalog.drop_partitions(identifier, to_unregister).await?;
+ }
+ Ok(())
+}
+
+/// Fail unless the table root is a directory; on object stores, a root
holding objects counts.
+async fn ensure_table_root_is_directory(table: &Table) -> paimon::Result<()> {
+ let file_io = table.file_io();
+ let root = table.location();
+ let problem = match file_io.get_status(root).await {
+ Ok(status) if status.is_dir => return Ok(()),
+ Ok(_) => "is not a directory",
+ // A store without directory entries cannot stat the root but lists
what it holds.
+ Err(_) => {
+ if !file_io.list_status(root).await?.is_empty() {
+ return Ok(());
+ }
+ "does not exist"
+ }
+ };
+ Err(paimon::Error::DataInvalid {
+ message: format!(
+ "MSCK REPAIR TABLE cannot drop partitions of Format Table {}: \
+ its location {root} {problem}",
+ table.identifier().full_name()
+ ),
+ source: None,
+ })
+}
+
+fn index_specs_by_name(
+ partition_paths: &FormatTablePartitionPaths,
+ specs: Vec<HashMap<String, String>>,
+) -> paimon::Result<BTreeMap<String, HashMap<String, String>>> {
+ specs
+ .into_iter()
+ .map(|spec| Ok((partition_paths.partition_name(&spec)?, spec)))
+ .collect()
+}
diff --git a/crates/integrations/datafusion/src/lib.rs
b/crates/integrations/datafusion/src/lib.rs
index 4ec76c45..dca1d409 100644
--- a/crates/integrations/datafusion/src/lib.rs
+++ b/crates/integrations/datafusion/src/lib.rs
@@ -44,6 +44,7 @@ mod delete;
mod error;
mod filter_pushdown;
mod format_partition_ddl;
+mod format_partition_repair;
#[cfg(feature = "fulltext")]
mod full_text_search;
mod hybrid_search;
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index b75534e9..09cf9764 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -37,6 +37,7 @@
//! - `ALTER TABLE db.t ADD [IF NOT EXISTS] PARTITION (...) [PARTITION (...)]`
//! - `ALTER TABLE db.t DROP [IF EXISTS] PARTITION (...)`
//! - `SHOW PARTITIONS db.t [PARTITION (...)]`
+//! - `MSCK REPAIR TABLE db.t [{ADD|DROP|SYNC} PARTITIONS]`
//! - `CREATE VIEW [IF NOT EXISTS] view [(col, ...)] AS query`
//! - `DROP VIEW [IF EXISTS] view`
//! - `CREATE FUNCTION name(args) RETURNS type [LANGUAGE SQL] RETURN
expression`
@@ -609,6 +610,7 @@ impl SQLContext {
self.handle_truncate_table(truncate,
enable_ident_normalization)
.await
}
+ Statement::Msck(msck) =>
crate::format_partition_repair::execute_msck(self, msck).await,
Statement::CreateView(create_view) => {
if create_view.temporary {
// Temporary views are always handled by us (Paimon
catalog temp storage)
diff --git a/crates/integrations/datafusion/tests/rest_format_partition_sql.rs
b/crates/integrations/datafusion/tests/rest_format_partition_sql.rs
index 27005727..9a3073f7 100644
--- a/crates/integrations/datafusion/tests/rest_format_partition_sql.rs
+++ b/crates/integrations/datafusion/tests/rest_format_partition_sql.rs
@@ -484,6 +484,115 @@ async fn
test_drop_partition_leaves_a_custom_location_in_place() {
assert!(temp_dir.path().join("dt=b").is_dir());
}
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_msck_repair_reconciles_registrations_with_directories() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ std::fs::create_dir_all(temp_dir.path().join("dt=2026-07-21")).unwrap();
+ let (_server, context) =
+ setup_rest_table(&temp_dir, format_table_schema(&[("dt",
varchar())])).await;
+ common::exec(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} ADD PARTITION (dt = '2026-07-22')"),
+ )
+ .await;
+
+ // ADD registers a directory the catalog does not know yet.
+ common::exec(
+ &context,
+ &format!("MSCK REPAIR TABLE {TABLE_NAME} ADD PARTITIONS"),
+ )
+ .await;
+ assert_eq!(
+ show_partitions(&context, "").await,
+ ["dt=2026-07-21", "dt=2026-07-22"]
+ );
+
+ // SYNC also unregisters a partition whose directory is gone, without
deleting anything.
+ std::fs::remove_dir_all(temp_dir.path().join("dt=2026-07-22")).unwrap();
+ common::exec(
+ &context,
+ &format!("MSCK REPAIR TABLE {TABLE_NAME} SYNC PARTITIONS"),
+ )
+ .await;
+ assert_eq!(show_partitions(&context, "").await, ["dt=2026-07-21"]);
+ assert!(temp_dir.path().join("dt=2026-07-21").is_dir());
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_msck_repair_refuses_to_drop_without_a_table_directory() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let (server, context) =
+ setup_rest_table(&temp_dir, format_table_schema(&[("dt",
varchar())])).await;
+ common::exec(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} ADD PARTITION (dt = 'a')"),
+ )
+ .await;
+
+ // A moved root must not read as every partition directory gone.
+ let moved_dir = tempfile::tempdir().unwrap();
+ std::fs::rename(temp_dir.path(), moved_dir.path().join("events")).unwrap();
+ for action in ["DROP", "SYNC"] {
+ common::assert_sql_error(
+ &context,
+ &format!("MSCK REPAIR TABLE {TABLE_NAME} {action} PARTITIONS"),
+ "does not exist",
+ )
+ .await;
+ }
+
+ std::fs::write(temp_dir.path(), b"").unwrap();
+ for action in ["DROP", "SYNC"] {
+ common::assert_sql_error(
+ &context,
+ &format!("MSCK REPAIR TABLE {TABLE_NAME} {action} PARTITIONS"),
+ "is not a directory",
+ )
+ .await;
+ }
+ assert_eq!(
+ server.table_partition_specs(DATABASE, TABLE),
+ vec![spec(&[("dt", "a")])]
+ );
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_msck_repair_keeps_a_partition_at_a_custom_location() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let external_dir = tempfile::tempdir().unwrap();
+ let (server, context) =
+ setup_rest_table(&temp_dir, format_table_schema(&[("dt",
varchar())])).await;
+ common::exec(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} ADD PARTITION (dt = 'a') PARTITION
(dt = 'b')"),
+ )
+ .await;
+ server.set_table_partition_options(
+ DATABASE,
+ TABLE,
+ &spec(&[("dt", "b")]),
+ HashMap::from([(
+ "path".to_string(),
+ format!("file://{}", external_dir.path().display()),
+ )]),
+ );
+
+ // Its directory is not under the table, so repair does not read it as
missing.
+ std::fs::remove_dir_all(temp_dir.path().join("dt=b")).unwrap();
+ common::exec(
+ &context,
+ &format!("MSCK REPAIR TABLE {TABLE_NAME} SYNC PARTITIONS"),
+ )
+ .await;
+ assert_eq!(
+ server.table_partition_specs(DATABASE, TABLE),
+ vec![spec(&[("dt", "a")]), spec(&[("dt", "b")])]
+ );
+}
+
/// `SQLContext::sql` futures have to stay `Send` for callers that box or
spawn them; this stops
/// compiling when a stream over borrowed items anywhere below a statement
takes that away.
#[allow(dead_code)]
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index efc0634e..3a35fd20 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -99,7 +99,7 @@ mod partition;
pub use partition::Partition;
mod partition_utils;
pub(crate) use partition_utils::{
- bucket_path, bucket_path_under, escape_path_name, PartitionComputer,
+ bucket_path, bucket_path_under, escape_path_name, unescape_path_name,
PartitionComputer,
};
mod predicate;
pub(crate) use predicate::datum_cmp;
diff --git a/crates/paimon/src/spec/partition_utils.rs
b/crates/paimon/src/spec/partition_utils.rs
index 463ba653..69e1941d 100644
--- a/crates/paimon/src/spec/partition_utils.rs
+++ b/crates/paimon/src/spec/partition_utils.rs
@@ -522,6 +522,37 @@ pub(crate) fn escape_path_name(path: &str) -> String {
sb
}
+/// Unescape a path component following Java
`PartitionPathUtils.unescapePathName`.
+pub(crate) fn unescape_path_name(value: &str) -> Option<String> {
+ let bytes = value.as_bytes();
+ let mut out = Vec::with_capacity(bytes.len());
+ let mut i = 0;
+ while i < bytes.len() {
+ if bytes[i] == b'%' {
+ if i + 2 >= bytes.len() {
+ return None;
+ }
+ let hi = hex_value(bytes[i + 1])?;
+ let lo = hex_value(bytes[i + 2])?;
+ out.push((hi << 4) | lo);
+ i += 3;
+ } else {
+ out.push(bytes[i]);
+ i += 1;
+ }
+ }
+ String::from_utf8(out).ok()
+}
+
+fn hex_value(byte: u8) -> Option<u8> {
+ match byte {
+ b'0'..=b'9' => Some(byte - b'0'),
+ b'a'..=b'f' => Some(byte - b'a' + 10),
+ b'A'..=b'F' => Some(byte - b'A' + 10),
+ _ => None,
+ }
+}
+
/// Check if a character needs escaping in partition path names.
///
/// Matches Java `PartitionPathUtils.CHAR_TO_ESCAPE`:
@@ -691,6 +722,14 @@ mod tests {
assert_eq!(escape_path_name("a\x7Fb"), "a%7Fb");
}
+ #[test]
+ fn test_unescape_path_name() {
+ assert_eq!(unescape_path_name("a%2Fb"), Some("a/b".to_string()));
+ assert_eq!(unescape_path_name("%E4%B8%AD"), Some("中".to_string()));
+ assert_eq!(unescape_path_name("a%ZZb"), None);
+ assert_eq!(unescape_path_name("a%b"), None);
+ }
+
// ======================== PartitionComputer tests
========================
#[test]
diff --git a/crates/paimon/src/table/format_partition.rs
b/crates/paimon/src/table/format_partition.rs
index deba4111..0ee14316 100644
--- a/crates/paimon/src/table/format_partition.rs
+++ b/crates/paimon/src/table/format_partition.rs
@@ -22,7 +22,8 @@ use std::collections::HashMap;
use chrono::NaiveDate;
-use crate::spec::{escape_path_name, DataType, Datum};
+use crate::io::FileIO;
+use crate::spec::{escape_path_name, unescape_path_name, DataType, Datum};
const UNIX_EPOCH_DAYS_FROM_CE: i32 = 719_163;
@@ -110,6 +111,60 @@ impl FormatTablePartitionPaths {
.join("/"))
}
+ /// Discover complete raw partition specs from the table directory, sorted
and deduplicated.
+ /// Skips hidden or non-matching entries; a malformed or non-canonical
segment is an error.
+ pub async fn discover(
+ &self,
+ file_io: &FileIO,
+ table_path: &str,
+ default_partition_name: &str,
+ ) -> crate::Result<Vec<HashMap<String, String>>> {
+ let default_partition_path_name = self
+ .only_value_in_path
+ .then(|| escape_path_name(default_partition_name));
+ let mut frontier = vec![(table_path.trim_end_matches('/').to_string(),
HashMap::new())];
+ for key in &self.partition_keys {
+ let mut next = Vec::new();
+ for (path, spec) in frontier {
+ let statuses = match file_io.list_status(&path).await {
+ Ok(statuses) => statuses,
+ Err(error) if is_storage_not_found(&error) => continue,
+ Err(error) => return Err(error),
+ };
+ for status in statuses {
+ if !status.is_dir {
+ continue;
+ }
+ let Some(segment) = last_path_segment(&status.path) else {
+ continue;
+ };
+ let is_value_only_default =
+ default_partition_path_name.as_deref() ==
Some(segment);
+ if (segment.starts_with('.') || segment.starts_with('_'))
+ && !is_value_only_default
+ {
+ continue;
+ }
+ let Some(value) = self.partition_value_from_segment(key,
segment)? else {
+ continue;
+ };
+ let mut child_spec = spec.clone();
+ child_spec.insert(key.clone(), value);
+ next.push((status.path.trim_end_matches('/').to_string(),
child_spec));
+ }
+ }
+ frontier = next;
+ }
+
+ let mut partitions = frontier
+ .into_iter()
+ .map(|(_, spec)| Ok((self.partition_name(&spec)?, spec)))
+ .collect::<crate::Result<Vec<_>>>()?;
+ partitions.sort_by(|left, right| left.0.cmp(&right.0));
+ partitions.dedup_by(|left, right| left.0 == right.0);
+ Ok(partitions.into_iter().map(|(_, spec)| spec).collect())
+ }
+
fn ordered_values<'a>(&self, spec: &'a HashMap<String, String>) ->
crate::Result<Vec<&'a str>> {
if spec.len() != self.partition_keys.len() {
return Err(crate::Error::DataInvalid {
@@ -147,6 +202,44 @@ impl FormatTablePartitionPaths {
self.partition_keys
)
}
+
+ fn partition_value_from_segment(
+ &self,
+ key: &str,
+ segment: &str,
+ ) -> crate::Result<Option<String>> {
+ let value = if self.only_value_in_path {
+ decode_canonical_path_name(segment)?
+ } else {
+ let Some((segment_key, value)) = segment.split_once('=') else {
+ return Ok(None);
+ };
+ let decoded_key = decode_canonical_path_name(segment_key)?;
+ if decoded_key != key {
+ return Ok(None);
+ }
+ decode_canonical_path_name(value)?
+ };
+ Ok(Some(value))
+ }
+}
+
+fn decode_canonical_path_name(value: &str) -> crate::Result<String> {
+ let decoded = unescape_path_name(value).ok_or_else(||
crate::Error::DataInvalid {
+ message: format!("Invalid escaped partition path segment {value:?}"),
+ source: None,
+ })?;
+ let canonical = escape_path_name(&decoded);
+ if canonical != value {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Partition path segment {value:?} cannot round-trip through
catalog metadata; \
+ its canonical escaped form is {canonical:?}"
+ ),
+ source: None,
+ });
+ }
+ Ok(decoded)
}
/// Parse a raw Format Table partition value from a path or catalog
registration.
@@ -232,6 +325,18 @@ fn format_partition_date(epoch_days: i32) ->
Option<String> {
.map(|date| date.format("%Y-%m-%d").to_string())
}
+fn is_storage_not_found(error: &crate::Error) -> bool {
+ matches!(
+ error,
+ crate::Error::IoUnexpected { source, .. }
+ if source.kind() == opendal::ErrorKind::NotFound
+ )
+}
+
+fn last_path_segment(path: &str) -> Option<&str> {
+ path.trim_end_matches('/').rsplit('/').next()
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -349,4 +454,18 @@ mod tests {
// rather than silently widened.
assert_eq!(paths.name_prefix_pattern(&["2026/07".to_string()]), None);
}
+
+ #[test]
+ fn test_storage_not_found_matches_only_not_found() {
+ for (kind, expected) in [
+ (opendal::ErrorKind::NotFound, true),
+ (opendal::ErrorKind::PermissionDenied, false),
+ ] {
+ let error = crate::Error::IoUnexpected {
+ message: "list partition directory".to_string(),
+ source: Box::new(opendal::Error::new(kind, "test")),
+ };
+ assert_eq!(is_storage_not_found(&error), expected);
+ }
+ }
}
diff --git a/crates/paimon/src/table/format_table_scan.rs
b/crates/paimon/src/table/format_table_scan.rs
index aabadcec..de2c02e0 100644
--- a/crates/paimon/src/table/format_table_scan.rs
+++ b/crates/paimon/src/table/format_table_scan.rs
@@ -26,9 +26,9 @@ use super::{Plan, RESTEnv, ScanTrace, Table};
use crate::api::RestError;
use crate::spec::stats::BinaryTableStats;
use crate::spec::{
- escape_path_name, extract_datum, BinaryRow, BinaryRowBuilder, CoreOptions,
DataField,
- DataFileMeta, DataType, Datum, Partition, PartitionComputer, Predicate,
PredicateOperator,
- PATH_OPTION,
+ escape_path_name, extract_datum, unescape_path_name, BinaryRow,
BinaryRowBuilder, CoreOptions,
+ DataField, DataFileMeta, DataType, Datum, Partition, PartitionComputer,
Predicate,
+ PredicateOperator, PATH_OPTION,
};
use crate::table::partition_filter::PartitionFilter;
use crate::table::source::{DataSplitBuilder, RowRange};
@@ -831,36 +831,6 @@ fn partition_segment_value(segment: &str, key: &str) ->
Option<String> {
}
}
-fn unescape_path_name(value: &str) -> Option<String> {
- let bytes = value.as_bytes();
- let mut out = Vec::with_capacity(bytes.len());
- let mut i = 0;
- while i < bytes.len() {
- if bytes[i] == b'%' {
- if i + 2 >= bytes.len() {
- return None;
- }
- let hi = hex_value(bytes[i + 1])?;
- let lo = hex_value(bytes[i + 2])?;
- out.push((hi << 4) | lo);
- i += 3;
- } else {
- out.push(bytes[i]);
- i += 1;
- }
- }
- String::from_utf8(out).ok()
-}
-
-fn hex_value(byte: u8) -> Option<u8> {
- match byte {
- b'0'..=b'9' => Some(byte - b'0'),
- b'a'..=b'f' => Some(byte - b'a' + 10),
- b'A'..=b'F' => Some(byte - b'A' + 10),
- _ => None,
- }
-}
-
fn supported_format_table_formats() -> Vec<&'static str> {
vec![
"parquet",
diff --git a/crates/paimon/tests/format_partition_test.rs
b/crates/paimon/tests/format_partition_test.rs
new file mode 100644
index 00000000..d9ad49ca
--- /dev/null
+++ b/crates/paimon/tests/format_partition_test.rs
@@ -0,0 +1,132 @@
+// 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.
+
+use std::collections::HashMap;
+#[cfg(not(windows))]
+use std::path::Path;
+
+use paimon::io::FileIO;
+use paimon::table::FormatTablePartitionPaths;
+
+#[cfg(not(windows))]
+async fn discover_partitions(
+ root: &Path,
+ partition_keys: &[&str],
+ only_value_in_path: bool,
+ default_partition_name: &str,
+) -> paimon::Result<Vec<HashMap<String, String>>> {
+ let table_path = format!("file://{}", root.display());
+ let file_io = FileIO::from_path(&table_path)?.build()?;
+ FormatTablePartitionPaths::new(partition_keys.iter().copied(),
only_value_in_path)
+ .discover(&file_io, &table_path, default_partition_name)
+ .await
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn discover_format_partitions_returns_complete_sorted_specs() {
+ let tmp = tempfile::tempdir().unwrap();
+ for relative in [
+ "dt=2026-07-22/hour=10",
+ "dt=2026-07-21/hour=09",
+ "dt=2026-07-20",
+ ".staging/dt=2026-07-19/hour=08",
+ ] {
+ std::fs::create_dir_all(tmp.path().join(relative)).unwrap();
+ }
+
+ let partitions =
+ discover_partitions(tmp.path(), &["dt", "hour"], false,
"__DEFAULT_PARTITION__")
+ .await
+ .unwrap();
+
+ assert_eq!(
+ partitions,
+ vec![
+ HashMap::from([
+ ("dt".to_string(), "2026-07-21".to_string()),
+ ("hour".to_string(), "09".to_string()),
+ ]),
+ HashMap::from([
+ ("dt".to_string(), "2026-07-22".to_string()),
+ ("hour".to_string(), "10".to_string()),
+ ]),
+ ]
+ );
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn discover_format_partitions_treats_missing_root_as_empty() {
+ let tmp = tempfile::tempdir().unwrap();
+
+ let partitions = discover_partitions(
+ &tmp.path().join("missing"),
+ &["dt"],
+ false,
+ "__DEFAULT_PARTITION__",
+ )
+ .await
+ .unwrap();
+
+ assert!(partitions.is_empty());
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn discover_value_only_partitions_rejects_parent_traversal_value() {
+ let tmp = tempfile::tempdir().unwrap();
+ std::fs::create_dir_all(tmp.path().join("%2E%2E")).unwrap();
+
+ let error = discover_partitions(tmp.path(), &["dt"], true,
"__DEFAULT_PARTITION__")
+ .await
+ .unwrap_err();
+
+ assert!(
+ error.to_string().contains(".."),
+ "expected traversal value in error, got: {error}"
+ );
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn discover_value_only_partitions_keeps_hidden_default_directory() {
+ for (default_name, directory, ignored_directory) in [
+ ("__DEFAULT_PARTITION__", "__DEFAULT_PARTITION__", None),
+ (".NULL", ".NULL", Some(".staging")),
+ ("_NULL/%NA", "_NULL%2F%25NA", Some("_temporary")),
+ ] {
+ let tmp = tempfile::tempdir().unwrap();
+ std::fs::create_dir_all(tmp.path().join(directory)).unwrap();
+ if let Some(ignored_directory) = ignored_directory {
+
std::fs::create_dir_all(tmp.path().join(ignored_directory)).unwrap();
+ }
+
+ let partitions = discover_partitions(tmp.path(), &["dt"], true,
default_name)
+ .await
+ .unwrap();
+
+ assert_eq!(
+ partitions,
+ vec![HashMap::from([(
+ "dt".to_string(),
+ default_name.to_string()
+ )])],
+ "{default_name}"
+ );
+ }
+}
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 88e23f68..75ac8f69 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -40,7 +40,7 @@ Mosaic support is always available and currently read-only.
SQL queries can read
SQL support has two layers:
- DataFusion provides the parser, query planner, optimizer, execution engine,
expressions, scalar functions, aggregate functions, and window functions. SQL
statements that `SQLContext` does not intercept are delegated to DataFusion.
This includes the DataFusion SQL surface for `SELECT` queries, CTEs (including
recursive CTEs), subqueries, joins including `LATERAL` joins, SQL lambda
functions, grouping, `HAVING`, window clauses, `QUALIFY`, set operations,
`ORDER BY`, `LIMIT`/`OFFSET`, `EX [...]
-- Paimon-specific table management and row-level writes are implemented by
`SQLContext`. This includes Paimon `CREATE TABLE`, `ALTER TABLE`, `DROP TABLE`,
`CREATE TEMPORARY TABLE`, `CREATE TEMPORARY VIEW`, REST Catalog persistent
`CREATE VIEW`, `DROP VIEW`, and `CREATE FUNCTION`, `DROP TEMPORARY TABLE` /
`VIEW`, `INSERT OVERWRITE ... PARTITION`, `UPDATE`, `DELETE`, `MERGE INTO`,
`TRUNCATE TABLE`, `ALTER TABLE ... ADD PARTITION`, `ALTER TABLE ... DROP
PARTITION`, `SHOW PARTITIONS`, `CALL [...]
+- Paimon-specific table management and row-level writes are implemented by
`SQLContext`. This includes Paimon `CREATE TABLE`, `ALTER TABLE`, `DROP TABLE`,
`CREATE TEMPORARY TABLE`, `CREATE TEMPORARY VIEW`, REST Catalog persistent
`CREATE VIEW`, `DROP VIEW`, and `CREATE FUNCTION`, `DROP TEMPORARY TABLE` /
`VIEW`, `INSERT OVERWRITE ... PARTITION`, `UPDATE`, `DELETE`, `MERGE INTO`,
`TRUNCATE TABLE`, `ALTER TABLE ... ADD PARTITION`, `ALTER TABLE ... DROP
PARTITION`, `SHOW PARTITIONS`, `MSCK [...]
Not every DataFusion DDL/DML statement maps to a Paimon table operation. For
Paimon catalogs, `CREATE EXTERNAL TABLE`, `LOCATION`, `CREATE MATERIALIZED
VIEW`, and persistent `CREATE TABLE AS SELECT` are rejected or not implemented.
Persistent `CREATE FUNCTION` is supported only for the REST Catalog SQL scalar
form documented below. DataFusion `COPY` can export query results to files; it
does not create or commit Paimon table files.
@@ -968,6 +968,23 @@ and the directory may survive; repair the file system and
remove it there rather
re-registering it. A partition at a custom location is only unregistered; its
directory
is left where it is.
+### MSCK REPAIR TABLE
+
+Reconcile the catalog registrations with the directories that actually exist:
+
+```sql
+MSCK REPAIR TABLE paimon.my_db.events; -- same as ADD
PARTITIONS
+MSCK REPAIR TABLE paimon.my_db.events ADD PARTITIONS; -- register discovered
directories
+MSCK REPAIR TABLE paimon.my_db.events DROP PARTITIONS; -- unregister vanished
directories
+MSCK REPAIR TABLE paimon.my_db.events SYNC PARTITIONS; -- both
+```
+
+Repair is metadata-only: it never deletes data. Directory discovery and the
catalog
+listing both complete before any change is made, so a listing failure cannot
turn a
+truncated view of the table into a `DROP` diff. There is no dry-run and no
scope
+argument — repair always covers the whole table. A partition at a custom
location is
+never unregistered by repair.
+
## Procedures
Use `CALL` to invoke built-in procedures. All procedures are under the `sys`
namespace.
