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 cecd3dc8 feat(datafusion): add SHOW, ADD and DROP PARTITION for
catalog-managed format tables (#816)
cecd3dc8 is described below
commit cecd3dc80512734311946fe2ab37b1f399bd7d6c
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Sun Sep 13 13:22:15 2026 +0800
feat(datafusion): add SHOW, ADD and DROP PARTITION for catalog-managed
format tables (#816)
---
.../datafusion/src/format_partition_ddl.rs | 570 +++++++++++++++++++++
crates/integrations/datafusion/src/lib.rs | 1 +
crates/integrations/datafusion/src/sql_context.rs | 277 +++++++---
.../datafusion/tests/rest_format_partition_sql.rs | 497 ++++++++++++++++++
crates/paimon/src/catalog/mod.rs | 12 +
crates/paimon/src/catalog/rest/rest_catalog.rs | 36 ++
crates/paimon/src/table/format_partition.rs | 16 +-
crates/paimon/src/table/mod.rs | 3 +
crates/paimon/tests/mock_server.rs | 20 +
crates/paimon/tests/rest_catalog_test.rs | 99 ++++
docs/src/sql.md | 72 ++-
11 files changed, 1520 insertions(+), 83 deletions(-)
diff --git a/crates/integrations/datafusion/src/format_partition_ddl.rs
b/crates/integrations/datafusion/src/format_partition_ddl.rs
new file mode 100644
index 00000000..b611c779
--- /dev/null
+++ b/crates/integrations/datafusion/src/format_partition_ddl.rs
@@ -0,0 +1,570 @@
+// 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.
+
+//! SHOW, ADD and DROP PARTITION for Format Tables with catalog-managed
partitions.
+//! DROP PARTITION on a Paimon table stays in `SQLContext`, where it is a
snapshot commit.
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+
+use datafusion::arrow::array::StringArray;
+use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, Schema};
+use datafusion::arrow::record_batch::RecordBatch;
+use datafusion::error::{DataFusionError, Result as DFResult};
+use datafusion::prelude::DataFrame;
+use datafusion::sql::sqlparser::ast::{
+ Expr as SqlExpr, ObjectName, Partition as SqlPartition, Value as SqlValue,
+};
+use datafusion::sql::sqlparser::dialect::GenericDialect;
+use datafusion::sql::sqlparser::keywords::Keyword;
+use datafusion::sql::sqlparser::parser::Parser;
+use datafusion::sql::sqlparser::tokenizer::{Token, Tokenizer};
+use paimon::catalog::{Catalog, Identifier};
+use paimon::spec::{CoreOptions, DataType as PaimonDataType};
+use paimon::table::{
+ format_partition_value, parse_format_partition_value,
FormatTablePartitionPaths,
+};
+
+use crate::error::to_datafusion_error;
+use crate::sql_context::{is_table_not_exist, ok_result, partition_assignment,
SQLContext};
+
+#[derive(Debug)]
+pub(crate) struct ShowPartitionsStatement {
+ table_name: ObjectName,
+ partition_filter: Vec<SqlExpr>,
+}
+
+pub(crate) fn parse_show_partitions(sql: &str) ->
DFResult<Option<ShowPartitionsStatement>> {
+ let dialect = GenericDialect {};
+ let tokens = Tokenizer::new(&dialect, sql)
+ .tokenize_with_location()
+ .map_err(sql_parse_error)?;
+ let significant = tokens
+ .iter()
+ .filter_map(|token| match &token.token {
+ Token::Whitespace(_) => None,
+ Token::Word(word) => Some(word.keyword),
+ _ => Some(Keyword::NoKeyword),
+ })
+ .take(2)
+ .collect::<Vec<_>>();
+ if !matches!(significant.as_slice(), [Keyword::SHOW, Keyword::PARTITIONS])
{
+ return Ok(None);
+ }
+
+ let mut parser = Parser::new(&dialect).with_tokens_with_locations(tokens);
+ parser
+ .expect_keyword_is(Keyword::SHOW)
+ .map_err(sql_parse_error)?;
+ parser
+ .expect_keyword_is(Keyword::PARTITIONS)
+ .map_err(sql_parse_error)?;
+ let table_name = parser.parse_object_name(false).map_err(sql_parse_error)?;
+ let partition_filter = if parser.parse_keyword(Keyword::PARTITION) {
+ parser
+ .expect_token(&Token::LParen)
+ .map_err(sql_parse_error)?;
+ let expressions = parser
+ .parse_comma_separated(Parser::parse_expr)
+ .map_err(sql_parse_error)?;
+ parser
+ .expect_token(&Token::RParen)
+ .map_err(sql_parse_error)?;
+ expressions
+ } else {
+ Vec::new()
+ };
+ let _ = parser.consume_token(&Token::SemiColon);
+ if parser.peek_token().token != Token::EOF {
+ return Err(DataFusionError::Plan(format!(
+ "SQL parse error: unexpected token {} after SHOW PARTITIONS
statement",
+ parser.peek_token().token
+ )));
+ }
+ Ok(Some(ShowPartitionsStatement {
+ table_name,
+ partition_filter,
+ }))
+}
+
+fn sql_parse_error(error: impl std::fmt::Display) -> DataFusionError {
+ DataFusionError::Plan(format!("SQL parse error: {error}"))
+}
+
+pub(crate) async fn execute_show_partitions(
+ ctx: &SQLContext,
+ show_partitions: &ShowPartitionsStatement,
+ enable_ident_normalization: bool,
+) -> DFResult<DataFrame> {
+ SQLContext::ensure_partition_command_target(&show_partitions.table_name,
"SHOW PARTITIONS")?;
+ let (catalog, _catalog_name, identifier) =
+ ctx.resolve_catalog_and_table(&show_partitions.table_name)?;
+ let table = catalog
+ .get_table(&identifier)
+ .await
+ .map_err(to_datafusion_error)?;
+ ensure_catalog_managed_format_table(&table, "SHOW PARTITIONS")?;
+ let filter = if show_partitions.partition_filter.is_empty() {
+ None
+ } else {
+ let spec = parse_format_partition_spec(
+ &show_partitions.partition_filter,
+ &table,
+ false,
+ None,
+ enable_ident_normalization,
+ )?;
+ Some(display_partition_values(&spec, &table)?)
+ };
+
+ let partition_paths = FormatTablePartitionPaths::new(
+ table.schema().partition_keys().iter().cloned(),
+
CoreOptions::new(table.schema().options()).format_table_partition_only_value_in_path(),
+ );
+ let mut names = Vec::new();
+ for partition in catalog
+ .list_partitions(&identifier)
+ .await
+ .map_err(to_datafusion_error)?
+ {
+ let values = display_partition_values(&partition.spec, &table)?;
+ if !filter.as_ref().is_none_or(|filter| {
+ filter
+ .iter()
+ .all(|(key, value)| values.get(key) == Some(value))
+ }) {
+ continue;
+ }
+ let display_spec = values
+ .into_iter()
+ .map(|(key, value)| (key, value.unwrap_or_else(||
"null".to_string())))
+ .collect();
+ let name = partition_paths
+ .partition_name(&display_spec)
+ .map_err(to_datafusion_error)?;
+ names.push(name);
+ }
+ names.sort();
+
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "partition",
+ ArrowDataType::Utf8,
+ false,
+ )]));
+ let batch = RecordBatch::try_new(schema,
vec![Arc::new(StringArray::from(names))])?;
+ ctx.ctx().read_batch(batch)
+}
+
+pub(crate) async fn execute_add_partitions(
+ ctx: &SQLContext,
+ catalog: &Arc<dyn Catalog>,
+ identifier: &Identifier,
+ partitions: &[SqlPartition],
+ ignore_if_exists: bool,
+ ignore_if_table_not_exists: bool,
+ enable_ident_normalization: bool,
+) -> DFResult<DataFrame> {
+ let table = match catalog.get_table(identifier).await {
+ Ok(table) => table,
+ Err(error) if ignore_if_table_not_exists && is_table_not_exist(&error)
=> {
+ return ok_result(ctx.ctx());
+ }
+ Err(error) => return Err(to_datafusion_error(error)),
+ };
+ ensure_catalog_managed_format_table(&table, "ALTER TABLE ADD PARTITION")?;
+ if partitions.is_empty() {
+ return Err(DataFusionError::Plan(
+ "ADD PARTITION requires at least one partition
specification".to_string(),
+ ));
+ }
+
+ 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();
+ let mut specs = Vec::with_capacity(partitions.len());
+ let mut directories = Vec::with_capacity(partitions.len());
+ for partition in partitions {
+ let expressions = match partition {
+ SqlPartition::Partitions(expressions) => expressions,
+ other => {
+ return Err(DataFusionError::Plan(format!(
+ "Unsupported ADD PARTITION specification: {other}"
+ )))
+ }
+ };
+ let spec = parse_format_partition_spec(
+ expressions,
+ &table,
+ true,
+ Some("ADD PARTITION"),
+ enable_ident_normalization,
+ )?;
+ let relative_path = partition_paths
+ .relative_path(&spec)
+ .map_err(to_datafusion_error)?;
+ directories.push(format!(
+ "{}/{}",
+ table_path.trim_end_matches('/'),
+ relative_path
+ ));
+ specs.push(spec);
+ }
+
+ catalog
+ .create_partitions(identifier, specs, ignore_if_exists)
+ .await
+ .map_err(to_datafusion_error)?;
+ for directory in directories {
+ table
+ .file_io()
+ .mkdirs(&directory)
+ .await
+ .map_err(to_datafusion_error)?;
+ }
+ ok_result(ctx.ctx())
+}
+
+/// Unregister catalog-managed partitions and then delete their directories.
+/// A partial specification, keys in any position, drops every registered
partition it matches.
+pub(crate) async fn drop_catalog_managed_partitions(
+ ctx: &SQLContext,
+ catalog: &Arc<dyn Catalog>,
+ identifier: &Identifier,
+ table: &paimon::Table,
+ requests: &[(&[SqlExpr], bool)],
+ enable_ident_normalization: bool,
+) -> DFResult<DataFrame> {
+ ensure_catalog_managed_format_table(table, "ALTER TABLE DROP PARTITION")?;
+ let partition_key_count = table.schema().partition_keys().len();
+ let mut requested = Vec::with_capacity(requests.len());
+ for (expressions, ignore_if_not_exists) in requests {
+ let spec = parse_format_partition_spec(
+ expressions,
+ table,
+ false,
+ Some("DROP PARTITION"),
+ enable_ident_normalization,
+ )?;
+ requested.push((spec, *ignore_if_not_exists));
+ }
+
+ // Only a partial specification needs the whole registry; complete ones
are looked up by
+ // name, which keeps the common exact drop from reading every registration.
+ let complete_specs = requested
+ .iter()
+ .filter(|(spec, _)| spec.len() == partition_key_count)
+ .map(|(spec, _)| spec.clone())
+ .collect::<Vec<_>>();
+ let registered = if complete_specs.len() == requested.len() {
+ catalog
+ .list_partitions_by_names(identifier, complete_specs)
+ .await
+ } else {
+ catalog.list_partitions(identifier).await
+ }
+ .map_err(to_datafusion_error)?;
+
+ 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().trim_end_matches('/');
+ // Every directory is resolved before the first mutation, so a
registration that cannot
+ // be turned into a path fails the statement as a whole.
+ let registered = registered
+ .into_iter()
+ .map(|partition| {
+ let relative_path = partition_paths
+ .relative_path(&partition.spec)
+ .map_err(to_datafusion_error)?;
+ let custom_located = has_custom_location(&partition);
+ Ok((
+ partition.spec,
+ format!("{table_path}/{relative_path}"),
+ custom_located,
+ ))
+ })
+ .collect::<DFResult<Vec<_>>>()?;
+
+ let mut selected: Vec<(HashMap<String, String>, String, bool)> =
Vec::new();
+ let mut selected_paths = HashSet::new();
+ for (spec, ignore_if_not_exists) in &requested {
+ // Values are compared as the catalog holds them, so a partition
registered as
+ // `month=01` is not the partition `month = 1` names.
+ let mut matched = false;
+ for (registered_spec, path, custom_located) in ®istered {
+ if !spec
+ .iter()
+ .all(|(key, value)| registered_spec.get(key) == Some(value))
+ {
+ continue;
+ }
+ matched = true;
+ if selected_paths.insert(path.clone()) {
+ selected.push((registered_spec.clone(), path.clone(),
*custom_located));
+ }
+ }
+ // Only a complete specification can be reported as missing; a partial
one may match
+ // nothing, as in Java.
+ if !matched && spec.len() == partition_key_count &&
!ignore_if_not_exists {
+ return Err(DataFusionError::Plan(format!(
+ "Partition {spec:?} does not exist in table {}",
+ identifier.full_name()
+ )));
+ }
+ }
+
+ if selected.is_empty() {
+ return ok_result(ctx.ctx());
+ }
+
+ catalog
+ .drop_partitions(
+ identifier,
+ selected.iter().map(|(spec, _, _)| spec.clone()).collect(),
+ )
+ .await
+ .map_err(to_datafusion_error)?;
+ for (_, path, custom_located) in selected {
+ // A partition registered at a location of its own is only
unregistered, as in Java,
+ // and keeps its data there.
+ if custom_located {
+ continue;
+ }
+ table
+ .file_io()
+ .delete_dir(&path)
+ .await
+ .map_err(to_datafusion_error)?;
+ }
+ ok_result(ctx.ctx())
+}
+
+/// Whether the catalog registered a partition at a location of its own rather
than under the
+/// table directory.
+fn has_custom_location(partition: &paimon::spec::Partition) -> bool {
+ partition
+ .options
+ .as_ref()
+ .is_some_and(|options| options.contains_key("path"))
+}
+
+pub(crate) fn ensure_catalog_managed_format_table(
+ table: &paimon::Table,
+ operation: &str,
+) -> DFResult<()> {
+ if table.schema().partition_keys().is_empty() {
+ return Err(DataFusionError::Plan(format!(
+ "{operation} requires a partitioned table, but {} is not
partitioned",
+ table.identifier().full_name()
+ )));
+ }
+ if !table.has_catalog_managed_partitions() {
+ return Err(DataFusionError::Plan(format!(
+ "{operation} is supported only for catalog-managed partitions on
an internal Format Table loaded from REST Catalog; table {} does not have this
configuration",
+ table.identifier().full_name()
+ )));
+ }
+ Ok(())
+}
+
+/// `mutating_operation` names the statement when it changes partitions (ADD
or DROP PARTITION),
+/// which refuses a blank string for a string partition column.
+fn parse_format_partition_spec(
+ exprs: &[SqlExpr],
+ table: &paimon::Table,
+ require_complete: bool,
+ mutating_operation: Option<&str>,
+ enable_ident_normalization: bool,
+) -> DFResult<HashMap<String, String>> {
+ let fields = table
+ .schema()
+ .fields()
+ .iter()
+ .map(|field| (field.name(), field))
+ .collect::<HashMap<_, _>>();
+ let partition_keys = table.schema().partition_keys();
+ let options = CoreOptions::new(table.schema().options());
+ let mut spec = HashMap::with_capacity(exprs.len());
+
+ for expr in exprs {
+ let (column, literal) = partition_assignment(expr,
enable_ident_normalization)?;
+ if !partition_keys.contains(&column) {
+ return Err(DataFusionError::Plan(format!(
+ "Column '{column}' is not a partition column"
+ )));
+ }
+ if spec.contains_key(&column) {
+ return Err(DataFusionError::Plan(format!(
+ "Duplicate partition column '{column}'"
+ )));
+ }
+ let field = fields.get(column.as_str()).ok_or_else(|| {
+ DataFusionError::Plan(format!("Column '{column}' not found in
table schema"))
+ })?;
+ let data_type = field.data_type();
+ let value = match partition_literal_to_string(literal)? {
+ None => options.partition_default_name().to_string(),
+ // Read with the column type as Java `TypeUtils.castFromString`
does, then written back
+ // in the spelling ADD PARTITION registers.
+ Some(text) => {
+ let text = match data_type {
+ PaimonDataType::Char(_) | PaimonDataType::VarChar(_) => {
+ // A blank string would address the default partition;
Java
+ // `PaimonFormatTable.requireNameablePartitionValues`
refuses it too.
+ if let Some(operation) = mutating_operation {
+ if text.trim().is_empty() {
+ return Err(DataFusionError::Plan(format!(
+ "{operation} does not support an empty or
whitespace-only \
+ string for partition column '{column}' of
Format Table {}. \
+ Such a value is written to the partition
named {}, name it \
+ directly to address it",
+ table.identifier().full_name(),
+ options.partition_default_name()
+ )));
+ }
+ }
+ text
+ }
+ _ => text.trim().to_string(),
+ };
+ parse_format_partition_value(&text, data_type)
+ .and_then(|datum| {
+ format_partition_value(
+ &datum,
+ data_type,
+ options.partition_default_name(),
+ options.legacy_partition_name(),
+ )
+ })
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "Cannot use {literal} as a value of partition
column '{column}' \
+ with type {data_type:?}"
+ ))
+ })?
+ }
+ };
+ spec.insert(column, value);
+ }
+
+ if require_complete {
+ let missing = partition_keys
+ .iter()
+ .filter(|key| !spec.contains_key(key.as_str()))
+ .cloned()
+ .collect::<Vec<_>>();
+ if !missing.is_empty() {
+ return Err(DataFusionError::Plan(format!(
+ "Incomplete partition spec: missing keys [{}]",
+ missing.join(", ")
+ )));
+ }
+ }
+ Ok(spec)
+}
+
+/// The string a SQL partition literal stands for, as Spark hands it to
Paimon; `None` is NULL.
+/// A number is canonical, so `month = 01` and `month = 1` name the same
partition.
+fn partition_literal_to_string(expr: &SqlExpr) -> DFResult<Option<String>> {
+ use datafusion::sql::sqlparser::ast::{DataType as SqlDataType,
UnaryOperator};
+
+ let unsupported =
+ || DataFusionError::Plan(format!("Unsupported partition value
expression: {expr}"));
+ let (sign, value, signed) = match expr {
+ SqlExpr::UnaryOp {
+ op: UnaryOperator::Minus,
+ expr,
+ } => ("-", expr.as_ref(), true),
+ SqlExpr::UnaryOp {
+ op: UnaryOperator::Plus,
+ expr,
+ } => ("", expr.as_ref(), true),
+ other => ("", other, false),
+ };
+ match value {
+ SqlExpr::Value(value) => match &value.value {
+ SqlValue::Number(number, _) => {
+ let canonical = number
+ .parse::<i128>()
+ .map_or_else(|_| number.clone(), |number|
number.to_string());
+ Ok(Some(format!("{sign}{canonical}")))
+ }
+ _ if signed => Err(unsupported()),
+ SqlValue::Null => Ok(None),
+ SqlValue::Boolean(value) => Ok(Some(value.to_string())),
+ other => other
+ .clone()
+ .into_string()
+ .map(Some)
+ .ok_or_else(unsupported),
+ },
+ SqlExpr::TypedString(typed) if !signed && matches!(typed.data_type,
SqlDataType::Date) => {
+ typed
+ .value
+ .value
+ .clone()
+ .into_string()
+ .map(Some)
+ .ok_or_else(unsupported)
+ }
+ _ => Err(unsupported()),
+ }
+}
+
+/// The values of a partition spec as SHOW PARTITIONS prints and filters them:
read with the column
+/// type, so `month=01` and `month=1` are the same partition, and `None` for
the default partition.
+fn display_partition_values(
+ spec: &HashMap<String, String>,
+ table: &paimon::Table,
+) -> DFResult<HashMap<String, Option<String>>> {
+ let options = CoreOptions::new(table.schema().options());
+ let default_partition_name = options.partition_default_name();
+ let fields = table.schema().partition_fields();
+ spec.iter()
+ .map(|(key, raw)| {
+ if raw == default_partition_name {
+ return Ok((key.clone(), None));
+ }
+ let data_type = fields
+ .iter()
+ .find(|field| field.name() == key)
+ .map(|field| field.data_type())
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "Invalid partition spec {spec:?} for table {}",
+ table.identifier().full_name()
+ ))
+ })?;
+ parse_format_partition_value(raw, data_type)
+ .and_then(|datum| {
+ format_partition_value(&datum, data_type,
default_partition_name, false)
+ })
+ .map(|value| (key.clone(), Some(value)))
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "Invalid catalog partition value {raw:?} for column
'{key}' with type \
+ {data_type:?}"
+ ))
+ })
+ })
+ .collect()
+}
diff --git a/crates/integrations/datafusion/src/lib.rs
b/crates/integrations/datafusion/src/lib.rs
index 985ad31f..4ec76c45 100644
--- a/crates/integrations/datafusion/src/lib.rs
+++ b/crates/integrations/datafusion/src/lib.rs
@@ -43,6 +43,7 @@ mod catalog;
mod delete;
mod error;
mod filter_pushdown;
+mod format_partition_ddl;
#[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 74a0df45..b75534e9 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -34,7 +34,9 @@
//! - `ALTER TABLE db.t ALTER COLUMN col TYPE new_type`
//! - `ALTER TABLE db.t ALTER COLUMN col SET|DROP NOT NULL`
//! - `ALTER TABLE db.t RENAME TO new_name`
-//! - `ALTER TABLE db.t DROP PARTITION (col = val, ...)`
+//! - `ALTER TABLE db.t ADD [IF NOT EXISTS] PARTITION (...) [PARTITION (...)]`
+//! - `ALTER TABLE db.t DROP [IF EXISTS] PARTITION (...)`
+//! - `SHOW PARTITIONS db.t [PARTITION (...)]`
//! - `CREATE VIEW [IF NOT EXISTS] view [(col, ...)] AS query`
//! - `DROP VIEW [IF EXISTS] view`
//! - `CREATE FUNCTION name(args) RETURNS type [LANGUAGE SQL] RETURN
expression`
@@ -72,11 +74,11 @@ use datafusion::sql::sqlparser::keywords::Keyword;
use datafusion::sql::sqlparser::parser::Parser;
use datafusion::sql::sqlparser::tokenizer::{Token, Tokenizer};
use futures::StreamExt;
-use paimon::catalog::{parse_object_name, Catalog, Identifier};
+use paimon::catalog::{parse_object_name, Catalog, Identifier,
ParsedObjectName};
use paimon::spec::{
ArrayType as PaimonArrayType, BigIntType, BinaryType, BlobType,
BooleanType, CharType,
- DataField as PaimonDataField, DataType as PaimonDataType, DateType, Datum,
DecimalType,
- DoubleType, FloatType, IntType, LocalZonedTimestampType, MapType as
PaimonMapType,
+ CoreOptions, DataField as PaimonDataField, DataType as PaimonDataType,
DateType, Datum,
+ DecimalType, DoubleType, FloatType, IntType, LocalZonedTimestampType,
MapType as PaimonMapType,
RowType as PaimonRowType, SchemaChange, SmallIntType, TimestampType,
TinyIntType,
VarBinaryType, VarCharType, VariantType,
};
@@ -452,6 +454,16 @@ impl SQLContext {
// Time-travel queries are not DDL; skip our own parsing and
handle directly.
return self.handle_time_travel_query(&rewritten_sql).await;
}
+ if let Some(show_partitions) =
+ crate::format_partition_ddl::parse_show_partitions(&rewritten_sql)?
+ {
+ return crate::format_partition_ddl::execute_show_partitions(
+ self,
+ &show_partitions,
+ enable_ident_normalization,
+ )
+ .await;
+ }
let statements = parse_sql_statements(&rewritten_sql)?;
@@ -524,6 +536,15 @@ impl SQLContext {
obj_name,
} => self.handle_show_create_table(sql, obj_name).await,
Statement::AlterTable(alter_table) => {
+ if alter_table.location.is_some()
+ && alter_table.operations.iter().any(|operation| {
+ matches!(operation, AlterTableOperation::AddPartitions
{ .. })
+ })
+ {
+ return Err(DataFusionError::Plan(
+ "LOCATION is not supported for Format Table
partitions".to_string(),
+ ));
+ }
let (catalog, _catalog_name, _) =
self.resolve_catalog_and_table(&alter_table.name)?;
self.handle_alter_table(
@@ -1262,20 +1283,64 @@ impl SQLContext {
if_exists: bool,
enable_ident_normalization: bool,
) -> DFResult<DataFrame> {
- Self::ensure_main_branch_write_target(name, "ALTER TABLE")?;
+ let has_partition_operation = operations.iter().any(|operation| {
+ matches!(
+ operation,
+ AlterTableOperation::AddPartitions { .. }
+ | AlterTableOperation::DropPartitions { .. }
+ )
+ });
+ if has_partition_operation {
+ Self::ensure_partition_command_target(name, "ALTER TABLE")?;
+ } else {
+ Self::ensure_main_branch_write_target(name, "ALTER TABLE")?;
+ }
let identifier = self.resolve_table_name(name)?;
if operations.len() > 1
- && operations.iter().any(|operation| {
- matches!(
- operation,
- AlterTableOperation::RenameTable { .. }
- | AlterTableOperation::DropPartitions { .. }
- )
+ && operations
+ .iter()
+ .any(|operation| matches!(operation,
AlterTableOperation::RenameTable { .. }))
+ {
+ return Err(DataFusionError::Plan(
+ "ALTER TABLE RENAME TO must be used alone".to_string(),
+ ));
+ }
+ // A statement may drop several partitions, but only partitions:
mixing the drop
+ // with schema changes would commit two unrelated changes under one
statement.
+ let drop_partition_requests = operations
+ .iter()
+ .filter_map(|operation| match operation {
+ AlterTableOperation::DropPartitions {
+ partitions,
+ if_exists: partition_if_exists,
+ } => Some((partitions.as_slice(), *partition_if_exists)),
+ _ => None,
})
+ .collect::<Vec<_>>();
+ if !drop_partition_requests.is_empty() {
+ if drop_partition_requests.len() != operations.len() {
+ return Err(DataFusionError::Plan(
+ "ALTER TABLE DROP PARTITION must be used
alone".to_string(),
+ ));
+ }
+ return self
+ .handle_drop_partitions(
+ catalog,
+ &identifier,
+ &drop_partition_requests,
+ if_exists,
+ enable_ident_normalization,
+ )
+ .await;
+ }
+ if operations
+ .iter()
+ .any(|operation| matches!(operation,
AlterTableOperation::AddPartitions { .. }))
+ && operations.len() != 1
{
return Err(DataFusionError::Plan(
- "ALTER TABLE RENAME TO and DROP PARTITION must be used
alone".to_string(),
+ "ALTER TABLE ADD PARTITION cannot be combined with other
operations".to_string(),
));
}
@@ -1335,19 +1400,20 @@ impl SQLContext {
}
}
}
- AlterTableOperation::DropPartitions {
- partitions,
- if_exists: partition_if_exists,
+ AlterTableOperation::AddPartitions {
+ if_not_exists,
+ new_partitions,
} => {
- return self
- .handle_drop_partitions(
- catalog,
- &identifier,
- partitions,
- if_exists || *partition_if_exists,
- enable_ident_normalization,
- )
- .await;
+ return crate::format_partition_ddl::execute_add_partitions(
+ self,
+ catalog,
+ &identifier,
+ new_partitions,
+ *if_not_exists,
+ if_exists,
+ enable_ident_normalization,
+ )
+ .await;
}
other => {
return Err(DataFusionError::Plan(format!(
@@ -1930,32 +1996,57 @@ impl SQLContext {
&self,
catalog: &Arc<dyn Catalog>,
identifier: &Identifier,
- partitions: &[SqlExpr],
- if_exists: bool,
+ requests: &[(&[SqlExpr], bool)],
+ ignore_if_table_not_exists: bool,
enable_ident_normalization: bool,
) -> DFResult<DataFrame> {
- if partitions.is_empty() {
+ if requests
+ .iter()
+ .any(|(expressions, _)| expressions.is_empty())
+ {
return Err(DataFusionError::Plan(
- "DROP PARTITIONS requires at least one partition
specification".to_string(),
+ "DROP PARTITION requires a partition
specification".to_string(),
));
}
let table = match catalog.get_table(identifier).await {
- Ok(t) => t,
- Err(e) if if_exists && is_table_not_exist(&e) => {
+ Ok(table) => table,
+ Err(error) if ignore_if_table_not_exists &&
is_table_not_exist(&error) => {
return ok_result(&self.ctx);
}
- Err(e) => return Err(to_datafusion_error(e)),
+ Err(error) => return Err(to_datafusion_error(error)),
};
- let partition_values = parse_partition_values(
- partitions,
- table.schema().fields(),
- table.schema().partition_keys(),
- enable_ident_normalization,
- )?;
+ if table.has_catalog_managed_partitions() {
+ return
crate::format_partition_ddl::drop_catalog_managed_partitions(
+ self,
+ catalog,
+ identifier,
+ &table,
+ requests,
+ enable_ident_normalization,
+ )
+ .await;
+ }
+ if CoreOptions::new(table.schema().options()).is_format_table() {
+ crate::format_partition_ddl::ensure_catalog_managed_format_table(
+ &table,
+ "ALTER TABLE DROP PARTITION",
+ )?;
+ }
- let wb = table.new_write_builder();
- let commit = wb.try_new_commit().map_err(to_datafusion_error)?;
+ let mut partition_values = Vec::with_capacity(requests.len());
+ for (expressions, _) in requests {
+ partition_values.extend(parse_partition_values(
+ expressions,
+ table.schema().fields(),
+ table.schema().partition_keys(),
+ enable_ident_normalization,
+ )?);
+ }
+ let commit = table
+ .new_write_builder()
+ .try_new_commit()
+ .map_err(to_datafusion_error)?;
commit
.truncate_partitions(partition_values)
.await
@@ -2099,7 +2190,7 @@ impl SQLContext {
}
/// Resolve an ObjectName like `catalog.db.table` or `db.table` to a
catalog and Identifier.
- fn resolve_catalog_and_table(
+ pub(crate) fn resolve_catalog_and_table(
&self,
name: &ObjectName,
) -> DFResult<(Arc<dyn Catalog>, String, Identifier)> {
@@ -2149,13 +2240,7 @@ impl SQLContext {
}
fn ensure_main_branch_write_target(name: &ObjectName, operation: &str) ->
DFResult<()> {
- let object = name
- .0
- .last()
- .and_then(|part| part.as_ident())
- .map(|ident| ident.value.as_str())
- .ok_or_else(|| DataFusionError::Plan(format!("Invalid table
reference: {name}")))?;
- let parsed = parse_object_name(object).map_err(to_datafusion_error)?;
+ let parsed = Self::parse_target_object_name(name)?;
if let Some(branch) = parsed.branch() {
return Err(DataFusionError::NotImplemented(format!(
"{operation} on Paimon branch '{branch}' is not supported"
@@ -2164,6 +2249,34 @@ impl SQLContext {
Ok(())
}
+ pub(crate) fn ensure_partition_command_target(
+ name: &ObjectName,
+ operation: &str,
+ ) -> DFResult<()> {
+ let parsed = Self::parse_target_object_name(name)?;
+ if let Some(branch) = parsed.branch() {
+ return Err(DataFusionError::NotImplemented(format!(
+ "{operation} on Paimon branch '{branch}' is not supported"
+ )));
+ }
+ if let Some(system_table) = parsed.system_table() {
+ return Err(DataFusionError::NotImplemented(format!(
+ "{operation} on Paimon system table '{system_table}' is not
supported"
+ )));
+ }
+ Ok(())
+ }
+
+ fn parse_target_object_name(name: &ObjectName) ->
DFResult<ParsedObjectName> {
+ let object = name
+ .0
+ .last()
+ .and_then(|part| part.as_ident())
+ .map(|ident| ident.value.as_str())
+ .ok_or_else(|| DataFusionError::Plan(format!("Invalid table
reference: {name}")))?;
+ parse_object_name(object).map_err(to_datafusion_error)
+ }
+
/// Resolve an ObjectName to just the Identifier (for backward compat in
handle_alter_table).
fn resolve_table_name(&self, name: &ObjectName) -> DFResult<Identifier> {
let (_catalog, _catalog_name, identifier) =
self.resolve_catalog_and_table(name)?;
@@ -3029,10 +3142,35 @@ fn extract_options(opts: &CreateTableOptions) ->
DFResult<Vec<(String, String)>>
.collect()
}
-fn is_table_not_exist(e: &paimon::Error) -> bool {
+pub(crate) fn is_table_not_exist(e: &paimon::Error) -> bool {
matches!(e, paimon::Error::TableNotExist { .. })
}
+pub(crate) fn partition_assignment(
+ expr: &SqlExpr,
+ enable_ident_normalization: bool,
+) -> DFResult<(String, &SqlExpr)> {
+ let SqlExpr::BinaryOp {
+ left,
+ op: datafusion::sql::sqlparser::ast::BinaryOperator::Eq,
+ right,
+ } = expr
+ else {
+ return Err(DataFusionError::Plan(format!(
+ "Expected 'column = value' in partition spec, got: {expr}"
+ )));
+ };
+ let SqlExpr::Identifier(identifier) = left.as_ref() else {
+ return Err(DataFusionError::Plan(format!(
+ "Expected column name in partition spec, got: {left}"
+ )));
+ };
+ Ok((
+ normalize_schema_identifier(identifier, enable_ident_normalization),
+ right.as_ref(),
+ ))
+}
+
/// Parse partition expressions (`col = val, ...`) into partition value maps
/// suitable for `TableCommit::truncate_partitions`.
///
@@ -3050,37 +3188,14 @@ fn parse_partition_values(
let mut partition = HashMap::new();
let mut seen_columns = HashSet::new();
for expr in exprs {
- let (col_name, val_expr) = match expr {
- SqlExpr::BinaryOp {
- left,
- op: datafusion::sql::sqlparser::ast::BinaryOperator::Eq,
- right,
- } => {
- let col = match left.as_ref() {
- SqlExpr::Identifier(ident) => {
- normalize_schema_identifier(ident,
enable_ident_normalization)
- }
- other => {
- return Err(DataFusionError::Plan(format!(
- "Expected column name in partition spec, got:
{other}"
- )))
- }
- };
- (col, right.as_ref())
- }
- other => {
- return Err(DataFusionError::Plan(format!(
- "Expected 'column = value' in partition spec, got: {other}"
- )))
- }
- };
+ let (col_name, val_expr) = partition_assignment(expr,
enable_ident_normalization)?;
if !seen_columns.insert(col_name.clone()) {
return Err(DataFusionError::Plan(format!(
"Duplicate partition column '{col_name}'"
)));
}
- if !partition_keys.iter().any(|k| k == &col_name) {
+ if !partition_keys.contains(&col_name) {
return Err(DataFusionError::Plan(format!(
"Column '{col_name}' is not a partition column"
)));
@@ -3633,7 +3748,7 @@ fn extract_all_timestamp_as_of(sql: &str) ->
Vec<TimestampAsOfInfo> {
}
/// Return an empty DataFrame with a single "result" column containing "OK".
-fn ok_result(ctx: &SessionContext) -> DFResult<DataFrame> {
+pub(crate) fn ok_result(ctx: &SessionContext) -> DFResult<DataFrame> {
let schema = Arc::new(Schema::new(vec![Field::new(
"result",
ArrowDataType::Utf8,
@@ -7753,6 +7868,20 @@ mod tests {
.unwrap();
}
+ #[tokio::test]
+ async fn test_drop_if_exists_partition_does_not_ignore_missing_table() {
+ let (_tmp, sql_context) = setup_fs_sql_context().await;
+
+ let err = sql_context
+ .sql("ALTER TABLE paimon.test_db.nonexistent DROP IF EXISTS
PARTITION (pt = 'a')")
+ .await
+ .unwrap_err();
+ assert!(
+ err.to_string().contains("does not exist"),
+ "Expected table-not-exist error, got: {err}"
+ );
+ }
+
#[tokio::test]
async fn test_drop_partition_incomplete_spec() {
let (_tmp, sql_context) = setup_fs_sql_context().await;
diff --git a/crates/integrations/datafusion/tests/rest_format_partition_sql.rs
b/crates/integrations/datafusion/tests/rest_format_partition_sql.rs
new file mode 100644
index 00000000..27005727
--- /dev/null
+++ b/crates/integrations/datafusion/tests/rest_format_partition_sql.rs
@@ -0,0 +1,497 @@
+// 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;
+
+#[path = "../../../paimon/tests/mock_server.rs"]
+mod mock_server;
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use paimon::api::ConfigResponse;
+use paimon::catalog::RESTCatalog;
+use paimon::spec::{BigIntType, BooleanType, DataType, DateType, IntType,
Schema, VarCharType};
+use paimon::{CatalogOptions, Options};
+use paimon_datafusion::SQLContext;
+use tempfile::TempDir;
+
+use mock_server::{start_mock_server, RESTServer};
+
+const DATABASE: &str = "default";
+const TABLE: &str = "events";
+const TABLE_NAME: &str = "paimon.default.events";
+const WAREHOUSE: &str = "test_warehouse";
+
+async fn setup_rest_table(temp_dir: &TempDir, schema: Schema) -> (RESTServer,
SQLContext) {
+ let server = start_mock_server(
+ WAREHOUSE.to_string(),
+ temp_dir.path().to_string_lossy().into_owned(),
+ ConfigResponse::new(HashMap::from([(
+ CatalogOptions::PREFIX.to_string(),
+ "mock-test".to_string(),
+ )])),
+ vec![DATABASE.to_string()],
+ )
+ .await;
+ server.add_table_with_schema(
+ DATABASE,
+ TABLE,
+ schema,
+ &format!("file://{}", temp_dir.path().display()),
+ );
+ server.set_table_external(DATABASE, TABLE, false);
+
+ let mut options = Options::new();
+ options.set(CatalogOptions::URI, server.url().unwrap());
+ options.set(CatalogOptions::WAREHOUSE, WAREHOUSE);
+ options.set(CatalogOptions::TOKEN_PROVIDER, "bear");
+ options.set(CatalogOptions::TOKEN, "test-token");
+ let catalog = Arc::new(RESTCatalog::new(options, true).await.unwrap());
+ let mut context = SQLContext::new();
+ context.register_catalog("paimon", catalog).await.unwrap();
+ (server, context)
+}
+
+fn format_table_schema(partition_columns: &[(&str, DataType)]) -> Schema {
+ let partition_keys = partition_columns
+ .iter()
+ .map(|(name, _)| (*name).to_string())
+ .collect::<Vec<_>>();
+ partition_columns
+ .iter()
+ .fold(Schema::builder(), |builder, (name, data_type)| {
+ builder.column(*name, data_type.clone())
+ })
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .partition_keys(partition_keys)
+ .option("type", "format-table")
+ .option("file.format", "parquet")
+ .option("metastore.partitioned-table", "true")
+ .build()
+ .unwrap()
+}
+
+fn varchar() -> DataType {
+ DataType::VarChar(VarCharType::new(255).unwrap())
+}
+
+/// A `(dt, hh)` table with the given partitions added through ADD PARTITION.
+async fn dt_hh_table(partitions: &[(&str, &str)]) -> (TempDir, RESTServer,
SQLContext) {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let schema = format_table_schema(&[("dt", varchar()), ("hh", varchar())]);
+ let (server, context) = setup_rest_table(&temp_dir, schema).await;
+ for (dt, hh) in partitions {
+ common::exec(
+ &context,
+ &format!(
+ "ALTER TABLE {TABLE_NAME} ADD IF NOT EXISTS PARTITION (dt =
'{dt}', hh = '{hh}')"
+ ),
+ )
+ .await;
+ }
+ (temp_dir, server, context)
+}
+
+async fn show_partitions(context: &SQLContext, partition_clause: &str) ->
Vec<String> {
+ let batches = context
+ .sql(&format!("SHOW PARTITIONS {TABLE_NAME}{partition_clause}"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ batches
+ .iter()
+ .flat_map(|batch| {
+ (0..batch.num_rows()).map(|row|
common::string_value(batch.column(0), row).to_string())
+ })
+ .collect()
+}
+
+fn spec(values: &[(&str, &str)]) -> HashMap<String, String> {
+ values
+ .iter()
+ .map(|(key, value)| ((*key).to_string(), (*value).to_string()))
+ .collect()
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_partition_commands_update_rest_metadata_and_directories() {
+ 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') PARTITION
(dt = 'b')"),
+ )
+ .await;
+ assert_eq!(show_partitions(&context, "").await, ["dt=a", "dt=b"]);
+ assert_eq!(
+ show_partitions(&context, " PARTITION (dt = 'b')").await,
+ ["dt=b"]
+ );
+ assert!(temp_dir.path().join("dt=a").is_dir());
+
+ common::exec(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} DROP PARTITION (dt = 'a')"),
+ )
+ .await;
+ assert_eq!(show_partitions(&context, "").await, ["dt=b"]);
+ assert!(!temp_dir.path().join("dt=a").exists());
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_partition_literals() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let schema = format_table_schema(&[
+ ("dt", DataType::Date(DateType::new())),
+ ("month", DataType::Int(IntType::new())),
+ ("active", DataType::Boolean(BooleanType::new())),
+ ("label", varchar()),
+ ]);
+ let (_server, context) = setup_rest_table(&temp_dir, schema).await;
+
+ // Literals are read with the column type the way Java reads partition
strings. Each case adds
+ // a partition, checks its name and directory, and drops it again with the
same literals.
+ for (literals, name, directory) in [
+ // A typed DATE, a zero-padded INT, a boolean in capitals and a number
for a string
+ // column. DATE directories hold Unix epoch days.
+ (
+ "dt = DATE '2026-07-22', month = '01', active = 'TRUE', label =
20260722",
+ "dt=2026-07-22/month=1/active=true/label=20260722",
+ "dt=20656/month=1/active=true/label=20260722",
+ ),
+ // Java's boolean spellings, and an unquoted zero-padded number.
+ (
+ "dt = '2026-07-23', month = 01, active = 'yes', label = 'a'",
+ "dt=2026-07-23/month=1/active=true/label=a",
+ "dt=20657/month=1/active=true/label=a",
+ ),
+ (
+ "dt = '2026-07-24', month = -1, active = '0', label = 'b'",
+ "dt=2026-07-24/month=-1/active=false/label=b",
+ "dt=20658/month=-1/active=false/label=b",
+ ),
+ // NULL is the default partition.
+ (
+ "dt = NULL, month = NULL, active = NULL, label = NULL",
+ "dt=null/month=null/active=null/label=null",
+ "dt=__DEFAULT_PARTITION__/month=__DEFAULT_PARTITION__/\
+ active=__DEFAULT_PARTITION__/label=__DEFAULT_PARTITION__",
+ ),
+ ] {
+ common::exec(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} ADD PARTITION ({literals})"),
+ )
+ .await;
+ assert_eq!(show_partitions(&context, "").await, [name], "{literals}");
+ assert!(temp_dir.path().join(directory).is_dir(), "{literals}");
+ common::exec(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} DROP PARTITION ({literals})"),
+ )
+ .await;
+ assert!(show_partitions(&context, "").await.is_empty(), "{literals}");
+ }
+
+ for (literals, column) in [
+ (
+ "dt = '2026-07-22', month = '1.5', active = 'true', label = 'a'",
+ "'month'",
+ ),
+ (
+ "dt = '2026-07-22', month = 1, active = 'maybe', label = 'a'",
+ "'active'",
+ ),
+ (
+ "dt = 'yesterday', month = 1, active = 'true', label = 'a'",
+ "'dt'",
+ ),
+ ] {
+ common::assert_sql_error(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} ADD PARTITION ({literals})"),
+ column,
+ )
+ .await;
+ }
+}
+
+/// A blank string for a string partition column is written to the default
partition, so ADD and
+/// DROP refuse it rather than address the NULL partition, as Java does.
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_partition_ddl_refuses_blank_string_values() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let schema = format_table_schema(&[("label", varchar())]);
+ let (_server, context) = setup_rest_table(&temp_dir, schema).await;
+ common::exec(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} ADD PARTITION (label = NULL)"),
+ )
+ .await;
+ let default_directory =
temp_dir.path().join("label=__DEFAULT_PARTITION__");
+ assert!(default_directory.is_dir());
+
+ for statement in [
+ "DROP PARTITION (label = '')",
+ "DROP IF EXISTS PARTITION (label = ' ')",
+ "ADD PARTITION (label = '')",
+ "ADD IF NOT EXISTS PARTITION (label = ' ')",
+ ] {
+ common::assert_sql_error(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} {statement}"),
+ "empty or whitespace-only string for partition column 'label'",
+ )
+ .await;
+ }
+
+ assert_eq!(show_partitions(&context, "").await, ["label=null"]);
+ assert!(default_directory.is_dir());
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_drop_partition_specifications() {
+ const REGISTERED: &[(&str, &str)] =
+ &[("20260722", "10"), ("20260722", "11"), ("20260723", "10")];
+ const ALL: &[&str] = &[
+ "dt=20260722/hh=10",
+ "dt=20260722/hh=11",
+ "dt=20260723/hh=10",
+ ];
+
+ for (operation, error, remaining) in [
+ // A partial specification expands to every registered partition it
matches.
+ ("DROP PARTITION (dt = '20260722')", None, &["dt=20260723/hh=10"][..]),
+ // The fixed keys need not be a leading prefix.
+ ("DROP PARTITION (hh = '10')", None, &["dt=20260722/hh=11"][..]),
+ // One statement may carry several specifications.
+ (
+ "DROP PARTITION (dt = '20260722', hh = '11'), DROP PARTITION (dt =
'20260723')",
+ None,
+ &["dt=20260722/hh=10"][..],
+ ),
+ // A complete specification names one partition, so a missing one is
an error,
+ (
+ "DROP PARTITION (dt = '20260724', hh = '10')",
+ Some("does not exist"),
+ ALL,
+ ),
+ // unless IF EXISTS is given.
+ ("DROP IF EXISTS PARTITION (dt = '20260724', hh = '10')", None, ALL),
+ // A partial specification describes a set that may come out empty.
+ (
+ "DROP PARTITION (dt = '20260724'), DROP PARTITION (dt =
'20260723')",
+ None,
+ &["dt=20260722/hh=10", "dt=20260722/hh=11"][..],
+ ),
+ // One failing specification leaves the whole statement unapplied.
+ (
+ "DROP PARTITION (dt = '20260722', hh = '10'), DROP PARTITION (dt =
'20260724', hh = '10')",
+ Some("does not exist"),
+ ALL,
+ ),
+ // Dropping partitions cannot be combined with a schema change.
+ (
+ "DROP PARTITION (dt = '20260722'), ADD COLUMN c INT",
+ Some("must be used alone"),
+ ALL,
+ ),
+ ] {
+ let (temp_dir, _server, context) = dt_hh_table(REGISTERED).await;
+ let sql = format!("ALTER TABLE {TABLE_NAME} {operation}");
+ match error {
+ None => common::exec(&context, &sql).await,
+ Some(expected) => common::assert_sql_error(&context, &sql,
expected).await,
+ }
+ assert_eq!(show_partitions(&context, "").await, remaining,
"{operation}");
+ // A dropped partition loses its directory and a kept one keeps it.
+ for (dt, hh) in REGISTERED {
+ let name = format!("dt={dt}/hh={hh}");
+ assert_eq!(
+ temp_dir.path().join(&name).exists(),
+ remaining.contains(&name.as_str()),
+ "{operation}: {name}"
+ );
+ }
+ }
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_drop_partition_matches_values_as_the_catalog_holds_them() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let schema = format_table_schema(&[
+ ("year", varchar()),
+ ("month", DataType::Int(IntType::new())),
+ ]);
+ let (server, context) = setup_rest_table(&temp_dir, schema).await;
+ // Repair keeps directory spellings, so both registrations are legitimate
and distinct.
+ for directory in ["year=2025/month=01", "year=2026/month=1"] {
+ std::fs::create_dir_all(temp_dir.path().join(directory)).unwrap();
+ }
+ server.set_table_partitions(
+ DATABASE,
+ TABLE,
+ vec![
+ spec(&[("year", "2025"), ("month", "01")]),
+ spec(&[("year", "2026"), ("month", "1")]),
+ ],
+ );
+
+ // SHOW PARTITIONS reads both with the column type.
+ assert_eq!(
+ show_partitions(&context, " PARTITION (month = 1)").await,
+ ["year=2025/month=1", "year=2026/month=1"]
+ );
+
+ // A request is spelled the way ADD PARTITION registers it, so `month = 1`
is not `month=01`.
+ common::assert_sql_error(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} DROP PARTITION (year = '2025',
month = 1)"),
+ "does not exist",
+ )
+ .await;
+ common::exec(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} DROP IF EXISTS PARTITION (year =
'2025', month = 1)"),
+ )
+ .await;
+
+ common::exec(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} DROP PARTITION (month = 1)"),
+ )
+ .await;
+ assert_eq!(
+ server.table_partition_specs(DATABASE, TABLE),
+ vec![spec(&[("year", "2025"), ("month", "01")])]
+ );
+ assert!(temp_dir.path().join("year=2025/month=01").is_dir());
+ assert!(!temp_dir.path().join("year=2026/month=1").exists());
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_drop_partition_looks_up_complete_specifications_by_name() {
+ let (_temp_dir, server, context) =
+ dt_hh_table(&[("20260722", "10"), ("20260722", "11"), ("20260723",
"10")]).await;
+ let listings = server
+ .table_partition_list_name_patterns(DATABASE, TABLE)
+ .len();
+
+ common::exec(
+ &context,
+ &format!(
+ "ALTER TABLE {TABLE_NAME} \
+ DROP PARTITION (dt = '20260722', hh = '10'), DROP PARTITION (dt =
'20260723', hh = '10')"
+ ),
+ )
+ .await;
+ assert_eq!(
+ server
+ .table_partition_list_name_patterns(DATABASE, TABLE)
+ .len(),
+ listings,
+ "complete specifications should not read the registry"
+ );
+ assert_eq!(
+ server.table_partition_list_by_names_calls(DATABASE, TABLE),
+ vec![vec![
+ spec(&[("dt", "20260722"), ("hh", "10")]),
+ spec(&[("dt", "20260723"), ("hh", "10")]),
+ ]]
+ );
+
+ // A partial specification needs the registry, and reads it once.
+ common::exec(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} DROP PARTITION (dt = '20260722')"),
+ )
+ .await;
+ assert_eq!(
+ server
+ .table_partition_list_name_patterns(DATABASE, TABLE)
+ .len(),
+ listings + 1
+ );
+ assert_eq!(
+ server
+ .table_partition_list_by_names_calls(DATABASE, TABLE)
+ .len(),
+ 1
+ );
+ assert!(server.table_partition_specs(DATABASE, TABLE).is_empty());
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_drop_partition_leaves_a_custom_location_in_place() {
+ 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;
+ // Another engine registered dt=b somewhere else; the table directory
still has its own dt=b.
+ std::fs::write(external_dir.path().join("part-0.parquet"),
b"data").unwrap();
+ server.set_table_partition_options(
+ DATABASE,
+ TABLE,
+ &spec(&[("dt", "b")]),
+ HashMap::from([(
+ "path".to_string(),
+ format!("file://{}", external_dir.path().display()),
+ )]),
+ );
+
+ // Dropping it unregisters it and deletes nothing, least of all its own
data.
+ common::exec(
+ &context,
+ &format!("ALTER TABLE {TABLE_NAME} DROP PARTITION (dt = 'b')"),
+ )
+ .await;
+ assert_eq!(
+ server.table_partition_specs(DATABASE, TABLE),
+ vec![spec(&[("dt", "a")])]
+ );
+ assert!(external_dir.path().join("part-0.parquet").exists());
+ assert!(temp_dir.path().join("dt=b").is_dir());
+}
+
+/// `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)]
+fn sql_future_is_send<'a>(
+ context: &'a SQLContext,
+ sql: &'a str,
+) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + 'a>> {
+ Box::pin(async move {
+ let _ = context.sql(sql).await;
+ })
+}
diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs
index 308b78dc..38bd9f13 100644
--- a/crates/paimon/src/catalog/mod.rs
+++ b/crates/paimon/src/catalog/mod.rs
@@ -606,6 +606,18 @@ pub trait Catalog: Send + Sync {
})
}
+ /// Unregister table partition metadata from the catalog, keeping
directories and data files.
+ /// Missing specs are ignored so callers can safely retry the request.
+ async fn drop_partitions(
+ &self,
+ _identifier: &Identifier,
+ _partition_specs: Vec<HashMap<String, String>>,
+ ) -> Result<()> {
+ Err(Error::Unsupported {
+ message: "Catalog does not support dropping
partitions".to_string(),
+ })
+ }
+
/// Return those of the given complete partition specs that are registered.
///
/// Specs are compared with the registered values as they are, without
normalizing them.
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index b4b9a510..09f9691e 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -455,6 +455,35 @@ impl Catalog for RESTCatalog {
Ok(())
}
+ async fn drop_partitions(
+ &self,
+ identifier: &Identifier,
+ partition_specs: Vec<HashMap<String, String>>,
+ ) -> Result<()> {
+ if partition_specs.is_empty() {
+ return Ok(());
+ }
+ // The endpoint only unregisters metadata, which is the whole drop
only for a Format Table
+ // whose catalog owns the partitions; any other table would keep its
data.
+ let table = self.get_table(identifier).await?;
+ if !table.has_catalog_managed_partitions() {
+ return Err(Error::Unsupported {
+ message: format!(
+ "Dropping partitions through the REST catalog is supported
only for Format \
+ Tables with catalog-managed partitions, and {} is not
one",
+ identifier.full_name()
+ ),
+ });
+ }
+ for batch in partition_specs.chunks(PARTITION_BATCH_SIZE) {
+ self.api
+ .drop_partitions(identifier, batch.to_vec(), true)
+ .await
+ .map_err(|error| map_rest_error_for_partition_request(error,
identifier))?;
+ }
+ Ok(())
+ }
+
async fn list_partitions_by_names(
&self,
identifier: &Identifier,
@@ -656,6 +685,13 @@ fn map_rest_error_for_create_partitions(err: Error,
identifier: &Identifier) ->
),
source: None,
},
+ other => map_rest_error_for_partition_request(other, identifier),
+ }
+}
+
+/// Map a REST API error from a partition request other than a create conflict.
+fn map_rest_error_for_partition_request(err: Error, identifier: &Identifier)
-> Error {
+ match err {
Error::RestApi {
source: RestError::BadRequest { message },
} => Error::DataInvalid {
diff --git a/crates/paimon/src/table/format_partition.rs
b/crates/paimon/src/table/format_partition.rs
index 90133d3b..deba4111 100644
--- a/crates/paimon/src/table/format_partition.rs
+++ b/crates/paimon/src/table/format_partition.rs
@@ -15,8 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-//! Format Table partition names, paths and values, shared by the scan and the
catalog
-//! registrations it reads.
+//! Format Table partition names, paths and values, shared by the scan, the
catalog
+//! registrations it reads and the SQL statements that administer them.
use std::collections::HashMap;
@@ -28,14 +28,14 @@ const UNIX_EPOCH_DAYS_FROM_CE: i32 = 719_163;
/// Generates canonical names and physical paths for Format Table partitions.
#[derive(Debug, Clone, PartialEq, Eq)]
-pub(crate) struct FormatTablePartitionPaths {
+pub struct FormatTablePartitionPaths {
partition_keys: Vec<String>,
only_value_in_path: bool,
}
impl FormatTablePartitionPaths {
/// Create a helper for the declared partition-key order and physical
layout.
- pub(crate) fn new<I, S>(partition_keys: I, only_value_in_path: bool) ->
Self
+ pub fn new<I, S>(partition_keys: I, only_value_in_path: bool) -> Self
where
I: IntoIterator<Item = S>,
S: Into<String>,
@@ -47,7 +47,7 @@ impl FormatTablePartitionPaths {
}
/// Return the canonical logical partition name (`key=value/...`).
- pub(crate) fn partition_name(&self, spec: &HashMap<String, String>) ->
crate::Result<String> {
+ pub fn partition_name(&self, spec: &HashMap<String, String>) ->
crate::Result<String> {
let values = self.ordered_values(spec)?;
Ok(self
.partition_keys
@@ -98,7 +98,7 @@ impl FormatTablePartitionPaths {
}
/// Return the physical partition path relative to the table location.
- pub(crate) fn relative_path(&self, spec: &HashMap<String, String>) ->
crate::Result<String> {
+ pub fn relative_path(&self, spec: &HashMap<String, String>) ->
crate::Result<String> {
if !self.only_value_in_path {
return self.partition_name(spec);
}
@@ -150,7 +150,7 @@ impl FormatTablePartitionPaths {
}
/// Parse a raw Format Table partition value from a path or catalog
registration.
-pub(crate) fn parse_format_partition_value(value: &str, data_type: &DataType)
-> Option<Datum> {
+pub fn parse_format_partition_value(value: &str, data_type: &DataType) ->
Option<Datum> {
match data_type {
DataType::Boolean(_) => parse_partition_bool(value).map(Datum::Bool),
DataType::TinyInt(_) => value.parse::<i8>().ok().map(Datum::TinyInt),
@@ -165,7 +165,7 @@ pub(crate) fn parse_format_partition_value(value: &str,
data_type: &DataType) ->
}
/// Format a typed value for Format Table partition metadata and paths.
-pub(crate) fn format_partition_value(
+pub fn format_partition_value(
datum: &Datum,
data_type: &DataType,
default_partition_name: &str,
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 708aa6b3..bd98ba8e 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -118,6 +118,9 @@ pub use commit_message::CommitMessage;
pub use consumer_manager::ConsumerManager;
pub use cow_writer::{CopyOnWriteMergeWriter, FileInfo};
pub use data_evolution_writer::{DataEvolutionDeleteWriter,
DataEvolutionWriter};
+pub use format_partition::{
+ format_partition_value, parse_format_partition_value,
FormatTablePartitionPaths,
+};
#[cfg(feature = "fulltext")]
pub use full_text_search_builder::FullTextSearchBuilder;
use futures::stream::BoxStream;
diff --git a/crates/paimon/tests/mock_server.rs
b/crates/paimon/tests/mock_server.rs
index bd73993b..523541b8 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -1435,6 +1435,26 @@ impl RESTServer {
inner.partition_list_call_counts.remove(&key);
}
+ /// Return the partition specs registered for a table, in registration
order.
+ pub fn table_partition_specs(
+ &self,
+ database: &str,
+ table: &str,
+ ) -> Vec<HashMap<String, String>> {
+ self.inner
+ .lock()
+ .unwrap()
+ .partitions
+ .get(&format!("{database}.{table}"))
+ .map(|partitions| {
+ partitions
+ .iter()
+ .map(|partition| partition.spec.clone())
+ .collect()
+ })
+ .unwrap_or_default()
+ }
+
/// Set whether a stored table is external.
pub fn set_table_external(&self, database: &str, table: &str, is_external:
bool) {
let key = format!("{database}.{table}");
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index b3f6cf53..63729cc5 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -433,6 +433,105 @@ async fn
test_rest_catalog_create_partitions_maps_missing_table() {
));
}
+#[tokio::test]
+async fn test_rest_catalog_drop_partitions_maps_missing_table() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "missing");
+
+ let error = ctx
+ .catalog
+ .drop_partitions(
+ &identifier,
+ vec![HashMap::from([(
+ "dt".to_string(),
+ "2026-07-22".to_string(),
+ )])],
+ )
+ .await
+ .unwrap_err();
+
+ assert!(matches!(
+ error,
+ paimon::Error::TableNotExist { full_name } if full_name ==
"default.missing"
+ ));
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_rest_catalog_skips_empty_and_batches_drop_partitions() {
+ let tmp = tempfile::tempdir().unwrap();
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "managed_table");
+ add_internal_table_with_schema(
+ &ctx.server,
+ "managed_table",
+ format_table_schema(&[]),
+ &format!("file://{}", tmp.path().display()),
+ );
+ let partition_specs = (0..1001)
+ .map(|value| HashMap::from([("dt".to_string(), value.to_string())]))
+ .collect::<Vec<_>>();
+ ctx.server
+ .set_table_partitions("default", "managed_table",
partition_specs.clone());
+
+ ctx.catalog
+ .drop_partitions(&identifier, Vec::new())
+ .await
+ .unwrap();
+ ctx.catalog
+ .drop_partitions(&identifier, partition_specs.clone())
+ .await
+ .unwrap();
+
+ let calls = ctx.server.drop_partitions_calls();
+ assert_eq!(calls.len(), 2);
+ assert_eq!(calls[0].2.partition_specs, partition_specs[..1000]);
+ assert_eq!(calls[1].2.partition_specs, partition_specs[1000..]);
+ assert!(calls
+ .iter()
+ .all(|(_, _, request)| request.ignore_if_not_exists));
+}
+
+#[tokio::test]
+async fn test_rest_catalog_refuses_to_drop_partitions_it_does_not_manage() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "paimon_table");
+ let schema = Schema::builder()
+ .column("dt", DataType::VarChar(VarCharType::new(255).unwrap()))
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .partition_keys(["dt"])
+ .build()
+ .unwrap();
+ ctx.server.add_table_with_schema(
+ "default",
+ "paimon_table",
+ schema,
+ "file:///tmp/test_warehouse/default.db/paimon_table",
+ );
+
+ // Unregistering would report success and leave every data file of a
Paimon table in place.
+ let error = ctx
+ .catalog
+ .drop_partitions(
+ &identifier,
+ vec![HashMap::from([(
+ "dt".to_string(),
+ "2026-07-22".to_string(),
+ )])],
+ )
+ .await
+ .unwrap_err();
+
+ assert!(
+ matches!(
+ &error,
+ paimon::Error::Unsupported { message } if
message.contains("default.paimon_table")
+ ),
+ "{error}"
+ );
+ assert!(ctx.server.drop_partitions_calls().is_empty());
+}
+
#[tokio::test]
async fn test_rest_catalog_looks_up_partitions_by_names_in_batches() {
let ctx = setup_catalog(vec!["default"]).await;
diff --git a/docs/src/sql.md b/docs/src/sql.md
index c37ac51c..c2c02e12 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 ... DROP PARTITION`, `CALL sys.*`, Paimon time
travel, and `SET` / `RESET 'paimon.*'`.
+- 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 [...]
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.
@@ -898,6 +898,76 @@ Multiple partition key-value pairs can be specified:
ALTER TABLE paimon.my_db.events DROP PARTITION (dt = '2024-01-01', region =
'us');
```
+## Format Table Partitions
+
+A `type=format-table` table loaded from a REST Catalog can have its partitions
managed
+by the catalog instead of discovered from the directory layout. The catalog
registration
+is then the authoritative partition set: it decides what a scan reads, and a
directory
+nobody registered is not part of the table.
+
+A table opts in with `'metastore.partitioned-table' = 'true'`. The statements
below apply
+only to such a table — a partitioned internal Format Table, loaded from a REST
Catalog,
+with a non-`engine` `format-table.implementation`.
+
+A partition the catalog holds at a custom location (the partition option
`path`) is not
+read from that location yet: a scan that reaches it fails rather than reading
the table
+directory in its place.
+
+### SHOW PARTITIONS
+
+```sql
+SHOW PARTITIONS paimon.my_db.events;
+SHOW PARTITIONS paimon.my_db.events PARTITION (dt = '2024-01-01');
+```
+
+Partition names are returned in the escaped `key=value/...` form, sorted. The
optional
+`PARTITION` clause keeps only the partitions matching the given values; it may
fix any
+subset of the partition keys.
+
+### ADD PARTITION
+
+```sql
+ALTER TABLE paimon.my_db.events ADD PARTITION (dt = '2024-01-01', region =
'us');
+ALTER TABLE paimon.my_db.events ADD IF NOT EXISTS PARTITION (dt = '2024-01-01')
+ PARTITION (dt = '2024-01-02');
+```
+
+Every specification must fix all partition keys. A value is read with its
column type the
+way Paimon reads partition strings, so `month = '01'` and `month = 01` both
register the INT
+value `1`, and a BOOLEAN column accepts `t`, `true`, `y`, `yes` and `1` or
their false
+counterparts. The partitions are registered with the catalog first and their
directories
+are created afterwards, so a failure to create a directory leaves the
registration in place
+and re-running the statement with `IF NOT EXISTS` completes it. Without `IF
NOT EXISTS`, an
+already registered partition is an error. A custom `LOCATION` is not
supported: the
+directory always follows the table's partition layout.
+
+### DROP PARTITION
+
+```sql
+ALTER TABLE paimon.my_db.events DROP PARTITION (dt = '2024-01-01', region =
'us');
+ALTER TABLE paimon.my_db.events DROP IF EXISTS PARTITION (dt = '2024-01-01');
+```
+
+A specification that fixes only some of the partition keys drops every
registered
+partition it matches, and the fixed keys need not be a leading prefix — on a
+`(dt, region)` table, `DROP PARTITION (region = 'us')` drops the `us`
partition of every
+date. A specification that fixes all keys names one partition, so a missing
one is an
+error unless `IF EXISTS` is given; a partial one describes a set that is
allowed to come
+out empty.
+
+One statement may carry several specifications, each `DROP PARTITION` in its
own clause:
+
+```sql
+ALTER TABLE paimon.my_db.events DROP PARTITION (dt = '2024-01-01'),
+ DROP PARTITION (dt = '2024-01-02');
+```
+
+The catalog registration is removed first, then the directory is deleted by
the client —
+the catalog never deletes data. If the deletion fails the partition is already
invisible
+and the directory may survive; repair the file system and remove it there
rather than
+re-registering it. A partition at a custom location is only unregistered; its
directory
+is left where it is.
+
## Procedures
Use `CALL` to invoke built-in procedures. All procedures are under the `sys`
namespace.
