Skip to content

Commit f80fa3b

Browse files
924060929924060929
committed
[fix](fe) Reuse Iceberg dictionary freshness per metadata generation
### What problem does this PR solve? Issue Number: N/A Related PR: #69299 Problem Summary: Repeated Iceberg dictionary freshness checks detach the latest snapshot by eagerly loading its partition view. When the partition entry is disabled, rejected by its weight limit or retired, three checks of the same immutable generation invoke the complete view loader three times. Use the existing scoped snapshot callback and retain only the exact maximum partition-update timestamp in that generation. A synchronized lazy getter computes it once, including an empty view's zero result; a new metadata generation computes its own value. Reserve the fixed scalar/flag fields in snapshot accounting and keep complete partition maps subject to the existing admission policy. Focused tests reduce three loader invocations to one without retaining an unadmitted full view; this is a loader-count result, not a latency benchmark. ### Release note Avoid repeated partition metadata enumeration during Iceberg dictionary freshness checks. ### Check List (For Author) - Test: Unit Test: 207 related FE tests; full FE packaging and Checkstyle via build.sh --fe - Behavior changed: Yes; reuse the same freshness value within one immutable metadata generation without retaining full partition views - Does this need documentation: No
1 parent 1385a1e commit f80fa3b

4 files changed

Lines changed: 113 additions & 4 deletions

File tree

