Skip to content

Commit e303ab6

Browse files
Youngwbmergify[bot]
authored andcommitted
[BugFix] Bound and expose the Iceberg partition cache memory (#76165)
Signed-off-by: Youngwb <yangwenbo_mailbox@163.com> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com> (cherry picked from commit 893d888) # Conflicts: # fe/fe-core/src/main/java/com/starrocks/connector/iceberg/CachingIcebergCatalog.java
1 parent cda4f22 commit e303ab6

3 files changed

Lines changed: 67 additions & 1 deletion

File tree

fe/fe-core/src/main/java/com/starrocks/connector/iceberg/CachingIcebergCatalog.java

Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -146,9 +146,19 @@ public Table reload(IcebergTableName key, Table oldValue) {
146146
}
147147
}
148148
});
149+
<<<<<<< HEAD
149150
this.partitionCache = newCacheBuilderWithMaximumSize(
150151
icebergProperties.getIcebergMetaCacheTtlSec(), NEVER_CACHE,
151152
enableCache ? DEFAULT_CACHE_NUM : NEVER_CACHE).build(
153+
=======
154+
long partitionCacheSize = Math.round(Runtime.getRuntime().maxMemory() *
155+
icebergProperties.getIcebergPartitionCacheMemoryUsageRatio());
156+
this.partitionCache = newCacheBuilder(icebergProperties.getIcebergMetaCacheTtlSec(), NEVER_CACHE)
157+
.executor(executorService)
158+
.maximumWeight(partitionCacheSize)
159+
.weigher((Weigher<IcebergTableName, Map<String, Partition>>) this::weighPartitionEntry)
160+
.build(
161+
>>>>>>> 893d888783 ([BugFix] Bound and expose the Iceberg partition cache memory (#76165))
152162
new com.github.benmanes.caffeine.cache.CacheLoader<IcebergTableName, Map<String, Partition>>() {
153163
@Override
154164
public Map<String, Partition> load(IcebergTableName key) throws Exception {
@@ -650,6 +660,17 @@ private int weighTableEntry(IcebergTableName key, Table table) {
650660
return (int) size;
651661
}
652662

663+
// Each entry holds a whole table's partition map, which is unbounded in bytes (a table can have
664+
// millions of partitions). Sample the partition values so weighing stays cheap on large maps.
665+
private int weighPartitionEntry(IcebergTableName key, Map<String, Partition> partitions) {
666+
long size = Estimator.estimate(key) + Estimator.estimate(partitions, 3);
667+
if (size > Integer.MAX_VALUE) {
668+
LOG.warn("Partition cache entry size for key {} is too large: {} bytes", key, size);
669+
return Integer.MAX_VALUE;
670+
}
671+
return (int) size;
672+
}
673+
653674
private long estimateContentFilesSize(Set<? extends ContentFile<?>> files) {
654675
if (files == null || files.isEmpty()) {
655676
return 0L;
@@ -694,7 +715,12 @@ public long estimateSize() {
694715
return Estimator.estimate(tables.asMap(), 10) +
695716
Estimator.estimate(databases.asMap(), 10) +
696717
estimateDataFileCacheSize() +
697-
estimateDeleteFileCacheSize() + Estimator.estimate(metaFileCacheMap, 10);
718+
estimateDeleteFileCacheSize() +
719+
Estimator.estimate(partitionCache.asMap(), 10) +
720+
Estimator.estimate(metaFileCacheMap, 10) +
721+
Estimator.estimate(tableLatestAccessTime, 10) +
722+
Estimator.estimate(tableLatestRefreshTime, 10) +
723+
Estimator.estimate(tableRefreshLockMap, 10);
698724
}
699725

700726
private long estimateDataFileCacheSize() {

fe/fe-core/src/main/java/com/starrocks/connector/iceberg/IcebergCatalogProperties.java

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ public class IcebergCatalogProperties {
4444
public static final String ICEBERG_DATA_FILE_CACHE_MEMORY_SIZE_RATIO = "iceberg_data_file_cache_memory_usage_ratio";
4545
public static final String ICEBERG_DELETE_FILE_CACHE_MEMORY_SIZE_RATIO = "iceberg_delete_file_cache_memory_usage_ratio";
4646
public static final String ICEBERG_TABLE_CACHE_MEMORY_SIZE_RATIO = "iceberg_table_cache_memory_usage_ratio";
47+
public static final String ICEBERG_PARTITION_CACHE_MEMORY_SIZE_RATIO = "iceberg_partition_cache_memory_usage_ratio";
4748

4849
// internal config
4950
public static final String REFRESH_ICEBERG_MANIFEST_MIN_LENGTH = "refresh_iceberg_manifest_min_length";
@@ -68,6 +69,7 @@ public class IcebergCatalogProperties {
6869
private double icebergDataFileCacheMemoryUsageRatio;
6970
private double icebergDeleteFileCacheMemoryUsageRatio;
7071
private double icebergTableCacheMemoryUsageRatio;
72+
private double icebergPartitionCacheMemoryUsageRatio;
7173
private long icebergTableCacheRefreshIntervalSec;
7274

7375
public IcebergCatalogProperties(Map<String, String> catalogProperties) {
@@ -109,6 +111,8 @@ private void initIcebergMetadataCache() {
109111
properties, ICEBERG_DELETE_FILE_CACHE_MEMORY_SIZE_RATIO, 0.1);
110112
this.icebergTableCacheMemoryUsageRatio = PropertyUtil.propertyAsDouble(
111113
properties, ICEBERG_TABLE_CACHE_MEMORY_SIZE_RATIO, 0.1);
114+
this.icebergPartitionCacheMemoryUsageRatio = PropertyUtil.propertyAsDouble(
115+
properties, ICEBERG_PARTITION_CACHE_MEMORY_SIZE_RATIO, 0.1);
112116
this.icebergManifestCacheWithColumnStatistics = PropertyUtil.propertyAsBoolean(
113117
properties, ICEBERG_MANIFEST_CACHE_WITH_COLUMN_STATISTICS, true);
114118
this.refreshIcebergManifestMinLength = PropertyUtil.propertyAsLong(properties, REFRESH_ICEBERG_MANIFEST_MIN_LENGTH,
@@ -184,6 +188,10 @@ public double getIcebergTableCacheMemoryUsageRatio() {
184188
return icebergTableCacheMemoryUsageRatio;
185189
}
186190

191+
public double getIcebergPartitionCacheMemoryUsageRatio() {
192+
return icebergPartitionCacheMemoryUsageRatio;
193+
}
194+
187195
public long getRefreshIcebergManifestMinLength() {
188196
return refreshIcebergManifestMinLength;
189197
}

fe/fe-core/src/test/java/com/starrocks/connector/iceberg/CachingIcebergCatalogTest.java

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -217,6 +217,38 @@ public Table getTable(ConnectContext context, String dbName, String tableName) {
217217
}
218218
}
219219

220+
@Test
221+
public void testPartitionCacheCountedInEstimateSize(@Mocked IcebergCatalog icebergCatalog) {
222+
PartitionSpec spec = Mockito.mock(PartitionSpec.class);
223+
Mockito.when(spec.isUnpartitioned()).thenReturn(false);
224+
Table nativeTable = createBaseTableWithManifests(1, 0, spec);
225+
Map<String, Partition> partitionMap = new HashMap<>();
226+
for (int i = 0; i < 1000; i++) {
227+
partitionMap.put("dt=part-" + i, new Partition(1234L, 1L));
228+
}
229+
new Expectations() {
230+
{
231+
icebergCatalog.getTable((ConnectContext) any, "db", "test");
232+
result = nativeTable;
233+
minTimes = 0;
234+
icebergCatalog.getPartitions((IcebergTable) any, anyLong, null);
235+
result = partitionMap;
236+
minTimes = 0;
237+
}
238+
};
239+
CachingIcebergCatalog cachingIcebergCatalog = new CachingIcebergCatalog(CATALOG_NAME, icebergCatalog,
240+
DEFAULT_CATALOG_PROPERTIES, Executors.newSingleThreadExecutor());
241+
IcebergTable table = IcebergTable.builder().setSrTableName("test")
242+
.setCatalogDBName("db").setCatalogTableName("test").setNativeTable(nativeTable).build();
243+
244+
long before = cachingIcebergCatalog.estimateSize();
245+
cachingIcebergCatalog.getPartitions(table, 1L, null);
246+
long after = cachingIcebergCatalog.estimateSize();
247+
// partitionCache used to be excluded from estimateSize, so a full partition map was invisible.
248+
Assertions.assertTrue(after > before,
249+
"partitionCache must be counted in estimateSize; before=" + before + " after=" + after);
250+
}
251+
220252
@Test
221253
public void testGetDB(@Mocked IcebergCatalog icebergCatalog, @Mocked Database db) {
222254
new Expectations() {

0 commit comments

Comments
 (0)