[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #770: [flink] Assign batch splits for ContinuousFileSplitEnumerator

["JingsongLi (via GitHub)" <gi...@apache.org>]

JingsongLi commented on code in PR #770:
URL: https://github.com/apache/incubator-paimon/pull/770#discussion_r1152896661


##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumerator.java:
##########
@@ -205,9 +210,13 @@ private Map<Integer, List<FileStoreSourceSplit>> createAssignment() {
                                 readersAwaitingSplit.remove(task);
                                 return;
                             }
+                            List<FileStoreSourceSplit> taskAssignment = new ArrayList<>();
+                            while (!splits.isEmpty() && taskAssignment.size() < splitBatchSize) {

Review Comment:
   I think it is OK to just assign one split to a task for a bucket.
   Maybe here we can just limit assignment batch.
   ```
   List<FileStoreSourceSplit> taskAssignment = assignment.computeIfAbsent(task, i -> new ArrayList<>());
   if (taskAssignment.size() < splitBatchSize) {
      taskAssignment.add(splits.poll());
   }
   ```
   
   This can be limited to a large number of buckets, and only a few tasks can work properly.
   For example, buckets = 1000, but there is only one task.



-- 
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-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org