JingsongLi commented on code in PR #1106:
URL: https://github.com/apache/paimon-rust/pull/1106#discussion_r4249004906
##########
crates/integrations/datafusion/src/procedures.rs:
##########
@@ -800,6 +804,190 @@ 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()?
+ .to_rfc3339()
Review Comment:
[P2] Preserve the LTZ instant before RFC3339 serialization
Chrono's to_rfc3339 rounds timezone offsets containing seconds to whole
minutes, so this can change the correctly parsed instant before constructing
the predicate. Reproduced through real SQL under TZ=Africa/Monrovia with
nonlegacy TIMESTAMP(3) WITH TIME ZONE partitions at 1971-01-01 00:00:00Z and
00:00:30Z. $partitions displays the first as ts=1970-12-31 23:15:30.000 using
the historical -00:44:30 offset. CALL sys.drop_global_index(..., partitions =>
'ts=1970-12-31 23:15:30.000') serializes that as -00:45, selects UTC 00:00:30,
deletes the second partition's BTree index, and leaves the requested first
index (remaining millis 31536000000; expected 31536030000). Please convert the
resolved datetime to UTC before serialization, or use its epoch components
directly. Adding .with_timezone(&chrono::Utc) before to_rfc3339 fixes the
actual selective-drop probe without changing contemporary timezone/DST cases.
Cover historical second offsets, including an explicit-Z input, as Arrow also c
onverts explicit offsets into the supplied Local timezone here.
--
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]
