sundapeng commented on code in PR #8728: URL: https://github.com/apache/paimon/pull/8728#discussion_r3610937571
########## paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCatalogProvider.java: ########## @@ -0,0 +1,191 @@ +/* + * 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.paimon.table.format; + +import org.apache.paimon.PagedList; +import org.apache.paimon.annotation.Experimental; +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.CatalogLoader; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.partition.Partition; +import org.apache.paimon.utils.StringUtils; + +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache; +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine; + +import javax.annotation.Nullable; + +import java.io.Serializable; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicLong; + +/** Serializable catalog access for a managed format table. */ +@Experimental +public class FormatTableCatalogProvider implements Serializable { + + private static final long serialVersionUID = 1L; + + private static final int PARTITION_PAGE_SIZE = 1000; + private static final int MAX_CACHED_PATTERNS = 128; + private static final int MAX_TRACKED_TABLE_GENERATIONS = 10_000; + private static final Duration PARTITION_CACHE_TTL = Duration.ofSeconds(30); + private static final Duration TABLE_GENERATION_TTL = Duration.ofHours(1); + + // Keyed by identifier full name: listings themselves live in per-instance caches, so sharing + // a generation counter across table incarnations (or across catalogs using the same name) can + // only cause an extra invalidation, never a stale read. + private static final Cache<String, AtomicLong> GENERATIONS = + Caffeine.newBuilder() + .expireAfterAccess(TABLE_GENERATION_TTL) + .maximumSize(MAX_TRACKED_TABLE_GENERATIONS) + .executor(Runnable::run) + .build(); + + private final Identifier identifier; + private final CatalogLoader catalogLoader; + + @Nullable private transient Cache<String, List<Partition>> partitionCache; + // Reused across list/create calls: constructing a Catalog (HTTP client, auth provider, and + // for some auth providers an initial token fetch) per partition operation is expensive, and + // RESTCatalog.close() is a no-op so a long-lived instance leaks nothing. Recreated lazily + // after deserialization. + @Nullable private transient Catalog catalog; Review Comment: 🟢 **[suggestion] architecture — Catalog 实例无生命周期管理** FormatTableCatalogProvider 通过 catalogLoader.load() 懒创建 Catalog 实例并无限期持有。注释以 "RESTCatalog.close() is a no-op" 为由不关闭它。然而 CatalogLoader 是通用的 Serializable 工厂，可以产生任何 Catalog 实现。如果未来的 catalog（或包装的 CachingCatalog/PrivilegedCatalog）持有连接池、线程池或文件句柄，provider 将静默泄漏资源。 该类实现了 Serializable 但未实现 AutoCloseable/Closeable，调用者没有生命周期钩子来释放资源。 **建议**: 实现 AutoCloseable 并委托给 catalog.close()（使生命周期显式化），或在 CatalogLoader 上文档化约定：加载的 catalog 必须可以安全地不关闭。前者更安全，与 Paimon 其他组件管理 catalog 实例的方式一致。 ########## paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java: ########## @@ -192,12 +247,68 @@ private Method getHiveCreatePartitionsInHmsMethod() throws NoSuchMethodException private LinkedHashMap<String, String> extractPartitionSpecFromPath( Review Comment: 🟢 **[suggestion] architecture — 分区 spec 提取逻辑重复** FormatTableCommit.extractPartitionSpecFromPath（private，约 65 行）重新实现了与 PartitionPathUtils.extractPartitionSpecFromPath(Path, List<String>)（package-private，同一 MR 中添加）相同的尾部组件遍历算法。两者都向后遍历路径组件、匹配声明的分区键、反转义值、生成 LinkedHashMap。Commit 版本额外处理 value-only 模式并抛出描述性异常而非返回 null。 核心算法（向后遍历、key 校验、反转义）重复，如果任一路径解析契约变更将独立漂移。 **建议**: 将共享的向后遍历-校验算法提取到 PartitionPathUtils 的单一方法中，接受错误处理策略（return-null vs throw）和 onlyValueInPath 标志。FormatTableCommit 委托给它，通过包装器添加表标识符上下文到错误消息中。 ########## paimon-core/src/main/java/org/apache/paimon/table/format/ManagedFormatTableScan.java: ########## @@ -0,0 +1,209 @@ +/* + * 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.paimon.table.format; + +import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.Path; +import org.apache.paimon.manifest.PartitionEntry; +import org.apache.paimon.partition.Partition; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.PartitionPathUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** A format table scan whose partition visibility is owned by a catalog. */ +public class ManagedFormatTableScan extends FormatTableScan { + + private static final Logger LOG = LoggerFactory.getLogger(ManagedFormatTableScan.class); + + private final FormatTableCatalogProvider catalogProvider; + + public ManagedFormatTableScan( + FormatTable table, + @Nullable PartitionPredicate partitionFilter, + @Nullable Integer limit) { + super(table, partitionFilter, limit); + FormatTableCatalogProvider provider = table.catalogProvider(); + if (provider == null) { + throw new IllegalStateException( + String.format( + "Managed format table %s has no catalog partition provider.", + table.fullName())); + } + this.catalogProvider = provider; + } + + @Override + protected List<Pair<LinkedHashMap<String, String>, Path>> findPartitions() { + String partitionNamePattern = createPartitionNamePattern(); + List<Partition> partitions = catalogProvider.listPartitions(partitionNamePattern); + if (partitions.isEmpty() && partitionNamePattern == null) { + warnIfFilesystemPartitionsExist(); + } + List<Pair<LinkedHashMap<String, String>, Path>> result = new ArrayList<>(partitions.size()); + Path tablePath = new Path(table.location()); + // Do not trust the catalog to be duplicate-free: a repeated spec would double every split + // of that partition and silently duplicate query results. + Set<String> seenPartitionPaths = new HashSet<>(partitions.size()); + for (Partition partition : partitions) { Review Comment: 🟢 **[suggestion] performance — 每次 managed scan 为每个分区构建转义路径字符串仅用于去重** 在 findPartitions() 中，对 catalog 返回的每个分区：先调用 normalizeSpec()（已对每个值执行 validatePartitionValueForPath），再调用 PartitionPathUtils.generatePartitionPathUtil() 构建完整转义的 `k=v/...` 路径字符串。该字符串**仅用作** seenPartitionPaths 去重集合的 key；实际的 Path 对象在下一行从它重建。 generatePartitionPathUtil 对每个 key 和 value 重新执行 escapePathName（逐字符遍历）并重新运行 validatePartitionValueForPath，因此每个分区的 spec 在每次 scan 中被转义/校验**两次**。由于列表是完整的 catalog 分区集（pattern pushdown 仅在有前导等值前缀时缩窄；range/IN/非前缀谓词列出所有分区），对于大型 managed table 这是 O(totalPartitions * partitionKeys) 的字符串分配 + 转义开销。 **建议**: 用已规范化的原始 spec（如 LinkedHashMap 的 values 或原始值的 join）作为去重 key，避免第二次 escapePathName 遍历和 generatePartitionPathUtil 内冗余的 validatePartitionValueForPath。每个唯一分区仅构建一次 Path。 -- 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]
