JingsongLi commented on code in PR #9368:
URL: https://github.com/apache/paimon/pull/9368#discussion_r3859278197
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java:
##########
@@ -227,10 +230,55 @@ private DataStream<RowData> buildStaticFileSource() {
options.get(FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_ASSIGN_MODE),
dynamicPartitionFilteringInfo,
outerProject(),
+ splitWeightFunc(options),
+ null,
options.get(CoreOptions.BLOB_AS_DESCRIPTOR),
skipPreloadTargetSnapshot));
}
+ private static SerializableFunction<FileStoreSourceSplit, Long>
splitWeightFunc(
+ Options options) {
+ if (isFileSizeWeightMode(options)) {
+ return FlinkSourceBuilder::splitFileSizeOrRowCount;
Review Comment:
[P2] Please assign the largest byte-weighted splits first. The supplied
weight is consumed by BinPacking.packForFixedBinNumber, which sorts items in
ascending order before placing each item in the lightest bin. With two readers,
100 splits of weight 1, and one split of weight 100, the current algorithm
produces loads of 50 and 150, while largest-first placement produces 100 and
100. A large file plus many small files is a realistic case for this option, so
the new mode can preserve the long-tail skew it is intended to remove. Please
use descending/LPT order for this path (or fix the shared packer if compatible)
and add this distribution as a regression test.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java:
##########
@@ -227,10 +230,55 @@ private DataStream<RowData> buildStaticFileSource() {
options.get(FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_ASSIGN_MODE),
dynamicPartitionFilteringInfo,
outerProject(),
+ splitWeightFunc(options),
+ null,
options.get(CoreOptions.BLOB_AS_DESCRIPTOR),
skipPreloadTargetSnapshot));
}
+ private static SerializableFunction<FileStoreSourceSplit, Long>
splitWeightFunc(
+ Options options) {
+ if (isFileSizeWeightMode(options)) {
+ return FlinkSourceBuilder::splitFileSizeOrRowCount;
+ }
+ switch
(options.get(FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_WEIGHT_MODE)) {
+ case ROW_COUNT:
+ return split -> split.split().rowCount();
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported split weight mode "
+ + options.get(
+
FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_WEIGHT_MODE));
+ }
+ }
+
+ private static void validateSplitWeightMode(Options options) {
+ checkArgument(
+ !isFileSizeWeightMode(options)
+ ||
options.get(FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_ASSIGN_MODE)
+ == FlinkConnectorOptions.SplitAssignMode.FAIR,
+ "'%s' = '%s' only works with '%s' = '%s'.",
+ FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_WEIGHT_MODE.key(),
+ FlinkConnectorOptions.SplitWeightMode.FILE_SIZE,
+ FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_ASSIGN_MODE.key(),
+ FlinkConnectorOptions.SplitAssignMode.FAIR);
+ }
+
+ private static boolean isFileSizeWeightMode(Options options) {
+ return
options.get(FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_WEIGHT_MODE)
+ == FlinkConnectorOptions.SplitWeightMode.FILE_SIZE;
+ }
+
+ @VisibleForTesting
+ static long splitFileSizeOrRowCount(FileStoreSourceSplit sourceSplit) {
+ Split split = sourceSplit.split();
+ if (split instanceof DataSplit) {
Review Comment:
[P2] Please preserve file-size weighting through QueryAuthSplit. When
query-auth.enabled is true and REST authorization returns a row filter or
column mask, TableQueryAuthResult.convertPlan wraps each underlying DataSplit
in QueryAuthSplit. This outer-type check then falls back to rowCount, so
authenticated bounded reads silently ignore file-size mode even though
QueryAuthSplit exposes the wrapped split. Please unwrap transparent
QueryAuthSplit layers before checking for DataSplit and add a wrapped-split
test.
--
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]
