CalvinKirs commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3829420614
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -251,36 +469,85 @@ private IcebergMetadataOps resolveMetadataOps(CatalogIf
catalog) {
throw new RuntimeException("Only support 'hms' and 'iceberg' type for
iceberg table");
}
+ /**
+ * Execute on the authenticated context captured with the table
generation, falling back to
+ * the catalog's current authenticator for values that predate the
capture. A concurrent
+ * property ALTER resets the catalog before retiring the group, so a
lookup that already
+ * owns the old generation must not resolve authentication from the
resetting catalog.
+ */
+ private <T> T executeForGeneration(
+ IcebergTableCacheValue tableValue, long catalogId, Callable<T>
task) {
+ org.apache.doris.common.security.authentication.ExecutionAuthenticator
authenticator =
+ tableValue.getAuthenticator();
+ if (authenticator == null) {
+ return executeAuthenticated(catalogId, task);
+ }
+ try {
+ return authenticator.execute(task);
+ } catch (Exception e) {
+ throw new RuntimeException(ExceptionUtils.getRootCauseMessage(e),
e);
+ }
+ }
+
+ private <T> T executeAuthenticated(long catalogId, Callable<T> task) {
+ CatalogIf<?> catalog = getCatalog(catalogId);
+ if (catalog == null) {
+ throw new RuntimeException("Cannot find catalog " + catalogId + "
when loading Iceberg metadata.");
+ }
+ return executeAuthenticated(catalog, task);
+ }
+
+ private <T> T executeAuthenticated(CatalogIf<?> catalog, Callable<T> task)
{
+ if (!(catalog instanceof ExternalCatalog)) {
+ throw new RuntimeException("Iceberg metadata cache requires an
external catalog");
+ }
+ try {
+ return ((ExternalCatalog)
catalog).getExecutionAuthenticator().execute(task);
+ } catch (Exception e) {
+ throw new RuntimeException(ExceptionUtils.getRootCauseMessage(e),
e);
+ }
+ }
+
@Override
protected Map<String, String> catalogPropertyCompatibilityMap() {
- return singleCompatibilityMap(ExternalCatalog.SCHEMA_CACHE_TTL_SECOND,
ENTRY_SCHEMA);
+ Map<String, String> compatibility = new java.util.HashMap<>(
+
singleCompatibilityMap(ExternalCatalog.SCHEMA_CACHE_TTL_SECOND, ENTRY_SCHEMA));
+ compatibility.put("meta.cache.iceberg.table.enable",
"meta.cache.iceberg.snapshot.enable");
+ compatibility.put("meta.cache.iceberg.table.ttl-second",
"meta.cache.iceberg.snapshot.ttl-second");
+ compatibility.put("meta.cache.iceberg.table.capacity",
"meta.cache.iceberg.snapshot.capacity");
+ return compatibility;
}
- private List<org.apache.iceberg.DataFile>
loadDataFiles(org.apache.iceberg.ManifestFile manifest, Table table)
+ private ManifestCacheValue loadDataFiles(
+ org.apache.iceberg.ManifestFile manifest, Table table, boolean
accountRetainedSize)
throws IOException {
- List<org.apache.iceberg.DataFile> dataFiles =
com.google.common.collect.Lists.newArrayList();
+ ManifestCacheValue.Builder builder =
ManifestCacheValue.dataFilesBuilder(accountRetainedSize);
try (ManifestReader<org.apache.iceberg.DataFile> reader =
ManifestFiles.read(manifest, table.io())) {
for (org.apache.iceberg.DataFile dataFile : reader) {
- dataFiles.add(dataFile.copy());
+ builder.addDataFile(dataFile.copy());
}
}
- return dataFiles;
+ return builder.build();
}
- private List<org.apache.iceberg.DeleteFile>
loadDeleteFiles(org.apache.iceberg.ManifestFile manifest, Table table)
+ private ManifestCacheValue loadDeleteFiles(
+ org.apache.iceberg.ManifestFile manifest, Table table, boolean
accountRetainedSize)
throws IOException {
- List<org.apache.iceberg.DeleteFile> deleteFiles =
com.google.common.collect.Lists.newArrayList();
+ ManifestCacheValue.Builder builder =
ManifestCacheValue.deleteFilesBuilder(accountRetainedSize);
try (ManifestReader<org.apache.iceberg.DeleteFile> reader =
ManifestFiles.readDeleteManifest(manifest,
table.io(), table.specs())) {
for (org.apache.iceberg.DeleteFile deleteFile : reader) {
- deleteFiles.add(deleteFile.copy());
+ builder.addDeleteFile(deleteFile.copy());
}
}
- return deleteFiles;
+ return builder.build();
}
private void dropManifestFileIoCacheForCatalog(long catalogId) {
- tableEntry.get(catalogId).forEach((key, value) ->
dropManifestFileIoCache(value));
+ MetaCacheEntry<NameMapping, IcebergTableCacheValue> tables =
tableEntry.getIfInitialized(catalogId);
Review Comment:
Fixed in 9c193c7a14a: catalog invalidation now collects FileIOs from both
the table and snapshot entries (identity de-duplicated) while they are still
enumerable, and calls ManifestFiles.dropCache only after
super.invalidateCatalog*() has detached the entries, so a snapshot-only frozen
generation is covered and a racing load can no longer repopulate a FileIO
behind the pre-detach pass. Regression added:
testCatalogInvalidationDropsManifestCacheForSnapshotOnlyFileIo (snapshot-only
generation dropped; shared generation dropped exactly once).
--
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]
