JingsongLi commented on a change in pull request #11796:
URL: https://github.com/apache/flink/pull/11796#discussion_r415197817
##########
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemTableSink.java
##########
@@ -284,4 +337,125 @@ public boolean configurePartitionGrouping(boolean
supportsGrouping) {
this.dynamicGrouping = supportsGrouping;
return dynamicGrouping;
}
+
+ /**
+ * Table bucket assigner, wrap {@link PartitionComputer}.
+ */
+ private static class TableBucketAssigner implements
BucketAssigner<BaseRow, String> {
+
+ private final PartitionComputer<BaseRow> computer;
+
+ private TableBucketAssigner(PartitionComputer<BaseRow>
computer) {
+ this.computer = computer;
+ }
+
+ @Override
+ public String getBucketId(BaseRow element, Context context) {
+ try {
+ return PartitionPathUtils.generatePartitionPath(
+
computer.generatePartValues(element));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public SimpleVersionedSerializer<String> getSerializer() {
+ return SimpleVersionedStringSerializer.INSTANCE;
+ }
+ }
+
+ /**
+ * Table {@link RollingPolicy}, now it is a {@link
CheckpointRollingPolicy}.
+ * Because partition commit is hard to support false.
Review comment:
Sorry for out of date comments, should be:
```
Table {@link RollingPolicy}, it extends {@link CheckpointRollingPolicy} for
bulk writers.
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
