Copilot commented on code in PR #4188: URL: https://github.com/apache/amoro/pull/4188#discussion_r3100096800
########## amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/ParquetFileMergeRunner.java: ########## @@ -0,0 +1,196 @@ +/* + * 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.amoro.optimizing; + +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.Metrics; +import org.apache.iceberg.MetricsConfig; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.encryption.EncryptedOutputFile; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.parquet.ParquetIOBridge; +import org.apache.iceberg.parquet.ParquetUtil; +import org.apache.parquet.column.ParquetProperties; +import org.apache.parquet.hadoop.ParquetFileWriter; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.metadata.BlockMetaData; +import org.apache.parquet.hadoop.metadata.ParquetMetadata; +import org.apache.parquet.schema.MessageType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Closeable; +import java.io.IOException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** + * This component merges one or more parquet files into a single output parquet file without + * row-level decode/encode, then builds an Iceberg {@link org.apache.iceberg.DataFile} for commit. + * + * <p>Lifecycle: + * + * <ol> + * <li>{@link #start(org.apache.iceberg.encryption.EncryptedOutputFile)} to initialize writer + * <li>{@link #appendFile(org.apache.iceberg.io.InputFile)} for each source file + * <li>{@link #result()} to finalize and build output {@link org.apache.iceberg.DataFile} + * <li>{@link #close()} for safe cleanup (also supports failure cleanup) + * </ol> + * + * <p>Failure behavior: + * + * <ul> + * <li>If writer is not successful, {@link #close()} attempts to delete output file + * <li>If writer close fails, the exception is propagated + * </ul> + */ +public final class ParquetFileMergeRunner implements Closeable { + + private static final Logger LOG = LoggerFactory.getLogger(ParquetFileMergeRunner.class); + + private final MessageType parquetSchema; + private final Map<String, String> keyValueMetadata; + private final PartitionSpec spec; + private final StructLike partition; + private final MetricsConfig metricsConfig; + private final FileIO io; + + private ParquetFileWriter writer; + private EncryptedOutputFile outputFile; + private boolean success; + private final long rowGroupSize; + + public ParquetFileMergeRunner( + FileIO io, + MessageType parquetSchema, + Map<String, String> keyValueMetadata, + PartitionSpec spec, + StructLike partition, + long rowGroupSize, + MetricsConfig metricsConfig) { + this.io = io; + this.parquetSchema = parquetSchema; + this.keyValueMetadata = + keyValueMetadata == null ? new HashMap<>() : new HashMap<>(keyValueMetadata); + this.spec = spec; + this.partition = partition; + this.rowGroupSize = rowGroupSize; + this.metricsConfig = metricsConfig; + } + + /** + * Initializes the ParquetFileWriter for row-group merge. + * + * @param output encrypted output file destination + * @throws java.io.IOException if writer creation or writer start fails + */ + public void start(EncryptedOutputFile output) throws IOException { + this.outputFile = output; + writer = + new ParquetFileWriter( + ParquetIOBridge.file(output.encryptingOutputFile()), + parquetSchema, + ParquetFileWriter.Mode.CREATE, + rowGroupSize, + ParquetWriter.MAX_PADDING_SIZE_DEFAULT, + null, + ParquetProperties.builder().build()); + writer.start(); + } + + public void appendFile(InputFile inputFile) throws IOException { + if (writer == null) { + throw new IllegalStateException("ParquetFileWriter not initialized. Call start() first."); + } + + org.apache.parquet.io.InputFile file = ParquetIOBridge.file(inputFile); + writer.appendFile(file); + } + + public DataFile result() throws IOException { + if (writer == null) { + throw new IllegalStateException("ParquetFileWriter not initialized. Call start() first."); + } + if (success) { + throw new IllegalStateException("result() already called."); + } + + writer.end(keyValueMetadata); + ParquetMetadata footer = writer.getFooter(); + writer.close(); + writer = null; + + Metrics metrics = ParquetUtil.footerMetrics(footer, Stream.empty(), metricsConfig); + List<Long> splitOffsets = + footer.getBlocks().stream().map(BlockMetaData::getStartingPos).collect(Collectors.toList()); + + DataFile result = + DataFiles.builder(spec) + .withEncryptedOutputFile(outputFile) + .withFormat(FileFormat.PARQUET) + .withMetrics(metrics) + .withPartition(partition) + .withSplitOffsets(splitOffsets) + .build(); Review Comment: ParquetFileMergeRunner builds the output DataFile without setting fileSizeInBytes (and potentially recordCount). Iceberg relies on these fields for manifest entries/planning; leaving them unset can result in 0/unknown sizes and incorrect planning/splitting behavior. Consider computing the actual output length (via FileIO/InputFile length) and total row count (sum row-group row counts from the footer) and setting them on the DataFiles.Builder. ########## amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/IcebergRewriteExecutor.java: ########## @@ -112,4 +151,247 @@ protected long targetSize() { private PartitionSpec fileSpec() { return table.asUnkeyedTable().specs().get(input.allFiles()[0].specId()); } + + @Override + protected List<DataFile> rewriterDataFiles() throws Exception { + if (!canParquetRowGroupMerge()) { + return super.rewriterDataFiles(); + } + + try { + LOG.debug( + "Starting parquet row-group merge for {} files.", input.rewrittenDataFiles().length); + List<DataFile> dataFiles = parquetRowGroupMergeFiles(); + LOG.debug("Parquet row-group merge completed successfully"); + + return dataFiles; + } catch (Exception e) { + LOG.warn("Parquet row-group merge failed, falling back to row-based rewrite", e); + return super.rewriterDataFiles(); + } + } + + protected boolean canParquetRowGroupMerge() { + return isParquetRowGroupMergeEnabled() + && isTableUnsorted() + && isParquetFormat() + && hasNoEncryptedDataFiles() + && hasNoReadOnlyDeleteFiles() + && hasNoRewrittenDeleteFiles() + && hasNoBloomFilter() + && allFilesHaveCurrentSpecId(); + } + + private boolean isTableUnsorted() { + return checkCondition(!table.asUnkeyedTable().sortOrder().isSorted(), "table has sort order"); + } + + private boolean hasNoEncryptedDataFiles() { + for (DataFile file : input.rewrittenDataFiles()) { + if (!checkCondition( + file.keyMetadata() == null, "input file is encrypted: " + file.location())) { + return false; + } + } + return true; + } + + /** + * Check whether all input Parquet file schemas are consistent; if so, build and return + * FileSchemaContext, otherwise return null. + */ + private FileSchemaContext checkSchemaAndBuildContext() { + MessageType firstSchema = null; + Map<String, String> firstMetadata = null; + List<InputFile> inputFiles = new ArrayList<>(input.rewrittenDataFiles().length); + List<Long> fileSizes = new ArrayList<>(input.rewrittenDataFiles().length); + + try { + for (DataFile file : input.rewrittenDataFiles()) { + InputFile inputFile = io.newInputFile(file.location(), file.fileSizeInBytes()); + inputFiles.add(inputFile); + fileSizes.add(file.fileSizeInBytes()); + + try (ParquetFileReader reader = ParquetFileReader.open(ParquetIOBridge.file(inputFile))) { + // Get parquet schema from file metadata. + MessageType currentSchema = reader.getFileMetaData().getSchema(); + if (firstSchema == null) { + firstSchema = currentSchema; + firstMetadata = new HashMap<>(reader.getFileMetaData().getKeyValueMetaData()); + continue; + } + + if (!firstSchema.equals(currentSchema)) { + LOG.debug( + "Skip parquet row-group merge due to schema mismatch between {} and {}", + input.rewrittenDataFiles()[0].location(), + file.location()); + return null; + } + } + } + + if (firstSchema == null) { + LOG.warn("Skip parquet row-group merge due to empty input schema"); + return null; + } + + return new FileSchemaContext(firstSchema, firstMetadata, inputFiles, fileSizes); + } catch (RuntimeException | IOException e) { + LOG.warn("Skip parquet row-group merge due to parquet metadata read failure", e); + return null; + } + } + + protected List<DataFile> parquetRowGroupMergeFiles() throws Exception { + List<DataFile> outputFiles = new ArrayList<>(); + OutputFileFactory outputFileFactory = newRowGroupMergeOutputFileFactory(); + FileSchemaContext fileSchemaContext = checkSchemaAndBuildContext(); + if (fileSchemaContext == null) { + throw new IllegalStateException( + "The input parquet files have inconsistent schemas and cannot be merged."); + } Review Comment: parquetRowGroupMergeFiles() throws IllegalStateException("inconsistent schemas") whenever checkSchemaAndBuildContext() returns null, but that method also returns null for other reasons (e.g., Parquet metadata read failure or empty input). This makes the exception message inaccurate for non-schema cases. Consider distinguishing schema mismatch vs metadata read errors (e.g., return a result with a failure reason, or throw a more accurate exception message when metadata cannot be read). ########## amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/IcebergRewriteExecutor.java: ########## @@ -112,4 +151,247 @@ protected long targetSize() { private PartitionSpec fileSpec() { return table.asUnkeyedTable().specs().get(input.allFiles()[0].specId()); } + + @Override + protected List<DataFile> rewriterDataFiles() throws Exception { + if (!canParquetRowGroupMerge()) { + return super.rewriterDataFiles(); + } + + try { + LOG.debug( + "Starting parquet row-group merge for {} files.", input.rewrittenDataFiles().length); + List<DataFile> dataFiles = parquetRowGroupMergeFiles(); + LOG.debug("Parquet row-group merge completed successfully"); + + return dataFiles; + } catch (Exception e) { + LOG.warn("Parquet row-group merge failed, falling back to row-based rewrite", e); + return super.rewriterDataFiles(); + } + } + + protected boolean canParquetRowGroupMerge() { + return isParquetRowGroupMergeEnabled() + && isTableUnsorted() + && isParquetFormat() + && hasNoEncryptedDataFiles() + && hasNoReadOnlyDeleteFiles() + && hasNoRewrittenDeleteFiles() + && hasNoBloomFilter() + && allFilesHaveCurrentSpecId(); + } + + private boolean isTableUnsorted() { + return checkCondition(!table.asUnkeyedTable().sortOrder().isSorted(), "table has sort order"); + } + + private boolean hasNoEncryptedDataFiles() { + for (DataFile file : input.rewrittenDataFiles()) { + if (!checkCondition( + file.keyMetadata() == null, "input file is encrypted: " + file.location())) { + return false; + } + } + return true; + } + + /** + * Check whether all input Parquet file schemas are consistent; if so, build and return + * FileSchemaContext, otherwise return null. + */ + private FileSchemaContext checkSchemaAndBuildContext() { + MessageType firstSchema = null; + Map<String, String> firstMetadata = null; + List<InputFile> inputFiles = new ArrayList<>(input.rewrittenDataFiles().length); + List<Long> fileSizes = new ArrayList<>(input.rewrittenDataFiles().length); + + try { + for (DataFile file : input.rewrittenDataFiles()) { + InputFile inputFile = io.newInputFile(file.location(), file.fileSizeInBytes()); + inputFiles.add(inputFile); + fileSizes.add(file.fileSizeInBytes()); + + try (ParquetFileReader reader = ParquetFileReader.open(ParquetIOBridge.file(inputFile))) { + // Get parquet schema from file metadata. + MessageType currentSchema = reader.getFileMetaData().getSchema(); + if (firstSchema == null) { + firstSchema = currentSchema; + firstMetadata = new HashMap<>(reader.getFileMetaData().getKeyValueMetaData()); + continue; + } + + if (!firstSchema.equals(currentSchema)) { + LOG.debug( + "Skip parquet row-group merge due to schema mismatch between {} and {}", + input.rewrittenDataFiles()[0].location(), + file.location()); + return null; + } + } + } + + if (firstSchema == null) { + LOG.warn("Skip parquet row-group merge due to empty input schema"); + return null; + } + + return new FileSchemaContext(firstSchema, firstMetadata, inputFiles, fileSizes); + } catch (RuntimeException | IOException e) { + LOG.warn("Skip parquet row-group merge due to parquet metadata read failure", e); + return null; + } + } + + protected List<DataFile> parquetRowGroupMergeFiles() throws Exception { + List<DataFile> outputFiles = new ArrayList<>(); + OutputFileFactory outputFileFactory = newRowGroupMergeOutputFileFactory(); + FileSchemaContext fileSchemaContext = checkSchemaAndBuildContext(); + if (fileSchemaContext == null) { + throw new IllegalStateException( + "The input parquet files have inconsistent schemas and cannot be merged."); + } + long maxOutputSize = super.targetSize(); + long currentOutputSize = 0L; + ParquetFileMergeRunner parquetFileMergeRunner = null; + + try { + for (int i = 0; i < fileSchemaContext.inputFiles.size(); i++) { + InputFile inputFile = fileSchemaContext.inputFiles.get(i); + long fileSize = fileSchemaContext.fileSizes.get(i); + // Roll to a new output file when appending the next input would exceed target size. + if (currentOutputSize > 0 && currentOutputSize + fileSize > maxOutputSize) { + outputFiles.add(parquetFileMergeRunner.result()); + parquetFileMergeRunner = null; + } + if (parquetFileMergeRunner == null) { + parquetFileMergeRunner = + new ParquetFileMergeRunner( + io, + fileSchemaContext.parquetSchema, + fileSchemaContext.keyValueMetadata, + table.spec(), + partition(), + parquetRowGroupSize(), + MetricsConfig.forTable(table.asUnkeyedTable())); + EncryptedOutputFile outputFile = newRowGroupMergeOutputFile(outputFileFactory); + parquetFileMergeRunner.start(outputFile); + currentOutputSize = 0L; + } + + parquetFileMergeRunner.appendFile(inputFile); + currentOutputSize += fileSize; + } + + if (parquetFileMergeRunner != null) { + outputFiles.add(parquetFileMergeRunner.result()); + parquetFileMergeRunner = null; + } + + return outputFiles; + } finally { + if (parquetFileMergeRunner != null) { + parquetFileMergeRunner.close(); + } + } + } + + private boolean isParquetRowGroupMergeEnabled() { + return checkCondition( + PropertyUtil.propertyAsBoolean( + table.properties(), + SELF_OPTIMIZING_REWRITE_USE_PARQUET_ROW_GROUP_MERGE_ENABLED, + SELF_OPTIMIZING_REWRITE_USE_PARQUET_ROW_GROUP_MERGE_ENABLED_DEFAULT), + SELF_OPTIMIZING_REWRITE_USE_PARQUET_ROW_GROUP_MERGE_ENABLED + " is not enabled"); + } + + /** + * check if all files are in parquet format, and the table default file format is parquet. Only + * when both conditions are satisfied, we can do parquet row-group merge. + * + * @return result of the check + */ + private boolean isParquetFormat() { + // 1、table default file format is parquet + if (!checkCondition( + dataFileFormat() == FileFormat.PARQUET, "table default file format is not PARQUET")) { + return false; + } + + // 2、all files are in parquet format + for (DataFile file : input.rewrittenDataFiles()) { + if (!checkCondition( + file.format() == FileFormat.PARQUET, + "input file format is not PARQUET: " + file.format())) { + return false; Review Comment: isParquetFormat() requires both (a) the table default file format is PARQUET and (b) all input files are PARQUET. The PR description/docs state eligibility is based on input files being Parquet, but don’t mention the additional requirement on table default format. Either document this extra precondition (and why it’s needed), or consider relaxing it if the intent is to allow merging Parquet files even when the table default format differs. -- 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]
