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..ce150229d 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` 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 { + 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 { + 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(), 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 { + 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 abc06c636..21bfc73e7 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 { + 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_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 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 [