JingsongLi commented on code in PR #1106:
URL: https://github.com/apache/paimon-rust/pull/1106#discussion_r4261113667
##########
crates/integrations/datafusion/src/procedures.rs:
##########
@@ -800,6 +804,192 @@ async fn proc_drop_global_index(
ok_result(ctx)
}
+/// Semicolon-separated alternatives, each containing comma-separated
key=value constraints.
+fn parse_global_index_partitions(table: &Table, partitions: &str) ->
DFResult<Predicate> {
+ let partition_fields = table.schema().partition_fields();
+ if partition_fields.is_empty() {
+ return Err(DataFusionError::Plan(
+ "Cannot filter global index partitions on an unpartitioned
table".to_string(),
+ ));
+ }
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let options = CoreOptions::new(table.schema().options());
+ let mut alternatives = Vec::new();
+ for partition in partitions.split(';') {
+ let mut constraints = Vec::new();
+ let mut keys = std::collections::HashSet::new();
+ for entry in partition.split(',') {
+ let (key, value) = entry.split_once('=').ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "Invalid partitions entry '{entry}'. Expected key=value"
+ ))
+ })?;
+ // Java ParameterUtils uses String.trim(), not Unicode whitespace
trimming.
+ let key = key.trim_matches(|ch| ch <= ' ');
+ let value = value.trim_matches(|ch| ch <= ' ');
+ let field = partition_fields
+ .iter()
+ .find(|field| field.name() == key)
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!("Column '{key}' is not a
partition column"))
+ })?;
+ if !keys.insert(key) {
+ return Err(DataFusionError::Plan(format!(
+ "Duplicate partition column '{key}'"
+ )));
+ }
+ if value.is_empty() {
+ return Err(DataFusionError::Plan(format!(
+ "Empty partitions value for column '{key}'"
+ )));
+ }
+ let predicate = if value == options.partition_default_name() {
+ pb.is_null(key)
+ } else {
+ let datum = parse_global_index_partition_value(
+ value,
+ field.data_type(),
+ options.legacy_partition_name(),
+ )
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "Invalid partitions value '{value}' for column '{key}'
with type {:?}",
+ field.data_type()
+ ))
+ })?;
+ pb.equal(key, datum)
+ };
+ constraints.push(predicate.map_err(to_datafusion_error)?);
+ }
+ alternatives.push(Predicate::and(constraints));
+ }
+ Ok(Predicate::or(alternatives))
+}
+
+fn parse_global_index_partition_value(
+ value: &str,
+ data_type: &DataType,
+ legacy_partition_name: bool,
+) -> Option<Datum> {
+ match data_type {
+ DataType::Char(typ) => {
+ (value.chars().count() <= typ.length()).then(||
Datum::String(value.to_string()))
+ }
+ DataType::VarChar(typ) => (value.chars().count() <= typ.length() as
usize)
+ .then(|| Datum::String(value.to_string())),
+ DataType::Binary(_) => Some(Datum::Bytes(value.as_bytes().to_vec())),
+ DataType::VarBinary(typ) => {
+ (value.len() <= typ.length() as usize).then(||
Datum::Bytes(value.as_bytes().to_vec()))
+ }
+ DataType::Float(_) => {
+ // Match TypeUtils.castFromString's check for FLOAT precision loss.
+ let double = value.parse::<f64>().ok()?;
+ let float = value.parse::<f32>().ok()?;
+ (f64::from(float) == double
+ || (float.is_nan() && double.is_nan())
+ || paimon::spec::format_java_float(float).parse::<f64>().ok()?
== double)
+ .then_some(Datum::Float(float))
+ }
+ DataType::Double(_) => value.parse().ok().map(Datum::Double),
+ DataType::Decimal(typ) => {
+ // Java Decimal.fromBigDecimal rounds HALF_UP; Arrow's string cast
truncates.
+ if value.contains('_') {
+ return None;
+ }
+ let decimal = value
+ .parse::<bigdecimal::BigDecimal>()
+ .ok()?
+ .with_scale_round(i64::from(typ.scale()),
bigdecimal::RoundingMode::HalfUp);
+ let unscaled = decimal
+ .as_bigint_and_exponent()
+ .0
+ .to_string()
+ .parse::<i128>()
+ .ok()?;
+ if unscaled
+ .unsigned_abs()
+ .checked_ilog10()
+ .map_or(1, |log| log + 1)
+ > typ.precision()
+ {
+ return None;
+ }
+ Some(Datum::Decimal {
+ unscaled,
+ precision: typ.precision(),
+ scale: typ.scale(),
+ })
+ }
+ DataType::Time(_) => {
+ paimon::table::parse_format_partition_value(value,
data_type).or_else(|| {
+ let arrow_type =
paimon::arrow::paimon_type_to_arrow(data_type).ok()?;
+ let scalar = ScalarValue::try_from_string(value.to_string(),
&arrow_type).ok()?;
+ crate::filter_pushdown::scalar_to_datum(&scalar, data_type)
+ })
+ }
+ DataType::Timestamp(typ) => {
+ parse_global_index_partition_timestamp(value, data_type,
typ.precision(), false)
+ }
+ DataType::LocalZonedTimestamp(typ) =>
parse_global_index_partition_timestamp(
+ value,
+ data_type,
+ typ.precision(),
+ !legacy_partition_name,
+ ),
+ _ => paimon::table::parse_format_partition_value(value, data_type),
+ }
+}
+
+fn parse_global_index_partition_timestamp(
+ value: &str,
+ data_type: &DataType,
+ precision: u32,
+ local_timezone: bool,
+) -> Option<Datum> {
+ 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 {
+ // 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()?
+ // 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;
Review Comment:
[P2] Preserve the exact stored timestamp partition before deletion
This truncates a selector to the declared precision even when existing
partitions retain finer values. I reproduced through public APIs: create a
partitioned table with Catalog::create_table / Schema::builder and
TimestampType::new(1), then ordinary SQL INSERT rows at 1970-01-01 00:00:00.100
and .199 and build BTree indexes. Both are valid, readable stored partitions;
$partitions displays ts=...00.1 and ts=...00.199. CALL
sys.drop_global_index(..., partitions => 'ts=1970-01-01 00:00:00.199') succeeds
but deletes the .100 index, leaving the requested .199 index (remaining
serialized millis [199], expected [100]). SQL CREATE rounds precisions to Arrow
units, but native/imported table schemas retain intermediate precisions and are
served by this same SQL procedure. The writer stores the raw milliseconds and
partition formatting intentionally preserves nonzero fractions beyond the
declared precision. Please preserve exact stored identity or reject a selector
that changes during conver
sion before any deletion. An actual persisted-data regression passes when this
new truncation is omitted; cover native/imported intermediate-precision schemas
as well as SQL-created tables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
