diff --git a/.github/workflows/utitcase-iceberg-ga.yml b/.github/workflows/utitcase-iceberg-ga.yml new file mode 100644 index 000000000000..deca2434fc89 --- /dev/null +++ b/.github/workflows/utitcase-iceberg-ga.yml @@ -0,0 +1,66 @@ +################################################################################ +# 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. +################################################################################ + +name: UTCase Iceberg GA row lineage on JDK 17 + +on: + push: + paths: + - 'paimon-iceberg/**' + - 'paimon-core/**' + - 'paimon-common/**' + - 'paimon-api/**' + - 'paimon-format/**' + - 'pom.xml' + - '.github/workflows/utitcase-iceberg-ga.yml' + pull_request: + paths: + - 'paimon-iceberg/**' + - 'paimon-core/**' + - 'paimon-common/**' + - 'paimon-api/**' + - 'paimon-format/**' + - 'pom.xml' + - '.github/workflows/utitcase-iceberg-ga.yml' + +env: + JDK_VERSION: 17 + MAVEN_OPTS: -Dmaven.wagon.httpconnectionManager.ttlSeconds=30 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true + +concurrency: + group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.event.number || github.run_id }} + cancel-in-progress: true + +jobs: + build: + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v6 + - name: Set up JDK ${{ env.JDK_VERSION }} + uses: actions/setup-java@v5 + with: + java-version: ${{ env.JDK_VERSION }} + distribution: 'temurin' + - name: Build + run: mvn -T 1C -B -ntp clean install -DskipTests -pl paimon-iceberg -am -Ppaimon-iceberg,iceberg-ga + - name: Test against GA Iceberg + run: mvn -B -ntp test -pl paimon-iceberg -Ppaimon-iceberg,iceberg-ga + env: + MAVEN_OPTS: -Xmx4096m diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java index 42e6ce2a8f58..ea6d256a4efb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java @@ -538,6 +538,16 @@ private void createMetadataWithoutBase( metrics.totalPositionDeletes = totalPositionDeleteRecords; metrics.totalEqualityDeletes = 0; + // a rebuild replaces metadata whose ids are already out with readers: never reuse them + Long snapshotFirstRowId = computeSnapshotFirstRowId(nextRowIdFloor); + ManifestRowIdAssignment rowIdAssignment = + assignManifestFirstRowIds(allManifestFileMetas, snapshotFirstRowId); + allManifestFileMetas = rowIdAssignment.manifests; + Long addedRows = snapshotFirstRowId == null ? null : rowIdAssignment.assignedRows; + Long nextRowId = + snapshotFirstRowId == null + ? null + : snapshotFirstRowId + rowIdAssignment.assignedRows; String manifestListFileName = manifestList.writeWithoutRolling(allManifestFileMetas); // current schema follows the latest; the snapshot entry records its own schema @@ -551,8 +561,6 @@ private void createMetadataWithoutBase( computeSnapshotSummary( IcebergSnapshotSummary.APPEND.operation(), paimonSnapshot, metrics); - // a rebuild replaces metadata whose ids are already out with readers: never reuse them - RowLineage rowLineage = computeRowLineage(nextRowIdFloor, metrics.addedRecords); IcebergSnapshot snapshot = new IcebergSnapshot( snapshotId, @@ -563,8 +571,8 @@ private void createMetadataWithoutBase( snapshotSummary, pathFactory.toManifestListPath(manifestListFileName).toString(), snapshotSchemaId, - rowLineage.firstRowId, - rowLineage.addedRows); + snapshotFirstRowId, + addedRows); // Tags can only be included in Iceberg if they point to an Iceberg snapshot that // exists. Otherwise, an Iceberg client fails to parse the metadata and all reads fail. @@ -607,7 +615,7 @@ private void createMetadataWithoutBase( IcebergPartitionField.FIRST_FIELD_ID - 1), Collections.singletonList(snapshot), (int) snapshotId, - rowLineage.nextRowId, + nextRowId, refs); Path metadataPath = pathFactory.toMetadataPath(snapshotId); @@ -725,7 +733,7 @@ private List getPartitionFields( return result; } - /** VARIANT needs Iceberg row lineage, which Paimon Iceberg compatibility cannot publish. */ + /** VARIANT is an Iceberg format-version-3 type; reject publishing it into v2 metadata. */ static void checkVariantNotPublishable(RowType rowType) { Collection variantFields = new LinkedHashSet<>(); for (DataField field : rowType.getFields()) { @@ -733,9 +741,8 @@ static void checkVariantNotPublishable(RowType rowType) { } Preconditions.checkArgument( variantFields.isEmpty(), - "Columns %s use the VARIANT type, which Paimon Iceberg compatibility cannot " - + "publish: it is an Iceberg format-version-3 type that requires row " - + "lineage.", + "Columns %s use the VARIANT type, which requires Iceberg format version 3. " + + "Set 'metadata.iceberg.format-version' = '3' to publish this table.", variantFields); } @@ -1086,13 +1093,6 @@ private void createMetadataWithBase( // compact data manifest file if needed newDataManifestFileMetas = compactMetadataIfNeeded(newDataManifestFileMetas, snapshotId); - String manifestListFileName = - manifestList.writeWithoutRolling( - Stream.concat( - newDataManifestFileMetas.stream(), - newDVManifestFileMetas.stream()) - .collect(Collectors.toList())); - SummaryMetrics metrics = new SummaryMetrics(); metrics.addedDataFiles = addedFiles.size(); metrics.addedRecords = @@ -1143,6 +1143,24 @@ private void createMetadataWithBase( metrics.totalPositionDeletes = computeLiveRowCount(newDVManifestFileMetas); metrics.totalEqualityDeletes = 0; + Long snapshotFirstRowId = computeSnapshotFirstRowId(rowIdFloor); + + ManifestRowIdAssignment rowIdAssignment = + assignManifestFirstRowIds( + Stream.concat( + newDataManifestFileMetas.stream(), + newDVManifestFileMetas.stream()) + .collect(Collectors.toList()), + snapshotFirstRowId); + List newManifestFileMetasWithRowIds = rowIdAssignment.manifests; + Long addedRows = snapshotFirstRowId == null ? null : rowIdAssignment.assignedRows; + Long nextRowId = + snapshotFirstRowId == null + ? null + : snapshotFirstRowId + rowIdAssignment.assignedRows; + String manifestListFileName = + manifestList.writeWithoutRolling(newManifestFileMetasWithRowIds); + IcebergSnapshotSummary snapshotSummary = computeSnapshotSummary(operation, snapshot, metrics); @@ -1161,8 +1179,6 @@ private void createMetadataWithBase( } // a schema-pointer rollback (validated above): only the current pointer moves - RowLineage rowLineage = computeRowLineage(rowIdFloor, metrics.addedRecords); - List snapshots = new ArrayList<>(baseMetadata.snapshots()); snapshots.add( new IcebergSnapshot( @@ -1175,8 +1191,8 @@ private void createMetadataWithBase( pathFactory.toManifestListPath(manifestListFileName).toString(), // the snapshot's own schema, for time travel snapshotSchemaId, - rowLineage.firstRowId, - rowLineage.addedRows)); + snapshotFirstRowId, + addedRows)); // all snapshots in this list, except the last one, need to expire List toExpireExceptLast = new ArrayList<>(); @@ -1221,7 +1237,7 @@ private void createMetadataWithBase( baseMetadata.lastPartitionId(), snapshots, (int) snapshotId, - rowLineage.nextRowId, + nextRowId, refs); Path metadataPath = pathFactory.toMetadataPath(snapshotId); @@ -1426,8 +1442,10 @@ private Pair, String> createWithDeleteManifestFile commitKind == Snapshot.CommitKind.COMPACT ? IcebergSnapshotSummary.REPLACE.operation() : IcebergSnapshotSummary.OVERWRITE.operation(); + List sourceEntries = + materializeFirstRowIds(fileMeta, entries); List newEntries = new ArrayList<>(); - for (IcebergManifestEntry entry : entries) { + for (IcebergManifestEntry entry : sourceEntries) { if (entry.isLive()) { boolean removed = removedFiles.containsKey(entry.file().filePath()); newEntries.add( @@ -1489,10 +1507,13 @@ private List compactMetadataIfNeeded( Function> processor = meta -> { + List sourceEntries = + materializeFirstRowIds( + meta, + IcebergManifestFile.create(table, pathFactory) + .read(new Path(meta.manifestPath()).getName())); List entries = new ArrayList<>(); - for (IcebergManifestEntry entry : - IcebergManifestFile.create(table, pathFactory) - .read(new Path(meta.manifestPath()).getName())) { + for (IcebergManifestEntry entry : sourceEntries) { // a deletion made by this commit is recorded against the current // snapshot but keeps the file sequence number of the older snapshot // that added the file, so it has to be recognised by snapshot id @@ -1553,9 +1574,14 @@ private boolean shouldExpire(IcebergSnapshot snapshot, long currentSnapshotId) { } private void expireManifestList(String toExpire, String next) { - Set metaInUse = new HashSet<>(manifestList.read(next)); + // compare by physical path: a carried-over manifest may be re-listed with different + // list-level fields (e.g. an assigned first_row_id) while sharing the same file + Set pathsInUse = new HashSet<>(); + for (IcebergManifestFileMeta meta : manifestList.read(next)) { + pathsInUse.add(meta.manifestPath()); + } for (IcebergManifestFileMeta meta : manifestList.read(toExpire)) { - if (metaInUse.contains(meta)) { + if (pathsInUse.contains(meta.manifestPath())) { continue; } table.fileIO().deleteQuietly(new Path(meta.manifestPath())); @@ -2001,24 +2027,102 @@ private boolean isSameFormatVersion(int baseFormatVersion) { /** * Row-lineage bookkeeping for a new snapshot, mandatory in Iceberg format version 3: the - * snapshot's first-row-id starts at the base metadata's next-row-id watermark and the table's - * next-row-id advances by the snapshot's added records. For format version 2 all fields stay - * null so nothing is written. + * snapshot's first-row-id starts at the base metadata's next-row-id watermark. The snapshot's + * added-rows and the table's next-row-id are NOT derived here: they depend on how many rows + * {@link #assignManifestFirstRowIds} actually assigns (which can exceed this commit's added + * records when a carried-over manifest is assigned for the first time, e.g. a Layer-1-written + * manifest being upgraded), so callers must recompute them from the assignment's result. For + * format version 2 the field stays null so nothing is written. */ - private RowLineage computeRowLineage(long baseNextRowId, long addedRecords) { - RowLineage lineage = new RowLineage(); - if (formatVersion >= IcebergMetadata.FORMAT_VERSION_V3) { - lineage.firstRowId = baseNextRowId; - lineage.addedRows = addedRecords; - lineage.nextRowId = baseNextRowId + addedRecords; - } - return lineage; + @Nullable + private Long computeSnapshotFirstRowId(long baseNextRowId) { + return formatVersion >= IcebergMetadata.FORMAT_VERSION_V3 ? baseNextRowId : null; } - private static class RowLineage { - @Nullable private Long firstRowId; - @Nullable private Long addedRows; - @Nullable private Long nextRowId; + /** + * Result of {@link #assignManifestFirstRowIds}: the manifests with first_row_id assigned, and + * the total number of rows actually consumed from the row-id space by that assignment (which + * may be larger than this commit's added-records count; see the class-level note there). + */ + private static class ManifestRowIdAssignment { + private final List manifests; + private final long assignedRows; + + private ManifestRowIdAssignment( + List manifests, long assignedRows) { + this.manifests = manifests; + this.assignedRows = assignedRows; + } + } + + /** + * Iceberg v3: assign first_row_id (field 520) to data manifests that do not have one yet. + * Manifests carried over from base metadata that are already assigned keep their value; delete + * manifests stay null. The watermark starts at the snapshot's first-row-id and advances by each + * newly-assigned manifest's TRUE inheriting-rows count (see {@link #trueInheritingRowsCount}), + * returned as {@link ManifestRowIdAssignment#assignedRows}. + * + *

