github-actions[bot] commented on code in PR #66711:
URL: https://github.com/apache/doris/pull/66711#discussion_r3772739212
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java:
##########
@@ -356,6 +356,8 @@ public void createScanRangeLocations() throws UserException
{
executor.getSummaryProfile().setGetSplitsStartTime();
}
TFileFormatType fileFormatType = getFileFormatType();
+ boolean fileAffinitySupported = fileFormatType ==
TFileFormatType.FORMAT_PARQUET
Review Comment:
[P2] Gate affinity by each range's physical format
This scan-wide flag is false for every Iceberg system table because
`getFileFormatType()` returns `FORMAT_JNI`, but the `position_deletes` path is
not JNI: it splits each `PositionDeletesScanTask`, creates real
path/start/length `IcebergSplit` ranges, and then overrides each range to
native Parquet/ORC. Those same-file ranges therefore miss the affinity promised
for native Parquet/ORC. Please derive eligibility from the split's physical
format while excluding logical JNI splits, and cover two ranges from one
position-delete file.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FederationBackendPolicy.java:
##########
@@ -309,6 +316,24 @@ public Multimap<Backend, Split>
computeScanRangeAssignment(List<Split> splits) t
return assignment;
}
+ private void assignSplit(ListMultimap<Backend, Split> assignment, Backend
backend, Split split) {
+ assignment.put(backend, split);
+ assignedWeightPerBackend.put(backend,
+ assignedWeightPerBackend.get(backend) +
split.getSplitWeight().getRawValue());
+ }
+
+ private Backend getFileAffinityBackend(Split split) {
+ if (!isFileAffinityEligible(split)) {
+ return null;
+ }
+ return consistentHash.getNode(split, 1).get(0);
Review Comment:
[P2] Validate the ring before indexing its owner
`split_assigner_virtual_node_number` is mutable and has no positive
validator. With a zero or negative value, a cold/new-backend-set
`ConsistentHash` has an empty ring, `getNode(split, 1)` returns an empty list,
and this new affinity path throws `IndexOutOfBoundsException` even for default
`ROUND_ROBIN` external scans (the base path never dereferenced the ring there).
Please validate the value at config/init time or safely fall back when no node
is returned, and add a zero-value regression test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java:
##########
@@ -410,7 +412,7 @@ public void createScanRangeLocations() throws UserException
{
// File splits are generated lazily, and fetched by backends while
scanning.
// Only provide the unique ID of split source to backend.
splitAssignment = new SplitAssignment(backendPolicy, this,
this::splitToScanRange,
- locationProperties, pathPartitionKeys, admissionResult);
+ locationProperties, pathPartitionKeys, admissionResult,
fileAffinitySupported);
Review Comment:
[P1] Prevent empty non-owner sources from failing lazy scans
Once this flag is enabled, all ranges in the first eligible file batch hash
to one backend, but lines 436-439 still register a `SplitSource` for every
backend before asynchronous generation finishes. A non-owner queue can
therefore stay empty; after the default 1000 ms, `SplitSource.getNextBatch()`
throws while scheduling is still active, FE returns `INTERNAL_ERROR`, and BE
aborts with a scan `IOError`. Hive/Hudi/Iceberg planning can validly exceed
that interval. Please keep an empty source retriable while its producer is
active, or delay exposing it until data/completion, and add a delayed
two-backend test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/SplitAssignment.java:
##########
@@ -145,6 +148,7 @@ public void addToQueue(List<Split> splits) throws
UserException {
}
Multimap<Backend, Split> batch = null;
synchronized (assignLock) {
+ enableFileAffinity(splits, fileAffinitySupported);
Review Comment:
[P2] Preserve the requested split-batch bound
After this enables affinity, a partition with more than 1000 ranges from one
large file can be assigned wholly to one backend. `appendBatch` enqueues that
backend's entire assignment as one collection, and
`SplitSource.getNextBatch(maxBatchSize)` uses `addAll`, so the BE request
(`remote_split_source_batch_size`, default 1000) can receive arbitrarily more
ranges in one Thrift response. Round-robin previously divided those ranges
among backends. Please chunk queue entries or retain remainders so each fetch
respects `maxBatchSize`.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
