924060929 commented on code in PR #66473:
URL: https://github.com/apache/doris/pull/66473#discussion_r3782811785
##########
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:
Fixed on the current head `4df1422a510`.
`HiveExternalMetaCache.loadFileCacheValue()` stamps every newly loaded global
file-cache value with a monotonically increasing per-catalog generation.
`HiveFileScanTaskCacheKey` now includes the generations of the exact current
file entries returned during planning, in addition to the explicit invalidation
generation. Therefore an automatic refresh, expiry reload, or capacity-eviction
reload produces a different statement key and cannot reuse a listing from the
replaced entry.
`HiveScanNodeTest.testStatementCacheSeparatesAutomaticFileCacheReplacement`
covers equal entries versus a replacement generation.
##########
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:
Fixed on the current head `4df1422a510`. Hudi freezes
`StorageProperties.combinedFsCacheFingerprint(hmsTable.getStoragePropertiesMap().values())`
while initializing the relation snapshot and carries that fingerprint in
`HudiFileScanTaskCacheKey` equality and hashing. A catalog storage-property
update therefore cannot reuse a split whose `LocationPath` was normalized under
earlier properties.
`HudiScanNodeTest.testPartitionCacheKeySeparatesHmsMetadataGenerations` now
also verifies separation for a changed storage fingerprint.
--
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]
