This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 0a48a16872 [core] Clarify live file row ID range collection (#8892)
0a48a16872 is described below
commit 0a48a16872cc6e19e431f319aab1e9683356a9ac
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jul 28 22:48:00 2026 +0800
[core] Clarify live file row ID range collection (#8892)
---
.../DataEvolutionRowIdAssignmentPlanner.java | 63 +++-----
...tries.java => LiveFileRowIdRangeCollector.java} | 149 +++++++++++-------
.../dataevolution/CurrentRowIdEntriesTest.java | 76 ----------
.../DataEvolutionRowIdReassignerTest.java | 10 +-
.../LiveFileRowIdRangeCollectorTest.java | 167 +++++++++++++++++++++
5 files changed, 294 insertions(+), 171 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignmentPlanner.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignmentPlanner.java
index 22721811a8..09c2a11362 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignmentPlanner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignmentPlanner.java
@@ -53,6 +53,8 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import static
org.apache.paimon.append.dataevolution.LiveFileRowIdRangeCollector.FileRole.DEDICATED;
+import static
org.apache.paimon.append.dataevolution.LiveFileRowIdRangeCollector.FileRole.NORMAL;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkState;
@@ -65,7 +67,7 @@ import static
org.apache.paimon.utils.Preconditions.checkState;
final class DataEvolutionRowIdAssignmentPlanner {
private static final int EXCLUDED_PARTITION_CACHE_SIZE = 1024;
- private static final int MAX_INITIAL_CURRENT_ENTRIES = 1 << 24;
+ private static final int MAX_INITIAL_LIVE_FILE_RANGES = 1 << 24;
private static final BinaryString ROW_ID_FIELD =
BinaryString.fromString(SpecialFields.ROW_ID.name());
private static final BinaryString BLOB_FILE_SUFFIX =
BinaryString.fromString(".blob");
@@ -161,18 +163,18 @@ final class DataEvolutionRowIdAssignmentPlanner {
partitionPredicate,
EXCLUDED_PARTITION_CACHE_SIZE,
partitionPredicate == null
- ? initialCurrentEntryCapacity(manifestGroup)
+ ? initialLiveFileRangeCapacity(manifestGroup)
: 0);
ReusableIdentifier identifier = new ReusableIdentifier();
long[] rowRangeScratch = new long[2];
collectDeletedIdentifiers(manifestGroup, group, identifier);
- collectCurrentEntries(
+ collectLiveFileRanges(
manifestGroup, group, identifier, manifestGroupOrdinal,
rowRangeScratch);
identifier.release();
group.releaseDeletedIdentifiers();
- List<PartitionState> selections = group.finishAddPass();
+ List<PartitionState> selections = group.selectFragmentedPartitions();
for (PartitionState selection : selections) {
mergeSelectedPartition(selection);
}
@@ -209,7 +211,7 @@ final class DataEvolutionRowIdAssignmentPlanner {
}
}
- private void collectCurrentEntries(
+ private void collectLiveFileRanges(
List<ManifestFileMeta> manifestGroup,
GroupState group,
ReusableIdentifier identifier,
@@ -262,9 +264,9 @@ final class DataEvolutionRowIdAssignmentPlanner {
maxSequenceNumber,
fileName,
retainedAddScanOrdinal);
- group.currentEntries.add(
+ group.liveFileRanges.add(
partition.id,
- fileOrder != 0,
+ fileOrder == 0 ? NORMAL : DEDICATED,
rowRangeScratch[0],
inclusiveRangeCount(rowRangeScratch[0],
rowRangeScratch[1]));
}
@@ -427,7 +429,7 @@ final class DataEvolutionRowIdAssignmentPlanner {
return ordinal;
}
- private static int initialCurrentEntryCapacity(List<ManifestFileMeta>
manifestGroup) {
+ private static int initialLiveFileRangeCapacity(List<ManifestFileMeta>
manifestGroup) {
long addedCount = 0L;
long deletedCount = 0L;
for (ManifestFileMeta manifestMeta : manifestGroup) {
@@ -437,17 +439,17 @@ final class DataEvolutionRowIdAssignmentPlanner {
addedCount = Math.addExact(addedCount,
manifestMeta.numAddedFiles());
deletedCount = Math.addExact(deletedCount,
manifestMeta.numDeletedFiles());
}
- return initialCurrentEntryCapacity(addedCount, deletedCount);
+ return initialLiveFileRangeCapacity(addedCount, deletedCount);
}
- static int initialCurrentEntryCapacity(long addedCount, long deletedCount)
{
+ static int initialLiveFileRangeCapacity(long addedCount, long
deletedCount) {
checkArgument(addedCount >= 0, "Added entry count cannot be
negative.");
checkArgument(deletedCount >= 0, "Deleted entry count cannot be
negative.");
// Counts are only a sizing hint: DELETE entries may be duplicated or
may not match an ADD
// in this group. Estimate the live set, cap the eager allocation, and
let
- // CurrentRowIdEntries grow if the actual number of retained ADD
entries is larger.
+ // LiveFileRowIdRangeCollector grow if the actual number of retained
ADD entries is larger.
long estimatedLiveCount = addedCount > deletedCount ? addedCount -
deletedCount : 0L;
- return (int) Math.min(estimatedLiveCount, MAX_INITIAL_CURRENT_ENTRIES);
+ return (int) Math.min(estimatedLiveCount,
MAX_INITIAL_LIVE_FILE_RANGES);
}
private void validateGroups(List<List<ManifestFileMeta>> groups) {
@@ -508,17 +510,17 @@ final class DataEvolutionRowIdAssignmentPlanner {
private final GroupPartitionDictionary partitions;
private final DeletedIdentifierSet deletedIdentifiers = new
DeletedIdentifierSet();
- private final CurrentRowIdEntries currentEntries;
+ private final LiveFileRowIdRangeCollector liveFileRanges;
private GroupState(
int partitionArity,
@Nullable PartitionPredicate partitionPredicate,
int excludedPartitionCacheSize,
- int expectedAddEntryCount) {
+ int expectedLiveFileCount) {
this.partitions =
new GroupPartitionDictionary(
partitionArity, partitionPredicate,
excludedPartitionCacheSize);
- this.currentEntries = new
CurrentRowIdEntries(expectedAddEntryCount);
+ this.liveFileRanges = new
LiveFileRowIdRangeCollector(expectedLiveFileCount);
}
private @Nullable PartitionState internPartition(byte[] serialized) {
@@ -529,31 +531,14 @@ final class DataEvolutionRowIdAssignmentPlanner {
deletedIdentifiers.release();
}
- private List<PartitionState> finishAddPass() {
- currentEntries.sort();
+ private List<PartitionState> selectFragmentedPartitions() {
List<PartitionState> selections = new ArrayList<>();
- long[] rangeScratch = new long[2];
- int groupStart = 0;
- while (groupStart < currentEntries.size()) {
- int partitionId = currentEntries.partitionId(groupStart);
- int groupEnd = groupStart + 1;
- while (groupEnd < currentEntries.size()
- && currentEntries.partitionId(groupEnd) ==
partitionId) {
- groupEnd++;
- }
- int rangeScan =
- currentEntries.scanLogicalRanges(groupStart, groupEnd,
rangeScratch);
- if (rangeScan > 0) {
- PrimitiveRowRanges logicalRanges =
- currentEntries.materializeLogicalRanges(
- groupStart, groupEnd, rangeScan,
rangeScratch);
- PartitionState partition =
partitions.partition(partitionId);
- partition.select(logicalRanges);
- selections.add(partition);
- }
- groupStart = groupEnd;
- }
- currentEntries.release();
+ liveFileRanges.finish(
+ (partitionId, logicalRanges) -> {
+ PartitionState partition =
partitions.partition(partitionId);
+ partition.select(logicalRanges);
+ selections.add(partition);
+ });
return selections;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/CurrentRowIdEntries.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/LiveFileRowIdRangeCollector.java
similarity index 61%
rename from
paimon-core/src/main/java/org/apache/paimon/append/dataevolution/CurrentRowIdEntries.java
rename to
paimon-core/src/main/java/org/apache/paimon/append/dataevolution/LiveFileRowIdRangeCollector.java
index de64624ad4..a9debe2d02 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/CurrentRowIdEntries.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/LiveFileRowIdRangeCollector.java
@@ -21,38 +21,49 @@ package org.apache.paimon.append.dataevolution;
import org.apache.paimon.utils.LongTripleArrayList;
import org.apache.paimon.utils.PrimitiveRowRanges;
-import javax.annotation.Nullable;
-
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkState;
-/** Object-free storage for current row-id entries. */
-final class CurrentRowIdEntries {
+/**
+ * Collects live file row-id ranges and emits logical ranges for fragmented
partitions.
+ *
+ * <p>Normal files define the logical range of an overlapping file group.
Dedicated files, such as
+ * blob and vector files, must be contained in that logical range. If a group
contains only
+ * dedicated files, their spanning range is used.
+ *
+ * <p>The hot collection path retains three primitive words per file and does
not retain manifest
+ * objects. {@link #finish(FragmentedPartitionConsumer)} is terminal and
releases this storage.
+ */
+final class LiveFileRowIdRangeCollector {
- private static final long SPECIAL = 1L << 32;
+ private static final long DEDICATED_FILE_FLAG = 1L << 32;
private final LongTripleArrayList entries;
+ private boolean finished;
- CurrentRowIdEntries() {
+ LiveFileRowIdRangeCollector() {
this(0);
}
- CurrentRowIdEntries(int expectedEntries) {
- checkArgument(expectedEntries >= 0, "Expected current entry count
cannot be negative.");
- this.entries = new LongTripleArrayList(expectedEntries);
+ LiveFileRowIdRangeCollector(int expectedFileCount) {
+ checkArgument(expectedFileCount >= 0, "Expected live file count cannot
be negative.");
+ this.entries = new LongTripleArrayList(expectedFileCount);
}
- void add(int partitionId, boolean special, long firstRowId, long rowCount)
{
+ void add(int partitionId, FileRole role, long firstRowId, long rowCount) {
+ checkState(!finished, "Cannot add a file range after the collector is
finished.");
checkArgument(partitionId >= 0, "Partition id cannot be negative.");
+ checkArgument(role != null, "File role cannot be null.");
checkArgument(rowCount > 0, "Row count must be positive.");
Math.addExact(firstRowId, rowCount - 1L);
entries.add(
- Integer.toUnsignedLong(partitionId) | (special ? SPECIAL : 0L),
+ Integer.toUnsignedLong(partitionId)
+ | (role == FileRole.DEDICATED ? DEDICATED_FILE_FLAG :
0L),
firstRowId,
rowCount);
}
- int size() {
+ int fileCount() {
return entries.size();
}
@@ -64,16 +75,12 @@ final class CurrentRowIdEntries {
return entries.usedLongCount();
}
- void release() {
- entries.release();
- }
-
- int partitionId(int index) {
+ private int partitionId(int index) {
return (int) entries.first(index);
}
- private boolean special(int index) {
- return (entries.first(index) & SPECIAL) != 0;
+ private boolean dedicatedFile(int index) {
+ return (entries.first(index) & DEDICATED_FILE_FLAG) != 0;
}
private long firstRowId(int index) {
@@ -88,7 +95,45 @@ final class CurrentRowIdEntries {
return firstRowId(index) + rowCount(index) - 1L;
}
- void sort() {
+ /**
+ * Emits only partitions whose logical row-id ranges contain gaps.
+ *
+ * <p>The callback owns each emitted {@link PrimitiveRowRanges}. This
collector is released even
+ * when range validation or the callback fails.
+ */
+ void finish(FragmentedPartitionConsumer consumer) {
+ checkState(!finished, "Live file row-id range collector is already
finished.");
+ checkArgument(consumer != null, "Fragmented partition consumer cannot
be null.");
+ finished = true;
+ try {
+ sortByPartitionAndRange();
+ long[] rangeScratch = new long[2];
+ LogicalRangeAnalysis analysis = new LogicalRangeAnalysis();
+ int partitionStart = 0;
+ while (partitionStart < entries.size()) {
+ int partitionId = partitionId(partitionStart);
+ int partitionEnd = partitionStart + 1;
+ while (partitionEnd < entries.size() &&
partitionId(partitionEnd) == partitionId) {
+ partitionEnd++;
+ }
+ analyzeLogicalRanges(partitionStart, partitionEnd,
rangeScratch, analysis);
+ if (analysis.fragmented) {
+ consumer.accept(
+ partitionId,
+ materializeLogicalRanges(
+ partitionStart,
+ partitionEnd,
+ analysis.rangeCount,
+ rangeScratch));
+ }
+ partitionStart = partitionEnd;
+ }
+ } finally {
+ entries.release();
+ }
+ }
+
+ private void sortByPartitionAndRange() {
if (entries.size() > 1) {
sort(0, entries.size() - 1);
}
@@ -151,11 +196,10 @@ final class CurrentRowIdEntries {
/**
* Scans logical ranges without retaining one object (or even one
primitive pair) per range.
*
- * <p>The absolute return value is the number of logical ranges. A
negative result means that
- * all logical ranges are contiguous and therefore this partition does not
need a plan. A
- * positive result means that the ranges are fragmented and need
materialization.
+ * <p>The result records both the number of logical ranges and whether
gaps exist between them.
*/
- int scanLogicalRanges(int from, int to, long[] rangeScratch) {
+ private void analyzeLogicalRanges(
+ int from, int to, long[] rangeScratch, LogicalRangeAnalysis
analysis) {
checkArgument(from >= 0 && from < to && to <= entries.size(), "Invalid
entry slice.");
int overlapStart = from;
long currentEnd = lastRowId(from);
@@ -184,10 +228,11 @@ final class CurrentRowIdEntries {
if (hasPrevious && (previousEnd == Long.MAX_VALUE || rangeScratch[0]
!= previousEnd + 1L)) {
contiguous = false;
}
- return contiguous ? -rangeCount : rangeCount;
+ analysis.rangeCount = rangeCount;
+ analysis.fragmented = !contiguous;
}
- PrimitiveRowRanges materializeLogicalRanges(
+ private PrimitiveRowRanges materializeLogicalRanges(
int from, int to, int expectedRangeCount, long[] rangeScratch) {
checkArgument(
from >= 0 && from < to && to <= entries.size() &&
expectedRangeCount > 0,
@@ -214,9 +259,9 @@ final class CurrentRowIdEntries {
}
private void computeLogicalRange(int from, int to, long[] result) {
- boolean hasOrdinary = false;
- long ordinaryStart = 0L;
- long ordinaryEnd = 0L;
+ boolean hasNormalFile = false;
+ long normalStart = 0L;
+ long normalEnd = 0L;
long spanningStart = Long.MAX_VALUE;
long spanningEnd = Long.MIN_VALUE;
for (int i = from; i < to; i++) {
@@ -224,17 +269,17 @@ final class CurrentRowIdEntries {
long end = lastRowId(i);
spanningStart = Math.min(spanningStart, start);
spanningEnd = Math.max(spanningEnd, end);
- if (!special(i)) {
+ if (!dedicatedFile(i)) {
checkState(
- !hasOrdinary || (ordinaryStart == start && ordinaryEnd
== end),
- "Data files in one overlapping row-id group must have
the same row-id range.");
- ordinaryStart = start;
- ordinaryEnd = end;
- hasOrdinary = true;
+ !hasNormalFile || (normalStart == start && normalEnd
== end),
+ "Normal files in one overlapping row-id group must
have the same row-id range.");
+ normalStart = start;
+ normalEnd = end;
+ hasNormalFile = true;
}
}
- long logicalStart = hasOrdinary ? ordinaryStart : spanningStart;
- long logicalEnd = hasOrdinary ? ordinaryEnd : spanningEnd;
+ long logicalStart = hasNormalFile ? normalStart : spanningStart;
+ long logicalEnd = hasNormalFile ? normalEnd : spanningEnd;
for (int i = from; i < to; i++) {
checkState(
firstRowId(i) >= logicalStart && lastRowId(i) <=
logicalEnd,
@@ -244,20 +289,20 @@ final class CurrentRowIdEntries {
result[1] = logicalEnd;
}
- @Nullable
- PrimitiveRowRanges selectedRangesForTesting() {
- checkState(entries.size() > 0, "Cannot inspect an empty current-entry
buffer.");
- sort();
- int partitionId = partitionId(0);
- for (int i = 1; i < entries.size(); i++) {
- checkState(
- partitionId(i) == partitionId,
- "The structural range test helper requires one
partition.");
- }
- long[] rangeScratch = new long[2];
- int rangeScan = scanLogicalRanges(0, entries.size(), rangeScratch);
- return rangeScan < 0
- ? null
- : materializeLogicalRanges(0, entries.size(), rangeScan,
rangeScratch);
+ enum FileRole {
+ NORMAL,
+ DEDICATED
+ }
+
+ @FunctionalInterface
+ interface FragmentedPartitionConsumer {
+
+ void accept(int partitionId, PrimitiveRowRanges logicalRanges);
+ }
+
+ private static final class LogicalRangeAnalysis {
+
+ private int rangeCount;
+ private boolean fragmented;
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CurrentRowIdEntriesTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CurrentRowIdEntriesTest.java
deleted file mode 100644
index 548b519833..0000000000
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CurrentRowIdEntriesTest.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.append.dataevolution;
-
-import org.apache.paimon.utils.PrimitiveRowRanges;
-
-import org.junit.jupiter.api.Test;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for {@link CurrentRowIdEntries}. */
-class CurrentRowIdEntriesTest {
-
- @Test
- void testHotStateUsesThreePrimitiveWords() {
- CurrentRowIdEntries entries = new CurrentRowIdEntries(10_000);
-
- for (int i = 0; i < 10_000; i++) {
- entries.add(i % 7, (i & 1) == 0, i * 10L, 3L);
- }
-
- assertThat(entries.size()).isEqualTo(10_000);
- assertThat(entries.usedWordCount()).isEqualTo(30_000);
- assertThat(entries.retainedWordCount()).isEqualTo(30_000);
- }
-
- @Test
- void testContiguousEntriesDoNotMaterializePerEntryRanges() {
- CurrentRowIdEntries entries = new CurrentRowIdEntries(10_000);
- for (int i = 0; i < 10_000; i++) {
- entries.add(0, false, i, 1L);
- }
-
- PrimitiveRowRanges ranges = entries.selectedRangesForTesting();
-
- assertThat(ranges).isNull();
- assertThat(entries.usedWordCount()).isEqualTo(30_000);
- assertThat(entries.retainedWordCount()).isEqualTo(30_000);
- }
-
- @Test
- void testFragmentedEntriesUsePrimitiveLogicalRanges() {
- CurrentRowIdEntries entries = new CurrentRowIdEntries(10_000);
- for (int i = 0; i < 10_000; i++) {
- entries.add(0, false, i * 2L, 1L);
- }
-
- PrimitiveRowRanges ranges = entries.selectedRangesForTesting();
-
- assertThat(ranges).isNotNull();
- assertThat(ranges.size()).isEqualTo(10_000);
- assertThat(ranges.retainedWordCount()).isEqualTo(20_000);
- assertThat(ranges.start(0)).isZero();
- assertThat(ranges.end(0)).isZero();
- assertThat(ranges.start(9_999)).isEqualTo(19_998L);
- assertThat(ranges.end(9_999)).isEqualTo(19_998L);
- assertThat(entries.usedWordCount()).isEqualTo(30_000);
- assertThat(entries.retainedWordCount()).isEqualTo(30_000);
- }
-}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
index 6fe5497585..b4a5c2396c 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
@@ -186,15 +186,17 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
}
@Test
- public void testInitialCapacityEstimatesLiveEntriesAfterDeletes() {
-
assertThat(DataEvolutionRowIdAssignmentPlanner.initialCurrentEntryCapacity(13_572_157L,
0L))
+ public void testInitialCapacityEstimatesLiveFileRangesAfterDeletes() {
+ assertThat(
+
DataEvolutionRowIdAssignmentPlanner.initialLiveFileRangeCapacity(
+ 13_572_157L, 0L))
.isEqualTo(13_572_157);
assertThat(
-
DataEvolutionRowIdAssignmentPlanner.initialCurrentEntryCapacity(
+
DataEvolutionRowIdAssignmentPlanner.initialLiveFileRangeCapacity(
800_000_000L, 799_999_990L))
.isEqualTo(10);
assertThat(
-
DataEvolutionRowIdAssignmentPlanner.initialCurrentEntryCapacity(
+
DataEvolutionRowIdAssignmentPlanner.initialLiveFileRangeCapacity(
800_000_000L, 0L))
.isEqualTo(1 << 24);
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/LiveFileRowIdRangeCollectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/LiveFileRowIdRangeCollectorTest.java
new file mode 100644
index 0000000000..7a05836d25
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/LiveFileRowIdRangeCollectorTest.java
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.append.dataevolution;
+
+import org.apache.paimon.utils.PrimitiveRowRanges;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static
org.apache.paimon.append.dataevolution.LiveFileRowIdRangeCollector.FileRole.DEDICATED;
+import static
org.apache.paimon.append.dataevolution.LiveFileRowIdRangeCollector.FileRole.NORMAL;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link LiveFileRowIdRangeCollector}. */
+class LiveFileRowIdRangeCollectorTest {
+
+ @Test
+ void testHotStateUsesThreePrimitiveWords() {
+ LiveFileRowIdRangeCollector ranges = new
LiveFileRowIdRangeCollector(10_000);
+
+ for (int i = 0; i < 10_000; i++) {
+ ranges.add(i % 7, (i & 1) == 0 ? NORMAL : DEDICATED, i * 10L, 3L);
+ }
+
+ assertThat(ranges.fileCount()).isEqualTo(10_000);
+ assertThat(ranges.usedWordCount()).isEqualTo(30_000);
+ assertThat(ranges.retainedWordCount()).isEqualTo(30_000);
+ }
+
+ @Test
+ void testEmptyCollectorCanFinishOnlyOnce() {
+ LiveFileRowIdRangeCollector ranges = new LiveFileRowIdRangeCollector();
+
+ assertThat(finish(ranges)).isEmpty();
+
+ assertThatThrownBy(() -> ranges.add(0, NORMAL, 0L, 1L))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("after the collector is finished");
+ assertThatThrownBy(() -> finish(ranges))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("already finished");
+ }
+
+ @Test
+ void testContiguousRangesDoNotSelectPartitionAndReleaseStorage() {
+ LiveFileRowIdRangeCollector ranges = new
LiveFileRowIdRangeCollector(10_000);
+ for (int i = 0; i < 10_000; i++) {
+ ranges.add(0, NORMAL, i, 1L);
+ }
+
+ Map<Integer, PrimitiveRowRanges> selections = finish(ranges);
+
+ assertThat(selections).isEmpty();
+ assertThat(ranges.usedWordCount()).isZero();
+ assertThat(ranges.retainedWordCount()).isZero();
+ }
+
+ @Test
+ void testFragmentedRangesSelectPartitionAndUsePrimitiveStorage() {
+ LiveFileRowIdRangeCollector ranges = new
LiveFileRowIdRangeCollector(10_000);
+ for (int i = 0; i < 10_000; i++) {
+ ranges.add(7, NORMAL, i * 2L, 1L);
+ }
+
+ Map<Integer, PrimitiveRowRanges> selections = finish(ranges);
+ PrimitiveRowRanges selected = selections.get(7);
+
+ assertThat(selections).containsOnlyKeys(7);
+ assertThat(selected.size()).isEqualTo(10_000);
+ assertThat(selected.retainedWordCount()).isEqualTo(20_000);
+ assertThat(selected.start(0)).isZero();
+ assertThat(selected.end(0)).isZero();
+ assertThat(selected.start(9_999)).isEqualTo(19_998L);
+ assertThat(selected.end(9_999)).isEqualTo(19_998L);
+ assertThat(ranges.usedWordCount()).isZero();
+ assertThat(ranges.retainedWordCount()).isZero();
+ }
+
+ @Test
+ void testNormalFilesDefineLogicalRanges() {
+ LiveFileRowIdRangeCollector ranges = new LiveFileRowIdRangeCollector();
+ ranges.add(3, DEDICATED, 2L, 3L);
+ ranges.add(3, NORMAL, 0L, 10L);
+ ranges.add(3, DEDICATED, 22L, 4L);
+ ranges.add(3, NORMAL, 20L, 10L);
+
+ PrimitiveRowRanges selected = finish(ranges).get(3);
+
+ assertRanges(selected, 0L, 9L, 20L, 29L);
+ }
+
+ @Test
+ void testDedicatedFilesDefineSpanningRangeWithoutNormalFile() {
+ LiveFileRowIdRangeCollector ranges = new LiveFileRowIdRangeCollector();
+ ranges.add(5, DEDICATED, 0L, 5L);
+ ranges.add(5, DEDICATED, 3L, 7L);
+ ranges.add(5, DEDICATED, 20L, 5L);
+
+ PrimitiveRowRanges selected = finish(ranges).get(5);
+
+ assertRanges(selected, 0L, 9L, 20L, 24L);
+ }
+
+ @Test
+ void testOverlappingNormalFilesMustHaveIdenticalRanges() {
+ LiveFileRowIdRangeCollector ranges = new LiveFileRowIdRangeCollector();
+ ranges.add(0, NORMAL, 0L, 10L);
+ ranges.add(0, NORMAL, 5L, 10L);
+
+ assertThatThrownBy(() -> finish(ranges))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining(
+ "Normal files in one overlapping row-id group must
have the same row-id range");
+ assertThat(ranges.retainedWordCount()).isZero();
+ }
+
+ @Test
+ void testDedicatedFileMustBeInsideLogicalRange() {
+ LiveFileRowIdRangeCollector ranges = new LiveFileRowIdRangeCollector();
+ ranges.add(0, DEDICATED, 0L, 7L);
+ ranges.add(0, NORMAL, 5L, 6L);
+
+ assertThatThrownBy(() -> finish(ranges))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("File row-id range is outside its
logical row-id range");
+ assertThat(ranges.retainedWordCount()).isZero();
+ }
+
+ private static Map<Integer, PrimitiveRowRanges>
finish(LiveFileRowIdRangeCollector collector) {
+ Map<Integer, PrimitiveRowRanges> selections = new LinkedHashMap<>();
+ collector.finish(
+ (partitionId, logicalRanges) -> {
+ PrimitiveRowRanges previous = selections.put(partitionId,
logicalRanges);
+ assertThat(previous).isNull();
+ });
+ return selections;
+ }
+
+ private static void assertRanges(PrimitiveRowRanges ranges, long...
boundaries) {
+ assertThat(ranges).isNotNull();
+ assertThat(boundaries.length % 2).isZero();
+ assertThat(ranges.size()).isEqualTo(boundaries.length / 2);
+ for (int i = 0; i < ranges.size(); i++) {
+ assertThat(ranges.start(i)).isEqualTo(boundaries[i * 2]);
+ assertThat(ranges.end(i)).isEqualTo(boundaries[i * 2 + 1]);
+ }
+ }
+}
