This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 992f0b5f90 [core] Add compaction duration log (#8273)
992f0b5f90 is described below
commit 992f0b5f903192aa741be706a4cf4254562521b6
Author: dwangatt <[email protected]>
AuthorDate: Fri Jul 3 22:16:59 2026 +1000
[core] Add compaction duration log (#8273)
- Add logs to trace the duration for compaction writer, so that in job
log, we can analyse slow compaction task, bucket size distribution, slow
partitions etc.
- Log real partition info instead of BinaryRow object address when
Paimon detects inconsistent bucket setup during committing data to
tables.
---
.../org/apache/paimon/compact/CompactTask.java | 33 ++++++++++++++++-
.../paimon/io/KeyValueFileReaderFactory.java | 4 ++
.../mergetree/compact/MergeTreeCompactManager.java | 8 ++++
.../compact/MergeTreeCompactManagerFactory.java | 43 ++++++++++++++--------
.../paimon/operation/AbstractFileStoreWrite.java | 42 +++++++++++++--------
.../org/apache/paimon/operation/WriteRestore.java | 11 +++++-
.../paimon/table/AppendOnlySimpleTableTest.java | 3 +-
.../sink/coordinator/TableWriteCoordinator.java | 20 +++++++++-
8 files changed, 128 insertions(+), 36 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compact/CompactTask.java
b/paimon-core/src/main/java/org/apache/paimon/compact/CompactTask.java
index 69e68949c3..faaf038106 100644
--- a/paimon-core/src/main/java/org/apache/paimon/compact/CompactTask.java
+++ b/paimon-core/src/main/java/org/apache/paimon/compact/CompactTask.java
@@ -37,22 +37,33 @@ public abstract class CompactTask implements
Callable<CompactResult> {
@Nullable private final CompactionMetrics.Reporter metricsReporter;
+ private String logInfo = "";
+
public CompactTask(@Nullable CompactionMetrics.Reporter metricsReporter) {
this.metricsReporter = metricsReporter;
}
+ /** Set additional compact task information for logging purposes. */
+ public void setLogInfo(String logInfo) {
+ this.logInfo = logInfo;
+ }
+
@Override
public CompactResult call() throws Exception {
MetricUtils.safeCall(this::startTimer, LOG);
+ LOG.info(
+ "Paimon compact task started: {}, taskType={}",
+ logInfo,
+ getClass().getSimpleName());
try {
long startMillis = System.currentTimeMillis();
CompactResult result = doCompact();
+ long durationMs = System.currentTimeMillis() - startMillis;
MetricUtils.safeCall(
() -> {
if (metricsReporter != null) {
- metricsReporter.reportCompactionTime(
- System.currentTimeMillis() - startMillis);
+ metricsReporter.reportCompactionTime(durationMs);
metricsReporter.increaseCompactionsCompletedCount();
metricsReporter.reportCompactionInputSize(
result.before().stream()
@@ -68,10 +79,28 @@ public abstract class CompactTask implements
Callable<CompactResult> {
},
LOG);
+ LOG.info(
+ "Paimon compact task finished: {}, taskType={}, "
+ + "inputFiles={}, inputBytes={}, outputFiles={},
outputBytes={}, durationMs={}",
+ logInfo,
+ getClass().getSimpleName(),
+ result.before().size(),
+
result.before().stream().mapToLong(DataFileMeta::fileSize).sum(),
+ result.after().size(),
+
result.after().stream().mapToLong(DataFileMeta::fileSize).sum(),
+ durationMs);
+
if (LOG.isDebugEnabled()) {
LOG.debug(logMetric(startMillis, result.before(),
result.after()));
}
return result;
+ } catch (Exception e) {
+ LOG.warn(
+ "Paimon compact task failed: {}, taskType={}",
+ logInfo,
+ getClass().getSimpleName(),
+ e);
+ throw e;
} finally {
MetricUtils.safeCall(this::stopTimer, LOG);
MetricUtils.safeCall(this::decreaseCompactionsQueuedCount, LOG);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java
b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java
index cae12ae82c..17aa99866f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java
@@ -300,6 +300,10 @@ public class KeyValueFileReaderFactory implements
FileReaderFactory<KeyValue> {
return readValueType;
}
+ public FileStorePathFactory pathFactory() {
+ return pathFactory;
+ }
+
public KeyValueFileReaderFactory build(
BinaryRow partition, int bucket, DeletionVector.Factory
dvFactory) {
return build(partition, bucket, dvFactory, true,
Collections.emptyList());
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
index ee431f4e03..6ee834acdd 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
@@ -72,6 +72,8 @@ public class MergeTreeCompactManager extends
CompactFutureManager {
@Nullable private final RecordLevelExpire recordLevelExpire;
+ private String logInfo = "";
+
public MergeTreeCompactManager(
ExecutorService executor,
Levels levels,
@@ -105,6 +107,11 @@ public class MergeTreeCompactManager extends
CompactFutureManager {
MetricUtils.safeCall(this::reportMetrics, LOG);
}
+ /** Set additional compact task information for logging purposes. */
+ public void setLogInfo(String logInfo) {
+ this.logInfo = logInfo;
+ }
+
@Override
public boolean shouldWaitForLatestCompaction() {
return levels.numberOfSortedRuns() > numSortedRunStopTrigger;
@@ -244,6 +251,7 @@ public class MergeTreeCompactManager extends
CompactFutureManager {
file.fileName(),
file.level(), file.fileSize()))
.collect(Collectors.joining(", ")));
}
+ task.setLogInfo(logInfo);
taskFuture = executor.submit(task);
if (metricsReporter != null) {
metricsReporter.increaseCompactionsQueuedCount();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerFactory.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerFactory.java
index 1daa51a399..23d997c0cf 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerFactory.java
@@ -173,21 +173,34 @@ public class MergeTreeCompactManagerFactory implements
KvCompactionManagerFactor
if (metricsReporter != null) {
rewriter.setMetricsReporter(metricsReporter);
}
- return new MergeTreeCompactManager(
- compactExecutor,
- levels,
- compactStrategy,
- keyComparator,
- options.compactionFileSize(true),
- options.numSortedRunStopTrigger(),
- rewriter,
- metricsReporter,
- dvMaintainer,
- options.prepareCommitWaitCompaction(),
- options.needLookup(),
- recordLevelExpire,
- options.forceRewriteAllFiles(),
- options.isChainTable());
+ MergeTreeCompactManager compactManager =
+ new MergeTreeCompactManager(
+ compactExecutor,
+ levels,
+ compactStrategy,
+ keyComparator,
+ options.compactionFileSize(true),
+ options.numSortedRunStopTrigger(),
+ rewriter,
+ metricsReporter,
+ dvMaintainer,
+ options.prepareCommitWaitCompaction(),
+ options.needLookup(),
+ recordLevelExpire,
+ options.forceRewriteAllFiles(),
+ options.isChainTable());
+ compactManager.setLogInfo(compactTaskLogInfo(partition, bucket));
+ return compactManager;
+ }
+
+ private String compactTaskLogInfo(BinaryRow partition, int bucket) {
+ String partitionString;
+ try {
+ partitionString =
readerFactoryBuilder.pathFactory().getPartitionString(partition);
+ } catch (Exception e) {
+ partitionString = partition.toString();
+ }
+ return String.format("partition=%s, bucket=%d", partitionString,
bucket);
}
private CompactStrategy createCompactStrategy(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
index 953942d738..2961a37029 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
@@ -540,27 +540,28 @@ public abstract class AbstractFileStoreWrite<T>
implements FileStoreWrite<T> {
}
private RestoreFiles scanExistingFileMetas(BinaryRow partition, int
bucket) {
- RestoreFiles restored =
- restore.restoreFiles(
- partition,
- bucket,
- dbMaintainerFactory != null,
- dvMaintainerFactory != null);
+ String partInfo = partitionInfo(partition);
+ RestoreFiles restored;
+ try {
+ restored =
+ restore.restoreFiles(
+ partition,
+ bucket,
+ dbMaintainerFactory != null,
+ dvMaintainerFactory != null);
+ } catch (RuntimeException e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to restore existing files for %s, bucket
%d.",
+ partInfo, bucket),
+ e);
+ }
Integer restoredTotalBuckets = restored.totalBuckets();
int totalBuckets = numBuckets;
if (restoredTotalBuckets != null) {
totalBuckets = restoredTotalBuckets;
}
if (!ignoreNumBucketCheck && totalBuckets != numBuckets) {
- String partInfo =
- partitionType.getFieldCount() > 0
- ? "partition "
- + getPartitionComputer(
- partitionType,
-
PARTITION_DEFAULT_NAME.defaultValue(),
- legacyPartitionName)
- .generatePartValues(partition)
- : "table";
throw new RuntimeException(
String.format(
"Try to write %s with a new bucket num %d, but the
previous bucket num is %d. "
@@ -570,6 +571,17 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
return restored;
}
+ private String partitionInfo(BinaryRow partition) {
+ return partitionType.getFieldCount() > 0
+ ? "partition "
+ + getPartitionComputer(
+ partitionType,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ legacyPartitionName)
+ .generatePartValues(partition)
+ : "table";
+ }
+
private ExecutorService compactExecutor() {
if (lazyCompactExecutor == null) {
lazyCompactExecutor =
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/WriteRestore.java
b/paimon-core/src/main/java/org/apache/paimon/operation/WriteRestore.java
index 5d4e335571..0d8db17313 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/WriteRestore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/WriteRestore.java
@@ -39,13 +39,20 @@ public interface WriteRestore {
@Nullable
static Integer extractDataFiles(List<ManifestEntry> entries,
List<DataFileMeta> dataFiles) {
+ return extractDataFiles(entries, dataFiles, null);
+ }
+
+ @Nullable
+ static Integer extractDataFiles(
+ List<ManifestEntry> entries, List<DataFileMeta> dataFiles,
@Nullable String context) {
Integer totalBuckets = null;
for (ManifestEntry entry : entries) {
if (totalBuckets != null && totalBuckets != entry.totalBuckets()) {
+ String contextInfo = context == null ? "" : " for " + context;
throw new RuntimeException(
String.format(
- "Bucket data files has different total bucket
number, %s vs %s, this should be a bug.",
- totalBuckets, entry.totalBuckets()));
+ "Bucket data files%s has different total
bucket number, %s vs %s, this should be a bug.",
+ contextInfo, totalBuckets,
entry.totalBuckets()));
}
totalBuckets = entry.totalBuckets();
dataFiles.add(entry.file());
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
index fca17b5b98..c0c3f5e738 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
@@ -195,7 +195,8 @@ public class AppendOnlySimpleTableTest extends
SimpleTableTestBase {
try (BatchTableWrite write = writeBuilder.newWrite()) {
if (ordered) {
assertThatThrownBy(() -> write.write(rowData(1, 10, 100L)))
- .hasMessageContaining("FileNotFoundException");
+ .hasMessageContaining("Failed to restore existing
files")
+
.hasRootCauseInstanceOf(java.io.FileNotFoundException.class);
} else {
// no exception
write.write(rowData(1, 10, 100L));
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java
index 1cff11eeff..4ca6228124 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java
@@ -28,6 +28,7 @@ import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.operation.FileStoreScan;
import org.apache.paimon.operation.WriteRestore;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.FileStorePathFactory;
import
org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache;
import
org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine;
@@ -42,6 +43,7 @@ import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
+import static org.apache.paimon.CoreOptions.PARTITION_DEFAULT_NAME;
import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
import static org.apache.paimon.utils.InstantiationUtil.deserializeObject;
import static org.apache.paimon.utils.InstantiationUtil.serializeObject;
@@ -169,7 +171,11 @@ public class TableWriteCoordinator {
List<DataFileMeta> restoreFiles = new ArrayList<>();
List<ManifestEntry> entries = scan.withPartitionBucket(partition,
bucket).plan().files();
- Integer totalBuckets = WriteRestore.extractDataFiles(entries,
restoreFiles);
+ Integer totalBuckets =
+ WriteRestore.extractDataFiles(
+ entries,
+ restoreFiles,
+ String.format("%s, bucket %d",
partitionInfo(partition), bucket));
IndexFileMeta dynamicBucketIndex = null;
if (request.scanDynamicBucketIndex()) {
@@ -212,6 +218,18 @@ public class TableWriteCoordinator {
latestCommittedIdentifiers.clear();
}
+ private String partitionInfo(BinaryRow partition) {
+ if (table.schema().logicalPartitionType().getFieldCount() == 0) {
+ return "table";
+ }
+ return "partition "
+ + FileStorePathFactory.getPartitionComputer(
+ table.schema().logicalPartitionType(),
+
table.coreOptions().toConfiguration().get(PARTITION_DEFAULT_NAME),
+ table.coreOptions().legacyPartitionName())
+ .generatePartValues(partition);
+ }
+
private static class CoordinationKey {
private final byte[] content;
