Copilot commented on code in PR #4077:
URL: https://github.com/apache/amoro/pull/4077#discussion_r3117078065
##########
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();
+ }
+ }
Review Comment:
The PR title/description focuses on persisting cleanup process info
(AMORO-4048), but this PR also introduces a substantial Parquet row-group merge
rewrite feature (new property, new runner, bridge class, and extensive executor
logic). This is a significant scope expansion; please either update the PR
title/description/change log to reflect the Parquet merge feature or split this
into a separate PR to keep changesets focused.
##########
amoro-ams/src/main/java/org/apache/amoro/server/scheduler/PeriodicTableScheduler.java:
##########
@@ -178,6 +206,124 @@ private void persistUpdatingCleanupTime(TableRuntime
tableRuntime) {
}
}
+ @VisibleForTesting
+ protected TableProcessMeta createCleanupProcessInfo(
+ TableRuntime tableRuntime, CleanupOperation cleanupOperation) {
+
+ if (shouldSkipOperation(tableRuntime, cleanupOperation)) {
+ return null;
+ }
+
+ TableProcessMeta cleanProcessMeta = buildProcessMeta(tableRuntime,
cleanupOperation);
+ persistenceHelper.beginAndPersistCleanupProcess(cleanProcessMeta);
+
+ logger.debug(
+ "Successfully persist cleanup process [processId={}, tableId={},
processType={}]",
+ cleanProcessMeta.getProcessId(),
+ cleanProcessMeta.getTableId(),
+ cleanProcessMeta.getProcessType());
+
+ return cleanProcessMeta;
+ }
+
+ @VisibleForTesting
+ protected void persistCleanupResult(
+ TableRuntime tableRuntime,
+ CleanupOperation cleanupOperation,
+ TableProcessMeta cleanProcessMeta,
+ Throwable executionError) {
+
+ if (cleanProcessMeta == null) {
+ return;
+ }
+
+ if (executionError != null) {
+ cleanProcessMeta.setStatus(ProcessStatus.FAILED);
+
cleanProcessMeta.setFailMessage(ExceptionUtil.getErrorMessage(executionError,
4000));
+ } else {
+ cleanProcessMeta.setStatus(ProcessStatus.SUCCESS);
+ }
+
+ long endTime = System.currentTimeMillis();
+ persistenceHelper.persistAndSetCompleted(
+ tableRuntime, cleanupOperation, cleanProcessMeta, endTime);
+
+ logger.debug(
+ "Successfully updated lastCleanTime and cleanupProcess for table {}
with cleanup operation {}",
+ tableRuntime.getTableIdentifier().getTableName(),
+ cleanupOperation);
+ }
+
+ private TableProcessMeta buildProcessMeta(
+ TableRuntime tableRuntime, CleanupOperation cleanupOperation) {
+
+ TableProcessMeta cleanProcessMeta = new TableProcessMeta();
+ cleanProcessMeta.setTableId(tableRuntime.getTableIdentifier().getId());
+ cleanProcessMeta.setProcessId(ID_GENERATOR.generateId());
+ cleanProcessMeta.setExternalProcessIdentifier(EXTERNAL_PROCESS_IDENTIFIER);
+ cleanProcessMeta.setStatus(ProcessStatus.RUNNING);
+ cleanProcessMeta.setProcessType(cleanupOperation.name());
+ cleanProcessMeta.setProcessStage(CLEANUP_PROCESS_STAGE);
+ cleanProcessMeta.setExecutionEngine(CLEANUP_EXECUTION_ENGINE);
+ cleanProcessMeta.setRetryNumber(0);
+ cleanProcessMeta.setCreateTime(System.currentTimeMillis());
+ cleanProcessMeta.setProcessParameters(new HashMap<>());
+ cleanProcessMeta.setSummary(new HashMap<>());
+
+ return cleanProcessMeta;
+ }
+
+ private static class PersistenceHelper extends PersistentBase {
+
+ public PersistenceHelper() {}
+
+ private void beginAndPersistCleanupProcess(TableProcessMeta meta) {
+ doAsTransaction(
+ () ->
+ doAs(
+ TableProcessMapper.class,
+ mapper ->
+ mapper.insertProcess(
+ meta.getTableId(),
+ meta.getProcessId(),
+ meta.getExternalProcessIdentifier(),
+ meta.getStatus(),
+ meta.getProcessType(),
+ meta.getProcessStage(),
+ meta.getExecutionEngine(),
+ meta.getRetryNumber(),
+ meta.getCreateTime(),
+ meta.getProcessParameters(),
+ meta.getSummary())));
+ }
+
+ private void persistAndSetCompleted(
+ TableRuntime tableRuntime,
+ CleanupOperation cleanupOperation,
+ TableProcessMeta meta,
+ long endTime) {
+
+ doAsTransaction(
+ () ->
+ doAs(
+ TableProcessMapper.class,
+ mapper ->
+ mapper.updateProcess(
+ meta.getTableId(),
+ meta.getProcessId(),
+ meta.getExternalProcessIdentifier(),
+ meta.getStatus(),
+ meta.getProcessStage(),
+ meta.getRetryNumber(),
+ endTime,
+ meta.getFailMessage(),
+ meta.getProcessParameters(),
+ meta.getSummary())),
+ () ->
+ ((DefaultTableRuntime)
tableRuntime).updateLastCleanTime(cleanupOperation, endTime));
Review Comment:
`persistAndSetCompleted` always updates `lastCleanTime` regardless of
whether the process `meta` ended in SUCCESS or FAILED. Given
`persistCleanupResult` sets FAILED on errors, this will still advance
`lastCleanTime` on failures, which is likely incorrect for
scheduling/visibility. Consider moving `updateLastCleanTime` behind a
success-only condition, or passing a flag/status into this helper to decide
whether `lastCleanTime` should be updated.
```suggestion
() -> {
if (meta.getStatus() == ProcessStatus.SUCCESS) {
((DefaultTableRuntime) tableRuntime)
.updateLastCleanTime(cleanupOperation, endTime);
}
});
```
##########
amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/PeriodicTableSchedulerTestBase.java:
##########
@@ -88,4 +104,37 @@ public boolean shouldExecuteTaskForTest(
TableRuntime tableRuntime, CleanupOperation cleanupOperation) {
return shouldExecuteTask(tableRuntime, cleanupOperation);
}
+
+ public TableProcessMeta createCleanupProcessInfoForTest(
+ TableRuntime tableRuntime, CleanupOperation operation) {
+ return createCleanupProcessInfo(tableRuntime, operation);
+ }
+
+ public void persistCleanupResultForTest(
+ TableRuntime tableRuntime,
+ CleanupOperation operation,
+ TableProcessMeta meta,
+ Throwable error) {
+ persistCleanupResult(tableRuntime, operation, meta, error);
+ }
+
+ public void executeTaskForTest(TableRuntime tableRuntime) {
+ try {
+ Method executeTask =
+ PeriodicTableScheduler.class.getDeclaredMethod("executeTask",
TableRuntime.class);
+ executeTask.setAccessible(true);
+ executeTask.invoke(this, tableRuntime);
+ } catch (InvocationTargetException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof RuntimeException) {
+ throw (RuntimeException) cause;
+ }
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ throw new RuntimeException(cause);
+ } catch (ReflectiveOperationException e) {
+ throw new RuntimeException("Failed to invoke executeTask for test", e);
+ }
+ }
Review Comment:
The test relies on reflective access to a private method (`executeTask`) via
`setAccessible(true)`, which is brittle under stricter JVM access rules and can
fail in certain build/test environments. Consider making `executeTask`
`protected` (or package-private) and annotating it with `@VisibleForTesting`,
similar to `createCleanupProcessInfo`/`persistCleanupResult`, so tests can call
it directly without reflection.
##########
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:
The `DataFile` built here does not explicitly set required physical
attributes like file path/location and file size (and potentially record
count). In Iceberg, `DataFiles.builder(...)` commonly requires `withPath(...)`
and `withFileSizeInBytes(...)` (and often `withRecordCount(...)`) for correct
manifest entries and planning. Depending on Iceberg version, this may throw at
build time or create incomplete `DataFile`s that break reads/plans. Recommended
fix: compute output file size after writer end/close (via `FileIO` /
`InputFile.getLength()` for the output location) and set `withPath(...)` +
`withFileSizeInBytes(...)` (+ `withRecordCount(...)` if available from footer
metrics).
##########
amoro-format-iceberg/src/main/java/org/apache/iceberg/parquet/ParquetIOBridge.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.iceberg.parquet;
+
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+
+/**
+ * Bridge for reusing Iceberg's package-scoped {@link ParquetIO} conversions
from parquet row-group
+ * merge path.
+ *
+ * <p>This class lives in the same package as {@link ParquetIO} on purpose to
convert Iceberg {@link
+ * org.apache.iceberg.io.InputFile}/{@link org.apache.iceberg.io.OutputFile}
instances into Parquet
+ * file abstractions without re-implementing Iceberg's adapter logic.
+ */
+public final class ParquetIOBridge {
+ private ParquetIOBridge() {}
+
+ public static org.apache.parquet.io.InputFile file(InputFile file) {
+ return ParquetIO.file(file);
+ }
+
+ public static org.apache.parquet.io.OutputFile file(OutputFile file) {
+ return ParquetIO.file(file);
+ }
Review Comment:
Introducing a new class under `org.apache.iceberg.parquet` inside this repo
creates a split-package with the Iceberg dependency, which can cause classpath
conflicts and is fragile if Iceberg later adds a class with the same
name/package. A safer approach is to keep Amoro code under
`org.apache.amoro...` and access `ParquetIO.file(...)` via reflection (or
contribute an upstream/public adapter to Iceberg), avoiding reliance on
package-private access through split packages.
```suggestion
package org.apache.amoro.format.iceberg.parquet;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
/**
* Bridge for reusing Iceberg's internal {@code ParquetIO} conversions from
parquet row-group merge
* path without introducing an Amoro class into Iceberg's package.
*/
public final class ParquetIOBridge {
private static final String PARQUET_IO_CLASS_NAME =
"org.apache.iceberg.parquet.ParquetIO";
private static final Method INPUT_FILE_METHOD =
initFileMethod(InputFile.class);
private static final Method OUTPUT_FILE_METHOD =
initFileMethod(OutputFile.class);
private ParquetIOBridge() {}
public static org.apache.parquet.io.InputFile file(InputFile file) {
return (org.apache.parquet.io.InputFile)
invokeFileMethod(INPUT_FILE_METHOD, file);
}
public static org.apache.parquet.io.OutputFile file(OutputFile file) {
return (org.apache.parquet.io.OutputFile)
invokeFileMethod(OUTPUT_FILE_METHOD, file);
}
private static Method initFileMethod(Class<?> parameterType) {
try {
Class<?> parquetIOClass = Class.forName(PARQUET_IO_CLASS_NAME);
Method method = parquetIOClass.getDeclaredMethod("file",
parameterType);
method.setAccessible(true);
return method;
} catch (ReflectiveOperationException e) {
throw new IllegalStateException(
"Failed to access Iceberg ParquetIO.file(" +
parameterType.getName() + ")", e);
}
}
private static Object invokeFileMethod(Method method, Object file) {
try {
return method.invoke(null, file);
} catch (IllegalAccessException | InvocationTargetException e) {
throw new IllegalStateException(
"Failed to invoke Iceberg " + method.getDeclaringClass().getName()
+ "." + method.getName(),
e);
}
}
```
--
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]
