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 5b248b70 fix(datafusion): truncate format tables instead of reporting
success (#863)
5b248b70 is described below
commit 5b248b7016fbbea33f67adbb50e83c7d39d747d9
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Fri Sep 18 14:32:50 2026 +0800
fix(datafusion): truncate format tables instead of reporting success (#863)
---
.../datafusion/src/format_partition_analyze.rs | 69 +-----
.../datafusion/src/format_partition_ddl.rs | 61 ++++-
.../datafusion/src/format_table_truncate.rs | 180 +++++++++++++++
crates/integrations/datafusion/src/lib.rs | 1 +
crates/integrations/datafusion/src/sql_context.rs | 11 +
.../datafusion/tests/rest_format_partition_sql.rs | 249 ++++++++++++++++++++-
.../paimon/src/table/format_partition_truncate.rs | 101 +++++++++
crates/paimon/src/table/format_table_scan.rs | 20 +-
crates/paimon/src/table/mod.rs | 2 +
crates/paimon/src/table/table_commit.rs | 17 ++
docs/src/sql.md | 22 ++
11 files changed, 662 insertions(+), 71 deletions(-)
diff --git a/crates/integrations/datafusion/src/format_partition_analyze.rs
b/crates/integrations/datafusion/src/format_partition_analyze.rs
index 71a8c610..a3277d17 100644
--- a/crates/integrations/datafusion/src/format_partition_analyze.rs
+++ b/crates/integrations/datafusion/src/format_partition_analyze.rs
@@ -17,20 +17,16 @@
//! ANALYZE TABLE for Format Tables with catalog-managed partitions.
-use std::collections::HashSet;
-
use datafusion::error::{DataFusionError, Result as DFResult};
use datafusion::prelude::DataFrame;
-use datafusion::sql::sqlparser::ast::{Analyze, Expr as SqlExpr};
+use datafusion::sql::sqlparser::ast::Analyze;
use paimon::table::FormatTablePartitionStatsCollector;
use crate::error::to_datafusion_error;
use crate::format_partition_ddl::{
- ensure_catalog_managed_format_table, has_custom_location,
parse_format_partition_spec,
-};
-use crate::sql_context::{
- normalize_schema_identifier, ok_result, partition_assignment, SQLContext,
+ ensure_catalog_managed_format_table, has_custom_location,
leading_partition_prefix,
};
+use crate::sql_context::{ok_result, SQLContext};
/// `ANALYZE TABLE t [PARTITION (...)] COMPUTE STATISTICS [NOSCAN]` on a
Format Table with
/// catalog-managed partitions. Mirrors Java
`PaimonAnalyzeFormatTablePartitionsCommand`.
@@ -68,9 +64,10 @@ pub(crate) async fn execute_analyze(
.await
.map_err(to_datafusion_error)?;
ensure_catalog_managed_format_table(&table, "ANALYZE TABLE")?;
- let prefix = analyze_partition_prefix(
+ let prefix = leading_partition_prefix(
analyze.partitions.as_deref().unwrap_or_default(),
&table,
+ "ANALYZE TABLE",
enable_ident_normalization,
)?;
@@ -138,59 +135,3 @@ fn format_table_statistics_parallelism(ctx: &SQLContext)
-> usize {
.map(|value| value.max(1) as usize)
.unwrap_or(8)
}
-
-/// The values an `ANALYZE ... PARTITION (...)` clause fixes, in partition-key
order; valued
-/// columns must be a leading run of the keys, so `PARTITION (hour = '00')` is
rejected.
-fn analyze_partition_prefix(
- expressions: &[SqlExpr],
- table: &paimon::Table,
- enable_ident_normalization: bool,
-) -> DFResult<Vec<(String, String)>> {
- let partition_keys = table.schema().partition_keys();
- let mut named = HashSet::with_capacity(expressions.len());
- let mut assignments = Vec::with_capacity(expressions.len());
- for expression in expressions {
- let column = match expression {
- SqlExpr::Identifier(identifier) => {
- normalize_schema_identifier(identifier,
enable_ident_normalization)
- }
- other => {
- let (column, _) = partition_assignment(other,
enable_ident_normalization)?;
- assignments.push(other.clone());
- column
- }
- };
- if !partition_keys.contains(&column) {
- return Err(DataFusionError::Plan(format!(
- "Column '{column}' is not a partition column"
- )));
- }
- if !named.insert(column.clone()) {
- return Err(DataFusionError::Plan(format!(
- "Duplicate partition column '{column}'"
- )));
- }
- }
- let spec = parse_format_partition_spec(
- &assignments,
- table,
- false,
- Some("ANALYZE TABLE"),
- enable_ident_normalization,
- )?;
- let leading = partition_keys
- .iter()
- .take_while(|key| spec.contains_key(key.as_str()))
- .count();
- if leading != spec.len() {
- return Err(DataFusionError::Plan(format!(
- "ANALYZE TABLE {} PARTITION must give values for a leading run of
its partition \
- columns {partition_keys:?}",
- table.identifier().full_name()
- )));
- }
- Ok(partition_keys[..leading]
- .iter()
- .map(|key| (key.clone(), spec[key].clone()))
- .collect())
-}
diff --git a/crates/integrations/datafusion/src/format_partition_ddl.rs
b/crates/integrations/datafusion/src/format_partition_ddl.rs
index 5c08a359..7d0c5bed 100644
--- a/crates/integrations/datafusion/src/format_partition_ddl.rs
+++ b/crates/integrations/datafusion/src/format_partition_ddl.rs
@@ -40,7 +40,9 @@ use paimon::table::{
};
use crate::error::to_datafusion_error;
-use crate::sql_context::{is_table_not_exist, ok_result, partition_assignment,
SQLContext};
+use crate::sql_context::{
+ is_table_not_exist, normalize_schema_identifier, ok_result,
partition_assignment, SQLContext,
+};
#[derive(Debug)]
pub(crate) struct ShowPartitionsStatement {
@@ -358,6 +360,63 @@ pub(crate) async fn drop_catalog_managed_partitions(
ok_result(ctx.ctx())
}
+/// The values a `PARTITION (...)` clause of ANALYZE or TRUNCATE fixes, in
partition-key order.
+/// Valued columns must be a leading run of the keys, so `PARTITION (hour =
'00')` is rejected.
+pub(crate) fn leading_partition_prefix(
+ expressions: &[SqlExpr],
+ table: &paimon::Table,
+ operation: &str,
+ enable_ident_normalization: bool,
+) -> DFResult<Vec<(String, String)>> {
+ let partition_keys = table.schema().partition_keys();
+ let mut named = HashSet::with_capacity(expressions.len());
+ let mut assignments = Vec::with_capacity(expressions.len());
+ for expression in expressions {
+ let column = match expression {
+ SqlExpr::Identifier(identifier) => {
+ normalize_schema_identifier(identifier,
enable_ident_normalization)
+ }
+ other => {
+ let (column, _) = partition_assignment(other,
enable_ident_normalization)?;
+ assignments.push(other.clone());
+ column
+ }
+ };
+ if !partition_keys.contains(&column) {
+ return Err(DataFusionError::Plan(format!(
+ "Column '{column}' is not a partition column"
+ )));
+ }
+ if !named.insert(column.clone()) {
+ return Err(DataFusionError::Plan(format!(
+ "Duplicate partition column '{column}'"
+ )));
+ }
+ }
+ let spec = parse_format_partition_spec(
+ &assignments,
+ table,
+ false,
+ Some(operation),
+ enable_ident_normalization,
+ )?;
+ let leading = partition_keys
+ .iter()
+ .take_while(|key| spec.contains_key(key.as_str()))
+ .count();
+ if leading != spec.len() {
+ return Err(DataFusionError::Plan(format!(
+ "{operation} {} PARTITION must give values for a leading run of
its partition \
+ columns {partition_keys:?}",
+ table.identifier().full_name()
+ )));
+ }
+ Ok(partition_keys[..leading]
+ .iter()
+ .map(|key| (key.clone(), spec[key].clone()))
+ .collect())
+}
+
/// Whether the catalog registered a partition at a location of its own rather
than under the
/// table directory.
pub(crate) fn has_custom_location(partition: &paimon::spec::Partition) -> bool
{
diff --git a/crates/integrations/datafusion/src/format_table_truncate.rs
b/crates/integrations/datafusion/src/format_table_truncate.rs
new file mode 100644
index 00000000..19051239
--- /dev/null
+++ b/crates/integrations/datafusion/src/format_table_truncate.rs
@@ -0,0 +1,180 @@
+// 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.
+
+//! TRUNCATE TABLE for Format Tables.
+
+use std::collections::HashMap;
+
+use datafusion::error::{DataFusionError, Result as DFResult};
+use datafusion::prelude::DataFrame;
+use datafusion::sql::sqlparser::ast::Expr as SqlExpr;
+use paimon::catalog::{Catalog, Identifier};
+use paimon::spec::{Partition, PartitionStatistics};
+use paimon::table::{FormatTableTruncator, Table};
+
+use crate::error::to_datafusion_error;
+use crate::format_partition_ddl::{has_custom_location,
leading_partition_prefix};
+use crate::sql_context::{ok_result, SQLContext};
+
+/// `TRUNCATE TABLE t [PARTITION (...)]` on a Format Table. Deletes data files
only: directories and
+/// registrations stay. Mirrors Java `FormatTableCommit#truncateTable` and
`#truncatePartitions`.
+pub(crate) async fn execute_truncate(
+ ctx: &SQLContext,
+ catalog: &dyn Catalog,
+ identifier: &Identifier,
+ table: &Table,
+ partitions: Option<&[SqlExpr]>,
+ enable_ident_normalization: bool,
+) -> DFResult<DataFrame> {
+ let truncator = FormatTableTruncator::new(table);
+ if table.schema().partition_keys().is_empty() {
+ if partitions.is_some() {
+ return Err(DataFusionError::Plan(format!(
+ "TRUNCATE TABLE ... PARTITION requires a partitioned table,
but {} is not \
+ partitioned",
+ identifier.full_name()
+ )));
+ }
+ truncator
+ .truncate_unpartitioned()
+ .await
+ .map_err(to_datafusion_error)?;
+ return ok_result(ctx.ctx());
+ }
+ let prefix = leading_partition_prefix(
+ partitions.unwrap_or_default(),
+ table,
+ "TRUNCATE TABLE",
+ enable_ident_normalization,
+ )?;
+ // Unlike ANALYZE, a bare `PARTITION (dt)` is not read as every value:
that empties the table.
+ if partitions.is_some() && prefix.is_empty() {
+ return Err(DataFusionError::Plan(
+ "PARTITION clause requires at least one column =
value".to_string(),
+ ));
+ }
+
+ // A table with catalog-managed partitions has the partitions its catalog
holds; any other
+ // table has the partition directories below it, as its scan reads them.
+ let catalog_managed = table.has_catalog_managed_partitions();
+ let selected = if catalog_managed {
+ registered_partitions(catalog, identifier, table, &prefix).await?
+ } else {
+ truncator
+ .discover_partitions()
+ .await
+ .map_err(to_datafusion_error)?
+ .into_iter()
+ .filter(|spec| matches_prefix(spec, &prefix))
+ .collect()
+ };
+ if selected.is_empty() && !prefix.is_empty() {
+ return Err(DataFusionError::Plan(format!(
+ "Partition {prefix:?} does not exist in table {}",
+ identifier.full_name()
+ )));
+ }
+
+ let truncated_at = chrono::Utc::now().timestamp_millis();
+ let mut emptied = Vec::with_capacity(selected.len());
+ let mut failure = None;
+ for spec in selected {
+ match truncator.truncate_partition(&spec).await {
+ Ok(()) => emptied.push(spec),
+ Err(error) => {
+ failure = Some(error);
+ break;
+ }
+ }
+ }
+ // What was emptied is reported even when a later partition failed, so the
catalog stops
+ // describing files that are gone; the deletion failure is the one
returned.
+ if catalog_managed && !emptied.is_empty() {
+ let statistics = emptied
+ .iter()
+ .map(|spec| empty_statistics(spec, truncated_at))
+ .collect();
+ let reported = catalog
+ .create_partitions_with_statistics(identifier, emptied, true,
Some(statistics), true)
+ .await;
+ if let (Err(error), None) = (reported, &failure) {
+ return Err(to_datafusion_error(error));
+ }
+ }
+ if let Some(error) = failure {
+ return Err(to_datafusion_error(error));
+ }
+ ok_result(ctx.ctx())
+}
+
+/// The registered partitions under `prefix`. Nothing is deleted when one of
them sits at a custom
+/// location, which this client cannot reset to its default directory as Java
does.
+async fn registered_partitions(
+ catalog: &dyn Catalog,
+ identifier: &Identifier,
+ table: &Table,
+ prefix: &[(String, String)],
+) -> DFResult<Vec<HashMap<String, String>>> {
+ let complete = prefix.len() == table.schema().partition_keys().len();
+ let partitions = if complete {
+ let spec = prefix.iter().cloned().collect::<HashMap<_, _>>();
+ catalog
+ .list_partitions_by_names(identifier, vec![spec])
+ .await
+ } else {
+ catalog.list_partitions(identifier).await
+ }
+ .map_err(to_datafusion_error)?
+ .into_iter()
+ .filter(|partition| matches_prefix(&partition.spec, prefix))
+ .collect::<Vec<_>>();
+
+ let custom_located = partitions
+ .iter()
+ .filter(|partition| has_custom_location(partition))
+ .map(|partition| &partition.spec)
+ .collect::<Vec<_>>();
+ if !custom_located.is_empty() {
+ return Err(DataFusionError::NotImplemented(format!(
+ "TRUNCATE TABLE cannot empty partitions with a custom location in
Format Table {}: \
+ {custom_located:?}",
+ identifier.full_name()
+ )));
+ }
+ Ok(partitions
+ .into_iter()
+ .map(|partition| partition.spec)
+ .collect())
+}
+
+fn matches_prefix(spec: &HashMap<String, String>, prefix: &[(String, String)])
-> bool {
+ prefix
+ .iter()
+ .all(|(key, value)| spec.get(key) == Some(value))
+}
+
+/// What an emptied partition holds, dated to the truncation.
+fn empty_statistics(spec: &HashMap<String, String>, truncated_at: i64) ->
PartitionStatistics {
+ PartitionStatistics {
+ spec: spec.clone(),
+ record_count: 0,
+ file_size_in_bytes: 0,
+ file_count: 0,
+ last_file_creation_time: truncated_at,
+ total_buckets: Partition::UNKNOWN_TOTAL_BUCKETS,
+ }
+}
diff --git a/crates/integrations/datafusion/src/lib.rs
b/crates/integrations/datafusion/src/lib.rs
index ad1e8c32..95d1efc9 100644
--- a/crates/integrations/datafusion/src/lib.rs
+++ b/crates/integrations/datafusion/src/lib.rs
@@ -46,6 +46,7 @@ mod filter_pushdown;
mod format_partition_analyze;
mod format_partition_ddl;
mod format_partition_repair;
+mod format_table_truncate;
#[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 41a425eb..aba77b15 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -1736,6 +1736,17 @@ impl SQLContext {
}
Err(e) => return Err(to_datafusion_error(e)),
};
+ if CoreOptions::new(table.schema().options()).is_format_table() {
+ return crate::format_table_truncate::execute_truncate(
+ self,
+ catalog.as_ref(),
+ &identifier,
+ &table,
+ truncate.partitions.as_deref(),
+ enable_ident_normalization,
+ )
+ .await;
+ }
let wb = table.new_write_builder();
let commit = wb.try_new_commit().map_err(to_datafusion_error)?;
diff --git a/crates/integrations/datafusion/tests/rest_format_partition_sql.rs
b/crates/integrations/datafusion/tests/rest_format_partition_sql.rs
index e07b1f4b..78c2cdc5 100644
--- a/crates/integrations/datafusion/tests/rest_format_partition_sql.rs
+++ b/crates/integrations/datafusion/tests/rest_format_partition_sql.rs
@@ -27,7 +27,7 @@ use std::sync::Arc;
use arrow_array::{Int64Array, RecordBatch};
use arrow_schema::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
use paimon::api::ConfigResponse;
-use paimon::catalog::RESTCatalog;
+use paimon::catalog::{Catalog, Identifier, RESTCatalog};
use paimon::spec::{BigIntType, BooleanType, DataType, DateType, IntType,
Schema, VarCharType};
use paimon::{CatalogOptions, Options};
use paimon_datafusion::SQLContext;
@@ -60,15 +60,19 @@ async fn setup_rest_table(temp_dir: &TempDir, schema:
Schema) -> (RESTServer, SQ
);
server.set_table_external(DATABASE, TABLE, false);
+ let catalog = rest_catalog(&server).await;
+ let mut context = SQLContext::new();
+ context.register_catalog("paimon", catalog).await.unwrap();
+ (server, context)
+}
+
+async fn rest_catalog(server: &RESTServer) -> Arc<RESTCatalog> {
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)
+ Arc::new(RESTCatalog::new(options, true).await.unwrap())
}
fn format_table_schema(partition_columns: &[(&str, DataType)]) -> Schema {
@@ -931,6 +935,241 @@ async fn test_analyze_refuses_what_it_cannot_measure() {
.await;
}
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn
test_truncate_table_empties_registered_partitions_and_keeps_them_registered() {
+ let (temp_dir, server, context) = dt_hh_table(&[("a", "00"), ("a", "01"),
("b", "00")]).await;
+ let root = temp_dir.path();
+ write_ids(&root.join("dt=a/hh=00"), &[1, 2]);
+ write_ids(&root.join("dt=a/hh=01"), &[3]);
+ // A Java reader reads a file whatever its name ends with, so truncating
deletes it too.
+ std::fs::write(root.join("dt=b/hh=00/part-00000"), b"data").unwrap();
+ // Neither another writer's staging output nor a directory awaiting MSCK
REPAIR is table data.
+ write_ids_file(&root.join("dt=a/hh=00/_temporary/0/part-1.parquet"), &[8]);
+ write_ids(&root.join("dt=c/hh=00"), &[9]);
+ common::exec(
+ &context,
+ &format!("ANALYZE TABLE {TABLE_NAME} COMPUTE STATISTICS NOSCAN"),
+ )
+ .await;
+ assert_eq!(counts(&partition_statistics(&server)["dt=a/hh=00"]).1, 1);
+ let select = format!("SELECT id FROM {TABLE_NAME}");
+ assert_eq!(ids(&context, &select).await, [1, 2, 3]);
+
+ common::exec(&context, &format!("TRUNCATE TABLE {TABLE_NAME}")).await;
+
+ assert!(ids(&context, &select).await.is_empty());
+ assert!(!root.join("dt=a/hh=00/part-0.parquet").exists());
+ assert!(!root.join("dt=b/hh=00/part-00000").exists());
+ assert!(root.join("dt=a/hh=00/_temporary/0/part-1.parquet").exists());
+ assert!(root.join("dt=c/hh=00/part-0.parquet").exists());
+ assert_eq!(server.table_partition_specs(DATABASE, TABLE).len(), 3);
+ let emptied = partition_statistics(&server);
+ for name in ["dt=a/hh=00", "dt=a/hh=01", "dt=b/hh=00"] {
+ assert!(root.join(name).is_dir(), "{name}");
+ assert_eq!(counts(&emptied[name]), (0, 0), "{name}");
+ assert_eq!(emptied[name].file_size_in_bytes, 0, "{name}");
+ assert!(emptied[name].last_file_creation_time > 0, "{name}");
+ }
+ let calls = server.create_partitions_calls();
+ let (_, _, request) = calls.last().unwrap();
+ assert_eq!(request.replace_statistics, Some(true));
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn
test_truncate_partition_clause_empties_a_leading_run_of_partition_values() {
+ let (temp_dir, server, context) = dt_hh_table(&[("a", "00"), ("a", "01"),
("b", "00")]).await;
+ let root = temp_dir.path();
+ for directory in ["dt=a/hh=00", "dt=a/hh=01", "dt=b/hh=00"] {
+ write_ids(&root.join(directory), &[1]);
+ }
+ common::exec(
+ &context,
+ &format!("ANALYZE TABLE {TABLE_NAME} COMPUTE STATISTICS NOSCAN"),
+ )
+ .await;
+ let has_data = || {
+ ["dt=a/hh=00", "dt=a/hh=01", "dt=b/hh=00"]
+ .map(|directory|
root.join(directory).join("part-0.parquet").exists())
+ };
+
+ common::exec(
+ &context,
+ &format!("TRUNCATE TABLE {TABLE_NAME} PARTITION (dt = 'a', hh =
'01')"),
+ )
+ .await;
+ assert_eq!(has_data(), [true, false, true]);
+ let measured = partition_statistics(&server);
+ assert_eq!(counts(&measured["dt=a/hh=00"]), (UNKNOWN, 1));
+ assert_eq!(counts(&measured["dt=a/hh=01"]), (0, 0));
+
+ common::exec(
+ &context,
+ &format!("TRUNCATE TABLE {TABLE_NAME} PARTITION (dt = 'a')"),
+ )
+ .await;
+ assert_eq!(has_data(), [false, false, true]);
+ assert_eq!(
+ counts(&partition_statistics(&server)["dt=b/hh=00"]),
+ (UNKNOWN, 1)
+ );
+
+ for (clause, message) in [
+ ("PARTITION (hh = '00')", "leading run"),
+ ("PARTITION (id = 1)", "not a partition column"),
+ ("PARTITION (dt = 'zzz')", "does not exist"),
+ ("PARTITION (dt = 'b', hh = '99')", "does not exist"),
+ ("PARTITION (dt, hh)", "at least one column = value"),
+ ] {
+ common::assert_sql_error(
+ &context,
+ &format!("TRUNCATE TABLE {TABLE_NAME} {clause}"),
+ message,
+ )
+ .await;
+ }
+ assert_eq!(has_data(), [false, false, true]);
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_truncate_refuses_a_custom_location_before_deleting_anything() {
+ 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;
+ let root = temp_dir.path();
+ write_ids(&root.join("dt=a"), &[1]);
+ write_ids(&root.join("dt=b"), &[2]);
+ write_ids(external_dir.path(), &[3]);
+ server.set_table_partition_options(
+ DATABASE,
+ TABLE,
+ &spec(&[("dt", "b")]),
+ HashMap::from([(
+ "path".to_string(),
+ format!("file://{}", external_dir.path().display()),
+ )]),
+ );
+ let reports = server.create_partitions_calls().len();
+
+ common::assert_sql_error(
+ &context,
+ &format!("TRUNCATE TABLE {TABLE_NAME}"),
+ "custom location",
+ )
+ .await;
+ assert!(root.join("dt=a/part-0.parquet").exists());
+ assert!(root.join("dt=b/part-0.parquet").exists());
+ assert!(external_dir.path().join("part-0.parquet").exists());
+ assert_eq!(server.create_partitions_calls().len(), reports);
+
+ // A partition clause that leaves it out does not meet it.
+ common::exec(
+ &context,
+ &format!("TRUNCATE TABLE {TABLE_NAME} PARTITION (dt = 'a')"),
+ )
+ .await;
+ assert!(!root.join("dt=a/part-0.parquet").exists());
+ assert!(root.join("dt=b/part-0.parquet").exists());
+ assert!(external_dir.path().join("part-0.parquet").exists());
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn
test_truncate_a_format_table_that_discovers_partitions_from_directories() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let schema = Schema::builder()
+ .column("dt", varchar())
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .partition_keys(["dt"])
+ .option("type", "format-table")
+ .option("file.format", "parquet")
+ .build()
+ .unwrap();
+ let (server, context) = setup_rest_table(&temp_dir, schema).await;
+ let root = temp_dir.path();
+ write_ids(&root.join("dt=a"), &[1]);
+ write_ids(&root.join("dt=b"), &[2]);
+ // Not a partition directory, so not part of the table.
+ write_ids(&root.join("tmp/unknown"), &[3]);
+
+ common::exec(
+ &context,
+ &format!("TRUNCATE TABLE {TABLE_NAME} PARTITION (dt = 'a')"),
+ )
+ .await;
+ assert!(!root.join("dt=a/part-0.parquet").exists());
+ assert!(root.join("dt=b/part-0.parquet").exists());
+ common::assert_sql_error(
+ &context,
+ &format!("TRUNCATE TABLE {TABLE_NAME} PARTITION (dt = 'zzz')"),
+ "does not exist",
+ )
+ .await;
+
+ common::exec(&context, &format!("TRUNCATE TABLE {TABLE_NAME}")).await;
+ assert!(!root.join("dt=b/part-0.parquet").exists());
+ assert!(root.join("dt=a").is_dir() && root.join("dt=b").is_dir());
+ assert!(root.join("tmp/unknown/part-0.parquet").exists());
+ assert!(server.create_partitions_calls().is_empty());
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_truncate_an_unpartitioned_format_table() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let schema = Schema::builder()
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .option("type", "format-table")
+ .option("file.format", "parquet")
+ .build()
+ .unwrap();
+ let (_server, context) = setup_rest_table(&temp_dir, schema).await;
+ let root = temp_dir.path();
+ write_ids(root, &[1]);
+ write_ids_file(&root.join("_temporary/0/part-1.parquet"), &[2]);
+
+ common::assert_sql_error(
+ &context,
+ &format!("TRUNCATE TABLE {TABLE_NAME} PARTITION (id = 1)"),
+ "not partitioned",
+ )
+ .await;
+ assert!(root.join("part-0.parquet").exists());
+
+ common::exec(&context, &format!("TRUNCATE TABLE {TABLE_NAME}")).await;
+ assert!(!root.join("part-0.parquet").exists());
+ assert!(root.join("_temporary/0/part-1.parquet").exists());
+}
+
+#[tokio::test]
+async fn test_a_snapshot_commit_refuses_to_truncate_a_format_table() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let (server, _context) =
+ setup_rest_table(&temp_dir, format_table_schema(&[("dt",
varchar())])).await;
+ let catalog = rest_catalog(&server).await;
+ let table = catalog
+ .get_table(&Identifier::new(DATABASE, TABLE))
+ .await
+ .unwrap();
+
+ // Without snapshots such a commit finds nothing to delete, and used to
report success.
+ let commit = paimon::table::WriteBuilder::new(&table).new_commit();
+ let error = commit.truncate_table().await.unwrap_err();
+ assert!(error.to_string().contains("snapshot commit"), "{error}");
+ assert!(commit
+ .truncate_partitions(vec![HashMap::from([("dt".to_string(), None)])])
+ .await
+ .is_err());
+}
+
/// `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/table/format_partition_truncate.rs
b/crates/paimon/src/table/format_partition_truncate.rs
new file mode 100644
index 00000000..f30a2ba8
--- /dev/null
+++ b/crates/paimon/src/table/format_partition_truncate.rs
@@ -0,0 +1,101 @@
+// 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.
+
+//! Deletes the data files of a Format Table for TRUNCATE TABLE.
+
+use std::collections::HashMap;
+
+use super::format_partition::FormatTablePartitionPaths;
+use super::format_table_scan::list_format_table_files;
+use super::Table;
+use crate::spec::CoreOptions;
+
+/// Empties a Format Table or its partitions, keeping every directory and
catalog registration.
+/// Mirrors the file side of Java `FormatTableCommit#truncateTable` and
`#truncatePartitions`.
+#[derive(Debug)]
+pub struct FormatTableTruncator<'a> {
+ table: &'a Table,
+ table_path: String,
+ partition_paths: FormatTablePartitionPaths,
+ default_partition_name: String,
+}
+
+impl<'a> FormatTableTruncator<'a> {
+ pub fn new(table: &'a Table) -> Self {
+ let options = CoreOptions::new(table.schema().options());
+ let table_path = options
+ .path()
+ .unwrap_or_else(|| table.location())
+ .trim_end_matches('/')
+ .to_string();
+ let partition_paths = FormatTablePartitionPaths::new(
+ table.schema().partition_keys().iter().cloned(),
+ options.format_table_partition_only_value_in_path(),
+ );
+ let default_partition_name =
options.partition_default_name().to_string();
+ Self {
+ table,
+ table_path,
+ partition_paths,
+ default_partition_name,
+ }
+ }
+
+ /// Deletes the data files of an unpartitioned table.
+ pub async fn truncate_unpartitioned(&self) -> crate::Result<()> {
+ if !self.table.schema().partition_keys().is_empty() {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Format Table {} is partitioned, so it is truncated
partition by partition",
+ self.table.identifier().full_name()
+ ),
+ source: None,
+ });
+ }
+ self.delete_data_files(&self.table_path).await
+ }
+
+ /// Deletes the data files of one complete partition in its directory
below the table.
+ pub async fn truncate_partition(&self, spec: &HashMap<String, String>) ->
crate::Result<()> {
+ let relative_path = self.partition_paths.relative_path(spec)?;
+ self.delete_data_files(&format!("{}/{relative_path}", self.table_path))
+ .await
+ }
+
+ /// The complete partitions whose directories sit below the table
directory, for a table that
+ /// does not keep its partitions in a catalog.
+ pub async fn discover_partitions(&self) ->
crate::Result<Vec<HashMap<String, String>>> {
+ self.partition_paths
+ .discover(
+ self.table.file_io(),
+ &self.table_path,
+ &self.default_partition_name,
+ )
+ .await
+ }
+
+ /// Every non-hidden file below `directory`, whatever its extension: a
Java reader reads those
+ /// too. Staging trees such as `_temporary` hold another writer's
uncommitted output and stay.
+ async fn delete_data_files(&self, directory: &str) -> crate::Result<()> {
+ let file_io = self.table.file_io();
+ let files = list_format_table_files(file_io, directory, 0,
None).await?;
+ for file in files.iter().filter(|file| !file.is_dir) {
+ file_io.delete_file(&file.path).await?;
+ }
+ Ok(())
+ }
+}
diff --git a/crates/paimon/src/table/format_table_scan.rs
b/crates/paimon/src/table/format_table_scan.rs
index b002e88a..7297b1c5 100644
--- a/crates/paimon/src/table/format_table_scan.rs
+++ b/crates/paimon/src/table/format_table_scan.rs
@@ -457,6 +457,23 @@ pub(crate) async fn list_format_table_data_files(
root: &str,
partition_levels_below_root: usize,
format_extension: &str,
+) -> crate::Result<Vec<crate::io::FileStatus>> {
+ list_format_table_files(
+ file_io,
+ root,
+ partition_levels_below_root,
+ Some(format_extension),
+ )
+ .await
+}
+
+/// As [`list_format_table_data_files`], with every non-hidden file when
`format_extension` is
+/// `None`: the files Java treats as table data, whatever their names end with.
+pub(crate) async fn list_format_table_files(
+ file_io: &crate::io::FileIO,
+ root: &str,
+ partition_levels_below_root: usize,
+ format_extension: Option<&str>,
) -> crate::Result<Vec<crate::io::FileStatus>> {
let statuses = match file_io.list_status_recursive_stream(root,
None).await {
Ok(listing) => collect_listing(listing).await?,
@@ -471,7 +488,8 @@ pub(crate) async fn list_format_table_data_files(
}
let is_data_file =
split_parent_and_file(&status.path).is_some_and(|(_, file_name)| {
is_format_table_data_file_name(file_name)
- && file_name.to_ascii_lowercase().ends_with(format_extension)
+ && format_extension
+ .is_none_or(|extension|
file_name.to_ascii_lowercase().ends_with(extension))
});
if !is_data_file {
continue;
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index f46c835f..caa9971a 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -45,6 +45,7 @@ mod de_vector_scan;
mod dedicated_format_file_writer;
mod format_partition;
mod format_partition_stats;
+mod format_partition_truncate;
mod format_read_builder;
mod format_table_read;
mod format_table_scan;
@@ -137,6 +138,7 @@ pub use format_partition::{
format_partition_value, parse_format_partition_value,
FormatTablePartitionPaths,
};
pub use format_partition_stats::FormatTablePartitionStatsCollector;
+pub use format_partition_truncate::FormatTableTruncator;
#[cfg(feature = "fulltext")]
pub use full_text_search_builder::FullTextSearchBuilder;
use futures::stream::BoxStream;
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index a77267a7..2b747483 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -597,6 +597,7 @@ impl TableCommit {
) -> Result<()> {
// A commit validates against the existing snapshot.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+ self.ensure_not_format_table()?;
self.table.ensure_not_branch_reference_for_write()?;
if partitions.is_empty() {
@@ -678,6 +679,7 @@ impl TableCommit {
) -> Result<()> {
// A commit validates against the existing snapshot.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+ self.ensure_not_format_table()?;
self.table.ensure_not_branch_reference_for_write()?;
self.try_commit(
@@ -698,6 +700,21 @@ impl TableCommit {
.await
}
+ /// A Format Table has no snapshots, so an overwrite commit would find
nothing to delete and
+ /// report success; see [`crate::table::FormatTableTruncator`].
+ fn ensure_not_format_table(&self) -> Result<()> {
+ if CoreOptions::new(self.table.schema().options()).is_format_table() {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Format Table {} cannot be truncated through a snapshot
commit; use \
+ FormatTableTruncator or TRUNCATE TABLE in SQL",
+ self.table.identifier().full_name()
+ ),
+ });
+ }
+ Ok(())
+ }
+
/// Abort a prepared commit by deleting newly written data, changelog and
index files.
///
/// Deletion is best-effort and mirrors Python `FileStoreCommit.abort`:
missing
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 7d35a05a..f35345dd 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1023,6 +1023,28 @@ listings and footer reads alike. Set it for the session:
SET 'paimon.format-table.statistics.parallelism' = '16';
```
+### TRUNCATE TABLE
+
+```sql
+TRUNCATE TABLE paimon.my_db.events;
+TRUNCATE TABLE paimon.my_db.events PARTITION (dt = '2024-01-01');
+```
+
+Unlike the statements above, this works on any Format Table, including one
that discovers
+its partitions from the directory layout. It deletes data files only:
partition directories
+and catalog registrations stay. Every non-hidden file in a partition directory
is deleted,
+whatever its extension, because other engines read those files too; staging
entries such as
+`_temporary` are left alone.
+
+With catalog-managed partitions, the statement empties the registered
partitions and
+reports zero statistics for each of them, replacing what the catalog holds. A
directory
+nobody registered is not part of the table and is not touched. Without them,
it empties the
+partition directories found below the table, and an unpartitioned table loses
the data
+files in its table directory. `PARTITION (...)` must give values for a leading
run of the
+partition keys, as in `ANALYZE TABLE`, but a column named without a value is
rejected
+rather than read as every value. It is an error when no partition matches. A
selected partition at a custom location fails the statement before
+anything is deleted.
+
## Procedures
Use `CALL` to invoke built-in procedures. All procedures are under the `sys`
namespace.
