JingsongLi commented on code in PR #9135: URL: https://github.com/apache/paimon/pull/9135#discussion_r3772631706
########## paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java: ########## @@ -0,0 +1,797 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.operation; + +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.serializer.InternalRowSerializer; +import org.apache.paimon.format.avro.AvroRawBlock; +import org.apache.paimon.manifest.CompactFileIdentifierSet; +import org.apache.paimon.manifest.FileEntry.ReusableIdentifier; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.ManifestAvroReader; +import org.apache.paimon.manifest.ManifestAvroReader.RawBlock; +import org.apache.paimon.manifest.ManifestAvroReader.RowIterator; +import org.apache.paimon.manifest.ManifestAvroWriter; +import org.apache.paimon.manifest.ManifestAvroWriter.EncodedBlock; +import org.apache.paimon.manifest.ManifestAvroWriter.EncodedEntry; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.manifest.ManifestFile; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ProjectedManifestEntry; +import org.apache.paimon.utils.CloseableIterator; +import org.apache.paimon.utils.Pair; + +import javax.annotation.Nullable; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.apache.paimon.utils.Preconditions.checkState; + +final class ManifestEntryRunMergePlan { + + final List<Source.Spec> sources; + final ManifestEntryRunMergeEntry.PartitionDictionary partitions; + + ManifestEntryRunMergePlan( + List<Source.Spec> sources, ManifestEntryRunMergeEntry.PartitionDictionary partitions) { + this.sources = sources; + this.partitions = partitions; + } + + List<ManifestFileMeta> mergeToManifest( + ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestFile manifestFile, + ManifestEntryRunMergeEntry.Filter filter, + List<ManifestFileMeta> newFilesForAbort) + throws Exception { + List<Cursor> cursors = new ArrayList<>(sources.size()); + Exception failure = null; + try { + for (Source.Spec source : sources) { + Cursor cursor = source.open(manifestFile, sortKey, filter, partitions); + cursors.add(cursor); + cursor.advance(); + } + SelectionTree selectionTree = new SelectionTree(cursors); + if (selectionTree.winner() < 0) { + return Collections.emptyList(); + } + List<ManifestFileMeta> files = writeSelected(selectionTree, manifestFile); + newFilesForAbort.addAll(files); + return files; + } catch (Exception e) { + failure = e; + throw e; + } finally { + try { + closeCursors(cursors); + } catch (Exception closeFailure) { + if (failure == null) { + throw closeFailure; + } + failure.addSuppressed(closeFailure); + } + } + } + + Pair<List<ManifestFileMeta>, List<ManifestFileMeta>> mergeMinorToManifest( + ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestFile manifestFile, + ManifestEntryRunMergeEntry.Filter filter, + CompactFileIdentifierSet deletedIdentifiers, + ManifestFileSorter.DeletedRowIdSet deletedRowIds, + List<ManifestFileMeta> newFilesForAbort) + throws Exception { + List<Cursor> cursors = new ArrayList<>(sources.size()); + Exception failure = null; + try { + for (Source.Spec source : sources) { + Cursor cursor = source.open(manifestFile, sortKey, filter, partitions); + cursors.add(cursor); + cursor.advance(); + } + SelectionTree selectionTree = new SelectionTree(cursors); + if (selectionTree.winner() < 0) { + return Pair.of(Collections.emptyList(), Collections.emptyList()); + } + Pair<List<ManifestFileMeta>, List<ManifestFileMeta>> files = + writeMinorSelected( + selectionTree, manifestFile, deletedIdentifiers, deletedRowIds); + newFilesForAbort.addAll(files.getLeft()); + newFilesForAbort.addAll(files.getRight()); + return files; + } catch (Exception e) { + failure = e; + throw e; + } finally { + try { + closeCursors(cursors); + } catch (Exception closeFailure) { + if (failure == null) { + throw closeFailure; + } + failure.addSuppressed(closeFailure); + } + } + } + + static List<ManifestFileMeta> writeSelected( + SelectionTree selectionTree, ManifestFile manifestFile) throws Exception { + ManifestAvroWriter writer = manifestFile.createAvroWriter(); + Exception failure = null; + try { + int winner; + while ((winner = selectionTree.winner()) >= 0) { + Cursor cursor = selectionTree.cursor(winner); + if (cursor.hasCopyableBlock() + && selectionTree.blockPrecedesOthers(winner, cursor.blockLastKey())) { + writer.writeEncodedBlock(cursor.encodedBlock(), cursor.blockMetadata()); + selectionTree.update(winner, cursor.advanceAfterBlock()); + continue; + } + cursor.materializeCurrent(); + ByteBuffer encodedRecord = cursor.encodedRecord(); + if (encodedRecord == null) { + writer.write(cursor.current()); + } else { + writer.writeEncoded(encodedRecord, cursor.metadata()); + } + selectionTree.update(winner, cursor.advance()); + } + } catch (Exception e) { + failure = e; + } finally { + if (failure != null) { + writer.abort(); + throw failure; + } + writer.close(); + } + return writer.result(); + } + + private static Pair<List<ManifestFileMeta>, List<ManifestFileMeta>> writeMinorSelected( + SelectionTree selectionTree, + ManifestFile manifestFile, + CompactFileIdentifierSet deletedIdentifiers, + ManifestFileSorter.DeletedRowIdSet deletedRowIds) + throws Exception { + ManifestAvroWriter addWriter = manifestFile.createAvroWriter(); + ManifestAvroWriter deleteWriter = manifestFile.createAvroWriter(); + CompactFileIdentifierSet matchedEntries = new CompactFileIdentifierSet(); + CompactFileIdentifierSet emittedDeletes = new CompactFileIdentifierSet(); + Exception failure = null; + try { + int winner; + while ((winner = selectionTree.winner()) >= 0) { + Cursor cursor = selectionTree.cursor(winner); + if (cursor.hasCopyableBlock() + && selectionTree.blockPrecedesOthers(winner, cursor.blockLastKey())) { + addWriter.writeEncodedBlock(cursor.encodedBlock(), cursor.blockMetadata()); + selectionTree.update(winner, cursor.advanceAfterBlock()); + continue; + } + + cursor.materializeCurrent(); + if (cursor.key().kind == FileKind.ADD.toByteValue()) { + if (!deletedRowIds.contains(cursor.key().firstRowId)) { + writeCurrent(addWriter, cursor); + } else { + ReusableIdentifier identifier = cursor.identifier(); + if (deletedIdentifiers.contains(identifier)) { + matchedEntries.add(identifier); + } else { + writeCurrent(addWriter, cursor); + } + } + } else { + ReusableIdentifier identifier = cursor.identifier(); + if (!matchedEntries.contains(identifier) + && !emittedDeletes.contains(identifier)) { + emittedDeletes.add(identifier); + writeCurrent(deleteWriter, cursor); + } + } + selectionTree.update(winner, cursor.advance()); + } + addWriter.close(); + deleteWriter.close(); + } catch (Exception e) { + failure = e; + } finally { + matchedEntries.release(); + emittedDeletes.release(); + if (failure != null) { + addWriter.abort(); + deleteWriter.abort(); + throw failure; + } + } + return Pair.of(addWriter.result(), deleteWriter.result()); + } + + private static void writeCurrent(ManifestAvroWriter writer, Cursor cursor) throws Exception { + ByteBuffer encodedRecord = cursor.encodedRecord(); + if (encodedRecord == null) { + writer.write(cursor.current()); + } else { + writer.writeEncoded(encodedRecord, cursor.metadata()); + } + } + + static void closeCursors(List<Cursor> cursors) throws Exception { + Exception failure = null; + for (Cursor cursor : cursors) { + try { + cursor.close(); + } catch (Exception e) { + if (failure == null) { + failure = e; + } else { + failure.addSuppressed(e); + } + } + } + if (failure != null) { + throw failure; + } + } + + /** Describes the manifest inputs which become cursors when this plan starts executing. */ + static final class Source { + + private Source() {} + + interface Spec { + + Cursor open( + ManifestFile manifestFile, + ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestEntryRunMergeEntry.Filter filter, + ManifestEntryRunMergeEntry.PartitionDictionary partitions) + throws Exception; + } + + static final class ManifestRunSpec implements Spec { + + final ManifestFileMeta meta; + final long start; + final long end; + final List<ManifestEntryRunMerge.Discovery.BlockInfo> blocks; + + ManifestRunSpec( + ManifestFileMeta meta, + long start, + long end, + List<ManifestEntryRunMerge.Discovery.BlockInfo> blocks) { + this.meta = meta; + this.start = start; + this.end = end; + this.blocks = blocks; + } + + long prefixBlockCount() { + long lastBlockOrdinal = -1; + for (ManifestEntryRunMerge.Discovery.BlockInfo block : blocks) { + if (block.start >= end) { + break; + } + if (block.end > start) { + lastBlockOrdinal = block.ordinal; + } + } + checkState(lastBlockOrdinal >= 0, "Manifest run does not contain an Avro block."); + return lastBlockOrdinal + 1; + } + + @Override + public Cursor open( + ManifestFile manifestFile, + ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestEntryRunMergeEntry.Filter filter, + ManifestEntryRunMergeEntry.PartitionDictionary partitions) + throws Exception { + return new PrimitiveManifestRunCursor( + manifestFile, meta, start, end, blocks, filter, partitions); + } + } + + static final class FragmentedManifestSpec implements Spec { + + final ManifestFileMeta meta; + + FragmentedManifestSpec(ManifestFileMeta meta) { + this.meta = meta; + } + + @Override + public Cursor open( + ManifestFile manifestFile, + ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestEntryRunMergeEntry.Filter filter, + ManifestEntryRunMergeEntry.PartitionDictionary partitions) + throws Exception { + return new InMemoryManifestCursor(manifestFile, meta, sortKey, filter, partitions); + } + } + } + + interface Cursor extends AutoCloseable { + + boolean advance() throws Exception; + + boolean hasCurrent(); + + @Nullable + ProjectedManifestEntry current(); + + @Nullable + EncodedEntry metadata(); + + ManifestEntryRunMergeEntry.Key key(); + + @Nullable + ByteBuffer encodedRecord(); + + ReusableIdentifier identifier(); + + default boolean hasCopyableBlock() { + return false; + } + + default ManifestEntryRunMergeEntry.Key blockLastKey() { + throw new UnsupportedOperationException(); + } + + default AvroRawBlock encodedBlock() { + throw new UnsupportedOperationException(); + } + + default EncodedBlock blockMetadata() { + throw new UnsupportedOperationException(); + } + + default boolean advanceAfterBlock() throws Exception { + throw new UnsupportedOperationException(); + } + + default void materializeCurrent() throws Exception {} + + @Override + void close() throws Exception; + } + + static final class PrimitiveManifestRunCursor implements Cursor { + + final ManifestAvroReader reader; + final ManifestEntryRunMergeEntry.Filter filter; + final ManifestEntryRunMergeEntry.PartitionDictionary partitions; + final ManifestEntryRunMergeEntry.Key key = new ManifestEntryRunMergeEntry.Key(); + final EncodedEntry metadata = new EncodedEntry(); + final List<ManifestEntryRunMerge.Discovery.BlockInfo> blocks; + final long runStart; + final long runEnd; + int blockIndex; + long nextReaderBlockOrdinal; + long decodedRemaining; + boolean rawBlock; + boolean current; + @Nullable RawBlock currentRawBlock; + @Nullable RowIterator currentRows; + @Nullable GenericRow currentRow; + @Nullable ManifestEntryRunMerge.Discovery.BlockInfo currentBlock; + boolean closed; + + PrimitiveManifestRunCursor( + ManifestFile manifestFile, + ManifestFileMeta meta, + long start, + long end, + List<ManifestEntryRunMerge.Discovery.BlockInfo> blocks, + ManifestEntryRunMergeEntry.Filter filter, + ManifestEntryRunMergeEntry.PartitionDictionary partitions) + throws Exception { + this.reader = manifestFile.scanForRunMerge(meta.fileName(), meta.fileSize()); + this.filter = filter; + this.partitions = partitions; + this.blocks = blocks; + this.runStart = start; + this.runEnd = end; + try { + while (blockIndex < blocks.size() && blocks.get(blockIndex).end <= start) { + blockIndex++; + } + checkState( + blockIndex < blocks.size(), + "Manifest run starts after the end of the file."); + } catch (Exception e) { + try { + reader.close(); + } catch (Exception closeFailure) { + e.addSuppressed(closeFailure); + } + throw e; + } + } + + @Override + public boolean advance() throws Exception { + current = false; + while (true) { + if (decodedRemaining == 0) { + if (!prepareNextBlock()) { + key.clear(); + close(); + return false; + } + if (rawBlock) { + return true; + } + } + checkState( + currentRows != null && currentRows.hasNext(), + "Manifest block ends before its discovered boundary."); + currentRow = currentRows.next(); + decodedRemaining--; + key.replace(currentRow, partitions); + if (filter.include(currentRow, key)) { + current = true; + InternalRow file = ManifestEntryRunMergeEntry.file(currentRow); + metadata.replace( + key.kind, + partitions.partition(key.partitionId), + currentRow.getInt(ManifestEntryRunMerge.BUCKET), + file.getInt(ManifestEntryRunMerge.LEVEL), + file.getLong(ManifestEntryRunMerge.SCHEMA_ID), + key.firstRowId, + file.getLong(ManifestEntryRunMerge.ROW_COUNT)); + return true; + } + } + } + + boolean prepareNextBlock() throws Exception { + rawBlock = false; + current = false; + currentRows = null; + currentRow = null; + while (blockIndex < blocks.size()) { + ManifestEntryRunMerge.Discovery.BlockInfo info = blocks.get(blockIndex); + if (info.start >= runEnd) { + return false; + } + while (nextReaderBlockOrdinal < info.ordinal) { + checkState(reader.hasNext(), "Manifest block ordinal is missing."); + reader.next(); + nextReaderBlockOrdinal++; + } + checkState(reader.hasNext(), "Manifest run ends after the end of the file."); + currentRawBlock = reader.next(); + nextReaderBlockOrdinal++; + currentBlock = info; + if (info.copyable(runStart, runEnd)) { + rawBlock = true; + key.copyFrom(info.firstKey); + return true; + } + + long overlapStart = Math.max(runStart, info.start); + long overlapEnd = Math.min(runEnd, info.end); + long prefix = overlapStart - info.start; + currentRows = currentRawBlock.toRows(ManifestEntryRunMerge.ENTRY_LAYOUT); + for (long i = 0; i < prefix; i++) { + checkState( + currentRows.hasNext(), + "Manifest run starts after the end of its block."); + currentRows.next(); + } + decodedRemaining = overlapEnd - overlapStart; + blockIndex++; + if (decodedRemaining > 0) { + return true; + } + } + return false; + } + + @Override + public boolean hasCurrent() { + return current || rawBlock; + } + + @Override + public ProjectedManifestEntry current() { + return null; + } + + @Override + public EncodedEntry metadata() { + return metadata; + } + + @Override + public ManifestEntryRunMergeEntry.Key key() { + return key; + } + + @Override + public ByteBuffer encodedRecord() { + return current ? currentRows.encodedRecord() : null; + } Review Comment: **[P1] Do not copy encoded records across Avro writer schemas** `rawBlockCopySupported()` prevents whole-block copying when the input writer schema differs from the current manifest schema, but this path still returns the encoded bytes from the source record. `writeSelected` then passes those bytes to `ManifestAvroWriter.writeEncoded`, which appends them under the current schema header without transcoding or validation. An upgraded data-evolution table can have legacy manifests that already contain `_FIRST_ROW_ID` (so run merge is enabled) but lack a later field such as `_WRITE_COLS`. Copying such records can produce a malformed manifest; a focused legacy-schema reproduction fails with `EOFException` when the output is read. Please fall back to the existing external sorter whenever a source writer schema is not exactly compatible, or fully materialize and re-encode those records with the current schema. Please also add full/minor compaction coverage for a legacy RowID manifest containing fields 0..18 but not `_WRITE_COLS`. This blocks enabling the optimization in production by default. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
