github-actions[bot] commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3829367764
##########
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:
[P2] Clean up ManifestFiles caches retained only by snapshots
A weighted/expired/collected table entry can disappear while its
independently admitted snapshot value still retains the same frozen `FileIO`.
On catalog reset this helper enumerates only `tableEntry`, so that
snapshot-only generation is missed; `super.invalidateCatalog*()` then removes
the last Doris owner without calling `ManifestFiles.dropCache()` for its
per-FileIO content cache. Those cached manifest bytes can therefore survive the
reset until soft-reference/weak-key cleanup. A load already holding the group
can also populate this SDK cache after the current pre-close pass, so please
make cleanup post-detach/fenced or generation-aware (including FileIOs from
both table and snapshot entries), and cover table eviction followed by catalog
invalidation.
--
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]
