You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2025/06/16 07:20:35 UTC
(paimon) branch release-1.2 updated (f8d89e67cb -> b99bf0b623)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a change to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
from f8d89e67cb [spark] Upgrade spark version to 4.0.0 (#5711)
new 2aabf52d0f [hive] Hive catalog configuration to get owner for table creation (#5668)
new 95322af748 [flink] Add 'scan.dedicated-split-generation' to Flink Batch Source (#5715)
new 0033a73721 [hotfix] Minor refactor for Flink Dedicated Split Gen Source
new 4efd31c85d [test][spark] Fix the unstable testHiveCatalogOptions (#5738)
new 958e85bac2 [flink] Add support for AvroType enum for debezium format (#5742)
new dc779c1512 [rest] Adjust writeTimeout of HttpClient to 3 minutes
new 79a3ab2b58 [flink] Postpone mode should support 'partition.sink-strategy' (#5743)
new 1e00d28699 [doc] fix docs gcs in filesystem.md (#5747)
new 68e2692f48 [core] Limit max task number in ManifestReadThreadPool (#5737)
new 08050cf2bc [fs] Add a requiredOptions method to COSNLoader (#5750)
new a44ae96578 [core] Support dedicated full compact to external paths (#5674)
new 7306a0ac4b [core] Simplify force rewrite files in Compact Task (#5751)
new 9b4791014a [flink] Make PostponeBucketSink no state and no intended failure (#5746)
new c2bdec5f6f [core] Introduce auto-buffer-spill for postpone bucket write (#5744)
new b99bf0b623 [flink] Introduce 'memory' type for 'lookup.cache' to in-memory cache (#5748)
The 15 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/content/flink/sql-query.md | 17 ++
docs/content/maintenance/filesystems.md | 5 +-
.../shortcodes/generated/core_configuration.html | 8 +-
.../generated/flink_connector_configuration.html | 20 +-
.../main/java/org/apache/paimon/CoreOptions.java | 13 +-
.../java/org/apache/paimon/rest/HttpClient.java | 1 +
.../org/apache/paimon/utils/ThreadPoolUtils.java | 16 +-
.../paimon/utils/SemaphoredDelegatingExecutor.java | 184 ++++++++++++++++
.../java/org/apache/paimon/KeyValueFileStore.java | 2 +
.../org/apache/paimon/append/AppendOnlyWriter.java | 216 +++----------------
.../append/BucketedAppendCompactManager.java | 19 +-
.../org/apache/paimon/compact/CompactUnit.java | 43 ++--
.../paimon/crosspartition/GlobalIndexAssigner.java | 16 +-
.../java/org/apache/paimon/lookup/BulkLoader.java | 101 +--------
.../LevelSortedRun.java => lookup/ByteArray.java} | 39 ++--
.../org/apache/paimon/lookup/ListBulkLoader.java | 11 +-
.../java/org/apache/paimon/lookup/ListState.java | 14 +-
.../java/org/apache/paimon/lookup/SetState.java | 17 +-
.../main/java/org/apache/paimon/lookup/State.java | 15 +-
.../org/apache/paimon/lookup/StateFactory.java | 31 ++-
.../org/apache/paimon/lookup/ValueBulkLoader.java | 10 +-
.../java/org/apache/paimon/lookup/ValueState.java | 14 +-
.../paimon/lookup/memory/InMemoryListState.java | 77 +++++++
.../paimon/lookup/memory/InMemorySetState.java | 69 ++++++
.../apache/paimon/lookup/memory/InMemoryState.java | 64 ++++++
.../paimon/lookup/memory/InMemoryStateFactory.java | 66 ++++++
.../paimon/lookup/memory/InMemoryValueState.java | 76 +++++++
.../RocksDBBulkLoader.java} | 33 ++-
.../lookup/{ => rocksdb}/RocksDBListState.java | 17 +-
.../lookup/{ => rocksdb}/RocksDBOptions.java | 2 +-
.../lookup/{ => rocksdb}/RocksDBSetState.java | 10 +-
.../paimon/lookup/{ => rocksdb}/RocksDBState.java | 52 ++---
.../lookup/{ => rocksdb}/RocksDBStateFactory.java | 14 +-
.../lookup/{ => rocksdb}/RocksDBValueState.java | 21 +-
.../paimon/mergetree/compact/CompactStrategy.java | 39 ++--
.../mergetree/compact/FileRewriteCompactTask.java | 67 ++++++
.../mergetree/compact/MergeTreeCompactManager.java | 47 +++--
.../mergetree/compact/MergeTreeCompactTask.java | 56 ++---
.../paimon/operation/AbstractFileStoreWrite.java | 2 +-
.../operation/BucketedAppendFileStoreWrite.java | 1 +
.../paimon/operation/KeyValueFileStoreWrite.java | 3 +-
.../postpone/PostponeBucketFileStoreWrite.java | 71 ++++++-
.../paimon/postpone/PostponeBucketWriter.java | 141 +++++++++++--
.../paimon/utils/ManifestReadThreadPool.java | 27 ++-
.../java/org/apache/paimon/utils/SinkWriter.java | 234 +++++++++++++++++++++
.../apache/paimon/append/AppendOnlyWriterTest.java | 1 +
.../append/BucketedAppendCompactManagerTest.java | 1 +
.../apache/paimon/append/FullCompactTaskTest.java | 2 +-
.../apache/paimon/format/FileFormatSuffixTest.java | 2 +-
.../apache/paimon/mergetree/MergeTreeTestBase.java | 6 +-
.../compact/MergeTreeCompactManagerTest.java | 9 +-
.../paimon/table/PrimaryKeySimpleTableTest.java | 38 ++++
.../configuration/ConfigOptionsDocGenerator.java | 2 +-
.../java/org/apache/paimon/cosn/COSNLoader.java | 12 ++
.../cdc/format/debezium/DebeziumSchemaUtils.java | 1 +
.../paimon/flink/sink/cdc/CdcAppendTableSink.java | 7 +
.../sink/cdc/CdcAppendTableWriteOperator.java | 13 ++
.../flink/sink/cdc/FlinkCdcMultiTableSink.java | 2 +-
.../format/debezium/DebeziumSchemaUtilsTest.java | 153 ++++++++++++++
.../apache/paimon/flink/FlinkConnectorOptions.java | 20 +-
.../flink/lookup/FileStoreLookupFunction.java | 4 +-
.../paimon/flink/lookup/FullCacheLookupTable.java | 38 +++-
.../flink/lookup/NoPrimaryKeyLookupTable.java | 19 +-
.../paimon/flink/lookup/PrimaryKeyLookupTable.java | 11 +-
.../flink/lookup/SecondaryIndexLookupTable.java | 4 +-
.../postpone/PostponeBucketCompactSplitSource.java | 2 +-
.../apache/paimon/flink/sink/AppendTableSink.java | 2 +
.../apache/paimon/flink/sink/FlinkSinkBuilder.java | 13 +-
.../apache/paimon/flink/sink/FlinkWriteSink.java | 50 +++++
...ixedBucketSink.java => PostponeBucketSink.java} | 20 +-
.../RestoreAndFailCommittableStateManager.java | 56 +----
...er.java => RestoreCommittableStateManager.java} | 28 +--
.../paimon/flink/sink/RowAppendTableSink.java | 40 +---
.../sink/RowDataHashPartitionChannelComputer.java | 10 +-
.../flink/source/ContinuousFileStoreSource.java | 2 +-
.../paimon/flink/source/FlinkSourceBuilder.java | 17 +-
.../align/AlignedContinuousFileStoreSource.java | 2 +-
.../flink/source/operator/MonitorSource.java | 38 +++-
.../paimon/flink/source/operator/ReadOperator.java | 25 ++-
.../apache/paimon/flink/utils/TableScanUtils.java | 16 +-
.../apache/paimon/flink/BatchFileStoreITCase.java | 38 ++++
.../org/apache/paimon/flink/LookupJoinITCase.java | 6 +-
.../paimon/flink/PostponeBucketTableITCase.java | 11 +-
.../flink/lookup/FileStoreLookupFunctionTest.java | 2 +-
.../paimon/flink/lookup/LookupTableTest.java | 90 +++++---
.../paimon/flink/lookup/RocksDBListStateTest.java | 4 +-
.../flink/procedure/CompactProcedureITCase.java | 67 ++++++
.../sink/BatchWriteGeneratorTagOperatorTest.java | 4 +-
.../paimon/flink/sink/CommitterOperatorTest.java | 2 +-
.../paimon/flink/sink/StoreMultiCommitterTest.java | 2 +-
.../flink/source/operator/OperatorSourceTest.java | 60 +++++-
.../java/org/apache/paimon/hive/HiveCatalog.java | 20 +-
.../paimon/spark/SparkCatalogWithHiveTest.java | 1 +
.../spark/SparkGenericCatalogWithHiveTest.java | 30 ++-
.../catalog/functions/BucketFunctionTest.java | 6 +
95 files changed, 2166 insertions(+), 844 deletions(-)
create mode 100644 paimon-common/src/main/java/org/apache/paimon/utils/SemaphoredDelegatingExecutor.java
copy paimon-core/src/main/java/org/apache/paimon/{mergetree/LevelSortedRun.java => lookup/ByteArray.java} (61%)
copy paimon-api/src/main/java/org/apache/paimon/rest/responses/PagedResponse.java => paimon-core/src/main/java/org/apache/paimon/lookup/ListBulkLoader.java (78%)
copy paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreWriter.java => paimon-core/src/main/java/org/apache/paimon/lookup/ListState.java (76%)
copy paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreWriter.java => paimon-core/src/main/java/org/apache/paimon/lookup/SetState.java (71%)
copy paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreReader.java => paimon-core/src/main/java/org/apache/paimon/lookup/State.java (76%)
copy paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreReader.java => paimon-core/src/main/java/org/apache/paimon/lookup/StateFactory.java (52%)
copy paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFTokenLoader.java => paimon-core/src/main/java/org/apache/paimon/lookup/ValueBulkLoader.java (78%)
copy paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreReader.java => paimon-core/src/main/java/org/apache/paimon/lookup/ValueState.java (74%)
create mode 100644 paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryListState.java
create mode 100644 paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemorySetState.java
create mode 100644 paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryState.java
create mode 100644 paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryStateFactory.java
create mode 100644 paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryValueState.java
copy paimon-core/src/main/java/org/apache/paimon/lookup/{BulkLoader.java => rocksdb/RocksDBBulkLoader.java} (82%)
rename paimon-core/src/main/java/org/apache/paimon/lookup/{ => rocksdb}/RocksDBListState.java (87%)
rename paimon-core/src/main/java/org/apache/paimon/lookup/{ => rocksdb}/RocksDBOptions.java (99%)
rename paimon-core/src/main/java/org/apache/paimon/lookup/{ => rocksdb}/RocksDBSetState.java (95%)
rename paimon-core/src/main/java/org/apache/paimon/lookup/{ => rocksdb}/RocksDBState.java (82%)
rename paimon-core/src/main/java/org/apache/paimon/lookup/{ => rocksdb}/RocksDBStateFactory.java (94%)
rename paimon-core/src/main/java/org/apache/paimon/lookup/{ => rocksdb}/RocksDBValueState.java (86%)
create mode 100644 paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FileRewriteCompactTask.java
create mode 100644 paimon-core/src/main/java/org/apache/paimon/utils/SinkWriter.java
create mode 100644 paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtilsTest.java
copy paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/{FixedBucketSink.java => PostponeBucketSink.java} (73%)
copy paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/{RestoreAndFailCommittableStateManager.java => RestoreCommittableStateManager.java} (75%)
(paimon) 13/15: [flink] Make PostponeBucketSink no state and no intended failure (#5746)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 9b4791014a3fe4dcaeb714e5e62c945f8173ef1d
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Jun 16 10:20:11 2025 +0800
[flink] Make PostponeBucketSink no state and no intended failure (#5746)
---
.../paimon/flink/sink/cdc/CdcAppendTableSink.java | 7 +++
.../sink/cdc/CdcAppendTableWriteOperator.java | 13 +++++
.../flink/sink/cdc/FlinkCdcMultiTableSink.java | 2 +-
.../apache/paimon/flink/sink/AppendTableSink.java | 2 +
.../apache/paimon/flink/sink/FlinkSinkBuilder.java | 2 +-
.../apache/paimon/flink/sink/FlinkWriteSink.java | 50 +++++++++++++++++++
.../paimon/flink/sink/PostponeBucketSink.java} | 34 ++++++-------
.../RestoreAndFailCommittableStateManager.java | 56 ++--------------------
...er.java => RestoreCommittableStateManager.java} | 28 ++---------
.../paimon/flink/sink/RowAppendTableSink.java | 40 ++++------------
.../sink/BatchWriteGeneratorTagOperatorTest.java | 4 +-
.../paimon/flink/sink/CommitterOperatorTest.java | 2 +-
.../paimon/flink/sink/StoreMultiCommitterTest.java | 2 +-
13 files changed, 112 insertions(+), 130 deletions(-)
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableSink.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableSink.java
index d43690e5b2..7a5b2a0f11 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableSink.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableSink.java
@@ -19,8 +19,10 @@
package org.apache.paimon.flink.sink.cdc;
import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.flink.sink.CommittableStateManager;
import org.apache.paimon.flink.sink.FlinkWriteSink;
import org.apache.paimon.flink.sink.StoreSinkWrite;
+import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.streaming.api.datastream.DataStream;
@@ -52,4 +54,9 @@ public class CdcAppendTableSink extends FlinkWriteSink<CdcRecord> {
DataStream<CdcRecord> input, String initialCommitUser, @Nullable Integer parallelism) {
return super.doWrite(input, initialCommitUser, this.parallelism);
}
+
+ @Override
+ protected CommittableStateManager<ManifestCommittable> createCommittableStateManager() {
+ return createRestoreOnlyCommittableStateManager(table);
+ }
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableWriteOperator.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableWriteOperator.java
index 96b9fefcdf..b3e4153727 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableWriteOperator.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableWriteOperator.java
@@ -19,11 +19,14 @@
package org.apache.paimon.flink.sink.cdc;
import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.flink.sink.NoopStoreSinkWriteState;
import org.apache.paimon.flink.sink.PrepareCommitOperator;
import org.apache.paimon.flink.sink.StoreSinkWrite;
+import org.apache.paimon.flink.sink.StoreSinkWriteState;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.RowKind;
+import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
@@ -40,6 +43,16 @@ public class CdcAppendTableWriteOperator extends CdcRecordStoreWriteOperator {
super(parameters, table, storeSinkWriteProvider, initialCommitUser);
}
+ @Override
+ protected StoreSinkWriteState createState(
+ int subtaskId,
+ StateInitializationContext context,
+ StoreSinkWriteState.StateValueFilter stateFilter) {
+ // No conflicts will occur in append only unaware bucket writer, so no state
+ // is needed.
+ return new NoopStoreSinkWriteState(subtaskId, stateFilter);
+ }
+
@Override
public void processElement(StreamRecord<CdcRecord> element) throws Exception {
// only accepts INSERT record
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
index 0cd2638179..a67e74ef55 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
@@ -177,6 +177,6 @@ public class FlinkCdcMultiTableSink implements Serializable {
protected CommittableStateManager<WrappedManifestCommittable> createCommittableStateManager() {
return new RestoreAndFailCommittableStateManager<>(
- WrappedManifestCommittableSerializer::new);
+ WrappedManifestCommittableSerializer::new, true);
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java
index 24d0e2db6f..094d61f6ea 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java
@@ -48,6 +48,8 @@ import static org.apache.paimon.flink.utils.ParallelismUtils.setParallelism;
*/
public abstract class AppendTableSink<T> extends FlinkWriteSink<T> {
+ private static final long serialVersionUID = 1L;
+
protected final FileStoreTable table;
protected final LogSinkFunction logSinkFunction;
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 7f211870b9..3484210014 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
@@ -300,7 +300,7 @@ public class FlinkSinkBuilder {
channelComputer = new PostponeBucketChannelComputer(table.schema());
}
DataStream<InternalRow> partitioned = partition(input, channelComputer, parallelism);
- FixedBucketSink sink = new FixedBucketSink(table, overwritePartition, null);
+ PostponeBucketSink sink = new PostponeBucketSink(table, overwritePartition);
return sink.sinkFrom(partitioned);
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkWriteSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkWriteSink.java
index 74f1febd90..8dc2044734 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkWriteSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkWriteSink.java
@@ -18,11 +18,17 @@
package org.apache.paimon.flink.sink;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.manifest.ManifestCommittableSerializer;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
import javax.annotation.Nullable;
import java.util.Map;
@@ -63,4 +69,48 @@ public abstract class FlinkWriteSink<T> extends FlinkSink<T> {
ManifestCommittableSerializer::new,
options.get(PARTITION_MARK_DONE_RECOVER_FROM_STATE));
}
+
+ protected static OneInputStreamOperatorFactory<InternalRow, Committable>
+ createNoStateRowWriteOperatorFactory(
+ FileStoreTable table,
+ LogSinkFunction logSinkFunction,
+ StoreSinkWrite.Provider writeProvider,
+ String commitUser) {
+ return new RowDataStoreWriteOperator.Factory(
+ table, logSinkFunction, writeProvider, commitUser) {
+ @Override
+ @SuppressWarnings("unchecked, rawtypes")
+ public StreamOperator createStreamOperator(StreamOperatorParameters parameters) {
+ return new RowDataStoreWriteOperator(
+ parameters, table, logSinkFunction, writeProvider, commitUser) {
+
+ @Override
+ protected StoreSinkWriteState createState(
+ int subtaskId,
+ StateInitializationContext context,
+ StoreSinkWriteState.StateValueFilter stateFilter) {
+ // No conflicts will occur in append only unaware bucket writer, so no state
+ // is needed.
+ return new NoopStoreSinkWriteState(subtaskId, stateFilter);
+ }
+
+ @Override
+ protected String getCommitUser(StateInitializationContext context)
+ throws Exception {
+ // No conflicts will occur in append only unaware bucket writer, so
+ // commitUser does not matter.
+ return commitUser;
+ }
+ };
+ }
+ };
+ }
+
+ protected static CommittableStateManager<ManifestCommittable>
+ createRestoreOnlyCommittableStateManager(FileStoreTable table) {
+ Options options = table.coreOptions().toConfiguration();
+ return new RestoreCommittableStateManager<>(
+ ManifestCommittableSerializer::new,
+ options.get(PARTITION_MARK_DONE_RECOVER_FROM_STATE));
+ }
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeBucketSink.java
similarity index 51%
copy from paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableSink.java
copy to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeBucketSink.java
index d43690e5b2..1ab6f398c5 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeBucketSink.java
@@ -16,40 +16,36 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.sink.cdc;
+package org.apache.paimon.flink.sink;
-import org.apache.paimon.flink.sink.Committable;
-import org.apache.paimon.flink.sink.FlinkWriteSink;
-import org.apache.paimon.flink.sink.StoreSinkWrite;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.table.FileStoreTable;
-import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
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 CdcAppendTableSink extends FlinkWriteSink<CdcRecord> {
+import java.util.Map;
+
+/** {@link FlinkSink} for writing records into fixed bucket Paimon table. */
+public class PostponeBucketSink extends FlinkWriteSink<InternalRow> {
- private final Integer parallelism;
+ private static final long serialVersionUID = 1L;
- public CdcAppendTableSink(FileStoreTable table, Integer parallelism) {
- super(table, null);
- this.parallelism = parallelism;
+ public PostponeBucketSink(
+ FileStoreTable table, @Nullable Map<String, String> overwritePartition) {
+ super(table, overwritePartition);
}
@Override
- protected OneInputStreamOperatorFactory<CdcRecord, Committable> createWriteOperatorFactory(
+ protected OneInputStreamOperatorFactory<InternalRow, Committable> createWriteOperatorFactory(
StoreSinkWrite.Provider writeProvider, String commitUser) {
- return new CdcAppendTableWriteOperator.Factory(table, writeProvider, commitUser);
+ return createNoStateRowWriteOperatorFactory(table, null, writeProvider, commitUser);
}
@Override
- public DataStream<Committable> doWrite(
- DataStream<CdcRecord> input, String initialCommitUser, @Nullable Integer parallelism) {
- return super.doWrite(input, initialCommitUser, this.parallelism);
+ protected CommittableStateManager<ManifestCommittable> createCommittableStateManager() {
+ return createRestoreOnlyCommittableStateManager(table);
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RestoreAndFailCommittableStateManager.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RestoreAndFailCommittableStateManager.java
index a9b0922d0b..8556cb6e77 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RestoreAndFailCommittableStateManager.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RestoreAndFailCommittableStateManager.java
@@ -19,18 +19,9 @@
package org.apache.paimon.flink.sink;
import org.apache.paimon.data.serializer.VersionedSerializer;
-import org.apache.paimon.flink.VersionedSerializerWrapper;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.utils.SerializableSupplier;
-import org.apache.flink.api.common.state.ListState;
-import org.apache.flink.api.common.state.ListStateDescriptor;
-import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
-import org.apache.flink.runtime.state.StateInitializationContext;
-import org.apache.flink.runtime.state.StateSnapshotContext;
-import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
-
-import java.util.ArrayList;
import java.util.List;
/**
@@ -44,52 +35,20 @@ import java.util.List;
* store writers.
*/
public class RestoreAndFailCommittableStateManager<GlobalCommitT>
- implements CommittableStateManager<GlobalCommitT> {
+ extends RestoreCommittableStateManager<GlobalCommitT> {
private static final long serialVersionUID = 1L;
- /** The committable's serializer. */
- private final SerializableSupplier<VersionedSerializer<GlobalCommitT>> committableSerializer;
-
- private final boolean partitionMarkDoneRecoverFromState;
-
- /** GlobalCommitT state of this job. Used to filter out previous successful commits. */
- private ListState<GlobalCommitT> streamingCommitterState;
-
- public RestoreAndFailCommittableStateManager(
- SerializableSupplier<VersionedSerializer<GlobalCommitT>> committableSerializer) {
- this(committableSerializer, true);
- }
-
public RestoreAndFailCommittableStateManager(
SerializableSupplier<VersionedSerializer<GlobalCommitT>> committableSerializer,
boolean partitionMarkDoneRecoverFromState) {
- this.committableSerializer = committableSerializer;
- this.partitionMarkDoneRecoverFromState = partitionMarkDoneRecoverFromState;
+ super(committableSerializer, partitionMarkDoneRecoverFromState);
}
@Override
- public void initializeState(
- StateInitializationContext context, Committer<?, GlobalCommitT> committer)
+ protected int recover(List<GlobalCommitT> committables, Committer<?, GlobalCommitT> committer)
throws Exception {
- streamingCommitterState =
- new SimpleVersionedListState<>(
- context.getOperatorStateStore()
- .getListState(
- new ListStateDescriptor<>(
- "streaming_committer_raw_states",
- BytePrimitiveArraySerializer.INSTANCE)),
- new VersionedSerializerWrapper<>(committableSerializer.get()));
- List<GlobalCommitT> restored = new ArrayList<>();
- streamingCommitterState.get().forEach(restored::add);
- streamingCommitterState.clear();
- recover(restored, committer);
- }
-
- private void recover(List<GlobalCommitT> committables, Committer<?, GlobalCommitT> committer)
- throws Exception {
- int numCommitted =
- committer.filterAndCommit(committables, true, partitionMarkDoneRecoverFromState);
+ int numCommitted = super.recover(committables, committer);
if (numCommitted > 0) {
throw new RuntimeException(
"This exception is intentionally thrown "
@@ -97,11 +56,6 @@ public class RestoreAndFailCommittableStateManager<GlobalCommitT>
+ "By restarting the job we hope that "
+ "writers can start writing based on these new commits.");
}
- }
-
- @Override
- public void snapshotState(StateSnapshotContext context, List<GlobalCommitT> committables)
- throws Exception {
- streamingCommitterState.update(committables);
+ return numCommitted;
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RestoreAndFailCommittableStateManager.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RestoreCommittableStateManager.java
similarity index 75%
copy from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RestoreAndFailCommittableStateManager.java
copy to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RestoreCommittableStateManager.java
index a9b0922d0b..b1ed396bdc 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RestoreAndFailCommittableStateManager.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RestoreCommittableStateManager.java
@@ -36,14 +36,9 @@ import java.util.List;
/**
* A {@link CommittableStateManager} which stores uncommitted {@link ManifestCommittable}s in state.
*
- * <p>When the job restarts, these {@link ManifestCommittable}s will be restored and committed, then
- * an intended failure will occur, hoping that after the job restarts, all writers can start writing
- * based on the restored snapshot.
- *
- * <p>Useful for committing snapshots containing records. For example snapshots produced by table
- * store writers.
+ * <p>When the job restarts, these {@link ManifestCommittable}s will be restored and committed.
*/
-public class RestoreAndFailCommittableStateManager<GlobalCommitT>
+public class RestoreCommittableStateManager<GlobalCommitT>
implements CommittableStateManager<GlobalCommitT> {
private static final long serialVersionUID = 1L;
@@ -56,12 +51,7 @@ public class RestoreAndFailCommittableStateManager<GlobalCommitT>
/** GlobalCommitT state of this job. Used to filter out previous successful commits. */
private ListState<GlobalCommitT> streamingCommitterState;
- public RestoreAndFailCommittableStateManager(
- SerializableSupplier<VersionedSerializer<GlobalCommitT>> committableSerializer) {
- this(committableSerializer, true);
- }
-
- public RestoreAndFailCommittableStateManager(
+ public RestoreCommittableStateManager(
SerializableSupplier<VersionedSerializer<GlobalCommitT>> committableSerializer,
boolean partitionMarkDoneRecoverFromState) {
this.committableSerializer = committableSerializer;
@@ -86,17 +76,9 @@ public class RestoreAndFailCommittableStateManager<GlobalCommitT>
recover(restored, committer);
}
- private void recover(List<GlobalCommitT> committables, Committer<?, GlobalCommitT> committer)
+ protected int recover(List<GlobalCommitT> committables, Committer<?, GlobalCommitT> committer)
throws Exception {
- int numCommitted =
- committer.filterAndCommit(committables, true, partitionMarkDoneRecoverFromState);
- if (numCommitted > 0) {
- throw new RuntimeException(
- "This exception is intentionally thrown "
- + "after committing the restored checkpoints. "
- + "By restarting the job we hope that "
- + "writers can start writing based on these new commits.");
- }
+ return committer.filterAndCommit(committables, true, partitionMarkDoneRecoverFromState);
}
@Override
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
index 69a339a411..a58839a841 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
@@ -19,18 +19,18 @@
package org.apache.paimon.flink.sink;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.table.FileStoreTable;
-import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
-import org.apache.flink.streaming.api.operators.StreamOperator;
-import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import java.util.Map;
/** An {@link AppendTableSink} which handles {@link InternalRow}. */
public class RowAppendTableSink extends AppendTableSink<InternalRow> {
+ private static final long serialVersionUID = 1L;
+
public RowAppendTableSink(
FileStoreTable table,
Map<String, String> overwritePartitions,
@@ -42,34 +42,12 @@ public class RowAppendTableSink extends AppendTableSink<InternalRow> {
@Override
protected OneInputStreamOperatorFactory<InternalRow, Committable> createWriteOperatorFactory(
StoreSinkWrite.Provider writeProvider, String commitUser) {
- return new RowDataStoreWriteOperator.Factory(
- table, logSinkFunction, writeProvider, commitUser) {
- @Override
- @SuppressWarnings("unchecked, rawtypes")
- public StreamOperator createStreamOperator(StreamOperatorParameters parameters) {
- return new RowDataStoreWriteOperator(
- parameters, table, logSinkFunction, writeProvider, commitUser) {
-
- @Override
- protected StoreSinkWriteState createState(
- int subtaskId,
- StateInitializationContext context,
- StoreSinkWriteState.StateValueFilter stateFilter)
- throws Exception {
- // No conflicts will occur in append only unaware bucket writer, so no state
- // is needed.
- return new NoopStoreSinkWriteState(subtaskId, stateFilter);
- }
+ return createNoStateRowWriteOperatorFactory(
+ table, logSinkFunction, writeProvider, commitUser);
+ }
- @Override
- protected String getCommitUser(StateInitializationContext context)
- throws Exception {
- // No conflicts will occur in append only unaware bucket writer, so
- // commitUser does not matter.
- return commitUser;
- }
- };
- }
- };
+ @Override
+ protected CommittableStateManager<ManifestCommittable> createCommittableStateManager() {
+ return createRestoreOnlyCommittableStateManager(table);
}
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/BatchWriteGeneratorTagOperatorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/BatchWriteGeneratorTagOperatorTest.java
index 32ee2f42af..1bc68477a3 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/BatchWriteGeneratorTagOperatorTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/BatchWriteGeneratorTagOperatorTest.java
@@ -67,7 +67,7 @@ public class BatchWriteGeneratorTagOperatorTest extends CommitterOperatorTest {
table,
initialCommitUser,
new RestoreAndFailCommittableStateManager<>(
- ManifestCommittableSerializer::new));
+ ManifestCommittableSerializer::new, true));
OneInputStreamOperator<Committable, Committable> committerOperator =
committerOperatorFactory.createStreamOperator(
@@ -143,7 +143,7 @@ public class BatchWriteGeneratorTagOperatorTest extends CommitterOperatorTest {
table,
initialCommitUser,
new RestoreAndFailCommittableStateManager<>(
- ManifestCommittableSerializer::new));
+ ManifestCommittableSerializer::new, true));
OneInputStreamOperator<Committable, Committable> committerOperator =
committerOperatorFactory.createStreamOperator(
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
index 4ad1dff9aa..220af9a73b 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
@@ -665,7 +665,7 @@ public class CommitterOperatorTest extends CommitterOperatorTestBase {
table,
null,
new RestoreAndFailCommittableStateManager<>(
- ManifestCommittableSerializer::new));
+ ManifestCommittableSerializer::new, true));
OneInputStreamOperatorTestHarness<Committable, Committable> testHarness =
createTestHarness(operatorFactory);
testHarness.open();
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreMultiCommitterTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreMultiCommitterTest.java
index a61a379bde..041a692d9e 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreMultiCommitterTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreMultiCommitterTest.java
@@ -652,7 +652,7 @@ class StoreMultiCommitterTest {
initialCommitUser,
context -> new StoreMultiCommitter(catalogLoader, context),
new RestoreAndFailCommittableStateManager<>(
- WrappedManifestCommittableSerializer::new));
+ WrappedManifestCommittableSerializer::new, true));
return createTestHarness(operator);
}
(paimon) 03/15: [hotfix] Minor refactor for Flink Dedicated Split Gen Source
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 0033a7372171c5eeb6d08cbb533f720fcb78d69e
Author: JingsongLi <ji...@gmail.com>
AuthorDate: Thu Jun 12 15:48:33 2025 +0800
[hotfix] Minor refactor for Flink Dedicated Split Gen Source
---
docs/content/append-table/query-performance.md | 11 -----------
docs/content/flink/sql-query.md | 17 +++++++++++++++++
docs/content/primary-key-table/query-performance.md | 12 ------------
.../apache/paimon/flink/source/FlinkSourceBuilder.java | 7 ++++---
4 files changed, 21 insertions(+), 26 deletions(-)
diff --git a/docs/content/append-table/query-performance.md b/docs/content/append-table/query-performance.md
index d488e4e999..e2128bbd89 100644
--- a/docs/content/append-table/query-performance.md
+++ b/docs/content/append-table/query-performance.md
@@ -75,14 +75,3 @@ we use the procedure, you should config appropriate configurations in target tab
`file-index.<filter-type>.columns` to the table.
How to invoke: see [flink procedures]({{< ref "flink/procedures#procedures" >}})
-
-## Dedicated Split Generation
-When Paimon table snapshots contain large amount of source splits, Flink jobs reading from this table might endure long initialization time or even OOM in JobManagers. In this case, you can configure `'scan.dedicated-split-generation' = 'true'` to avoid such problem. This option would enable executing the source split generation process in a dedicated subtask that runs on TaskManager, instead of in the source coordinator on the JobManager.
-
-Note that this feature could have some side effects on your Flink jobs. For example:
-
-1. It will change the DAG of the flink job, thus breaking checkpoint compatibility if enabled on an existing job.
-2. It may lead to the Flink AdaptiveBatchScheduler inferring a small parallelism for the source reader operator. you can configure `scan.infer-parallelism` to avoid this possible drawback.
-3. The failover strategy of the Flink job would be forced into global failover instead of regional failover, given that the dedicated source split generation task would be connected to all downstream subtasks.
-
-So please make sure these side effects are acceptable to you before enabling it.
diff --git a/docs/content/flink/sql-query.md b/docs/content/flink/sql-query.md
index 85940a36e8..c974527358 100644
--- a/docs/content/flink/sql-query.md
+++ b/docs/content/flink/sql-query.md
@@ -288,3 +288,20 @@ SELECT * FROM orders WHERE order_id=29495;
SELECT * FROM orders WHERE catalog_id=1025 OR order_id=29495;
```
+
+## Dedicated Split Generation
+
+When Paimon table snapshots contain large amount of source splits, Flink jobs reading from this table might endure long
+initialization time or even OOM in JobManagers. In this case, you can configure `'scan.dedicated-split-generation' = 'true'`
+to avoid such problem. This option would enable executing the source split generation process in a dedicated subtask
+that runs on TaskManager, instead of in the source coordinator on the JobManager.
+
+Note that this feature could have some side effects on your Flink jobs. For example:
+
+1. It will change the DAG of the flink job, thus breaking checkpoint compatibility if enabled on an existing job.
+2. It may lead to the Flink AdaptiveBatchScheduler inferring a small parallelism for the source reader operator. you can
+ configure `scan.infer-parallelism` to avoid this possible drawback.
+3. The failover strategy of the Flink job would be forced into global failover instead of regional failover, given that
+ the dedicated source split generation task would be connected to all downstream subtasks.
+
+So please make sure these side effects are acceptable to you before enabling it.
diff --git a/docs/content/primary-key-table/query-performance.md b/docs/content/primary-key-table/query-performance.md
index ed0e357fc6..d27c4868fe 100644
--- a/docs/content/primary-key-table/query-performance.md
+++ b/docs/content/primary-key-table/query-performance.md
@@ -73,15 +73,3 @@ we use the procedure, you should config appropriate configurations in target tab
`file-index.<filter-type>.columns` to the table.
How to invoke: see [flink procedures]({{< ref "flink/procedures#procedures" >}})
-
-## Dedicated Split Generation
-When Paimon table snapshots contain large amount of source splits, Flink jobs reading from this table might endure long initialization time or even OOM in JobManagers. In this case, you can configure `'scan.dedicated-split-generation' = 'true'` to avoid such problem. This option would enable executing the source split generation process in a dedicated subtask that runs on TaskManager, instead of in the source coordinator on the JobManager.
-
-Note that this feature could have some side effects on your Flink jobs. For example:
-
-1. It will change the DAG of the flink job, thus breaking checkpoint compatibility if enabled on an existing job.
-2. It may lead to the Flink AdaptiveBatchScheduler inferring a small parallelism for the source reader operator. you can configure `scan.infer-parallelism` to avoid this possible drawback.
-3. The failover strategy of the Flink job would be forced into global failover instead of regional failover, given that the dedicated source split generation task would be connected to all downstream subtasks.
-
-So please make sure these side effects are acceptable to you before enabling it.
-
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
index ac776b7cd3..44b5790472 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
@@ -78,6 +78,7 @@ import static org.apache.paimon.utils.Preconditions.checkState;
* @since 0.8
*/
public class FlinkSourceBuilder {
+
private static final String SOURCE_NAME = "Source";
private final Table table;
@@ -295,7 +296,7 @@ public class FlinkSourceBuilder {
if (sourceBounded) {
if (conf.get(FlinkConnectorOptions.SCAN_DEDICATED_SPLIT_GENERATION)) {
- return buildContinuousStreamOperator(true);
+ return buildDedicatedSplitGenSource(true);
}
return buildStaticFileSource();
}
@@ -328,14 +329,14 @@ public class FlinkSourceBuilder {
} else if (conf.contains(CoreOptions.CONSUMER_ID)
&& conf.get(CoreOptions.CONSUMER_CONSISTENCY_MODE)
== CoreOptions.ConsumerMode.EXACTLY_ONCE) {
- return buildContinuousStreamOperator(false);
+ return buildDedicatedSplitGenSource(false);
} else {
return buildContinuousFileSource();
}
}
}
- private DataStream<RowData> buildContinuousStreamOperator(boolean isBounded) {
+ private DataStream<RowData> buildDedicatedSplitGenSource(boolean isBounded) {
DataStream<RowData> dataStream;
if (limit != null && !isBounded) {
throw new IllegalArgumentException(
(paimon) 11/15: [core] Support dedicated full compact to external paths (#5674)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit a44ae965784106f293320979c282c9c9661d74fb
Author: Jiao Mingye <35...@users.noreply.github.com>
AuthorDate: Sun Jun 15 09:38:27 2025 +0800
[core] Support dedicated full compact to external paths (#5674)
---
.../shortcodes/generated/core_configuration.html | 6 ++
.../main/java/org/apache/paimon/CoreOptions.java | 11 ++++
.../append/BucketedAppendCompactManager.java | 19 ++++--
.../paimon/mergetree/compact/CompactStrategy.java | 8 ++-
.../mergetree/compact/MergeTreeCompactManager.java | 14 ++++-
.../mergetree/compact/MergeTreeCompactTask.java | 19 +++---
.../operation/BucketedAppendFileStoreWrite.java | 1 +
.../paimon/operation/KeyValueFileStoreWrite.java | 3 +-
.../apache/paimon/append/AppendOnlyWriterTest.java | 1 +
.../append/BucketedAppendCompactManagerTest.java | 1 +
.../apache/paimon/append/FullCompactTaskTest.java | 2 +-
.../apache/paimon/format/FileFormatSuffixTest.java | 2 +-
.../apache/paimon/mergetree/MergeTreeTestBase.java | 6 +-
.../compact/MergeTreeCompactManagerTest.java | 9 ++-
.../flink/procedure/CompactProcedureITCase.java | 67 ++++++++++++++++++++++
15 files changed, 144 insertions(+), 25 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
index 8cd61f61d6..6bee1104a1 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -182,6 +182,12 @@ under the License.
<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-compact-all-files</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to force pick all files for a full compaction. Usually seen in a compaction task to external paths.</td>
+ </tr>
<tr>
<td><h5>compaction.force-up-level-0</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 1df7949acc..47967be549 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -151,6 +151,13 @@ public class CoreOptions implements Serializable {
+ ExternalPathStrategy.SPECIFIC_FS
+ ", should be the prefix scheme of the external path, now supported are s3 and oss.");
+ public static final ConfigOption<Boolean> COMPACTION_FORCE_COMPACT_ALL_FILES =
+ key("compaction.force-compact-all-files")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to force pick all files for a full compaction. Usually seen in a compaction task to external paths.");
+
@ExcludeFromDocumentation("Internal use only")
public static final ConfigOption<String> PATH =
key("path")
@@ -2466,6 +2473,10 @@ public class CoreOptions implements Serializable {
return options.get(DATA_FILE_EXTERNAL_PATHS_SPECIFIC_FS);
}
+ public Boolean forceCompactAllFiles() {
+ return options.get(COMPACTION_FORCE_COMPACT_ALL_FILES);
+ }
+
public String partitionTimestampFormatter() {
return options.get(PARTITION_TIMESTAMP_FORMATTER);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java b/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java
index d5c47c3b88..520fd8edb3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java
@@ -60,6 +60,7 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
private final PriorityQueue<DataFileMeta> toCompact;
private final int minFileNum;
private final long targetFileSize;
+ private final boolean forceCompactAllFiles;
private final CompactRewriter rewriter;
private List<DataFileMeta> compacting;
@@ -72,6 +73,7 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
@Nullable DeletionVectorsMaintainer dvMaintainer,
int minFileNum,
long targetFileSize,
+ boolean forceCompactAllFiles,
CompactRewriter rewriter,
@Nullable CompactionMetrics.Reporter metricsReporter) {
this.executor = executor;
@@ -80,6 +82,7 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
this.toCompact.addAll(restored);
this.minFileNum = minFileNum;
this.targetFileSize = targetFileSize;
+ this.forceCompactAllFiles = forceCompactAllFiles;
this.rewriter = rewriter;
this.metricsReporter = metricsReporter;
}
@@ -98,9 +101,10 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
taskFuture == null,
"A compaction task is still running while the user "
+ "forces a new compaction. This is unexpected.");
- // if deletion vector enables, always trigger compaction.
- if (toCompact.isEmpty()
- || (dvMaintainer == null && toCompact.size() < FULL_COMPACT_MIN_FILE)) {
+ // if all files are force picked or deletion vector enables, always trigger compaction.
+ if (!forceCompactAllFiles
+ && (toCompact.isEmpty()
+ || (dvMaintainer == null && toCompact.size() < FULL_COMPACT_MIN_FILE))) {
return;
}
@@ -114,6 +118,7 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
dvMaintainer,
toCompact,
targetFileSize,
+ forceCompactAllFiles,
rewriter,
metricsReporter));
recordCompactionsQueuedRequest();
@@ -238,25 +243,28 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
private final DeletionVectorsMaintainer dvMaintainer;
private final LinkedList<DataFileMeta> toCompact;
private final long targetFileSize;
+ private final boolean forceCompactAllFiles;
private final CompactRewriter rewriter;
public FullCompactTask(
DeletionVectorsMaintainer dvMaintainer,
Collection<DataFileMeta> inputs,
long targetFileSize,
+ boolean forceCompactAllFiles,
CompactRewriter rewriter,
@Nullable CompactionMetrics.Reporter metricsReporter) {
super(metricsReporter);
this.dvMaintainer = dvMaintainer;
this.toCompact = new LinkedList<>(inputs);
this.targetFileSize = targetFileSize;
+ this.forceCompactAllFiles = forceCompactAllFiles;
this.rewriter = rewriter;
}
@Override
protected CompactResult doCompact() throws Exception {
// remove large files
- while (!toCompact.isEmpty()) {
+ while (!forceCompactAllFiles && !toCompact.isEmpty()) {
DataFileMeta file = toCompact.peekFirst();
// the data file with deletion file always need to be compacted.
if (file.fileSize() >= targetFileSize && !hasDeletionFile(file)) {
@@ -281,7 +289,8 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
small++;
}
}
- if (small > big && toCompact.size() >= FULL_COMPACT_MIN_FILE) {
+ if (forceCompactAllFiles
+ || (small > big && toCompact.size() >= FULL_COMPACT_MIN_FILE)) {
return compact(null, toCompact, rewriter);
} else {
return result(emptyList(), emptyList());
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactStrategy.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactStrategy.java
index ec82e9e530..ef56196e0a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactStrategy.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactStrategy.java
@@ -55,7 +55,8 @@ public interface CompactStrategy {
int numLevels,
List<LevelSortedRun> runs,
@Nullable RecordLevelExpire recordLevelExpire,
- @Nullable DeletionVectorsMaintainer dvMaintainer) {
+ @Nullable DeletionVectorsMaintainer dvMaintainer,
+ boolean forceCompactAllFiles) {
int maxLevel = numLevels - 1;
if (runs.isEmpty()) {
// no sorted run, no need to compact
@@ -64,7 +65,10 @@ public interface CompactStrategy {
List<DataFileMeta> filesToBeCompacted = new ArrayList<>();
for (DataFileMeta file : runs.get(0).run().files()) {
- if (recordLevelExpire != null && recordLevelExpire.isExpireFile(file)) {
+ if (forceCompactAllFiles) {
+ // add all files when force compacted
+ filesToBeCompacted.add(file);
+ } else if (recordLevelExpire != null && recordLevelExpire.isExpireFile(file)) {
// check record level expire for large files
filesToBeCompacted.add(file);
} else if (dvMaintainer != null
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
index a4852e7346..953a235391 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
@@ -65,6 +65,7 @@ public class MergeTreeCompactManager extends CompactFutureManager {
@Nullable private final DeletionVectorsMaintainer dvMaintainer;
private final boolean lazyGenDeletionFile;
private final boolean needLookup;
+ private final boolean forceCompactAllFiles;
@Nullable private final RecordLevelExpire recordLevelExpire;
@@ -80,7 +81,8 @@ public class MergeTreeCompactManager extends CompactFutureManager {
@Nullable DeletionVectorsMaintainer dvMaintainer,
boolean lazyGenDeletionFile,
boolean needLookup,
- @Nullable RecordLevelExpire recordLevelExpire) {
+ @Nullable RecordLevelExpire recordLevelExpire,
+ boolean forceCompactAllFiles) {
this.executor = executor;
this.levels = levels;
this.strategy = strategy;
@@ -93,6 +95,7 @@ public class MergeTreeCompactManager extends CompactFutureManager {
this.lazyGenDeletionFile = lazyGenDeletionFile;
this.recordLevelExpire = recordLevelExpire;
this.needLookup = needLookup;
+ this.forceCompactAllFiles = forceCompactAllFiles;
MetricUtils.safeCall(this::reportMetrics, LOG);
}
@@ -135,7 +138,11 @@ public class MergeTreeCompactManager extends CompactFutureManager {
}
optionalUnit =
CompactStrategy.pickFullCompaction(
- levels.numberOfLevels(), runs, recordLevelExpire, dvMaintainer);
+ levels.numberOfLevels(),
+ runs,
+ recordLevelExpire,
+ dvMaintainer,
+ forceCompactAllFiles);
} else {
if (taskFuture != null) {
return;
@@ -210,7 +217,8 @@ public class MergeTreeCompactManager extends CompactFutureManager {
metricsReporter,
compactDfSupplier,
dvMaintainer,
- recordLevelExpire);
+ recordLevelExpire,
+ forceCompactAllFiles);
if (LOG.isDebugEnabled()) {
LOG.debug(
"Pick these files (name, level, size) for compaction: {}",
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactTask.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactTask.java
index 0ce9fbb4c2..4aa10e9b79 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactTask.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactTask.java
@@ -55,6 +55,7 @@ public class MergeTreeCompactTask extends CompactTask {
private int upgradeFilesNum;
@Nullable private final RecordLevelExpire recordLevelExpire;
+ private final boolean forceCompactAllFiles;
@Nullable private final DeletionVectorsMaintainer dvMaintainer;
public MergeTreeCompactTask(
@@ -67,7 +68,8 @@ public class MergeTreeCompactTask extends CompactTask {
@Nullable CompactionMetrics.Reporter metricsReporter,
Supplier<CompactDeletionFile> compactDfSupplier,
@Nullable DeletionVectorsMaintainer dvMaintainer,
- @Nullable RecordLevelExpire recordLevelExpire) {
+ @Nullable RecordLevelExpire recordLevelExpire,
+ boolean forceCompactAllFiles) {
super(metricsReporter);
this.minFileSize = minFileSize;
this.rewriter = rewriter;
@@ -78,6 +80,7 @@ public class MergeTreeCompactTask extends CompactTask {
this.dropDelete = dropDelete;
this.maxLevel = maxLevel;
this.recordLevelExpire = recordLevelExpire;
+ this.forceCompactAllFiles = forceCompactAllFiles;
this.upgradeFilesNum = 0;
}
@@ -126,12 +129,14 @@ public class MergeTreeCompactTask extends CompactTask {
private void upgrade(DataFileMeta file, CompactResult toUpdate) throws Exception {
if (file.level() == outputLevel) {
- if (isContainExpiredRecords(file)
+ if (forceCompactAllFiles
+ || isContainExpiredRecords(file)
|| (dvMaintainer != null
&& dvMaintainer.deletionVectorOf(file.fileName()).isPresent())) {
/*
- * 1. if the large file in maxLevel has expired records, we need to rewrite it.
- * 2. if the large file in maxLevel has corresponding deletion vector, we need to rewrite it.
+ * 1. if files are force picked, we need to rewrite all files.
+ * 2. if the large file in maxLevel has expired records, we need to rewrite it.
+ * 3. if the large file in maxLevel has corresponding deletion vector, we need to rewrite it.
*/
rewriteFile(file, toUpdate);
}
@@ -139,9 +144,9 @@ public class MergeTreeCompactTask extends CompactTask {
}
if (outputLevel != maxLevel || file.deleteRowCount().map(d -> d == 0).orElse(false)) {
- if (isContainExpiredRecords(file)) {
- // if the file which could be directly upgraded has expired records, we need to
- // rewrite it
+ if (forceCompactAllFiles || isContainExpiredRecords(file)) {
+ // if all files are force picked, or the file which could be directly upgraded has
+ // expired records, we need to rewrite it
rewriteFile(file, toUpdate);
} else {
CompactResult upgradeResult = rewriter.upgrade(outputLevel, file);
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java
index 36201c7f7f..d8f2d9211b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java
@@ -91,6 +91,7 @@ public class BucketedAppendFileStoreWrite extends BaseAppendFileStoreWrite {
dvMaintainer,
options.compactionMinFileNum(),
options.targetFileSize(false),
+ options.forceCompactAllFiles(),
files -> compactRewrite(partition, bucket, dvFactory, files),
compactionMetrics == null
? null
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
index 9a65e83d06..7c0b36701f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
@@ -290,7 +290,8 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
dvMaintainer,
options.prepareCommitWaitCompaction(),
options.needLookup(),
- recordLevelExpire);
+ recordLevelExpire,
+ options.forceCompactAllFiles());
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
index 2b280220ee..8cb293cd8e 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
@@ -652,6 +652,7 @@ public class AppendOnlyWriterTest {
null,
MIN_FILE_NUM,
targetFileSize,
+ false,
compactBefore -> {
latch.await();
return compactBefore.isEmpty()
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/BucketedAppendCompactManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/append/BucketedAppendCompactManagerTest.java
index cfdf38558f..2f031548a8 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/BucketedAppendCompactManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/BucketedAppendCompactManagerTest.java
@@ -207,6 +207,7 @@ public class BucketedAppendCompactManagerTest {
null,
minFileNum,
targetFileSize,
+ false,
null, // not used
null);
Optional<List<DataFileMeta>> actual = manager.pickCompactBefore();
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/FullCompactTaskTest.java b/paimon-core/src/test/java/org/apache/paimon/append/FullCompactTaskTest.java
index e7c3cce01d..262acd5d4f 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/FullCompactTaskTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/FullCompactTaskTest.java
@@ -123,7 +123,7 @@ public class FullCompactTaskTest {
Collection<DataFileMeta> inputs,
long targetFileSize,
BucketedAppendCompactManager.CompactRewriter rewriter) {
- super(null, inputs, targetFileSize, rewriter, null);
+ super(null, inputs, targetFileSize, false, rewriter, null);
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/format/FileFormatSuffixTest.java b/paimon-core/src/test/java/org/apache/paimon/format/FileFormatSuffixTest.java
index c43b3c20c6..c6761d9c92 100644
--- a/paimon-core/src/test/java/org/apache/paimon/format/FileFormatSuffixTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/format/FileFormatSuffixTest.java
@@ -87,7 +87,7 @@ public class FileFormatSuffixTest extends KeyValueFileReadWriteTest {
SCHEMA,
0,
new BucketedAppendCompactManager(
- null, toCompact, null, 4, 10, null, null), // not used
+ null, toCompact, null, 4, 10, false, null, null), // not used
null,
false,
dataFilePathFactory,
diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java
index 4af6c39800..95daf384e7 100644
--- a/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java
+++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java
@@ -454,7 +454,8 @@ public abstract class MergeTreeTestBase {
null,
false,
options.needLookup(),
- null);
+ null,
+ false);
}
static class MockFailResultCompactionManager extends MergeTreeCompactManager {
@@ -478,7 +479,8 @@ public abstract class MergeTreeTestBase {
null,
false,
false,
- null);
+ null,
+ false);
}
protected CompactResult obtainCompactResult()
diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerTest.java
index 4240555977..4adff94778 100644
--- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerTest.java
@@ -208,7 +208,8 @@ public class MergeTreeCompactManagerTest {
null,
false,
true,
- null);
+ null,
+ false);
MergeTreeCompactManager defaultManager =
new MergeTreeCompactManager(
@@ -223,7 +224,8 @@ public class MergeTreeCompactManagerTest {
null,
false,
false,
- null);
+ null,
+ false);
assertThat(lookupManager.compactNotCompleted()).isTrue();
assertThat(defaultManager.compactNotCompleted()).isFalse();
@@ -259,7 +261,8 @@ public class MergeTreeCompactManagerTest {
null,
false,
false,
- null);
+ null,
+ false);
manager.triggerCompaction(false);
manager.getCompactionResult(true);
List<LevelMinMax> outputs =
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java
index d79d13f026..d58c09d3a8 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java
@@ -20,6 +20,7 @@ package org.apache.paimon.flink.procedure;
import org.apache.paimon.Snapshot;
import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.StreamTableScan;
@@ -35,6 +36,7 @@ import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.List;
+import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -196,6 +198,71 @@ public class CompactProcedureITCase extends CatalogITCaseBase {
}
}
+ @Test
+ public void testForceCompactToExternalPath() throws Exception {
+ // test for pk table
+ String tmpPath = getTempDirPath("external/" + UUID.randomUUID());
+ sql(
+ "CREATE TABLE Tpk ("
+ + " k INT,"
+ + " v INT,"
+ + " hh INT,"
+ + " dt STRING,"
+ + " PRIMARY KEY (k, dt, hh) NOT ENFORCED"
+ + ") PARTITIONED BY (dt, hh) WITH ("
+ + " 'write-only' = 'true',"
+ + " 'bucket' = '1'"
+ + ")");
+ FileStoreTable pkTable = paimonTable("Tpk");
+
+ sql(
+ "INSERT INTO Tpk VALUES (1, 100, 15, '20221208'), (1, 100, 16, '20221208'), (1, 100, 15, '20221209')");
+ sql(
+ "INSERT INTO Tpk VALUES (2, 100, 15, '20221208'), (2, 100, 16, '20221208'), (2, 100, 15, '20221209')");
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ sql(
+ "CALL sys.compact(`table` => 'default.Tpk',"
+ + " options => 'compaction.force-compact-all-files=true,data-file.external-paths=file://%s,data-file.external-paths.strategy=specific-fs,data-file.external-paths.specific-fs=file')",
+ tmpPath);
+ List<DataSplit> splits = pkTable.newSnapshotReader().read().dataSplits();
+ for (DataSplit split : splits) {
+ for (DataFileMeta meta : split.dataFiles()) {
+ assertThat(meta.externalPath().get().startsWith("file:" + tmpPath)).isTrue();
+ }
+ }
+
+ // test for append table
+ tmpPath = getTempDirPath("external/" + UUID.randomUUID());
+ sql(
+ "CREATE TABLE Tap ("
+ + " k INT,"
+ + " v INT,"
+ + " hh INT,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt, hh) WITH ("
+ + " 'write-only' = 'true',"
+ + " 'bucket' = '1',"
+ + " 'bucket-key' = 'k'"
+ + ")");
+ FileStoreTable apTable = paimonTable("Tap");
+
+ sql(
+ "INSERT INTO Tap VALUES (1, 100, 15, '20221208'), (1, 100, 16, '20221208'), (1, 100, 15, '20221209')");
+ sql(
+ "INSERT INTO Tap VALUES (2, 100, 15, '20221208'), (2, 100, 16, '20221208'), (2, 100, 15, '20221209')");
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ sql(
+ "CALL sys.compact(`table` => 'default.Tap',"
+ + " options => 'compaction.force-compact-all-files=true,data-file.external-paths=file://%s,data-file.external-paths.strategy=specific-fs,data-file.external-paths.specific-fs=file')",
+ tmpPath);
+ splits = apTable.newSnapshotReader().read().dataSplits();
+ for (DataSplit split : splits) {
+ for (DataFileMeta meta : split.dataFiles()) {
+ assertThat(meta.externalPath().get().startsWith("file:" + tmpPath)).isTrue();
+ }
+ }
+ }
+
// ----------------------- Sort Compact -----------------------
@Test
(paimon) 06/15: [rest] Adjust writeTimeout of HttpClient to 3 minutes
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit dc779c1512179a57ec56f550135b481a14a11278
Author: JingsongLi <ji...@gmail.com>
AuthorDate: Fri Jun 13 15:33:53 2025 +0800
[rest] Adjust writeTimeout of HttpClient to 3 minutes
---
paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java | 1 +
1 file changed, 1 insertion(+)
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java
index 8aba11d52c..7d0f07b80b 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java
@@ -59,6 +59,7 @@ public class HttpClient implements RESTClient {
.addInterceptor(new LoggingInterceptor())
.connectTimeout(Duration.ofMinutes(3))
.readTimeout(Duration.ofMinutes(3))
+ .writeTimeout(Duration.ofMinutes(3))
.build();
private static final MediaType MEDIA_TYPE = MediaType.parse("application/json");
(paimon) 07/15: [flink] Postpone mode should support 'partition.sink-strategy' (#5743)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 79a3ab2b58ae523799ddc584d24999b4c7378c10
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Fri Jun 13 19:41:10 2025 +0800
[flink] Postpone mode should support 'partition.sink-strategy' (#5743)
---
docs/layouts/shortcodes/generated/core_configuration.html | 2 +-
paimon-api/src/main/java/org/apache/paimon/CoreOptions.java | 2 +-
.../java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java | 11 +++++++++--
.../flink/sink/RowDataHashPartitionChannelComputer.java | 10 ++++------
.../org/apache/paimon/flink/PostponeBucketTableITCase.java | 11 ++++++++++-
5 files changed, 25 insertions(+), 11 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
index 020b03c0c8..8cd61f61d6 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -765,7 +765,7 @@ This config option does not affect the default filesystem metastore.</td>
<td><h5>partition.sink-strategy</h5></td>
<td style="word-wrap: break-word;">NONE</td>
<td><p>Enum</p></td>
- <td>This is only for partitioned unaware-buckets append table, and the purpose is to reduce small files and improve write performance. Through this repartitioning strategy to reduce the number of partitions written by each task to as few as possible.<ul><li>none: Rebalanced or Forward partitioning, this is the default behavior, this strategy is suitable for the number of partitions you write in a batch is much smaller than write parallelism.</li><li>hash: Hash the partitions [...]
+ <td>This is only for partitioned append table or postpone pk table, and the purpose is to reduce small files and improve write performance. Through this repartitioning strategy to reduce the number of partitions written by each task to as few as possible.<ul><li>none: Rebalanced or Forward partitioning, this is the default behavior, this strategy is suitable for the number of partitions you write in a batch is much smaller than write parallelism.</li><li>hash: Hash the partit [...]
</tr>
<tr>
<td><h5>partition.timestamp-formatter</h5></td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index e79d1593d5..1df7949acc 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1325,7 +1325,7 @@ public class CoreOptions implements Serializable {
.withDescription(
Description.builder()
.text(
- "This is only for partitioned unaware-buckets append table, and the purpose is to reduce small files and improve write performance."
+ "This is only for partitioned append table or postpone pk table, and the purpose is to reduce small files and improve write performance."
+ " Through this repartitioning strategy to reduce the number of partitions written by each task to as few as possible.")
.list(
text(
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 e83bbeecbb..7f211870b9 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
@@ -30,6 +30,7 @@ import org.apache.paimon.flink.sorter.TableSorter;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.ChannelComputer;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
@@ -291,8 +292,14 @@ public class FlinkSinkBuilder {
}
private DataStreamSink<?> buildPostponeBucketSink(DataStream<InternalRow> input) {
- DataStream<InternalRow> partitioned =
- partition(input, new PostponeBucketChannelComputer(table.schema()), parallelism);
+ ChannelComputer<InternalRow> channelComputer;
+ if (!table.partitionKeys().isEmpty()
+ && table.coreOptions().partitionSinkStrategy() == PartitionSinkStrategy.HASH) {
+ channelComputer = new RowDataHashPartitionChannelComputer(table.schema());
+ } else {
+ channelComputer = new PostponeBucketChannelComputer(table.schema());
+ }
+ DataStream<InternalRow> partitioned = partition(input, channelComputer, parallelism);
FixedBucketSink sink = new FixedBucketSink(table, overwritePartition, null);
return sink.sinkFrom(partitioned);
}
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 73258e2966..8943c549ce 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,8 @@ 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.RowPartitionKeyExtractor;
/** This is only for partitioned unaware-buckets Append only table. */
public class RowDataHashPartitionChannelComputer implements ChannelComputer<InternalRow> {
@@ -32,7 +31,7 @@ public class RowDataHashPartitionChannelComputer implements ChannelComputer<Inte
private final TableSchema schema;
private transient int numChannels;
- private transient KeyAndBucketExtractor<InternalRow> extractor;
+ private transient RowPartitionKeyExtractor extractor;
public RowDataHashPartitionChannelComputer(TableSchema schema) {
this.schema = schema;
@@ -41,13 +40,12 @@ public class RowDataHashPartitionChannelComputer implements ChannelComputer<Inte
@Override
public void setup(int numChannels) {
this.numChannels = numChannels;
- this.extractor = new AppendTableRowKeyExtractor(schema);
+ this.extractor = new RowPartitionKeyExtractor(schema);
}
@Override
public int channel(InternalRow record) {
- extractor.setRecord(record);
- return ChannelComputer.select(extractor.partition(), 0, numChannels);
+ return ChannelComputer.select(extractor.partition(record), 0, numChannels);
}
@Override
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java
index cf319ac1c2..f66c2f0998 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java
@@ -31,6 +31,7 @@ import org.junit.jupiter.api.Timeout;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.assertj.core.api.Assertions.assertThat;
@@ -75,7 +76,15 @@ public class PostponeBucketTableITCase extends AbstractTestBase {
values.add(String.format("(%d, %d, %d)", i, j, i * numKeys + j));
}
}
- tEnv.executeSql("INSERT INTO T VALUES " + String.join(", ", values)).await();
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ if (random.nextBoolean()) {
+ tEnv.executeSql("INSERT INTO T VALUES " + String.join(", ", values)).await();
+ } else {
+ tEnv.executeSql(
+ "INSERT INTO T /*+ OPTIONS('partition.sink-strategy'='hash') */ VALUES "
+ + String.join(", ", values))
+ .await();
+ }
assertThat(collect(tEnv.executeSql("SELECT * FROM T"))).isEmpty();
tEnv.executeSql("CALL sys.compact(`table` => 'default.T')").await();
(paimon) 01/15: [hive] Hive catalog configuration to get owner for table creation (#5668)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 2aabf52d0f84187320d1482dd9118343587e698e
Author: cxzl25 <38...@users.noreply.github.com>
AuthorDate: Thu Jun 12 15:39:45 2025 +0800
[hive] Hive catalog configuration to get owner for table creation (#5668)
---
.../java/org/apache/paimon/hive/HiveCatalog.java | 20 ++++++++++++++++++--
1 file changed, 18 insertions(+), 2 deletions(-)
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
index e0f94d92da..35747cdb63 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
@@ -74,6 +74,7 @@ import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.api.UnknownTableException;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -1358,6 +1359,22 @@ public class HiveCatalog extends AbstractCatalog {
return table != null && TableType.EXTERNAL_TABLE.name().equals(table.getTableType());
}
+ private static String currentUser() {
+ String username = null;
+ try {
+ username = UserGroupInformation.getCurrentUser().getShortUserName();
+ } catch (IOException e) {
+ LOG.warn("Failed to get Hadoop user", e);
+ }
+
+ if (username != null) {
+ return username;
+ } else {
+ LOG.warn("Hadoop user is null, defaulting to user.name");
+ return System.getProperty("user.name");
+ }
+ }
+
private Table newHmsTable(
Identifier identifier,
Map<String, String> tableParameters,
@@ -1368,8 +1385,7 @@ public class HiveCatalog extends AbstractCatalog {
new Table(
identifier.getTableName(),
identifier.getDatabaseName(),
- // current linux user
- System.getProperty("user.name"),
+ currentUser(),
(int) (currentTimeMillis / 1000),
(int) (currentTimeMillis / 1000),
Integer.MAX_VALUE,
(paimon) 12/15: [core] Simplify force rewrite files in Compact Task (#5751)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 7306a0ac4b4df5fe84034835df087f63937aca45
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Jun 16 10:16:53 2025 +0800
[core] Simplify force rewrite files in Compact Task (#5751)
---
.../shortcodes/generated/core_configuration.html | 2 +-
.../main/java/org/apache/paimon/CoreOptions.java | 8 +--
.../append/BucketedAppendCompactManager.java | 20 +++----
.../org/apache/paimon/compact/CompactUnit.java | 43 ++++++++------
.../paimon/mergetree/compact/CompactStrategy.java | 35 ++++-------
.../mergetree/compact/FileRewriteCompactTask.java | 67 ++++++++++++++++++++++
.../mergetree/compact/MergeTreeCompactManager.java | 45 +++++++++------
.../mergetree/compact/MergeTreeCompactTask.java | 59 ++++++-------------
.../operation/BucketedAppendFileStoreWrite.java | 2 +-
.../paimon/operation/KeyValueFileStoreWrite.java | 2 +-
.../flink/procedure/CompactProcedureITCase.java | 4 +-
11 files changed, 167 insertions(+), 120 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
index 6bee1104a1..d7b21f18f7 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -183,7 +183,7 @@ under the License.
<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-compact-all-files</h5></td>
+ <td><h5>compaction.force-rewrite-all-files</h5></td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
<td>Whether to force pick all files for a full compaction. Usually seen in a compaction task to external paths.</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 47967be549..4c446dc4e5 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -151,8 +151,8 @@ public class CoreOptions implements Serializable {
+ ExternalPathStrategy.SPECIFIC_FS
+ ", should be the prefix scheme of the external path, now supported are s3 and oss.");
- public static final ConfigOption<Boolean> COMPACTION_FORCE_COMPACT_ALL_FILES =
- key("compaction.force-compact-all-files")
+ public static final ConfigOption<Boolean> COMPACTION_FORCE_REWRITE_ALL_FILES =
+ key("compaction.force-rewrite-all-files")
.booleanType()
.defaultValue(false)
.withDescription(
@@ -2473,8 +2473,8 @@ public class CoreOptions implements Serializable {
return options.get(DATA_FILE_EXTERNAL_PATHS_SPECIFIC_FS);
}
- public Boolean forceCompactAllFiles() {
- return options.get(COMPACTION_FORCE_COMPACT_ALL_FILES);
+ public Boolean forceRewriteAllFiles() {
+ return options.get(COMPACTION_FORCE_REWRITE_ALL_FILES);
}
public String partitionTimestampFormatter() {
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java b/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java
index 520fd8edb3..a27a40e8a9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java
@@ -60,7 +60,7 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
private final PriorityQueue<DataFileMeta> toCompact;
private final int minFileNum;
private final long targetFileSize;
- private final boolean forceCompactAllFiles;
+ private final boolean forceRewriteAllFiles;
private final CompactRewriter rewriter;
private List<DataFileMeta> compacting;
@@ -73,7 +73,7 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
@Nullable DeletionVectorsMaintainer dvMaintainer,
int minFileNum,
long targetFileSize,
- boolean forceCompactAllFiles,
+ boolean forceRewriteAllFiles,
CompactRewriter rewriter,
@Nullable CompactionMetrics.Reporter metricsReporter) {
this.executor = executor;
@@ -82,7 +82,7 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
this.toCompact.addAll(restored);
this.minFileNum = minFileNum;
this.targetFileSize = targetFileSize;
- this.forceCompactAllFiles = forceCompactAllFiles;
+ this.forceRewriteAllFiles = forceRewriteAllFiles;
this.rewriter = rewriter;
this.metricsReporter = metricsReporter;
}
@@ -102,7 +102,7 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
"A compaction task is still running while the user "
+ "forces a new compaction. This is unexpected.");
// if all files are force picked or deletion vector enables, always trigger compaction.
- if (!forceCompactAllFiles
+ if (!forceRewriteAllFiles
&& (toCompact.isEmpty()
|| (dvMaintainer == null && toCompact.size() < FULL_COMPACT_MIN_FILE))) {
return;
@@ -118,7 +118,7 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
dvMaintainer,
toCompact,
targetFileSize,
- forceCompactAllFiles,
+ forceRewriteAllFiles,
rewriter,
metricsReporter));
recordCompactionsQueuedRequest();
@@ -243,28 +243,28 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
private final DeletionVectorsMaintainer dvMaintainer;
private final LinkedList<DataFileMeta> toCompact;
private final long targetFileSize;
- private final boolean forceCompactAllFiles;
+ private final boolean forceRewriteAllFiles;
private final CompactRewriter rewriter;
public FullCompactTask(
DeletionVectorsMaintainer dvMaintainer,
Collection<DataFileMeta> inputs,
long targetFileSize,
- boolean forceCompactAllFiles,
+ boolean forceRewriteAllFiles,
CompactRewriter rewriter,
@Nullable CompactionMetrics.Reporter metricsReporter) {
super(metricsReporter);
this.dvMaintainer = dvMaintainer;
this.toCompact = new LinkedList<>(inputs);
this.targetFileSize = targetFileSize;
- this.forceCompactAllFiles = forceCompactAllFiles;
+ this.forceRewriteAllFiles = forceRewriteAllFiles;
this.rewriter = rewriter;
}
@Override
protected CompactResult doCompact() throws Exception {
// remove large files
- while (!forceCompactAllFiles && !toCompact.isEmpty()) {
+ while (!forceRewriteAllFiles && !toCompact.isEmpty()) {
DataFileMeta file = toCompact.peekFirst();
// the data file with deletion file always need to be compacted.
if (file.fileSize() >= targetFileSize && !hasDeletionFile(file)) {
@@ -289,7 +289,7 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
small++;
}
}
- if (forceCompactAllFiles
+ if (forceRewriteAllFiles
|| (small > big && toCompact.size() >= FULL_COMPACT_MIN_FILE)) {
return compact(null, toCompact, rewriter);
} else {
diff --git a/paimon-core/src/main/java/org/apache/paimon/compact/CompactUnit.java b/paimon-core/src/main/java/org/apache/paimon/compact/CompactUnit.java
index 9b0fdd26ad..ae367f9171 100644
--- a/paimon-core/src/main/java/org/apache/paimon/compact/CompactUnit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/compact/CompactUnit.java
@@ -25,31 +25,40 @@ import java.util.ArrayList;
import java.util.List;
/** A files unit for compaction. */
-public interface CompactUnit {
+public class CompactUnit {
- int outputLevel();
+ private final int outputLevel;
+ private final List<DataFileMeta> files;
+ private final boolean fileRewrite;
- List<DataFileMeta> files();
+ public CompactUnit(int outputLevel, List<DataFileMeta> files, boolean fileRewrite) {
+ this.outputLevel = outputLevel;
+ this.files = files;
+ this.fileRewrite = fileRewrite;
+ }
+
+ public int outputLevel() {
+ return outputLevel;
+ }
+
+ public List<DataFileMeta> files() {
+ return files;
+ }
+
+ public boolean fileRewrite() {
+ return fileRewrite;
+ }
- static CompactUnit fromLevelRuns(int outputLevel, List<LevelSortedRun> runs) {
+ public static CompactUnit fromLevelRuns(int outputLevel, List<LevelSortedRun> runs) {
List<DataFileMeta> files = new ArrayList<>();
for (LevelSortedRun run : runs) {
files.addAll(run.run().files());
}
- return fromFiles(outputLevel, files);
+ return fromFiles(outputLevel, files, false);
}
- static CompactUnit fromFiles(int outputLevel, List<DataFileMeta> files) {
- return new CompactUnit() {
- @Override
- public int outputLevel() {
- return outputLevel;
- }
-
- @Override
- public List<DataFileMeta> files() {
- return files;
- }
- };
+ public static CompactUnit fromFiles(
+ int outputLevel, List<DataFileMeta> files, boolean fileRewrite) {
+ return new CompactUnit(outputLevel, files, fileRewrite);
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactStrategy.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactStrategy.java
index ef56196e0a..0ab0981963 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactStrategy.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactStrategy.java
@@ -32,7 +32,6 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
-import java.util.stream.Collectors;
/** Compact strategy to decide which files to select for compaction. */
public interface CompactStrategy {
@@ -56,16 +55,19 @@ public interface CompactStrategy {
List<LevelSortedRun> runs,
@Nullable RecordLevelExpire recordLevelExpire,
@Nullable DeletionVectorsMaintainer dvMaintainer,
- boolean forceCompactAllFiles) {
+ boolean forceRewriteAllFiles) {
int maxLevel = numLevels - 1;
if (runs.isEmpty()) {
// no sorted run, no need to compact
return Optional.empty();
- } else if ((runs.size() == 1 && runs.get(0).level() == maxLevel)) {
+ }
+
+ // only max level files
+ if ((runs.size() == 1 && runs.get(0).level() == maxLevel)) {
List<DataFileMeta> filesToBeCompacted = new ArrayList<>();
for (DataFileMeta file : runs.get(0).run().files()) {
- if (forceCompactAllFiles) {
+ if (forceRewriteAllFiles) {
// add all files when force compacted
filesToBeCompacted.add(file);
} else if (recordLevelExpire != null && recordLevelExpire.isExpireFile(file)) {
@@ -78,27 +80,14 @@ public interface CompactStrategy {
}
}
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "Pick these files which have expired records or dv index for full compaction: {}",
- filesToBeCompacted.stream()
- .map(
- file ->
- String.format(
- "(%s, %d, %d)",
- file.fileName(),
- file.level(),
- file.fileSize()))
- .collect(Collectors.joining(", ")));
- }
-
- if (!filesToBeCompacted.isEmpty()) {
- return Optional.of(CompactUnit.fromFiles(maxLevel, filesToBeCompacted));
- } else {
+ if (filesToBeCompacted.isEmpty()) {
return Optional.empty();
}
- } else {
- return Optional.of(CompactUnit.fromLevelRuns(maxLevel, runs));
+
+ return Optional.of(CompactUnit.fromFiles(maxLevel, filesToBeCompacted, true));
}
+
+ // full compaction
+ return Optional.of(CompactUnit.fromLevelRuns(maxLevel, runs));
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FileRewriteCompactTask.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FileRewriteCompactTask.java
new file mode 100644
index 0000000000..620ea0748a
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FileRewriteCompactTask.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.mergetree.compact;
+
+import org.apache.paimon.compact.CompactResult;
+import org.apache.paimon.compact.CompactTask;
+import org.apache.paimon.compact.CompactUnit;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.mergetree.SortedRun;
+import org.apache.paimon.operation.metrics.CompactionMetrics;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+import static java.util.Collections.singletonList;
+
+/** Compact task for file rewrite compaction. */
+public class FileRewriteCompactTask extends CompactTask {
+
+ private final CompactRewriter rewriter;
+ private final int outputLevel;
+ private final List<DataFileMeta> files;
+ private final boolean dropDelete;
+
+ public FileRewriteCompactTask(
+ CompactRewriter rewriter,
+ CompactUnit unit,
+ boolean dropDelete,
+ @Nullable CompactionMetrics.Reporter metricsReporter) {
+ super(metricsReporter);
+ this.rewriter = rewriter;
+ this.outputLevel = unit.outputLevel();
+ this.files = unit.files();
+ this.dropDelete = dropDelete;
+ }
+
+ @Override
+ protected CompactResult doCompact() throws Exception {
+ CompactResult result = new CompactResult();
+ for (DataFileMeta file : files) {
+ rewriteFile(file, result);
+ }
+ return result;
+ }
+
+ private void rewriteFile(DataFileMeta file, CompactResult toUpdate) throws Exception {
+ List<List<SortedRun>> candidate = singletonList(singletonList(SortedRun.fromSingle(file)));
+ toUpdate.merge(rewriter.rewrite(outputLevel, dropDelete, candidate));
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
index 953a235391..025fd20156 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
@@ -23,6 +23,7 @@ import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.compact.CompactDeletionFile;
import org.apache.paimon.compact.CompactFutureManager;
import org.apache.paimon.compact.CompactResult;
+import org.apache.paimon.compact.CompactTask;
import org.apache.paimon.compact.CompactUnit;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.deletionvectors.DeletionVectorsMaintainer;
@@ -65,7 +66,7 @@ public class MergeTreeCompactManager extends CompactFutureManager {
@Nullable private final DeletionVectorsMaintainer dvMaintainer;
private final boolean lazyGenDeletionFile;
private final boolean needLookup;
- private final boolean forceCompactAllFiles;
+ private final boolean forceRewriteAllFiles;
@Nullable private final RecordLevelExpire recordLevelExpire;
@@ -82,7 +83,7 @@ public class MergeTreeCompactManager extends CompactFutureManager {
boolean lazyGenDeletionFile,
boolean needLookup,
@Nullable RecordLevelExpire recordLevelExpire,
- boolean forceCompactAllFiles) {
+ boolean forceRewriteAllFiles) {
this.executor = executor;
this.levels = levels;
this.strategy = strategy;
@@ -95,7 +96,7 @@ public class MergeTreeCompactManager extends CompactFutureManager {
this.lazyGenDeletionFile = lazyGenDeletionFile;
this.recordLevelExpire = recordLevelExpire;
this.needLookup = needLookup;
- this.forceCompactAllFiles = forceCompactAllFiles;
+ this.forceRewriteAllFiles = forceRewriteAllFiles;
MetricUtils.safeCall(this::reportMetrics, LOG);
}
@@ -142,7 +143,7 @@ public class MergeTreeCompactManager extends CompactFutureManager {
runs,
recordLevelExpire,
dvMaintainer,
- forceCompactAllFiles);
+ forceRewriteAllFiles);
} else {
if (taskFuture != null) {
return;
@@ -152,7 +153,7 @@ public class MergeTreeCompactManager extends CompactFutureManager {
}
optionalUnit =
strategy.pick(levels.numberOfLevels(), runs)
- .filter(unit -> unit.files().size() > 0)
+ .filter(unit -> !unit.files().isEmpty())
.filter(
unit ->
unit.files().size() > 1
@@ -206,22 +207,28 @@ public class MergeTreeCompactManager extends CompactFutureManager {
: () -> CompactDeletionFile.generateFiles(dvMaintainer);
}
- MergeTreeCompactTask task =
- new MergeTreeCompactTask(
- keyComparator,
- compactionFileSize,
- rewriter,
- unit,
- dropDelete,
- levels.maxLevel(),
- metricsReporter,
- compactDfSupplier,
- dvMaintainer,
- recordLevelExpire,
- forceCompactAllFiles);
+ CompactTask task;
+ if (unit.fileRewrite()) {
+ task = new FileRewriteCompactTask(rewriter, unit, dropDelete, metricsReporter);
+ } else {
+ task =
+ new MergeTreeCompactTask(
+ keyComparator,
+ compactionFileSize,
+ rewriter,
+ unit,
+ dropDelete,
+ levels.maxLevel(),
+ metricsReporter,
+ compactDfSupplier,
+ recordLevelExpire,
+ forceRewriteAllFiles);
+ }
+
if (LOG.isDebugEnabled()) {
LOG.debug(
- "Pick these files (name, level, size) for compaction: {}",
+ "Pick these files (name, level, size) for {} compaction: {}",
+ task.getClass().getSimpleName(),
unit.files().stream()
.map(
file ->
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactTask.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactTask.java
index 4aa10e9b79..667a965c62 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactTask.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactTask.java
@@ -23,7 +23,6 @@ import org.apache.paimon.compact.CompactResult;
import org.apache.paimon.compact.CompactTask;
import org.apache.paimon.compact.CompactUnit;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.deletionvectors.DeletionVectorsMaintainer;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.RecordLevelExpire;
import org.apache.paimon.mergetree.SortedRun;
@@ -45,19 +44,15 @@ public class MergeTreeCompactTask extends CompactTask {
private final CompactRewriter rewriter;
private final int outputLevel;
private final Supplier<CompactDeletionFile> compactDfSupplier;
-
private final List<List<SortedRun>> partitioned;
-
private final boolean dropDelete;
private final int maxLevel;
+ @Nullable private final RecordLevelExpire recordLevelExpire;
+ private final boolean forceRewriteAllFiles;
// metric
private int upgradeFilesNum;
- @Nullable private final RecordLevelExpire recordLevelExpire;
- private final boolean forceCompactAllFiles;
- @Nullable private final DeletionVectorsMaintainer dvMaintainer;
-
public MergeTreeCompactTask(
Comparator<InternalRow> keyComparator,
long minFileSize,
@@ -67,20 +62,18 @@ public class MergeTreeCompactTask extends CompactTask {
int maxLevel,
@Nullable CompactionMetrics.Reporter metricsReporter,
Supplier<CompactDeletionFile> compactDfSupplier,
- @Nullable DeletionVectorsMaintainer dvMaintainer,
@Nullable RecordLevelExpire recordLevelExpire,
- boolean forceCompactAllFiles) {
+ boolean forceRewriteAllFiles) {
super(metricsReporter);
this.minFileSize = minFileSize;
this.rewriter = rewriter;
this.outputLevel = unit.outputLevel();
this.compactDfSupplier = compactDfSupplier;
- this.dvMaintainer = dvMaintainer;
this.partitioned = new IntervalPartition(unit.files(), keyComparator).partition();
this.dropDelete = dropDelete;
this.maxLevel = maxLevel;
this.recordLevelExpire = recordLevelExpire;
- this.forceCompactAllFiles = forceCompactAllFiles;
+ this.forceRewriteAllFiles = forceRewriteAllFiles;
this.upgradeFilesNum = 0;
}
@@ -128,34 +121,19 @@ public class MergeTreeCompactTask extends CompactTask {
}
private void upgrade(DataFileMeta file, CompactResult toUpdate) throws Exception {
- if (file.level() == outputLevel) {
- if (forceCompactAllFiles
- || isContainExpiredRecords(file)
- || (dvMaintainer != null
- && dvMaintainer.deletionVectorOf(file.fileName()).isPresent())) {
- /*
- * 1. if files are force picked, we need to rewrite all files.
- * 2. if the large file in maxLevel has expired records, we need to rewrite it.
- * 3. if the large file in maxLevel has corresponding deletion vector, we need to rewrite it.
- */
- rewriteFile(file, toUpdate);
- }
+ if ((outputLevel == maxLevel && containsDeleteRecords(file))
+ || forceRewriteAllFiles
+ || containsExpiredRecords(file)) {
+ List<List<SortedRun>> candidate = new ArrayList<>();
+ candidate.add(singletonList(SortedRun.fromSingle(file)));
+ rewriteImpl(candidate, toUpdate);
return;
}
- if (outputLevel != maxLevel || file.deleteRowCount().map(d -> d == 0).orElse(false)) {
- if (forceCompactAllFiles || isContainExpiredRecords(file)) {
- // if all files are force picked, or the file which could be directly upgraded has
- // expired records, we need to rewrite it
- rewriteFile(file, toUpdate);
- } else {
- CompactResult upgradeResult = rewriter.upgrade(outputLevel, file);
- toUpdate.merge(upgradeResult);
- upgradeFilesNum++;
- }
- } else {
- // files with delete records should not be upgraded directly to max level
- rewriteFile(file, toUpdate);
+ if (file.level() != outputLevel) {
+ CompactResult upgradeResult = rewriter.upgrade(outputLevel, file);
+ toUpdate.merge(upgradeResult);
+ upgradeFilesNum++;
}
}
@@ -185,14 +163,11 @@ public class MergeTreeCompactTask extends CompactTask {
candidate.clear();
}
- private void rewriteFile(DataFileMeta file, CompactResult toUpdate) throws Exception {
- List<List<SortedRun>> candidate = new ArrayList<>();
- candidate.add(new ArrayList<>());
- candidate.get(0).add(SortedRun.fromSingle(file));
- rewriteImpl(candidate, toUpdate);
+ private boolean containsDeleteRecords(DataFileMeta file) {
+ return file.deleteRowCount().map(d -> d > 0).orElse(true);
}
- private boolean isContainExpiredRecords(DataFileMeta file) {
+ private boolean containsExpiredRecords(DataFileMeta file) {
return recordLevelExpire != null && recordLevelExpire.isExpireFile(file);
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java
index d8f2d9211b..03320ad6e1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java
@@ -91,7 +91,7 @@ public class BucketedAppendFileStoreWrite extends BaseAppendFileStoreWrite {
dvMaintainer,
options.compactionMinFileNum(),
options.targetFileSize(false),
- options.forceCompactAllFiles(),
+ options.forceRewriteAllFiles(),
files -> compactRewrite(partition, bucket, dvFactory, files),
compactionMetrics == null
? null
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
index 7c0b36701f..30fce29821 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
@@ -291,7 +291,7 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
options.prepareCommitWaitCompaction(),
options.needLookup(),
recordLevelExpire,
- options.forceCompactAllFiles());
+ options.forceRewriteAllFiles());
}
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java
index d58c09d3a8..02001ff1af 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java
@@ -222,7 +222,7 @@ public class CompactProcedureITCase extends CatalogITCaseBase {
tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
sql(
"CALL sys.compact(`table` => 'default.Tpk',"
- + " options => 'compaction.force-compact-all-files=true,data-file.external-paths=file://%s,data-file.external-paths.strategy=specific-fs,data-file.external-paths.specific-fs=file')",
+ + " options => 'compaction.force-rewrite-all-files=true,data-file.external-paths=file://%s,data-file.external-paths.strategy=specific-fs,data-file.external-paths.specific-fs=file')",
tmpPath);
List<DataSplit> splits = pkTable.newSnapshotReader().read().dataSplits();
for (DataSplit split : splits) {
@@ -253,7 +253,7 @@ public class CompactProcedureITCase extends CatalogITCaseBase {
tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
sql(
"CALL sys.compact(`table` => 'default.Tap',"
- + " options => 'compaction.force-compact-all-files=true,data-file.external-paths=file://%s,data-file.external-paths.strategy=specific-fs,data-file.external-paths.specific-fs=file')",
+ + " options => 'compaction.force-rewrite-all-files=true,data-file.external-paths=file://%s,data-file.external-paths.strategy=specific-fs,data-file.external-paths.specific-fs=file')",
tmpPath);
splits = apTable.newSnapshotReader().read().dataSplits();
for (DataSplit split : splits) {
(paimon) 09/15: [core] Limit max task number in ManifestReadThreadPool (#5737)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 68e2692f48d8d21835d50bf329d9c24ef1c69f32
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Sat Jun 14 08:57:37 2025 +0800
[core] Limit max task number in ManifestReadThreadPool (#5737)
---
.../org/apache/paimon/utils/ThreadPoolUtils.java | 16 +-
.../paimon/utils/SemaphoredDelegatingExecutor.java | 184 +++++++++++++++++++++
.../paimon/utils/ManifestReadThreadPool.java | 27 ++-
3 files changed, 205 insertions(+), 22 deletions(-)
diff --git a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java
index a4790583c5..7ad3425367 100644
--- a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java
+++ b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java
@@ -21,8 +21,6 @@ package org.apache.paimon.utils;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
-import javax.annotation.Nullable;
-
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collection;
@@ -36,8 +34,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -78,20 +74,12 @@ public class ThreadPoolUtils {
return executor;
}
- public static ScheduledExecutorService createScheduledThreadPool(
- int threadNum, String namePrefix) {
- return new ScheduledThreadPoolExecutor(threadNum, newDaemonThreadFactory(namePrefix));
- }
-
/** This method aims to parallel process tasks with memory control and sequentially. */
public static <T, U> Iterable<T> sequentialBatchedExecute(
- ThreadPoolExecutor executor,
+ ExecutorService executor,
Function<U, List<T>> processor,
List<U> input,
- @Nullable Integer queueSize) {
- if (queueSize == null) {
- queueSize = executor.getMaximumPoolSize();
- }
+ int queueSize) {
if (queueSize <= 0) {
throw new NegativeArraySizeException("queue size should not be negative");
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/SemaphoredDelegatingExecutor.java b/paimon-common/src/main/java/org/apache/paimon/utils/SemaphoredDelegatingExecutor.java
new file mode 100644
index 0000000000..bdbb23796b
--- /dev/null
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/SemaphoredDelegatingExecutor.java
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.utils;
+
+import org.apache.paimon.shade.guava30.com.google.common.util.concurrent.ForwardingExecutorService;
+import org.apache.paimon.shade.guava30.com.google.common.util.concurrent.Futures;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A {@link ForwardingExecutorService} to delegate tasks to limit the number of tasks executed
+ * concurrently.
+ */
+public class SemaphoredDelegatingExecutor extends ForwardingExecutorService {
+
+ private final Semaphore queueingPermits;
+ private final ExecutorService executorDelegated;
+ private final int permitCount;
+
+ public SemaphoredDelegatingExecutor(
+ ExecutorService executorDelegated, int permitCount, boolean fair) {
+ this.permitCount = permitCount;
+ this.queueingPermits = new Semaphore(permitCount, fair);
+ this.executorDelegated = executorDelegated;
+ }
+
+ @Override
+ protected ExecutorService delegate() {
+ return this.executorDelegated;
+ }
+
+ @Override
+ public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
+ throw new RuntimeException("Not implemented");
+ }
+
+ @Override
+ public <T> List<Future<T>> invokeAll(
+ Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) {
+ throw new RuntimeException("Not implemented");
+ }
+
+ @Override
+ public <T> T invokeAny(Collection<? extends Callable<T>> tasks) {
+ throw new RuntimeException("Not implemented");
+ }
+
+ @Override
+ public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) {
+ throw new RuntimeException("Not implemented");
+ }
+
+ @Override
+ public <T> Future<T> submit(Callable<T> task) {
+ try {
+ this.queueingPermits.acquire();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return Futures.immediateFailedFuture(e);
+ }
+
+ return super.submit(new CallableWithPermitRelease(task));
+ }
+
+ @Override
+ public <T> Future<T> submit(Runnable task, T result) {
+ try {
+ this.queueingPermits.acquire();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return Futures.immediateFailedFuture(e);
+ }
+
+ return super.submit(new RunnableWithPermitRelease(task), result);
+ }
+
+ @Override
+ public Future<?> submit(Runnable task) {
+ try {
+ this.queueingPermits.acquire();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return Futures.immediateFailedFuture(e);
+ }
+
+ return super.submit(new RunnableWithPermitRelease(task));
+ }
+
+ @Override
+ public void execute(Runnable command) {
+ try {
+ this.queueingPermits.acquire();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ super.execute(new RunnableWithPermitRelease(command));
+ }
+
+ public int getAvailablePermits() {
+ return this.queueingPermits.availablePermits();
+ }
+
+ public int getWaitingCount() {
+ return this.queueingPermits.getQueueLength();
+ }
+
+ public int getPermitCount() {
+ return this.permitCount;
+ }
+
+ @Override
+ public String toString() {
+ return "SemaphoredDelegatingExecutor{"
+ + "permitCount="
+ + getPermitCount()
+ + ", available="
+ + getAvailablePermits()
+ + ", waiting="
+ + getWaitingCount()
+ + '}';
+ }
+
+ private class RunnableWithPermitRelease implements Runnable {
+
+ private final Runnable delegated;
+
+ RunnableWithPermitRelease(Runnable delegated) {
+ this.delegated = delegated;
+ }
+
+ @Override
+ public void run() {
+ try {
+ this.delegated.run();
+ } finally {
+ SemaphoredDelegatingExecutor.this.queueingPermits.release();
+ }
+ }
+ }
+
+ private class CallableWithPermitRelease<T> implements Callable<T> {
+
+ private final Callable<T> delegated;
+
+ CallableWithPermitRelease(Callable<T> delegated) {
+ this.delegated = delegated;
+ }
+
+ @Override
+ public T call() throws Exception {
+ T result;
+ try {
+ result = this.delegated.call();
+ } finally {
+ SemaphoredDelegatingExecutor.this.queueingPermits.release();
+ }
+
+ return result;
+ }
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java b/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java
index 49fcfc8bd9..16bd73ee80 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java
@@ -22,6 +22,7 @@ import javax.annotation.Nullable;
import java.util.Iterator;
import java.util.List;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Function;
@@ -35,28 +36,38 @@ public class ManifestReadThreadPool {
private static ThreadPoolExecutor executorService =
createCachedThreadPool(Runtime.getRuntime().availableProcessors(), THREAD_NAME);
- public static synchronized ThreadPoolExecutor getExecutorService(@Nullable Integer threadNum) {
- if (threadNum == null || threadNum <= executorService.getMaximumPoolSize()) {
+ public static synchronized ExecutorService getExecutorService(@Nullable Integer threadNum) {
+ if (threadNum == null || threadNum == executorService.getMaximumPoolSize()) {
return executorService;
}
- // we don't need to close previous pool
- // it is just cached pool
- executorService = createCachedThreadPool(threadNum, THREAD_NAME);
+ if (threadNum < executorService.getMaximumPoolSize()) {
+ return new SemaphoredDelegatingExecutor(executorService, threadNum, false);
+ } else {
+ // we don't need to close previous pool
+ // it is just cached pool
+ executorService = createCachedThreadPool(threadNum, THREAD_NAME);
- return executorService;
+ return executorService;
+ }
}
/** This method aims to parallel process tasks with memory control and sequentially. */
public static <T, U> Iterable<T> sequentialBatchedExecute(
Function<U, List<T>> processor, List<U> input, @Nullable Integer threadNum) {
- ThreadPoolExecutor executor = getExecutorService(threadNum);
+ ExecutorService executor = getExecutorService(threadNum);
+ if (threadNum == null) {
+ threadNum =
+ executor instanceof ThreadPoolExecutor
+ ? ((ThreadPoolExecutor) executor).getMaximumPoolSize()
+ : ((SemaphoredDelegatingExecutor) executor).getPermitCount();
+ }
return ThreadPoolUtils.sequentialBatchedExecute(executor, processor, input, threadNum);
}
/** This method aims to parallel process tasks with randomly but return values sequentially. */
public static <T, U> Iterator<T> randomlyExecuteSequentialReturn(
Function<U, List<T>> processor, List<U> input, @Nullable Integer threadNum) {
- ThreadPoolExecutor executor = getExecutorService(threadNum);
+ ExecutorService executor = getExecutorService(threadNum);
return ThreadPoolUtils.randomlyExecuteSequentialReturn(executor, processor, input);
}
}
(paimon) 04/15: [test][spark] Fix the unstable testHiveCatalogOptions (#5738)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 4efd31c85dda0e07e3d6723c8e09213e9080217c
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Thu Jun 12 18:12:32 2025 +0800
[test][spark] Fix the unstable testHiveCatalogOptions (#5738)
---
.../paimon/spark/SparkCatalogWithHiveTest.java | 1 +
.../spark/SparkGenericCatalogWithHiveTest.java | 30 ++++++++++------------
.../catalog/functions/BucketFunctionTest.java | 6 +++++
3 files changed, 20 insertions(+), 17 deletions(-)
diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithHiveTest.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithHiveTest.java
index 9f383a54c2..c7c07b1aef 100644
--- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithHiveTest.java
+++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithHiveTest.java
@@ -109,6 +109,7 @@ public class SparkCatalogWithHiveTest {
.collect(Collectors.toList()))
.containsExactlyInAnyOrder("[1,1,1]", "[2,2,2]");
}
+ spark.stop();
}
@Test
diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkGenericCatalogWithHiveTest.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkGenericCatalogWithHiveTest.java
index 3bb013648e..9e13dd7be5 100644
--- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkGenericCatalogWithHiveTest.java
+++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkGenericCatalogWithHiveTest.java
@@ -60,12 +60,15 @@ public class SparkGenericCatalogWithHiveTest {
.config(
"spark.sql.catalog.spark_catalog",
SparkGenericCatalog.class.getName())
+ .config("spark.sql.catalog.paimon.warehouse", warehousePath.toString())
+ .config("spark.sql.catalog.paimon", SparkCatalog.class.getName())
.config(
"spark.sql.extensions",
"org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions")
.master("local[2]")
.getOrCreate();
+ spark.sql("USE spark_catalog");
spark.sql("CREATE DATABASE my_db");
spark.sql("USE my_db");
spark.sql(
@@ -80,29 +83,22 @@ public class SparkGenericCatalogWithHiveTest {
.map(s -> s.get(1))
.map(Object::toString))
.containsExactlyInAnyOrder("t1");
- spark.close();
- // secondly, we close catalog with hive metastore, and start a filesystem metastore to check
- // the result.
- SparkSession spark2 =
- SparkSession.builder()
- .config("spark.sql.catalog.paimon.warehouse", warehousePath.toString())
- .config("spark.sql.catalogImplementation", "in-memory")
- .config("spark.sql.catalog.paimon", SparkCatalog.class.getName())
- .config(
- "spark.sql.extensions",
- "org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions")
- .master("local[2]")
- .getOrCreate();
- spark2.sql("USE paimon");
- spark2.sql("USE my_db");
- assertThat(spark2.sql("SHOW NAMESPACES").collectAsList().stream().map(Object::toString))
+ // secondly, use filesystem metastore to check the result.
+ spark.sql("USE paimon");
+ spark.sql("USE my_db");
+ assertThat(spark.sql("SHOW NAMESPACES").collectAsList().stream().map(Object::toString))
.containsExactlyInAnyOrder("[default]", "[my_db]");
assertThat(
- spark2.sql("SHOW TABLES").collectAsList().stream()
+ spark.sql("SHOW TABLES").collectAsList().stream()
.map(s -> s.get(1))
.map(Object::toString))
.containsExactlyInAnyOrder("t1");
+
+ spark.sql("USE spark_catalog");
+ spark.sql("DROP TABLE my_db.t1");
+ spark.sql("DROP DATABASE my_db");
+ spark.close();
}
@Test
diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/catalog/functions/BucketFunctionTest.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/catalog/functions/BucketFunctionTest.java
index cf02654063..b8fbcdae42 100644
--- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/catalog/functions/BucketFunctionTest.java
+++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/catalog/functions/BucketFunctionTest.java
@@ -56,6 +56,7 @@ import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
import org.apache.spark.sql.SparkSession;
import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -190,6 +191,11 @@ public class BucketFunctionTest {
spark.sql(String.format("DROP TABLE IF EXISTS %s", TABLE_NAME));
}
+ @AfterAll
+ public static void tearDown() {
+ spark.stop();
+ }
+
public static void setupTable(String... bucketColumns) {
String commitUser = UUID.randomUUID().toString();
try {
(paimon) 14/15: [core] Introduce auto-buffer-spill for postpone bucket write (#5744)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit c2bdec5f6fd2d74eb4bc056a0558e98c7be5fb10
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Jun 16 10:31:54 2025 +0800
[core] Introduce auto-buffer-spill for postpone bucket write (#5744)
---
.../java/org/apache/paimon/KeyValueFileStore.java | 2 +
.../org/apache/paimon/append/AppendOnlyWriter.java | 216 +++----------------
.../paimon/operation/AbstractFileStoreWrite.java | 2 +-
.../postpone/PostponeBucketFileStoreWrite.java | 71 ++++++-
.../paimon/postpone/PostponeBucketWriter.java | 141 +++++++++++--
.../java/org/apache/paimon/utils/SinkWriter.java | 234 +++++++++++++++++++++
.../paimon/table/PrimaryKeySimpleTableTest.java | 38 ++++
7 files changed, 495 insertions(+), 209 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
index 81039da369..be4416af9b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
@@ -177,12 +177,14 @@ public class KeyValueFileStore extends AbstractFileStore<KeyValue> {
if (options.bucket() == BucketMode.POSTPONE_BUCKET) {
return new PostponeBucketFileStoreWrite(
fileIO,
+ pathFactory(),
schema,
commitUser,
partitionType,
keyType,
valueType,
this::pathFactory,
+ newReaderFactoryBuilder(),
snapshotManager(),
newScan(ScanType.FOR_WRITE).withManifestCacheFilter(manifestFilter),
options,
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
index 694937c88f..3534ac4e9e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
@@ -23,7 +23,6 @@ import org.apache.paimon.compact.CompactDeletionFile;
import org.apache.paimon.compact.CompactManager;
import org.apache.paimon.compression.CompressOptions;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.disk.RowBuffer;
import org.apache.paimon.fileindex.FileIndexOptions;
@@ -45,14 +44,15 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.BatchRecordWriter;
import org.apache.paimon.utils.CommitIncrement;
import org.apache.paimon.utils.IOFunction;
-import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.LongCounter;
import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.RecordWriter;
+import org.apache.paimon.utils.SinkWriter;
+import org.apache.paimon.utils.SinkWriter.BufferedSinkWriter;
+import org.apache.paimon.utils.SinkWriter.DirectSinkWriter;
import javax.annotation.Nullable;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -72,7 +72,7 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
private final RowType writeSchema;
private final DataFilePathFactory pathFactory;
private final CompactManager compactManager;
- private final IOFunction<List<DataFileMeta>, RecordReaderIterator<InternalRow>> bucketFileRead;
+ private final IOFunction<List<DataFileMeta>, RecordReaderIterator<InternalRow>> dataFileRead;
private final boolean forceCompact;
private final boolean asyncFileWrite;
private final boolean statsDenseStore;
@@ -80,17 +80,17 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
private final List<DataFileMeta> deletedFiles;
private final List<DataFileMeta> compactBefore;
private final List<DataFileMeta> compactAfter;
- @Nullable private CompactDeletionFile compactDeletionFile;
private final LongCounter seqNumCounter;
private final String fileCompression;
private final CompressOptions spillCompression;
- private SinkWriter sinkWriter;
private final SimpleColStatsCollector.Factory[] statsCollectors;
@Nullable private final IOManager ioManager;
private final FileIndexOptions fileIndexOptions;
+ private final MemorySize maxDiskSize;
+ @Nullable private CompactDeletionFile compactDeletionFile;
+ private SinkWriter<InternalRow> sinkWriter;
private MemorySegmentPool memorySegmentPool;
- private final MemorySize maxDiskSize;
public AppendOnlyWriter(
FileIO fileIO,
@@ -101,7 +101,7 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
RowType writeSchema,
long maxSequenceNumber,
CompactManager compactManager,
- IOFunction<List<DataFileMeta>, RecordReaderIterator<InternalRow>> bucketFileRead,
+ IOFunction<List<DataFileMeta>, RecordReaderIterator<InternalRow>> dataFileRead,
boolean forceCompact,
DataFilePathFactory pathFactory,
@Nullable CommitIncrement increment,
@@ -121,7 +121,7 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
this.writeSchema = writeSchema;
this.pathFactory = pathFactory;
this.compactManager = compactManager;
- this.bucketFileRead = bucketFileRead;
+ this.dataFileRead = dataFileRead;
this.forceCompact = forceCompact;
this.asyncFileWrite = asyncFileWrite;
this.statsDenseStore = statsDenseStore;
@@ -139,8 +139,8 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
this.sinkWriter =
useWriteBuffer
- ? new BufferedSinkWriter(spillable, maxDiskSize, spillCompression)
- : new DirectSinkWriter();
+ ? createBufferedSinkWriter(spillable)
+ : new DirectSinkWriter<>(this::createRollingRowWriter);
if (increment != null) {
newFiles.addAll(increment.newFilesIncrement().newFiles());
@@ -151,6 +151,18 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
}
}
+ private BufferedSinkWriter<InternalRow> createBufferedSinkWriter(boolean spillable) {
+ return new BufferedSinkWriter<>(
+ this::createRollingRowWriter,
+ t -> t,
+ t -> t,
+ ioManager,
+ writeSchema,
+ spillable,
+ maxDiskSize,
+ spillCompression);
+ }
+
@Override
public void write(InternalRow rowData) throws Exception {
Preconditions.checkArgument(
@@ -178,7 +190,7 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
write(row);
}
} else {
- ((DirectSinkWriter) sinkWriter).writeBundle(bundle);
+ ((DirectSinkWriter<?>) sinkWriter).writeBundle(bundle);
}
}
@@ -252,16 +264,16 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
}
public void toBufferedWriter() throws Exception {
- if (sinkWriter != null && !sinkWriter.bufferSpillableWriter() && bucketFileRead != null) {
+ if (sinkWriter != null && !sinkWriter.bufferSpillableWriter() && dataFileRead != null) {
// fetch the written results
List<DataFileMeta> files = sinkWriter.flush();
sinkWriter.close();
- sinkWriter = new BufferedSinkWriter(true, maxDiskSize, spillCompression);
+ sinkWriter = createBufferedSinkWriter(true);
sinkWriter.setMemoryPool(memorySegmentPool);
// rewrite small files
- try (RecordReaderIterator<InternalRow> reader = bucketFileRead.apply(files)) {
+ try (RecordReaderIterator<InternalRow> reader = dataFileRead.apply(files)) {
while (reader.hasNext()) {
sinkWriter.write(reader.next());
}
@@ -356,7 +368,7 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
@VisibleForTesting
public RowBuffer getWriteBuffer() {
if (sinkWriter instanceof BufferedSinkWriter) {
- return ((BufferedSinkWriter) sinkWriter).writeBuffer;
+ return ((BufferedSinkWriter<?>) sinkWriter).rowBuffer();
} else {
return null;
}
@@ -366,176 +378,4 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
List<DataFileMeta> getNewFiles() {
return newFiles;
}
-
- /** Internal interface to Sink Data from input. */
- private interface SinkWriter {
-
- boolean write(InternalRow data) throws IOException;
-
- List<DataFileMeta> flush() throws IOException;
-
- boolean flushMemory() throws IOException;
-
- long memoryOccupancy();
-
- void close();
-
- void setMemoryPool(MemorySegmentPool memoryPool);
-
- boolean bufferSpillableWriter();
- }
-
- /**
- * Directly sink data to file, no memory cache here, use OrcWriter/ParquetWrite/etc directly
- * write data. May cause out-of-memory.
- */
- private class DirectSinkWriter implements SinkWriter {
-
- private RowDataRollingFileWriter writer;
-
- @Override
- public boolean write(InternalRow data) throws IOException {
- if (writer == null) {
- writer = createRollingRowWriter();
- }
- writer.write(data);
- return true;
- }
-
- public void writeBundle(BundleRecords bundle) throws IOException {
- if (writer == null) {
- writer = createRollingRowWriter();
- }
- writer.writeBundle(bundle);
- }
-
- @Override
- public List<DataFileMeta> flush() throws IOException {
- List<DataFileMeta> flushedFiles = new ArrayList<>();
- if (writer != null) {
- writer.close();
- flushedFiles.addAll(writer.result());
- writer = null;
- }
- return flushedFiles;
- }
-
- @Override
- public boolean flushMemory() throws IOException {
- return false;
- }
-
- @Override
- public long memoryOccupancy() {
- return 0;
- }
-
- @Override
- public void close() {
- if (writer != null) {
- writer.abort();
- writer = null;
- }
- }
-
- @Override
- public void setMemoryPool(MemorySegmentPool memoryPool) {
- // do nothing
- }
-
- @Override
- public boolean bufferSpillableWriter() {
- return false;
- }
- }
-
- /**
- * Use buffered writer, segment pooled from segment pool. When spillable, may delay checkpoint
- * acknowledge time. When non-spillable, may cause too many small files.
- */
- private class BufferedSinkWriter implements SinkWriter {
-
- private final boolean spillable;
-
- private final MemorySize maxDiskSize;
-
- private final CompressOptions compression;
-
- private RowBuffer writeBuffer;
-
- private BufferedSinkWriter(
- boolean spillable, MemorySize maxDiskSize, CompressOptions compression) {
- this.spillable = spillable;
- this.maxDiskSize = maxDiskSize;
- this.compression = compression;
- }
-
- @Override
- public boolean write(InternalRow data) throws IOException {
- return writeBuffer.put(data);
- }
-
- @Override
- public List<DataFileMeta> flush() throws IOException {
- List<DataFileMeta> flushedFiles = new ArrayList<>();
- if (writeBuffer != null) {
- writeBuffer.complete();
- RowDataRollingFileWriter writer = createRollingRowWriter();
- IOException exception = null;
- try (RowBuffer.RowBufferIterator iterator = writeBuffer.newIterator()) {
- while (iterator.advanceNext()) {
- writer.write(iterator.getRow());
- }
- } catch (IOException e) {
- exception = e;
- } finally {
- if (exception != null) {
- IOUtils.closeQuietly(writer);
- // cleanup code that might throw another exception
- throw exception;
- }
- writer.close();
- }
- flushedFiles.addAll(writer.result());
- // reuse writeBuffer
- writeBuffer.reset();
- }
- return flushedFiles;
- }
-
- @Override
- public long memoryOccupancy() {
- return writeBuffer.memoryOccupancy();
- }
-
- @Override
- public void close() {
- if (writeBuffer != null) {
- writeBuffer.reset();
- writeBuffer = null;
- }
- }
-
- @Override
- public void setMemoryPool(MemorySegmentPool memoryPool) {
- writeBuffer =
- RowBuffer.getBuffer(
- ioManager,
- memoryPool,
- new InternalRowSerializer(writeSchema),
- spillable,
- maxDiskSize,
- compression);
- }
-
- @Override
- public boolean bufferSpillableWriter() {
- return spillable;
- }
-
- @Override
- public boolean flushMemory() throws IOException {
- return writeBuffer.flushMemory();
- }
- }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
index fad898da45..a7bac3f12a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
@@ -567,7 +567,7 @@ public abstract class AbstractFileStoreWrite<T> implements FileStoreWrite<T> {
}
@VisibleForTesting
- Map<BinaryRow, Map<Integer, WriterContainer<T>>> writers() {
+ public Map<BinaryRow, Map<Integer, WriterContainer<T>>> writers() {
return writers;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java
index 09b3afdff2..97b5dfe7c2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java
@@ -26,11 +26,15 @@ import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.avro.AvroSchemaConverter;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.KeyValueFileReaderFactory;
import org.apache.paimon.io.KeyValueFileWriterFactory;
-import org.apache.paimon.operation.AbstractFileStoreWrite;
+import org.apache.paimon.mergetree.compact.ConcatRecordReader;
import org.apache.paimon.operation.FileStoreScan;
import org.apache.paimon.operation.FileStoreWrite;
+import org.apache.paimon.operation.MemoryFileStoreWrite;
import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.ReaderSupplier;
+import org.apache.paimon.reader.RecordReaderIterator;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.types.RowType;
@@ -39,10 +43,17 @@ import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.SnapshotManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.BiFunction;
@@ -52,25 +63,37 @@ import static org.apache.paimon.format.FileFormat.fileFormat;
import static org.apache.paimon.utils.FileStorePathFactory.createFormatPathFactories;
/** {@link FileStoreWrite} for {@code bucket = -2} tables. */
-public class PostponeBucketFileStoreWrite extends AbstractFileStoreWrite<KeyValue> {
+public class PostponeBucketFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PostponeBucketFileStoreWrite.class);
private final CoreOptions options;
private final KeyValueFileWriterFactory.Builder writerFactoryBuilder;
+ private final FileIO fileIO;
+ private final FileStorePathFactory pathFactory;
+ private final KeyValueFileReaderFactory.Builder readerFactoryBuilder;
+
+ private boolean forceBufferSpill = false;
public PostponeBucketFileStoreWrite(
FileIO fileIO,
+ FileStorePathFactory pathFactory,
TableSchema schema,
String commitUser,
RowType partitionType,
RowType keyType,
RowType valueType,
BiFunction<CoreOptions, String, FileStorePathFactory> formatPathFactory,
+ KeyValueFileReaderFactory.Builder readerFactoryBuilder,
SnapshotManager snapshotManager,
FileStoreScan scan,
CoreOptions options,
String tableName,
@Nullable Integer writeId) {
- super(snapshotManager, scan, null, null, tableName, options, partitionType);
+ super(snapshotManager, scan, options, partitionType, null, null, tableName);
+ this.fileIO = fileIO;
+ this.pathFactory = pathFactory;
+ this.readerFactoryBuilder = readerFactoryBuilder;
Options newOptions = new Options(options.toMap());
try {
@@ -120,6 +143,25 @@ public class PostponeBucketFileStoreWrite extends AbstractFileStoreWrite<KeyValu
withIgnorePreviousFiles(true);
}
+ @Override
+ protected void forceBufferSpill() throws Exception {
+ if (ioManager == null) {
+ return;
+ }
+ if (forceBufferSpill) {
+ return;
+ }
+ forceBufferSpill = true;
+ LOG.info(
+ "Force buffer spill for postpone file store write, writer number is: {}",
+ writers.size());
+ for (Map<Integer, WriterContainer<KeyValue>> bucketWriters : writers.values()) {
+ for (WriterContainer<KeyValue> writerContainer : bucketWriters.values()) {
+ ((PostponeBucketWriter) writerContainer.writer).toBufferedWriter();
+ }
+ }
+ }
+
@Override
public void withIgnorePreviousFiles(boolean ignorePrevious) {
// see comments in constructor
@@ -141,7 +183,28 @@ public class PostponeBucketFileStoreWrite extends AbstractFileStoreWrite<KeyValu
"Postpone bucket writers should not restore previous files. This is unexpected.");
KeyValueFileWriterFactory writerFactory =
writerFactoryBuilder.build(partition, bucket, options);
- return new PostponeBucketWriter(writerFactory, restoreIncrement);
+ return new PostponeBucketWriter(
+ fileIO,
+ pathFactory.createDataFilePathFactory(partition, bucket),
+ options.spillCompressOptions(),
+ options.writeBufferSpillDiskSize(),
+ ioManager,
+ writerFactory,
+ files -> newFileRead(partition, bucket, files),
+ forceBufferSpill,
+ forceBufferSpill,
+ restoreIncrement);
+ }
+
+ private RecordReaderIterator<KeyValue> newFileRead(
+ BinaryRow partition, int bucket, List<DataFileMeta> files) throws IOException {
+ KeyValueFileReaderFactory readerFactory =
+ readerFactoryBuilder.build(partition, bucket, name -> Optional.empty());
+ List<ReaderSupplier<KeyValue>> suppliers = new ArrayList<>();
+ for (DataFileMeta file : files) {
+ suppliers.add(() -> readerFactory.createRecordReader(file));
+ }
+ return new RecordReaderIterator<>(ConcatRecordReader.create(suppliers));
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketWriter.java b/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketWriter.java
index 10259650d4..c41db8aac7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketWriter.java
@@ -19,14 +19,29 @@
package org.apache.paimon.postpone;
import org.apache.paimon.KeyValue;
+import org.apache.paimon.KeyValueSerializer;
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.compression.CompressOptions;
+import org.apache.paimon.disk.IOManager;
+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.io.KeyValueFileWriterFactory;
import org.apache.paimon.io.RollingFileWriter;
import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.memory.MemoryOwner;
+import org.apache.paimon.memory.MemorySegmentPool;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.reader.RecordReaderIterator;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CommitIncrement;
+import org.apache.paimon.utils.IOFunction;
import org.apache.paimon.utils.RecordWriter;
+import org.apache.paimon.utils.SinkWriter;
+import org.apache.paimon.utils.SinkWriter.BufferedSinkWriter;
+import org.apache.paimon.utils.SinkWriter.DirectSinkWriter;
import javax.annotation.Nullable;
@@ -36,30 +51,69 @@ import java.util.Collections;
import java.util.List;
/** {@link RecordWriter} for {@code bucket = -2} tables. */
-public class PostponeBucketWriter implements RecordWriter<KeyValue> {
+public class PostponeBucketWriter implements RecordWriter<KeyValue>, MemoryOwner {
+ private final FileIO fileIO;
+ private final DataFilePathFactory pathFactory;
private final KeyValueFileWriterFactory writerFactory;
private final List<DataFileMeta> files;
+ private final IOFunction<List<DataFileMeta>, RecordReaderIterator<KeyValue>> fileRead;
+ private final @Nullable IOManager ioManager;
+ private final CompressOptions spillCompression;
+ private final MemorySize maxDiskSize;
- private RollingFileWriter<KeyValue, DataFileMeta> writer;
+ private SinkWriter<KeyValue> sinkWriter;
+ private MemorySegmentPool memorySegmentPool;
public PostponeBucketWriter(
- KeyValueFileWriterFactory writerFactory, @Nullable CommitIncrement restoreIncrement) {
+ FileIO fileIO,
+ DataFilePathFactory pathFactory,
+ CompressOptions spillCompression,
+ MemorySize maxDiskSize,
+ @Nullable IOManager ioManager,
+ KeyValueFileWriterFactory writerFactory,
+ IOFunction<List<DataFileMeta>, RecordReaderIterator<KeyValue>> fileRead,
+ boolean useWriteBuffer,
+ boolean spillable,
+ @Nullable CommitIncrement restoreIncrement) {
+ this.ioManager = ioManager;
this.writerFactory = writerFactory;
+ this.fileRead = fileRead;
+ this.fileIO = fileIO;
+ this.pathFactory = pathFactory;
+ this.spillCompression = spillCompression;
+ this.maxDiskSize = maxDiskSize;
this.files = new ArrayList<>();
if (restoreIncrement != null) {
files.addAll(restoreIncrement.newFilesIncrement().newFiles());
}
+ this.sinkWriter =
+ useWriteBuffer
+ ? createBufferedSinkWriter(spillable)
+ : new DirectSinkWriter<>(this::createRollingRowWriter);
+ }
- this.writer = null;
+ private RollingFileWriter<KeyValue, DataFileMeta> createRollingRowWriter() {
+ return writerFactory.createRollingMergeTreeFileWriter(0, FileSource.APPEND);
}
@Override
public void write(KeyValue record) throws Exception {
- if (writer == null) {
- writer = writerFactory.createRollingMergeTreeFileWriter(0, FileSource.APPEND);
+ boolean success = sinkWriter.write(record);
+ if (!success) {
+ flush();
+ success = sinkWriter.write(record);
+ if (!success) {
+ // Should not get here, because writeBuffer will throw too big exception out.
+ // But we throw again in case of something unexpected happens. (like someone changed
+ // code in SpillableBuffer.)
+ throw new RuntimeException("Mem table is too small to hold a single element.");
+ }
}
- writer.write(record);
+ }
+
+ private void flush() throws Exception {
+ files.addAll(sinkWriter.flush());
}
@Override
@@ -82,13 +136,66 @@ public class PostponeBucketWriter implements RecordWriter<KeyValue> {
}
@Override
- public CommitIncrement prepareCommit(boolean waitCompaction) throws Exception {
- if (writer != null) {
- writer.close();
- files.addAll(writer.result());
- writer = null;
+ public void setMemoryPool(MemorySegmentPool memoryPool) {
+ this.memorySegmentPool = memoryPool;
+ sinkWriter.setMemoryPool(memoryPool);
+ }
+
+ @Override
+ public long memoryOccupancy() {
+ return sinkWriter.memoryOccupancy();
+ }
+
+ @Override
+ public void flushMemory() throws Exception {
+ boolean success = sinkWriter.flushMemory();
+ if (!success) {
+ flush();
}
+ }
+ private BufferedSinkWriter<KeyValue> createBufferedSinkWriter(boolean spillable) {
+ RowType keyType = writerFactory.keyType();
+ RowType valueType = writerFactory.valueType();
+ RowType kvRowType = KeyValue.schema(keyType, valueType);
+ KeyValueSerializer serializer = new KeyValueSerializer(keyType, valueType);
+ return new BufferedSinkWriter<>(
+ this::createRollingRowWriter,
+ serializer::toRow,
+ serializer::fromRow,
+ ioManager,
+ kvRowType,
+ spillable,
+ maxDiskSize,
+ spillCompression);
+ }
+
+ public void toBufferedWriter() throws Exception {
+ if (sinkWriter != null && !sinkWriter.bufferSpillableWriter() && fileRead != null) {
+ // fetch the written results
+ List<DataFileMeta> files = sinkWriter.flush();
+
+ sinkWriter.close();
+ sinkWriter = createBufferedSinkWriter(true);
+ sinkWriter.setMemoryPool(memorySegmentPool);
+
+ // rewrite small files
+ try (RecordReaderIterator<KeyValue> reader = fileRead.apply(files)) {
+ while (reader.hasNext()) {
+ sinkWriter.write(reader.next());
+ }
+ } finally {
+ // remove small files
+ for (DataFileMeta file : files) {
+ fileIO.deleteQuietly(pathFactory.toPath(file));
+ }
+ }
+ }
+ }
+
+ @Override
+ public CommitIncrement prepareCommit(boolean waitCompaction) throws Exception {
+ flush();
List<DataFileMeta> result = new ArrayList<>(files);
files.clear();
return new CommitIncrement(
@@ -97,6 +204,11 @@ public class PostponeBucketWriter implements RecordWriter<KeyValue> {
null);
}
+ @VisibleForTesting
+ public boolean useBufferedSinkWriter() {
+ return sinkWriter instanceof BufferedSinkWriter;
+ }
+
@Override
public boolean compactNotCompleted() {
return false;
@@ -107,9 +219,6 @@ public class PostponeBucketWriter implements RecordWriter<KeyValue> {
@Override
public void close() throws Exception {
- if (writer != null) {
- writer.abort();
- writer = null;
- }
+ sinkWriter.close();
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/SinkWriter.java b/paimon-core/src/main/java/org/apache/paimon/utils/SinkWriter.java
new file mode 100644
index 0000000000..c4596a1df4
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/SinkWriter.java
@@ -0,0 +1,234 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.utils;
+
+import org.apache.paimon.compression.CompressOptions;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.disk.RowBuffer;
+import org.apache.paimon.io.BundleRecords;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.memory.MemorySegmentPool;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.types.RowType;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+/** Internal interface to Sink Data from input. */
+public interface SinkWriter<T> {
+
+ boolean write(T data) throws IOException;
+
+ List<DataFileMeta> flush() throws IOException;
+
+ boolean flushMemory() throws IOException;
+
+ long memoryOccupancy();
+
+ void close();
+
+ void setMemoryPool(MemorySegmentPool memoryPool);
+
+ boolean bufferSpillableWriter();
+
+ /**
+ * Directly sink data to file, no memory cache here, use OrcWriter/ParquetWrite/etc directly
+ * write data. May cause out-of-memory.
+ */
+ class DirectSinkWriter<T> implements SinkWriter<T> {
+
+ private final Supplier<RollingFileWriter<T, DataFileMeta>> writerSupplier;
+
+ private RollingFileWriter<T, DataFileMeta> writer;
+
+ public DirectSinkWriter(Supplier<RollingFileWriter<T, DataFileMeta>> writerSupplier) {
+ this.writerSupplier = writerSupplier;
+ }
+
+ @Override
+ public boolean write(T data) throws IOException {
+ if (writer == null) {
+ writer = writerSupplier.get();
+ }
+ writer.write(data);
+ return true;
+ }
+
+ public void writeBundle(BundleRecords bundle) throws IOException {
+ if (writer == null) {
+ writer = writerSupplier.get();
+ }
+ writer.writeBundle(bundle);
+ }
+
+ @Override
+ public List<DataFileMeta> flush() throws IOException {
+ List<DataFileMeta> flushedFiles = new ArrayList<>();
+ if (writer != null) {
+ writer.close();
+ flushedFiles.addAll(writer.result());
+ writer = null;
+ }
+ return flushedFiles;
+ }
+
+ @Override
+ public boolean flushMemory() throws IOException {
+ return false;
+ }
+
+ @Override
+ public long memoryOccupancy() {
+ return 0;
+ }
+
+ @Override
+ public void close() {
+ if (writer != null) {
+ writer.abort();
+ writer = null;
+ }
+ }
+
+ @Override
+ public void setMemoryPool(MemorySegmentPool memoryPool) {
+ // do nothing
+ }
+
+ @Override
+ public boolean bufferSpillableWriter() {
+ return false;
+ }
+ }
+
+ /**
+ * Use buffered writer, segment pooled from segment pool. When spillable, may delay checkpoint
+ * acknowledge time. When non-spillable, may cause too many small files.
+ */
+ class BufferedSinkWriter<T> implements SinkWriter<T> {
+
+ private final Supplier<RollingFileWriter<T, DataFileMeta>> writerSupplier;
+ private final Function<T, InternalRow> toRow;
+ private final Function<InternalRow, T> fromRow;
+ private final IOManager ioManager;
+ private final RowType rowType;
+ private final boolean spillable;
+ private final MemorySize maxDiskSize;
+ private final CompressOptions compression;
+
+ private RowBuffer writeBuffer;
+
+ public BufferedSinkWriter(
+ Supplier<RollingFileWriter<T, DataFileMeta>> writerSupplier,
+ Function<T, InternalRow> toRow,
+ Function<InternalRow, T> fromRow,
+ IOManager ioManager,
+ RowType rowType,
+ boolean spillable,
+ MemorySize maxDiskSize,
+ CompressOptions compression) {
+ this.writerSupplier = writerSupplier;
+ this.toRow = toRow;
+ this.fromRow = fromRow;
+ this.ioManager = ioManager;
+ this.rowType = rowType;
+ this.spillable = spillable;
+ this.maxDiskSize = maxDiskSize;
+ this.compression = compression;
+ }
+
+ public RowBuffer rowBuffer() {
+ return writeBuffer;
+ }
+
+ @Override
+ public boolean write(T data) throws IOException {
+ return writeBuffer.put(toRow.apply(data));
+ }
+
+ @Override
+ public List<DataFileMeta> flush() throws IOException {
+ List<DataFileMeta> flushedFiles = new ArrayList<>();
+ if (writeBuffer != null) {
+ writeBuffer.complete();
+ RollingFileWriter<T, DataFileMeta> writer = writerSupplier.get();
+ IOException exception = null;
+ try (RowBuffer.RowBufferIterator iterator = writeBuffer.newIterator()) {
+ while (iterator.advanceNext()) {
+ writer.write(fromRow.apply(iterator.getRow()));
+ }
+ } catch (IOException e) {
+ exception = e;
+ } finally {
+ if (exception != null) {
+ IOUtils.closeQuietly(writer);
+ // cleanup code that might throw another exception
+ throw exception;
+ }
+ writer.close();
+ }
+ flushedFiles.addAll(writer.result());
+ // reuse writeBuffer
+ writeBuffer.reset();
+ }
+ return flushedFiles;
+ }
+
+ @Override
+ public long memoryOccupancy() {
+ return writeBuffer.memoryOccupancy();
+ }
+
+ @Override
+ public void close() {
+ if (writeBuffer != null) {
+ writeBuffer.reset();
+ writeBuffer = null;
+ }
+ }
+
+ @Override
+ public void setMemoryPool(MemorySegmentPool memoryPool) {
+ writeBuffer =
+ RowBuffer.getBuffer(
+ ioManager,
+ memoryPool,
+ new InternalRowSerializer(rowType),
+ spillable,
+ maxDiskSize,
+ compression);
+ }
+
+ @Override
+ public boolean bufferSpillableWriter() {
+ return spillable;
+ }
+
+ @Override
+ public boolean flushMemory() throws IOException {
+ return writeBuffer.flushMemory();
+ }
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
index 0bf4971f94..35928e2335 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
@@ -21,6 +21,7 @@ package org.apache.paimon.table;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.CoreOptions.ChangelogProducer;
import org.apache.paimon.CoreOptions.LookupLocalFileType;
+import org.apache.paimon.KeyValue;
import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
@@ -34,9 +35,12 @@ import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.AbstractFileStoreWrite;
import org.apache.paimon.operation.FileStoreScan;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
+import org.apache.paimon.postpone.PostponeBucketFileStoreWrite;
+import org.apache.paimon.postpone.PostponeBucketWriter;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.RecordReader;
@@ -54,6 +58,7 @@ import org.apache.paimon.table.sink.InnerTableCommit;
import org.apache.paimon.table.sink.StreamTableCommit;
import org.apache.paimon.table.sink.StreamTableWrite;
import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.table.sink.TableWriteImpl;
import org.apache.paimon.table.sink.WriteSelector;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.InnerTableRead;
@@ -136,6 +141,39 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link PrimaryKeyFileStoreTable}. */
public class PrimaryKeySimpleTableTest extends SimpleTableTestBase {
+ @Test
+ public void testPostponeBucketWithManyPartitions() throws Exception {
+ FileStoreTable table =
+ createFileStoreTable(options -> options.set(BUCKET, BucketMode.POSTPONE_BUCKET));
+
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.withIOManager(new IOManagerImpl(tempDir.toString()));
+ for (int i = 0; i < 100; i++) {
+ write.write(rowData(i, i, (long) i));
+ }
+
+ for (Map<Integer, AbstractFileStoreWrite.WriterContainer<KeyValue>> bucketWriters :
+ ((PostponeBucketFileStoreWrite) ((TableWriteImpl<?>) write).getWrite())
+ .writers()
+ .values()) {
+ for (AbstractFileStoreWrite.WriterContainer<KeyValue> writerContainer :
+ bucketWriters.values()) {
+ PostponeBucketWriter writer = (PostponeBucketWriter) writerContainer.writer;
+ assertThat(writer.useBufferedSinkWriter()).isTrue();
+ }
+ }
+ commit.commit(write.prepareCommit());
+ }
+
+ Snapshot snapshot = table.latestSnapshot().get();
+ ManifestFileMeta manifest =
+ table.manifestListReader().read(snapshot.deltaManifestList()).get(0);
+ List<ManifestEntry> entries = table.manifestFileReader().read(manifest.fileName());
+ assertThat(entries.size()).isEqualTo(100);
+ }
+
@Test
public void testPostponeBucket() throws Exception {
FileStoreTable table =
(paimon) 05/15: [flink] Add support for AvroType enum for debezium format (#5742)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 958e85bac23e5778ce038d54ac1012a4a1c8aa35
Author: jsingh-yelp <jk...@yelp.com>
AuthorDate: Thu Jun 12 21:55:35 2025 -0400
[flink] Add support for AvroType enum for debezium format (#5742)
---
.../cdc/format/debezium/DebeziumSchemaUtils.java | 1 +
.../format/debezium/DebeziumSchemaUtilsTest.java | 153 +++++++++++++++++++++
2 files changed, 154 insertions(+)
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java
index 114344bade..80f99165e6 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java
@@ -475,6 +475,7 @@ public class DebeziumSchemaUtils {
return DataTypes.INT();
case LONG:
return DataTypes.BIGINT();
+ case ENUM:
case STRING:
return DataTypes.STRING();
case RECORD:
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtilsTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtilsTest.java
new file mode 100644
index 0000000000..7381bb4890
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtilsTest.java
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action.cdc.format.debezium;
+
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+
+import org.apache.avro.LogicalTypes;
+import org.apache.avro.Schema;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+/** Test class for DebeziumSchemaUtils. */
+public class DebeziumSchemaUtilsTest {
+ @Test
+ public void testFromDebeziumAvroTypeWithGenericUnion() {
+ // Create a union schema with multiple non-null types
+ Schema stringSchema = Schema.create(Schema.Type.STRING);
+ Schema intSchema = Schema.create(Schema.Type.INT);
+ Schema unionSchema = Schema.createUnion(Arrays.asList(stringSchema, intSchema));
+
+ // Test that an exception is thrown for generic unions
+ UnsupportedOperationException exception =
+ Assertions.assertThrows(
+ UnsupportedOperationException.class,
+ () -> DebeziumSchemaUtils.avroToPaimonDataType(unionSchema));
+
+ Assertions.assertEquals("Generic unions are not supported", exception.getMessage());
+ }
+
+ @Test
+ public void testFromDebeziumAvroTypeWithLogicalTypes() {
+ // Test date logical type
+ Schema dateSchema = Schema.create(Schema.Type.INT);
+ LogicalTypes.date().addToSchema(dateSchema);
+ DataType dateType = DebeziumSchemaUtils.avroToPaimonDataType(dateSchema);
+ Assertions.assertEquals(DataTypes.DATE(), dateType);
+
+ // Test decimal logical type
+ Schema decimalSchema = Schema.create(Schema.Type.BYTES);
+ LogicalTypes.decimal(10, 2).addToSchema(decimalSchema);
+ DataType decimalType = DebeziumSchemaUtils.avroToPaimonDataType(decimalSchema);
+ Assertions.assertEquals(DataTypes.DECIMAL(10, 2), decimalType);
+
+ // Test timestamp-millis logical type
+ Schema timestampMillisSchema = Schema.create(Schema.Type.LONG);
+ LogicalTypes.timestampMillis().addToSchema(timestampMillisSchema);
+ DataType timestampMillisType =
+ DebeziumSchemaUtils.avroToPaimonDataType(timestampMillisSchema);
+ Assertions.assertEquals(DataTypes.TIMESTAMP_MILLIS(), timestampMillisType);
+
+ // Test timestamp-micros logical type
+ Schema timestampMicrosSchema = Schema.create(Schema.Type.LONG);
+ LogicalTypes.timestampMicros().addToSchema(timestampMicrosSchema);
+ DataType timestampMicrosType =
+ DebeziumSchemaUtils.avroToPaimonDataType(timestampMicrosSchema);
+ Assertions.assertEquals(DataTypes.TIMESTAMP(), timestampMicrosType);
+ }
+
+ @Test
+ public void testFromDebeziumAvroTypeWithPrimitiveTypes() {
+ // Test boolean type
+ Schema booleanSchema = Schema.create(Schema.Type.BOOLEAN);
+ DataType booleanType = DebeziumSchemaUtils.avroToPaimonDataType(booleanSchema);
+ Assertions.assertEquals(DataTypes.BOOLEAN(), booleanType);
+
+ // Test int type
+ Schema intSchema = Schema.create(Schema.Type.INT);
+ DataType intType = DebeziumSchemaUtils.avroToPaimonDataType(intSchema);
+ Assertions.assertEquals(DataTypes.INT(), intType);
+
+ // Test long type
+ Schema longSchema = Schema.create(Schema.Type.LONG);
+ DataType longType = DebeziumSchemaUtils.avroToPaimonDataType(longSchema);
+ Assertions.assertEquals(DataTypes.BIGINT(), longType);
+
+ // Test float type
+ Schema floatSchema = Schema.create(Schema.Type.FLOAT);
+ DataType floatType = DebeziumSchemaUtils.avroToPaimonDataType(floatSchema);
+ Assertions.assertEquals(DataTypes.FLOAT(), floatType);
+
+ // Test double type
+ Schema doubleSchema = Schema.create(Schema.Type.DOUBLE);
+ DataType doubleType = DebeziumSchemaUtils.avroToPaimonDataType(doubleSchema);
+ Assertions.assertEquals(DataTypes.DOUBLE(), doubleType);
+
+ // Test enum type
+ Schema enumSchema =
+ Schema.createEnum("TestEnum", null, null, Arrays.asList("VALUE1", "VALUE2"));
+ DataType enumType = DebeziumSchemaUtils.avroToPaimonDataType(enumSchema);
+ Assertions.assertEquals(DataTypes.STRING(), enumType);
+
+ // Test string type
+ Schema stringSchema = Schema.create(Schema.Type.STRING);
+ DataType stringType = DebeziumSchemaUtils.avroToPaimonDataType(stringSchema);
+ Assertions.assertEquals(DataTypes.STRING(), stringType);
+
+ // Test bytes type
+ Schema bytesSchema = Schema.create(Schema.Type.BYTES);
+ DataType bytesType = DebeziumSchemaUtils.avroToPaimonDataType(bytesSchema);
+ Assertions.assertEquals(DataTypes.BYTES(), bytesType);
+ }
+
+ @Test
+ public void testFromDebeziumAvroTypeWithComplexTypes() {
+ // Test array type
+ Schema stringSchema = Schema.create(Schema.Type.STRING);
+ Schema arraySchema = Schema.createArray(stringSchema);
+ DataType arrayType = DebeziumSchemaUtils.avroToPaimonDataType(arraySchema);
+ Assertions.assertEquals(DataTypes.ARRAY(DataTypes.STRING()), arrayType);
+
+ // Test map type
+ Schema mapSchema = Schema.createMap(stringSchema);
+ DataType mapType = DebeziumSchemaUtils.avroToPaimonDataType(mapSchema);
+ Assertions.assertEquals(DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()), mapType);
+
+ // Test record type
+ Schema recordSchema = Schema.createRecord("TestRecord", null, null, false);
+ List<Schema.Field> fields =
+ Arrays.asList(
+ new Schema.Field("field1", stringSchema, null, null),
+ new Schema.Field("field2", Schema.create(Schema.Type.INT), null, null));
+ recordSchema.setFields(fields);
+ DataType recordType = DebeziumSchemaUtils.avroToPaimonDataType(recordSchema);
+
+ DataField[] expectedFields =
+ new DataField[] {
+ DataTypes.FIELD(0, "field1", DataTypes.STRING(), null),
+ DataTypes.FIELD(1, "field2", DataTypes.INT(), null)
+ };
+ Assertions.assertEquals(DataTypes.ROW(expectedFields), recordType);
+ }
+}
(paimon) 10/15: [fs] Add a requiredOptions method to COSNLoader (#5750)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 08050cf2bc53be060499278d1985cbcb63b98b40
Author: andybj0228 <an...@126.com>
AuthorDate: Sun Jun 15 09:29:00 2025 +0800
[fs] Add a requiredOptions method to COSNLoader (#5750)
---
.../src/main/java/org/apache/paimon/cosn/COSNLoader.java | 12 ++++++++++++
1 file changed, 12 insertions(+)
diff --git a/paimon-filesystems/paimon-cosn/src/main/java/org/apache/paimon/cosn/COSNLoader.java b/paimon-filesystems/paimon-cosn/src/main/java/org/apache/paimon/cosn/COSNLoader.java
index 6d86fdc179..35b6c1ff28 100644
--- a/paimon-filesystems/paimon-cosn/src/main/java/org/apache/paimon/cosn/COSNLoader.java
+++ b/paimon-filesystems/paimon-cosn/src/main/java/org/apache/paimon/cosn/COSNLoader.java
@@ -25,6 +25,9 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PluginFileIO;
import org.apache.paimon.plugin.PluginLoader;
+import java.util.ArrayList;
+import java.util.List;
+
/** A {@link PluginLoader} to load cosn. */
public class COSNLoader implements FileIOLoader {
private static final String COSN_JAR = "paimon-plugin-cosn";
@@ -45,6 +48,15 @@ public class COSNLoader implements FileIOLoader {
return "cosn";
}
+ @Override
+ public List<String[]> requiredOptions() {
+ List<String[]> options = new ArrayList<>();
+ options.add(new String[] {"fs.cosn.bucket.region"});
+ options.add(new String[] {"fs.cosn.userinfo.secretId"});
+ options.add(new String[] {"fs.cosn.userinfo.secretKey"});
+ return options;
+ }
+
@Override
public FileIO load(Path path) {
return new COSNPluginFileIO();
(paimon) 02/15: [flink] Add 'scan.dedicated-split-generation' to Flink Batch Source (#5715)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 95322af7480021215efc5c953c301842c08163f1
Author: wangwj <ho...@gmail.com>
AuthorDate: Thu Jun 12 15:44:56 2025 +0800
[flink] Add 'scan.dedicated-split-generation' to Flink Batch Source (#5715)
---
docs/content/append-table/query-performance.md | 11 ++++
.../content/primary-key-table/query-performance.md | 12 +++++
.../generated/flink_connector_configuration.html | 18 ++++---
.../apache/paimon/flink/FlinkConnectorOptions.java | 15 ++++--
.../postpone/PostponeBucketCompactSplitSource.java | 2 +-
.../flink/source/ContinuousFileStoreSource.java | 2 +-
.../paimon/flink/source/FlinkSourceBuilder.java | 16 +++---
.../align/AlignedContinuousFileStoreSource.java | 2 +-
.../flink/source/operator/MonitorSource.java | 38 ++++++++++----
.../paimon/flink/source/operator/ReadOperator.java | 25 ++++++++-
.../apache/paimon/flink/utils/TableScanUtils.java | 16 ++++--
.../apache/paimon/flink/BatchFileStoreITCase.java | 38 ++++++++++++++
.../flink/source/operator/OperatorSourceTest.java | 60 +++++++++++++++++++---
13 files changed, 216 insertions(+), 39 deletions(-)
diff --git a/docs/content/append-table/query-performance.md b/docs/content/append-table/query-performance.md
index e2128bbd89..d488e4e999 100644
--- a/docs/content/append-table/query-performance.md
+++ b/docs/content/append-table/query-performance.md
@@ -75,3 +75,14 @@ we use the procedure, you should config appropriate configurations in target tab
`file-index.<filter-type>.columns` to the table.
How to invoke: see [flink procedures]({{< ref "flink/procedures#procedures" >}})
+
+## Dedicated Split Generation
+When Paimon table snapshots contain large amount of source splits, Flink jobs reading from this table might endure long initialization time or even OOM in JobManagers. In this case, you can configure `'scan.dedicated-split-generation' = 'true'` to avoid such problem. This option would enable executing the source split generation process in a dedicated subtask that runs on TaskManager, instead of in the source coordinator on the JobManager.
+
+Note that this feature could have some side effects on your Flink jobs. For example:
+
+1. It will change the DAG of the flink job, thus breaking checkpoint compatibility if enabled on an existing job.
+2. It may lead to the Flink AdaptiveBatchScheduler inferring a small parallelism for the source reader operator. you can configure `scan.infer-parallelism` to avoid this possible drawback.
+3. The failover strategy of the Flink job would be forced into global failover instead of regional failover, given that the dedicated source split generation task would be connected to all downstream subtasks.
+
+So please make sure these side effects are acceptable to you before enabling it.
diff --git a/docs/content/primary-key-table/query-performance.md b/docs/content/primary-key-table/query-performance.md
index d27c4868fe..ed0e357fc6 100644
--- a/docs/content/primary-key-table/query-performance.md
+++ b/docs/content/primary-key-table/query-performance.md
@@ -73,3 +73,15 @@ we use the procedure, you should config appropriate configurations in target tab
`file-index.<filter-type>.columns` to the table.
How to invoke: see [flink procedures]({{< ref "flink/procedures#procedures" >}})
+
+## Dedicated Split Generation
+When Paimon table snapshots contain large amount of source splits, Flink jobs reading from this table might endure long initialization time or even OOM in JobManagers. In this case, you can configure `'scan.dedicated-split-generation' = 'true'` to avoid such problem. This option would enable executing the source split generation process in a dedicated subtask that runs on TaskManager, instead of in the source coordinator on the JobManager.
+
+Note that this feature could have some side effects on your Flink jobs. For example:
+
+1. It will change the DAG of the flink job, thus breaking checkpoint compatibility if enabled on an existing job.
+2. It may lead to the Flink AdaptiveBatchScheduler inferring a small parallelism for the source reader operator. you can configure `scan.infer-parallelism` to avoid this possible drawback.
+3. The failover strategy of the Flink job would be forced into global failover instead of regional failover, given that the dedicated source split generation task would be connected to all downstream subtasks.
+
+So please make sure these side effects are acceptable to you before enabling it.
+
diff --git a/docs/layouts/shortcodes/generated/flink_connector_configuration.html b/docs/layouts/shortcodes/generated/flink_connector_configuration.html
index 8b80512fc4..1a67d4e60f 100644
--- a/docs/layouts/shortcodes/generated/flink_connector_configuration.html
+++ b/docs/layouts/shortcodes/generated/flink_connector_configuration.html
@@ -122,12 +122,24 @@ under the License.
<td>Boolean</td>
<td>If true, it will add a compact coordinator and worker operator after the writer operator,in order to compact several changelog files (for primary key tables) or newly created data files (for unaware bucket tables) from the same partition into large ones, which can decrease the number of small files.</td>
</tr>
+ <tr>
+ <td><h5>read.shuffle-bucket-with-partition</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether shuffle by partition and bucket when read.</td>
+ </tr>
<tr>
<td><h5>scan.bounded</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>Boolean</td>
<td>Bounded mode for Paimon consumer. By default, Paimon automatically selects bounded mode based on the mode of the Flink job.</td>
</tr>
+ <tr>
+ <td><h5>scan.dedicated-split-generation</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If true, the split generation process would be performed during runtime on a Flink task, instead of on the JobManager during initialization phase.</td>
+ </tr>
<tr>
<td><h5>scan.infer-parallelism</h5></td>
<td style="word-wrap: break-word;">true</td>
@@ -314,12 +326,6 @@ under the License.
<td>String</td>
<td>Set the uid suffix for the source operators. After setting, the uid format is ${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. If the uid suffix is not set, flink will automatically generate the operator uid, which may be incompatible when the topology changes.</td>
</tr>
- <tr>
- <td><h5>streaming-read.shuffle-bucket-with-partition</h5></td>
- <td style="word-wrap: break-word;">true</td>
- <td>Boolean</td>
- <td>Whether shuffle by partition and bucket when streaming read.</td>
- </tr>
<tr>
<td><h5>unaware-bucket.compaction.parallelism</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
index 9c0e9d6191..ed8d9a7ac8 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
@@ -224,12 +224,12 @@ public class FlinkConnectorOptions {
+ " Note: This is dangerous and is likely to cause data errors if downstream"
+ " is used to calculate aggregation and the input is not complete changelog.");
- public static final ConfigOption<Boolean> STREAMING_READ_SHUFFLE_BUCKET_WITH_PARTITION =
- key("streaming-read.shuffle-bucket-with-partition")
+ public static final ConfigOption<Boolean> READ_SHUFFLE_BUCKET_WITH_PARTITION =
+ key("read.shuffle-bucket-with-partition")
.booleanType()
.defaultValue(true)
- .withDescription(
- "Whether shuffle by partition and bucket when streaming read.");
+ .withFallbackKeys("streaming-read.shuffle-bucket-with-partition")
+ .withDescription("Whether shuffle by partition and bucket when read.");
/**
* Weight of writer buffer in managed memory, Flink will compute the memory size for writer
@@ -508,6 +508,13 @@ public class FlinkConnectorOptions {
.withDescription(
"Bucket number for the partitions compacted for the first time in postpone bucket tables.");
+ public static final ConfigOption<Boolean> SCAN_DEDICATED_SPLIT_GENERATION =
+ key("scan.dedicated-split-generation")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "If true, the split generation process would be performed during runtime on a Flink task, instead of on the JobManager during initialization phase.");
+
public static List<ConfigOption<?>> getOptions() {
final Field[] fields = FlinkConnectorOptions.class.getFields();
final List<ConfigOption<?>> list = new ArrayList<>(fields.length);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/postpone/PostponeBucketCompactSplitSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/postpone/PostponeBucketCompactSplitSource.java
index 53bc7a50a7..833eff8120 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/postpone/PostponeBucketCompactSplitSource.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/postpone/PostponeBucketCompactSplitSource.java
@@ -155,7 +155,7 @@ public class PostponeBucketCompactSplitSource extends AbstractNonCoordinatedSour
table.fullName(), partitionSpec),
InternalTypeInfo.of(
LogicalTypeConversion.toLogicalType(table.rowType())),
- new ReadOperator(table::newRead, null)),
+ new ReadOperator(table::newRead, null, null)),
source.forward()
.transform(
"Remove new files",
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileStoreSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileStoreSource.java
index 3587313748..2c1273e811 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileStoreSource.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileStoreSource.java
@@ -111,7 +111,7 @@ public class ContinuousFileStoreSource extends FlinkSource {
scan,
unawareBucket,
options.get(CoreOptions.SCAN_MAX_SPLITS_PER_TASK),
- options.get(FlinkConnectorOptions.STREAMING_READ_SHUFFLE_BUCKET_WITH_PARTITION),
+ options.get(FlinkConnectorOptions.READ_SHUFFLE_BUCKET_WITH_PARTITION),
options.get(FlinkConnectorOptions.SCAN_MAX_SNAPSHOT_COUNT));
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
index feb9176c85..ac776b7cd3 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
@@ -294,6 +294,9 @@ public class FlinkSourceBuilder {
}
if (sourceBounded) {
+ if (conf.get(FlinkConnectorOptions.SCAN_DEDICATED_SPLIT_GENERATION)) {
+ return buildContinuousStreamOperator(true);
+ }
return buildStaticFileSource();
}
TableScanUtils.streamingReadingValidate(table);
@@ -325,16 +328,16 @@ public class FlinkSourceBuilder {
} else if (conf.contains(CoreOptions.CONSUMER_ID)
&& conf.get(CoreOptions.CONSUMER_CONSISTENCY_MODE)
== CoreOptions.ConsumerMode.EXACTLY_ONCE) {
- return buildContinuousStreamOperator();
+ return buildContinuousStreamOperator(false);
} else {
return buildContinuousFileSource();
}
}
}
- private DataStream<RowData> buildContinuousStreamOperator() {
+ private DataStream<RowData> buildContinuousStreamOperator(boolean isBounded) {
DataStream<RowData> dataStream;
- if (limit != null) {
+ if (limit != null && !isBounded) {
throw new IllegalArgumentException(
"Cannot limit streaming source, please use batch execution mode.");
}
@@ -346,10 +349,11 @@ public class FlinkSourceBuilder {
createReadBuilder(projectedRowType()),
conf.get(CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL).toMillis(),
watermarkStrategy == null,
- conf.get(
- FlinkConnectorOptions.STREAMING_READ_SHUFFLE_BUCKET_WITH_PARTITION),
+ conf.get(FlinkConnectorOptions.READ_SHUFFLE_BUCKET_WITH_PARTITION),
unawareBucket,
- outerProject());
+ outerProject(),
+ isBounded,
+ limit);
if (parallelism != null) {
dataStream.getTransformation().setParallelism(parallelism);
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileStoreSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileStoreSource.java
index 29ff63a31c..59845901ff 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileStoreSource.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileStoreSource.java
@@ -84,7 +84,7 @@ public class AlignedContinuousFileStoreSource extends ContinuousFileStoreSource
unawareBucket,
options.get(FlinkConnectorOptions.SOURCE_CHECKPOINT_ALIGN_TIMEOUT).toMillis(),
options.get(CoreOptions.SCAN_MAX_SPLITS_PER_TASK),
- options.get(FlinkConnectorOptions.STREAMING_READ_SHUFFLE_BUCKET_WITH_PARTITION),
+ options.get(FlinkConnectorOptions.READ_SHUFFLE_BUCKET_WITH_PARTITION),
options.get(FlinkConnectorOptions.SCAN_MAX_SNAPSHOT_COUNT));
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
index 228d0d5a44..d664b526ff 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
@@ -30,6 +30,7 @@ import org.apache.paimon.table.source.EndOfScanException;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.StreamTableScan;
+import org.apache.paimon.table.source.TableScan;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
@@ -48,6 +49,8 @@ import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.List;
import java.util.NavigableMap;
@@ -59,20 +62,22 @@ import java.util.TreeMap;
*
* <ol>
* <li>Monitoring snapshots of the Paimon table.
- * <li>Creating the {@link Split splits} corresponding to the incremental files
+ * <li>Creating the {@link Split splits} corresponding to the DateFiles
* <li>Assigning them to downstream tasks for further processing.
* </ol>
*
* <p>The splits to be read are forwarded to the downstream {@link ReadOperator} which can have
* parallelism greater than one.
*
- * <p>Currently, there are two features that rely on this monitor:
+ * <p>Currently, there are three features that rely on this monitor:
*
* <ol>
* <li>Consumer-id: rely on this source to do aligned snapshot consumption, and ensure that all
* data in a snapshot is consumed within each checkpoint.
* <li>Snapshot-watermark: when there is no watermark definition, the default Paimon table will
* pass the watermark recorded in the snapshot.
+ * <li>Optimize-coordinator-memory: rely on this source to get splits on a single task, this can
+ * reduce the memory pressure of source coordinator.
* </ol>
*/
public class MonitorSource extends AbstractNonCoordinatedSource<Split> {
@@ -84,17 +89,22 @@ public class MonitorSource extends AbstractNonCoordinatedSource<Split> {
private final ReadBuilder readBuilder;
private final long monitorInterval;
private final boolean emitSnapshotWatermark;
+ private final boolean isBounded;
public MonitorSource(
- ReadBuilder readBuilder, long monitorInterval, boolean emitSnapshotWatermark) {
+ ReadBuilder readBuilder,
+ long monitorInterval,
+ boolean emitSnapshotWatermark,
+ boolean isBounded) {
this.readBuilder = readBuilder;
this.monitorInterval = monitorInterval;
this.emitSnapshotWatermark = emitSnapshotWatermark;
+ this.isBounded = isBounded;
}
@Override
public Boundedness getBoundedness() {
- return Boundedness.CONTINUOUS_UNBOUNDED;
+ return isBounded ? Boundedness.BOUNDED : Boundedness.CONTINUOUS_UNBOUNDED;
}
@Override
@@ -108,6 +118,7 @@ public class MonitorSource extends AbstractNonCoordinatedSource<Split> {
private static final String NEXT_SNAPSHOT_STATE = "NSS";
private final StreamTableScan scan = readBuilder.newStreamScan();
+ private final TableScan batchScan = readBuilder.newScan();
private final SplitListState<Long> checkpointState =
new SplitListState<>(CHECKPOINT_STATE, x -> Long.toString(x), Long::parseLong);
private final SplitListState<Tuple2<Long, Long>> nextSnapshotState =
@@ -178,11 +189,11 @@ public class MonitorSource extends AbstractNonCoordinatedSource<Split> {
public InputStatus pollNext(ReaderOutput<Split> readerOutput) throws Exception {
boolean isEmpty;
try {
- List<Split> splits = scan.plan().splits();
+ List<Split> splits = isBounded ? batchScan.plan().splits() : scan.plan().splits();
isEmpty = splits.isEmpty();
splits.forEach(readerOutput::collect);
- if (emitSnapshotWatermark) {
+ if (emitSnapshotWatermark && !isBounded) {
Long watermark = scan.watermark();
if (watermark != null) {
readerOutput.emitWatermark(new Watermark(watermark));
@@ -193,6 +204,10 @@ public class MonitorSource extends AbstractNonCoordinatedSource<Split> {
return InputStatus.END_OF_INPUT;
}
+ if (isBounded) {
+ return InputStatus.END_OF_INPUT;
+ }
+
if (isEmpty) {
Thread.sleep(monitorInterval);
}
@@ -209,11 +224,16 @@ public class MonitorSource extends AbstractNonCoordinatedSource<Split> {
boolean emitSnapshotWatermark,
boolean shuffleBucketWithPartition,
boolean unawareBucket,
- NestedProjectedRowData nestedProjectedRowData) {
+ NestedProjectedRowData nestedProjectedRowData,
+ boolean isBounded,
+ @Nullable Long limit) {
SingleOutputStreamOperator<Split> singleOutputStreamOperator =
env.fromSource(
new MonitorSource(
- readBuilder, monitorInterval, emitSnapshotWatermark),
+ readBuilder,
+ monitorInterval,
+ emitSnapshotWatermark,
+ isBounded),
WatermarkStrategy.noWatermarks(),
name + "-Monitor",
new JavaTypeInfo<>(Split.class))
@@ -228,7 +248,7 @@ public class MonitorSource extends AbstractNonCoordinatedSource<Split> {
return sourceDataStream.transform(
name + "-Reader",
typeInfo,
- new ReadOperator(readBuilder::newRead, nestedProjectedRowData));
+ new ReadOperator(readBuilder::newRead, nestedProjectedRowData, limit));
}
private static DataStream<Split> shuffleUnawareBucket(
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
index 685b324088..a9b9767041 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
@@ -36,6 +36,8 @@ import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
@@ -47,6 +49,8 @@ import javax.annotation.Nullable;
public class ReadOperator extends AbstractStreamOperator<RowData>
implements OneInputStreamOperator<Split, RowData> {
+ private static final Logger LOG = LoggerFactory.getLogger(ReadOperator.class);
+
private static final long serialVersionUID = 2L;
private final SerializableSupplier<TableRead> readSupplier;
@@ -64,12 +68,15 @@ public class ReadOperator extends AbstractStreamOperator<RowData>
private transient long emitEventTimeLag = FileStoreSourceReaderMetrics.UNDEFINED;
private transient long idleStartTime = FileStoreSourceReaderMetrics.ACTIVE;
private transient Counter numRecordsIn;
+ @Nullable private final Long limit;
public ReadOperator(
SerializableSupplier<TableRead> readSupplier,
- @Nullable NestedProjectedRowData nestedProjectedRowData) {
+ @Nullable NestedProjectedRowData nestedProjectedRowData,
+ @Nullable Long limit) {
this.readSupplier = readSupplier;
this.nestedProjectedRowData = nestedProjectedRowData;
+ this.limit = limit;
}
@Override
@@ -98,6 +105,10 @@ public class ReadOperator extends AbstractStreamOperator<RowData>
@Override
public void processElement(StreamRecord<Split> record) throws Exception {
+ if (reachLimit()) {
+ return;
+ }
+
Split split = record.getValue();
// update metric when reading a new split
long eventTime =
@@ -122,6 +133,10 @@ public class ReadOperator extends AbstractStreamOperator<RowData>
numRecordsIn.inc();
}
+ if (reachLimit()) {
+ return;
+ }
+
reuseRow.replace(iterator.next());
if (nestedProjectedRowData == null) {
reuseRecord.replace(reuseRow);
@@ -144,6 +159,14 @@ public class ReadOperator extends AbstractStreamOperator<RowData>
}
}
+ private boolean reachLimit() {
+ if (limit != null && numRecordsIn.getCount() > limit) {
+ LOG.info("Reader {} reach the limit record {}.", this, limit);
+ return true;
+ }
+ return false;
+ }
+
private void idlingStarted() {
if (!isIdling()) {
idleStartTime = System.currentTimeMillis();
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java
index 30b7bbdd5d..ca901ad1bf 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java
@@ -30,12 +30,14 @@ import java.util.HashSet;
import java.util.Optional;
import java.util.Set;
+import static org.apache.paimon.flink.FlinkConnectorOptions.SCAN_DEDICATED_SPLIT_GENERATION;
+
/** Utility methods for {@link TableScan}, such as validating. */
public class TableScanUtils {
public static void streamingReadingValidate(Table table) {
- CoreOptions options = CoreOptions.fromMap(table.options());
- CoreOptions.MergeEngine mergeEngine = options.mergeEngine();
+ CoreOptions coreOptions = CoreOptions.fromMap(table.options());
+ CoreOptions.MergeEngine mergeEngine = coreOptions.mergeEngine();
HashMap<CoreOptions.MergeEngine, String> mergeEngineDesc =
new HashMap<CoreOptions.MergeEngine, String>() {
{
@@ -45,7 +47,7 @@ public class TableScanUtils {
}
};
if (table.primaryKeys().size() > 0 && mergeEngineDesc.containsKey(mergeEngine)) {
- if (options.changelogProducer() == CoreOptions.ChangelogProducer.NONE) {
+ if (coreOptions.changelogProducer() == CoreOptions.ChangelogProducer.NONE) {
throw new RuntimeException(
mergeEngineDesc.get(mergeEngine)
+ " streaming reading is not supported. You can use "
@@ -53,6 +55,14 @@ public class TableScanUtils {
+ "('input' changelog producer is also supported, but only returns input records.)");
}
}
+
+ Options options = Options.fromMap(table.options());
+ if (options.get(SCAN_DEDICATED_SPLIT_GENERATION)) {
+ throw new RuntimeException(
+ "The option "
+ + SCAN_DEDICATED_SPLIT_GENERATION.key()
+ + " can only used in batch mode.");
+ }
}
/** Get snapshot id from {@link FileStoreSourceSplit}. */
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
index 2b846b68bc..e431b69841 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
@@ -815,4 +815,42 @@ public class BatchFileStoreITCase extends CatalogITCaseBase {
assertThat(sql("SELECT rowkind, b FROM `test_table$binlog`"))
.containsExactly(Row.of("+I", new String[] {"A"}));
}
+
+ @Test
+ public void testBatchReadSourceWithSnapshot() {
+ batchSql("INSERT INTO T VALUES (1, 11, 111), (2, 22, 222), (3, 33, 333), (4, 44, 444)");
+ assertThat(
+ batchSql(
+ "SELECT * FROM T /*+ OPTIONS('scan.snapshot-id'='1', 'scan.dedicated-split-generation'='true') */"))
+ .containsExactlyInAnyOrder(
+ Row.of(1, 11, 111),
+ Row.of(2, 22, 222),
+ Row.of(3, 33, 333),
+ Row.of(4, 44, 444));
+
+ batchSql("INSERT INTO T VALUES (5, 55, 555), (6, 66, 666)");
+ assertThat(
+ batchSql(
+ "SELECT * FROM T /*+ OPTIONS('scan.dedicated-split-generation'='true') */"))
+ .containsExactlyInAnyOrder(
+ Row.of(1, 11, 111),
+ Row.of(2, 22, 222),
+ Row.of(3, 33, 333),
+ Row.of(4, 44, 444),
+ Row.of(5, 55, 555),
+ Row.of(6, 66, 666));
+
+ assertThat(
+ batchSql(
+ "SELECT * FROM T /*+ OPTIONS('scan.dedicated-split-generation'='true') */ limit 2"))
+ .containsExactlyInAnyOrder(Row.of(1, 11, 111), Row.of(2, 22, 222));
+ }
+
+ @Test
+ public void testBatchReadSourceWithoutSnapshot() {
+ assertThat(
+ batchSql(
+ "SELECT * FROM T /*+ OPTIONS('scan.dedicated-split-generation'='true') */"))
+ .hasSize(0);
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
index 0d14b60e0c..e4ab4ec157 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
@@ -122,13 +122,26 @@ public class OperatorSourceTest {
return result;
}
+ @Test
+ public void testMonitorSourceWhenIsBoundedIsTrue() throws Exception {
+ MonitorSource source = new MonitorSource(table.newReadBuilder(), 10, false, true);
+ TestingSourceOperator<Split> operator =
+ (TestingSourceOperator<Split>)
+ TestingSourceOperator.createTestOperator(
+ source.createReader(null), WatermarkStrategy.noWatermarks(), false);
+ AbstractStreamOperatorTestHarness<Split> testHarness =
+ new AbstractStreamOperatorTestHarness<>(operator, 1, 1, 0);
+ testHarness.open();
+ testReadSplit(operator, () -> 1, 1, 1, 1);
+ }
+
@Test
public void testMonitorSource() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 1. run first
OperatorSubtaskState snapshot;
{
- MonitorSource source = new MonitorSource(table.newReadBuilder(), 10, false);
+ MonitorSource source = new MonitorSource(table.newReadBuilder(), 10, false, false);
TestingSourceOperator<Split> operator =
(TestingSourceOperator<Split>)
TestingSourceOperator.createTestOperator(
@@ -143,7 +156,7 @@ public class OperatorSourceTest {
// 2. restore from state
{
- MonitorSource sourceCopy1 = new MonitorSource(table.newReadBuilder(), 10, false);
+ MonitorSource sourceCopy1 = new MonitorSource(table.newReadBuilder(), 10, false, false);
TestingSourceOperator<Split> operatorCopy1 =
(TestingSourceOperator<Split>)
TestingSourceOperator.createTestOperator(
@@ -168,7 +181,7 @@ public class OperatorSourceTest {
// 3. restore from consumer id
{
- MonitorSource sourceCopy2 = new MonitorSource(table.newReadBuilder(), 10, false);
+ MonitorSource sourceCopy2 = new MonitorSource(table.newReadBuilder(), 10, false, false);
TestingSourceOperator<Split> operatorCopy2 =
(TestingSourceOperator<Split>)
TestingSourceOperator.createTestOperator(
@@ -184,7 +197,8 @@ public class OperatorSourceTest {
@Test
public void testReadOperator() throws Exception {
- ReadOperator readOperator = new ReadOperator(() -> table.newReadBuilder().newRead(), null);
+ ReadOperator readOperator =
+ new ReadOperator(() -> table.newReadBuilder().newRead(), null, null);
OneInputStreamOperatorTestHarness<Split, RowData> harness =
new OneInputStreamOperatorTestHarness<>(readOperator);
harness.setup(
@@ -204,9 +218,41 @@ public class OperatorSourceTest {
new StreamRecord<>(GenericRowData.of(2, 2, 2)));
}
+ @Test
+ public void testReadOperatorWithLimit() throws Exception {
+ ReadOperator readOperator =
+ new ReadOperator(() -> table.newReadBuilder().newRead(), null, 2L);
+ OneInputStreamOperatorTestHarness<Split, RowData> harness =
+ new OneInputStreamOperatorTestHarness<>(readOperator);
+ harness.setup(
+ InternalSerializers.create(
+ RowType.of(new IntType(), new IntType(), new IntType())));
+ writeToTable(1, 1, 1);
+ writeToTable(2, 2, 2);
+ writeToTable(3, 3, 3);
+ writeToTable(4, 4, 4);
+ List<Split> splits = table.newReadBuilder().newScan().plan().splits();
+ harness.open();
+ for (Split split : splits) {
+ harness.processElement(new StreamRecord<>(split));
+ }
+ ArrayList<Object> values = new ArrayList<>(harness.getOutput());
+
+ // In ReadOperator each Split is already counted as one input record. But in this case it
+ // will not happen.
+ // So in this case the result values's size if 3 even if the limit is 2.
+ // The IT case see BatchFileStoreITCase#testBatchReadSourceWithSnapshot.
+ assertThat(values)
+ .containsExactlyInAnyOrder(
+ new StreamRecord<>(GenericRowData.of(1, 1, 1)),
+ new StreamRecord<>(GenericRowData.of(2, 2, 2)),
+ new StreamRecord<>(GenericRowData.of(3, 3, 3)));
+ }
+
@Test
public void testReadOperatorMetricsRegisterAndUpdate() throws Exception {
- ReadOperator readOperator = new ReadOperator(() -> table.newReadBuilder().newRead(), null);
+ ReadOperator readOperator =
+ new ReadOperator(() -> table.newReadBuilder().newRead(), null, null);
OneInputStreamOperatorTestHarness<Split, RowData> harness =
new OneInputStreamOperatorTestHarness<>(readOperator);
harness.setup(
@@ -305,6 +351,8 @@ public class OperatorSourceTest {
public void emitWatermark(WatermarkEvent watermarkEvent) {}
};
+ writeToTable(a, b, c);
+
AtomicBoolean isRunning = new AtomicBoolean(true);
Thread runner =
new Thread(
@@ -320,8 +368,6 @@ public class OperatorSourceTest {
});
runner.start();
- writeToTable(a, b, c);
-
Split split = queue.poll(1, TimeUnit.MINUTES);
assertThat(readSplit(split)).containsExactlyInAnyOrder(Arrays.asList(a, b, c));
(paimon) 15/15: [flink] Introduce 'memory' type for 'lookup.cache' to in-memory cache (#5748)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit b99bf0b623dd2999ba4d268b99e44a9e8071026b
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Jun 16 14:02:03 2025 +0800
[flink] Introduce 'memory' type for 'lookup.cache' to in-memory cache (#5748)
---
.../generated/flink_connector_configuration.html | 2 +-
.../paimon/crosspartition/GlobalIndexAssigner.java | 16 ++--
.../java/org/apache/paimon/lookup/BulkLoader.java | 101 +--------------------
.../java/org/apache/paimon/lookup/ByteArray.java | 61 +++++++++++++
.../org/apache/paimon/lookup/ListBulkLoader.java | 27 ++++++
.../java/org/apache/paimon/lookup/ListState.java | 32 +++++++
.../java/org/apache/paimon/lookup/SetState.java | 35 +++++++
.../main/java/org/apache/paimon/lookup/State.java | 31 +++++++
.../org/apache/paimon/lookup/StateFactory.java | 51 +++++++++++
.../org/apache/paimon/lookup/ValueBulkLoader.java | 25 +++++
.../java/org/apache/paimon/lookup/ValueState.java | 36 ++++++++
.../paimon/lookup/memory/InMemoryListState.java | 77 ++++++++++++++++
.../paimon/lookup/memory/InMemorySetState.java | 69 ++++++++++++++
.../apache/paimon/lookup/memory/InMemoryState.java | 64 +++++++++++++
.../paimon/lookup/memory/InMemoryStateFactory.java | 66 ++++++++++++++
.../paimon/lookup/memory/InMemoryValueState.java | 76 ++++++++++++++++
.../RocksDBBulkLoader.java} | 33 +++++--
.../lookup/{ => rocksdb}/RocksDBListState.java | 17 +---
.../lookup/{ => rocksdb}/RocksDBOptions.java | 2 +-
.../lookup/{ => rocksdb}/RocksDBSetState.java | 10 +-
.../paimon/lookup/{ => rocksdb}/RocksDBState.java | 52 ++++-------
.../lookup/{ => rocksdb}/RocksDBStateFactory.java | 14 ++-
.../lookup/{ => rocksdb}/RocksDBValueState.java | 21 ++---
.../configuration/ConfigOptionsDocGenerator.java | 2 +-
.../apache/paimon/flink/FlinkConnectorOptions.java | 5 +-
.../flink/lookup/FileStoreLookupFunction.java | 4 +-
.../paimon/flink/lookup/FullCacheLookupTable.java | 38 +++++---
.../flink/lookup/NoPrimaryKeyLookupTable.java | 19 ++--
.../paimon/flink/lookup/PrimaryKeyLookupTable.java | 11 ++-
.../flink/lookup/SecondaryIndexLookupTable.java | 4 +-
.../org/apache/paimon/flink/LookupJoinITCase.java | 6 +-
.../flink/lookup/FileStoreLookupFunctionTest.java | 2 +-
.../paimon/flink/lookup/LookupTableTest.java | 90 +++++++++++-------
.../paimon/flink/lookup/RocksDBListStateTest.java | 4 +-
34 files changed, 856 insertions(+), 247 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/flink_connector_configuration.html b/docs/layouts/shortcodes/generated/flink_connector_configuration.html
index 1a67d4e60f..fa9b8e4d9c 100644
--- a/docs/layouts/shortcodes/generated/flink_connector_configuration.html
+++ b/docs/layouts/shortcodes/generated/flink_connector_configuration.html
@@ -60,7 +60,7 @@ under the License.
<td><h5>lookup.cache</h5></td>
<td style="word-wrap: break-word;">AUTO</td>
<td><p>Enum</p></td>
- <td>The cache mode of lookup join.<br /><br />Possible values:<ul><li>"AUTO"</li><li>"FULL"</li></ul></td>
+ <td>The cache mode of lookup join.<br /><br />Possible values:<ul><li>"AUTO"</li><li>"FULL"</li><li>"MEMORY"</li></ul></td>
</tr>
<tr>
<td><h5>lookup.dynamic-partition.refresh-interval</h5></td>
diff --git a/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java b/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
index 892e27e966..fe14048310 100644
--- a/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
+++ b/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
@@ -28,11 +28,11 @@ import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.data.serializer.RowCompactedSerializer;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.disk.RowBuffer;
-import org.apache.paimon.lookup.BulkLoader;
-import org.apache.paimon.lookup.RocksDBOptions;
-import org.apache.paimon.lookup.RocksDBState;
-import org.apache.paimon.lookup.RocksDBStateFactory;
-import org.apache.paimon.lookup.RocksDBValueState;
+import org.apache.paimon.lookup.rocksdb.RocksDBBulkLoader;
+import org.apache.paimon.lookup.rocksdb.RocksDBOptions;
+import org.apache.paimon.lookup.rocksdb.RocksDBState;
+import org.apache.paimon.lookup.rocksdb.RocksDBStateFactory;
+import org.apache.paimon.lookup.rocksdb.RocksDBValueState;
import org.apache.paimon.memory.HeapMemorySegmentPool;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
@@ -71,7 +71,7 @@ import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.stream.IntStream;
-import static org.apache.paimon.lookup.RocksDBOptions.BLOCK_CACHE_SIZE;
+import static org.apache.paimon.lookup.rocksdb.RocksDBOptions.BLOCK_CACHE_SIZE;
import static org.apache.paimon.utils.ListUtils.pickRandomly;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -210,14 +210,14 @@ public class GlobalIndexAssigner implements Serializable, Closeable {
bootstrapRecords.complete();
boolean isEmpty = true;
if (bootstrapKeys.size() > 0) {
- BulkLoader bulkLoader = keyIndex.createBulkLoader();
+ RocksDBBulkLoader bulkLoader = keyIndex.createBulkLoader();
MutableObjectIterator<BinaryRow> keyIterator = bootstrapKeys.sortedIterator();
BinaryRow row = new BinaryRow(2);
try {
while ((row = keyIterator.next(row)) != null) {
bulkLoader.write(row.getBinary(0), row.getBinary(1));
}
- } catch (BulkLoader.WriteException e) {
+ } catch (RocksDBBulkLoader.WriteException e) {
throw new RuntimeException(
"Exception in bulkLoad, the most suspicious reason is that "
+ "your data contains duplicates, please check your sink table. "
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/BulkLoader.java b/paimon-core/src/main/java/org/apache/paimon/lookup/BulkLoader.java
index 0bcdbbb5d2..f0ea474f52 100644
--- a/paimon-core/src/main/java/org/apache/paimon/lookup/BulkLoader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/BulkLoader.java
@@ -18,106 +18,13 @@
package org.apache.paimon.lookup;
-import org.rocksdb.ColumnFamilyHandle;
-import org.rocksdb.EnvOptions;
-import org.rocksdb.IngestExternalFileOptions;
-import org.rocksdb.Options;
-import org.rocksdb.RocksDB;
-import org.rocksdb.RocksDBException;
-import org.rocksdb.SstFileWriter;
-import org.rocksdb.TtlDB;
+/** Bulk loader for {@link State}, incoming keys must be sorted, and there must be no repetition. */
+public interface BulkLoader {
-import java.io.File;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
-
-/** Bulk loader for RocksDB. */
-public class BulkLoader {
-
- private final String uuid = UUID.randomUUID().toString();
-
- private final ColumnFamilyHandle columnFamily;
- private final String path;
- private final RocksDB db;
- private final boolean isTtlEnabled;
- private final Options options;
- private final List<String> files = new ArrayList<>();
- private final int currentTimeSeconds;
-
- private SstFileWriter writer = null;
- private int sstIndex = 0;
- private long recordNum = 0;
-
- public BulkLoader(RocksDB db, Options options, ColumnFamilyHandle columnFamily, String path) {
- this.db = db;
- this.isTtlEnabled = db instanceof TtlDB;
- this.options = options;
- this.columnFamily = columnFamily;
- this.path = path;
- this.currentTimeSeconds = (int) (System.currentTimeMillis() / 1000);
- }
-
- public void write(byte[] key, byte[] value) throws WriteException {
- try {
- if (writer == null) {
- writer = new SstFileWriter(new EnvOptions(), options);
- String path = new File(this.path, "sst-" + uuid + "-" + (sstIndex++)).getPath();
- writer.open(path);
- files.add(path);
- }
-
- if (isTtlEnabled) {
- value = appendTimestamp(value);
- }
-
- try {
- writer.put(key, value);
- } catch (RocksDBException e) {
- throw new WriteException(e);
- }
-
- recordNum++;
- if (recordNum % 1000 == 0 && writer.fileSize() >= options.targetFileSizeBase()) {
- writer.finish();
- writer.close();
- writer = null;
- recordNum = 0;
- }
- } catch (RocksDBException e) {
- throw new RuntimeException(e);
- }
- }
-
- private byte[] appendTimestamp(byte[] value) {
- byte[] newValue = new byte[value.length + 4];
- System.arraycopy(value, 0, newValue, 0, value.length);
- newValue[value.length] = (byte) (currentTimeSeconds & 0xff);
- newValue[value.length + 1] = (byte) ((currentTimeSeconds >> 8) & 0xff);
- newValue[value.length + 2] = (byte) ((currentTimeSeconds >> 16) & 0xff);
- newValue[value.length + 3] = (byte) ((currentTimeSeconds >> 24) & 0xff);
- return newValue;
- }
-
- public void finish() {
- try {
- if (writer != null) {
- writer.finish();
- writer.close();
- }
-
- if (files.size() > 0) {
- IngestExternalFileOptions ingestOptions = new IngestExternalFileOptions();
- db.ingestExternalFile(columnFamily, files, ingestOptions);
- ingestOptions.close();
- }
- } catch (RocksDBException e) {
- throw new RuntimeException(e);
- }
- }
+ void finish();
/** Exception during writing. */
- public static class WriteException extends Exception {
+ class WriteException extends Exception {
public WriteException(Throwable cause) {
super(cause);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/ByteArray.java b/paimon-core/src/main/java/org/apache/paimon/lookup/ByteArray.java
new file mode 100644
index 0000000000..95a279d20f
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/ByteArray.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.lookup;
+
+import org.apache.paimon.utils.SortUtil;
+
+import org.jetbrains.annotations.NotNull;
+
+import java.util.Arrays;
+
+/** A class wraps byte[] to implement equals and hashCode. */
+public class ByteArray implements Comparable<ByteArray> {
+
+ public final byte[] bytes;
+
+ public ByteArray(byte[] bytes) {
+ this.bytes = bytes;
+ }
+
+ @Override
+ public int hashCode() {
+ return Arrays.hashCode(bytes);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ByteArray byteArray = (ByteArray) o;
+ return Arrays.equals(bytes, byteArray.bytes);
+ }
+
+ public static ByteArray wrapBytes(byte[] bytes) {
+ return new ByteArray(bytes);
+ }
+
+ @Override
+ public int compareTo(@NotNull ByteArray o) {
+ return SortUtil.compareBinary(bytes, o.bytes);
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/ListBulkLoader.java b/paimon-core/src/main/java/org/apache/paimon/lookup/ListBulkLoader.java
new file mode 100644
index 0000000000..6496d5186b
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/ListBulkLoader.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.lookup;
+
+import java.util.List;
+
+/** List bulk loader to load key values, incoming keys must be sorted. */
+public interface ListBulkLoader extends BulkLoader {
+
+ void write(byte[] key, List<byte[]> value) throws WriteException;
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/ListState.java b/paimon-core/src/main/java/org/apache/paimon/lookup/ListState.java
new file mode 100644
index 0000000000..8e3d7c2c8b
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/ListState.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.lookup;
+
+import java.io.IOException;
+import java.util.List;
+
+/** {@link State} interface for list state in Operations. */
+public interface ListState<K, V> extends State<K, V> {
+
+ void add(K key, V value) throws IOException;
+
+ List<V> get(K key) throws IOException;
+
+ ListBulkLoader createBulkLoader();
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/SetState.java b/paimon-core/src/main/java/org/apache/paimon/lookup/SetState.java
new file mode 100644
index 0000000000..7874fb7178
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/SetState.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.lookup;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * {@link State} interface for set state in Operations, the values must be sorted by byte array to
+ * be returned.
+ */
+public interface SetState<K, V> extends State<K, V> {
+
+ List<V> get(K key) throws IOException;
+
+ void retract(K key, V value) throws IOException;
+
+ void add(K key, V value) throws IOException;
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/State.java b/paimon-core/src/main/java/org/apache/paimon/lookup/State.java
new file mode 100644
index 0000000000..e4174a9e21
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/State.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.lookup;
+
+import java.io.IOException;
+
+/** Interface that different types of state must implement. */
+public interface State<K, V> {
+
+ byte[] serializeKey(K key) throws IOException;
+
+ byte[] serializeValue(V value) throws IOException;
+
+ V deserializeValue(byte[] valueBytes) throws IOException;
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/StateFactory.java b/paimon-core/src/main/java/org/apache/paimon/lookup/StateFactory.java
new file mode 100644
index 0000000000..478c6f9f3b
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/StateFactory.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.lookup;
+
+import org.apache.paimon.data.serializer.Serializer;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/** State factory to create {@link State}. */
+public interface StateFactory extends Closeable {
+
+ <K, V> ValueState<K, V> valueState(
+ String name,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer,
+ long lruCacheSize)
+ throws IOException;
+
+ <K, V> SetState<K, V> setState(
+ String name,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer,
+ long lruCacheSize)
+ throws IOException;
+
+ <K, V> ListState<K, V> listState(
+ String name,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer,
+ long lruCacheSize)
+ throws IOException;
+
+ boolean preferBulkLoad();
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/ValueBulkLoader.java b/paimon-core/src/main/java/org/apache/paimon/lookup/ValueBulkLoader.java
new file mode 100644
index 0000000000..eef737078f
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/ValueBulkLoader.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.lookup;
+
+/** Value bulk loader to load key values, incoming keys must be sorted. */
+public interface ValueBulkLoader extends BulkLoader {
+
+ void write(byte[] key, byte[] value) throws WriteException;
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/ValueState.java b/paimon-core/src/main/java/org/apache/paimon/lookup/ValueState.java
new file mode 100644
index 0000000000..f04e456b3b
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/ValueState.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.lookup;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/** {@link State} interface for single-value state. The value can be deleted or updated. */
+public interface ValueState<K, V> extends State<K, V> {
+
+ @Nullable
+ V get(K key) throws IOException;
+
+ void put(K key, V value) throws IOException;
+
+ void delete(K key) throws IOException;
+
+ ValueBulkLoader createBulkLoader();
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryListState.java b/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryListState.java
new file mode 100644
index 0000000000..992fbc1abd
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryListState.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.lookup.memory;
+
+import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.lookup.ByteArray;
+import org.apache.paimon.lookup.ListBulkLoader;
+import org.apache.paimon.lookup.ListState;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.lookup.ByteArray.wrapBytes;
+
+/** In-memory list state. */
+public class InMemoryListState<K, V> extends InMemoryState<K, V> implements ListState<K, V> {
+
+ private final Map<ByteArray, List<byte[]>> values;
+
+ public InMemoryListState(Serializer<K> keySerializer, Serializer<V> valueSerializer) {
+ super(keySerializer, valueSerializer);
+ this.values = new HashMap<>();
+ }
+
+ @Override
+ public void add(K key, V value) throws IOException {
+ byte[] keyBytes = serializeKey(key);
+ byte[] valueBytes = serializeValue(value);
+ values.computeIfAbsent(wrapBytes(keyBytes), k -> new ArrayList<>()).add(valueBytes);
+ }
+
+ @Override
+ public List<V> get(K key) throws IOException {
+ List<byte[]> list = this.values.get(wrapBytes(serializeKey(key)));
+ List<V> result = new ArrayList<>();
+ if (list != null) {
+ for (byte[] value : list) {
+ result.add(deserializeValue(value));
+ }
+ }
+ return result;
+ }
+
+ @Override
+ public ListBulkLoader createBulkLoader() {
+ return new ListBulkLoader() {
+
+ @Override
+ public void write(byte[] key, List<byte[]> value) {
+ // copy the list, outside will reuse list
+ values.put(wrapBytes(key), new ArrayList<>(value));
+ }
+
+ @Override
+ public void finish() {}
+ };
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemorySetState.java b/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemorySetState.java
new file mode 100644
index 0000000000..f694a57643
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemorySetState.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.lookup.memory;
+
+import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.lookup.ByteArray;
+import org.apache.paimon.lookup.SetState;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+
+import static org.apache.paimon.lookup.ByteArray.wrapBytes;
+
+/** In-memory set state. */
+public class InMemorySetState<K, V> extends InMemoryState<K, V> implements SetState<K, V> {
+
+ private final Map<ByteArray, TreeSet<ByteArray>> values;
+
+ public InMemorySetState(Serializer<K> keySerializer, Serializer<V> valueSerializer) {
+ super(keySerializer, valueSerializer);
+ this.values = new HashMap<>();
+ }
+
+ @Override
+ public List<V> get(K key) throws IOException {
+ Set<ByteArray> set = values.get(wrapBytes(serializeKey(key)));
+ List<V> result = new ArrayList<>();
+ if (set != null) {
+ for (ByteArray value : set) {
+ result.add(deserializeValue(value.bytes));
+ }
+ }
+ return result;
+ }
+
+ @Override
+ public void retract(K key, V value) throws IOException {
+ values.get(wrapBytes(serializeKey(key))).remove(wrapBytes(serializeValue(value)));
+ }
+
+ @Override
+ public void add(K key, V value) throws IOException {
+ byte[] keyBytes = serializeKey(key);
+ byte[] valueBytes = serializeValue(value);
+ values.computeIfAbsent(wrapBytes(keyBytes), k -> new TreeSet<>())
+ .add(wrapBytes(valueBytes));
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryState.java b/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryState.java
new file mode 100644
index 0000000000..6b79776a70
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryState.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.lookup.memory;
+
+import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.io.DataInputDeserializer;
+import org.apache.paimon.io.DataOutputSerializer;
+import org.apache.paimon.lookup.State;
+
+import java.io.IOException;
+
+/** In-memory state. */
+public abstract class InMemoryState<K, V> implements State<K, V> {
+
+ protected final Serializer<K> keySerializer;
+ protected final Serializer<V> valueSerializer;
+ protected final DataOutputSerializer keyOutView;
+ protected final DataInputDeserializer valueInputView;
+ protected final DataOutputSerializer valueOutputView;
+
+ public InMemoryState(Serializer<K> keySerializer, Serializer<V> valueSerializer) {
+ this.keySerializer = keySerializer;
+ this.valueSerializer = valueSerializer;
+ this.keyOutView = new DataOutputSerializer(32);
+ this.valueInputView = new DataInputDeserializer();
+ this.valueOutputView = new DataOutputSerializer(32);
+ }
+
+ @Override
+ public byte[] serializeKey(K key) throws IOException {
+ keyOutView.clear();
+ keySerializer.serialize(key, keyOutView);
+ return keyOutView.getCopyOfBuffer();
+ }
+
+ @Override
+ public byte[] serializeValue(V value) throws IOException {
+ valueOutputView.clear();
+ valueSerializer.serialize(value, valueOutputView);
+ return valueOutputView.getCopyOfBuffer();
+ }
+
+ @Override
+ public V deserializeValue(byte[] valueBytes) throws IOException {
+ valueInputView.setBuffer(valueBytes);
+ return valueSerializer.deserialize(valueInputView);
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryStateFactory.java b/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryStateFactory.java
new file mode 100644
index 0000000000..77aa1cf17e
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryStateFactory.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.lookup.memory;
+
+import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.lookup.ListState;
+import org.apache.paimon.lookup.SetState;
+import org.apache.paimon.lookup.StateFactory;
+import org.apache.paimon.lookup.ValueState;
+
+import java.io.IOException;
+
+/** Factory to create in-memory state. */
+public class InMemoryStateFactory implements StateFactory {
+
+ @Override
+ public <K, V> ValueState<K, V> valueState(
+ String name,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer,
+ long lruCacheSize) {
+ return new InMemoryValueState<>(keySerializer, valueSerializer);
+ }
+
+ @Override
+ public <K, V> SetState<K, V> setState(
+ String name,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer,
+ long lruCacheSize) {
+ return new InMemorySetState<>(keySerializer, valueSerializer);
+ }
+
+ @Override
+ public <K, V> ListState<K, V> listState(
+ String name,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer,
+ long lruCacheSize) {
+ return new InMemoryListState<>(keySerializer, valueSerializer);
+ }
+
+ @Override
+ public boolean preferBulkLoad() {
+ return false;
+ }
+
+ @Override
+ public void close() throws IOException {}
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryValueState.java b/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryValueState.java
new file mode 100644
index 0000000000..07db364ef1
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/memory/InMemoryValueState.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.lookup.memory;
+
+import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.lookup.ByteArray;
+import org.apache.paimon.lookup.ValueBulkLoader;
+import org.apache.paimon.lookup.ValueState;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.paimon.lookup.ByteArray.wrapBytes;
+
+/** In-memory value state. */
+public class InMemoryValueState<K, V> extends InMemoryState<K, V> implements ValueState<K, V> {
+
+ private final Map<ByteArray, byte[]> values;
+
+ public InMemoryValueState(Serializer<K> keySerializer, Serializer<V> valueSerializer) {
+ super(keySerializer, valueSerializer);
+ this.values = new HashMap<>();
+ }
+
+ @Override
+ public @Nullable V get(K key) throws IOException {
+ byte[] bytes = values.get(wrapBytes(serializeKey(key)));
+ if (bytes == null) {
+ return null;
+ }
+ return deserializeValue(bytes);
+ }
+
+ @Override
+ public void put(K key, V value) throws IOException {
+ values.put(wrapBytes(serializeKey(key)), serializeValue(value));
+ }
+
+ @Override
+ public void delete(K key) throws IOException {
+ values.remove(wrapBytes(serializeKey(key)));
+ }
+
+ @Override
+ public ValueBulkLoader createBulkLoader() {
+ return new ValueBulkLoader() {
+
+ @Override
+ public void write(byte[] key, byte[] value) {
+ values.put(wrapBytes(key), value);
+ }
+
+ @Override
+ public void finish() {}
+ };
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/BulkLoader.java b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBBulkLoader.java
similarity index 82%
copy from paimon-core/src/main/java/org/apache/paimon/lookup/BulkLoader.java
copy to paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBBulkLoader.java
index 0bcdbbb5d2..8278376f73 100644
--- a/paimon-core/src/main/java/org/apache/paimon/lookup/BulkLoader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBBulkLoader.java
@@ -16,7 +16,11 @@
* limitations under the License.
*/
-package org.apache.paimon.lookup;
+package org.apache.paimon.lookup.rocksdb;
+
+import org.apache.paimon.lookup.ListBulkLoader;
+import org.apache.paimon.lookup.ValueBulkLoader;
+import org.apache.paimon.utils.ListDelimitedSerializer;
import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.EnvOptions;
@@ -28,14 +32,16 @@ import org.rocksdb.SstFileWriter;
import org.rocksdb.TtlDB;
import java.io.File;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
/** Bulk loader for RocksDB. */
-public class BulkLoader {
+public class RocksDBBulkLoader implements ValueBulkLoader, ListBulkLoader {
private final String uuid = UUID.randomUUID().toString();
+ private final ListDelimitedSerializer listSerializer = new ListDelimitedSerializer();
private final ColumnFamilyHandle columnFamily;
private final String path;
@@ -49,7 +55,8 @@ public class BulkLoader {
private int sstIndex = 0;
private long recordNum = 0;
- public BulkLoader(RocksDB db, Options options, ColumnFamilyHandle columnFamily, String path) {
+ public RocksDBBulkLoader(
+ RocksDB db, Options options, ColumnFamilyHandle columnFamily, String path) {
this.db = db;
this.isTtlEnabled = db instanceof TtlDB;
this.options = options;
@@ -58,6 +65,7 @@ public class BulkLoader {
this.currentTimeSeconds = (int) (System.currentTimeMillis() / 1000);
}
+ @Override
public void write(byte[] key, byte[] value) throws WriteException {
try {
if (writer == null) {
@@ -89,6 +97,17 @@ public class BulkLoader {
}
}
+ @Override
+ public void write(byte[] key, List<byte[]> value) throws WriteException {
+ byte[] bytes;
+ try {
+ bytes = listSerializer.serializeList(value);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ write(key, bytes);
+ }
+
private byte[] appendTimestamp(byte[] value) {
byte[] newValue = new byte[value.length + 4];
System.arraycopy(value, 0, newValue, 0, value.length);
@@ -99,6 +118,7 @@ public class BulkLoader {
return newValue;
}
+ @Override
public void finish() {
try {
if (writer != null) {
@@ -115,11 +135,4 @@ public class BulkLoader {
throw new RuntimeException(e);
}
}
-
- /** Exception during writing. */
- public static class WriteException extends Exception {
- public WriteException(Throwable cause) {
- super(cause);
- }
- }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBListState.java b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBListState.java
similarity index 87%
rename from paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBListState.java
rename to paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBListState.java
index 7fe29d209a..70716c40cb 100644
--- a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBListState.java
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBListState.java
@@ -16,9 +16,10 @@
* limitations under the License.
*/
-package org.apache.paimon.lookup;
+package org.apache.paimon.lookup.rocksdb;
import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.lookup.ListState;
import org.apache.paimon.utils.ListDelimitedSerializer;
import org.rocksdb.ColumnFamilyHandle;
@@ -29,7 +30,7 @@ import java.util.Collections;
import java.util.List;
/** RocksDB state for key -> List of value. */
-public class RocksDBListState<K, V> extends RocksDBState<K, V, List<V>> {
+public class RocksDBListState<K, V> extends RocksDBState<K, V, List<V>> implements ListState<K, V> {
private final ListDelimitedSerializer listSerializer = new ListDelimitedSerializer();
@@ -42,6 +43,7 @@ public class RocksDBListState<K, V> extends RocksDBState<K, V, List<V>> {
super(stateFactory, columnFamily, keySerializer, valueSerializer, lruCacheSize);
}
+ @Override
public void add(K key, V value) throws IOException {
byte[] keyBytes = serializeKey(key);
byte[] valueBytes = serializeValue(value);
@@ -53,6 +55,7 @@ public class RocksDBListState<K, V> extends RocksDBState<K, V, List<V>> {
cache.invalidate(wrap(keyBytes));
}
+ @Override
public List<V> get(K key) throws IOException {
byte[] keyBytes = serializeKey(key);
return cache.get(
@@ -71,14 +74,4 @@ public class RocksDBListState<K, V> extends RocksDBState<K, V, List<V>> {
return rows;
});
}
-
- public byte[] serializeValue(V value) throws IOException {
- valueOutputView.clear();
- valueSerializer.serialize(value, valueOutputView);
- return valueOutputView.getCopyOfBuffer();
- }
-
- public byte[] serializeList(List<byte[]> valueList) throws IOException {
- return listSerializer.serializeList(valueList);
- }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBOptions.java b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBOptions.java
similarity index 99%
rename from paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBOptions.java
rename to paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBOptions.java
index e8c3fecdf8..2cde6d6e7b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBOptions.java
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBOptions.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.paimon.lookup;
+package org.apache.paimon.lookup.rocksdb;
import org.apache.paimon.annotation.Documentation;
import org.apache.paimon.options.ConfigOption;
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBSetState.java b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBSetState.java
similarity index 95%
rename from paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBSetState.java
rename to paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBSetState.java
index 5c06cbd51a..9f654e152c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBSetState.java
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBSetState.java
@@ -16,9 +16,11 @@
* limitations under the License.
*/
-package org.apache.paimon.lookup;
+package org.apache.paimon.lookup.rocksdb;
import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.lookup.ByteArray;
+import org.apache.paimon.lookup.SetState;
import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.RocksDBException;
@@ -32,7 +34,8 @@ import java.util.List;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Rocksdb state for key -> Set values. */
-public class RocksDBSetState<K, V> extends RocksDBState<K, V, List<byte[]>> {
+public class RocksDBSetState<K, V> extends RocksDBState<K, V, List<byte[]>>
+ implements SetState<K, V> {
private static final byte[] EMPTY = new byte[0];
@@ -45,6 +48,7 @@ public class RocksDBSetState<K, V> extends RocksDBState<K, V, List<byte[]>> {
super(stateFactory, columnFamily, keySerializer, valueSerializer, lruCacheSize);
}
+ @Override
public List<V> get(K key) throws IOException {
ByteArray keyBytes = wrap(serializeKey(key));
List<byte[]> valueBytes = cache.getIfPresent(keyBytes);
@@ -73,6 +77,7 @@ public class RocksDBSetState<K, V> extends RocksDBState<K, V, List<byte[]>> {
return values;
}
+ @Override
public void retract(K key, V value) throws IOException {
try {
byte[] bytes = invalidKeyAndGetKVBytes(key, value);
@@ -84,6 +89,7 @@ public class RocksDBSetState<K, V> extends RocksDBState<K, V, List<byte[]>> {
}
}
+ @Override
public void add(K key, V value) throws IOException {
try {
byte[] bytes = invalidKeyAndGetKVBytes(key, value);
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBState.java b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBState.java
similarity index 82%
rename from paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBState.java
rename to paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBState.java
index 0181917a7a..62257cfd34 100644
--- a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBState.java
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBState.java
@@ -16,13 +16,15 @@
* limitations under the License.
*/
-package org.apache.paimon.lookup;
+package org.apache.paimon.lookup.rocksdb;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.serializer.Serializer;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.io.DataInputDeserializer;
import org.apache.paimon.io.DataOutputSerializer;
+import org.apache.paimon.lookup.ByteArray;
+import org.apache.paimon.lookup.State;
import org.apache.paimon.sort.BinaryExternalSortBuffer;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
@@ -37,10 +39,9 @@ import org.rocksdb.WriteOptions;
import javax.annotation.Nullable;
import java.io.IOException;
-import java.util.Arrays;
/** Rocksdb state for key value. */
-public abstract class RocksDBState<K, V, CacheV> {
+public abstract class RocksDBState<K, V, CacheV> implements State<K, V> {
protected final RocksDBStateFactory stateFactory;
@@ -85,12 +86,26 @@ public abstract class RocksDBState<K, V, CacheV> {
.build();
}
+ @Override
public byte[] serializeKey(K key) throws IOException {
keyOutView.clear();
keySerializer.serialize(key, keyOutView);
return keyOutView.getCopyOfBuffer();
}
+ @Override
+ public byte[] serializeValue(V value) throws IOException {
+ valueOutputView.clear();
+ valueSerializer.serialize(value, valueOutputView);
+ return valueOutputView.getCopyOfBuffer();
+ }
+
+ @Override
+ public V deserializeValue(byte[] valueBytes) throws IOException {
+ valueInputView.setBuffer(valueBytes);
+ return valueSerializer.deserialize(valueInputView);
+ }
+
protected ByteArray wrap(byte[] bytes) {
return new ByteArray(bytes);
}
@@ -99,8 +114,8 @@ public abstract class RocksDBState<K, V, CacheV> {
return new Reference(bytes);
}
- public BulkLoader createBulkLoader() {
- return new BulkLoader(db, stateFactory.options(), columnFamily, stateFactory.path());
+ public RocksDBBulkLoader createBulkLoader() {
+ return new RocksDBBulkLoader(db, stateFactory.options(), columnFamily, stateFactory.path());
}
public static BinaryExternalSortBuffer createBulkLoadSorter(
@@ -117,33 +132,6 @@ public abstract class RocksDBState<K, V, CacheV> {
options.sequenceFieldSortOrderIsAscending());
}
- /** A class wraps byte[] to implement equals and hashCode. */
- protected static class ByteArray {
-
- protected final byte[] bytes;
-
- protected ByteArray(byte[] bytes) {
- this.bytes = bytes;
- }
-
- @Override
- public int hashCode() {
- return Arrays.hashCode(bytes);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- ByteArray byteArray = (ByteArray) o;
- return Arrays.equals(bytes, byteArray.bytes);
- }
- }
-
/** A class wraps byte[] to indicate contain or not contain. */
protected static class Reference {
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBStateFactory.java b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBStateFactory.java
similarity index 94%
rename from paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBStateFactory.java
rename to paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBStateFactory.java
index 9168fbf4cb..883c82906b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBStateFactory.java
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBStateFactory.java
@@ -16,9 +16,10 @@
* limitations under the License.
*/
-package org.apache.paimon.lookup;
+package org.apache.paimon.lookup.rocksdb;
import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.lookup.StateFactory;
import org.rocksdb.ColumnFamilyDescriptor;
import org.rocksdb.ColumnFamilyHandle;
@@ -31,13 +32,12 @@ import org.rocksdb.TtlDB;
import javax.annotation.Nullable;
-import java.io.Closeable;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
/** Factory to create state. */
-public class RocksDBStateFactory implements Closeable {
+public class RocksDBStateFactory implements StateFactory {
public static final String MERGE_OPERATOR_NAME = "stringappendtest";
@@ -85,6 +85,7 @@ public class RocksDBStateFactory implements Closeable {
return path;
}
+ @Override
public <K, V> RocksDBValueState<K, V> valueState(
String name,
Serializer<K> keySerializer,
@@ -95,6 +96,7 @@ public class RocksDBStateFactory implements Closeable {
this, createColumnFamily(name), keySerializer, valueSerializer, lruCacheSize);
}
+ @Override
public <K, V> RocksDBSetState<K, V> setState(
String name,
Serializer<K> keySerializer,
@@ -105,6 +107,7 @@ public class RocksDBStateFactory implements Closeable {
this, createColumnFamily(name), keySerializer, valueSerializer, lruCacheSize);
}
+ @Override
public <K, V> RocksDBListState<K, V> listState(
String name,
Serializer<K> keySerializer,
@@ -116,6 +119,11 @@ public class RocksDBStateFactory implements Closeable {
this, createColumnFamily(name), keySerializer, valueSerializer, lruCacheSize);
}
+ @Override
+ public boolean preferBulkLoad() {
+ return true;
+ }
+
private ColumnFamilyHandle createColumnFamily(String name) throws IOException {
try {
return db.createColumnFamily(
diff --git a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBValueState.java b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBValueState.java
similarity index 86%
rename from paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBValueState.java
rename to paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBValueState.java
index 444ed5ad06..3d8768a9c4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBValueState.java
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBValueState.java
@@ -16,9 +16,11 @@
* limitations under the License.
*/
-package org.apache.paimon.lookup;
+package org.apache.paimon.lookup.rocksdb;
import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.lookup.ByteArray;
+import org.apache.paimon.lookup.ValueState;
import org.rocksdb.ColumnFamilyHandle;
@@ -29,7 +31,8 @@ import java.io.IOException;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Rocksdb state for key -> a single value. */
-public class RocksDBValueState<K, V> extends RocksDBState<K, V, RocksDBState.Reference> {
+public class RocksDBValueState<K, V> extends RocksDBState<K, V, RocksDBState.Reference>
+ implements ValueState<K, V> {
public RocksDBValueState(
RocksDBStateFactory stateFactory,
@@ -41,6 +44,7 @@ public class RocksDBValueState<K, V> extends RocksDBState<K, V, RocksDBState.Ref
}
@Nullable
+ @Override
public V get(K key) throws IOException {
try {
Reference valueRef = get(wrap(serializeKey(key)));
@@ -60,6 +64,7 @@ public class RocksDBValueState<K, V> extends RocksDBState<K, V, RocksDBState.Ref
return valueRef;
}
+ @Override
public void put(K key, V value) throws IOException {
checkArgument(value != null);
@@ -73,6 +78,7 @@ public class RocksDBValueState<K, V> extends RocksDBState<K, V, RocksDBState.Ref
}
}
+ @Override
public void delete(K key) throws IOException {
try {
byte[] keyBytes = serializeKey(key);
@@ -85,15 +91,4 @@ public class RocksDBValueState<K, V> extends RocksDBState<K, V, RocksDBState.Ref
throw new IOException(e);
}
}
-
- public V deserializeValue(byte[] valueBytes) throws IOException {
- valueInputView.setBuffer(valueBytes);
- return valueSerializer.deserialize(valueInputView);
- }
-
- public byte[] serializeValue(V value) throws IOException {
- valueOutputView.clear();
- valueSerializer.serialize(value, valueOutputView);
- return valueOutputView.getCopyOfBuffer();
- }
}
diff --git a/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java b/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java
index 71792962b8..219e7df99c 100644
--- a/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java
+++ b/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java
@@ -74,7 +74,7 @@ public class ConfigOptionsDocGenerator {
new OptionsClassLocation[] {
new OptionsClassLocation("paimon-api", "org.apache.paimon.options"),
new OptionsClassLocation("paimon-api", "org.apache.paimon"),
- new OptionsClassLocation("paimon-core", "org.apache.paimon.lookup"),
+ new OptionsClassLocation("paimon-core", "org.apache.paimon.lookup.rocksdb"),
new OptionsClassLocation("paimon-core", "org.apache.paimon.jdbc"),
new OptionsClassLocation("paimon-core", "org.apache.paimon.table"),
new OptionsClassLocation("paimon-core", "org.apache.paimon.iceberg"),
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
index ed8d9a7ac8..d524920e82 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
@@ -541,7 +541,10 @@ public class FlinkConnectorOptions {
AUTO,
/** Use full caching mode. */
- FULL
+ FULL,
+
+ /** Use in-memory caching mode. */
+ MEMORY
}
/** Watermark emit strategy for scan. */
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
index dec278fb70..e3e15c4ce6 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
@@ -71,8 +71,8 @@ import static org.apache.paimon.CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL;
import static org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_CACHE_MODE;
import static org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_REFRESH_TIME_PERIODS_BLACKLIST;
import static org.apache.paimon.flink.query.RemoteTableQuery.isRemoteServiceAvailable;
-import static org.apache.paimon.lookup.RocksDBOptions.LOOKUP_CACHE_ROWS;
-import static org.apache.paimon.lookup.RocksDBOptions.LOOKUP_CONTINUOUS_DISCOVERY_INTERVAL;
+import static org.apache.paimon.lookup.rocksdb.RocksDBOptions.LOOKUP_CACHE_ROWS;
+import static org.apache.paimon.lookup.rocksdb.RocksDBOptions.LOOKUP_CONTINUOUS_DISCOVERY_INTERVAL;
import static org.apache.paimon.predicate.PredicateBuilder.transformFieldMapping;
/** A lookup {@link TableFunction} for file store. */
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
index 4154b6742c..27795a00ff 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
@@ -24,9 +24,11 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
-import org.apache.paimon.lookup.BulkLoader;
-import org.apache.paimon.lookup.RocksDBState;
-import org.apache.paimon.lookup.RocksDBStateFactory;
+import org.apache.paimon.lookup.StateFactory;
+import org.apache.paimon.lookup.memory.InMemoryStateFactory;
+import org.apache.paimon.lookup.rocksdb.RocksDBBulkLoader;
+import org.apache.paimon.lookup.rocksdb.RocksDBState;
+import org.apache.paimon.lookup.rocksdb.RocksDBStateFactory;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
@@ -64,11 +66,14 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import static org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_CACHE_MODE;
import static org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_REFRESH_ASYNC;
import static org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_REFRESH_ASYNC_PENDING_SNAPSHOT_COUNT;
+import static org.apache.paimon.flink.FlinkConnectorOptions.LookupCacheMode.MEMORY;
/** Lookup table of full cache. */
public abstract class FullCacheLookupTable implements LookupTable {
+
private static final Logger LOG = LoggerFactory.getLogger(FullCacheLookupTable.class);
protected final Object lock = new Object();
@@ -79,7 +84,7 @@ public abstract class FullCacheLookupTable implements LookupTable {
@Nullable protected final FieldsComparator userDefinedSeqComparator;
protected final int appendUdsFieldNumber;
- protected RocksDBStateFactory stateFactory;
+ protected StateFactory stateFactory;
@Nullable private ExecutorService refreshExecutor;
private final AtomicReference<Exception> cachedException;
private final int maxPendingSnapshotCount;
@@ -143,11 +148,7 @@ public abstract class FullCacheLookupTable implements LookupTable {
}
protected void init() throws Exception {
- this.stateFactory =
- new RocksDBStateFactory(
- context.tempPath.toString(),
- context.table.coreOptions().toConfiguration(),
- null);
+ this.stateFactory = createStateFactory();
this.refreshExecutor =
this.refreshAsync
? Executors.newSingleThreadExecutor(
@@ -158,6 +159,16 @@ public abstract class FullCacheLookupTable implements LookupTable {
: null;
}
+ private StateFactory createStateFactory() throws IOException {
+ String diskDir = context.tempPath.toString();
+ Options options = context.table.coreOptions().toConfiguration();
+ if (options.get(LOOKUP_CACHE_MODE) == MEMORY) {
+ return new InMemoryStateFactory();
+ } else {
+ return new RocksDBStateFactory(diskDir, options, null);
+ }
+ }
+
protected void bootstrap() throws Exception {
Predicate scanPredicate =
PredicateBuilder.andNullable(context.tablePredicate, specificPartition);
@@ -168,6 +179,11 @@ public abstract class FullCacheLookupTable implements LookupTable {
scanPredicate,
context.requiredCachedBucketIds,
cacheRowFilter);
+ if (!stateFactory.preferBulkLoad()) {
+ doRefresh();
+ return;
+ }
+
BinaryExternalSortBuffer bulkLoadSorter =
RocksDBState.createBulkLoadSorter(
IOManager.create(context.tempPath.toString()), context.table.coreOptions());
@@ -189,7 +205,7 @@ public abstract class FullCacheLookupTable implements LookupTable {
while ((row = keyIterator.next(row)) != null) {
bulkLoader.write(row.getBinary(0), row.getBinary(1));
}
- } catch (BulkLoader.WriteException e) {
+ } catch (RocksDBBulkLoader.WriteException e) {
throw new RuntimeException(
"Exception in bulkLoad, the most suspicious reason is that "
+ "your data contains duplicates, please check your lookup table. ",
@@ -331,7 +347,7 @@ public abstract class FullCacheLookupTable implements LookupTable {
/** Bulk loader for the table. */
public interface TableBulkLoader {
- void write(byte[] key, byte[] value) throws BulkLoader.WriteException, IOException;
+ void write(byte[] key, byte[] value) throws RocksDBBulkLoader.WriteException, IOException;
void finish() throws IOException;
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/NoPrimaryKeyLookupTable.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/NoPrimaryKeyLookupTable.java
index 63af4f3506..546bc5a60a 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/NoPrimaryKeyLookupTable.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/NoPrimaryKeyLookupTable.java
@@ -20,8 +20,9 @@ package org.apache.paimon.flink.lookup;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalSerializers;
-import org.apache.paimon.lookup.BulkLoader;
-import org.apache.paimon.lookup.RocksDBListState;
+import org.apache.paimon.lookup.ListBulkLoader;
+import org.apache.paimon.lookup.ListState;
+import org.apache.paimon.lookup.rocksdb.RocksDBBulkLoader;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.utils.KeyProjectedRow;
@@ -40,7 +41,7 @@ public class NoPrimaryKeyLookupTable extends FullCacheLookupTable {
private final KeyProjectedRow joinKeyRow;
- private RocksDBListState<InternalRow, InternalRow> state;
+ private ListState<InternalRow, InternalRow> state;
public NoPrimaryKeyLookupTable(Context context, long lruCacheSize) {
super(context);
@@ -105,7 +106,7 @@ public class NoPrimaryKeyLookupTable extends FullCacheLookupTable {
@Override
public TableBulkLoader createBulkLoader() {
- BulkLoader bulkLoader = state.createBulkLoader();
+ ListBulkLoader bulkLoader = state.createBulkLoader();
return new TableBulkLoader() {
private final List<byte[]> values = new ArrayList<>();
@@ -113,7 +114,7 @@ public class NoPrimaryKeyLookupTable extends FullCacheLookupTable {
private byte[] currentKey;
@Override
- public void write(byte[] key, byte[] value) throws IOException {
+ public void write(byte[] key, byte[] value) {
if (currentKey != null && !Arrays.equals(key, currentKey)) {
flush();
}
@@ -122,16 +123,16 @@ public class NoPrimaryKeyLookupTable extends FullCacheLookupTable {
}
@Override
- public void finish() throws IOException {
+ public void finish() {
flush();
bulkLoader.finish();
}
- private void flush() throws IOException {
+ private void flush() {
if (currentKey != null && values.size() > 0) {
try {
- bulkLoader.write(currentKey, state.serializeList(values));
- } catch (BulkLoader.WriteException e) {
+ bulkLoader.write(currentKey, values);
+ } catch (RocksDBBulkLoader.WriteException e) {
throw new RuntimeException(e);
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyLookupTable.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyLookupTable.java
index 2a3099e9a6..485a9cfaa6 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyLookupTable.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyLookupTable.java
@@ -20,8 +20,9 @@ package org.apache.paimon.flink.lookup;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalSerializers;
-import org.apache.paimon.lookup.BulkLoader;
-import org.apache.paimon.lookup.RocksDBValueState;
+import org.apache.paimon.lookup.ValueBulkLoader;
+import org.apache.paimon.lookup.ValueState;
+import org.apache.paimon.lookup.rocksdb.RocksDBBulkLoader;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.RowKind;
@@ -44,7 +45,7 @@ public class PrimaryKeyLookupTable extends FullCacheLookupTable {
@Nullable private final ProjectedRow keyRearrange;
- protected RocksDBValueState<InternalRow, InternalRow> tableState;
+ protected ValueState<InternalRow, InternalRow> tableState;
public PrimaryKeyLookupTable(Context context, long lruCacheSize, List<String> joinKey) {
super(context);
@@ -129,12 +130,12 @@ public class PrimaryKeyLookupTable extends FullCacheLookupTable {
@Override
public TableBulkLoader createBulkLoader() {
- BulkLoader bulkLoader = tableState.createBulkLoader();
+ ValueBulkLoader bulkLoader = tableState.createBulkLoader();
return new TableBulkLoader() {
@Override
public void write(byte[] key, byte[] value)
- throws BulkLoader.WriteException, IOException {
+ throws RocksDBBulkLoader.WriteException, IOException {
bulkLoader.write(key, value);
bulkLoadWritePlus(key, value);
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/SecondaryIndexLookupTable.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/SecondaryIndexLookupTable.java
index 11c9cba24b..668570d77d 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/SecondaryIndexLookupTable.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/SecondaryIndexLookupTable.java
@@ -20,7 +20,7 @@ package org.apache.paimon.flink.lookup;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalSerializers;
-import org.apache.paimon.lookup.RocksDBSetState;
+import org.apache.paimon.lookup.SetState;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.utils.KeyProjectedRow;
@@ -35,7 +35,7 @@ public class SecondaryIndexLookupTable extends PrimaryKeyLookupTable {
private final KeyProjectedRow secKeyRow;
- private RocksDBSetState<InternalRow, InternalRow> indexState;
+ private SetState<InternalRow, InternalRow> indexState;
public SecondaryIndexLookupTable(Context context, long lruCacheSize) {
super(context, lruCacheSize / 2, context.table.primaryKeys());
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
index dea30d4d13..652ef5bd9c 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
@@ -57,8 +57,8 @@ public class LookupJoinITCase extends CatalogITCaseBase {
+ "PARTITIONED BY (`i`) WITH ('continuous.discovery-interval'='1 ms' %s)";
String fullOption = ", 'lookup.cache' = 'full'";
-
String lruOption = ", 'changelog-producer'='lookup'";
+ String memoryOption = ", 'lookup.cache' = 'memory'";
switch (cacheMode) {
case FULL:
@@ -69,6 +69,10 @@ public class LookupJoinITCase extends CatalogITCaseBase {
tEnv.executeSql(String.format(dim, lruOption));
tEnv.executeSql(String.format(partitioned, lruOption));
break;
+ case MEMORY:
+ tEnv.executeSql(String.format(dim, memoryOption));
+ tEnv.executeSql(String.format(partitioned, memoryOption));
+ break;
default:
throw new UnsupportedOperationException();
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/FileStoreLookupFunctionTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/FileStoreLookupFunctionTest.java
index dcbc405d31..4f7ff334f6 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/FileStoreLookupFunctionTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/FileStoreLookupFunctionTest.java
@@ -26,7 +26,7 @@ import org.apache.paimon.flink.FlinkRowData;
import org.apache.paimon.flink.lookup.PrimaryKeyPartialLookupTable.LocalQueryExecutor;
import org.apache.paimon.flink.lookup.PrimaryKeyPartialLookupTable.QueryExecutor;
import org.apache.paimon.flink.lookup.PrimaryKeyPartialLookupTable.RemoteQueryExecutor;
-import org.apache.paimon.lookup.RocksDBOptions;
+import org.apache.paimon.lookup.rocksdb.RocksDBOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
index 88b9471330..4a3ca45c3b 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
@@ -42,6 +42,8 @@ import org.apache.paimon.table.TableTestBase;
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.testutils.junit.parameterized.ParameterizedTestExtension;
+import org.apache.paimon.testutils.junit.parameterized.Parameters;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
@@ -50,15 +52,15 @@ import org.apache.paimon.utils.SortUtil;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.io.TempDir;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
import org.testcontainers.shaded.com.google.common.collect.ImmutableList;
import java.io.IOException;
import java.nio.file.Path;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -71,21 +73,34 @@ import java.util.concurrent.TimeoutException;
import static java.util.Collections.emptyList;
import static java.util.Collections.singletonList;
+import static org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_CACHE_MODE;
+import static org.apache.paimon.flink.FlinkConnectorOptions.LookupCacheMode.FULL;
+import static org.apache.paimon.flink.FlinkConnectorOptions.LookupCacheMode.MEMORY;
import static org.apache.paimon.types.DataTypes.INT;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link LookupTable}. */
+@ExtendWith(ParameterizedTestExtension.class)
public class LookupTableTest extends TableTestBase {
- @TempDir Path tempDir;
+ private final boolean inMemory;
+ @TempDir Path tempDir;
private RowType rowType;
-
private IOManager ioManager;
-
private FullCacheLookupTable table;
+ public LookupTableTest(boolean inMemory) {
+ this.inMemory = inMemory;
+ }
+
+ @SuppressWarnings("unused")
+ @Parameters(name = "{0}")
+ public static List<Boolean> getVarSeg() {
+ return Arrays.asList(true, false);
+ }
+
@BeforeEach
public void before() throws IOException {
this.rowType = RowType.of(new IntType(), new IntType(), new IntType());
@@ -100,6 +115,9 @@ public class LookupTableTest extends TableTestBase {
}
private FileStoreTable createTable(List<String> primaryKeys, Options options) throws Exception {
+ if (inMemory) {
+ options.set(LOOKUP_CACHE_MODE, MEMORY);
+ }
Identifier identifier = new Identifier("default", "t");
Schema schema =
new Schema(
@@ -112,7 +130,7 @@ public class LookupTableTest extends TableTestBase {
return (FileStoreTable) catalog.getTable(identifier);
}
- @Test
+ @TestTemplate
public void testPkTable() throws Exception {
FileStoreTable storeTable = createTable(singletonList("f0"), new Options());
FullCacheLookupTable.Context context =
@@ -132,7 +150,7 @@ public class LookupTableTest extends TableTestBase {
table.open();
// test bulk load error
- {
+ if (!inMemory) {
TableBulkLoader bulkLoader = table.createBulkLoader();
bulkLoader.write(new byte[] {1}, new byte[] {1});
assertThatThrownBy(() -> bulkLoader.write(new byte[] {1}, new byte[] {2}))
@@ -172,7 +190,7 @@ public class LookupTableTest extends TableTestBase {
assertThat(table.get(row(3))).hasSize(0);
}
- @Test
+ @TestTemplate
public void testPkTableWithSequenceField() throws Exception {
Options options = new Options();
options.set(CoreOptions.SEQUENCE_FIELD, "f1");
@@ -223,7 +241,7 @@ public class LookupTableTest extends TableTestBase {
assertRow(result.get(0), 1, 22, 222);
}
- @Test
+ @TestTemplate
public void testPkTableWithSequenceFieldProjection() throws Exception {
Options options = new Options();
options.set(CoreOptions.SEQUENCE_FIELD, "f2");
@@ -263,7 +281,7 @@ public class LookupTableTest extends TableTestBase {
assertRow(result.get(0), 1, 22);
}
- @Test
+ @TestTemplate
public void testPkTablePkFilter() throws Exception {
FileStoreTable storeTable = createTable(singletonList("f0"), new Options());
FullCacheLookupTable.Context context =
@@ -299,7 +317,7 @@ public class LookupTableTest extends TableTestBase {
assertThat(table.get(row(3))).hasSize(0);
}
- @Test
+ @TestTemplate
public void testPkTableNonPkFilter() throws Exception {
FileStoreTable storeTable = createTable(singletonList("f0"), new Options());
FullCacheLookupTable.Context context =
@@ -328,7 +346,7 @@ public class LookupTableTest extends TableTestBase {
assertThat(result).hasSize(0);
}
- @Test
+ @TestTemplate
public void testSecKeyTable() throws Exception {
FileStoreTable storeTable = createTable(singletonList("f0"), new Options());
FullCacheLookupTable.Context context =
@@ -376,7 +394,7 @@ public class LookupTableTest extends TableTestBase {
assertThat(result.stream().map(row -> row.getInt(0))).contains(1);
}
- @Test
+ @TestTemplate
public void testSecKeyTableWithSequenceField() throws Exception {
Options options = new Options();
options.set(CoreOptions.SEQUENCE_FIELD, "f1");
@@ -431,7 +449,7 @@ public class LookupTableTest extends TableTestBase {
assertThat(result.stream().map(row -> row.getInt(2))).doesNotContain(333);
}
- @Test
+ @TestTemplate
public void testSecKeyTablePkFilter() throws Exception {
FileStoreTable storeTable = createTable(singletonList("f0"), new Options());
FullCacheLookupTable.Context context =
@@ -476,7 +494,7 @@ public class LookupTableTest extends TableTestBase {
assertThat(table.get(row(33))).hasSize(0);
}
- @Test
+ @TestTemplate
public void testNoPrimaryKeyTable() throws Exception {
FileStoreTable storeTable = createTable(emptyList(), new Options());
FullCacheLookupTable.Context context =
@@ -524,7 +542,7 @@ public class LookupTableTest extends TableTestBase {
assertThat(result.stream().map(row -> row.getInt(0))).contains(1);
}
- @Test
+ @TestTemplate
public void testNoPrimaryKeyTableFilter() throws Exception {
FileStoreTable storeTable = createTable(emptyList(), new Options());
FullCacheLookupTable.Context context =
@@ -559,7 +577,7 @@ public class LookupTableTest extends TableTestBase {
assertRow(result.get(1), 1, 11, 111);
}
- @Test
+ @TestTemplate
public void testPkTableWithCacheRowFilter() throws Exception {
FileStoreTable storeTable = createTable(singletonList("f0"), new Options());
writeWithBucketAssigner(
@@ -600,7 +618,7 @@ public class LookupTableTest extends TableTestBase {
assertThat(res).isEmpty();
}
- @Test
+ @TestTemplate
public void testRefreshExecutorRebuildAfterReopen() throws Exception {
Options options = new Options();
options.set(FlinkConnectorOptions.LOOKUP_REFRESH_ASYNC, true);
@@ -635,7 +653,7 @@ public class LookupTableTest extends TableTestBase {
assertRow(res.get(0), 1, 22, 222);
}
- @Test
+ @TestTemplate
public void testNoPkTableWithCacheRowFilter() throws Exception {
FileStoreTable storeTable = createTable(emptyList(), new Options());
writeWithBucketAssigner(
@@ -676,7 +694,7 @@ public class LookupTableTest extends TableTestBase {
assertThat(res).isEmpty();
}
- @Test
+ @TestTemplate
public void testSecKeyTableWithCacheRowFilter() throws Exception {
FileStoreTable storeTable = createTable(singletonList("f0"), new Options());
writeWithBucketAssigner(
@@ -717,7 +735,7 @@ public class LookupTableTest extends TableTestBase {
assertThat(res).isEmpty();
}
- @Test
+ @TestTemplate
public void testPartialLookupTable() throws Exception {
FileStoreTable dimTable = createDimTable();
PrimaryKeyPartialLookupTable table =
@@ -750,7 +768,7 @@ public class LookupTableTest extends TableTestBase {
assertThat(result).hasSize(0);
}
- @Test
+ @TestTemplate
public void testPartialLookupTableWithRowFilter() throws Exception {
Options options = new Options();
options.set(CoreOptions.BUCKET.key(), "2");
@@ -771,7 +789,7 @@ public class LookupTableTest extends TableTestBase {
assertThat(result).isEmpty();
}
- @Test
+ @TestTemplate
public void testPartialLookupTableWithProjection() throws Exception {
FileStoreTable dimTable = createDimTable();
PrimaryKeyPartialLookupTable table =
@@ -803,7 +821,7 @@ public class LookupTableTest extends TableTestBase {
assertRow(result.get(0), 22, -2);
}
- @Test
+ @TestTemplate
public void testPartialLookupTableJoinKeyOrder() throws Exception {
FileStoreTable dimTable = createDimTable();
PrimaryKeyPartialLookupTable table =
@@ -835,9 +853,17 @@ public class LookupTableTest extends TableTestBase {
assertRow(result.get(0), 22, -2);
}
- @ParameterizedTest
- @ValueSource(booleans = {false, true})
- public void testPKLookupTableRefreshAsync(boolean refreshAsync) throws Exception {
+ @TestTemplate
+ public void testPKLookupTableNotRefreshAsync() throws Exception {
+ innerTestPKLookupTableRefreshAsync(false);
+ }
+
+ @TestTemplate
+ public void testPKLookupTableRefreshAsync() throws Exception {
+ innerTestPKLookupTableRefreshAsync(true);
+ }
+
+ private void innerTestPKLookupTableRefreshAsync(boolean refreshAsync) throws Exception {
Options options = new Options();
options.set(FlinkConnectorOptions.LOOKUP_REFRESH_ASYNC, refreshAsync);
FileStoreTable storeTable = createTable(singletonList("f0"), options);
@@ -902,13 +928,11 @@ public class LookupTableTest extends TableTestBase {
table.close();
}
- @Test
+ @TestTemplate
public void testFullCacheLookupTableWithForceLookup() throws Exception {
Options options = new Options();
options.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.PARTIAL_UPDATE);
- options.set(
- FlinkConnectorOptions.LOOKUP_CACHE_MODE,
- FlinkConnectorOptions.LookupCacheMode.FULL);
+ options.set(LOOKUP_CACHE_MODE, inMemory ? MEMORY : FULL);
options.set(CoreOptions.WRITE_ONLY, true);
options.set(CoreOptions.FORCE_LOOKUP, true);
options.set(CoreOptions.BUCKET, 1);
@@ -964,7 +988,7 @@ public class LookupTableTest extends TableTestBase {
assertRow(result.get(0), 1, 22, 222); // new value
}
- @Test
+ @TestTemplate
public void testPartialLookupTableWithForceLookup() throws Exception {
Options options = new Options();
options.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.PARTIAL_UPDATE);
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/RocksDBListStateTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/RocksDBListStateTest.java
index 7a942ddb56..d98d199623 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/RocksDBListStateTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/RocksDBListStateTest.java
@@ -24,8 +24,8 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
-import org.apache.paimon.lookup.RocksDBListState;
-import org.apache.paimon.lookup.RocksDBStateFactory;
+import org.apache.paimon.lookup.rocksdb.RocksDBListState;
+import org.apache.paimon.lookup.rocksdb.RocksDBStateFactory;
import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
(paimon) 08/15: [doc] fix docs gcs in filesystem.md (#5747)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 1e00d2869964afb1f11df9e56338ff3e65073b62
Author: andybj0228 <an...@126.com>
AuthorDate: Sat Jun 14 08:56:09 2025 +0800
[doc] fix docs gcs in filesystem.md (#5747)
---
docs/content/maintenance/filesystems.md | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/docs/content/maintenance/filesystems.md b/docs/content/maintenance/filesystems.md
index 673fe1f073..4a0f5e6f38 100644
--- a/docs/content/maintenance/filesystems.md
+++ b/docs/content/maintenance/filesystems.md
@@ -44,6 +44,7 @@ FileSystem pluggable jars for user to query tables from Spark/Hive side.
| Tencent Cloud Object Storage | cosn:// | Y | |
| Microsoft Azure Storage | abfs:// | Y | |
| Huawei OBS | obs:// | Y | |
+| Google Cloud Storage | gs:// | Y | |
## Dependency
@@ -405,7 +406,7 @@ Download [paimon-gs-{{< version >}}.jar](https://repository.apache.org/snapshots
{{< tab "Flink" >}}
{{< hint info >}}
-If you have already configured [oss access through Flink](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/deployment/filesystems/gcs/) (Via Flink FileSystem),
+If you have already configured [gcs access through Flink](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/deployment/filesystems/gcs/) (Via Flink FileSystem),
here you can skip the following configuration.
{{< /hint >}}
@@ -414,7 +415,7 @@ Put `paimon-gs-{{< version >}}.jar` into `lib` directory of your Flink home, and
```sql
CREATE CATALOG my_catalog WITH (
'type' = 'paimon',
- 'warehouse' = 'oss://<bucket>/<path>',
+ 'warehouse' = 'gs://<bucket>/<path>',
'fs.gs.auth.type' = 'SERVICE_ACCOUNT_JSON_KEYFILE',
'fs.gs.auth.service.account.json.keyfile' = '/path/to/service-account-.json'
);