924060929 commented on code in PR #66473:
URL: https://github.com/apache/doris/pull/66473#discussion_r3801152433
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -743,52 +745,147 @@ 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();
}
- return splits;
+ int[] projected = projectedColumns;
+ PaimonSplitTaskCacheKey cacheKey = createPaimonSplitTaskCacheKey(
+ relationSnapshot, paimonTable, resolvedOptions, projected);
+ return getOrLoadExternalScanTasks(cacheKey,
Review Comment:
Addressed on the current head. Paimon splits are retained only as serialized
payloads under the PAIMON_SERIALIZED_BYTES budget; the native
DataSplit/DataFileMeta graph is not kept in StatementContext after conversion.
##########
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:
Resolved on the current head. The statement cache is bypassed when
`max_external_file_cache_num <= 0` (`isBatchMode || !withCache`), so a disabled
global file cache never turns statement retention into an uncapped replacement.
##########
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:
Addressed on the current head. Ordinary Hudi partition planning uses the
TASK_COUNT cumulative budget with `maxRetainedExternalScanTasks` (default 10k)
and an oversize direct-use fallback, so a large single-partition plan is not
retained.
##########
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:
Addressed on the current head. The WeightedLoader receives the generation's
remaining budget (`reserveBeforeLoad=true`), and Paimon/Iceberg serializers
stop encoding once the remaining bytes are exhausted
(PaimonTaskCacheLimitException/IcebergTaskCacheLimitException), so a plan that
cannot be retained is never fully encoded.
--
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]
