JingsongLi commented on code in PR #8904:
URL: https://github.com/apache/paimon/pull/8904#discussion_r3672848862
##########
paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignmentPlanner.java:
##########
@@ -297,27 +329,87 @@ private void markRewrittenManifests(
}
}
+ private boolean manifestMayContainSelectedRange(ManifestFileMeta
manifestMeta) {
+ Long minimum = manifestMeta.minRowId();
+ Long maximum = manifestMeta.maxRowId();
+ if (minimum == null || maximum == null) {
+ return true;
+ }
+ for (SelectedPartition partition : selectedPartitions.values()) {
+ if (rangesOverlap(minimum, maximum, partition.logicalRanges)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean rangesOverlap(
+ long rangeStart, long rangeEnd, PrimitiveRowRanges ranges) {
+ int lower = 0;
+ int upper = ranges.size();
+ while (lower < upper) {
+ int middle = lower + ((upper - lower) >>> 1);
+ if (ranges.end(middle) < rangeStart) {
+ lower = middle + 1;
+ } else {
+ upper = middle;
+ }
+ }
+ return lower < ranges.size() && ranges.start(lower) <= rangeEnd;
+ }
+
private Result buildResult() {
if (selectedPartitions.isEmpty()) {
return new Result(new int[0], Collections.emptyMap(), 0L);
}
+ long skippedRangeCount = 0L;
+ long skippedRowCount = 0L;
+ long[] skipped = new long[2];
+ Iterator<Map.Entry<ByteArrayKey, SelectedPartition>> selectedIterator =
+ selectedPartitions.entrySet().iterator();
+ while (selectedIterator.hasNext()) {
+ SelectedPartition partition = selectedIterator.next().getValue();
+ partition.logicalRanges.normalizeOverlapping();
+ if (skipContiguousRowCount > 0) {
+ partition.removeLargeContiguousRuns(skipContiguousRowCount,
skipped);
+ skippedRangeCount = Math.addExact(skippedRangeCount,
skipped[0]);
+ skippedRowCount = Math.addExact(skippedRowCount, skipped[1]);
+ }
+ if (!partition.hasFragmentedLogicalRanges()) {
+ selectedIterator.remove();
+ }
+ }
+ if (skippedRangeCount > 0) {
+ LOG.info(
+ "Excluded {} logical ranges containing {} rows from row-id
reassignment "
+ + "because their strictly contiguous
same-partition runs exceed {} rows.",
+ skippedRangeCount,
+ skippedRowCount,
+ skipContiguousRowCount);
+ }
+ if (selectedPartitions.isEmpty()) {
+ return new Result(new int[0], Collections.emptyMap(), 0L);
+ }
+
+ markRewrittenManifests();
List<SelectedPartition> partitions = new
ArrayList<>(selectedPartitions.values());
RecordComparator typedComparator =
CodeGenUtils.newRecordComparator(
table.schema().logicalPartitionType().getFieldTypes());
partitions.sort(
(left, right) -> {
int comparison = typedComparator.compare(left.partition,
right.partition);
+ // Row IDs are globally unique, so this also orders
binary-distinct
Review Comment:
Add tests for delete entries. They might move the deletion to the beginning
here, but we should put it at the end. I understand that compaction will be
performed here, so it won’t have any impact, but we want to make sure the tests
run without any issues.
--
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]
