github-actions[bot] commented on code in PR #68839:
URL: https://github.com/apache/doris/pull/68839#discussion_r4236050867
##########
fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalogTest.java:
##########
@@ -422,6 +422,39 @@ void
realFileStoreTableIsAdmittedByTheWeightGovernedCache(@TempDir java.nio.file
Assertions.assertEquals(0L, budgetManager.getGlobalUsedWeight());
}
+ @Test
+ void
tableInvalidationReleasesTheReservationAndTheReloadIsReadmitted(@TempDir
java.nio.file.Path warehouse)
+ throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ org.apache.paimon.fs.Path tablePath =
+ createFileStoreTable(fileIO,
warehouse.resolve("refreshed-table"), "payload").location();
+ RecordingCatalog recording = new RecordingCatalog();
+ recording.tableSupplier = () -> FileStoreTableFactory.create(fileIO,
tablePath);
+ MetaCacheBudgetManager budgetManager = new
MetaCacheBudgetManager(OptionalLong.of(1024L * 1024L));
+ try (CatalogMetaCache owner = new CatalogMetaCache(
+ budgetManager, 66717L, "paimon", Collections.emptyMap())) {
+ PaimonMetaCacheCatalog catalog = new
PaimonMetaCacheCatalog(recording.catalog(), owner,
+ 100, 100, cacheOptions(Duration.ofDays(1),
Duration.ofDays(1)),
+ true, System::nanoTime);
+
+ Table first = catalog.getTable(TABLE);
+ long admittedWeight = budgetManager.getGlobalUsedWeight();
+ Assertions.assertTrue(admittedWeight > 0L);
+
+ // REFRESH TABLE reaches the connector as a table-scoped
invalidation of its cache owner.
+ owner.invalidateTable("db", "t");
Review Comment:
[P2] Exercise the connector hook in this invalidation test. Calling
`owner.invalidateTable` here bypasses the `PaimonConnector.invalidateTable`
path used by REFRESH TABLE. If that hook stops forwarding table invalidation to
the shared owner, this test still releases the reservation and reloads
successfully while production continues to serve a stale weighted
`paimon-table` entry. Load the table through a connector-owned catalog, call
`connector.invalidateTable("db", "t")`, and assert that its reservation is
released and the next read reloads.
##########
fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/MetaCacheBudgetManagerTest.java:
##########
@@ -0,0 +1,492 @@
+// 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.cache;
+
+import
org.apache.doris.connector.cache.MetaCacheBudgetManager.AdmissionReservation;
+import org.apache.doris.connector.cache.MetaCacheBudgetManager.EntryBudget;
+import
org.apache.doris.connector.cache.MetaCacheBudgetManager.ReservationReplacement;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Budget arithmetic of {@link MetaCacheBudgetManager} without a cache on top:
the global, catalog and
+ * entry-group limits, the reservation hand-off between two generations of one
key, release on close,
+ * and which peers a reclamation request may ask.
+ */
+class MetaCacheBudgetManagerTest {
+ private static final OptionalLong NONE = OptionalLong.empty();
+
+ @Test
+ void reservationsRespectGlobalCatalogAndEntryLimits() {
+ MetaCacheBudgetManager manager = new
MetaCacheBudgetManager(OptionalLong.of(100L));
+ EntryBudget file = manager.createEntryBudget(
+ 1L, "hive", "file", "file", OptionalLong.of(80L),
OptionalLong.of(60L));
+ EntryBudget partitionView = manager.createEntryBudget(
+ 1L, "hive", "partition_view", "partition_view",
OptionalLong.of(80L), OptionalLong.of(50L));
+
+ AdmissionReservation fileReservation = file.tryReserve(60L).get();
+ Assertions.assertFalse(partitionView.tryReserve(30L).isPresent(),
+ "the catalog limit of 80 is already 60 used");
+ AdmissionReservation viewReservation =
partitionView.tryReserve(20L).get();
+ Assertions.assertEquals(80L, manager.getGlobalUsedWeight());
+ Assertions.assertFalse(partitionView.tryReplace(viewReservation,
30L).isPresent());
+
+ fileReservation.release();
+ ReservationReplacement grown =
partitionView.tryReplace(viewReservation, 30L).get();
+ grown.commit();
+ Assertions.assertEquals(30L, manager.getGlobalUsedWeight());
+ Assertions.assertEquals(30L, partitionView.getUsedWeight());
+
+ grown.current().release();
+ Assertions.assertFalse(partitionView.tryReserve(51L).isPresent(),
+ "the entry limit of 50 applies even when the catalog and
global budgets have room");
+ file.close();
+ partitionView.close();
+ Assertions.assertEquals(0L, manager.getGlobalUsedWeight());
+ }
+
+ @Test
+ void concurrentReservationsNeverExceedTheGlobalLimit() throws Exception {
+ MetaCacheBudgetManager manager = new
MetaCacheBudgetManager(OptionalLong.of(100L));
+ EntryBudget budget = manager.createEntryBudget(1L, "iceberg",
"manifest", "manifest", NONE, NONE);
+ ExecutorService executor = Executors.newFixedThreadPool(8);
+ CountDownLatch start = new CountDownLatch(1);
+ List<AdmissionReservation> reservations =
Collections.synchronizedList(new ArrayList<>());
+ List<Future<?>> attempts = new ArrayList<>();
+ try {
+ for (int i = 0; i < 200; i++) {
+ attempts.add(executor.submit(() -> {
+ await(start);
+ budget.tryReserve(1L).ifPresent(reservations::add);
+ }));
+ }
+ start.countDown();
+ executor.shutdown();
+ Assertions.assertTrue(executor.awaitTermination(10L,
TimeUnit.SECONDS));
+ // A refused attempt returns an empty Optional; an attempt that
threw would surface here.
+ for (Future<?> attempt : attempts) {
+ attempt.get();
+ }
+ Assertions.assertEquals(100, reservations.size());
+ Assertions.assertEquals(100L, manager.getGlobalUsedWeight());
+ } finally {
+ executor.shutdownNow();
+ reservations.forEach(AdmissionReservation::release);
+ budget.close();
+ }
+ Assertions.assertEquals(0L, manager.getGlobalUsedWeight());
+ }
+
+ @Test
+ void entryBudgetIsClampedToTheSmallestEnclosingLimit() {
+ // A catalog created on an FE with a larger global quota replays here
with limits above ours.
+ MetaCacheBudgetManager manager = new
MetaCacheBudgetManager(OptionalLong.of(100L));
+ EntryBudget replayed = manager.createEntryBudget(
+ 1L, "iceberg", "table", "table", OptionalLong.of(400L),
OptionalLong.of(300L));
+ Assertions.assertEquals(100L, replayed.getEffectiveMaxWeight());
+ AdmissionReservation reservation = replayed.tryReserve(100L).get();
+ Assertions.assertFalse(replayed.tryReserve(1L).isPresent());
+ reservation.release();
+ replayed.close();
+
+ EntryBudget entryAboveCatalog = manager.createEntryBudget(
+ 2L, "iceberg", "table", "table", OptionalLong.of(40L),
OptionalLong.of(60L));
+ Assertions.assertEquals(40L,
entryAboveCatalog.getEffectiveMaxWeight());
+ Assertions.assertFalse(entryAboveCatalog.tryReserve(41L).isPresent());
+ entryAboveCatalog.close();
+ Assertions.assertEquals(0L, manager.getGlobalUsedWeight());
+ }
+
+ @Test
+ void weightParsingAcceptsBinaryUnitsAndHeapPercentages() {
+ long heap = 4L * 1024L * 1024L * 1024L;
+ String key = "external_meta_cache_max_weight";
Review Comment:
[P2] Test that the FE global quota reaches managed connector caches. These
assertions call `CacheSpec.parseWeight` directly, and the budget tests
instantiate their own managers; none exercises `ExternalMetaCacheMgr` passing
`Config.external_meta_cache_max_weight` to `MetaCacheGovernance`. If that
bridge is removed, an operator's global limit stops bounding connector caches
while every new test still passes. Add an FE-side case that configures a small
global limit, initializes the cache manager, and verifies combined admission
across managed catalogs.
##########
fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorCacheTest.java:
##########
@@ -226,6 +229,50 @@ private static Table fakeTable(String name) {
PartitionSpec.unpartitioned(), "s3://b/" + name,
Collections.emptyMap());
}
+ @Test
+ public void weightLimitsReachEveryConnectorOwnedCache() throws Exception {
+ long oneMb = 1024L * 1024L;
+ Map<String, Long> catalogLimited =
maxWeightsOf(props("meta.cache.max-weight", "1MB"));
+ for (String cache : new String[] {"iceberg-table",
"iceberg-latest-snapshot", "iceberg-partition",
+ "iceberg-format", "iceberg-manifest",
"iceberg-equality-delete-field-ids",
+ "iceberg.mvcc-partition-view",
"iceberg.list-partitions-view"}) {
+ Assertions.assertEquals(Long.valueOf(oneMb),
catalogLimited.get(cache),
+ "a catalog limit must bound " + cache);
+ }
+
+ Map<String, Long> manifestLimited =
maxWeightsOf(props("meta.cache.iceberg.manifest.max-weight", "1MB"));
+ Assertions.assertEquals(Long.valueOf(oneMb),
manifestLimited.get("iceberg-manifest"));
+ Assertions.assertEquals(Long.valueOf(-1L),
manifestLimited.get("iceberg-table"),
+ "an entry limit must bound only its own entry");
+
+ Map<String, Long> viewLimited =
maxWeightsOf(props("meta.cache.iceberg.partition_view.max-weight", "1MB"));
+ Assertions.assertEquals(Long.valueOf(oneMb),
viewLimited.get("iceberg.mvcc-partition-view"));
+ Assertions.assertEquals(Long.valueOf(oneMb),
viewLimited.get("iceberg.list-partitions-view"),
Review Comment:
[P2] Verify that both partition views share one budget. `getMaxWeight()`
reports each cache's effective limit, so these two 1 MB assertions still pass
if `ConnectorMetadataCache` groups by the physical cache name instead of
`partition_view`. That would give the MVCC and list caches separate buckets and
let them retain about 2 MB under one 1 MB entry limit. Exercise admission
through both caches with payloads that fit individually but exceed the limit
together, then assert their combined retained weight stays within 1 MB.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
