github-actions[bot] commented on code in PR #66473:
URL: https://github.com/apache/doris/pull/66473#discussion_r3772121355
##########
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:
[P1] Cap ordinary Iceberg task retention
Batch, explicit-size, count, and generic metadata-table paths now preserve
streaming, but the ordinary non-batch path still materializes every
`FileScanTask` and retains the unweighted list until statement cleanup. After
`TableScanUtil.splitFiles`/`createIcebergSplit` build the execution splits, the
cached SDK task wrappers plus residual/schema/spec state remain additionally
owned even though this node no longer needs them, and a large plan has no task
or byte cap; before this change the materialized list was method-local. Please
use a cumulative retention limit with an oversize direct-consumption fallback,
or cache a compact converted representation, and cover a large non-batch plan.
##########
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:
[P1] Keep position-delete planning bounded
`position_deletes` bypasses the generic system-table streaming helper: this
call first collects every `PositionDeletesScanTask` into the unweighted
statement cache, then splits and converts the complete list into native
`IcebergSplit` ranges. Those ranges carry the file, offset, partition, and
deletion-vector fields needed by BE, but the original SDK task graph remains
owned until statement cleanup, so a table with many delete files keeps both
representations. Please stream this special path through conversion where
possible, or apply a bounded compact cache/oversize fallback, and cover a
many-delete-file plan.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -380,29 +383,65 @@ 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(
+ HivePartition partition, List<Split> splits, boolean
useStatementCache) throws Exception {
+ List<HudiSplit> plannedSplits;
+ if (useStatementCache) {
+ HudiFileScanTaskCacheKey cacheKey = new HudiFileScanTaskCacheKey(
+ hmsTable.getCatalog().getId(), hmsTable.getId(),
queryInstant,
+ canUseNativeReader(),
sessionVariable.isEnableRuntimeFilterPartitionPrune(), partition);
+ plannedSplits = getOrLoadExternalScanTasks(
+ cacheKey, () -> planPartitionSplits(partition));
Review Comment:
[P1] Bound ordinary Hudi task retention
The new bypasses cover batch and incremental planning, but an ordinary
snapshot scan below the partition-count batch threshold still stores every
planned `HudiSplit` in the unweighted statement cache and then deep-copies
every split for the consumer. A large single-partition MOR table therefore
keeps the complete source graph plus per-file copies (including copied
schema/log lists) until statement cleanup even when no alias reuses it;
previously there was only the directly planned graph. Please add a cumulative
task/byte cap with an oversize direct-use fallback, or cache a compact
representation, and exercise a many-file non-batch snapshot scan.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -743,52 +747,235 @@ public List<org.apache.paimon.table.source.Split>
getPaimonSplitFromAPI() throws
if (PaimonScanParams.isPinnedEmptyScan(resolvedOptions)) {
return Collections.emptyList();
}
- Optional<Long> fileCreationTime =
PaimonScanParams.getPinnedFileCreationTime(resolvedOptions);
- if (fileCreationTime.isPresent()) {
- if (!(paimonTable instanceof FileStoreTable)) {
- throw new UserException("Paimon file-creation OPTIONS
require a data table.");
+ int[] projectedColumns = new int[0];
+ if
(!PaimonScanParams.getPinnedFileCreationTime(resolvedOptions).isPresent()) {
+ List<String> fieldNames =
paimonTable.rowType().getFieldNames();
+ projectedColumns = desc.getSlots().stream().mapToInt(
+ slot -> getFieldIndex(fieldNames,
slot.getColumn().getName()))
+ .toArray();
+ if (Arrays.stream(projectedColumns).anyMatch(index -> index <
0)) {
+ throw new UserException("Paimon scan schema does not
contain all bound Doris columns.");
}
- FileStoreTable fileStoreTable = (FileStoreTable) paimonTable;
- SnapshotReader snapshotReader =
fileStoreTable.newSnapshotReader()
- .withMode(ScanMode.ALL)
- .withSnapshot(Long.parseLong(
-
paimonTable.options().get(CoreOptions.SCAN_SNAPSHOT_ID.key())))
- .withManifestEntryFilter(entry ->
- entry.file().creationTimeEpochMillis() >=
fileCreationTime.get());
- preserveBatchScanFilters(fileStoreTable, snapshotReader);
- if (predicates != null) {
- predicates.forEach(snapshotReader::withFilter);
- }
- return snapshotReader.read().splits();
- }
- List<String> fieldNames = paimonTable.rowType().getFieldNames();
- int[] projected = desc.getSlots().stream().mapToInt(
- slot -> getFieldIndex(fieldNames,
slot.getColumn().getName()))
- .toArray();
- if (Arrays.stream(projected).anyMatch(index -> index < 0)) {
- throw new UserException("Paimon scan schema does not contain
all bound Doris columns.");
}
- ReadBuilder readBuilder = paimonTable.newReadBuilder();
- TableScan scan = readBuilder.withFilter(predicates)
- .withProjection(projected)
- .newScan();
- PaimonMetricRegistry registry = new PaimonMetricRegistry();
- if (scan instanceof InnerTableScan) {
- scan = ((InnerTableScan) scan).withMetricRegistry(registry);
- }
- List<org.apache.paimon.table.source.Split> splits =
scan.plan().splits();
- PaimonScanMetricsReporter.report(source.getTargetTable(),
paimonTable.name(), registry);
- if (!registry.getAllGroups().isEmpty()) {
- registry.clear();
+ int[] projected = projectedColumns;
+ PaimonSplitTaskCacheKey cacheKey = createPaimonSplitTaskCacheKey(
+ relationSnapshot, paimonTable, resolvedOptions,
+ scanParams != null && scanParams.incrementalRead()
+ ? getIncrReadParams() : Collections.emptyMap(),
+ projected);
+ List<PaimonSerializedScanTask> serializedSplits;
+ try {
+ serializedSplits = getOrLoadExternalScanTasks(cacheKey,
+ () -> serializePaimonSplitsWithinLimit(
Review Comment:
[P2] Stop encoding once the generation budget is spent
The serializer sees the fixed 16 MiB per-node limit, not this cache
generation's remaining allowance. If one key has already retained 10 MiB, a
distinct 10 MiB plan is therefore completely Java-serialized here; only
afterward does the weighted cache reject/remove it against the remaining 6 MiB,
return the byte arrays, and make this caller deserialize every task. Each later
distinct alias repeats that throwaway encode/decode and transient allocation,
whereas the individual-oversize path stops early and returns the original
planned list. Please reserve/expose the remaining budget to the loader, or
report non-retention so the raw tasks can be consumed directly, and test two
individually fitting keys whose sum exceeds the cap.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java:
##########
@@ -320,8 +323,25 @@ private void
getFileSplitByPartitions(HiveExternalMetaCache cache, List<HivePart
}
} else {
boolean withCache = Config.max_external_file_cache_num > 0;
- fileCaches = cache.getFilesByPartitions(partitions, withCache,
partitions.size() > 1,
- directoryLister, hmsTable);
+ if (isBatchMode) {
+ // Batch mode bounds FE memory by retaining only the
partitions currently in flight.
+ // Keeping every completed partition in the statement cache
would materialize the
+ // full scan again and defeat that bound.
+ fileCaches = cache.getFilesByPartitions(partitions, withCache,
partitions.size() > 1,
+ directoryLister, hmsTable);
+ } else {
+ HiveFileScanTaskCacheKey cacheKey = new
HiveFileScanTaskCacheKey(
+ hmsTable.getCatalog().getId(), hmsTable.getId(),
partitions);
+ try {
+ fileCaches = getOrLoadExternalScanTasks(cacheKey,
+ () -> cache.getFilesByPartitions(partitions,
withCache, partitions.size() > 1,
Review Comment:
[P1] Honor the disabled file-cache memory fence
When `max_external_file_cache_num <= 0`, `withCache` deliberately disables
the global Hive file cache, but this unweighted statement-cache call still
retains the full `FileCacheValue`/`HiveFileStatus` graph until statement
cleanup. Under the default 1024-partition threshold, an unpartitioned table
remains non-batch regardless of file count, so an arbitrarily large listing
stays owned even after it has been converted into independent `FileSplit`s;
before this change the method-local list could be collected. Please bypass
statement retention when the global file cache is disabled, or apply a real
cumulative file/status weight cap with an oversize fallback, and cover a
one-partition many-file scan under the disabled-cache setting.
--
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]
