JingsongLi commented on code in PR #8631:
URL: https://github.com/apache/paimon/pull/8631#discussion_r3638316592
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SortCompactSinkBuilder.java:
##########
@@ -18,17 +18,90 @@
package org.apache.paimon.flink.sink;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.DataSplit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
/** A special version {@link FlinkSinkBuilder} for sort compact. */
public class SortCompactSinkBuilder extends FlinkSinkBuilder {
+ private static final Logger LOG =
LoggerFactory.getLogger(SortCompactSinkBuilder.class);
+
+ private long baseSnapshotId;
+ @Nullable private List<DataSplit> compactInputSplits;
+ private boolean sortCompactInputSet = false;
+
public SortCompactSinkBuilder(Table table) {
super(table);
}
- public FlinkSinkBuilder forCompact(boolean compactSink) {
+ public SortCompactSinkBuilder forCompact(boolean compactSink) {
this.compactSink = compactSink;
return this;
}
+
+ /**
+ * Capture the base snapshot id and the planned compact input splits of
the sort compact. The
+ * splits (serializable) are carried into the job graph and used at commit
time to rewrite the
+ * written append files into a compact commit.
+ *
+ * <p><b>Scale note:</b> each {@link DataSplit} embeds full file metadata
and is serialized into
+ * the committer factory closure. For tables with very large numbers of
input files, this can
+ * significantly inflate the Flink job graph. Consider compacting in
smaller partition batches
+ * when approaching hundreds of thousands of files.
+ */
+ public SortCompactSinkBuilder withSortCompactInput(
+ long baseSnapshotId, List<DataSplit> compactInputSplits) {
+ validateSortCompactInput(compactInputSplits);
+ this.baseSnapshotId = baseSnapshotId;
+ this.compactInputSplits = compactInputSplits;
+ this.sortCompactInputSet = true;
+ return this;
+ }
+
+ private void validateSortCompactInput(List<DataSplit> compactInputSplits) {
Review Comment:
Remove this method and two options. We don't need to restrict this.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
