From 511bbc7545d6ab28ccc751c7ad5393bd5da736be Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Sun, 11 Oct 2026 23:33:04 +0800 Subject: [PATCH 1/4] feat(index): support partition-scoped global index drops Add a partition predicate to the core drop builder and parse structured partition specifications in the SQL procedure. Reject invalid parameters before deletion and preserve dry-run and snapshot-guarded commits. Reuse Java-compatible FLOAT formatting and compare floating partition keys with signed-zero and NaN semantics. Cover partition selection, unchanged index definitions, invalid inputs, and non-committing previews. Signed-off-by: QuakeWang --- Cargo.lock | 1 + crates/integrations/datafusion/Cargo.toml | 1 + .../integrations/datafusion/src/procedures.rs | 250 ++++++++- .../datafusion/tests/procedures.rs | 514 +++++++++++++++++- crates/paimon/src/spec/mod.rs | 2 +- crates/paimon/src/spec/partition_utils.rs | 11 +- .../src/table/global_index_drop_builder.rs | 441 ++++++++++++++- 7 files changed, 1207 insertions(+), 13 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 69911e8bc..9f4e429c4 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 a2f78ae27..dfca7278e 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 b908da7d6..9545aafd8 100644 --- a/crates/integrations/datafusion/src/procedures.rs +++ b/crates/integrations/datafusion/src/procedures.rs @@ -41,6 +41,9 @@ //! //! 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. //! //! `column_names` and `excluded_column_names` are comma-separated lists, not SQL arrays: //! `column_names => 'id, region'`. Java's Spark procedures take `ARRAY` there, but a @@ -53,6 +56,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 +69,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 +177,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 +778,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 +802,162 @@ 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 { + 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()) + .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) -> Option { + 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::().ok()?; + let float = value.parse::().ok()?; + (f64::from(float) == double + || (float.is_nan() && double.is_nan()) + || paimon::spec::format_java_float(float).parse::().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::() + .ok()? + .with_scale_round(i64::from(typ.scale()), bigdecimal::RoundingMode::HalfUp); + let unscaled = decimal + .as_bigint_and_exponent() + .0 + .to_string() + .parse::() + .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()) + } + DataType::LocalZonedTimestamp(typ) => { + parse_global_index_partition_timestamp(value, data_type, typ.precision()) + } + _ => paimon::table::parse_format_partition_value(value, data_type), + } +} + +fn parse_global_index_partition_timestamp( + value: &str, + data_type: &DataType, + precision: u32, +) -> Option { + let arrow_type = paimon::arrow::paimon_type_to_arrow(data_type).ok()?; + let scalar = 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; + } + ScalarValue::Int64(Some(value.parse().ok()?)) + .cast_to(&arrow_type) + .ok()? + } else { + ScalarValue::try_from_string(value.to_string(), &arrow_type).ok()? + }; + let mut datum = crate::filter_pushdown::scalar_to_datum(&scalar, data_type)?; + let (Datum::Timestamp { millis, nanos } | Datum::LocalZonedTimestamp { millis, nanos }) = + &mut datum + else { + return None; + }; + // Paimon truncates to the declared precision, which may be between Arrow time units. + let millis_quantum = 10_i64.pow(3_u32.saturating_sub(precision)); + *millis = millis.div_euclid(millis_quantum) * millis_quantum; + let nanos_quantum = 10_i32.pow(9 - precision); + *nanos = *nanos / nanos_quantum * nanos_quantum; + Some(datum) +} + /// 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 +1430,82 @@ 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())), + Some(Datum::Int(1)) + ); + for text in ["1.5", "2147483648", "bad"] { + assert!( + parse_global_index_partition_value(text, &DataType::Int(IntType::new())).is_none() + ); + } + assert!( + parse_global_index_partition_value("2026-02-30", &DataType::Date(DateType::new())) + .is_none() + ); + assert_eq!( + parse_global_index_partition_value("1970-01-02", &DataType::Date(DateType::new())), + Some(Datum::Date(1)) + ); + assert_eq!( + parse_global_index_partition_value( + "a=b", + &DataType::VarChar(VarCharType::new(3).unwrap()) + ), + Some(Datum::String("a=b".into())) + ); + assert!(parse_global_index_partition_value( + "abcd", + &DataType::Char(CharType::new(3).unwrap()) + ) + .is_none()); + assert_eq!( + parse_global_index_partition_value( + "12:00:00.123", + &DataType::Time(TimeType::new(3).unwrap()) + ), + 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), + 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).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), + Some(Datum::Timestamp { + millis: 100, + nanos: 0 + }) + ); + assert!(parse_global_index_partition_value("123", ×tamp).is_none()); + assert!(parse_global_index_partition_value("bad", ×tamp).is_none()); + let float = DataType::Float(FloatType::new()); + assert_eq!( + parse_global_index_partition_value("1.1", &float), + Some(Datum::Float(1.1)) + ); + for text in ["1e100", "1e-100", "1.234567891"] { + assert!(parse_global_index_partition_value(text, &float).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 abc06c636..fb0c6fcfa 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,514 @@ 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 { + 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::>() + .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::>(); + 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::().unwrap()) + } else { + value.parse::().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_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 call = "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!("{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 a108f89ea..d09bc3cf8 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 e6fe1dda1..f534c1f40 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 7b2f811bb..d0bb2bf59 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, index_type: String, + partition_filter: Option, 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::>(); + 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 { + 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::>(); + let mut null = BinaryRowBuilder::new(1); + null.set_null_at(0); + rows.push(null.build()); + let leaf = |op, literals: Vec| 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::>(); + let remaining = latest_index_entries(&table) + .await + .into_iter() + .map(|entry| (entry.index_file.file_name.clone(), entry)) + .collect::>(); + 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 [ From 855ca729cab322c0fe730e827c0cee1b935168bd Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Mon, 12 Oct 2026 13:24:04 +0800 Subject: [PATCH 2/4] fix(index): align LTZ partition drop parsing with displayed names Parse non-legacy LTZ partition values in the system-local timezone while preserving UTC semantics for legacy names and explicit timezone offsets. Accept legacy names with omitted zero seconds and reject ambiguous or nonexistent local times before deletion. Cover SQL deletion in isolated timezones, timestamp precision, retained index definitions, invalid parameters, and non-committing dry runs. Signed-off-by: QuakeWang --- .../integrations/datafusion/src/procedures.rs | 100 +++++--- .../datafusion/tests/procedures.rs | 235 ++++++++++++++++++ 2 files changed, 305 insertions(+), 30 deletions(-) diff --git a/crates/integrations/datafusion/src/procedures.rs b/crates/integrations/datafusion/src/procedures.rs index 9545aafd8..977c06bc8 100644 --- a/crates/integrations/datafusion/src/procedures.rs +++ b/crates/integrations/datafusion/src/procedures.rs @@ -44,6 +44,8 @@ //! `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` there, but a @@ -844,13 +846,17 @@ fn parse_global_index_partitions(table: &Table, partitions: &str) -> DFResult DFResult Option { +fn parse_global_index_partition_value( + value: &str, + data_type: &DataType, + legacy_partition_name: bool, +) -> Option { match data_type { DataType::Char(typ) => { (value.chars().count() <= typ.length()).then(|| Datum::String(value.to_string())) @@ -918,11 +928,14 @@ fn parse_global_index_partition_value(value: &str, data_type: &DataType) -> Opti }) } DataType::Timestamp(typ) => { - parse_global_index_partition_timestamp(value, data_type, typ.precision()) - } - DataType::LocalZonedTimestamp(typ) => { - parse_global_index_partition_timestamp(value, data_type, typ.precision()) + 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), } } @@ -931,6 +944,7 @@ fn parse_global_index_partition_timestamp( value: &str, data_type: &DataType, precision: u32, + local_timezone: bool, ) -> Option { let arrow_type = paimon::arrow::paimon_type_to_arrow(data_type).ok()?; let scalar = if value.bytes().all(|byte| byte.is_ascii_digit()) { @@ -942,7 +956,23 @@ fn parse_global_index_partition_timestamp( .cast_to(&arrow_type) .ok()? } else { - ScalarValue::try_from_string(value.to_string(), &arrow_type).ok()? + // 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 value = 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()? + .to_rfc3339() + } else { + value + }; + ScalarValue::try_from_string(value, &arrow_type).ok()? }; let mut datum = crate::filter_pushdown::scalar_to_datum(&scalar, data_type)?; let (Datum::Timestamp { millis, nanos } | Datum::LocalZonedTimestamp { millis, nanos }) = @@ -1437,45 +1467,55 @@ mod tests { VarCharType, }; assert_eq!( - parse_global_index_partition_value("01", &DataType::Int(IntType::new())), + 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())).is_none() + 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())) - .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())), + 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()) + &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()) + &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()) + &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), + parse_global_index_partition_value(text, &decimal, true), Some(Datum::Decimal { unscaled, precision: 4, @@ -1484,25 +1524,25 @@ mod tests { ); } for text in ["99.995", "bad", "1_2.34"] { - assert!(parse_global_index_partition_value(text, &decimal).is_none()); + 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), + parse_global_index_partition_value("1970-01-01 00:00:00.199", ×tamp, true), Some(Datum::Timestamp { millis: 100, nanos: 0 }) ); - assert!(parse_global_index_partition_value("123", ×tamp).is_none()); - assert!(parse_global_index_partition_value("bad", ×tamp).is_none()); + 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), + 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).is_none()); + assert!(parse_global_index_partition_value(text, &float, true).is_none()); } } diff --git a/crates/integrations/datafusion/tests/procedures.rs b/crates/integrations/datafusion/tests/procedures.rs index fb0c6fcfa..6627ee031 100644 --- a/crates/integrations/datafusion/tests/procedures.rs +++ b/crates/integrations/datafusion/tests/procedures.rs @@ -1562,6 +1562,241 @@ async fn assert_floating_partition_drops(sql_type: &str, values: &[&str]) { } } +#[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"] { + 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 { + // 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; + } +} + +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(); From 0a659a4e7046da588b61902b1f2504f537ca3b5f Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Mon, 12 Oct 2026 23:47:29 +0800 Subject: [PATCH 3/4] fix(index): preserve LTZ instants with historical timezone offsets Convert resolved LTZ values to UTC before RFC3339 serialization so second-level timezone offsets cannot shift the partition predicate. Cover historical Monrovia partition names and explicit UTC inputs through SQL, including retained index definitions, dry runs, and snapshot updates. Signed-off-by: QuakeWang --- crates/integrations/datafusion/src/procedures.rs | 2 ++ .../integrations/datafusion/tests/procedures.rs | 16 ++++++++++++++-- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/crates/integrations/datafusion/src/procedures.rs b/crates/integrations/datafusion/src/procedures.rs index 977c06bc8..a91a3af55 100644 --- a/crates/integrations/datafusion/src/procedures.rs +++ b/crates/integrations/datafusion/src/procedures.rs @@ -968,6 +968,8 @@ fn parse_global_index_partition_timestamp( &value, ) .ok()? + // RFC3339 rounds offsets containing seconds; UTC preserves the resolved instant. + .with_timezone(&chrono::Utc) .to_rfc3339() } else { value diff --git a/crates/integrations/datafusion/tests/procedures.rs b/crates/integrations/datafusion/tests/procedures.rs index 6627ee031..6c894a69a 100644 --- a/crates/integrations/datafusion/tests/procedures.rs +++ b/crates/integrations/datafusion/tests/procedures.rs @@ -1567,7 +1567,7 @@ 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"] { + for timezone in ["Asia/Shanghai", "America/New_York", "Africa/Monrovia"] { let output = std::process::Command::new(std::env::current_exe().unwrap()) .args([ "--exact", @@ -1639,7 +1639,7 @@ async fn test_drop_global_index_ltz_partitions_use_display_timezone() { ) .await; } - } else { + } 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", @@ -1657,6 +1657,18 @@ async fn test_drop_global_index_ltz_partitions_use_display_timezone() { (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; + } } } From 3a4f6a906ffc08a053b888538baf4830f5b6f7ca Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Wed, 14 Oct 2026 10:32:45 +0800 Subject: [PATCH 4/4] fix(index): preserve exact timestamp partition identity Parse timestamp partition text directly into epoch milliseconds and nanos instead of truncating to the declared precision or Arrow time units. Reject nonzero fractional digits beyond nanosecond precision before deletion. Cover native TIMESTAMP/LTZ intermediate precisions, unmatched selectors, invalid parameters, retained index definitions, and non-committing dry runs. Signed-off-by: QuakeWang --- .../integrations/datafusion/src/procedures.rs | 106 ++++++---- .../datafusion/tests/procedures.rs | 182 +++++++++++++++++- 2 files changed, 253 insertions(+), 35 deletions(-) diff --git a/crates/integrations/datafusion/src/procedures.rs b/crates/integrations/datafusion/src/procedures.rs index a91a3af55..ce150229d 100644 --- a/crates/integrations/datafusion/src/procedures.rs +++ b/crates/integrations/datafusion/src/procedures.rs @@ -946,48 +946,49 @@ fn parse_global_index_partition_timestamp( precision: u32, local_timezone: bool, ) -> Option { - let arrow_type = paimon::arrow::paimon_type_to_arrow(data_type).ok()?; - let scalar = if value.bytes().all(|byte| byte.is_ascii_digit()) { + 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; } - ScalarValue::Int64(Some(value.parse().ok()?)) + 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 { - // 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 value = 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, - ) + datafusion::arrow::compute::kernels::cast_utils::string_to_datetime(&chrono::Utc, &value) .ok()? - // RFC3339 rounds offsets containing seconds; UTC preserves the resolved instant. - .with_timezone(&chrono::Utc) - .to_rfc3339() - } else { - value - }; - ScalarValue::try_from_string(value, &arrow_type).ok()? - }; - let mut datum = crate::filter_pushdown::scalar_to_datum(&scalar, data_type)?; - let (Datum::Timestamp { millis, nanos } | Datum::LocalZonedTimestamp { millis, nanos }) = - &mut datum - else { - return None; }; - // Paimon truncates to the declared precision, which may be between Arrow time units. - let millis_quantum = 10_i64.pow(3_u32.saturating_sub(precision)); - *millis = millis.div_euclid(millis_quantum) * millis_quantum; - let nanos_quantum = 10_i32.pow(9 - precision); - *nanos = *nanos / nanos_quantum * nanos_quantum; - Some(datum) + // 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. @@ -1532,7 +1533,7 @@ mod tests { assert_eq!( parse_global_index_partition_value("1970-01-01 00:00:00.199", ×tamp, true), Some(Datum::Timestamp { - millis: 100, + millis: 199, nanos: 0 }) ); @@ -1548,6 +1549,43 @@ mod tests { } } + #[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 6c894a69a..21bfc73e7 100644 --- a/crates/integrations/datafusion/tests/procedures.rs +++ b/crates/integrations/datafusion/tests/procedures.rs @@ -1562,6 +1562,171 @@ async fn assert_floating_partition_drops(sql_type: &str, values: &[&str]) { } } +#[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"; @@ -1844,8 +2009,23 @@ async fn test_drop_global_index_partitions_timestamp_and_decimal_are_exact() { !(row.get_long(0).unwrap() == millis && row.get_long(1).unwrap() == 124) }); assert_eq!(expected.len(), 2); - let call = "CALL sys.drop_global_index(table => 'test_db.drop_typed_partitions', index_column => 'id', \ + 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