github-actions[bot] commented on code in PR #67996:
URL: https://github.com/apache/doris/pull/67996#discussion_r4059855205
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java:
##########
@@ -543,11 +541,15 @@ private Catalog createCatalogFromContext(CatalogContext
catalogContext, String f
try {
Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
return context.executeAuthenticated(() -> {
+ // PaimonMetaCacheCatalog installs PrivilegedCatalog after the
raw metadata cache.
Catalog catalog = PaimonCatalogProperties.HMS.equals(flavor)
? createHmsCatalog(catalogContext, hmsAuth,
catalogProps.getRaw(),
storageHadoopConfig)
- : CatalogFactory.createCatalog(catalogContext);
- return catalog;
+ :
CatalogFactory.createUnwrappedCatalog(catalogContext,
getClass().getClassLoader());
+ return PaimonMetaCacheCatalog.tryToCreate(catalog, metaCache,
Review Comment:
[P2] Fence this cache-backed catalog publication against close
A statement can retain this connector while `resetToUninitialized()` closes
it. `ensureCatalog()` constructs under the connector monitor, but `close()`
does not take that monitor: it can close `metaCache`, observe the still-null
`catalog`, and return after these two caches have registered but before
`tryToCreate` returns. `ensureCatalog()` then publishes a wrapper backed by
closed caches, so its first table/database lookup throws `Scoped meta cache ...
is closed`, and no later close owns the raw catalog. Please add a connector
lifecycle fence so a successful construction cannot publish after close
(closing the losing raw/decorated catalog exactly once), plus a latch-based
close-versus-construction test.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimator.java:
##########
@@ -0,0 +1,163 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.connector.paimon;
+
+import org.apache.doris.connector.cache.JvmSizeUtils;
+import org.apache.doris.connector.cache.MetaCacheSizeEstimate;
+import org.apache.doris.connector.cache.ReflectiveObjectSizeEstimator;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.privilege.PrivilegedFileStoreTable;
+import org.apache.paimon.table.CatalogEnvironment;
+import org.apache.paimon.table.DelegatedFileStoreTable;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.iceberg.IcebergTable;
+import org.apache.paimon.table.lance.LanceTable;
+import org.apache.paimon.table.object.ObjectTable;
+
+import java.net.URI;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.Set;
+
+/**
+ * Retained-size formulas for Paimon table-cache entries.
+ *
+ * <p>The table's shallow size includes references to FileIO, catalog loaders,
and lock factories,
+ * but their graphs are catalog-scoped executable services rather than
entry-owned metadata. Walking
+ * those graphs both double-counts shared state and reaches strongly
encapsulated JDK objects. The
+ * estimator therefore expands only immutable metadata owned by the entry.
+ */
+final class PaimonCacheSizeEstimator {
+ private PaimonCacheSizeEstimator() {
+ }
+
+ static MetaCacheSizeEstimate estimateTable(Identifier key, Table table,
long entryOverheadBytes) {
+ if (table instanceof PrivilegedFileStoreTable) {
+ return MetaCacheSizeEstimate.incomplete(
+ "authorization decorators must be applied outside the
metadata cache");
+ }
+ long bytes = add(entryOverheadBytes,
ReflectiveObjectSizeEstimator.estimateComplete(key));
+ if (table instanceof FileStoreTable) {
+ Set<Object> visited = Collections.newSetFromMap(new
IdentityHashMap<>());
+ bytes = add(bytes, estimateFileStoreTable((FileStoreTable) table,
visited));
+ } else {
+ if (!isSupportedNonFileStoreTable(table)) {
+ return MetaCacheSizeEstimate.incomplete(
+ "unsupported retained graph for " +
table.getClass().getName());
+ }
+ bytes = add(bytes, JvmSizeUtils.instanceSize(table.getClass()));
+ bytes = add(bytes,
ReflectiveObjectSizeEstimator.estimateComplete(table.rowType()));
+ bytes = add(bytes,
ReflectiveObjectSizeEstimator.estimateComplete(table.partitionKeys()));
+ bytes = add(bytes,
ReflectiveObjectSizeEstimator.estimateComplete(table.primaryKeys()));
+ bytes = add(bytes,
ReflectiveObjectSizeEstimator.estimateComplete(table.options()));
+ bytes = add(bytes,
ReflectiveObjectSizeEstimator.estimateComplete(table.comment()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(location(table)));
+ }
+ return MetaCacheSizeEstimate.complete(bytes);
+ }
+
+ private static boolean isSupportedNonFileStoreTable(Table table) {
+ return table instanceof FormatTable
+ || table instanceof ObjectTable
+ || table instanceof LanceTable
+ || table instanceof IcebergTable;
+ }
+
+ private static long estimateFileStoreTable(FileStoreTable table,
Set<Object> visited) {
+ if (!visited.add(table)) {
+ return 0L;
+ }
+ long bytes = JvmSizeUtils.instanceSize(table.getClass());
+ if (table instanceof FallbackReadFileStoreTable) {
+ FallbackReadFileStoreTable fallback = (FallbackReadFileStoreTable)
table;
+ bytes = add(bytes, estimateFileStoreTable(fallback.wrapped(),
visited));
+ return add(bytes, estimateFileStoreTable(fallback.fallback(),
visited));
+ }
+ if (table instanceof DelegatedFileStoreTable) {
+ return add(bytes, estimateFileStoreTable(
+ ((DelegatedFileStoreTable) table).wrapped(), visited));
+ }
+ bytes = add(bytes, estimateCompleteOnce(table.schema(), visited));
+ bytes = add(bytes, estimatePath(table.location(), visited));
+ return add(bytes,
estimateCatalogEnvironment(table.catalogEnvironment(), visited));
Review Comment:
[P2] Account for the lazily retained FileStore before admission
This marks a normal `FileStoreTable` complete while its entry-owned graph
can still grow after the one-time weight reservation. In Paimon 1.3.1, both
`AppendOnlyFileStoreTable` and `PrimaryKeyFileStoreTable` start with `lazyStore
== null`; the first ordinary `latestSnapshot()` call from
`beginQuerySnapshot()` invokes `store()` and retains a newly allocated
`AppendOnlyFileStore`/`KeyValueFileStore` plus its managers and factories on
the cached raw table. Under an enclosing weight limit the three explicit SDK
caches are skipped, but this lazy field is still populated and
`ScopedMetaCache` never reweighs reads, so many cached tables can exceed the
configured hard byte budget. Please initialize and charge this retained graph
before returning `complete`, reject the entry under weight governance, or cache
an immutable descriptor. Extend the governed real-table test past `getTable()`
into `latestSnapshot()`/`newScan()` to cover post-admission growth.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalog.java:
##########
@@ -0,0 +1,410 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.connector.paimon;
+
+import org.apache.doris.connector.cache.CacheSpec;
+import org.apache.doris.connector.cache.CatalogMetaCache;
+import org.apache.doris.connector.cache.JvmSizeUtils;
+import org.apache.doris.connector.cache.MetaCache;
+import org.apache.doris.connector.cache.MetaCacheDefinition;
+import org.apache.doris.connector.cache.MetaCacheSizeEstimators;
+import org.apache.doris.connector.cache.ScopePath;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogLoader;
+import org.apache.paimon.catalog.Database;
+import org.apache.paimon.catalog.DelegateCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.catalog.PropertyChange;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.privilege.PrivilegedCatalog;
+import org.apache.paimon.schema.SchemaChange;
+import
org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.system.SystemTableLoader;
+import org.apache.paimon.utils.SegmentsCache;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+import java.util.function.LongSupplier;
+
+/**
+ * Doris-owned replacement for Paimon's {@code CachingCatalog}. Table and
database entries live in
+ * {@link CatalogMetaCache}, so a Doris catalog/database/table invalidation
fences every matching
+ * in-flight load and cached value.
+ *
+ * <p>The cache retains raw table metadata. Paimon's privilege catalog is
applied outside this
+ * wrapper so every lookup receives a fresh checker instead of caching one
authorization snapshot.
+ *
+ * <p>The user's {@code paimon.cache-enabled} and access/write expiry settings
remain authoritative.
+ * The Paimon SDK wrapper itself is disabled because a second hidden table
cache cannot participate
+ * in Doris invalidation. Under a Doris weight budget, mutable SDK
snapshot/stats/manifest caches are
+ * not attached: their post-publication growth cannot be reweighed by the
enclosing budget.
+ */
+final class PaimonMetaCacheCatalog extends DelegateCatalog {
+
+ private static final int DATABASE_CACHE_CAPACITY = 100;
+ static final long TABLE_ENTRY_OVERHEAD_BYTES = JvmSizeUtils.saturatedAdd(
+ JvmSizeUtils.instanceSize(ExpiringValue.class),
JvmSizeUtils.instanceSize(AtomicLong.class));
+
+ private final CatalogMetaCache metaCache;
+ private final MetaCache<Identifier, ExpiringValue<Table>> tableCache;
+ private final MetaCache<String, ExpiringValue<Database>> databaseCache;
+ private final SegmentsCache<Path> manifestCache;
+ private final long tableExpireAfterAccessNanos;
+ private final long databaseExpireAfterAccessNanos;
+ private final long expireAfterWriteNanos;
+ private final int snapshotMaxNumPerTable;
+ private final boolean attachSdkCaches;
+ private final LongSupplier nanoTime;
+ private final BiConsumer<String, Object> cacheMissObserver;
+
+ static Catalog tryToCreate(Catalog wrapped, CatalogMetaCache metaCache,
int tableCacheMaxSize,
+ long tableCacheTtlSecond, Options catalogOptions, boolean
cacheEnabled,
+ boolean hasEnclosingWeightLimit) {
+ return tryToCreate(wrapped, metaCache, tableCacheMaxSize,
tableCacheTtlSecond,
+ catalogOptions, cacheEnabled, hasEnclosingWeightLimit,
+ catalog -> PrivilegedCatalog.tryToCreate(catalog,
catalogOptions));
+ }
+
+ static Catalog tryToCreate(Catalog wrapped, CatalogMetaCache metaCache,
int tableCacheMaxSize,
+ long tableCacheTtlSecond, Options catalogOptions, boolean
cacheEnabled,
+ boolean hasEnclosingWeightLimit, Function<Catalog, Catalog>
decorator) {
+ PaimonMetaCacheCatalog cached = null;
+ try {
+ cached = new PaimonMetaCacheCatalog(wrapped, metaCache,
tableCacheMaxSize,
+ tableCacheTtlSecond, catalogOptions, cacheEnabled,
hasEnclosingWeightLimit,
+ System::nanoTime, (name, key) -> { });
+ return decorator.apply(cached);
+ } catch (RuntimeException | Error throwable) {
+ if (cached != null) {
+ try {
+ cached.unregisterCaches();
+ } catch (RuntimeException | Error rollbackFailure) {
+ throwable.addSuppressed(rollbackFailure);
+ }
+ }
+ try {
+ wrapped.close();
+ } catch (Exception | Error closeFailure) {
+ throwable.addSuppressed(closeFailure);
+ }
+ throw throwable;
+ }
+ }
+
+ PaimonMetaCacheCatalog(Catalog wrapped, CatalogMetaCache metaCache, int
tableCacheMaxSize,
+ long tableCacheTtlSecond, Options catalogOptions, boolean
hasEnclosingWeightLimit,
+ LongSupplier nanoTime) {
+ this(wrapped, metaCache, tableCacheMaxSize, tableCacheTtlSecond,
catalogOptions,
+ true, hasEnclosingWeightLimit, nanoTime, (name, key) -> { });
+ }
+
+ PaimonMetaCacheCatalog(Catalog wrapped, CatalogMetaCache metaCache, int
tableCacheMaxSize,
+ long tableCacheTtlSecond, Options catalogOptions, boolean
hasEnclosingWeightLimit,
+ LongSupplier nanoTime, BiConsumer<String, Object>
cacheMissObserver) {
+ this(wrapped, metaCache, tableCacheMaxSize, tableCacheTtlSecond,
catalogOptions,
+ true, hasEnclosingWeightLimit, nanoTime, cacheMissObserver);
+ }
+
+ private PaimonMetaCacheCatalog(Catalog wrapped, CatalogMetaCache
metaCache, int tableCacheMaxSize,
+ long tableCacheTtlSecond, Options catalogOptions, boolean
cacheEnabled,
+ boolean hasEnclosingWeightLimit, LongSupplier nanoTime,
+ BiConsumer<String, Object> cacheMissObserver) {
+ super(wrapped);
+ this.metaCache = metaCache;
+ this.nanoTime = nanoTime;
+ this.cacheMissObserver = cacheMissObserver;
+
+ Duration expireAfterAccess = cacheEnabled
+ ? catalogOptions.get(CatalogOptions.CACHE_EXPIRE_AFTER_ACCESS)
: null;
+ Duration expireAfterWrite = cacheEnabled
+ ? catalogOptions.get(CatalogOptions.CACHE_EXPIRE_AFTER_WRITE)
: null;
+ if (cacheEnabled) {
+ requirePositive(expireAfterAccess,
CatalogOptions.CACHE_EXPIRE_AFTER_ACCESS.key());
+ requirePositive(expireAfterWrite,
CatalogOptions.CACHE_EXPIRE_AFTER_WRITE.key());
+ }
+ long paimonAccessNanos = cacheEnabled ?
saturatedNanos(expireAfterAccess) : Long.MAX_VALUE;
+ this.tableExpireAfterAccessNanos = cacheEnabled && tableCacheTtlSecond
> 0
+ ? Math.min(paimonAccessNanos,
saturatedNanos(Duration.ofSeconds(tableCacheTtlSecond)))
+ : paimonAccessNanos;
+ this.databaseExpireAfterAccessNanos = paimonAccessNanos;
+ this.expireAfterWriteNanos = cacheEnabled ?
saturatedNanos(expireAfterWrite) : Long.MAX_VALUE;
+ this.attachSdkCaches = cacheEnabled && !hasEnclosingWeightLimit;
+ this.manifestCache = attachSdkCaches ?
buildManifestCache(catalogOptions) : null;
+ this.snapshotMaxNumPerTable = attachSdkCaches
+ ?
catalogOptions.get(CatalogOptions.CACHE_SNAPSHOT_MAX_NUM_PER_TABLE) : 0;
+
+ CacheSpec tableSpec = CacheSpec.of(cacheEnabled,
+ cacheEnabled && tableCacheTtlSecond > 0
+ ? CacheSpec.CACHE_NO_TTL :
CacheSpec.CACHE_TTL_DISABLE_CACHE,
+ tableCacheMaxSize);
+ MetaCache<Identifier, ExpiringValue<Table>> createdTableCache =
metaCache.create(
+ MetaCacheDefinition
+ .<Identifier,
ExpiringValue<Table>>builder("paimon-table", tableSpec,
+ id -> ScopePath.table(id.getDatabaseName(),
id.getTableName()))
+ .sizeEstimator((id, value) ->
PaimonCacheSizeEstimator.estimateTable(
+ id, value.value, TABLE_ENTRY_OVERHEAD_BYTES))
+ .build());
+ try {
+ CacheSpec dbSpec = CacheSpec.of(cacheEnabled, cacheEnabled
+ ? CacheSpec.CACHE_NO_TTL :
CacheSpec.CACHE_TTL_DISABLE_CACHE, DATABASE_CACHE_CAPACITY);
+ this.databaseCache = metaCache.create(MetaCacheDefinition
+ .<String,
ExpiringValue<Database>>builder("paimon-database", dbSpec, ScopePath::database)
+ .sizeEstimator(MetaCacheSizeEstimators.reflective())
+ .build());
+ } catch (RuntimeException | Error throwable) {
+ try {
+ metaCache.remove(createdTableCache);
+ } catch (RuntimeException | Error rollbackFailure) {
+ throwable.addSuppressed(rollbackFailure);
+ }
+ throw throwable;
+ }
+ this.tableCache = createdTableCache;
+ }
+
+ @Override
+ public Table getTable(Identifier identifier) throws TableNotExistException
{
+ if (identifier.isSystemTable()) {
+ Identifier origin = new Identifier(identifier.getDatabaseName(),
identifier.getTableName(),
+ identifier.getBranchName(), null);
+ Table originTable = getTable(origin);
+ if (!(originTable instanceof FileStoreTable)) {
+ return super.getTable(identifier);
+ }
+ Table systemTable =
SystemTableLoader.load(identifier.getSystemTableName(),
+ (FileStoreTable) originTable);
+ if (systemTable == null) {
+ throw new TableNotExistException(identifier);
+ }
+ return systemTable;
+ }
+
+ while (true) {
+ ExpiringValue<Table> cached = tableCache.getIfPresent(identifier);
+ if (cached == null) {
+ cacheMissObserver.accept("table", identifier);
+ try {
+ cached = tableCache.get(identifier, ignored -> {
+ try {
+ Table loaded =
attachPerTableCaches(super.getTable(identifier));
+ return new ExpiringValue<>(loaded,
nanoTime.getAsLong());
+ } catch (TableNotExistException e) {
+ throw new CatalogLoadException(e);
+ }
+ });
+ } catch (CatalogLoadException e) {
+ throw (TableNotExistException) e.getCause();
+ }
+ }
+ if (cached.tryAccess(nanoTime.getAsLong(),
tableExpireAfterAccessNanos,
Review Comment:
[P2] Return a freshly loaded value before re-expiring it
This unconditional recheck can make a cold lookup nonterminating for a valid
short positive expiry. For example, with `cache.expire-after-access=1 ns`, the
loader records one `nanoTime`, publication completes, and this second clock
read makes `tryAccess` reject the just-loaded value. The CAS removes it and
`while (true)` repeats the remote load forever; the database path has the same
loop, and disabled/rejected physical admission cannot break it. Please
distinguish the current shared load's fresh result from a pre-existing miss-gap
result (or age it from publication) so this caller returns once, while
retaining the resolved miss-gap revalidation. A deterministic advancing-clock
test should cover both table and database lookups.
--
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]
