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 27f4656bb0 [core] Optimize data evolution row-id assignment (#8904)
27f4656bb0 is described below
commit 27f4656bb033e8fdb9b9574559914fdcc09d5485
Author: YeJunHao <[email protected]>
AuthorDate: Thu Jul 30 13:19:27 2026 +0800
[core] Optimize data evolution row-id assignment (#8904)
---
docs/generated/core_configuration.html | 18 +-
.../main/java/org/apache/paimon/CoreOptions.java | 18 +
.../apache/paimon/utils/PrimitiveRowRanges.java | 20 +
.../paimon/utils/PrimitiveRowRangesTest.java | 18 +
.../DataEvolutionRowIdAssignmentPlanner.java | 348 +++++------
.../DataEvolutionRowIdReassigner.java | 123 ++--
.../dataevolution/LiveFileRowIdRangeCollector.java | 635 +++++++++++++++------
.../paimon/manifest/BinaryManifestEntry.java | 17 +
.../DataEvolutionRowIdReassignerTest.java | 371 ++++++++++--
.../LiveFileRowIdRangeCollectorTest.java | 74 ++-
10 files changed, 1221 insertions(+), 421 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 6e80db093f..5670cc65c8 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -170,6 +170,12 @@ under the License.
<td>Boolean</td>
<td>Whether enabled chain table.</td>
</tr>
+ <tr>
+ <td><h5>chain-table.split.key-range-enabled</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>If true, a batch chain-table scan splits each bucket's
snapshot and delta files into multiple splits by key range to improve read
parallelism. Files with intersecting key ranges always stay in the same split
so that all versions of a key across the snapshot and delta branches are merged
together. Set to false to fall back to one split per bucket.</td>
+ </tr>
<tr>
<td><h5>chain-table.streaming.merge-snapshot</h5></td>
<td style="word-wrap: break-word;">false</td>
@@ -506,6 +512,12 @@ under the License.
<td>Boolean</td>
<td>Whether to persist source when process merge into action on
data evolution table.</td>
</tr>
+ <tr>
+ <td><h5>data-evolution.reassign.skip-contiguous-row-count</h5></td>
+ <td style="word-wrap: break-word;">1000000000</td>
+ <td>Long</td>
+ <td>Strictly contiguous same-partition logical row-id runs
containing more than this number of rows are excluded from row-id reassignment.
Set to 0 to disable this filtering.</td>
+ </tr>
<tr>
<td><h5>data-evolution.row-sidecar.enabled</h5></td>
<td style="word-wrap: break-word;">false</td>
@@ -1416,12 +1428,6 @@ For an internal format table in a REST catalog, it also
makes the catalog own th
<td>Long</td>
<td>After configuring this time, only the data files created after
this time will be read. It is independent of snapshots, but it is imprecise
filtering (depending on whether or not compaction occurs).</td>
</tr>
- <tr>
- <td><h5>chain-table.split.key-range-enabled</h5></td>
- <td style="word-wrap: break-word;">true</td>
- <td>Boolean</td>
- <td>If true, a batch chain-table scan splits each bucket's
snapshot and delta files into multiple splits by key range to improve read
parallelism. Files with intersecting key ranges always stay in the same split
so that all versions of a key across the snapshot and delta branches are merged
together. Set to false to fall back to one split per bucket.</td>
- </tr>
<tr>
<td><h5>scan.ignore-corrupt-files</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index f3f22b6e7b..7449142e8f 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2497,6 +2497,15 @@ public class CoreOptions implements Serializable {
.defaultValue(false)
.withDescription("Whether enable data evolution for row
tracking table.");
+ public static final ConfigOption<Long>
DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT =
+ key("data-evolution.reassign.skip-contiguous-row-count")
+ .longType()
+ .defaultValue(1_000_000_000L)
+ .withDescription(
+ "Strictly contiguous same-partition logical row-id
runs containing "
+ + "more than this number of rows are
excluded from row-id "
+ + "reassignment. Set to 0 to disable this
filtering.");
+
public static final ConfigOption<Boolean>
DATA_EVOLUTION_ROW_SIDECAR_ENABLED =
key("data-evolution.row-sidecar.enabled")
.booleanType()
@@ -4217,6 +4226,15 @@ public class CoreOptions implements Serializable {
return options.get(DATA_EVOLUTION_ENABLED);
}
+ public long dataEvolutionReassignSkipContiguousRowCount() {
+ long threshold =
options.get(DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT);
+ checkArgument(
+ threshold >= 0,
+ "The option %s cannot be negative.",
+ DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT.key());
+ return threshold;
+ }
+
public boolean dataEvolutionRowSidecarEnabled() {
return options.get(DATA_EVOLUTION_ROW_SIDECAR_ENABLED);
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/utils/PrimitiveRowRanges.java
b/paimon-common/src/main/java/org/apache/paimon/utils/PrimitiveRowRanges.java
index 423f996f9b..da63c1be22 100644
---
a/paimon-common/src/main/java/org/apache/paimon/utils/PrimitiveRowRanges.java
+++
b/paimon-common/src/main/java/org/apache/paimon/utils/PrimitiveRowRanges.java
@@ -119,6 +119,26 @@ public final class PrimitiveRowRanges {
normalized = true;
}
+ /**
+ * Returns whether any stored range overlaps the inclusive range from
{@code start} to {@code
+ * end}.
+ */
+ public boolean overlaps(long start, long end) {
+ checkArgument(start <= end, "Invalid row range [%s, %s].", start, end);
+ normalizeOverlapping();
+ int lower = 0;
+ int upper = size;
+ while (lower < upper) {
+ int middle = lower + ((upper - lower) >>> 1);
+ if (ends[middle] < start) {
+ lower = middle + 1;
+ } else {
+ upper = middle;
+ }
+ }
+ return lower < size && starts[lower] <= end;
+ }
+
/**
* Returns whether these ranges fully cover the inclusive range from
{@code start} to {@code
* end}.
diff --git
a/paimon-common/src/test/java/org/apache/paimon/utils/PrimitiveRowRangesTest.java
b/paimon-common/src/test/java/org/apache/paimon/utils/PrimitiveRowRangesTest.java
index 2a16b00e3b..e39adb5a64 100644
---
a/paimon-common/src/test/java/org/apache/paimon/utils/PrimitiveRowRangesTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/utils/PrimitiveRowRangesTest.java
@@ -78,6 +78,22 @@ class PrimitiveRowRangesTest {
assertThat(ranges.size()).isEqualTo(2);
}
+ @Test
+ void testOverlaps() {
+ PrimitiveRowRanges ranges = new PrimitiveRowRanges(4);
+ ranges.add(20L, 25L);
+ ranges.add(5L, 8L);
+ ranges.add(8L, 12L);
+ ranges.add(30L, 35L);
+
+ assertThat(ranges.overlaps(0L, 4L)).isFalse();
+ assertThat(ranges.overlaps(4L, 5L)).isTrue();
+ assertThat(ranges.overlaps(12L, 19L)).isTrue();
+ assertThat(ranges.overlaps(13L, 19L)).isFalse();
+ assertThat(ranges.overlaps(24L, 31L)).isTrue();
+ assertThat(ranges.overlaps(36L, Long.MAX_VALUE)).isFalse();
+ }
+
@Test
void testCovers() {
PrimitiveRowRanges ranges = new PrimitiveRowRanges(4);
@@ -111,6 +127,8 @@ class PrimitiveRowRangesTest {
assertThatThrownBy(() -> ranges.add(2L,
1L)).isInstanceOf(IllegalArgumentException.class);
assertThatThrownBy(() ->
ranges.start(0)).isInstanceOf(IllegalArgumentException.class);
assertThatThrownBy(() ->
ranges.append(null)).isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> ranges.overlaps(2L, 1L))
+ .isInstanceOf(IllegalArgumentException.class);
assertThatThrownBy(() -> ranges.covers(2L, 1L))
.isInstanceOf(IllegalArgumentException.class);
}
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 d40e0fd52c..0b241afde5 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
@@ -39,15 +39,20 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.ByteArrayKey;
import org.apache.paimon.utils.ByteArrayLookupKey;
import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.PrimitiveRowRanges;
import org.apache.paimon.utils.SerializationUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -65,6 +70,8 @@ import static
org.apache.paimon.utils.Preconditions.checkState;
*/
final class DataEvolutionRowIdAssignmentPlanner {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(DataEvolutionRowIdAssignmentPlanner.class);
private static final int EXCLUDED_PARTITION_CACHE_SIZE = 1024;
private static final int MAX_INITIAL_LIVE_FILE_RANGES = 1 << 24;
private static final BinaryString ROW_ID_FIELD =
@@ -81,16 +88,14 @@ final class DataEvolutionRowIdAssignmentPlanner {
DataFileMeta.EMBEDDED_FILE_INDEX,
DataFileMeta.EXTERNAL_PATH,
DataFileMeta.FIRST_ROW_ID,
- DataFileMeta.WRITE_COLS,
- DataFileMeta.MAX_SEQUENCE_NUMBER);
+ DataFileMeta.WRITE_COLS);
private static final Projection COMPACT_ADD_PROJECTION =
manifestProjection(
false,
DataFileMeta.FILE_NAME,
DataFileMeta.ROW_COUNT,
DataFileMeta.FIRST_ROW_ID,
- DataFileMeta.WRITE_COLS,
- DataFileMeta.MAX_SEQUENCE_NUMBER);
+ DataFileMeta.WRITE_COLS);
private static final Projection REWRITE_PROJECTION =
manifestProjection(
false,
@@ -103,10 +108,10 @@ final class DataEvolutionRowIdAssignmentPlanner {
private final @Nullable PartitionPredicate partitionPredicate;
private final List<ManifestFileMeta> manifestMetas;
private final Map<String, Integer> manifestOrdinals;
+ private final boolean[] plannedManifests;
private final boolean[] rewrittenManifests;
private final Map<ByteArrayKey, SelectedPartition> selectedPartitions;
- private long nextManifestGroupOrdinal;
- private long nextRetainedAddScanOrdinal;
+ private final long skipContiguousRowCount;
DataEvolutionRowIdAssignmentPlanner(
FileStoreTable table,
@@ -117,8 +122,11 @@ final class DataEvolutionRowIdAssignmentPlanner {
this.partitionPredicate = partitionPredicate;
this.manifestMetas = manifestMetas;
this.manifestOrdinals = manifestOrdinals(manifestMetas);
+ this.plannedManifests = new boolean[manifestMetas.size()];
this.rewrittenManifests = new boolean[manifestMetas.size()];
this.selectedPartitions = new LinkedHashMap<>();
+ this.skipContiguousRowCount =
+
table.coreOptions().dataEvolutionReassignSkipContiguousRowCount();
}
private static Projection manifestProjection(
@@ -145,8 +153,6 @@ final class DataEvolutionRowIdAssignmentPlanner {
}
private void planGroup(List<ManifestFileMeta> manifestGroup) {
- long manifestGroupOrdinal = nextManifestGroupOrdinal;
- nextManifestGroupOrdinal = Math.addExact(nextManifestGroupOrdinal, 1L);
GroupState group =
new GroupState(
table.schema().logicalPartitionType().getFieldCount(),
@@ -155,21 +161,26 @@ final class DataEvolutionRowIdAssignmentPlanner {
partitionPredicate == null
? initialLiveFileRangeCapacity(manifestGroup)
: 0);
+ for (ManifestFileMeta manifestMeta : manifestGroup) {
+ plannedManifests[ordinal(manifestMeta)] = true;
+ }
ReusableIdentifier identifier = new ReusableIdentifier();
long[] rowRangeScratch = new long[2];
- collectDeletedIdentifiers(manifestGroup, group, identifier);
- collectLiveFileRanges(
- manifestGroup, group, identifier, manifestGroupOrdinal,
rowRangeScratch);
- identifier.release();
- group.releaseDeletedIdentifiers();
+ try {
+ collectDeletedIdentifiers(manifestGroup, group, identifier);
+ collectLiveFileRanges(manifestGroup, group, identifier,
rowRangeScratch);
+ identifier.release();
+ group.releaseDeletedIdentifiers();
- List<PartitionState> selections = group.selectFragmentedPartitions();
- for (PartitionState selection : selections) {
- mergeSelectedPartition(selection);
- }
- if (!selections.isEmpty()) {
- markRewrittenManifests(manifestGroup, group, rowRangeScratch);
+ List<PartitionState> selections =
group.selectFragmentedPartitions();
+ for (PartitionState selection : selections) {
+ mergeSelectedPartition(selection);
+ }
+ } catch (RuntimeException | Error e) {
+ identifier.release();
+ group.abort();
+ throw e;
}
}
@@ -206,7 +217,6 @@ final class DataEvolutionRowIdAssignmentPlanner {
List<ManifestFileMeta> manifestGroup,
GroupState group,
ReusableIdentifier identifier,
- long manifestGroupOrdinal,
long[] rowRangeScratch) {
Projection addProjection =
group.deletedIdentifiers.isEmpty()
@@ -243,17 +253,8 @@ final class DataEvolutionRowIdAssignmentPlanner {
!file.containsWriteColumn(ROW_ID_FIELD),
"Cannot reassign row IDs for file '%s' because it
physically stores the row-id field.",
fileName);
- long maxSequenceNumber = file.maxSequenceNumber();
- long retainedAddScanOrdinal = nextRetainedAddScanOrdinal;
- nextRetainedAddScanOrdinal =
Math.addExact(nextRetainedAddScanOrdinal, 1L);
int fileOrder = fileOrder(fileName);
- partition.considerLegacyOrderKey(
- manifestGroupOrdinal,
- rowRangeScratch[0],
- fileOrder,
- maxSequenceNumber,
- fileName,
- retainedAddScanOrdinal);
+ partition.setMinFirstRowId(rowRangeScratch[0]);
group.liveFileRanges.add(
partition.id,
fileOrder == 0 ? NORMAL : DEDICATED,
@@ -266,22 +267,35 @@ final class DataEvolutionRowIdAssignmentPlanner {
}
}
- private void markRewrittenManifests(
- List<ManifestFileMeta> manifestGroup, GroupState group, long[]
rowRangeScratch) {
- for (ManifestFileMeta manifestMeta : manifestGroup) {
- int manifestOrdinal = ordinal(manifestMeta);
+ private void markRewrittenManifests() {
+ ByteArrayLookupKey lookup = new ByteArrayLookupKey();
+ long[] rowRangeScratch = new long[2];
+ for (int manifestOrdinal = 0; manifestOrdinal < manifestMetas.size();
manifestOrdinal++) {
+ if (!plannedManifests[manifestOrdinal]) {
+ continue;
+ }
+ ManifestFileMeta manifestMeta = manifestMetas.get(manifestOrdinal);
+ if (!manifestMayContainSelectedRange(manifestMeta)) {
+ continue;
+ }
try (CloseableIterator<BinaryManifestEntry> entries =
manifestFile.scan(
manifestMeta.fileName(), manifestMeta.fileSize(),
REWRITE_PROJECTION)) {
while (entries.hasNext()) {
BinaryManifestEntry entry = entries.next();
- PartitionState partition =
group.internPartition(entry.partitionBytes());
- if (partition == null || partition.logicalRanges == null) {
+ lookup.reset(entry.partitionBytes());
+ SelectedPartition selection;
+ try {
+ selection = selectedPartitions.get(lookup);
+ } finally {
+ lookup.clear();
+ }
+ if (selection == null) {
continue;
}
BinaryDataFileMeta file = entry.file();
readRowRange(file, manifestOrdinal, null, rowRangeScratch);
- if (!partition.logicalRanges.covers(rowRangeScratch[0],
rowRangeScratch[1])) {
+ if (!selection.logicalRanges.covers(rowRangeScratch[0],
rowRangeScratch[1])) {
continue;
}
checkState(
@@ -297,11 +311,55 @@ final class DataEvolutionRowIdAssignmentPlanner {
}
}
+ 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 (partition.logicalRanges.overlaps(minimum, maximum)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
private Result buildResult() {
if (selectedPartitions.isEmpty()) {
return new Result(new int[0], Collections.emptyMap(), 0L);
}
+ long skippedRangeCount = 0L;
+ long skippedRowCount = 0L;
+ Iterator<Map.Entry<ByteArrayKey, SelectedPartition>> selectedIterator =
+ selectedPartitions.entrySet().iterator();
+ while (selectedIterator.hasNext()) {
+ SelectedPartition partition = selectedIterator.next().getValue();
+ partition.logicalRanges.normalizeOverlapping();
+ if (skipContiguousRowCount > 0) {
+ Pair<Long, Long> skipped =
+
partition.removeLargeContiguousRuns(skipContiguousRowCount);
+ skippedRangeCount = Math.addExact(skippedRangeCount,
skipped.getLeft());
+ skippedRowCount = Math.addExact(skippedRowCount,
skipped.getRight());
+ }
+ 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(
@@ -309,15 +367,16 @@ final class DataEvolutionRowIdAssignmentPlanner {
partitions.sort(
(left, right) -> {
int comparison = typedComparator.compare(left.partition,
right.partition);
+ // Row IDs are globally unique, so this also orders
binary-distinct
+ // partitions which compare equal, for example different
NaN payloads.
return comparison != 0
? comparison
- :
left.legacyOrderKey.compareTo(right.legacyOrderKey);
+ : Long.compare(left.minFirstRowId,
right.minFirstRowId);
});
Map<BinaryRow, RowRangeMappingIndex> mappings = new LinkedHashMap<>();
long nextOffset = 0L;
for (SelectedPartition partition : partitions) {
- partition.logicalRanges.normalizeOverlapping();
int rangeCount = partition.logicalRanges.size();
checkState(rangeCount > 0, "Selected partition has no logical
row-id ranges.");
PrimitiveRowRanges.Owned ownedRanges =
partition.logicalRanges.takeOwned();
@@ -358,15 +417,13 @@ final class DataEvolutionRowIdAssignmentPlanner {
ByteArrayLookupKey lookup = new
ByteArrayLookupKey(selection.serialized);
SelectedPartition selected = selectedPartitions.get(lookup);
if (selected == null) {
- LegacyPartitionOrderKey legacyOrderKey =
selection.requiredLegacyOrderKey();
- selected = new SelectedPartition(selection.partition,
legacyOrderKey, logicalRanges);
+ selected =
+ new SelectedPartition(
+ selection.partition, selection.minFirstRowId,
logicalRanges);
selectedPartitions.put(new ByteArrayKey(selection.serialized),
selected);
return;
}
- LegacyPartitionOrderKey incomingOrderKey =
selection.requiredLegacyOrderKey();
- if (incomingOrderKey.compareTo(selected.legacyOrderKey) < 0) {
- selected.legacyOrderKey = incomingOrderKey;
- }
+ selected.minFirstRowId = Math.min(selected.minFirstRowId,
selection.minFirstRowId);
selected.logicalRanges.append(logicalRanges);
selected.logicalRanges.normalizeOverlapping();
}
@@ -531,6 +588,11 @@ final class DataEvolutionRowIdAssignmentPlanner {
});
return selections;
}
+
+ private void abort() {
+ deletedIdentifiers.release();
+ liveFileRanges.abort();
+ }
}
private static final class GroupPartitionDictionary {
@@ -623,7 +685,7 @@ final class DataEvolutionRowIdAssignmentPlanner {
private final int id;
private final byte[] serialized;
private final BinaryRow partition;
- private @Nullable LegacyPartitionOrderKey legacyOrderKey;
+ private long minFirstRowId = Long.MAX_VALUE;
private @Nullable PrimitiveRowRanges logicalRanges;
private PartitionState(int id, byte[] serialized, BinaryRow partition)
{
@@ -632,66 +694,8 @@ final class DataEvolutionRowIdAssignmentPlanner {
this.partition = partition;
}
- private void considerLegacyOrderKey(
- long manifestGroupOrdinal,
- long firstRowId,
- int fileOrder,
- long maxSequenceNumber,
- BinaryString fileName,
- long retainedAddScanOrdinal) {
- if (legacyOrderKey == null) {
- legacyOrderKey =
- new LegacyPartitionOrderKey(
- manifestGroupOrdinal,
- firstRowId,
- fileOrder,
- maxSequenceNumber,
- fileName.toString(),
- retainedAddScanOrdinal);
- return;
- }
-
- int comparison =
- Long.compare(manifestGroupOrdinal,
legacyOrderKey.manifestGroupOrdinal);
- if (comparison == 0) {
- comparison = Long.compare(firstRowId,
legacyOrderKey.firstRowId);
- }
- if (comparison == 0) {
- comparison = Integer.compare(fileOrder,
legacyOrderKey.fileOrder);
- }
- if (comparison == 0) {
- comparison = Long.compare(legacyOrderKey.maxSequenceNumber,
maxSequenceNumber);
- }
-
- String stableFileName = null;
- if (comparison == 0) {
- stableFileName = fileName.toString();
- comparison = stableFileName.compareTo(legacyOrderKey.fileName);
- }
- if (comparison == 0) {
- comparison =
- Long.compare(retainedAddScanOrdinal,
legacyOrderKey.retainedAddScanOrdinal);
- }
- if (comparison < 0) {
- if (stableFileName == null) {
- stableFileName = fileName.toString();
- }
- legacyOrderKey =
- new LegacyPartitionOrderKey(
- manifestGroupOrdinal,
- firstRowId,
- fileOrder,
- maxSequenceNumber,
- stableFileName,
- retainedAddScanOrdinal);
- }
- }
-
- private LegacyPartitionOrderKey requiredLegacyOrderKey() {
- checkState(
- legacyOrderKey != null,
- "Selected partition does not have a retained ADD ordering
key.");
- return legacyOrderKey;
+ private void setMinFirstRowId(long firstRowId) {
+ minFirstRowId = Math.min(minFirstRowId, firstRowId);
}
private void select(PrimitiveRowRanges logicalRanges) {
@@ -705,69 +709,91 @@ final class DataEvolutionRowIdAssignmentPlanner {
}
}
- private static final class LegacyPartitionOrderKey
- implements Comparable<LegacyPartitionOrderKey> {
-
- private final long manifestGroupOrdinal;
- private final long firstRowId;
- private final int fileOrder;
- private final long maxSequenceNumber;
- private final String fileName;
- private final long retainedAddScanOrdinal;
-
- private LegacyPartitionOrderKey(
- long manifestGroupOrdinal,
- long firstRowId,
- int fileOrder,
- long maxSequenceNumber,
- String fileName,
- long retainedAddScanOrdinal) {
- this.manifestGroupOrdinal = manifestGroupOrdinal;
- this.firstRowId = firstRowId;
- this.fileOrder = fileOrder;
- this.maxSequenceNumber = maxSequenceNumber;
- this.fileName = fileName;
- this.retainedAddScanOrdinal = retainedAddScanOrdinal;
- }
-
- @Override
- public int compareTo(LegacyPartitionOrderKey other) {
- int comparison = Long.compare(manifestGroupOrdinal,
other.manifestGroupOrdinal);
- if (comparison != 0) {
- return comparison;
+ private static final class SelectedPartition {
+
+ private final BinaryRow partition;
+ private long minFirstRowId;
+ private PrimitiveRowRanges logicalRanges;
+
+ private SelectedPartition(
+ BinaryRow partition, long minFirstRowId, PrimitiveRowRanges
logicalRanges) {
+ this.partition = partition;
+ this.minFirstRowId = minFirstRowId;
+ this.logicalRanges = logicalRanges;
+ }
+
+ private Pair<Long, Long> removeLargeContiguousRuns(long threshold) {
+ checkArgument(threshold > 0, "Skip threshold must be positive.");
+ int originalRangeCount = logicalRanges.size();
+ int retainedRangeCount = 0;
+ long skippedRangeCount = 0L;
+ long skippedRowCount = 0L;
+
+ int index = 0;
+ while (index < originalRangeCount) {
+ int runEnd = contiguousRunEnd(index);
+ long start = logicalRanges.start(index);
+ long end = logicalRanges.end(runEnd);
+ if (rangeCountExceeds(start, end, threshold)) {
+ skippedRangeCount =
+ Math.addExact(skippedRangeCount, (long) runEnd -
index + 1L);
+ skippedRowCount =
+ Math.addExact(skippedRowCount,
inclusiveRangeCount(start, end));
+ } else {
+ retainedRangeCount = Math.addExact(retainedRangeCount,
runEnd - index + 1);
+ }
+ index = runEnd + 1;
}
- comparison = Long.compare(firstRowId, other.firstRowId);
- if (comparison != 0) {
- return comparison;
+
+ if (skippedRangeCount == 0L) {
+ return Pair.of(0L, 0L);
}
- comparison = Integer.compare(fileOrder, other.fileOrder);
- if (comparison != 0) {
- return comparison;
+
+ PrimitiveRowRanges retained = new
PrimitiveRowRanges(retainedRangeCount);
+ index = 0;
+ while (index < originalRangeCount) {
+ int runEnd = contiguousRunEnd(index);
+ long start = logicalRanges.start(index);
+ long end = logicalRanges.end(runEnd);
+ if (!rangeCountExceeds(start, end, threshold)) {
+ for (int rangeIndex = index; rangeIndex <= runEnd;
rangeIndex++) {
+ retained.add(
+ logicalRanges.start(rangeIndex),
logicalRanges.end(rangeIndex));
+ }
+ }
+ index = runEnd + 1;
}
- comparison = Long.compare(other.maxSequenceNumber,
maxSequenceNumber);
- if (comparison != 0) {
- return comparison;
+ logicalRanges = retained;
+ return Pair.of(skippedRangeCount, skippedRowCount);
+ }
+
+ private boolean hasFragmentedLogicalRanges() {
+ for (int index = 1; index < logicalRanges.size(); index++) {
+ if (!adjacent(logicalRanges.end(index - 1),
logicalRanges.start(index))) {
+ return true;
+ }
}
- comparison = fileName.compareTo(other.fileName);
- return comparison != 0
- ? comparison
- : Long.compare(retainedAddScanOrdinal,
other.retainedAddScanOrdinal);
+ return false;
}
- }
- private static final class SelectedPartition {
+ private int contiguousRunEnd(int runStart) {
+ int runEnd = runStart;
+ while (runEnd + 1 < logicalRanges.size()
+ && adjacent(logicalRanges.end(runEnd),
logicalRanges.start(runEnd + 1))) {
+ runEnd++;
+ }
+ return runEnd;
+ }
- private final BinaryRow partition;
- private LegacyPartitionOrderKey legacyOrderKey;
- private final PrimitiveRowRanges logicalRanges;
+ private static boolean adjacent(long leftEnd, long rightStart) {
+ return leftEnd != Long.MAX_VALUE && rightStart == leftEnd + 1L;
+ }
- private SelectedPartition(
- BinaryRow partition,
- LegacyPartitionOrderKey legacyOrderKey,
- PrimitiveRowRanges logicalRanges) {
- this.partition = partition;
- this.legacyOrderKey = legacyOrderKey;
- this.logicalRanges = logicalRanges;
+ private static boolean rangeCountExceeds(long start, long end, long
threshold) {
+ if (start > Long.MAX_VALUE - threshold) {
+ return false;
+ }
+ return end >= start + threshold;
}
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
index 97c19bd66f..3a731143f8 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
@@ -26,6 +26,7 @@ import org.apache.paimon.codegen.RecordComparator;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.manifest.BinaryManifestEntry;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.IndexManifestFile;
@@ -39,7 +40,7 @@ import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.SpecialFields;
-import org.apache.paimon.utils.Filter;
+import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
@@ -401,10 +402,15 @@ public class DataEvolutionRowIdReassigner {
previous.id(),
latest.id());
+ Set<String> previousManifestFiles = new HashSet<>();
+ for (ManifestFileMeta manifestMeta :
manifestList.readDataManifests(previous)) {
+ previousManifestFiles.add(manifestMeta.fileName());
+ }
Map<String, ManifestFileMeta> manifestMetasToRewrite = new
LinkedHashMap<>();
for (ManifestFileMeta manifestMeta :
assignmentPlan.manifestMetasToRewrite) {
manifestMetasToRewrite.put(manifestMeta.fileName(), manifestMeta);
}
+ Map<String, Boolean> newManifestNeedsReassign = new HashMap<>();
for (long id = previous.id() + 1; id <= latest.id(); id++) {
Snapshot snapshot;
try {
@@ -433,18 +439,10 @@ public class DataEvolutionRowIdReassigner {
snapshot.commitKind());
for (ManifestFileMeta manifestMeta :
manifestList.readDeltaManifests(snapshot)) {
- boolean needsReassign = false;
- for (ManifestEntry entry :
- readPlanningManifestEntries(manifestFile,
manifestMeta)) {
- checkState(
- entry.kind() == FileKind.ADD,
- "APPEND snapshot %s contains non-ADD manifest
entry %s.",
- snapshot.id(),
- entry);
- if (appendedEntryNeedsReassign(assignmentPlan, entry)) {
- needsReassign = true;
- }
- }
+ boolean needsReassign =
+ appendedManifestNeedsReassign(
+ assignmentPlan, manifestFile, manifestMeta,
snapshot.id());
+ newManifestNeedsReassign.put(manifestMeta.fileName(),
needsReassign);
if (needsReassign) {
manifestMetasToRewrite.put(manifestMeta.fileName(),
manifestMeta);
}
@@ -452,21 +450,90 @@ public class DataEvolutionRowIdReassigner {
}
List<ManifestFileMeta> latestManifestMetas =
manifestList.readDataManifests(latest);
- Set<String> latestManifestFiles = new HashSet<>();
+ Map<String, ManifestFileMeta> reboundManifestMetasToRewrite = new
LinkedHashMap<>();
for (ManifestFileMeta manifestMeta : latestManifestMetas) {
- latestManifestFiles.add(manifestMeta.fileName());
- }
- for (String plannedManifestFile : manifestMetasToRewrite.keySet()) {
- checkState(
- latestManifestFiles.contains(plannedManifestFile),
- "Cannot advance row-id assignment because planned manifest
%s no longer exists after APPEND manifest merge.",
- plannedManifestFile);
+ String manifestFileName = manifestMeta.fileName();
+ boolean needsReassign =
manifestMetasToRewrite.containsKey(manifestFileName);
+ if (!needsReassign &&
!previousManifestFiles.contains(manifestFileName)) {
+ Boolean cached =
newManifestNeedsReassign.get(manifestFileName);
+ needsReassign =
+ cached != null
+ ? cached
+ : manifestContainsMappedEntry(
+ assignmentPlan, manifestFile,
manifestMeta);
+ }
+ if (needsReassign) {
+ reboundManifestMetasToRewrite.put(manifestFileName,
manifestMeta);
+ }
}
+ checkState(
+ !reboundManifestMetasToRewrite.isEmpty(),
+ "Cannot advance row-id assignment because no current manifest
contains the planned row-id ranges.");
return new AssignmentPlan(
- new ArrayList<>(manifestMetasToRewrite.values()),
+ new ArrayList<>(reboundManifestMetasToRewrite.values()),
assignmentPlan.relativeRowIdMappings);
}
+ private boolean manifestContainsMappedEntry(
+ AssignmentPlan assignmentPlan,
+ ManifestFile manifestFile,
+ ManifestFileMeta manifestMeta) {
+ try (CloseableIterator<BinaryManifestEntry> entries =
+ manifestFile.scan(
+ manifestMeta.fileName(),
+ manifestMeta.fileSize(),
+ BinaryManifestEntry.ROW_RANGE_PROJECTION)) {
+ while (entries.hasNext()) {
+ BinaryManifestEntry entry = entries.next();
+ RowRangeMappingIndex mapping =
+
assignmentPlan.relativeRowIdMappings.mappings.get(entry.partition());
+ if (mapping != null &&
mapping.map(entry.file().nonNullRowIdRange()).isPresent()) {
+ return true;
+ }
+ }
+ return false;
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to scan manifest file " + manifestMeta.fileName(),
e);
+ }
+ }
+
+ private boolean appendedManifestNeedsReassign(
+ AssignmentPlan assignmentPlan,
+ ManifestFile manifestFile,
+ ManifestFileMeta manifestMeta,
+ long appendSnapshotId) {
+ boolean needsReassign = false;
+ try (CloseableIterator<BinaryManifestEntry> entries =
+ manifestFile.scan(
+ manifestMeta.fileName(),
+ manifestMeta.fileSize(),
+ BinaryManifestEntry.ROW_RANGE_PROJECTION)) {
+ while (entries.hasNext()) {
+ BinaryManifestEntry entry = entries.next();
+ if (partitionPredicate != null &&
!partitionPredicate.test(entry.partition())) {
+ continue;
+ }
+ checkState(
+ entry.isAdd(),
+ "APPEND snapshot %s contains a non-ADD entry in
manifest %s.",
+ appendSnapshotId,
+ manifestMeta.fileName());
+ if (appendedEntryNeedsReassign(assignmentPlan, entry)) {
+ needsReassign = true;
+ }
+ }
+ return needsReassign;
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to scan manifest file " + manifestMeta.fileName(),
e);
+ }
+ }
+
private boolean appendedEntryNeedsReassign(
AssignmentPlan assignmentPlan, ManifestEntry appendedEntry) {
RowRangeMappingIndex mapping =
@@ -622,18 +689,6 @@ public class DataEvolutionRowIdReassigner {
indexManifestFile.writeWithoutRolling(rewritten),
globalIndexFileCount);
}
- private List<ManifestEntry> readPlanningManifestEntries(
- ManifestFile manifestFile, ManifestFileMeta manifestMeta) {
- return manifestFile.read(
- manifestMeta.fileName(),
- manifestMeta.fileSize(),
- partitionPredicate,
- null,
- Filter.alwaysTrue(),
- entry -> partitionPredicate == null ||
partitionPredicate.test(entry.partition()),
- ManifestEntry::copyWithoutStats);
- }
-
private RecordComparator partitionComparator() {
return CodeGenUtils.newRecordComparator(
table.schema().logicalPartitionType().getFieldTypes());
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/LiveFileRowIdRangeCollector.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/LiveFileRowIdRangeCollector.java
index a9debe2d02..94465ec7c5 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/LiveFileRowIdRangeCollector.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/LiveFileRowIdRangeCollector.java
@@ -18,9 +18,15 @@
package org.apache.paimon.append.dataevolution;
-import org.apache.paimon.utils.LongTripleArrayList;
import org.apache.paimon.utils.PrimitiveRowRanges;
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.PriorityQueue;
+
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkState;
@@ -31,14 +37,25 @@ import static
org.apache.paimon.utils.Preconditions.checkState;
* 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.
+ * <p>The hot collection path retains three primitive words per file. Entries
are synchronously
+ * radix-sorted in fixed-size chunks, compressed, and merged without retaining
manifest objects or
+ * allocating one Java object per range. {@link
#finish(FragmentedPartitionConsumer)} is terminal
+ * and releases this storage.
*/
final class LiveFileRowIdRangeCollector {
+ private static final int ENTRY_WORDS = 3;
+ private static final int ENTRY_CHUNK_SIZE = 1 << 21;
+ private static final int RADIX_BITS = 16;
+ private static final int RADIX_BUCKETS = 1 << RADIX_BITS;
+ private static final long RADIX_MASK = RADIX_BUCKETS - 1L;
private static final long DEDICATED_FILE_FLAG = 1L << 32;
- private final LongTripleArrayList entries;
+ private final int expectedFileCount;
+ private final List<SortedEntryChunk> sortedChunks = new ArrayList<>();
+ private long[] words;
+ private int chunkSize;
+ private int fileCount;
private boolean finished;
LiveFileRowIdRangeCollector() {
@@ -47,7 +64,9 @@ final class LiveFileRowIdRangeCollector {
LiveFileRowIdRangeCollector(int expectedFileCount) {
checkArgument(expectedFileCount >= 0, "Expected live file count cannot
be negative.");
- this.entries = new LongTripleArrayList(expectedFileCount);
+ this.expectedFileCount = expectedFileCount;
+ int initialEntries = Math.max(16, Math.min(expectedFileCount,
ENTRY_CHUNK_SIZE));
+ this.words = new long[Math.multiplyExact(initialEntries, ENTRY_WORDS)];
}
void add(int partitionId, FileRole role, long firstRowId, long rowCount) {
@@ -56,43 +75,34 @@ final class LiveFileRowIdRangeCollector {
checkArgument(role != null, "File role cannot be null.");
checkArgument(rowCount > 0, "Row count must be positive.");
Math.addExact(firstRowId, rowCount - 1L);
- entries.add(
+ if (chunkSize == ENTRY_CHUNK_SIZE) {
+ flushCurrentChunk(false);
+ }
+ ensureCapacity(Math.addExact(chunkSize, 1));
+ int offset = chunkSize * ENTRY_WORDS;
+ words[offset] =
Integer.toUnsignedLong(partitionId)
- | (role == FileRole.DEDICATED ? DEDICATED_FILE_FLAG :
0L),
- firstRowId,
- rowCount);
+ | (role == FileRole.DEDICATED ? DEDICATED_FILE_FLAG :
0L);
+ words[offset + 1] = firstRowId;
+ words[offset + 2] = rowCount;
+ chunkSize++;
+ fileCount = Math.addExact(fileCount, 1);
}
int fileCount() {
- return entries.size();
+ return fileCount;
}
int retainedWordCount() {
- return entries.retainedLongCount();
+ int retained = words.length;
+ for (SortedEntryChunk chunk : sortedChunks) {
+ retained = Math.addExact(retained, chunk.words.length);
+ }
+ return retained;
}
int usedWordCount() {
- return entries.usedLongCount();
- }
-
- private int partitionId(int index) {
- return (int) entries.first(index);
- }
-
- private boolean dedicatedFile(int index) {
- return (entries.first(index) & DEDICATED_FILE_FLAG) != 0;
- }
-
- private long firstRowId(int index) {
- return entries.second(index);
- }
-
- private long rowCount(int index) {
- return entries.third(index);
- }
-
- private long lastRowId(int index) {
- return firstRowId(index) + rowCount(index) - 1L;
+ return Math.multiplyExact(fileCount, ENTRY_WORDS);
}
/**
@@ -106,187 +116,336 @@ final class LiveFileRowIdRangeCollector {
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));
+ List<SortedEntryChunk> chunks = finishSortedChunks();
+ if (chunks.isEmpty()) {
+ return;
+ }
+
+ PartitionRangeAccumulator accumulator = new
PartitionRangeAccumulator(consumer);
+ if (chunks.size() == 1) {
+ SortedEntryChunk chunk = chunks.get(0);
+ for (int index = 0; index < chunk.size; index++) {
+ addToAccumulator(accumulator, chunk, index);
}
- partitionStart = partitionEnd;
+ } else if (chunks.size() == 2) {
+ mergeTwoChunks(accumulator, chunks.get(0), chunks.get(1));
+ } else {
+ mergeChunks(accumulator, chunks);
}
+ accumulator.finish();
} finally {
- entries.release();
+ release();
}
}
- private void sortByPartitionAndRange() {
- if (entries.size() > 1) {
- sort(0, entries.size() - 1);
+ void abort() {
+ if (!finished) {
+ finished = true;
+ release();
}
}
- private void sort(int left, int right) {
- while (left < right) {
- int middle = left + ((right - left) >>> 1);
- long pivotPartition = entries.first(middle);
- long pivotFirst = entries.second(middle);
- long pivotCount = entries.third(middle);
- int lower = left;
- int current = left;
- int upper = right;
- while (current <= upper) {
- int comparison = compare(current, pivotPartition, pivotFirst,
pivotCount);
- if (comparison < 0) {
- swap(lower++, current++);
- } else if (comparison > 0) {
- swap(current, upper--);
- } else {
- current++;
- }
- }
-
- if (lower - left < right - upper) {
- if (left < lower - 1) {
- sort(left, lower - 1);
- }
- left = upper + 1;
- } else {
- if (upper + 1 < right) {
- sort(upper + 1, right);
- }
- right = lower - 1;
- }
+ private static void mergeTwoChunks(
+ PartitionRangeAccumulator accumulator,
+ SortedEntryChunk leftChunk,
+ SortedEntryChunk rightChunk) {
+ EntryChunkCursor left = new EntryChunkCursor(leftChunk);
+ EntryChunkCursor right = new EntryChunkCursor(rightChunk);
+ while (left.index < left.chunk.size && right.index < right.chunk.size)
{
+ EntryChunkCursor next = compareCursors(left, right) <= 0 ? left :
right;
+ addToAccumulator(accumulator, next.chunk, next.index++);
+ }
+ while (left.index < left.chunk.size) {
+ addToAccumulator(accumulator, left.chunk, left.index++);
+ }
+ while (right.index < right.chunk.size) {
+ addToAccumulator(accumulator, right.chunk, right.index++);
}
}
- private int compare(int index, long pivotPartition, long pivotFirst, long
pivotCount) {
- int result =
- Long.compare(entries.first(index) & 0xFFFF_FFFFL,
pivotPartition & 0xFFFF_FFFFL);
- if (result != 0) {
- return result;
+ private static void mergeChunks(
+ PartitionRangeAccumulator accumulator, List<SortedEntryChunk>
chunks) {
+ PriorityQueue<EntryChunkCursor> queue =
+ new
PriorityQueue<>(LiveFileRowIdRangeCollector::compareCursors);
+ for (SortedEntryChunk chunk : chunks) {
+ if (chunk.size > 0) {
+ queue.add(new EntryChunkCursor(chunk));
+ }
}
- long first = entries.second(index);
- result = Long.compare(first, pivotFirst);
- if (result != 0) {
- return result;
+ while (!queue.isEmpty()) {
+ EntryChunkCursor cursor = queue.poll();
+ addToAccumulator(accumulator, cursor.chunk, cursor.index++);
+ if (cursor.index < cursor.chunk.size) {
+ queue.add(cursor);
+ }
}
- long end = first + entries.third(index) - 1L;
- long pivotEnd = pivotFirst + pivotCount - 1L;
- return Long.compare(end, pivotEnd);
}
- private void swap(int left, int right) {
- entries.swap(left, right);
+ private static void addToAccumulator(
+ PartitionRangeAccumulator accumulator, SortedEntryChunk chunk, int
index) {
+ int offset = index * ENTRY_WORDS;
+ long metadata = chunk.words[offset];
+ long start = chunk.words[offset + 1];
+ long rowCount = chunk.words[offset + 2];
+ accumulator.add(
+ (int) metadata,
+ (metadata & DEDICATED_FILE_FLAG) != 0,
+ start,
+ start + rowCount - 1L);
}
- /**
- * Scans logical ranges without retaining one object (or even one
primitive pair) per range.
- *
- * <p>The result records both the number of logical ranges and whether
gaps exist between them.
- */
- 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);
- int rangeCount = 0;
- boolean contiguous = true;
- boolean hasPrevious = false;
- long previousEnd = 0L;
- for (int i = from + 1; i < to; i++) {
- if (firstRowId(i) <= currentEnd) {
- currentEnd = Math.max(currentEnd, lastRowId(i));
+ private List<SortedEntryChunk> finishSortedChunks() {
+ if (chunkSize > 0) {
+ if (sortedChunks.isEmpty()) {
+ int usedWords = chunkSize * ENTRY_WORDS;
+ long[] finalWords =
+ words.length == usedWords ? words :
Arrays.copyOf(words, usedWords);
+ sortedChunks.add(sortAndCompress(finalWords, chunkSize));
+ words = new long[0];
+ chunkSize = 0;
} else {
- computeLogicalRange(overlapStart, i, rangeScratch);
- rangeCount++;
- if (hasPrevious
- && (previousEnd == Long.MAX_VALUE || rangeScratch[0]
!= previousEnd + 1L)) {
- contiguous = false;
- }
- previousEnd = rangeScratch[1];
- hasPrevious = true;
- overlapStart = i;
- currentEnd = lastRowId(i);
+ flushCurrentChunk(true);
}
}
- computeLogicalRange(overlapStart, to, rangeScratch);
- rangeCount++;
- if (hasPrevious && (previousEnd == Long.MAX_VALUE || rangeScratch[0]
!= previousEnd + 1L)) {
- contiguous = false;
- }
- analysis.rangeCount = rangeCount;
- analysis.fragmented = !contiguous;
+ return sortedChunks;
}
- private PrimitiveRowRanges materializeLogicalRanges(
- int from, int to, int expectedRangeCount, long[] rangeScratch) {
- checkArgument(
- from >= 0 && from < to && to <= entries.size() &&
expectedRangeCount > 0,
- "Invalid fragmented entry slice.");
- PrimitiveRowRanges ranges = new PrimitiveRowRanges(expectedRangeCount);
- int overlapStart = from;
- long currentEnd = lastRowId(from);
- for (int i = from + 1; i < to; i++) {
- if (firstRowId(i) <= currentEnd) {
- currentEnd = Math.max(currentEnd, lastRowId(i));
- } else {
- computeLogicalRange(overlapStart, i, rangeScratch);
- ranges.add(rangeScratch[0], rangeScratch[1]);
- overlapStart = i;
- currentEnd = lastRowId(i);
+ private void flushCurrentChunk(boolean finalChunk) {
+ if (chunkSize == 0) {
+ if (finalChunk) {
+ words = new long[0];
}
+ return;
}
- computeLogicalRange(overlapStart, to, rangeScratch);
- ranges.add(rangeScratch[0], rangeScratch[1]);
+ int usedWords = chunkSize * ENTRY_WORDS;
+ long[] chunkWords = words.length == usedWords ? words :
Arrays.copyOf(words, usedWords);
+ int entries = chunkSize;
+ words =
+ finalChunk
+ ? new long[0]
+ : new long[Math.multiplyExact(nextChunkCapacity(),
ENTRY_WORDS)];
+ chunkSize = 0;
+ sortedChunks.add(sortAndCompress(chunkWords, entries));
+ }
+
+ private int nextChunkCapacity() {
+ if (expectedFileCount <= fileCount) {
+ return ENTRY_CHUNK_SIZE;
+ }
+ int remainingHint = Math.max(0, expectedFileCount - fileCount);
+ return Math.max(16, Math.min(remainingHint, ENTRY_CHUNK_SIZE));
+ }
+
+ private void ensureCapacity(int requiredEntries) {
checkState(
- ranges.size() == expectedRangeCount,
- "Logical range count changed between scan and
materialization.");
- return ranges;
+ requiredEntries <= ENTRY_CHUNK_SIZE,
+ "Live file row-id range chunk exceeds its fixed capacity.");
+ long requiredWords = (long) requiredEntries * ENTRY_WORDS;
+ if (requiredWords <= words.length) {
+ return;
+ }
+ int maximumWords = ENTRY_CHUNK_SIZE * ENTRY_WORDS;
+ int newLength = Math.max(16 * ENTRY_WORDS, words.length);
+ while (newLength < requiredWords) {
+ int grown = newLength + (newLength >>> 1);
+ if (grown <= newLength || grown > maximumWords) {
+ newLength = maximumWords;
+ break;
+ }
+ newLength = grown;
+ }
+ words = Arrays.copyOf(words, newLength);
}
- private void computeLogicalRange(int from, int to, long[] result) {
- 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++) {
- long start = firstRowId(i);
- long end = lastRowId(i);
- spanningStart = Math.min(spanningStart, start);
- spanningEnd = Math.max(spanningEnd, end);
- if (!dedicatedFile(i)) {
- checkState(
- !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;
+ private static SortedEntryChunk sortAndCompress(long[] words, int size) {
+ if (size > 1) {
+ radixSort(words, size);
+ }
+ if (size == 0) {
+ return new SortedEntryChunk(new long[0], 0);
+ }
+
+ int outputSize = 0;
+ int currentPartition = partitionId(words, 0);
+ long spanningStart = firstRowId(words, 0);
+ long spanningEnd = lastRowId(words, 0);
+ boolean hasNormalFile = !dedicatedFile(words, 0);
+ long normalStart = spanningStart;
+ long normalEnd = spanningEnd;
+ for (int i = 1; i < size; i++) {
+ int partitionId = partitionId(words, i);
+ long start = firstRowId(words, i);
+ long end = lastRowId(words, i);
+ if (partitionId == currentPartition && start <= spanningEnd) {
+ spanningEnd = Math.max(spanningEnd, end);
+ if (!dedicatedFile(words, i)) {
+ checkState(
+ !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;
+ }
+ continue;
}
+ outputSize =
+ writeLogicalComponent(
+ words,
+ outputSize,
+ currentPartition,
+ spanningStart,
+ spanningEnd,
+ hasNormalFile,
+ normalStart,
+ normalEnd);
+ currentPartition = partitionId;
+ spanningStart = start;
+ spanningEnd = end;
+ hasNormalFile = !dedicatedFile(words, i);
+ normalStart = start;
+ normalEnd = end;
}
+ outputSize =
+ writeLogicalComponent(
+ words,
+ outputSize,
+ currentPartition,
+ spanningStart,
+ spanningEnd,
+ hasNormalFile,
+ normalStart,
+ normalEnd);
+ int outputWords = outputSize * ENTRY_WORDS;
+ return new SortedEntryChunk(
+ outputWords == words.length ? words : Arrays.copyOf(words,
outputWords),
+ outputSize);
+ }
+
+ private static int writeLogicalComponent(
+ long[] words,
+ int outputIndex,
+ int partitionId,
+ long spanningStart,
+ long spanningEnd,
+ boolean hasNormalFile,
+ long normalStart,
+ long normalEnd) {
long logicalStart = hasNormalFile ? normalStart : spanningStart;
long logicalEnd = hasNormalFile ? normalEnd : spanningEnd;
- for (int i = from; i < to; i++) {
- checkState(
- firstRowId(i) >= logicalStart && lastRowId(i) <=
logicalEnd,
- "File row-id range is outside its logical row-id range.");
+ checkState(
+ spanningStart >= logicalStart && spanningEnd <= logicalEnd,
+ "File row-id range is outside its logical row-id range.");
+ int outputOffset = outputIndex * ENTRY_WORDS;
+ words[outputOffset] =
+ Integer.toUnsignedLong(partitionId) | (hasNormalFile ? 0L :
DEDICATED_FILE_FLAG);
+ words[outputOffset + 1] = logicalStart;
+ words[outputOffset + 2] = inclusiveRangeCount(logicalStart,
logicalEnd);
+ return outputIndex + 1;
+ }
+
+ /**
+ * Stable LSD radix sort by unsigned partition id, signed first row id,
and signed last row id.
+ */
+ private static void radixSort(long[] words, int size) {
+ long[] auxiliary = new long[Math.multiplyExact(size, ENTRY_WORDS)];
+ int[] counts = new int[RADIX_BUCKETS];
+ long[] source = words;
+ long[] target = auxiliary;
+
+ // Four 16-bit passes for last row id, four for first row id, then two
for partition id.
+ for (int pass = 0; pass < 10; pass++) {
+ Arrays.fill(counts, 0);
+ for (int index = 0; index < size; index++) {
+ counts[radixBucket(source, index, pass)]++;
+ }
+
+ int position = 0;
+ for (int bucket = 0; bucket < RADIX_BUCKETS; bucket++) {
+ int bucketSize = counts[bucket];
+ counts[bucket] = position;
+ position += bucketSize;
+ }
+
+ for (int index = 0; index < size; index++) {
+ int sourceOffset = index * ENTRY_WORDS;
+ int targetOffset = counts[radixBucket(source, index, pass)]++
* ENTRY_WORDS;
+ target[targetOffset] = source[sourceOffset];
+ target[targetOffset + 1] = source[sourceOffset + 1];
+ target[targetOffset + 2] = source[sourceOffset + 2];
+ }
+
+ long[] swap = source;
+ source = target;
+ target = swap;
+ }
+
+ checkState(source == words, "Radix sort must finish in its input
buffer.");
+ }
+
+ private static int radixBucket(long[] words, int index, int pass) {
+ int offset = index * ENTRY_WORDS;
+ long value;
+ int shift;
+ if (pass < 4) {
+ value = (words[offset + 1] + words[offset + 2] - 1L) ^
Long.MIN_VALUE;
+ shift = pass * RADIX_BITS;
+ } else if (pass < 8) {
+ value = words[offset + 1] ^ Long.MIN_VALUE;
+ shift = (pass - 4) * RADIX_BITS;
+ } else {
+ value = words[offset] & 0xFFFF_FFFFL;
+ shift = (pass - 8) * RADIX_BITS;
+ }
+ return (int) ((value >>> shift) & RADIX_MASK);
+ }
+
+ private static int compareCursors(EntryChunkCursor left, EntryChunkCursor
right) {
+ int leftOffset = left.index * ENTRY_WORDS;
+ int rightOffset = right.index * ENTRY_WORDS;
+ long[] leftWords = left.chunk.words;
+ long[] rightWords = right.chunk.words;
+ int result =
+ Long.compare(
+ leftWords[leftOffset] & 0xFFFF_FFFFL,
+ rightWords[rightOffset] & 0xFFFF_FFFFL);
+ if (result != 0) {
+ return result;
+ }
+ result = Long.compare(leftWords[leftOffset + 1],
rightWords[rightOffset + 1]);
+ if (result != 0) {
+ return result;
}
- result[0] = logicalStart;
- result[1] = logicalEnd;
+ long leftEnd = leftWords[leftOffset + 1] + leftWords[leftOffset + 2] -
1L;
+ long rightEnd = rightWords[rightOffset + 1] + rightWords[rightOffset +
2] - 1L;
+ return Long.compare(leftEnd, rightEnd);
+ }
+
+ private static int partitionId(long[] words, int index) {
+ return (int) words[index * ENTRY_WORDS];
+ }
+
+ private static boolean dedicatedFile(long[] words, int index) {
+ return (words[index * ENTRY_WORDS] & DEDICATED_FILE_FLAG) != 0;
+ }
+
+ private static long firstRowId(long[] words, int index) {
+ return words[index * ENTRY_WORDS + 1];
+ }
+
+ private static long lastRowId(long[] words, int index) {
+ int offset = index * ENTRY_WORDS;
+ return words[offset + 1] + words[offset + 2] - 1L;
+ }
+
+ private static long inclusiveRangeCount(long start, long end) {
+ return Math.addExact(Math.subtractExact(end, start), 1L);
+ }
+
+ private void release() {
+ words = new long[0];
+ chunkSize = 0;
+ fileCount = 0;
+ sortedChunks.clear();
}
enum FileRole {
@@ -300,9 +459,121 @@ final class LiveFileRowIdRangeCollector {
void accept(int partitionId, PrimitiveRowRanges logicalRanges);
}
- private static final class LogicalRangeAnalysis {
+ private static final class SortedEntryChunk {
+
+ private final long[] words;
+ private final int size;
- private int rangeCount;
+ private SortedEntryChunk(long[] words, int size) {
+ this.words = words;
+ this.size = size;
+ }
+ }
+
+ private static final class EntryChunkCursor {
+
+ private final SortedEntryChunk chunk;
+ private int index;
+
+ private EntryChunkCursor(SortedEntryChunk chunk) {
+ this.chunk = chunk;
+ }
+ }
+
+ private static final class PartitionRangeAccumulator {
+
+ private final FragmentedPartitionConsumer consumer;
+ private int partitionId = -1;
+ private @Nullable PrimitiveRowRanges ranges;
private boolean fragmented;
+ private boolean hasPreviousRange;
+ private long previousRangeEnd;
+ private boolean hasComponent;
+ private long spanningStart;
+ private long spanningEnd;
+ private boolean hasNormalFile;
+ private long normalStart;
+ private long normalEnd;
+
+ private PartitionRangeAccumulator(FragmentedPartitionConsumer
consumer) {
+ this.consumer = consumer;
+ }
+
+ private void add(int incomingPartitionId, boolean dedicated, long
start, long end) {
+ if (!hasComponent) {
+ startPartition(incomingPartitionId);
+ startComponent(dedicated, start, end);
+ return;
+ }
+ if (incomingPartitionId == partitionId && start <= spanningEnd) {
+ spanningEnd = Math.max(spanningEnd, end);
+ if (!dedicated) {
+ checkState(
+ !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;
+ }
+ return;
+ }
+
+ finishComponent();
+ if (incomingPartitionId != partitionId) {
+ finishPartition();
+ startPartition(incomingPartitionId);
+ }
+ startComponent(dedicated, start, end);
+ }
+
+ private void startPartition(int incomingPartitionId) {
+ partitionId = incomingPartitionId;
+ ranges = new PrimitiveRowRanges(16);
+ fragmented = false;
+ hasPreviousRange = false;
+ }
+
+ private void startComponent(boolean dedicated, long start, long end) {
+ spanningStart = start;
+ spanningEnd = end;
+ hasNormalFile = !dedicated;
+ normalStart = start;
+ normalEnd = end;
+ hasComponent = true;
+ }
+
+ private void finishComponent() {
+ long logicalStart = hasNormalFile ? normalStart : spanningStart;
+ long logicalEnd = hasNormalFile ? normalEnd : spanningEnd;
+ checkState(
+ spanningStart >= logicalStart && spanningEnd <= logicalEnd,
+ "File row-id range is outside its logical row-id range.");
+ checkState(ranges != null, "Missing logical range buffer.");
+ if (hasPreviousRange
+ && (previousRangeEnd == Long.MAX_VALUE
+ || logicalStart != previousRangeEnd + 1L)) {
+ fragmented = true;
+ }
+ ranges.add(logicalStart, logicalEnd);
+ previousRangeEnd = logicalEnd;
+ hasPreviousRange = true;
+ hasComponent = false;
+ }
+
+ private void finishPartition() {
+ checkState(partitionId >= 0 && ranges != null, "Missing partition
state.");
+ if (fragmented) {
+ consumer.accept(partitionId, ranges);
+ }
+ ranges = null;
+ }
+
+ private void finish() {
+ if (!hasComponent) {
+ return;
+ }
+ finishComponent();
+ finishPartition();
+ }
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryManifestEntry.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryManifestEntry.java
index 217d7846a2..c66f165c96 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryManifestEntry.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryManifestEntry.java
@@ -49,6 +49,7 @@ public final class BinaryManifestEntry implements
ManifestEntry {
private static final Projection FULL_PROJECTION =
Projection.create(ManifestEntry.MANIFEST_ROW_TYPE);
public static final Projection DELETE_ENTRY_PROJECTION =
createDeleteEntryProjection();
+ public static final Projection ROW_RANGE_PROJECTION =
createRowRangeProjection();
private final Projection projection;
private final @Nullable BinaryDataFileMeta file;
@@ -115,6 +116,22 @@ public final class BinaryManifestEntry implements
ManifestEntry {
DataFileMeta.EXTERNAL_PATH)))));
}
+ private static Projection createRowRangeProjection() {
+ RowType manifestType = ManifestEntry.MANIFEST_ROW_TYPE;
+ return Projection.create(
+ new RowType(
+ false,
+ Arrays.asList(
+ manifestType.getField(ManifestEntry.KIND),
+ manifestType.getField(ManifestEntry.PARTITION),
+ manifestType
+ .getField(ManifestEntry.FILE)
+ .newType(
+ DataFileMeta.SCHEMA.project(
+ DataFileMeta.ROW_COUNT,
+
DataFileMeta.FIRST_ROW_ID)))));
+ }
+
/** Drops references to the current row before its reader batch is
released. */
public void clear() {
row = null;
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 b4a5c2396c..5e5b2ad923 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
@@ -459,6 +459,71 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
assertThat(plan.totalOffset).isEqualTo(8L);
}
+ @Test
+ public void testDeleteEntryAtEndDoesNotAffectEqualPartitionOrdering()
throws Exception {
+ Schema.Builder schemaBuilder = Schema.newBuilder();
+ schemaBuilder.column("float_pt", DataTypes.FLOAT());
+ schemaBuilder.column("id", DataTypes.INT());
+ schemaBuilder.partitionKeys("float_pt");
+ schemaBuilder.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ schemaBuilder.option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ catalog.createTable(identifier(), schemaBuilder.build(), true);
+ FileStoreTable table = getTableDefault();
+
+ InternalRowSerializer partitionSerializer =
+ new
InternalRowSerializer(table.schema().logicalPartitionType());
+ BinaryRow laterPartition =
+ partitionSerializer
+
.toBinaryRow(GenericRow.of(Float.intBitsToFloat(0x7fc00001)))
+ .copy();
+ BinaryRow earlierPartition =
+ partitionSerializer
+
.toBinaryRow(GenericRow.of(Float.intBitsToFloat(0x7fc00002)))
+ .copy();
+
assertThat(laterPartition.toBytes()).isNotEqualTo(earlierPartition.toBytes());
+
+ DataFileMeta deletedFile =
+ planningDataFile("deleted-0.parquet", 0L, 1,
Collections.emptyList(), null, null);
+ List<ManifestEntry> entries =
+ Arrays.asList(
+ ManifestEntry.create(FileKind.ADD, laterPartition, 0,
1, deletedFile),
+ planningManifestEntry(earlierPartition,
"earlier-live-10.parquet", 10L, 1L),
+ planningManifestEntry(earlierPartition,
"earlier-live-12.parquet", 12L, 1L),
+ planningManifestEntry(laterPartition,
"later-live-100.parquet", 100L, 1L),
+ planningManifestEntry(laterPartition,
"later-live-102.parquet", 102L, 1L),
+ // Keep DELETE after its ADD, matching manifest
compaction ordering.
+ ManifestEntry.create(FileKind.DELETE, laterPartition,
0, 1, deletedFile));
+ ManifestFile manifestFile =
table.store().manifestFileFactory().create();
+ List<ManifestFileMeta> manifestMetas = manifestFile.write(entries);
+ assertThat(manifestMetas).hasSize(1);
+ assertThat(
+ manifestFile.read(
+ manifestMetas.get(0).fileName(),
manifestMetas.get(0).fileSize()))
+ .extracting(ManifestEntry::kind)
+ .containsExactly(
+ FileKind.ADD,
+ FileKind.ADD,
+ FileKind.ADD,
+ FileKind.ADD,
+ FileKind.ADD,
+ FileKind.DELETE);
+
+ Optional<AssignmentPlanView> compactPlan = compactPlanView(table,
manifestMetas, null);
+ Optional<AssignmentPlanView> legacyPlan = legacyPlanView(table,
manifestMetas, null);
+ assertPlanViewsEqual(compactPlan, legacyPlan);
+
+ assertThat(compactPlan).isPresent();
+ AssignmentPlanView plan = compactPlan.get();
+ assertThat(plan.partitionMappings).hasSize(2);
+ PartitionMappingView earlierMapping = plan.partitionMappings.get(0);
+
assertThat(earlierMapping.partitionBytes).containsExactly(earlierPartition.toBytes());
+ assertThat(earlierMapping.oldStarts).containsExactly(10L, 12L);
+ PartitionMappingView laterMapping = plan.partitionMappings.get(1);
+
assertThat(laterMapping.partitionBytes).containsExactly(laterPartition.toBytes());
+ assertThat(laterMapping.oldStarts).containsExactly(100L, 102L);
+ assertThat(plan.totalOffset).isEqualTo(4L);
+ }
+
@Test
public void testCompactAndLegacyPlansMatchForComplexDeleteIdentifier()
throws Exception {
createTableDefault();
@@ -768,8 +833,8 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
}
@Test
- public void testReassignAbortsWhenConcurrentAppendMergesPlannedManifests()
throws Exception {
- FileStoreTable originalTable = createTableWithInterleavedPartitions();
+ public void
testReassignRetriesAfterConcurrentAppendMergesPlannedManifests() throws
Exception {
+ FileStoreTable originalTable =
createTableWithPartiallyOverlappedPartitions();
List<String> plannedManifestFiles =
dataManifestFileNames(originalTable);
assertThat(plannedManifestFiles).hasSizeGreaterThan(1);
@@ -777,43 +842,41 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
Snapshot before = table.snapshotManager().latestSnapshot();
AtomicBoolean merged = new AtomicBoolean();
- assertThatThrownBy(
- () ->
- new DataEvolutionRowIdReassigner(
- table,
- null,
- () -> {
- if
(merged.compareAndSet(false, true)) {
- try {
- writeOneRow(table,
"c", 100);
- Snapshot
appendSnapshot =
-
table.snapshotManager()
-
.latestSnapshot();
-
assertThat(appendSnapshot.commitKind())
- .isEqualTo(
-
Snapshot.CommitKind
-
.APPEND);
-
assertThat(dataManifestFileNames(table))
-
.doesNotContainAnyElementsOf(
-
plannedManifestFiles);
- } catch (Exception e) {
- throw new
RuntimeException(e);
- }
- }
- })
-
.reassign("test-reassign-append-manifest-merge"))
- .isInstanceOf(RuntimeException.class)
- .hasMessageContaining("planned manifest")
- .hasMessageContaining("no longer exists");
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+ if (merged.compareAndSet(false, true)) {
+ try {
+ writeOneRow(table, "d", 100);
+ Snapshot appendSnapshot =
+
table.snapshotManager().latestSnapshot();
+
assertThat(appendSnapshot.commitKind())
+
.isEqualTo(Snapshot.CommitKind.APPEND);
+
assertThat(dataManifestFileNames(table))
+
.doesNotContainAnyElementsOf(
+
plannedManifestFiles);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ })
+ .reassign("test-reassign-append-manifest-merge");
assertThat(merged).isTrue();
-
assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(before.id()
+ 1);
- assertThat(rowIdsByPartition(table))
- .containsEntry("pt=a/", Arrays.asList(0L, 2L, 4L))
- .containsEntry("pt=b/", Arrays.asList(1L, 3L))
- .containsEntry("pt=c/", Collections.singletonList(5L));
- assertThat(readTableRows(table))
- .containsExactly("0|a|v0", "100|c|v100", "1|b|v1", "2|a|v2",
"3|b|v3", "4|a|v4");
+ assertThat(result.reassigned).isTrue();
+ assertThat(result.previousSnapshotId).isEqualTo(before.id() + 1);
+ assertThat(result.newSnapshotId).isEqualTo(before.id() + 2);
+ assertThat(result.firstAssignedRowId).isEqualTo(6L);
+ assertThat(result.nextRowId).isEqualTo(8L);
+ assertThat(result.fileCount).isEqualTo(2L);
+ assertThat(result.rowCount).isEqualTo(2L);
+ assertThat(expandedRowIdsByPartition(table))
+ .containsEntry("pt=a/", Arrays.asList(6L, 7L))
+ .containsEntry("pt=b/", Collections.singletonList(3L))
+ .containsEntry("pt=c/", Arrays.asList(0L, 1L))
+ .containsEntry("pt=d/", Collections.singletonList(5L));
}
@Test
@@ -1297,6 +1360,199 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
.containsEntry("pt=b/", Arrays.asList(3L, 4L));
}
+ @Test
+ public void testExcludeLargeContiguousRunFromReassignment() throws
Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeOneRow(table, "seed", -1);
+ InternalRowSerializer partitionSerializer =
+ new
InternalRowSerializer(table.schema().logicalPartitionType());
+ replaceWithManifestEntries(
+ table,
+ Arrays.asList(
+ manifestEntry(partitionSerializer, "a",
"a-long-1.parquet", 0, 5),
+ manifestEntry(partitionSerializer, "a",
"a-long-2.parquet", 6, 11),
+ manifestEntry(partitionSerializer, "a",
"a-short-1.parquet", 20, 21),
+ manifestEntry(partitionSerializer, "a",
"a-short-2.parquet", 30, 31),
+ manifestEntry(partitionSerializer, "b", "b.parquet",
40, 41)),
+ 42L);
+
+ FileStoreTable configured =
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT.key(),
+ "10"));
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(configured)
+ .reassign("test-exclude-large-contiguous-run");
+
+ assertThat(result.reassigned).isTrue();
+ assertThat(result.firstAssignedRowId).isEqualTo(42L);
+ assertThat(result.nextRowId).isEqualTo(46L);
+ assertThat(result.fileCount).isEqualTo(2L);
+ assertThat(result.rowCount).isEqualTo(4L);
+ assertThat(expandedRowIdsByPartition(configured))
+ .containsEntry(
+ "pt=a/",
+ Arrays.asList(
+ 0L, 1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L,
11L, 42L, 43L, 44L,
+ 45L))
+ .containsEntry("pt=b/", Arrays.asList(40L, 41L));
+
+ DataEvolutionRowIdReassigner.Result secondResult =
+ new DataEvolutionRowIdReassigner(configured)
+ .reassign("test-exclude-large-contiguous-run-again");
+ assertThat(secondResult.reassigned).isFalse();
+ assertThat(secondResult.skipReason).isEqualTo("no partition requires
row-id reassignment");
+ assertThat(secondResult.firstAssignedRowId).isEqualTo(46L);
+ assertThat(secondResult.nextRowId).isEqualTo(46L);
+ }
+
+ @Test
+ public void testContiguousRunAtThresholdStillReassigned() throws Exception
{
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeOneRow(table, "seed", -1);
+ InternalRowSerializer partitionSerializer =
+ new
InternalRowSerializer(table.schema().logicalPartitionType());
+ replaceWithManifestEntries(
+ table,
+ Arrays.asList(
+ manifestEntry(partitionSerializer, "a",
"a-boundary-1.parquet", 0, 4),
+ manifestEntry(partitionSerializer, "a",
"a-boundary-2.parquet", 5, 9),
+ manifestEntry(partitionSerializer, "a",
"a-short.parquet", 20, 21)),
+ 22L);
+
+ FileStoreTable configured =
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT.key(),
+ "10"));
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(configured)
+ .reassign("test-retain-threshold-contiguous-run");
+
+ assertThat(result.reassigned).isTrue();
+ assertThat(result.firstAssignedRowId).isEqualTo(22L);
+ assertThat(result.nextRowId).isEqualTo(34L);
+ assertThat(result.fileCount).isEqualTo(3L);
+ assertThat(result.rowCount).isEqualTo(12L);
+ assertThat(expandedRowIdsByPartition(configured).get("pt=a/"))
+ .containsExactly(22L, 23L, 24L, 25L, 26L, 27L, 28L, 29L, 30L,
31L, 32L, 33L);
+ }
+
+ @Test
+ public void testZeroThresholdDisablesLargeContiguousRunFiltering() throws
Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeOneRow(table, "seed", -1);
+ InternalRowSerializer partitionSerializer =
+ new
InternalRowSerializer(table.schema().logicalPartitionType());
+ replaceWithManifestEntries(
+ table,
+ Arrays.asList(
+ manifestEntry(partitionSerializer, "a",
"a-long-1.parquet", 0, 5),
+ manifestEntry(partitionSerializer, "a",
"a-long-2.parquet", 6, 11),
+ manifestEntry(partitionSerializer, "a",
"a-short-1.parquet", 20, 21),
+ manifestEntry(partitionSerializer, "a",
"a-short-2.parquet", 30, 31),
+ manifestEntry(partitionSerializer, "b", "b.parquet",
40, 41)),
+ 42L);
+
+ FileStoreTable configured =
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT.key(),
+ "0"));
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(configured)
+ .reassign("test-disable-contiguous-run-filtering");
+
+ assertThat(result.reassigned).isTrue();
+ assertThat(result.firstAssignedRowId).isEqualTo(42L);
+ assertThat(result.nextRowId).isEqualTo(58L);
+ assertThat(result.fileCount).isEqualTo(4L);
+ assertThat(result.rowCount).isEqualTo(16L);
+ assertThat(expandedRowIdsByPartition(configured))
+ .containsEntry(
+ "pt=a/",
+ Arrays.asList(
+ 42L, 43L, 44L, 45L, 46L, 47L, 48L, 49L, 50L,
51L, 52L, 53L, 54L,
+ 55L, 56L, 57L))
+ .containsEntry("pt=b/", Arrays.asList(40L, 41L));
+ }
+
+ @Test
+ public void testAllLargeContiguousRunsCanBeExcluded() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeOneRow(table, "seed", -1);
+ InternalRowSerializer partitionSerializer =
+ new
InternalRowSerializer(table.schema().logicalPartitionType());
+ replaceWithManifestEntries(
+ table,
+ Arrays.asList(
+ manifestEntry(partitionSerializer, "a",
"a-first-1.parquet", 0, 5),
+ manifestEntry(partitionSerializer, "a",
"a-first-2.parquet", 6, 11),
+ manifestEntry(partitionSerializer, "a",
"a-second-1.parquet", 20, 25),
+ manifestEntry(partitionSerializer, "a",
"a-second-2.parquet", 26, 31)),
+ 32L);
+
+ FileStoreTable configured =
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT.key(),
+ "10"));
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(configured)
+ .reassign("test-exclude-all-large-contiguous-runs");
+
+ assertThat(result.reassigned).isFalse();
+ assertThat(result.skipReason).isEqualTo("no partition requires row-id
reassignment");
+ assertThat(expandedRowIdsByPartition(configured).get("pt=a/"))
+ .containsExactly(
+ 0L, 1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L, 11L, 20L,
21L, 22L, 23L, 24L,
+ 25L, 26L, 27L, 28L, 29L, 30L, 31L);
+ }
+
+ @Test
+ public void testSingleRetainedRunIsNotReassigned() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeOneRow(table, "seed", -1);
+ InternalRowSerializer partitionSerializer =
+ new
InternalRowSerializer(table.schema().logicalPartitionType());
+ replaceWithManifestEntries(
+ table,
+ Arrays.asList(
+ manifestEntry(partitionSerializer, "a",
"a-large-1.parquet", 0, 7),
+ manifestEntry(partitionSerializer, "a",
"a-large-2.parquet", 8, 15),
+ manifestEntry(partitionSerializer, "a",
"a-short.parquet", 30, 31),
+ manifestEntry(partitionSerializer, "b",
"b-adjacent.parquet", 16, 17)),
+ 32L);
+
+ FileStoreTable configured =
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT.key(),
+ "10"));
+ DataEvolutionRowIdReassigner.Result result =
+ new
DataEvolutionRowIdReassigner(configured).reassign("test-single-retained-run");
+
+ assertThat(result.reassigned).isFalse();
+ assertThat(result.skipReason).isEqualTo("no partition requires row-id
reassignment");
+ assertThat(result.firstAssignedRowId).isEqualTo(32L);
+ assertThat(result.nextRowId).isEqualTo(32L);
+ assertThat(result.fileCount).isEqualTo(0L);
+ assertThat(result.rowCount).isEqualTo(0L);
+ assertThat(expandedRowIdsByPartition(configured))
+ .containsEntry(
+ "pt=a/",
+ Arrays.asList(
+ 0L, 1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L,
11L, 12L, 13L, 14L,
+ 15L, 30L, 31L))
+ .containsEntry("pt=b/", Arrays.asList(16L, 17L));
+ }
+
@Test
public void testSkipUnpartitionedTable() throws Exception {
Identifier identifier = Identifier.create(database,
"unpartitioned_table");
@@ -1837,6 +2093,47 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
return ManifestEntry.create(FileKind.ADD, partition, 0, 1, file);
}
+ private ManifestEntry manifestEntry(
+ InternalRowSerializer partitionSerializer,
+ String partitionValue,
+ String fileName,
+ long firstRowId,
+ long lastRowId) {
+ BinaryRow partition =
+ partitionSerializer
+
.toBinaryRow(GenericRow.of(BinaryString.fromString(partitionValue)))
+ .copy();
+ return planningManifestEntry(partition, fileName, firstRowId,
lastRowId - firstRowId + 1L);
+ }
+
+ private void replaceWithManifestEntries(
+ FileStoreTable table, List<ManifestEntry> entries, long nextRowId)
throws Exception {
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ ManifestFile manifestFile =
table.store().manifestFileFactory().create();
+ ManifestList manifestList =
table.store().manifestListFactory().create();
+ List<ManifestFileMeta> manifestMetas = new ArrayList<>();
+ long totalRecordCount = 0L;
+ for (ManifestEntry entry : entries) {
+
manifestMetas.addAll(manifestFile.write(Collections.singletonList(entry)));
+ totalRecordCount += entry.file().rowCount();
+ }
+ Pair<String, Long> baseManifestList =
manifestList.write(manifestMetas);
+ Pair<String, Long> deltaManifestList =
manifestList.write(Collections.emptyList());
+ try (FileStoreCommitImpl commit =
+ (FileStoreCommitImpl)
+ table.store().newCommit("test-contiguous-run-source",
table)) {
+ assertThat(
+ commit.replaceManifestList(
+ latest,
+ totalRecordCount,
+ baseManifestList,
+ deltaManifestList,
+ latest.indexManifest(),
+ nextRowId))
+ .isTrue();
+ }
+ }
+
private DataFileMeta planningDataFile(
String fileName,
long firstRowId,
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
index 7a05836d25..fbda56b8c8 100644
---
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
@@ -33,6 +33,8 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link LiveFileRowIdRangeCollector}. */
class LiveFileRowIdRangeCollectorTest {
+ private static final int FIXED_CHUNK_SIZE = 1 << 21;
+
@Test
void testHotStateUsesThreePrimitiveWords() {
LiveFileRowIdRangeCollector ranges = new
LiveFileRowIdRangeCollector(10_000);
@@ -60,6 +62,24 @@ class LiveFileRowIdRangeCollectorTest {
.hasMessageContaining("already finished");
}
+ @Test
+ void testAbortReleasesStorageAndIsIdempotent() {
+ LiveFileRowIdRangeCollector ranges = new
LiveFileRowIdRangeCollector(10_000);
+ ranges.add(0, NORMAL, 0L, 1L);
+
+ ranges.abort();
+ ranges.abort();
+
+ assertThat(ranges.usedWordCount()).isZero();
+ assertThat(ranges.retainedWordCount()).isZero();
+ assertThatThrownBy(() -> ranges.add(0, NORMAL, 1L, 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);
@@ -86,7 +106,7 @@ class LiveFileRowIdRangeCollectorTest {
assertThat(selections).containsOnlyKeys(7);
assertThat(selected.size()).isEqualTo(10_000);
- assertThat(selected.retainedWordCount()).isEqualTo(20_000);
+ assertThat(selected.retainedWordCount()).isBetween(20_000, 30_000);
assertThat(selected.start(0)).isZero();
assertThat(selected.end(0)).isZero();
assertThat(selected.start(9_999)).isEqualTo(19_998L);
@@ -95,6 +115,40 @@ class LiveFileRowIdRangeCollectorTest {
assertThat(ranges.retainedWordCount()).isZero();
}
+ @Test
+ void testRadixSortOrdersPartitionsAndSignedRanges() {
+ LiveFileRowIdRangeCollector ranges = new LiveFileRowIdRangeCollector();
+ ranges.add(9, NORMAL, 10L, 2L);
+ ranges.add(1, NORMAL, -4L, 2L);
+ ranges.add(9, NORMAL, 0L, 2L);
+ ranges.add(1, NORMAL, -10L, 2L);
+
+ Map<Integer, PrimitiveRowRanges> selections = finish(ranges);
+
+ assertThat(selections.keySet()).containsExactly(1, 9);
+ assertRanges(selections.get(1), -10L, -9L, -4L, -3L);
+ assertRanges(selections.get(9), 0L, 1L, 10L, 11L);
+ }
+
+ @Test
+ void testFixedChunksAreCompressedAndMergedWithKWayMerge() {
+ int repeatedFiles = Math.multiplyExact(FIXED_CHUNK_SIZE, 2);
+ LiveFileRowIdRangeCollector ranges =
+ new LiveFileRowIdRangeCollector(Math.addExact(repeatedFiles,
2));
+ for (int i = 0; i < repeatedFiles; i++) {
+ ranges.add(4, NORMAL, 100L, 10L);
+ }
+ ranges.add(4, NORMAL, 300L, 10L);
+ ranges.add(4, NORMAL, 200L, 10L);
+
+ Map<Integer, PrimitiveRowRanges> selections = finish(ranges);
+
+ assertThat(selections).containsOnlyKeys(4);
+ assertRanges(selections.get(4), 100L, 109L, 200L, 209L, 300L, 309L);
+ assertThat(ranges.usedWordCount()).isZero();
+ assertThat(ranges.retainedWordCount()).isZero();
+ }
+
@Test
void testNormalFilesDefineLogicalRanges() {
LiveFileRowIdRangeCollector ranges = new LiveFileRowIdRangeCollector();
@@ -145,6 +199,24 @@ class LiveFileRowIdRangeCollectorTest {
assertThat(ranges.retainedWordCount()).isZero();
}
+ @Test
+ void testConsumerFailureReleasesStorage() {
+ LiveFileRowIdRangeCollector ranges = new LiveFileRowIdRangeCollector();
+ ranges.add(0, NORMAL, 0L, 1L);
+ ranges.add(0, NORMAL, 2L, 1L);
+
+ assertThatThrownBy(
+ () ->
+ ranges.finish(
+ (partitionId, logicalRanges) -> {
+ throw new
RuntimeException("consumer failure");
+ }))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("consumer failure");
+ assertThat(ranges.usedWordCount()).isZero();
+ assertThat(ranges.retainedWordCount()).isZero();
+ }
+
private static Map<Integer, PrimitiveRowRanges>
finish(LiveFileRowIdRangeCollector collector) {
Map<Integer, PrimitiveRowRanges> selections = new LinkedHashMap<>();
collector.finish(
