This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new aab45278f [flink] support one or multiple databases in a dedicated
compaction job with one combined source and sink (#1887)
aab45278f is described below
commit aab45278f6e5ee4a016dd55a393aba2cb3df8cce
Author: LsomeYeah <[email protected]>
AuthorDate: Tue Sep 12 12:21:49 2023 +0800
[flink] support one or multiple databases in a dedicated compaction job
with one combined source and sink (#1887)
---
docs/content/maintenance/multiple-writers.md | 42 +-
.../apache/paimon/table/system/BucketsTable.java | 35 +-
.../paimon/flink/action/CompactDatabaseAction.java | 94 +++-
.../flink/action/CompactDatabaseActionFactory.java | 17 +-
.../flink/sink/MultiTablesCompactorSink.java | 190 +++++++
.../sink/MultiTablesStoreCompactOperator.java | 286 +++++++++++
.../paimon/flink/sink/StoreMultiCommitter.java | 23 +-
.../source/MultiTablesCompactorSourceBuilder.java | 98 ++++
.../MultiTablesBatchCompactorSourceFunction.java | 133 +++++
.../MultiTablesCompactorSourceFunction.java | 158 ++++++
.../source/operator/MultiTablesReadOperator.java | 125 +++++
...ultiTablesStreamingCompactorSourceFunction.java | 144 ++++++
.../flink/utils/MultiTablesCompactorUtil.java | 76 +++
.../flink/action/CompactDatabaseActionITCase.java | 563 +++++++++++++--------
.../MultiTablesCompactorSourceBuilderITCase.java | 553 ++++++++++++++++++++
15 files changed, 2278 insertions(+), 259 deletions(-)
diff --git a/docs/content/maintenance/multiple-writers.md
b/docs/content/maintenance/multiple-writers.md
index 2de3a7f97..1db009d79 100644
--- a/docs/content/maintenance/multiple-writers.md
+++ b/docs/content/maintenance/multiple-writers.md
@@ -103,16 +103,29 @@ Or run the following command to submit a compaction job
for multiple database.
/path/to/paimon-flink-action-{{< version >}}.jar \
compact-database \
--warehouse <warehouse-path> \
- --database <database-name> \
+ --including-databases <database-name|name-regular-expr> \
[--including-tables <paimon-table-name|name-regular-expr>] \
[--excluding-tables <paimon-table-name|name-regular-expr>] \
- [--catalog-conf <paimon-catalog-conf> [--catalog-conf
<paimon-catalog-conf> ...]]
+ [--mode <compact-mode>] \
+ [--catalog-conf <paimon-catalog-conf> [--catalog-conf
<paimon-catalog-conf> ...]] \
+ [--compact-conf <paimon-compact-conf> [--compact-conf
<paimon-compact-conf> ...]]
```
-* `--database` is used to specify which database is to be compacted. In
compact mode, you need to specify a database name, in compact-database mode,
you could specify multiple database, regular expression is supported.
+* `--including-databases` is used to specify which database is to be
compacted. In compact mode, you need to specify a database name, in
compact-database mode, you could specify multiple database, regular expression
is supported.
* `--including-tables` is used to specify which source tables are to be
compacted, you must use '|' to separate multiple tables, the format is
`databaseName.tableName`, regular expression is supported. For example,
specifying "--including-tables db1.t1|db2.+" means to compact table 'db1.t1'
and all tables in the db2 database.
* `--excluding-tables` is used to specify which source tables are not to be
compacted. The usage is same as "--including-tables". "--excluding-tables" has
higher priority than "--including-tables" if you specified both.
+* `mode` is used to specify compaction mode. Possible values:
+ * "divided" (the default mode if you haven't specified one): start a sink
for each table, the compaction of the new table requires restarting the job.
+ * "combined": start a single combined sink for all tables, the new table
will be automatically compacted.
* `--catalog-conf` is the configuration for Paimon catalog. Each configuration
should be specified in the format `key=value`. See [here]({{< ref
"maintenance/configurations" >}}) for a complete list of catalog configurations.
+* `--compact-conf` is the configuration for compaction in combined mode. Each
configuration should be specified in the format `key=value`. Compact
configuration is listed below:
+
+| Key | Default | Type | Description
|
+|-----------------------------------|---------|------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| continuous.discovery-interval | 10 s | Duration | The discovery
interval of continuous reading.
|
+| sink.parallelism | (none) | Integer | Defines a custom
parallelism for the sink. By default, if this option is not defined, the
planner will derive the parallelism for each statement individually by also
considering the global configuration. |
+| sink.use-managed-memory-allocator | false | Boolean | If true, flink
sink will use managed memory for merge tree; otherwise, it will create an
independent memory allocator.
|
+| sink.managed.writer-buffer-memory | 256 mb | MemorySize | Weight of writer
buffer in managed memory, Flink will compute the memory size for writer
according to the weight, the actual memory used depends on the running
environment. |
If you submit a batch job (set `execution.runtime-mode: batch` in Flink's
configuration), all current table files will be compacted. If you submit a
streaming job (set `execution.runtime-mode: streaming` in Flink's
configuration), the job will continuously monitor new changes to the table and
perform compactions as needed.
@@ -122,6 +135,12 @@ If you only want to submit the compaction job and don't
want to wait until the j
{{< /hint >}}
+{{< hint info >}}
+
+You can set `--mode combined` to enable compacting newly added tables without
restarting job.
+
+{{< /hint >}}
+
Example1: compact table
```bash
@@ -145,12 +164,27 @@ Example2: compact database
/path/to/paimon-flink-action-{{< version >}}.jar \
compact-database \
--warehouse s3:///path/to/warehouse \
- --database test_db \
+ --including-databases test_db \
--catalog-conf s3.endpoint=https://****.com \
--catalog-conf s3.access-key=***** \
--catalog-conf s3.secret-key=*****
```
+Example3: compact database in combined mode
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-{{< version >}}.jar \
+ compact-database \
+ --warehouse s3:///path/to/warehouse \
+ --including-databases test_db \
+ --mode combined \
+ --catalog-conf s3.endpoint=https://****.com \
+ --catalog-conf s3.access-key=***** \
+ --catalog-conf s3.secret-key=***** \
+ --compact-conf continuous.discovery-interval=*****
+```
+
For more usage of the compact action, see
```bash
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
index 79712aa17..062a440ce 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
@@ -40,7 +40,7 @@ import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.apache.paimon.types.BigIntType;
-import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.VarCharType;
@@ -51,7 +51,6 @@ import org.apache.paimon.utils.TagManager;
import javax.annotation.Nullable;
import java.io.IOException;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -73,6 +72,25 @@ public class BucketsTable implements DataTable,
ReadonlyTable {
@Nullable private final String databaseName;
+ private static final RowType ROW_TYPE =
+ RowType.of(
+ new DataType[] {
+ new BigIntType(false),
+ newBytesType(false),
+ new IntType(false),
+ newBytesType(false),
+ new VarCharType(true, Integer.MAX_VALUE),
+ new VarCharType(false, Integer.MAX_VALUE)
+ },
+ new String[] {
+ "_SNAPSHOT_ID",
+ "_PARTITION",
+ "_BUCKET",
+ "_FILES",
+ "_DATABASE_NAME",
+ "_TABLE_NAME"
+ });
+
public BucketsTable(FileStoreTable wrapped, boolean isContinuous) {
this(wrapped, isContinuous, null);
}
@@ -106,14 +124,7 @@ public class BucketsTable implements DataTable,
ReadonlyTable {
@Override
public RowType rowType() {
- List<DataField> fields = new ArrayList<>();
- fields.add(new DataField(0, "_SNAPSHOT_ID", new BigIntType(false)));
- fields.add(new DataField(1, "_PARTITION", newBytesType(false)));
- fields.add(new DataField(2, "_BUCKET", new IntType(false)));
- fields.add(new DataField(3, "_FILES", newBytesType(false)));
- fields.add(new DataField(4, "_DATABASE_NAME", new VarCharType(true,
Integer.MAX_VALUE)));
- fields.add(new DataField(5, "_TABLE_NAME", new VarCharType(false,
Integer.MAX_VALUE)));
- return new RowType(fields);
+ return ROW_TYPE;
}
@Override
@@ -161,6 +172,10 @@ public class BucketsTable implements DataTable,
ReadonlyTable {
return wrapped.fileIO();
}
+ public static RowType getRowType() {
+ return ROW_TYPE;
+ }
+
private class BucketsRead implements InnerTableRead {
private final DataFileMetaSerializer dataFileMetaSerializer = new
DataFileMetaSerializer();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
index abbeb5e33..e56c1f549 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
@@ -21,17 +21,24 @@ package org.apache.paimon.flink.action;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.flink.compact.UnawareBucketCompactionTopoBuilder;
+import org.apache.paimon.flink.sink.BucketsRowChannelComputer;
import org.apache.paimon.flink.sink.CompactorSinkBuilder;
+import org.apache.paimon.flink.sink.MultiTablesCompactorSink;
import org.apache.paimon.flink.source.CompactorSourceBuilder;
+import org.apache.paimon.flink.source.MultiTablesCompactorSourceBuilder;
import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
+import org.apache.paimon.options.Options;
import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
+import org.apache.paimon.utils.Preconditions;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.ExecutionOptions;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.RowData;
@@ -47,25 +54,53 @@ import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import static org.apache.paimon.flink.sink.FlinkStreamPartitioner.partition;
+
/** Database compact action for Flink. */
public class CompactDatabaseAction extends ActionBase {
private static final Logger LOG =
LoggerFactory.getLogger(CompactDatabaseAction.class);
- private final Pattern includingPattern;
- private final Pattern excludingPattern;
- private final String database;
+ private Pattern includingPattern = Pattern.compile(".*");
+ @Nullable private Pattern excludingPattern;
+ private Pattern databasePattern = Pattern.compile(".*");
+
+ private MultiTablesSinkMode databaseCompactMode =
MultiTablesSinkMode.DIVIDED;
+
private final Map<String, Table> tableMap = new HashMap<>();
- public CompactDatabaseAction(
- String warehouse,
- String database,
- @Nullable String includingTables,
- @Nullable String excludingTables,
- Map<String, String> catalogConfig) {
+ private Options compactOptions = new Options();
+
+ public CompactDatabaseAction(String warehouse, Map<String, String>
catalogConfig) {
super(warehouse, catalogConfig);
- this.database = database;
- this.includingPattern = Pattern.compile(includingTables == null ? ".*"
: includingTables);
+ }
+
+ public CompactDatabaseAction includingDatabases(@Nullable String
includingDatabases) {
+ if (includingDatabases != null) {
+ this.databasePattern = Pattern.compile(includingDatabases);
+ }
+ return this;
+ }
+
+ public CompactDatabaseAction includingTables(@Nullable String
includingTables) {
+ if (includingTables != null) {
+ this.includingPattern = Pattern.compile(includingTables);
+ }
+ return this;
+ }
+
+ public CompactDatabaseAction excludingTables(@Nullable String
excludingTables) {
this.excludingPattern = excludingTables == null ? null :
Pattern.compile(excludingTables);
+ return this;
+ }
+
+ public CompactDatabaseAction withDatabaseCompactMode(@Nullable String
mode) {
+ this.databaseCompactMode = MultiTablesSinkMode.fromString(mode);
+ return this;
+ }
+
+ public CompactDatabaseAction withCompactOptions(Map<String, String>
compactOptions) {
+ this.compactOptions = Options.fromMap(compactOptions);
+ return this;
}
private boolean shouldCompactionTable(String paimonFullTableName) {
@@ -82,8 +117,15 @@ public class CompactDatabaseAction extends ActionBase {
@Override
public void build(StreamExecutionEnvironment env) {
+ if (databaseCompactMode == MultiTablesSinkMode.DIVIDED) {
+ buildForDividedMode(env);
+ } else {
+ buildForCombinedMode(env);
+ }
+ }
+
+ private void buildForDividedMode(StreamExecutionEnvironment env) {
try {
- Pattern databasePattern = Pattern.compile(database);
List<String> databases = catalog.listDatabases();
for (String databaseName : databases) {
Matcher databaseMatcher =
databasePattern.matcher(databaseName);
@@ -116,6 +158,10 @@ public class CompactDatabaseAction extends ActionBase {
throw new RuntimeException(e);
}
+ Preconditions.checkState(
+ !tableMap.isEmpty(),
+ "no tables to be compacted. possible cause is that there are
no tables detected after pattern matching");
+
ReadableConfig conf =
StreamExecutionEnvironmentUtils.getConfiguration(env);
boolean isStreaming =
conf.get(ExecutionOptions.RUNTIME_MODE) ==
RuntimeExecutionMode.STREAMING;
@@ -142,6 +188,30 @@ public class CompactDatabaseAction extends ActionBase {
}
}
+ private void buildForCombinedMode(StreamExecutionEnvironment env) {
+
+ ReadableConfig conf =
StreamExecutionEnvironmentUtils.getConfiguration(env);
+ boolean isStreaming =
+ conf.get(ExecutionOptions.RUNTIME_MODE) ==
RuntimeExecutionMode.STREAMING;
+ // TODO: Currently, multi-tables compaction don't support tables which
bucketmode is UNWARE.
+ MultiTablesCompactorSourceBuilder sourceBuilder =
+ new MultiTablesCompactorSourceBuilder(
+ catalogLoader(),
+ databasePattern,
+ includingPattern,
+ excludingPattern,
+
compactOptions.get(CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL).toMillis());
+ DataStream<RowData> source =
+
sourceBuilder.withEnv(env).withContinuousMode(isStreaming).build();
+
+ DataStream<RowData> partitioned =
+ partition(
+ source,
+ new BucketsRowChannelComputer(),
+
compactOptions.get(FlinkConnectorOptions.SINK_PARALLELISM));
+ new MultiTablesCompactorSink(catalogLoader(),
compactOptions).sinkFrom(partitioned);
+ }
+
private void buildForTraditionalCompaction(
StreamExecutionEnvironment env,
String fullName,
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseActionFactory.java
index 345733616..559c28ed4 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseActionFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseActionFactory.java
@@ -20,7 +20,6 @@ package org.apache.paimon.flink.action;
import org.apache.flink.api.java.utils.MultipleParameterTool;
-import java.util.Map;
import java.util.Optional;
/** Factory to create {@link CompactDatabaseAction}. */
@@ -35,15 +34,17 @@ public class CompactDatabaseActionFactory implements
ActionFactory {
@Override
public Optional<Action> create(MultipleParameterTool params) {
- String warehouse = params.get("warehouse");
- String database = params.get("database");
- String includingTables = params.get("including-tables");
- String excludingTables = params.get("excluding-tables");
- Map<String, String> catalogConfig = optionalConfigMap(params,
"catalog-conf");
-
CompactDatabaseAction action =
new CompactDatabaseAction(
- warehouse, database, includingTables, excludingTables,
catalogConfig);
+ getRequiredValue(params, "warehouse"),
+ optionalConfigMap(params, "catalog-conf"));
+
+ action.includingDatabases(params.get("including-databases"))
+ .includingTables(params.get("including-tables"))
+ .excludingTables(params.get("excluding-tables"))
+ .withDatabaseCompactMode(params.get("mode"))
+ .withCompactOptions(optionalConfigMap(params, "compact-conf"));
+
return Optional.of(action);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesCompactorSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesCompactorSink.java
new file mode 100644
index 000000000..9651a86b9
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesCompactorSink.java
@@ -0,0 +1,190 @@
+/*
+ * 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.sink;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.VersionedSerializerWrapper;
+import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
+import org.apache.paimon.manifest.WrappedManifestCommittable;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.table.data.RowData;
+
+import java.io.Serializable;
+import java.util.UUID;
+
+import static
org.apache.paimon.flink.FlinkConnectorOptions.SINK_MANAGED_WRITER_BUFFER_MEMORY;
+import static
org.apache.paimon.flink.FlinkConnectorOptions.SINK_USE_MANAGED_MEMORY;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** A sink for processing multi-tables in dedicated compaction job. */
+public class MultiTablesCompactorSink implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private static final String WRITER_NAME = "Writer";
+ private static final String GLOBAL_COMMITTER_NAME = "Global Committer";
+
+ private final Catalog.Loader catalogLoader;
+ private final boolean ignorePreviousFiles;
+
+ private final Options options;
+
+ public MultiTablesCompactorSink(Catalog.Loader catalogLoader, Options
options) {
+ this.catalogLoader = catalogLoader;
+ this.ignorePreviousFiles = false;
+ this.options = options;
+ }
+
+ public DataStreamSink<?> sinkFrom(DataStream<RowData> input) {
+ // This commitUser is valid only for new jobs.
+ // After the job starts, this commitUser will be recorded into the
states of write and
+ // commit operators.
+ // When the job restarts, commitUser will be recovered from states and
this value is
+ // ignored.
+ String initialCommitUser = UUID.randomUUID().toString();
+ return sinkFrom(input, initialCommitUser);
+ }
+
+ public DataStreamSink<?> sinkFrom(DataStream<RowData> input, String
initialCommitUser) {
+ // do the actually writing action, no snapshot generated in this stage
+ SingleOutputStreamOperator<MultiTableCommittable> written =
+ doWrite(input, initialCommitUser, input.getParallelism());
+
+ // commit the committable to generate a new snapshot
+ return doCommit(written, initialCommitUser);
+ }
+
+ public SingleOutputStreamOperator<MultiTableCommittable> doWrite(
+ DataStream<RowData> input, String commitUser, Integer parallelism)
{
+ StreamExecutionEnvironment env = input.getExecutionEnvironment();
+ boolean isStreaming =
+ StreamExecutionEnvironmentUtils.getConfiguration(env)
+ .get(ExecutionOptions.RUNTIME_MODE)
+ == RuntimeExecutionMode.STREAMING;
+
+ SingleOutputStreamOperator<MultiTableCommittable> written =
+ input.transform(
+ WRITER_NAME,
+ new MultiTableCommittableTypeInfo(),
+ createWriteOperator(
+ env.getCheckpointConfig(),
isStreaming, commitUser))
+ .setParallelism(parallelism == null ?
input.getParallelism() : parallelism);
+
+ if (!isStreaming) {
+ assertBatchConfiguration(env, written.getParallelism());
+ }
+
+ if (options.get(SINK_USE_MANAGED_MEMORY)) {
+ MemorySize memorySize =
options.get(SINK_MANAGED_WRITER_BUFFER_MEMORY);
+ written.getTransformation()
+ .declareManagedMemoryUseCaseAtOperatorScope(
+ ManagedMemoryUseCase.OPERATOR,
memorySize.getMebiBytes());
+ }
+ return written;
+ }
+
+ protected DataStreamSink<?> doCommit(
+ DataStream<MultiTableCommittable> written, String commitUser) {
+ StreamExecutionEnvironment env = written.getExecutionEnvironment();
+ ReadableConfig conf =
StreamExecutionEnvironmentUtils.getConfiguration(env);
+ CheckpointConfig checkpointConfig = env.getCheckpointConfig();
+ boolean isStreaming =
+ conf.get(ExecutionOptions.RUNTIME_MODE) ==
RuntimeExecutionMode.STREAMING;
+ boolean streamingCheckpointEnabled =
+ isStreaming && checkpointConfig.isCheckpointingEnabled();
+ if (streamingCheckpointEnabled) {
+ assertStreamingConfiguration(env);
+ }
+
+ SingleOutputStreamOperator<?> committed =
+ written.transform(
+ GLOBAL_COMMITTER_NAME,
+ new MultiTableCommittableTypeInfo(),
+ new CommitterOperator<>(
+ streamingCheckpointEnabled,
+ commitUser,
+ createCommitterFactory(),
+ createCommittableStateManager()))
+ .setParallelism(1)
+ .setMaxParallelism(1);
+ return committed.addSink(new
DiscardingSink<>()).name("end").setParallelism(1);
+ }
+
+ public static void assertStreamingConfiguration(StreamExecutionEnvironment
env) {
+ checkArgument(
+ !env.getCheckpointConfig().isUnalignedCheckpointsEnabled(),
+ "Paimon sink currently does not support unaligned checkpoints.
Please set "
+ + ExecutionCheckpointingOptions.ENABLE_UNALIGNED.key()
+ + " to false.");
+ checkArgument(
+ env.getCheckpointConfig().getCheckpointingMode() ==
CheckpointingMode.EXACTLY_ONCE,
+ "Paimon sink currently only supports EXACTLY_ONCE checkpoint
mode. Please set "
+ +
ExecutionCheckpointingOptions.CHECKPOINTING_MODE.key()
+ + " to exactly-once");
+ }
+
+ private void assertBatchConfiguration(StreamExecutionEnvironment env, int
sinkParallelism) {
+ try {
+ checkArgument(
+ sinkParallelism != -1 ||
!AdaptiveParallelism.isEnabled(env),
+ "Paimon Sink does not support Flink's Adaptive Parallelism
mode. "
+ + "Please manually turn it off or set Paimon
`sink.parallelism` manually.");
+ } catch (NoClassDefFoundError ignored) {
+ // before 1.17, there is no adaptive parallelism
+ }
+ }
+
+ // TODO:refactor FlinkSink to adopt this sink
+ protected OneInputStreamOperator<RowData, MultiTableCommittable>
createWriteOperator(
+ CheckpointConfig checkpointConfig, boolean isStreaming, String
commitUser) {
+ return new MultiTablesStoreCompactOperator(
+ catalogLoader,
+ commitUser,
+ checkpointConfig,
+ isStreaming,
+ ignorePreviousFiles,
+ options);
+ }
+
+ protected Committer.Factory<MultiTableCommittable,
WrappedManifestCommittable>
+ createCommitterFactory() {
+ return (user, metricGroup) ->
+ new StoreMultiCommitter(
+ catalogLoader, user, new
CommitterMetrics(metricGroup), true);
+ }
+
+ protected CommittableStateManager<WrappedManifestCommittable>
createCommittableStateManager() {
+ return new RestoreAndFailCommittableStateManager<>(
+ () -> new VersionedSerializerWrapper<>(new
WrappedManifestCommittableSerializer()));
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesStoreCompactOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesStoreCompactOperator.java
new file mode 100644
index 000000000..8ce822bc2
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesStoreCompactOperator.java
@@ -0,0 +1,286 @@
+/*
+ * 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.sink;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFileMetaSerializer;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.data.RowData;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.CoreOptions.FULL_COMPACTION_DELTA_COMMITS;
+import static
org.apache.paimon.flink.FlinkConnectorOptions.CHANGELOG_PRODUCER_FULL_COMPACTION_TRIGGER_INTERVAL;
+import static
org.apache.paimon.flink.FlinkConnectorOptions.CHANGELOG_PRODUCER_LOOKUP_WAIT;
+import static
org.apache.paimon.flink.sink.cdc.CdcRecordStoreWriteOperator.RETRY_SLEEP_TIME;
+import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
+
+/**
+ * A dedicated operator for manual triggered compaction.
+ *
+ * <p>In-coming records are generated by sources built from {@link
+ * org.apache.paimon.flink.source.MultiTablesCompactorSourceBuilder}. The
records will contain
+ * partition keys, bucket number, table name and database name.
+ */
+public class MultiTablesStoreCompactOperator
+ extends PrepareCommitOperator<RowData, MultiTableCommittable> {
+
+ private static final long serialVersionUID = 1L;
+
+ private StoreSinkWrite.Provider storeSinkWriteProvider;
+ private final CheckpointConfig checkpointConfig;
+ private final boolean isStreaming;
+ private final boolean ignorePreviousFiles;
+ private final String initialCommitUser;
+
+ private transient StoreSinkWriteState state;
+ private transient DataFileMetaSerializer dataFileMetaSerializer;
+
+ private final Catalog.Loader catalogLoader;
+
+ protected Catalog catalog;
+ protected Map<Identifier, FileStoreTable> tables;
+ protected Map<Identifier, StoreSinkWrite> writes;
+ protected String commitUser;
+
+ public MultiTablesStoreCompactOperator(
+ Catalog.Loader catalogLoader,
+ String initialCommitUser,
+ CheckpointConfig checkpointConfig,
+ boolean isStreaming,
+ boolean ignorePreviousFiles,
+ Options options) {
+ super(options);
+ this.catalogLoader = catalogLoader;
+ this.initialCommitUser = initialCommitUser;
+ this.checkpointConfig = checkpointConfig;
+ this.isStreaming = isStreaming;
+ this.ignorePreviousFiles = ignorePreviousFiles;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws
Exception {
+ super.initializeState(context);
+
+ catalog = catalogLoader.load();
+
+ // Each job can only have one user name and this name must be
consistent across restarts.
+ // We cannot use job id as commit user name here because user may
change job id by creating
+ // a savepoint, stop the job and then resume from savepoint.
+ commitUser =
+ StateUtils.getSingleValueFromState(
+ context, "commit_user_state", String.class,
initialCommitUser);
+
+ state =
+ new StoreSinkWriteState(
+ context,
+ (tableName, partition, bucket) ->
+ ChannelComputer.select(
+ partition,
+ bucket,
+
getRuntimeContext().getNumberOfParallelSubtasks())
+ ==
getRuntimeContext().getIndexOfThisSubtask());
+
+ tables = new HashMap<>();
+ writes = new HashMap<>();
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ dataFileMetaSerializer = new DataFileMetaSerializer();
+ }
+
+ @Override
+ public void processElement(StreamRecord<RowData> element) throws Exception
{
+ RowData record = element.getValue();
+
+ long snapshotId = record.getLong(0);
+ BinaryRow partition = deserializeBinaryRow(record.getBinary(1));
+ int bucket = record.getInt(2);
+ byte[] serializedFiles = record.getBinary(3);
+ List<DataFileMeta> files =
dataFileMetaSerializer.deserializeList(serializedFiles);
+ String databaseName = record.getString(4).toString();
+ String tableName = record.getString(5).toString();
+
+ Identifier tableId = Identifier.create(databaseName, tableName);
+ FileStoreTable table = getTable(tableId);
+
+ Preconditions.checkArgument(
+ !table.coreOptions().writeOnly(),
+ CoreOptions.WRITE_ONLY.key()
+ + " should not be true for
MultiTablesStoreCompactOperator.");
+
+ storeSinkWriteProvider =
+ createWriteProvider(table, checkpointConfig, isStreaming,
ignorePreviousFiles);
+
+ StoreSinkWrite write =
+ writes.computeIfAbsent(
+ tableId,
+ id ->
+ storeSinkWriteProvider.provide(
+ table,
+ commitUser,
+ state,
+
getContainingTask().getEnvironment().getIOManager(),
+ memoryPool));
+
+ if (write.streamingMode()) {
+ write.notifyNewFiles(snapshotId, partition, bucket, files);
+ write.compact(partition, bucket, false);
+ } else {
+ Preconditions.checkArgument(
+ files.isEmpty(),
+ "Batch compact job does not concern what files are
compacted. "
+ + "They only need to know what buckets are
compacted.");
+ write.compact(partition, bucket, true);
+ }
+ }
+
+ @Override
+ protected List<MultiTableCommittable> prepareCommit(boolean
waitCompaction, long checkpointId)
+ throws IOException {
+
+ List<MultiTableCommittable> committables = new LinkedList<>();
+ for (Map.Entry<Identifier, StoreSinkWrite> entry : writes.entrySet()) {
+ Identifier key = entry.getKey();
+ StoreSinkWrite write = entry.getValue();
+ committables.addAll(
+ write.prepareCommit(waitCompaction, checkpointId).stream()
+ .map(
+ committable ->
+
MultiTableCommittable.fromCommittable(key, committable))
+ .collect(Collectors.toList()));
+ }
+ return committables;
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ for (StoreSinkWrite write : writes.values()) {
+ write.snapshotState();
+ }
+ state.snapshotState();
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ for (StoreSinkWrite write : writes.values()) {
+ write.close();
+ }
+ }
+
+ private FileStoreTable getTable(Identifier tableId) throws
InterruptedException {
+ FileStoreTable table = tables.get(tableId);
+ if (table == null) {
+ while (true) {
+ try {
+ table = (FileStoreTable) catalog.getTable(tableId);
+ table =
+ table.copy(
+ Collections.singletonMap(
+ CoreOptions.WRITE_ONLY.key(),
"false"));
+ tables.put(tableId, table);
+ break;
+ } catch (Catalog.TableNotExistException e) {
+ // table not found, waiting until table is created by
+ // upstream operators
+ }
+ Thread.sleep(RETRY_SLEEP_TIME.defaultValue().toMillis());
+ }
+ }
+ return table;
+ }
+
+ private StoreSinkWrite.Provider createWriteProvider(
+ FileStoreTable fileStoreTable,
+ CheckpointConfig checkpointConfig,
+ boolean isStreaming,
+ boolean ignorePreviousFiles) {
+ boolean waitCompaction;
+ if (fileStoreTable.coreOptions().writeOnly()) {
+ waitCompaction = false;
+ } else {
+ Options options = fileStoreTable.coreOptions().toConfiguration();
+ CoreOptions.ChangelogProducer changelogProducer =
+ fileStoreTable.coreOptions().changelogProducer();
+ waitCompaction =
+ changelogProducer == CoreOptions.ChangelogProducer.LOOKUP
+ && options.get(CHANGELOG_PRODUCER_LOOKUP_WAIT);
+
+ int deltaCommits = -1;
+ if (options.contains(FULL_COMPACTION_DELTA_COMMITS)) {
+ deltaCommits = options.get(FULL_COMPACTION_DELTA_COMMITS);
+ } else if
(options.contains(CHANGELOG_PRODUCER_FULL_COMPACTION_TRIGGER_INTERVAL)) {
+ long fullCompactionThresholdMs =
+
options.get(CHANGELOG_PRODUCER_FULL_COMPACTION_TRIGGER_INTERVAL).toMillis();
+ deltaCommits =
+ (int)
+ (fullCompactionThresholdMs
+ /
checkpointConfig.getCheckpointInterval());
+ }
+
+ if (changelogProducer ==
CoreOptions.ChangelogProducer.FULL_COMPACTION
+ || deltaCommits >= 0) {
+ int finalDeltaCommits = Math.max(deltaCommits, 1);
+ return (table, commitUser, state, ioManager, memoryPool) ->
+ new GlobalFullCompactionSinkWrite(
+ table,
+ commitUser,
+ state,
+ ioManager,
+ ignorePreviousFiles,
+ waitCompaction,
+ finalDeltaCommits,
+ isStreaming,
+ memoryPool);
+ }
+ }
+
+ return (table, commitUser, state, ioManager, memoryPool) ->
+ new StoreSinkWriteImpl(
+ table,
+ commitUser,
+ state,
+ ioManager,
+ ignorePreviousFiles,
+ waitCompaction,
+ isStreaming,
+ memoryPool);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java
index afecf7548..6559e9969 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.sink;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.manifest.ManifestCommittable;
@@ -32,6 +33,7 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -54,13 +56,24 @@ public class StoreMultiCommitter
// referenced by table id.
private final Map<Identifier, StoreCommitter> tableCommitters;
+ // compact job needs set "write-only" of a table to false
+ private final boolean isCompactJob;
+
public StoreMultiCommitter(
Catalog.Loader catalogLoader, String commitUser, @Nullable
CommitterMetrics metrics) {
+ this(catalogLoader, commitUser, metrics, false);
+ }
+
+ public StoreMultiCommitter(
+ Catalog.Loader catalogLoader,
+ String commitUser,
+ @Nullable CommitterMetrics metrics,
+ boolean isCompactJob) {
this.catalog = catalogLoader.load();
this.commitUser = commitUser;
this.metrics = metrics;
-
this.tableCommitters = new HashMap<>();
+ this.isCompactJob = isCompactJob;
}
@Override
@@ -148,6 +161,12 @@ public class StoreMultiCommitter
FileStoreTable table;
try {
table = (FileStoreTable) catalog.getTable(tableId);
+ if (isCompactJob) {
+ table =
+ table.copy(
+ Collections.singletonMap(
+ CoreOptions.WRITE_ONLY.key(),
"false"));
+ }
} catch (Catalog.TableNotExistException e) {
throw new RuntimeException(
String.format(
@@ -156,7 +175,7 @@ public class StoreMultiCommitter
}
committer =
new StoreCommitter(
-
table.newCommit(commitUser).ignoreEmptyCommit(false), metrics);
+
table.newCommit(commitUser).ignoreEmptyCommit(isCompactJob), metrics);
tableCommitters.put(tableId, committer);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/MultiTablesCompactorSourceBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/MultiTablesCompactorSourceBuilder.java
new file mode 100644
index 000000000..b631d3b7a
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/MultiTablesCompactorSourceBuilder.java
@@ -0,0 +1,98 @@
+/*
+ * 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.source;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.LogicalTypeConversion;
+import
org.apache.paimon.flink.source.operator.MultiTablesBatchCompactorSourceFunction;
+import
org.apache.paimon.flink.source.operator.MultiTablesStreamingCompactorSourceFunction;
+import org.apache.paimon.table.system.BucketsTable;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+import java.util.regex.Pattern;
+
+/**
+ * source builder to build a Flink compactor source for multi-tables. This is
for dedicated
+ * compactor jobs.
+ */
+public class MultiTablesCompactorSourceBuilder {
+ private final Catalog.Loader catalogLoader;
+ private final Pattern includingPattern;
+ private final Pattern excludingPattern;
+ private final Pattern databasePattern;
+ private final long monitorInterval;
+
+ private boolean isContinuous = false;
+ private StreamExecutionEnvironment env;
+
+ public MultiTablesCompactorSourceBuilder(
+ Catalog.Loader catalogLoader,
+ Pattern databasePattern,
+ Pattern includingPattern,
+ Pattern excludingPattern,
+ long monitorInterval) {
+ this.catalogLoader = catalogLoader;
+ this.includingPattern = includingPattern;
+ this.excludingPattern = excludingPattern;
+ this.databasePattern = databasePattern;
+ this.monitorInterval = monitorInterval;
+ }
+
+ public MultiTablesCompactorSourceBuilder withContinuousMode(boolean
isContinuous) {
+ this.isContinuous = isContinuous;
+ return this;
+ }
+
+ public MultiTablesCompactorSourceBuilder
withEnv(StreamExecutionEnvironment env) {
+ this.env = env;
+ return this;
+ }
+
+ public DataStream<RowData> build() {
+ Preconditions.checkArgument(env != null, "StreamExecutionEnvironment
should not be null.");
+ RowType produceType = BucketsTable.getRowType();
+ if (isContinuous) {
+ return MultiTablesStreamingCompactorSourceFunction.buildSource(
+ env,
+ "MultiTables-StreamingCompactorSource",
+
InternalTypeInfo.of(LogicalTypeConversion.toLogicalType(produceType)),
+ catalogLoader,
+ includingPattern,
+ excludingPattern,
+ databasePattern,
+ monitorInterval);
+ } else {
+ return MultiTablesBatchCompactorSourceFunction.buildSource(
+ env,
+ "MultiTables-BatchCompactorSource",
+
InternalTypeInfo.of(LogicalTypeConversion.toLogicalType(produceType)),
+ catalogLoader,
+ includingPattern,
+ excludingPattern,
+ databasePattern,
+ monitorInterval);
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesBatchCompactorSourceFunction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesBatchCompactorSourceFunction.java
new file mode 100644
index 000000000..3da9fb2e6
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesBatchCompactorSourceFunction.java
@@ -0,0 +1,133 @@
+/*
+ * 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.source.operator;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.utils.JavaTypeInfo;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.EndOfScanException;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.StreamTableScan;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.operators.StreamSource;
+import org.apache.flink.table.data.RowData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+/** It is responsible for monitoring compactor source in batch mode. */
+public class MultiTablesBatchCompactorSourceFunction extends
MultiTablesCompactorSourceFunction {
+
+ private static final Logger LOG =
+
LoggerFactory.getLogger(MultiTablesBatchCompactorSourceFunction.class);
+
+ public MultiTablesBatchCompactorSourceFunction(
+ Catalog.Loader catalogLoader,
+ Pattern includingPattern,
+ Pattern excludingPattern,
+ Pattern databasePattern,
+ long monitorInterval) {
+ super(
+ catalogLoader,
+ includingPattern,
+ excludingPattern,
+ databasePattern,
+ false,
+ monitorInterval);
+ }
+
+ @Override
+ public void run(SourceContext<Tuple2<Split, String>> ctx) throws Exception
{
+ this.ctx = ctx;
+ if (isRunning) {
+ boolean isEmpty;
+ synchronized (ctx.getCheckpointLock()) {
+ if (!isRunning) {
+ return;
+ }
+ try {
+ // batch mode do not need check for new tables
+ List<Tuple2<Split, String>> splits = new ArrayList<>();
+ for (Map.Entry<Identifier, StreamTableScan> entry :
scansMap.entrySet()) {
+ Identifier identifier = entry.getKey();
+ StreamTableScan scan = entry.getValue();
+ splits.addAll(
+ scan.plan().splits().stream()
+ .map(split -> new Tuple2<>(split,
identifier.getFullName()))
+ .collect(Collectors.toList()));
+ }
+
+ isEmpty = splits.isEmpty();
+ splits.forEach(ctx::collect);
+
+ } catch (EndOfScanException esf) {
+ LOG.info("Catching EndOfStreamException, the stream is
finished.");
+ return;
+ }
+ }
+ if (isEmpty) {
+ throw new Exception(
+ "No splits were collected. Please ensure there are
tables detected after pattern matching");
+ }
+ }
+ }
+
+ public static DataStream<RowData> buildSource(
+ StreamExecutionEnvironment env,
+ String name,
+ TypeInformation<RowData> typeInfo,
+ Catalog.Loader catalogLoader,
+ Pattern includingPattern,
+ Pattern excludingPattern,
+ Pattern databasePattern,
+ long monitorInterval) {
+ MultiTablesBatchCompactorSourceFunction function =
+ new MultiTablesBatchCompactorSourceFunction(
+ catalogLoader,
+ includingPattern,
+ excludingPattern,
+ databasePattern,
+ monitorInterval);
+ StreamSource<Tuple2<Split, String>, ?> sourceOperator = new
StreamSource<>(function);
+ TupleTypeInfo<Tuple2<Split, String>> tupleTypeInfo =
+ new TupleTypeInfo<>(
+ new JavaTypeInfo<>(Split.class),
BasicTypeInfo.STRING_TYPE_INFO);
+ return new DataStreamSource<>(
+ env, tupleTypeInfo, sourceOperator, false, name,
Boundedness.BOUNDED)
+ .forceNonParallel()
+ .partitionCustom(
+ (key, numPartitions) -> key % numPartitions,
+ split -> ((DataSplit) split.f0).bucket())
+ .transform(name, typeInfo, new
MultiTablesReadOperator(catalogLoader, false));
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesCompactorSourceFunction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesCompactorSourceFunction.java
new file mode 100644
index 000000000..b1f9637c6
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesCompactorSourceFunction.java
@@ -0,0 +1,158 @@
+/*
+ * 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.source.operator;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.StreamTableScan;
+import org.apache.paimon.table.system.BucketsTable;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+
+import static
org.apache.paimon.flink.utils.MultiTablesCompactorUtil.compactOptions;
+import static
org.apache.paimon.flink.utils.MultiTablesCompactorUtil.shouldCompactTable;
+
+/**
+ * This is the single (non-parallel) monitoring task, it is responsible for:
+ *
+ * <ol>
+ * <li>Monitoring snapshots of the Paimon table.
+ * <li>Creating the Tuple2<{@link Split}, String> splits corresponding to
the incremental files
+ * <li>Assigning them to downstream tasks for further processing.
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link
MultiTablesReadOperator} which
+ * can have parallelism greater than one.
+ *
+ * <p>Currently, only dedicated compaction job for multi-tables rely on this
monitor.
+ */
+public abstract class MultiTablesCompactorSourceFunction
+ extends RichSourceFunction<Tuple2<Split, String>> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(MultiTablesCompactorSourceFunction.class);
+
+ protected final Catalog.Loader catalogLoader;
+ protected final Pattern includingPattern;
+ protected final Pattern excludingPattern;
+ protected final Pattern databasePattern;
+ protected final boolean isStreaming;
+ protected final long monitorInterval;
+
+ protected transient Catalog catalog;
+ protected transient Map<Identifier, BucketsTable> tablesMap;
+ protected transient Map<Identifier, StreamTableScan> scansMap;
+
+ public MultiTablesCompactorSourceFunction(
+ Catalog.Loader catalogLoader,
+ Pattern includingPattern,
+ Pattern excludingPattern,
+ Pattern databasePattern,
+ boolean isStreaming,
+ long monitorInterval) {
+ this.catalogLoader = catalogLoader;
+ this.includingPattern = includingPattern;
+ this.excludingPattern = excludingPattern;
+ this.databasePattern = databasePattern;
+ this.isStreaming = isStreaming;
+ this.monitorInterval = monitorInterval;
+ }
+
+ protected volatile boolean isRunning = true;
+
+ protected transient SourceContext<Tuple2<Split, String>> ctx;
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ tablesMap = new HashMap<>();
+ scansMap = new HashMap<>();
+ catalog = catalogLoader.load();
+
+ updateTableMap();
+ }
+
+ @Override
+ public void cancel() {
+ // this is to cover the case where cancel() is called before the run()
+ if (ctx != null) {
+ synchronized (ctx.getCheckpointLock()) {
+ isRunning = false;
+ }
+ } else {
+ isRunning = false;
+ }
+ }
+
+ protected void updateTableMap()
+ throws Catalog.DatabaseNotExistException,
Catalog.TableNotExistException {
+ List<String> databases = catalog.listDatabases();
+
+ for (String databaseName : databases) {
+ if (databasePattern.matcher(databaseName).matches()) {
+ List<String> tables = catalog.listTables(databaseName);
+ for (String tableName : tables) {
+ Identifier identifier = Identifier.create(databaseName,
tableName);
+ if (shouldCompactTable(identifier, includingPattern,
excludingPattern)
+ && (!tablesMap.containsKey(identifier))) {
+ Table table = catalog.getTable(identifier);
+ if (!(table instanceof FileStoreTable)) {
+ LOG.error(
+ String.format(
+ "Only FileStoreTable supports
compact action. The table type is '%s'.",
+ table.getClass().getName()));
+ continue;
+ }
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ if (fileStoreTable.bucketMode() == BucketMode.UNAWARE)
{
+ LOG.info(
+ String.format(
+ "the bucket mode of %s is
unware. ",
+ identifier.getFullName())
+ + "currently, the table with
unware bucket mode is not support in combined mode.");
+ continue;
+ }
+ BucketsTable bucketsTable =
+ new BucketsTable(
+ fileStoreTable,
+ isStreaming,
+ identifier.getDatabaseName())
+ .copy(compactOptions(isStreaming));
+ tablesMap.put(identifier, bucketsTable);
+ scansMap.put(identifier,
bucketsTable.newReadBuilder().newStreamScan());
+ }
+ }
+ }
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesReadOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesReadOperator.java
new file mode 100644
index 000000000..37366a239
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesReadOperator.java
@@ -0,0 +1,125 @@
+/*
+ * 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.source.operator;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManagerImpl;
+import org.apache.paimon.flink.FlinkRowData;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.table.system.BucketsTable;
+import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+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 java.util.HashMap;
+import java.util.Map;
+
+import static
org.apache.paimon.flink.utils.MultiTablesCompactorUtil.compactOptions;
+
+/**
+ * The operator that reads the Tuple2<{@link Split}, String> received from the
preceding {@link
+ * MultiTablesBatchCompactorSourceFunction} or {@link
MultiTablesStreamingCompactorSourceFunction}.
+ * Contrary to the {@link MultiTablesCompactorSourceFunction} which has a
parallelism of 1, this
+ * operator can have DOP > 1.
+ */
+public class MultiTablesReadOperator extends AbstractStreamOperator<RowData>
+ implements OneInputStreamOperator<Tuple2<Split, String>, RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Catalog.Loader catalogLoader;
+ private final boolean isStreaming;
+
+ public MultiTablesReadOperator(Catalog.Loader catalogLoader, boolean
isStreaming) {
+ this.catalogLoader = catalogLoader;
+ this.isStreaming = isStreaming;
+ }
+
+ private transient Catalog catalog;
+ private transient IOManagerImpl ioManager;
+ private transient Map<Identifier, BucketsTable> tablesMap;
+ private transient Map<Identifier, TableRead> readsMap;
+ private transient StreamRecord<RowData> reuseRecord;
+ private transient FlinkRowData reuseRow;
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ ioManager =
+ new IOManagerImpl(
+ getContainingTask()
+ .getEnvironment()
+ .getIOManager()
+ .getSpillingDirectoriesPaths());
+ tablesMap = new HashMap<>();
+ readsMap = new HashMap<>();
+ catalog = catalogLoader.load();
+
+ this.reuseRow = new FlinkRowData(null);
+ this.reuseRecord = new StreamRecord<>(reuseRow);
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple2<Split, String>> record)
throws Exception {
+ Identifier identifier = Identifier.fromString(record.getValue().f1);
+ TableRead read = getTableRead(identifier);
+ try (CloseableIterator<InternalRow> iterator =
+ read.createReader(record.getValue().f0).toCloseableIterator())
{
+ while (iterator.hasNext()) {
+ reuseRow.replace(iterator.next());
+ output.collect(reuseRecord);
+ }
+ }
+ }
+
+ private TableRead getTableRead(Identifier tableId) {
+ BucketsTable table = tablesMap.get(tableId);
+ if (table == null) {
+ try {
+ Table newTable = catalog.getTable(tableId);
+ Preconditions.checkArgument(
+ newTable instanceof FileStoreTable,
+ "Only FileStoreTable supports compact action. The
table type is '%s'.",
+ newTable.getClass().getName());
+ table =
+ new BucketsTable(
+ (FileStoreTable) newTable,
+ isStreaming,
+ tableId.getDatabaseName())
+ .copy(compactOptions(isStreaming));
+ tablesMap.put(tableId, table);
+ readsMap.put(tableId,
table.newReadBuilder().newRead().withIOManager(ioManager));
+ } catch (Catalog.TableNotExistException e) {
+ LOG.error(String.format("table: %s not found.",
tableId.getFullName()));
+ }
+ }
+
+ return readsMap.get(tableId);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesStreamingCompactorSourceFunction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesStreamingCompactorSourceFunction.java
new file mode 100644
index 000000000..61e085f32
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesStreamingCompactorSourceFunction.java
@@ -0,0 +1,144 @@
+/*
+ * 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.source.operator;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.utils.JavaTypeInfo;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.EndOfScanException;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.StreamTableScan;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.operators.StreamSource;
+import org.apache.flink.table.data.RowData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+/** It is responsible for monitoring compactor source in streaming mode. */
+public class MultiTablesStreamingCompactorSourceFunction
+ extends MultiTablesCompactorSourceFunction {
+
+ private static final Logger LOG =
+
LoggerFactory.getLogger(MultiTablesStreamingCompactorSourceFunction.class);
+
+ public MultiTablesStreamingCompactorSourceFunction(
+ Catalog.Loader catalogLoader,
+ Pattern includingPattern,
+ Pattern excludingPattern,
+ Pattern databasePattern,
+ long monitorInterval) {
+ super(
+ catalogLoader,
+ includingPattern,
+ excludingPattern,
+ databasePattern,
+ true,
+ monitorInterval);
+ }
+
+ @SuppressWarnings("BusyWait")
+ @Override
+ public void run(SourceContext<Tuple2<Split, String>> ctx) throws Exception
{
+ this.ctx = ctx;
+ while (isRunning) {
+ boolean isEmpty;
+ synchronized (ctx.getCheckpointLock()) {
+ if (!isRunning) {
+ return;
+ }
+ try {
+ // check for new tables
+ updateTableMap();
+
+ List<Tuple2<Split, String>> splits = new ArrayList<>();
+ for (Map.Entry<Identifier, StreamTableScan> entry :
scansMap.entrySet()) {
+ Identifier identifier = entry.getKey();
+ StreamTableScan scan = entry.getValue();
+ splits.addAll(
+ scan.plan().splits().stream()
+ .map(split -> new Tuple2<>(split,
identifier.getFullName()))
+ .collect(Collectors.toList()));
+ }
+
+ isEmpty = splits.isEmpty();
+ splits.forEach(ctx::collect);
+
+ } catch (EndOfScanException esf) {
+ LOG.info("Catching EndOfStreamException, the stream is
finished.");
+ return;
+ }
+ }
+
+ if (isEmpty) {
+ Thread.sleep(monitorInterval);
+ }
+ }
+ }
+
+ public static DataStream<RowData> buildSource(
+ StreamExecutionEnvironment env,
+ String name,
+ TypeInformation<RowData> typeInfo,
+ Catalog.Loader catalogLoader,
+ Pattern includingPattern,
+ Pattern excludingPattern,
+ Pattern databasePattern,
+ long monitorInterval) {
+
+ MultiTablesStreamingCompactorSourceFunction function =
+ new MultiTablesStreamingCompactorSourceFunction(
+ catalogLoader,
+ includingPattern,
+ excludingPattern,
+ databasePattern,
+ monitorInterval);
+ StreamSource<Tuple2<Split, String>, ?> sourceOperator = new
StreamSource<>(function);
+ boolean isParallel = false;
+ TupleTypeInfo<Tuple2<Split, String>> tupleTypeInfo =
+ new TupleTypeInfo<>(
+ new JavaTypeInfo<>(Split.class),
BasicTypeInfo.STRING_TYPE_INFO);
+ return new DataStreamSource<>(
+ env,
+ tupleTypeInfo,
+ sourceOperator,
+ isParallel,
+ name,
+ Boundedness.CONTINUOUS_UNBOUNDED)
+ .forceNonParallel()
+ .partitionCustom(
+ (key, numPartitions) -> key % numPartitions,
+ split -> ((DataSplit) split.f0).bucket())
+ .transform(name, typeInfo, new
MultiTablesReadOperator(catalogLoader, true));
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/MultiTablesCompactorUtil.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/MultiTablesCompactorUtil.java
new file mode 100644
index 000000000..d26a9fe12
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/MultiTablesCompactorUtil.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.flink.utils;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Identifier;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.regex.Pattern;
+
+/** Utility methods for dedicated compaction job for multi-tables. */
+public class MultiTablesCompactorUtil {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(MultiTablesCompactorUtil.class);
+
+ public static Map<String, String> compactOptions(boolean isStreaming) {
+ if (isStreaming) {
+ // set 'streaming-compact' and remove 'scan.bounded.watermark'
+ return new HashMap<String, String>() {
+ {
+ put(
+ CoreOptions.STREAMING_COMPACT.key(),
+
CoreOptions.StreamingCompactionType.NORMAL.getValue());
+ put(CoreOptions.SCAN_BOUNDED_WATERMARK.key(), null);
+ put(CoreOptions.WRITE_ONLY.key(), "false");
+ }
+ };
+ } else {
+ // batch compactor source will compact all current files
+ return new HashMap<String, String>() {
+ {
+ put(CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), null);
+ put(CoreOptions.SCAN_SNAPSHOT_ID.key(), null);
+ put(
+ CoreOptions.SCAN_MODE.key(),
+ CoreOptions.StartupMode.LATEST_FULL.toString());
+ put(CoreOptions.WRITE_ONLY.key(), "false");
+ }
+ };
+ }
+ }
+
+ public static boolean shouldCompactTable(
+ Identifier tableIdentifier, Pattern includingPattern, Pattern
excludingPattern) {
+ String paimonFullTableName = tableIdentifier.getFullName();
+ boolean shouldCompaction =
includingPattern.matcher(paimonFullTableName).matches();
+ if (excludingPattern != null) {
+ shouldCompaction =
+ shouldCompaction &&
!excludingPattern.matcher(paimonFullTableName).matches();
+ }
+ if (!shouldCompaction) {
+ LOG.debug("Source table '{}' is excluded.", paimonFullTableName);
+ }
+ return shouldCompaction;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
index b78eec829..0d15e9606 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
@@ -22,10 +22,13 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.operation.FileStoreScan;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
+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.source.DataSplit;
import org.apache.paimon.table.source.StreamTableScan;
@@ -39,9 +42,10 @@ import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.util.ArrayList;
@@ -59,23 +63,14 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
private static final String[] DATABASE_NAMES = new String[] {"db1", "db2"};
private static final String[] TABLE_NAMES = new String[] {"t1", "t2"};
- private static final Map<String, RowType> ROW_TYPE_MAP = new
HashMap<>(TABLE_NAMES.length);
-
- @BeforeAll
- public static void beforeAll() {
- // set different datatype and RowType
- DataType[] dataTypes1 =
- new DataType[] {
- DataTypes.INT(), DataTypes.INT(), DataTypes.INT(),
DataTypes.STRING()
- };
- DataType[] dataTypes2 =
- new DataType[] {
- DataTypes.INT(), DataTypes.BIGINT(), DataTypes.INT(),
DataTypes.STRING()
- };
-
- ROW_TYPE_MAP.put("t1", RowType.of(dataTypes1, new String[] {"k", "v",
"hh", "dt"}));
- ROW_TYPE_MAP.put("t2", RowType.of(dataTypes2, new String[] {"k", "v1",
"hh", "dt"}));
- }
+ private static final String[] New_DATABASE_NAMES = new String[] {"db3",
"db4"};
+ private static final String[] New_TABLE_NAMES = new String[] {"t3", "t4"};
+ private static final RowType ROW_TYPE =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT(), DataTypes.INT(), DataTypes.INT(),
DataTypes.STRING()
+ },
+ new String[] {"k", "v", "hh", "dt"});
private FileStoreTable createTable(
String databaseName,
@@ -95,12 +90,15 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
return (FileStoreTable) catalog.getTable(identifier);
}
- @Test
+ @ParameterizedTest(name = "mode = {0}")
+ @ValueSource(strings = {"divided", "combined"})
@Timeout(60)
- public void testBatchCompact() throws Exception {
+ public void testBatchCompact(String mode) throws Exception {
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.WRITE_ONLY.key(), "true");
+ Map<String, String> compactOptions = new HashMap<>();
+
List<FileStoreTable> tables = new ArrayList<>();
for (String dbName : DATABASE_NAMES) {
@@ -109,7 +107,7 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
createTable(
dbName,
tableName,
- ROW_TYPE_MAP.get(tableName),
+ ROW_TYPE,
Arrays.asList("dt", "hh"),
Arrays.asList("dt", "hh", "k"),
options);
@@ -120,22 +118,15 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
write = streamWriteBuilder.newWrite();
commit = streamWriteBuilder.newCommit();
- Object value = null;
- if (tableName.equals("t1")) {
- value = 100;
- } else if (tableName.equals("t2")) {
- value = 100L;
- }
-
writeData(
- rowData(1, value, 15,
BinaryString.fromString("20221208")),
- rowData(1, value, 16,
BinaryString.fromString("20221208")),
- rowData(1, value, 15,
BinaryString.fromString("20221209")));
+ rowData(1, 100, 15,
BinaryString.fromString("20221208")),
+ rowData(1, 100, 16,
BinaryString.fromString("20221208")),
+ rowData(1, 100, 15,
BinaryString.fromString("20221209")));
writeData(
- rowData(2, value, 15,
BinaryString.fromString("20221208")),
- rowData(2, value, 16,
BinaryString.fromString("20221208")),
- rowData(2, value, 15,
BinaryString.fromString("20221209")));
+ rowData(2, 100, 15,
BinaryString.fromString("20221208")),
+ rowData(2, 100, 16,
BinaryString.fromString("20221208")),
+ rowData(2, 100, 15,
BinaryString.fromString("20221209")));
Snapshot snapshot =
snapshotManager.snapshot(snapshotManager.latestSnapshotId());
assertThat(snapshot.id()).isEqualTo(2);
@@ -146,10 +137,26 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
env.setParallelism(ThreadLocalRandom.current().nextInt(2) + 1);
- new CompactDatabaseAction(warehouse, "db1|db2", null, null, new
HashMap<>()).build(env);
+ if (mode.equals("divided")) {
+ new CompactDatabaseAction(warehouse, new HashMap<>())
+ .includingDatabases(null)
+ .includingTables(null)
+ .excludingTables(null)
+ .build(env);
+ } else {
+ new CompactDatabaseAction(warehouse, new HashMap<>())
+ .includingDatabases(null)
+ .includingTables(null)
+ .excludingTables(null)
+ .withDatabaseCompactMode("combined")
+ .withCompactOptions(compactOptions)
+ .build(env);
+ }
+
env.execute();
for (FileStoreTable table : tables) {
+ snapshotManager = table.snapshotManager();
Snapshot snapshot =
table.snapshotManager().snapshot(snapshotManager.latestSnapshotId());
assertThat(snapshot.id()).isEqualTo(3);
@@ -163,10 +170,232 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
}
}
- @Test
+ @ParameterizedTest(name = "mode = {0}")
+ @ValueSource(strings = {"divided", "combined"})
+ public void testStreamingCompact(String mode) throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.CHANGELOG_PRODUCER.key(), "full-compaction");
+ options.put(
+
FlinkConnectorOptions.CHANGELOG_PRODUCER_FULL_COMPACTION_TRIGGER_INTERVAL.key(),
+ "1s");
+ options.put(CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL.key(), "1s");
+ options.put(CoreOptions.WRITE_ONLY.key(), "true");
+ // test that dedicated compact job will expire snapshots
+ options.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "3");
+ options.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "3");
+
+ Map<String, String> compactOptions = new HashMap<>();
+ // if CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL.key() use default
value, the cost time in
+ // combined mode will be over 1mins
+ compactOptions.put(CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL.key(),
"1s");
+
+ List<FileStoreTable> tables = new ArrayList<>();
+ for (String dbName : DATABASE_NAMES) {
+ for (String tableName : TABLE_NAMES) {
+ FileStoreTable table =
+ createTable(
+ dbName,
+ tableName,
+ ROW_TYPE,
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ options);
+ tables.add(table);
+ snapshotManager = table.snapshotManager();
+ StreamWriteBuilder streamWriteBuilder =
+
table.newStreamWriteBuilder().withCommitUser(commitUser);
+ write = streamWriteBuilder.newWrite();
+ commit = streamWriteBuilder.newCommit();
+
+ // base records
+ writeData(
+ rowData(1, 100, 15,
BinaryString.fromString("20221208")),
+ rowData(1, 100, 16,
BinaryString.fromString("20221208")),
+ rowData(1, 100, 15,
BinaryString.fromString("20221209")));
+
+ Snapshot snapshot =
snapshotManager.snapshot(snapshotManager.latestSnapshotId());
+ assertThat(snapshot.id()).isEqualTo(1);
+
assertThat(snapshot.commitKind()).isEqualTo(Snapshot.CommitKind.APPEND);
+
+ // no full compaction has happened, so plan should be empty
+ StreamTableScan scan = table.newReadBuilder().newStreamScan();
+ TableScan.Plan plan = scan.plan();
+ assertThat(plan.splits()).isEmpty();
+ }
+ }
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
+
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
+ env.getCheckpointConfig().setCheckpointInterval(500);
+ env.setParallelism(ThreadLocalRandom.current().nextInt(2) + 1);
+ if (mode.equals("divided")) {
+ new CompactDatabaseAction(warehouse, new HashMap<>())
+ .includingDatabases(null)
+ .includingTables(null)
+ .excludingTables(null)
+ .build(env);
+ } else {
+ new CompactDatabaseAction(warehouse, new HashMap<>())
+ .includingDatabases(null)
+ .includingTables(null)
+ .excludingTables(null)
+ .withDatabaseCompactMode("combined")
+ .withCompactOptions(compactOptions)
+ .build(env);
+ }
+ JobClient client = env.executeAsync();
+
+ for (FileStoreTable table : tables) {
+ StreamTableScan scan = table.newReadBuilder().newStreamScan();
+ // first full compaction
+ validateResult(
+ table,
+ ROW_TYPE,
+ scan,
+ Arrays.asList(
+ "+I[1, 100, 15, 20221208]",
+ "+I[1, 100, 15, 20221209]",
+ "+I[1, 100, 16, 20221208]"),
+ 60_000);
+
+ snapshotManager = table.snapshotManager();
+ StreamWriteBuilder streamWriteBuilder =
+ table.newStreamWriteBuilder().withCommitUser(commitUser);
+ write = streamWriteBuilder.newWrite();
+ commit = streamWriteBuilder.newCommit();
+
+ // incremental records
+ writeData(
+ rowData(1, 101, 15, BinaryString.fromString("20221208")),
+ rowData(1, 101, 16, BinaryString.fromString("20221208")),
+ rowData(1, 101, 15, BinaryString.fromString("20221209")));
+
+ // second full compaction
+ validateResult(
+ table,
+ ROW_TYPE,
+ scan,
+ Arrays.asList(
+ "+U[1, 101, 15, 20221208]",
+ "+U[1, 101, 15, 20221209]",
+ "+U[1, 101, 16, 20221208]",
+ "-U[1, 100, 15, 20221208]",
+ "-U[1, 100, 15, 20221209]",
+ "-U[1, 100, 16, 20221208]"),
+ 60_000);
+
+ // assert dedicated compact job will expire snapshots
+ CommonTestUtils.waitUtil(
+ () ->
+ snapshotManager.latestSnapshotId() - 2
+ == snapshotManager.earliestSnapshotId(),
+ Duration.ofSeconds(60_000),
+ Duration.ofSeconds(100),
+ String.format(
+ "Cannot validate snapshot expiration in %s
milliseconds.", 60_000));
+ }
+
+ // In combined mode, check whether newly created table can be detected
+ if (mode.equals("combined")) {
+ // second create tables and write data to tables
+ List<FileStoreTable> newtables = new ArrayList<>();
+ for (String dbName : New_DATABASE_NAMES) {
+ for (String tableName : New_TABLE_NAMES) {
+ FileStoreTable table =
+ createTable(
+ dbName,
+ tableName,
+ ROW_TYPE,
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ options);
+ newtables.add(table);
+ snapshotManager = table.snapshotManager();
+ StreamWriteBuilder streamWriteBuilder =
+
table.newStreamWriteBuilder().withCommitUser(commitUser);
+ write = streamWriteBuilder.newWrite();
+ commit = streamWriteBuilder.newCommit();
+
+ // base records
+ writeData(
+ write,
+ commit,
+ 0,
+ rowData(1, 100, 15,
BinaryString.fromString("20221208")),
+ rowData(1, 100, 16,
BinaryString.fromString("20221208")),
+ rowData(1, 100, 15,
BinaryString.fromString("20221209")));
+
+ Snapshot snapshot =
+
snapshotManager.snapshot(snapshotManager.latestSnapshotId());
+ assertThat(snapshot.id()).isEqualTo(1);
+
assertThat(snapshot.commitKind()).isEqualTo(Snapshot.CommitKind.APPEND);
+ }
+ }
+
+ for (FileStoreTable table : newtables) {
+ StreamTableScan scan = table.newReadBuilder().newStreamScan();
+ // first full compaction for new tables
+ validateResult(
+ table,
+ ROW_TYPE,
+ scan,
+ Arrays.asList(
+ "+I[1, 100, 15, 20221208]",
+ "+I[1, 100, 15, 20221209]",
+ "+I[1, 100, 16, 20221208]"),
+ 60_000);
+
+ snapshotManager = table.snapshotManager();
+ StreamWriteBuilder streamWriteBuilder =
+
table.newStreamWriteBuilder().withCommitUser(commitUser);
+ write = streamWriteBuilder.newWrite();
+ commit = streamWriteBuilder.newCommit();
+
+ // incremental records
+ writeData(
+ write,
+ commit,
+ 1,
+ rowData(1, 101, 15,
BinaryString.fromString("20221208")),
+ rowData(1, 101, 16,
BinaryString.fromString("20221208")),
+ rowData(1, 101, 15,
BinaryString.fromString("20221209")));
+
+ // second full compaction for new tables
+ validateResult(
+ table,
+ ROW_TYPE,
+ scan,
+ Arrays.asList(
+ "+U[1, 101, 15, 20221208]",
+ "+U[1, 101, 15, 20221209]",
+ "+U[1, 101, 16, 20221208]",
+ "-U[1, 100, 15, 20221208]",
+ "-U[1, 100, 15, 20221209]",
+ "-U[1, 100, 16, 20221208]"),
+ 60_000);
+
+ // assert dedicated compact job will expire snapshots
+ CommonTestUtils.waitUtil(
+ () ->
+ snapshotManager.latestSnapshotId() - 2
+ ==
snapshotManager.earliestSnapshotId(),
+ Duration.ofSeconds(60_000),
+ Duration.ofSeconds(100),
+ String.format(
+ "Cannot validate snapshot expiration in %s
milliseconds.", 60_000));
+ }
+ }
+
+ client.cancel();
+ }
+
+ @ParameterizedTest(name = "mode = {0}")
+ @ValueSource(strings = {"divided", "combined"})
@Timeout(60)
- public void includeTableCompaction() throws Exception {
+ public void includeTableCompaction(String mode) throws Exception {
includingAndExcludingTablesImpl(
+ mode,
"db1.t1",
null,
Collections.singletonList(Identifier.fromString("db1.t1")),
@@ -176,10 +405,12 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
Identifier.fromString("db2.t2")));
}
- @Test
+ @ParameterizedTest(name = "mode = {0}")
+ @ValueSource(strings = {"divided", "combined"})
@Timeout(60)
- public void excludeTableCompaction() throws Exception {
+ public void excludeTableCompaction(String mode) throws Exception {
includingAndExcludingTablesImpl(
+ mode,
null,
"db2.t2",
Arrays.asList(
@@ -189,10 +420,12 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
Collections.singletonList(Identifier.fromString("db2.t2")));
}
- @Test
+ @ParameterizedTest(name = "mode = {0}")
+ @ValueSource(strings = {"divided", "combined"})
@Timeout(60)
- public void includeAndExcludeTableCompaction() throws Exception {
+ public void includeAndExcludeTableCompaction(String mode) throws Exception
{
includingAndExcludingTablesImpl(
+ mode,
"db1.+|db2.t1",
"db1.t2",
Arrays.asList(Identifier.fromString("db1.t1"),
Identifier.fromString("db2.t1")),
@@ -200,6 +433,7 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
}
private void includingAndExcludingTablesImpl(
+ String mode,
String includingPattern,
String excludesPattern,
List<Identifier> includeTables,
@@ -208,6 +442,9 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.WRITE_ONLY.key(), "true");
+ Map<String, String> compactOptions = new HashMap<>();
+ compactOptions.put(CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL.key(),
"1s");
+
List<FileStoreTable> compactionTables = new ArrayList<>();
List<FileStoreTable> noCompactionTables = new ArrayList<>();
@@ -217,7 +454,7 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
createTable(
dbName,
tableName,
- ROW_TYPE_MAP.get(tableName),
+ ROW_TYPE,
Arrays.asList("dt", "hh"),
Arrays.asList("dt", "hh", "k"),
options);
@@ -233,22 +470,15 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
write = streamWriteBuilder.newWrite();
commit = streamWriteBuilder.newCommit();
- Object value = null;
- if (tableName.equals("t1")) {
- value = 100;
- } else if (tableName.equals("t2")) {
- value = 100L;
- }
-
writeData(
- rowData(1, value, 15,
BinaryString.fromString("20221208")),
- rowData(1, value, 16,
BinaryString.fromString("20221208")),
- rowData(1, value, 15,
BinaryString.fromString("20221209")));
+ rowData(1, 100, 15,
BinaryString.fromString("20221208")),
+ rowData(1, 100, 16,
BinaryString.fromString("20221208")),
+ rowData(1, 100, 15,
BinaryString.fromString("20221209")));
writeData(
- rowData(2, value, 15,
BinaryString.fromString("20221208")),
- rowData(2, value, 16,
BinaryString.fromString("20221208")),
- rowData(2, value, 15,
BinaryString.fromString("20221209")));
+ rowData(2, 100, 15,
BinaryString.fromString("20221208")),
+ rowData(2, 100, 16,
BinaryString.fromString("20221208")),
+ rowData(2, 100, 15,
BinaryString.fromString("20221209")));
Snapshot snapshot =
snapshotManager.snapshot(snapshotManager.latestSnapshotId());
assertThat(snapshot.id()).isEqualTo(2);
@@ -259,9 +489,22 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
env.setParallelism(ThreadLocalRandom.current().nextInt(2) + 1);
- new CompactDatabaseAction(
- warehouse, "db1|db2", includingPattern,
excludesPattern, new HashMap<>())
- .build(env);
+
+ if (mode.equals("divided")) {
+ new CompactDatabaseAction(warehouse, new HashMap<>())
+ .includingDatabases(null)
+ .includingTables(includingPattern)
+ .excludingTables(excludesPattern)
+ .build(env);
+ } else {
+ new CompactDatabaseAction(warehouse, new HashMap<>())
+ .includingDatabases(null)
+ .includingTables(includingPattern)
+ .excludingTables(excludesPattern)
+ .withDatabaseCompactMode("combined")
+ .withCompactOptions(compactOptions)
+ .build(env);
+ }
env.execute();
for (FileStoreTable table : compactionTables) {
@@ -295,130 +538,6 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
}
}
- @Test
- public void testStreamingCompact() throws Exception {
- Map<String, String> options = new HashMap<>();
- options.put(CoreOptions.CHANGELOG_PRODUCER.key(), "full-compaction");
- options.put(
-
FlinkConnectorOptions.CHANGELOG_PRODUCER_FULL_COMPACTION_TRIGGER_INTERVAL.key(),
- "1s");
- options.put(CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL.key(), "1s");
- options.put(CoreOptions.WRITE_ONLY.key(), "true");
- // test that dedicated compact job will expire snapshots
- options.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "3");
- options.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "3");
-
- List<FileStoreTable> tables = new ArrayList<>();
- for (String dbName : DATABASE_NAMES) {
- for (String tableName : TABLE_NAMES) {
- FileStoreTable table =
- createTable(
- dbName,
- tableName,
- ROW_TYPE_MAP.get(tableName),
- Arrays.asList("dt", "hh"),
- Arrays.asList("dt", "hh", "k"),
- options);
- tables.add(table);
- snapshotManager = table.snapshotManager();
- StreamWriteBuilder streamWriteBuilder =
-
table.newStreamWriteBuilder().withCommitUser(commitUser);
- write = streamWriteBuilder.newWrite();
- commit = streamWriteBuilder.newCommit();
-
- Object value = null;
- if (tableName.equals("t1")) {
- value = 100;
- } else if (tableName.equals("t2")) {
- value = 100L;
- }
-
- // base records
- writeData(
- rowData(1, value, 15,
BinaryString.fromString("20221208")),
- rowData(1, value, 16,
BinaryString.fromString("20221208")),
- rowData(1, value, 15,
BinaryString.fromString("20221209")));
-
- Snapshot snapshot =
snapshotManager.snapshot(snapshotManager.latestSnapshotId());
- assertThat(snapshot.id()).isEqualTo(1);
-
assertThat(snapshot.commitKind()).isEqualTo(Snapshot.CommitKind.APPEND);
-
- // no full compaction has happened, so plan should be empty
- StreamTableScan scan = table.newReadBuilder().newStreamScan();
- TableScan.Plan plan = scan.plan();
- assertThat(plan.splits()).isEmpty();
- }
- }
-
- StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
- env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
-
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
- env.getCheckpointConfig().setCheckpointInterval(500);
- env.setParallelism(ThreadLocalRandom.current().nextInt(2) + 1);
- new CompactDatabaseAction(warehouse, "db1|db2", null, null, new
HashMap<>()).build(env);
- JobClient client = env.executeAsync();
-
- for (FileStoreTable table : tables) {
- StreamTableScan scan = table.newReadBuilder().newStreamScan();
- // first full compaction
- validateResult(
- table,
- ROW_TYPE_MAP.get(table.name()),
- scan,
- Arrays.asList(
- "+I[1, 100, 15, 20221208]",
- "+I[1, 100, 15, 20221209]",
- "+I[1, 100, 16, 20221208]"),
- 60_000);
-
- Object value = null;
- String tName = table.name();
- if (tName.equals("t1")) {
- value = 101;
- } else if (tName.equals("t2")) {
- value = 101L;
- }
-
- snapshotManager = table.snapshotManager();
- StreamWriteBuilder streamWriteBuilder =
- table.newStreamWriteBuilder().withCommitUser(commitUser);
- write = streamWriteBuilder.newWrite();
- commit = streamWriteBuilder.newCommit();
-
- // incremental records
- writeData(
- rowData(1, value, 15, BinaryString.fromString("20221208")),
- rowData(1, value, 16, BinaryString.fromString("20221208")),
- rowData(1, value, 15,
BinaryString.fromString("20221209")));
-
- // second full compaction
- validateResult(
- table,
- ROW_TYPE_MAP.get(table.name()),
- scan,
- Arrays.asList(
- "+U[1, 101, 15, 20221208]",
- "+U[1, 101, 15, 20221209]",
- "+U[1, 101, 16, 20221208]",
- "-U[1, 100, 15, 20221208]",
- "-U[1, 100, 15, 20221209]",
- "-U[1, 100, 16, 20221208]"),
- 60_000);
-
- // assert dedicated compact job will expire snapshots
- CommonTestUtils.waitUtil(
- () ->
- snapshotManager.latestSnapshotId() - 2
- == snapshotManager.earliestSnapshotId(),
- Duration.ofSeconds(60_000),
- Duration.ofSeconds(100),
- String.format(
- "Cannot validate snapshot expiration in %s
milliseconds.", 60_000));
- }
-
- client.cancel();
- }
-
@Test
public void testUnawareBucketStreamingCompact() throws Exception {
Map<String, String> options = new HashMap<>();
@@ -434,7 +553,7 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
createTable(
database,
tableName,
- ROW_TYPE_MAP.get(tableName),
+ ROW_TYPE,
Arrays.asList("k"),
Collections.emptyList(),
options);
@@ -445,23 +564,16 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
write = streamWriteBuilder.newWrite();
commit = streamWriteBuilder.newCommit();
- Object value = null;
- if (tableName.equals("t1")) {
- value = 100;
- } else if (tableName.equals("t2")) {
- value = 100L;
- }
-
// base records
writeData(
- rowData(1, value, 15, BinaryString.fromString("20221208")),
- rowData(1, value, 16, BinaryString.fromString("20221208")),
- rowData(1, value, 15,
BinaryString.fromString("20221209")));
+ rowData(1, 100, 15, BinaryString.fromString("20221208")),
+ rowData(1, 100, 16, BinaryString.fromString("20221208")),
+ rowData(1, 100, 15, BinaryString.fromString("20221209")));
writeData(
- rowData(1, value, 15, BinaryString.fromString("20221208")),
- rowData(1, value, 16, BinaryString.fromString("20221208")),
- rowData(1, value, 15,
BinaryString.fromString("20221209")));
+ rowData(1, 100, 15, BinaryString.fromString("20221208")),
+ rowData(1, 100, 16, BinaryString.fromString("20221208")),
+ rowData(1, 100, 15, BinaryString.fromString("20221209")));
Snapshot snapshot =
snapshotManager.snapshot(snapshotManager.latestSnapshotId());
assertThat(snapshot.id()).isEqualTo(2);
@@ -473,7 +585,11 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointInterval(500);
env.setParallelism(ThreadLocalRandom.current().nextInt(2) + 1);
- new CompactDatabaseAction(warehouse, database, null, null, new
HashMap<>()).build(env);
+ new CompactDatabaseAction(warehouse, new HashMap<>())
+ .includingDatabases(null)
+ .includingTables(null)
+ .excludingTables(null)
+ .build(env);
JobClient client = env.executeAsync();
for (FileStoreTable table : tables) {
@@ -488,18 +604,10 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
// first compaction, snapshot will be 3
checkFileAndRowSize(storeScan, 3L, 30_000L, 1, 6);
- Object value = null;
- String tName = table.name();
- if (tName.equals("t1")) {
- value = 101;
- } else if (tName.equals("t2")) {
- value = 101L;
- }
-
writeData(
- rowData(1, value, 15, BinaryString.fromString("20221208")),
- rowData(1, value, 16, BinaryString.fromString("20221208")),
- rowData(1, value, 15,
BinaryString.fromString("20221209")));
+ rowData(1, 101, 15, BinaryString.fromString("20221208")),
+ rowData(1, 101, 16, BinaryString.fromString("20221208")),
+ rowData(1, 101, 15, BinaryString.fromString("20221209")));
// second compaction, snapshot will be 5
checkFileAndRowSize(storeScan, 5L, 30_000L, 1, 9);
@@ -522,7 +630,7 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
createTable(
database,
tableName,
- ROW_TYPE_MAP.get(tableName),
+ ROW_TYPE,
Collections.singletonList("k"),
Collections.emptyList(),
options);
@@ -533,23 +641,16 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
write = streamWriteBuilder.newWrite();
commit = streamWriteBuilder.newCommit();
- Object value = null;
- if (tableName.equals("t1")) {
- value = 100;
- } else if (tableName.equals("t2")) {
- value = 100L;
- }
-
// base records
writeData(
- rowData(1, value, 15, BinaryString.fromString("20221208")),
- rowData(1, value, 16, BinaryString.fromString("20221208")),
- rowData(1, value, 15,
BinaryString.fromString("20221209")));
+ rowData(1, 100, 15, BinaryString.fromString("20221208")),
+ rowData(1, 100, 16, BinaryString.fromString("20221208")),
+ rowData(1, 100, 15, BinaryString.fromString("20221209")));
writeData(
- rowData(1, value, 15, BinaryString.fromString("20221208")),
- rowData(1, value, 16, BinaryString.fromString("20221208")),
- rowData(1, value, 15,
BinaryString.fromString("20221209")));
+ rowData(1, 100, 15, BinaryString.fromString("20221208")),
+ rowData(1, 100, 16, BinaryString.fromString("20221208")),
+ rowData(1, 100, 15, BinaryString.fromString("20221209")));
Snapshot snapshot =
snapshotManager.snapshot(snapshotManager.latestSnapshotId());
assertThat(snapshot.id()).isEqualTo(2);
@@ -559,7 +660,11 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
env.setParallelism(ThreadLocalRandom.current().nextInt(2) + 1);
- new CompactDatabaseAction(warehouse, database, null, null, new
HashMap<>()).build(env);
+ new CompactDatabaseAction(warehouse, new HashMap<>())
+ .includingDatabases(null)
+ .includingTables(null)
+ .excludingTables(null)
+ .build(env);
env.execute();
for (FileStoreTable table : tables) {
@@ -569,4 +674,16 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
checkFileAndRowSize(storeScan, 3L, 0L, 1, 6);
}
}
+
+ private void writeData(
+ StreamTableWrite write,
+ StreamTableCommit commit,
+ long incrementalIdentifier,
+ GenericRow... data)
+ throws Exception {
+ for (GenericRow d : data) {
+ write.write(d);
+ }
+ commit.commit(incrementalIdentifier, write.prepareCommit(true,
incrementalIdentifier));
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/MultiTablesCompactorSourceBuilderITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/MultiTablesCompactorSourceBuilderITCase.java
new file mode 100644
index 000000000..1a7127305
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/MultiTablesCompactorSourceBuilderITCase.java
@@ -0,0 +1,553 @@
+/*
+ * 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.source;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.flink.util.AbstractTestBase;
+import org.apache.paimon.io.DataFileMetaSerializer;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+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.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.CloseableIterator;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.regex.Pattern;
+
+import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link MultiTablesCompactorSourceBuilder}. */
+public class MultiTablesCompactorSourceBuilderITCase extends AbstractTestBase
+ implements Serializable {
+ private String warehouse;
+ private Options catalogOptions;
+
+ private String commitUser;
+
+ private static final String[] DATABASE_NAMES = new String[] {"db1", "db2"};
+ private static final String[] TABLE_NAMES = new String[] {"t1", "t2"};
+ private static final String[] New_DATABASE_NAMES = new String[] {"db3"};
+ private static final String[] New_TABLE_NAMES = new String[] {"t1", "t2"};
+ private static final Map<String, RowType> ROW_TYPE_MAP = new
HashMap<>(TABLE_NAMES.length);
+
+ @BeforeAll
+ public static void beforeAll() {
+ // set different datatype and RowType
+ DataType[] dataTypes1 =
+ new DataType[] {
+ DataTypes.INT(), DataTypes.INT(), DataTypes.INT(),
DataTypes.STRING()
+ };
+ DataType[] dataTypes2 =
+ new DataType[] {
+ DataTypes.INT(), DataTypes.INT(), DataTypes.INT(),
DataTypes.STRING()
+ };
+
+ ROW_TYPE_MAP.put("t1", RowType.of(dataTypes1, new String[] {"k", "v",
"hh", "dt"}));
+ ROW_TYPE_MAP.put("t2", RowType.of(dataTypes2, new String[] {"k", "v1",
"hh", "dt"}));
+ }
+
+ @BeforeEach
+ public void before() throws IOException {
+ warehouse = getTempDirPath();
+ catalogOptions = new Options();
+ commitUser = UUID.randomUUID().toString();
+ }
+
+ @ParameterizedTest(name = "defaultOptions = {0}")
+ @ValueSource(booleans = {true, false})
+ public void testBatchRead(boolean defaultOptions) throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.WRITE_ONLY.key(), "true");
+ if (!defaultOptions) {
+ // change options to test whether CompactorSourceBuilder work
normally
+ options.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), "2");
+ }
+ long monitorInterval = 1000;
+
+ for (String dbName : DATABASE_NAMES) {
+ for (String tableName : TABLE_NAMES) {
+ FileStoreTable table =
+ createTable(
+ dbName,
+ tableName,
+ ROW_TYPE_MAP.get(tableName),
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ options);
+ SnapshotManager snapshotManager = table.snapshotManager();
+ StreamWriteBuilder streamWriteBuilder =
+
table.newStreamWriteBuilder().withCommitUser(commitUser);
+ StreamTableWrite write = streamWriteBuilder.newWrite();
+ StreamTableCommit commit = streamWriteBuilder.newCommit();
+
+ writeData(
+ write,
+ commit,
+ 0,
+ rowData(1, 100, 15,
BinaryString.fromString("20221208")),
+ rowData(1, 100, 16,
BinaryString.fromString("20221208")),
+ rowData(1, 100, 15,
BinaryString.fromString("20221209")));
+
+ writeData(
+ write,
+ commit,
+ 1,
+ rowData(2, 100, 15,
BinaryString.fromString("20221208")),
+ rowData(2, 100, 16,
BinaryString.fromString("20221208")),
+ rowData(2, 100, 15,
BinaryString.fromString("20221209")));
+
+ Snapshot snapshot =
snapshotManager.snapshot(snapshotManager.latestSnapshotId());
+ assertThat(snapshot.id()).isEqualTo(2);
+
assertThat(snapshot.commitKind()).isEqualTo(Snapshot.CommitKind.APPEND);
+
+ write.close();
+ commit.close();
+ }
+ }
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+ env.setParallelism(ThreadLocalRandom.current().nextInt(2) + 1);
+ DataStream<RowData> source =
+ new MultiTablesCompactorSourceBuilder(
+ catalogLoader(),
+ Pattern.compile("db1|db2"),
+ Pattern.compile(".*"),
+ null,
+ monitorInterval)
+ .withContinuousMode(false)
+ .withEnv(env)
+ .build();
+ CloseableIterator<RowData> it = source.executeAndCollect();
+ List<String> actual = new ArrayList<>();
+ while (it.hasNext()) {
+ actual.add(toString(it.next()));
+ }
+ assertThat(actual)
+ .hasSameElementsAs(
+ Arrays.asList(
+ "+I 2|20221208|15|0|0|db1|t1",
+ "+I 2|20221209|15|0|0|db1|t1",
+ "+I 2|20221208|16|0|0|db1|t1",
+ "+I 2|20221208|15|0|0|db1|t2",
+ "+I 2|20221209|15|0|0|db1|t2",
+ "+I 2|20221208|16|0|0|db1|t2",
+ "+I 2|20221208|15|0|0|db2|t1",
+ "+I 2|20221209|15|0|0|db2|t1",
+ "+I 2|20221208|16|0|0|db2|t1",
+ "+I 2|20221208|15|0|0|db2|t2",
+ "+I 2|20221209|15|0|0|db2|t2",
+ "+I 2|20221208|16|0|0|db2|t2"));
+ it.close();
+ }
+
+ @ParameterizedTest(name = "defaultOptions = {0}")
+ @ValueSource(booleans = {true, false})
+ public void testStreamingRead(boolean defaultOptions) throws Exception {
+ Map<String, String> options = new HashMap<>();
+ if (!defaultOptions) {
+ // change options to test whether CompactorSourceBuilder work
normally
+ options.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), "2");
+ options.put(
+ CoreOptions.CHANGELOG_PRODUCER.key(),
+ CoreOptions.ChangelogProducer.NONE.toString());
+ options.put(CoreOptions.SCAN_BOUNDED_WATERMARK.key(), "0");
+ }
+ long monitorInterval = 1000;
+ List<FileStoreTable> tables = new ArrayList<>();
+ for (String dbName : DATABASE_NAMES) {
+ for (String tableName : TABLE_NAMES) {
+ FileStoreTable table =
+ createTable(
+ dbName,
+ tableName,
+ ROW_TYPE_MAP.get(tableName),
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ options);
+ tables.add(table);
+ StreamWriteBuilder streamWriteBuilder =
+
table.newStreamWriteBuilder().withCommitUser(commitUser);
+ StreamTableWrite write = streamWriteBuilder.newWrite();
+ StreamTableCommit commit = streamWriteBuilder.newCommit();
+
+ writeData(
+ write,
+ commit,
+ 0,
+ rowData(1, 1510, 15,
BinaryString.fromString("20221208")),
+ rowData(2, 1620, 16,
BinaryString.fromString("20221208")));
+
+ write.write(rowData(1, 1511, 15,
BinaryString.fromString("20221208")));
+ write.write(rowData(1, 1510, 15,
BinaryString.fromString("20221209")));
+ write.compact(binaryRow("20221208", 15), 0, true);
+ write.compact(binaryRow("20221209", 15), 0, true);
+ commit.commit(1, write.prepareCommit(true, 1));
+
+ writeData(
+ write,
+ commit,
+ 2,
+ rowData(2, 1520, 15,
BinaryString.fromString("20221208")),
+ rowData(2, 1621, 16,
BinaryString.fromString("20221208")));
+ writeData(
+ write,
+ commit,
+ 3,
+ rowData(1, 1512, 15,
BinaryString.fromString("20221208")),
+ rowData(2, 1620, 16,
BinaryString.fromString("20221209")));
+
+ write.close();
+ commit.close();
+ }
+ }
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream<RowData> compactorSource =
+ new MultiTablesCompactorSourceBuilder(
+ catalogLoader(),
+ Pattern.compile(".*"),
+ Pattern.compile(".*"),
+ null,
+ monitorInterval)
+ .withContinuousMode(true)
+ .withEnv(env)
+ .build();
+ CloseableIterator<RowData> it = compactorSource.executeAndCollect();
+
+ List<String> actual = new ArrayList<>();
+ for (int i = 0; i < 16; i++) {
+ actual.add(toString(it.next()));
+ }
+ assertThat(actual)
+ .hasSameElementsAs(
+ Arrays.asList(
+ "+I 4|20221208|15|0|1|db1|t1",
+ "+I 4|20221208|16|0|1|db1|t1",
+ "+I 4|20221208|15|0|1|db1|t2",
+ "+I 4|20221208|16|0|1|db1|t2",
+ "+I 4|20221208|15|0|1|db2|t1",
+ "+I 4|20221208|16|0|1|db2|t1",
+ "+I 4|20221208|15|0|1|db2|t2",
+ "+I 4|20221208|16|0|1|db2|t2",
+ "+I 5|20221209|16|0|1|db1|t1",
+ "+I 5|20221208|15|0|1|db1|t1",
+ "+I 5|20221209|16|0|1|db1|t2",
+ "+I 5|20221208|15|0|1|db1|t2",
+ "+I 5|20221209|16|0|1|db2|t1",
+ "+I 5|20221208|15|0|1|db2|t1",
+ "+I 5|20221209|16|0|1|db2|t2",
+ "+I 5|20221208|15|0|1|db2|t2"));
+
+ for (FileStoreTable table : tables) {
+ StreamWriteBuilder streamWriteBuilder =
+ table.newStreamWriteBuilder().withCommitUser(commitUser);
+ StreamTableWrite write = streamWriteBuilder.newWrite();
+ StreamTableCommit commit = streamWriteBuilder.newCommit();
+ writeData(
+ write,
+ commit,
+ 4,
+ rowData(2, 1520, 15, BinaryString.fromString("20221209")),
+ rowData(1, 1510, 16, BinaryString.fromString("20221208")),
+ rowData(1, 1511, 15, BinaryString.fromString("20221209")));
+ }
+ actual.clear();
+ for (int i = 0; i < 8; i++) {
+ actual.add(toString(it.next()));
+ }
+ assertThat(actual)
+ .hasSameElementsAs(
+ Arrays.asList(
+ "+I 6|20221209|15|0|1|db1|t1",
+ "+I 6|20221208|16|0|1|db1|t1",
+ "+I 6|20221209|15|0|1|db1|t2",
+ "+I 6|20221208|16|0|1|db1|t2",
+ "+I 6|20221209|15|0|1|db2|t1",
+ "+I 6|20221208|16|0|1|db2|t1",
+ "+I 6|20221209|15|0|1|db2|t2",
+ "+I 6|20221208|16|0|1|db2|t2"));
+
+ // check if newly created tables can be detected
+ for (String dbName : New_DATABASE_NAMES) {
+ for (String tableName : New_TABLE_NAMES) {
+ FileStoreTable table =
+ createTable(
+ dbName,
+ tableName,
+ ROW_TYPE_MAP.get(tableName),
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ options);
+ tables.add(table);
+ StreamWriteBuilder streamWriteBuilder =
+
table.newStreamWriteBuilder().withCommitUser(commitUser);
+ StreamTableWrite write = streamWriteBuilder.newWrite();
+ StreamTableCommit commit = streamWriteBuilder.newCommit();
+ writeData(
+ write,
+ commit,
+ 0,
+ rowData(2, 1520, 15,
BinaryString.fromString("20221209")),
+ rowData(1, 1510, 16,
BinaryString.fromString("20221208")));
+ }
+ }
+ actual.clear();
+ for (int i = 0; i < 4; i++) {
+ actual.add(toString(it.next()));
+ }
+ assertThat(actual)
+ .hasSameElementsAs(
+ Arrays.asList(
+ "+I 1|20221209|15|0|1|db3|t2",
+ "+I 1|20221208|16|0|1|db3|t2",
+ "+I 1|20221209|15|0|1|db3|t1",
+ "+I 1|20221208|16|0|1|db3|t1"));
+ it.close();
+ }
+
+ @ParameterizedTest(name = "defaultOptions = {0}")
+ @ValueSource(booleans = {true, false})
+ public void testIncludeAndExcludeTableRead(boolean defaultOptions) throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ if (!defaultOptions) {
+ // change options to test whether CompactorSourceBuilder work
normally
+ options.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), "2");
+ options.put(
+ CoreOptions.CHANGELOG_PRODUCER.key(),
+ CoreOptions.ChangelogProducer.NONE.toString());
+ options.put(CoreOptions.SCAN_BOUNDED_WATERMARK.key(), "0");
+ }
+ long monitorInterval = 1000;
+
+ for (String dbName : DATABASE_NAMES) {
+ for (String tableName : TABLE_NAMES) {
+ FileStoreTable table =
+ createTable(
+ dbName,
+ tableName,
+ ROW_TYPE_MAP.get(tableName),
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ options);
+ StreamWriteBuilder streamWriteBuilder =
+
table.newStreamWriteBuilder().withCommitUser(commitUser);
+ StreamTableWrite write = streamWriteBuilder.newWrite();
+ StreamTableCommit commit = streamWriteBuilder.newCommit();
+
+ writeData(
+ write,
+ commit,
+ 0,
+ rowData(1, 1510, 15,
BinaryString.fromString("20221208")),
+ rowData(2, 1620, 16,
BinaryString.fromString("20221208")));
+
+ write.write(rowData(1, 1511, 15,
BinaryString.fromString("20221208")));
+ write.write(rowData(1, 1510, 15,
BinaryString.fromString("20221209")));
+ write.compact(binaryRow("20221208", 15), 0, true);
+ write.compact(binaryRow("20221209", 15), 0, true);
+ commit.commit(1, write.prepareCommit(true, 1));
+
+ writeData(
+ write,
+ commit,
+ 2,
+ rowData(2, 1520, 15,
BinaryString.fromString("20221208")),
+ rowData(2, 1621, 16,
BinaryString.fromString("20221208")));
+ writeData(
+ write,
+ commit,
+ 3,
+ rowData(1, 1512, 15,
BinaryString.fromString("20221208")),
+ rowData(2, 1620, 16,
BinaryString.fromString("20221209")));
+
+ write.close();
+ commit.close();
+ }
+ }
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream<RowData> compactorSource =
+ new MultiTablesCompactorSourceBuilder(
+ catalogLoader(),
+ Pattern.compile(".*"),
+ Pattern.compile("db1.+|db2.t1|db3.t1"),
+ Pattern.compile("db1.t2"),
+ monitorInterval)
+ .withContinuousMode(true)
+ .withEnv(env)
+ .build();
+ CloseableIterator<RowData> it = compactorSource.executeAndCollect();
+
+ List<String> actual = new ArrayList<>();
+ for (int i = 0; i < 8; i++) {
+ actual.add(toString(it.next()));
+ }
+ assertThat(actual)
+ .hasSameElementsAs(
+ Arrays.asList(
+ "+I 4|20221208|15|0|1|db1|t1",
+ "+I 4|20221208|16|0|1|db1|t1",
+ "+I 4|20221208|15|0|1|db2|t1",
+ "+I 4|20221208|16|0|1|db2|t1",
+ "+I 5|20221209|16|0|1|db1|t1",
+ "+I 5|20221208|15|0|1|db1|t1",
+ "+I 5|20221209|16|0|1|db2|t1",
+ "+I 5|20221208|15|0|1|db2|t1"));
+
+ // check if newly created tables conform to the pattern rule
+ for (String dbName : New_DATABASE_NAMES) {
+ for (String tableName : New_TABLE_NAMES) {
+ FileStoreTable table =
+ createTable(
+ dbName,
+ tableName,
+ ROW_TYPE_MAP.get(tableName),
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ options);
+ StreamWriteBuilder streamWriteBuilder =
+
table.newStreamWriteBuilder().withCommitUser(commitUser);
+ StreamTableWrite write = streamWriteBuilder.newWrite();
+ StreamTableCommit commit = streamWriteBuilder.newCommit();
+ writeData(
+ write,
+ commit,
+ 0,
+ rowData(2, 1520, 15,
BinaryString.fromString("20221209")),
+ rowData(1, 1510, 16,
BinaryString.fromString("20221208")));
+ }
+ }
+ actual.clear();
+ for (int i = 0; i < 2; i++) {
+ actual.add(toString(it.next()));
+ }
+ assertThat(actual)
+ .hasSameElementsAs(
+ Arrays.asList(
+ "+I 1|20221209|15|0|1|db3|t1", "+I
1|20221208|16|0|1|db3|t1"));
+ it.close();
+ }
+
+ private FileStoreTable createTable(
+ String databaseName,
+ String tableName,
+ RowType rowType,
+ List<String> partitionKeys,
+ List<String> primaryKeys,
+ Map<String, String> options)
+ throws Exception {
+ Catalog catalog = catalogLoader().load();
+ Identifier identifier = Identifier.create(databaseName, tableName);
+ catalog.createDatabase(databaseName, true);
+ catalog.createTable(
+ identifier,
+ new Schema(rowType.getFields(), partitionKeys, primaryKeys,
options, ""),
+ false);
+ return (FileStoreTable) catalog.getTable(identifier);
+ }
+
+ private GenericRow rowData(Object... values) {
+ return GenericRow.of(values);
+ }
+
+ private void writeData(
+ StreamTableWrite write,
+ StreamTableCommit commit,
+ long incrementalIdentifier,
+ GenericRow... data)
+ throws Exception {
+ for (GenericRow d : data) {
+ write.write(d);
+ }
+ commit.commit(incrementalIdentifier, write.prepareCommit(true,
incrementalIdentifier));
+ }
+
+ private String toString(RowData rowData) {
+ int numFiles;
+ DataFileMetaSerializer dataFileMetaSerializer = new
DataFileMetaSerializer();
+ try {
+ numFiles =
dataFileMetaSerializer.deserializeList(rowData.getBinary(3)).size();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+
+ BinaryRow partition = deserializeBinaryRow(rowData.getBinary(1));
+
+ return String.format(
+ "%s %d|%s|%d|%d|%d|%s|%s",
+ rowData.getRowKind().shortString(),
+ rowData.getLong(0),
+ partition.getString(0),
+ partition.getInt(1),
+ rowData.getInt(2),
+ numFiles,
+ rowData.getString(4),
+ rowData.getString(5));
+ }
+
+ private BinaryRow binaryRow(String dt, int hh) {
+ BinaryRow b = new BinaryRow(2);
+ BinaryRowWriter writer = new BinaryRowWriter(b);
+ writer.writeString(0, BinaryString.fromString(dt));
+ writer.writeInt(1, hh);
+ writer.complete();
+ return b;
+ }
+
+ private Catalog.Loader catalogLoader() {
+ // to make the action workflow serializable
+ catalogOptions.set(CatalogOptions.WAREHOUSE, warehouse);
+ return () -> FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
+ }
+}
