[core] Build expired partitions from their values instead of a joined string

[PDGGK]

Purpose

NormalPartitionExpire decides which partitions to drop, then rebuilds each one by joining its values with , and splitting the string back apart:

// NormalPartitionExpire:219-226
return expiredPartValues.stream()
        .map(values -> String.join(DELIMITER, values))     // DELIMITER = ","
        .sorted()
        // Use split(DELIMITER, -1) to preserve trailing empty strings
        .map(s -> s.split(DELIMITER, -1))
        .map(strategy::toPartitionString)
        .limit(Math.min(expiredPartValues.size(), maxExpireNum))
        .collect(Collectors.toList());

The values come from strategy.toPartitionValue(...) on real partition data (:162) and are never escaped, so a partition value that itself contains a comma produces extra tokens. toPartitionString reads only the first partitionKeys.size() of them, so every field after the comma shifts left and the trailing token is dropped. It never throws — a join of N values always splits into at least N tokens, so the index is always in range and the corruption is silent.

The list this produces is not just a log line. It is handed straight to doBatchExpire, which passes it to commit.dropPartitions(...) (:195), or to partitionModification.dropPartitions(...) (:186) on a metastore-partitioned table.

What it does

Two reproductions, both added as tests.

A live partition is deleted and the expired one is kept. Single partition key, partition.expiration-strategy = update-time (which accepts any value, not just dates), partition.expiration-time = 1 s. Write us,ca, wait, write us, expire — only us,ca is old enough:

expired        = [{f0=us}]        <- "us" was written a moment ago and is not expired
remaining rows = [us,ca:old]      <- the partition that should have gone is still there

Exactly inverted: the live partition is dropped, the expired one survives. us is gone from the table.

Multi-key, default date strategy. Keys (f0, f1), partition.timestamp-formatter = yyyyMMdd, write ("20230101","us,ca") and ("20230105","51"):

expired        = [{f0=20230101, f1=us}]      <- a partition that does not exist
remaining rows = [20230101:us,ca, 20230105:51]

Here the drop targets a nonexistent partition, so nothing is deleted — but expiration silently no-ops for 20230101:us,ca on every subsequent run, and the partition never ages out. The first token is still the expired date, which is why this direction does not lose data; under update-time there is no such coincidence.

Reading further, the same corrupted map is what toDonePartitions (:199) appends .done to, so on a metastore-partitioned table the suffix lands on the wrong field.

The round trip has already leaked once: the split(DELIMITER, -1) and its comment were added by #7643 so that a trailing empty partition value would survive it. That fixed one consequence of the join/split; a value containing the delimiter is another.

What changes

Sort on the joined form exactly as before, but build the partition from the values that were already in hand:

return expiredPartValues.stream()
        .sorted(Comparator.comparing(values -> String.join(DELIMITER, values)))
        .map(values -> strategy.toPartitionString(values.toArray()))
        .limit(Math.min(expiredPartValues.size(), maxExpireNum))
        .collect(Collectors.toList());

The sort key is the same string as today and Stream.sorted is stable in both forms, so the ordering — and therefore which partitions survive the maxExpireNum truncation — is unchanged for every input. Only the reconstruction goes away.

toPartitionString takes Object[] and calls toString() on each element, so passing values.toArray() is the same work split was feeding it. toPartitionValue (PartitionExpireStrategy:63-72) substitutes partitionDefaultName for nulls and always returns exactly partitionKeys.size() elements, so there is no null or length case the old path handled and this one does not.

Test evidence

PartitionExpireTest — 10 tests, 0 failures with the change.

Mutation control, on a forced clean rebuild of paimon-core (rm -rf target/classes target/test-classes) so the result is not an incremental-build artefact: with the two new tests kept and NormalPartitionExpire reverted, both fail, and on the shift itself —

Expecting actual:
  {"f0"="20230101", "f1"="us"}
to contain exactly (and in same order):
  ["f0"="20230101", "f1"="us,ca"]

Expecting actual:
  {"f0"="us"}
to contain exactly (and in same order):
  ["f0"="us,ca"]

Wider run: *PartitionExpire* and *Partition*Test across paimon-core — 275 tests, 0 failures.

The update-time test needs the two writes to fall on opposite sides of the cut-off, so it sleeps 2s against a 1s expiration time and pins the check time immediately after the second write — a later stall cannot then move the cut-off past the live partition.

API and Format

No change to any option, on-disk format or public signature. Output is identical for every partition value that does not contain a comma.