This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new a34b2c6068a [fix](fe) Keep Iceberg and Paimon metadata caches within
their weight limits (#68842)
a34b2c6068a is described below
commit a34b2c6068ac91f232d73e5379dc0a37bbcd159b
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Mon Oct 12 14:28:22 2026 +0800
[fix](fe) Keep Iceberg and Paimon metadata caches within their weight
limits (#68842)
### What problem does this PR solve?
Issue Number: None
Related PR: #66717, #67726
Problem Summary:
**In short.** This ports four fixes from branch-4.1 #66717 (memory
governance for external metadata caches) that master's re-implementation
(#67726) does not have. Under a metadata cache weight limit, master
still lets the Iceberg SDK keep its own unbounded manifest cache,
fetches the whole snapshot history of REST `snapshot-loading-mode=refs`
tables on every table cache miss, and under-weighs cached Paimon tables
by 2-5x because they keep growing after admission. Catalogs without a
weight limit are not affected.
**Background.** #66717 bounds the estimated retained memory of external
metadata caches with the FE-wide `external_meta_cache_max_weight`, the
catalog property `meta.cache.max-weight`, and per-entry
`meta.cache.<engine>.<entry>.max-weight`. A value is weighed once, when
it is admitted; if it does not fit, it is returned to the query but not
cached. master re-implemented this in #67726 on the connector cache
framework, and #68839 ports the 4.x tests. Comparing the two
implementations found the gaps below; branch-4.1 handles all of them.
**The problem, and what it cost.**
- **Iceberg SDK manifest cache outside the limit.**
`IcebergCatalogFactory` derives `io.manifest.cache-enabled=true`
whenever `meta.cache.iceberg.manifest.*` is enabled. That SDK cache
keeps up to `io.manifest.cache.max-total-bytes` (100 MB by default) of
manifest content per FileIO, outside every Doris limit, and Glue / S3
Tables catalogs have one FileIO per table. A user who sets a limit gets
a second, unbounded manifest cache next to the bounded one.
- **Remote IO while weighing REST refs-mode tables.** A weight-limited
`IcebergTableCache` serializes the loaded metadata
(`TableMetadataParser.toJson`) and weighs it. Both call
`TableMetadata.snapshots()`, which in refs mode runs the catalog's
supplier and fetches every snapshot. So each table cache miss pays one
extra full-history REST load, exactly the cost refs mode exists to
avoid, and the cached value then holds the whole history.
- **Paimon tables grow after admission.** `paimon-table` is weighed from
the schema, location and catalog environment. A cached `FileStoreTable`
then grows: the first `latestSnapshot()` or scan builds its transient
store (`RowType` copies of the schema, a copy of the options, and for a
partial-update table a merge function factory with sequence-group and
aggregator maps for every field), and field lookups build the four lazy
index maps of every `RowType`. Measured on Paimon 1.4.2, the grown graph
is 2-5x the admitted weight, so the heap the cache really holds can
exceed `MAX_WEIGHT` by that factor.
- **Paimon tables can own their FileIO, which is not charged.** A REST
catalog gives each table its own `RESTTokenFileIO` with data tokens (the
vended token, about 2 KB, arrives on first data access) or
`ResolvingFileIO` without them (`RESTCatalog.fileIOFromOptions`), and a
Hive catalog resolves a FileIO per table location
(`HiveCatalog.fileIO(Path)`). Their tokens and backend maps appear after
admission.
**How this PR fixes it.**
- `IcebergConnector#hasMetaCacheWeightLimit` decides whether any limit
applies: the FE-global limit, the catalog limit, or the `max-weight` of
one of the entries the connector reads (`table`, `partition`,
`manifest`, `partition_view`, now one constant shared with the DDL
validation). The three catalog option builders take that flag, and
`appendManifestCacheProperties` does not derive
`io.manifest.cache-enabled` under a limit. An explicit user value still
wins, as in branch-4.1.
- `IcebergCacheSizeEstimator.areSnapshotsLoaded` reads
`TableMetadata.snapshotsLoaded` (Iceberg exposes no accessor for it). A
weight-limited `TableOwner` checks it first and, if the list is still
behind its supplier, declines retention with
`incomplete_estimate:iceberg_snapshots_not_loaded` instead of
serializing. The statement keeps the live table, whose supplier still
serves historical snapshots on demand.
- `PaimonCacheSizeEstimator` reserves the post-admission growth of every
concrete `FileStoreTable` from its schema shape: a base, each field,
each field nested in a row, each child type of an array / map / multiset
/ vector, each option and each key. The constants are calibrated against
a retained-size oracle that walks the grown graph (new test helper
`EstimatorCalibrationAssertions`). A partial-update table reserves
another 256 bytes per field for its merge function. Every concrete table
also carries a fixed 16 KB allowance for its FileIO, as on branch-4.x:
Paimon offers no IO-free way to tell an owned FileIO from one the
catalog shares, and charging a shared FileIO to each table only makes
the estimate conservative. The allowance covers an 8,192-character REST
token and a `ResolvingFileIO` with its backend; a token refresh replaces
the token rather than adding to it.
**Results.**
| Case | Before | After |
|---|---|---|
| `meta.cache.iceberg.manifest.enable=true` plus any weight limit | SDK
manifest cache derived on, up to 100 MB per FileIO outside the limit |
not derived; `io.manifest.cache-enabled=true` still turns it on |
| Weight-limited REST catalog, `snapshot-loading-mode=refs`, table cache
miss | one extra remote load of the full snapshot list, full history
retained | no extra load; table not retained, statement reads the live
table |
| Paimon table with 301 INT columns, after one scan | 35 KB admitted,
114 KB retained | 134 KB admitted, 114 KB retained |
| Paimon table with a 200-field ROW column, after field lookups | 25 KB
admitted, 117 KB retained | 143 KB admitted, 117 KB retained |
| Paimon partial-update table with 303 INT columns, after one scan |
merge function not reserved | 233 KB admitted, 143 KB retained |
| Paimon table on its own FileIO (REST with or without data tokens,
Hive) | FileIO and token not charged | 16 KB allowance per table |
Across 13 Paimon fixtures (append and primary-key tables, wide schemas,
many options, nested rows, arrays and maps of rows, multisets, wide
partial-update and aggregation tables) the admitted weight without the
FileIO allowance is now 1.2-1.7x the fully grown graph.
Not in this PR: wide schemas that exceed the 10,000-visit budget of
`ReflectiveObjectSizeEstimator.estimateComplete` (for example HMS tables
with about 1,000 primitive columns, or Paimon tables with about 3,400)
are still never admitted under a limit. The same budget is shared by
several connectors and the FE schema cache, so it needs its own change.
**Classes, and how they call each other.**
- `IcebergConnector`: `WEIGHTED_CACHE_ENTRIES`,
`hasMetaCacheWeightLimit()`, and passes the flag at the three catalog
builds.
- `IcebergCatalogFactory`: `buildBaseCatalogProperties`,
`buildCatalogProperties` and `buildS3TablesCatalogProperties` take
`metaCacheWeightLimited` and stay pure.
- `IcebergCatalogProperties`: validates `max-weight` keys against the
shared entry list.
- `IcebergCacheSizeEstimator` / `IcebergTableCache.TableOwner`: snapshot
probe and the guard before serialization.
- `PaimonCacheSizeEstimator`: `estimateStoreGrowth` and
`estimateTableFileIO` / `estimateFileIO`.
```
IcebergConnector.createCatalog / createS3TablesCatalog / createDlfCatalog
└─ hasMetaCacheWeightLimit(): metaCache.hasEnclosingWeightLimit() ||
cacheSpec(entry).isWeightBounded()
└─ IcebergCatalogFactory.build*CatalogProperties(...,
metaCacheWeightLimited)
└─ appendManifestCacheProperties: derive io.manifest.cache-enabled
only when not limited
IcebergTableCache.borrow ─ loader ─ new TableOwner(table, ..., weighted)
└─ areSnapshotsLoaded(table) ? toJson + estimateTable :
incomplete("iceberg_snapshots_not_loaded")
PaimonMetaCacheCatalog "paimon-table" size estimator
└─ PaimonCacheSizeEstimator.estimateTable ─ estimateFileStoreTable (each
concrete table)
├─ estimateStoreGrowth(schema)
└─ estimateTableFileIO(fileIO): fixed allowance, once per FileIO
```
Nothing changes for catalogs without a weight limit: the flag is false,
and the probe and the reservations only run when an entry is weighed.
Independent of #68839 (tests only); both add tests to
`IcebergConnectorCacheTest` and `IcebergTableCacheTest`, in different
places.
Co-authored-by: Calvin Kirs <[email protected]>
---
.../iceberg/IcebergCacheSizeEstimator.java | 31 ++
.../connector/iceberg/IcebergCatalogFactory.java | 50 +--
.../iceberg/IcebergCatalogProperties.java | 2 +-
.../doris/connector/iceberg/IcebergConnector.java | 29 +-
.../doris/connector/iceberg/IcebergTableCache.java | 6 +
.../iceberg/IcebergCatalogFactoryTest.java | 66 ++--
.../iceberg/IcebergCatalogOptionsSnapshotTest.java | 6 +-
.../iceberg/IcebergConnectorCacheTest.java | 21 ++
.../connector/iceberg/IcebergTableCacheTest.java | 84 +++++
.../connector/paimon/PaimonCacheSizeEstimator.java | 113 ++++++-
.../paimon/EstimatorCalibrationAssertions.java | 181 +++++++++++
.../paimon/PaimonCacheSizeEstimatorTest.java | 351 +++++++++++++++++++++
12 files changed, 893 insertions(+), 47 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCacheSizeEstimator.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCacheSizeEstimator.java
index 3b88516b5b5..fdffb7dc184 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCacheSizeEstimator.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCacheSizeEstimator.java
@@ -54,6 +54,7 @@ import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
+import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
@@ -91,6 +92,8 @@ final class IcebergCacheSizeEstimator {
private static final long HASH_MAP_NODE_SHALLOW_BYTES =
classSize("java.util.HashMap$Node");
private static final long LINKED_HASH_MAP_ENTRY_SHALLOW_BYTES =
classSize("java.util.LinkedHashMap$Entry");
private static final long CONTENT_FILE_SCHEMA_BYTES =
estimateContentFileSchema();
+ // TableMetadata does not expose whether a lazily supplied snapshot list
has been fetched yet.
+ private static final Field TABLE_METADATA_SNAPSHOTS_LOADED =
tableMetadataField("snapshotsLoaded");
private IcebergCacheSizeEstimator() {
}
@@ -115,6 +118,24 @@ final class IcebergCacheSizeEstimator {
return add(bytes, estimateTableMetadata(operations.current()));
}
+ /**
+ * Whether the table's snapshot list is materialized, so serializing or
weighing its metadata does no catalog IO.
+ * A REST catalog with {@code snapshot-loading-mode=refs} returns metadata
holding only the referenced snapshots
+ * plus a supplier that loads the complete list remotely on the first
{@code snapshots()} call. A table without
+ * {@link TableOperations} has no {@link TableMetadata}, so nothing of it
is loaded lazily.
+ */
+ static boolean areSnapshotsLoaded(Table table) {
+ if (!(table instanceof HasTableOperations)) {
+ return true;
+ }
+ TableMetadata metadata = ((HasTableOperations)
table).operations().current();
+ try {
+ return TABLE_METADATA_SNAPSHOTS_LOADED.getBoolean(metadata);
+ } catch (IllegalAccessException e) {
+ throw new IllegalStateException("Can not read the snapshot state
of Iceberg table metadata", e);
+ }
+ }
+
static long estimateSerializedTableMetadata(String metadataJson) {
return JvmSizeUtils.stringSize(metadataJson);
}
@@ -728,6 +749,16 @@ final class IcebergCacheSizeEstimator {
}
}
+ private static Field tableMetadataField(String name) {
+ try {
+ Field field = TableMetadata.class.getDeclaredField(name);
+ field.setAccessible(true);
+ return field;
+ } catch (NoSuchFieldException e) {
+ throw new IllegalStateException("Required Iceberg TableMetadata
field is missing: " + name, e);
+ }
+ }
+
private static long add(long left, long right) {
return JvmSizeUtils.saturatedAdd(left, right);
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java
index 054bfb76ea8..6374647c60f 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java
@@ -142,16 +142,18 @@ public final class IcebergCatalogFactory {
* raw props (legacy {@code getOrigProps()} copy-all — arbitrary
user/iceberg keys pass through to
* the SDK; the per-flavor appenders + the connector add the derived keys
on top), (2) map
* {@code warehouse} to {@link CatalogProperties#WAREHOUSE_LOCATION}, (3)
add manifest-cache keys.
- * PURE: depends only on {@code props}. The flavor's {@code catalog-impl}
and the {@code type}
+ * PURE: depends only on its arguments; {@code metaCacheWeightLimited} is
+ * {@code IcebergConnector#hasMetaCacheWeightLimit}. The flavor's {@code
catalog-impl} and the {@code type}
* removal are applied by the caller (the connector / per-flavor path).
*/
- public static Map<String, String> buildBaseCatalogProperties(Map<String,
String> props) {
+ public static Map<String, String> buildBaseCatalogProperties(Map<String,
String> props,
+ boolean metaCacheWeightLimited) {
Map<String, String> opts = new HashMap<>(props);
String warehouse = props.get(CatalogProperties.WAREHOUSE_LOCATION);
if (StringUtils.isNotBlank(warehouse)) {
opts.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse);
}
- appendManifestCacheProperties(props, opts);
+ appendManifestCacheProperties(props, opts, metaCacheWeightLimited);
return opts;
}
@@ -161,21 +163,31 @@ public final class IcebergCatalogFactory {
* {@code io.manifest.cache-enabled} directly — derive it to {@code
"true"} from the FE meta-cache
* spec ({@code meta.cache.iceberg.manifest.*}) using the same {@code
enable && ttl != 0 &&
* capacity != 0} rule. Default-disabled (legacy {@code
DEFAULT_ICEBERG_MANIFEST_CACHE_ENABLE}).
+ * Under a metadata cache weight limit the SDK cache, which keeps up to
+ * {@code io.manifest.cache.max-total-bytes} of manifest content per
FileIO outside that limit, is not
+ * derived but switched off explicitly: an Iceberg REST server's {@code
/config} defaults fill in a key the
+ * client leaves out. The server's overrides and a table's own config
still take precedence.
*/
- private static void appendManifestCacheProperties(Map<String, String>
props, Map<String, String> opts) {
+ private static void appendManifestCacheProperties(Map<String, String>
props, Map<String, String> opts,
+ boolean metaCacheWeightLimited) {
boolean hasExplicitEnabled =
StringUtils.isNotBlank(props.get(CatalogProperties.IO_MANIFEST_CACHE_ENABLED));
copyIfPresent(props, opts,
CatalogProperties.IO_MANIFEST_CACHE_ENABLED);
copyIfPresent(props, opts,
CatalogProperties.IO_MANIFEST_CACHE_EXPIRATION_INTERVAL_MS);
copyIfPresent(props, opts,
CatalogProperties.IO_MANIFEST_CACHE_MAX_TOTAL_BYTES);
copyIfPresent(props, opts,
CatalogProperties.IO_MANIFEST_CACHE_MAX_CONTENT_LENGTH);
- if (!hasExplicitEnabled) {
- CacheSpec spec = CacheSpec.fromProperties(props,
- IcebergConnector.MANIFEST_CACHE_ENABLE,
DEFAULT_MANIFEST_CACHE_ENABLE,
- IcebergConnector.MANIFEST_CACHE_TTL_SECOND,
DEFAULT_MANIFEST_CACHE_TTL_SECOND,
- IcebergConnector.MANIFEST_CACHE_CAPACITY,
DEFAULT_MANIFEST_CACHE_CAPACITY);
- if (CacheSpec.isCacheEnabled(spec.isEnable(), spec.getTtlSecond(),
spec.getCapacity())) {
- opts.put(CatalogProperties.IO_MANIFEST_CACHE_ENABLED, "true");
- }
+ if (hasExplicitEnabled) {
+ return;
+ }
+ if (metaCacheWeightLimited) {
+ opts.put(CatalogProperties.IO_MANIFEST_CACHE_ENABLED, "false");
+ return;
+ }
+ CacheSpec spec = CacheSpec.fromProperties(props,
+ IcebergConnector.MANIFEST_CACHE_ENABLE,
DEFAULT_MANIFEST_CACHE_ENABLE,
+ IcebergConnector.MANIFEST_CACHE_TTL_SECOND,
DEFAULT_MANIFEST_CACHE_TTL_SECOND,
+ IcebergConnector.MANIFEST_CACHE_CAPACITY,
DEFAULT_MANIFEST_CACHE_CAPACITY);
+ if (CacheSpec.isCacheEnabled(spec.isEnable(), spec.getTtlSecond(),
spec.getCapacity())) {
+ opts.put(CatalogProperties.IO_MANIFEST_CACHE_ENABLED, "true");
}
}
@@ -376,7 +388,8 @@ public final class IcebergCatalogFactory {
* the common base (copy-all + warehouse + manifest cache), the flavor's
{@code catalog-impl}, the
* per-flavor derivations, the S3FileIO dialect (for rest/hadoop/jdbc;
glue emits its own), the jdbc
* {@code catalog_name} positional removal, and finally the removal of the
{@code type} key (the iceberg SDK
- * forbids both {@code type} and {@code catalog-impl}). PURE: a function
of {@code props} + {@code chosenS3}.
+ * forbids both {@code type} and {@code catalog-impl}). PURE: a function
of {@code props} + {@code chosenS3} +
+ * {@code metaCacheWeightLimited} (see {@link
#buildBaseCatalogProperties}).
*
* <p>The metastore connection (HMS {@code HiveConf}) and storage {@code
Configuration} are SEPARATE sinks
* built by the connector ({@link #assembleHiveConf} / {@link
#buildHadoopConfiguration}); they are not part
@@ -384,10 +397,10 @@ public final class IcebergCatalogFactory {
* impl only here (the existing skeleton behavior), so this method covers
exactly the five SDK-built flavors.
*/
public static Map<String, String>
buildCatalogProperties(IcebergCatalogProperties catalogProps,
- Optional<S3CompatibleFileSystemProperties> chosenS3) {
+ Optional<S3CompatibleFileSystemProperties> chosenS3, boolean
metaCacheWeightLimited) {
Map<String, String> props = catalogProps.getRaw();
String flavor = catalogProps.getFlavor();
- Map<String, String> opts = buildBaseCatalogProperties(props);
+ Map<String, String> opts = buildBaseCatalogProperties(props,
metaCacheWeightLimited);
opts.put(CatalogProperties.CATALOG_IMPL, resolveCatalogImpl(flavor));
switch (flavor) {
case IcebergCatalogProperties.TYPE_REST:
@@ -431,7 +444,8 @@ public final class IcebergCatalogFactory {
* plus the {@code S3FileIO} dialect ({@code client.region} + {@code
s3.*}) and the EXPLICIT-wins credential
* block ({@link #appendS3TablesFileIOProperties}) — which, unlike the
generic rest/hadoop/jdbc FileIO path,
* suppresses the assume-role keys when static AK/SK are present (legacy
{@code putS3FileIOCredentialProperties}
- * returns early for EXPLICIT). PURE: a function of {@code props} + {@code
chosenS3}.
+ * returns early for EXPLICIT). PURE: a function of {@code props} + {@code
chosenS3} +
+ * {@code metaCacheWeightLimited} (see {@link
#buildBaseCatalogProperties}).
*
* <p>Unlike {@link #buildCatalogProperties}, this does NOT add a {@code
catalog-impl} and does NOT remove the
* {@code type} key: s3tables is built by the connector via {@code new
S3TablesCatalog().initialize(name, opts,
@@ -442,9 +456,9 @@ public final class IcebergCatalogFactory {
* connector ({@code IcebergConnector.buildS3TablesClient}); it is not
part of this pure options map.
*/
public static Map<String, String>
buildS3TablesCatalogProperties(IcebergCatalogProperties catalogProps,
- Optional<S3CompatibleFileSystemProperties> chosenS3) {
+ Optional<S3CompatibleFileSystemProperties> chosenS3, boolean
metaCacheWeightLimited) {
Map<String, String> props = catalogProps.getRaw();
- Map<String, String> opts = buildBaseCatalogProperties(props);
+ Map<String, String> opts = buildBaseCatalogProperties(props,
metaCacheWeightLimited);
if (chosenS3.isPresent()) {
appendS3TablesFileIOProperties(opts, chosenS3.get(), props);
} else {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogProperties.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogProperties.java
index 0774e131a16..fe9844a82c3 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogProperties.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogProperties.java
@@ -163,7 +163,7 @@ public final class IcebergCatalogProperties {
private static void checkMetaCacheProperties(
Map<String, String> properties, Map<String, String>
submittedProperties) {
CacheSpec.checkWeightProperties(properties, submittedProperties,
"iceberg",
- "table", "partition", "manifest", "partition_view");
+ IcebergConnector.WEIGHTED_CACHE_ENTRIES);
CacheSpec.checkBooleanProperty(properties.get(IcebergConnector.TABLE_CACHE_ENABLE),
IcebergConnector.TABLE_CACHE_ENABLE);
CacheSpec.checkLongProperty(properties.get(IcebergConnector.TABLE_CACHE_TTL_SECOND),
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
index d26b359c601..8e910135238 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
@@ -161,6 +161,8 @@ public class IcebergConnector implements Connector {
static final String MANIFEST_CACHE_CAPACITY =
"meta.cache.iceberg.manifest.capacity";
static final long DEFAULT_TABLE_CACHE_TTL_SECOND = 86400L;
static final int DEFAULT_TABLE_CACHE_CAPACITY = 1000;
+ // Entries whose meta.cache.iceberg.<entry>.max-weight this connector
reads.
+ static final String[] WEIGHTED_CACHE_ENTRIES = {"table", "partition",
"manifest", "partition_view"};
// Doris storage property keys (mirror StorageProperties without a fe-core
dependency).
// Catalog property key gating the plugin-side Kerberos authenticator
(value matches AuthType.KERBEROS).
@@ -370,6 +372,24 @@ public class IcebergConnector implements Connector {
return CacheSpec.fromProperties(cacheProperties, "iceberg", entryName,
defaults);
}
+ /**
+ * Whether a metadata cache weight limit applies to this catalog: the
FE-global limit, the catalog's
+ * {@code meta.cache.max-weight}, or the {@code max-weight} of one of
{@link #WEIGHTED_CACHE_ENTRIES}. The
+ * catalog options then leave the Iceberg SDK manifest content cache off
unless the user enables it, because
+ * that cache keeps manifests per FileIO outside every Doris limit.
+ */
+ boolean hasMetaCacheWeightLimit() {
+ if (metaCache.hasEnclosingWeightLimit()) {
+ return true;
+ }
+ for (String entry : WEIGHTED_CACHE_ENTRIES) {
+ if (cacheSpec(entry).isWeightBounded()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@Override
public ConnectorMetadata getMetadata(ConnectorSession session) {
return new IcebergConnectorMetadata(newCatalogBackedOps(session),
catalogProps, context,
@@ -1032,7 +1052,7 @@ public class IcebergConnector implements Connector {
}
Map<String, String> catalogOptions =
- IcebergCatalogFactory.buildCatalogProperties(catalogProps,
chosenS3);
+ IcebergCatalogFactory.buildCatalogProperties(catalogProps,
chosenS3, hasMetaCacheWeightLimit());
Map<String, String> storageHadoopConfig = buildStorageHadoopConfig();
Configuration conf;
@@ -1107,7 +1127,8 @@ public class IcebergConnector implements Connector {
DlfMetaStoreProperties dlf = (DlfMetaStoreProperties)
MetaStoreProviders.bindForType(
IcebergCatalogProperties.TYPE_DLF, properties,
buildStorageHadoopConfig());
Configuration conf =
IcebergCatalogFactory.buildDlfConfiguration(dlf.toDlfCatalogConf());
- Map<String, String> catalogOptions =
IcebergCatalogFactory.buildBaseCatalogProperties(properties);
+ Map<String, String> catalogOptions =
+ IcebergCatalogFactory.buildBaseCatalogProperties(properties,
hasMetaCacheWeightLimit());
return buildCatalogAuthenticated(IcebergCatalogProperties.TYPE_DLF, ()
-> {
DLFCatalog catalog = new DLFCatalog(chosenS3.get());
catalog.setConf(conf);
@@ -1232,8 +1253,8 @@ public class IcebergConnector implements Connector {
*/
private Catalog createS3TablesCatalog(String catalogName,
Optional<S3CompatibleFileSystemProperties> chosenS3) {
String region = resolveS3TablesRegion(chosenS3, properties);
- Map<String, String> catalogOptions =
-
IcebergCatalogFactory.buildS3TablesCatalogProperties(catalogProps, chosenS3);
+ Map<String, String> catalogOptions =
IcebergCatalogFactory.buildS3TablesCatalogProperties(
+ catalogProps, chosenS3, hasMetaCacheWeightLimit());
LOG.info("Creating Iceberg s3tables catalog '{}' region='{}'
boundStorage={}",
catalogName, region, chosenS3.isPresent());
return
buildCatalogAuthenticated(IcebergCatalogProperties.TYPE_S3_TABLES, () -> {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergTableCache.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergTableCache.java
index 43be412bd04..5526a52587d 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergTableCache.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergTableCache.java
@@ -270,6 +270,12 @@ final class IcebergTableCache {
this.sizeEstimate = estimateWeight
?
MetaCacheSizeEstimator.estimateSafely("iceberg_table_estimator_failure",
() -> {
+ if
(!IcebergCacheSizeEstimator.areSnapshotsLoaded(table)) {
+ // Serializing or weighing a lazily
supplied snapshot list would fetch every
+ // snapshot remotely on each miss. Decline
retention; borrowers keep reading
+ // the live table, whose supplier still
serves historical snapshots.
+ return
MetaCacheSizeEstimate.incomplete("iceberg_snapshots_not_loaded");
+ }
long serializedMetadataBytes = 0L;
if (table instanceof BaseTable) {
TableMetadata metadata = ((BaseTable)
table).operations().current();
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java
index 26921878e6b..6a3f31dbb1f 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java
@@ -27,6 +27,7 @@ import
org.apache.doris.filesystem.properties.StorageProperties;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.iceberg.aws.AwsClientProperties;
+import org.apache.iceberg.rest.responses.ConfigResponse;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -177,7 +178,7 @@ public class IcebergCatalogFactoryTest {
// CatalogProperties.WAREHOUSE_LOCATION ("warehouse"). MUTATION:
dropping the copy-all (selective
// re-key) loses "foo"; a wrong warehouse key loses "warehouse" -> red.
Map<String, String> opts =
IcebergCatalogFactory.buildBaseCatalogProperties(
- props("iceberg.catalog.type", "hadoop", "warehouse",
"s3://b/wh", "foo", "bar"));
+ props("iceberg.catalog.type", "hadoop", "warehouse",
"s3://b/wh", "foo", "bar"), false);
Assertions.assertEquals("bar", opts.get("foo"));
Assertions.assertEquals("s3://b/wh", opts.get("warehouse"));
}
@@ -188,7 +189,7 @@ public class IcebergCatalogFactoryTest {
// io.manifest.cache-enabled and no
meta.cache.iceberg.manifest.enable, the key must stay ABSENT
// (not default-on). MUTATION: unconditionally putting "true" -> red.
Map<String, String> opts =
IcebergCatalogFactory.buildBaseCatalogProperties(
- props("iceberg.catalog.type", "rest"));
+ props("iceberg.catalog.type", "rest"), false);
Assertions.assertNull(opts.get("io.manifest.cache-enabled"));
}
@@ -199,7 +200,7 @@ public class IcebergCatalogFactoryTest {
// The key is DOTTED ("io.manifest.cache-enabled"); the recon agent
guessed a hyphenated spelling.
// MUTATION: wrong key spelling OR skipping the derivation -> red.
Map<String, String> opts =
IcebergCatalogFactory.buildBaseCatalogProperties(
- props("iceberg.catalog.type", "rest",
"meta.cache.iceberg.manifest.enable", "true"));
+ props("iceberg.catalog.type", "rest",
"meta.cache.iceberg.manifest.enable", "true"), false);
Assertions.assertEquals("true", opts.get("io.manifest.cache-enabled"));
}
@@ -210,7 +211,7 @@ public class IcebergCatalogFactoryTest {
// the derivation overwrite it to "true" -> red.
Map<String, String> opts =
IcebergCatalogFactory.buildBaseCatalogProperties(
props("iceberg.catalog.type", "rest",
"io.manifest.cache-enabled", "false",
- "meta.cache.iceberg.manifest.enable", "true"));
+ "meta.cache.iceberg.manifest.enable", "true"), false);
Assertions.assertEquals("false",
opts.get("io.manifest.cache-enabled"));
}
@@ -221,10 +222,39 @@ public class IcebergCatalogFactoryTest {
// MUTATION: deriving on enable alone (ignoring ttl/capacity==0) ->
"true" -> red.
Map<String, String> opts =
IcebergCatalogFactory.buildBaseCatalogProperties(
props("iceberg.catalog.type", "rest",
"meta.cache.iceberg.manifest.enable", "true",
- "meta.cache.iceberg.manifest.ttl-second", "0"));
+ "meta.cache.iceberg.manifest.ttl-second", "0"), false);
Assertions.assertNull(opts.get("io.manifest.cache-enabled"));
}
+ @Test
+ public void metaCacheWeightLimitTurnsTheSdkManifestCacheOff() {
+ // WHY: the SDK manifest content cache keeps up to
io.manifest.cache.max-total-bytes of manifests per FileIO
+ // outside every Doris metadata cache weight limit, so under a limit
it is not derived from
+ // meta.cache.iceberg.manifest.* (branch-4.x
AbstractIcebergProperties) but switched off outright, because
+ // an Iceberg REST server's /config defaults fill in a key the client
leaves out. An explicit user value
+ // still wins. MUTATION: ignoring the limit flag -> "true" is derived
-> red; leaving the key out -> the
+ // REST server's default turns the cache on -> red.
+ Map<String, String> metaCacheOnly = props("iceberg.catalog.type",
"hadoop", "warehouse", "s3://b/wh",
+ "meta.cache.iceberg.manifest.enable", "true");
+ Assertions.assertEquals("false",
IcebergCatalogFactory.buildBaseCatalogProperties(metaCacheOnly, true)
+ .get("io.manifest.cache-enabled"));
+ Assertions.assertEquals("false",
IcebergCatalogFactory.buildCatalogProperties(
+ IcebergCatalogProperties.of(metaCacheOnly), Optional.empty(),
true).get("io.manifest.cache-enabled"));
+
+ Map<String, String> rest =
IcebergCatalogFactory.buildCatalogProperties(
+ IcebergCatalogProperties.of(props("iceberg.catalog.type",
"rest", "uri", "https://rest";)),
+ Optional.empty(), true);
+ ConfigResponse serverDefaultsOn = ConfigResponse.builder()
+ .withDefault("io.manifest.cache-enabled", "true").build();
+ Assertions.assertEquals("false",
serverDefaultsOn.merge(rest).get("io.manifest.cache-enabled"),
+ "a REST server's /config default must not turn the SDK
manifest cache back on");
+
+ Map<String, String> explicit = new HashMap<>(metaCacheOnly);
+ explicit.put("io.manifest.cache-enabled", "true");
+ Assertions.assertEquals("true",
IcebergCatalogFactory.buildBaseCatalogProperties(explicit, true)
+ .get("io.manifest.cache-enabled"));
+ }
+
// ---------------------------------------------------------------------
// appendS3FileIOProperties — storage creds -> iceberg S3FileIO dialect
(D-061)
// ---------------------------------------------------------------------
@@ -471,7 +501,7 @@ public class IcebergCatalogFactoryTest {
.accessKey("OSS_AK")
.secretKey("OSS_SK")
.sessionToken("OSS_TOKEN")
- .usePathStyle("false")));
+ .usePathStyle("false")), false);
Assertions.assertEquals("https://cn-hangzhou.oss-tables.aliyuncs.com/iceberg";,
opts.get("uri"));
Assertions.assertEquals("acs:osstables:cn-hangzhou:1234567890:bucket/my-table-bucket",
@@ -764,7 +794,7 @@ public class IcebergCatalogFactoryTest {
// iceberg.catalog.type key is a separate raw key carried by copy-all
and is harmless. MUTATION: not
// setting impl, or leaving "type" -> red.
Map<String, String> opts =
IcebergCatalogFactory.buildCatalogProperties(
- IcebergCatalogProperties.of(props("iceberg.catalog.type",
"hadoop", "warehouse", "s3://b/wh", "type", "hadoop")), Optional.empty());
+ IcebergCatalogProperties.of(props("iceberg.catalog.type",
"hadoop", "warehouse", "s3://b/wh", "type", "hadoop")), Optional.empty(),
false);
Assertions.assertEquals("org.apache.iceberg.hadoop.HadoopCatalog",
opts.get("catalog-impl"));
Assertions.assertNull(opts.get("type"), "the SDK 'type' key must be
removed before building");
}
@@ -775,7 +805,7 @@ public class IcebergCatalogFactoryTest {
// initCatalog. MUTATION: leaving it in the map -> red (iceberg would
treat it as an unknown option).
Map<String, String> opts =
IcebergCatalogFactory.buildCatalogProperties(
IcebergCatalogProperties.of(props("iceberg.catalog.type",
"jdbc", "uri", "jdbc:mysql://h/db", "warehouse", "s3://b/wh",
- "iceberg.jdbc.catalog_name", "mycat")),
Optional.empty());
+ "iceberg.jdbc.catalog_name", "mycat")),
Optional.empty(), false);
Assertions.assertEquals("org.apache.iceberg.jdbc.JdbcCatalog",
opts.get("catalog-impl"));
Assertions.assertNull(opts.get("iceberg.jdbc.catalog_name"),
"the jdbc catalog_name must be consumed positionally, not left
in the options map");
@@ -787,7 +817,7 @@ public class IcebergCatalogFactoryTest {
// the s3.* options. MUTATION: appending the base S3FileIO for HMS ->
s3.endpoint present -> red.
Map<String, String> opts =
IcebergCatalogFactory.buildCatalogProperties(
IcebergCatalogProperties.of(props("iceberg.catalog.type",
"hms")),
- Optional.of(new
FakeS3CompatibleStorageProperties("S3").endpoint("https://s3";).accessKey("AK")));
+ Optional.of(new
FakeS3CompatibleStorageProperties("S3").endpoint("https://s3";).accessKey("AK")),
false);
Assertions.assertEquals(DorisHiveCatalog.class.getName(),
opts.get("catalog-impl"));
Assertions.assertNull(opts.get("s3.endpoint"), "HMS must not emit
S3FileIO options");
Assertions.assertNull(opts.get("s3.access-key-id"));
@@ -803,7 +833,7 @@ public class IcebergCatalogFactoryTest {
Map<String, String> opts =
IcebergCatalogFactory.buildCatalogProperties(
IcebergCatalogProperties.of(props("iceberg.catalog.type",
"rest", "uri", "https://rest";,
"iceberg.rest.vended-credentials-enabled", "true",
"s3.endpoint", "https://minio:9000";,
- "s3.region", "us-east-1")), Optional.empty());
+ "s3.region", "us-east-1")), Optional.empty(), false);
Assertions.assertEquals("us-east-1", opts.get("client.region"),
"vended REST (no bound S3) must still translate s3.region ->
client.region");
}
@@ -818,14 +848,14 @@ public class IcebergCatalogFactoryTest {
// iceberg.rest.signing-region is absent from the narrow 4-alias set
-> client.region null.
Map<String, String> viaAwsRegion =
IcebergCatalogFactory.buildCatalogProperties(
IcebergCatalogProperties.of(props("iceberg.catalog.type",
"rest", "uri", "https://rest";,
- "iceberg.rest.vended-credentials-enabled", "true",
"AWS_REGION", "us-east-1")), Optional.empty());
+ "iceberg.rest.vended-credentials-enabled", "true",
"AWS_REGION", "us-east-1")), Optional.empty(), false);
Assertions.assertEquals("us-east-1", viaAwsRegion.get("client.region"),
"region supplied only via AWS_REGION must translate to
client.region");
Map<String, String> viaSigningRegion =
IcebergCatalogFactory.buildCatalogProperties(
IcebergCatalogProperties.of(props("iceberg.catalog.type",
"rest", "uri", "https://rest";,
"iceberg.rest.vended-credentials-enabled", "true",
- "iceberg.rest.signing-region", "eu-west-1")),
Optional.empty());
+ "iceberg.rest.signing-region", "eu-west-1")),
Optional.empty(), false);
Assertions.assertEquals("eu-west-1",
viaSigningRegion.get("client.region"),
"region supplied only via iceberg.rest.signing-region must
translate to client.region");
}
@@ -845,7 +875,7 @@ public class IcebergCatalogFactoryTest {
"warehouse", "arn:aws:s3tables:us-east-1:1:bucket/b")),
Optional.of(new FakeS3CompatibleStorageProperties("S3")
.endpoint("https://s3.us-east-1.amazonaws.com";).region("us-east-1")
-
.accessKey("AK").secretKey("SK").sessionToken("TK").usePathStyle("true")));
+
.accessKey("AK").secretKey("SK").sessionToken("TK").usePathStyle("true")),
false);
Assertions.assertEquals("arn:aws:s3tables:us-east-1:1:bucket/b",
opts.get("warehouse"),
"the table-bucket ARN warehouse must be carried through for
the 3-arg initialize");
Assertions.assertEquals("AK", opts.get("s3.access-key-id"));
@@ -869,7 +899,7 @@ public class IcebergCatalogFactoryTest {
Map<String, String> opts =
IcebergCatalogFactory.buildS3TablesCatalogProperties(
IcebergCatalogProperties.of(props("iceberg.catalog.type",
"s3tables", "warehouse", "arn:aws:s3tables:us-west-2:1:bucket/b")),
Optional.of(new
FakeS3CompatibleStorageProperties("S3").region("us-west-2")
- .roleArn("arn:aws:iam::1:role/r").externalId("eid")));
+ .roleArn("arn:aws:iam::1:role/r").externalId("eid")),
false);
Assertions.assertEquals("org.apache.iceberg.aws.AssumeRoleAwsClientFactory",
opts.get("client.factory"));
Assertions.assertEquals("arn:aws:iam::1:role/r",
opts.get("client.assume-role.arn"));
Assertions.assertEquals("us-west-2",
opts.get("client.assume-role.region"));
@@ -886,7 +916,7 @@ public class IcebergCatalogFactoryTest {
Map<String, String> opts =
IcebergCatalogFactory.buildS3TablesCatalogProperties(
IcebergCatalogProperties.of(props("iceberg.catalog.type",
"s3tables", "warehouse", "arn:aws:s3tables:us-west-2:1:bucket/b",
"s3.credentials_provider_type", "ENV")),
- Optional.of(new
FakeS3CompatibleStorageProperties("S3").region("us-west-2")));
+ Optional.of(new
FakeS3CompatibleStorageProperties("S3").region("us-west-2")), false);
Assertions.assertEquals(EnvironmentVariableCredentialsProvider.class.getName(),
opts.get("client.credentials-provider"));
Assertions.assertNull(opts.get("client.factory"), "no role -> no
assume-role block");
@@ -902,7 +932,7 @@ public class IcebergCatalogFactoryTest {
Map<String, String> opts =
IcebergCatalogFactory.buildS3TablesCatalogProperties(
IcebergCatalogProperties.of(props("iceberg.catalog.type",
"s3tables", "warehouse", "arn:aws:s3tables:us-west-2:1:bucket/b")),
Optional.of(new
FakeS3CompatibleStorageProperties("S3").region("us-west-2")
-
.accessKey("AK").secretKey("SK").roleArn("arn:aws:iam::1:role/r")));
+
.accessKey("AK").secretKey("SK").roleArn("arn:aws:iam::1:role/r")), false);
Assertions.assertEquals("AK", opts.get("s3.access-key-id"));
Assertions.assertEquals("SK", opts.get("s3.secret-access-key"));
Assertions.assertNull(opts.get("client.factory"),
@@ -918,7 +948,7 @@ public class IcebergCatalogFactoryTest {
//
buildS3TablesCatalogPropertiesPropagatesClientRegionWithoutBoundS3.) MUTATION:
fabricating any s3.* -> red.
Map<String, String> opts =
IcebergCatalogFactory.buildS3TablesCatalogProperties(
IcebergCatalogProperties.of(props("iceberg.catalog.type",
"s3tables", "warehouse", "arn:aws:s3tables:us-east-1:1:bucket/b")),
- Optional.empty());
+ Optional.empty(), false);
Assertions.assertEquals("arn:aws:s3tables:us-east-1:1:bucket/b",
opts.get("warehouse"));
Assertions.assertNull(opts.get("s3.access-key-id"));
Assertions.assertNull(opts.get("client.region"));
@@ -934,7 +964,7 @@ public class IcebergCatalogFactoryTest {
Map<String, String> opts =
IcebergCatalogFactory.buildS3TablesCatalogProperties(
IcebergCatalogProperties.of(props("iceberg.catalog.type",
"s3tables", "warehouse", "arn:aws:s3tables:us-east-1:1:bucket/b",
"s3.region", "us-east-1")),
- Optional.empty());
+ Optional.empty(), false);
Assertions.assertEquals("us-east-1", opts.get("client.region"),
"no-storage s3tables must propagate s3.region -> client.region
for the data-plane S3FileIO");
Assertions.assertNull(opts.get("s3.access-key-id"), "no credentials
are bound");
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogOptionsSnapshotTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogOptionsSnapshotTest.java
index c0f918d29a2..cd3b22b0d4f 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogOptionsSnapshotTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogOptionsSnapshotTest.java
@@ -64,7 +64,7 @@ public class IcebergCatalogOptionsSnapshotTest {
private static void assertOptions(Map<String, String> expected,
Map<String, String> input,
Optional<S3CompatibleFileSystemProperties> chosenS3) {
Assertions.assertEquals(new TreeMap<>(expected),
- new
TreeMap<>(IcebergCatalogFactory.buildCatalogProperties(IcebergCatalogProperties.of(input),
chosenS3)));
+ new
TreeMap<>(IcebergCatalogFactory.buildCatalogProperties(IcebergCatalogProperties.of(input),
chosenS3, false)));
}
private static Optional<S3CompatibleFileSystemProperties>
s3(FakeS3CompatibleStorageProperties fake) {
@@ -586,7 +586,7 @@ public class IcebergCatalogOptionsSnapshotTest {
"warehouse", "arn:aws:s3tables:us-east-1:1:bucket/b",
"s3.role_arn", "arn:aws:iam::1:role/ignored")),
Optional.of(new
FakeS3CompatibleStorageProperties("S3").region("us-east-1")
-
.accessKey("AK").secretKey("SK").roleArn("arn:aws:iam::1:role/ignored")));
+
.accessKey("AK").secretKey("SK").roleArn("arn:aws:iam::1:role/ignored")),
false);
Assertions.assertEquals(new TreeMap<>(expected), new
TreeMap<>(actual));
}
@@ -601,7 +601,7 @@ public class IcebergCatalogOptionsSnapshotTest {
IcebergCatalogProperties.of(props("iceberg.catalog.type",
"s3tables",
"warehouse", "arn:aws:s3tables:us-east-1:1:bucket/b",
"s3.region", "us-east-1")),
- Optional.empty());
+ Optional.empty(), false);
Assertions.assertEquals(new TreeMap<>(expected), new
TreeMap<>(actual));
}
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorCacheTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorCacheTest.java
index ec352b02ca4..74ec5b7589b 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorCacheTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorCacheTest.java
@@ -129,6 +129,27 @@ public class IcebergConnectorCacheTest {
new
RecordingConnectorContext()).schemaCacheTtlSecondOverride());
}
+ @Test
+ public void metaCacheWeightLimitSeesCatalogAndEveryConsumedEntryLimit()
throws Exception {
+ // The SDK manifest content cache is derived on only when no Doris
weight limit applies to the catalog. Keys
+ // that create no budget (an entry this connector does not read, an
unparsable persisted value) do not count.
+ // MUTATION: checking only the catalog limit, or dropping an entry
from the list -> red.
+ assertMetaCacheWeightLimit(false, Collections.emptyMap());
+ assertMetaCacheWeightLimit(true, props("meta.cache.max-weight",
"128MB"));
+ for (String entry : IcebergConnector.WEIGHTED_CACHE_ENTRIES) {
+ assertMetaCacheWeightLimit(true, props("meta.cache.iceberg." +
entry + ".max-weight", "16MB"));
+ }
+ assertMetaCacheWeightLimit(false,
props("meta.cache.iceberg.snapshot.max-weight", "16MB"));
+ assertMetaCacheWeightLimit(false, props("meta.cache.max-weight",
"not-a-size"));
+ }
+
+ private static void assertMetaCacheWeightLimit(boolean expected,
Map<String, String> properties)
+ throws Exception {
+ try (IcebergConnector connector = new IcebergConnector(properties, new
RecordingConnectorContext())) {
+ Assertions.assertEquals(expected,
connector.hasMetaCacheWeightLimit(), properties.toString());
+ }
+ }
+
@Test
public void invalidateHooksAreNoThrowOnFreshConnector() {
// Smoke: the REFRESH TABLE / REFRESH CATALOG hooks must be safe to
call (they only touch the
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTableCacheTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTableCacheTest.java
index 9d1218c9ef5..8ee2b2e6db6 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTableCacheTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTableCacheTest.java
@@ -48,6 +48,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.Collections;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -414,6 +415,89 @@ public class IcebergTableCacheTest {
}
}
+ @Test
+ public void weightedPublicationDoesNotFetchLazilySuppliedSnapshots() {
+ // REST snapshot-loading-mode=refs keeps the full snapshot list behind
a remote supplier. Serialization and
+ // estimation both read snapshots(), so weighted publication declines
the table instead of fetching the list
+ // on every miss; the borrower still reaches history through the
supplier. MUTATION: dropping the guard ->
+ // the supplier runs during publication -> red.
+ AtomicInteger snapshotLoads = new AtomicInteger();
+ Table table = tableWithLazySnapshots(snapshotLoads);
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ IcebergTableCache cache = new IcebergTableCache(
+ owner, CacheSpec.ofWeight(true, 100L, 1000L, 10L * 1024L *
1024L),
+ ignored -> () -> { }, new IcebergCatalogResourceTracker());
+ try (IcebergTableCache.TableLease lease = cache.borrow(id(), () ->
table)) {
+ Assertions.assertEquals(0, snapshotLoads.get(), "publication
must not fetch the snapshot list");
+ Assertions.assertEquals(0, cache.size());
+
Assertions.assertEquals("incomplete_estimate:iceberg_snapshots_not_loaded",
+
owner.entries().get("iceberg-table").metrics().getLastWeightRejectReason());
+
+ Table statement = lease.snapshotReadTable();
+ Assertions.assertEquals(102L,
statement.currentSnapshot().snapshotId());
+ Assertions.assertEquals(0, snapshotLoads.get());
+ Assertions.assertEquals(101L,
statement.snapshot(101L).snapshotId());
+ Assertions.assertEquals(1, snapshotLoads.get());
+ }
+ // Once the list has been fetched the same metadata is weighed and
retained like any other.
+ try (IcebergTableCache.TableLease lease = cache.borrow(id(), () ->
table)) {
+ Assertions.assertEquals(1, cache.size());
+ }
+ Assertions.assertEquals(1, snapshotLoads.get());
+ }
+ }
+
+ @Test
+ public void
unweightedCacheRetainsLazilySuppliedSnapshotsWithoutFetchingThem() {
+ AtomicInteger snapshotLoads = new AtomicInteger();
+ Table table = tableWithLazySnapshots(snapshotLoads);
+ IcebergTableCache cache = new IcebergTableCache(100L, 1000);
+
+ Assertions.assertSame(table, cache.getOrLoad(id(), () -> table));
+ Assertions.assertEquals(1, cache.size());
+ Assertions.assertEquals(0, snapshotLoads.get());
+ }
+
+ /** What a REST catalog returns under snapshot-loading-mode=refs: the
referenced snapshot plus a supplier. */
+ private static Table tableWithLazySnapshots(AtomicInteger snapshotLoads) {
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ TableMetadata base = TableMetadata.newTableMetadata(
+ schema, PartitionSpec.unpartitioned(),
"file:///tmp/lazy-snapshots-table", Collections.emptyMap());
+ Snapshot older = SnapshotParser.fromJson("{"
+ + "\"sequence-number\":1,"
+ + "\"snapshot-id\":101,"
+ + "\"timestamp-ms\":1000,"
+ + "\"summary\":{\"operation\":\"append\"},"
+ + "\"manifest-list\":\"file:///tmp/snap-101.avro\","
+ + "\"schema-id\":0}");
+ Snapshot newer = SnapshotParser.fromJson("{"
+ + "\"sequence-number\":2,"
+ + "\"snapshot-id\":102,"
+ + "\"parent-snapshot-id\":101,"
+ + "\"timestamp-ms\":2000,"
+ + "\"summary\":{\"operation\":\"append\"},"
+ + "\"manifest-list\":\"file:///tmp/snap-102.avro\","
+ + "\"schema-id\":0}");
+ TableMetadata full = TableMetadata.buildFrom(base)
+ .upgradeFormatVersion(2)
+ .setBranchSnapshot(older, "main")
+ .setBranchSnapshot(newer, "main")
+ .discardChanges()
+ .build();
+ List<Snapshot> allSnapshots = full.snapshots();
+ // The server drops unreferenced snapshots; the client attaches a
supplier that loads all of them.
+ TableMetadata refsOnly = TableMetadata.buildFrom(
+
TableMetadata.buildFrom(full).suppressHistoricalSnapshots().build())
+
.withMetadataLocation("file:///tmp/lazy-snapshots.metadata.json")
+ .setSnapshotsSupplier(() -> {
+ snapshotLoads.incrementAndGet();
+ return allSnapshots;
+ })
+ .discardChanges()
+ .build();
+ return new BaseTable(new StaticTableOperations(refsOnly),
"lazy-snapshots");
+ }
+
@Test
public void ttlZeroDisablesCachingAlwaysLive() {
AtomicInteger loads = new AtomicInteger();
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimator.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimator.java
index 83641254968..e00a1509c69 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimator.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimator.java
@@ -21,9 +21,12 @@ import org.apache.doris.connector.cache.JvmSizeUtils;
import org.apache.doris.connector.cache.MetaCacheSizeEstimate;
import org.apache.doris.connector.cache.ReflectiveObjectSizeEstimator;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.privilege.PrivilegedFileStoreTable;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.CatalogEnvironment;
import org.apache.paimon.table.DelegatedFileStoreTable;
import org.apache.paimon.table.FallbackReadFileStoreTable;
@@ -33,21 +36,60 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.iceberg.IcebergTable;
import org.apache.paimon.table.lance.LanceTable;
import org.apache.paimon.table.object.ObjectTable;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.MultisetType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VectorType;
import java.net.URI;
import java.util.Collections;
import java.util.IdentityHashMap;
+import java.util.List;
import java.util.Set;
/**
* Retained-size formulas for Paimon table-cache entries.
*
* <p>The table's shallow size includes references to FileIO, catalog loaders,
and lock factories,
- * but their graphs are catalog-scoped executable services rather than
entry-owned metadata. Walking
- * those graphs both double-counts shared state and reaches strongly
encapsulated JDK objects. The
- * estimator therefore expands only immutable metadata owned by the entry.
+ * but their graphs are executable services rather than entry-owned metadata.
Walking those graphs
+ * both double-counts shared state and reaches strongly encapsulated JDK
objects. The estimator
+ * therefore expands only immutable metadata owned by the entry, plus a fixed
allowance for a FileIO
+ * the table may own.
+ *
+ * <p>A cached value is weighed once, at admission, but a {@link
FileStoreTable} keeps growing afterwards: the
+ * first {@code latestSnapshot()} or scan builds its transient store (a {@code
FileStore} holding its own
+ * {@code RowType} copies of the schema and a copy of the options) and field
lookups build the four lazy index
+ * maps of every {@code RowType}. That growth is reserved from the schema's
shape. On Paimon 1.4.2 it measures
+ * about 1-2 KB per table, 200-270 bytes per field, another 200 bytes per
field nested in a row, 40 bytes per
+ * option and, for a partial-update table's merge function, about 150 more
bytes per field; the constants below
+ * keep a margin above that.
*/
final class PaimonCacheSizeEstimator {
+ private static final long STORE_GROWTH_BASE_BYTES = 2048L;
+ private static final long STORE_GROWTH_FIELD_BYTES = 320L;
+ private static final long STORE_GROWTH_NESTED_FIELD_BYTES = 256L;
+ // A child type of an array, map, multiset or vector: the store's copy of
it.
+ private static final long STORE_GROWTH_TYPE_NODE_BYTES = 128L;
+ private static final long STORE_GROWTH_OPTION_BYTES = 64L;
+ // A partition, primary or bucket key: its slot in the derived key types;
a primary key is also copied as a
+ // renamed key field.
+ private static final long STORE_GROWTH_KEY_BYTES = 192L;
+ // A partial-update table's merge function factory, built with its store,
keeps sequence-group and aggregator
+ // maps for every field: about 150 bytes per field more than the other
merge engines on Paimon 1.4.2.
+ private static final long STORE_GROWTH_PARTIAL_UPDATE_FIELD_BYTES = 256L;
+ // A table can own its FileIO: a REST catalog gives each table a
RESTTokenFileIO (holding the vended token it
+ // fetches on first data access, an STS key id, secret and security token
of about 2 KB) or a ResolvingFileIO,
+ // and an external table of any kind (format, object, Lance and Iceberg
tables included) always a
+ // ResolvingFileIO; a Hive catalog resolves a FileIO for each table
location. Their tokens and backend maps
+ // appear only after admission, and Paimon offers no IO-free way to read
that state or to tell an owned FileIO
+ // from one the catalog shares. Every table, whatever its kind, is
therefore charged a fixed 16 KB for its
+ // FileIO, several times any STS token seen so far; a token refresh
replaces the token rather than adding to
+ // it. Charging a catalog-shared FileIO to each table only makes the
estimate conservative.
+ private static final long FILE_IO_BYTES = 16L * 1024L;
+
private PaimonCacheSizeEstimator() {
}
@@ -72,6 +114,7 @@ final class PaimonCacheSizeEstimator {
bytes = add(bytes,
ReflectiveObjectSizeEstimator.estimateComplete(table.options()));
bytes = add(bytes,
ReflectiveObjectSizeEstimator.estimateComplete(table.comment()));
bytes = add(bytes, JvmSizeUtils.stringSize(location(table)));
+ bytes = add(bytes, estimateFileIO(table.fileIO()));
}
return MetaCacheSizeEstimate.complete(bytes);
}
@@ -98,10 +141,70 @@ final class PaimonCacheSizeEstimator {
((DelegatedFileStoreTable) table).wrapped(), visited));
}
bytes = add(bytes, estimateCompleteOnce(table.schema(), visited));
+ bytes = add(bytes, estimateStoreGrowth(table.schema()));
+ bytes = add(bytes, estimateTableFileIO(table.fileIO(), visited));
bytes = add(bytes, estimatePath(table.location(), visited));
return add(bytes,
estimateCatalogEnvironment(table.catalogEnvironment(), visited));
}
+ /** Reserves the store and index graph this table builds after admission;
no IO, no store access. */
+ private static long estimateStoreGrowth(TableSchema schema) {
+ long bytes = add(STORE_GROWTH_BASE_BYTES,
+ multiply(schema.options().size(), STORE_GROWTH_OPTION_BYTES));
+ long keys = (long) schema.partitionKeys().size() +
schema.primaryKeys().size() + schema.bucketKeys().size();
+ bytes = add(bytes, multiply(keys, STORE_GROWTH_KEY_BYTES));
+ for (String primaryKey : schema.primaryKeys()) {
+ bytes = add(bytes, JvmSizeUtils.stringSize(primaryKey));
+ }
+ if (CoreOptions.fromMap(schema.options()).mergeEngine() ==
CoreOptions.MergeEngine.PARTIAL_UPDATE) {
+ bytes = add(bytes, multiply(schema.fields().size(),
STORE_GROWTH_PARTIAL_UPDATE_FIELD_BYTES));
+ }
+ return add(bytes, estimateFieldGrowth(schema.fields(),
STORE_GROWTH_FIELD_BYTES));
+ }
+
+ private static long estimateFieldGrowth(List<DataField> fields, long
perFieldBytes) {
+ long bytes = multiply(fields.size(), perFieldBytes);
+ for (DataField field : fields) {
+ bytes = add(bytes, estimateTypeGrowth(field.type()));
+ }
+ return bytes;
+ }
+
+ private static long estimateTypeGrowth(DataType type) {
+ if (type instanceof RowType) {
+ return estimateFieldGrowth(((RowType) type).getFields(),
+ STORE_GROWTH_FIELD_BYTES +
STORE_GROWTH_NESTED_FIELD_BYTES);
+ }
+ if (type instanceof ArrayType) {
+ return estimateChildTypeGrowth(((ArrayType)
type).getElementType());
+ }
+ if (type instanceof MultisetType) {
+ return estimateChildTypeGrowth(((MultisetType)
type).getElementType());
+ }
+ if (type instanceof VectorType) {
+ return estimateChildTypeGrowth(((VectorType)
type).getElementType());
+ }
+ if (type instanceof MapType) {
+ MapType mapType = (MapType) type;
+ return add(estimateChildTypeGrowth(mapType.getKeyType()),
+ estimateChildTypeGrowth(mapType.getValueType()));
+ }
+ return 0L;
+ }
+
+ private static long estimateChildTypeGrowth(DataType child) {
+ return add(STORE_GROWTH_TYPE_NODE_BYTES, estimateTypeGrowth(child));
+ }
+
+ /** The allowance a table carries for its FileIO; the fallback branches of
one table share theirs. */
+ private static long estimateTableFileIO(FileIO fileIO, Set<Object>
visited) {
+ return visited.add(fileIO) ? estimateFileIO(fileIO) : 0L;
+ }
+
+ static long estimateFileIO(FileIO fileIO) {
+ return add(JvmSizeUtils.instanceSize(fileIO.getClass()),
FILE_IO_BYTES);
+ }
+
private static long estimateCompleteOnce(Object value, Set<Object>
visited) {
return value == null || !visited.add(value)
? 0L : ReflectiveObjectSizeEstimator.estimateComplete(value);
@@ -160,4 +263,8 @@ final class PaimonCacheSizeEstimator {
private static long add(long left, long right) {
return JvmSizeUtils.saturatedAdd(left, right);
}
+
+ private static long multiply(long left, long right) {
+ return JvmSizeUtils.saturatedMultiply(left, right);
+ }
}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/EstimatorCalibrationAssertions.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/EstimatorCalibrationAssertions.java
new file mode 100644
index 00000000000..a384fb53eac
--- /dev/null
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/EstimatorCalibrationAssertions.java
@@ -0,0 +1,181 @@
+// 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.JvmSizeUtils;
+
+import org.junit.jupiter.api.Assertions;
+
+import java.lang.reflect.Array;
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Offline calibration checks for the cache weight formulas.
+ *
+ * <p>Estimates are admission weights, not exact retained sizes, so the
assertions verify order-of-magnitude
+ * sanity rather than byte accuracy: a populated fixture must weigh more than
an empty one, the growth must not
+ * be under-estimated by more than a small factor, and it must not be absurdly
over-estimated. The retained-size
+ * oracle walks the object graph with identity dedup and the running JVM's
layout. Unlike
+ * {@code ReflectiveObjectSizeEstimator} it never gives up: JDK collections
and maps, whose fields are strongly
+ * encapsulated without {@code --add-opens}, are walked through their public
API, other inaccessible JDK state
+ * counts as its shallow size, and hidden classes (lambdas) are walked like
any other class.
+ */
+final class EstimatorCalibrationAssertions {
+ // estimatedDelta / actualDelta must stay within this band.
+ private static final double MIN_ESTIMATE_FACTOR = 0.34D;
+ private static final double MAX_ESTIMATE_FACTOR = 12.0D;
+ private static final long HASH_MAP_NODE_BYTES = hashMapNodeBytes();
+ private static final Map<Class<?>, List<Field>> REFERENCE_FIELD_CACHE =
new HashMap<>();
+
+ private EstimatorCalibrationAssertions() {
+ }
+
+ static void assertConservativeDelta(String fixture, long emptyEstimate,
long populatedEstimate,
+ Object emptyGraph, Object populatedGraph) {
+ long actualDelta = graphSize(populatedGraph) - graphSize(emptyGraph);
+ long estimatedDelta = populatedEstimate - emptyEstimate;
+ Assertions.assertTrue(actualDelta > 0L, fixture + " must add retained
heap");
+ Assertions.assertTrue(estimatedDelta > 0L, fixture + " must add
estimated weight");
+ Assertions.assertTrue((double) estimatedDelta >= actualDelta *
MIN_ESTIMATE_FACTOR,
+ fixture + " grossly under-estimates retained heap: estimated="
+ estimatedDelta
+ + ", probed=" + actualDelta);
+ Assertions.assertTrue((double) estimatedDelta <= actualDelta *
MAX_ESTIMATE_FACTOR,
+ fixture + " absurdly over-estimates retained heap: estimated="
+ estimatedDelta
+ + ", probed=" + actualDelta);
+ }
+
+ /**
+ * Retained size of the graph reachable from {@code graph}. Shared JVM
state (classes, class loaders, threads,
+ * enum constants, references) is treated as leaves so deltas between two
fixtures cancel it out.
+ */
+ static synchronized long graphSize(Object graph) {
+ if (graph == null) {
+ return 0L;
+ }
+ IdentityHashMap<Object, Boolean> seen = new IdentityHashMap<>();
+ Deque<Object> pending = new ArrayDeque<>();
+ pending.add(graph);
+ seen.put(graph, Boolean.TRUE);
+ long bytes = 0L;
+ while (!pending.isEmpty()) {
+ Object current = pending.poll();
+ Class<?> type = current.getClass();
+ if (type.isArray()) {
+ int length = Array.getLength(current);
+ Class<?> component = type.getComponentType();
+ if (component.isPrimitive()) {
+ bytes += JvmSizeUtils.primitiveArraySize(component,
length);
+ } else {
+ bytes += JvmSizeUtils.objectArraySize(length);
+ for (int i = 0; i < length; i++) {
+ enqueue(Array.get(current, i), seen, pending);
+ }
+ }
+ continue;
+ }
+ boolean jdkType = type.getModule().isNamed();
+ if (jdkType && current instanceof String) {
+ bytes += JvmSizeUtils.stringSize((String) current);
+ continue;
+ }
+ bytes += JvmSizeUtils.instanceSize(type);
+ if (jdkType && current instanceof Map) {
+ Map<?, ?> map = (Map<?, ?>) current;
+ bytes += JvmSizeUtils.objectArraySize(tableLength(map.size()));
+ for (Map.Entry<?, ?> entry : map.entrySet()) {
+ bytes += HASH_MAP_NODE_BYTES;
+ enqueue(entry.getKey(), seen, pending);
+ enqueue(entry.getValue(), seen, pending);
+ }
+ } else if (jdkType && current instanceof Collection) {
+ Collection<?> collection = (Collection<?>) current;
+ bytes += JvmSizeUtils.objectArraySize(collection.size());
+ for (Object element : collection) {
+ enqueue(element, seen, pending);
+ }
+ } else {
+ for (Field field : referenceFields(type)) {
+ try {
+ field.setAccessible(true);
+ enqueue(field.get(current), seen, pending);
+ } catch (IllegalAccessException | RuntimeException
ignored) {
+ // Inaccessible JDK state counts as its shallow size;
the acceptance band absorbs it.
+ }
+ }
+ }
+ }
+ return bytes;
+ }
+
+ private static void enqueue(Object value, IdentityHashMap<Object, Boolean>
seen, Deque<Object> pending) {
+ if (value == null || isSharedLeaf(value)) {
+ return;
+ }
+ if (seen.put(value, Boolean.TRUE) == null) {
+ pending.add(value);
+ }
+ }
+
+ private static boolean isSharedLeaf(Object value) {
+ return value instanceof Class || value instanceof ClassLoader
+ || value instanceof Thread || value instanceof Enum
+ || value instanceof java.lang.ref.Reference;
+ }
+
+ private static List<Field> referenceFields(Class<?> type) {
+ List<Field> cached = REFERENCE_FIELD_CACHE.get(type);
+ if (cached != null) {
+ return cached;
+ }
+ List<Field> fields = new ArrayList<>();
+ for (Class<?> owner = type; owner != null && owner != Object.class;
owner = owner.getSuperclass()) {
+ for (Field field : owner.getDeclaredFields()) {
+ if (!Modifier.isStatic(field.getModifiers()) &&
!field.getType().isPrimitive()) {
+ fields.add(field);
+ }
+ }
+ }
+ REFERENCE_FIELD_CACHE.put(type, fields);
+ return fields;
+ }
+
+ private static int tableLength(int size) {
+ int length = 1;
+ while (length * 3L < size * 4L) {
+ length <<= 1;
+ }
+ return length;
+ }
+
+ private static long hashMapNodeBytes() {
+ try {
+ return
JvmSizeUtils.instanceSize(Class.forName("java.util.HashMap$Node"));
+ } catch (ClassNotFoundException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimatorTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimatorTest.java
new file mode 100644
index 00000000000..3ba694bc7c1
--- /dev/null
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimatorTest.java
@@ -0,0 +1,351 @@
+// 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.MetaCacheSizeEstimate;
+
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.ResolvingFileIO;
+import org.apache.paimon.fs.hadoop.HadoopFileIO;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.rest.RESTToken;
+import org.apache.paimon.rest.RESTTokenFileIO;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.CatalogEnvironment;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.iceberg.IcebergTable;
+import org.apache.paimon.table.lance.LanceTable;
+import org.apache.paimon.table.object.ObjectTable;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.MultisetType;
+import org.apache.paimon.types.RowType;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+class PaimonCacheSizeEstimatorTest {
+ private static final Identifier TABLE = Identifier.create("db", "t");
+
+ @TempDir
+ java.nio.file.Path warehouse;
+
+ @Test
+ void admissionWeightCoversTheStoreAndIndexesBuiltAfterAdmission() throws
Exception {
+ // A cached table is weighed once, but the first snapshot lookup or
scan builds its store (RowType copies
+ // of the schema, a copy of the options, a partial-update table's
merge function) and field lookups build
+ // every RowType's lazy index maps. The admission weight must already
cover that grown graph without the
+ // FileIO allowance, which pays for the FileIO. MUTATION: dropping the
store-growth reservation -> every
+ // fixture's grown graph exceeds its weight -> red; dropping the
partial-update reservation -> the wide
+ // partial-update fixture does -> red.
+ List<FileStoreTable> fixtures = new ArrayList<>();
+ fixtures.add(table("append_narrow", 10, 0, 0, 0, null));
+ fixtures.add(table("append_wide", 300, 0, 0, 0, null));
+ fixtures.add(table("pk_narrow", 10, 2, 0, 0, null));
+ fixtures.add(table("pk_wide_options", 300, 2, 50, 0, null));
+ fixtures.add(table("pk_many_keys", 64, 30, 8, 0, null));
+ fixtures.add(table("options", 10, 0, 100, 0, null));
+ fixtures.add(table("nested_narrow", 0, 0, 0, 1, null));
+ fixtures.add(table("nested_wide", 0, 0, 0, 200, null));
+ fixtures.add(table("array_of_rows", 0, 0, 0, 0, new
ArrayType(rowOf(20))));
+ fixtures.add(table("map_of_rows", 0, 0, 0, 0, new
MapType(DataTypes.INT(), rowOf(20))));
+ fixtures.add(table("multiset", 0, 0, 0, 0, new
MultisetType(DataTypes.STRING())));
+ fixtures.add(mergeEngineTable("partial_update_wide", 300,
"partial-update"));
+ fixtures.add(mergeEngineTable("aggregation_wide", 300, "aggregation"));
+ for (FileStoreTable table : fixtures) {
+ long weight = weight(table) -
PaimonCacheSizeEstimator.estimateFileIO(table.fileIO());
+ Assertions.assertNull(readField(table, table.getClass(),
"lazyStore"),
+ table.name() + ": estimation must not build the store");
+ grow(table);
+ long grown = EstimatorCalibrationAssertions.graphSize(table);
+ Assertions.assertTrue(weight >= grown,
+ table.name() + ": weight " + weight + " does not cover the
grown graph " + grown);
+ }
+ }
+
+ @Test
+ void storeGrowthReservationScalesWithTheSchema() throws Exception {
+ assertGrownDelta("fields", table("fields_narrow", 10, 0, 0, 0, null),
+ table("fields_wide", 300, 0, 0, 0, null));
+ assertGrownDelta("primary-key fields", table("pk_fields_narrow", 10,
2, 0, 0, null),
+ table("pk_fields_wide", 300, 2, 0, 0, null));
+ assertGrownDelta("options", table("options_none", 10, 0, 0, 0, null),
+ table("options_many", 10, 0, 100, 0, null));
+ assertGrownDelta("nested fields", table("nested_one", 0, 0, 0, 1,
null),
+ table("nested_many", 0, 0, 0, 200, null));
+ }
+
+ @Test
+ void everyTableCarriesAnAllowanceForItsFileIo() throws Exception {
+ // A REST catalog gives each table its own RESTTokenFileIO or
ResolvingFileIO and a Hive catalog a FileIO
+ // per table location; all of them grow after admission, and none can
be told apart from a FileIO the
+ // catalog shares. So a table's weight carries the same allowance
whatever its FileIO is. MUTATION:
+ // charging only RESTTokenFileIO -> the other tables weigh a whole
allowance less -> red.
+ FileStoreTable local = table("file_io", 10, 0, 0, 0, null);
+ CatalogContext context = CatalogContext.create(new Options());
+ ResolvingFileIO resolving = resolvingFileIO(context);
+ HadoopFileIO hadoop = new HadoopFileIO(local.location());
+ hadoop.configure(context);
+ List<FileIO> fileIOs = Arrays.asList(local.fileIO(), resolving, hadoop,
+ new RESTTokenFileIO(context, null, TABLE, local.location()));
+ long withoutFileIO = weight(local) -
PaimonCacheSizeEstimator.estimateFileIO(local.fileIO());
+ for (FileIO fileIO : fileIOs) {
+ FileStoreTable table = FileStoreTableFactory.create(
+ fileIO, local.location(), local.schema(),
CatalogEnvironment.empty());
+ long allowance = PaimonCacheSizeEstimator.estimateFileIO(fileIO);
+ Assertions.assertTrue(allowance >= 16L * 1024L);
+ Assertions.assertEquals(withoutFileIO + allowance, weight(table),
+ fileIO.getClass().getSimpleName() + " must be charged to
its table");
+ }
+
+ // A ResolvingFileIO builds its backend on the first data access,
after admission.
+ resolving.exists(local.location());
+ long resolved = EstimatorCalibrationAssertions.graphSize(resolving)
+ - EstimatorCalibrationAssertions.graphSize(context);
+
Assertions.assertTrue(PaimonCacheSizeEstimator.estimateFileIO(resolving) >=
resolved,
+ "the allowance does not cover a ResolvingFileIO with its
backend: " + resolved);
+ }
+
+ @Test
+ void tablesOutsideTheFileStoreCarryTheFileIoAllowanceToo() throws
Exception {
+ // A REST catalog builds an external table of any kind on a
ResolvingFileIO of its own, even with data
+ // tokens enabled, so format, object, Lance and Iceberg tables grow
after admission like a FileStoreTable.
+ // MUTATION: charging the FileIO only on the FileStoreTable path ->
each weight falls below the allowance
+ // alone -> red.
+ CatalogContext context = CatalogContext.create(new Options());
+ String location = warehouse.resolve("external").toUri().toString();
+ List<Table> tables = Arrays.asList(
+
FormatTable.builder().fileIO(resolvingFileIO(context)).identifier(TABLE).rowType(rowOf(10))
+
.partitionKeys(Collections.emptyList()).location(location).format(FormatTable.Format.CSV)
+ .options(Collections.emptyMap()).build(),
+
ObjectTable.builder().fileIO(resolvingFileIO(context)).identifier(TABLE).location(location)
+ .options(Collections.emptyMap()).build(),
+
LanceTable.builder().fileIO(resolvingFileIO(context)).identifier(TABLE).rowType(rowOf(10))
+
.location(location).options(Collections.emptyMap()).build(),
+
IcebergTable.builder().fileIO(resolvingFileIO(context)).identifier(TABLE).rowType(rowOf(10))
+
.partitionKeys(Collections.emptyList()).location(location)
+ .options(Collections.emptyMap()).build());
+ for (Table table : tables) {
+ String kind = table.getClass().getSimpleName();
+ long weight = weight(table);
+ long allowance =
PaimonCacheSizeEstimator.estimateFileIO(table.fileIO());
+ Assertions.assertTrue(weight > allowance,
+ kind + ": weight " + weight + " leaves out the FileIO
allowance " + allowance);
+ // A ResolvingFileIO builds its backend on the first data access,
after admission.
+ table.fileIO().exists(new Path(location));
+ long grown = EstimatorCalibrationAssertions.graphSize(table)
+ - EstimatorCalibrationAssertions.graphSize(context);
+ Assertions.assertTrue(weight >= grown,
+ kind + ": weight " + weight + " does not cover the grown
graph " + grown);
+ }
+ }
+
+ @Test
+ void fileIoAllowanceCoversARestDataTokenAndItsRefresh() throws Exception {
+ // The token arrives after admission and a refresh replaces it, so the
one allowance must cover the token
+ // the first data access fetches as well as a later, larger one. The
1,200-character security token is the
+ // usual STS size; the 8,192-character one stands for a credential
several times larger.
+ // MUTATION: a 4 KB allowance -> the refreshed token outgrows it ->
red.
+ FileStoreTable local = table("rest_token", 10, 0, 0, 0, null);
+ CatalogContext context = CatalogContext.create(new Options());
+ RESTTokenFileIO fileIO = new RESTTokenFileIO(context, null, TABLE,
local.location());
+ long allowance = PaimonCacheSizeEstimator.estimateFileIO(fileIO);
+ for (int securityTokenChars : new int[] {1200, 8192}) {
+ Map<String, String> token = new HashMap<>();
+ token.put("fs.oss.accessKeyId", "STS." + "a".repeat(28));
+ token.put("fs.oss.accessKeySecret", "b".repeat(44));
+ token.put("fs.oss.securityToken", "c".repeat(securityTokenChars));
+ token.put("fs.oss.endpoint",
"oss-cn-hangzhou-internal.aliyuncs.com");
+ setToken(fileIO, new RESTToken(token, System.currentTimeMillis() +
3_600_000L));
+ long owned = EstimatorCalibrationAssertions.graphSize(fileIO)
+ - EstimatorCalibrationAssertions.graphSize(context);
+ Assertions.assertTrue(allowance >= owned, "allowance " + allowance
+ + " does not cover the table-owned graph with a " +
securityTokenChars + "-character token "
+ + owned);
+ }
+ }
+
+ /**
+ * A primary-key table with {@code fieldCount} INT value fields under
{@code mergeEngine}: a partial-update
+ * table puts them all in one sequence group with a default aggregate
function, an aggregation table sums them.
+ */
+ private FileStoreTable mergeEngineTable(String name, int fieldCount,
String mergeEngine) throws Exception {
+ Schema.Builder builder = Schema.newBuilder()
+ .column("part", DataTypes.INT())
+ .column("key_0", DataTypes.INT().notNull())
+ .column("seq_0", DataTypes.INT());
+ List<String> grouped = new ArrayList<>();
+ for (int i = 0; i < fieldCount; i++) {
+ builder.column("field_" + i, DataTypes.INT());
+ grouped.add("field_" + i);
+ }
+ builder.primaryKey("key_0",
"part").partitionKeys("part").option("merge-engine", mergeEngine);
+ if ("partial-update".equals(mergeEngine)) {
+ builder.option("fields.seq_0.sequence-group", String.join(",",
grouped));
+ builder.option("fields.default-aggregate-function",
"last_non_null_value");
+ } else if ("aggregation".equals(mergeEngine)) {
+ builder.option("fields.default-aggregate-function", "sum");
+ }
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new Path(warehouse.resolve(name).toUri());
+ new SchemaManager(fileIO, path).createTable(builder.build());
+ return FileStoreTableFactory.create(fileIO, path);
+ }
+
+ private static ResolvingFileIO resolvingFileIO(CatalogContext context) {
+ ResolvingFileIO fileIO = new ResolvingFileIO();
+ fileIO.configure(context);
+ return fileIO;
+ }
+
+ private static void setToken(RESTTokenFileIO fileIO, RESTToken token)
throws ReflectiveOperationException {
+ Field tokenField = RESTTokenFileIO.class.getDeclaredField("token");
+ tokenField.setAccessible(true);
+ tokenField.set(fileIO, token);
+ }
+
+ private void assertGrownDelta(String fixture, FileStoreTable small,
FileStoreTable large) throws Exception {
+ long smallWeight = weight(small);
+ long largeWeight = weight(large);
+ grow(small);
+ grow(large);
+ EstimatorCalibrationAssertions.assertConservativeDelta(fixture,
smallWeight, largeWeight, small, large);
+ }
+
+ private static long weight(Table table) {
+ MetaCacheSizeEstimate estimate =
PaimonCacheSizeEstimator.estimateTable(
+ TABLE, table,
PaimonMetaCacheCatalog.TABLE_ENTRY_OVERHEAD_BYTES);
+ Assertions.assertTrue(estimate.isComplete(),
estimate.getIncompleteReason());
+ return estimate.getBytes();
+ }
+
+ /**
+ * A table with a partition column, {@code fieldCount} INT fields, {@code
primaryKeyCount} primary key fields
+ * (besides the partition column), {@code optionCount} options, an
optional ROW of {@code nestedFieldCount}
+ * fields and an optional {@code extraType} column.
+ */
+ private FileStoreTable table(String name, int fieldCount, int
primaryKeyCount, int optionCount,
+ int nestedFieldCount, DataType extraType) throws Exception {
+ Schema.Builder builder = Schema.newBuilder().column("part",
DataTypes.INT());
+ for (int i = 0; i < fieldCount; i++) {
+ builder.column("field_" + i, DataTypes.INT());
+ }
+ List<String> primaryKeys = new ArrayList<>();
+ for (int i = 0; i < primaryKeyCount; i++) {
+ builder.column("key_" + i, DataTypes.INT().notNull());
+ primaryKeys.add("key_" + i);
+ }
+ if (nestedFieldCount > 0) {
+ builder.column("payload", rowOf(nestedFieldCount));
+ }
+ if (extraType != null) {
+ builder.column("extra", extraType);
+ }
+ if (!primaryKeys.isEmpty()) {
+ primaryKeys.add("part");
+ builder.primaryKey(primaryKeys);
+ }
+ builder.partitionKeys("part");
+ for (int i = 0; i < optionCount; i++) {
+ builder.option("option_" + i, "value_" + i);
+ }
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new Path(warehouse.resolve(name).toUri());
+ new SchemaManager(fileIO, path).createTable(builder.build());
+ return FileStoreTableFactory.create(fileIO, path);
+ }
+
+ private static RowType rowOf(int fieldCount) {
+ List<DataField> fields = new ArrayList<>();
+ for (int i = 0; i < fieldCount; i++) {
+ fields.add(new DataField(1000 + i, "nested_" + i,
DataTypes.INT()));
+ }
+ return new RowType(fields);
+ }
+
+ /** What query planning builds on a cached table: the store, then every
reachable RowType's index maps. */
+ private static void grow(FileStoreTable table) throws Exception {
+ table.latestSnapshot();
+ table.newReadBuilder().newScan();
+ Object store = readField(table, table.getClass(), "lazyStore");
+ for (Class<?> owner = store.getClass(); owner != null && owner !=
Object.class;
+ owner = owner.getSuperclass()) {
+ for (Field field : owner.getDeclaredFields()) {
+ if (RowType.class.isAssignableFrom(field.getType())) {
+ field.setAccessible(true);
+ Object rowType = field.get(store);
+ if (rowType != null) {
+ buildIndexes((RowType) rowType);
+ }
+ }
+ }
+ }
+ for (DataField field : table.schema().fields()) {
+ buildIndexes(field.type());
+ }
+ }
+
+ private static void buildIndexes(DataType type) {
+ if (type instanceof RowType) {
+ RowType rowType = (RowType) type;
+ // The store also holds empty row types (for example the bucket
key type of an unbucketed table).
+ if (!rowType.getFields().isEmpty()) {
+ DataField first = rowType.getFields().get(0);
+ rowType.getField(first.name());
+ rowType.getFieldIndex(first.name());
+ rowType.getField(first.id());
+ rowType.getFieldIndexByFieldId(first.id());
+ }
+ for (DataField field : rowType.getFields()) {
+ buildIndexes(field.type());
+ }
+ } else if (type instanceof ArrayType) {
+ buildIndexes(((ArrayType) type).getElementType());
+ } else if (type instanceof MultisetType) {
+ buildIndexes(((MultisetType) type).getElementType());
+ } else if (type instanceof MapType) {
+ buildIndexes(((MapType) type).getKeyType());
+ buildIndexes(((MapType) type).getValueType());
+ }
+ }
+
+ private static Object readField(Object target, Class<?> owner, String
fieldName) throws Exception {
+ Field field = owner.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return field.get(target);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
