github-actions[bot] commented on code in PR #66473:
URL: https://github.com/apache/doris/pull/66473#discussion_r3782475400
##########
fe/fe-core/src/main/java/org/apache/doris/job/executor/TaskProcessor.java:
##########
@@ -82,6 +84,22 @@ private void runTask(AbstractTask task) {
task.runTask();
} catch (Exception e) {
log.warn("Execute task error, task id: {}", task.getTaskId(), e);
+ } finally {
+ closeTaskContext();
Review Comment:
[P1] Clean up scheduled dictionary refresh contexts too
This cleanup only runs for jobs submitted to TaskProcessor. Auto-refreshing
dictionaries use DictionaryManager's separate fixed pool: dataLoad(null, ...)
installs a thread-local context through InsertTask.makeConnectContext(),
creates a StatementContext, and directly runs an INSERT ... SELECT from the
dictionary's arbitrary source catalog, but never closes that statement context
or removes/restores the worker context. An external dictionary refresh can
therefore leave the new scan-task cache pinned on an idle dictionary worker
after success or failure, up to the retained task/byte budgets. This is a
separate execution surface from the TaskProcessor/MTMV path covered by
discussion_r3780581260. Please put the scheduled dictionary context under a
cleanup finally and cover an external-source refresh on a reused dictionary
worker.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -380,29 +389,70 @@ 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(),
+ basePath, inputFormat, serdeLib, columnNames, columnTypes,
+ partitionColumnNames, partition);
+ plannedSplits = getOrLoadExternalScanTasks(
Review Comment:
[P1] Include storage normalization in the Hudi cache identity
The cached HudiSplit embeds a LocationPath normalized with the catalog's
current storage properties, but this key has no storage-property identity and
copyHudiSplit reuses that LocationPath. A catalog property update between two
alias planning calls can leave every key field equal while changing HDFS/S3
normalization; the second alias then receives the first alias's normalized
endpoint alongside its own newly derived backend/Hadoop properties. This is
distinct from discussion_r3773239592, which covers HMS table metadata fields.
Please include an immutable storage-configuration fingerprint in the key, or
cache raw paths and normalize them per consumer, and cover equal-instant
aliases across a storage-property update.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveExternalMetaCache.java:
##########
@@ -194,6 +210,19 @@ public void invalidateTable(long catalogId, String dbName,
String tableName) {
partitionEntry.get(catalogId).invalidateIf(key ->
matchTable(key.getNameMapping(), dbName, tableName));
long tableId = Util.genIdByName(dbName, tableName);
fileEntry.get(catalogId).invalidateIf(key -> key.isSameTable(tableId));
+ advanceFileCacheInvalidationGeneration(catalogId);
+ }
+
+ public long getFileCacheInvalidationGeneration(long catalogId) {
Review Comment:
[P1] Advance the generation on automatic file-cache replacement
This generation only changes in the explicit Hive invalidators, but the file
entry is an auto-refreshing, expiring, capacity-bounded MetaCacheEntry.
Caffeine can therefore refresh, expire, or evict and reload a FileCacheValue
without changing this value. If alias A retained the old listing under
generation G and the global entry is then replaced, alias B still builds
generation G and reuses A's removed files or misses newly added files instead
of observing the replacement. This is separate from discussion_r3774982788,
which covers explicit invalidation paths. Please couple the generation to
automatic file-entry replacement/removal (with the same ordering fence) and
test an alias after refresh or eviction.
--
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]