A manifest written entirely under manifest-level assignment satisfies "inheriting rows == + * ADDED rows", so the bound is exact for it. A manifest carried over from before assignment + * existed may hold EXISTING entries whose field 142 is also still null; the bound covers them + * without reading the manifest, at the cost of spec-legal id gaps when some of those entries + * were already materialized. DELETED entries never inherit ids and are excluded. Callers MUST + * use {@code assignedRows} (not this commit's added-records count) to advance the snapshot's + * added-rows / table next-row-id, precisely because of that mismatch. + */ + private ManifestRowIdAssignment assignManifestFirstRowIds( + List manifests, @Nullable Long snapshotFirstRowId) { + if (snapshotFirstRowId == null) { + return new ManifestRowIdAssignment(manifests, 0L); + } + List result = new ArrayList<>(); + long watermark = snapshotFirstRowId; + for (IcebergManifestFileMeta meta : manifests) { + if (meta.content() == IcebergManifestFileMeta.Content.DATA + && meta.firstRowId() == null) { + result.add(meta.withFirstRowId(watermark)); + // spec-sanctioned upper bound: only ADDED and EXISTING rows can inherit + // ids from this manifest (readers never assign ids to DELETED entries). + // Rows whose field 142 is already materialized merely widen the reserved + // range, leaving legal id gaps - in exchange the commit path never has to + // read manifest contents. + watermark += meta.addedRowsCount() + meta.existingRowsCount(); + } else { + result.add(meta); + } + } + return new ManifestRowIdAssignment(result, watermark - snapshotFirstRowId); + } + + /** + * Iceberg v3 requires the inherited first_row_id to be written into file metadata when entries + * are copied into a rewritten manifest. Computes each entry's effective id in base manifest + * order (explicit field 142, or inherited from the manifest's first_row_id, skipping DELETED + * entries exactly like GA readers do) and returns entries with the id materialized. No-op for + * delete manifests and for base manifests without an assigned first_row_id (v2 metadata, or v3 + * metadata written before manifest-level assignment existed — those stay in the spec's + * upgraded-table state). + */ + private static List materializeFirstRowIds( + IcebergManifestFileMeta baseMeta, List entries) { + if (baseMeta.content() != IcebergManifestFileMeta.Content.DATA + || baseMeta.firstRowId() == null) { + return entries; + } + List result = new ArrayList<>(); + long watermark = baseMeta.firstRowId(); + for (IcebergManifestEntry entry : entries) { + if (entry.status() != IcebergManifestEntry.Status.DELETED + && entry.file().firstRowId() == null) { + result.add(entry.withFile(entry.file().withFirstRowId(watermark))); + watermark += entry.file().recordCount(); + } else { + // DELETED entries never inherit an id (GA readers skip them when + // assigning), so their field 142 stays null and the walk does not advance + result.add(entry); + } + } + return result; } private class SchemaCache { @@ -2031,8 +2135,11 @@ private IcebergSchema get(long schemaId) { schemaId, id -> { TableSchema schema = schemaManager.schema(id); - // backstop: reject variant on each schema as it is emitted - checkVariantNotPublishable(schema.logicalRowType()); + if (formatVersion < IcebergMetadata.FORMAT_VERSION_V3) { + // VARIANT is an Iceberg format-version-3 type; v2 metadata cannot + // represent it + checkVariantNotPublishable(schema.logicalRowType()); + } SchemaValidation.validateIcebergGeospatialTypes( schema.logicalRowType(), table.coreOptions()); return IcebergSchema.create(schema); diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMeta.java index 950da63b9d11..f9ce30bad4e3 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMeta.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMeta.java @@ -87,6 +87,7 @@ public static Content fromId(int id) { private final String referencedDataFile; private final Long contentOffset; private final Long contentSizeInBytes; + @Nullable private final Long firstRowId; IcebergDataFileMeta( Content content, @@ -110,6 +111,7 @@ public static Content fromId(int id) { upperBounds, null, null, + null, null); } @@ -125,7 +127,8 @@ public static Content fromId(int id) { InternalMap upperBounds, String referencedDataFile, Long contentOffset, - Long contentSizeInBytes) { + Long contentSizeInBytes, + @Nullable Long firstRowId) { this.content = content; this.filePath = filePath; this.fileFormat = fileFormat; @@ -139,6 +142,7 @@ public static Content fromId(int id) { this.referencedDataFile = referencedDataFile; this.contentOffset = contentOffset; this.contentSizeInBytes = contentSizeInBytes; + this.firstRowId = firstRowId; } public static IcebergDataFileMeta create( @@ -245,7 +249,8 @@ public static IcebergDataFileMeta createForDeleteFile( null, referencedDataFile, contentOffset, - contentSizeInBytes); + contentSizeInBytes, + null); } public Content content() { @@ -296,7 +301,33 @@ public Long contentSizeInBytes() { return contentSizeInBytes; } + @Nullable + public Long firstRowId() { + return firstRowId; + } + + public IcebergDataFileMeta withFirstRowId(long firstRowId) { + return new IcebergDataFileMeta( + content, + filePath, + fileFormat, + partition, + recordCount, + fileSizeInBytes, + nullValueCounts, + lowerBounds, + upperBounds, + referencedDataFile, + contentOffset, + contentSizeInBytes, + firstRowId); + } + public static RowType schema(RowType partitionType) { + return schema(partitionType, false); + } + + public static RowType schema(RowType partitionType, boolean withFirstRowId) { List fields = new ArrayList<>(); fields.add(new DataField(134, "content", DataTypes.INT().notNull())); fields.add(new DataField(100, "file_path", DataTypes.STRING().notNull())); @@ -322,6 +353,9 @@ public static RowType schema(RowType partitionType) { fields.add(new DataField(143, "referenced_data_file", DataTypes.STRING())); fields.add(new DataField(144, "content_offset", DataTypes.BIGINT())); fields.add(new DataField(145, "content_size_in_bytes", DataTypes.BIGINT())); + if (withFirstRowId) { + fields.add(new DataField(142, "first_row_id", DataTypes.BIGINT())); + } return new RowType(false, fields); } @@ -345,7 +379,8 @@ public boolean equals(Object o) { && Objects.equals(upperBounds, that.upperBounds) && Objects.equals(referencedDataFile, that.referencedDataFile) && Objects.equals(contentOffset, that.contentOffset) - && Objects.equals(contentSizeInBytes, that.contentSizeInBytes); + && Objects.equals(contentSizeInBytes, that.contentSizeInBytes) + && Objects.equals(firstRowId, that.firstRowId); } @Override @@ -362,6 +397,7 @@ public int hashCode() { upperBounds, referencedDataFile, contentOffset, - contentSizeInBytes); + contentSizeInBytes, + firstRowId); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaSerializer.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaSerializer.java index 020b5e1b44e3..5db5dea27db7 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaSerializer.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaSerializer.java @@ -36,18 +36,40 @@ public class IcebergDataFileMetaSerializer extends ObjectSerializer { public IcebergManifestFile( FileIO fileIO, RowType partitionType, + boolean withFirstRowId, FormatReaderFactory readerFactory, FormatWriterFactory writerFactory, String compression, @@ -79,8 +81,8 @@ public IcebergManifestFile( MemorySize targetFileSize) { super( fileIO, - new IcebergManifestEntrySerializer(partitionType), - IcebergManifestEntry.schema(partitionType), + new IcebergManifestEntrySerializer(partitionType, withFirstRowId), + IcebergManifestEntry.schema(partitionType, withFirstRowId), readerFactory, writerFactory, compression, @@ -98,8 +100,10 @@ public String compression() { public static IcebergManifestFile create(FileStoreTable table, IcebergPathFactory pathFactory) { RowType partitionType = table.schema().logicalPartitionType(); - RowType entryType = IcebergManifestEntry.schema(partitionType); Options avroOptions = Options.fromMap(table.options()); + boolean withFirstRowId = + avroOptions.get(IcebergOptions.FORMAT_VERSION) >= IcebergMetadata.FORMAT_VERSION_V3; + RowType entryType = IcebergManifestEntry.schema(partitionType, withFirstRowId); // https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/ManifestReader.java avroOptions.set( "avro.row-name-mapping", @@ -120,6 +124,7 @@ public static IcebergManifestFile create(FileStoreTable table, IcebergPathFactor return new IcebergManifestFile( table.fileIO(), partitionType, + withFirstRowId, manifestFileAvro.createReaderFactory(entryType, entryType, new ArrayList<>()), manifestFileAvro.createWriterFactory(entryType), avroOptions.get(IcebergOptions.MANIFEST_COMPRESSION), @@ -298,7 +303,8 @@ public IcebergManifestFileMeta result() throws IOException { addedRowsCount, existingRowsCount, deletedRowsCount, - partitionSummaries); + partitionSummaries, + null); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java index da3e0c24029e..d5ac07c962ed 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java @@ -22,6 +22,8 @@ import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.List; import java.util.Objects; @@ -73,6 +75,7 @@ public static Content fromId(int id) { private final long existingRowsCount; private final long deletedRowsCount; private final List partitions; + @Nullable private final Long firstRowId; public IcebergManifestFileMeta( String manifestPath, @@ -88,7 +91,8 @@ public IcebergManifestFileMeta( long addedRowsCount, long existingRowsCount, long deletedRowsCount, - List partitions) { + List partitions, + @Nullable Long firstRowId) { this.manifestPath = manifestPath; this.manifestLength = manifestLength; this.partitionSpecId = partitionSpecId; @@ -103,6 +107,7 @@ public IcebergManifestFileMeta( this.existingRowsCount = existingRowsCount; this.deletedRowsCount = deletedRowsCount; this.partitions = partitions; + this.firstRowId = firstRowId; } public String manifestPath() { @@ -165,8 +170,42 @@ public List partitions() { return partitions; } + @Nullable + public Long firstRowId() { + return firstRowId; + } + + public IcebergManifestFileMeta withFirstRowId(long firstRowId) { + return new IcebergManifestFileMeta( + manifestPath, + manifestLength, + partitionSpecId, + content, + sequenceNumber, + minSequenceNumber, + addedSnapshotId, + addedFilesCount, + existingFilesCount, + deletedFilesCount, + addedRowsCount, + existingRowsCount, + deletedRowsCount, + partitions, + firstRowId); + } + public static RowType schema(boolean legacyVersion) { - return legacyVersion ? schemaForIceberg1_4() : schemaForIcebergNew(); + return schema(legacyVersion, false); + } + + public static RowType schema(boolean legacyVersion, boolean withFirstRowId) { + RowType base = legacyVersion ? schemaForIceberg1_4() : schemaForIcebergNew(); + if (!withFirstRowId) { + return base; + } + List fields = new ArrayList<>(base.getFields()); + fields.add(new DataField(520, "first_row_id", DataTypes.BIGINT())); + return new RowType(false, fields); } private static RowType schemaForIcebergNew() { @@ -235,7 +274,8 @@ public boolean equals(Object o) { && addedRowsCount == that.addedRowsCount && existingRowsCount == that.existingRowsCount && deletedRowsCount == that.deletedRowsCount - && Objects.equals(partitions, that.partitions); + && Objects.equals(partitions, that.partitions) + && Objects.equals(firstRowId, that.firstRowId); } @Override @@ -254,6 +294,7 @@ public int hashCode() { addedRowsCount, existingRowsCount, deletedRowsCount, - partitions); + partitions, + firstRowId); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMetaSerializer.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMetaSerializer.java index 2b4c9b771c59..14b6bbf91bed 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMetaSerializer.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMetaSerializer.java @@ -36,14 +36,41 @@ public class IcebergManifestFileMetaSerializer extends ObjectSerializer toPartitionSummaries(InternalArray array) { diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestList.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestList.java index 9eb1194c6f59..0439aba08716 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestList.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestList.java @@ -23,6 +23,7 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.iceberg.IcebergOptions; import org.apache.paimon.iceberg.IcebergPathFactory; +import org.apache.paimon.iceberg.metadata.IcebergMetadata; import org.apache.paimon.options.Options; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.types.RowType; @@ -69,9 +70,11 @@ public static IcebergManifestList create(FileStoreTable table, IcebergPathFactor + "manifest_file_partitions:r508," + "array_id_r508:508"); FileFormat fileFormat = FileFormat.fromIdentifier("avro", avroOptions); + boolean withFirstRowId = + avroOptions.get(IcebergOptions.FORMAT_VERSION) >= IcebergMetadata.FORMAT_VERSION_V3; RowType manifestType = IcebergManifestFileMeta.schema( - avroOptions.get(IcebergOptions.MANIFEST_LEGACY_VERSION)); + avroOptions.get(IcebergOptions.MANIFEST_LEGACY_VERSION), withFirstRowId); return new IcebergManifestList( table.fileIO(), fileFormat, diff --git a/paimon-iceberg/pom.xml b/paimon-iceberg/pom.xml index 403419d69b42..bf02653af2d2 100644 --- a/paimon-iceberg/pom.xml +++ b/paimon-iceberg/pom.xml @@ -38,6 +38,7 @@ under the License. 1.19 2.3.10 1.19.0 + 11.0.24 @@ -192,6 +193,14 @@ under the License. commons-io commons-io + + + org.eclipse.jetty + * + + + org.eclipse.jetty.websocket + * @@ -229,6 +238,14 @@ under the License. commons-io commons-io + + + org.eclipse.jetty + * + + + org.eclipse.jetty.websocket + * @@ -275,7 +292,7 @@ under the License. org.eclipse.jetty jetty-server - 11.0.24 + ${jetty.server.version} test @@ -356,4 +373,79 @@ under the License. + + + + + iceberg-ga + + 1.11.0 + 1.20 + 1.20.0 + + 12.1.8 + + 3.4.3 + + + + + com.fasterxml.jackson.core + jackson-annotations + 2.21 + provided + + + + org.eclipse.jetty.ee10 + jetty-ee10-servlet + 12.1.8 + test + + + + + org.eclipse.jetty + jetty-security + 12.1.8 + test + + + + org.eclipse.jetty.compression + jetty-compression-server + 12.1.8 + test + + + + org.eclipse.jetty.compression + jetty-compression-gzip + 12.1.8 + test + + + + \ No newline at end of file diff --git a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java index 643839625cd4..327361b9d3f8 100644 --- a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java +++ b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java @@ -20,6 +20,7 @@ import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.iceberg.metadata.IcebergMetadata; import org.apache.paimon.iceberg.metadata.IcebergSchema; @@ -45,6 +46,7 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.rest.Endpoint; import org.apache.iceberg.rest.RESTCatalog; import org.apache.iceberg.types.Types; import org.apache.iceberg.types.Types.NestedField; @@ -53,6 +55,8 @@ import javax.annotation.Nullable; +import java.io.IOException; +import java.lang.reflect.Field; import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; @@ -60,6 +64,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.UUID; import java.util.stream.Collectors; import static org.apache.iceberg.CatalogUtil.ICEBERG_CATALOG_TYPE; @@ -79,6 +84,8 @@ public class IcebergRestMetadataCommitter implements IcebergMetadataCommitter { private static final String REST_CATALOG_NAME = "rest-catalog"; private final RESTCatalog restCatalog; + private final FileIO fileIO; + private final Path metadataDirectory; private final String icebergDatabaseName; private final TableIdentifier icebergTableIdentifier; private final IcebergOptions icebergOptions; @@ -88,6 +95,8 @@ public class IcebergRestMetadataCommitter implements IcebergMetadataCommitter { public IcebergRestMetadataCommitter(FileStoreTable table) { Options options = new Options(table.options()); icebergOptions = new IcebergOptions(options); + this.fileIO = table.fileIO(); + this.metadataDirectory = IcebergCommitCallback.catalogTableMetadataPath(table); Identifier identifier = Preconditions.checkNotNull(table.catalogEnvironment().identifier()); String icebergDatabase = options.get(IcebergOptions.METASTORE_DATABASE); @@ -151,6 +160,11 @@ private void commitMetadataImpl( try { if (!tableExists()) { + if (requiresRegistration(newIcebergMetadata)) { + LOG.info("Table {} does not exist, register it.", icebergTableIdentifier); + registerAsCurrent(newIcebergMetadata, newMetadata, false); + return; + } LOG.info("Table {} does not exist, create it.", icebergTableIdentifier); icebergTable = createTable(newMetadata); updateBuilder = @@ -191,6 +205,12 @@ private void commitMetadataImpl( LOG.info( "Iceberg table {} exists but has no snapshots, treating as new table.", icebergTableIdentifier); + if (requiresRegistration(newIcebergMetadata)) { + // registration has no post-create commit step, so the drop+create + // failure loop this branch guards against cannot occur + registerAsCurrent(newIcebergMetadata, newMetadata, true); + return; + } updateBuilder = updatesForCorrectBase(metadata, newMetadata, true); } else { boolean withBase = checkBase(metadata, newMetadata, baseIcebergMetadata); @@ -204,6 +224,13 @@ private void commitMetadataImpl( newMetadata.currentSnapshot() != null ? newMetadata.currentSnapshot().snapshotId() : "No snapshot"); + if (requiresRegistration(newIcebergMetadata)) { + LOG.info( + "the base metadata is incorrect, re-registering the iceberg" + + " table from local metadata."); + registerAsCurrent(newIcebergMetadata, newMetadata, true); + return; + } updateBuilder = updatesForIncorrectBase(newMetadata); } } @@ -321,6 +348,172 @@ void createDatabase() { } } + /** + * Whether publishing {@code metadata} to a new or recreated catalog table must go through + * {@link RESTCatalog#registerTable}. The create/update path replays only the current snapshot + * through {@link TableMetadata.Builder}, which derives the table's next-row-id from that + * snapshot's added-rows alone, so the server ends at {@code added-rows} while the local + * watermark is {@code first-row-id + added-rows}. Any nonzero first-row-id (rollback and + * self-heal rebuilds) would leave the server below ids already assigned in manifests, and a + * later external writer could reuse them; registration imports the metadata verbatim. Format + * version 2 tables and zero-based v3 metadata keep the create/update path, which every REST + * catalog supports. + */ + private static boolean requiresRegistration(IcebergMetadata metadata) { + IcebergSnapshot current = metadata.currentSnapshot(); + return metadata.formatVersion() >= IcebergMetadata.FORMAT_VERSION_V3 + && current != null + && current.firstRowId() != null + && current.firstRowId() != 0; + } + + /** + * Makes the catalog's state exactly the (REST-adjusted) local metadata by registering a + * metadata file, instead of rebuilding the table through {@link TableMetadata.Builder}; see + * {@link #requiresRegistration}. Fails before touching the catalog table when the server does + * not support registration. + */ + private void registerAsCurrent( + IcebergMetadata adjustedMetadata, TableMetadata newMetadata, boolean dropFirst) { + if (!registerTableAdvertised()) { + throw registerTableUnsupported( + "The Iceberg REST catalog does not advertise registerTable", + adjustedMetadata, + null); + } + try { + Path registerPath = writeRegisterFile(adjustedMetadata); + if (dropFirst) { + if (probeRegisterTable(registerPath)) { + // the table disappeared concurrently and the probe registered it + verifyRegistered(newMetadata); + return; + } + dropTable(); + } + icebergTable = + restCatalog.registerTable(icebergTableIdentifier, registerPath.toString()); + verifyRegistered(newMetadata); + } catch (UnsupportedOperationException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException( + "Fail to register iceberg table " + icebergTableIdentifier, e); + } + } + + /** + * Whether the server advertises the register-table endpoint. Iceberg's REST client keeps the + * endpoint set from the server's {@code /v1/config} response (or its backwards-compatible + * defaults, register-table included, when the server advertises none) in {@code + * RESTSessionCatalog#endpoints}, without a public accessor. + */ + private boolean registerTableAdvertised() { + try { + Field sessionField = RESTCatalog.class.getDeclaredField("sessionCatalog"); + sessionField.setAccessible(true); + Object sessionCatalog = sessionField.get(restCatalog); + Field endpointsField = null; + for (Class c = sessionCatalog.getClass(); c != null; c = c.getSuperclass()) { + try { + endpointsField = c.getDeclaredField("endpoints"); + break; + } catch (NoSuchFieldException ignored) { + // keep looking in the superclass + } + } + if (endpointsField == null) { + throw new NoSuchFieldException("endpoints"); + } + endpointsField.setAccessible(true); + Set endpoints = (Set) endpointsField.get(sessionCatalog); + return endpoints.contains(Endpoint.V1_REGISTER_TABLE); + } catch (Exception | LinkageError e) { + LOG.warn( + "Cannot read the endpoints advertised by the Iceberg REST catalog, " + + "assuming registerTable is unsupported.", + e); + return false; + } + } + + /** + * A server that advertises no endpoint list gets the client's default list, register-table + * included, so {@link #registerTableAdvertised} alone cannot prove support. Before the existing + * table is dropped, the registration is attempted against it: a server that implements the + * endpoint rejects it with {@link AlreadyExistsException}; anything else means the endpoint is + * unavailable and the existing table is left untouched. + * + * @return true if the registration went through because the table no longer existed + */ + private boolean probeRegisterTable(Path registerPath) { + try { + icebergTable = + restCatalog.registerTable(icebergTableIdentifier, registerPath.toString()); + return true; + } catch (AlreadyExistsException e) { + return false; + } catch (RuntimeException e) { + throw registerTableUnsupported( + "registerTable against the existing Iceberg REST catalog table failed with " + + "something other than AlreadyExists, so the endpoint is assumed " + + "unsupported", + null, + e); + } + } + + private UnsupportedOperationException registerTableUnsupported( + String reason, @Nullable IcebergMetadata metadata, @Nullable Throwable cause) { + return new UnsupportedOperationException( + String.format( + "%s; registerTable is required to publish format version 3 metadata " + + "whose row-id space does not start at 0%s. The catalog table " + + "%s was left untouched.", + reason, + metadata == null + ? "" + : String.format( + " (current snapshot first-row-id %s, next-row-id %s)", + metadata.currentSnapshot().firstRowId(), + metadata.nextRowId()), + icebergTableIdentifier), + cause); + } + + /** + * Writes the metadata to register into a fresh, never-overwritten file. A catalog may keep + * referencing the registered location, and rollback and rebuild paths reuse Paimon snapshot ids + * for different timelines, so the name carries a UUID and an existing file is never replaced. + */ + private Path writeRegisterFile(IcebergMetadata metadata) throws IOException { + Path registerPath = + new Path( + metadataDirectory, + String.format( + "rest-register-v%d-%s.metadata.json", + metadata.currentSnapshotId(), UUID.randomUUID())); + if (!fileIO.tryToWriteAtomic(registerPath, metadata.toJson())) { + throw new IOException("Metadata file to register already exists: " + registerPath); + } + return registerPath; + } + + private void verifyRegistered(TableMetadata newMetadata) { + long registered = + ((BaseTable) icebergTable).operations().current().currentSnapshot().snapshotId(); + if (newMetadata.currentSnapshot() == null + || registered != newMetadata.currentSnapshot().snapshotId()) { + throw new IllegalStateException( + String.format( + "Registered catalog table is at snapshot %s instead of %s", + registered, + newMetadata.currentSnapshot() == null + ? "null" + : newMetadata.currentSnapshot().snapshotId())); + } + } + private Table createTable(TableMetadata newMetadata) { /* Handles fieldId incompatibility between Paimon (starts at 0) and Iceberg (starts at 1). @@ -410,15 +603,6 @@ static String toRestLocation(String location) { return location; } - private Table getTable() { - return restCatalog.loadTable(icebergTableIdentifier); - } - - private void dropTable() { - // set purge to false, because we don't need to delete the data files - restCatalog.dropTable(icebergTableIdentifier, false); - } - private Table recreateTable(TableMetadata newMetadata) { try { dropTable(); @@ -428,6 +612,15 @@ private Table recreateTable(TableMetadata newMetadata) { } } + private Table getTable() { + return restCatalog.loadTable(icebergTableIdentifier); + } + + private void dropTable() { + // set purge to false, because we don't need to delete the data files + restCatalog.dropTable(icebergTableIdentifier, false); + } + // ------------------------------------------------------------------------------------- // metadata updates // ------------------------------------------------------------------------------------- @@ -682,9 +875,6 @@ private static String updateToString(MetadataUpdate update) { return String.format( "AddSnapshot(%s)", ((MetadataUpdate.AddSnapshot) update).snapshot().snapshotId()); - } else if (update instanceof MetadataUpdate.RemoveSnapshot) { - return String.format( - "RemoveSnapshot(%s)", ((MetadataUpdate.RemoveSnapshot) update).snapshotId()); } else if (update instanceof MetadataUpdate.SetSnapshotRef) { return String.format( "SetSnapshotRef(%s, %s, %s)", diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java index cab2fe10d4b1..b10cb345d19d 100644 --- a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java +++ b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java @@ -21,34 +21,63 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.catalog.FileSystemCatalog; import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.variant.GenericVariant; import org.apache.paimon.disk.IOManagerImpl; import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.iceberg.IcebergOptions; +import org.apache.paimon.iceberg.IcebergPathFactory; +import org.apache.paimon.iceberg.manifest.IcebergManifestEntry; +import org.apache.paimon.iceberg.manifest.IcebergManifestFile; +import org.apache.paimon.iceberg.manifest.IcebergManifestFileMeta; +import org.apache.paimon.iceberg.manifest.IcebergManifestList; import org.apache.paimon.iceberg.metadata.IcebergMetadata; import org.apache.paimon.iceberg.metadata.IcebergSnapshot; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.schema.Schema; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.FixedBucketRowKeyExtractor; import org.apache.paimon.table.sink.TableCommitImpl; import org.apache.paimon.table.sink.TableWriteImpl; +import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.IOUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestReader; +import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.TableMetadataParser; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.IcebergGenerics; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.hadoop.HadoopCatalog; +import org.apache.iceberg.io.CloseableIterable; +import org.junit.jupiter.api.Assumptions; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; +import java.util.List; import java.util.Map; import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for Iceberg format-version 3 row-lineage metadata fields. */ public class IcebergRowLineageCompatibilityTest { @@ -207,6 +236,775 @@ public void testTagPreservesNextRowId() throws Exception { assertThat(metadata.nextRowId()).isEqualTo(3L); } + @Test + public void testVariantPublishableWithFormatVersion3() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.VARIANT()}, + new String[] {"k", "payload"}); + FileStoreTable table = createPaimonTable(rowType, formatVersionOptions(3), "parquet"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + + write.write(GenericRow.of(1, GenericVariant.fromJson("{\"a\": 1}"))); + commit.commit(1, write.prepareCommit(false, 1)); + write.close(); + commit.close(); + + IcebergMetadata metadata = readIcebergMetadata(table, 1); + assertThat(metadata.nextRowId()).isEqualTo(1L); + assertThat(metadata.schemas().get(metadata.currentSchemaId()).fields().get(1).type()) + .isEqualTo("variant"); + } + + @Test + public void testVariantRejectedWithFormatVersion2() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.VARIANT()}, + new String[] {"k", "payload"}); + FileStoreTable table = createPaimonTable(rowType, formatVersionOptions(2), "parquet"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + + write.write(GenericRow.of(1, GenericVariant.fromJson("{\"a\": 1}"))); + // hasStackTraceContaining: robust whether or not the commit path wraps the + // IllegalArgumentException from the guard + assertThatThrownBy(() -> commit.commit(1, write.prepareCommit(false, 1))) + .hasStackTraceContaining("VARIANT"); + write.close(); + commit.close(); + } + + @Test + public void testManifestListCarriesFirstRowIdColumn() throws Exception { + FileStoreTable table = createPaimonTable(defaultRowType(), formatVersionOptions(3), "avro"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + write.write(GenericRow.of(2, 20)); + commit.commit(1, write.prepareCommit(false, 1)); + write.close(); + commit.close(); + + // the v3 manifest list must round-trip the first_row_id column + IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergManifestList manifestList = IcebergManifestList.create(table, paths); + List metas = + manifestList.read( + new Path(readIcebergMetadata(table, 1).currentSnapshot().manifestList()) + .getName()); + assertThat(metas).isNotEmpty(); + // the v3 manifest list round-trips the first_row_id column; data manifests are assigned + // at manifest-list write time (this table's single commit starts at row id 0) + for (IcebergManifestFileMeta meta : metas) { + if (meta.content() == IcebergManifestFileMeta.Content.DATA) { + assertThat(meta.firstRowId()).isEqualTo(0L); + } else { + assertThat(meta.firstRowId()).isNull(); + } + } + } + + @Test + public void testManifestFirstRowIdAssignment() throws Exception { + FileStoreTable table = createPaimonTable(defaultRowType(), formatVersionOptions(3), "avro"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + + write.write(GenericRow.of(1, 10)); + write.write(GenericRow.of(2, 20)); + commit.commit(1, write.prepareCommit(false, 1)); + + write.write(GenericRow.of(3, 30)); + commit.commit(2, write.prepareCommit(false, 2)); + write.close(); + commit.close(); + + IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergManifestList manifestList = IcebergManifestList.create(table, paths); + List metas = + manifestList.read( + new Path(readIcebergMetadata(table, 2).currentSnapshot().manifestList()) + .getName()); + + // every data manifest is assigned; watermark walks addedRowsCount in list order + long watermark = -1; + long totalAssigned = 0; + for (IcebergManifestFileMeta meta : metas) { + if (meta.content() == IcebergManifestFileMeta.Content.DATA) { + assertThat(meta.firstRowId()).isNotNull(); + assertThat(meta.firstRowId()).isGreaterThan(watermark); + watermark = meta.firstRowId(); + totalAssigned += meta.addedRowsCount(); + } else { + assertThat(meta.firstRowId()).isNull(); + } + } + // commit1 assigned rows [0,2), commit2 assigned [2,3): manifests carry 0 and 2 + assertThat( + metas.stream() + .filter(m -> m.content() == IcebergManifestFileMeta.Content.DATA) + .map(IcebergManifestFileMeta::firstRowId)) + .containsExactlyInAnyOrder(0L, 2L); + assertThat(totalAssigned).isEqualTo(3L); + } + + @Test + public void testReadsManifestListWrittenWithoutFirstRowIdColumn() throws Exception { + // A Layer-1 manifest list physically lacks column 520. The v3 reader must resolve + // the missing column to null (Avro schema resolution), not fail. + FileStoreTable table = createPaimonTable(defaultRowType(), formatVersionOptions(3), "avro"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + write.close(); + commit.close(); + + IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata")); + String listName = + new Path(readIcebergMetadata(table, 1).currentSnapshot().manifestList()).getName(); + + // rewrite the manifest list through the OLD (14-column) serializer to simulate Layer 1 + FileStoreTable v2SchemaView = + table.copy(Collections.singletonMap(IcebergOptions.FORMAT_VERSION.key(), "2")); + IcebergManifestList oldWriter = IcebergManifestList.create(v2SchemaView, paths); + IcebergManifestList newReader = IcebergManifestList.create(table, paths); + List metas = newReader.read(listName); + String rewritten = oldWriter.writeWithoutRolling(metas); + LocalFileIO.create().deleteQuietly(paths.toManifestListPath(listName)); + LocalFileIO.create() + .rename(paths.toManifestListPath(rewritten), paths.toManifestListPath(listName)); + + // the v3 reader resolves the absent column to null for every meta + for (IcebergManifestFileMeta meta : newReader.read(listName)) { + assertThat(meta.firstRowId()).isNull(); + } + } + + @Test + public void testExpireKeepsManifestSharedAcrossRowIdAssignment() throws Exception { + // A manifest written before manifest-level lineage is re-listed with an assigned + // first_row_id but the same physical path. Expiring the pre-assignment manifest list + // must not delete the shared file, so liveness is decided by path, not value equality. + Map options = formatVersionOptions(3); + options.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "1"); + options.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "2"); + FileStoreTable table = createPaimonTable(defaultRowType(), options, "avro"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + + IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata")); + String listName = + new Path(readIcebergMetadata(table, 1).currentSnapshot().manifestList()).getName(); + + // strip column 520 from snapshot 1's manifest list to simulate a pre-assignment writer + FileStoreTable v2SchemaView = + table.copy(Collections.singletonMap(IcebergOptions.FORMAT_VERSION.key(), "2")); + IcebergManifestList oldWriter = IcebergManifestList.create(v2SchemaView, paths); + IcebergManifestList reader = IcebergManifestList.create(table, paths); + String rewritten = oldWriter.writeWithoutRolling(reader.read(listName)); + LocalFileIO.create().deleteQuietly(paths.toManifestListPath(listName)); + LocalFileIO.create() + .rename(paths.toManifestListPath(rewritten), paths.toManifestListPath(listName)); + List sharedManifestPaths = new ArrayList<>(); + for (IcebergManifestFileMeta meta : reader.read(listName)) { + sharedManifestPaths.add(meta.manifestPath()); + } + assertThat(sharedManifestPaths).isNotEmpty(); + + // commit 2 carries the manifest over, assigning first_row_id at the same path; + // commit 3 expires snapshot 1's manifest list against snapshot 2's + write.write(GenericRow.of(2, 20)); + commit.commit(2, write.prepareCommit(false, 2)); + write.write(GenericRow.of(3, 30)); + commit.commit(3, write.prepareCommit(false, 3)); + write.close(); + commit.close(); + + IcebergMetadata metadata = readIcebergMetadata(table, 3); + assertThat(metadata.snapshots()).hasSize(2); + for (String manifestPath : sharedManifestPaths) { + assertThat(LocalFileIO.create().exists(new Path(manifestPath))).isTrue(); + } + // every retained snapshot must stay readable end to end + IcebergManifestFile manifestFile = IcebergManifestFile.create(table, paths); + for (IcebergSnapshot snapshot : metadata.snapshots()) { + for (IcebergManifestFileMeta meta : + reader.read(new Path(snapshot.manifestList()).getName())) { + assertThat(manifestFile.read(meta)).isNotEmpty(); + } + } + } + + @Test + public void testManifestEntriesCarryFirstRowIdColumn() throws Exception { + FileStoreTable table = createPaimonTable(defaultRowType(), formatVersionOptions(3), "avro"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + write.close(); + commit.close(); + + IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergManifestList manifestList = IcebergManifestList.create(table, paths); + IcebergManifestFile manifestFile = IcebergManifestFile.create(table, paths); + List metas = + manifestList.read( + new Path(readIcebergMetadata(table, 1).currentSnapshot().manifestList()) + .getName()); + for (IcebergManifestFileMeta meta : metas) { + for (IcebergManifestEntry entry : manifestFile.read(meta)) { + // ADDED entries are unassigned by definition; the column must round-trip as null + assertThat(entry.file().firstRowId()).isNull(); + } + } + } + + @Test + public void testFirstRowIdStableAcrossManifestRewrite() throws Exception { + // A single-bucket table always rewrites its one-and-only file whole, so no entry ever + // survives a rewrite unchanged (the same-path invariant this test targets never fires). + // Use two buckets instead, and only touch one of them: the other bucket's file must + // then be carried, byte-for-byte unchanged, as an EXISTING entry into the manifest that + // gets rewritten because its sibling entry was removed. + RowType rowType = defaultRowType(); + Map customOptions = formatVersionOptions(3); + customOptions.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); + customOptions.put(CoreOptions.DELETION_VECTOR_BITMAP64.key(), "true"); + FileStoreTable table = createPkPaimonTable(rowType, customOptions); + + int keyBucket0 = findKeyForBucket(table, 0); + int keyBucket1 = findKeyForBucket(table, 1); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + + write.write(GenericRow.of(keyBucket0, 10)); + write.write(GenericRow.of(keyBucket1, 20)); + write.compact(BinaryRow.EMPTY_ROW, 0, true); + write.compact(BinaryRow.EMPTY_ROW, 1, true); + commit.commit(1, write.prepareCommit(true, 1)); + + // an append bundled with a full compaction is committed as two separate physical + // snapshots (append, then compact), so the Iceberg metadata id to inspect is whatever + // snapshot id is actually latest now, not the external commit identifier used above + Map idsBefore = + effectiveFileFirstRowIds(table, table.snapshotManager().latestSnapshotId()); + assertThat(idsBefore).isNotEmpty(); + + // delete the bucket-0 key and compact only bucket 0: bucket 1's file is left entirely + // untouched, so the shared manifest is rewritten (bucket 0's entry removed) while + // bucket 1's entry must survive, under the same file path, as a materialized EXISTING + // entry + write.write(GenericRow.ofKind(RowKind.DELETE, keyBucket0, 10)); + write.compact(BinaryRow.EMPTY_ROW, 0, true); + commit.commit(2, write.prepareCommit(true, 2)); + write.close(); + commit.close(); + + Map idsAfter = + effectiveFileFirstRowIds(table, table.snapshotManager().latestSnapshotId()); + assertThat(idsAfter).isNotEmpty(); + boolean checkedAtLeastOneSurvivor = false; + for (Map.Entry e : idsAfter.entrySet()) { + Long before = idsBefore.get(e.getKey()); + if (before != null) { + checkedAtLeastOneSurvivor = true; + // a file carried across the rewrite keeps its effective first row id + assertThat(e.getValue()).as("file %s", e.getKey()).isEqualTo(before); + } + } + assertThat(checkedAtLeastOneSurvivor) + .as("expected bucket 1's file to survive the manifest rewrite") + .isTrue(); + } + + @Test + public void testDeletedEntriesDoNotShiftInheritedIds() throws Exception { + // A legacy (pre-assignment) manifest holding a DELETED entry before a live one: GA + // readers skip DELETED entries when assigning inherited ids, so the live entry + // inherits the manifest's first_row_id itself, NOT shifted by the deleted rows. + RowType rowType = defaultRowType(); + Map customOptions = formatVersionOptions(3); + customOptions.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); + customOptions.put(CoreOptions.DELETION_VECTOR_BITMAP64.key(), "true"); + FileStoreTable table = createPkPaimonTable(rowType, customOptions); + + int keyBucket0 = findKeyForBucket(table, 0); + int keyBucket1 = findKeyForBucket(table, 1); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + + write.write(GenericRow.of(keyBucket0, 10)); + write.write(GenericRow.of(keyBucket1, 20)); + write.compact(BinaryRow.EMPTY_ROW, 0, true); + write.compact(BinaryRow.EMPTY_ROW, 1, true); + commit.commit(1, write.prepareCommit(true, 1)); + + // rewrite the shared manifest: bucket 0's entry becomes DELETED, bucket 1's is + // carried as EXISTING behind it + write.write(GenericRow.ofKind(RowKind.DELETE, keyBucket0, 10)); + write.compact(BinaryRow.EMPTY_ROW, 0, true); + commit.commit(2, write.prepareCommit(true, 2)); + + long rewriteSnapshot = table.snapshotManager().latestSnapshotId(); + IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergManifestList listReader = IcebergManifestList.create(table, paths); + IcebergManifestFile entryReader = IcebergManifestFile.create(table, paths); + String listName = + new Path( + readIcebergMetadata(table, rewriteSnapshot) + .currentSnapshot() + .manifestList()) + .getName(); + + // strip lineage from the manifest with the DELETED entry and from the list, + // simulating metadata written before manifest-level assignment existed + FileStoreTable v2SchemaView = + table.copy(Collections.singletonMap(IcebergOptions.FORMAT_VERSION.key(), "2")); + IcebergManifestFile oldEntryWriter = IcebergManifestFile.create(v2SchemaView, paths); + String strippedPath = null; + for (IcebergManifestFileMeta meta : listReader.read(listName)) { + if (meta.content() != IcebergManifestFileMeta.Content.DATA + || meta.deletedFilesCount() == 0) { + continue; + } + strippedPath = meta.manifestPath(); + List entries = entryReader.read(meta); + assertThat(entries.get(0).status()).isEqualTo(IcebergManifestEntry.Status.DELETED); + List rewritten = + oldEntryWriter.rollingWrite( + entries.iterator(), meta.sequenceNumber(), meta.content()); + Path target = new Path(meta.manifestPath()); + LocalFileIO.create().deleteQuietly(target); + LocalFileIO.create().rename(new Path(rewritten.get(0).manifestPath()), target); + } + assertThat(strippedPath).isNotNull(); + final String stripped = strippedPath; + IcebergManifestList oldListWriter = IcebergManifestList.create(v2SchemaView, paths); + String rewrittenList = oldListWriter.writeWithoutRolling(listReader.read(listName)); + LocalFileIO.create().deleteQuietly(paths.toManifestListPath(listName)); + LocalFileIO.create() + .rename( + paths.toManifestListPath(rewrittenList), + paths.toManifestListPath(listName)); + + // next commit re-assigns the stripped manifest at list level + write.write(GenericRow.of(keyBucket0, 11)); + commit.commit(3, write.prepareCommit(false, 3)); + write.close(); + commit.close(); + + long latest = table.snapshotManager().latestSnapshotId(); + Long assignedFirst = null; + long upperBoundAdvance = 0; + String latestList = + new Path(readIcebergMetadata(table, latest).currentSnapshot().manifestList()) + .getName(); + for (IcebergManifestFileMeta meta : listReader.read(latestList)) { + if (meta.manifestPath().equals(strippedPath)) { + assignedFirst = meta.firstRowId(); + upperBoundAdvance = meta.addedRowsCount() + meta.existingRowsCount(); + } + } + assertThat(assignedFirst).isNotNull(); + // the manifest reserves only its ADDED+EXISTING rows; the DELETED entry is excluded + assertThat(upperBoundAdvance).isEqualTo(1L); + + // the live entry inherits the manifest's own first_row_id, not shifted by the + // 1-row DELETED entry sitting before it + Map effective = effectiveFileFirstRowIds(table, latest); + boolean sawSurvivor = false; + for (IcebergManifestEntry entry : + entryReader.read( + listReader.read(latestList).stream() + .filter(m -> m.manifestPath().equals(stripped)) + .findFirst() + .get())) { + if (entry.isLive()) { + sawSurvivor = true; + assertThat(entry.file().firstRowId()).isNull(); + assertThat(effective.get(entry.file().filePath())).isEqualTo(assignedFirst); + } + } + assertThat(sawSurvivor).isTrue(); + + // GA reader cross-check: Iceberg 1.11 resolves the same id for the live file + if (GA_ROW_LINEAGE_READER) { + HadoopCatalog icebergCatalog = + new HadoopCatalog(new Configuration(), tempDir.toString()); + Table icebergTable = icebergCatalog.loadTable(TableIdentifier.of("mydb2.db", "t")); + boolean checked = false; + for (ManifestFile manifest : + icebergTable.currentSnapshot().dataManifests(icebergTable.io())) { + if (!manifest.path().equals(strippedPath)) { + continue; + } + try (ManifestReader gaReader = + ManifestFiles.read(manifest, icebergTable.io(), icebergTable.specs())) { + for (DataFile file : gaReader) { + checked = true; + assertThat(dataFileFirstRowId(file)).isEqualTo(assignedFirst); + } + } + } + assertThat(checked).isTrue(); + } + } + + @Test + public void testGaReaderSeesAssignedManifests() throws Exception { + assumeGaRowLineageReader(); + FileStoreTable table = createPaimonTable(defaultRowType(), formatVersionOptions(3), "avro"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + write.write(GenericRow.of(2, 20)); + commit.commit(1, write.prepareCommit(false, 1)); + write.close(); + commit.close(); + + HadoopCatalog icebergCatalog = new HadoopCatalog(new Configuration(), tempDir.toString()); + Table icebergTable = icebergCatalog.loadTable(TableIdentifier.of("mydb.db", "t")); + assertThat(icebergTable.currentSnapshot().firstRowId()).isEqualTo(0L); + for (ManifestFile manifest : + icebergTable.currentSnapshot().dataManifests(icebergTable.io())) { + assertThat(manifestFirstRowId(manifest)).isNotNull(); + } + } + + @Test + public void testLayer1TableUpgradesOnNextCommit() throws Exception { + FileStoreTable table = createPaimonTable(defaultRowType(), formatVersionOptions(3), "avro"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + + // simulate a Layer-1-written manifest list: strip the assigned first_row_id + IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergManifestList manifestList = IcebergManifestList.create(table, paths); + String listName = + new Path(readIcebergMetadata(table, 1).currentSnapshot().manifestList()).getName(); + List stripped = new ArrayList<>(); + for (IcebergManifestFileMeta meta : manifestList.read(listName)) { + stripped.add( + new IcebergManifestFileMeta( + meta.manifestPath(), + meta.manifestLength(), + meta.partitionSpecId(), + meta.content(), + meta.sequenceNumber(), + meta.minSequenceNumber(), + meta.addedSnapshotId(), + meta.addedFilesCount(), + meta.existingFilesCount(), + meta.deletedFilesCount(), + meta.addedRowsCount(), + meta.existingRowsCount(), + meta.deletedRowsCount(), + meta.partitions(), + null)); + } + // overwrite the manifest list in place with unassigned metas + LocalFileIO.create().deleteQuietly(paths.toManifestListPath(listName)); + // writeWithoutRolling creates a new file; rename it over the original + String rewritten = manifestList.writeWithoutRolling(stripped); + LocalFileIO.create() + .rename(paths.toManifestListPath(rewritten), paths.toManifestListPath(listName)); + assertThat(manifestList.read(listName).get(0).firstRowId()).isNull(); + // the stripped Layer-1 manifest has no existing/deleted entries, so its addedRowsCount() + // is an exact count of the legacy rows it carries and still needs a real id for + long legacyManifestRows = stripped.get(0).addedRowsCount(); + assertThat(legacyManifestRows).isEqualTo(1L); + + // next commit re-lists carried-over manifests: unassigned metas get assigned now + write.write(GenericRow.of(2, 20)); + commit.commit(2, write.prepareCommit(false, 2)); + + IcebergMetadata metadataAfterCommit2 = readIcebergMetadata(table, 2); + IcebergSnapshot snapshotAfterCommit2 = metadataAfterCommit2.currentSnapshot(); + List dataManifestsAfterCommit2 = new ArrayList<>(); + for (IcebergManifestFileMeta meta : + manifestList.read(new Path(snapshotAfterCommit2.manifestList()).getName())) { + if (meta.content() == IcebergManifestFileMeta.Content.DATA) { + assertThat(meta.firstRowId()).isNotNull(); + dataManifestsAfterCommit2.add(meta); + } + } + // the re-assigned legacy manifest (M1) and this commit's freshly written manifest (M2) + // stay separate: only 2 data manifests, well under the metadata-compaction threshold + assertThat(dataManifestsAfterCommit2).hasSize(2); + + // this is the corruption scenario from the review: added-rows/next-row-id must count + // the legacy manifest's re-assigned rows in addition to this commit's own new rows, not + // just this commit's `metrics.addedRecords` (which is only the new row) + long newRowsThisCommit = 1L; + long expectedAddedRows = legacyManifestRows + newRowsThisCommit; + assertThat(snapshotAfterCommit2.addedRows()) + .as("snapshot added-rows must include the re-assigned legacy manifest's rows") + .isEqualTo(expectedAddedRows); + assertThat(metadataAfterCommit2.nextRowId()) + .as("next-row-id must equal first-row-id plus the true assigned-rows total") + .isEqualTo(snapshotAfterCommit2.firstRowId() + snapshotAfterCommit2.addedRows()); + for (IcebergManifestFileMeta meta : dataManifestsAfterCommit2) { + assertThat(metadataAfterCommit2.nextRowId()) + .as( + "next-row-id must be at or beyond every manifest's assigned range " + + "(manifest %s)", + meta.manifestPath()) + .isGreaterThanOrEqualTo(meta.firstRowId() + meta.addedRowsCount()); + } + + // a third commit: its first-row-id must continue exactly where commit 2 left off, and no + // manifest's assigned id range may overlap another's (i.e. no file gets a duplicate + // effective first-row-id) + write.write(GenericRow.of(3, 30)); + commit.commit(3, write.prepareCommit(false, 3)); + write.close(); + commit.close(); + + IcebergMetadata metadataAfterCommit3 = readIcebergMetadata(table, 3); + IcebergSnapshot snapshotAfterCommit3 = metadataAfterCommit3.currentSnapshot(); + assertThat(snapshotAfterCommit3.firstRowId()).isEqualTo(metadataAfterCommit2.nextRowId()); + + Map effectiveIdsAfterCommit3 = effectiveFileFirstRowIds(table, 3); + assertThat(effectiveIdsAfterCommit3).hasSize(3); + assertThat(new HashSet<>(effectiveIdsAfterCommit3.values())) + .as("no two files may share an effective first-row-id") + .hasSameSizeAs(effectiveIdsAfterCommit3.values()); + } + + @Test + public void testRowTrackingTableUsesSyntheticIds() throws Exception { + Map options = formatVersionOptions(3); + options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true"); + FileStoreTable table = createPaimonTable(defaultRowType(), options, "avro"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + write.write(GenericRow.of(2, 20)); + commit.commit(1, write.prepareCommit(false, 1)); + write.close(); + commit.close(); + + // documented independence: assignment behaves exactly as for any other table + IcebergMetadata metadata = readIcebergMetadata(table, 1); + assertThat(metadata.nextRowId()).isEqualTo(2L); + assertThat(effectiveFileFirstRowIds(table, 1).values()).containsExactly(0L); + } + + @Test + public void testCompactMetadataIfNeededMaterializesRowLineageUnderV3() throws Exception { + // exercises the `compactMetadataIfNeeded` manifest-metadata-merge call site under v3, + // which no existing test hits (all format-version-3 tests leave COMPACT_MIN_FILE_NUM + // at its default of 10, and all tests that force compaction stay on format version 2). + RowType rowType = defaultRowType(); + Map customOptions = formatVersionOptions(3); + customOptions.put(IcebergOptions.COMPACT_MIN_FILE_NUM.key(), "2"); + customOptions.put(IcebergOptions.COMPACT_MAX_FILE_NUM.key(), "2"); + // large enough that manifests are never excluded as "already big enough", so the + // min/max file-count thresholds above are what actually triggers the merge + customOptions.put(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "64 mb"); + FileStoreTable table = createPaimonTable(rowType, customOptions, "avro"); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + + // commit 1: single manifest M1 (candidates=1 < COMPACT_MIN_FILE_NUM, no compaction) + write.write(GenericRow.of(1, 10)); + write.write(GenericRow.of(2, 20)); + commit.commit(1, write.prepareCommit(false, 1)); + Map idsAfterCommit1 = + effectiveFileFirstRowIds(table, table.snapshotManager().latestSnapshotId()); + assertThat(idsAfterCommit1).hasSize(1); + + // commit 2: M1 (already assigned) + fresh M2 => 2 candidates, meets both thresholds, + // manifest metadata compaction merges them into a single manifest this same commit + write.write(GenericRow.of(3, 30)); + commit.commit(2, write.prepareCommit(false, 2)); + assertThat(dataManifestCount(table, table.snapshotManager().latestSnapshotId())) + .as("commit 2 should have merged M1+M2 into a single data manifest") + .isEqualTo(1); + Map idsAfterCommit2 = + effectiveFileFirstRowIds(table, table.snapshotManager().latestSnapshotId()); + assertThat(idsAfterCommit2).hasSize(2); + for (Map.Entry e : idsAfterCommit1.entrySet()) { + // every file's effective first-row-id survives the metadata-compaction commit + // unchanged, whether it was inherited or already explicit before the merge + assertThat(idsAfterCommit2) + .as("file %s", e.getKey()) + .containsEntry(e.getKey(), e.getValue()); + } + assertMergedManifestExistingEntriesHaveExplicitFirstRowId(table); + + // commit 3: merges again, this time the base manifest already contains an + // EXISTING entry with an explicit field 142 from commit 2's merge (file 1/2) sitting + // alongside an ADDED entry with inherited-only field 142 (file from commit 2) -- this + // is the "explicit-142 passthrough" branch of materializeFirstRowIds that no other + // test reaches + write.write(GenericRow.of(4, 40)); + commit.commit(3, write.prepareCommit(false, 3)); + write.close(); + commit.close(); + assertThat(dataManifestCount(table, table.snapshotManager().latestSnapshotId())) + .as("commit 3 should merge again into a single data manifest") + .isEqualTo(1); + Map idsAfterCommit3 = + effectiveFileFirstRowIds(table, table.snapshotManager().latestSnapshotId()); + assertThat(idsAfterCommit3).hasSize(3); + for (Map.Entry e : idsAfterCommit2.entrySet()) { + assertThat(idsAfterCommit3) + .as("file %s", e.getKey()) + .containsEntry(e.getKey(), e.getValue()); + } + assertMergedManifestExistingEntriesHaveExplicitFirstRowId(table); + } + + @Test + public void testGaReaderResolvesPerFileRowLineage() throws Exception { + assumeGaRowLineageReader(); + // standard two-commit 2+1-row setup (matches testNextRowIdAdvancesAcrossCommits): + // commit 1's file gets effective first-row-id 0, commit 2's file gets 2 + FileStoreTable table = createPaimonTable(defaultRowType(), formatVersionOptions(3), "avro"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + write.write(GenericRow.of(2, 20)); + commit.commit(1, write.prepareCommit(false, 1)); + write.write(GenericRow.of(3, 30)); + commit.commit(2, write.prepareCommit(false, 2)); + write.close(); + commit.close(); + + HadoopCatalog icebergCatalog = new HadoopCatalog(new Configuration(), tempDir.toString()); + Table icebergTable = icebergCatalog.loadTable(TableIdentifier.of("mydb.db", "t")); + + // Attempt 1: resolve a per-row `_row_id` value through iceberg-data's GA generics + // reader. `select(...)` does not throw, but the requested metadata column is silently + // dropped from the projected schema: GenericReader/InternalRecordWrapper in + // iceberg-data 1.11.0 have no wiring for MetadataColumns.ROW_ID (unlike _pos/_file/ + // _deleted/_spec_id, which the same reader stack does resolve), so every record's + // "_row_id" field comes back null instead of the assigned/inherited value. This is + // verified here rather than assumed: if a future Iceberg release adds real support, + // this loop will start observing non-null values and the assertion below must change. + boolean anyRowIdResolved = false; + try (CloseableIterable records = + IcebergGenerics.read(icebergTable).select("k", "v", "_row_id").build()) { + for (Record record : records) { + if (record.getField("_row_id") != null) { + anyRowIdResolved = true; + } + } + } + assertThat(anyRowIdResolved) + .as( + "iceberg-data 1.11.0 GA generics do not resolve the _row_id metadata " + + "column; if this starts failing, generics gained support and " + + "the fallback below can be simplified/removed") + .isFalse(); + + // Fallback: GA's ManifestFiles/DataFile APIs DO resolve the per-file first_row_id + // (including inheritance from the manifest-level value), so assert actual values, + // not just non-nullity. + List resolvedFirstRowIds = new ArrayList<>(); + for (ManifestFile manifest : + icebergTable.currentSnapshot().dataManifests(icebergTable.io())) { + try (ManifestReader reader = + ManifestFiles.read(manifest, icebergTable.io(), icebergTable.specs())) { + for (DataFile file : reader) { + resolvedFirstRowIds.add(dataFileFirstRowId(file)); + } + } + } + assertThat(resolvedFirstRowIds).containsExactlyInAnyOrder(0L, 2L); + } + + @Test + public void testV2ManifestListSchemaHasNoFirstRowIdColumn() throws Exception { + // pins the v2 manifest-list shape: 14 columns, no first_row_id anywhere, so a future + // change to the v3 schema construction cannot silently leak into v2's byte-identical + // output + assertThat(IcebergManifestFileMeta.schema(false).getFieldCount()).isEqualTo(14); + assertThat(IcebergManifestFileMeta.schema(false).getFields().stream().map(DataField::name)) + .doesNotContain("first_row_id"); + assertThat(IcebergManifestFileMeta.schema(true).getFieldCount()).isEqualTo(14); + assertThat(IcebergManifestFileMeta.schema(true).getFields().stream().map(DataField::name)) + .doesNotContain("first_row_id"); + + FileStoreTable table = createPaimonTable(defaultRowType(), formatVersionOptions(2), "avro"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + write.close(); + commit.close(); + + // the written manifest-list file's raw bytes must not contain "first_row_id" anywhere; + // Avro embeds the writer schema as JSON in the file header, so a plain byte-scan of the + // whole file is a valid (and stronger-than-parsed) check of the physical shape + Path manifestListPath = + new Path(readIcebergMetadata(table, 1).currentSnapshot().manifestList()); + byte[] bytes; + try (SeekableInputStream in = table.fileIO().newInputStream(manifestListPath)) { + bytes = IOUtils.readFully(in, false); + } + String content = new String(bytes, StandardCharsets.ISO_8859_1); + assertThat(content).doesNotContain("first_row_id"); + } + // ------------------------------------------------------------------------ // helpers // ------------------------------------------------------------------------ @@ -330,6 +1128,135 @@ private FileStoreTable createPaimonTable( } } + private FileStoreTable createPkPaimonTable(RowType rowType, Map customOptions) + throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path path = new Path(tempDir.toString()); + Options options = new Options(customOptions); + // two fixed buckets so a manifest rewrite can leave one bucket's file untouched + // (see testFirstRowIdStableAcrossManifestRewrite) + options.set(CoreOptions.BUCKET, 2); + options.set( + IcebergOptions.METADATA_ICEBERG_STORAGE, IcebergOptions.StorageType.TABLE_LOCATION); + options.set(CoreOptions.FILE_FORMAT, "avro"); + Schema schema = + new Schema( + rowType.getFields(), + Collections.emptyList(), + Collections.singletonList("k"), + options.toMap(), + ""); + try (FileSystemCatalog paimonCatalog = new FileSystemCatalog(fileIO, path)) { + paimonCatalog.createDatabase("mydb2", false); + Identifier id = Identifier.create("mydb2", "t"); + paimonCatalog.createTable(id, schema, false); + return (FileStoreTable) paimonCatalog.getTable(id); + } + } + + /** Finds the smallest positive key whose fixed-bucket hash lands in {@code targetBucket}. */ + private int findKeyForBucket(FileStoreTable table, int targetBucket) { + FixedBucketRowKeyExtractor extractor = new FixedBucketRowKeyExtractor(table.schema()); + for (int k = 1; k < 10_000; k++) { + extractor.setRecord(GenericRow.of(k, 0)); + if (extractor.bucket() == targetBucket) { + return k; + } + } + throw new IllegalStateException("No key found for bucket " + targetBucket); + } + + /** Effective per-file first row id: explicit field 142, or inherited per the spec rules. */ + private Map effectiveFileFirstRowIds(FileStoreTable table, long snapshotId) + throws Exception { + IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergManifestList manifestList = IcebergManifestList.create(table, paths); + IcebergManifestFile manifestFile = IcebergManifestFile.create(table, paths); + Map result = new HashMap<>(); + for (IcebergManifestFileMeta meta : + manifestList.read( + new Path( + readIcebergMetadata(table, snapshotId) + .currentSnapshot() + .manifestList()) + .getName())) { + if (meta.content() != IcebergManifestFileMeta.Content.DATA) { + continue; + } + long watermark = meta.firstRowId() == null ? -1 : meta.firstRowId(); + for (IcebergManifestEntry entry : manifestFile.read(meta)) { + if (entry.status() == IcebergManifestEntry.Status.DELETED) { + // GA readers never assign ids to DELETED entries; they must not + // advance the inheritance walk either + continue; + } + long effective; + if (entry.file().firstRowId() != null) { + effective = entry.file().firstRowId(); + } else { + effective = watermark; + watermark += entry.file().recordCount(); + } + if (entry.isLive()) { + result.put(entry.file().filePath(), effective); + } + } + } + return result; + } + + /** Number of DATA-content manifests referenced by the given snapshot's manifest list. */ + private int dataManifestCount(FileStoreTable table, long snapshotId) throws Exception { + return dataManifestMetas(table, snapshotId).size(); + } + + private List dataManifestMetas(FileStoreTable table, long snapshotId) + throws Exception { + IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergManifestList manifestList = IcebergManifestList.create(table, paths); + List result = new ArrayList<>(); + for (IcebergManifestFileMeta meta : + manifestList.read( + new Path( + readIcebergMetadata(table, snapshotId) + .currentSnapshot() + .manifestList()) + .getName())) { + if (meta.content() == IcebergManifestFileMeta.Content.DATA) { + result.add(meta); + } + } + return result; + } + + /** + * After a manifest-metadata-compaction merge, every EXISTING entry in the merged manifest(s) + * for the table's current snapshot must carry an explicit (non-null) field 142: EXISTING + * entries are, by definition, carried-over/rewritten entries, so their first_row_id must have + * been materialized by {@code materializeFirstRowIds} rather than left to be inherited from the + * (now-merged-away) original manifest. + */ + private void assertMergedManifestExistingEntriesHaveExplicitFirstRowId(FileStoreTable table) + throws Exception { + IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergManifestFile manifestFile = IcebergManifestFile.create(table, paths); + long snapshotId = table.snapshotManager().latestSnapshotId(); + boolean checkedAtLeastOneExistingEntry = false; + for (IcebergManifestFileMeta meta : dataManifestMetas(table, snapshotId)) { + for (IcebergManifestEntry entry : manifestFile.read(meta)) { + if (entry.status() == IcebergManifestEntry.Status.EXISTING) { + checkedAtLeastOneExistingEntry = true; + assertThat(entry.file().firstRowId()) + .as("EXISTING entry for file %s", entry.file().filePath()) + .isNotNull(); + } + } + } + assertThat(checkedAtLeastOneExistingEntry) + .as("expected at least one materialized EXISTING entry after the merge") + .isTrue(); + } + private Path metadataPath(FileStoreTable table, long snapshotId) { return new Path(table.location(), String.format("metadata/v%d.metadata.json", snapshotId)); } @@ -341,4 +1268,42 @@ private IcebergMetadata readIcebergMetadata(FileStoreTable table, long snapshotI private String readMetadataJson(FileStoreTable table, long snapshotId) throws Exception { return LocalFileIO.create().readFileUtf8(metadataPath(table, snapshotId)); } + + /** + * Iceberg exposes per-manifest / per-file {@code firstRowId()} only from the GA row-lineage + * line (1.10+). The module compiles against 1.8.1 by default, so GA reader assertions look the + * method up reflectively and the tests skip when the API is absent (run with -Piceberg-ga). + */ + private static final boolean GA_ROW_LINEAGE_READER = detectGaRowLineageReader(); + + private static boolean detectGaRowLineageReader() { + try { + ManifestFile.class.getMethod("firstRowId"); + return true; + } catch (NoSuchMethodException e) { + return false; + } + } + + private static void assumeGaRowLineageReader() { + Assumptions.assumeTrue( + GA_ROW_LINEAGE_READER, + "Iceberg on the test classpath predates GA row lineage; run with -Piceberg-ga"); + } + + private static Long manifestFirstRowId(ManifestFile manifest) { + try { + return (Long) ManifestFile.class.getMethod("firstRowId").invoke(manifest); + } catch (ReflectiveOperationException e) { + throw new IllegalStateException(e); + } + } + + private static Long dataFileFirstRowId(DataFile file) { + try { + return (Long) DataFile.class.getMethod("firstRowId").invoke(file); + } catch (ReflectiveOperationException e) { + throw new IllegalStateException(e); + } + } } diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergMetadataTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergMetadataTest.java index a6205dc107d5..79a2909df18c 100644 --- a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergMetadataTest.java +++ b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergMetadataTest.java @@ -389,10 +389,18 @@ void testReadMetadataWithIcebergPartitioningEvolution() throws Exception { void testFormatVersionV3Table() throws Exception { // Create a v3 format version Iceberg table Table icebergTable = createIcebergTableV3("v3_snapshot_table"); - TableMetadata base = ((HasTableOperations) icebergTable).operations().current(); - ((HasTableOperations) icebergTable) - .operations() - .commit(base, TableMetadata.buildFrom(base).enableRowLineage().build()); + try { + // pre-GA Iceberg (< 1.10) required opting into v3 row lineage; the builder + // method was removed in GA, where row lineage is always on for v3 + java.lang.reflect.Method enableRowLineage = + TableMetadata.Builder.class.getMethod("enableRowLineage"); + TableMetadata base = ((HasTableOperations) icebergTable).operations().current(); + TableMetadata.Builder builder = TableMetadata.buildFrom(base); + enableRowLineage.invoke(builder); + ((HasTableOperations) icebergTable).operations().commit(base, builder.build()); + } catch (NoSuchMethodException e) { + // GA Iceberg: nothing to opt into + } // Read metadata using Paimon's IcebergMetadata IcebergMetadata paimonIcebergMetadata = readIcebergMetadata(icebergTable); diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java index f374c8c119d4..7fd3e5e17065 100644 --- a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java +++ b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java @@ -28,6 +28,8 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.iceberg.manifest.IcebergManifestFileMeta; +import org.apache.paimon.iceberg.manifest.IcebergManifestList; import org.apache.paimon.iceberg.metadata.IcebergMetadata; import org.apache.paimon.iceberg.metadata.IcebergSnapshot; import org.apache.paimon.options.MemorySize; @@ -43,9 +45,11 @@ import org.apache.iceberg.BaseTable; import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.ManifestFile; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; import org.apache.iceberg.Table; +import org.apache.iceberg.TableMetadata; import org.apache.iceberg.TableUtil; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; @@ -54,6 +58,7 @@ import org.apache.iceberg.hadoop.HadoopCatalog; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.rest.Endpoint; import org.apache.iceberg.rest.RESTCatalog; import org.apache.iceberg.rest.RESTCatalogServer; import org.apache.iceberg.rest.RESTServerExtension; @@ -68,9 +73,11 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ThreadLocalRandom; import java.util.function.BiFunction; @@ -79,6 +86,7 @@ import static org.apache.paimon.iceberg.IcebergCommitCallback.catalogTableMetadataPath; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test for {@link IcebergRestMetadataCommitter}. */ public class IcebergRestMetadataCommitterTest { @@ -1348,6 +1356,277 @@ public void testRecreateWithNonZeroLineageWatermark() throws Exception { // Known Layer 1 limitation (documented in the spec): the server-side next-row-id // watermark restarts on recreation and stays behind the local metadata; commits // must keep succeeding regardless (validation is first-row-id >= next-row-id). + + // reader-visible lineage from the REST catalog matches the file-based mirror: + // snapshot first-row-id and manifest assignments come from local metadata, never + // from the server's table-level watermark. Compare by value, not just non-nullity, + // against the locally-written IcebergMetadata + manifest list under the paimon + // table's own metadata dir (catalogTableMetadataPath), which is the source of truth + // the REST-registered table's metadata-location actually points at. + long latestSnapshotId = table.snapshotManager().latestSnapshotId(); + IcebergMetadata localMetadata = + IcebergMetadata.fromPath( + table.fileIO(), + new Path( + catalogTableMetadataPath(table), + String.format("v%d.metadata.json", latestSnapshotId))); + IcebergSnapshot localSnapshot = localMetadata.currentSnapshot(); + assertThat(localSnapshot.firstRowId()).isNotNull(); + + IcebergPathFactory pathFactory = new IcebergPathFactory(catalogTableMetadataPath(table)); + IcebergManifestList localManifestList = IcebergManifestList.create(table, pathFactory); + List localDataManifestFirstRowIds = + localManifestList.read(new Path(localSnapshot.manifestList()).getName()).stream() + .filter(m -> m.content() == IcebergManifestFileMeta.Content.DATA) + .map(IcebergManifestFileMeta::firstRowId) + .collect(Collectors.toList()); + assertThat(localDataManifestFirstRowIds).isNotEmpty().doesNotContainNull(); + + Table reloaded = restCatalog.loadTable(TableIdentifier.of("mydb", "t")); + assertThat(reloaded.currentSnapshot().firstRowId()).isEqualTo(localSnapshot.firstRowId()); + // manifest-level first_row_id is only exposed by the GA (1.10+) reader API + if (GA_ROW_LINEAGE_READER) { + List restDataManifestFirstRowIds = new ArrayList<>(); + for (ManifestFile manifest : reloaded.currentSnapshot().dataManifests(reloaded.io())) { + assertThat(manifestFirstRowId(manifest)).isNotNull(); + restDataManifestFirstRowIds.add(manifestFirstRowId(manifest)); + } + assertThat(restDataManifestFirstRowIds) + .containsExactlyInAnyOrderElementsOf(localDataManifestFirstRowIds); + + // the registered server table must carry the full row-id high-water mark: + // external REST writers allocate from it, so anything lower reuses ids + Long serverNextRowId = + tableMetadataNextRowId(((BaseTable) reloaded).operations().current()); + assertThat(serverNextRowId).isNotNull(); + assertThat(serverNextRowId).isGreaterThanOrEqualTo(localMetadata.nextRowId()); + + // one external append through the Iceberg API allocates above the watermark + reloaded.newAppend().commit(); + Table afterExternal = restCatalog.loadTable(TableIdentifier.of("mydb", "t")); + assertThat(afterExternal.currentSnapshot().firstRowId()) + .isGreaterThanOrEqualTo(localMetadata.nextRowId()); + } + } + + /** + * Some REST catalogs (AWS Glue's Iceberg REST endpoint, for one) implement create, load, update + * and delete but not registerTable. Simulated by removing the endpoint from the set the client + * took from the server's config response. + */ + @Test + public void testWithoutRegisterTableEndpoint() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + Map customOptions = new HashMap<>(); + customOptions.put(IcebergOptions.FORMAT_VERSION.key(), "3"); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.emptyList(), + -1, + "avro", + customOptions); + // Iceberg metadata is only produced locally; the committer under test publishes it + FileStoreTable localTable = + table.copy( + Collections.singletonMap( + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "table-location")); + TableIdentifier identifier = TableIdentifier.of("mydb", "t"); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = localTable.newWrite(commitUser); + TableCommitImpl commit = localTable.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + write.write(GenericRow.of(2, 20)); + commit.commit(1, write.prepareCommit(true, 1)); + write.write(GenericRow.of(3, 30)); + commit.commit(2, write.prepareCommit(true, 2)); + IcebergMetadata v1 = localMetadata(localTable, 1); + IcebergMetadata v2 = localMetadata(localTable, 2); + + IcebergRestMetadataCommitter committer = new IcebergRestMetadataCommitter(table); + removeRegisterTableEndpoint(committer); + + // zero-based v3 metadata publishes through the create/update path every catalog has + assertThat(v1.currentSnapshot().firstRowId()).isEqualTo(0L); + committer.commitMetadata(v1, null); + committer.commitMetadata(v2, v1); + Table icebergTable = restCatalog.loadTable(identifier); + assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(2); + assertThat(TableUtil.formatVersion(icebergTable)).isEqualTo(3); + assertThat(getIcebergResult()) + .containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)", "Record(3, 30)"); + + // a rebuild whose row-id space does not start at 0 can only be published by + // registration: fail closed and leave the catalog table untouched + write.write(GenericRow.of(4, 40)); + commit.commit(3, write.prepareCommit(true, 3)); + IcebergMetadata v3 = localMetadata(localTable, 3); + assertThat(v3.currentSnapshot().firstRowId()).isEqualTo(3L); + assertThatThrownBy(() -> committer.commitMetadata(v3, null)) + .hasStackTraceContaining("does not advertise registerTable"); + assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(2); + assertThat(registerFiles(localTable)).isEmpty(); + + // the same metadata is a plain update when the base is correct + committer.commitMetadata(v3, v2); + assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(3); + + // recreating a lost table needs registration as well: nothing is created + restCatalog.dropTable(identifier, false); + write.write(GenericRow.of(5, 50)); + commit.commit(4, write.prepareCommit(true, 4)); + IcebergMetadata v4 = localMetadata(localTable, 4); + assertThatThrownBy(() -> committer.commitMetadata(v4, v3)) + .hasStackTraceContaining("does not advertise registerTable"); + assertThat(restCatalog.tableExists(identifier)).isFalse(); + + write.close(); + commit.close(); + } + + @Test + public void testRecreateWithoutRegisterTableEndpointOnV2() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.emptyList(), + -1, + "avro", + Collections.emptyMap()); + FileStoreTable localTable = + table.copy( + Collections.singletonMap( + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "table-location")); + TableIdentifier identifier = TableIdentifier.of("mydb", "t"); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = localTable.newWrite(commitUser); + TableCommitImpl commit = localTable.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(true, 1)); + write.write(GenericRow.of(2, 20)); + commit.commit(2, write.prepareCommit(true, 2)); + write.close(); + commit.close(); + IcebergMetadata v1 = localMetadata(localTable, 1); + IcebergMetadata v2 = localMetadata(localTable, 2); + + IcebergRestMetadataCommitter committer = new IcebergRestMetadataCommitter(table); + removeRegisterTableEndpoint(committer); + + committer.commitMetadata(v1, null); + assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(1); + // no base: the table is dropped and recreated, which needs no registration on v2 + committer.commitMetadata(v2, null); + Table icebergTable = restCatalog.loadTable(identifier); + assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(2); + assertThat(TableUtil.formatVersion(icebergTable)).isEqualTo(2); + assertThat(getIcebergResult()).containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)"); + assertThat(registerFiles(localTable)).isEmpty(); + } + + /** + * A catalog may keep referencing the registered metadata location, and rollback and rebuild + * paths reuse Paimon snapshot ids, so every registration writes a new file and never replaces + * an existing one. + */ + @Test + public void testRegisteredMetadataFilesAreWriteOnce() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + Map customOptions = new HashMap<>(); + customOptions.put(IcebergOptions.FORMAT_VERSION.key(), "3"); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.emptyList(), + -1, + "avro", + customOptions); + TableIdentifier identifier = TableIdentifier.of("mydb", "t"); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(true, 1)); + write.write(GenericRow.of(2, 20)); + commit.commit(2, write.prepareCommit(true, 2)); + IcebergMetadata v2 = localMetadata(table, 2); + assertThat(registerFiles(table)).isEmpty(); + + // the lost table is recreated by registration + restCatalog.dropTable(identifier, false); + write.write(GenericRow.of(3, 30)); + commit.commit(3, write.prepareCommit(true, 3)); + write.close(); + commit.close(); + IcebergMetadata v3 = localMetadata(table, 3); + assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(3); + List firstRegistration = registerFiles(table); + assertThat(firstRegistration).hasSize(1); + assertThat(firstRegistration.get(0).getName()).startsWith("rest-register-v3-"); + String firstContent = table.fileIO().readFileUtf8(firstRegistration.get(0)); + + // registering the same snapshot id again writes a second file, the first is untouched + restCatalog.dropTable(identifier, false); + new IcebergRestMetadataCommitter(table).commitMetadata(v3, v2); + assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(3); + List registrations = registerFiles(table); + assertThat(registrations).hasSize(2).contains(firstRegistration.get(0)); + assertThat(table.fileIO().readFileUtf8(firstRegistration.get(0))).isEqualTo(firstContent); + for (Path path : registrations) { + assertThat(path.getName()).startsWith("rest-register-v3-"); + assertThat(IcebergMetadata.fromPath(table.fileIO(), path).currentSnapshotId()) + .isEqualTo(3); + } + } + + private static IcebergMetadata localMetadata(FileStoreTable table, long snapshotId) { + return IcebergMetadata.fromPath( + table.fileIO(), + new Path( + catalogTableMetadataPath(table), + String.format("v%d.metadata.json", snapshotId))); + } + + private static List registerFiles(FileStoreTable table) throws Exception { + List files = new ArrayList<>(); + for (org.apache.paimon.fs.FileStatus status : + table.fileIO().listStatus(catalogTableMetadataPath(table))) { + if (status.getPath().getName().startsWith("rest-register-")) { + files.add(status.getPath()); + } + } + return files; + } + + /** Makes the committer's REST client see a server that does not advertise registerTable. */ + private static void removeRegisterTableEndpoint(IcebergRestMetadataCommitter committer) + throws Exception { + java.lang.reflect.Field catalogField = + IcebergRestMetadataCommitter.class.getDeclaredField("restCatalog"); + catalogField.setAccessible(true); + Object restCatalog = catalogField.get(committer); + java.lang.reflect.Field sessionField = RESTCatalog.class.getDeclaredField("sessionCatalog"); + sessionField.setAccessible(true); + Object sessionCatalog = sessionField.get(restCatalog); + java.lang.reflect.Field endpointsField = + sessionCatalog.getClass().getDeclaredField("endpoints"); + endpointsField.setAccessible(true); + Set endpoints = new HashSet<>((Set) endpointsField.get(sessionCatalog)); + assertThat(endpoints.remove(Endpoint.V1_REGISTER_TABLE)).isTrue(); + endpointsField.set(sessionCatalog, endpoints); } private static class TestRecord { @@ -1431,4 +1710,35 @@ private String randomFormat() { String[] formats = new String[] {"orc", "parquet", "avro"}; return formats[i]; } + + /** See IcebergRowLineageCompatibilityTest: GA reader API (1.10+) looked up reflectively. */ + private static final boolean GA_ROW_LINEAGE_READER = detectGaRowLineageReader(); + + private static boolean detectGaRowLineageReader() { + try { + ManifestFile.class.getMethod("firstRowId"); + return true; + } catch (NoSuchMethodException e) { + return false; + } + } + + private static Long manifestFirstRowId(ManifestFile manifest) { + try { + return (Long) ManifestFile.class.getMethod("firstRowId").invoke(manifest); + } catch (ReflectiveOperationException e) { + throw new IllegalStateException(e); + } + } + + /** TableMetadata#nextRowId is a GA (1.10+) API; resolve reflectively. */ + private static Long tableMetadataNextRowId(TableMetadata metadata) { + try { + return (Long) TableMetadata.class.getMethod("nextRowId").invoke(metadata); + } catch (NoSuchMethodException e) { + return null; + } catch (ReflectiveOperationException e) { + throw new IllegalStateException(e); + } + } }