‎fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergCacheSizeEstimator.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,8 @@ final class IcebergCacheSizeEstimator {
9696
private static final long METADATA_ENTRY_WEIGHT = 128L;
9797
private static final long SNAPSHOT_WEIGHT = 512L;
9898
private static final long CURRENT_SNAPSHOT_WEIGHT = 1024L;
99+
// Reserve the projection's lazy maximum partition-update timestamp and its loaded flag.
100+
private static final long SNAPSHOT_UPDATE_TIME_STATE_WEIGHT = 32L;
99101
private static final long SNAPSHOT_LOG_WEIGHT = 64L;
100102
private static final long METADATA_LOG_WEIGHT = 160L;
101103
private static final long SNAPSHOT_REF_WEIGHT = 256L;
@@ -161,7 +163,7 @@ static MetaCacheSizeEstimate estimateTableEntry(NameMapping key, IcebergTableCac
161163

162164
static MetaCacheSizeEstimate estimateSnapshotEntry(
163165
IcebergSnapshotEntryKey key, IcebergSnapshotCacheValue value) {
164-
long bytes = KEY_BASE_WEIGHT;
166+
long bytes = KEY_BASE_WEIGHT + SNAPSHOT_UPDATE_TIME_STATE_WEIGHT;
165167
bytes = MetaCacheWeightUtils.saturatedAdd(bytes,
166168
MetaCacheWeightUtils.estimatedNameMappingBytes(key.getNameMapping()));
167169
bytes = MetaCacheWeightUtils.saturatedAdd(bytes,

‎fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalTable.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -245,9 +245,8 @@ public MTMVSnapshotIf getTableSnapshot(Optional<MvccSnapshot> snapshot) throws A
245245

246246
@Override
247247
public long getNewestUpdateVersionOrTime() {
248-
return IcebergUtils.getLatestSnapshotCacheValue(this)
249-
.getPartitionInfo().getNameToIcebergPartition().values().stream()
250-
.mapToLong(IcebergPartition::getLastUpdateTime).max().orElse(0);
248+
return IcebergUtils.withLatestSnapshotCacheValue(this,
249+
IcebergSnapshotCacheValue::getNewestPartitionUpdateTime);
251250
}
252251

253252
@Override

‎fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSnapshotCacheValue.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,8 @@ public class IcebergSnapshotCacheValue {
5151

5252
private final IcebergPartitionInfo partitionInfo;
5353
private Supplier<IcebergPartitionInfo> partitionInfoLoader;
54+
private long newestPartitionUpdateTime;
55+
private boolean newestPartitionUpdateTimeLoaded;
5456
private final IcebergSnapshot snapshot;
5557
private final Optional<Map<Integer, List<String>>> nameMapping;
5658
private Optional<Table> icebergTable;
@@ -142,6 +144,17 @@ public IcebergPartitionInfo getPartitionInfo() {
142144
return partitionInfoLoader == null ? partitionInfo : partitionInfoLoader.get();
143145
}
144146

147+
synchronized long getNewestPartitionUpdateTime() {
148+
if (!newestPartitionUpdateTimeLoaded) {
149+
// A dictionary check needs only this scalar. Do not retain an unadmitted full view
150+
// in the shared projection, and keep zero as a valid result for an empty view.
151+
newestPartitionUpdateTime = getPartitionInfo().getNameToIcebergPartition().values().stream()
152+
.mapToLong(IcebergPartition::getLastUpdateTime).max().orElse(0);
153+
newestPartitionUpdateTimeLoaded = true;
154+
}
155+
return newestPartitionUpdateTime;
156+
}
157+
145158
IcebergPartitionInfo getRetainedPartitionInfo() {
146159
return partitionInfo;
147160
}

‎fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCacheTest.java‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -899,6 +899,101 @@ private void assertMvBulkPartitionSnapshotsReuseUnadmittedView(Map<String, Strin
899899
}
900900
}
901901

902+
@Test
903+
public void testDictionaryChecksReuseVersionWhenPartitionCacheDisabled() throws Exception {
904+
assertDictionaryChecksReuseScalar(Collections.singletonMap("meta.cache.iceberg.partition.enable", "false"),
905+
false);
906+
}
907+
908+
@Test
909+
public void testDictionaryChecksReuseVersionWhenPartitionViewRejected() throws Exception {
910+
assertDictionaryChecksReuseScalar(Collections.singletonMap("meta.cache.iceberg.partition.max-weight", "1B"),
911+
false);
912+
}
913+
914+
@Test
915+
public void testDictionaryChecksReuseVersionAfterPartitionEntryRetirement() throws Exception {
916+
assertDictionaryChecksReuseScalar(Collections.emptyMap(), true);
917+
}
918+
919+
private void assertDictionaryChecksReuseScalar(Map<String, String> properties, boolean retired) throws Exception {
920+
ExecutorService executor = Executors.newSingleThreadExecutor();
921+
IcebergExternalMetaCache cache = new IcebergExternalMetaCache(executor);
922+
cache.initCatalog(1L, properties);
923+
NameMapping identity = NameMapping.createForTest(1L, "db", "tbl");
924+
IcebergExternalTable table = partitionConsumerForTest(identity);
925+
Mockito.when(table.getNewestUpdateVersionOrTime()).thenCallRealMethod();
926+
Table first = tableWithMetadataLocation("/metadata/dictionary-v1.json");
927+
cache.entry(1L, IcebergExternalMetaCache.ENTRY_TABLE, NameMapping.class, IcebergTableCacheValue.class)
928+
.put(identity, new IcebergTableCacheValue(first));
929+
MetaCacheEntry<IcebergSnapshotEntryKey, IcebergPartitionInfo> partitions = cache.entry(
930+
1L, IcebergExternalMetaCache.ENTRY_PARTITION,
931+
IcebergSnapshotEntryKey.class, IcebergPartitionInfo.class);
932+
AtomicInteger loads = new AtomicInteger();
933+
AtomicReference<IcebergSnapshotCacheValue> projection = new AtomicReference<>();
934+
AtomicReference<IcebergPartitionInfo> info = new AtomicReference<>(partitionUpdateTimes(3L, 7L, 5L));
935+
ExecutionAuthenticator authenticator = new ExecutionAuthenticator() {};
936+
try (MockedStatic<IcebergUtils> utils = Mockito.mockStatic(IcebergUtils.class, Mockito.CALLS_REAL_METHODS)) {
937+
utils.when(() -> IcebergUtils.isValidRelatedTable(Mockito.any(Table.class))).thenReturn(true);
938+
utils.when(() -> IcebergUtils.loadPartitionInfo(Mockito.eq("ice.db.tbl"), Mockito.any(Table.class),
939+
Mockito.anyLong(), Mockito.anyLong(), Mockito.same(authenticator), Mockito.eq(false),
940+
Mockito.eq(false))).thenAnswer(call -> {
941+
loads.incrementAndGet();
942+
return info.get();
943+
});
944+
utils.when(() -> IcebergUtils.getLatestSnapshotCacheValue(table))
945+
.thenAnswer(call -> projection.get().withoutRetainedTable());
946+
utils.when(() -> IcebergUtils.withLatestSnapshotCacheValue(Mockito.same(table),
947+
Mockito.<java.util.function.Function<IcebergSnapshotCacheValue, Long>>any()))
948+
.thenAnswer(call -> {
949+
java.util.function.Function<IcebergSnapshotCacheValue, Long> action = call.getArgument(1);
950+
return action.apply(projection.get());
951+
});
952+
projection.set(partitionProjectionForTest(cache, table, first, authenticator));
953+
if (retired) {
954+
cache.invalidateCatalog(1L);
955+
}
956+
for (int check = 0; check < 3; check++) {
957+
Assert.assertEquals(7L, table.getNewestUpdateVersionOrTime());
958+
}
959+
Assert.assertEquals("repeated dictionary checks must not reload the full view", 1, loads.get());
960+
Assert.assertSame(IcebergPartitionInfo.empty(), projection.get().getRetainedPartitionInfo());
961+
Assert.assertNull(partitions.peekIfPresent(IcebergSnapshotEntryKey.tryCreate(identity, first).get()));
962+
963+
cache.initCatalog(1L, properties);
964+
Table second = tableWithMetadataLocation("/metadata/dictionary-v2.json");
965+
cache.entry(1L, IcebergExternalMetaCache.ENTRY_TABLE, NameMapping.class, IcebergTableCacheValue.class)
966+
.put(identity, new IcebergTableCacheValue(second));
967+
info.set(partitionUpdateTimes(11L, 9L));
968+
projection.set(partitionProjectionForTest(cache, table, second, authenticator));
969+
Assert.assertEquals(11L, table.getNewestUpdateVersionOrTime());
970+
Assert.assertEquals(11L, table.getNewestUpdateVersionOrTime());
971+
Assert.assertEquals("a new metadata generation must compute its own version", 2, loads.get());
972+
Assert.assertSame(IcebergPartitionInfo.empty(), projection.get().getRetainedPartitionInfo());
973+
Table empty = tableWithMetadataLocation("/metadata/dictionary-empty.json");
974+
cache.entry(1L, IcebergExternalMetaCache.ENTRY_TABLE, NameMapping.class, IcebergTableCacheValue.class)
975+
.put(identity, new IcebergTableCacheValue(empty));
976+
info.set(IcebergPartitionInfo.empty());
977+
projection.set(partitionProjectionForTest(cache, table, empty, authenticator));
978+
Assert.assertEquals(0L, table.getNewestUpdateVersionOrTime());
979+
Assert.assertEquals(0L, table.getNewestUpdateVersionOrTime());
980+
Assert.assertEquals("an empty view's zero version is memoized too", 3, loads.get());
981+
} finally {
982+
cache.close();
983+
executor.shutdownNow();
984+
}
985+
}
986+
987+
private IcebergPartitionInfo partitionUpdateTimes(long... updateTimes) {
988+
Map<String, IcebergPartition> partitions = new HashMap<>();
989+
for (int i = 0; i < updateTimes.length; i++) {
990+
String name = "part=" + i;
991+
partitions.put(name, new IcebergPartition(name, 0, 1L, 1L, 1L, updateTimes[i], 1L,
992+
Collections.singletonList(Integer.toString(i)), Collections.singletonList("day")));
993+
}
994+
return new IcebergPartitionInfo(Collections.emptyMap(), partitions, Collections.emptyMap());
995+
}
996+
902997
@Test
903998
public void testTableReplacementRetiresOnlyObsoletePartitionViews() throws Exception {
904999
ExecutorService executor = Executors.newSingleThreadExecutor();

0 commit comments

Comments
 (0)