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 a17688d5 feat(index): support partition-scoped global index drops
(#1106)
a17688d5 is described below
commit a17688d5e85747f9a9323bd2f83f625a3a982da7
Author: QuakeWang <[email protected]>
AuthorDate: Thu Oct 15 14:17:47 2026 +0800
feat(index): support partition-scoped global index drops (#1106)
---
Cargo.lock | 1 +
crates/integrations/datafusion/Cargo.toml | 1 +
crates/integrations/datafusion/src/procedures.rs | 330 +++++++-
crates/integrations/datafusion/tests/procedures.rs | 941 ++++++++++++++++++++-
crates/paimon/src/spec/mod.rs | 2 +-
crates/paimon/src/spec/partition_utils.rs | 11 +-
.../paimon/src/table/global_index_drop_builder.rs | 441 +++++++++-
7 files changed, 1714 insertions(+), 13 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 22ea40a7..e1b2da67 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4758,6 +4758,7 @@ dependencies = [
"arrow-select",
"async-trait",
"axum",
+ "bigdecimal",
"bytes",
"chrono",
"constant_time_eq",
diff --git a/crates/integrations/datafusion/Cargo.toml
b/crates/integrations/datafusion/Cargo.toml
index a2f78ae2..dfca7278 100644
--- a/crates/integrations/datafusion/Cargo.toml
+++ b/crates/integrations/datafusion/Cargo.toml
@@ -35,6 +35,7 @@ vortex = ["paimon/vortex"]
[dependencies]
arrow-select = { workspace = true }
async-trait = "0.1"
+bigdecimal = "0.4"
chrono = "0.4"
constant_time_eq = { workspace = true }
datafusion = { workspace = true }
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index b908da7d..ce150229 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -41,6 +41,11 @@
//!
//! The `index_type` argument of the three global index procedures is
//! case-insensitive and surrounding whitespace is ignored.
+//! `drop_global_index` accepts partition alternatives such as
+//! `partitions => 'dt=2026-10-10,hr=1;dt=2026-10-11,hr=2'`; omitted keys are
unrestricted.
+//! Explicit empty or invalid partition specifications fail before deleting
any index.
+//! Timezone-less LTZ partition values use local time for non-legacy names and
UTC for
+//! legacy names. Ambiguous or nonexistent local times require an explicit
offset.
//!
//! `column_names` and `excluded_column_names` are comma-separated lists, not
SQL arrays:
//! `column_names => 'id, region'`. Java's Spark procedures take
`ARRAY<STRING>` there, but a
@@ -53,6 +58,7 @@ use std::sync::Arc;
use datafusion::arrow::array::{ArrayRef, StringArray};
use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, Schema};
use datafusion::arrow::record_batch::RecordBatch;
+use datafusion::common::ScalarValue;
use datafusion::error::{DataFusionError, Result as DFResult};
use datafusion::prelude::{DataFrame, SessionContext};
use datafusion::sql::sqlparser::ast::{
@@ -65,7 +71,7 @@ use paimon::api::{
};
use paimon::catalog::{Catalog, Identifier, RESTCatalog};
use paimon::lumina::{is_lumina_index_type, LUMINA_IDENTIFIER};
-use paimon::spec::Snapshot;
+use paimon::spec::{CoreOptions, DataType, Datum, Predicate, PredicateBuilder,
Snapshot};
use paimon::table::{
normalize_global_index_type, BranchManager, SnapshotManager, Table,
TagManager,
SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP,
@@ -173,8 +179,6 @@ fn declared_parameters(proc_name: &str) -> Option<&'static
[&'static str]> {
"create_tag_from_timestamp" => &["table", "tag", "timestamp"],
"rename_branch" => &["table", "from_branch", "to_branch"],
"create_global_index" => &["table", "index_column", "index_type",
"options"],
- // `partitions` is declared but not yet implemented; it still reaches
its
- // own "not supported yet" error rather than being reported as unknown.
"drop_global_index" => &[
"table",
"index_column",
@@ -776,17 +780,17 @@ async fn proc_drop_global_index(
"unsupported global index type '{index_type_arg}'; supported:
{SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP}"
)));
}
- if args.contains_key("partitions") {
- return Err(DataFusionError::NotImplemented(
- "drop_global_index partitions are not supported yet".to_string(),
- ));
- }
let dry_run = bool_arg(args, "dry_run")?;
let mut builder = table.new_global_index_drop_builder();
builder.with_index_column(index_column);
builder.with_index_type(index_type);
builder.with_dry_run(dry_run);
+ if let Some(partitions) = args.get("partitions") {
+ builder
+ .with_partition_filter(parse_global_index_partitions(&table,
partitions)?)
+ .map_err(to_datafusion_error)?;
+ }
let dropped = builder.execute().await.map_err(to_datafusion_error)?;
if dry_run {
return utf8_result(
@@ -800,6 +804,193 @@ async fn proc_drop_global_index(
ok_result(ctx)
}
+/// Semicolon-separated alternatives, each containing comma-separated
key=value constraints.
+fn parse_global_index_partitions(table: &Table, partitions: &str) ->
DFResult<Predicate> {
+ let partition_fields = table.schema().partition_fields();
+ if partition_fields.is_empty() {
+ return Err(DataFusionError::Plan(
+ "Cannot filter global index partitions on an unpartitioned
table".to_string(),
+ ));
+ }
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let options = CoreOptions::new(table.schema().options());
+ let mut alternatives = Vec::new();
+ for partition in partitions.split(';') {
+ let mut constraints = Vec::new();
+ let mut keys = std::collections::HashSet::new();
+ for entry in partition.split(',') {
+ let (key, value) = entry.split_once('=').ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "Invalid partitions entry '{entry}'. Expected key=value"
+ ))
+ })?;
+ // Java ParameterUtils uses String.trim(), not Unicode whitespace
trimming.
+ let key = key.trim_matches(|ch| ch <= ' ');
+ let value = value.trim_matches(|ch| ch <= ' ');
+ let field = partition_fields
+ .iter()
+ .find(|field| field.name() == key)
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!("Column '{key}' is not a
partition column"))
+ })?;
+ if !keys.insert(key) {
+ return Err(DataFusionError::Plan(format!(
+ "Duplicate partition column '{key}'"
+ )));
+ }
+ if value.is_empty() {
+ return Err(DataFusionError::Plan(format!(
+ "Empty partitions value for column '{key}'"
+ )));
+ }
+ let predicate = if value == options.partition_default_name() {
+ pb.is_null(key)
+ } else {
+ let datum = parse_global_index_partition_value(
+ value,
+ field.data_type(),
+ options.legacy_partition_name(),
+ )
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "Invalid partitions value '{value}' for column '{key}'
with type {:?}",
+ field.data_type()
+ ))
+ })?;
+ pb.equal(key, datum)
+ };
+ constraints.push(predicate.map_err(to_datafusion_error)?);
+ }
+ alternatives.push(Predicate::and(constraints));
+ }
+ Ok(Predicate::or(alternatives))
+}
+
+fn parse_global_index_partition_value(
+ value: &str,
+ data_type: &DataType,
+ legacy_partition_name: bool,
+) -> Option<Datum> {
+ match data_type {
+ DataType::Char(typ) => {
+ (value.chars().count() <= typ.length()).then(||
Datum::String(value.to_string()))
+ }
+ DataType::VarChar(typ) => (value.chars().count() <= typ.length() as
usize)
+ .then(|| Datum::String(value.to_string())),
+ DataType::Binary(_) => Some(Datum::Bytes(value.as_bytes().to_vec())),
+ DataType::VarBinary(typ) => {
+ (value.len() <= typ.length() as usize).then(||
Datum::Bytes(value.as_bytes().to_vec()))
+ }
+ DataType::Float(_) => {
+ // Match TypeUtils.castFromString's check for FLOAT precision loss.
+ let double = value.parse::<f64>().ok()?;
+ let float = value.parse::<f32>().ok()?;
+ (f64::from(float) == double
+ || (float.is_nan() && double.is_nan())
+ || paimon::spec::format_java_float(float).parse::<f64>().ok()?
== double)
+ .then_some(Datum::Float(float))
+ }
+ DataType::Double(_) => value.parse().ok().map(Datum::Double),
+ DataType::Decimal(typ) => {
+ // Java Decimal.fromBigDecimal rounds HALF_UP; Arrow's string cast
truncates.
+ if value.contains('_') {
+ return None;
+ }
+ let decimal = value
+ .parse::<bigdecimal::BigDecimal>()
+ .ok()?
+ .with_scale_round(i64::from(typ.scale()),
bigdecimal::RoundingMode::HalfUp);
+ let unscaled = decimal
+ .as_bigint_and_exponent()
+ .0
+ .to_string()
+ .parse::<i128>()
+ .ok()?;
+ if unscaled
+ .unsigned_abs()
+ .checked_ilog10()
+ .map_or(1, |log| log + 1)
+ > typ.precision()
+ {
+ return None;
+ }
+ Some(Datum::Decimal {
+ unscaled,
+ precision: typ.precision(),
+ scale: typ.scale(),
+ })
+ }
+ DataType::Time(_) => {
+ paimon::table::parse_format_partition_value(value,
data_type).or_else(|| {
+ let arrow_type =
paimon::arrow::paimon_type_to_arrow(data_type).ok()?;
+ let scalar = ScalarValue::try_from_string(value.to_string(),
&arrow_type).ok()?;
+ crate::filter_pushdown::scalar_to_datum(&scalar, data_type)
+ })
+ }
+ DataType::Timestamp(typ) => {
+ parse_global_index_partition_timestamp(value, data_type,
typ.precision(), false)
+ }
+ DataType::LocalZonedTimestamp(typ) =>
parse_global_index_partition_timestamp(
+ value,
+ data_type,
+ typ.precision(),
+ !legacy_partition_name,
+ ),
+ _ => paimon::table::parse_format_partition_value(value, data_type),
+ }
+}
+
+fn parse_global_index_partition_timestamp(
+ value: &str,
+ data_type: &DataType,
+ precision: u32,
+ local_timezone: bool,
+) -> Option<Datum> {
+ if value.bytes().all(|byte| byte.is_ascii_digit()) {
+ // Java only accepts numeric timestamps at these precisions.
+ if !matches!(precision, 0 | 3 | 6 | 9) {
+ return None;
+ }
+ let arrow_type = paimon::arrow::paimon_type_to_arrow(data_type).ok()?;
+ let scalar = ScalarValue::Int64(Some(value.parse().ok()?))
+ .cast_to(&arrow_type)
+ .ok()?;
+ return crate::filter_pushdown::scalar_to_datum(&scalar, data_type);
+ }
+ // Arrow accepts extra fractional digits by truncating them; reject any
loss of identity.
+ if value.split_once('.').is_some_and(|(_, fraction)| {
+ fraction
+ .bytes()
+ .take_while(u8::is_ascii_digit)
+ .skip(9)
+ .any(|byte| byte != b'0')
+ }) {
+ return None;
+ }
+ // Legacy partition names omit seconds when both seconds and fractions are
zero.
+ let value = chrono::NaiveDateTime::parse_from_str(value, "%Y-%m-%dT%H:%M")
+ .map(|value| value.format("%Y-%m-%dT%H:%M:%S").to_string())
+ .unwrap_or_else(|_| value.to_string());
+ let timestamp = if local_timezone {
+ // Non-legacy LTZ names use the system-local timezone. Preserve
explicit offsets
+ // and fail on ambiguous/nonexistent local times instead of falling
back to UTC.
+
datafusion::arrow::compute::kernels::cast_utils::string_to_datetime(&chrono::Local,
&value)
+ .ok()?
+ .with_timezone(&chrono::Utc)
+ } else {
+
datafusion::arrow::compute::kernels::cast_utils::string_to_datetime(&chrono::Utc,
&value)
+ .ok()?
+ };
+ // Stored partitions can retain fractions beyond the declared precision or
Arrow unit.
+ let millis = timestamp.timestamp_millis();
+ let nanos = (timestamp.timestamp_subsec_nanos() % 1_000_000) as i32;
+ match data_type {
+ DataType::Timestamp(_) => Some(Datum::Timestamp { millis, nanos }),
+ DataType::LocalZonedTimestamp(_) => Some(Datum::LocalZonedTimestamp {
millis, nanos }),
+ _ => None,
+ }
+}
+
/// Canonicalize a procedure's `index_type` argument: trim, then lowercase.
/// Mirrors `indexType.toLowerCase(Locale.ROOT).trim()` in Java's Flink and
Spark
/// `CreateGlobalIndexProcedure` / `DropGlobalIndexProcedure`. Normalizing at
this
@@ -1272,6 +1463,129 @@ mod tests {
use paimon::io::FileIOBuilder;
use paimon::spec::CommitKind;
+ #[test]
+ fn test_global_index_partition_values_are_typed_and_strict() {
+ use paimon::spec::{
+ CharType, DateType, DecimalType, FloatType, IntType, TimeType,
TimestampType,
+ VarCharType,
+ };
+ assert_eq!(
+ parse_global_index_partition_value("01",
&DataType::Int(IntType::new()), true),
+ Some(Datum::Int(1))
+ );
+ for text in ["1.5", "2147483648", "bad"] {
+ assert!(
+ parse_global_index_partition_value(text,
&DataType::Int(IntType::new()), true)
+ .is_none()
+ );
+ }
+ assert!(parse_global_index_partition_value(
+ "2026-02-30",
+ &DataType::Date(DateType::new()),
+ true
+ )
+ .is_none());
+ assert_eq!(
+ parse_global_index_partition_value(
+ "1970-01-02",
+ &DataType::Date(DateType::new()),
+ true
+ ),
+ Some(Datum::Date(1))
+ );
+ assert_eq!(
+ parse_global_index_partition_value(
+ "a=b",
+ &DataType::VarChar(VarCharType::new(3).unwrap()),
+ true,
+ ),
+ Some(Datum::String("a=b".into()))
+ );
+ assert!(parse_global_index_partition_value(
+ "abcd",
+ &DataType::Char(CharType::new(3).unwrap()),
+ true,
+ )
+ .is_none());
+ assert_eq!(
+ parse_global_index_partition_value(
+ "12:00:00.123",
+ &DataType::Time(TimeType::new(3).unwrap()),
+ true,
+ ),
+ Some(Datum::Time(43_200_123))
+ );
+ let decimal = DataType::Decimal(DecimalType::new(4, 2).unwrap());
+ for (text, unscaled) in [("1.235", 124), ("-1.235", -124), ("1.235e1",
1235)] {
+ assert_eq!(
+ parse_global_index_partition_value(text, &decimal, true),
+ Some(Datum::Decimal {
+ unscaled,
+ precision: 4,
+ scale: 2
+ })
+ );
+ }
+ for text in ["99.995", "bad", "1_2.34"] {
+ assert!(parse_global_index_partition_value(text, &decimal,
true).is_none());
+ }
+ let timestamp = DataType::Timestamp(TimestampType::new(1).unwrap());
+ assert_eq!(
+ parse_global_index_partition_value("1970-01-01 00:00:00.199",
×tamp, true),
+ Some(Datum::Timestamp {
+ millis: 199,
+ nanos: 0
+ })
+ );
+ assert!(parse_global_index_partition_value("123", ×tamp,
true).is_none());
+ assert!(parse_global_index_partition_value("bad", ×tamp,
true).is_none());
+ let float = DataType::Float(FloatType::new());
+ assert_eq!(
+ parse_global_index_partition_value("1.1", &float, true),
+ Some(Datum::Float(1.1))
+ );
+ for text in ["1e100", "1e-100", "1.234567891"] {
+ assert!(parse_global_index_partition_value(text, &float,
true).is_none());
+ }
+ }
+
+ #[test]
+ fn test_global_index_timestamp_partition_values_preserve_identity() {
+ use paimon::spec::{LocalZonedTimestampType, TimestampType};
+
+ for precision in 0..=9 {
+ for zoned in [false, true] {
+ let data_type = if zoned {
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(precision).unwrap())
+ } else {
+ DataType::Timestamp(TimestampType::new(precision).unwrap())
+ };
+ for (text, millis, nanos) in [
+ ("1970-01-01T00:00:00.123456789Z", 123, 456_789),
+ ("1969-12-31T23:59:59.999999999Z", -1, 999_999),
+ ("2500-01-01T00:00:00Z", 16_725_225_600_000, 0),
+ ] {
+ let expected = if zoned {
+ Datum::LocalZonedTimestamp { millis, nanos }
+ } else {
+ Datum::Timestamp { millis, nanos }
+ };
+ assert_eq!(
+ parse_global_index_partition_value(text, &data_type,
true),
+ Some(expected),
+ "{data_type:?}: {text}"
+ );
+ }
+ assert!(parse_global_index_partition_value(
+ "1970-01-01T00:00:00.1234567891Z",
+ &data_type,
+ true,
+ )
+ .is_none());
+ }
+ }
+ }
+
fn test_file_io() -> paimon::io::FileIO {
FileIOBuilder::new("memory").build().unwrap()
}
diff --git a/crates/integrations/datafusion/tests/procedures.rs
b/crates/integrations/datafusion/tests/procedures.rs
index abc06c63..21bfc73e 100644
--- a/crates/integrations/datafusion/tests/procedures.rs
+++ b/crates/integrations/datafusion/tests/procedures.rs
@@ -1133,7 +1133,11 @@ async fn
test_drop_global_index_dry_run_rejects_invalid_args() {
("dry_run => NULL", "Unsupported argument value"),
(
"dry_run => true, partitions => 'pt=p1'",
- "drop_global_index partitions are not supported yet",
+ "unpartitioned table",
+ ),
+ (
+ "dry_run => true, partitions => NULL",
+ "Unsupported argument value",
),
] {
assert_sql_error(
@@ -1148,6 +1152,941 @@ async fn
test_drop_global_index_dry_run_rejects_invalid_args() {
}
}
+async fn setup_partitioned_global_index_table(
+ table_name: &str,
+ populated: bool,
+) -> (
+ tempfile::TempDir,
+ paimon_datafusion::SQLContext,
+ paimon::Table,
+) {
+ let (tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ exec(&sql_context, "CREATE SCHEMA paimon.test_db").await;
+ exec(
+ &sql_context,
+ &format!(
+ "CREATE TABLE paimon.test_db.{table_name} (id INT, dt DATE, name
VARCHAR(100), hr INT) \
+ PARTITIONED BY (hr, dt) WITH (\
+ 'row-tracking.enabled' = 'true',\
+ 'data-evolution.enabled' = 'true',\
+ 'global-index.enabled' = 'true',\
+ 'sorted-index.records-per-range' = '10'\
+ )"
+ ),
+ )
+ .await;
+ if populated {
+ exec(
+ &sql_context,
+ &format!(
+ "INSERT INTO paimon.test_db.{table_name} (id, dt, name, hr)
VALUES \
+ (1, DATE '2026-10-10', 'a', 1), (2, DATE '2026-10-10', 'b',
2), \
+ (3, DATE '2026-10-11', 'c', 1), (4, DATE '2026-10-11', 'd',
2), \
+ (5, DATE '2026-10-12', 'e', 1), (6, NULL, 'f', 1)"
+ ),
+ )
+ .await;
+ for (column, index_type) in [
+ ("id", "btree"),
+ ("name", "btree"),
+ ("id,name", "btree"),
+ ("id", "bitmap"),
+ ] {
+ exec(
+ &sql_context,
+ &format!(
+ "CALL sys.create_global_index(table =>
'test_db.{table_name}', \
+ index_column => '{column}', index_type => '{index_type}')"
+ ),
+ )
+ .await;
+ }
+ }
+ let table = catalog
+ .get_table(&Identifier::new("test_db", table_name))
+ .await
+ .unwrap();
+ (tmp, sql_context, table)
+}
+
+async fn global_index_entries(
+ table: &paimon::Table,
+) -> BTreeMap<String, paimon::spec::IndexManifestEntry> {
+ let sm = table.snapshot_manager();
+ let snapshot = sm.get_latest_snapshot().await.unwrap().unwrap();
+ let Some(manifest_name) = snapshot.index_manifest() else {
+ return BTreeMap::new();
+ };
+ paimon::spec::IndexManifest::read(table.file_io(),
&sm.manifest_path(manifest_name))
+ .await
+ .unwrap()
+ .into_iter()
+ .map(|entry| (entry.index_file.file_name.clone(), entry))
+ .collect()
+}
+
+#[tokio::test]
+async fn
test_drop_global_index_partitions_preserve_other_partitions_and_definitions() {
+ let day = |text: &str| {
+ (chrono::NaiveDate::parse_from_str(text, "%Y-%m-%d").unwrap()
+ - chrono::DateTime::UNIX_EPOCH.date_naive())
+ .num_days() as i32
+ };
+ let d10 = Some(day("2026-10-10"));
+ let d11 = Some(day("2026-10-11"));
+ for (case, columns, partitions, selected) in [
+ (
+ "dates",
+ "id",
+ "dt=2026-10-10;dt=2026-10-11",
+ vec![(1, d10), (2, d10), (1, d11), (2, d11)],
+ ),
+ (
+ "pairs",
+ "id",
+ " dt = 2026-10-10, hr = 01 ; hr=2,dt=2026-10-11 ",
+ vec![(1, d10), (2, d11)],
+ ),
+ (
+ "composite",
+ "id,name",
+ "dt=2026-10-10,hr=1;dt=2026-10-11,hr=2",
+ vec![(1, d10), (2, d11)],
+ ),
+ ("null", "id", "dt=__DEFAULT_PARTITION__", vec![(1, None)]),
+ ] {
+ let name = format!("drop_partitions_{case}");
+ let (tmp, sql_context, table) =
setup_partitioned_global_index_table(&name, true).await;
+ let before = table_files(tmp.path());
+ let entries = global_index_entries(&table).await;
+ let id = table
+ .schema()
+ .fields()
+ .iter()
+ .find(|field| field.name() == "id")
+ .unwrap()
+ .id();
+ let name_id = table
+ .schema()
+ .fields()
+ .iter()
+ .find(|field| field.name() == "name")
+ .unwrap()
+ .id();
+ let extra_ids = if columns == "id" {
+ vec![]
+ } else {
+ vec![name_id]
+ };
+ let mut expected = entries.clone();
+ expected.retain(|_, entry| {
+ let row =
paimon::spec::BinaryRow::from_serialized_bytes(&entry.partition).unwrap();
+ let partition = (
+ row.get_int(0).unwrap(),
+ if row.is_null_at(1) {
+ None
+ } else {
+ Some(row.get_int(1).unwrap())
+ },
+ );
+ let target = entry.index_file.index_type == "btree"
+ && entry
+ .index_file
+ .global_index_meta
+ .as_ref()
+ .is_some_and(|meta| {
+ meta.index_field_id == id
+ &&
meta.extra_field_ids.as_deref().unwrap_or_default() == extra_ids
+ })
+ && selected.contains(&partition);
+ !target
+ });
+ let matched = entries.len() - expected.len();
+ assert_eq!(matched, selected.len(), "{case}");
+ let snapshot_id = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id();
+ let call = format!(
+ "CALL sys.drop_global_index(table => 'test_db.{name}',
index_column => '{columns}', \
+ partitions => '{partitions}'"
+ );
+ let result = sql_context
+ .sql(&format!("{call}, dry_run => true)"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ string_value(result[0].column(0).as_ref(), 0),
+ format!("Would drop {matched} global index file(s)")
+ );
+ assert_eq!(table_files(tmp.path()), before, "{case}");
+ exec(&sql_context, &format!("{call})")).await;
+ assert_eq!(global_index_entries(&table).await, expected, "{case}");
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ snapshot_id + 1
+ );
+ }
+}
+
+#[tokio::test]
+async fn test_drop_global_index_partitions_no_match_does_not_commit() {
+ for populated in [false, true] {
+ let (tmp, sql_context, _) =
+ setup_partitioned_global_index_table("drop_partitions_empty",
populated).await;
+ let before = table_files(tmp.path());
+ for dry_run in [true, false] {
+ let result = sql_context.sql(&format!(
+ "CALL sys.drop_global_index(table =>
'test_db.drop_partitions_empty', index_column => 'id', \
+ partitions => 'dt=2026-10-09', dry_run => {dry_run})"
+ )).await.unwrap().collect().await.unwrap();
+ assert_eq!(
+ string_value(result[0].column(0).as_ref(), 0),
+ if dry_run {
+ "Would drop 0 global index file(s)"
+ } else {
+ "OK"
+ }
+ );
+ assert_eq!(table_files(tmp.path()), before);
+ }
+ }
+}
+
+#[tokio::test]
+async fn test_drop_global_index_partitions_string_keys_and_custom_null_name() {
+ let (tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ exec(&sql_context, "CREATE SCHEMA paimon.test_db").await;
+ exec(
+ &sql_context,
+ "CREATE TABLE paimon.test_db.drop_string_partitions (id INT, dt
VARCHAR(20)) \
+ PARTITIONED BY (dt) WITH ('row-tracking.enabled'='true',
'data-evolution.enabled'='true', \
+ 'global-index.enabled'='true', 'partition.default-name'='NULL_PART')",
+ )
+ .await;
+ exec(&sql_context,
+ "INSERT INTO paimon.test_db.drop_string_partitions (id, dt) VALUES \
+ (1, '2026-10-10'), (2, '2026-10-11'), (3, '2026-10-12'), (4, 'a=b'),
(5, NULL), (6, 'NULL')"
+ ).await;
+ exec(&sql_context,
+ "CALL sys.create_global_index(table =>
'test_db.drop_string_partitions', index_column => 'id')"
+ ).await;
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "drop_string_partitions"))
+ .await
+ .unwrap();
+ let mut expected = global_index_entries(&table).await;
+ assert_eq!(expected.len(), 6);
+ for (partitions, selected, matched) in [
+ (
+ "dt=2026-10-10;dt=2026-10-11",
+ vec![Some("2026-10-10"), Some("2026-10-11")],
+ 2,
+ ),
+ ("dt=a=b", vec![Some("a=b")], 1),
+ ("dt=NULL_PART", vec![None], 1),
+ ] {
+ let before = table_files(tmp.path());
+ let call = format!(
+ "CALL sys.drop_global_index(table =>
'test_db.drop_string_partitions', \
+ index_column => 'id', partitions => '{partitions}'"
+ );
+ let result = sql_context
+ .sql(&format!("{call}, dry_run => true)"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ string_value(result[0].column(0).as_ref(), 0),
+ format!("Would drop {matched} global index file(s)")
+ );
+ assert_eq!(table_files(tmp.path()), before);
+ exec(&sql_context, &format!("{call})")).await;
+ expected.retain(|_, entry| {
+ let row =
paimon::spec::BinaryRow::from_serialized_bytes(&entry.partition).unwrap();
+ let dt = if row.is_null_at(0) {
+ None
+ } else {
+ Some(row.get_string(0).unwrap())
+ };
+ !selected.contains(&dt)
+ });
+ assert_eq!(global_index_entries(&table).await, expected);
+ }
+}
+
+#[tokio::test]
+async fn test_drop_global_index_floating_partition_identity() {
+ for sql_type in ["FLOAT", "DOUBLE"] {
+ assert_floating_partition_drops(sql_type, &["-0.0", "NaN", "0.0",
"1.5"]).await;
+ }
+}
+
+#[tokio::test]
+async fn test_drop_global_index_float_generated_partition_names() {
+ assert_floating_partition_drops("FLOAT", &["147318.12", "-2096723.2",
"1.4E-45"]).await;
+}
+
+async fn assert_floating_partition_drops(sql_type: &str, values: &[&str]) {
+ let (tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ exec(&sql_context, "CREATE SCHEMA paimon.test_db").await;
+ exec(
+ &sql_context,
+ &format!(
+ "CREATE TABLE paimon.test_db.drop_float_partitions (id INT, f
{sql_type}) \
+ PARTITIONED BY (f) WITH ('row-tracking.enabled'='true',
'data-evolution.enabled'='true', \
+ 'global-index.enabled'='true')"
+ ),
+ )
+ .await;
+ let rows = values
+ .iter()
+ .enumerate()
+ .map(|(i, value)| format!("({}, CAST('{value}' AS {sql_type}))", i +
1))
+ .collect::<Vec<_>>()
+ .join(", ");
+ exec(
+ &sql_context,
+ &format!("INSERT INTO paimon.test_db.drop_float_partitions (id, f)
VALUES {rows}"),
+ )
+ .await;
+ exec(&sql_context,
+ "CALL sys.create_global_index(table =>
'test_db.drop_float_partitions', index_column => 'id')"
+ ).await;
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "drop_float_partitions"))
+ .await
+ .unwrap();
+ let directory_names =
std::fs::read_dir(table.location().strip_prefix("file://").unwrap())
+ .unwrap()
+ .map(|entry| entry.unwrap().file_name().into_string().unwrap())
+ .filter(|name| name.starts_with("f="))
+ .collect::<std::collections::BTreeSet<_>>();
+ assert_eq!(
+ directory_names,
+ values.iter().map(|value| format!("f={value}")).collect()
+ );
+ let mut expected = global_index_entries(&table).await;
+ assert_eq!(expected.len(), values.len());
+ if sql_type == "FLOAT" {
+ let before = table_files(tmp.path());
+ for partitions in ["f=1e100", "f=1e-100", "f=1.234567891",
"f=-0.0;f=bad"] {
+ for dry_run in [true, false] {
+ assert_sql_error(
+ &sql_context,
+ &format!(
+ "CALL sys.drop_global_index(table =>
'test_db.drop_float_partitions', \
+ index_column => 'id', partitions => '{partitions}',
dry_run => {dry_run})"
+ ),
+ "Invalid partitions value",
+ )
+ .await;
+ assert_eq!(table_files(tmp.path()), before);
+ }
+ }
+ }
+ for value in values {
+ // Use the directory name produced by the writer as the procedure
parameter.
+ let partition_name =
directory_names.get(&format!("f={value}")).unwrap();
+ let selected = if sql_type == "FLOAT" {
+ f64::from(value.parse::<f32>().unwrap())
+ } else {
+ value.parse::<f64>().unwrap()
+ };
+ let before = table_files(tmp.path());
+ let snapshot_id = table
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap()
+ .unwrap();
+ let call = format!(
+ "CALL sys.drop_global_index(table =>
'test_db.drop_float_partitions', \
+ index_column => 'id', partitions => '{partition_name}'"
+ );
+ let result = sql_context
+ .sql(&format!("{call}, dry_run => true)"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ string_value(result[0].column(0).as_ref(), 0),
+ "Would drop 1 global index file(s)",
+ "{sql_type}: {partition_name}"
+ );
+ assert_eq!(table_files(tmp.path()), before);
+ exec(&sql_context, &format!("{call})")).await;
+ let previous_len = expected.len();
+ expected.retain(|_, entry| {
+ let row =
paimon::spec::BinaryRow::from_serialized_bytes(&entry.partition).unwrap();
+ let actual = if sql_type == "FLOAT" {
+ f64::from(row.get_float(0).unwrap())
+ } else {
+ row.get_double(0).unwrap()
+ };
+ !(actual.to_bits() == selected.to_bits() || (actual.is_nan() &&
selected.is_nan()))
+ });
+ assert_eq!(expected.len(), previous_len - 1);
+ assert_eq!(
+ global_index_entries(&table).await,
+ expected,
+ "{sql_type}: {partition_name}"
+ );
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap(),
+ Some(snapshot_id + 1)
+ );
+ }
+}
+
+#[tokio::test]
+async fn test_drop_global_index_native_timestamp_partition_identity() {
+ use paimon::spec::{DataType, Datum, IntType, LocalZonedTimestampType,
Schema, TimestampType};
+
+ for precision in [1, 2, 4, 5, 7, 8] {
+ for zoned in [false, true] {
+ let (tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ exec(&sql_context, "CREATE SCHEMA paimon.test_db").await;
+ let data_type = if zoned {
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(precision).unwrap())
+ } else {
+ DataType::Timestamp(TimestampType::new(precision).unwrap())
+ };
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("other", DataType::Int(IntType::new()))
+ .column("ts", data_type.clone())
+ .partition_keys(["ts"])
+ .options([
+ ("row-tracking.enabled".into(), "true".into()),
+ ("data-evolution.enabled".into(), "true".into()),
+ ("global-index.enabled".into(), "true".into()),
+ ("partition.legacy-name".into(), "false".into()),
+ ])
+ .build()
+ .unwrap();
+ let identifier = Identifier::new("test_db",
"drop_native_timestamp");
+ catalog
+ .create_table(&identifier, schema, false)
+ .await
+ .unwrap();
+ let (first, selected, millis, nanos) = match precision {
+ 1 | 2 => ("100", "199", 199, 0),
+ 4 | 5 => ("123400", "123499", 123, 499_000),
+ _ => ("123456700", "123456799", 123, 456_799),
+ };
+ exec(
+ &sql_context,
+ &format!(
+ "INSERT INTO paimon.test_db.drop_native_timestamp (id,
other, ts) VALUES \
+ (1, 10, TIMESTAMP '1970-01-01 00:00:00.{first}Z'), \
+ (2, 20, TIMESTAMP '1970-01-01 00:00:00.{selected}Z')"
+ ),
+ )
+ .await;
+ for (column, index_type) in [("id", "btree"), ("other", "btree"),
("id", "bitmap")] {
+ exec(
+ &sql_context,
+ &format!(
+ "CALL sys.create_global_index(table =>
'test_db.drop_native_timestamp', \
+ index_column => '{column}', index_type => '{index_type}')"
+ ),
+ )
+ .await;
+ }
+ let table = catalog.get_table(&identifier).await.unwrap();
+ assert_eq!(table.schema().partition_fields()[0].data_type(),
&data_type);
+ assert_eq!(
+ row_count(
+ &sql_context,
+ &format!(
+ "SELECT * FROM
paimon.test_db.`drop_native_timestamp$partitions` \
+ WHERE partition LIKE 'ts=%.{selected}'"
+ )
+ )
+ .await,
+ 1
+ );
+ let selected_datum = if zoned {
+ Datum::LocalZonedTimestamp { millis, nanos }
+ } else {
+ Datum::Timestamp { millis, nanos }
+ };
+ let mut expected = global_index_entries(&table).await;
+ assert_eq!(expected.len(), 6);
+ expected.retain(|_, entry| {
+ let row =
paimon::spec::BinaryRow::from_serialized_bytes(&entry.partition).unwrap();
+ !(entry.index_file.index_type == "btree"
+ && entry
+ .index_file
+ .global_index_meta
+ .as_ref()
+ .unwrap()
+ .index_field_id
+ == 0
+ && row.get_datum(0, &data_type).unwrap().as_ref() ==
Some(&selected_datum))
+ });
+ assert_eq!(expected.len(), 5);
+ let before = table_files(tmp.path());
+ let snapshot_id = table
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap()
+ .unwrap();
+ let suffix = if zoned { "Z" } else { "" };
+ let absent = format!("{}8", &selected[..selected.len() - 1]);
+ let no_match = format!(
+ "CALL sys.drop_global_index(table =>
'test_db.drop_native_timestamp', \
+ index_column => 'id', partitions =>
'ts=1970-01-01T00:00:00.{absent}{suffix}'"
+ );
+ let batches = sql_context
+ .sql(&format!("{no_match}, dry_run => true)"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ string_value(batches[0].column(0).as_ref(), 0),
+ "Would drop 0 global index file(s)"
+ );
+ exec(&sql_context, &format!("{no_match})")).await;
+ assert_eq!(table_files(tmp.path()), before);
+ for dry_run in [true, false] {
+ assert_sql_error(&sql_context, &format!(
+ "CALL sys.drop_global_index(table =>
'test_db.drop_native_timestamp', \
+ index_column => 'id', partitions =>
'ts=1970-01-01T00:00:00.{selected}{suffix};\
+ ts=1970-01-01T00:00:00.1234567891{suffix}', dry_run =>
{dry_run})"
+ ), "Invalid partitions value").await;
+ assert_eq!(table_files(tmp.path()), before);
+ }
+ let call = format!(
+ "CALL sys.drop_global_index(table =>
'test_db.drop_native_timestamp', \
+ index_column => 'id', partitions =>
'ts=1970-01-01T00:00:00.{selected}{suffix}'"
+ );
+ let batches = sql_context
+ .sql(&format!("{call}, dry_run => true)"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ string_value(batches[0].column(0).as_ref(), 0),
+ "Would drop 1 global index file(s)"
+ );
+ assert_eq!(table_files(tmp.path()), before);
+ exec(&sql_context, &format!("{call})")).await;
+ assert_eq!(
+ global_index_entries(&table).await,
+ expected,
+ "{data_type:?}"
+ );
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap(),
+ Some(snapshot_id + 1)
+ );
+ assert_eq!(
+ row_count(
+ &sql_context,
+ "SELECT * FROM paimon.test_db.drop_native_timestamp"
+ )
+ .await,
+ 2
+ );
+ }
+ }
+}
+
+#[tokio::test]
+async fn test_drop_global_index_ltz_partitions_use_display_timezone() {
+ const CHILD: &str = "PAIMON_DROP_LTZ_PARTITIONS_CHILD";
+ if std::env::var_os(CHILD).is_none() {
+ // Local timezone is process-global; isolate it from parallel SQL
tests.
+ for timezone in ["Asia/Shanghai", "America/New_York",
"Africa/Monrovia"] {
+ let output =
std::process::Command::new(std::env::current_exe().unwrap())
+ .args([
+ "--exact",
+
"test_drop_global_index_ltz_partitions_use_display_timezone",
+ ])
+ .env("TZ", timezone)
+ .env(CHILD, timezone)
+ .output()
+ .unwrap();
+ assert!(
+ output.status.success(),
+ "{timezone}:\n{}{}",
+ String::from_utf8_lossy(&output.stdout),
+ String::from_utf8_lossy(&output.stderr)
+ );
+ }
+ return;
+ }
+ let timezone = std::env::var(CHILD).unwrap();
+ if timezone == "Asia/Shanghai" {
+ let instants = ["1970-01-01 00:00:00+00:00", "1970-01-01
08:00:00+00:00"];
+ for (legacy, input, selected) in [
+ (false, "1970-01-01 08:00:00.000", 0),
+ (false, "1970-01-01T08:00:00.000+08:00", 0),
+ (false, "1970-01-01T08:00:00.000Z", 28_800_000),
+ (false, "0", 0),
+ (true, "1970-01-01T00:00", 0),
+ ] {
+ assert_timestamp_partition_drop(
+ "TIMESTAMP(3) WITH TIME ZONE",
+ legacy,
+ instants,
+ input,
+ (selected, 0),
+ )
+ .await;
+ }
+ assert_timestamp_partition_drop(
+ "TIMESTAMP(3)",
+ false,
+ instants,
+ "1970-01-01 00:00:00.000",
+ (0, 0),
+ )
+ .await;
+ assert_timestamp_partition_drop("TIMESTAMP(3)", true, instants,
"1970-01-01T00:00", (0, 0))
+ .await;
+ for (sql_type, input, expected_nanos) in [
+ (
+ "TIMESTAMP(6) WITH TIME ZONE",
+ "1970-01-01 08:00:00.123456",
+ 456_000,
+ ),
+ (
+ "TIMESTAMP(9) WITH TIME ZONE",
+ "1970-01-01 08:00:00.123456789",
+ 456_789,
+ ),
+ ] {
+ assert_timestamp_partition_drop(
+ sql_type,
+ false,
+ [
+ "1970-01-01 00:00:00.123456789+00:00",
+ "1970-01-01 08:00:00.123456789+00:00",
+ ],
+ input,
+ (123, expected_nanos),
+ )
+ .await;
+ }
+ } else if timezone == "America/New_York" {
+ // Summer offset is -04:00, not the timezone's winter offset of -05:00.
+ assert_timestamp_partition_drop(
+ "TIMESTAMP(3) WITH TIME ZONE",
+ false,
+ ["2024-07-01 16:00:00+00:00", "2024-07-01 17:00:00+00:00"],
+ "2024-07-01 12:00:00.000",
+ (1_719_849_600_000, 0),
+ )
+ .await;
+ assert_timestamp_partition_drop(
+ "TIMESTAMP(3) WITH TIME ZONE",
+ false,
+ ["2024-11-03 05:30:00+00:00", "2024-11-03 06:30:00+00:00"],
+ "2024-11-03T01:30:00-04:00",
+ (1_730_611_800_000, 0),
+ )
+ .await;
+ } else {
+ // Monrovia's historical offset of -00:44:30 must not shift the
instant by 30 seconds.
+ for input in ["1970-12-31 23:15:30.000", "1971-01-01T00:00:00Z"] {
+ assert_timestamp_partition_drop(
+ "TIMESTAMP(3) WITH TIME ZONE",
+ false,
+ ["1971-01-01 00:00:00+00:00", "1971-01-01 00:00:30+00:00"],
+ input,
+ (31_536_000_000, 0),
+ )
+ .await;
+ }
+ }
+}
+
+async fn assert_timestamp_partition_drop(
+ sql_type: &str,
+ legacy: bool,
+ instants: [&str; 2],
+ input: &str,
+ selected: (i64, i32),
+) {
+ let (tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ exec(&sql_context, "CREATE SCHEMA paimon.test_db").await;
+ exec(
+ &sql_context,
+ &format!(
+ "CREATE TABLE paimon.test_db.drop_ltz_partitions (id INT, name
STRING, ts {sql_type}) \
+ PARTITIONED BY (ts) WITH ('row-tracking.enabled'='true',
'data-evolution.enabled'='true', \
+ 'global-index.enabled'='true', 'partition.legacy-name'='{legacy}')"
+ ),
+ )
+ .await;
+ exec(
+ &sql_context,
+ &format!(
+ "INSERT INTO paimon.test_db.drop_ltz_partitions (id, name, ts)
VALUES \
+ (1, 'a', TIMESTAMP '{}'), (2, 'b', TIMESTAMP '{}')",
+ instants[0], instants[1]
+ ),
+ )
+ .await;
+ for (column, index_type) in [("id", "btree"), ("name", "btree"), ("id",
"bitmap")] {
+ exec(
+ &sql_context,
+ &format!(
+ "CALL sys.create_global_index(table =>
'test_db.drop_ltz_partitions', \
+ index_column => '{column}', index_type => '{index_type}')"
+ ),
+ )
+ .await;
+ }
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "drop_ltz_partitions"))
+ .await
+ .unwrap();
+ let partition_type =
table.schema().partition_fields()[0].data_type().clone();
+ let selected_datum = match partition_type {
+ paimon::spec::DataType::Timestamp(_) => paimon::spec::Datum::Timestamp
{
+ millis: selected.0,
+ nanos: selected.1,
+ },
+ paimon::spec::DataType::LocalZonedTimestamp(_) => {
+ paimon::spec::Datum::LocalZonedTimestamp {
+ millis: selected.0,
+ nanos: selected.1,
+ }
+ }
+ other => panic!("Unexpected partition type: {other:?}"),
+ };
+ let mut expected = global_index_entries(&table).await;
+ assert_eq!(expected.len(), 6);
+ if input.contains(' ') || legacy {
+ assert_eq!(row_count(&sql_context, &format!(
+ "SELECT * FROM paimon.test_db.`drop_ltz_partitions$partitions`
WHERE partition = 'ts={input}'"
+ )).await, 1);
+ }
+ expected.retain(|_, entry| {
+ let row =
paimon::spec::BinaryRow::from_serialized_bytes(&entry.partition).unwrap();
+ !(entry.index_file.index_type == "btree"
+ && entry
+ .index_file
+ .global_index_meta
+ .as_ref()
+ .unwrap()
+ .index_field_id
+ == 0
+ && row.get_datum(0, &partition_type).unwrap().as_ref() ==
Some(&selected_datum))
+ });
+ assert_eq!(expected.len(), 5);
+ let before = table_files(tmp.path());
+ let snapshot_id = table
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap()
+ .unwrap();
+ if std::env::var("TZ").unwrap() == "America/New_York" {
+ for invalid in [
+ "2024-03-10 02:30:00",
+ "2024-11-03 01:30:00",
+ "not-a-timestamp",
+ ] {
+ for dry_run in [true, false] {
+ assert_sql_error(&sql_context, &format!(
+ "CALL sys.drop_global_index(table =>
'test_db.drop_ltz_partitions', index_column => 'id', \
+ partitions => 'ts={input};ts={invalid}', dry_run =>
{dry_run})"
+ ), "Invalid partitions value").await;
+ assert_eq!(table_files(tmp.path()), before);
+ }
+ }
+ }
+ let call = format!(
+ "CALL sys.drop_global_index(table => 'test_db.drop_ltz_partitions',
index_column => 'id', partitions => 'ts={input}'"
+ );
+ let result = sql_context
+ .sql(&format!("{call}, dry_run => true)"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ string_value(result[0].column(0).as_ref(), 0),
+ "Would drop 1 global index file(s)"
+ );
+ assert_eq!(table_files(tmp.path()), before);
+ exec(&sql_context, &format!("{call})")).await;
+ assert_eq!(
+ global_index_entries(&table).await,
+ expected,
+ "{sql_type}, legacy={legacy}, {input}"
+ );
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap(),
+ Some(snapshot_id + 1)
+ );
+ assert_eq!(
+ row_count(
+ &sql_context,
+ "SELECT * FROM paimon.test_db.drop_ltz_partitions"
+ )
+ .await,
+ 2
+ );
+}
+
+#[tokio::test]
+async fn test_drop_global_index_partitions_timestamp_and_decimal_are_exact() {
+ let (tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ exec(&sql_context, "CREATE SCHEMA paimon.test_db").await;
+ exec(&sql_context,
+ "CREATE TABLE paimon.test_db.drop_typed_partitions (id INT, ts
TIMESTAMP(3), price DECIMAL(10,2)) \
+ PARTITIONED BY (ts, price) WITH ('row-tracking.enabled'='true',
'data-evolution.enabled'='true', \
+ 'global-index.enabled'='true')"
+ ).await;
+ exec(&sql_context,
+ "INSERT INTO paimon.test_db.drop_typed_partitions (id, ts, price)
VALUES \
+ (1, TIMESTAMP '2026-10-10 12:00:00.199', 1.24), (2, TIMESTAMP
'2026-10-10 12:00:00.199', 1.23), \
+ (3, TIMESTAMP '2026-10-10 12:00:00.2', 1.24)"
+ ).await;
+ exec(&sql_context,
+ "CALL sys.create_global_index(table =>
'test_db.drop_typed_partitions', index_column => 'id')"
+ ).await;
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "drop_typed_partitions"))
+ .await
+ .unwrap();
+ let before = table_files(tmp.path());
+ let mut expected = global_index_entries(&table).await;
+ assert_eq!(expected.len(), 3);
+ let millis =
+ chrono::NaiveDateTime::parse_from_str("2026-10-10 12:00:00.199",
"%Y-%m-%d %H:%M:%S%.f")
+ .unwrap()
+ .and_utc()
+ .timestamp_millis();
+ expected.retain(|_, entry| {
+ let row =
paimon::spec::BinaryRow::from_serialized_bytes(&entry.partition).unwrap();
+ !(row.get_long(0).unwrap() == millis && row.get_long(1).unwrap() ==
124)
+ });
+ assert_eq!(expected.len(), 2);
+ let no_match = "CALL sys.drop_global_index(table =>
'test_db.drop_typed_partitions', index_column => 'id', \
+ partitions => 'ts=2026-10-10 12:00:00.1999,price=1.235'";
+ let result = sql_context
+ .sql(&format!("{no_match}, dry_run => true)"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ string_value(result[0].column(0).as_ref(), 0),
+ "Would drop 0 global index file(s)"
+ );
+ exec(&sql_context, &format!("{no_match})")).await;
+ assert_eq!(table_files(tmp.path()), before);
+ let call = "CALL sys.drop_global_index(table =>
'test_db.drop_typed_partitions', index_column => 'id', \
+ partitions => 'ts=2026-10-10 12:00:00.199,price=1.235'";
+ let result = sql_context
+ .sql(&format!("{call}, dry_run => true)"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ string_value(result[0].column(0).as_ref(), 0),
+ "Would drop 1 global index file(s)"
+ );
+ assert_eq!(table_files(tmp.path()), before);
+ exec(&sql_context, &format!("{call})")).await;
+ assert_eq!(global_index_entries(&table).await, expected);
+}
+
+#[tokio::test]
+async fn test_drop_global_index_partitions_invalid_parameters_never_commit() {
+ for populated in [false, true] {
+ let (tmp, sql_context, _) =
+ setup_partitioned_global_index_table("drop_partitions_invalid",
populated).await;
+ let before = table_files(tmp.path());
+ for partitions in [
+ "",
+ " ",
+ ";",
+ "dt",
+ "dt=",
+ "=2026-10-10",
+ "missing=x",
+ "id=1",
+ "dt=not-a-date",
+ "dt=2026-02-30",
+ "hr=not-an-int",
+ "hr=1.5",
+ "hr=2147483648",
+ "dt=2026-10-10;",
+ ";dt=2026-10-10",
+ "dt=2026-10-10;;dt=2026-10-11",
+ "dt=2026-10-10,",
+ ",dt=2026-10-10",
+ "dt=2026-10-10,,hr=1",
+ "dt=2026-10-10,dt=2026-10-11",
+ "dt=2026-10-10;missing=x",
+ "dt=2026-10-10;hr=bad",
+ ] {
+ for dry_run in [true, false] {
+ assert_sql_error(&sql_context, &format!(
+ "CALL sys.drop_global_index(table =>
'test_db.drop_partitions_invalid', index_column => 'id', \
+ partitions => '{partitions}', dry_run => {dry_run})"
+ ), "partition").await;
+ assert_eq!(
+ table_files(tmp.path()),
+ before,
+ "{partitions:?}, dry_run={dry_run}"
+ );
+ }
+ }
+ }
+}
+
#[tokio::test]
async fn test_drop_global_index_removes_btree_and_reads_fallback() {
let (_tmp, sql_context) =
setup_btree_global_index_table("btree_drop").await;
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index 743004c7..02c50c9c 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -119,11 +119,11 @@ pub use types::*;
mod partition;
pub use partition::Partition;
mod partition_utils;
-pub use partition_utils::is_java_whitespace_only;
pub(crate) use partition_utils::{
bucket_path, bucket_path_under, data_file_path, escape_path_name,
relative_bucket_path,
unescape_path_name, PartitionComputer,
};
+pub use partition_utils::{format_java_float, is_java_whitespace_only};
mod predicate;
pub(crate) use predicate::datum_cmp;
pub(crate) use predicate::eval_row;
diff --git a/crates/paimon/src/spec/partition_utils.rs
b/crates/paimon/src/spec/partition_utils.rs
index e6fe1dda..f534c1f4 100644
--- a/crates/paimon/src/spec/partition_utils.rs
+++ b/crates/paimon/src/spec/partition_utils.rs
@@ -24,7 +24,7 @@
//! - Java `PartitionPathUtils` for path escaping and assembly
use crate::common::date_time::{format_time, format_timestamp};
-use crate::common::number::format_java_float;
+use crate::common::number::format_java_float as format_java_floating;
use crate::error::*;
use crate::spec::types::DataType;
use crate::spec::BinaryRow;
@@ -393,7 +393,7 @@ fn format_partition_value(
DataType::Int(_) => row.get_int(pos)?.to_string(),
DataType::BigInt(_) => row.get_long(pos)?.to_string(),
DataType::Float(_) => format_java_float(row.get_float(pos)?),
- DataType::Double(_) => format_java_float(row.get_double(pos)?),
+ DataType::Double(_) => format_java_floating(row.get_double(pos)?),
DataType::Char(_) | DataType::VarChar(_) => {
let s = row.get_string(pos)?;
@@ -485,6 +485,13 @@ fn format_partition_value(
Ok(value)
}
+/// Java's shortest-decimal string contract (JDK 19+), including its notation
+/// thresholds and closest choice among one- and two-digit decimals. Rust's
+/// shortest Display chooses different digits at ties; Ryū uses ties-to-even.
+pub fn format_java_float(value: f32) -> String {
+ format_java_floating(value)
+}
+
/// Decode as Java's UTF-8 decoder does for `BinaryString.toString()`.
/// Rust's lossy decoder replaces each byte of a UTF-8 encoded surrogate,
/// while Java replaces the complete malformed surrogate sequence once.
diff --git a/crates/paimon/src/table/global_index_drop_builder.rs
b/crates/paimon/src/table/global_index_drop_builder.rs
index 7b2f811b..d0bb2bf5 100644
--- a/crates/paimon/src/table/global_index_drop_builder.rs
+++ b/crates/paimon/src/table/global_index_drop_builder.rs
@@ -19,7 +19,10 @@ use super::global_index_build_common::IndexColumns;
use super::global_index_types::{
normalize_global_index_type, BTREE_GLOBAL_INDEX_TYPE,
SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP,
};
-use crate::spec::{FileKind, IndexFileMeta, IndexManifest};
+use crate::spec::{
+ eval_row, extract_datum, field_idx_to_partition_idx, BinaryRow, DataType,
Datum, FileKind,
+ IndexFileMeta, IndexManifest, Predicate, PredicateOperator,
+};
use crate::table::{CommitMessage, Table, TableCommit};
use crate::{Error, Result};
use std::collections::HashMap;
@@ -28,6 +31,7 @@ pub struct GlobalIndexDropBuilder<'a> {
table: &'a Table,
index_column: Option<IndexColumns>,
index_type: String,
+ partition_filter: Option<Predicate>,
dry_run: bool,
}
@@ -37,6 +41,7 @@ impl<'a> GlobalIndexDropBuilder<'a> {
table,
index_column: None,
index_type: BTREE_GLOBAL_INDEX_TYPE.to_string(),
+ partition_filter: None,
dry_run: false,
}
}
@@ -59,6 +64,16 @@ impl<'a> GlobalIndexDropBuilder<'a> {
self
}
+ /// Restrict the drop using table-field indices. Repeated calls combine
filters with AND.
+ pub fn with_partition_filter(&mut self, filter: Predicate) -> Result<&mut
Self> {
+ super::partition_filter::validate_partition_filter(self.table,
&filter)?;
+ self.partition_filter = Some(match self.partition_filter.take() {
+ Some(previous) => Predicate::and(vec![previous, filter]),
+ None => filter,
+ });
+ Ok(self)
+ }
+
/// Return the matched index file count without committing when enabled.
pub fn with_dry_run(&mut self, dry_run: bool) -> &mut Self {
self.dry_run = dry_run;
@@ -95,6 +110,21 @@ impl<'a> GlobalIndexDropBuilder<'a> {
.skip(1)
.map(|field| field.id())
.collect::<Vec<_>>();
+ let partition_filter = self
+ .partition_filter
+ .as_ref()
+ .map(|filter| {
+ let mapping = field_idx_to_partition_idx(
+ self.table.schema().fields(),
+ self.table.schema().partition_keys(),
+ );
+ filter
+ .remap_field_index(&mapping)
+ .ok_or_else(|| Error::ConfigInvalid {
+ message: "Partition filter must reference only
partition keys".to_string(),
+ })
+ })
+ .transpose()?;
// The table's manager resolves a REST-managed table's latest snapshot
// through the catalog, like reads and commit validation do.
@@ -132,6 +162,20 @@ impl<'a> GlobalIndexDropBuilder<'a> {
{
continue;
}
+ if let Some(filter) = &partition_filter {
+ // Deletion requires exact matching, not conservative scan
pruning.
+ let partition =
BinaryRow::from_serialized_bytes(&entry.partition)?;
+ if partition.arity() as usize !=
self.table.schema().partition_keys().len() {
+ return Err(Error::DataInvalid {
+ message: "Index partition row arity does not match
partition keys"
+ .to_string(),
+ source: None,
+ });
+ }
+ if !eval_partition_filter(filter, &partition)? {
+ continue;
+ }
+ }
dropped += 1;
deletions_by_partition_bucket
.entry((entry.partition, entry.bucket))
@@ -172,6 +216,92 @@ impl<'a> GlobalIndexDropBuilder<'a> {
}
}
+/// Preserve Java partition comparisons without changing general row
evaluation.
+fn eval_partition_filter(predicate: &Predicate, row: &BinaryRow) ->
Result<bool> {
+ use std::cmp::Ordering::{Equal, Greater, Less};
+ use PredicateOperator::*;
+
+ match predicate {
+ Predicate::And(children) => {
+ for child in children {
+ if !eval_partition_filter(child, row)? {
+ return Ok(false);
+ }
+ }
+ Ok(true)
+ }
+ Predicate::Or(children) => {
+ for child in children {
+ if eval_partition_filter(child, row)? {
+ return Ok(true);
+ }
+ }
+ Ok(false)
+ }
+ Predicate::Not(inner) => Ok(!eval_partition_filter(inner, row)?),
+ Predicate::Leaf {
+ index,
+ data_type,
+ op,
+ literals,
+ ..
+ } if matches!(data_type, DataType::Float(_) | DataType::Double(_))
+ && !matches!(op, IsNull | IsNotNull) =>
+ {
+ let Some(value) = extract_datum(row, *index, data_type)? else {
+ return Ok(false);
+ };
+ let compare = |literal: &Datum| {
+ // Widening FLOAT to DOUBLE is exact, including the sign of
zero.
+ let (left, right) = match (&value, literal) {
+ (Datum::Float(left), Datum::Float(right)) => {
+ (f64::from(*left), f64::from(*right))
+ }
+ (Datum::Double(left), Datum::Double(right)) => (*left,
*right),
+ _ => return None,
+ };
+ Some(match (left.is_nan(), right.is_nan()) {
+ // Java canonicalizes every NaN before comparing
floating-point bits.
+ (true, true) => Equal,
+ (true, false) => Greater,
+ (false, true) => Less,
+ (false, false) => left.total_cmp(&right),
+ })
+ };
+ let first = || literals.first().and_then(compare);
+ Ok(match op {
+ Eq => first() == Some(Equal),
+ NotEq => matches!(first(), Some(Less | Greater)),
+ Lt => first() == Some(Less),
+ LtEq => matches!(first(), Some(Less | Equal)),
+ Gt => first() == Some(Greater),
+ GtEq => matches!(first(), Some(Greater | Equal)),
+ In => literals
+ .iter()
+ .any(|literal| compare(literal) == Some(Equal)),
+ NotIn => !literals
+ .iter()
+ .any(|literal| compare(literal) == Some(Equal)),
+ Between | NotBetween => {
+ let between = matches!(first(), Some(Greater | Equal))
+ && matches!(literals.get(1).and_then(compare),
Some(Less | Equal));
+ if *op == Between {
+ between
+ } else {
+ !between
+ }
+ }
+ _ => {
+ return Err(Error::ConfigInvalid {
+ message: format!("Unsupported floating partition
operator '{op}'"),
+ })
+ }
+ })
+ }
+ _ => eval_row(predicate, row),
+ }
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -187,6 +317,83 @@ mod tests {
use indexmap::IndexMap;
use std::collections::BTreeMap;
+ #[test]
+ fn test_floating_partition_comparisons_follow_java_ordering() {
+ use crate::spec::{BinaryRowBuilder, DoubleType, FloatType};
+ use PredicateOperator::*;
+
+ for data_type in [
+ DataType::Float(FloatType::new()),
+ DataType::Double(DoubleType::new()),
+ ] {
+ let datum = |value| match data_type {
+ DataType::Float(_) => Datum::Float(value as f32),
+ _ => Datum::Double(value),
+ };
+ // Java compares all NaN signs and payloads equally, above
positive infinity.
+ let values = [
+ f64::NEG_INFINITY,
+ -0.0,
+ 0.0,
+ f64::INFINITY,
+ f64::NAN,
+ f64::from_bits(0xfff8_1234_5678_0000),
+ ];
+ let mut rows = values
+ .iter()
+ .map(|value| {
+ let mut row = BinaryRowBuilder::new(1);
+ row.write_datum(0, &datum(*value), &data_type);
+ row.build()
+ })
+ .collect::<Vec<_>>();
+ let mut null = BinaryRowBuilder::new(1);
+ null.set_null_at(0);
+ rows.push(null.build());
+ let leaf = |op, literals: Vec<f64>| Predicate::Leaf {
+ column: "f".into(),
+ index: 0,
+ data_type: data_type.clone(),
+ op,
+ literals: literals.into_iter().map(datum).collect(),
+ };
+ for (op, literals, expected) in [
+ (Eq, vec![0.0], vec![2]),
+ (Eq, vec![f64::NAN], vec![4, 5]),
+ (NotEq, vec![0.0], vec![0, 1, 3, 4, 5]),
+ (Lt, vec![0.0], vec![0, 1]),
+ (LtEq, vec![0.0], vec![0, 1, 2]),
+ (Gt, vec![0.0], vec![3, 4, 5]),
+ (GtEq, vec![0.0], vec![2, 3, 4, 5]),
+ (In, vec![-0.0, f64::NAN], vec![1, 4, 5]),
+ (NotIn, vec![-0.0, f64::NAN], vec![0, 2, 3]),
+ (Between, vec![-0.0, 0.0], vec![1, 2]),
+ (NotBetween, vec![-0.0, 0.0], vec![0, 3, 4, 5]),
+ (IsNull, vec![], vec![6]),
+ (IsNotNull, vec![], vec![0, 1, 2, 3, 4, 5]),
+ ] {
+ let predicate = leaf(op, literals);
+ for (i, row) in rows.iter().enumerate() {
+ assert_eq!(
+ eval_partition_filter(&predicate, row).unwrap(),
+ expected.contains(&i),
+ "{data_type:?}: {op}, row {i}"
+ );
+ }
+ }
+ let nested = Predicate::and(vec![
+ Predicate::or(vec![leaf(Eq, vec![0.0]), leaf(Eq,
vec![f64::NAN])]),
+ Predicate::Not(Box::new(leaf(Eq, vec![0.0]))),
+ ]);
+ for (i, row) in rows.iter().enumerate() {
+ assert_eq!(
+ eval_partition_filter(&nested, row).unwrap(),
+ matches!(i, 4 | 5)
+ );
+ }
+ }
+ }
+
#[tokio::test]
async fn test_query_auth_table_refuses_index_builds_and_drops() {
let table = crate::table::query_auth_table();
@@ -394,6 +601,238 @@ mod tests {
files
}
+ async fn partitioned_index_table(path: &str) -> Table {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("dt", DataType::VarChar(VarCharType::string_type()))
+ .column("name", DataType::VarChar(VarCharType::string_type()))
+ .column("hr", DataType::Int(IntType::new()))
+ .partition_keys(["hr", "dt"])
+ .build()
+ .unwrap();
+ let table = Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "partitioned_indexes"),
+ path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ setup_dirs(&table).await;
+ let mut messages = Vec::new();
+ for (i, (dt, hr)) in [
+ (Some("2026-10-10"), 1),
+ (Some("2026-10-10"), 2),
+ (Some("2026-10-11"), 1),
+ (Some("2026-10-11"), 2),
+ (None, 1),
+ ]
+ .into_iter()
+ .enumerate()
+ {
+ let mut partition = crate::spec::BinaryRowBuilder::new(2);
+ partition.write_int(0, hr);
+ match dt {
+ Some(dt) => partition.write_string(1, dt),
+ None => partition.set_null_at(1),
+ }
+ let start = i as i64 * 10;
+ let end = start + 9;
+ let mut composite =
+ global_index_file("btree", &format!("composite-{i}"), 0,
start, end);
+ composite
+ .global_index_meta
+ .as_mut()
+ .unwrap()
+ .extra_field_ids = Some(vec![2]);
+ let mut message = CommitMessage::new(
+ partition.build_serialized(),
+ 0,
+ vec![data_file(&format!("data-{i}.parquet"))],
+ );
+ message.new_index_files = vec![
+ global_index_file("btree", &format!("id-{i}"), 0, start, end),
+ composite,
+ global_index_file("btree", &format!("name-{i}"), 2, start,
end),
+ global_index_file("bitmap", &format!("bitmap-{i}"), 0, start,
end),
+ hash_index_file(&format!("hash-{i}")),
+ ];
+ messages.push(message);
+ }
+ TableCommit::new(table.clone(), "setup".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ table
+ }
+
+ #[tokio::test]
+ async fn test_drop_global_index_partition_filter_preserves_other_entries()
{
+ let table =
partitioned_index_table("memory:/drop_partition_alternatives").await;
+ let pb = crate::spec::PredicateBuilder::new(table.schema().fields());
+ let filter = Predicate::or(vec![
+ pb.partition_predicate(&[
+ ("dt", Some(crate::spec::Datum::String("2026-10-10".into()))),
+ ("hr", Some(crate::spec::Datum::Int(1))),
+ ])
+ .unwrap(),
+ pb.partition_predicate(&[
+ ("dt", Some(crate::spec::Datum::String("2026-10-11".into()))),
+ ("hr", Some(crate::spec::Datum::Int(2))),
+ ])
+ .unwrap(),
+ ]);
+ for columns in ["id", "id,name"] {
+ let before = table_files(&table).await;
+ let previous_entries = latest_index_entries(&table).await;
+ let snapshot_id = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id();
+ let mut builder = table.new_global_index_drop_builder();
+ builder.with_index_column(columns).with_dry_run(true);
+ builder.with_partition_filter(filter.clone()).unwrap();
+ assert_eq!(builder.execute().await.unwrap(), 2);
+ assert_eq!(table_files(&table).await, before);
+
+ builder.with_dry_run(false);
+ assert_eq!(builder.execute().await.unwrap(), 2);
+ let prefix = if columns == "id" { "id" } else { "composite" };
+ let expected = previous_entries
+ .into_iter()
+ .filter(|entry| {
+ ![format!("{prefix}-0"), format!("{prefix}-3")]
+ .contains(&entry.index_file.file_name)
+ })
+ .map(|entry| (entry.index_file.file_name.clone(), entry))
+ .collect::<BTreeMap<_, _>>();
+ let remaining = latest_index_entries(&table)
+ .await
+ .into_iter()
+ .map(|entry| (entry.index_file.file_name.clone(), entry))
+ .collect::<BTreeMap<_, _>>();
+ assert_eq!(remaining, expected);
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ snapshot_id + 1
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn test_drop_global_index_partition_filters_combine_with_and() {
+ let table =
partitioned_index_table("memory:/drop_partition_and").await;
+ let pb = crate::spec::PredicateBuilder::new(table.schema().fields());
+ let mut builder = table.new_global_index_drop_builder();
+ builder.with_index_column("id").with_dry_run(true);
+ builder
+ .with_partition_filter(
+ pb.equal("dt", crate::spec::Datum::String("2026-10-10".into()))
+ .unwrap(),
+ )
+ .unwrap();
+ assert_eq!(builder.execute().await.unwrap(), 2);
+ builder
+ .with_partition_filter(pb.equal("hr",
crate::spec::Datum::Int(2)).unwrap())
+ .unwrap();
+ assert_eq!(builder.execute().await.unwrap(), 1);
+ builder.with_dry_run(false);
+ assert_eq!(builder.execute().await.unwrap(), 1);
+ let remaining = latest_index_entries(&table).await;
+ assert!(!remaining
+ .iter()
+ .any(|entry| entry.index_file.file_name == "id-1"));
+ assert!(remaining
+ .iter()
+ .any(|entry| entry.index_file.file_name == "id-0"));
+ }
+
+ #[tokio::test]
+ async fn test_drop_global_index_partition_no_match_and_null() {
+ let table =
partitioned_index_table("memory:/drop_partition_no_match").await;
+ let pb = crate::spec::PredicateBuilder::new(table.schema().fields());
+ let before = table_files(&table).await;
+ for dry_run in [true, false] {
+ let mut builder = table.new_global_index_drop_builder();
+ builder.with_index_column("id").with_dry_run(dry_run);
+ builder
+ .with_partition_filter(pb.equal("hr",
crate::spec::Datum::Int(99)).unwrap())
+ .unwrap();
+ assert_eq!(builder.execute().await.unwrap(), 0);
+ assert_eq!(table_files(&table).await, before);
+ }
+ let mut builder = table.new_global_index_drop_builder();
+ builder.with_index_column("id");
+ builder
+ .with_partition_filter(pb.is_null("dt").unwrap())
+ .unwrap();
+ assert_eq!(builder.execute().await.unwrap(), 1);
+ assert!(!latest_index_entries(&table)
+ .await
+ .iter()
+ .any(|entry| entry.index_file.file_name == "id-4"));
+ }
+
+ #[tokio::test]
+ async fn test_drop_global_index_partition_filter_rejects_data_columns() {
+ let table =
partitioned_index_table("memory:/drop_partition_invalid").await;
+ let pb = crate::spec::PredicateBuilder::new(table.schema().fields());
+ let before = table_files(&table).await;
+ for filter in [
+ pb.equal("id", crate::spec::Datum::Int(1)).unwrap(),
+ Predicate::or(vec![pb.is_null("dt").unwrap(),
pb.is_null("name").unwrap()]),
+ ] {
+ assert!(table
+ .new_global_index_drop_builder()
+ .with_partition_filter(filter)
+ .is_err());
+ }
+ assert_eq!(table_files(&table).await, before);
+ let unpartitioned = test_table("memory:/drop_unpartitioned_filter");
+ assert!(unpartitioned
+ .new_global_index_drop_builder()
+ .with_partition_filter(Predicate::AlwaysTrue)
+ .is_err());
+ }
+
+ #[tokio::test]
+ async fn test_drop_global_index_partition_filter_rejects_invalid_rows() {
+ for partition in [vec![0, 0], BinaryRow::new(0).to_serialized_bytes()]
{
+ let table =
partitioned_index_table("memory:/drop_partition_invalid_row").await;
+ let sm = table.snapshot_manager();
+ let snapshot = sm.get_latest_snapshot().await.unwrap().unwrap();
+ let path = sm.manifest_path(snapshot.index_manifest().unwrap());
+ let mut entries = latest_index_entries(&table).await;
+ entries
+ .iter_mut()
+ .find(|entry| entry.index_file.file_name == "id-0")
+ .unwrap()
+ .partition = partition;
+ IndexManifest::write(table.file_io(), &path, &entries)
+ .await
+ .unwrap();
+ let before = table_files(&table).await;
+ for dry_run in [true, false] {
+ let pb =
crate::spec::PredicateBuilder::new(table.schema().fields());
+ let mut builder = table.new_global_index_drop_builder();
+ builder.with_index_column("id").with_dry_run(dry_run);
+ builder
+ .with_partition_filter(pb.is_null("dt").unwrap())
+ .unwrap();
+ assert!(builder.execute().await.is_err());
+ assert_eq!(table_files(&table).await, before);
+ }
+ }
+ }
+
#[tokio::test]
async fn test_drop_global_index_dry_run_preserves_files() {
for index_type in [
