Copilot commented on code in PR #4188:
URL: https://github.com/apache/amoro/pull/4188#discussion_r3099437853
##########
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()
+ && hasNoEqualityDeleteFiles()
+ && 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.warn(
+ "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;
+ }
+ }
+
+ return true;
+ }
+
+ private boolean hasNoReadOnlyDeleteFiles() {
+ return checkCondition(
+ ArrayUtils.isEmpty(input.readOnlyDeleteFiles()), "has read-only delete
files");
+ }
+
+ private boolean hasNoEqualityDeleteFiles() {
+ return checkCondition(
+ ArrayUtils.isEmpty(input.rewrittenDeleteFiles()), "has equality delete
files");
Review Comment:
Method name `hasNoEqualityDeleteFiles` is misleading: it checks
`input.rewrittenDeleteFiles()` which can contain both equality and position
delete files (it’s typed as `ContentFile<?>[]`). Consider renaming it (and the
debug message) to reflect that it disallows any rewritten delete files, not
just equality deletes, to avoid confusion and future misuse.
```suggestion
private boolean hasNoRewrittenDeleteFiles() {
return checkCondition(
ArrayUtils.isEmpty(input.rewrittenDeleteFiles()), "has rewritten
delete files");
```
##########
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()
+ && hasNoEqualityDeleteFiles()
+ && 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.warn(
+ "Skip parquet row-group merge due to schema mismatch between
{} and {}",
+ input.rewrittenDataFiles()[0].location(),
+ file.location());
Review Comment:
The log statement arguments are indented with tab characters, which can
violate code style checks and makes alignment inconsistent. Replace the tabs
with spaces and format the arguments consistently with the surrounding code.
```suggestion
input.rewrittenDataFiles()[0].location(),
file.location());
```
##########
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()
+ && hasNoEqualityDeleteFiles()
+ && 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.warn(
+ "Skip parquet row-group merge due to schema mismatch between
{} and {}",
+ input.rewrittenDataFiles()[0].location(),
+ file.location());
+ return null;
Review Comment:
`checkSchemaAndBuildContext` logs schema mismatches at WARN even though this
is an expected eligibility failure (similar to other precondition checks that
log at DEBUG). This can create noisy logs during normal schema evolution.
Consider lowering this to DEBUG (or routing through `checkCondition`) and avoid
treating it as a warning-level event.
##########
amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/ParquetFileMergeRunner.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * 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 {
Review Comment:
`appendFile` will throw a NullPointerException if it’s called before
`start()` initializes `writer`. Even though the current call sites invoke
`start()` first, adding a defensive state check (as in `result()`) would make
this utility safer to reuse and fail with a clearer error message.
```suggestion
public void appendFile(InputFile inputFile) throws IOException {
if (writer == null) {
throw new IllegalStateException("ParquetFileWriter not initialized.
Call start() first.");
}
```
##########
amoro-format-iceberg/src/test/java/org/apache/amoro/optimizing/IcebergRewriteExecutorTest.java:
##########
@@ -208,6 +213,296 @@ public void readAllDataWithPartitionEvolution() throws
IOException {
readAllData();
}
+ @Test
+ public void testParquetRowGroupMergeEnableCondition() throws IOException {
+ // parquet row-group merge should only be enabled for Parquet files.
+ if (fileFormat != FileFormat.PARQUET) {
+ Assert.assertFalse(newExecutor(dataScanTask).canParquetRowGroupMerge());
+ return;
+ }
+
+ // All parquet row-group merge preconditions are satisfied.
+ prepareParquetRowGroupMergeTableConditions(true);
+ Assert.assertTrue(newExecutor(dataScanTask).canParquetRowGroupMerge());
+
+ // parquet row-group merge should be disabled if the table property is not
enabled.
+ prepareParquetRowGroupMergeTableConditions(false);
+ Assert.assertFalse(newExecutor(dataScanTask).canParquetRowGroupMerge());
+
+ // parquet row-group merge should be disabled if there are delete files.
+ prepareParquetRowGroupMergeTableConditions(true);
+ RewriteFilesInput rewrittenDeleteInput =
+ new RewriteFilesInput(
+ dataScanTask.rewrittenDataFiles(),
+ dataScanTask.rePosDeletedDataFiles(),
+ new DeleteFile[] {},
+ new DeleteFile[] {(DeleteFile) scanTask.readOnlyDeleteFiles()[0]},
+ getMixedTable());
+
Assert.assertFalse(newExecutor(rewrittenDeleteInput).canParquetRowGroupMerge());
+
+ rewrittenDeleteInput =
+ new RewriteFilesInput(
+ dataScanTask.rewrittenDataFiles(),
+ dataScanTask.rePosDeletedDataFiles(),
+ new DeleteFile[] {(DeleteFile) scanTask.readOnlyDeleteFiles()[0]},
+ new DeleteFile[] {},
+ getMixedTable());
+
Assert.assertFalse(newExecutor(rewrittenDeleteInput).canParquetRowGroupMerge());
+
+ // parquet row-group merge should be disabled if the bloom filter is
enabled.
+ resetParquetRowGroupMergeTestState();
+ Assert.assertTrue(newExecutor(dataScanTask).canParquetRowGroupMerge());
+ getMixedTable()
+ .asUnkeyedTable()
+ .updateProperties()
+ .set(
+
org.apache.iceberg.TableProperties.PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX +
"id",
+ "true")
+ .commit();
+ Assert.assertFalse(newExecutor(dataScanTask).canParquetRowGroupMerge());
+
+ // parquet row-group merge should be disabled if the table spec evolves so
that the source files
+ // no longer match the partition spec.
+ resetParquetRowGroupMergeTestState();
+ Assert.assertTrue(newExecutor(dataScanTask).canParquetRowGroupMerge());
+
getMixedTable().asUnkeyedTable().updateSpec().addField(Expressions.month("op_time")).commit();
+ Assert.assertFalse(newExecutor(dataScanTask).canParquetRowGroupMerge());
+
+ // parquet row-group merge should be disabled if table has a sort order.
+ resetParquetRowGroupMergeTestState();
+ Assert.assertTrue(newExecutor(dataScanTask).canParquetRowGroupMerge());
+ getMixedTable().asUnkeyedTable().replaceSortOrder().asc("id").commit();
+ Assert.assertFalse(newExecutor(dataScanTask).canParquetRowGroupMerge());
+ }
+
+ @Test
+ public void testParquetRowGroupMergeWithEncryptedDataFile() throws
IOException {
+ Assume.assumeTrue(fileFormat == FileFormat.PARQUET);
+ prepareParquetRowGroupMergeTableConditions(true);
+ Assert.assertTrue(newExecutor(dataScanTask).canParquetRowGroupMerge());
+
+ DataFile encryptedDataFile =
+ DataFiles.builder(getMixedTable().spec())
+ .copy(dataScanTask.rewrittenDataFiles()[0])
+ .withEncryptionKeyMetadata(ByteBuffer.wrap(new byte[] {1}))
+ .build();
+
+ RewriteFilesInput encryptedInput =
+ new RewriteFilesInput(
+ new DataFile[] {encryptedDataFile},
+ new DataFile[] {},
+ new DeleteFile[] {},
+ new DeleteFile[] {},
+ getMixedTable());
+ // parquet row-group merge should be disabled if the data file is
encrypted.
+ Assert.assertFalse(newExecutor(encryptedInput).canParquetRowGroupMerge());
+ }
+
+ @Test
+ public void testParquetRowGroupMergeExecuteSplitAndCommitDataIntegrity()
throws IOException {
+ Assume.assumeTrue(fileFormat == FileFormat.PARQUET);
+ prepareParquetRowGroupMergeTableConditions(true);
+
+ // Create 5 data files, and each data file has 3 records.
+ List<List<Record>> inputRecordGroups =
parquetRowGroupMergeSourceRecordGroups();
+ List<DataFile> sourceDataFiles =
Lists.newArrayListWithExpectedSize(inputRecordGroups.size());
+ long txId = 1L;
+ for (List<Record> records : inputRecordGroups) {
+ sourceDataFiles.addAll(
+ MixedDataTestHelpers.writeAndCommitBaseStore(getMixedTable(),
txId++, records, false));
+ }
+
+ // Set the targetSize
+ long splitTargetSize =
+ sourceDataFiles.get(0).fileSizeInBytes()
+ + sourceDataFiles.get(1).fileSizeInBytes()
+ + sourceDataFiles.get(2).fileSizeInBytes();
+ getMixedTable()
+ .asUnkeyedTable()
+ .updateProperties()
+ .set(
+ org.apache.amoro.table.TableProperties.SELF_OPTIMIZING_TARGET_SIZE,
+ String.valueOf(splitTargetSize))
+ .commit();
+
+ // Build a rewrite input with only data files and no delete files.
+ RewriteFilesInput rewriteFilesInput = newDataOnlyInput(sourceDataFiles);
+ IcebergRewriteExecutor executor = newExecutor(rewriteFilesInput);
+
+ // Verify parquet row-group merge eligibility and execute the merge.
+ Assert.assertTrue(executor.canParquetRowGroupMerge());
+
+ RewriteFilesOutput output = executor.execute();
+ // With this target, the first 3 source files are grouped into one output
and
+ // the last 2 into another.
+ Assert.assertEquals(2, output.getDataFiles().length);
+
+ // Commit merge result and validate the iceberg table data contains all
records from the
+ // source files.
+ getMixedTable()
+ .asUnkeyedTable()
+ .newRewrite()
+ .rewriteFiles(Sets.newHashSet(sourceDataFiles),
Sets.newHashSet(output.getDataFiles()))
+ .commit();
+
+ // Get the source records
+ Map<Integer, Record> expectedRecordsById = new HashMap<>();
+ for (List<Record> recordGroup : inputRecordGroups) {
+ for (Record record : recordGroup) {
+ expectedRecordsById.put((Integer) record.get(0), record);
+ }
+ }
+
+ // Get the current records after merge and commit, and compare with the
source records.
+ List<Record> currentRecords =
+ MixedDataTestHelpers.readBaseStore(getMixedTable(),
Expressions.alwaysTrue());
+ Assert.assertEquals(15, currentRecords.size());
+ for (Record currentRecord : currentRecords) {
+ Integer id = (Integer) currentRecord.get(0);
+ Record expectedRecord = expectedRecordsById.get(id);
+ // Compare all fields to ensure the record is correctly merge.
+ Assert.assertEquals(expectedRecord.get(0), currentRecord.get(0));
+ Assert.assertEquals(expectedRecord.get(1), currentRecord.get(1));
+ Assert.assertEquals(expectedRecord.get(2), currentRecord.get(2));
+ Assert.assertEquals(expectedRecord.get(3), currentRecord.get(3));
+ }
+ }
+
+ @Test
+ public void testParquetRowGroupMergeWithSchemaDifferent() throws IOException
{
+ Assume.assumeTrue(fileFormat == FileFormat.PARQUET);
+ prepareParquetRowGroupMergeTableConditions(true);
+
+ // Scenario 1: schema differs by adding new field.
+ List<DataFile> sourceDataFiles = Lists.newArrayList();
+ sourceDataFiles.addAll(
+ MixedDataTestHelpers.writeAndCommitBaseStore(
+ getMixedTable(),
+ 301L,
+ Collections.singletonList(
+ MixedDataTestHelpers.createRecord(301, "field_old", 11L,
"1970-01-03T08:00:00")),
+ false));
+ // add new field
+ getMixedTable()
+ .asUnkeyedTable()
+ .updateSchema()
+ .addColumn("new_col", Types.StringType.get())
+ .commit();
+ sourceDataFiles.addAll(
+ MixedDataTestHelpers.writeAndCommitBaseStore(
+ getMixedTable(),
+ 302L,
+ Collections.singletonList(
+ MixedDataTestHelpers.createRecord(
+ getMixedTable().schema(), 302, "field_new", 12L,
"1970-01-04T08:00:00", "x")),
+ false));
+
+ RewriteFilesInput rewriteInput = newDataOnlyInput(sourceDataFiles);
+ IcebergRewriteExecutor executor = newExecutor(rewriteInput);
+
+ Assert.assertTrue(executor.canParquetRowGroupMerge());
+ String exceptedErrorMessage =
+ "The input parquet files have inconsistent schemas and cannot be
merged.";
+ String errorMessage =
+ Assert.assertThrows(IllegalStateException.class,
executor::parquetRowGroupMergeFiles)
+ .getMessage();
+ Assert.assertEquals(exceptedErrorMessage, errorMessage);
Review Comment:
Typo in variable name `exceptedErrorMessage`; rename to
`expectedErrorMessage` for clarity.
##########
amoro-format-iceberg/src/test/java/org/apache/amoro/optimizing/IcebergRewriteExecutorTest.java:
##########
@@ -208,6 +213,296 @@ public void readAllDataWithPartitionEvolution() throws
IOException {
readAllData();
}
+ @Test
+ public void testParquetRowGroupMergeEnableCondition() throws IOException {
+ // parquet row-group merge should only be enabled for Parquet files.
+ if (fileFormat != FileFormat.PARQUET) {
+ Assert.assertFalse(newExecutor(dataScanTask).canParquetRowGroupMerge());
+ return;
+ }
+
+ // All parquet row-group merge preconditions are satisfied.
+ prepareParquetRowGroupMergeTableConditions(true);
+ Assert.assertTrue(newExecutor(dataScanTask).canParquetRowGroupMerge());
+
+ // parquet row-group merge should be disabled if the table property is not
enabled.
+ prepareParquetRowGroupMergeTableConditions(false);
+ Assert.assertFalse(newExecutor(dataScanTask).canParquetRowGroupMerge());
+
+ // parquet row-group merge should be disabled if there are delete files.
+ prepareParquetRowGroupMergeTableConditions(true);
+ RewriteFilesInput rewrittenDeleteInput =
+ new RewriteFilesInput(
+ dataScanTask.rewrittenDataFiles(),
+ dataScanTask.rePosDeletedDataFiles(),
+ new DeleteFile[] {},
+ new DeleteFile[] {(DeleteFile) scanTask.readOnlyDeleteFiles()[0]},
+ getMixedTable());
+
Assert.assertFalse(newExecutor(rewrittenDeleteInput).canParquetRowGroupMerge());
+
+ rewrittenDeleteInput =
+ new RewriteFilesInput(
+ dataScanTask.rewrittenDataFiles(),
+ dataScanTask.rePosDeletedDataFiles(),
+ new DeleteFile[] {(DeleteFile) scanTask.readOnlyDeleteFiles()[0]},
+ new DeleteFile[] {},
+ getMixedTable());
+
Assert.assertFalse(newExecutor(rewrittenDeleteInput).canParquetRowGroupMerge());
+
+ // parquet row-group merge should be disabled if the bloom filter is
enabled.
+ resetParquetRowGroupMergeTestState();
+ Assert.assertTrue(newExecutor(dataScanTask).canParquetRowGroupMerge());
+ getMixedTable()
+ .asUnkeyedTable()
+ .updateProperties()
+ .set(
+
org.apache.iceberg.TableProperties.PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX +
"id",
+ "true")
+ .commit();
+ Assert.assertFalse(newExecutor(dataScanTask).canParquetRowGroupMerge());
+
+ // parquet row-group merge should be disabled if the table spec evolves so
that the source files
+ // no longer match the partition spec.
+ resetParquetRowGroupMergeTestState();
+ Assert.assertTrue(newExecutor(dataScanTask).canParquetRowGroupMerge());
+
getMixedTable().asUnkeyedTable().updateSpec().addField(Expressions.month("op_time")).commit();
+ Assert.assertFalse(newExecutor(dataScanTask).canParquetRowGroupMerge());
+
+ // parquet row-group merge should be disabled if table has a sort order.
+ resetParquetRowGroupMergeTestState();
+ Assert.assertTrue(newExecutor(dataScanTask).canParquetRowGroupMerge());
+ getMixedTable().asUnkeyedTable().replaceSortOrder().asc("id").commit();
+ Assert.assertFalse(newExecutor(dataScanTask).canParquetRowGroupMerge());
+ }
+
+ @Test
+ public void testParquetRowGroupMergeWithEncryptedDataFile() throws
IOException {
+ Assume.assumeTrue(fileFormat == FileFormat.PARQUET);
+ prepareParquetRowGroupMergeTableConditions(true);
+ Assert.assertTrue(newExecutor(dataScanTask).canParquetRowGroupMerge());
+
+ DataFile encryptedDataFile =
+ DataFiles.builder(getMixedTable().spec())
+ .copy(dataScanTask.rewrittenDataFiles()[0])
+ .withEncryptionKeyMetadata(ByteBuffer.wrap(new byte[] {1}))
+ .build();
+
+ RewriteFilesInput encryptedInput =
+ new RewriteFilesInput(
+ new DataFile[] {encryptedDataFile},
+ new DataFile[] {},
+ new DeleteFile[] {},
+ new DeleteFile[] {},
+ getMixedTable());
+ // parquet row-group merge should be disabled if the data file is
encrypted.
+ Assert.assertFalse(newExecutor(encryptedInput).canParquetRowGroupMerge());
+ }
+
+ @Test
+ public void testParquetRowGroupMergeExecuteSplitAndCommitDataIntegrity()
throws IOException {
+ Assume.assumeTrue(fileFormat == FileFormat.PARQUET);
+ prepareParquetRowGroupMergeTableConditions(true);
+
+ // Create 5 data files, and each data file has 3 records.
+ List<List<Record>> inputRecordGroups =
parquetRowGroupMergeSourceRecordGroups();
+ List<DataFile> sourceDataFiles =
Lists.newArrayListWithExpectedSize(inputRecordGroups.size());
+ long txId = 1L;
+ for (List<Record> records : inputRecordGroups) {
+ sourceDataFiles.addAll(
+ MixedDataTestHelpers.writeAndCommitBaseStore(getMixedTable(),
txId++, records, false));
+ }
+
+ // Set the targetSize
+ long splitTargetSize =
+ sourceDataFiles.get(0).fileSizeInBytes()
+ + sourceDataFiles.get(1).fileSizeInBytes()
+ + sourceDataFiles.get(2).fileSizeInBytes();
+ getMixedTable()
+ .asUnkeyedTable()
+ .updateProperties()
+ .set(
+ org.apache.amoro.table.TableProperties.SELF_OPTIMIZING_TARGET_SIZE,
+ String.valueOf(splitTargetSize))
+ .commit();
+
+ // Build a rewrite input with only data files and no delete files.
+ RewriteFilesInput rewriteFilesInput = newDataOnlyInput(sourceDataFiles);
+ IcebergRewriteExecutor executor = newExecutor(rewriteFilesInput);
+
+ // Verify parquet row-group merge eligibility and execute the merge.
+ Assert.assertTrue(executor.canParquetRowGroupMerge());
+
+ RewriteFilesOutput output = executor.execute();
+ // With this target, the first 3 source files are grouped into one output
and
+ // the last 2 into another.
+ Assert.assertEquals(2, output.getDataFiles().length);
+
+ // Commit merge result and validate the iceberg table data contains all
records from the
+ // source files.
+ getMixedTable()
+ .asUnkeyedTable()
+ .newRewrite()
+ .rewriteFiles(Sets.newHashSet(sourceDataFiles),
Sets.newHashSet(output.getDataFiles()))
+ .commit();
+
+ // Get the source records
+ Map<Integer, Record> expectedRecordsById = new HashMap<>();
+ for (List<Record> recordGroup : inputRecordGroups) {
+ for (Record record : recordGroup) {
+ expectedRecordsById.put((Integer) record.get(0), record);
+ }
+ }
+
+ // Get the current records after merge and commit, and compare with the
source records.
+ List<Record> currentRecords =
+ MixedDataTestHelpers.readBaseStore(getMixedTable(),
Expressions.alwaysTrue());
+ Assert.assertEquals(15, currentRecords.size());
+ for (Record currentRecord : currentRecords) {
+ Integer id = (Integer) currentRecord.get(0);
+ Record expectedRecord = expectedRecordsById.get(id);
+ // Compare all fields to ensure the record is correctly merge.
Review Comment:
Grammar in comment: “ensure the record is correctly merge” should be
“correctly merged”.
```suggestion
// Compare all fields to ensure the record is correctly merged.
```
--
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]
