wuchong commented on code in PR #20469:
URL: https://github.com/apache/flink/pull/20469#discussion_r940004377
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemConnectorOptions.java:
##########
@@ -245,6 +245,13 @@ public class FileSystemConnectorOptions {
.withDescription(
"The compaction target file size, the default
value is the rolling file size.");
+ public static final ConfigOption<Boolean>
HIVE_SINK_PARTITION_COMMIT_SUCCESS_FILE_ENABLED =
+ key("hive.sink.partition-commit.success-file.enabled")
Review Comment:
Do we really need this configuration? I think this can be covered by
`sink.partition-commit.policy.kind=success-file`.
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionLoader.java:
##########
@@ -154,8 +169,69 @@ private void moveFiles(List<Path> srcDirs, Path destDir)
throws Exception {
}
}
+ private void createSuccessFile(LinkedHashMap<String, String>
partitionSpec, Path path)
+ throws Exception {
+ if (writeSuccessFileEnabled) {
+ PartitionCommitPolicy.Context context =
+ new CommitPolicyContextImpl(partitionSpec, path);
+ commitPolicy.commit(context);
+ }
+ }
+
@Override
public void close() throws IOException {
metaStore.close();
}
+
+ private class CommitPolicyContextImpl implements
PartitionCommitPolicy.Context {
+
+ private final Path partitionPath;
+ private final LinkedHashMap<String, String> partitionSpec;
+
+ private CommitPolicyContextImpl(
+ LinkedHashMap<String, String> partitionSpec, Path
partitionPath) {
+ this.partitionSpec = partitionSpec;
+ this.partitionPath = partitionPath;
+ }
+
+ @Override
+ public String catalogName() {
+ throw new UnsupportedOperationException(
+ "Hive batch sink should not call
PartitionCommitPolicy.Context#catalogName.");
+ }
+
+ @Override
+ public String databaseName() {
+ throw new UnsupportedOperationException(
+ "Hive batch sink should not call
PartitionCommitPolicy.Context#databaseName.");
+ }
+
+ @Override
+ public String tableName() {
+ throw new UnsupportedOperationException(
+ "Hive batch sink should not call
PartitionCommitPolicy.Context#tableName.");
Review Comment:
I think these can be easily supported by passing in the `tableIdentifier`.
--
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-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
