github-actions[bot] commented on code in PR #67996: URL: https://github.com/apache/doris/pull/67996#discussion_r4055864624
########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalog.java: ########## @@ -0,0 +1,176 @@ +// 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.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.fs.Path; +import org.apache.paimon.options.CatalogOptions; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.options.Options; +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.utils.SegmentsCache; + +import java.time.Duration; +import java.util.Optional; + +/** + * Doris-owned replacement for the Paimon SDK {@code CachingCatalog}. Every catalog-level cache + * ({@code tableCache}, {@code databaseCache}) lives in Doris's {@link CatalogMetaCache} framework + * with a per-catalog scope, so {@code REFRESH TABLE}/{@code REFRESH DATABASE}/ + * {@code REFRESH CATALOG} invalidates them through the same registry path as every other + * connector-owned cache. The per-{@link FileStoreTable} caches ({@code snapshotCache}, + * {@code statsCache}, {@code manifestCache}) are built from the same {@link CatalogOptions} that + * {@code CachingCatalog} reads and attached on {@link #getTable}, preserving scan-time performance. + * + * <p><b>Why not the SDK's CachingCatalog?</b> Its {@code tableCache} freezes schema/snapshot + * state at load time and exposes only per-table {@code invalidateTable(Identifier)} — no + * db/catalog-level eviction. After an external same-name drop/recreate the stale frozen + * {@link Table} survives every Doris-side {@code REFRESH}. + */ +final class PaimonMetaCacheCatalog extends DelegateCatalog { + + private final MetaCache<Identifier, Table> tableCache; + private final MetaCache<String, Database> databaseCache; + private final SegmentsCache<Path> manifestCache; + private final Duration expireAfterAccess; + private final Duration expireAfterWrite; + private final int snapshotMaxNumPerTable; + + PaimonMetaCacheCatalog(Catalog wrapped, CatalogMetaCache metaCache, int tableCacheMaxSize, + long tableCacheTtlSecond, Options catalogOptions) { + super(wrapped); + CacheSpec tableSpec = CacheSpec.ofConnectorTtl(tableCacheTtlSecond, tableCacheMaxSize); + this.tableCache = metaCache.create(MetaCacheDefinition + .<Identifier, Table>builder("paimon-table", tableSpec, + id -> ScopePath.table(id.getDatabaseName(), id.getObjectName())) Review Comment: [P1] Scope branch and system variants under the base table `Identifier#getObjectName()` includes Paimon's branch/system suffixes (for example `t$branch_dev` and `t$snapshots`), but `invalidateTable(db, t)` invalidates only the exact base-table scope. Ordinary branch resolution and deserialized system-table handles both populate these variant keys, so `REFRESH TABLE` or a same-name drop/recreate can leave them serving the old table/schema generation. Use `id.getTableName()` for the semantic scope (and preserve the SDK's origin-table rebuild behavior for system tables), then test refresh after both variant paths have been cached. ########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java: ########## @@ -547,7 +546,9 @@ private Catalog createCatalogFromContext(CatalogContext catalogContext, String f ? createHmsCatalog(catalogContext, hmsAuth, catalogProps.getRaw(), storageHadoopConfig) : CatalogFactory.createCatalog(catalogContext); - return catalog; + return new PaimonMetaCacheCatalog(catalog, metaCache, Review Comment: [P1] Preserve REST partition dispatch through this wrapper This now wraps every catalog flavor in `PaimonMetaCacheCatalog`, so `CatalogBackedPaimonCatalogOps#listPartitions` can no longer satisfy its `catalog instanceof RESTCatalog` check. For REST catalogs that previously returned the raw catalog (for example `paimon.cache-enabled=false` or the enclosing-weight-limit path), partition listing therefore switches from `PaimonRestCatalogPartitions` to unconditional filesystem enumeration. That bypasses the REST server's authoritative partition set and permission errors. Please unwrap the delegate (or expose a flavor-neutral partition operation) before this dispatch and cover a wrapped REST catalog's success/forbidden/fallback paths. ########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalog.java: ########## @@ -0,0 +1,176 @@ +// 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.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.fs.Path; +import org.apache.paimon.options.CatalogOptions; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.options.Options; +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.utils.SegmentsCache; + +import java.time.Duration; +import java.util.Optional; + +/** + * Doris-owned replacement for the Paimon SDK {@code CachingCatalog}. Every catalog-level cache + * ({@code tableCache}, {@code databaseCache}) lives in Doris's {@link CatalogMetaCache} framework + * with a per-catalog scope, so {@code REFRESH TABLE}/{@code REFRESH DATABASE}/ + * {@code REFRESH CATALOG} invalidates them through the same registry path as every other + * connector-owned cache. The per-{@link FileStoreTable} caches ({@code snapshotCache}, + * {@code statsCache}, {@code manifestCache}) are built from the same {@link CatalogOptions} that + * {@code CachingCatalog} reads and attached on {@link #getTable}, preserving scan-time performance. + * + * <p><b>Why not the SDK's CachingCatalog?</b> Its {@code tableCache} freezes schema/snapshot + * state at load time and exposes only per-table {@code invalidateTable(Identifier)} — no + * db/catalog-level eviction. After an external same-name drop/recreate the stale frozen + * {@link Table} survives every Doris-side {@code REFRESH}. + */ +final class PaimonMetaCacheCatalog extends DelegateCatalog { + + private final MetaCache<Identifier, Table> tableCache; + private final MetaCache<String, Database> databaseCache; + private final SegmentsCache<Path> manifestCache; + private final Duration expireAfterAccess; + private final Duration expireAfterWrite; + private final int snapshotMaxNumPerTable; + + PaimonMetaCacheCatalog(Catalog wrapped, CatalogMetaCache metaCache, int tableCacheMaxSize, + long tableCacheTtlSecond, Options catalogOptions) { + super(wrapped); + CacheSpec tableSpec = CacheSpec.ofConnectorTtl(tableCacheTtlSecond, tableCacheMaxSize); + this.tableCache = metaCache.create(MetaCacheDefinition + .<Identifier, Table>builder("paimon-table", tableSpec, + id -> ScopePath.table(id.getDatabaseName(), id.getObjectName())) + .sizeEstimator(MetaCacheSizeEstimators.reflective()) + .build()); + CacheSpec dbSpec = CacheSpec.ofConnectorTtl(86400L, 100); + this.databaseCache = metaCache.create(MetaCacheDefinition + .<String, Database>builder("paimon-database", dbSpec, + ScopePath::database) + .sizeEstimator(MetaCacheSizeEstimators.reflective()) + .build()); + + this.manifestCache = buildManifestCache(catalogOptions); + this.expireAfterAccess = catalogOptions.get( + CatalogOptions.CACHE_EXPIRE_AFTER_ACCESS); + this.expireAfterWrite = catalogOptions.get( + CatalogOptions.CACHE_EXPIRE_AFTER_WRITE); + this.snapshotMaxNumPerTable = catalogOptions.get( + CatalogOptions.CACHE_SNAPSHOT_MAX_NUM_PER_TABLE); + } + + @Override + public Table getTable(Identifier identifier) throws TableNotExistException { + try { + return tableCache.get(identifier, ignored -> { + try { + return attachPerTableCaches(super.getTable(identifier)); + } catch (TableNotExistException e) { + throw new RuntimeException(e); + } + }); + } catch (RuntimeException e) { + if (e.getCause() instanceof TableNotExistException) { + throw (TableNotExistException) e.getCause(); + } + throw e; + } + } + + @Override + public Database getDatabase(String name) throws DatabaseNotExistException { + try { + return databaseCache.get(name, ignored -> { + try { + return super.getDatabase(name); + } catch (DatabaseNotExistException e) { + throw new RuntimeException(e); + } + }); + } catch (RuntimeException e) { + if (e.getCause() instanceof DatabaseNotExistException) { + throw (DatabaseNotExistException) e.getCause(); + } + throw e; + } + } + + @Override Review Comment: [P1] Evict each table after its remote drop commits `DROP DATABASE ... FORCE` drops tables sequentially, while FE calls `connector.invalidateDb` only if the whole operation returns successfully. Because this wrapper inherits `DelegateCatalog.dropTable`, a failure on a later table (or the final database drop) leaves earlier deletions committed remotely but their `paimon-table` entries still reachable for the TTL. The SDK `CachingCatalog` evicted inside each successful `dropTable`, which avoided this partial-failure stale state. Please restore mutation-local eviction in the wrapper and test a forced drop whose second remote table deletion fails. ########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalog.java: ########## @@ -0,0 +1,176 @@ +// 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.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.fs.Path; +import org.apache.paimon.options.CatalogOptions; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.options.Options; +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.utils.SegmentsCache; + +import java.time.Duration; +import java.util.Optional; + +/** + * Doris-owned replacement for the Paimon SDK {@code CachingCatalog}. Every catalog-level cache + * ({@code tableCache}, {@code databaseCache}) lives in Doris's {@link CatalogMetaCache} framework + * with a per-catalog scope, so {@code REFRESH TABLE}/{@code REFRESH DATABASE}/ + * {@code REFRESH CATALOG} invalidates them through the same registry path as every other + * connector-owned cache. The per-{@link FileStoreTable} caches ({@code snapshotCache}, + * {@code statsCache}, {@code manifestCache}) are built from the same {@link CatalogOptions} that + * {@code CachingCatalog} reads and attached on {@link #getTable}, preserving scan-time performance. + * + * <p><b>Why not the SDK's CachingCatalog?</b> Its {@code tableCache} freezes schema/snapshot + * state at load time and exposes only per-table {@code invalidateTable(Identifier)} — no + * db/catalog-level eviction. After an external same-name drop/recreate the stale frozen + * {@link Table} survives every Doris-side {@code REFRESH}. + */ +final class PaimonMetaCacheCatalog extends DelegateCatalog { + + private final MetaCache<Identifier, Table> tableCache; + private final MetaCache<String, Database> databaseCache; + private final SegmentsCache<Path> manifestCache; + private final Duration expireAfterAccess; + private final Duration expireAfterWrite; + private final int snapshotMaxNumPerTable; + + PaimonMetaCacheCatalog(Catalog wrapped, CatalogMetaCache metaCache, int tableCacheMaxSize, + long tableCacheTtlSecond, Options catalogOptions) { + super(wrapped); + CacheSpec tableSpec = CacheSpec.ofConnectorTtl(tableCacheTtlSecond, tableCacheMaxSize); Review Comment: [P1] Preserve the existing Paimon cache controls The replacement always creates these caches even when a persisted catalog explicitly has `paimon.cache-enabled=false`; before this change that setting returned the raw catalog with no SDK caches. It also stops applying `paimon.cache.expire-after-access/write` to the cached `Table` and `Database`: a catalog configured for a 1-second bound can now retain those objects for the 24-hour Doris default, while the Paimon durations affect only nested snapshot/stats caches. Please carry the original enable flag and expiry semantics into the replacement (separately from the forced SDK flag), and add behavioral tests for disabled and short-expiry catalogs. ########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalog.java: ########## @@ -0,0 +1,176 @@ +// 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.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.fs.Path; +import org.apache.paimon.options.CatalogOptions; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.options.Options; +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.utils.SegmentsCache; + +import java.time.Duration; +import java.util.Optional; + +/** + * Doris-owned replacement for the Paimon SDK {@code CachingCatalog}. Every catalog-level cache + * ({@code tableCache}, {@code databaseCache}) lives in Doris's {@link CatalogMetaCache} framework + * with a per-catalog scope, so {@code REFRESH TABLE}/{@code REFRESH DATABASE}/ + * {@code REFRESH CATALOG} invalidates them through the same registry path as every other + * connector-owned cache. The per-{@link FileStoreTable} caches ({@code snapshotCache}, + * {@code statsCache}, {@code manifestCache}) are built from the same {@link CatalogOptions} that + * {@code CachingCatalog} reads and attached on {@link #getTable}, preserving scan-time performance. + * + * <p><b>Why not the SDK's CachingCatalog?</b> Its {@code tableCache} freezes schema/snapshot + * state at load time and exposes only per-table {@code invalidateTable(Identifier)} — no + * db/catalog-level eviction. After an external same-name drop/recreate the stale frozen + * {@link Table} survives every Doris-side {@code REFRESH}. + */ +final class PaimonMetaCacheCatalog extends DelegateCatalog { + + private final MetaCache<Identifier, Table> tableCache; + private final MetaCache<String, Database> databaseCache; + private final SegmentsCache<Path> manifestCache; + private final Duration expireAfterAccess; + private final Duration expireAfterWrite; + private final int snapshotMaxNumPerTable; + + PaimonMetaCacheCatalog(Catalog wrapped, CatalogMetaCache metaCache, int tableCacheMaxSize, + long tableCacheTtlSecond, Options catalogOptions) { + super(wrapped); + CacheSpec tableSpec = CacheSpec.ofConnectorTtl(tableCacheTtlSecond, tableCacheMaxSize); + this.tableCache = metaCache.create(MetaCacheDefinition + .<Identifier, Table>builder("paimon-table", tableSpec, + id -> ScopePath.table(id.getDatabaseName(), id.getObjectName())) + .sizeEstimator(MetaCacheSizeEstimators.reflective()) + .build()); + CacheSpec dbSpec = CacheSpec.ofConnectorTtl(86400L, 100); + this.databaseCache = metaCache.create(MetaCacheDefinition + .<String, Database>builder("paimon-database", dbSpec, + ScopePath::database) + .sizeEstimator(MetaCacheSizeEstimators.reflective()) + .build()); + + this.manifestCache = buildManifestCache(catalogOptions); Review Comment: [P1] Keep the SDK caches inside the configured weight budget Under `meta.cache.max-weight`, the managed table entry is weighed only once when it is published. The snapshot/stats caches attached later can grow without reweighing, and this shared `SegmentsCache` is retained directly by the wrapper, outside `CatalogMetaCache` entirely (the SDK default can retain 128 MiB). The old weight-limited path disabled these SDK caches by default, so the replacement can now exceed the catalog/global hard limit. Please either account for their live growth in the Doris budget or disable them whenever an enclosing weight limit is active, with a fill-to-limit test. -- 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]
