You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "LsomeYeah (via GitHub)" <gi...@apache.org> on 2023/08/25 06:23:58 UTC
[GitHub] [incubator-paimon] LsomeYeah opened a new pull request, #1887: [flink] support one or multiple databases in dedicated compaction job with one mixed source and sink
LsomeYeah opened a new pull request, #1887:
URL: https://github.com/apache/incubator-paimon/pull/1887
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #xxx
It would be beneficial for Paimon to support compaction across multiple tables in a single job. Specially, it should have the ability to detect newly created tables after the running compaction job.
This pr support：
- support one or multiple databases compaction in a dedicated job with one combined source and combined sink.
- support detecting newly created tables after the running compaction job
- support including-tables.
- support excluding-tables.
<!-- What is the purpose of the change -->
### Tests
<!-- List UT and IT cases to verify this change -->
`org.apache.paimon.flink.source.MultiTablesCompactorSourceBuilderITCase`
This is for testing if the combined source can work normally.
`org.apache.paimon.flink.action.MultiTablesCompactActionITCase`
This is for testing if the compaction job for multi-tables can work normally.
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] yuzelin commented on a diff in pull request #1887: [flink] support one or multiple databases in a dedicated compaction job with one combined source and sink
Posted by "yuzelin (via GitHub)" <gi...@apache.org>.
yuzelin commented on code in PR #1887:
URL: https://github.com/apache/incubator-paimon/pull/1887#discussion_r1312789473
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java:
##########
@@ -142,6 +197,28 @@ public void build(StreamExecutionEnvironment env) {
}
}
+ private void buildForCombinedMode(StreamExecutionEnvironment env) {
+ Preconditions.checkArgument(
+ compactOptions != null, "in combined mode, compact options need to be set.");
Review Comment:
Remove this.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java:
##########
@@ -142,6 +197,28 @@ public void build(StreamExecutionEnvironment env) {
}
}
+ private void buildForCombinedMode(StreamExecutionEnvironment env) {
+ Preconditions.checkArgument(
+ compactOptions != null, "in combined mode, compact options need to be set.");
+
+ ReadableConfig conf = StreamExecutionEnvironmentUtils.getConfiguration(env);
+ boolean isStreaming =
+ conf.get(ExecutionOptions.RUNTIME_MODE) == RuntimeExecutionMode.STREAMING;
+ // Currently, multi-tables compaction do not 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(), null);
Review Comment:
Parallelism `sink.parallelism`
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesBatchCompactorSourceFunction.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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 + "-Monitor",
Review Comment:
No suffix.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesCompactorSourceFunction.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.common.state.CheckpointListener;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.util.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.OptionalLong;
+import java.util.TreeMap;
+import java.util.regex.Pattern;
+
+import static org.apache.paimon.flink.utils.MultiTablesCompactorUtil.compactOptions;
+import static org.apache.paimon.flink.utils.MultiTablesCompactorUtil.shouldCompactionTable;
+
+/**
+ * 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>>
+ implements CheckpointedFunction, CheckpointListener {
+ 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;
+
+ protected transient ListState<Long> checkpointState;
+ protected transient ListState<Tuple2<Long, Long>> nextSnapshotState;
+ protected transient TreeMap<Long, Long> nextSnapshotPerCheckpoint;
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ tablesMap = new HashMap<>();
+ scansMap = new HashMap<>();
+
+ catalog = catalogLoader.load();
+
+ updateTableMap();
+
+ this.checkpointState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "next-snapshot", LongSerializer.INSTANCE));
+
+ @SuppressWarnings("unchecked")
+ final Class<Tuple2<Long, Long>> typedTuple =
+ (Class<Tuple2<Long, Long>>) (Class<?>) Tuple2.class;
+ this.nextSnapshotState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "next-snapshot-per-checkpoint",
+ new TupleSerializer<>(
+ typedTuple,
+ new TypeSerializer[] {
+ LongSerializer.INSTANCE, LongSerializer.INSTANCE
+ })));
+
+ this.nextSnapshotPerCheckpoint = new TreeMap<>();
+
+ if (context.isRestored()) {
+ LOG.info("Restoring state for the {}.", getClass().getSimpleName());
+
+ List<Long> retrievedStates = new ArrayList<>();
+ for (Long entry : this.checkpointState.get()) {
+ retrievedStates.add(entry);
+ }
+
+ // given that the parallelism of the function is 1, we can only have 1 retrieved items.
+
+ Preconditions.checkArgument(
+ retrievedStates.size() <= 1,
+ getClass().getSimpleName() + " retrieved invalid state.");
+
+ if (retrievedStates.size() == 1) {
+ for (StreamTableScan scan : this.scansMap.values()) {
+ scan.restore(retrievedStates.get(0));
+ }
+ }
+
+ for (Tuple2<Long, Long> tuple2 : nextSnapshotState.get()) {
+ nextSnapshotPerCheckpoint.put(tuple2.f0, tuple2.f1);
+ }
+ } else {
+ LOG.info("No state to restore for the {}.", getClass().getSimpleName());
+ }
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
+ this.checkpointState.clear();
+ // TODO: how to get nextSnapshot better?
+ Long nextSnapshot = this.scansMap.values().iterator().next().checkpoint();
+ if (nextSnapshot != null) {
+ this.checkpointState.add(nextSnapshot);
+ this.nextSnapshotPerCheckpoint.put(ctx.getCheckpointId(), nextSnapshot);
+ }
+
+ List<Tuple2<Long, Long>> nextSnapshots = new ArrayList<>();
+ this.nextSnapshotPerCheckpoint.forEach((k, v) -> nextSnapshots.add(new Tuple2<>(k, v)));
+ this.nextSnapshotState.update(nextSnapshots);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("{} checkpoint {}.", getClass().getSimpleName(), nextSnapshot);
+ }
+ }
+
+ @Override
+ public void run(SourceContext<Tuple2<Split, String>> ctx) throws Exception {}
Review Comment:
remove this.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java:
##########
@@ -47,25 +53,63 @@
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 final Map<String, Table> tableMap = new HashMap<>();
+ private Pattern includingPattern;
+ @Nullable private Pattern excludingPattern;
+ private Pattern databasePattern;
+
+ private MultiTablesSinkMode databaseCompactMode = MultiTablesSinkMode.DIVIDED;
+
+ private Map<String, Table> tableMap = new HashMap<>();
+
+ @Nullable private Options compactOptions = null;
public CompactDatabaseAction(
Review Comment:
Remove this.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java:
##########
@@ -47,25 +53,63 @@
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 final Map<String, Table> tableMap = new HashMap<>();
+ private Pattern includingPattern;
+ @Nullable private Pattern excludingPattern;
+ private Pattern databasePattern;
+
+ private MultiTablesSinkMode databaseCompactMode = MultiTablesSinkMode.DIVIDED;
+
+ private Map<String, Table> tableMap = new HashMap<>();
+
+ @Nullable private Options compactOptions = null;
Review Comment:
default empty options.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java:
##########
@@ -47,25 +53,63 @@
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 final Map<String, Table> tableMap = new HashMap<>();
+ private Pattern includingPattern;
+ @Nullable private Pattern excludingPattern;
+ private Pattern databasePattern;
+
+ private MultiTablesSinkMode databaseCompactMode = MultiTablesSinkMode.DIVIDED;
+
+ private Map<String, Table> tableMap = new HashMap<>();
+
+ @Nullable private Options compactOptions = null;
public CompactDatabaseAction(
String warehouse,
- String database,
+ @Nullable String includingDatabases,
@Nullable String includingTables,
@Nullable String excludingTables,
Map<String, String> catalogConfig) {
super(warehouse, catalogConfig);
- this.database = database;
+ this.databasePattern =
+ Pattern.compile(includingDatabases == null ? ".*" : includingDatabases);
+ this.includingPattern = Pattern.compile(includingTables == null ? ".*" : includingTables);
+ this.excludingPattern = excludingTables == null ? null : Pattern.compile(excludingTables);
+ }
+
+ public CompactDatabaseAction(String warehouse, Map<String, String> catalogConfig) {
+ super(warehouse, catalogConfig);
+ }
+
+ public CompactDatabaseAction includingtDatabases(@Nullable String includingDatabases) {
+ this.databasePattern =
+ Pattern.compile(includingDatabases == null ? ".*" : includingDatabases);
+ return this;
+ }
+
+ public CompactDatabaseAction includingtTables(@Nullable String includingTables) {
Review Comment:
Should improve.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java:
##########
@@ -47,25 +53,63 @@
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 final Map<String, Table> tableMap = new HashMap<>();
+ private Pattern includingPattern;
+ @Nullable private Pattern excludingPattern;
+ private Pattern databasePattern;
+
+ private MultiTablesSinkMode databaseCompactMode = MultiTablesSinkMode.DIVIDED;
+
+ private Map<String, Table> tableMap = new HashMap<>();
Review Comment:
final
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesCompactorSourceFunction.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.common.state.CheckpointListener;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.util.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.OptionalLong;
+import java.util.TreeMap;
+import java.util.regex.Pattern;
+
+import static org.apache.paimon.flink.utils.MultiTablesCompactorUtil.compactOptions;
+import static org.apache.paimon.flink.utils.MultiTablesCompactorUtil.shouldCompactionTable;
+
+/**
+ * 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>>
+ implements CheckpointedFunction, CheckpointListener {
Review Comment:
Currently, we can not to support these checkpoint interfaces.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesBatchCompactorSourceFunction.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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 =
Review Comment:
Add an empty line.
##########
paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java:
##########
@@ -161,6 +161,17 @@ public FileIO fileIO() {
return wrapped.fileIO();
}
+ public static RowType getRowType() {
+ 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(Integer.MAX_VALUE)));
+ fields.add(new DataField(5, "_TABLE_NAME", new VarCharType(Integer.MAX_VALUE)));
+ return new RowType(fields);
Review Comment:
Common variable.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java:
##########
@@ -142,6 +197,28 @@ public void build(StreamExecutionEnvironment env) {
}
}
+ private void buildForCombinedMode(StreamExecutionEnvironment env) {
+ Preconditions.checkArgument(
+ compactOptions != null, "in combined mode, compact options need to be set.");
+
+ ReadableConfig conf = StreamExecutionEnvironmentUtils.getConfiguration(env);
+ boolean isStreaming =
+ conf.get(ExecutionOptions.RUNTIME_MODE) == RuntimeExecutionMode.STREAMING;
+ // Currently, multi-tables compaction do not support tables which bucketmode is UNWARE.
Review Comment:
Use todo.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] LsomeYeah commented on a diff in pull request #1887: [flink] support one or multiple databases in a dedicated compaction job with one combined source and sink
Posted by "LsomeYeah (via GitHub)" <gi...@apache.org>.
LsomeYeah commented on code in PR #1887:
URL: https://github.com/apache/incubator-paimon/pull/1887#discussion_r1311380751
##########
paimon-core/src/main/java/org/apache/paimon/table/system/BucketsMultiTable.java:
##########
@@ -0,0 +1,215 @@
+/*
+ * 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.table.system;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFileMetaSerializer;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.DataTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.ReadonlyTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.InnerStreamTableScan;
+import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.InnerTableScan;
+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.IntType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.IteratorRecordReader;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TagManager;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.SerializationUtils.newBytesType;
+import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
+
+/** this is a doc. */
+public class BucketsMultiTable implements DataTable, ReadonlyTable {
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] LsomeYeah commented on a diff in pull request #1887: [flink] support one or multiple databases in a dedicated compaction job with one combined source and sink
Posted by "LsomeYeah (via GitHub)" <gi...@apache.org>.
LsomeYeah commented on code in PR #1887:
URL: https://github.com/apache/incubator-paimon/pull/1887#discussion_r1312484816
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesCompactorSinkBuilder.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.options.Options;
+
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.table.data.RowData;
+
+import static org.apache.paimon.flink.sink.FlinkStreamPartitioner.partition;
+
+/** this is a doc. */
+public class MultiTablesCompactorSinkBuilder {
Review Comment:
`MultiTablesCompactorSinkBuilder` has been removed and `MultiTablesCompactorSink` is built in `CompactDatabaseAction` now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi merged pull request #1887: [flink] support one or multiple databases in a dedicated compaction job with one combined source and sink
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #1887:
URL: https://github.com/apache/incubator-paimon/pull/1887
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] LsomeYeah commented on a diff in pull request #1887: [flink] support one or multiple databases in a dedicated compaction job with one combined source and sink
Posted by "LsomeYeah (via GitHub)" <gi...@apache.org>.
LsomeYeah commented on code in PR #1887:
URL: https://github.com/apache/incubator-paimon/pull/1887#discussion_r1312484901
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java:
##########
@@ -54,13 +56,29 @@ 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.catalog = catalogLoader.load();
this.commitUser = commitUser;
this.metrics = metrics;
this.tableCommitters = new HashMap<>();
+ this.isCompactJob = 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;
Review Comment:
done
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java:
##########
@@ -148,15 +166,27 @@ private StoreCommitter getStoreCommitter(Identifier tableId) {
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(
"Failed to get committer for table %s", tableId.getFullName()),
e);
}
- committer =
- new StoreCommitter(
- table.newCommit(commitUser).ignoreEmptyCommit(false), metrics);
+ if (isCompactJob) {
+ committer =
+ new StoreCommitter(
+ table.newCommit(commitUser).ignoreEmptyCommit(true), metrics);
+ } else {
+ committer =
+ new StoreCommitter(
+ table.newCommit(commitUser).ignoreEmptyCommit(false), metrics);
+ }
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] LsomeYeah commented on a diff in pull request #1887: [flink] support one or multiple databases in a dedicated compaction job with one combined source and sink
Posted by "LsomeYeah (via GitHub)" <gi...@apache.org>.
LsomeYeah commented on code in PR #1887:
URL: https://github.com/apache/incubator-paimon/pull/1887#discussion_r1312506995
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesCompactorSink.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.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;
+
+/** this is a doc. */
+public class MultiTablesCompactorSink implements Serializable {
Review Comment:
Waiting for refactoring `FlinkSink`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] yuzelin commented on a diff in pull request #1887: [flink] support one or multiple databases in a dedicated compaction job with one combined source and sink
Posted by "yuzelin (via GitHub)" <gi...@apache.org>.
yuzelin commented on code in PR #1887:
URL: https://github.com/apache/incubator-paimon/pull/1887#discussion_r1309611577
##########
paimon-core/src/main/java/org/apache/paimon/table/system/BucketsMultiTable.java:
##########
@@ -0,0 +1,215 @@
+/*
+ * 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.table.system;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFileMetaSerializer;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.DataTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.ReadonlyTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.InnerStreamTableScan;
+import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.InnerTableScan;
+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.IntType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.IteratorRecordReader;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TagManager;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.SerializationUtils.newBytesType;
+import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
+
+/** this is a doc. */
+public class BucketsMultiTable implements DataTable, ReadonlyTable {
Review Comment:
This class can be merged with BucketsTable
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java:
##########
@@ -53,21 +58,37 @@ public class CompactDatabaseAction extends ActionBase {
private final Pattern includingPattern;
private final Pattern excludingPattern;
Review Comment:
@Nullable
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java:
##########
@@ -53,21 +58,37 @@ public class CompactDatabaseAction extends ActionBase {
private final Pattern includingPattern;
private final Pattern excludingPattern;
- private final String database;
- private final Map<String, Table> tableMap = new HashMap<>();
+ private final Pattern databasePattern;
+
+ private DatabaseCompactMode databaseCompactMode = DatabaseCompactMode.DIVIDED;
+
+ private Map<String, Table> tableMap = new HashMap<>();
+
+ @Nullable private Options compactOptions = null;
public CompactDatabaseAction(
String warehouse,
- String database,
+ @Nullable String includingDatabases,
@Nullable String includingTables,
@Nullable String excludingTables,
Map<String, String> catalogConfig) {
super(warehouse, catalogConfig);
- this.database = database;
+ this.databasePattern =
+ Pattern.compile(includingDatabases == null ? ".*" : includingDatabases);
this.includingPattern = Pattern.compile(includingTables == null ? ".*" : includingTables);
this.excludingPattern = excludingTables == null ? null : Pattern.compile(excludingTables);
}
+ public CompactDatabaseAction withDatabaseCompactMode(String mode) {
+ this.databaseCompactMode = DatabaseCompactMode.fromString(mode);
+ return this;
+ }
+
+ public CompactDatabaseAction withCompactOptions(Map<String, String> compactOptions) {
+ this.compactOptions = Options.fromMap(compactOptions);
+ return this;
+ }
+
Review Comment:
The action factory should be modified too. We can also put the nullable arguments here.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesCompactorSinkBuilder.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.options.Options;
+
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.table.data.RowData;
+
+import static org.apache.paimon.flink.sink.FlinkStreamPartitioner.partition;
+
+/** this is a doc. */
+public class MultiTablesCompactorSinkBuilder {
Review Comment:
Seems unnecessary.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesReadOperator.java:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.BucketsMultiTable;
+import org.apache.paimon.utils.CloseableIterator;
+
+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.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.shouldCompactionTable;
+
+/** this is a doc. */
+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 Pattern includingPattern;
+ private final Pattern excludingPattern;
+ private final Pattern databasePattern;
Review Comment:
The upstream should make sure the data is valid.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesCompactorSink.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.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;
+
+/** this is a doc. */
+public class MultiTablesCompactorSink implements Serializable {
Review Comment:
implement FlinkSink.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DatabaseCompactMode.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+
+/** this is a doc. */
+public enum DatabaseCompactMode implements Serializable {
Review Comment:
Merge with DatabaseSyncMode.
We can rename it to `MultiTablesSinkMode`.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/MultiTablesCompactorSourceBuilder.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.BucketsMultiTable;
+import org.apache.paimon.types.RowType;
+
+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;
+
+/** this is a doc. */
+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() {
+ if (env == null) {
+ throw new IllegalArgumentException("StreamExecutionEnvironment should not be null.");
Review Comment:
checkArgument
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java:
##########
@@ -54,13 +56,29 @@ 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.catalog = catalogLoader.load();
this.commitUser = commitUser;
this.metrics = metrics;
this.tableCommitters = new HashMap<>();
+ this.isCompactJob = 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;
Review Comment:
Use `this(...)`
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java:
##########
@@ -148,15 +166,27 @@ private StoreCommitter getStoreCommitter(Identifier tableId) {
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(
"Failed to get committer for table %s", tableId.getFullName()),
e);
}
- committer =
- new StoreCommitter(
- table.newCommit(commitUser).ignoreEmptyCommit(false), metrics);
+ if (isCompactJob) {
+ committer =
+ new StoreCommitter(
+ table.newCommit(commitUser).ignoreEmptyCommit(true), metrics);
+ } else {
+ committer =
+ new StoreCommitter(
+ table.newCommit(commitUser).ignoreEmptyCommit(false), metrics);
+ }
Review Comment:
`table.newCommit().ignoreEmptyCommit(isCompactJob)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] LsomeYeah commented on a diff in pull request #1887: [flink] support one or multiple databases in a dedicated compaction job with one combined source and sink
Posted by "LsomeYeah (via GitHub)" <gi...@apache.org>.
LsomeYeah commented on code in PR #1887:
URL: https://github.com/apache/incubator-paimon/pull/1887#discussion_r1312484102
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java:
##########
@@ -53,21 +58,37 @@ public class CompactDatabaseAction extends ActionBase {
private final Pattern includingPattern;
private final Pattern excludingPattern;
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] LsomeYeah commented on a diff in pull request #1887: [flink] support one or multiple databases in a dedicated compaction job with one combined source and sink
Posted by "LsomeYeah (via GitHub)" <gi...@apache.org>.
LsomeYeah commented on code in PR #1887:
URL: https://github.com/apache/incubator-paimon/pull/1887#discussion_r1311380997
##########
paimon-core/src/main/java/org/apache/paimon/table/system/BucketsMultiTable.java:
##########
@@ -0,0 +1,215 @@
+/*
+ * 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.table.system;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFileMetaSerializer;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.DataTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.ReadonlyTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.InnerStreamTableScan;
+import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.InnerTableScan;
+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.IntType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.IteratorRecordReader;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TagManager;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.SerializationUtils.newBytesType;
+import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
+
+/** this is a doc. */
+public class BucketsMultiTable implements DataTable, ReadonlyTable {
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] LsomeYeah commented on a diff in pull request #1887: [flink] support one or multiple databases in a dedicated compaction job with one combined source and sink
Posted by "LsomeYeah (via GitHub)" <gi...@apache.org>.
LsomeYeah commented on code in PR #1887:
URL: https://github.com/apache/incubator-paimon/pull/1887#discussion_r1311381578
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DatabaseCompactMode.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+
+/** this is a doc. */
+public enum DatabaseCompactMode implements Serializable {
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] LsomeYeah commented on a diff in pull request #1887: [flink] support one or multiple databases in a dedicated compaction job with one combined source and sink
Posted by "LsomeYeah (via GitHub)" <gi...@apache.org>.
LsomeYeah commented on code in PR #1887:
URL: https://github.com/apache/incubator-paimon/pull/1887#discussion_r1312485001
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/MultiTablesCompactorSourceBuilder.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.BucketsMultiTable;
+import org.apache.paimon.types.RowType;
+
+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;
+
+/** this is a doc. */
+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() {
+ if (env == null) {
+ throw new IllegalArgumentException("StreamExecutionEnvironment should not be null.");
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] LsomeYeah commented on a diff in pull request #1887: [flink] support one or multiple databases in a dedicated compaction job with one combined source and sink
Posted by "LsomeYeah (via GitHub)" <gi...@apache.org>.
LsomeYeah commented on code in PR #1887:
URL: https://github.com/apache/incubator-paimon/pull/1887#discussion_r1312486102
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesReadOperator.java:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.BucketsMultiTable;
+import org.apache.paimon.utils.CloseableIterator;
+
+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.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.shouldCompactionTable;
+
+/** this is a doc. */
+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 Pattern includingPattern;
+ private final Pattern excludingPattern;
+ private final Pattern databasePattern;
Review Comment:
The step of initializing table map in `MultiTablesReadOperator` has been removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org