github-actions[bot] commented on code in PR #66473:
URL: https://github.com/apache/doris/pull/66473#discussion_r3765146685
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -1367,9 +1461,11 @@ private List<Split> doGetSystemTableSplits() throws
UserException {
List<Split> splits = new ArrayList<>();
TableScan scan = createTableScan();
long startTime = System.currentTimeMillis();
- try (CloseableIterable<FileScanTask> fileScanTasks = scan.planFiles())
{
+ try {
+ List<FileScanTask> fileScanTasks = getOrLoadExternalScanTasks(
Review Comment:
[P1] Keep metadata-table task planning streaming
This changes the metadata-table path from consuming `scan.planFiles()` one
task at a time to materializing every native task in the statement cache before
any are serialized into Doris splits. System tables are hard-excluded from
batch mode, so large `$files`/`$entries` scans—and especially
`$all_files`/`$all_entries`, which traverse manifests across snapshots—now
retain both the full native task graph and every Base64-serialized
`IcebergSplit` until statement end. In Iceberg 1.10.1 these tables create one
`ManifestReadTask` per manifest and each task copies the table's spec map, so
long table histories can exhaust FE heap even though the previous loop let each
native task become collectible after conversion. Please keep this system-table
conversion on the streaming `planFiles()` iterable, or otherwise bypass
statement caching here.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -380,29 +384,80 @@ 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 {
+ HudiIncrementalScanTaskCacheKey cacheKey = new
HudiIncrementalScanTaskCacheKey(
+ hmsTable.getCatalog().getId(), hmsTable.getId(),
+ incrementalRelation.getStartTs(),
incrementalRelation.getEndTs(),
+ canUseNativeReader(),
incrementalRelation.getHoodieParams());
+ List<HudiSplit> plannedSplits =
getOrLoadExternalScanTasks(cacheKey, () -> {
+ if (canUseNativeReader()) {
+ return incrementalRelation.collectSplits().stream()
+ .map(split -> {
+ Preconditions.checkState(split instanceof
HudiSplit,
+ "Hudi COW incremental relation must
produce HudiSplit");
+ return (HudiSplit) split;
+ })
+ .collect(Collectors.toList());
+ }
+ Option<String[]> partitionColumns =
hudiClient.getTableConfig().getPartitionFields();
+ List<String> partitionNames = partitionColumns.isPresent()
+ ? Arrays.asList(partitionColumns.get()) :
Collections.emptyList();
+ return incrementalRelation.collectFileSlices().stream()
+ .map(fileSlice -> generateHudiSplit(fileSlice,
+ HudiPartitionUtils.parsePartitionValues(
+ partitionNames,
fileSlice.getPartitionPath()),
+ incrementalRelation.getEndTs()))
+ .collect(Collectors.toList());
+ });
+ List<Split> splits = plannedSplits.stream()
Review Comment:
[P1] Avoid duplicating every incremental Hudi split
`getIncrementalSplits()` is hard-excluded from batch mode, but this now
retains the complete `plannedSplits` list in the statement cache and
deep-copies every element into the scan's output. For MOR/JNI reads,
`generateHudiSplit()` shares the node's column-name/type lists across source
splits, while `copyHudiSplit()` allocates a fresh copy of both lists for every
file, so a wide incremental range holds an O(files * columns) copy in addition
to the original split graph until statement close. COW ranges also retain two
complete split graphs. Previously the one planned list could be released after
range conversion. Please bypass statement caching for incremental scans, or
cache an immutable compact representation that doesn't require retaining and
copying the full graph.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -380,29 +384,80 @@ 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 {
+ HudiIncrementalScanTaskCacheKey cacheKey = new
HudiIncrementalScanTaskCacheKey(
Review Comment:
[P1] Key incremental tasks by the relation snapshot
`HudiIncrementalScanTaskCacheKey` omits this node's resolved `queryInstant`,
even though the relation grammar permits `table@incr(...) FOR TIME AS OF ...`
and each alias carries its own `HudiMvccSnapshot`. For MOR/JNI planning,
`doInitialize()` loads `columnNames`/`columnTypes` from that historical instant
and `generateHudiSplit()` embeds them in every cached split. Two aliases with
the same incremental begin/end/options but different `FOR TIME AS OF` instants
therefore compare equal, so the second skips its loader and sends the first
alias's historical schema vectors to BE. With schema evolution between the
instants this can decode the second scan against the wrong schema. Please
include `queryInstant` (or an equivalent snapshot/schema identity) in this key
and cover two historical MOR aliases in one statement.
--
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]
