morningman commented on code in PR #68839: URL: https://github.com/apache/doris/pull/68839#discussion_r4235999737
########## fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/MetaCacheBudgetManagerTest.java: ########## @@ -0,0 +1,464 @@ +// 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"; + Assertions.assertEquals(1L, CacheSpec.parseWeight("1", key, true, heap)); + Assertions.assertEquals(2L * 1024L * 1024L, CacheSpec.parseWeight(" 2 mb ", key, true, heap)); + Assertions.assertEquals(heap / 4L, CacheSpec.parseWeight("25%", key, true, heap)); + Assertions.assertEquals(heap / 8L, CacheSpec.parseWeight("12.5%", key, true, heap)); + // 0 disables the global quota; rejecting 0% is the caller's job (ExternalMetaCacheMgr). + Assertions.assertEquals(0L, CacheSpec.parseWeight("0", key, true, heap)); + Assertions.assertEquals(0L, CacheSpec.parseWeight("0%", key, true, heap)); + + for (String invalid : new String[] {"", "1.5GB", "101%", "-1%", "-1", "8EB", "1PB000", + "8192PB", "999999999999999999PB"}) { + Assertions.assertThrows(IllegalArgumentException.class, + () -> CacheSpec.parseWeight(invalid, key, true, heap), invalid); + } + IllegalArgumentException catalogPercentage = Assertions.assertThrows(IllegalArgumentException.class, + () -> CacheSpec.parseWeight("10%", "meta.cache.max-weight", false, heap)); + Assertions.assertEquals("Invalid cache weight for 'meta.cache.max-weight': 10%", + catalogPercentage.getMessage()); + } + + @Test + void releasingAReservationTwiceReleasesItOnce() { + MetaCacheBudgetManager manager = new MetaCacheBudgetManager(OptionalLong.of(100L)); + EntryBudget first = manager.createEntryBudget(1L, "hive", "partition_view", "partition_view", NONE, NONE); + EntryBudget second = manager.createEntryBudget(1L, "hive", "file", "file", NONE, NONE); + AdmissionReservation released = first.tryReserve(40L).get(); + AdmissionReservation kept = second.tryReserve(30L).get(); + + released.release(); + released.release(); + + Assertions.assertEquals(30L, manager.getGlobalUsedWeight(), + "a repeated release must not take bytes from another reservation"); + Assertions.assertEquals(0L, released.getBytes()); + kept.release(); + first.close(); + second.close(); + Assertions.assertEquals(0L, manager.getGlobalUsedWeight()); + } + + @Test + void replacementHoldsTheLargerGenerationUntilCommitOrRollback() { + MetaCacheBudgetManager manager = new MetaCacheBudgetManager(OptionalLong.of(100L)); + EntryBudget budget = manager.createEntryBudget(1L, "iceberg", "table", "table", NONE, NONE); + AdmissionReservation first = budget.tryReserve(30L).get(); + + ReservationReplacement grow = budget.tryReplace(first, 50L).get(); + Assertions.assertEquals(50L, manager.getGlobalUsedWeight(), "max(old, new) is held, never old + new"); + Assertions.assertFalse(budget.tryReplace(first, 10L).isPresent(), + "a generation that is being replaced can not be replaced again"); + grow.commit(); + AdmissionReservation second = grow.current(); + Assertions.assertEquals(50L, second.getBytes()); + + ReservationReplacement shrink = budget.tryReplace(second, 10L).get(); + Assertions.assertEquals(50L, manager.getGlobalUsedWeight(), + "the old generation stays charged until the new one is published"); + shrink.commit(); + Assertions.assertEquals(10L, manager.getGlobalUsedWeight()); + + AdmissionReservation third = shrink.current(); + ReservationReplacement abandoned = budget.tryReplace(third, 40L).get(); + Assertions.assertEquals(40L, manager.getGlobalUsedWeight()); + abandoned.rollback(); + Assertions.assertEquals(10L, manager.getGlobalUsedWeight()); + Assertions.assertThrows(IllegalStateException.class, abandoned::commit); + third.release(); + Assertions.assertEquals(0L, manager.getGlobalUsedWeight(), "rollback reactivates the previous generation"); + budget.close(); + } + + @Test + void closedBudgetRejectsNewReservationsAndReplacements() { + MetaCacheBudgetManager manager = new MetaCacheBudgetManager(OptionalLong.of(100L)); + EntryBudget stale = manager.createEntryBudget(1L, "hive", "partition_view", "partition_view", NONE, NONE); + AdmissionReservation zeroBytes = stale.tryReserve(0L).get(); + + stale.close(); + stale.close(); + + Assertions.assertFalse(stale.tryReserve(1L).isPresent()); + Assertions.assertFalse(stale.tryReplace(zeroBytes, 1L).isPresent()); + zeroBytes.release(); + + EntryBudget replacement = manager.createEntryBudget( + 1L, "hive", "partition_view", "partition_view", NONE, NONE); + AdmissionReservation full = replacement.tryReserve(100L).get(); + Assertions.assertEquals(100L, manager.getGlobalUsedWeight()); + full.release(); + replacement.close(); + Assertions.assertEquals(0L, manager.getGlobalUsedWeight()); + } + + @Test + void closeForceReleasesOutstandingReservations() { + MetaCacheBudgetManager manager = new MetaCacheBudgetManager(OptionalLong.of(100L)); + EntryBudget closing = manager.createEntryBudget(1L, "hive", "partition_view", "partition_view", NONE, NONE); + EntryBudget sibling = manager.createEntryBudget(1L, "hive", "file", "file", NONE, NONE); + AdmissionReservation leaked = closing.tryReserve(40L).get(); + AdmissionReservation live = sibling.tryReserve(30L).get(); + + closing.close(); + Assertions.assertEquals(30L, manager.getGlobalUsedWeight()); + leaked.release(); + Assertions.assertEquals(30L, manager.getGlobalUsedWeight(), + "a late release from a closed owner must not be subtracted twice"); + Assertions.assertEquals(30L, sibling.getUsedWeight()); + + EntryBudget reopened = manager.createEntryBudget(1L, "hive", "partition_view", "partition_view", NONE, NONE); + AdmissionReservation reReserved = reopened.tryReserve(70L).get(); + Assertions.assertEquals(100L, manager.getGlobalUsedWeight()); + reReserved.release(); + live.release(); + reopened.close(); + sibling.close(); + Assertions.assertEquals(0L, manager.getGlobalUsedWeight()); + } + + @Test + void peerReclaimCoalescesQueuedRequestsToTheLargestAdmission() throws Exception { + MetaCacheBudgetManager manager = new MetaCacheBudgetManager(OptionalLong.of(100L)); + EntryBudget owner = manager.createEntryBudget(1L, "iceberg", "partition", "partition", NONE, NONE); + EntryBudget requester = manager.createEntryBudget(2L, "hive", "partition_view", "partition_view", NONE, NONE); + AdmissionReservation reservation = owner.tryReserve(100L).get(); + CountDownLatch firstReclaimStarted = new CountDownLatch(1); + CountDownLatch releaseFirstReclaim = new CountDownLatch(1); + CountDownLatch secondReclaimFinished = new CountDownLatch(1); + AtomicInteger invocations = new AtomicInteger(); + List<Long> targets = Collections.synchronizedList(new ArrayList<>()); + owner.setReclaimer(target -> { + targets.add(target); + if (invocations.getAndIncrement() == 0) { + firstReclaimStarted.countDown(); + await(releaseFirstReclaim); + } else { + secondReclaimFinished.countDown(); + } + return 0L; + }); + try { + requester.requestPeerReclaim(10L); + Assertions.assertTrue(firstReclaimStarted.await(10L, TimeUnit.SECONDS)); + + requester.requestPeerReclaim(10L); + requester.requestPeerReclaim(20L); + requester.requestPeerReclaim(15L); + releaseFirstReclaim.countDown(); + + Assertions.assertTrue(secondReclaimFinished.await(10L, TimeUnit.SECONDS)); + Assertions.assertEquals(Arrays.asList(10L, 20L), new ArrayList<>(targets), Review Comment: Fixed in 3506202c47e for the coalescing test. Before it checks the targets, it now waits on two barriers. Each barrier is a drain of a separate budget manager on the same process-wide reclaim thread. The first runs after the second drain has finished and queued anything it left behind; the second runs after that queued drain. With a mutation that leaves the request pending and pauses 50 ms at the start of each drain, the previous version of the test passes and this one fails with `[10, 20, 20, 20]`. The catalog and group exclusion tests stay as they are. What they guard is the first drain's walk over the candidates, and that walk is already complete when their latch fires, because the second request is served only after the first drain returns. A regression that asked the other catalog or the other group would do so in that first drain (in the group test the other group's larger cache even sorts first), so the counter is already non-zero when the assertion runs. The unobserved tail of the second drain only repeats the same walk. -- 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]
