leaves12138 commented on code in PR #8872:
URL: https://github.com/apache/paimon/pull/8872#discussion_r3662055856
##########
paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/UniversalCompactor.java:
##########
@@ -317,7 +327,9 @@ private MergeResult mergeSortedRuns(
for (List<SstFileMetadata> group : mergedGroups) {
if (group.size() == 1) {
SstFileMetadata singleFile = group.get(0);
- boolean canSkip = !dropTombstones ||
!singleFile.hasTombstones();
+ boolean canSkip =
Review Comment:
Merge correctness currently depends on group sizing. `RecordCombiningWriter`
is created inside `mergeFileGroup`, so mergeable records on opposite sides of
two large non-overlapping groups are never presented to the same writer;
additionally, singleton groups are skipped even when `mergeOperator` is
configured. Reproduction: set `maxSstFileSize=1`, write and flush `a-0=expired`
and `a-1=expired`, configure `canMerge(a-0, a-1)=true`, `merge => live`, and a
TTL predicate matching `expired`, then call `compact()`. Both keys become null
instead of `a-0=live` / `a-1=null`, because each group evaluates TTL before the
cross-group merge. Please preserve combining state across group boundaries or
disable the grouping/skip optimization when a merge operator is set.
##########
paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/UniversalCompactor.java:
##########
@@ -625,6 +599,105 @@ private static void clearLevelsOfRuns(
}
}
+ /** Writes compacted records, combining adjacent values before enforcing
output file sizes. */
+ private final class CompactionSstOutput {
+
+ private final List<SstFileMetadata> result;
+ private final FileSupplier fileSupplier;
+ private final int outputLevel;
+ private final boolean dropTombstones;
+ private final RecordCombiningWriter combiningWriter;
+
+ @Nullable private SortLookupStoreWriter currentWriter;
+ @Nullable private File currentSstFile;
+ @Nullable private MemorySlice currentFileMinKey;
+ @Nullable private MemorySlice currentFileMaxKey;
+ private long currentBatchSize;
+ private long currentTombstoneCount;
+
+ private CompactionSstOutput(
+ List<SstFileMetadata> result,
+ FileSupplier fileSupplier,
+ int outputLevel,
+ boolean dropTombstones) {
+ this.result = result;
+ this.fileSupplier = fileSupplier;
+ this.outputLevel = outputLevel;
+ this.dropTombstones = dropTombstones;
+ this.combiningWriter =
+ new RecordCombiningWriter(mergeOperator,
this::writeCombinedRecord);
+ }
+
+ private void put(MemorySlice key, byte[] value) throws IOException {
+ combiningWriter.put(key, value);
+ }
+
+ private void finish() throws IOException {
+ combiningWriter.finish();
+ closeCurrentWriter();
+ }
+
+ private void abort(Throwable failure) {
+ if (currentWriter != null) {
+ try {
+ currentWriter.close();
+ } catch (IOException suppressed) {
+ failure.addSuppressed(suppressed);
+ }
+ currentWriter = null;
+ }
+ }
+
+ private void writeCombinedRecord(MemorySlice key, byte[] value) throws
IOException {
+ // Evaluate expiration after combining so a TTL-aware merge can
refresh the result.
+ boolean expired = expiredValuePredicate != null &&
expiredValuePredicate.test(value);
Review Comment:
The expiration predicate should not be invoked for tombstones. Tombstones
are internal markers rather than stored values, and a predicate that decodes
the value envelope can fail. I reproduced this with `value -> value[0] == 1`:
after put+flush, delete+flush, `compact()` throws
`ArrayIndexOutOfBoundsException`. Please short-circuit `isTombstone(value)`
before calling the predicate.
--
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]
