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/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 2a3410cdc [flink] Introduce partition mark done (#3317)
2a3410cdc is described below
commit 2a3410cdc5e6d876436f2b9cbfe09ce660171bf6
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue May 14 12:10:27 2024 +0800
[flink] Introduce partition mark done (#3317)
---
docs/content/flink/sql-write.md | 30 +++
.../generated/flink_connector_configuration.html | 18 ++
.../main/java/org/apache/paimon/utils/IOUtils.java | 14 ++
.../paimon/partition/PartitionTimeExtractor.java | 8 +-
.../flink/sink/cdc/FlinkCdcMultiTableSink.java | 2 +-
.../apache/paimon/flink/FlinkConnectorOptions.java | 35 ++++
.../org/apache/paimon/flink/sink/Committer.java | 55 +++++-
.../paimon/flink/sink/CommitterOperator.java | 9 +-
.../apache/paimon/flink/sink/CompactorSink.java | 5 +-
.../org/apache/paimon/flink/sink/FlinkSink.java | 5 +-
.../apache/paimon/flink/sink/FlinkWriteSink.java | 12 +-
.../flink/sink/MultiTablesCompactorSink.java | 3 +-
.../apache/paimon/flink/sink/StoreCommitter.java | 41 +++-
.../paimon/flink/sink/StoreMultiCommitter.java | 25 +--
.../flink/sink/UnawareBucketCompactionSink.java | 5 +-
.../sink/partition/AddDonePartitionAction.java | 61 ++++++
.../flink/sink/partition/PartitionMarkDone.java | 207 +++++++++++++++++++++
.../sink/partition/PartitionMarkDoneAction.java | 27 +++
.../sink/partition/PartitionMarkDoneTrigger.java | 118 ++++++++++++
.../paimon/flink/sink/partition/SuccessFile.java | 105 +++++++++++
.../sink/partition/SuccessFileMarkDoneAction.java | 54 ++++++
.../paimon/flink/sink/CommitterOperatorTest.java | 22 ++-
.../paimon/flink/sink/StoreMultiCommitterTest.java | 8 +-
.../sink/partition/AddDonePartitionActionTest.java | 77 ++++++++
.../partition/PartitionMarkDoneTriggerTest.java | 107 +++++++++++
.../partition/SuccessFileMarkDoneActionTest.java | 52 ++++++
.../apache/paimon/hive/HiveCatalogITCaseBase.java | 60 +++++-
27 files changed, 1105 insertions(+), 60 deletions(-)
diff --git a/docs/content/flink/sql-write.md b/docs/content/flink/sql-write.md
index 52b8e4a62..8b275dbbc 100644
--- a/docs/content/flink/sql-write.md
+++ b/docs/content/flink/sql-write.md
@@ -206,3 +206,33 @@ DELETE FROM my_table WHERE currency = 'UNKNOWN';
{{< /tab >}}
{{< /tabs >}}
+
+## Partition Mark Done
+
+For partitioned tables, each partition may need to be scheduled to trigger
downstream batch computation. Therefore,
+it is necessary to choose this timing to indicate that it is ready for
scheduling and to minimize the amount of data
+drift during scheduling. We call this process: "Partition Mark Done".
+
+Example to mark done:
+```sql
+CREATE TABLE my_partitioned_table (
+ f0 INT,
+ f1 INT,
+ f2 INT,
+ ...
+ dt STRING
+) PARTITIONED BY (dt) WITH (
+ 'partition.timestamp-formatter'='yyyyMMdd',
+ 'partition.timestamp-pattern'='$dt',
+ 'partition.time-interval'='1 d',
+ 'partition.idle-time-to-done'='15 m'
+);
+```
+
+1. Firstly, you need to define the time parser of the partition and the time
interval between partitions in order to
+ determine when the partition can be properly marked done.
+2. Secondly, you need to define idle-time, which determines how long it takes
for the partition to have no new data,
+ and then it will be marked as done.
+3. Thirdly, by default, partition mark done will create _SUCCESS file, the
content of _SUCCESS file is a json, contains
+ `creationTime` and `modificationTime`, they can help you understand if
there is any delayed data. You can also
+ configure other actions.
diff --git
a/docs/layouts/shortcodes/generated/flink_connector_configuration.html
b/docs/layouts/shortcodes/generated/flink_connector_configuration.html
index b6befc9a1..0280581d2 100644
--- a/docs/layouts/shortcodes/generated/flink_connector_configuration.html
+++ b/docs/layouts/shortcodes/generated/flink_connector_configuration.html
@@ -68,6 +68,24 @@ under the License.
<td>Duration</td>
<td>Specific dynamic partition refresh interval for lookup, scan
all partitions and obtain corresponding partition.</td>
</tr>
+ <tr>
+ <td><h5>partition.idle-time-to-done</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Duration</td>
+ <td>Set a time duration when a partition has no new data after
this time duration, mark the done status to indicate that the data is
ready.</td>
+ </tr>
+ <tr>
+ <td><h5>partition.mark-done-action</h5></td>
+ <td style="word-wrap: break-word;">"success-file"</td>
+ <td>String</td>
+ <td>Action to mark a partition done is to notify the downstream
application that the partition has finished writing, the partition is ready to
be read.<br />1. 'success-file': add '_success' file to directory.<br />2.
'done-partition': add 'xxx.done' partition to metastore.<br />Both can be
configured at the same time: 'done-partition,success-file'.</td>
+ </tr>
+ <tr>
+ <td><h5>partition.time-interval</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Duration</td>
+ <td>You can specify time interval for partition, for example,
daily partition is '1 d', hourly partition is '1 h'.</td>
+ </tr>
<tr>
<td><h5>scan.infer-parallelism</h5></td>
<td style="word-wrap: break-word;">true</td>
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/IOUtils.java
b/paimon-common/src/main/java/org/apache/paimon/utils/IOUtils.java
index f8a78eb21..9878dec36 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/IOUtils.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/IOUtils.java
@@ -21,10 +21,13 @@ package org.apache.paimon.utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
+import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintStream;
+import java.nio.charset.StandardCharsets;
import static java.util.Arrays.asList;
@@ -125,6 +128,17 @@ public final class IOUtils {
}
}
+ public static String readUTF8Fully(final InputStream in) throws
IOException {
+ BufferedReader reader =
+ new BufferedReader(new InputStreamReader(in,
StandardCharsets.UTF_8));
+ StringBuilder builder = new StringBuilder();
+ String line;
+ while ((line = reader.readLine()) != null) {
+ builder.append(line);
+ }
+ return builder.toString();
+ }
+
// ------------------------------------------------------------------------
// Silent I/O cleanup / closing
// ------------------------------------------------------------------------
diff --git
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionTimeExtractor.java
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionTimeExtractor.java
index caf0e7a1f..60976e8b5 100644
---
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionTimeExtractor.java
+++
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionTimeExtractor.java
@@ -34,6 +34,8 @@ import java.time.format.DateTimeParseException;
import java.time.format.ResolverStyle;
import java.time.format.SignStyle;
import java.time.temporal.ChronoField;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Objects;
@@ -90,7 +92,11 @@ public class PartitionTimeExtractor {
this.formatter = formatter;
}
- public LocalDateTime extract(List<String> partitionKeys, List<Object>
partitionValues) {
+ public LocalDateTime extract(LinkedHashMap<String, String> spec) {
+ return extract(new ArrayList<>(spec.keySet()), new
ArrayList<>(spec.values()));
+ }
+
+ public LocalDateTime extract(List<String> partitionKeys, List<?>
partitionValues) {
LocalDateTime dateTime = null;
try {
String timestampString;
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
index e8dd0fde3..fb2bfe950 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
@@ -153,7 +153,7 @@ public class FlinkCdcMultiTableSink implements Serializable
{
// commit new files list even if they're empty.
// Otherwise we can't tell if the commit is successful after
// a restart.
- return (user, metricGroup) -> new StoreMultiCommitter(catalogLoader,
user, metricGroup);
+ return context -> new StoreMultiCommitter(catalogLoader, context);
}
protected CommittableStateManager<WrappedManifestCommittable>
createCommittableStateManager() {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
index 90eb9ce32..e02b1c124 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
@@ -339,6 +339,41 @@ public class FlinkConnectorOptions {
.withDescription(
"Allow sink committer and writer operator to be
chained together");
+ public static final ConfigOption<Duration> PARTITION_IDLE_TIME_TO_DONE =
+ key("partition.idle-time-to-done")
+ .durationType()
+ .noDefaultValue()
+ .withDescription(
+ "Set a time duration when a partition has no new
data after this time duration, "
+ + "mark the done status to indicate that
the data is ready.");
+
+ public static final ConfigOption<Duration> PARTITION_TIME_INTERVAL =
+ key("partition.time-interval")
+ .durationType()
+ .noDefaultValue()
+ .withDescription(
+ "You can specify time interval for partition, for
example, "
+ + "daily partition is '1 d', hourly
partition is '1 h'.");
+
+ public static final ConfigOption<String> PARTITION_MARK_DONE_ACTION =
+ key("partition.mark-done-action")
+ .stringType()
+ .defaultValue("success-file")
+ .withDescription(
+ Description.builder()
+ .text(
+ "Action to mark a partition done
is to notify the downstream application that the partition"
+ + " has finished writing,
the partition is ready to be read.")
+ .linebreak()
+ .text("1. 'success-file': add '_success'
file to directory.")
+ .linebreak()
+ .text(
+ "2. 'done-partition': add
'xxx.done' partition to metastore.")
+ .linebreak()
+ .text(
+ "Both can be configured at the
same time: 'done-partition,success-file'.")
+ .build());
+
public static List<ConfigOption<?>> getOptions() {
final Field[] fields = FlinkConnectorOptions.class.getFields();
final List<ConfigOption<?>> list = new ArrayList<>(fields.length);
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/Committer.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/Committer.java
index d92417be7..8d08c2471 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/Committer.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/Committer.java
@@ -18,8 +18,11 @@
package org.apache.paimon.flink.sink;
+import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.metrics.groups.OperatorMetricGroup;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.io.Serializable;
import java.util.Collection;
@@ -57,7 +60,55 @@ public interface Committer<CommitT, GlobalCommitT> extends
AutoCloseable {
/** Factory to create {@link Committer}. */
interface Factory<CommitT, GlobalCommitT> extends Serializable {
- Committer<CommitT, GlobalCommitT> create(
- String commitUser, OperatorMetricGroup metricGroup);
+ Committer<CommitT, GlobalCommitT> create(Context context);
+ }
+
+ /** Context to create {@link Committer}. */
+ interface Context {
+
+ String commitUser();
+
+ @Nullable
+ OperatorMetricGroup metricGroup();
+
+ boolean streamingCheckpointEnabled();
+
+ boolean isRestored();
+
+ OperatorStateStore stateStore();
+ }
+
+ static Context createContext(
+ String commitUser,
+ @Nullable OperatorMetricGroup metricGroup,
+ boolean streamingCheckpointEnabled,
+ boolean isRestored,
+ OperatorStateStore stateStore) {
+ return new Committer.Context() {
+ @Override
+ public String commitUser() {
+ return commitUser;
+ }
+
+ @Override
+ public OperatorMetricGroup metricGroup() {
+ return metricGroup;
+ }
+
+ @Override
+ public boolean streamingCheckpointEnabled() {
+ return streamingCheckpointEnabled;
+ }
+
+ @Override
+ public boolean isRestored() {
+ return isRestored;
+ }
+
+ @Override
+ public OperatorStateStore stateStore() {
+ return stateStore;
+ }
+ };
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java
index facd3b398..3220512b0 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java
@@ -119,7 +119,14 @@ public class CommitterOperator<CommitT, GlobalCommitT>
extends AbstractStreamOpe
StateUtils.getSingleValueFromState(
context, "commit_user_state", String.class,
initialCommitUser);
// parallelism of commit operator is always 1, so commitUser will
never be null
- committer = committerFactory.create(commitUser, getMetricGroup());
+ committer =
+ committerFactory.create(
+ Committer.createContext(
+ commitUser,
+ getMetricGroup(),
+ streamingCheckpointEnabled,
+ context.isRestored(),
+ context.getOperatorStateStore()));
committableStateManager.initializeState(context, committer);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CompactorSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CompactorSink.java
index 8e0882456..7dc3ab115 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CompactorSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CompactorSink.java
@@ -40,9 +40,8 @@ public class CompactorSink extends FlinkSink<RowData> {
}
@Override
- protected Committer.Factory<Committable, ManifestCommittable>
createCommitterFactory(
- boolean streamingCheckpointEnabled) {
- return (user, metricGroup) -> new
StoreCommitter(table.newCommit(user), metricGroup);
+ protected Committer.Factory<Committable, ManifestCommittable>
createCommitterFactory() {
+ return context -> new StoreCommitter(table,
table.newCommit(context.commitUser()), context);
}
@Override
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
index 92683e8f7..cf9bd487a 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
@@ -233,7 +233,7 @@ public abstract class FlinkSink<T> implements Serializable {
true,
options.get(SINK_COMMITTER_OPERATOR_CHAINING),
commitUser,
- createCommitterFactory(streamingCheckpointEnabled),
+ createCommitterFactory(),
createCommittableStateManager());
if (options.get(SINK_AUTO_TAG_FOR_SAVEPOINT)) {
committerOperator =
@@ -310,8 +310,7 @@ public abstract class FlinkSink<T> implements Serializable {
protected abstract OneInputStreamOperator<T, Committable>
createWriteOperator(
StoreSinkWrite.Provider writeProvider, String commitUser);
- protected abstract Committer.Factory<Committable, ManifestCommittable>
createCommitterFactory(
- boolean streamingCheckpointEnabled);
+ protected abstract Committer.Factory<Committable, ManifestCommittable>
createCommitterFactory();
protected abstract CommittableStateManager<ManifestCommittable>
createCommittableStateManager();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkWriteSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkWriteSink.java
index b812c0491..0d6f245ba 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkWriteSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkWriteSink.java
@@ -40,18 +40,18 @@ public abstract class FlinkWriteSink<T> extends
FlinkSink<T> {
}
@Override
- protected Committer.Factory<Committable, ManifestCommittable>
createCommitterFactory(
- boolean streamingCheckpointEnabled) {
+ protected Committer.Factory<Committable, ManifestCommittable>
createCommitterFactory() {
// If checkpoint is enabled for streaming job, we have to
// commit new files list even if they're empty.
// Otherwise we can't tell if the commit is successful after
// a restart.
- return (user, metricGroup) ->
+ return context ->
new StoreCommitter(
- table.newCommit(user)
+ table,
+ table.newCommit(context.commitUser())
.withOverwrite(overwritePartition)
-
.ignoreEmptyCommit(!streamingCheckpointEnabled),
- metricGroup);
+
.ignoreEmptyCommit(!context.streamingCheckpointEnabled()),
+ context);
}
@Override
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
index f03b5b18b..cfc1ec1cd 100644
---
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
@@ -153,8 +153,7 @@ public class MultiTablesCompactorSink implements
Serializable {
createCommitterFactory() {
Map<String, String> dynamicOptions = options.toMap();
dynamicOptions.put(CoreOptions.WRITE_ONLY.key(), "false");
- return (user, metricGroup) ->
- new StoreMultiCommitter(catalogLoader, user, metricGroup,
true, dynamicOptions);
+ return context -> new StoreMultiCommitter(catalogLoader, context,
true, dynamicOptions);
}
protected CommittableStateManager<WrappedManifestCommittable>
createCommittableStateManager() {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
index 17c1d209a..922949cce 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
@@ -20,15 +20,15 @@ package org.apache.paimon.flink.sink;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.flink.metrics.FlinkMetricRegistry;
+import org.apache.paimon.flink.sink.partition.PartitionMarkDone;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.TableCommit;
import org.apache.paimon.table.sink.TableCommitImpl;
-import org.apache.flink.metrics.groups.OperatorMetricGroup;
-
import javax.annotation.Nullable;
import java.io.IOException;
@@ -43,16 +43,28 @@ public class StoreCommitter implements
Committer<Committable, ManifestCommittabl
private final TableCommitImpl commit;
@Nullable private final CommitterMetrics committerMetrics;
+ @Nullable private final PartitionMarkDone partitionMarkDone;
- public StoreCommitter(TableCommit commit, @Nullable OperatorMetricGroup
metricGroup) {
+ public StoreCommitter(FileStoreTable table, TableCommit commit, Context
context) {
this.commit = (TableCommitImpl) commit;
- if (metricGroup != null) {
- this.commit.withMetricRegistry(new
FlinkMetricRegistry(metricGroup));
- this.committerMetrics = new
CommitterMetrics(metricGroup.getIOMetricGroup());
+ if (context.metricGroup() != null) {
+ this.commit.withMetricRegistry(new
FlinkMetricRegistry(context.metricGroup()));
+ this.committerMetrics = new
CommitterMetrics(context.metricGroup().getIOMetricGroup());
} else {
this.committerMetrics = null;
}
+
+ try {
+ this.partitionMarkDone =
+ PartitionMarkDone.create(
+ context.streamingCheckpointEnabled(),
+ context.isRestored(),
+ context.stateStore(),
+ table);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
}
@VisibleForTesting
@@ -99,15 +111,30 @@ public class StoreCommitter implements
Committer<Committable, ManifestCommittabl
throws IOException, InterruptedException {
commit.commitMultiple(committables, false);
calcNumBytesAndRecordsOut(committables);
+ if (partitionMarkDone != null) {
+ partitionMarkDone.notifyCommittable(committables);
+ }
}
@Override
public int filterAndCommit(List<ManifestCommittable> globalCommittables) {
- return commit.filterAndCommitMultiple(globalCommittables);
+ int committed = commit.filterAndCommitMultiple(globalCommittables);
+ if (partitionMarkDone != null) {
+ partitionMarkDone.notifyCommittable(globalCommittables);
+ }
+ return committed;
}
@Override
public Map<Long, List<Committable>>
groupByCheckpoint(Collection<Committable> committables) {
+ if (partitionMarkDone != null) {
+ try {
+ partitionMarkDone.snapshotState();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
Map<Long, List<Committable>> grouped = new HashMap<>();
for (Committable c : committables) {
grouped.computeIfAbsent(c.checkpointId(), k -> new
ArrayList<>()).add(c);
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 ebe1fd645..efa2aefe3 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
@@ -26,9 +26,6 @@ import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.metrics.groups.OperatorMetricGroup;
-
-import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
@@ -48,8 +45,7 @@ public class StoreMultiCommitter
implements Committer<MultiTableCommittable,
WrappedManifestCommittable> {
private final Catalog catalog;
- private final String commitUser;
- @Nullable private final OperatorMetricGroup flinkMetricGroup;
+ private final Context context;
// To make the commit behavior consistent with that of Committer,
// StoreMultiCommitter manages multiple committers which are
@@ -60,22 +56,17 @@ public class StoreMultiCommitter
private final boolean ignoreEmptyCommit;
private final Map<String, String> dynamicOptions;
- public StoreMultiCommitter(
- Catalog.Loader catalogLoader,
- String commitUser,
- @Nullable OperatorMetricGroup flinkMetricGroup) {
- this(catalogLoader, commitUser, flinkMetricGroup, false,
Collections.emptyMap());
+ public StoreMultiCommitter(Catalog.Loader catalogLoader, Context context) {
+ this(catalogLoader, context, false, Collections.emptyMap());
}
public StoreMultiCommitter(
Catalog.Loader catalogLoader,
- String commitUser,
- @Nullable OperatorMetricGroup flinkMetricGroup,
+ Context context,
boolean ignoreEmptyCommit,
Map<String, String> dynamicOptions) {
this.catalog = catalogLoader.load();
- this.commitUser = commitUser;
- this.flinkMetricGroup = flinkMetricGroup;
+ this.context = context;
this.ignoreEmptyCommit = ignoreEmptyCommit;
this.dynamicOptions = dynamicOptions;
this.tableCommitters = new HashMap<>();
@@ -202,8 +193,10 @@ public class StoreMultiCommitter
}
committer =
new StoreCommitter(
-
table.newCommit(commitUser).ignoreEmptyCommit(ignoreEmptyCommit),
- flinkMetricGroup);
+ table,
+ table.newCommit(context.commitUser())
+ .ignoreEmptyCommit(ignoreEmptyCommit),
+ context);
tableCommitters.put(tableId, committer);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketCompactionSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketCompactionSink.java
index cf825cec7..6785c53fb 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketCompactionSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketCompactionSink.java
@@ -48,9 +48,8 @@ public class UnawareBucketCompactionSink extends
FlinkSink<AppendOnlyCompactionT
}
@Override
- protected Committer.Factory<Committable, ManifestCommittable>
createCommitterFactory(
- boolean streamingCheckpointEnabled) {
- return (s, metricGroup) -> new StoreCommitter(table.newCommit(s),
metricGroup);
+ protected Committer.Factory<Committable, ManifestCommittable>
createCommitterFactory() {
+ return context -> new StoreCommitter(table,
table.newCommit(context.commitUser()), context);
}
@Override
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/AddDonePartitionAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/AddDonePartitionAction.java
new file mode 100644
index 000000000..ea888c6d5
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/AddDonePartitionAction.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.sink.partition;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.metastore.MetastoreClient;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
+
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.Map.Entry;
+
+import static
org.apache.paimon.utils.PartitionPathUtils.extractPartitionSpecFromPath;
+
+/** A {@link PartitionMarkDoneAction} which add ".done" partition. */
+public class AddDonePartitionAction implements PartitionMarkDoneAction {
+
+ private final MetastoreClient metastoreClient;
+
+ public AddDonePartitionAction(MetastoreClient metastoreClient) {
+ this.metastoreClient = metastoreClient;
+ }
+
+ @Override
+ public void markDone(String partition) throws Exception {
+ LinkedHashMap<String, String> doneSpec =
extractPartitionSpecFromPath(new Path(partition));
+ Entry<String, String> lastField = tailEntry(doneSpec);
+ doneSpec.put(lastField.getKey(), lastField.getValue() + ".done");
+ metastoreClient.addPartition(doneSpec);
+ }
+
+ private Entry<String, String> tailEntry(LinkedHashMap<String, String>
partitionSpec) {
+ return Iterators.getLast(partitionSpec.entrySet().iterator());
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ metastoreClient.close();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDone.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDone.java
new file mode 100644
index 000000000..b4b4aa0bd
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDone.java
@@ -0,0 +1,207 @@
+/*
+ * 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.partition;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.metastore.MetastoreClient;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.PartitionTimeExtractor;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.utils.RowDataPartitionComputer;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.OperatorStateStore;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.table.utils.PartitionPathUtils;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.CoreOptions.METASTORE_PARTITIONED_TABLE;
+import static
org.apache.paimon.flink.FlinkConnectorOptions.PARTITION_IDLE_TIME_TO_DONE;
+import static
org.apache.paimon.flink.FlinkConnectorOptions.PARTITION_MARK_DONE_ACTION;
+import static
org.apache.paimon.flink.FlinkConnectorOptions.PARTITION_TIME_INTERVAL;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/** Mark partition done. */
+public class PartitionMarkDone implements Closeable {
+
+ private static final ListStateDescriptor<List<String>>
PENDING_PARTITIONS_STATE_DESC =
+ new ListStateDescriptor<>(
+ "mark-done-pending-partitions",
+ new ListSerializer<>(StringSerializer.INSTANCE));
+
+ private final RowDataPartitionComputer partitionComputer;
+ private final PartitionMarkDoneTrigger trigger;
+ private final List<PartitionMarkDoneAction> actions;
+
+ @Nullable
+ public static PartitionMarkDone create(
+ boolean isStreaming,
+ boolean isRestored,
+ OperatorStateStore stateStore,
+ FileStoreTable table)
+ throws Exception {
+ if (!isStreaming) {
+ return null;
+ }
+
+ List<String> partitionKeys = table.partitionKeys();
+ if (partitionKeys.isEmpty()) {
+ return null;
+ }
+
+ CoreOptions coreOptions = table.coreOptions();
+ Options options = coreOptions.toConfiguration();
+
+ Duration idleToDone = options.get(PARTITION_IDLE_TIME_TO_DONE);
+ if (idleToDone == null) {
+ return null;
+ }
+
+ MetastoreClient.Factory metastoreClientFactory =
+ table.catalogEnvironment().metastoreClientFactory();
+ checkNotNull(
+ metastoreClientFactory, "Cannot mark done partition for table
without metastore.");
+ checkArgument(
+ coreOptions.partitionedTableInMetastore(),
+ "Table should enable %s",
+ METASTORE_PARTITIONED_TABLE.key());
+
+ RowDataPartitionComputer partitionComputer =
+ new RowDataPartitionComputer(
+ coreOptions.partitionDefaultName(),
+ table.schema().logicalPartitionType(),
+ partitionKeys.toArray(new String[0]));
+
+ PartitionMarkDoneTrigger trigger =
+ new PartitionMarkDoneTrigger(
+ new PartitionMarkDoneTriggerState(isRestored,
stateStore),
+ new PartitionTimeExtractor(
+ coreOptions.partitionTimestampPattern(),
+ coreOptions.partitionTimestampFormatter()),
+ options.get(PARTITION_TIME_INTERVAL),
+ idleToDone);
+
+ List<PartitionMarkDoneAction> actions =
+
Arrays.asList(options.get(PARTITION_MARK_DONE_ACTION).split(",")).stream()
+ .map(
+ action -> {
+ switch (action) {
+ case "success-file":
+ return new
SuccessFileMarkDoneAction(
+ table.fileIO(),
table.location());
+ case "done-partition":
+ return new AddDonePartitionAction(
+
metastoreClientFactory.create());
+ default:
+ throw new
UnsupportedOperationException(action);
+ }
+ })
+ .collect(Collectors.toList());
+
+ return new PartitionMarkDone(partitionComputer, trigger, actions);
+ }
+
+ public PartitionMarkDone(
+ RowDataPartitionComputer partitionComputer,
+ PartitionMarkDoneTrigger trigger,
+ List<PartitionMarkDoneAction> actions) {
+ this.partitionComputer = partitionComputer;
+ this.trigger = trigger;
+ this.actions = actions;
+ }
+
+ public void notifyCommittable(List<ManifestCommittable> committables) {
+ Set<BinaryRow> partitions = new HashSet<>();
+ for (ManifestCommittable committable : committables) {
+ committable.fileCommittables().stream()
+ .map(CommitMessage::partition)
+ .forEach(partitions::add);
+ }
+
+ partitions.stream()
+ .map(partitionComputer::generatePartValues)
+ .map(PartitionPathUtils::generatePartitionPath)
+ .forEach(trigger::notifyPartition);
+
+ for (String partition : trigger.donePartitions()) {
+ try {
+ for (PartitionMarkDoneAction action : actions) {
+ action.markDone(partition);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ public void snapshotState() throws Exception {
+ trigger.snapshotState();
+ }
+
+ @Override
+ public void close() throws IOException {
+ for (PartitionMarkDoneAction action : actions) {
+ action.close();
+ }
+ }
+
+ private static class PartitionMarkDoneTriggerState implements
PartitionMarkDoneTrigger.State {
+
+ private final boolean isRestored;
+ private final ListState<List<String>> pendingPartitionsState;
+
+ private PartitionMarkDoneTriggerState(boolean isRestored,
OperatorStateStore stateStore)
+ throws Exception {
+ this.isRestored = isRestored;
+ this.pendingPartitionsState =
stateStore.getListState(PENDING_PARTITIONS_STATE_DESC);
+ }
+
+ @Override
+ public List<String> restore() throws Exception {
+ List<String> pendingPartitions = new ArrayList<>();
+ if (isRestored) {
+
pendingPartitions.addAll(pendingPartitionsState.get().iterator().next());
+ }
+ return pendingPartitions;
+ }
+
+ @Override
+ public void update(List<String> partitions) throws Exception {
+
pendingPartitionsState.update(Collections.singletonList(partitions));
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneAction.java
new file mode 100644
index 000000000..78d05ccc5
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneAction.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.sink.partition;
+
+import java.io.Closeable;
+
+/** Action to mark partitions done. */
+public interface PartitionMarkDoneAction extends Closeable {
+
+ void markDone(String partition) throws Exception;
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneTrigger.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneTrigger.java
new file mode 100644
index 000000000..946d2b209
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneTrigger.java
@@ -0,0 +1,118 @@
+/*
+ * 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.partition;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.partition.PartitionTimeExtractor;
+import org.apache.paimon.utils.StringUtils;
+
+import java.time.Duration;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.paimon.utils.PartitionPathUtils.extractPartitionSpecFromPath;
+
+/** Trigger to mark partitions done. */
+public class PartitionMarkDoneTrigger {
+
+ private final State state;
+ private final PartitionTimeExtractor timeExtractor;
+ private final long timeInternal;
+ private final long idleTime;
+ private final Map<String, Long> pendingPartitions;
+
+ public PartitionMarkDoneTrigger(
+ State state,
+ PartitionTimeExtractor timeExtractor,
+ Duration timeInternal,
+ Duration idleTime)
+ throws Exception {
+ this(state, timeExtractor, timeInternal, idleTime,
System.currentTimeMillis());
+ }
+
+ PartitionMarkDoneTrigger(
+ State state,
+ PartitionTimeExtractor timeExtractor,
+ Duration timeInternal,
+ Duration idleTime,
+ long currentTimeMillis)
+ throws Exception {
+ this.state = state;
+ this.timeExtractor = timeExtractor;
+ this.timeInternal = timeInternal.toMillis();
+ this.idleTime = idleTime.toMillis();
+ this.pendingPartitions = new HashMap<>();
+ state.restore().forEach(p -> pendingPartitions.put(p,
currentTimeMillis));
+ }
+
+ public void notifyPartition(String partition) {
+ notifyPartition(partition, System.currentTimeMillis());
+ }
+
+ void notifyPartition(String partition, long currentTimeMillis) {
+ if (!StringUtils.isNullOrWhitespaceOnly(partition)) {
+ this.pendingPartitions.put(partition, currentTimeMillis);
+ }
+ }
+
+ public List<String> donePartitions() {
+ return donePartitions(System.currentTimeMillis());
+ }
+
+ public List<String> donePartitions(long currentTimeMillis) {
+ List<String> needDone = new ArrayList<>();
+ Iterator<Map.Entry<String, Long>> iter =
pendingPartitions.entrySet().iterator();
+ while (iter.hasNext()) {
+ Map.Entry<String, Long> entry = iter.next();
+ String partition = entry.getKey();
+
+ long lastUpdateTime = entry.getValue();
+ long partitionStartTime =
+ timeExtractor
+ .extract(extractPartitionSpecFromPath(new
Path(partition)))
+ .atZone(ZoneId.systemDefault())
+ .toInstant()
+ .toEpochMilli();
+ long partitionEndTime = partitionStartTime + timeInternal;
+ lastUpdateTime = Math.max(lastUpdateTime, partitionEndTime);
+
+ if (currentTimeMillis - lastUpdateTime > idleTime) {
+ needDone.add(partition);
+ iter.remove();
+ }
+ }
+ return needDone;
+ }
+
+ public void snapshotState() throws Exception {
+ state.update(new ArrayList<>(pendingPartitions.keySet()));
+ }
+
+ /** State to store partitions. */
+ public interface State {
+
+ List<String> restore() throws Exception;
+
+ void update(List<String> partitions) throws Exception;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/SuccessFile.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/SuccessFile.java
new file mode 100644
index 000000000..4d3656e6f
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/SuccessFile.java
@@ -0,0 +1,105 @@
+/*
+ * 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.partition;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Objects;
+
+/** Json represents a success file. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class SuccessFile {
+
+ private static final String FIELD_CREATION_TIME = "creationTime";
+ private static final String FIELD_MODIFICATION_TIME = "modificationTime";
+
+ @JsonProperty(FIELD_CREATION_TIME)
+ private final long creationTime;
+
+ @JsonProperty(FIELD_MODIFICATION_TIME)
+ private final long modificationTime;
+
+ @JsonCreator
+ public SuccessFile(
+ @JsonProperty(FIELD_CREATION_TIME) long creationTime,
+ @JsonProperty(FIELD_MODIFICATION_TIME) long modificationTime) {
+ this.creationTime = creationTime;
+ this.modificationTime = modificationTime;
+ }
+
+ @JsonGetter(FIELD_CREATION_TIME)
+ public long creationTime() {
+ return creationTime;
+ }
+
+ @JsonGetter(FIELD_MODIFICATION_TIME)
+ public long modificationTime() {
+ return modificationTime;
+ }
+
+ public SuccessFile updateModificationTime(long modificationTime) {
+ return new SuccessFile(creationTime, modificationTime);
+ }
+
+ public String toJson() {
+ return JsonSerdeUtil.toJson(this);
+ }
+
+ public static SuccessFile fromJson(String json) {
+ return JsonSerdeUtil.fromJson(json, SuccessFile.class);
+ }
+
+ @Nullable
+ public static SuccessFile safelyFromPath(FileIO fileIO, Path path) throws
IOException {
+ try {
+ String json = fileIO.readFileUtf8(path);
+ return SuccessFile.fromJson(json);
+ } catch (FileNotFoundException e) {
+ return null;
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SuccessFile that = (SuccessFile) o;
+ return creationTime == that.creationTime && modificationTime ==
that.modificationTime;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(creationTime, modificationTime);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/SuccessFileMarkDoneAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/SuccessFileMarkDoneAction.java
new file mode 100644
index 000000000..ac97b6f14
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/SuccessFileMarkDoneAction.java
@@ -0,0 +1,54 @@
+/*
+ * 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.partition;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+
+/** A {@link PartitionMarkDoneAction} which create "_SUCCESS" file. */
+public class SuccessFileMarkDoneAction implements PartitionMarkDoneAction {
+
+ public static final String SUCCESS_FILE_NAME = "_SUCCESS";
+
+ private final FileIO fileIO;
+ private final Path tablePath;
+
+ public SuccessFileMarkDoneAction(FileIO fileIO, Path tablePath) {
+ this.fileIO = fileIO;
+ this.tablePath = tablePath;
+ }
+
+ @Override
+ public void markDone(String partition) throws Exception {
+ Path partitionPath = new Path(tablePath, partition);
+ Path successPath = new Path(partitionPath, SUCCESS_FILE_NAME);
+
+ long currentTime = System.currentTimeMillis();
+ SuccessFile successFile = SuccessFile.safelyFromPath(fileIO,
successPath);
+ if (successFile == null) {
+ successFile = new SuccessFile(currentTime, currentTime);
+ } else {
+ successFile = successFile.updateModificationTime(currentTime);
+ }
+ fileIO.overwriteFileUtf8(successPath, successFile.toJson());
+ }
+
+ @Override
+ public void close() {}
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
index 589eb773b..672ccfd39 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
@@ -561,7 +561,9 @@ public class CommitterOperatorTest extends
CommitterOperatorTestBase {
StreamTableCommit commit = table.newCommit(initialCommitUser);
OperatorMetricGroup metricGroup =
UnregisteredMetricsGroup.createOperatorMetricGroup();
- StoreCommitter committer = new StoreCommitter(commit, metricGroup);
+ StoreCommitter committer =
+ new StoreCommitter(
+ table, commit, Committer.createContext("",
metricGroup, true, false, null));
committer.commit(Collections.singletonList(manifestCommittable));
CommitterMetrics metrics = committer.getCommitterMetrics();
assertThat(metrics.getNumBytesOutCounter().getCount()).isEqualTo(285);
@@ -728,10 +730,13 @@ public class CommitterOperatorTest extends
CommitterOperatorTestBase {
true,
true,
commitUser == null ? initialCommitUser : commitUser,
- (user, metricGroup) ->
+ context ->
new StoreCommitter(
-
table.newStreamWriteBuilder().withCommitUser(user).newCommit(),
- metricGroup),
+ table,
+ table.newStreamWriteBuilder()
+ .withCommitUser(context.commitUser())
+ .newCommit(),
+ context),
committableStateManager);
}
@@ -745,10 +750,13 @@ public class CommitterOperatorTest extends
CommitterOperatorTestBase {
true,
true,
commitUser == null ? initialCommitUser : commitUser,
- (user, metricGroup) ->
+ context ->
new StoreCommitter(
-
table.newStreamWriteBuilder().withCommitUser(user).newCommit(),
- metricGroup),
+ table,
+ table.newStreamWriteBuilder()
+ .withCommitUser(context.commitUser())
+ .newCommit(),
+ context),
committableStateManager) {
@Override
public void initializeState(StateInitializationContext context)
throws Exception {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreMultiCommitterTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreMultiCommitterTest.java
index 36d31ff41..94cfb3670 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreMultiCommitterTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreMultiCommitterTest.java
@@ -642,9 +642,7 @@ class StoreMultiCommitterTest {
false,
true,
initialCommitUser,
- (user, metricGroup) ->
- new StoreMultiCommitter(
- catalogLoader, initialCommitUser,
metricGroup),
+ context -> new StoreMultiCommitter(catalogLoader,
context),
new RestoreAndFailCommittableStateManager<>(
() ->
new VersionedSerializerWrapper<>(
@@ -660,9 +658,7 @@ class StoreMultiCommitterTest {
false,
true,
initialCommitUser,
- (user, metricGroup) ->
- new StoreMultiCommitter(
- catalogLoader, initialCommitUser,
metricGroup),
+ context -> new StoreMultiCommitter(catalogLoader,
context),
new
CommittableStateManager<WrappedManifestCommittable>() {
@Override
public void initializeState(
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/AddDonePartitionActionTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/AddDonePartitionActionTest.java
new file mode 100644
index 000000000..4fc31e41d
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/AddDonePartitionActionTest.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.sink.partition;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.metastore.MetastoreClient;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.apache.paimon.utils.PartitionPathUtils.generatePartitionPath;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class AddDonePartitionActionTest {
+
+ @Test
+ public void test() throws Exception {
+ AtomicBoolean closed = new AtomicBoolean(false);
+ Set<String> donePartitions = new HashSet<>();
+ MetastoreClient metastoreClient =
+ new MetastoreClient() {
+ @Override
+ public void addPartition(BinaryRow partition) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void addPartition(LinkedHashMap<String, String>
partitionSpec) {
+
donePartitions.add(generatePartitionPath(partitionSpec));
+ }
+
+ @Override
+ public void deletePartition(LinkedHashMap<String, String>
partitionSpec) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() throws Exception {
+ closed.set(true);
+ }
+ };
+
+ AddDonePartitionAction action = new
AddDonePartitionAction(metastoreClient);
+
+ // test normal
+ action.markDone("dt=20201202");
+
assertThat(donePartitions).containsExactlyInAnyOrder("dt=20201202.done/");
+
+ // test multiple partition fields
+ action.markDone("dt=20201202/hour=02");
+ assertThat(donePartitions)
+ .containsExactlyInAnyOrder("dt=20201202.done/",
"dt=20201202/hour=02.done/");
+
+ action.close();
+ assertThat(closed).isTrue();
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneTriggerTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneTriggerTest.java
new file mode 100644
index 000000000..b3c636308
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneTriggerTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.partition;
+
+import org.apache.paimon.partition.PartitionTimeExtractor;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class PartitionMarkDoneTriggerTest {
+
+ @Test
+ public void test() throws Exception {
+ List<String> pendingPartitions = new ArrayList<>();
+ PartitionMarkDoneTrigger.State state =
+ new PartitionMarkDoneTrigger.State() {
+ @Override
+ public List<String> restore() {
+ return new ArrayList<>(pendingPartitions);
+ }
+
+ @Override
+ public void update(List<String> partitions) {
+ pendingPartitions.clear();
+ pendingPartitions.addAll(partitions);
+ }
+ };
+
+ PartitionTimeExtractor extractor = new PartitionTimeExtractor("$dt",
"yyyy-MM-dd");
+ Duration timeInternal = Duration.ofDays(1);
+ Duration idleTime = Duration.ofMinutes(15);
+ PartitionMarkDoneTrigger trigger =
+ new PartitionMarkDoneTrigger(
+ state, extractor, timeInternal, idleTime,
toEpochMillis("2024-02-01"));
+
+ // test not reach partition end + idle time
+ trigger.notifyPartition("dt=2024-02-02", toEpochMillis("2024-02-01"));
+ List<String> partitions =
trigger.donePartitions(toEpochMillis("2024-02-03"));
+ assertThat(partitions).isEmpty();
+
+ // test state
+ assertThat(pendingPartitions).isEmpty();
+ trigger.snapshotState();
+ assertThat(pendingPartitions).containsOnly("dt=2024-02-02");
+
+ // test trigger
+ partitions = trigger.donePartitions(toEpochMillis("2024-02-03") +
idleTime.toMillis());
+ assertThat(partitions).isEmpty();
+ partitions = trigger.donePartitions(toEpochMillis("2024-02-03") +
idleTime.toMillis() + 1);
+ assertThat(partitions).containsOnly("dt=2024-02-02");
+
+ // test state
+ trigger.snapshotState();
+ assertThat(pendingPartitions).isEmpty();
+
+ // test refresh
+ trigger.notifyPartition("dt=2024-02-03", toEpochMillis("2024-02-03"));
+ trigger.notifyPartition("dt=2024-02-03", toEpochMillis("2024-02-04") +
idleTime.toMillis());
+ partitions = trigger.donePartitions(toEpochMillis("2024-02-04") +
idleTime.toMillis() + 1);
+ assertThat(partitions).isEmpty();
+ partitions =
+ trigger.donePartitions(toEpochMillis("2024-02-04") + 2 *
idleTime.toMillis() + 1);
+ assertThat(partitions).containsOnly("dt=2024-02-03");
+
+ // test restore
+ pendingPartitions.add("dt=2024-02-04");
+ trigger =
+ new PartitionMarkDoneTrigger(
+ state, extractor, timeInternal, idleTime,
toEpochMillis("2024-02-06"));
+ partitions = trigger.donePartitions(toEpochMillis("2024-02-06"));
+ assertThat(partitions).isEmpty();
+ partitions = trigger.donePartitions(toEpochMillis("2024-02-06") +
idleTime.toMillis() + 1);
+ assertThat(partitions).containsOnly("dt=2024-02-04");
+ }
+
+ private long toEpochMillis(String dt) {
+ return LocalDateTime.of(LocalDate.parse(dt), LocalTime.MIN)
+ .atZone(ZoneId.systemDefault())
+ .toInstant()
+ .toEpochMilli();
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/SuccessFileMarkDoneActionTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/SuccessFileMarkDoneActionTest.java
new file mode 100644
index 000000000..5a6818b25
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/SuccessFileMarkDoneActionTest.java
@@ -0,0 +1,52 @@
+/*
+ * 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.partition;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class SuccessFileMarkDoneActionTest {
+
+ @TempDir java.nio.file.Path temp;
+
+ @Test
+ public void test() throws Exception {
+ LocalFileIO fileIO = new LocalFileIO();
+ Path path = new Path(temp.toUri());
+ SuccessFileMarkDoneAction action = new
SuccessFileMarkDoneAction(fileIO, path);
+ Path successPath = new Path(path, "dt=20240513/_SUCCESS");
+
+ action.markDone("dt=20240513");
+ SuccessFile successFile1 = SuccessFile.safelyFromPath(fileIO,
successPath);
+ assertThat(successFile1).isNotNull();
+
+ Thread.sleep(100);
+ action.markDone("dt=20240513");
+ SuccessFile successFile2 = SuccessFile.safelyFromPath(fileIO,
successPath);
+ assertThat(successFile2).isNotNull();
+
+ assertThat(successFile1.creationTime() ==
successFile2.creationTime()).isTrue();
+ assertThat(successFile1.modificationTime() <
successFile2.modificationTime()).isTrue();
+ }
+}
diff --git
a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
index e478d44cd..5a915d200 100644
---
a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
+++
b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
@@ -27,14 +27,19 @@ import org.apache.paimon.hive.annotation.Minio;
import org.apache.paimon.hive.runner.PaimonEmbeddedHiveRunner;
import org.apache.paimon.privilege.NoPrivilegeException;
import org.apache.paimon.s3.MinioTestContainer;
+import org.apache.paimon.utils.IOUtils;
import com.klarna.hiverunner.HiveShell;
import com.klarna.hiverunner.annotations.HiveSQL;
+import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.fs.Path;
+import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
import org.apache.flink.table.api.internal.TableEnvironmentImpl;
import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
@@ -55,6 +60,7 @@ import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -80,6 +86,7 @@ public abstract class HiveCatalogITCaseBase {
protected String path;
protected TableEnvironment tEnv;
+ protected TableEnvironment sEnv;
private boolean locationInProperties;
@HiveSQL(files = {})
@@ -100,6 +107,10 @@ public abstract class HiveCatalogITCaseBase {
tEnv.executeSql("DROP DATABASE IF EXISTS test_db CASCADE");
tEnv.executeSql("CREATE DATABASE test_db").await();
tEnv.executeSql("USE test_db").await();
+
+ sEnv.executeSql("USE CATALOG my_hive").await();
+ sEnv.executeSql("USE test_db").await();
+
hiveShell.execute("USE test_db");
hiveShell.execute("CREATE TABLE hive_table ( a INT, b STRING )");
hiveShell.execute("INSERT INTO hive_table VALUES (100, 'Hive'), (200,
'Table')");
@@ -117,8 +128,15 @@ public abstract class HiveCatalogITCaseBase {
catalogProperties.putAll(minioTestContainer.getS3ConfigOptions());
}
- EnvironmentSettings settings =
EnvironmentSettings.newInstance().inBatchMode().build();
- tEnv = TableEnvironmentImpl.create(settings);
+ tEnv =
TableEnvironmentImpl.create(EnvironmentSettings.newInstance().inBatchMode().build());
+ sEnv =
+ TableEnvironmentImpl.create(
+
EnvironmentSettings.newInstance().inStreamingMode().build());
+ sEnv.getConfig()
+ .getConfiguration()
+ .set(ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL,
Duration.ofSeconds(1));
+
sEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM,
1);
+
tEnv.executeSql(
String.join(
"\n",
@@ -132,6 +150,8 @@ public abstract class HiveCatalogITCaseBase {
.collect(Collectors.joining(",\n")),
")"))
.await();
+
+ sEnv.registerCatalog(catalogName, tEnv.getCatalog(catalogName).get());
}
private void after() {
@@ -1111,6 +1131,42 @@ public abstract class HiveCatalogITCaseBase {
assertNoPrivilege(() -> tEnv.executeSql("DROP TABLE t").await());
}
+ @Test
+ public void testMarkDone() throws Exception {
+ sEnv.executeSql(
+ "CREATE TABLE mark_done_t1 (a INT, dt STRING) WITH
('continuous.discovery-interval' = '1s')");
+ sEnv.executeSql(
+ "CREATE TABLE mark_done_t2 (a INT, dt STRING)
PARTITIONED BY (dt) WITH ("
+ + "'partition.timestamp-formatter'='yyyyMMdd',"
+ + "'partition.timestamp-pattern'='$dt',"
+ + "'partition.idle-time-to-done'='1 s',"
+ + "'partition.time-interval'='1 d',"
+ + "'metastore.partitioned-table'='true',"
+ +
"'partition.mark-done-action'='done-partition,success-file'"
+ + ")")
+ .await();
+
+ TableResult insertSql =
+ sEnv.executeSql("INSERT INTO mark_done_t2 SELECT * FROM
mark_done_t1");
+
+ tEnv.executeSql("INSERT INTO mark_done_t1 VALUES (5,
'20240501')").await();
+
+ Thread.sleep(10 * 1000);
+
+ assertThat(hiveShell.executeQuery("SHOW PARTITIONS mark_done_t2"))
+ .containsExactlyInAnyOrder("dt=20240501", "dt=20240501.done");
+
+ Path successFile = new Path(path,
"test_db.db/mark_done_t2/dt=20240501/_SUCCESS");
+ String successText;
+ try (FSDataInputStream in =
successFile.getFileSystem().open(successFile)) {
+ successText = IOUtils.readUTF8Fully(in);
+ }
+
+
assertThat(successText).contains("creationTime").contains("modificationTime");
+
+ insertSql.getJobClient().get().cancel();
+ }
+
private void assertNoPrivilege(Executable executable) {
Exception e = assertThrows(Exception.class, executable);
if (e.getCause() != null) {
