morrySnow commented on code in PR #66711:
URL: https://github.com/apache/doris/pull/66711#discussion_r3801995538
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java:
##########
@@ -356,6 +363,8 @@ public void createScanRangeLocations() throws UserException
{
executor.getSummaryProfile().setGetSplitsStartTime();
}
TFileFormatType fileFormatType = getFileFormatType();
+ boolean fileAffinitySupported = fileFormatType ==
TFileFormatType.FORMAT_PARQUET
+ || fileFormatType == TFileFormatType.FORMAT_ORC ||
supportsPerRangeFileAffinity();
Review Comment:
为什么两个 file format type 的判断要放在 supportsPerRangeFileAffinity 的外面？
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/SplitAssignment.java:
##########
@@ -154,6 +154,18 @@ public void addToQueue(List<Split> splits) throws
UserException {
appendBatch(batch);
}
+ public static void enableFileAffinity(List<Split> splits, boolean
supported) {
+
splits.stream().filter(FileSplit.class::isInstance).map(FileSplit.class::cast)
+ .forEach(split -> split.setFileAffinitySupported(supported));
Review Comment:
为什么是在这里更新split的标记位，而不是在生成split的时候就做好这个标记？
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FederationBackendPolicy.java:
##########
@@ -246,9 +260,7 @@ public Multimap<Backend, Split>
computeScanRangeAssignment(List<Split> splits) t
if (chosenNode.isPresent()) {
Review Comment:
这里应该也做亲和性处理
##########
fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java:
##########
@@ -134,6 +136,40 @@ protected List<Column> getColumns() {
return columns;
}
+ /** Whether this scan node has ranges that must stay together in one
fragment instance. */
+ public boolean hasScanRangeInstanceAffinity() {
+ return false;
+ }
+
+ /**
+ * Split this scan node's ranges into the requested number of fragment
instances.
+ * Subclasses may keep related ranges in the same partition by overriding
this method.
+ */
+ public List<List<Integer>> splitScanRangeParamsByInstance(
+ List<TScanRangeParams> scanRangeParams, int expectedInstanceNum) {
+ List<Integer> rangeIndexes = IntStream.range(0,
scanRangeParams.size()).boxed().collect(Collectors.toList());
+ return ListUtil.splitBySize(rangeIndexes, expectedInstanceNum);
+ }
+
+ /** Materialize the instance partitions used by the legacy coordinator. */
+ public List<List<TScanRangeParams>> materializeScanRangeParamsByInstance(
+ List<TScanRangeParams> scanRangeParams, int expectedInstanceNum) {
+ if (!hasScanRangeInstanceAffinity()) {
+ return ListUtil.splitBySize(scanRangeParams, expectedInstanceNum);
+ }
+ List<List<Integer>> rangeIndexesPerInstance
+ = splitScanRangeParamsByInstance(scanRangeParams,
expectedInstanceNum);
+ List<List<TScanRangeParams>> result =
Lists.newArrayListWithCapacity(rangeIndexesPerInstance.size());
+ for (List<Integer> rangeIndexes : rangeIndexesPerInstance) {
+ List<TScanRangeParams> instanceScanRanges =
Lists.newArrayListWithCapacity(rangeIndexes.size());
+ for (Integer rangeIndex : rangeIndexes) {
+ instanceScanRanges.add(scanRangeParams.get(rangeIndex));
+ }
+ result.add(instanceScanRanges);
+ }
+ return result;
Review Comment:
看起来和scanranges里面的代码一模一样
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java:
##########
@@ -409,6 +418,8 @@ public void createScanRangeLocations() throws UserException
{
if (isBatchMode()) {
// File splits are generated lazily, and fetched by backends while
scanning.
// Only provide the unique ID of split source to backend.
+ // Do not enable file affinity here. The current fetch protocol
uses an empty split list as EOS, so it
+ // cannot safely represent a non-owner source that is temporarily
empty while scheduling is still active.
Review Comment:
这里是不是可以优化一下呀。在batch mode 下，保证不发empty split 的情况下，尽可能的将同一个文件放到一个机器上
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergSplit.java:
##########
@@ -98,4 +99,18 @@ public static IcebergSplit
newPositionDeleteSysTableSplit(LocationPath file, lon
split.setSelfSplitWeight(Math.max(length, 1L));
return split;
}
+
+ @Override
+ public Optional<String> getFileAffinityKey() {
+ if (positionDeleteSystemTableSplit) {
+ return isFileAffinitySupported()
+ && positionDeleteContent ==
IcebergDeleteFileFilter.PositionDelete.type()
+ && (positionDeleteFileFormat ==
TFileFormatType.FORMAT_PARQUET
+ || positionDeleteFileFormat ==
TFileFormatType.FORMAT_ORC)
+ && getHosts().length == 0 && getFileLength() > getLength()
Review Comment:
整个PR上的host和affinity key 的关联关系，感觉逻辑上不太对劲。可能是想处理 local read
的情况。但是不应该在这里引入，而是在最外侧的split生成逻辑处处理
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/DefaultScanSource.java:
##########
@@ -66,7 +66,9 @@ public List<ScanSource> parallelize(List<ScanNode> scanNodes,
int instanceNum) {
return ImmutableList.of();
}
- List<ScanRanges> scanRangesPerInstance = scanRanges.split(instanceNum);
+ List<ScanRanges> scanRangesPerInstance =
scanNode.hasScanRangeInstanceAffinity()
+ ? scanRanges.split(scanNode, instanceNum)
+ : scanRanges.split(instanceNum);
Review Comment:
统一到 scanRanges.split(scanNode, instanceNum) 中做是否启用亲和性切分的判断？
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java:
##########
@@ -356,6 +363,8 @@ public void createScanRangeLocations() throws UserException
{
executor.getSummaryProfile().setGetSplitsStartTime();
}
TFileFormatType fileFormatType = getFileFormatType();
+ boolean fileAffinitySupported = fileFormatType ==
TFileFormatType.FORMAT_PARQUET
Review Comment:
这个局部变量距离使用的地方太远了
--
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]
