JingsongLi commented on code in PR #9660: URL: https://github.com/apache/paimon/pull/9660#discussion_r4254539480
########## paimon-core/src/main/java/org/apache/paimon/append/ParquetFastPathCompactRewriter.java: ########## @@ -0,0 +1,534 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.append; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.deletionvectors.DeletionVector; +import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.SimpleColStats; +import org.apache.paimon.format.parquet.ParquetRowGroupCopier; +import org.apache.paimon.format.parquet.ParquetRowGroupCopyChecker; +import org.apache.paimon.format.parquet.ParquetRowGroupCopyChecker.Incompatibility; +import org.apache.paimon.format.parquet.ParquetRowGroupCopySource; +import org.apache.paimon.format.parquet.ParquetSimpleStatsExtractor; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataFilePathFactory; +import org.apache.paimon.manifest.FileSource; +import org.apache.paimon.operation.metrics.CompactionFastPathMetrics; +import org.apache.paimon.operation.metrics.CompactionFastPathMetrics.MissReason; +import org.apache.paimon.statistics.SimpleColStatsCollector; +import org.apache.paimon.stats.SimpleStats; +import org.apache.paimon.stats.SimpleStatsConverter; +import org.apache.paimon.stats.SimpleStatsMerger; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.LongCounter; +import org.apache.paimon.utils.Pair; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.function.Function; + +import static org.apache.paimon.CoreOptions.FILE_FORMAT_PARQUET; +import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET; +import static org.apache.paimon.utils.StatsCollectorFactories.createStatsFactories; + +/** Fast-path rewriter that concatenates Parquet RowGroups for append-only compaction. */ +public class ParquetFastPathCompactRewriter { + + private static final Logger LOG = LoggerFactory.getLogger(ParquetFastPathCompactRewriter.class); + + private ParquetFastPathCompactRewriter() {} + + @Nullable + public static List<DataFileMeta> tryRewrite( + FileIO fileIO, + FileFormat fileFormat, + RowType writeType, + CoreOptions options, + BinaryRow partition, + int bucket, + @Nullable Function<String, DeletionVector> dvFactory, + List<DataFileMeta> toCompact, + DataFilePathFactory pathFactory, + long schemaId, + @Nullable CompactionFastPathMetrics metrics) { + long startNanos = System.nanoTime(); + long inputBytes = toCompact.stream().mapToLong(DataFileMeta::fileSize).sum(); + long inputRows = toCompact.stream().mapToLong(DataFileMeta::rowCount).sum(); + try { + if (dvFactory != null) { + reportMiss( + metrics, + MissReason.DV, + String.format( + "inputFiles=%d, inputRows=%d, inputBytes=%d", + toCompact.size(), inputRows, inputBytes)); + return null; + } + + ParquetRowGroupCopyChecker checker = + new ParquetRowGroupCopyChecker( + writeType, options.fileCompression(), options.toConfiguration()); + List<String> expectedValueStatsCols = toCompact.get(0).valueStatsCols(); + + MissReason tableMiss = + checkTableLevel( + options, + fileFormat, + bucket, + expectedValueStatsCols, + toCompact, + checker); + if (tableMiss != null) { + reportMiss( + metrics, + tableMiss, + String.format( + "bucket=%d, fileFormat=%s, inputFiles=%d", + bucket, options.fileFormatString(), toCompact.size())); + return null; + } + + List<FooterRequest> footerRequests = new ArrayList<>(toCompact.size()); + for (int i = 0; i < toCompact.size(); i++) { + DataFileMeta file = toCompact.get(i); + MissReason fileMiss = checkFileLevel(file, schemaId, expectedValueStatsCols); + if (fileMiss != null) { + reportMiss( + metrics, + fileMiss, + String.format( + "file=%s, fileIndex=%d, schemaId=%d", + file.fileName(), i, file.schemaId())); + return null; + } + Path filePath = pathFactory.toPath(file); + footerRequests.add(new FooterRequest(i, file, filePath)); + } + + FooterReadStats footerReadStats = readFooters(fileIO, options, footerRequests); + List<PreparedInput> preparedInputs = new ArrayList<>(footerRequests.size()); + for (FooterRequest request : footerRequests) { + IndexedFooter indexedFooter = footerReadStats.orderedFooters[request.fileIndex]; + Incompatibility incompatibility = indexedFooter.source.check(checker); + if (incompatibility != null) { + reportMiss( + metrics, + toMissReason(incompatibility), + String.format( + "file=%s, fileIndex=%d, expectedCodec=%s", + request.file.fileName(), + request.fileIndex, + checker.expectedCodec())); + return null; + } + preparedInputs.add(new PreparedInput(request.file, indexedFooter.source)); + } + + long prepareMs = elapsedMillis(startNanos); + long footerReadWallMs = nanosToMillis(footerReadStats.footerReadWallNanos); + long footerReadSumMs = nanosToMillis(footerReadStats.footerReadSumNanos); + int inputRowGroups = + preparedInputs.stream().mapToInt(input -> input.source.blockCount()).sum(); + List<ParquetRowGroupCopier.Input> copierInputs = new ArrayList<>(preparedInputs.size()); + for (PreparedInput preparedInput : preparedInputs) { + copierInputs.add(preparedInput.source.copierInput()); + } + + ParquetRowGroupCopier copier = + new ParquetRowGroupCopier( + fileIO, writeType, options.targetFileSize(false), pathFactory::newPath); + long copyStartNanos = System.nanoTime(); + List<ParquetRowGroupCopier.OutputFile> copiedFiles = copier.copy(copierInputs); + long copyMs = elapsedMillis(copyStartNanos); + try { + long buildStartNanos = System.nanoTime(); + List<DataFileMeta> result = + buildResult( + copiedFiles, + preparedInputs, + toCompact, + writeType, + schemaId, + expectedValueStatsCols, + options, + pathFactory); + long buildResultMs = elapsedMillis(buildStartNanos); + long outputBytes = result.stream().mapToLong(DataFileMeta::fileSize).sum(); + long outputRows = result.stream().mapToLong(DataFileMeta::rowCount).sum(); + if (metrics != null) { + metrics.reportHit(); + } + LOG.info( + "Append compaction fast path succeeded: inputFiles={}, inputRows={}, " + + "inputBytes={}, inputRowGroups={}, outputFiles={}, outputRows={}, " + + "outputBytes={}, prepareMs={}, footerReadWallMs={}, " + + "footerReadSumMs={}, copyMs={}, buildResultMs={}, totalMs={}", + toCompact.size(), + inputRows, + inputBytes, + inputRowGroups, + result.size(), + outputRows, + outputBytes, + prepareMs, + footerReadWallMs, + footerReadSumMs, + copyMs, + buildResultMs, + elapsedMillis(startNanos)); + return result; + } catch (IOException | RuntimeException e) { + cleanupCopiedFiles(fileIO, copiedFiles); + throw e; + } + } catch (IOException e) { + reportMiss( + metrics, + MissReason.IO_ERROR, + String.format( + "inputFiles=%d, inputRows=%d, inputBytes=%d, error=%s", + toCompact.size(), inputRows, inputBytes, e.toString())); + LOG.info("Append compaction fast path failed with IO error, fallback to rewrite", e); + return null; + } catch (RuntimeException e) { + reportMiss( + metrics, + MissReason.IO_ERROR, + String.format( + "inputFiles=%d, inputRows=%d, inputBytes=%d, error=%s", + toCompact.size(), inputRows, inputBytes, e.toString())); + LOG.info("Append compaction fast path failed, fallback to rewrite", e); + return null; + } + } + + private static List<DataFileMeta> buildResult( + List<ParquetRowGroupCopier.OutputFile> copiedFiles, + List<PreparedInput> preparedInputs, + List<DataFileMeta> toCompact, + RowType writeType, + long schemaId, + @Nullable List<String> valueStatsCols, + CoreOptions options, + DataFilePathFactory pathFactory) + throws IOException { + boolean isExternalPath = pathFactory.isExternalPath(); + LongCounter sequenceCounter = new LongCounter(toCompact.get(0).minSequenceNumber()); + List<DataFileMeta> result = new ArrayList<>(copiedFiles.size()); + for (ParquetRowGroupCopier.OutputFile copiedFile : copiedFiles) { + SimpleStats valueStats = + mergeOutputValueStats( + copiedFile.blockContributions(), + preparedInputs, + writeType, + valueStatsCols, + options); + + long minSequenceNumber = sequenceCounter.getValue(); + sequenceCounter.add(copiedFile.rowCount()); + long maxSequenceNumber = sequenceCounter.getValue() - 1; + + String externalPath = isExternalPath ? copiedFile.path().toString() : null; + result.add( + DataFileMeta.forAppend( + copiedFile.path().getName(), + copiedFile.fileSize(), + copiedFile.rowCount(), + valueStats, + minSequenceNumber, + maxSequenceNumber, + schemaId, + Collections.emptyList(), + null, + FileSource.COMPACT, + valueStatsCols, + externalPath, + null, + null)); + } + + long inputRowCount = toCompact.stream().mapToLong(DataFileMeta::rowCount).sum(); + long outputRowCount = result.stream().mapToLong(DataFileMeta::rowCount).sum(); + if (inputRowCount != outputRowCount) { + throw new IOException( + String.format( + "Row count mismatch after RowGroup copy: input %d but output %d", + inputRowCount, outputRowCount)); + } + return result; + } + + private static SimpleStats mergeOutputValueStats( + List<ParquetRowGroupCopier.BlockContribution> contributions, + List<PreparedInput> preparedInputs, + RowType writeType, + @Nullable List<String> valueStatsCols, + CoreOptions options) + throws IOException { + List<SimpleStats> statsToMerge = new ArrayList<>(); + Map<Integer, List<Integer>> blockIndicesByFile = new LinkedHashMap<>(); + for (ParquetRowGroupCopier.BlockContribution contribution : contributions) { + blockIndicesByFile + .computeIfAbsent(contribution.fileIndex(), ignored -> new ArrayList<>()) + .add(contribution.blockIndex()); + } + + for (Map.Entry<Integer, List<Integer>> entry : blockIndicesByFile.entrySet()) { + int fileIndex = entry.getKey(); + PreparedInput preparedInput = preparedInputs.get(fileIndex); + List<Integer> blockIndices = entry.getValue(); + if (isFullFile(blockIndices, preparedInput.source.blockCount())) { + statsToMerge.add(preparedInput.file.valueStats()); + } else { + statsToMerge.add( + statsFromBlocks( + preparedInput.source, + blockIndices, + writeType, + valueStatsCols, + options)); + } + } + return SimpleStatsMerger.merge(statsToMerge, writeType, valueStatsCols); + } + + private static boolean isFullFile(List<Integer> blockIndices, int totalBlocks) { + if (blockIndices.size() != totalBlocks) { + return false; + } + for (int i = 0; i < totalBlocks; i++) { + if (blockIndices.get(i) != i) { + return false; + } + } + return true; + } + + private static SimpleStats statsFromBlocks( + ParquetRowGroupCopySource source, + List<Integer> blockIndexes, + RowType writeType, + @Nullable List<String> valueStatsCols, + CoreOptions options) + throws IOException { + SimpleColStatsCollector.Factory[] collectors = + createStatsFactories(options.statsMode(), options, writeType.getFieldNames()); + ParquetSimpleStatsExtractor extractor = + new ParquetSimpleStatsExtractor(options.toConfiguration(), writeType, collectors); + SimpleColStats[] colStats = source.extractBlockStats(extractor, blockIndexes); + Pair<List<String>, SimpleStats> converted = + new SimpleStatsConverter(writeType, options.statsDenseStore()).toBinary(colStats); + if (!SimpleStatsMerger.sameValueStatsCols(valueStatsCols, converted.getLeft())) { + throw new IOException( + String.format( + "Partial block stats columns mismatch: expected %s but got %s", + valueStatsCols, converted.getLeft())); + } + return converted.getRight(); + } + + private static void cleanupCopiedFiles( + FileIO fileIO, List<ParquetRowGroupCopier.OutputFile> copiedFiles) { + for (ParquetRowGroupCopier.OutputFile copiedFile : copiedFiles) { + fileIO.deleteQuietly(copiedFile.path()); + } + } + + @Nullable + private static MissReason checkTableLevel( + CoreOptions options, + FileFormat fileFormat, + int bucket, + @Nullable List<String> expectedValueStatsCols, + List<DataFileMeta> toCompact, + ParquetRowGroupCopyChecker checker) { + if (bucket != UNAWARE_BUCKET) { + return MissReason.OTHER; + } + if (!FILE_FORMAT_PARQUET.equals(options.fileFormatString())) { + return MissReason.OTHER; + } + if (!FILE_FORMAT_PARQUET.equals(fileFormat.getFormatIdentifier())) { + return MissReason.OTHER; + } + if (options.rowTrackingEnabled() || options.dataEvolutionEnabled()) { + return MissReason.ROW_TRACKING; + } + Incompatibility configurationMiss = checker.checkConfiguration(); Review Comment: [P2] Keep corrupt-file skipping at the original file boundary Please reject `scan.ignore-corrupt-files=true` here (or validate/isolate each copied input). An input with a damaged data page and intact footer passes the footer/schema/codec checks, so raw row-group copying merges it with healthy inputs. `DataFileRecordReader.readBatch` treats the first page failure as the end of that entire output file, hiding later healthy row groups too. Real-file reproduction with two 32-row inputs: damage the first input's first data-page header while preserving its footer; with ignore-corrupt enabled, the original table scan and ordinary `compactRewrite` both retain all 32 healthy rows. The copy path produces a single 64-row output; after committing that compaction, the standard table scan returns **zero rows**. Adding only a guard to fall back under this option makes the same committed read retain the 32 healthy rows. This amplifies an existing bad file into lost visibility of an independent healthy file; it is not just continuing to skip the already corr upt input. -- 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]
