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.
--
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]
