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/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。 ########## paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java: ########## @@ -265,12 +357,45 @@ public static LinkedHashMap<String, String> extractPartitionSpecFromPath(Path cu return fullPartSpec; } + /** Extract exactly the trailing key-value components for the declared partition keys. */ + @Nullable + static LinkedHashMap<String, String> extractPartitionSpecFromPath( + Path currPath, List<String> partitionKeys) { + String[] values = new String[partitionKeys.size()]; + Path current = currPath; + for (int i = partitionKeys.size() - 1; i >= 0; i--) { + if (current == null) { + return null; + } + Matcher matcher = PARTITION_NAME_PATTERN.matcher(current.getName()); + if (!matcher.matches() + || !partitionKeys.get(i).equals(unescapePathName(matcher.group(1)))) { + return null; + } + values[i] = unescapePathName(matcher.group(2)); + current = current.getParent(); + } + + LinkedHashMap<String, String> spec = new LinkedHashMap<>(); + for (int i = 0; i < partitionKeys.size(); i++) { + spec.put(partitionKeys.get(i), values[i]); + } + return spec; + } + public static LinkedHashMap<String, String> extractPartitionSpecFromPathOnlyValue( Path currPath, List<String> partitionKeys) { LinkedHashMap<String, String> fullPartSpec = new LinkedHashMap<>(); String[] split = currPath.toString().split(Path.SEPARATOR); for (int i = 0; i < partitionKeys.size(); i++) { - fullPartSpec.put(partitionKeys.get(i), split[split.length - partitionKeys.size() + i]); + // Unescape the directory component so the extracted value is the RAW partition value, + // consistent with the key=value branch (extractPartitionSpecFromPath) and with the + // values the write path registers into a partition-managing catalog. Without this, + // directories containing escaped characters (e.g. a%3Ab) would round-trip to a + // different value than the one registered (a:b). + fullPartSpec.put( + partitionKeys.get(i), + unescapePathName(split[split.length - partitionKeys.size() + i])); Review Comment: 🟡 **[minor] migration — value-only 分区值新增反转义，改变现有 format table 的分区 spec** extractPartitionSpecFromPathOnlyValue 中新增的 `unescapePathName()` 改变了所有使用 `format-table.partition-only-value-in-path=true` 的现有 format table 的文件系统扫描返回的分区 spec 值。以前目录名 `a%3Ab` 产生分区值 `a%3Ab`；现在产生 `a:b`。 虽然这是正确性修复（对齐 key=value 分支），但对现有表是**行为变更**： 1. 缓存或存储了旧转义分区值的下游消费者（如 Hive metastore 同步、外部元数据存储）将出现不匹配 2. 使用旧转义表示的分区过滤器查询将停止匹配 3. 向外部系统注册分区（MSCK、Hive sync）将注册与之前不同的值，可能创建重复分区条目 这影响 searchPartSpecAndPaths 代码路径，用于所有非 managed format table 扫描，不仅是新的 managed format table。 **建议**: 考虑将反转义行为置于表选项或版本标志之后，使现有表保留之前的读取语义直到显式迁移。或者在 release notes 中记录为已知行为变更，并提供下游元数据存储的迁移步骤。至少添加集成测试验证具有转义目录名（如 `a%3Ab`）的现有 value-only 表在升级后仍返回正确查询结果。 -- 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]
