github-actions[bot] commented on code in PR #66717: URL: https://github.com/apache/doris/pull/66717#discussion_r3823928006
########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergCacheSizeEstimator.java: ########## @@ -0,0 +1,483 @@ +// 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.datasource.iceberg; + +import org.apache.doris.datasource.NameMapping; +import org.apache.doris.datasource.iceberg.cache.ManifestCacheValue; +import org.apache.doris.datasource.metacache.MetaCacheSizeEstimate; +import org.apache.doris.datasource.metacache.MetaCacheWeightUtils; + +import org.apache.iceberg.BlobMetadata; +import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.PartitionField; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.PartitionStatisticsFile; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.StatisticsFile; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.encryption.EncryptedKey; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; + +import java.lang.reflect.Field; +import java.lang.reflect.Modifier; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Publication-time approximate weight formulas for Iceberg cache entries. + * + * <p>The formulas follow a coarse cardinality model: a rounded-up constant per stable logical + * dimension (snapshot, schema field, partition field, metadata entry, ...) plus the + * skew-sensitive string payload the loader already materialized. Constants are calibrated + * offline and deliberately absorb the lazy state Iceberg materializes after admission (schema + * name/id/lower-case/accessor indexes, partition-type graphs) instead of modeling those objects + * individually, so weights track metadata size without depending on SDK-private layouts. + * {@code max-weight} is an estimated admission budget, not an exact heap limit. + */ +final class IcebergCacheSizeEstimator { + // Every metadata element visited (field, snapshot, summary entry, ...) costs a few reads; + // the bound only guards against pathological metadata and is far above real tables (a + // 10,000-snapshot history with 15 summary keys each is 160,000 elements). Exceeding it + // rejects weighted admission, so it must not be reachable by ordinary long-lived tables. + private static final long MAX_TABLE_ACCOUNTING_ELEMENTS = 2_000_000L; + // Total name characters the estimator may account for retained name indexes. + private static final long MAX_TABLE_ACCOUNTING_CHARACTERS = 4_000_000L; + private static final int MAX_TYPE_ACCOUNTING_DEPTH = 128; + + private static final long KEY_BASE_WEIGHT = 256L; + private static final long TABLE_BASE_WEIGHT = 16L * 1024L; + // One schema version: the Schema object, its lists, the schemasById entry and the struct it + // wraps, including their growth from singleton to regular immutable shapes. + private static final long SCHEMA_WEIGHT = 1024L; + // One nested field at any depth: the NestedField and its type node plus this field's share + // of every eager and lazy schema index (idToName, nameToId, idToField, lowerCaseNameToId, + // idToAccessor, struct field indexes and the secondary partition-type schema graph). + private static final long FIELD_WEIGHT = 1408L; + // Retained copies of one field name across the case-sensitive and lower-cased name indexes. + private static final long NAME_INDEX_COPIES = 4L; + // One partition spec: the spec object, its field list, javaClasses and the partitionType() + // graph with its own indexes. + private static final long SPEC_WEIGHT = 2048L; + // One partition field including its transform, index entries and the per-field share of the + // lazily built fieldsBySourceId multimap. + private static final long PARTITION_FIELD_WEIGHT = 1024L; + // The lazily built fieldsBySourceId multimap grows O(distinctSourceIds * fieldCount). + private static final long FIELDS_BY_SOURCE_SLOT_WEIGHT = 64L; + private static final long SORT_ORDER_WEIGHT = 512L; + private static final long SORT_FIELD_WEIGHT = 256L; + // One entry of any retained string map (table properties, snapshot summaries, blob or key + // properties): map node plus boxed/list slack; the strings are charged separately. + private static final long METADATA_ENTRY_WEIGHT = 128L; + private static final long SNAPSHOT_WEIGHT = 512L; + private static final long CURRENT_SNAPSHOT_WEIGHT = 1024L; + private static final long SNAPSHOT_LOG_WEIGHT = 64L; + private static final long METADATA_LOG_WEIGHT = 160L; + private static final long SNAPSHOT_REF_WEIGHT = 256L; + private static final long STATISTICS_FILE_WEIGHT = 640L; + private static final long BLOB_METADATA_WEIGHT = 256L; + private static final long BLOB_FIELD_WEIGHT = 32L; + private static final long PARTITION_STATISTICS_FILE_WEIGHT = 320L; + private static final long ENCRYPTED_KEY_WEIGHT = 320L; + // One retained IcebergPartition (value/transform lists) or one RangePartitionItem with a + // single partition column plus its map entry; extra columns are charged by IcebergPartitionInfo. + private static final long PARTITION_WEIGHT = 768L; + // Outer map entry and table share of one merged-overlap group; the alias set itself and its + // contents are charged by IcebergPartitionInfo per enclosed partition name. + private static final long PARTITION_ALIAS_WEIGHT = 160L; + // One name-mapping field: map node, boxed id and list object; alias arrays and Strings are + // charged by IcebergSnapshotCacheValue when the mapping is copied. + private static final long NAME_MAPPING_ENTRY_WEIGHT = 256L; + private static final long MANIFEST_ENTRY_BASE_WEIGHT = 512L; + private static final long DATA_FILE_WEIGHT = 1024L; + private static final long DELETE_FILE_WEIGHT = 1024L; + private static final long FILE_METRIC_ENTRY_WEIGHT = 128L; + + // TableMetadata.snapshots()/snapshot(id) load lazily through a catalog supplier + // (REST snapshot-loading-mode=refs). Publication must not perform that IO, so this single + // reflective probe is retained as an IO guard rather than a layout model. + private static final Field TABLE_METADATA_SNAPSHOTS_LOADED_FIELD = + loadTableMetadataField("snapshotsLoaded", boolean.class); + private static final Field TABLE_METADATA_SNAPSHOTS_SUPPLIER_FIELD = + loadTableMetadataField("snapshotsSupplier", null); + + private IcebergCacheSizeEstimator() { + } + + static MetaCacheSizeEstimate estimateTableEntry(NameMapping key, IcebergTableCacheValue value) { + Table table = value.getRetainedIcebergTable(); + MetaCacheSizeEstimate support = checkSupportedTable(table); + if (!support.isComplete()) { + return support; + } + long bytes = MetaCacheWeightUtils.saturatedAdd( + KEY_BASE_WEIGHT, MetaCacheWeightUtils.estimatedNameMappingBytes(key)); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, estimateTable(table)); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, value.getRetainedTablePayloadBytes()); + bytes = MetaCacheWeightUtils.saturatedAdd( + bytes, value.getRetainedCurrentSnapshotPayloadBytes()); + return MetaCacheSizeEstimate.complete(bytes); + } + + static MetaCacheSizeEstimate estimateSnapshotEntry( + IcebergSnapshotEntryKey key, IcebergSnapshotCacheValue value) { + long bytes = KEY_BASE_WEIGHT; + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, + MetaCacheWeightUtils.estimatedNameMappingBytes(key.getNameMapping())); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, + MetaCacheWeightUtils.estimatedStringBytes(key.getTableUuid())); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, + MetaCacheWeightUtils.estimatedStringBytes(key.getMetadataFileLocation())); + + IcebergPartitionInfo partitionInfo = value.getPartitionInfo(); + bytes = addCount(bytes, partitionInfo.getNameToPartitionItem().size(), PARTITION_WEIGHT); + bytes = addCount(bytes, partitionInfo.getNameToIcebergPartition().size(), PARTITION_WEIGHT); + bytes = addCount(bytes, partitionInfo.getNameToIcebergPartitionNames().size(), + PARTITION_ALIAS_WEIGHT); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, partitionInfo.getRetainedPayloadBytes()); + bytes = addCount(bytes, value.getNameMapping().map(Map::size).orElse(0), + NAME_MAPPING_ENTRY_WEIGHT); + bytes = MetaCacheWeightUtils.saturatedAdd( + bytes, value.getRetainedNameMappingPayloadBytes()); + + if (value.getRetainedIcebergTable().isPresent()) { + // The projection keeps its own reference to the frozen table generation. That graph + // is charged here as well as by the table entry that produced it: the two entries have + // independent lifetimes (TTL, weight eviction, soft collection) and either may outlive + // the other, so each must be able to carry the graph on its own. Budgets should be + // sized for the table metadata being counted once per dependent entry. + Table table = value.getRetainedIcebergTable().get(); + MetaCacheSizeEstimate support = checkSupportedTable(table); + if (!support.isComplete()) { + return support; + } + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, estimateTable(table)); + bytes = MetaCacheWeightUtils.saturatedAdd( + bytes, value.getRetainedTablePayloadBytes()); + bytes = MetaCacheWeightUtils.saturatedAdd( + bytes, value.getRetainedCurrentSnapshotPayloadBytes()); + } + return MetaCacheSizeEstimate.complete(bytes); + } + + static MetaCacheSizeEstimate estimateManifestEntry( + IcebergManifestEntryKey key, ManifestCacheValue value) { + if (!value.isAccountingComplete()) { + return MetaCacheSizeEstimate.incomplete("iceberg_manifest_accounting_incomplete"); + } + long bytes = MetaCacheWeightUtils.saturatedAdd( + MANIFEST_ENTRY_BASE_WEIGHT, + MetaCacheWeightUtils.estimatedStringBytes(key.getManifestPath())); + bytes = addCount(bytes, value.getDataFiles().size(), DATA_FILE_WEIGHT); + bytes = addCount(bytes, value.getDeleteFiles().size(), DELETE_FILE_WEIGHT); + bytes = addCount(bytes, value.getDataFileMetricEntryCount(), FILE_METRIC_ENTRY_WEIGHT); + bytes = addCount(bytes, value.getDeleteFileMetricEntryCount(), FILE_METRIC_ENTRY_WEIGHT); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, value.getRetainedPayloadBytes()); + return MetaCacheSizeEstimate.complete(bytes); + } + + private static MetaCacheSizeEstimate checkSupportedTable(Table table) { + if (table == null) { + return MetaCacheSizeEstimate.incomplete("missing_iceberg_table"); + } + if (!(table instanceof HasTableOperations)) { + return MetaCacheSizeEstimate.incomplete( + "unsupported_iceberg_table:" + table.getClass().getName()); + } + TableMetadata metadata = ((HasTableOperations) table).operations().current(); + if (metadata == null) { + return MetaCacheSizeEstimate.incomplete("missing_iceberg_table_metadata"); + } + if (!areSnapshotsLoaded(metadata)) { + return MetaCacheSizeEstimate.incomplete("iceberg_snapshots_not_loaded"); + } + if (metadata.metadataFileLocation() == null + || metadata.metadataFileLocation().isEmpty()) { + return MetaCacheSizeEstimate.incomplete("missing_iceberg_metadata_location"); + } + return MetaCacheSizeEstimate.complete(1L); + } + + /** Reads only metadata collection sizes and a constant number of strings; no FileIO is used. */ + private static long estimateTable(Table table) { + TableMetadata metadata = ((HasTableOperations) table).operations().current(); + long bytes = MetaCacheWeightUtils.saturatedAdd( + TABLE_BASE_WEIGHT, MetaCacheWeightUtils.estimatedStringBytes(table.name())); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, + MetaCacheWeightUtils.estimatedStringBytes(metadata.location())); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, + MetaCacheWeightUtils.estimatedStringBytes(metadata.metadataFileLocation())); + bytes = addCount(bytes, metadata.properties().size(), METADATA_ENTRY_WEIGHT); + if (metadata.currentSnapshot() != null) { + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, CURRENT_SNAPSHOT_WEIGHT); + } + return bytes; + } + + /** + * Weight of everything a retained table generation's metadata can grow into, computed from + * already-parsed metadata with bounded publication-time work and no IO. + */ + static long retainedTablePayloadBytes(Table table) { + if (!(table instanceof HasTableOperations)) { + return 0L; + } + TableMetadata metadata = ((HasTableOperations) table).operations().current(); + if (metadata == null) { + return 0L; + } + if (!areSnapshotsLoaded(metadata)) { + // snapshots()/refs() would call the catalog's lazy snapshot supplier: fail closed. + throw new IllegalStateException("Iceberg table snapshots are not loaded"); + } + + long bytes = 0L; + AccountingBudget budget = new AccountingBudget( + MAX_TABLE_ACCOUNTING_ELEMENTS, MAX_TABLE_ACCOUNTING_CHARACTERS); + for (PartitionSpec spec : metadata.specs()) { + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, partitionSpecBytes(spec, budget)); + } + for (SortOrder sortOrder : metadata.sortOrders()) { + budget.chargeElements(MetaCacheWeightUtils.saturatedAdd( + 1L, sortOrder.fields().size())); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, SORT_ORDER_WEIGHT); + bytes = addCount(bytes, sortOrder.fields().size(), SORT_FIELD_WEIGHT); + } + for (Schema schema : metadata.schemas()) { + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, schemaBytes(schema, budget)); + } + budget.chargeElements(metadata.properties().size()); + bytes = addCount(bytes, metadata.properties().size(), METADATA_ENTRY_WEIGHT); + for (Map.Entry<String, String> property : metadata.properties().entrySet()) { + bytes = addString(bytes, property.getKey()); + bytes = addString(bytes, property.getValue()); + } + for (Snapshot snapshot : metadata.snapshots()) { + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, snapshotBytes(snapshot, budget)); + } + budget.chargeElements(metadata.snapshotLog().size()); + bytes = addCount(bytes, metadata.snapshotLog().size(), SNAPSHOT_LOG_WEIGHT); + budget.chargeElements(metadata.previousFiles().size()); + for (TableMetadata.MetadataLogEntry previousFile : metadata.previousFiles()) { + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, METADATA_LOG_WEIGHT); + bytes = addString(bytes, previousFile.file()); + } + budget.chargeElements(metadata.refs().size()); + bytes = addCount(bytes, metadata.refs().size(), SNAPSHOT_REF_WEIGHT); + for (String refName : metadata.refs().keySet()) { + bytes = addString(bytes, refName); + } + budget.chargeElements(metadata.statisticsFiles().size()); + for (StatisticsFile statisticsFile : metadata.statisticsFiles()) { + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, STATISTICS_FILE_WEIGHT); + bytes = addString(bytes, statisticsFile.path()); + for (BlobMetadata blob : statisticsFile.blobMetadata()) { + budget.chargeElements(MetaCacheWeightUtils.saturatedAdd(1L, + MetaCacheWeightUtils.saturatedAdd( + blob.fields().size(), blob.properties().size()))); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, BLOB_METADATA_WEIGHT); + bytes = addString(bytes, blob.type()); + bytes = addCount(bytes, blob.fields().size(), BLOB_FIELD_WEIGHT); + bytes = addStringMap(bytes, blob.properties()); + } + } + budget.chargeElements(metadata.partitionStatisticsFiles().size()); + for (PartitionStatisticsFile statisticsFile : metadata.partitionStatisticsFiles()) { + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, PARTITION_STATISTICS_FILE_WEIGHT); + bytes = addString(bytes, statisticsFile.path()); + } + budget.chargeElements(metadata.encryptionKeys().size()); + for (EncryptedKey encryptedKey : metadata.encryptionKeys()) { + budget.chargeElements(encryptedKey.properties().size()); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, ENCRYPTED_KEY_WEIGHT); + bytes = addString(bytes, encryptedKey.keyId()); + bytes = addString(bytes, encryptedKey.encryptedById()); + if (encryptedKey.encryptedKeyMetadata() != null) { + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, + MetaCacheWeightUtils.estimatedByteArrayBytes( + encryptedKey.encryptedKeyMetadata().remaining())); + } + bytes = addStringMap(bytes, encryptedKey.properties()); + } + bytes = addString(bytes, metadata.uuid()); + return bytes; + } + + private static long partitionSpecBytes(PartitionSpec spec, AccountingBudget budget) { + List<PartitionField> fields = spec.fields(); + budget.chargeElements(MetaCacheWeightUtils.saturatedAdd(1L, fields.size())); + long bytes = SPEC_WEIGHT; + bytes = addCount(bytes, fields.size(), PARTITION_FIELD_WEIGHT); + Set<Integer> sourceIds = new HashSet<>(); + for (PartitionField field : fields) { + sourceIds.add(field.sourceId()); + budget.chargeCharacters(field.name().length()); + // The name is retained by the field and again by the partition-type name indexes. + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, MetaCacheWeightUtils.saturatedMultiply( + MetaCacheWeightUtils.estimatedStringBytes(field.name()), NAME_INDEX_COPIES)); + } + // Reserve the O(distinctSources * fields) growth of the lazy fieldsBySourceId index. + bytes = addCount(bytes, + MetaCacheWeightUtils.saturatedMultiply(sourceIds.size(), fields.size()), + FIELDS_BY_SOURCE_SLOT_WEIGHT); + return bytes; + } + + /** + * One schema version: a constant per nested field at any depth plus the retained name/doc + * payload. The per-field constant absorbs the type node and this field's share of every + * eager and lazily materialized schema index. + */ + private static long schemaBytes(Schema schema, AccountingBudget budget) { + long bytes = SCHEMA_WEIGHT; + bytes = addCount(bytes, schema.identifierFieldIds().size(), METADATA_ENTRY_WEIGHT); + for (Types.NestedField field : schema.columns()) { + bytes = MetaCacheWeightUtils.saturatedAdd( + bytes, fieldBytes(field, budget, 0)); + } + return bytes; + } + + private static long fieldBytes(Types.NestedField field, AccountingBudget budget, int depth) { + if (depth > MAX_TYPE_ACCOUNTING_DEPTH) { + throw new IllegalStateException("Iceberg schema type nesting is too deep"); + } + budget.chargeElements(1L); + long bytes = FIELD_WEIGHT; + String name = field.name(); + if (name != null) { + budget.chargeCharacters(name.length()); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, MetaCacheWeightUtils.saturatedMultiply( Review Comment: [P2] Account generated qualified field paths `fieldBytes()` charges and bounds only each local `field.name()`, but the lazy Schema name maps retained by this entry materialize fully qualified nested paths. For example, a supported 128-level chain with 30,000-character local names totals 3.84M characters and passes this guard; the four-copy formula covers about 15.36M characters, while one canonical qualified-path index alone retains about 247.68M characters (`30000 * (1 + ... + 128)`), before lower-case and other name maps. Ordinary schema lookups can initialize these maps after admission, bypassing the entry/catalog/global reservation. Please account the generated ancestor paths (and charge them to the work bound), or fail closed, with a deep long-name JOL case that materializes all Schema indexes. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonCacheSizeEstimator.java: ########## @@ -0,0 +1,355 @@ +// 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.datasource.paimon; + +import org.apache.doris.datasource.NameMapping; +import org.apache.doris.datasource.metacache.MetaCacheSizeEstimate; +import org.apache.doris.datasource.metacache.MetaCacheWeightUtils; + +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.DelegatedFileStoreTable; +import org.apache.paimon.table.FallbackReadFileStoreTable; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.RowType; + +import java.util.List; +import java.util.Map; + +/** + * Publication-time approximate weight formulas for Paimon table handles and snapshot projections. + * + * <p>The formulas follow a coarse cardinality model: a rounded-up constant per stable logical + * dimension (schema field, logical type node, option, key, partition, ...) plus the + * skew-sensitive string payload the loader already materialized. The per-node constants absorb + * the lazy state Paimon materializes after admission (the four RowType lookup maps, the store + * graph and its derived RowType copies) instead of modeling those objects individually, so + * weights track metadata size without depending on SDK-private layouts. {@code max-weight} is + * an estimated admission budget, not an exact heap limit. Estimation never opens the table + * store and performs no IO. + */ +final class PaimonCacheSizeEstimator { + // Bounds accounting work per publication; far above real schemas and option maps. + private static final long MAX_TABLE_ACCOUNTING_ELEMENTS = 50_000L; + private static final int MAX_TYPE_ACCOUNTING_DEPTH = 256; + + private static final long KEY_BASE_WEIGHT = 256L; + private static final long SNAPSHOT_BASE_WEIGHT = 4L * 1024L; + private static final long TABLE_BASE_WEIGHT = 16L * 1024L; + // PaimonTableCacheValue and its size-estimate holder. + private static final long TABLE_VALUE_BASE_WEIGHT = 128L; + // One schema field (DataField) at any depth, including its share of the enclosing RowType's + // four lazily built lookup maps, boxed ids and the field copies the lazily created store + // graph derives from it (append-only row copy, trimmed key/value types, merge row type). + private static final long FIELD_NODE_WEIGHT = 576L; + // One bare nested type node (array/map/multiset/vector element types and unknown future + // DataType implementations), including its store-graph copies; charged generically instead + // of disabling weighted caching for unknown types. + private static final long TYPE_NODE_WEIGHT = 128L; + // One nested RowType container: the RowType, its field list, its four lazily built lookup + // maps and the container copies the store graph derives. + private static final long ROW_CONTAINER_WEIGHT = 2048L; + // One schema/table-level entry: option map node, key list slot and boxes. + private static final long OPTION_WEIGHT = 192L; + private static final long KEY_WEIGHT = 192L; + // Wrapper tables (privileged, fallback-read) around the concrete FileStoreTable. + private static final long WRAPPER_WEIGHT = 512L; + private static final long PARTITION_WEIGHT = 320L; + private static final long PARTITION_ITEM_WEIGHT = 768L; + + private PaimonCacheSizeEstimator() { + } + + /** + * Retained weight of the base table entry. The table handle is owned independently of the + * snapshot projections that reference it (they may pin an older generation), so the same + * table graph is charged to both owners rather than shared. + */ + static MetaCacheSizeEstimate estimateTableEntry(NameMapping key, PaimonTableCacheValue value) { + String unsupported = unsupportedReason(value.getPaimonTable()); + if (unsupported != null) { + return MetaCacheSizeEstimate.incomplete(unsupported); + } + long bytes = MetaCacheWeightUtils.saturatedAdd( + KEY_BASE_WEIGHT, MetaCacheWeightUtils.estimatedNameMappingBytes(key)); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, TABLE_VALUE_BASE_WEIGHT); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, value.getRetainedTablePayloadBytes()); + return MetaCacheSizeEstimate.complete( + MetaCacheWeightUtils.saturatedAdd(bytes, estimateTable(value.getPaimonTable()))); + } + + static MetaCacheSizeEstimate estimateSnapshotEntry( + PaimonSnapshotEntryKey key, PaimonSnapshotCacheValue value) { + Table table = value.getSnapshot().getTable(); + String unsupported = unsupportedReason(table); + if (unsupported != null) { + return MetaCacheSizeEstimate.incomplete(unsupported); + } + long bytes = MetaCacheWeightUtils.saturatedAdd( + KEY_BASE_WEIGHT, MetaCacheWeightUtils.estimatedNameMappingBytes(key.getNameMapping())); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, SNAPSHOT_BASE_WEIGHT); + bytes = addCount(bytes, value.getPartitionInfo().getNameToPartition().size(), PARTITION_WEIGHT); + bytes = addCount(bytes, + value.getPartitionInfo().getNameToPartitionItem().size(), PARTITION_ITEM_WEIGHT); + bytes = MetaCacheWeightUtils.saturatedAdd( + bytes, value.getPartitionInfo().getRetainedPayloadBytes()); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, value.getRetainedTablePayloadBytes()); + return MetaCacheSizeEstimate.complete( + MetaCacheWeightUtils.saturatedAdd(bytes, estimateTable(table))); + } + + private static String unsupportedReason(Table table) { + if (table == null) { + return "unsupported_paimon_table:null"; + } + if (unwrap(table) == null) { + return "unsupported_paimon_table:" + table.getClass().getName(); + } + return null; + } + + /** The concrete FileStoreTable behind any known wrapper chain, or null. */ + private static FileStoreTable unwrap(Table table) { + if (table instanceof DelegatedFileStoreTable) { + return unwrap(((DelegatedFileStoreTable) table).wrapped()); + } + return table instanceof FileStoreTable ? (FileStoreTable) table : null; + } + + /** Uses TableSchema cardinalities only and deliberately never calls FileStoreTable.store(). */ + private static long estimateTable(Table table) { + long bytes = 0L; + Table current = table; + while (true) { + if (current instanceof FallbackReadFileStoreTable) { + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, WRAPPER_WEIGHT); + bytes = MetaCacheWeightUtils.saturatedAdd( + bytes, estimateTable(((FallbackReadFileStoreTable) current).other())); + current = ((FallbackReadFileStoreTable) current).wrapped(); + continue; + } + if (current instanceof DelegatedFileStoreTable) { + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, WRAPPER_WEIGHT); + current = ((DelegatedFileStoreTable) current).wrapped(); + continue; + } + break; + } + if (!(current instanceof FileStoreTable)) { + return bytes; + } + FileStoreTable fileStoreTable = (FileStoreTable) current; + TableSchema schema = fileStoreTable.schema(); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, TABLE_BASE_WEIGHT); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, + MetaCacheWeightUtils.estimatedStringBytes(current.name())); + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, + MetaCacheWeightUtils.estimatedStringBytes(fileStoreTable.location().toString())); + // The store graph the table lazily materializes derives several RowType copies of the + // schema; the per-node constants absorb those copies instead of modeling store classes. + NodeCounts nodes = new NodeCounts(); + countFieldNodes(schema.fields(), 0, nodes, + new AccountingBudget(MAX_TABLE_ACCOUNTING_ELEMENTS)); + bytes = addCount(bytes, nodes.fieldNodes, FIELD_NODE_WEIGHT); + bytes = addCount(bytes, nodes.bareTypeNodes, TYPE_NODE_WEIGHT); + bytes = addCount(bytes, nodes.rowContainers, ROW_CONTAINER_WEIGHT); + bytes = addCount(bytes, schema.options().size(), OPTION_WEIGHT); + bytes = addCount(bytes, schema.partitionKeys().size(), KEY_WEIGHT); + bytes = addCount(bytes, schema.primaryKeys().size(), KEY_WEIGHT); + bytes = addCount(bytes, schema.bucketKeys().size(), KEY_WEIGHT); + for (String primaryKey : schema.primaryKeys()) { + // Each trimmed key field of a primary-key store gets a fresh "_KEY_" + name string. + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, + MetaCacheWeightUtils.estimatedStringBytes(primaryKey)); + } + return bytes; + } + + private static final class NodeCounts { + private long fieldNodes; + private long bareTypeNodes; + private long rowContainers; + } + + private static void countFieldNodes( + List<DataField> fields, int depth, NodeCounts counts, AccountingBudget budget) { + if (depth > MAX_TYPE_ACCOUNTING_DEPTH) { + throw new IllegalStateException("Paimon schema type nesting is too deep"); + } + for (DataField field : fields) { + budget.charge(1L); + counts.fieldNodes = MetaCacheWeightUtils.saturatedAdd(counts.fieldNodes, 1L); + countTypeNodes(field.type(), depth, counts, budget); + } + } + + private static void countTypeNodes( + DataType type, int depth, NodeCounts counts, AccountingBudget budget) { + if (depth > MAX_TYPE_ACCOUNTING_DEPTH) { + throw new IllegalStateException("Paimon schema type nesting is too deep"); + } + if (type instanceof RowType) { + counts.rowContainers = MetaCacheWeightUtils.saturatedAdd(counts.rowContainers, 1L); + countFieldNodes(((RowType) type).getFields(), depth + 1, counts, budget); + return; + } + if (type != null) { + // Known container types (array, map, multiset) contribute their children as nodes; + // other types, including future implementations, are charged as a single node. + for (DataType child : childTypes(type)) { + budget.charge(1L); + counts.bareTypeNodes = MetaCacheWeightUtils.saturatedAdd(counts.bareTypeNodes, 1L); + countTypeNodes(child, depth + 1, counts, budget); + } + } + } + + private static List<DataType> childTypes(DataType type) { + if (type instanceof org.apache.paimon.types.ArrayType) { + return java.util.Collections.singletonList( + ((org.apache.paimon.types.ArrayType) type).getElementType()); + } + if (type instanceof org.apache.paimon.types.MultisetType) { + return java.util.Collections.singletonList( + ((org.apache.paimon.types.MultisetType) type).getElementType()); + } + if (type instanceof org.apache.paimon.types.MapType) { + org.apache.paimon.types.MapType mapType = (org.apache.paimon.types.MapType) type; + return java.util.Arrays.asList(mapType.getKeyType(), mapType.getValueType()); + } + return java.util.Collections.emptyList(); + } + + /** + * Captures skew-sensitive schema text once when the cache value is constructed. All + * collections are already materialized in TableSchema; this never opens the table store. + */ + static long retainedTablePayloadBytes(Table table) { + return retainedTablePayloadBytes( + table, new AccountingBudget(MAX_TABLE_ACCOUNTING_ELEMENTS)); + } + + private static long retainedTablePayloadBytes(Table table, AccountingBudget budget) { + budget.charge(1L); + if (table instanceof FallbackReadFileStoreTable) { + FallbackReadFileStoreTable fallback = (FallbackReadFileStoreTable) table; + return MetaCacheWeightUtils.saturatedAdd( + retainedTablePayloadBytes(fallback.wrapped(), budget), + retainedTablePayloadBytes(fallback.other(), budget)); + } + if (table instanceof DelegatedFileStoreTable) { + return retainedTablePayloadBytes( + ((DelegatedFileStoreTable) table).wrapped(), budget); + } + if (!(table instanceof FileStoreTable)) { + return 0L; + } + TableSchema schema = ((FileStoreTable) table).schema(); + if (schema == null) { + return 0L; + } + long bytes = addString(0L, schema.comment()); + for (DataField field : schema.fields()) { + bytes = addFieldPayload(bytes, field, budget, 0); + } + budget.charge(schema.options().size()); + for (Map.Entry<String, String> option : schema.options().entrySet()) { + bytes = addString(bytes, option.getKey()); + bytes = addString(bytes, option.getValue()); + } + bytes = addStrings(bytes, schema.partitionKeys(), budget); + bytes = addStrings(bytes, schema.primaryKeys(), budget); + return addStrings(bytes, schema.bucketKeys(), budget); + } + + private static long addFieldPayload( + long bytes, DataField field, AccountingBudget budget, int depth) { + if (depth > MAX_TYPE_ACCOUNTING_DEPTH) { + throw new IllegalStateException("Paimon schema type nesting is too deep"); + } + budget.charge(1L); + bytes = addString(bytes, field.name()); + bytes = addString(bytes, field.description()); Review Comment: [P2] Account retained Paimon field defaults Both table and snapshot estimates include this payload walk, but `addFieldPayload()` charges the field name and description and never charges `field.defaultValue()`. That nullable String is retained by each Paimon `DataField` and is used by the schema/write paths; a single long default, or defaults across a wide schema, can therefore exceed the fixed 576-byte per-field allowance and bypass the entry/catalog/global reservation by an arbitrary amount. Please include the default string in the skew-sensitive payload accounting and add fixed-field-count calibration cases whose default length grows. -- 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]
