This is an automated email from the ASF dual-hosted git repository.
junhao 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 6ff03646f [core] Optimize UnawareAppendTableCompactionCoordinator to
avoid OOM (#4260)
6ff03646f is described below
commit 6ff03646f57a8948f0066e741fee741a1cf9de72
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Sep 26 14:00:45 2024 +0800
[core] Optimize UnawareAppendTableCompactionCoordinator to avoid OOM (#4260)
---
.../UnawareAppendTableCompactionCoordinator.java | 173 +++++++++++++--------
.../java/org/apache/paimon/manifest/FileEntry.java | 24 ++-
.../paimon/operation/AbstractFileStoreScan.java | 20 +++
.../org/apache/paimon/operation/FileStoreScan.java | 3 +
.../table/source/snapshot/SnapshotReader.java | 3 +
.../table/source/snapshot/SnapshotReaderImpl.java | 6 +
.../apache/paimon/table/system/AuditLogTable.java | 6 +
.../append/AppendOnlyTableCompactionTest.java | 6 +-
.../paimon/spark/procedure/CompactProcedure.java | 10 +-
9 files changed, 177 insertions(+), 74 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/UnawareAppendTableCompactionCoordinator.java
b/paimon-core/src/main/java/org/apache/paimon/append/UnawareAppendTableCompactionCoordinator.java
index 53068cc63..9a54ea72e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/UnawareAppendTableCompactionCoordinator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/UnawareAppendTableCompactionCoordinator.java
@@ -30,12 +30,10 @@ import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.EndOfScanException;
import org.apache.paimon.table.source.ScanMode;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.apache.paimon.utils.Filter;
-import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.SnapshotManager;
import javax.annotation.Nullable;
@@ -45,13 +43,14 @@ import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
-import static java.util.Collections.emptyList;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
/**
* Compact coordinator for append only tables.
@@ -70,19 +69,18 @@ import static java.util.Collections.emptyList;
*/
public class UnawareAppendTableCompactionCoordinator {
+ private static final int FILES_BATCH = 100_000;
+
protected static final int REMOVE_AGE = 10;
protected static final int COMPACT_AGE = 5;
private final SnapshotManager snapshotManager;
- private final SnapshotReader snapshotReader;
private final long targetFileSize;
private final long compactionFileSize;
private final int minFileNum;
private final int maxFileNum;
- private final boolean streamingMode;
private final DvMaintainerCache dvMaintainerCache;
-
- @Nullable private Long nextSnapshot = null;
+ private final FilesIterator filesIterator;
final Map<BinaryRow, PartitionCompactCoordinator>
partitionCompactCoordinators =
new HashMap<>();
@@ -97,13 +95,8 @@ public class UnawareAppendTableCompactionCoordinator {
public UnawareAppendTableCompactionCoordinator(
FileStoreTable table, boolean isStreaming, @Nullable Predicate
filter) {
- Preconditions.checkArgument(table.primaryKeys().isEmpty());
+ checkArgument(table.primaryKeys().isEmpty());
this.snapshotManager = table.snapshotManager();
- this.snapshotReader = table.newSnapshotReader();
- if (filter != null) {
- snapshotReader.withFilter(filter);
- }
- this.streamingMode = isStreaming;
CoreOptions options = table.coreOptions();
this.targetFileSize = options.targetFileSize(false);
this.compactionFileSize = options.compactionFileSize(false);
@@ -114,6 +107,7 @@ public class UnawareAppendTableCompactionCoordinator {
options.deletionVectorsEnabled()
? new
DvMaintainerCache(table.store().newIndexFileHandler())
: null;
+ this.filesIterator = new FilesIterator(table, isStreaming, filter);
}
public List<UnawareAppendCompactionTask> run() {
@@ -128,62 +122,36 @@ public class UnawareAppendTableCompactionCoordinator {
@VisibleForTesting
boolean scan() {
- List<DataSplit> splits;
- boolean hasResult = false;
- while (!(splits = plan()).isEmpty()) {
- hasResult = true;
- splits.forEach(split -> notifyNewFiles(split.partition(),
split.dataFiles()));
- // batch mode, we don't do continuous scanning
- if (!streamingMode) {
- break;
- }
- }
- return hasResult;
- }
-
- @VisibleForTesting
- List<DataSplit> plan() {
- if (nextSnapshot == null) {
- nextSnapshot = snapshotManager.latestSnapshotId();
- if (nextSnapshot == null) {
- return emptyList();
+ Map<BinaryRow, List<DataFileMeta>> files = new HashMap<>();
+ for (int i = 0; i < FILES_BATCH; i++) {
+ ManifestEntry entry;
+ try {
+ entry = filesIterator.next();
+ } catch (EndOfScanException e) {
+ if (!files.isEmpty()) {
+ files.forEach(this::notifyNewFiles);
+ return true;
+ }
+ throw e;
}
- snapshotReader.withMode(ScanMode.ALL);
- } else {
- if (!streamingMode) {
- throw new EndOfScanException();
+ if (entry == null) {
+ break;
}
- snapshotReader.withMode(ScanMode.DELTA);
+ BinaryRow partition = entry.partition();
+ files.computeIfAbsent(partition, k -> new
ArrayList<>()).add(entry.file());
}
- if (!snapshotManager.snapshotExists(nextSnapshot)) {
- return emptyList();
+ if (files.isEmpty()) {
+ return false;
}
- Snapshot snapshot = snapshotManager.snapshot(nextSnapshot);
- nextSnapshot++;
-
- if (dvMaintainerCache != null) {
- dvMaintainerCache.refresh();
- }
- Filter<ManifestEntry> entryFilter =
- entry -> {
- if (entry.file().fileSize() < compactionFileSize) {
- return true;
- }
+ files.forEach(this::notifyNewFiles);
+ return true;
+ }
- if (dvMaintainerCache != null) {
- return dvMaintainerCache
- .dvMaintainer(entry.partition())
- .hasDeletionFile(entry.fileName());
- }
- return false;
- };
- return snapshotReader
- .withManifestEntryFilter(entryFilter)
- .withSnapshot(snapshot)
- .read()
- .dataSplits();
+ @VisibleForTesting
+ FilesIterator filesIterator() {
+ return filesIterator;
}
@VisibleForTesting
@@ -395,4 +363,85 @@ public class UnawareAppendTableCompactionCoordinator {
return maintainer;
}
}
+
+ /** Iterator to read files. */
+ class FilesIterator {
+
+ private final SnapshotReader snapshotReader;
+ private final boolean streamingMode;
+
+ @Nullable private Long nextSnapshot = null;
+ @Nullable private Iterator<ManifestEntry> currentIterator;
+
+ public FilesIterator(
+ FileStoreTable table, boolean isStreaming, @Nullable Predicate
filter) {
+ this.snapshotReader = table.newSnapshotReader();
+ if (filter != null) {
+ snapshotReader.withFilter(filter);
+ }
+ this.streamingMode = isStreaming;
+ }
+
+ private void assignNewIterator() {
+ currentIterator = null;
+ if (nextSnapshot == null) {
+ nextSnapshot = snapshotManager.latestSnapshotId();
+ if (nextSnapshot == null) {
+ return;
+ }
+ snapshotReader.withMode(ScanMode.ALL);
+ } else {
+ if (!streamingMode) {
+ throw new EndOfScanException();
+ }
+ snapshotReader.withMode(ScanMode.DELTA);
+ }
+
+ if (!snapshotManager.snapshotExists(nextSnapshot)) {
+ return;
+ }
+
+ Snapshot snapshot = snapshotManager.snapshot(nextSnapshot);
+ nextSnapshot++;
+
+ if (dvMaintainerCache != null) {
+ dvMaintainerCache.refresh();
+ }
+ Filter<ManifestEntry> entryFilter =
+ entry -> {
+ if (entry.file().fileSize() < compactionFileSize) {
+ return true;
+ }
+
+ if (dvMaintainerCache != null) {
+ return dvMaintainerCache
+ .dvMaintainer(entry.partition())
+ .hasDeletionFile(entry.fileName());
+ }
+ return false;
+ };
+ currentIterator =
+ snapshotReader
+ .withManifestEntryFilter(entryFilter)
+ .withSnapshot(snapshot)
+ .readFileIterator();
+ }
+
+ @Nullable
+ public ManifestEntry next() {
+ while (true) {
+ if (currentIterator == null) {
+ assignNewIterator();
+ if (currentIterator == null) {
+ return null;
+ }
+ }
+
+ if (currentIterator.hasNext()) {
+ return currentIterator.next();
+ }
+ currentIterator = null;
+ }
+ }
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java
index 7f5fc9de6..3b3e514e0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java
@@ -211,20 +211,28 @@ public interface FileEntry {
ManifestFile manifestFile,
List<ManifestFileMeta> manifestFiles,
@Nullable Integer manifestReadParallelism) {
+ return readDeletedEntries(
+ m ->
+ manifestFile.read(
+ m.fileName(), m.fileSize(),
Filter.alwaysTrue(), deletedFilter()),
+ manifestFiles,
+ manifestReadParallelism);
+ }
+
+ static <T extends FileEntry> Set<Identifier> readDeletedEntries(
+ Function<ManifestFileMeta, List<T>> manifestReader,
+ List<ManifestFileMeta> manifestFiles,
+ @Nullable Integer manifestReadParallelism) {
manifestFiles =
manifestFiles.stream()
.filter(file -> file.numDeletedFiles() > 0)
.collect(Collectors.toList());
Function<ManifestFileMeta, List<Identifier>> processor =
file ->
- manifestFile
- .read(
- file.fileName(),
- file.fileSize(),
- Filter.alwaysTrue(),
- deletedFilter())
- .stream()
- .map(ManifestEntry::identifier)
+ manifestReader.apply(file).stream()
+ // filter again, ensure is delete
+ .filter(e -> e.kind() == FileKind.DELETE)
+ .map(FileEntry::identifier)
.collect(Collectors.toList());
Iterable<Identifier> identifiers =
sequentialBatchedExecute(processor, manifestFiles,
manifestReadParallelism);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
index dd89944cb..f2c15e1ce 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
@@ -23,6 +23,7 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.manifest.BucketEntry;
import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.ManifestCacheFilter;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestEntrySerializer;
@@ -42,14 +43,18 @@ import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
+
import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Consumer;
@@ -280,6 +285,21 @@ public abstract class AbstractFileStoreScan implements
FileStoreScan {
.collect(Collectors.toList());
}
+ @Override
+ public Iterator<ManifestEntry> readFileIterator() {
+ List<ManifestFileMeta> manifests = readManifests().filteredManifests;
+ Set<FileEntry.Identifier> deleteEntries =
+ FileEntry.readDeletedEntries(this::readSimpleEntries,
manifests, parallelism);
+ Iterator<ManifestEntry> iterator =
+ sequentialBatchedExecute(this::readManifest, manifests,
parallelism).iterator();
+ return Iterators.filter(
+ iterator,
+ entry ->
+ entry != null
+ && entry.kind() == FileKind.ADD
+ &&
!deleteEntries.contains(entry.identifier()));
+ }
+
private Pair<Snapshot, List<ManifestEntry>> doPlan() {
long started = System.nanoTime();
ManifestsReader.Result manifestsResult = readManifests();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
index fdae3f75b..bc0d7ff27 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
@@ -38,6 +38,7 @@ import org.apache.paimon.utils.Filter;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -111,6 +112,8 @@ public interface FileStoreScan {
List<BucketEntry> readBucketEntries();
+ Iterator<ManifestEntry> readFileIterator();
+
default List<BinaryRow> listPartitions() {
return readPartitionEntries().stream()
.map(PartitionEntry::partition)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
index bd07f49fd..2dd02be04 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
@@ -39,6 +39,7 @@ import org.apache.paimon.utils.SnapshotManager;
import javax.annotation.Nullable;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -103,6 +104,8 @@ public interface SnapshotReader {
List<BucketEntry> bucketEntries();
+ Iterator<ManifestEntry> readFileIterator();
+
/** Result plan of this scan. */
interface Plan extends TableScan.Plan {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
index b2fd1c785..f4591734b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
@@ -56,6 +56,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -363,6 +364,11 @@ public class SnapshotReaderImpl implements SnapshotReader {
return scan.readBucketEntries();
}
+ @Override
+ public Iterator<ManifestEntry> readFileIterator() {
+ return scan.readFileIterator();
+ }
+
@Override
public Plan readChanges() {
withMode(ScanMode.DELTA);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
index eab5617bc..bbbea7441 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
@@ -70,6 +70,7 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -382,6 +383,11 @@ public class AuditLogTable implements DataTable,
ReadonlyTable {
public List<BucketEntry> bucketEntries() {
return wrapped.bucketEntries();
}
+
+ @Override
+ public Iterator<ManifestEntry> readFileIterator() {
+ return wrapped.readFileIterator();
+ }
}
private class AuditLogBatchScan implements DataTableScan {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionTest.java
index ed513626a..20f2382b0 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionTest.java
@@ -131,7 +131,7 @@ public class AppendOnlyTableCompactionTest {
commit(messages);
tableSchema = tableSchema.copy(singletonMap("target-file-size", "1
b"));
recreate();
- assertThat(compactionCoordinator.plan()).isEmpty();
+ assertThat(compactionCoordinator.filesIterator().next()).isNull();
}
@Test
@@ -143,7 +143,9 @@ public class AppendOnlyTableCompactionTest {
for (int i = 90; i < 100; i++) {
count += i;
commit(writeCommit(i));
- commit(doCompact(compactionCoordinator.run()));
+ List<UnawareAppendCompactionTask> tasks =
compactionCoordinator.run();
+ assertThat(tasks).hasSizeGreaterThan(0);
+ commit(doCompact(tasks));
// scan the file generated itself
assertThat(compactionCoordinator.scan()).isTrue();
assertThat(
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
index 5b5f4e33c..71cf04cf5 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
@@ -43,6 +43,7 @@ import org.apache.paimon.table.sink.CommitMessageSerializer;
import org.apache.paimon.table.sink.CompactionTaskSerializer;
import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.EndOfScanException;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.ParameterUtils;
@@ -340,8 +341,13 @@ public class CompactProcedure extends BaseProcedure {
@Nullable Predicate filter,
@Nullable Duration partitionIdleTime,
JavaSparkContext javaSparkContext) {
- List<UnawareAppendCompactionTask> compactionTasks =
- new UnawareAppendTableCompactionCoordinator(table, false,
filter).run();
+ List<UnawareAppendCompactionTask> compactionTasks;
+ try {
+ compactionTasks =
+ new UnawareAppendTableCompactionCoordinator(table, false,
filter).run();
+ } catch (EndOfScanException e) {
+ compactionTasks = new ArrayList<>();
+ }
if (partitionIdleTime != null) {
Map<BinaryRow, Long> partitionInfo =
table.newSnapshotReader().partitionEntries().stream()
