You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by ju...@apache.org on 2025/04/21 08:11:52 UTC
(paimon) branch master updated: [core] Introduce 'compaction.delete-ratio-threshold' to reduce compaction (#5492)
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 30a7495fc4 [core] Introduce 'compaction.delete-ratio-threshold' to reduce compaction (#5492)
30a7495fc4 is described below
commit 30a7495fc49c2393eb0ce7edd89047b27aa79e24
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Apr 21 16:11:48 2025 +0800
[core] Introduce 'compaction.delete-ratio-threshold' to reduce compaction (#5492)
---
.../shortcodes/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 12 ++
.../org/apache/paimon/AppendOnlyFileStore.java | 14 +-
...rdinator.java => AppendCompactCoordinator.java} | 220 ++++++++++++---------
...dCompactionTask.java => AppendCompactTask.java} | 20 +-
...java => AppendPreCommitCompactCoordinator.java} | 4 +-
...nTask.java => MultiTableAppendCompactTask.java} | 6 +-
...tainer.java => AppendDeleteFileMaintainer.java} | 26 +--
...er.java => BaseAppendDeleteFileMaintainer.java} | 10 +-
...ava => BucketedAppendDeleteFileMaintainer.java} | 6 +-
.../paimon/manifest/IndexManifestFileHandler.java | 10 +-
...leStoreWrite.java => AppendFileStoreWrite.java} | 6 +-
...oreWrite.java => BaseAppendFileStoreWrite.java} | 6 +-
...rite.java => BucketedAppendFileStoreWrite.java} | 6 +-
.../paimon/table/AbstractFileStoreTable.java | 4 +-
.../paimon/table/AppendOnlyFileStoreTable.java | 4 +-
...lizer.java => AppendCompactTaskSerializer.java} | 28 ++-
...ractor.java => AppendTableRowKeyExtractor.java} | 4 +-
.../sink/MultiTableCompactionTaskSerializer.java | 27 ++-
.../org/apache/paimon/TestAppendFileStore.java | 4 +-
...Test.java => AppendCompactCoordinatorTest.java} | 93 ++++++---
.../append/AppendOnlyTableCompactionTest.java | 54 +++--
.../append/AppendDeletionFileMaintainerHelper.java | 6 +-
.../append/AppendDeletionFileMaintainerTest.java | 2 +-
....java => BucketedAppendFileStoreWriteTest.java} | 15 +-
...t.java => AppendCompactTaskSerializerTest.java} | 23 ++-
...st.java => AppendTableRowKeyExtractorTest.java} | 10 +-
...OnlyTableITCase.java => AppendTableITCase.java} | 4 +-
...OnlyTableITCase.java => AppendTableITCase.java} | 4 +-
...wareBucketSink.java => CdcAppendTableSink.java} | 6 +-
...rator.java => CdcAppendTableWriteOperator.java} | 10 +-
.../paimon/flink/sink/cdc/CdcSinkBuilder.java | 2 +-
.../sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java | 2 +-
.../apache/paimon/flink/action/CompactAction.java | 23 ++-
.../paimon/flink/action/CompactDatabaseAction.java | 10 +-
...AppendPreCommitCompactCoordinatorOperator.java} | 31 ++-
...a => AppendPreCommitCompactWorkerOperator.java} | 23 ++-
...Builder.java => AppendTableCompactBuilder.java} | 35 ++--
...ketCompactor.java => AppendTableCompactor.java} | 18 +-
...eScan.java => MultiAppendCompactTableScan.java} | 26 ++-
.../paimon/flink/compact/MultiTableScanBase.java | 6 +-
.../sink/AppendBypassCompactWorkerOperator.java | 10 +-
.../flink/sink/AppendCompactWorkerOperator.java | 15 +-
...pendOnlyMultiTableCompactionWorkerOperator.java | 32 ++-
...endOnlySingleTableCompactionWorkerOperator.java | 15 +-
...actionSink.java => AppendTableCompactSink.java} | 12 +-
...UnawareBucketSink.java => AppendTableSink.java} | 12 +-
.../flink/sink/CombinedTableCompactorSink.java | 8 +-
.../flink/sink/CompactionTaskSimpleSerializer.java | 16 +-
.../paimon/flink/sink/CompactionTaskTypeInfo.java | 20 +-
.../apache/paimon/flink/sink/FlinkSinkBuilder.java | 2 +-
.../sink/MultiTableCompactionTaskTypeInfo.java | 30 ++-
...wareBucketSink.java => RowAppendTableSink.java} | 6 +-
.../sink/RowDataHashPartitionChannelComputer.java | 4 +-
.../source/AppendBypassCoordinateOperator.java | 21 +-
.../AppendBypassCoordinateOperatorFactory.java | 13 +-
...ctSource.java => AppendTableCompactSource.java} | 37 ++--
.../CombinedTableCompactorSourceBuilder.java | 12 +-
...ource.java => CombinedAppendCompactSource.java} | 44 ++---
...java => CombinedAppendCompactStreamSource.java} | 29 ++-
.../source/operator/CombinedCompactorSource.java | 4 +-
...va => MultiTableAppendCompactReadOperator.java} | 20 +-
...OnlyTableITCase.java => AppendTableITCase.java} | 4 +-
...RangePartitionAndSortForAppendTableITCase.java} | 2 +-
... => SortCompactActionForAppendTableITCase.java} | 2 +-
...ndPreCommitCompactCoordinatorOperatorTest.java} | 45 +++--
...Case.java => AppendPreCommitCompactITCase.java} | 6 +-
...ctorTest.java => AppendTableCompactorTest.java} | 20 +-
...OnlyMultiTableCompactionWorkerOperatorTest.java | 8 +-
...nlySingleTableCompactionWorkerOperatorTest.java | 25 ++-
.../sink/CompactionTaskSimpleSerializerTest.java | 13 +-
.../paimon/spark/procedure/CompactProcedure.java | 27 ++-
.../paimon/spark/commands/PaimonSparkWriter.scala | 8 +-
.../spark/procedure/CompactProcedureTestBase.scala | 2 +-
74 files changed, 690 insertions(+), 660 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
index ae6fded4f8..12af27f81f 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -164,6 +164,12 @@ under the License.
<td>String</td>
<td>Specifies the commit user prefix.</td>
</tr>
+ <tr>
+ <td><h5>compaction.delete-ratio-threshold</h5></td>
+ <td style="word-wrap: break-word;">0.2</td>
+ <td>Double</td>
+ <td>Ratio of the deleted rows in a data file to be forced compacted for append-only table.</td>
+ </tr>
<tr>
<td><h5>compaction.force-up-level-0</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
index 33fc36ee9d..c3e9b0fba5 100644
--- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -636,6 +636,14 @@ public class CoreOptions implements Serializable {
"For file set [f_0,...,f_N], the minimum file number to trigger a compaction for "
+ "append-only table.");
+ public static final ConfigOption<Double> COMPACTION_DELETE_RATIO_THRESHOLD =
+ key("compaction.delete-ratio-threshold")
+ .doubleType()
+ .defaultValue(0.2)
+ .withDescription(
+ "Ratio of the deleted rows in a data file to be forced compacted for "
+ + "append-only table.");
+
public static final ConfigOption<ChangelogProducer> CHANGELOG_PRODUCER =
key("changelog-producer")
.enumType(ChangelogProducer.class)
@@ -2195,6 +2203,10 @@ public class CoreOptions implements Serializable {
return options.get(COMPACTION_MIN_FILE_NUM);
}
+ public double compactionDeleteRatioThreshold() {
+ return options.get(COMPACTION_DELETE_RATIO_THRESHOLD);
+ }
+
public long dynamicBucketTargetRowNum() {
return options.get(DYNAMIC_BUCKET_TARGET_ROW_NUM);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java
index dc5171a744..b901681050 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java
@@ -23,11 +23,11 @@ import org.apache.paimon.deletionvectors.DeletionVectorsMaintainer;
import org.apache.paimon.format.FileFormatDiscover;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.manifest.ManifestCacheFilter;
+import org.apache.paimon.operation.AppendFileStoreWrite;
import org.apache.paimon.operation.AppendOnlyFileStoreScan;
-import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
-import org.apache.paimon.operation.AppendOnlyFixedBucketFileStoreWrite;
-import org.apache.paimon.operation.AppendOnlyUnawareBucketFileStoreWrite;
+import org.apache.paimon.operation.BaseAppendFileStoreWrite;
import org.apache.paimon.operation.BucketSelectConverter;
+import org.apache.paimon.operation.BucketedAppendFileStoreWrite;
import org.apache.paimon.operation.RawFileSplitRead;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.schema.SchemaManager;
@@ -88,19 +88,19 @@ public class AppendOnlyFileStore extends AbstractFileStore<InternalRow> {
}
@Override
- public AppendOnlyFileStoreWrite newWrite(String commitUser) {
+ public BaseAppendFileStoreWrite newWrite(String commitUser) {
return newWrite(commitUser, null);
}
@Override
- public AppendOnlyFileStoreWrite newWrite(
+ public BaseAppendFileStoreWrite newWrite(
String commitUser, ManifestCacheFilter manifestFilter) {
DeletionVectorsMaintainer.Factory dvMaintainerFactory =
options.deletionVectorsEnabled()
? DeletionVectorsMaintainer.factory(newIndexFileHandler())
: null;
if (bucketMode() == BucketMode.BUCKET_UNAWARE) {
- return new AppendOnlyUnawareBucketFileStoreWrite(
+ return new AppendFileStoreWrite(
fileIO,
newRead(),
schema.id(),
@@ -113,7 +113,7 @@ public class AppendOnlyFileStore extends AbstractFileStore<InternalRow> {
dvMaintainerFactory,
tableName);
} else {
- return new AppendOnlyFixedBucketFileStoreWrite(
+ return new BucketedAppendFileStoreWrite(
fileIO,
newRead(),
schema.id(),
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/UnawareAppendTableCompactionCoordinator.java b/paimon-core/src/main/java/org/apache/paimon/append/AppendCompactCoordinator.java
similarity index 66%
rename from paimon-core/src/main/java/org/apache/paimon/append/UnawareAppendTableCompactionCoordinator.java
rename to paimon-core/src/main/java/org/apache/paimon/append/AppendCompactCoordinator.java
index 565bd5d99d..1fead5e612 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/UnawareAppendTableCompactionCoordinator.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/AppendCompactCoordinator.java
@@ -22,19 +22,18 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.data.BinaryRow;
-import org.apache.paimon.deletionvectors.append.AppendDeletionFileMaintainer;
-import org.apache.paimon.deletionvectors.append.UnawareAppendDeletionFileMaintainer;
+import org.apache.paimon.deletionvectors.append.AppendDeleteFileMaintainer;
+import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
import org.apache.paimon.index.IndexFileHandler;
-import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DeletionFile;
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.SnapshotManager;
import javax.annotation.Nullable;
@@ -56,19 +55,19 @@ import static org.apache.paimon.utils.Preconditions.checkArgument;
/**
* Compact coordinator for append only tables.
*
- * <p>Note: {@link UnawareAppendTableCompactionCoordinator} scan files in snapshot, read APPEND and
- * COMPACT snapshot then load those new files. It will try it best to generate compaction task for
- * the restored files scanned in snapshot, but to reduce memory usage, it won't remain single file
- * for a long time. After ten times scan, single file with one partition will be ignored and removed
- * from memory, which means, it will not participate in compaction again until restart the
- * compaction job.
+ * <p>Note: {@link AppendCompactCoordinator} scan files in snapshot, read APPEND and COMPACT
+ * snapshot then load those new files. It will try it best to generate compaction task for the
+ * restored files scanned in snapshot, but to reduce memory usage, it won't remain single file for a
+ * long time. After ten times scan, single file with one partition will be ignored and removed from
+ * memory, which means, it will not participate in compaction again until restart the compaction
+ * job.
*
* <p>When a third task delete file in latest snapshot(including batch delete/update and overwrite),
* the file in coordinator will still remain and participate in compaction task. When this happens,
* compaction job will fail in commit stage, and fail-over to rescan the restored files in latest
* snapshot.
*/
-public class UnawareAppendTableCompactionCoordinator {
+public class AppendCompactCoordinator {
private static final int FILES_BATCH = 100_000;
@@ -78,29 +77,26 @@ public class UnawareAppendTableCompactionCoordinator {
private final SnapshotManager snapshotManager;
private final long targetFileSize;
private final long compactionFileSize;
+ private final double deleteThreshold;
private final long openFileCost;
private final int minFileNum;
private final DvMaintainerCache dvMaintainerCache;
private final FilesIterator filesIterator;
- final Map<BinaryRow, PartitionCompactCoordinator> partitionCompactCoordinators =
- new HashMap<>();
+ final Map<BinaryRow, SubCoordinator> subCoordinators = new HashMap<>();
- public UnawareAppendTableCompactionCoordinator(FileStoreTable table) {
- this(table, true);
- }
-
- public UnawareAppendTableCompactionCoordinator(FileStoreTable table, boolean isStreaming) {
+ public AppendCompactCoordinator(FileStoreTable table, boolean isStreaming) {
this(table, isStreaming, null);
}
- public UnawareAppendTableCompactionCoordinator(
+ public AppendCompactCoordinator(
FileStoreTable table, boolean isStreaming, @Nullable Predicate filter) {
checkArgument(table.primaryKeys().isEmpty());
this.snapshotManager = table.snapshotManager();
CoreOptions options = table.coreOptions();
this.targetFileSize = options.targetFileSize(false);
this.compactionFileSize = options.compactionFileSize(false);
+ this.deleteThreshold = options.compactionDeleteRatioThreshold();
this.openFileCost = options.splitOpenFileCost();
this.minFileNum = options.compactionMinFileNum();
this.dvMaintainerCache =
@@ -110,7 +106,7 @@ public class UnawareAppendTableCompactionCoordinator {
this.filesIterator = new FilesIterator(table, isStreaming, filter);
}
- public List<UnawareAppendCompactionTask> run() {
+ public List<AppendCompactTask> run() {
// scan files in snapshot
if (scan()) {
// do plan compact tasks
@@ -156,66 +152,53 @@ public class UnawareAppendTableCompactionCoordinator {
@VisibleForTesting
void notifyNewFiles(BinaryRow partition, List<DataFileMeta> files) {
- java.util.function.Predicate<DataFileMeta> filter =
- file -> {
- if (dvMaintainerCache == null
- || dvMaintainerCache
- .dvMaintainer(partition)
- .getDeletionFile(file.fileName())
- == null) {
- return file.fileSize() < compactionFileSize;
- }
- // if a data file has a deletion file, always be to compact.
- return true;
- };
- List<DataFileMeta> toCompact = files.stream().filter(filter).collect(Collectors.toList());
- partitionCompactCoordinators
- .computeIfAbsent(partition, pp -> new PartitionCompactCoordinator(partition))
+ List<DataFileMeta> toCompact =
+ files.stream()
+ .filter(file -> shouldCompact(partition, file))
+ .collect(Collectors.toList());
+ subCoordinators
+ .computeIfAbsent(partition, pp -> new SubCoordinator(partition))
.addFiles(toCompact);
}
@VisibleForTesting
// generate compaction task to the next stage
- List<UnawareAppendCompactionTask> compactPlan() {
+ List<AppendCompactTask> compactPlan() {
// first loop to found compaction tasks
- List<UnawareAppendCompactionTask> tasks =
- partitionCompactCoordinators.values().stream()
+ List<AppendCompactTask> tasks =
+ subCoordinators.values().stream()
.flatMap(s -> s.plan().stream())
.collect(Collectors.toList());
// second loop to eliminate empty or old(with only one file) coordinator
- new ArrayList<>(partitionCompactCoordinators.values())
+ new ArrayList<>(subCoordinators.values())
.stream()
- .filter(PartitionCompactCoordinator::readyToRemove)
- .map(PartitionCompactCoordinator::partition)
- .forEach(partitionCompactCoordinators::remove);
+ .filter(SubCoordinator::readyToRemove)
+ .map(SubCoordinator::partition)
+ .forEach(subCoordinators::remove);
return tasks;
}
@VisibleForTesting
HashSet<DataFileMeta> listRestoredFiles() {
- HashSet<DataFileMeta> sets = new HashSet<>();
- partitionCompactCoordinators
- .values()
- .forEach(
- partitionCompactCoordinator ->
- sets.addAll(partitionCompactCoordinator.toCompact));
- return sets;
+ HashSet<DataFileMeta> result = new HashSet<>();
+ subCoordinators.values().stream().map(SubCoordinator::toCompact).forEach(result::addAll);
+ return result;
}
/** Coordinator for a single partition. */
- class PartitionCompactCoordinator {
+ class SubCoordinator {
private final BinaryRow partition;
private final HashSet<DataFileMeta> toCompact = new HashSet<>();
int age = 0;
- public PartitionCompactCoordinator(BinaryRow partition) {
+ public SubCoordinator(BinaryRow partition) {
this.partition = partition;
}
- public List<UnawareAppendCompactionTask> plan() {
+ public List<AppendCompactTask> plan() {
return pickCompact();
}
@@ -223,10 +206,14 @@ public class UnawareAppendTableCompactionCoordinator {
return partition;
}
- private List<UnawareAppendCompactionTask> pickCompact() {
+ public HashSet<DataFileMeta> toCompact() {
+ return toCompact;
+ }
+
+ private List<AppendCompactTask> pickCompact() {
List<List<DataFileMeta>> waitCompact = agePack();
return waitCompact.stream()
- .map(files -> new UnawareAppendCompactionTask(partition, files))
+ .map(files -> new AppendCompactTask(partition, files))
.collect(Collectors.toList());
}
@@ -262,38 +249,40 @@ public class UnawareAppendTableCompactionCoordinator {
}
private List<List<DataFileMeta>> pack(Set<DataFileMeta> toCompact) {
- // we compact smaller files first
- // step 1, sort files by file size, pick the smaller first
+ // we don't know how many parallel compact works there should be, so in order to pack
+ // better, we will sort them first
ArrayList<DataFileMeta> files = new ArrayList<>(toCompact);
files.sort(Comparator.comparingLong(DataFileMeta::fileSize));
- // step 2, when files picked size greater than targetFileSize(meanwhile file num greater
- // than minFileNum) or file numbers bigger than maxFileNum, we pack it to a compaction
- // task
List<List<DataFileMeta>> result = new ArrayList<>();
FileBin fileBin = new FileBin();
for (DataFileMeta fileMeta : files) {
fileBin.addFile(fileMeta);
- if (fileBin.binFull()) {
- result.add(new ArrayList<>(fileBin.bin));
- // remove it from coordinator memory, won't join in compaction again
- fileBin.reset();
+ if (fileBin.enoughContent()) {
+ result.add(fileBin.drain());
}
}
- if (fileBin.fileNum >= minFileNum) {
- result.add(new ArrayList<>(fileBin.bin));
- fileBin.reset();
+
+ if (fileBin.enoughInputFiles()) {
+ result.add(fileBin.drain());
}
+ // else skip these small files that are too few
+
return result;
}
private List<List<DataFileMeta>> packInDeletionVectorVMode(Set<DataFileMeta> toCompact) {
// we group the data files by their related index files.
- Map<IndexFileMeta, List<DataFileMeta>> filesWithDV = new HashMap<>();
+ // In the subsequent compact task, if any files with deletion vectors are compacted, we
+ // need to rewrite their corresponding deleted files. To avoid duplicate deleted files,
+ // we must group them according to the deleted files
+ Map<String, List<DataFileMeta>> filesWithDV = new HashMap<>();
Set<DataFileMeta> rest = new HashSet<>();
for (DataFileMeta dataFile : toCompact) {
- IndexFileMeta indexFile =
- dvMaintainerCache.dvMaintainer(partition).getIndexFile(dataFile.fileName());
+ String indexFile =
+ dvMaintainerCache
+ .dvMaintainer(partition)
+ .getIndexFilePath(dataFile.fileName());
if (indexFile == null) {
rest.add(dataFile);
} else {
@@ -301,36 +290,62 @@ public class UnawareAppendTableCompactionCoordinator {
}
}
- List<List<DataFileMeta>> result = new ArrayList<>(filesWithDV.values());
+ // To avoid too small a compact task, merge them
+ List<List<DataFileMeta>> dvGroups = new ArrayList<>(filesWithDV.values());
+ dvGroups.sort(Comparator.comparingLong(this::fileSizeOfList));
+
+ List<List<DataFileMeta>> result = new ArrayList<>();
+ FileBin fileBin = new FileBin();
+ for (List<DataFileMeta> dvGroup : dvGroups) {
+ fileBin.addFiles(dvGroup);
+ if (fileBin.enoughContent()) {
+ result.add(fileBin.drain());
+ }
+ }
+
+ // for file with deletion vectors, must do compaction
+ if (!fileBin.bin.isEmpty()) {
+ result.add(fileBin.drain());
+ }
+
if (rest.size() > 1) {
result.addAll(pack(rest));
}
return result;
}
- /**
- * A file bin for {@link PartitionCompactCoordinator} determine whether ready to compact.
- */
+ private long fileSizeOfList(List<DataFileMeta> list) {
+ return list.stream().mapToLong(DataFileMeta::fileSize).sum();
+ }
+
+ /** A file bin for {@link SubCoordinator} determine whether ready to compact. */
private class FileBin {
List<DataFileMeta> bin = new ArrayList<>();
long totalFileSize = 0;
- int fileNum = 0;
- public void reset() {
+ public List<DataFileMeta> drain() {
+ List<DataFileMeta> result = new ArrayList<>(bin);
bin.forEach(toCompact::remove);
bin.clear();
totalFileSize = 0;
- fileNum = 0;
+ return result;
+ }
+
+ public void addFiles(List<DataFileMeta> files) {
+ files.forEach(this::addFile);
}
public void addFile(DataFileMeta file) {
totalFileSize += file.fileSize() + openFileCost;
- fileNum++;
bin.add(file);
}
- public boolean binFull() {
- return totalFileSize >= targetFileSize * 50 && fileNum >= minFileNum;
+ private boolean enoughContent() {
+ return bin.size() > 1 && totalFileSize >= targetFileSize * 2;
+ }
+
+ private boolean enoughInputFiles() {
+ return bin.size() >= minFileNum;
}
}
}
@@ -340,8 +355,7 @@ public class UnawareAppendTableCompactionCoordinator {
private final IndexFileHandler indexFileHandler;
/** Should be thread safe, ManifestEntryFilter will be invoked in many threads. */
- private final Map<BinaryRow, UnawareAppendDeletionFileMaintainer> cache =
- new ConcurrentHashMap<>();
+ private final Map<BinaryRow, AppendDeleteFileMaintainer> cache = new ConcurrentHashMap<>();
private DvMaintainerCache(IndexFileHandler indexFileHandler) {
this.indexFileHandler = indexFileHandler;
@@ -351,12 +365,12 @@ public class UnawareAppendTableCompactionCoordinator {
this.cache.clear();
}
- private UnawareAppendDeletionFileMaintainer dvMaintainer(BinaryRow partition) {
- UnawareAppendDeletionFileMaintainer maintainer = cache.get(partition);
+ private AppendDeleteFileMaintainer dvMaintainer(BinaryRow partition) {
+ AppendDeleteFileMaintainer maintainer = cache.get(partition);
if (maintainer == null) {
synchronized (this) {
maintainer =
- AppendDeletionFileMaintainer.forUnawareAppend(
+ BaseAppendDeleteFileMaintainer.forUnawareAppend(
indexFileHandler, snapshotManager.latestSnapshot(), partition);
}
cache.put(partition, maintainer);
@@ -415,22 +429,10 @@ public class UnawareAppendTableCompactionCoordinator {
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)
+ .withManifestEntryFilter(
+ entry -> shouldCompact(entry.partition(), entry.file()))
.withSnapshot(snapshot)
.readFileIterator();
}
@@ -457,4 +459,26 @@ public class UnawareAppendTableCompactionCoordinator {
}
}
}
+
+ private boolean shouldCompact(BinaryRow partition, DataFileMeta file) {
+ return file.fileSize() < compactionFileSize || tooHighDeleteRatio(partition, file);
+ }
+
+ private boolean tooHighDeleteRatio(BinaryRow partition, DataFileMeta file) {
+ if (dvMaintainerCache != null) {
+ DeletionFile deletionFile =
+ dvMaintainerCache.dvMaintainer(partition).getDeletionFile(file.fileName());
+ if (deletionFile != null) {
+ Long cardinality = deletionFile.cardinality();
+ long rowCount = file.rowCount();
+ return cardinality == null || cardinality > rowCount * deleteThreshold;
+ }
+ }
+ return false;
+ }
+
+ @VisibleForTesting
+ AppendDeleteFileMaintainer dvMaintainer(BinaryRow partition) {
+ return dvMaintainerCache.dvMaintainer(partition);
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/UnawareAppendCompactionTask.java b/paimon-core/src/main/java/org/apache/paimon/append/AppendCompactTask.java
similarity index 87%
rename from paimon-core/src/main/java/org/apache/paimon/append/UnawareAppendCompactionTask.java
rename to paimon-core/src/main/java/org/apache/paimon/append/AppendCompactTask.java
index 37c3b78af0..af8997f981 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/UnawareAppendCompactionTask.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/AppendCompactTask.java
@@ -19,8 +19,8 @@
package org.apache.paimon.append;
import org.apache.paimon.data.BinaryRow;
-import org.apache.paimon.deletionvectors.append.AppendDeletionFileMaintainer;
-import org.apache.paimon.deletionvectors.append.UnawareAppendDeletionFileMaintainer;
+import org.apache.paimon.deletionvectors.append.AppendDeleteFileMaintainer;
+import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
@@ -28,7 +28,7 @@ import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.io.IndexIncrement;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
-import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
+import org.apache.paimon.operation.BaseAppendFileStoreWrite;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
@@ -42,14 +42,14 @@ import java.util.stream.Collectors;
import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET;
-/** Compaction task generated by {@link UnawareAppendTableCompactionCoordinator}. */
-public class UnawareAppendCompactionTask {
+/** Compaction task generated by {@link AppendCompactCoordinator}. */
+public class AppendCompactTask {
private final BinaryRow partition;
private final List<DataFileMeta> compactBefore;
private final List<DataFileMeta> compactAfter;
- public UnawareAppendCompactionTask(BinaryRow partition, List<DataFileMeta> files) {
+ public AppendCompactTask(BinaryRow partition, List<DataFileMeta> files) {
Preconditions.checkArgument(files != null);
this.partition = partition;
compactBefore = new ArrayList<>(files);
@@ -68,7 +68,7 @@ public class UnawareAppendCompactionTask {
return compactAfter;
}
- public CommitMessage doCompact(FileStoreTable table, AppendOnlyFileStoreWrite write)
+ public CommitMessage doCompact(FileStoreTable table, BaseAppendFileStoreWrite write)
throws Exception {
boolean dvEnabled = table.coreOptions().deletionVectorsEnabled();
Preconditions.checkArgument(
@@ -76,8 +76,8 @@ public class UnawareAppendCompactionTask {
"AppendOnlyCompactionTask need more than one file input.");
IndexIncrement indexIncrement;
if (dvEnabled) {
- UnawareAppendDeletionFileMaintainer dvIndexFileMaintainer =
- AppendDeletionFileMaintainer.forUnawareAppend(
+ AppendDeleteFileMaintainer dvIndexFileMaintainer =
+ BaseAppendDeleteFileMaintainer.forUnawareAppend(
table.store().newIndexFileHandler(),
table.snapshotManager().latestSnapshot(),
partition);
@@ -130,7 +130,7 @@ public class UnawareAppendCompactionTask {
return false;
}
- UnawareAppendCompactionTask that = (UnawareAppendCompactionTask) o;
+ AppendCompactTask that = (AppendCompactTask) o;
return Objects.equals(partition, that.partition)
&& Objects.equals(compactBefore, that.compactBefore)
&& Objects.equals(compactAfter, that.compactAfter);
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/UnawareBucketNewFilesCompactionCoordinator.java b/paimon-core/src/main/java/org/apache/paimon/append/AppendPreCommitCompactCoordinator.java
similarity index 95%
rename from paimon-core/src/main/java/org/apache/paimon/append/UnawareBucketNewFilesCompactionCoordinator.java
rename to paimon-core/src/main/java/org/apache/paimon/append/AppendPreCommitCompactCoordinator.java
index 4425d12a25..ef49fa5636 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/UnawareBucketNewFilesCompactionCoordinator.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/AppendPreCommitCompactCoordinator.java
@@ -30,12 +30,12 @@ import java.util.Optional;
import java.util.stream.Collectors;
/** Buffer files from the same partition, until their total size reaches {@code targetFileSize}. */
-public class UnawareBucketNewFilesCompactionCoordinator {
+public class AppendPreCommitCompactCoordinator {
private final long targetFileSize;
private final Map<BinaryRow, PartitionFiles> partitions;
- public UnawareBucketNewFilesCompactionCoordinator(long targetFileSize) {
+ public AppendPreCommitCompactCoordinator(long targetFileSize) {
this.targetFileSize = targetFileSize;
this.partitions = new LinkedHashMap<>();
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/MultiTableUnawareAppendCompactionTask.java b/paimon-core/src/main/java/org/apache/paimon/append/MultiTableAppendCompactTask.java
similarity index 89%
rename from paimon-core/src/main/java/org/apache/paimon/append/MultiTableUnawareAppendCompactionTask.java
rename to paimon-core/src/main/java/org/apache/paimon/append/MultiTableAppendCompactTask.java
index 37f322a973..eb178b690c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/MultiTableUnawareAppendCompactionTask.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/MultiTableAppendCompactTask.java
@@ -26,11 +26,11 @@ import java.util.List;
import java.util.Objects;
/** Compaction task for multi table . */
-public class MultiTableUnawareAppendCompactionTask extends UnawareAppendCompactionTask {
+public class MultiTableAppendCompactTask extends AppendCompactTask {
private final Identifier tableIdentifier;
- public MultiTableUnawareAppendCompactionTask(
+ public MultiTableAppendCompactTask(
BinaryRow partition, List<DataFileMeta> files, Identifier identifier) {
super(partition, files);
this.tableIdentifier = identifier;
@@ -53,7 +53,7 @@ public class MultiTableUnawareAppendCompactionTask extends UnawareAppendCompacti
return false;
}
- MultiTableUnawareAppendCompactionTask that = (MultiTableUnawareAppendCompactionTask) o;
+ MultiTableAppendCompactTask that = (MultiTableAppendCompactTask) o;
return Objects.equals(partition(), that.partition())
&& Objects.equals(compactBefore(), that.compactBefore())
&& Objects.equals(compactAfter(), that.compactAfter())
diff --git a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/UnawareAppendDeletionFileMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeleteFileMaintainer.java
similarity index 91%
rename from paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/UnawareAppendDeletionFileMaintainer.java
rename to paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeleteFileMaintainer.java
index abbdeea205..38e1176de7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/UnawareAppendDeletionFileMaintainer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeleteFileMaintainer.java
@@ -40,8 +40,8 @@ import java.util.stream.Collectors;
import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET;
-/** A {@link AppendDeletionFileMaintainer} of unaware bucket append table. */
-public class UnawareAppendDeletionFileMaintainer implements AppendDeletionFileMaintainer {
+/** A {@link BaseAppendDeleteFileMaintainer} of unaware bucket append table. */
+public class AppendDeleteFileMaintainer implements BaseAppendDeleteFileMaintainer {
private final IndexFileHandler indexFileHandler;
@@ -56,13 +56,13 @@ public class UnawareAppendDeletionFileMaintainer implements AppendDeletionFileMa
private final DeletionVectorsMaintainer maintainer;
- UnawareAppendDeletionFileMaintainer(
+ AppendDeleteFileMaintainer(
IndexFileHandler indexFileHandler,
BinaryRow partition,
Map<String, DeletionFile> deletionFiles) {
this.indexFileHandler = indexFileHandler;
this.partition = partition;
- this.dataFileToDeletionFile = deletionFiles;
+ this.dataFileToDeletionFile = new HashMap<>(deletionFiles);
// the deletion of data files is independent
// just create an empty maintainer
this.maintainer = new DeletionVectorsMaintainer.Factory(indexFileHandler).create();
@@ -100,14 +100,14 @@ public class UnawareAppendDeletionFileMaintainer implements AppendDeletionFileMa
return UNAWARE_BUCKET;
}
- public boolean hasDeletionFile(String dataFile) {
- return this.dataFileToDeletionFile.containsKey(dataFile);
- }
-
public DeletionFile getDeletionFile(String dataFile) {
return this.dataFileToDeletionFile.get(dataFile);
}
+ public void putDeletionFile(String dataFile, DeletionFile deletionFile) {
+ this.dataFileToDeletionFile.put(dataFile, deletionFile);
+ }
+
public DeletionVector getDeletionVector(String dataFile) {
DeletionFile deletionFile = getDeletionFile(dataFile);
if (deletionFile != null) {
@@ -151,15 +151,9 @@ public class UnawareAppendDeletionFileMaintainer implements AppendDeletionFileMa
return result;
}
- public IndexFileMeta getIndexFile(String dataFile) {
+ public String getIndexFilePath(String dataFile) {
DeletionFile deletionFile = getDeletionFile(dataFile);
- if (deletionFile == null) {
- return null;
- } else {
- IndexManifestEntry entry =
- this.indexNameToEntry.get(new Path(deletionFile.path()).getName());
- return entry == null ? null : entry.indexFile();
- }
+ return deletionFile == null ? null : deletionFile.path();
}
@VisibleForTesting
diff --git a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BaseAppendDeleteFileMaintainer.java
similarity index 87%
rename from paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainer.java
rename to paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BaseAppendDeleteFileMaintainer.java
index 4455605e9a..d5f3d570c7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BaseAppendDeleteFileMaintainer.java
@@ -42,7 +42,7 @@ import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET;
* <li>{@link #persist}: persist deletion files to commit.
* </ul>
*/
-public interface AppendDeletionFileMaintainer {
+public interface BaseAppendDeleteFileMaintainer {
BinaryRow getPartition();
@@ -52,7 +52,7 @@ public interface AppendDeletionFileMaintainer {
List<IndexManifestEntry> persist();
- static BucketedAppendDeletionFileMaintainer forBucketedAppend(
+ static BucketedAppendDeleteFileMaintainer forBucketedAppend(
IndexFileHandler indexFileHandler,
@Nullable Snapshot snapshot,
BinaryRow partition,
@@ -62,13 +62,13 @@ public interface AppendDeletionFileMaintainer {
DeletionVectorsMaintainer maintainer =
new DeletionVectorsMaintainer.Factory(indexFileHandler)
.createOrRestore(snapshot, partition, bucket);
- return new BucketedAppendDeletionFileMaintainer(partition, bucket, maintainer);
+ return new BucketedAppendDeleteFileMaintainer(partition, bucket, maintainer);
}
- static UnawareAppendDeletionFileMaintainer forUnawareAppend(
+ static AppendDeleteFileMaintainer forUnawareAppend(
IndexFileHandler indexFileHandler, @Nullable Snapshot snapshot, BinaryRow partition) {
Map<String, DeletionFile> deletionFiles =
indexFileHandler.scanDVIndex(snapshot, partition, UNAWARE_BUCKET);
- return new UnawareAppendDeletionFileMaintainer(indexFileHandler, partition, deletionFiles);
+ return new AppendDeleteFileMaintainer(indexFileHandler, partition, deletionFiles);
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BucketedAppendDeletionFileMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BucketedAppendDeleteFileMaintainer.java
similarity index 91%
rename from paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BucketedAppendDeletionFileMaintainer.java
rename to paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BucketedAppendDeleteFileMaintainer.java
index 261573f843..847caae499 100644
--- a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BucketedAppendDeletionFileMaintainer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BucketedAppendDeleteFileMaintainer.java
@@ -27,14 +27,14 @@ import org.apache.paimon.manifest.IndexManifestEntry;
import java.util.List;
import java.util.stream.Collectors;
-/** A {@link AppendDeletionFileMaintainer} of bucketed append table. */
-public class BucketedAppendDeletionFileMaintainer implements AppendDeletionFileMaintainer {
+/** A {@link BaseAppendDeleteFileMaintainer} of bucketed append table. */
+public class BucketedAppendDeleteFileMaintainer implements BaseAppendDeleteFileMaintainer {
private final BinaryRow partition;
private final int bucket;
private final DeletionVectorsMaintainer maintainer;
- BucketedAppendDeletionFileMaintainer(
+ BucketedAppendDeleteFileMaintainer(
BinaryRow partition, int bucket, DeletionVectorsMaintainer maintainer) {
this.partition = partition;
this.bucket = bucket;
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java
index b5719555f8..bf334e4ad3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java
@@ -94,9 +94,9 @@ public class IndexManifestFileHandler {
private IndexManifestFileCombiner getIndexManifestFileCombine(String indexType) {
if (DELETION_VECTORS_INDEX.equals(indexType) && BucketMode.BUCKET_UNAWARE == bucketMode) {
- return new UnawareBucketCombiner();
+ return new GlobalCombiner();
} else {
- return new CommonBucketCombiner();
+ return new BucketedCombiner();
}
}
@@ -107,9 +107,9 @@ public class IndexManifestFileHandler {
/**
* We combine the previous and new index files by the file name. This is only used for tables
- * with UnawareBucket.
+ * without bucket.
*/
- static class UnawareBucketCombiner implements IndexManifestFileCombiner {
+ static class GlobalCombiner implements IndexManifestFileCombiner {
@Override
public List<IndexManifestEntry> combine(
@@ -131,7 +131,7 @@ public class IndexManifestFileHandler {
}
/** We combine the previous and new index files by {@link BucketIdentifier}. */
- static class CommonBucketCombiner implements IndexManifestFileCombiner {
+ static class BucketedCombiner implements IndexManifestFileCombiner {
@Override
public List<IndexManifestEntry> combine(
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyUnawareBucketFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/AppendFileStoreWrite.java
similarity index 93%
rename from paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyUnawareBucketFileStoreWrite.java
rename to paimon-core/src/main/java/org/apache/paimon/operation/AppendFileStoreWrite.java
index e509b58994..3f1e701255 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyUnawareBucketFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/AppendFileStoreWrite.java
@@ -37,11 +37,11 @@ import java.util.concurrent.ExecutorService;
import java.util.function.Function;
/**
- * {@link AppendOnlyFileStoreWrite} for {@link org.apache.paimon.table.BucketMode#BUCKET_UNAWARE}.
+ * {@link BaseAppendFileStoreWrite} for {@link org.apache.paimon.table.BucketMode#BUCKET_UNAWARE}.
*/
-public class AppendOnlyUnawareBucketFileStoreWrite extends AppendOnlyFileStoreWrite {
+public class AppendFileStoreWrite extends BaseAppendFileStoreWrite {
- public AppendOnlyUnawareBucketFileStoreWrite(
+ public AppendFileStoreWrite(
FileIO fileIO,
RawFileSplitRead read,
long schemaId,
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
similarity index 97%
rename from paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreWrite.java
rename to paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
index 83372d8dbf..5e57d9088a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
@@ -59,10 +59,10 @@ import java.util.concurrent.ExecutorService;
import java.util.function.Function;
/** {@link FileStoreWrite} for {@link AppendOnlyFileStore}. */
-public abstract class AppendOnlyFileStoreWrite extends MemoryFileStoreWrite<InternalRow>
+public abstract class BaseAppendFileStoreWrite extends MemoryFileStoreWrite<InternalRow>
implements BundleFileStoreWriter {
- private static final Logger LOG = LoggerFactory.getLogger(AppendOnlyFileStoreWrite.class);
+ private static final Logger LOG = LoggerFactory.getLogger(BaseAppendFileStoreWrite.class);
private final FileIO fileIO;
private final RawFileSplitRead read;
@@ -75,7 +75,7 @@ public abstract class AppendOnlyFileStoreWrite extends MemoryFileStoreWrite<Inte
private final FileIndexOptions fileIndexOptions;
private boolean forceBufferSpill = false;
- public AppendOnlyFileStoreWrite(
+ public BaseAppendFileStoreWrite(
FileIO fileIO,
RawFileSplitRead read,
long schemaId,
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFixedBucketFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java
similarity index 95%
rename from paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFixedBucketFileStoreWrite.java
rename to paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java
index c11390e588..36201c7f7f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFixedBucketFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java
@@ -38,12 +38,12 @@ import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.function.Function;
-/** {@link AppendOnlyFileStoreWrite} for {@link org.apache.paimon.table.BucketMode#HASH_FIXED}. */
-public class AppendOnlyFixedBucketFileStoreWrite extends AppendOnlyFileStoreWrite {
+/** {@link BaseAppendFileStoreWrite} for {@link org.apache.paimon.table.BucketMode#HASH_FIXED}. */
+public class BucketedAppendFileStoreWrite extends BaseAppendFileStoreWrite {
private final String commitUser;
- public AppendOnlyFixedBucketFileStoreWrite(
+ public BucketedAppendFileStoreWrite(
FileIO fileIO,
RawFileSplitRead read,
long schemaId,
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
index f68bd9615e..f8090918eb 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
@@ -36,13 +36,13 @@ import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.SchemaValidation;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.stats.Statistics;
+import org.apache.paimon.table.sink.AppendTableRowKeyExtractor;
import org.apache.paimon.table.sink.DynamicBucketRowKeyExtractor;
import org.apache.paimon.table.sink.FixedBucketRowKeyExtractor;
import org.apache.paimon.table.sink.FixedBucketWriteSelector;
import org.apache.paimon.table.sink.RowKeyExtractor;
import org.apache.paimon.table.sink.RowKindGenerator;
import org.apache.paimon.table.sink.TableCommitImpl;
-import org.apache.paimon.table.sink.UnawareBucketRowKeyExtractor;
import org.apache.paimon.table.sink.WriteSelector;
import org.apache.paimon.table.source.DataTableBatchScan;
import org.apache.paimon.table.source.DataTableStreamScan;
@@ -239,7 +239,7 @@ abstract class AbstractFileStoreTable implements FileStoreTable {
case CROSS_PARTITION:
return new DynamicBucketRowKeyExtractor(schema());
case BUCKET_UNAWARE:
- return new UnawareBucketRowKeyExtractor(schema());
+ return new AppendTableRowKeyExtractor(schema());
default:
throw new UnsupportedOperationException("Unsupported mode: " + bucketMode());
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
index 26e5dd5a2b..1fece2596f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
@@ -25,7 +25,7 @@ import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.ManifestCacheFilter;
import org.apache.paimon.operation.AppendOnlyFileStoreScan;
-import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
+import org.apache.paimon.operation.BaseAppendFileStoreWrite;
import org.apache.paimon.operation.FileStoreScan;
import org.apache.paimon.operation.RawFileSplitRead;
import org.apache.paimon.predicate.Predicate;
@@ -136,7 +136,7 @@ public class AppendOnlyFileStoreTable extends AbstractFileStoreTable {
@Override
public TableWriteImpl<InternalRow> newWrite(
String commitUser, ManifestCacheFilter manifestFilter) {
- AppendOnlyFileStoreWrite writer = store().newWrite(commitUser, manifestFilter);
+ BaseAppendFileStoreWrite writer = store().newWrite(commitUser, manifestFilter);
return new TableWriteImpl<>(
rowType(),
writer,
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/CompactionTaskSerializer.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/AppendCompactTaskSerializer.java
similarity index 76%
rename from paimon-core/src/main/java/org/apache/paimon/table/sink/CompactionTaskSerializer.java
rename to paimon-core/src/main/java/org/apache/paimon/table/sink/AppendCompactTaskSerializer.java
index 2f145b2982..ed78ec2f7f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/CompactionTaskSerializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/AppendCompactTaskSerializer.java
@@ -18,7 +18,7 @@
package org.apache.paimon.table.sink;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.data.serializer.VersionedSerializer;
import org.apache.paimon.io.DataFileMetaSerializer;
import org.apache.paimon.io.DataInputDeserializer;
@@ -34,14 +34,14 @@ import java.util.List;
import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
-/** Serializer for {@link UnawareAppendCompactionTask}. */
-public class CompactionTaskSerializer implements VersionedSerializer<UnawareAppendCompactionTask> {
+/** Serializer for {@link AppendCompactTask}. */
+public class AppendCompactTaskSerializer implements VersionedSerializer<AppendCompactTask> {
private static final int CURRENT_VERSION = 2;
private final DataFileMetaSerializer dataFileSerializer;
- public CompactionTaskSerializer() {
+ public AppendCompactTaskSerializer() {
this.dataFileSerializer = new DataFileMetaSerializer();
}
@@ -51,40 +51,38 @@ public class CompactionTaskSerializer implements VersionedSerializer<UnawareAppe
}
@Override
- public byte[] serialize(UnawareAppendCompactionTask obj) throws IOException {
+ public byte[] serialize(AppendCompactTask obj) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
serialize(obj, view);
return out.toByteArray();
}
- public void serializeList(List<UnawareAppendCompactionTask> list, DataOutputView view)
+ public void serializeList(List<AppendCompactTask> list, DataOutputView view)
throws IOException {
view.writeInt(list.size());
- for (UnawareAppendCompactionTask commitMessage : list) {
+ for (AppendCompactTask commitMessage : list) {
serialize(commitMessage, view);
}
}
- private void serialize(UnawareAppendCompactionTask task, DataOutputView view)
- throws IOException {
+ private void serialize(AppendCompactTask task, DataOutputView view) throws IOException {
serializeBinaryRow(task.partition(), view);
dataFileSerializer.serializeList(task.compactBefore(), view);
}
@Override
- public UnawareAppendCompactionTask deserialize(int version, byte[] serialized)
- throws IOException {
+ public AppendCompactTask deserialize(int version, byte[] serialized) throws IOException {
checkVersion(version);
DataInputDeserializer view = new DataInputDeserializer(serialized);
return deserialize(view);
}
- public List<UnawareAppendCompactionTask> deserializeList(int version, DataInputView view)
+ public List<AppendCompactTask> deserializeList(int version, DataInputView view)
throws IOException {
checkVersion(version);
int length = view.readInt();
- List<UnawareAppendCompactionTask> list = new ArrayList<>(length);
+ List<AppendCompactTask> list = new ArrayList<>(length);
for (int i = 0; i < length; i++) {
list.add(deserialize(view));
}
@@ -103,8 +101,8 @@ public class CompactionTaskSerializer implements VersionedSerializer<UnawareAppe
}
}
- private UnawareAppendCompactionTask deserialize(DataInputView view) throws IOException {
- return new UnawareAppendCompactionTask(
+ private AppendCompactTask deserialize(DataInputView view) throws IOException {
+ return new AppendCompactTask(
deserializeBinaryRow(view), dataFileSerializer.deserializeList(view));
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/UnawareBucketRowKeyExtractor.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/AppendTableRowKeyExtractor.java
similarity index 92%
rename from paimon-core/src/main/java/org/apache/paimon/table/sink/UnawareBucketRowKeyExtractor.java
rename to paimon-core/src/main/java/org/apache/paimon/table/sink/AppendTableRowKeyExtractor.java
index ad8a3cd830..b4cc41107c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/UnawareBucketRowKeyExtractor.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/AppendTableRowKeyExtractor.java
@@ -25,9 +25,9 @@ import org.apache.paimon.table.BucketMode;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Unaware bucket table put all records into bucket 0 (for comparability with old design). */
-public class UnawareBucketRowKeyExtractor extends RowKeyExtractor {
+public class AppendTableRowKeyExtractor extends RowKeyExtractor {
- public UnawareBucketRowKeyExtractor(TableSchema schema) {
+ public AppendTableRowKeyExtractor(TableSchema schema) {
super(schema);
int numBuckets = new CoreOptions(schema.options()).bucket();
checkArgument(
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/MultiTableCompactionTaskSerializer.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/MultiTableCompactionTaskSerializer.java
index c828ab56b6..8478d04ea3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/MultiTableCompactionTaskSerializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/MultiTableCompactionTaskSerializer.java
@@ -18,7 +18,7 @@
package org.apache.paimon.table.sink;
-import org.apache.paimon.append.MultiTableUnawareAppendCompactionTask;
+import org.apache.paimon.append.MultiTableAppendCompactTask;
import org.apache.paimon.data.serializer.VersionedSerializer;
import org.apache.paimon.io.DataFileMetaSerializer;
import org.apache.paimon.io.DataInputDeserializer;
@@ -35,9 +35,9 @@ import java.util.List;
import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
-/** Serializer for {@link MultiTableUnawareAppendCompactionTask}. */
+/** Serializer for {@link MultiTableAppendCompactTask}. */
public class MultiTableCompactionTaskSerializer
- implements VersionedSerializer<MultiTableUnawareAppendCompactionTask> {
+ implements VersionedSerializer<MultiTableAppendCompactTask> {
private static final int CURRENT_VERSION = 1;
@@ -56,14 +56,14 @@ public class MultiTableCompactionTaskSerializer
}
@Override
- public byte[] serialize(MultiTableUnawareAppendCompactionTask task) throws IOException {
+ public byte[] serialize(MultiTableAppendCompactTask task) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
serialize(task, view);
return out.toByteArray();
}
- private void serialize(MultiTableUnawareAppendCompactionTask task, DataOutputView view)
+ private void serialize(MultiTableAppendCompactTask task, DataOutputView view)
throws IOException {
serializeBinaryRow(task.partition(), view);
dataFileSerializer.serializeList(task.compactBefore(), view);
@@ -71,36 +71,35 @@ public class MultiTableCompactionTaskSerializer
}
@Override
- public MultiTableUnawareAppendCompactionTask deserialize(int version, byte[] serialized)
+ public MultiTableAppendCompactTask deserialize(int version, byte[] serialized)
throws IOException {
checkVersion(version);
DataInputDeserializer view = new DataInputDeserializer(serialized);
return deserialize(view);
}
- private MultiTableUnawareAppendCompactionTask deserialize(DataInputView view)
- throws IOException {
- return new MultiTableUnawareAppendCompactionTask(
+ private MultiTableAppendCompactTask deserialize(DataInputView view) throws IOException {
+ return new MultiTableAppendCompactTask(
deserializeBinaryRow(view),
dataFileSerializer.deserializeList(view),
identifierSerializer.deserialize(view));
}
- public List<MultiTableUnawareAppendCompactionTask> deserializeList(
- int version, DataInputView view) throws IOException {
+ public List<MultiTableAppendCompactTask> deserializeList(int version, DataInputView view)
+ throws IOException {
checkVersion(version);
int length = view.readInt();
- List<MultiTableUnawareAppendCompactionTask> list = new ArrayList<>(length);
+ List<MultiTableAppendCompactTask> list = new ArrayList<>(length);
for (int i = 0; i < length; i++) {
list.add(deserialize(view));
}
return list;
}
- public void serializeList(List<MultiTableUnawareAppendCompactionTask> list, DataOutputView view)
+ public void serializeList(List<MultiTableAppendCompactTask> list, DataOutputView view)
throws IOException {
view.writeInt(list.size());
- for (MultiTableUnawareAppendCompactionTask commitMessage : list) {
+ for (MultiTableAppendCompactTask commitMessage : list) {
serialize(commitMessage, view);
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/TestAppendFileStore.java b/paimon-core/src/test/java/org/apache/paimon/TestAppendFileStore.java
index aa629a67bc..1a5279c5bf 100644
--- a/paimon-core/src/test/java/org/apache/paimon/TestAppendFileStore.java
+++ b/paimon-core/src/test/java/org/apache/paimon/TestAppendFileStore.java
@@ -20,8 +20,8 @@ package org.apache.paimon;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.deletionvectors.DeletionVectorsMaintainer;
+import org.apache.paimon.deletionvectors.append.AppendDeleteFileMaintainer;
import org.apache.paimon.deletionvectors.append.AppendDeletionFileMaintainerHelper;
-import org.apache.paimon.deletionvectors.append.UnawareAppendDeletionFileMaintainer;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileIOFinder;
import org.apache.paimon.fs.Path;
@@ -120,7 +120,7 @@ public class TestAppendFileStore extends AppendOnlyFileStore {
return fileHandler.scan(latestSnapshot, DELETION_VECTORS_INDEX, partition, bucket);
}
- public UnawareAppendDeletionFileMaintainer createDVIFMaintainer(
+ public AppendDeleteFileMaintainer createDVIFMaintainer(
BinaryRow partition, Map<String, DeletionFile> dataFileToDeletionFiles) {
return AppendDeletionFileMaintainerHelper.fromDeletionFiles(
fileHandler, partition, dataFileToDeletionFiles);
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/UnawareAppendTableCompactionCoordinatorTest.java b/paimon-core/src/test/java/org/apache/paimon/append/AppendCompactCoordinatorTest.java
similarity index 68%
rename from paimon-core/src/test/java/org/apache/paimon/append/UnawareAppendTableCompactionCoordinatorTest.java
rename to paimon-core/src/test/java/org/apache/paimon/append/AppendCompactCoordinatorTest.java
index 13f21cdfe0..e166086515 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/UnawareAppendTableCompactionCoordinatorTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/AppendCompactCoordinatorTest.java
@@ -18,8 +18,8 @@
package org.apache.paimon.append;
-import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.deletionvectors.append.AppendDeleteFileMaintainer;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.DataFileMeta;
@@ -29,6 +29,7 @@ import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.table.source.EndOfScanException;
import org.apache.paimon.types.DataTypes;
@@ -42,24 +43,26 @@ import java.util.HashSet;
import java.util.List;
import java.util.UUID;
+import static org.apache.paimon.CoreOptions.COMPACTION_MIN_FILE_NUM;
+import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
import static org.apache.paimon.mergetree.compact.MergeTreeCompactManagerTest.row;
import static org.apache.paimon.stats.StatsTestUtils.newSimpleStats;
import static org.apache.paimon.testutils.assertj.PaimonAssertions.anyCauseMatches;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-/** Tests for {@link UnawareAppendTableCompactionCoordinator}. */
-public class UnawareAppendTableCompactionCoordinatorTest {
+/** Tests for {@link AppendCompactCoordinator}. */
+public class AppendCompactCoordinatorTest {
@TempDir Path tempDir;
private FileStoreTable appendOnlyFileStoreTable;
- private UnawareAppendTableCompactionCoordinator compactionCoordinator;
+ private AppendCompactCoordinator compactionCoordinator;
private BinaryRow partition;
@Test
public void testForCompactPlan() {
List<DataFileMeta> files = generateNewFiles(200, 0);
- assertTasks(files, 1);
+ assertTasks(files, 2);
}
@Test
@@ -73,7 +76,7 @@ public class UnawareAppendTableCompactionCoordinatorTest {
List<DataFileMeta> files =
generateNewFiles(
100, appendOnlyFileStoreTable.coreOptions().targetFileSize(false) / 3 + 1);
- assertTasks(files, 1);
+ assertTasks(files, 17);
}
@Test
@@ -91,14 +94,14 @@ public class UnawareAppendTableCompactionCoordinatorTest {
1000, appendOnlyFileStoreTable.coreOptions().targetFileSize(false) / 10);
compactionCoordinator.notifyNewFiles(partition, files);
- assertThat(compactionCoordinator.compactPlan().size()).isEqualTo(3);
+ assertThat(compactionCoordinator.compactPlan().size()).isEqualTo(56);
files.clear();
files =
generateNewFiles(
1050, appendOnlyFileStoreTable.coreOptions().targetFileSize(false) / 5);
compactionCoordinator.notifyNewFiles(partition, files);
- assertThat(compactionCoordinator.compactPlan().size()).isEqualTo(5);
+ assertThat(compactionCoordinator.compactPlan().size()).isEqualTo(105);
}
@Test
@@ -107,7 +110,7 @@ public class UnawareAppendTableCompactionCoordinatorTest {
generateNewFiles(
1089, appendOnlyFileStoreTable.coreOptions().targetFileSize(false) / 5);
compactionCoordinator.notifyNewFiles(partition, files);
- assertThat(compactionCoordinator.compactPlan().size()).isEqualTo(5);
+ assertThat(compactionCoordinator.compactPlan().size()).isEqualTo(109);
}
@Test
@@ -115,14 +118,14 @@ public class UnawareAppendTableCompactionCoordinatorTest {
List<DataFileMeta> files = generateNewFiles(1, 0);
compactionCoordinator.notifyNewFiles(partition, files);
- for (int i = 0; i < UnawareAppendTableCompactionCoordinator.REMOVE_AGE; i++) {
+ for (int i = 0; i < AppendCompactCoordinator.REMOVE_AGE; i++) {
assertThat(compactionCoordinator.compactPlan().size()).isEqualTo(0);
- assertThat(compactionCoordinator.partitionCompactCoordinators.size()).isEqualTo(1);
+ assertThat(compactionCoordinator.subCoordinators.size()).isEqualTo(1);
}
// age enough, eliminate partitionCoordinator
assertThat(compactionCoordinator.compactPlan().size()).isEqualTo(0);
- assertThat(compactionCoordinator.partitionCompactCoordinators.size()).isEqualTo(0);
+ assertThat(compactionCoordinator.subCoordinators.size()).isEqualTo(0);
}
@Test
@@ -130,17 +133,17 @@ public class UnawareAppendTableCompactionCoordinatorTest {
List<DataFileMeta> files = generateNewFiles(2, 0);
compactionCoordinator.notifyNewFiles(partition, files);
- for (int i = 0; i < UnawareAppendTableCompactionCoordinator.COMPACT_AGE; i++) {
+ for (int i = 0; i < AppendCompactCoordinator.COMPACT_AGE; i++) {
assertThat(compactionCoordinator.compactPlan().size()).isEqualTo(0);
- assertThat(compactionCoordinator.partitionCompactCoordinators.size()).isEqualTo(1);
+ assertThat(compactionCoordinator.subCoordinators.size()).isEqualTo(1);
}
// age enough, generate less file compaction
- List<UnawareAppendCompactionTask> tasks = compactionCoordinator.compactPlan();
+ List<AppendCompactTask> tasks = compactionCoordinator.compactPlan();
assertThat(tasks.size()).isEqualTo(1);
assertThat(new HashSet<>(files))
.containsExactlyInAnyOrderElementsOf(tasks.get(0).compactBefore());
- assertThat(compactionCoordinator.partitionCompactCoordinators.size()).isEqualTo(0);
+ assertThat(compactionCoordinator.subCoordinators.size()).isEqualTo(0);
}
@Test
@@ -148,33 +151,63 @@ public class UnawareAppendTableCompactionCoordinatorTest {
List<DataFileMeta> files = generateNewFiles(1, 0);
compactionCoordinator.notifyNewFiles(partition, files);
- for (int i = 0; i < UnawareAppendTableCompactionCoordinator.REMOVE_AGE; i++) {
+ for (int i = 0; i < AppendCompactCoordinator.REMOVE_AGE; i++) {
compactionCoordinator.compactPlan();
- assertThat(compactionCoordinator.partitionCompactCoordinators.size()).isEqualTo(1);
- assertThat(compactionCoordinator.partitionCompactCoordinators.get(partition).age)
- .isEqualTo(i + 1);
+ assertThat(compactionCoordinator.subCoordinators.size()).isEqualTo(1);
+ assertThat(compactionCoordinator.subCoordinators.get(partition).age).isEqualTo(i + 1);
}
// clear age
compactionCoordinator.notifyNewFiles(partition, generateNewFiles(1, 0));
- assertThat(compactionCoordinator.partitionCompactCoordinators.size()).isEqualTo(1);
+ assertThat(compactionCoordinator.subCoordinators.size()).isEqualTo(1);
// check whether age goes to zero again
- assertThat(compactionCoordinator.partitionCompactCoordinators.get(partition).age)
- .isEqualTo(0);
+ assertThat(compactionCoordinator.subCoordinators.get(partition).age).isEqualTo(0);
}
@Test
public void testBatchScanEmptyTable() {
- compactionCoordinator =
- new UnawareAppendTableCompactionCoordinator(appendOnlyFileStoreTable, false);
+ compactionCoordinator = new AppendCompactCoordinator(appendOnlyFileStoreTable, false);
assertThatThrownBy(() -> compactionCoordinator.scan())
.satisfies(anyCauseMatches(EndOfScanException.class));
}
+ @Test
+ public void testDeleteRatio() {
+ List<DataFileMeta> files =
+ generateNewFiles(2, appendOnlyFileStoreTable.coreOptions().targetFileSize(false));
+ AppendDeleteFileMaintainer dvMaintainer = compactionCoordinator.dvMaintainer(partition);
+ DeletionFile deletionFile = new DeletionFile("", 0, 0, 21L);
+ dvMaintainer.putDeletionFile(files.get(0).fileName(), deletionFile);
+ assertTasks(files, 1);
+ }
+
+ @Test
+ public void testDeleteRatioNoCompact() {
+ List<DataFileMeta> files =
+ generateNewFiles(2, appendOnlyFileStoreTable.coreOptions().targetFileSize(false));
+ AppendDeleteFileMaintainer dvMaintainer = compactionCoordinator.dvMaintainer(partition);
+ DeletionFile deletionFile = new DeletionFile("", 0, 0, 11L);
+ dvMaintainer.putDeletionFile(files.get(0).fileName(), deletionFile);
+ assertTasks(files, 0);
+ }
+
+ @Test
+ public void testDeleteRatioMergeTask() {
+ List<DataFileMeta> files = generateNewFiles(100, 1);
+ AppendDeleteFileMaintainer dvMaintainer = compactionCoordinator.dvMaintainer(partition);
+ for (int i = 0; i < files.size(); i++) {
+ DeletionFile deletionFile = new DeletionFile("" + i, 0, 0, 21L);
+ dvMaintainer.putDeletionFile(files.get(i).fileName(), deletionFile);
+ }
+
+ // with merging, should be 1
+ assertTasks(files, 1);
+ }
+
private void assertTasks(List<DataFileMeta> files, int taskNum) {
compactionCoordinator.notifyNewFiles(partition, files);
- List<UnawareAppendCompactionTask> tasks = compactionCoordinator.compactPlan();
+ List<AppendCompactTask> tasks = compactionCoordinator.compactPlan();
assertThat(tasks.size()).isEqualTo(taskNum);
}
@@ -184,7 +217,8 @@ public class UnawareAppendTableCompactionCoordinatorTest {
schemaBuilder.column("f1", DataTypes.STRING());
schemaBuilder.column("f2", DataTypes.STRING());
schemaBuilder.column("f3", DataTypes.STRING());
- schemaBuilder.option(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "3");
+ schemaBuilder.option(COMPACTION_MIN_FILE_NUM.key(), "3");
+ schemaBuilder.option(DELETION_VECTORS_ENABLED.key(), "true");
return schemaBuilder.build();
}
@@ -198,8 +232,7 @@ public class UnawareAppendTableCompactionCoordinatorTest {
appendOnlyFileStoreTable =
FileStoreTableFactory.create(
fileIO, new org.apache.paimon.fs.Path(tempDir.toString()), tableSchema);
- compactionCoordinator =
- new UnawareAppendTableCompactionCoordinator(appendOnlyFileStoreTable);
+ compactionCoordinator = new AppendCompactCoordinator(appendOnlyFileStoreTable, true);
partition = BinaryRow.EMPTY_ROW;
}
@@ -215,7 +248,7 @@ public class UnawareAppendTableCompactionCoordinatorTest {
return new DataFileMeta(
UUID.randomUUID().toString(),
fileSize,
- 1,
+ 100,
row(0),
row(0),
newSimpleStats(0, 1),
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 544684fd9f..5af16aa316 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
@@ -24,7 +24,7 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.DataFileMeta;
-import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
+import org.apache.paimon.operation.BaseAppendFileStoreWrite;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
@@ -59,8 +59,8 @@ public class AppendOnlyTableCompactionTest {
@TempDir private Path tempDir;
private FileStoreTable appendOnlyFileStoreTable;
private SnapshotManager snapshotManager;
- private UnawareAppendTableCompactionCoordinator compactionCoordinator;
- private AppendOnlyFileStoreWrite write;
+ private AppendCompactCoordinator coordinator;
+ private BaseAppendFileStoreWrite write;
private org.apache.paimon.fs.Path path;
private TableSchema tableSchema;
private final String commitUser = UUID.randomUUID().toString();
@@ -80,9 +80,8 @@ public class AppendOnlyTableCompactionTest {
LocalFileIO.create(),
new org.apache.paimon.fs.Path(tempDir.toString()),
tableSchema);
- compactionCoordinator =
- new UnawareAppendTableCompactionCoordinator(appendOnlyFileStoreTable);
- write = (AppendOnlyFileStoreWrite) appendOnlyFileStoreTable.store().newWrite(commitUser);
+ coordinator = new AppendCompactCoordinator(appendOnlyFileStoreTable, true);
+ write = (BaseAppendFileStoreWrite) appendOnlyFileStoreTable.store().newWrite(commitUser);
}
@Test
@@ -103,29 +102,29 @@ public class AppendOnlyTableCompactionTest {
// first compact, six files left after commit compact and update restored files
// test run method invoke scan and compactPlan
- List<UnawareAppendCompactionTask> tasks = compactionCoordinator.run();
+ List<AppendCompactTask> tasks = coordinator.run();
assertThat(tasks.size()).isEqualTo(1);
- UnawareAppendCompactionTask task = tasks.get(0);
+ AppendCompactTask task = tasks.get(0);
assertThat(task.compactBefore().size()).isEqualTo(11);
List<CommitMessage> result = doCompact(tasks);
assertThat(result.size()).isEqualTo(1);
commit(result);
- compactionCoordinator.scan();
- assertThat(compactionCoordinator.listRestoredFiles().size()).isEqualTo(1);
+ coordinator.scan();
+ assertThat(coordinator.listRestoredFiles().size()).isEqualTo(1);
messages = writeCommit(11);
commit(messages);
// second compact, only one file left after updateRestored
- compactionCoordinator.scan();
- assertThat(compactionCoordinator.listRestoredFiles().size()).isEqualTo(12);
- tasks = compactionCoordinator.compactPlan();
+ coordinator.scan();
+ assertThat(coordinator.listRestoredFiles().size()).isEqualTo(12);
+ tasks = coordinator.compactPlan();
assertThat(tasks.size()).isEqualTo(1);
// before update, zero file left
- assertThat(compactionCoordinator.listRestoredFiles().size()).isEqualTo(0);
+ assertThat(coordinator.listRestoredFiles().size()).isEqualTo(0);
commit(doCompact(tasks));
- compactionCoordinator.scan();
+ coordinator.scan();
// one file is loaded from delta
- List<DataFileMeta> last = new ArrayList<>(compactionCoordinator.listRestoredFiles());
+ List<DataFileMeta> last = new ArrayList<>(coordinator.listRestoredFiles());
assertThat(last.size()).isEqualTo(1);
assertThat(last.get(0).rowCount()).isEqualTo(22);
}
@@ -136,7 +135,7 @@ public class AppendOnlyTableCompactionTest {
commit(messages);
tableSchema = tableSchema.copy(singletonMap("target-file-size", "1 b"));
recreate();
- assertThat(compactionCoordinator.filesIterator().next()).isNull();
+ assertThat(coordinator.filesIterator().next()).isNull();
}
@Test
@@ -148,13 +147,13 @@ public class AppendOnlyTableCompactionTest {
for (int i = 90; i < 100; i++) {
count += i;
commit(writeCommit(i));
- List<UnawareAppendCompactionTask> tasks = compactionCoordinator.run();
+ List<AppendCompactTask> tasks = coordinator.run();
assertThat(tasks).hasSizeGreaterThan(0);
commit(doCompact(tasks));
// scan the file generated itself
- assertThat(compactionCoordinator.scan()).isTrue();
+ assertThat(coordinator.scan()).isTrue();
assertThat(
- compactionCoordinator.listRestoredFiles().stream()
+ coordinator.listRestoredFiles().stream()
.map(DataFileMeta::rowCount)
.reduce(Long::sum)
.get())
@@ -162,18 +161,16 @@ public class AppendOnlyTableCompactionTest {
}
assertThat(appendOnlyFileStoreTable.store().newScan().plan().files().size())
- .isEqualTo(compactionCoordinator.listRestoredFiles().size());
+ .isEqualTo(coordinator.listRestoredFiles().size());
- List<UnawareAppendCompactionTask> tasks = compactionCoordinator.run();
+ List<AppendCompactTask> tasks = coordinator.run();
while (tasks.size() != 0) {
commit(doCompact(tasks));
- tasks = compactionCoordinator.run();
+ tasks = coordinator.run();
}
int remainedSize = appendOnlyFileStoreTable.store().newScan().plan().files().size();
- assertThat(remainedSize)
- .isEqualTo(compactionCoordinator.listRestoredFiles().size())
- .isEqualTo(1);
+ assertThat(remainedSize).isEqualTo(coordinator.listRestoredFiles().size()).isEqualTo(1);
}
private static Schema schema() {
@@ -203,10 +200,9 @@ public class AppendOnlyTableCompactionTest {
return messages;
}
- private List<CommitMessage> doCompact(List<UnawareAppendCompactionTask> tasks)
- throws Exception {
+ private List<CommitMessage> doCompact(List<AppendCompactTask> tasks) throws Exception {
List<CommitMessage> result = new ArrayList<>();
- for (UnawareAppendCompactionTask task : tasks) {
+ for (AppendCompactTask task : tasks) {
result.add(task.doCompact(appendOnlyFileStoreTable, write));
}
return result;
diff --git a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainerHelper.java b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainerHelper.java
index dea50a7a16..c6048b95f4 100644
--- a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainerHelper.java
+++ b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainerHelper.java
@@ -24,13 +24,13 @@ import org.apache.paimon.table.source.DeletionFile;
import java.util.Map;
-/** Helper for {@link AppendDeletionFileMaintainer}. */
+/** Helper for {@link BaseAppendDeleteFileMaintainer}. */
public class AppendDeletionFileMaintainerHelper {
- public static UnawareAppendDeletionFileMaintainer fromDeletionFiles(
+ public static AppendDeleteFileMaintainer fromDeletionFiles(
IndexFileHandler indexFileHandler,
BinaryRow partition,
Map<String, DeletionFile> deletionFiles) {
- return new UnawareAppendDeletionFileMaintainer(indexFileHandler, partition, deletionFiles);
+ return new AppendDeleteFileMaintainer(indexFileHandler, partition, deletionFiles);
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainerTest.java b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainerTest.java
index a52819c805..e9ce57f30a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainerTest.java
@@ -70,7 +70,7 @@ class AppendDeletionFileMaintainerTest {
createDeletionFileMapFromIndexFileMetas(
indexPathFactory, commitMessage2.indexIncrement().newIndexFiles()));
- UnawareAppendDeletionFileMaintainer dvIFMaintainer =
+ AppendDeleteFileMaintainer dvIFMaintainer =
store.createDVIFMaintainer(BinaryRow.EMPTY_ROW, dataFileToDeletionFiles);
// no dv should be rewritten, because nothing is changed.
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/AppendOnlyFileStoreWriteTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/BucketedAppendFileStoreWriteTest.java
similarity index 95%
rename from paimon-core/src/test/java/org/apache/paimon/operation/AppendOnlyFileStoreWriteTest.java
rename to paimon-core/src/test/java/org/apache/paimon/operation/BucketedAppendFileStoreWriteTest.java
index 6384729b81..39f2ba79d6 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/AppendOnlyFileStoreWriteTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/BucketedAppendFileStoreWriteTest.java
@@ -51,8 +51,8 @@ import java.util.Random;
import static org.apache.paimon.CoreOptions.WRITE_MAX_WRITERS_TO_SPILL;
-/** Tests for {@link AppendOnlyFileStoreWrite}. */
-public class AppendOnlyFileStoreWriteTest {
+/** Tests for {@link BucketedAppendFileStoreWrite}. */
+public class BucketedAppendFileStoreWriteTest {
private static final Random RANDOM = new Random();
@@ -63,7 +63,8 @@ public class AppendOnlyFileStoreWriteTest {
FileStoreTable table = createFileStoreTable();
table = table.copy(Collections.singletonMap(WRITE_MAX_WRITERS_TO_SPILL.key(), "5"));
- AppendOnlyFileStoreWrite write = (AppendOnlyFileStoreWrite) table.store().newWrite("ss");
+ BucketedAppendFileStoreWrite write =
+ (BucketedAppendFileStoreWrite) table.store().newWrite("ss");
write.withIOManager(IOManager.create(tempDir.toString()));
write.withExecutionMode(false);
@@ -117,7 +118,7 @@ public class AppendOnlyFileStoreWriteTest {
public void testWritesInBatchWithNoExtraFiles() throws Exception {
FileStoreTable table = createFileStoreTable();
- AppendOnlyFileStoreWrite write = (AppendOnlyFileStoreWrite) table.store().newWrite("ss");
+ BaseAppendFileStoreWrite write = (BaseAppendFileStoreWrite) table.store().newWrite("ss");
write.withExecutionMode(false);
write.write(partition(0), 0, GenericRow.of(0, 0, 0));
@@ -185,7 +186,7 @@ public class AppendOnlyFileStoreWriteTest {
public void testScanFilterWithMixedPartitionWrite() throws Exception {
FileStoreTable table = createFileStoreTable();
- AppendOnlyFileStoreWrite write = (AppendOnlyFileStoreWrite) table.store().newWrite("ss");
+ BaseAppendFileStoreWrite write = (BaseAppendFileStoreWrite) table.store().newWrite("ss");
StreamTableCommit commit = table.newStreamWriteBuilder().newCommit();
write.withExecutionMode(false);
@@ -217,7 +218,7 @@ public class AppendOnlyFileStoreWriteTest {
public void testScanFilterWithAllNullPartitionWrite() throws Exception {
FileStoreTable table = createFileStoreTable();
- AppendOnlyFileStoreWrite write = (AppendOnlyFileStoreWrite) table.store().newWrite("ss");
+ BaseAppendFileStoreWrite write = (BaseAppendFileStoreWrite) table.store().newWrite("ss");
StreamTableCommit commit = table.newStreamWriteBuilder().newCommit();
write.withExecutionMode(false);
@@ -244,7 +245,7 @@ public class AppendOnlyFileStoreWriteTest {
public void testScanFilterWithNoneNullPartitionWrite() throws Exception {
FileStoreTable table = createFileStoreTable();
- AppendOnlyFileStoreWrite write = (AppendOnlyFileStoreWrite) table.store().newWrite("ss");
+ BaseAppendFileStoreWrite write = (BaseAppendFileStoreWrite) table.store().newWrite("ss");
StreamTableCommit commit = table.newStreamWriteBuilder().newCommit();
write.withExecutionMode(false);
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/sink/CompactionTaskSerializerTest.java b/paimon-core/src/test/java/org/apache/paimon/table/sink/AppendCompactTaskSerializerTest.java
similarity index 70%
rename from paimon-core/src/test/java/org/apache/paimon/table/sink/CompactionTaskSerializerTest.java
rename to paimon-core/src/test/java/org/apache/paimon/table/sink/AppendCompactTaskSerializerTest.java
index 53a2d82bcd..33b4b1b0c6 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/sink/CompactionTaskSerializerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/sink/AppendCompactTaskSerializerTest.java
@@ -18,8 +18,8 @@
package org.apache.paimon.table.sink;
-import org.apache.paimon.append.MultiTableUnawareAppendCompactionTask;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
+import org.apache.paimon.append.MultiTableAppendCompactTask;
import org.apache.paimon.catalog.Identifier;
import org.junit.jupiter.api.Test;
@@ -30,32 +30,31 @@ import static org.apache.paimon.manifest.ManifestCommittableSerializerTest.rando
import static org.apache.paimon.mergetree.compact.MergeTreeCompactManagerTest.row;
import static org.assertj.core.api.Assertions.assertThat;
-/** Tests for {@link CompactionTaskSerializer} and {@link MultiTableCompactionTaskSerializer}. */
-public class CompactionTaskSerializerTest {
+/** Tests for {@link AppendCompactTaskSerializer} and {@link MultiTableCompactionTaskSerializer}. */
+public class AppendCompactTaskSerializerTest {
@Test
public void testCompactionTaskSerializer() throws IOException {
- CompactionTaskSerializer serializer = new CompactionTaskSerializer();
- UnawareAppendCompactionTask task =
- new UnawareAppendCompactionTask(row(0), randomNewFilesIncrement().newFiles());
+ AppendCompactTaskSerializer serializer = new AppendCompactTaskSerializer();
+ AppendCompactTask task =
+ new AppendCompactTask(row(0), randomNewFilesIncrement().newFiles());
byte[] bytes = serializer.serialize(task);
- UnawareAppendCompactionTask task1 = serializer.deserialize(serializer.getVersion(), bytes);
+ AppendCompactTask task1 = serializer.deserialize(serializer.getVersion(), bytes);
assertThat(task).isEqualTo(task1);
}
@Test
public void testMultiTableCompactionTaskSerializer() throws IOException {
MultiTableCompactionTaskSerializer serializer = new MultiTableCompactionTaskSerializer();
- MultiTableUnawareAppendCompactionTask task =
- new MultiTableUnawareAppendCompactionTask(
+ MultiTableAppendCompactTask task =
+ new MultiTableAppendCompactTask(
row(0),
randomNewFilesIncrement().newFiles(),
Identifier.create("db", "table"));
byte[] bytes = serializer.serialize(task);
- MultiTableUnawareAppendCompactionTask task1 =
- serializer.deserialize(serializer.getVersion(), bytes);
+ MultiTableAppendCompactTask task1 = serializer.deserialize(serializer.getVersion(), bytes);
assertThat(task).isEqualTo(task1);
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/sink/UnawareBucketRowKeyExtractorTest.java b/paimon-core/src/test/java/org/apache/paimon/table/sink/AppendTableRowKeyExtractorTest.java
similarity index 89%
rename from paimon-core/src/test/java/org/apache/paimon/table/sink/UnawareBucketRowKeyExtractorTest.java
rename to paimon-core/src/test/java/org/apache/paimon/table/sink/AppendTableRowKeyExtractorTest.java
index 4fd402b807..4653546cdc 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/sink/UnawareBucketRowKeyExtractorTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/sink/AppendTableRowKeyExtractorTest.java
@@ -36,8 +36,8 @@ import java.util.Map;
import static org.apache.paimon.CoreOptions.BUCKET;
import static org.assertj.core.api.Assertions.assertThat;
-/** Test for {@link UnawareBucketRowKeyExtractor}. */
-public class UnawareBucketRowKeyExtractorTest {
+/** Test for {@link AppendTableRowKeyExtractor}. */
+public class AppendTableRowKeyExtractorTest {
@Test
public void testBucket() {
@@ -45,12 +45,12 @@ public class UnawareBucketRowKeyExtractorTest {
assertThat(bucket(extractor("a"), row)).isEqualTo(0);
}
- private int bucket(UnawareBucketRowKeyExtractor extractor, InternalRow row) {
+ private int bucket(AppendTableRowKeyExtractor extractor, InternalRow row) {
extractor.setRecord(row);
return extractor.bucket();
}
- private UnawareBucketRowKeyExtractor extractor(String partK) {
+ private AppendTableRowKeyExtractor extractor(String partK) {
RowType rowType =
new RowType(
Arrays.asList(
@@ -71,6 +71,6 @@ public class UnawareBucketRowKeyExtractorTest {
Collections.emptyList(),
options,
"");
- return new UnawareBucketRowKeyExtractor(schema);
+ return new AppendTableRowKeyExtractor(schema);
}
}
diff --git a/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/UnawareBucketAppendOnlyTableITCase.java b/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/AppendTableITCase.java
similarity index 95%
rename from paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/UnawareBucketAppendOnlyTableITCase.java
rename to paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/AppendTableITCase.java
index 49698343e2..f3b6dd051c 100644
--- a/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/UnawareBucketAppendOnlyTableITCase.java
+++ b/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/AppendTableITCase.java
@@ -28,8 +28,8 @@ import java.util.List;
import static org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL;
-/** Test case for append-only managed unaware-bucket table. */
-public class UnawareBucketAppendOnlyTableITCase extends CatalogITCaseBase {
+/** Test case for append table. */
+public class AppendTableITCase extends CatalogITCaseBase {
@Override
protected List<String> ddl() {
diff --git a/paimon-flink/paimon-flink-1.19/src/test/java/org/apache/paimon/flink/UnawareBucketAppendOnlyTableITCase.java b/paimon-flink/paimon-flink-1.19/src/test/java/org/apache/paimon/flink/AppendTableITCase.java
similarity index 95%
rename from paimon-flink/paimon-flink-1.19/src/test/java/org/apache/paimon/flink/UnawareBucketAppendOnlyTableITCase.java
rename to paimon-flink/paimon-flink-1.19/src/test/java/org/apache/paimon/flink/AppendTableITCase.java
index 49698343e2..f3b6dd051c 100644
--- a/paimon-flink/paimon-flink-1.19/src/test/java/org/apache/paimon/flink/UnawareBucketAppendOnlyTableITCase.java
+++ b/paimon-flink/paimon-flink-1.19/src/test/java/org/apache/paimon/flink/AppendTableITCase.java
@@ -28,8 +28,8 @@ import java.util.List;
import static org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL;
-/** Test case for append-only managed unaware-bucket table. */
-public class UnawareBucketAppendOnlyTableITCase extends CatalogITCaseBase {
+/** Test case for append table. */
+public class AppendTableITCase extends CatalogITCaseBase {
@Override
protected List<String> ddl() {
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketSink.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableSink.java
similarity index 88%
rename from paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketSink.java
rename to paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableSink.java
index 820ef7728f..d43690e5b2 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketSink.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableSink.java
@@ -32,11 +32,11 @@ import javax.annotation.Nullable;
* CDC Sink for unaware bucket table. It should not add compaction node, because the compaction may
* have old schema.
*/
-public class CdcUnawareBucketSink extends FlinkWriteSink<CdcRecord> {
+public class CdcAppendTableSink extends FlinkWriteSink<CdcRecord> {
private final Integer parallelism;
- public CdcUnawareBucketSink(FileStoreTable table, Integer parallelism) {
+ public CdcAppendTableSink(FileStoreTable table, Integer parallelism) {
super(table, null);
this.parallelism = parallelism;
}
@@ -44,7 +44,7 @@ public class CdcUnawareBucketSink extends FlinkWriteSink<CdcRecord> {
@Override
protected OneInputStreamOperatorFactory<CdcRecord, Committable> createWriteOperatorFactory(
StoreSinkWrite.Provider writeProvider, String commitUser) {
- return new CdcUnawareBucketWriteOperator.Factory(table, writeProvider, commitUser);
+ return new CdcAppendTableWriteOperator.Factory(table, writeProvider, commitUser);
}
@Override
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketWriteOperator.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableWriteOperator.java
similarity index 89%
rename from paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketWriteOperator.java
rename to paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableWriteOperator.java
index 26f65fdd09..96b9fefcdf 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketWriteOperator.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableWriteOperator.java
@@ -30,9 +30,9 @@ import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
/** A {@link PrepareCommitOperator} to write {@link CdcRecord} to unaware-bucket mode table. */
-public class CdcUnawareBucketWriteOperator extends CdcRecordStoreWriteOperator {
+public class CdcAppendTableWriteOperator extends CdcRecordStoreWriteOperator {
- private CdcUnawareBucketWriteOperator(
+ private CdcAppendTableWriteOperator(
StreamOperatorParameters<Committable> parameters,
FileStoreTable table,
StoreSinkWrite.Provider storeSinkWriteProvider,
@@ -48,7 +48,7 @@ public class CdcUnawareBucketWriteOperator extends CdcRecordStoreWriteOperator {
}
}
- /** {@link StreamOperatorFactory} of {@link CdcUnawareBucketWriteOperator}. */
+ /** {@link StreamOperatorFactory} of {@link CdcAppendTableWriteOperator}. */
public static class Factory extends CdcRecordStoreWriteOperator.Factory {
public Factory(
@@ -63,14 +63,14 @@ public class CdcUnawareBucketWriteOperator extends CdcRecordStoreWriteOperator {
public <T extends StreamOperator<Committable>> T createStreamOperator(
StreamOperatorParameters<Committable> parameters) {
return (T)
- new CdcUnawareBucketWriteOperator(
+ new CdcAppendTableWriteOperator(
parameters, table, storeSinkWriteProvider, initialCommitUser);
}
@Override
@SuppressWarnings("rawtypes")
public Class<? extends StreamOperator> getStreamOperatorClass(ClassLoader classLoader) {
- return CdcUnawareBucketWriteOperator.class;
+ return CdcAppendTableWriteOperator.class;
}
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
index 7b24ff8138..9297670d3a 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
@@ -149,6 +149,6 @@ public class CdcSinkBuilder<T> {
private DataStreamSink<?> buildForUnawareBucket(DataStream<CdcRecord> parsed) {
FileStoreTable dataTable = (FileStoreTable) table;
// rebalance it to make sure schema change work to avoid infinite loop
- return new CdcUnawareBucketSink(dataTable, parallelism).sinkFrom(parsed.rebalance());
+ return new CdcAppendTableSink(dataTable, parallelism).sinkFrom(parsed.rebalance());
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
index 196b9e4350..a6b272eaf8 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
@@ -206,7 +206,7 @@ public class FlinkCdcSyncDatabaseSinkBuilder<T> {
private void buildForUnawareBucket(FileStoreTable table, DataStream<CdcRecord> parsed) {
// rebalance it to make sure schema change work to avoid infinite loop
- new CdcUnawareBucketSink(table, parallelism).sinkFrom(parsed.rebalance());
+ new CdcAppendTableSink(table, parallelism).sinkFrom(parsed.rebalance());
}
private void buildDividedCdcSink() {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
index 316655613b..306216a57d 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
@@ -22,7 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.flink.FlinkConnectorOptions;
-import org.apache.paimon.flink.compact.UnawareBucketCompactionTopoBuilder;
+import org.apache.paimon.flink.compact.AppendTableCompactBuilder;
import org.apache.paimon.flink.postpone.PostponeBucketCompactSplitSource;
import org.apache.paimon.flink.postpone.RewritePostponeBucketCommittableOperator;
import org.apache.paimon.flink.predicate.SimpleSqlPredicateConvertor;
@@ -136,15 +136,15 @@ public class CompactAction extends TableActionBase {
if (fileStoreTable.coreOptions().bucket() == BucketMode.POSTPONE_BUCKET) {
return buildForPostponeBucketCompaction(env, fileStoreTable, isStreaming);
} else if (fileStoreTable.bucketMode() == BucketMode.BUCKET_UNAWARE) {
- buildForUnawareBucketCompaction(env, fileStoreTable, isStreaming);
+ buildForAppendTableCompact(env, fileStoreTable, isStreaming);
return true;
} else {
- buildForTraditionalCompaction(env, fileStoreTable, isStreaming);
+ buildForBucketedTableCompact(env, fileStoreTable, isStreaming);
return true;
}
}
- private void buildForTraditionalCompaction(
+ private void buildForBucketedTableCompact(
StreamExecutionEnvironment env, FileStoreTable table, boolean isStreaming)
throws Exception {
if (fullCompaction == null) {
@@ -180,16 +180,15 @@ public class CompactAction extends TableActionBase {
sinkBuilder.withInput(source).build();
}
- private void buildForUnawareBucketCompaction(
+ private void buildForAppendTableCompact(
StreamExecutionEnvironment env, FileStoreTable table, boolean isStreaming)
throws Exception {
- UnawareBucketCompactionTopoBuilder unawareBucketCompactionTopoBuilder =
- new UnawareBucketCompactionTopoBuilder(env, identifier.getFullName(), table);
-
- unawareBucketCompactionTopoBuilder.withPartitionPredicate(getPredicate());
- unawareBucketCompactionTopoBuilder.withContinuousMode(isStreaming);
- unawareBucketCompactionTopoBuilder.withPartitionIdleTime(partitionIdleTime);
- unawareBucketCompactionTopoBuilder.build();
+ AppendTableCompactBuilder builder =
+ new AppendTableCompactBuilder(env, identifier.getFullName(), table);
+ builder.withPartitionPredicate(getPredicate());
+ builder.withContinuousMode(isStreaming);
+ builder.withPartitionIdleTime(partitionIdleTime);
+ builder.build();
}
protected Predicate getPredicate() throws Exception {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
index c6ec8cfb90..a438a5f895 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
@@ -19,11 +19,11 @@
package org.apache.paimon.flink.action;
import org.apache.paimon.CoreOptions;
-import org.apache.paimon.append.MultiTableUnawareAppendCompactionTask;
+import org.apache.paimon.append.MultiTableAppendCompactTask;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.FlinkConnectorOptions;
-import org.apache.paimon.flink.compact.UnawareBucketCompactionTopoBuilder;
+import org.apache.paimon.flink.compact.AppendTableCompactBuilder;
import org.apache.paimon.flink.sink.BucketsRowChannelComputer;
import org.apache.paimon.flink.sink.CombinedTableCompactorSink;
import org.apache.paimon.flink.sink.CompactorSinkBuilder;
@@ -234,7 +234,7 @@ public class CompactDatabaseAction extends ActionBase {
parallelism);
// unaware bucket table
- DataStream<MultiTableUnawareAppendCompactionTask> unawareBucketTableSource =
+ DataStream<MultiTableAppendCompactTask> unawareBucketTableSource =
rebalance(
sourceBuilder
.withEnv(env)
@@ -278,8 +278,8 @@ public class CompactDatabaseAction extends ActionBase {
private void buildForUnawareBucketCompaction(
StreamExecutionEnvironment env, String fullName, FileStoreTable table) {
- UnawareBucketCompactionTopoBuilder unawareBucketCompactionTopoBuilder =
- new UnawareBucketCompactionTopoBuilder(env, fullName, table);
+ AppendTableCompactBuilder unawareBucketCompactionTopoBuilder =
+ new AppendTableCompactBuilder(env, fullName, table);
unawareBucketCompactionTopoBuilder.withContinuousMode(isStreaming);
unawareBucketCompactionTopoBuilder.withPartitionIdleTime(partitionIdleTime);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketNewFilesCompactionCoordinatorOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendPreCommitCompactCoordinatorOperator.java
similarity index 83%
rename from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketNewFilesCompactionCoordinatorOperator.java
rename to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendPreCommitCompactCoordinatorOperator.java
index a906f9b414..674dd59e7e 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketNewFilesCompactionCoordinatorOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendPreCommitCompactCoordinatorOperator.java
@@ -19,8 +19,8 @@
package org.apache.paimon.flink.compact;
import org.apache.paimon.CoreOptions;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
-import org.apache.paimon.append.UnawareBucketNewFilesCompactionCoordinator;
+import org.apache.paimon.append.AppendCompactTask;
+import org.apache.paimon.append.AppendPreCommitCompactCoordinator;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.flink.sink.Committable;
import org.apache.paimon.io.CompactIncrement;
@@ -46,27 +46,25 @@ import java.util.Optional;
/**
* Coordinator operator for compacting newly created files for unaware bucket tables.
*
- * <p>{@link UnawareBucketNewFilesCompactionCoordinatorOperator} calculates the file size of newly
- * created files contained in all buckets within each partition from {@link Committable} message
- * emitted from writer operator. And emit {@link UnawareAppendCompactionTask} to {@link
- * UnawareBucketNewFilesCompactionWorkerOperator}.
+ * <p>{@link AppendPreCommitCompactCoordinatorOperator} calculates the file size of newly created
+ * files contained in all buckets within each partition from {@link Committable} message emitted
+ * from writer operator. And emit {@link AppendCompactTask} to {@link
+ * AppendPreCommitCompactWorkerOperator}.
*/
-public class UnawareBucketNewFilesCompactionCoordinatorOperator
- extends AbstractStreamOperator<
- Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>>
+public class AppendPreCommitCompactCoordinatorOperator
+ extends AbstractStreamOperator<Either<Committable, Tuple2<Long, AppendCompactTask>>>
implements OneInputStreamOperator<
- Committable,
- Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>>,
+ Committable, Either<Committable, Tuple2<Long, AppendCompactTask>>>,
BoundedOneInput {
private final long targetFileSize;
private final long compactionFileSize;
private final int totalBuckets;
- private transient UnawareBucketNewFilesCompactionCoordinator coordinator;
+ private transient AppendPreCommitCompactCoordinator coordinator;
private transient long checkpointId;
- public UnawareBucketNewFilesCompactionCoordinatorOperator(CoreOptions options) {
+ public AppendPreCommitCompactCoordinatorOperator(CoreOptions options) {
this.targetFileSize = options.targetFileSize(false);
this.compactionFileSize = options.compactionFileSize(false);
this.totalBuckets = options.bucket();
@@ -76,7 +74,7 @@ public class UnawareBucketNewFilesCompactionCoordinatorOperator
public void open() throws Exception {
super.open();
- coordinator = new UnawareBucketNewFilesCompactionCoordinator(targetFileSize);
+ coordinator = new AppendPreCommitCompactCoordinator(targetFileSize);
checkpointId = Long.MIN_VALUE;
}
@@ -114,7 +112,7 @@ public class UnawareBucketNewFilesCompactionCoordinatorOperator
Either.Right(
Tuple2.of(
checkpointId,
- new UnawareAppendCompactionTask(
+ new AppendCompactTask(
p.getKey(), p.getValue())))));
} else {
skippedFiles.add(p.getValue().get(0));
@@ -159,8 +157,7 @@ public class UnawareBucketNewFilesCompactionCoordinatorOperator
Either.Right(
Tuple2.of(
checkpointId,
- new UnawareAppendCompactionTask(
- p.getKey(), p.getValue())))));
+ new AppendCompactTask(p.getKey(), p.getValue())))));
} else {
CommitMessageImpl message =
new CommitMessageImpl(
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketNewFilesCompactionWorkerOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendPreCommitCompactWorkerOperator.java
similarity index 81%
rename from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketNewFilesCompactionWorkerOperator.java
rename to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendPreCommitCompactWorkerOperator.java
index 2ada98f6b9..0a1a04f868 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketNewFilesCompactionWorkerOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendPreCommitCompactWorkerOperator.java
@@ -18,14 +18,14 @@
package org.apache.paimon.flink.compact;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.flink.sink.Committable;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
import org.apache.paimon.io.DataIncrement;
-import org.apache.paimon.operation.AppendOnlyUnawareBucketFileStoreWrite;
+import org.apache.paimon.operation.AppendFileStoreWrite;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
@@ -42,35 +42,34 @@ import java.util.HashMap;
import java.util.Map;
/**
- * Receive and process the {@link UnawareAppendCompactionTask}s emitted by {@link
- * UnawareBucketNewFilesCompactionCoordinatorOperator}.
+ * Receive and process the {@link AppendCompactTask}s emitted by {@link
+ * AppendPreCommitCompactCoordinatorOperator}.
*/
-public class UnawareBucketNewFilesCompactionWorkerOperator
- extends AbstractStreamOperator<Committable>
+public class AppendPreCommitCompactWorkerOperator extends AbstractStreamOperator<Committable>
implements OneInputStreamOperator<
- Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>, Committable> {
+ Either<Committable, Tuple2<Long, AppendCompactTask>>, Committable> {
private final FileStoreTable table;
- private transient AppendOnlyUnawareBucketFileStoreWrite write;
+ private transient AppendFileStoreWrite write;
private transient FileStorePathFactory pathFactory;
private transient FileIO fileIO;
- public UnawareBucketNewFilesCompactionWorkerOperator(FileStoreTable table) {
+ public AppendPreCommitCompactWorkerOperator(FileStoreTable table) {
this.table = table;
}
@Override
public void open() throws Exception {
super.open();
- this.write = (AppendOnlyUnawareBucketFileStoreWrite) table.store().newWrite(null);
+ this.write = (AppendFileStoreWrite) table.store().newWrite(null);
this.pathFactory = table.store().pathFactory();
this.fileIO = table.fileIO();
}
@Override
public void processElement(
- StreamRecord<Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>> record)
+ StreamRecord<Either<Committable, Tuple2<Long, AppendCompactTask>>> record)
throws Exception {
if (record.getValue().isLeft()) {
output.collect(new StreamRecord<>(record.getValue().left()));
@@ -83,7 +82,7 @@ public class UnawareBucketNewFilesCompactionWorkerOperator
}
}
- private CommitMessage doCompact(UnawareAppendCompactionTask task) throws Exception {
+ private CommitMessage doCompact(AppendCompactTask task) throws Exception {
CommitMessageImpl message = (CommitMessageImpl) task.doCompact(table, write);
Map<String, DataFileMeta> toDelete = new HashMap<>();
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactionTopoBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendTableCompactBuilder.java
similarity index 81%
rename from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactionTopoBuilder.java
rename to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendTableCompactBuilder.java
index a572354e89..007e625afd 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactionTopoBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendTableCompactBuilder.java
@@ -18,11 +18,11 @@
package org.apache.paimon.flink.compact;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.flink.FlinkConnectorOptions;
-import org.apache.paimon.flink.sink.UnawareBucketCompactionSink;
-import org.apache.paimon.flink.source.BucketUnawareCompactSource;
+import org.apache.paimon.flink.sink.AppendTableCompactSink;
+import org.apache.paimon.flink.source.AppendTableCompactSource;
import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
@@ -54,7 +54,7 @@ import java.util.stream.Collectors;
* snapshot. We need to enable checkpoint for this compaction job, checkpoint will trigger committer
* stage to commit all the compacted files.
*/
-public class UnawareBucketCompactionTopoBuilder {
+public class AppendTableCompactBuilder {
private final transient StreamExecutionEnvironment env;
private final String tableIdentifier;
@@ -65,7 +65,7 @@ public class UnawareBucketCompactionTopoBuilder {
@Nullable private Predicate partitionPredicate;
@Nullable private Duration partitionIdleTime = null;
- public UnawareBucketCompactionTopoBuilder(
+ public AppendTableCompactBuilder(
StreamExecutionEnvironment env, String tableIdentifier, FileStoreTable table) {
this.env = env;
this.tableIdentifier = tableIdentifier;
@@ -86,7 +86,7 @@ public class UnawareBucketCompactionTopoBuilder {
public void build() {
// build source from UnawareSourceFunction
- DataStreamSource<UnawareAppendCompactionTask> source = buildSource();
+ DataStreamSource<AppendCompactTask> source = buildSource();
if (isContinuous) {
Preconditions.checkArgument(
partitionIdleTime == null, "Streaming mode does not support partitionIdleTime");
@@ -98,7 +98,7 @@ public class UnawareBucketCompactionTopoBuilder {
.atZone(ZoneId.systemDefault())
.toInstant()
.toEpochMilli();
- SingleOutputStreamOperator<UnawareAppendCompactionTask> filterStream =
+ SingleOutputStreamOperator<AppendCompactTask> filterStream =
source.filter(
task -> {
BinaryRow partition = task.partition();
@@ -119,28 +119,27 @@ public class UnawareBucketCompactionTopoBuilder {
PartitionEntry::partition, PartitionEntry::lastFileCreationTime));
}
- private DataStreamSource<UnawareAppendCompactionTask> buildSource() {
+ private DataStreamSource<AppendCompactTask> buildSource() {
long scanInterval = table.coreOptions().continuousDiscoveryInterval().toMillis();
- BucketUnawareCompactSource source =
- new BucketUnawareCompactSource(
- table, isContinuous, scanInterval, partitionPredicate);
+ AppendTableCompactSource source =
+ new AppendTableCompactSource(table, isContinuous, scanInterval, partitionPredicate);
- return BucketUnawareCompactSource.buildSource(env, source, tableIdentifier);
+ return AppendTableCompactSource.buildSource(env, source, tableIdentifier);
}
- private void sinkFromSource(DataStreamSource<UnawareAppendCompactionTask> input) {
- DataStream<UnawareAppendCompactionTask> rebalanced = rebalanceInput(input);
+ private void sinkFromSource(DataStreamSource<AppendCompactTask> input) {
+ DataStream<AppendCompactTask> rebalanced = rebalanceInput(input);
- UnawareBucketCompactionSink.sink(table, rebalanced);
+ AppendTableCompactSink.sink(table, rebalanced);
}
- private DataStream<UnawareAppendCompactionTask> rebalanceInput(
- DataStreamSource<UnawareAppendCompactionTask> input) {
+ private DataStream<AppendCompactTask> rebalanceInput(
+ DataStreamSource<AppendCompactTask> input) {
Options conf = Options.fromMap(table.options());
Integer compactionWorkerParallelism =
conf.get(FlinkConnectorOptions.UNAWARE_BUCKET_COMPACTION_PARALLELISM);
- PartitionTransformation<UnawareAppendCompactionTask> transformation =
+ PartitionTransformation<AppendCompactTask> transformation =
new PartitionTransformation<>(
input.getTransformation(), new RebalancePartitioner<>());
if (compactionWorkerParallelism != null) {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactor.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendTableCompactor.java
similarity index 92%
rename from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactor.java
rename to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendTableCompactor.java
index e8021329c9..0f13c164f2 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactor.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendTableCompactor.java
@@ -19,11 +19,11 @@
package org.apache.paimon.flink.compact;
import org.apache.paimon.annotation.VisibleForTesting;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.flink.metrics.FlinkMetricRegistry;
import org.apache.paimon.flink.sink.Committable;
-import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
+import org.apache.paimon.operation.BaseAppendFileStoreWrite;
import org.apache.paimon.operation.metrics.CompactionMetrics;
import org.apache.paimon.operation.metrics.MetricUtils;
import org.apache.paimon.table.FileStoreTable;
@@ -46,15 +46,15 @@ import java.util.concurrent.Future;
import java.util.function.Supplier;
import java.util.stream.Collectors;
-/** The Compactor of unaware bucket table to execute {@link UnawareAppendCompactionTask}. */
-public class UnawareBucketCompactor {
+/** The Compactor of unaware bucket table to execute {@link AppendCompactTask}. */
+public class AppendTableCompactor {
- private static final Logger LOG = LoggerFactory.getLogger(UnawareBucketCompactor.class);
+ private static final Logger LOG = LoggerFactory.getLogger(AppendTableCompactor.class);
private final FileStoreTable table;
private final String commitUser;
- private final transient AppendOnlyFileStoreWrite write;
+ private final transient BaseAppendFileStoreWrite write;
protected final transient Queue<Future<CommitMessage>> result;
@@ -62,14 +62,14 @@ public class UnawareBucketCompactor {
@Nullable private final transient CompactionMetrics compactionMetrics;
@Nullable private final transient CompactionMetrics.Reporter metricsReporter;
- public UnawareBucketCompactor(
+ public AppendTableCompactor(
FileStoreTable table,
String commitUser,
Supplier<ExecutorService> lazyCompactExecutor,
@Nullable MetricGroup metricGroup) {
this.table = table;
this.commitUser = commitUser;
- this.write = (AppendOnlyFileStoreWrite) table.store().newWrite(commitUser);
+ this.write = (BaseAppendFileStoreWrite) table.store().newWrite(commitUser);
this.result = new LinkedList<>();
this.compactExecutorsupplier = lazyCompactExecutor;
this.compactionMetrics =
@@ -83,7 +83,7 @@ public class UnawareBucketCompactor {
: this.compactionMetrics.createReporter(BinaryRow.EMPTY_ROW, 0);
}
- public void processElement(UnawareAppendCompactionTask task) throws Exception {
+ public void processElement(AppendCompactTask task) throws Exception {
result.add(
compactExecutorsupplier
.get()
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiUnawareBucketTableScan.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiAppendCompactTableScan.java
similarity index 72%
rename from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiUnawareBucketTableScan.java
rename to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiAppendCompactTableScan.java
index 7ab893e24b..b9eeb45f3c 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiUnawareBucketTableScan.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiAppendCompactTableScan.java
@@ -18,8 +18,8 @@
package org.apache.paimon.flink.compact;
-import org.apache.paimon.append.MultiTableUnawareAppendCompactionTask;
-import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.append.AppendCompactCoordinator;
+import org.apache.paimon.append.MultiTableAppendCompactTask;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.table.BucketMode;
@@ -35,12 +35,11 @@ import java.util.regex.Pattern;
* This class is responsible for implementing the scanning logic {@link MultiTableScanBase} for the
* table with fix single bucket such as unaware bucket table.
*/
-public class MultiUnawareBucketTableScan
- extends MultiTableScanBase<MultiTableUnawareAppendCompactionTask> {
+public class MultiAppendCompactTableScan extends MultiTableScanBase<MultiTableAppendCompactTask> {
- protected transient Map<Identifier, UnawareAppendTableCompactionCoordinator> tablesMap;
+ protected transient Map<Identifier, AppendCompactCoordinator> tablesMap;
- public MultiUnawareBucketTableScan(
+ public MultiAppendCompactTableScan(
CatalogLoader catalogLoader,
Pattern includingPattern,
Pattern excludingPattern,
@@ -58,18 +57,17 @@ public class MultiUnawareBucketTableScan
}
@Override
- List<MultiTableUnawareAppendCompactionTask> doScan() {
+ List<MultiTableAppendCompactTask> doScan() {
// do scan and plan action, emit append-only compaction tasks.
- List<MultiTableUnawareAppendCompactionTask> tasks = new ArrayList<>();
- for (Map.Entry<Identifier, UnawareAppendTableCompactionCoordinator> tableIdAndCoordinator :
+ List<MultiTableAppendCompactTask> tasks = new ArrayList<>();
+ for (Map.Entry<Identifier, AppendCompactCoordinator> tableIdAndCoordinator :
tablesMap.entrySet()) {
Identifier tableId = tableIdAndCoordinator.getKey();
- UnawareAppendTableCompactionCoordinator compactionCoordinator =
- tableIdAndCoordinator.getValue();
+ AppendCompactCoordinator compactionCoordinator = tableIdAndCoordinator.getValue();
compactionCoordinator.run().stream()
.map(
task ->
- new MultiTableUnawareAppendCompactionTask(
+ new MultiTableAppendCompactTask(
task.partition(), task.compactBefore(), tableId))
.forEach(tasks::add);
}
@@ -84,9 +82,7 @@ public class MultiUnawareBucketTableScan
@Override
public void addScanTable(FileStoreTable fileStoreTable, Identifier identifier) {
if (fileStoreTable.bucketMode() == BucketMode.BUCKET_UNAWARE) {
- tablesMap.put(
- identifier,
- new UnawareAppendTableCompactionCoordinator(fileStoreTable, isStreaming));
+ tablesMap.put(identifier, new AppendCompactCoordinator(fileStoreTable, isStreaming));
}
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiTableScanBase.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiTableScanBase.java
index 6360172a61..67b0925a4f 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiTableScanBase.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiTableScanBase.java
@@ -18,7 +18,7 @@
package org.apache.paimon.flink.compact;
-import org.apache.paimon.append.MultiTableUnawareAppendCompactionTask;
+import org.apache.paimon.append.MultiTableAppendCompactTask;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.catalog.Identifier;
@@ -45,8 +45,8 @@ import static org.apache.paimon.flink.utils.MultiTablesCompactorUtil.shouldCompa
* <ol>
* <li>Tuple2<{@link Split},String> for the table with multi buckets, such as dynamic or fixed
* bucket table.
- * <li>{@link MultiTableUnawareAppendCompactionTask} for the table witch fixed single bucket
- * ,such as unaware bucket table.
+ * <li>{@link MultiTableAppendCompactTask} for the table witch fixed single bucket ,such as
+ * unaware bucket table.
* </ol>
*/
public abstract class MultiTableScanBase<T> implements AutoCloseable {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendBypassCompactWorkerOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendBypassCompactWorkerOperator.java
index 977511920a..a9ab225c92 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendBypassCompactWorkerOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendBypassCompactWorkerOperator.java
@@ -18,7 +18,7 @@
package org.apache.paimon.flink.sink;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.streaming.api.operators.StreamOperator;
@@ -29,7 +29,7 @@ import org.apache.flink.types.Either;
/** A {@link AppendCompactWorkerOperator} to bypass Committable inputs. */
public class AppendBypassCompactWorkerOperator
- extends AppendCompactWorkerOperator<Either<Committable, UnawareAppendCompactionTask>> {
+ extends AppendCompactWorkerOperator<Either<Committable, AppendCompactTask>> {
private AppendBypassCompactWorkerOperator(
StreamOperatorParameters<Committable> parameters,
@@ -44,8 +44,7 @@ public class AppendBypassCompactWorkerOperator
}
@Override
- public void processElement(
- StreamRecord<Either<Committable, UnawareAppendCompactionTask>> element)
+ public void processElement(StreamRecord<Either<Committable, AppendCompactTask>> element)
throws Exception {
if (element.getValue().isLeft()) {
output.collect(new StreamRecord<>(element.getValue().left()));
@@ -56,8 +55,7 @@ public class AppendBypassCompactWorkerOperator
/** {@link StreamOperatorFactory} of {@link AppendBypassCompactWorkerOperator}. */
public static class Factory
- extends AppendCompactWorkerOperator.Factory<
- Either<Committable, UnawareAppendCompactionTask>> {
+ extends AppendCompactWorkerOperator.Factory<Either<Committable, AppendCompactTask>> {
public Factory(FileStoreTable table, String initialCommitUser) {
super(table, initialCommitUser);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendCompactWorkerOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendCompactWorkerOperator.java
index 7a3c0231eb..4adb2b9dc8 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendCompactWorkerOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendCompactWorkerOperator.java
@@ -19,9 +19,9 @@
package org.apache.paimon.flink.sink;
import org.apache.paimon.annotation.VisibleForTesting;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
-import org.apache.paimon.flink.compact.UnawareBucketCompactor;
-import org.apache.paimon.flink.source.BucketUnawareCompactSource;
+import org.apache.paimon.append.AppendCompactTask;
+import org.apache.paimon.flink.compact.AppendTableCompactor;
+import org.apache.paimon.flink.source.AppendTableCompactSource;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
@@ -40,8 +40,8 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
/**
- * An abstract Operator to execute {@link UnawareAppendCompactionTask} passed from {@link
- * BucketUnawareCompactSource} for compacting table. This operator is always in async mode.
+ * An abstract Operator to execute {@link AppendCompactTask} passed from {@link
+ * AppendTableCompactSource} for compacting table. This operator is always in async mode.
*/
public abstract class AppendCompactWorkerOperator<IN>
extends PrepareCommitOperator<IN, Committable> {
@@ -51,7 +51,7 @@ public abstract class AppendCompactWorkerOperator<IN>
private final FileStoreTable table;
private final String commitUser;
- protected transient UnawareBucketCompactor unawareBucketCompactor;
+ protected transient AppendTableCompactor unawareBucketCompactor;
private transient ExecutorService lazyCompactExecutor;
@@ -73,8 +73,7 @@ public abstract class AppendCompactWorkerOperator<IN>
public void open() throws Exception {
LOG.debug("Opened a append-only table compaction worker.");
this.unawareBucketCompactor =
- new UnawareBucketCompactor(
- table, commitUser, this::workerExecutor, getMetricGroup());
+ new AppendTableCompactor(table, commitUser, this::workerExecutor, getMetricGroup());
}
@Override
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyMultiTableCompactionWorkerOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyMultiTableCompactionWorkerOperator.java
index 6d71e44a82..98787caba3 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyMultiTableCompactionWorkerOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyMultiTableCompactionWorkerOperator.java
@@ -18,12 +18,12 @@
package org.apache.paimon.flink.sink;
-import org.apache.paimon.append.MultiTableUnawareAppendCompactionTask;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
+import org.apache.paimon.append.MultiTableAppendCompactTask;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.flink.compact.UnawareBucketCompactor;
+import org.apache.paimon.flink.compact.AppendTableCompactor;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.ExceptionUtils;
@@ -46,12 +46,11 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
- * Operator to execute {@link UnawareAppendCompactionTask} passed for support compacting multi
- * unaware bucket tables in combined mode.
+ * Operator to execute {@link AppendCompactTask} passed for support compacting multi unaware bucket
+ * tables in combined mode.
*/
public class AppendOnlyMultiTableCompactionWorkerOperator
- extends PrepareCommitOperator<
- MultiTableUnawareAppendCompactionTask, MultiTableCommittable> {
+ extends PrepareCommitOperator<MultiTableAppendCompactTask, MultiTableCommittable> {
private static final Logger LOG =
LoggerFactory.getLogger(AppendOnlyMultiTableCompactionWorkerOperator.class);
@@ -60,7 +59,7 @@ public class AppendOnlyMultiTableCompactionWorkerOperator
private final CatalogLoader catalogLoader;
// support multi table compaction
- private transient Map<Identifier, UnawareBucketCompactor> compactorContainer;
+ private transient Map<Identifier, AppendTableCompactor> compactorContainer;
private transient ExecutorService lazyCompactExecutor;
@@ -87,10 +86,10 @@ public class AppendOnlyMultiTableCompactionWorkerOperator
protected List<MultiTableCommittable> prepareCommit(boolean waitCompaction, long checkpointId)
throws IOException {
List<MultiTableCommittable> result = new ArrayList<>();
- for (Map.Entry<Identifier, UnawareBucketCompactor> compactorWithTable :
+ for (Map.Entry<Identifier, AppendTableCompactor> compactorWithTable :
compactorContainer.entrySet()) {
Identifier tableId = compactorWithTable.getKey();
- UnawareBucketCompactor compactor = compactorWithTable.getValue();
+ AppendTableCompactor compactor = compactorWithTable.getValue();
for (Committable committable : compactor.prepareCommit(waitCompaction, checkpointId)) {
result.add(
@@ -107,17 +106,16 @@ public class AppendOnlyMultiTableCompactionWorkerOperator
}
@Override
- public void processElement(StreamRecord<MultiTableUnawareAppendCompactionTask> element)
- throws Exception {
+ public void processElement(StreamRecord<MultiTableAppendCompactTask> element) throws Exception {
Identifier identifier = element.getValue().tableIdentifier();
compactorContainer
.computeIfAbsent(identifier, this::compactor)
.processElement(element.getValue());
}
- private UnawareBucketCompactor compactor(Identifier tableId) {
+ private AppendTableCompactor compactor(Identifier tableId) {
try {
- return new UnawareBucketCompactor(
+ return new AppendTableCompactor(
(FileStoreTable) catalog.getTable(tableId).copy(options.toMap()),
commitUser,
this::workerExecutor,
@@ -158,10 +156,10 @@ public class AppendOnlyMultiTableCompactionWorkerOperator
exceptions.add(e);
}
- for (Map.Entry<Identifier, UnawareBucketCompactor> compactorEntry :
+ for (Map.Entry<Identifier, AppendTableCompactor> compactorEntry :
compactorContainer.entrySet()) {
try {
- UnawareBucketCompactor compactor = compactorEntry.getValue();
+ AppendTableCompactor compactor = compactorEntry.getValue();
compactor.close();
} catch (Exception e) {
Identifier id = compactorEntry.getKey();
@@ -186,7 +184,7 @@ public class AppendOnlyMultiTableCompactionWorkerOperator
/** {@link StreamOperatorFactory} of {@link AppendOnlyMultiTableCompactionWorkerOperator}. */
public static class Factory
extends PrepareCommitOperator.Factory<
- MultiTableUnawareAppendCompactionTask, MultiTableCommittable> {
+ MultiTableAppendCompactTask, MultiTableCommittable> {
private final String commitUser;
private final CatalogLoader catalogLoader;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlySingleTableCompactionWorkerOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlySingleTableCompactionWorkerOperator.java
index 917a7f64f1..da55344961 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlySingleTableCompactionWorkerOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlySingleTableCompactionWorkerOperator.java
@@ -18,8 +18,8 @@
package org.apache.paimon.flink.sink;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
-import org.apache.paimon.flink.source.BucketUnawareCompactSource;
+import org.apache.paimon.append.AppendCompactTask;
+import org.apache.paimon.flink.source.AppendTableCompactSource;
import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.streaming.api.operators.StreamOperator;
@@ -28,11 +28,11 @@ import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
/**
- * Operator to execute {@link UnawareAppendCompactionTask} passed from {@link
- * BucketUnawareCompactSource} for compacting single unaware bucket tables in divided mode.
+ * Operator to execute {@link AppendCompactTask} passed from {@link AppendTableCompactSource} for
+ * compacting single unaware bucket tables in divided mode.
*/
public class AppendOnlySingleTableCompactionWorkerOperator
- extends AppendCompactWorkerOperator<UnawareAppendCompactionTask> {
+ extends AppendCompactWorkerOperator<AppendCompactTask> {
private AppendOnlySingleTableCompactionWorkerOperator(
StreamOperatorParameters<Committable> parameters,
@@ -42,13 +42,12 @@ public class AppendOnlySingleTableCompactionWorkerOperator
}
@Override
- public void processElement(StreamRecord<UnawareAppendCompactionTask> element) throws Exception {
+ public void processElement(StreamRecord<AppendCompactTask> element) throws Exception {
this.unawareBucketCompactor.processElement(element.getValue());
}
/** {@link StreamOperatorFactory} of {@link AppendOnlySingleTableCompactionWorkerOperator}. */
- public static class Factory
- extends AppendCompactWorkerOperator.Factory<UnawareAppendCompactionTask> {
+ public static class Factory extends AppendCompactWorkerOperator.Factory<AppendCompactTask> {
public Factory(FileStoreTable table, String initialCommitUser) {
super(table, initialCommitUser);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketCompactionSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableCompactSink.java
similarity index 81%
rename from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketCompactionSink.java
rename to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableCompactSink.java
index 7a4095f896..3cf6f582c8 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketCompactionSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableCompactSink.java
@@ -18,7 +18,7 @@
package org.apache.paimon.flink.sink;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.table.FileStoreTable;
@@ -27,22 +27,22 @@ import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
/** Compaction Sink for unaware-bucket table. */
-public class UnawareBucketCompactionSink extends FlinkSink<UnawareAppendCompactionTask> {
+public class AppendTableCompactSink extends FlinkSink<AppendCompactTask> {
private final FileStoreTable table;
- public UnawareBucketCompactionSink(FileStoreTable table) {
+ public AppendTableCompactSink(FileStoreTable table) {
super(table, true);
this.table = table;
}
public static DataStreamSink<?> sink(
- FileStoreTable table, DataStream<UnawareAppendCompactionTask> input) {
- return new UnawareBucketCompactionSink(table).sinkFrom(input);
+ FileStoreTable table, DataStream<AppendCompactTask> input) {
+ return new AppendTableCompactSink(table).sinkFrom(input);
}
@Override
- protected OneInputStreamOperatorFactory<UnawareAppendCompactionTask, Committable>
+ protected OneInputStreamOperatorFactory<AppendCompactTask, Committable>
createWriteOperatorFactory(StoreSinkWrite.Provider writeProvider, String commitUser) {
return new AppendOnlySingleTableCompactionWorkerOperator.Factory(table, commitUser);
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java
similarity index 92%
rename from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketSink.java
rename to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java
index b53e2031e9..24d0e2db6f 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java
@@ -19,8 +19,8 @@
package org.apache.paimon.flink.sink;
import org.apache.paimon.flink.FlinkConnectorOptions;
-import org.apache.paimon.flink.compact.UnawareBucketNewFilesCompactionCoordinatorOperator;
-import org.apache.paimon.flink.compact.UnawareBucketNewFilesCompactionWorkerOperator;
+import org.apache.paimon.flink.compact.AppendPreCommitCompactCoordinatorOperator;
+import org.apache.paimon.flink.compact.AppendPreCommitCompactWorkerOperator;
import org.apache.paimon.flink.source.AppendBypassCoordinateOperatorFactory;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
@@ -46,14 +46,14 @@ import static org.apache.paimon.flink.utils.ParallelismUtils.setParallelism;
* <p>Note: in unaware-bucket mode, we don't shuffle by bucket in inserting. We can assign
* compaction to the inserting jobs aside.
*/
-public abstract class UnawareBucketSink<T> extends FlinkWriteSink<T> {
+public abstract class AppendTableSink<T> extends FlinkWriteSink<T> {
protected final FileStoreTable table;
protected final LogSinkFunction logSinkFunction;
@Nullable protected final Integer parallelism;
- public UnawareBucketSink(
+ public AppendTableSink(
FileStoreTable table,
@Nullable Map<String, String> overwritePartitions,
LogSinkFunction logSinkFunction,
@@ -79,14 +79,14 @@ public abstract class UnawareBucketSink<T> extends FlinkWriteSink<T> {
new TupleTypeInfo<>(
BasicTypeInfo.LONG_TYPE_INFO,
new CompactionTaskTypeInfo())),
- new UnawareBucketNewFilesCompactionCoordinatorOperator(
+ new AppendPreCommitCompactCoordinatorOperator(
table.coreOptions()))
.startNewChain()
.forceNonParallel()
.transform(
"New Files Compact Worker: " + table.name(),
new CommittableTypeInfo(),
- new UnawareBucketNewFilesCompactionWorkerOperator(table))
+ new AppendPreCommitCompactWorkerOperator(table))
.startNewChain();
forwardParallelism(newWritten, written);
written = newWritten;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CombinedTableCompactorSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CombinedTableCompactorSink.java
index 5d94c6fa2c..3d40bb6f49 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CombinedTableCompactorSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CombinedTableCompactorSink.java
@@ -19,7 +19,7 @@
package org.apache.paimon.flink.sink;
import org.apache.paimon.CoreOptions;
-import org.apache.paimon.append.MultiTableUnawareAppendCompactionTask;
+import org.apache.paimon.append.MultiTableAppendCompactTask;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.manifest.WrappedManifestCommittable;
import org.apache.paimon.options.Options;
@@ -72,7 +72,7 @@ public class CombinedTableCompactorSink implements Serializable {
public DataStreamSink<?> sinkFrom(
DataStream<RowData> awareBucketTableSource,
- DataStream<MultiTableUnawareAppendCompactionTask> unawareBucketTableSource) {
+ DataStream<MultiTableAppendCompactTask> unawareBucketTableSource) {
// This commitUser is valid only for new jobs.
// After the job starts, this commitUser will be recorded into the states of write and
// commit operators.
@@ -84,7 +84,7 @@ public class CombinedTableCompactorSink implements Serializable {
public DataStreamSink<?> sinkFrom(
DataStream<RowData> awareBucketTableSource,
- DataStream<MultiTableUnawareAppendCompactionTask> unawareBucketTableSource,
+ DataStream<MultiTableAppendCompactTask> unawareBucketTableSource,
String initialCommitUser) {
// do the actually writing action, no snapshot generated in this stage
DataStream<MultiTableCommittable> written =
@@ -96,7 +96,7 @@ public class CombinedTableCompactorSink implements Serializable {
public DataStream<MultiTableCommittable> doWrite(
DataStream<RowData> awareBucketTableSource,
- DataStream<MultiTableUnawareAppendCompactionTask> unawareBucketTableSource,
+ DataStream<MultiTableAppendCompactTask> unawareBucketTableSource,
String commitUser) {
StreamExecutionEnvironment env = awareBucketTableSource.getExecutionEnvironment();
boolean isStreaming =
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CompactionTaskSimpleSerializer.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CompactionTaskSimpleSerializer.java
index 0b86d2779b..1c6465b7b7 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CompactionTaskSimpleSerializer.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CompactionTaskSimpleSerializer.java
@@ -18,21 +18,21 @@
package org.apache.paimon.flink.sink;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
-import org.apache.paimon.table.sink.CompactionTaskSerializer;
+import org.apache.paimon.append.AppendCompactTask;
+import org.apache.paimon.table.sink.AppendCompactTaskSerializer;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import java.io.IOException;
import java.nio.ByteBuffer;
-/** {@link SimpleVersionedSerializer} for {@link UnawareAppendCompactionTask}. */
+/** {@link SimpleVersionedSerializer} for {@link AppendCompactTask}. */
public class CompactionTaskSimpleSerializer
- implements SimpleVersionedSerializer<UnawareAppendCompactionTask> {
+ implements SimpleVersionedSerializer<AppendCompactTask> {
- private final CompactionTaskSerializer compactionTaskSerializer;
+ private final AppendCompactTaskSerializer compactionTaskSerializer;
- public CompactionTaskSimpleSerializer(CompactionTaskSerializer compactionTaskSerializer) {
+ public CompactionTaskSimpleSerializer(AppendCompactTaskSerializer compactionTaskSerializer) {
this.compactionTaskSerializer = compactionTaskSerializer;
}
@@ -42,7 +42,7 @@ public class CompactionTaskSimpleSerializer
}
@Override
- public byte[] serialize(UnawareAppendCompactionTask compactionTask) throws IOException {
+ public byte[] serialize(AppendCompactTask compactionTask) throws IOException {
byte[] wrapped = compactionTaskSerializer.serialize(compactionTask);
int version = compactionTaskSerializer.getVersion();
@@ -50,7 +50,7 @@ public class CompactionTaskSimpleSerializer
}
@Override
- public UnawareAppendCompactionTask deserialize(int compactionTaskVersion, byte[] bytes)
+ public AppendCompactTask deserialize(int compactionTaskVersion, byte[] bytes)
throws IOException {
if (compactionTaskVersion != getVersion()) {
throw new RuntimeException("Can not deserialize version: " + compactionTaskVersion);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CompactionTaskTypeInfo.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CompactionTaskTypeInfo.java
index 6510a85b80..b5b9c851ef 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CompactionTaskTypeInfo.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CompactionTaskTypeInfo.java
@@ -18,16 +18,16 @@
package org.apache.paimon.flink.sink;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
-import org.apache.paimon.table.sink.CompactionTaskSerializer;
+import org.apache.paimon.append.AppendCompactTask;
+import org.apache.paimon.table.sink.AppendCompactTaskSerializer;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.serialization.SerializerConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-/** Type information of {@link UnawareAppendCompactionTask}. */
-public class CompactionTaskTypeInfo extends TypeInformation<UnawareAppendCompactionTask> {
+/** Type information of {@link AppendCompactTask}. */
+public class CompactionTaskTypeInfo extends TypeInformation<AppendCompactTask> {
@Override
public boolean isBasicType() {
@@ -50,8 +50,8 @@ public class CompactionTaskTypeInfo extends TypeInformation<UnawareAppendCompact
}
@Override
- public Class<UnawareAppendCompactionTask> getTypeClass() {
- return UnawareAppendCompactionTask.class;
+ public Class<AppendCompactTask> getTypeClass() {
+ return AppendCompactTask.class;
}
@Override
@@ -62,17 +62,17 @@ public class CompactionTaskTypeInfo extends TypeInformation<UnawareAppendCompact
/**
* Do not annotate with <code>@override</code> here to maintain compatibility with Flink 1.18-.
*/
- public TypeSerializer<UnawareAppendCompactionTask> createSerializer(SerializerConfig config) {
+ public TypeSerializer<AppendCompactTask> createSerializer(SerializerConfig config) {
return this.createSerializer((ExecutionConfig) null);
}
/**
* Do not annotate with <code>@override</code> here to maintain compatibility with Flink 2.0+.
*/
- public TypeSerializer<UnawareAppendCompactionTask> createSerializer(ExecutionConfig config) {
+ public TypeSerializer<AppendCompactTask> createSerializer(ExecutionConfig config) {
// we don't need copy for task
- return new NoneCopyVersionedSerializerTypeSerializerProxy<UnawareAppendCompactionTask>(
- () -> new CompactionTaskSimpleSerializer(new CompactionTaskSerializer())) {};
+ return new NoneCopyVersionedSerializerTypeSerializerProxy<AppendCompactTask>(
+ () -> new CompactionTaskSimpleSerializer(new AppendCompactTaskSerializer())) {};
}
@Override
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
index 19203294be..3b34289062 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
@@ -318,7 +318,7 @@ public class FlinkSinkBuilder {
parallelism);
}
- return new RowUnawareBucketSink(table, overwritePartition, logSinkFunction, parallelism)
+ return new RowAppendTableSink(table, overwritePartition, logSinkFunction, parallelism)
.sinkFrom(input);
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTableCompactionTaskTypeInfo.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTableCompactionTaskTypeInfo.java
index 0116ff1988..6a744006d0 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTableCompactionTaskTypeInfo.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTableCompactionTaskTypeInfo.java
@@ -18,7 +18,7 @@
package org.apache.paimon.flink.sink;
-import org.apache.paimon.append.MultiTableUnawareAppendCompactionTask;
+import org.apache.paimon.append.MultiTableAppendCompactTask;
import org.apache.paimon.flink.VersionedSerializerWrapper;
import org.apache.paimon.table.sink.MultiTableCompactionTaskSerializer;
@@ -28,9 +28,8 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.core.io.SimpleVersionedSerializerTypeSerializerProxy;
-/** Type information of {@link MultiTableUnawareAppendCompactionTask}. */
-public class MultiTableCompactionTaskTypeInfo
- extends TypeInformation<MultiTableUnawareAppendCompactionTask> {
+/** Type information of {@link MultiTableAppendCompactTask}. */
+public class MultiTableCompactionTaskTypeInfo extends TypeInformation<MultiTableAppendCompactTask> {
@Override
public boolean isBasicType() {
return false;
@@ -52,8 +51,8 @@ public class MultiTableCompactionTaskTypeInfo
}
@Override
- public Class<MultiTableUnawareAppendCompactionTask> getTypeClass() {
- return MultiTableUnawareAppendCompactionTask.class;
+ public Class<MultiTableAppendCompactTask> getTypeClass() {
+ return MultiTableAppendCompactTask.class;
}
@Override
@@ -64,7 +63,7 @@ public class MultiTableCompactionTaskTypeInfo
/**
* Do not annotate with <code>@override</code> here to maintain compatibility with Flink 1.18-.
*/
- public TypeSerializer<MultiTableUnawareAppendCompactionTask> createSerializer(
+ public TypeSerializer<MultiTableAppendCompactTask> createSerializer(
SerializerConfig serializerConfig) {
return this.createSerializer((ExecutionConfig) null);
}
@@ -72,21 +71,18 @@ public class MultiTableCompactionTaskTypeInfo
/**
* Do not annotate with <code>@override</code> here to maintain compatibility with Flink 2.0+.
*/
- public TypeSerializer<MultiTableUnawareAppendCompactionTask> createSerializer(
+ public TypeSerializer<MultiTableAppendCompactTask> createSerializer(
ExecutionConfig executionConfig) {
- return new SimpleVersionedSerializerTypeSerializerProxy<
- MultiTableUnawareAppendCompactionTask>(
+ return new SimpleVersionedSerializerTypeSerializerProxy<MultiTableAppendCompactTask>(
() -> new VersionedSerializerWrapper<>(new MultiTableCompactionTaskSerializer())) {
@Override
- public MultiTableUnawareAppendCompactionTask copy(
- MultiTableUnawareAppendCompactionTask from) {
+ public MultiTableAppendCompactTask copy(MultiTableAppendCompactTask from) {
return from;
}
@Override
- public MultiTableUnawareAppendCompactionTask copy(
- MultiTableUnawareAppendCompactionTask from,
- MultiTableUnawareAppendCompactionTask reuse) {
+ public MultiTableAppendCompactTask copy(
+ MultiTableAppendCompactTask from, MultiTableAppendCompactTask reuse) {
return from;
}
};
@@ -99,7 +95,7 @@ public class MultiTableCompactionTaskTypeInfo
@Override
public boolean equals(Object o) {
- return o instanceof MultiTableUnawareAppendCompactionTask;
+ return o instanceof MultiTableAppendCompactTask;
}
@Override
@@ -109,6 +105,6 @@ public class MultiTableCompactionTaskTypeInfo
@Override
public boolean canEqual(Object o) {
- return o instanceof MultiTableUnawareAppendCompactionTask;
+ return o instanceof MultiTableAppendCompactTask;
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowUnawareBucketSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
similarity index 94%
rename from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowUnawareBucketSink.java
rename to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
index fea8a382a9..b6f2381c9f 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowUnawareBucketSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
@@ -28,10 +28,10 @@ import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import java.util.Map;
-/** An {@link UnawareBucketSink} which handles {@link InternalRow}. */
-public class RowUnawareBucketSink extends UnawareBucketSink<InternalRow> {
+/** An {@link AppendTableSink} which handles {@link InternalRow}. */
+public class RowAppendTableSink extends AppendTableSink<InternalRow> {
- public RowUnawareBucketSink(
+ public RowAppendTableSink(
FileStoreTable table,
Map<String, String> overwritePartitions,
LogSinkFunction logSinkFunction,
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowDataHashPartitionChannelComputer.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowDataHashPartitionChannelComputer.java
index 964fad0955..73258e2966 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowDataHashPartitionChannelComputer.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowDataHashPartitionChannelComputer.java
@@ -20,9 +20,9 @@ package org.apache.paimon.flink.sink;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.sink.AppendTableRowKeyExtractor;
import org.apache.paimon.table.sink.ChannelComputer;
import org.apache.paimon.table.sink.KeyAndBucketExtractor;
-import org.apache.paimon.table.sink.UnawareBucketRowKeyExtractor;
/** This is only for partitioned unaware-buckets Append only table. */
public class RowDataHashPartitionChannelComputer implements ChannelComputer<InternalRow> {
@@ -41,7 +41,7 @@ public class RowDataHashPartitionChannelComputer implements ChannelComputer<Inte
@Override
public void setup(int numChannels) {
this.numChannels = numChannels;
- this.extractor = new UnawareBucketRowKeyExtractor(schema);
+ this.extractor = new AppendTableRowKeyExtractor(schema);
}
@Override
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperator.java
index b8b0d61e10..89bba8d87f 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperator.java
@@ -18,8 +18,8 @@
package org.apache.paimon.flink.source;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
-import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.append.AppendCompactCoordinator;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.flink.utils.RuntimeContextUtils;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.ExecutorUtils;
@@ -46,8 +46,8 @@ import static org.apache.paimon.utils.ThreadUtils.newDaemonThreadFactory;
* compact task to downstream operators.
*/
public class AppendBypassCoordinateOperator<CommitT>
- extends AbstractStreamOperator<Either<CommitT, UnawareAppendCompactionTask>>
- implements OneInputStreamOperator<CommitT, Either<CommitT, UnawareAppendCompactionTask>>,
+ extends AbstractStreamOperator<Either<CommitT, AppendCompactTask>>
+ implements OneInputStreamOperator<CommitT, Either<CommitT, AppendCompactTask>>,
ProcessingTimeCallback {
private static final long MAX_PENDING_TASKS = 5000;
@@ -55,10 +55,10 @@ public class AppendBypassCoordinateOperator<CommitT>
private final FileStoreTable table;
private transient ScheduledExecutorService executorService;
- private transient LinkedBlockingQueue<UnawareAppendCompactionTask> compactTasks;
+ private transient LinkedBlockingQueue<AppendCompactTask> compactTasks;
public AppendBypassCoordinateOperator(
- StreamOperatorParameters<Either<CommitT, UnawareAppendCompactionTask>> parameters,
+ StreamOperatorParameters<Either<CommitT, AppendCompactTask>> parameters,
FileStoreTable table,
ProcessingTimeService processingTimeService) {
this.table = table;
@@ -74,8 +74,7 @@ public class AppendBypassCoordinateOperator<CommitT>
"Compaction Coordinator parallelism in paimon MUST be one.");
long intervalMs = table.coreOptions().continuousDiscoveryInterval().toMillis();
this.compactTasks = new LinkedBlockingQueue<>();
- UnawareAppendTableCompactionCoordinator coordinator =
- new UnawareAppendTableCompactionCoordinator(table, true, null);
+ AppendCompactCoordinator coordinator = new AppendCompactCoordinator(table, true, null);
this.executorService =
Executors.newSingleThreadScheduledExecutor(
newDaemonThreadFactory("Compaction Coordinator"));
@@ -84,9 +83,9 @@ public class AppendBypassCoordinateOperator<CommitT>
this.getProcessingTimeService().scheduleWithFixedDelay(this, 0, intervalMs);
}
- private void asyncPlan(UnawareAppendTableCompactionCoordinator coordinator) {
+ private void asyncPlan(AppendCompactCoordinator coordinator) {
while (compactTasks.size() < MAX_PENDING_TASKS) {
- List<UnawareAppendCompactionTask> tasks = coordinator.run();
+ List<AppendCompactTask> tasks = coordinator.run();
compactTasks.addAll(tasks);
if (tasks.isEmpty()) {
break;
@@ -97,7 +96,7 @@ public class AppendBypassCoordinateOperator<CommitT>
@Override
public void onProcessingTime(long time) {
while (true) {
- UnawareAppendCompactionTask task = compactTasks.poll();
+ AppendCompactTask task = compactTasks.poll();
if (task == null) {
return;
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperatorFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperatorFactory.java
index a4c51e5b5a..5b9694e7f9 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperatorFactory.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperatorFactory.java
@@ -18,7 +18,7 @@
package org.apache.paimon.flink.source;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
@@ -29,9 +29,8 @@ import org.apache.flink.types.Either;
/** Factory of {@link AppendBypassCoordinateOperator}. */
public class AppendBypassCoordinateOperatorFactory<CommitT>
- extends AbstractStreamOperatorFactory<Either<CommitT, UnawareAppendCompactionTask>>
- implements OneInputStreamOperatorFactory<
- CommitT, Either<CommitT, UnawareAppendCompactionTask>> {
+ extends AbstractStreamOperatorFactory<Either<CommitT, AppendCompactTask>>
+ implements OneInputStreamOperatorFactory<CommitT, Either<CommitT, AppendCompactTask>> {
private final FileStoreTable table;
@@ -40,10 +39,8 @@ public class AppendBypassCoordinateOperatorFactory<CommitT>
}
@Override
- public <T extends StreamOperator<Either<CommitT, UnawareAppendCompactionTask>>>
- T createStreamOperator(
- StreamOperatorParameters<Either<CommitT, UnawareAppendCompactionTask>>
- parameters) {
+ public <T extends StreamOperator<Either<CommitT, AppendCompactTask>>> T createStreamOperator(
+ StreamOperatorParameters<Either<CommitT, AppendCompactTask>> parameters) {
AppendBypassCoordinateOperator<CommitT> operator =
new AppendBypassCoordinateOperator<>(parameters, table, processingTimeService);
return (T) operator;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BucketUnawareCompactSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendTableCompactSource.java
similarity index 76%
rename from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BucketUnawareCompactSource.java
rename to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendTableCompactSource.java
index 7954aad2df..7c707e850f 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BucketUnawareCompactSource.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendTableCompactSource.java
@@ -18,8 +18,8 @@
package org.apache.paimon.flink.source;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
-import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.append.AppendCompactCoordinator;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.flink.sink.CompactionTaskTypeInfo;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.FileStoreTable;
@@ -50,10 +50,9 @@ import java.util.List;
* we don't need to save state in this source, it will invoke a full scan when starting up, and scan
* continuously for the following snapshot.
*/
-public class BucketUnawareCompactSource
- extends AbstractNonCoordinatedSource<UnawareAppendCompactionTask> {
+public class AppendTableCompactSource extends AbstractNonCoordinatedSource<AppendCompactTask> {
- private static final Logger LOG = LoggerFactory.getLogger(BucketUnawareCompactSource.class);
+ private static final Logger LOG = LoggerFactory.getLogger(AppendTableCompactSource.class);
private static final String COMPACTION_COORDINATOR_NAME = "Compaction Coordinator";
private final FileStoreTable table;
@@ -61,7 +60,7 @@ public class BucketUnawareCompactSource
private final long scanInterval;
private final Predicate filter;
- public BucketUnawareCompactSource(
+ public AppendTableCompactSource(
FileStoreTable table,
boolean isStreaming,
long scanInterval,
@@ -78,34 +77,32 @@ public class BucketUnawareCompactSource
}
@Override
- public SourceReader<UnawareAppendCompactionTask, SimpleSourceSplit> createReader(
+ public SourceReader<AppendCompactTask, SimpleSourceSplit> createReader(
SourceReaderContext readerContext) throws Exception {
Preconditions.checkArgument(
readerContext.currentParallelism() == 1,
"Compaction Operator parallelism in paimon MUST be one.");
- return new BucketUnawareCompactSourceReader(table, streaming, filter, scanInterval);
+ return new CompactSourceReader(table, streaming, filter, scanInterval);
}
/** BucketUnawareCompactSourceReader. */
- public static class BucketUnawareCompactSourceReader
- extends AbstractNonCoordinatedSourceReader<UnawareAppendCompactionTask> {
- private final UnawareAppendTableCompactionCoordinator compactionCoordinator;
+ public static class CompactSourceReader
+ extends AbstractNonCoordinatedSourceReader<AppendCompactTask> {
+ private final AppendCompactCoordinator compactionCoordinator;
private final long scanInterval;
- public BucketUnawareCompactSourceReader(
+ public CompactSourceReader(
FileStoreTable table, boolean streaming, Predicate filter, long scanInterval) {
this.scanInterval = scanInterval;
- compactionCoordinator =
- new UnawareAppendTableCompactionCoordinator(table, streaming, filter);
+ compactionCoordinator = new AppendCompactCoordinator(table, streaming, filter);
}
@Override
- public InputStatus pollNext(ReaderOutput<UnawareAppendCompactionTask> readerOutput)
- throws Exception {
+ public InputStatus pollNext(ReaderOutput<AppendCompactTask> readerOutput) throws Exception {
boolean isEmpty;
try {
// do scan and plan action, emit append-only compaction tasks.
- List<UnawareAppendCompactionTask> tasks = compactionCoordinator.run();
+ List<AppendCompactTask> tasks = compactionCoordinator.run();
isEmpty = tasks.isEmpty();
tasks.forEach(readerOutput::collect);
} catch (EndOfScanException esf) {
@@ -120,11 +117,11 @@ public class BucketUnawareCompactSource
}
}
- public static DataStreamSource<UnawareAppendCompactionTask> buildSource(
+ public static DataStreamSource<AppendCompactTask> buildSource(
StreamExecutionEnvironment env,
- BucketUnawareCompactSource source,
+ AppendTableCompactSource source,
String tableIdentifier) {
- return (DataStreamSource<UnawareAppendCompactionTask>)
+ return (DataStreamSource<AppendCompactTask>)
env.fromSource(
source,
WatermarkStrategy.noWatermarks(),
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CombinedTableCompactorSourceBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CombinedTableCompactorSourceBuilder.java
index 0d5ec299f9..8f7d49b5eb 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CombinedTableCompactorSourceBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CombinedTableCompactorSourceBuilder.java
@@ -18,13 +18,13 @@
package org.apache.paimon.flink.source;
-import org.apache.paimon.append.MultiTableUnawareAppendCompactionTask;
+import org.apache.paimon.append.MultiTableAppendCompactTask;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.flink.LogicalTypeConversion;
+import org.apache.paimon.flink.source.operator.CombinedAppendCompactSource;
+import org.apache.paimon.flink.source.operator.CombinedAppendCompactStreamSource;
import org.apache.paimon.flink.source.operator.CombinedAwareBatchSource;
import org.apache.paimon.flink.source.operator.CombinedAwareStreamingSource;
-import org.apache.paimon.flink.source.operator.CombinedUnawareBatchSource;
-import org.apache.paimon.flink.source.operator.CombinedUnawareStreamingSource;
import org.apache.paimon.table.system.CompactBucketsTable;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Preconditions;
@@ -120,10 +120,10 @@ public class CombinedTableCompactorSourceBuilder {
}
}
- public DataStream<MultiTableUnawareAppendCompactionTask> buildForUnawareBucketsTableSource() {
+ public DataStream<MultiTableAppendCompactTask> buildForUnawareBucketsTableSource() {
Preconditions.checkArgument(env != null, "StreamExecutionEnvironment should not be null.");
if (isContinuous) {
- return CombinedUnawareStreamingSource.buildSource(
+ return CombinedAppendCompactStreamSource.buildSource(
env,
"Combined-UnawareBucketTables-StreamingCompactorSource",
catalogLoader,
@@ -133,7 +133,7 @@ public class CombinedTableCompactorSourceBuilder {
tableOptions,
monitorInterval);
} else {
- return CombinedUnawareBatchSource.buildSource(
+ return CombinedAppendCompactSource.buildSource(
env,
"Combined-UnawareBucketTables-BatchCompactorSource",
catalogLoader,
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareBatchSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAppendCompactSource.java
similarity index 83%
rename from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareBatchSource.java
rename to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAppendCompactSource.java
index 54231bab1c..32bece2c78 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareBatchSource.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAppendCompactSource.java
@@ -18,13 +18,13 @@
package org.apache.paimon.flink.source.operator;
-import org.apache.paimon.append.MultiTableUnawareAppendCompactionTask;
+import org.apache.paimon.append.MultiTableAppendCompactTask;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.compact.MultiAppendCompactTableScan;
import org.apache.paimon.flink.compact.MultiTableScanBase;
-import org.apache.paimon.flink.compact.MultiUnawareBucketTableScan;
import org.apache.paimon.flink.sink.MultiTableCompactionTaskTypeInfo;
import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
import org.apache.paimon.flink.source.SimpleSourceSplit;
@@ -58,13 +58,13 @@ import static org.apache.paimon.flink.compact.MultiTableScanBase.ScanResult.IS_E
* It is responsible for the batch compactor source of the table with unaware bucket in combined
* mode.
*/
-public class CombinedUnawareBatchSource
- extends CombinedCompactorSource<MultiTableUnawareAppendCompactionTask> {
+public class CombinedAppendCompactSource
+ extends CombinedCompactorSource<MultiTableAppendCompactTask> {
- private static final Logger LOGGER = LoggerFactory.getLogger(CombinedUnawareBatchSource.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(CombinedAppendCompactSource.class);
private final Map<String, String> tableOptions;
- public CombinedUnawareBatchSource(
+ public CombinedAppendCompactSource(
CatalogLoader catalogLoader,
Pattern includingPattern,
Pattern excludingPattern,
@@ -75,20 +75,19 @@ public class CombinedUnawareBatchSource
}
@Override
- public SourceReader<MultiTableUnawareAppendCompactionTask, SimpleSourceSplit> createReader(
+ public SourceReader<MultiTableAppendCompactTask, SimpleSourceSplit> createReader(
SourceReaderContext sourceReaderContext) throws Exception {
return new Reader();
}
- private class Reader
- extends AbstractNonCoordinatedSourceReader<MultiTableUnawareAppendCompactionTask> {
- private transient MultiTableScanBase<MultiTableUnawareAppendCompactionTask> tableScan;
+ private class Reader extends AbstractNonCoordinatedSourceReader<MultiTableAppendCompactTask> {
+ private transient MultiTableScanBase<MultiTableAppendCompactTask> tableScan;
@Override
public void start() {
super.start();
tableScan =
- new MultiUnawareBucketTableScan(
+ new MultiAppendCompactTableScan(
catalogLoader,
includingPattern,
excludingPattern,
@@ -98,8 +97,8 @@ public class CombinedUnawareBatchSource
}
@Override
- public InputStatus pollNext(
- ReaderOutput<MultiTableUnawareAppendCompactionTask> readerOutput) throws Exception {
+ public InputStatus pollNext(ReaderOutput<MultiTableAppendCompactTask> readerOutput)
+ throws Exception {
MultiTableScanBase.ScanResult scanResult = tableScan.scanTable(readerOutput);
if (scanResult == FINISHED) {
return InputStatus.END_OF_INPUT;
@@ -123,7 +122,7 @@ public class CombinedUnawareBatchSource
}
}
- public static DataStream<MultiTableUnawareAppendCompactionTask> buildSource(
+ public static DataStream<MultiTableAppendCompactTask> buildSource(
StreamExecutionEnvironment env,
String name,
CatalogLoader catalogLoader,
@@ -132,8 +131,8 @@ public class CombinedUnawareBatchSource
Pattern databasePattern,
Map<String, String> tableOptions,
@Nullable Duration partitionIdleTime) {
- CombinedUnawareBatchSource combinedUnawareBatchSource =
- new CombinedUnawareBatchSource(
+ CombinedAppendCompactSource source =
+ new CombinedAppendCompactSource(
catalogLoader,
includingPattern,
excludingPattern,
@@ -142,23 +141,24 @@ public class CombinedUnawareBatchSource
MultiTableCompactionTaskTypeInfo compactionTaskTypeInfo =
new MultiTableCompactionTaskTypeInfo();
- SingleOutputStreamOperator<MultiTableUnawareAppendCompactionTask> source =
+ SingleOutputStreamOperator<MultiTableAppendCompactTask> operator =
env.fromSource(
- combinedUnawareBatchSource,
+ source,
WatermarkStrategy.noWatermarks(),
name,
compactionTaskTypeInfo)
.forceNonParallel();
if (partitionIdleTime != null) {
- source =
- source.transform(
+ operator =
+ operator.transform(
name,
compactionTaskTypeInfo,
- new MultiUnawareTablesReadOperator(catalogLoader, partitionIdleTime));
+ new MultiTableAppendCompactReadOperator(
+ catalogLoader, partitionIdleTime));
}
- return source;
+ return operator;
}
private static Long getPartitionInfo(
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareStreamingSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAppendCompactStreamSource.java
similarity index 81%
rename from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareStreamingSource.java
rename to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAppendCompactStreamSource.java
index c4270228b3..34e062801b 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareStreamingSource.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAppendCompactStreamSource.java
@@ -18,10 +18,10 @@
package org.apache.paimon.flink.source.operator;
-import org.apache.paimon.append.MultiTableUnawareAppendCompactionTask;
+import org.apache.paimon.append.MultiTableAppendCompactTask;
import org.apache.paimon.catalog.CatalogLoader;
+import org.apache.paimon.flink.compact.MultiAppendCompactTableScan;
import org.apache.paimon.flink.compact.MultiTableScanBase;
-import org.apache.paimon.flink.compact.MultiUnawareBucketTableScan;
import org.apache.paimon.flink.sink.MultiTableCompactionTaskTypeInfo;
import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
import org.apache.paimon.flink.source.SimpleSourceSplit;
@@ -43,13 +43,13 @@ import static org.apache.paimon.flink.compact.MultiTableScanBase.ScanResult.IS_E
/**
* It is responsible for monitoring compactor source in stream mode for the table of unaware bucket.
*/
-public class CombinedUnawareStreamingSource
- extends CombinedCompactorSource<MultiTableUnawareAppendCompactionTask> {
+public class CombinedAppendCompactStreamSource
+ extends CombinedCompactorSource<MultiTableAppendCompactTask> {
private final long monitorInterval;
private final Map<String, String> tableOptions;
- public CombinedUnawareStreamingSource(
+ public CombinedAppendCompactStreamSource(
CatalogLoader catalogLoader,
Pattern includingPattern,
Pattern excludingPattern,
@@ -62,20 +62,19 @@ public class CombinedUnawareStreamingSource
}
@Override
- public SourceReader<MultiTableUnawareAppendCompactionTask, SimpleSourceSplit> createReader(
+ public SourceReader<MultiTableAppendCompactTask, SimpleSourceSplit> createReader(
SourceReaderContext sourceReaderContext) throws Exception {
return new Reader();
}
- private class Reader
- extends AbstractNonCoordinatedSourceReader<MultiTableUnawareAppendCompactionTask> {
- private MultiTableScanBase<MultiTableUnawareAppendCompactionTask> tableScan;
+ private class Reader extends AbstractNonCoordinatedSourceReader<MultiTableAppendCompactTask> {
+ private MultiTableScanBase<MultiTableAppendCompactTask> tableScan;
@Override
public void start() {
super.start();
tableScan =
- new MultiUnawareBucketTableScan(
+ new MultiAppendCompactTableScan(
catalogLoader,
includingPattern,
excludingPattern,
@@ -85,8 +84,8 @@ public class CombinedUnawareStreamingSource
}
@Override
- public InputStatus pollNext(
- ReaderOutput<MultiTableUnawareAppendCompactionTask> readerOutput) throws Exception {
+ public InputStatus pollNext(ReaderOutput<MultiTableAppendCompactTask> readerOutput)
+ throws Exception {
MultiTableScanBase.ScanResult scanResult = tableScan.scanTable(readerOutput);
if (scanResult == FINISHED) {
return InputStatus.END_OF_INPUT;
@@ -106,7 +105,7 @@ public class CombinedUnawareStreamingSource
}
}
- public static DataStream<MultiTableUnawareAppendCompactionTask> buildSource(
+ public static DataStream<MultiTableAppendCompactTask> buildSource(
StreamExecutionEnvironment env,
String name,
CatalogLoader catalogLoader,
@@ -116,8 +115,8 @@ public class CombinedUnawareStreamingSource
Map<String, String> tableOptions,
long monitorInterval) {
- CombinedUnawareStreamingSource source =
- new CombinedUnawareStreamingSource(
+ CombinedAppendCompactStreamSource source =
+ new CombinedAppendCompactStreamSource(
catalogLoader,
includingPattern,
excludingPattern,
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedCompactorSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedCompactorSource.java
index e292d2441c..9960c667b5 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedCompactorSource.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedCompactorSource.java
@@ -18,7 +18,7 @@
package org.apache.paimon.flink.source.operator;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.flink.source.AbstractNonCoordinatedSource;
import org.apache.paimon.table.source.Split;
@@ -32,7 +32,7 @@ import java.util.regex.Pattern;
*
* <ol>
* <li>Monitoring snapshots of the Paimon table.
- * <li>Creating the splits {@link Split} or compaction task {@link UnawareAppendCompactionTask}
+ * <li>Creating the splits {@link Split} or compaction task {@link AppendCompactTask}
* corresponding to the incremental files
* <li>Assigning them to downstream tasks for further processing.
* </ol>
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiUnawareTablesReadOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTableAppendCompactReadOperator.java
similarity index 85%
rename from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiUnawareTablesReadOperator.java
rename to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTableAppendCompactReadOperator.java
index 15fde93755..30ec85bea6 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiUnawareTablesReadOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTableAppendCompactReadOperator.java
@@ -18,7 +18,7 @@
package org.apache.paimon.flink.source.operator;
-import org.apache.paimon.append.MultiTableUnawareAppendCompactionTask;
+import org.apache.paimon.append.MultiTableAppendCompactTask;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.catalog.Identifier;
@@ -44,22 +44,24 @@ import java.util.stream.Collectors;
/**
* The operator is used for historical partition compaction. It reads {@link
- * MultiTableUnawareAppendCompactionTask} received from the preceding {@link
- * CombinedUnawareBatchSource} and filter partitions which is not historical.
+ * MultiTableAppendCompactTask} received from the preceding {@link CombinedAppendCompactSource} and
+ * filter partitions which is not historical.
*/
-public class MultiUnawareTablesReadOperator
- extends AbstractStreamOperator<MultiTableUnawareAppendCompactionTask>
+public class MultiTableAppendCompactReadOperator
+ extends AbstractStreamOperator<MultiTableAppendCompactTask>
implements OneInputStreamOperator<
- MultiTableUnawareAppendCompactionTask, MultiTableUnawareAppendCompactionTask> {
+ MultiTableAppendCompactTask, MultiTableAppendCompactTask> {
private static final long serialVersionUID = 1L;
- private static final Logger LOG = LoggerFactory.getLogger(MultiUnawareTablesReadOperator.class);
+ private static final Logger LOG =
+ LoggerFactory.getLogger(MultiTableAppendCompactReadOperator.class);
private final CatalogLoader catalogLoader;
private final Duration partitionIdleTime;
- public MultiUnawareTablesReadOperator(CatalogLoader catalogLoader, Duration partitionIdleTime) {
+ public MultiTableAppendCompactReadOperator(
+ CatalogLoader catalogLoader, Duration partitionIdleTime) {
this.catalogLoader = catalogLoader;
this.partitionIdleTime = partitionIdleTime;
}
@@ -75,7 +77,7 @@ public class MultiUnawareTablesReadOperator
}
@Override
- public void processElement(StreamRecord<MultiTableUnawareAppendCompactionTask> record) {
+ public void processElement(StreamRecord<MultiTableAppendCompactTask> record) {
Identifier identifier = record.getValue().tableIdentifier();
BinaryRow partition = record.getValue().partition();
FileStoreTable table = getTable(identifier);
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/UnawareBucketAppendOnlyTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableITCase.java
similarity index 99%
rename from paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/UnawareBucketAppendOnlyTableITCase.java
rename to paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableITCase.java
index c8a91431b2..025a7f4b5a 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/UnawareBucketAppendOnlyTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableITCase.java
@@ -67,8 +67,8 @@ import static org.apache.paimon.CoreOptions.PARTITION_SINK_STRATEGY;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-/** Test case for append-only managed unaware-bucket table. */
-public class UnawareBucketAppendOnlyTableITCase extends CatalogITCaseBase {
+/** Test case for append table. */
+public class AppendTableITCase extends CatalogITCaseBase {
private static final Random RANDOM = new Random();
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RangePartitionAndSortForUnawareBucketTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RangePartitionAndSortForAppendTableITCase.java
similarity index 99%
rename from paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RangePartitionAndSortForUnawareBucketTableITCase.java
rename to paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RangePartitionAndSortForAppendTableITCase.java
index c65f4ac12a..2dce791b88 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RangePartitionAndSortForUnawareBucketTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RangePartitionAndSortForAppendTableITCase.java
@@ -46,7 +46,7 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
* The IT test for the range partitioning and sort batch writing for append-only bucket unaware
* tables.
*/
-public class RangePartitionAndSortForUnawareBucketTableITCase extends CatalogITCaseBase {
+public class RangePartitionAndSortForAppendTableITCase extends CatalogITCaseBase {
private static final int SINK_ROW_NUMBER = 1000;
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForUnawareBucketITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForAppendTableITCase.java
similarity index 99%
rename from paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForUnawareBucketITCase.java
rename to paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForAppendTableITCase.java
index ed645a9079..d871c13910 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForUnawareBucketITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForAppendTableITCase.java
@@ -52,7 +52,7 @@ import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;
/** Order Rewrite Action tests for {@link SortCompactAction}. */
-public class SortCompactActionForUnawareBucketITCase extends ActionITCaseBase {
+public class SortCompactActionForAppendTableITCase extends ActionITCaseBase {
private static final Random RANDOM = new Random();
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/UnawareBucketNewFilesCompactionCoordinatorOperatorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/AppendPreCommitCompactCoordinatorOperatorTest.java
similarity index 87%
rename from paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/UnawareBucketNewFilesCompactionCoordinatorOperatorTest.java
rename to paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/AppendPreCommitCompactCoordinatorOperatorTest.java
index 326ed9b6a2..6675a1358a 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/UnawareBucketNewFilesCompactionCoordinatorOperatorTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/AppendPreCommitCompactCoordinatorOperatorTest.java
@@ -19,7 +19,7 @@
package org.apache.paimon.flink.compact;
import org.apache.paimon.CoreOptions;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.flink.sink.Committable;
@@ -57,17 +57,17 @@ import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
-/** Tests for {@link UnawareBucketNewFilesCompactionCoordinatorOperator}. */
-public class UnawareBucketNewFilesCompactionCoordinatorOperatorTest {
+/** Tests for {@link AppendPreCommitCompactCoordinatorOperator}. */
+public class AppendPreCommitCompactCoordinatorOperatorTest {
@Test
public void testPrepareSnapshotWithMultipleFiles() throws Exception {
Options options = new Options();
options.set(CoreOptions.TARGET_FILE_SIZE, MemorySize.ofMebiBytes(8));
- UnawareBucketNewFilesCompactionCoordinatorOperator operator =
- new UnawareBucketNewFilesCompactionCoordinatorOperator(new CoreOptions(options));
+ AppendPreCommitCompactCoordinatorOperator operator =
+ new AppendPreCommitCompactCoordinatorOperator(new CoreOptions(options));
OneInputStreamOperatorTestHarness<
- Committable, Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>>
+ Committable, Either<Committable, Tuple2<Long, AppendCompactTask>>>
testHarness = createTestHarness(operator);
testHarness.open();
@@ -91,10 +91,10 @@ public class UnawareBucketNewFilesCompactionCoordinatorOperatorTest {
public void testPrepareSnapshotWithSingleFile() throws Exception {
Options options = new Options();
options.set(CoreOptions.TARGET_FILE_SIZE, MemorySize.ofMebiBytes(8));
- UnawareBucketNewFilesCompactionCoordinatorOperator operator =
- new UnawareBucketNewFilesCompactionCoordinatorOperator(new CoreOptions(options));
+ AppendPreCommitCompactCoordinatorOperator operator =
+ new AppendPreCommitCompactCoordinatorOperator(new CoreOptions(options));
OneInputStreamOperatorTestHarness<
- Committable, Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>>
+ Committable, Either<Committable, Tuple2<Long, AppendCompactTask>>>
testHarness = createTestHarness(operator);
testHarness.open();
@@ -118,10 +118,10 @@ public class UnawareBucketNewFilesCompactionCoordinatorOperatorTest {
public void testPrepareSnapshotWithMultiplePartitions() throws Exception {
Options options = new Options();
options.set(CoreOptions.TARGET_FILE_SIZE, MemorySize.ofMebiBytes(8));
- UnawareBucketNewFilesCompactionCoordinatorOperator operator =
- new UnawareBucketNewFilesCompactionCoordinatorOperator(new CoreOptions(options));
+ AppendPreCommitCompactCoordinatorOperator operator =
+ new AppendPreCommitCompactCoordinatorOperator(new CoreOptions(options));
OneInputStreamOperatorTestHarness<
- Committable, Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>>
+ Committable, Either<Committable, Tuple2<Long, AppendCompactTask>>>
testHarness = createTestHarness(operator);
Function<Integer, BinaryRow> binaryRow =
@@ -159,10 +159,10 @@ public class UnawareBucketNewFilesCompactionCoordinatorOperatorTest {
public void testSkipLargeFiles() throws Exception {
Options options = new Options();
options.set(CoreOptions.TARGET_FILE_SIZE, MemorySize.ofMebiBytes(8));
- UnawareBucketNewFilesCompactionCoordinatorOperator operator =
- new UnawareBucketNewFilesCompactionCoordinatorOperator(new CoreOptions(options));
+ AppendPreCommitCompactCoordinatorOperator operator =
+ new AppendPreCommitCompactCoordinatorOperator(new CoreOptions(options));
OneInputStreamOperatorTestHarness<
- Committable, Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>>
+ Committable, Either<Committable, Tuple2<Long, AppendCompactTask>>>
testHarness = createTestHarness(operator);
testHarness.open();
@@ -180,8 +180,8 @@ public class UnawareBucketNewFilesCompactionCoordinatorOperatorTest {
@SuppressWarnings("unchecked")
private void assertCommittable(Object o, long checkpointId, BinaryRow partition, int... mbs) {
- StreamRecord<Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>> record =
- (StreamRecord<Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>>) o;
+ StreamRecord<Either<Committable, Tuple2<Long, AppendCompactTask>>> record =
+ (StreamRecord<Either<Committable, Tuple2<Long, AppendCompactTask>>>) o;
assertThat(record.getValue().isLeft()).isTrue();
Committable committable = record.getValue().left();
assertThat(committable.checkpointId()).isEqualTo(checkpointId);
@@ -201,11 +201,11 @@ public class UnawareBucketNewFilesCompactionCoordinatorOperatorTest {
@SuppressWarnings("unchecked")
private void assertCompactionTask(
Object o, long checkpointId, BinaryRow partition, int... mbs) {
- StreamRecord<Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>> record =
- (StreamRecord<Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>>) o;
+ StreamRecord<Either<Committable, Tuple2<Long, AppendCompactTask>>> record =
+ (StreamRecord<Either<Committable, Tuple2<Long, AppendCompactTask>>>) o;
assertThat(record.getValue().isRight()).isTrue();
assertThat(record.getValue().right().f0).isEqualTo(checkpointId);
- UnawareAppendCompactionTask task = record.getValue().right().f1;
+ AppendCompactTask task = record.getValue().right().f1;
assertThat(task.partition()).isEqualTo(partition);
assertThat(task.compactBefore().stream().map(DataFileMeta::fileSize))
.hasSameElementsAs(
@@ -248,9 +248,8 @@ public class UnawareBucketNewFilesCompactionCoordinatorOperatorTest {
@SuppressWarnings({"rawtypes", "unchecked"})
private OneInputStreamOperatorTestHarness<
- Committable, Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>>
- createTestHarness(UnawareBucketNewFilesCompactionCoordinatorOperator operator)
- throws Exception {
+ Committable, Either<Committable, Tuple2<Long, AppendCompactTask>>>
+ createTestHarness(AppendPreCommitCompactCoordinatorOperator operator) throws Exception {
TypeSerializer serializer =
new EitherSerializer<>(
new CommittableTypeInfo().createSerializer(new ExecutionConfig()),
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/UnawareBucketNewFilesCompactionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/AppendPreCommitCompactITCase.java
similarity index 96%
rename from paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/UnawareBucketNewFilesCompactionITCase.java
rename to paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/AppendPreCommitCompactITCase.java
index 54f3737316..d52ea26a4e 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/UnawareBucketNewFilesCompactionITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/AppendPreCommitCompactITCase.java
@@ -37,10 +37,10 @@ import java.util.function.Supplier;
import static org.assertj.core.api.Assertions.assertThat;
/**
- * IT cases for {@link UnawareBucketNewFilesCompactionCoordinatorOperator} and {@link
- * UnawareBucketNewFilesCompactionWorkerOperator}.
+ * IT cases for {@link AppendPreCommitCompactCoordinatorOperator} and {@link
+ * AppendPreCommitCompactWorkerOperator}.
*/
-public class UnawareBucketNewFilesCompactionITCase extends AbstractTestBase {
+public class AppendPreCommitCompactITCase extends AbstractTestBase {
@Test
public void testCompactNewFiles() throws Exception {
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/UnawareBucketCompactorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/AppendTableCompactorTest.java
similarity index 92%
rename from paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/UnawareBucketCompactorTest.java
rename to paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/AppendTableCompactorTest.java
index 1ec26afabc..6e534f832c 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/UnawareBucketCompactorTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/AppendTableCompactorTest.java
@@ -18,14 +18,14 @@
package org.apache.paimon.flink.compact;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.flink.source.FileStoreSourceReaderTest;
-import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
+import org.apache.paimon.operation.BaseAppendFileStoreWrite;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
@@ -54,8 +54,8 @@ import static org.apache.paimon.operation.metrics.CompactionMetrics.COMPACTION_C
import static org.apache.paimon.operation.metrics.CompactionMetrics.COMPACTION_QUEUED_COUNT;
import static org.apache.paimon.operation.metrics.CompactionMetrics.COMPACTION_THREAD_BUSY;
-/** Test for {@link UnawareBucketCompactor}. */
-public class UnawareBucketCompactorTest {
+/** Test for {@link AppendTableCompactor}. */
+public class AppendTableCompactorTest {
@TempDir private Path dir;
private String tableName = "Orders1";
@@ -71,8 +71,8 @@ public class UnawareBucketCompactorTest {
Thread.currentThread().getName() + "-append-only-compact-worker"));
Map<String, Gauge> gaugeMap = new HashMap<>();
Map<String, Counter> counterMap = new HashMap<>();
- UnawareBucketCompactor unawareBucketCompactor =
- new UnawareBucketCompactor(
+ AppendTableCompactor unawareBucketCompactor =
+ new AppendTableCompactor(
(FileStoreTable) catalog.getTable(identifier()),
"10086",
() -> executorService,
@@ -102,7 +102,7 @@ public class UnawareBucketCompactorTest {
});
for (int i = 0; i < 320; i++) {
- unawareBucketCompactor.processElement(new MockCompactionTask());
+ unawareBucketCompactor.processElement(new MockCompactTask());
Thread.sleep(250);
}
@@ -158,14 +158,14 @@ public class UnawareBucketCompactorTest {
}
/** Mock compaction task for test. */
- private static class MockCompactionTask extends UnawareAppendCompactionTask {
+ private static class MockCompactTask extends AppendCompactTask {
- public MockCompactionTask() {
+ public MockCompactTask() {
super(BinaryRow.EMPTY_ROW, Collections.emptyList());
}
@Override
- public CommitMessage doCompact(FileStoreTable table, AppendOnlyFileStoreWrite write)
+ public CommitMessage doCompact(FileStoreTable table, BaseAppendFileStoreWrite write)
throws Exception {
Thread.sleep(125);
return null;
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AppendOnlyMultiTableCompactionWorkerOperatorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AppendOnlyMultiTableCompactionWorkerOperatorTest.java
index 949c2c7a66..b2268cec99 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AppendOnlyMultiTableCompactionWorkerOperatorTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AppendOnlyMultiTableCompactionWorkerOperatorTest.java
@@ -18,7 +18,7 @@
package org.apache.paimon.flink.sink;
-import org.apache.paimon.append.MultiTableUnawareAppendCompactionTask;
+import org.apache.paimon.append.MultiTableAppendCompactTask;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.TableTestBase;
@@ -63,7 +63,7 @@ public class AppendOnlyMultiTableCompactionWorkerOperatorTest extends TableTestB
null,
null));
- List<StreamRecord<MultiTableUnawareAppendCompactionTask>> records = new ArrayList<>();
+ List<StreamRecord<MultiTableAppendCompactTask>> records = new ArrayList<>();
// create table and write
for (String table : tables) {
Identifier identifier = identifier(table);
@@ -76,7 +76,7 @@ public class AppendOnlyMultiTableCompactionWorkerOperatorTest extends TableTestB
.map(
task ->
new StreamRecord<>(
- new MultiTableUnawareAppendCompactionTask(
+ new MultiTableAppendCompactTask(
task.partition(),
task.compactBefore(),
identifier)))
@@ -86,7 +86,7 @@ public class AppendOnlyMultiTableCompactionWorkerOperatorTest extends TableTestB
Assertions.assertThat(records.size()).isEqualTo(8);
workerOperator.open();
- for (StreamRecord<MultiTableUnawareAppendCompactionTask> record : records) {
+ for (StreamRecord<MultiTableAppendCompactTask> record : records) {
workerOperator.processElement(record);
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AppendOnlySingleTableCompactionWorkerOperatorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AppendOnlySingleTableCompactionWorkerOperatorTest.java
index c30bfdc03a..ae98f2dd34 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AppendOnlySingleTableCompactionWorkerOperatorTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AppendOnlySingleTableCompactionWorkerOperatorTest.java
@@ -19,7 +19,7 @@
package org.apache.paimon.flink.sink;
import org.apache.paimon.CoreOptions;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
@@ -71,14 +71,14 @@ public class AppendOnlySingleTableCompactionWorkerOperatorTest extends TableTest
// write 200 files
List<CommitMessage> commitMessages = writeDataDefault(200, 20);
- List<UnawareAppendCompactionTask> tasks = packTask(commitMessages, 5);
- List<StreamRecord<UnawareAppendCompactionTask>> records =
+ List<AppendCompactTask> tasks = packTask(commitMessages, 5);
+ List<StreamRecord<AppendCompactTask>> records =
tasks.stream().map(StreamRecord::new).collect(Collectors.toList());
assertThat(tasks.size()).isEqualTo(4);
workerOperator.open();
- for (StreamRecord<UnawareAppendCompactionTask> record : records) {
+ for (StreamRecord<AppendCompactTask> record : records) {
workerOperator.processElement(record);
}
@@ -138,14 +138,14 @@ public class AppendOnlySingleTableCompactionWorkerOperatorTest extends TableTest
// write 200 files
List<CommitMessage> commitMessages = writeDataDefault(200, 40);
- List<UnawareAppendCompactionTask> tasks = packTask(commitMessages, 5);
- List<StreamRecord<UnawareAppendCompactionTask>> records =
+ List<AppendCompactTask> tasks = packTask(commitMessages, 5);
+ List<StreamRecord<AppendCompactTask>> records =
tasks.stream().map(StreamRecord::new).collect(Collectors.toList());
assertThat(tasks.size()).isEqualTo(8);
workerOperator.open();
- for (StreamRecord<UnawareAppendCompactionTask> record : records) {
+ for (StreamRecord<AppendCompactTask> record : records) {
workerOperator.processElement(record);
}
@@ -216,9 +216,8 @@ public class AppendOnlySingleTableCompactionWorkerOperatorTest extends TableTest
return GenericRow.of(RANDOM.nextInt(), RANDOM.nextLong(), randomString());
}
- public static List<UnawareAppendCompactionTask> packTask(
- List<CommitMessage> messages, int fileSize) {
- List<UnawareAppendCompactionTask> result = new ArrayList<>();
+ public static List<AppendCompactTask> packTask(List<CommitMessage> messages, int fileSize) {
+ List<AppendCompactTask> result = new ArrayList<>();
List<DataFileMeta> metas =
messages.stream()
.flatMap(
@@ -229,12 +228,10 @@ public class AppendOnlySingleTableCompactionWorkerOperatorTest extends TableTest
for (int i = 0; i < metas.size(); i += fileSize) {
if (i < metas.size() - fileSize) {
result.add(
- new UnawareAppendCompactionTask(
- BinaryRow.EMPTY_ROW, metas.subList(i, i + fileSize)));
+ new AppendCompactTask(BinaryRow.EMPTY_ROW, metas.subList(i, i + fileSize)));
} else {
result.add(
- new UnawareAppendCompactionTask(
- BinaryRow.EMPTY_ROW, metas.subList(i, metas.size())));
+ new AppendCompactTask(BinaryRow.EMPTY_ROW, metas.subList(i, metas.size())));
}
}
return result;
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CompactionTaskSimpleSerializerTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CompactionTaskSimpleSerializerTest.java
index 9f6e8159a1..0c90b54dee 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CompactionTaskSimpleSerializerTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CompactionTaskSimpleSerializerTest.java
@@ -18,11 +18,11 @@
package org.apache.paimon.flink.sink;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileSource;
-import org.apache.paimon.table.sink.CompactionTaskSerializer;
+import org.apache.paimon.table.sink.AppendCompactTaskSerializer;
import org.junit.jupiter.api.Test;
@@ -38,8 +38,8 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link CompactionTaskSimpleSerializer}. */
public class CompactionTaskSimpleSerializerTest {
- private final CompactionTaskSerializer compactionTaskSerializer =
- new CompactionTaskSerializer();
+ private final AppendCompactTaskSerializer compactionTaskSerializer =
+ new AppendCompactTaskSerializer();
private final CompactionTaskSimpleSerializer serializer =
new CompactionTaskSimpleSerializer(compactionTaskSerializer);
@@ -49,9 +49,8 @@ public class CompactionTaskSimpleSerializerTest {
@Test
public void testSerializer() throws IOException {
- UnawareAppendCompactionTask task1 =
- new UnawareAppendCompactionTask(partition, newFiles(20));
- UnawareAppendCompactionTask task2 = serializer.deserialize(2, serializer.serialize(task1));
+ AppendCompactTask task1 = new AppendCompactTask(partition, newFiles(20));
+ AppendCompactTask task2 = serializer.deserialize(2, serializer.serialize(task1));
assertThat(task1).isEqualTo(task2);
}
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 50cc5ad986..196f682cd5 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
@@ -21,12 +21,12 @@ package org.apache.paimon.spark.procedure;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.CoreOptions.OrderType;
import org.apache.paimon.annotation.VisibleForTesting;
-import org.apache.paimon.append.UnawareAppendCompactionTask;
-import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.append.AppendCompactCoordinator;
+import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.manifest.PartitionEntry;
-import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
+import org.apache.paimon.operation.BaseAppendFileStoreWrite;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.spark.PaimonSplitScan;
import org.apache.paimon.spark.SparkUtils;
@@ -36,12 +36,12 @@ import org.apache.paimon.spark.commands.PaimonSparkWriter;
import org.apache.paimon.spark.sort.TableSorter;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.AppendCompactTaskSerializer;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.CommitMessage;
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;
@@ -361,10 +361,9 @@ public class CompactProcedure extends BaseProcedure {
@Nullable Predicate filter,
@Nullable Duration partitionIdleTime,
JavaSparkContext javaSparkContext) {
- List<UnawareAppendCompactionTask> compactionTasks;
+ List<AppendCompactTask> compactionTasks;
try {
- compactionTasks =
- new UnawareAppendTableCompactionCoordinator(table, false, filter).run();
+ compactionTasks = new AppendCompactCoordinator(table, false, filter).run();
} catch (EndOfScanException e) {
compactionTasks = new ArrayList<>();
}
@@ -391,10 +390,10 @@ public class CompactProcedure extends BaseProcedure {
return;
}
- CompactionTaskSerializer serializer = new CompactionTaskSerializer();
+ AppendCompactTaskSerializer serializer = new AppendCompactTaskSerializer();
List<byte[]> serializedTasks = new ArrayList<>();
try {
- for (UnawareAppendCompactionTask compactionTask : compactionTasks) {
+ for (AppendCompactTask compactionTask : compactionTasks) {
serializedTasks.add(serializer.serialize(compactionTask));
}
} catch (IOException e) {
@@ -409,17 +408,17 @@ public class CompactProcedure extends BaseProcedure {
.mapPartitions(
(FlatMapFunction<Iterator<byte[]>, byte[]>)
taskIterator -> {
- AppendOnlyFileStoreWrite write =
- (AppendOnlyFileStoreWrite)
+ BaseAppendFileStoreWrite write =
+ (BaseAppendFileStoreWrite)
table.store().newWrite(commitUser);
- CompactionTaskSerializer ser =
- new CompactionTaskSerializer();
+ AppendCompactTaskSerializer ser =
+ new AppendCompactTaskSerializer();
List<byte[]> messages = new ArrayList<>();
try {
CommitMessageSerializer messageSer =
new CommitMessageSerializer();
while (taskIterator.hasNext()) {
- UnawareAppendCompactionTask task =
+ AppendCompactTask task =
ser.deserialize(
ser.getVersion(),
taskIterator.next());
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
index 2366ff9e02..e134548667 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
@@ -24,7 +24,7 @@ import org.apache.paimon.codegen.CodeGenUtils
import org.apache.paimon.crosspartition.{IndexBootstrap, KeyPartOrRow}
import org.apache.paimon.data.serializer.InternalSerializers
import org.apache.paimon.deletionvectors.DeletionVector
-import org.apache.paimon.deletionvectors.append.AppendDeletionFileMaintainer
+import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer
import org.apache.paimon.index.{BucketAssigner, SimpleHashBucketAssigner}
import org.apache.paimon.io.{CompactIncrement, DataIncrement, IndexIncrement}
import org.apache.paimon.manifest.{FileKind, IndexManifestEntry}
@@ -280,15 +280,15 @@ case class PaimonSparkWriter(table: FileStoreTable) {
.mapGroups {
(_, iter: Iterator[SparkDeletionVectors]) =>
val indexHandler = table.store().newIndexFileHandler()
- var dvIndexFileMaintainer: AppendDeletionFileMaintainer = null
+ var dvIndexFileMaintainer: BaseAppendDeleteFileMaintainer = null
while (iter.hasNext) {
val sdv: SparkDeletionVectors = iter.next()
if (dvIndexFileMaintainer == null) {
val partition = SerializationUtils.deserializeBinaryRow(sdv.partition)
dvIndexFileMaintainer = if (bucketMode == BUCKET_UNAWARE) {
- AppendDeletionFileMaintainer.forUnawareAppend(indexHandler, snapshot, partition)
+ BaseAppendDeleteFileMaintainer.forUnawareAppend(indexHandler, snapshot, partition)
} else {
- AppendDeletionFileMaintainer.forBucketedAppend(
+ BaseAppendDeleteFileMaintainer.forBucketedAppend(
indexHandler,
snapshot,
partition,
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
index 9bea013589..4da80dc2da 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
@@ -772,7 +772,7 @@ abstract class CompactProcedureTestBase extends PaimonSparkTestBase with StreamT
spark.sql(
"CALL sys.compact(table => 'T', options => 'source.split.open-file-cost=3200M, compaction.min.file-num=2')")
- assertResult(Seq(2, 1))(taskBuffer)
+ assertResult(Seq(2, 3))(taskBuffer)
} finally {
spark.sparkContext.removeSparkListener(listener)
}