JingsongLi commented on code in PR #1106:
URL: https://github.com/apache/paimon-rust/pull/1106#discussion_r4244699091
##########
crates/integrations/datafusion/src/procedures.rs:
##########
@@ -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<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())
+ .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<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())
+ }
+ DataType::LocalZonedTimestamp(typ) => {
+ parse_global_index_partition_timestamp(value, data_type,
typ.precision())
Review Comment:
[P2] Parse LTZ partition strings in the timezone used to display them
This delegates LTZ to a parser whose Arrow type carries UTC, while
partition.legacy-name=false formats LTZ partition values using chrono::Local
(partition_utils.rs), and $partitions exposes those local values. Reproduced
with real SQL under TZ=Asia/Shanghai: a partitioned TIMESTAMP(3) WITH TIME ZONE
table containing UTC instants 1970-01-01 00:00:00 and 08:00:00 displays
partitions ts=1970-01-01 08:00:00.000 and ts=1970-01-01 16:00:00.000. After
building BTree indexes, CALL sys.drop_global_index(..., partitions =>
'ts=1970-01-01 08:00:00.000') deletes the second partition's index and leaves
the requested first one (remaining serialized partition millis: [0], expected
[28800000]). Please interpret timezone-less LTZ specifications with the same
local timezone contract as partition formatting / Java TypeUtils, and add an
end-to-end non-UTC regression test.
--
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]
