github-actions[bot] commented on code in PR #66717: URL: https://github.com/apache/doris/pull/66717#discussion_r3826016608
########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergCacheSizeEstimator.java: ########## @@ -0,0 +1,549 @@ +// 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 path across the case-sensitive, lower-cased and alias + // name indexes. + private static final long NAME_INDEX_COPIES = 5L; + // Per nesting level of a field: accessor wrappers and the enclosing struct's index shares. + private static final long NESTED_LEVEL_WEIGHT = 128L; + // String object and array overhead per retained generated path copy. + private static final long STRING_OVERHEAD_WEIGHT = 48L; + // 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 MetaCacheWeightUtils.saturatedAdd(bytes, fileIoBytes(table)); + } + + /** + * The retained operations strongly own the handle's FileIO with its configuration and any + * vended storage credentials; those maps grow independently of table metadata, so their + * payload is charged per owner. A FileIO that cannot expose its configuration makes the + * estimate fail closed via {@code estimateSafely}. + */ + private static long fileIoBytes(Table table) { + org.apache.iceberg.io.FileIO fileIo = table.io(); + if (fileIo == null) { + return 0L; + } + long bytes = addStringMapWithEntries(0L, fileIo.properties()); + if (fileIo instanceof org.apache.iceberg.io.SupportsStorageCredentials) { + for (org.apache.iceberg.io.StorageCredential credential + : ((org.apache.iceberg.io.SupportsStorageCredentials) fileIo).credentials()) { + bytes = MetaCacheWeightUtils.saturatedAdd(bytes, METADATA_ENTRY_WEIGHT); + bytes = addString(bytes, credential.prefix()); + bytes = addStringMapWithEntries(bytes, credential.config()); + } + } + return bytes; + } + + private static long addStringMapWithEntries(long bytes, Map<String, String> values) { + if (values == null) { + return bytes; + } + bytes = addCount(bytes, values.size(), METADATA_ENTRY_WEIGHT); + return addStringMap(bytes, values); + } + + /** + * 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 (org.apache.iceberg.SortField field : sortOrder.fields()) { + bytes = addTransformPayload(bytes, field.transform(), budget); + } + } + 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()); Review Comment: [P2] Charge the structure of blob and key property maps `METADATA_ENTRY_WEIGHT` explicitly covers blob/key property-map entries, but this loop only charges their strings; unlike table properties, snapshot summaries, and FileIO maps, it never adds `blob.properties().size() * METADATA_ENTRY_WEIGHT`. The encrypted-key loop below has the same omission. A fixed number of blobs/keys with many short properties is therefore undercharged by 128 bytes per property, allowing the retained graph to exceed its entry/catalog/global reservation. Please add the per-entry structural charge in both loops and calibrate a fixed-blob/key case that increases only the property count. -- 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]
