This is an automated email from the ASF dual-hosted git repository.
924060929 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 69b803b32a5 [refactor](paimon) Replace Paimon SDK CachingCatalog with
PaimonMetaCacheCatalog (#67996)
69b803b32a5 is described below
commit 69b803b32a5ec914a4f4bfdaf45f6347c48435a0
Author: 924060929 <[email protected]>
AuthorDate: Mon Sep 21 17:41:41 2026 +0800
[refactor](paimon) Replace Paimon SDK CachingCatalog with
PaimonMetaCacheCatalog (#67996)
### What problem does this PR solve?
Problem Summary:
The Paimon SDK `CachingCatalog` wraps every catalog by default and
caches
Table objects with frozen schema/snapshot state, exposing only per-table
`invalidateTable(Identifier)` — no db/catalog-level eviction. After an
external same-name drop/recreate of a Paimon table, `REFRESH TABLE` only
flushes Doris-owned caches; the SDK's frozen Table survives and keeps
serving stale data
This PR replaces the SDK `CachingCatalog` entirely with a new
`PaimonMetaCacheCatalog` (extends `DelegateCatalog`) that places every
catalog-level cache inside Doris's own `CatalogMetaCache` framework so
`REFRESH TABLE/DATABASE/CATALOG` invalidates them through the same
registry path as every other connector-owned cache.
### Design
`cache-enabled=false` is forced unconditionally — PaimonMetaCacheCatalog
and the SDK CachingCatalog cannot coexist: a Doris-side MetaCache miss
falls through to `super.getTable()` which hits the SDK cache and returns
a frozen Table, so the stale-read bug persists even after REFRESH.
The `hasEnclosingMetaCacheWeightLimit` parameter is retained in the API
signature for compatibility but is a no-op (the SDK cache is always
off).
Key cache layers provided by PaimonMetaCacheCatalog:
- `tableCache` → `MetaCache<Identifier,Table>` (scope: table)
- `databaseCache` → `MetaCache<String,Database>` (scope: database)
- per-FileStoreTable caches (snapshot/stats/manifest) retained on load
### Release note
Fix stale reads after an external drop/recreate of a same-name Paimon
table:
Doris-side `REFRESH TABLE` now fully invalidates the Paimon SDK table
cache.
---
.../doris/connector/cache/CatalogMetaCache.java | 15 +
.../apache/doris/connector/cache/MetaCache.java | 4 +
.../connector/paimon/PaimonCacheSizeEstimator.java | 163 +++++
.../connector/paimon/PaimonCatalogFactory.java | 17 +-
.../doris/connector/paimon/PaimonCatalogOps.java | 13 +-
.../doris/connector/paimon/PaimonConnector.java | 15 +-
.../connector/paimon/PaimonMetaCacheCatalog.java | 410 +++++++++++++
.../connector/paimon/PaimonCatalogFactoryTest.java | 29 +-
.../paimon/PaimonCatalogOptionsSnapshotTest.java | 32 +-
.../paimon/PaimonMetaCacheCatalogTest.java | 679 +++++++++++++++++++++
.../paimon/PaimonRestCatalogPartitionsTest.java | 20 +
11 files changed, 1361 insertions(+), 36 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CatalogMetaCache.java
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CatalogMetaCache.java
index 84b5022952e..d84c389b04b 100644
---
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CatalogMetaCache.java
+++
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CatalogMetaCache.java
@@ -132,6 +132,21 @@ public final class CatalogMetaCache implements
AutoCloseable {
}
}
+ /**
+ * Removes one cache registration if it is still owned by this catalog.
This is intended for
+ * rolling back a multi-step connector construction without closing
unrelated sibling caches.
+ */
+ public void remove(MetaCache<?, ?> cache) {
+ MetaCache<?, ?> nonNullCache = Objects.requireNonNull(cache, "cache
can not be null");
+ if (entries.remove(nonNullCache.name(), nonNullCache)) {
+ try {
+ nonNullCache.closeFromOwner();
+ } finally {
+ names.remove(nonNullCache.name());
+ }
+ }
+ }
+
public void invalidateCatalog() {
registry.invalidate(ScopePath.catalog());
}
diff --git
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java
index 1447fd5a83f..07afd00576d 100644
---
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java
+++
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java
@@ -135,6 +135,10 @@ public final class MetaCache<K, V> {
delegate.forEach(consumer);
}
+ void closeFromOwner() {
+ delegate.close();
+ }
+
public static final class BulkLoad<K, V> implements AutoCloseable {
private final MetaCache<K, V> owner;
private final ScopedMetaCache.BulkLoadHandle delegate;
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimator.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimator.java
new file mode 100644
index 00000000000..ffd4cdfe10d
--- /dev/null
+++
b/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));
+ }
+
+ private static long estimateCompleteOnce(Object value, Set<Object>
visited) {
+ return value == null || !visited.add(value)
+ ? 0L : ReflectiveObjectSizeEstimator.estimateComplete(value);
+ }
+
+ private static long estimateCatalogEnvironment(CatalogEnvironment
environment, Set<Object> visited) {
+ if (environment == null || !visited.add(environment)) {
+ return 0L;
+ }
+ long bytes = JvmSizeUtils.instanceSize(environment.getClass());
+ bytes = add(bytes, estimateCompleteOnce(environment.identifier(),
visited));
+ return add(bytes, estimateStringOnce(environment.uuid(), visited));
+ }
+
+ private static long estimatePath(Path path, Set<Object> visited) {
+ if (path == null || !visited.add(path)) {
+ return 0L;
+ }
+ URI uri = path.toUri();
+ long bytes = JvmSizeUtils.instanceSize(path.getClass());
+ if (visited.add(uri)) {
+ bytes = add(bytes, JvmSizeUtils.instanceSize(uri.getClass()));
+ bytes = add(bytes, estimateStringOnce(uri.toString(), visited));
+ bytes = add(bytes, estimateStringOnce(uri.getScheme(), visited));
+ bytes = add(bytes, estimateStringOnce(uri.getUserInfo(), visited));
+ bytes = add(bytes, estimateStringOnce(uri.getHost(), visited));
+ bytes = add(bytes, estimateStringOnce(uri.getPath(), visited));
+ bytes = add(bytes, estimateStringOnce(uri.getQuery(), visited));
+ bytes = add(bytes, estimateStringOnce(uri.getFragment(), visited));
+ bytes = add(bytes, estimateStringOnce(uri.getAuthority(),
visited));
+ bytes = add(bytes, estimateStringOnce(uri.getSchemeSpecificPart(),
visited));
+ }
+ return bytes;
+ }
+
+ private static long estimateStringOnce(String value, Set<Object> visited) {
+ return value == null || !visited.add(value) ? 0L :
JvmSizeUtils.stringSize(value);
+ }
+
+ private static String location(Table table) {
+ if (table instanceof FormatTable) {
+ return ((FormatTable) table).location();
+ }
+ if (table instanceof ObjectTable) {
+ return ((ObjectTable) table).location();
+ }
+ if (table instanceof LanceTable) {
+ return ((LanceTable) table).location();
+ }
+ if (table instanceof IcebergTable) {
+ return ((IcebergTable) table).location();
+ }
+ return null;
+ }
+
+ private static long add(long left, long right) {
+ return JvmSizeUtils.saturatedAdd(left, right);
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogFactory.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogFactory.java
index 3e52564feec..e6c96cdab9d 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogFactory.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogFactory.java
@@ -103,11 +103,19 @@ public final class PaimonCatalogFactory {
* plus each flavor's {@code appendCustomCatalogOptions()}.
*/
public static Options buildCatalogOptions(PaimonCatalogProperties
catalogProperties) {
- return buildCatalogOptions(catalogProperties, false);
+ Options options = assembleCatalogOptions(catalogProperties);
+ // PaimonMetaCacheCatalog replaces the SDK CachingCatalog. The two
cache layers cannot
+ // coexist because a Doris invalidation cannot evict a frozen Table
from the SDK wrapper.
+ // Preserve the user's flag separately, but always disable the SDK
wrapper itself.
+ options.set(CatalogOptions.CACHE_ENABLED, false);
+ return options;
+ }
+
+ static boolean isCatalogCacheEnabled(PaimonCatalogProperties
catalogProperties) {
+ return
assembleCatalogOptions(catalogProperties).get(CatalogOptions.CACHE_ENABLED);
}
- static Options buildCatalogOptions(
- PaimonCatalogProperties catalogProperties, boolean
hasEnclosingMetaCacheWeightLimit) {
+ private static Options assembleCatalogOptions(PaimonCatalogProperties
catalogProperties) {
Options options = new Options();
Map<String, String> props = catalogProperties.getRaw();
String flavor = catalogProperties.getFlavor();
@@ -141,9 +149,6 @@ public final class PaimonCatalogFactory {
// filesystem: nothing custom.
break;
}
- if (hasEnclosingMetaCacheWeightLimit &&
!options.contains(CatalogOptions.CACHE_ENABLED)) {
- options.set(CatalogOptions.CACHE_ENABLED, false);
- }
return options;
}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java
index b52ffd8f360..584eb9f7e83 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java
@@ -22,6 +22,7 @@ import org.apache.paimon.Snapshot;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogUtils;
import org.apache.paimon.catalog.Database;
+import org.apache.paimon.catalog.DelegateCatalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.privilege.PrivilegedFileStoreTable;
@@ -304,25 +305,29 @@ public interface PaimonCatalogOps {
public Table getTable(Identifier identifier) throws
Catalog.TableNotExistException {
Table table = catalog.getTable(identifier);
Map<String, String> optionsForCopy =
PaimonTableOptions.forCopy(tableOptions);
- // Relation options are applied after this cached handle is
returned. Defer final
- // validation so a safe relation value can override an unsafe
physical value.
return optionsForCopy.isEmpty() ? table :
table.copy(optionsForCopy);
}
@Override
public List<Partition> listPartitions(Identifier identifier, Table
table)
throws Catalog.TableNotExistException {
- if (catalog instanceof RESTCatalog) {
+ RESTCatalog restCatalog = restCatalog(catalog);
+ if (restCatalog != null) {
// REST owns partition visibility when its endpoint is
implemented; the bridge
// retains this effective relation copy only for the
endpoint's filesystem fallback.
return PaimonRestCatalogPartitions.listPartitions(
- (RESTCatalog) catalog, identifier, table);
+ restCatalog, identifier, table);
}
// The supplied handle already contains catalog and relation
policy. Reloading by identifier
// would discard those copies before manifest enumeration reaches
the final scan guard.
return CatalogUtils.listPartitionsFromFileSystem(table);
}
+ static RESTCatalog restCatalog(Catalog catalog) {
+ Catalog rootCatalog = DelegateCatalog.rootCatalog(catalog);
+ return rootCatalog instanceof RESTCatalog ? (RESTCatalog)
rootCatalog : null;
+ }
+
@Override
public void createDatabase(String name, boolean ignoreIfExists,
Map<String, String> properties)
throws Catalog.DatabaseAlreadyExistException {
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java
index b00efbcb577..a47f90e0951 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java
@@ -50,7 +50,6 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
-import org.apache.paimon.catalog.CachingCatalog;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
@@ -60,7 +59,6 @@ import org.apache.paimon.hive.HiveCatalog;
import org.apache.paimon.hive.HiveCatalogOptions;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
-import org.apache.paimon.privilege.PrivilegedCatalog;
import java.io.IOException;
import java.io.UncheckedIOException;
@@ -499,7 +497,7 @@ public class PaimonConnector implements Connector {
}
Options buildCatalogOptions() {
- return PaimonCatalogFactory.buildCatalogOptions(catalogProps,
metaCache.hasEnclosingWeightLimit());
+ return PaimonCatalogFactory.buildCatalogOptions(catalogProps);
}
/**
@@ -543,11 +541,15 @@ public class PaimonConnector implements Connector {
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,
+ DEFAULT_TABLE_CACHE_CAPACITY,
resolveTableCacheTtlSecond(catalogProps.getRaw()),
+ catalogContext.options(),
PaimonCatalogFactory.isCatalogCacheEnabled(catalogProps),
+ metaCache.hasEnclosingWeightLimit());
});
} catch (Exception e) {
throw new RuntimeException(failureMessage + " (flavor=" + flavor +
"): " + e.getMessage(), e);
@@ -579,8 +581,7 @@ public class PaimonConnector implements Connector {
fileIO, hiveConf, clientClass, options,
warehousePath.toUri().toString()));
catalog = PaimonHmsClientPool.install(catalog, hmsAuth);
catalog = PaimonHmsCatalog.install(catalog, properties,
storageHadoopConfig);
- catalog = CachingCatalog.tryToCreate(catalog, options);
- return PrivilegedCatalog.tryToCreate(catalog, options);
+ return catalog;
} catch (IOException e) {
throw new UncheckedIOException(e);
}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalog.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalog.java
new file mode 100644
index 00000000000..b537b6f1e13
--- /dev/null
+++
b/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,
+ expireAfterWriteNanos)) {
+ return cached.value;
+ }
+ tableCache.compareAndSet(identifier, cached, null);
+ }
+ }
+
+ @Override
+ public Database getDatabase(String name) throws DatabaseNotExistException {
+ while (true) {
+ ExpiringValue<Database> cached = databaseCache.getIfPresent(name);
+ if (cached == null) {
+ cacheMissObserver.accept("database", name);
+ try {
+ cached = databaseCache.get(name, ignored -> {
+ try {
+ return new
ExpiringValue<>(super.getDatabase(name), nanoTime.getAsLong());
+ } catch (DatabaseNotExistException e) {
+ throw new CatalogLoadException(e);
+ }
+ });
+ } catch (CatalogLoadException e) {
+ throw (DatabaseNotExistException) e.getCause();
+ }
+ }
+ if (cached.tryAccess(nanoTime.getAsLong(),
databaseExpireAfterAccessNanos,
+ expireAfterWriteNanos)) {
+ return cached.value;
+ }
+ databaseCache.compareAndSet(name, cached, null);
+ }
+ }
+
+ @Override
+ public void dropDatabase(String name, boolean ignoreIfNotExists, boolean
cascade)
+ throws DatabaseNotExistException, DatabaseNotEmptyException {
+ try {
+ super.dropDatabase(name, ignoreIfNotExists, cascade);
+ } finally {
+ metaCache.invalidateDatabase(name);
+ }
+ }
+
+ @Override
+ public void alterDatabase(String name, List<PropertyChange> changes,
boolean ignoreIfNotExists)
+ throws DatabaseNotExistException {
+ try {
+ super.alterDatabase(name, changes, ignoreIfNotExists);
+ } finally {
+ metaCache.invalidateDatabase(name);
+ }
+ }
+
+ @Override
+ public void dropTable(Identifier identifier, boolean ignoreIfNotExists)
+ throws TableNotExistException {
+ try {
+ super.dropTable(identifier, ignoreIfNotExists);
+ } finally {
+ invalidateTable(identifier);
+ }
+ }
+
+ @Override
+ public void renameTable(Identifier fromTable, Identifier toTable, boolean
ignoreIfNotExists)
+ throws TableNotExistException, TableAlreadyExistException {
+ try {
+ super.renameTable(fromTable, toTable, ignoreIfNotExists);
+ } finally {
+ invalidateTable(fromTable);
+ invalidateTable(toTable);
+ }
+ }
+
+ @Override
+ public void alterTable(Identifier identifier, List<SchemaChange> changes,
boolean ignoreIfNotExists)
+ throws TableNotExistException, ColumnAlreadyExistException,
ColumnNotExistException {
+ try {
+ super.alterTable(identifier, changes, ignoreIfNotExists);
+ } finally {
+ invalidateTable(identifier);
+ }
+ }
+
+ @Override
+ public void invalidateTable(Identifier identifier) {
+ metaCache.invalidateTable(identifier.getDatabaseName(),
identifier.getTableName());
+ super.invalidateTable(identifier);
+ }
+
+ @Override
+ public CatalogLoader catalogLoader() {
+ return wrapped.catalogLoader();
+ }
+
+ private void unregisterCaches() {
+ metaCache.remove(databaseCache);
+ metaCache.remove(tableCache);
+ }
+
+ private Table attachPerTableCaches(Table table) {
+ if (!attachSdkCaches || !(table instanceof FileStoreTable)) {
+ return table;
+ }
+ FileStoreTable storeTable = (FileStoreTable) table;
+ Duration expireAfterAccess =
Duration.ofNanos(databaseExpireAfterAccessNanos);
+ Duration expireAfterWrite = Duration.ofNanos(expireAfterWriteNanos);
+ storeTable.setSnapshotCache(Caffeine.newBuilder()
+ .softValues()
+ .expireAfterAccess(expireAfterAccess)
+ .expireAfterWrite(expireAfterWrite)
+ .maximumSize(snapshotMaxNumPerTable)
+ .executor(Runnable::run)
+ .build());
+ storeTable.setStatsCache(Caffeine.newBuilder()
+ .softValues()
+ .expireAfterAccess(expireAfterAccess)
+ .expireAfterWrite(expireAfterWrite)
+ .maximumSize(5)
+ .executor(Runnable::run)
+ .build());
+ storeTable.setManifestCache(manifestCache);
+ return table;
+ }
+
+ private static SegmentsCache<Path> buildManifestCache(Options options) {
+ MemorySize manifestMaxMemory =
options.get(CatalogOptions.CACHE_MANIFEST_SMALL_FILE_MEMORY);
+ long manifestCacheThreshold = options.get(
+ CatalogOptions.CACHE_MANIFEST_SMALL_FILE_THRESHOLD).getBytes();
+ Optional<MemorySize> maxMemory =
options.getOptional(CatalogOptions.CACHE_MANIFEST_MAX_MEMORY);
+ if (maxMemory.isPresent() &&
maxMemory.get().compareTo(manifestMaxMemory) > 0) {
+ manifestMaxMemory = maxMemory.get();
+ manifestCacheThreshold = Long.MAX_VALUE;
+ }
+ return SegmentsCache.create(manifestMaxMemory, manifestCacheThreshold);
+ }
+
+ private static void requirePositive(Duration duration, String option) {
+ if (duration.isZero() || duration.isNegative()) {
+ throw new IllegalArgumentException("When '" + option
+ + "' is set to negative or 0, the catalog cache should be
disabled.");
+ }
+ }
+
+ private static long saturatedNanos(Duration duration) {
+ try {
+ return duration.toNanos();
+ } catch (ArithmeticException e) {
+ return Long.MAX_VALUE;
+ }
+ }
+
+ private static final class ExpiringValue<V> {
+ private final V value;
+ private final long createdNanos;
+ private final AtomicLong lastAccessNanos;
+
+ private ExpiringValue(V value, long createdNanos) {
+ this.value = value;
+ this.createdNanos = createdNanos;
+ this.lastAccessNanos = new AtomicLong(createdNanos);
+ }
+
+ private boolean tryAccess(long now, long expireAfterAccessNanos, long
expireAfterWriteNanos) {
+ while (true) {
+ long lastAccess = lastAccessNanos.get();
+ if (now - createdNanos >= expireAfterWriteNanos
+ || now - lastAccess >= expireAfterAccessNanos) {
+ return false;
+ }
+ if (now <= lastAccess) {
+ return true;
+ }
+ if (lastAccessNanos.compareAndSet(lastAccess, now)) {
+ return true;
+ }
+ }
+ }
+ }
+
+ private static final class CatalogLoadException extends RuntimeException {
+ private CatalogLoadException(Exception cause) {
+ super(cause);
+ }
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCatalogFactoryTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCatalogFactoryTest.java
index c1c13f67e54..525f4855b9f 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCatalogFactoryTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCatalogFactoryTest.java
@@ -97,10 +97,17 @@ public class PaimonCatalogFactoryTest {
Map<String, String> defaults = props(
"paimon.catalog.type", "filesystem", "warehouse", "/wh");
+ PaimonCatalogProperties defaultProperties =
PaimonCatalogProperties.of(defaults);
+
Assertions.assertTrue(PaimonCatalogFactory.isCatalogCacheEnabled(defaultProperties),
+ "Paimon caching remains enabled by default through the
Doris-owned wrapper");
+
+ // The SDK wrapper itself is always disabled; governance changes only
whether mutable SDK
+ // child caches can be attached to the Doris-governed table entry.
try (PaimonConnector ungoverned = new PaimonConnector(defaults, new
RecordingConnectorContext())) {
Options options = ungoverned.buildCatalogOptions();
-
Assertions.assertFalse(options.contains(CatalogOptions.CACHE_ENABLED),
- "without a global/catalog total, Doris must preserve the
Paimon SDK default");
+
Assertions.assertTrue(options.contains(CatalogOptions.CACHE_ENABLED));
+ Assertions.assertFalse(options.get(CatalogOptions.CACHE_ENABLED),
+ "the Paimon SDK CachingCatalog must be disabled even
without a Doris weight limit");
}
Map<String, String> catalogLimited = new HashMap<>(defaults);
@@ -109,25 +116,29 @@ public class PaimonCatalogFactoryTest {
Options options = governed.buildCatalogOptions();
Assertions.assertTrue(options.contains(CatalogOptions.CACHE_ENABLED));
Assertions.assertFalse(options.get(CatalogOptions.CACHE_ENABLED),
- "an enclosing Doris hard limit must not be bypassed by an
unaccounted SDK cache");
+ "the Paimon SDK CachingCatalog must stay disabled under an
enclosing Doris hard limit");
}
Map<String, String> entryLimited = new HashMap<>(defaults);
entryLimited.put("meta.cache.paimon.partition_view.max-weight", "1MB");
try (PaimonConnector entryOnly = new PaimonConnector(entryLimited, new
RecordingConnectorContext())) {
-
Assertions.assertFalse(entryOnly.buildCatalogOptions().contains(CatalogOptions.CACHE_ENABLED),
- "an entry-only limit must preserve the Paimon SDK
default");
+
Assertions.assertTrue(entryOnly.buildCatalogOptions().contains(CatalogOptions.CACHE_ENABLED));
+
Assertions.assertFalse(entryOnly.buildCatalogOptions().get(CatalogOptions.CACHE_ENABLED),
+ "the Paimon SDK CachingCatalog must be disabled
unconditionally");
}
+ // The user flag controls whether the Doris wrapper is created, while
the SDK flag remains off.
PaimonCatalogProperties explicitTrue =
PaimonCatalogProperties.of(props(
"paimon.catalog.type", "filesystem", "warehouse", "/wh",
"paimon.cache-enabled", "true"));
-
Assertions.assertTrue(PaimonCatalogFactory.buildCatalogOptions(explicitTrue,
true)
- .get(CatalogOptions.CACHE_ENABLED), "an explicit Paimon
setting must win");
+
Assertions.assertTrue(PaimonCatalogFactory.isCatalogCacheEnabled(explicitTrue));
+
Assertions.assertFalse(PaimonCatalogFactory.buildCatalogOptions(explicitTrue)
+ .get(CatalogOptions.CACHE_ENABLED));
PaimonCatalogProperties explicitFalse =
PaimonCatalogProperties.of(props(
"paimon.catalog.type", "filesystem", "warehouse", "/wh",
"paimon.cache-enabled", "false"));
-
Assertions.assertFalse(PaimonCatalogFactory.buildCatalogOptions(explicitFalse,
true)
- .get(CatalogOptions.CACHE_ENABLED), "an explicit Paimon
setting must win");
+
Assertions.assertFalse(PaimonCatalogFactory.isCatalogCacheEnabled(explicitFalse));
+
Assertions.assertFalse(PaimonCatalogFactory.buildCatalogOptions(explicitFalse)
+ .get(CatalogOptions.CACHE_ENABLED));
}
@Test
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCatalogOptionsSnapshotTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCatalogOptionsSnapshotTest.java
index 027b37e9e9e..9e4910002fa 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCatalogOptionsSnapshotTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCatalogOptionsSnapshotTest.java
@@ -85,7 +85,10 @@ public class PaimonCatalogOptionsSnapshotTest {
props("metastore", "filesystem",
"catalog.type", "filesystem",
"warehouse", "/wh",
- "read.batch-size", "4096"),
+ "read.batch-size", "4096",
+ // Doris owns the paimon Table cache
(PaimonTableCache); the SDK CachingCatalog is
+ // always disabled so frozen Tables cannot outlive a
REFRESH (DORIS-29032).
+ "cache-enabled", "false"),
withExcludedNamespaces(props(
"paimon.catalog.type", "filesystem",
"warehouse", "/wh",
@@ -99,7 +102,8 @@ public class PaimonCatalogOptionsSnapshotTest {
public void defaultFlavorSnapshotIsFilesystem() {
assertOptions(
props("metastore", "filesystem",
- "warehouse", "/wh"),
+ "warehouse", "/wh",
+ "cache-enabled", "false"),
props("warehouse", "/wh"));
}
@@ -113,7 +117,8 @@ public class PaimonCatalogOptionsSnapshotTest {
// Both are emitted unconditionally, at their legacy
defaults when unset.
"client-pool-cache.eviction-interval-ms", "300000",
"location-in-properties", "false",
- "read.batch-size", "4096"),
+ "read.batch-size", "4096",
+ "cache-enabled", "false"),
withExcludedNamespaces(props(
"paimon.catalog.type", "hms",
"warehouse", "/wh",
@@ -132,7 +137,8 @@ public class PaimonCatalogOptionsSnapshotTest {
"warehouse", "/wh",
"uri", "thrift://alias:9083",
"client-pool-cache.eviction-interval-ms", "60000",
- "location-in-properties", "true"),
+ "location-in-properties", "true",
+ "cache-enabled", "false"),
props("paimon.catalog.type", "hms",
"warehouse", "/wh",
"uri", "thrift://alias:9083",
@@ -157,7 +163,8 @@ public class PaimonCatalogOptionsSnapshotTest {
"warehouse", "/wh",
"uri", "thrift://nn:9083",
"client-pool-cache.eviction-interval-ms", "300000",
- "location-in-properties", "false"),
+ "location-in-properties", "false",
+ "cache-enabled", "false"),
props("paimon.catalog.type", "hms",
"warehouse", " /wh ",
"hive.metastore.uris", " thrift://nn:9083 "));
@@ -181,7 +188,8 @@ public class PaimonCatalogOptionsSnapshotTest {
"rest.uri", "http://rest:8080";,
"rest.token.provider", "dlf",
"rest.dlf.access-key-id", "ak",
- "read.batch-size", "4096"),
+ "read.batch-size", "4096",
+ "cache-enabled", "false"),
withExcludedNamespaces(props(
"paimon.catalog.type", "rest",
"warehouse", "/wh",
@@ -198,7 +206,8 @@ public class PaimonCatalogOptionsSnapshotTest {
props("metastore", "rest",
"catalog.type", "rest",
"warehouse", "/wh",
- "uri", "http://rest:8080";),
+ "uri", "http://rest:8080";,
+ "cache-enabled", "false"),
props("paimon.catalog.type", "rest",
"warehouse", "/wh",
"uri", "http://rest:8080";));
@@ -220,7 +229,8 @@ public class PaimonCatalogOptionsSnapshotTest {
"catalog.type", "rest",
"warehouse", "/wh",
"uri", "http://rest:8080";,
- "rest.uri", " http://rest:8080 "),
+ "rest.uri", " http://rest:8080 ",
+ "cache-enabled", "false"),
props("paimon.catalog.type", "rest",
"warehouse", " /wh ",
"paimon.rest.uri", " http://rest:8080 "));
@@ -244,7 +254,8 @@ public class PaimonCatalogOptionsSnapshotTest {
"jdbc.driver_url", "mysql.jar",
"jdbc.driver_class", "com.mysql.cj.jdbc.Driver",
"jdbc.foo", "bar",
- "read.batch-size", "4096"),
+ "read.batch-size", "4096",
+ "cache-enabled", "false"),
withExcludedNamespaces(props(
"paimon.catalog.type", "jdbc",
"warehouse", "/wh",
@@ -265,7 +276,8 @@ public class PaimonCatalogOptionsSnapshotTest {
"catalog.type", "jdbc",
"warehouse", "/wh",
"uri", "jdbc:mysql://db:3306/meta",
- "jdbc.uri", "jdbc:mysql://db:3306/meta"),
+ "jdbc.uri", "jdbc:mysql://db:3306/meta",
+ "cache-enabled", "false"),
props("paimon.catalog.type", "jdbc",
"warehouse", "/wh",
"paimon.jdbc.uri", "jdbc:mysql://db:3306/meta"));
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalogTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalogTest.java
new file mode 100644
index 00000000000..d7c4a8ae7e3
--- /dev/null
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalogTest.java
@@ -0,0 +1,679 @@
+// 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.MetaCache;
+import org.apache.doris.connector.cache.MetaCacheBudgetManager;
+import org.apache.doris.connector.cache.MetaCacheDefinition;
+import org.apache.doris.connector.cache.ScopePath;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.Database;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.privilege.PrivilegeChecker;
+import org.apache.paimon.privilege.PrivilegeManager;
+import org.apache.paimon.privilege.PrivilegedCatalog;
+import org.apache.paimon.privilege.PrivilegedFileStoreTable;
+import org.apache.paimon.rest.RESTCatalog;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.system.AllTableOptionsTable;
+import org.apache.paimon.types.DataTypes;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.lang.reflect.Proxy;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+
+class PaimonMetaCacheCatalogTest {
+ private static final Identifier TABLE = Identifier.create("db", "t");
+
+ @Test
+ void disabledCacheOnlyKeepsTheInvalidationDecorator() throws Exception {
+ RecordingCatalog recording = new RecordingCatalog();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ MetaCache<Identifier, String> sibling =
owner.create(MetaCacheDefinition
+ .<Identifier, String>builder("paimon-sibling",
CacheSpec.of(true, -1, 10),
+ id -> ScopePath.table(id.getDatabaseName(),
id.getTableName()))
+ .build());
+ Catalog result =
PaimonMetaCacheCatalog.tryToCreate(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ZERO, Duration.ZERO),
+ false, false);
+
+ Assertions.assertNotSame(recording.catalog(), result);
+ Assertions.assertNotSame(result.getTable(TABLE),
result.getTable(TABLE));
+ Assertions.assertEquals(2, recording.tableLoads.get());
+ sibling.put(TABLE, "stale");
+ result.dropTable(TABLE, true);
+ Assertions.assertNull(sibling.getIfPresent(TABLE));
+ }
+ }
+
+ @Test
+ void failedDecorationRollsBackCacheRegistrationsAndClosesCatalog() {
+ RecordingCatalog failed = new RecordingCatalog();
+ RecordingCatalog retry = new RecordingCatalog();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ Assertions.assertThrows(IllegalStateException.class,
+ () -> PaimonMetaCacheCatalog.tryToCreate(failed.catalog(),
owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ true, false, ignored -> {
+ throw new IllegalStateException("privilege
metadata is temporarily unavailable");
+ }));
+
+ Assertions.assertTrue(owner.entries().isEmpty());
+ Assertions.assertEquals(1, failed.closeCalls.get());
+ Assertions.assertDoesNotThrow(() ->
PaimonMetaCacheCatalog.tryToCreate(retry.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ true, false, catalog -> catalog));
+ Assertions.assertEquals(2, owner.entries().size());
+ }
+ }
+
+ @Test
+ void disabledCacheDoesNotParseUnusedSettings() {
+ Options options = new Options();
+ options.set(CatalogOptions.CACHE_EXPIRE_AFTER_ACCESS.key(),
"invalid-duration");
+ options.set(CatalogOptions.CACHE_EXPIRE_AFTER_WRITE.key(),
"invalid-duration");
+ options.set(CatalogOptions.CACHE_SNAPSHOT_MAX_NUM_PER_TABLE.key(),
"invalid-integer");
+ options.set(CatalogOptions.CACHE_MANIFEST_SMALL_FILE_MEMORY.key(),
"invalid-memory");
+ RecordingCatalog recording = new RecordingCatalog();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ Assertions.assertDoesNotThrow(() ->
PaimonMetaCacheCatalog.tryToCreate(
+ recording.catalog(), owner, 100, 100, options,
+ false, false, catalog -> catalog));
+ }
+ }
+
+ @Test
+ void enclosingWeightLimitDoesNotParseUnusedSdkCacheSettings() {
+ Options options = cacheOptions(Duration.ofDays(1), Duration.ofDays(1));
+ options.set(CatalogOptions.CACHE_SNAPSHOT_MAX_NUM_PER_TABLE.key(),
"invalid-integer");
+ options.set(CatalogOptions.CACHE_MANIFEST_SMALL_FILE_MEMORY.key(),
"invalid-memory");
+ RecordingCatalog recording = new RecordingCatalog();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ Assertions.assertDoesNotThrow(() -> new PaimonMetaCacheCatalog(
+ recording.catalog(), owner, 100, 100, options, true,
System::nanoTime));
+ }
+ }
+
+ @Test
+ void tableAndDatabaseHonorAccessAndWriteExpiry() throws Exception {
+ AtomicLong clock = new AtomicLong();
+ RecordingCatalog accessRecording = new RecordingCatalog();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(accessRecording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofSeconds(5),
Duration.ofSeconds(100)),
+ false, clock::get);
+
+ Assertions.assertSame(catalog.getTable(TABLE),
catalog.getTable(TABLE));
+ Assertions.assertSame(catalog.getDatabase("db"),
catalog.getDatabase("db"));
+ clock.set(Duration.ofSeconds(6).toNanos());
+ catalog.getTable(TABLE);
+ catalog.getDatabase("db");
+
+ Assertions.assertEquals(2, accessRecording.tableLoads.get());
+ Assertions.assertEquals(2, accessRecording.databaseLoads.get());
+ }
+
+ clock.set(0);
+ RecordingCatalog writeRecording = new RecordingCatalog();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(writeRecording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofSeconds(100),
Duration.ofSeconds(10)),
+ false, clock::get);
+
+ Table first = catalog.getTable(TABLE);
+ clock.set(Duration.ofSeconds(5).toNanos());
+ Assertions.assertSame(first, catalog.getTable(TABLE));
+ clock.set(Duration.ofSeconds(11).toNanos());
+ Assertions.assertNotSame(first, catalog.getTable(TABLE));
+ Assertions.assertEquals(2, writeRecording.tableLoads.get());
+ }
+ }
+
+ @Test
+ void
tableMissRaceDoesNotReturnAValueThatExpiredBeforePublicationWasObserved()
throws Exception {
+ AtomicLong clock = new AtomicLong();
+ AtomicInteger misses = new AtomicInteger();
+ CountDownLatch firstMiss = new CountDownLatch(1);
+ CountDownLatch releaseFirstMiss = new CountDownLatch(1);
+ RecordingCatalog recording = new RecordingCatalog();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofSeconds(1),
Duration.ofDays(1)),
+ false, clock::get, (cache, key) -> {
+ if (cache.equals("table") && misses.incrementAndGet()
== 1) {
+ firstMiss.countDown();
+ await(releaseFirstMiss);
+ }
+ });
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<Table> racing = executor.submit(() ->
catalog.getTable(TABLE));
+ Assertions.assertTrue(firstMiss.await(10, TimeUnit.SECONDS));
+ Table expired = catalog.getTable(TABLE);
+ clock.set(Duration.ofSeconds(2).toNanos());
+ releaseFirstMiss.countDown();
+
+ Assertions.assertNotSame(expired, racing.get(10,
TimeUnit.SECONDS));
+ Assertions.assertEquals(2, recording.tableLoads.get());
+ } finally {
+ releaseFirstMiss.countDown();
+ executor.shutdownNow();
+ }
+ }
+ }
+
+ @Test
+ void
databaseMissRaceDoesNotReturnAValueThatExpiredBeforePublicationWasObserved()
throws Exception {
+ AtomicLong clock = new AtomicLong();
+ AtomicInteger misses = new AtomicInteger();
+ CountDownLatch firstMiss = new CountDownLatch(1);
+ CountDownLatch releaseFirstMiss = new CountDownLatch(1);
+ RecordingCatalog recording = new RecordingCatalog();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofSeconds(1),
Duration.ofDays(1)),
+ false, clock::get, (cache, key) -> {
+ if (cache.equals("database") &&
misses.incrementAndGet() == 1) {
+ firstMiss.countDown();
+ await(releaseFirstMiss);
+ }
+ });
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<Database> racing = executor.submit(() ->
catalog.getDatabase("db"));
+ Assertions.assertTrue(firstMiss.await(10, TimeUnit.SECONDS));
+ Database expired = catalog.getDatabase("db");
+ clock.set(Duration.ofSeconds(2).toNanos());
+ releaseFirstMiss.countDown();
+
+ Assertions.assertNotSame(expired, racing.get(10,
TimeUnit.SECONDS));
+ Assertions.assertEquals(2, recording.databaseLoads.get());
+ } finally {
+ releaseFirstMiss.countDown();
+ executor.shutdownNow();
+ }
+ }
+ }
+
+ @Test
+ void acceptedLargeDurationsSaturateInsteadOfFailingCatalogCreation()
throws Exception {
+ RecordingCatalog recording = new RecordingCatalog();
+ Options options = new Options();
+ options.set(CatalogOptions.CACHE_EXPIRE_AFTER_ACCESS.key(),
"9223372037s");
+ options.set(CatalogOptions.CACHE_EXPIRE_AFTER_WRITE.key(),
"9223372037s");
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, Long.MAX_VALUE, options, false, System::nanoTime);
+
+ Assertions.assertSame(catalog.getTable(TABLE),
catalog.getTable(TABLE));
+ Assertions.assertEquals(1, recording.tableLoads.get());
+ }
+ }
+
+ @Test
+ void lastAccessTimeNeverMovesBackward() throws Exception {
+ AtomicLong clock = new AtomicLong();
+ RecordingCatalog recording = new RecordingCatalog();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofSeconds(10),
Duration.ofSeconds(100)),
+ false, clock::get);
+
+ Table first = catalog.getTable(TABLE);
+ clock.set(Duration.ofSeconds(5).toNanos());
+ catalog.getTable(TABLE);
+ clock.set(Duration.ofSeconds(1).toNanos());
+ catalog.getTable(TABLE);
+ clock.set(Duration.ofSeconds(12).toNanos());
+
+ Assertions.assertSame(first, catalog.getTable(TABLE));
+ Assertions.assertEquals(1, recording.tableLoads.get());
+ }
+ }
+
+ @Test
+ void baseTableInvalidationEvictsEveryBranchVariant() throws Exception {
+ AtomicLong clock = new AtomicLong();
+ RecordingCatalog recording = new RecordingCatalog();
+ Identifier branch = new Identifier("db", "t", "dev", null);
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ false, clock::get);
+
+ Table mainBefore = catalog.getTable(TABLE);
+ Table branchBefore = catalog.getTable(branch);
+ owner.invalidateTable("db", "t");
+
+ Assertions.assertNotSame(mainBefore, catalog.getTable(TABLE));
+ Assertions.assertNotSame(branchBefore, catalog.getTable(branch));
+ Assertions.assertEquals(4, recording.tableLoads.get());
+ }
+ }
+
+ @Test
+ void systemTableIsRebuiltFromTheCachedOriginTable() throws Exception {
+ AtomicLong clock = new AtomicLong();
+ RecordingCatalog recording = new RecordingCatalog();
+ recording.fileStoreTables = true;
+ Identifier systemTable = Identifier.create("db", "t$snapshots");
+ Assertions.assertTrue(systemTable.isSystemTable());
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ true, clock::get);
+
+ Assertions.assertNotNull(catalog.getTable(systemTable));
+ Assertions.assertEquals(TABLE, recording.lastLoadedTable.get());
+ Assertions.assertEquals(1, recording.tableLoads.get());
+ }
+ }
+
+ @Test
+ void nonFileStoreSystemTableIsDelegatedToTheWrappedCatalog() throws
Exception {
+ AtomicLong clock = new AtomicLong();
+ RecordingCatalog recording = new RecordingCatalog();
+ Identifier systemTable = Identifier.create("db", "t$snapshots");
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ false, clock::get);
+
+ Assertions.assertNotNull(catalog.getTable(systemTable));
+ Assertions.assertEquals(systemTable,
recording.lastLoadedTable.get());
+ Assertions.assertEquals(2, recording.tableLoads.get());
+ }
+ }
+
+ @Test
+ void eachSuccessfulDropEvictsBeforeALaterDropFails() throws Exception {
+ AtomicLong clock = new AtomicLong();
+ RecordingCatalog recording = new RecordingCatalog();
+ Identifier first = Identifier.create("db", "first");
+ Identifier second = Identifier.create("db", "second");
+ recording.failDrop.set(second);
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ false, clock::get);
+
+ Table staleFirst = catalog.getTable(first);
+ catalog.getTable(second);
+ catalog.dropTable(first, true);
+ Assertions.assertThrows(Catalog.TableNotExistException.class,
+ () -> catalog.dropTable(second, true));
+
+ Assertions.assertNotSame(staleFirst, catalog.getTable(first));
+ }
+ }
+
+ @Test
+ void mutationEvictsWhenTheDelegateThrowsAfterTheRemoteChange() throws
Exception {
+ AtomicLong clock = new AtomicLong();
+ RecordingCatalog recording = new RecordingCatalog();
+ recording.failAfterDrop.set(TABLE);
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ false, clock::get);
+
+ Table stale = catalog.getTable(TABLE);
+ Assertions.assertThrows(IllegalStateException.class, () ->
catalog.dropTable(TABLE, true));
+
+ Assertions.assertNotSame(stale, catalog.getTable(TABLE));
+ Assertions.assertEquals(2, recording.tableLoads.get());
+ }
+ }
+
+ @Test
+ void enclosingWeightLimitDoesNotAttachMutableSdkCaches() throws Exception {
+ AtomicLong clock = new AtomicLong();
+ RecordingCatalog governedRecording = new RecordingCatalog();
+ governedRecording.fileStoreTables = true;
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog governed = new
PaimonMetaCacheCatalog(governedRecording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ true, clock::get);
+ governed.getTable(TABLE);
+ Assertions.assertEquals(0,
governedRecording.sdkCacheAttachments.get());
+ }
+
+ RecordingCatalog ungovernedRecording = new RecordingCatalog();
+ ungovernedRecording.fileStoreTables = true;
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog ungoverned = new
PaimonMetaCacheCatalog(ungovernedRecording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ false, clock::get);
+ ungoverned.getTable(TABLE);
+ Assertions.assertEquals(3,
ungovernedRecording.sdkCacheAttachments.get());
+ }
+ }
+
+ @Test
+ void realFileStoreTableIsAdmittedByTheWeightGovernedCache(@TempDir
java.nio.file.Path warehouse)
+ throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ org.apache.paimon.fs.Path tablePath = new org.apache.paimon.fs.Path(
+ warehouse.resolve("weighted-table").toUri());
+ Schema schema = Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("payload", DataTypes.STRING())
+ .option("file.format", "parquet")
+ .build();
+ new SchemaManager(fileIO, tablePath).createTable(schema);
+
+ RecordingCatalog recording = new RecordingCatalog();
+ recording.tableSupplier = () -> FileStoreTableFactory.create(fileIO,
tablePath);
+ MetaCacheBudgetManager budgetManager = new
MetaCacheBudgetManager(OptionalLong.of(1024L * 1024L));
+ try (CatalogMetaCache owner = new CatalogMetaCache(
+ budgetManager, 67996L, "paimon", Collections.emptyMap())) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ true, System::nanoTime);
+
+ Table first = catalog.getTable(TABLE);
+ Assertions.assertSame(first, catalog.getTable(TABLE));
+ Assertions.assertEquals(1, recording.tableLoads.get());
+ Assertions.assertTrue(budgetManager.getGlobalUsedWeight() > 0L);
+ }
+ Assertions.assertEquals(0L, budgetManager.getGlobalUsedWeight());
+ }
+
+ @Test
+ void allTableOptionsIsNotAdmittedWithoutACompleteRetainedSizeEstimate()
throws Exception {
+ Map<Identifier, Map<String, String>> allOptions = new HashMap<>();
+ for (int i = 0; i < 100; i++) {
+ allOptions.put(Identifier.create("db", "table_" + i),
+ Collections.singletonMap("large-option", "x".repeat(100)));
+ }
+ RecordingCatalog recording = new RecordingCatalog();
+ recording.tableSupplier = () -> new AllTableOptionsTable(allOptions);
+ MetaCacheBudgetManager budgetManager = new
MetaCacheBudgetManager(OptionalLong.of(512L));
+ try (CatalogMetaCache owner = new CatalogMetaCache(
+ budgetManager, 67996L, "paimon", Collections.emptyMap())) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ true, System::nanoTime);
+
+ catalog.getTable(Identifier.create("sys",
AllTableOptionsTable.ALL_TABLE_OPTIONS));
+ catalog.getTable(Identifier.create("sys",
AllTableOptionsTable.ALL_TABLE_OPTIONS));
+ Assertions.assertEquals(2, recording.tableLoads.get());
+ Assertions.assertEquals(0L, budgetManager.getGlobalUsedWeight());
+ }
+ }
+
+ @Test
+ void fallbackBranchesAreIncludedInWeightGovernedAdmission(@TempDir
java.nio.file.Path warehouse)
+ throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ FileStoreTable main = createFileStoreTable(fileIO,
warehouse.resolve("main"), "main_payload");
+ FileStoreTable fallback = createFileStoreTable(fileIO,
warehouse.resolve("fallback"), "fallback_payload");
+ FileStoreTable decorated = new FallbackReadFileStoreTable(main,
fallback);
+ long mainWeight = PaimonCacheSizeEstimator.estimateTable(
+ TABLE, main,
PaimonMetaCacheCatalog.TABLE_ENTRY_OVERHEAD_BYTES).getBytes();
+ long decoratedWeight = PaimonCacheSizeEstimator.estimateTable(
+ TABLE, decorated,
PaimonMetaCacheCatalog.TABLE_ENTRY_OVERHEAD_BYTES).getBytes();
+ long sharedBranchWeight = PaimonCacheSizeEstimator.estimateTable(
+ TABLE, new FallbackReadFileStoreTable(main, main),
+ PaimonMetaCacheCatalog.TABLE_ENTRY_OVERHEAD_BYTES).getBytes();
+
+ Assertions.assertTrue(decoratedWeight > mainWeight);
+ Assertions.assertTrue(decoratedWeight > sharedBranchWeight,
+ "the same branch object must be counted once by identity");
+ Assertions.assertFalse(PaimonCacheSizeEstimator.estimateTable(TABLE,
+ PrivilegedFileStoreTable.wrap(decorated,
privilegeChecker(true), TABLE),
+
PaimonMetaCacheCatalog.TABLE_ENTRY_OVERHEAD_BYTES).isComplete(),
+ "an authorization snapshot must never be admitted to the raw
metadata cache");
+
+ RecordingCatalog recording = new RecordingCatalog();
+ recording.tableSupplier = () -> new FallbackReadFileStoreTable(main,
fallback);
+ MetaCacheBudgetManager budgetManager = new MetaCacheBudgetManager(
+ OptionalLong.of(mainWeight));
+ try (CatalogMetaCache owner = new CatalogMetaCache(
+ budgetManager, 67996L, "paimon", Collections.emptyMap())) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ true, System::nanoTime);
+
+ Assertions.assertNotSame(catalog.getTable(TABLE),
catalog.getTable(TABLE));
+ Assertions.assertEquals(2, recording.tableLoads.get());
+ Assertions.assertEquals(0L, budgetManager.getGlobalUsedWeight());
+ }
+ }
+
+ @Test
+ void privilegeCheckerIsRefreshedOutsideTheRawTableCache() throws Exception
{
+ AtomicReference<Boolean> canSelect = new AtomicReference<>(true);
+ RecordingCatalog recording = new RecordingCatalog();
+ recording.fileStoreTables = true;
+ PrivilegeManager privilegeManager = (PrivilegeManager)
Proxy.newProxyInstance(
+ PrivilegeManager.class.getClassLoader(), new Class<?>[]
{PrivilegeManager.class},
+ (proxy, method, args) -> {
+ if (method.getName().equals("getPrivilegeChecker")) {
+ boolean snapshot = canSelect.get();
+ return privilegeChecker(snapshot);
+ }
+ return defaultValue(method.getReturnType());
+ });
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog cached = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ false, System::nanoTime);
+ Catalog privileged = new PrivilegedCatalog(cached, () ->
privilegeManager);
+
+ FileStoreTable beforeRevoke = (FileStoreTable)
privileged.getTable(TABLE);
+ Assertions.assertDoesNotThrow(beforeRevoke::newScan);
+ canSelect.set(false);
+ FileStoreTable afterRevoke = (FileStoreTable)
privileged.getTable(TABLE);
+
+ Assertions.assertNotSame(beforeRevoke, afterRevoke);
+ Assertions.assertThrows(IllegalStateException.class,
afterRevoke::newScan);
+ Assertions.assertEquals(1, recording.tableLoads.get(),
+ "revocation must refresh authorization without reloading
raw metadata");
+ }
+ }
+
+ @Test
+ void restDispatchSeesThroughTheMetaCacheWrapper() {
+ Options options = cacheOptions(Duration.ofDays(1), Duration.ofDays(1));
+ options.set("uri", "http://localhost:1";);
+ options.set("prefix", "test-prefix");
+ options.set("token.provider", "bear");
+ options.set("token", "test-token");
+ RESTCatalog rest = new RESTCatalog(CatalogContext.create(options),
false);
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ PaimonMetaCacheCatalog wrapped = new PaimonMetaCacheCatalog(rest,
owner,
+ 100, 100, options, true, System::nanoTime);
+
+ Assertions.assertSame(rest,
+
PaimonCatalogOps.CatalogBackedPaimonCatalogOps.restCatalog(wrapped));
+ }
+ }
+
+ private static Options cacheOptions(Duration access, Duration write) {
+ Options options = new Options();
+ options.set(CatalogOptions.CACHE_EXPIRE_AFTER_ACCESS, access);
+ options.set(CatalogOptions.CACHE_EXPIRE_AFTER_WRITE, write);
+ return options;
+ }
+
+ private static FileStoreTable createFileStoreTable(
+ LocalFileIO fileIO, java.nio.file.Path path, String payloadColumn)
throws Exception {
+ org.apache.paimon.fs.Path tablePath = new
org.apache.paimon.fs.Path(path.toUri());
+ Schema schema = Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(payloadColumn, DataTypes.STRING())
+ .option("file.format", "parquet")
+ .build();
+ new SchemaManager(fileIO, tablePath).createTable(schema);
+ return FileStoreTableFactory.create(fileIO, tablePath);
+ }
+
+ private static PrivilegeChecker privilegeChecker(boolean canSelect) {
+ return (PrivilegeChecker) Proxy.newProxyInstance(
+ PrivilegeChecker.class.getClassLoader(), new Class<?>[]
{PrivilegeChecker.class},
+ (proxy, method, args) -> {
+ if (method.getName().equals("assertCanSelect") &&
!canSelect) {
+ throw new IllegalStateException("SELECT privilege was
revoked");
+ }
+ return null;
+ });
+ }
+
+ private static Object defaultValue(Class<?> returnType) {
+ if (!returnType.isPrimitive()) {
+ if (returnType == Optional.class) {
+ return Optional.empty();
+ }
+ if (returnType == Map.class) {
+ return Collections.emptyMap();
+ }
+ return null;
+ }
+ if (returnType == boolean.class) {
+ return false;
+ }
+ if (returnType == char.class) {
+ return '\0';
+ }
+ if (returnType == byte.class) {
+ return (byte) 0;
+ }
+ if (returnType == short.class) {
+ return (short) 0;
+ }
+ if (returnType == int.class) {
+ return 0;
+ }
+ if (returnType == long.class) {
+ return 0L;
+ }
+ if (returnType == float.class) {
+ return 0F;
+ }
+ return 0D;
+ }
+
+ private static void await(CountDownLatch latch) {
+ try {
+ if (!latch.await(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("timed out waiting for cache
race");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("interrupted while waiting for
cache race", e);
+ }
+ }
+
+ private static final class RecordingCatalog {
+ private final AtomicInteger tableLoads = new AtomicInteger();
+ private final AtomicInteger databaseLoads = new AtomicInteger();
+ private final AtomicInteger sdkCacheAttachments = new AtomicInteger();
+ private final AtomicInteger closeCalls = new AtomicInteger();
+ private final AtomicReference<Identifier> lastLoadedTable = new
AtomicReference<>();
+ private final AtomicReference<Identifier> failDrop = new
AtomicReference<>();
+ private final AtomicReference<Identifier> failAfterDrop = new
AtomicReference<>();
+ private final Catalog catalog;
+ private boolean fileStoreTables;
+ private Supplier<Table> tableSupplier = this::newTable;
+
+ private RecordingCatalog() {
+ AtomicReference<Catalog> self = new AtomicReference<>();
+ this.catalog = (Catalog)
Proxy.newProxyInstance(Catalog.class.getClassLoader(),
+ new Class<?>[] {Catalog.class}, (proxy, method, args) -> {
+ switch (method.getName()) {
+ case "getTable":
+ Identifier identifier = (Identifier) args[0];
+ lastLoadedTable.set(identifier);
+ tableLoads.incrementAndGet();
+ return tableSupplier.get();
+ case "getDatabase":
+ databaseLoads.incrementAndGet();
+ return Database.of((String) args[0]);
+ case "dropTable":
+ if (args[0].equals(failDrop.get())) {
+ throw new
Catalog.TableNotExistException((Identifier) args[0]);
+ }
+ if (args[0].equals(failAfterDrop.get())) {
+ throw new
IllegalStateException("post-mutation cleanup failed");
+ }
+ return null;
+ case "catalogLoader":
+ return
(org.apache.paimon.catalog.CatalogLoader) self::get;
+ case "close":
+ closeCalls.incrementAndGet();
+ return null;
+ case "options":
+ return Collections.emptyMap();
+ case "toString":
+ return "RecordingCatalog";
+ default:
+ return defaultValue(method.getReturnType());
+ }
+ });
+ self.set(catalog);
+ }
+
+ private Catalog catalog() {
+ return catalog;
+ }
+
+ private Table newTable() {
+ Class<?> tableType = fileStoreTables ? FileStoreTable.class :
Table.class;
+ return (Table) Proxy.newProxyInstance(tableType.getClassLoader(),
new Class<?>[] {tableType},
+ (proxy, method, args) -> {
+ if (method.getName().startsWith("set") &&
method.getName().endsWith("Cache")) {
+ sdkCacheAttachments.incrementAndGet();
+ return null;
+ }
+ if (method.getName().equals("toString")) {
+ return "RecordingTable";
+ }
+ return defaultValue(method.getReturnType());
+ });
+ }
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonRestCatalogPartitionsTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonRestCatalogPartitionsTest.java
index cb035de0b5b..d94692f6e9c 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonRestCatalogPartitionsTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonRestCatalogPartitionsTest.java
@@ -17,8 +17,10 @@
package org.apache.doris.connector.paimon;
+import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.partition.Partition;
+import org.apache.paimon.rest.exceptions.ForbiddenException;
import org.apache.paimon.rest.exceptions.NotImplementedException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -59,4 +61,22 @@ class PaimonRestCatalogPartitionsTest {
Assertions.assertSame(manifestPartitions, result);
}
+
+ @Test
+ void forbiddenEndpointDoesNotFallBackToFilesystem() {
+ AtomicBoolean fallbackCalled = new AtomicBoolean();
+
+ Assertions.assertThrows(Catalog.TableNoPermissionException.class,
+ () -> PaimonRestCatalogPartitions.listPartitions(
+ identifier -> {
+ throw new ForbiddenException("forbidden");
+ },
+ TABLE,
+ () -> {
+ fallbackCalled.set(true);
+ return Collections.emptyList();
+ }));
+
+ Assertions.assertFalse(fallbackCalled.get());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
