924060929 commented on code in PR #66473:
URL: https://github.com/apache/doris/pull/66473#discussion_r3781299886
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -910,6 +917,7 @@ protected void finalize() throws Throwable {
@Override
public void close() {
+ clearExternalScanTasks();
Review Comment:
Follow-up fixed on ba943660b0db. In addition to the TaskProcessor fallback,
MTMVTask.exec now closes its own StatementContext and removes the thread-local
ConnectContext after every internal refresh chunk/retry, including failures
before command execution. This prevents an earlier chunk's external scan-task
cache from surviving until the task-level finally can only see the last
context. The regression test runs two successive contexts and verifies both
caches are invalidated and both thread-local contexts are removed.
TaskProcessorTest + the complete StatementContextTest passed (28 tests),
Checkstyle passed, and ./build.sh --fe completed successfully.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -870,15 +873,24 @@ private CloseableIterable<FileScanTask>
splitFiles(TableScan scan) {
return TableScanUtil.splitFiles(scan.planFiles(),
sessionVariable.getFileSplitSize());
}
- if (isBatchMode()) {
+ if (isBatchMode() || tableLevelPushDownCount) {
// Currently iceberg batch split mode will use max split size.
// TODO: dynamic split size in batch split mode need to customize
iceberg splitter.
+ // A metadata COUNT(*) also consumes only a bounded number of
representative splits.
+ // Keep planFiles lazy for that path instead of materializing and
retaining the whole
+ // table in the statement cache.
return TableScanUtil.splitFiles(scan.planFiles(),
sessionVariable.getMaxSplitSize());
}
// Non Batch Mode
// Materialize planFiles() into a list to avoid iterating the
CloseableIterable twice.
// RISK: It will cost memory if the table is large.
+ List<FileScanTask> fileScanTaskList = getOrPlanFileScanTasks(scan, ()
-> materializeFileScanTasks(scan));
+ targetSplitSize = determineTargetFileSplitSize(fileScanTaskList);
Review Comment:
Addressed on the current head. Ordinary Iceberg file tasks are retained only
as serialized task payloads under the statement-wide ICEBERG_SERIALIZED_BYTES
cumulative budget. If adding the plan exceeds the byte limit,
IcebergTaskCacheLimitException aborts cache population and that execution falls
back to the directly planned task list, so the oversized plan is not retained
in StatementContext. Covered by testOversizedIcebergPlanIsNotRetained and
testIcebergTaskSerializationStopsAtByteLimit.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -1416,14 +1522,30 @@ private List<Split>
doGetPositionDeletesSystemTableSplits() throws UserException
long startTime = System.currentTimeMillis();
scan = scan.planWith(source.getCatalog().getThreadPoolWithPreAuth());
- try (CloseableIterable<ScanTask> scanTasks = scan.planFiles()) {
- for (ScanTask task : scanTasks) {
- if (!(task instanceof PositionDeletesScanTask)) {
- throw new UserException("Unexpected Iceberg
position_deletes scan task: " + task);
+ BatchScan plannedScan = scan;
+ Snapshot snapshot = plannedScan.snapshot();
+ IcebergScanTaskCacheKey<PositionDeletesScanTask> cacheKey = new
IcebergScanTaskCacheKey<>(
+ source.getCatalog().getId(),
+ source.getTargetTable().getId(),
+ snapshot == null ? null : snapshot.snapshotId(),
+ plannedScan.schema().schemaId(),
+ plannedScan.filter(),
+ plannedScan.isCaseSensitive(),
+ PositionDeletesScanTask.class.getName());
+ try {
+ positionDeleteTasks = getOrLoadExternalScanTasks(cacheKey, () -> {
+ List<PositionDeletesScanTask> tasks = new ArrayList<>();
Review Comment:
Addressed on the current head. Position-delete tasks use the same
serialized-task cache path and the same cumulative ICEBERG_SERIALIZED_BYTES
budget as ordinary file tasks. On overflow the partially serialized entry is
discarded and the caller uses the direct planning result, rather than retaining
the full task graph. Covered by testOversizedPositionDeletePlanIsNotRetained.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -380,29 +385,68 @@ private List<HivePartition>
getPrunedPartitions(HoodieTableMetaClient metaClient
private List<Split> getIncrementalSplits() {
long startTime = System.currentTimeMillis();
- if (canUseNativeReader()) {
- List<Split> splits = incrementalRelation.collectSplits();
- noLogsSplitNum.addAndGet(splits.size());
+ try {
+ if (canUseNativeReader()) {
+ List<Split> splits = incrementalRelation.collectSplits();
+ noLogsSplitNum.addAndGet(splits.size());
+ return splits;
+ }
+ Option<String[]> partitionColumns =
hudiClient.getTableConfig().getPartitionFields();
+ List<String> partitionNames = partitionColumns.isPresent()
+ ? Arrays.asList(partitionColumns.get()) :
Collections.emptyList();
+ List<Split> splits =
incrementalRelation.collectFileSlices().stream()
+ .map(fileSlice -> generateHudiSplit(fileSlice,
+ HudiPartitionUtils.parsePartitionValues(
+ partitionNames,
fileSlice.getPartitionPath()),
+ incrementalRelation.getEndTs()))
+ .collect(Collectors.toList());
+ if (!sessionVariable.isForceJniScanner()) {
+ splits.stream()
+ .map(split -> (HudiSplit) split)
+ .filter(split -> split.getHudiDeltaLogs().isEmpty())
+ .forEach(split -> noLogsSplitNum.incrementAndGet());
+ }
+ return splits;
+ } finally {
if (getSummaryProfile() != null) {
getSummaryProfile().addExternalTableGetFileScanTasksTime(System.currentTimeMillis()
- startTime);
}
- return splits;
}
- Option<String[]> partitionColumns =
hudiClient.getTableConfig().getPartitionFields();
- List<String> partitionNames = partitionColumns.isPresent() ?
Arrays.asList(partitionColumns.get())
- : Collections.emptyList();
- List<Split> splits = incrementalRelation.collectFileSlices().stream()
- .map(fileSlice -> generateHudiSplit(fileSlice,
-
HudiPartitionUtils.parsePartitionValues(partitionNames,
fileSlice.getPartitionPath()),
- incrementalRelation.getEndTs()))
- .collect(Collectors.toList());
- if (getSummaryProfile() != null) {
-
getSummaryProfile().addExternalTableGetFileScanTasksTime(System.currentTimeMillis()
- startTime);
+ }
+
+ private void getPartitionSplits(HivePartition partition, List<Split>
splits) throws Exception {
+ getPartitionSplits(partition, splits, true);
+ }
+
+ private void getPartitionSplits(
Review Comment:
Addressed on the current head. HudiPartitionScanKey now includes serdeLib
together with the base path, input format, column name/type vectors, instant,
partition and reader/pruning modes, so two HMS metadata generations with
different SerDe identity cannot share a partition plan. Covered by
testPartitionCacheKeySeparatesHmsMetadataGenerations.
--
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]
