github-actions[bot] commented on code in PR #68839: URL: https://github.com/apache/doris/pull/68839#discussion_r4233435702
########## 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: [P2] Wait for the reclaim drain to finish before asserting coalescing. `secondReclaimFinished` is counted down inside the second reclaimer, before that callback returns and before `drainPeerReclaim` decides whether to queue another drain. If coalescing regresses and leaves a request pending, this assertion can see `[10, 20]` and pass before a third callback runs; the `finally` cleanup can also close the requester before that work is observed. Add a deterministic drain-completion barrier or test hook, then assert the final callback list. The catalog/group exclusion checks have the same end-of-second-drain observation gap. -- 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]
