diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/LocalDynamicFilterConsumer.java b/core/trino-main/src/main/java/io/trino/sql/planner/LocalDynamicFilterConsumer.java index 98bfa93158a3..ed0dd63440ad 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/LocalDynamicFilterConsumer.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/LocalDynamicFilterConsumer.java @@ -14,9 +14,6 @@ package io.trino.sql.planner; import com.google.common.collect.ImmutableMap; -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.SettableFuture; import io.trino.spi.predicate.Domain; import io.trino.spi.predicate.TupleDomain; import io.trino.spi.type.Type; @@ -24,6 +21,7 @@ import io.trino.sql.planner.plan.JoinNode; import io.trino.sql.planner.plan.PlanNode; +import javax.annotation.Nullable; import javax.annotation.concurrent.GuardedBy; import java.util.ArrayList; @@ -31,13 +29,13 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.function.Consumer; import static com.google.common.base.MoreObjects.toStringHelper; import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkState; import static com.google.common.base.Verify.verify; import static com.google.common.collect.ImmutableMap.toImmutableMap; -import static com.google.common.util.concurrent.MoreExecutors.directExecutor; import static java.util.Objects.requireNonNull; import static java.util.function.Function.identity; @@ -51,71 +49,106 @@ public class LocalDynamicFilterConsumer // Mapping from dynamic filter ID to its build channel type. private final Map filterBuildTypes; - private final SettableFuture> resultFuture; + private final List>> collectors; // Number of build-side partitions to be collected, must be provided by setPartitionCount @GuardedBy("this") private int expectedPartitionCount = PARTITION_COUNT_INITIAL_VALUE; + @GuardedBy("this") + private boolean collected; + // The resulting predicates from each build-side partition. + @Nullable @GuardedBy("this") - private final List> partitions; + private List> partitions; - public LocalDynamicFilterConsumer(Map buildChannels, Map filterBuildTypes) + public LocalDynamicFilterConsumer( + Map buildChannels, + Map filterBuildTypes, + List>> collectors) { this.buildChannels = requireNonNull(buildChannels, "buildChannels is null"); this.filterBuildTypes = requireNonNull(filterBuildTypes, "filterBuildTypes is null"); verify(buildChannels.keySet().equals(filterBuildTypes.keySet()), "filterBuildTypes and buildChannels must have same keys"); - this.resultFuture = SettableFuture.create(); + requireNonNull(collectors, "collectors is null"); + checkArgument(!collectors.isEmpty(), "collectors is empty"); + this.collectors = collectors; this.partitions = new ArrayList<>(); } - public ListenableFuture> getDynamicFilterDomains() - { - return Futures.transform(resultFuture, this::convertTupleDomain, directExecutor()); - } - @Override public void addPartition(TupleDomain tupleDomain) { - if (resultFuture.isDone()) { - return; - } - TupleDomain result = null; + TupleDomain result; synchronized (this) { + if (collected) { + return; + } + requireNonNull(partitions, "partitions is null"); // Called concurrently by each DynamicFilterSourceOperator instance (when collection is over). verify(expectedPartitionCount == PARTITION_COUNT_INITIAL_VALUE || partitions.size() < expectedPartitionCount); // NOTE: may result in a bit more relaxed constraint if there are multiple columns and multiple rows. // See the comment at TupleDomain::columnWiseUnion() for more details. partitions.add(tupleDomain); - if (partitions.size() == expectedPartitionCount || tupleDomain.isAll()) { + if (tupleDomain.isAll()) { + result = tupleDomain; + } + else if (partitions.size() == expectedPartitionCount) { // No more partitions are left to be processed. - result = TupleDomain.columnWiseUnion(partitions); + if (partitions.isEmpty()) { + result = TupleDomain.none(); + } + else { + result = TupleDomain.columnWiseUnion(partitions); + } } + else { + return; + } + collected = true; + partitions = null; } - if (result != null) { - resultFuture.set(result); - } + notifyConsumers(result); } @Override public void setPartitionCount(int partitionCount) { - TupleDomain result = null; + TupleDomain result; synchronized (this) { + if (collected) { + return; + } checkState(expectedPartitionCount == PARTITION_COUNT_INITIAL_VALUE, "setPartitionCount should be called only once"); + requireNonNull(partitions, "partitions is null"); expectedPartitionCount = partitionCount; if (partitions.size() == expectedPartitionCount) { // No more partitions are left to be processed. - result = TupleDomain.columnWiseUnion(partitions); + if (partitions.isEmpty()) { + result = TupleDomain.none(); + } + else { + result = TupleDomain.columnWiseUnion(partitions); + } + collected = true; + partitions = null; + } + else { + return; } } - if (result != null) { - resultFuture.set(result); - } + notifyConsumers(result); + } + + private void notifyConsumers(TupleDomain result) + { + requireNonNull(result, "result is null"); + Map dynamicFilterDomains = convertTupleDomain(result); + collectors.forEach(consumer -> consumer.accept(dynamicFilterDomains)); } private Map convertTupleDomain(TupleDomain result) @@ -135,7 +168,8 @@ private Map convertTupleDomain(TupleDomain buildSourceTypes, - Set collectedFilters) + Set collectedFilters, + List>> collectors) { checkArgument(!planNode.getDynamicFilters().isEmpty(), "Join node dynamicFilters is empty."); checkArgument(!collectedFilters.isEmpty(), "Collected dynamic filters set is empty"); @@ -159,7 +193,7 @@ public static LocalDynamicFilterConsumer create( .collect(toImmutableMap( Map.Entry::getKey, entry -> buildSourceTypes.get(entry.getValue()))); - return new LocalDynamicFilterConsumer(buildChannels, filterBuildTypes); + return new LocalDynamicFilterConsumer(buildChannels, filterBuildTypes, collectors); } public Map getBuildChannels() @@ -168,11 +202,12 @@ public Map getBuildChannels() } @Override - public String toString() + public synchronized String toString() { return toStringHelper(this) .add("buildChannels", buildChannels) .add("expectedPartitionCount", expectedPartitionCount) + .add("collected", collected) .add("partitions", partitions) .toString(); } diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/LocalExecutionPlanner.java b/core/trino-main/src/main/java/io/trino/sql/planner/LocalExecutionPlanner.java index edd2fbb9f1bc..817739eff5b5 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/LocalExecutionPlanner.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/LocalExecutionPlanner.java @@ -27,7 +27,6 @@ import com.google.common.collect.Multimap; import com.google.common.collect.SetMultimap; import com.google.common.primitives.Ints; -import com.google.common.util.concurrent.ListenableFuture; import io.airlift.log.Logger; import io.airlift.units.DataSize; import io.trino.Session; @@ -268,6 +267,7 @@ import java.util.OptionalInt; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; import java.util.stream.Collectors; @@ -286,7 +286,6 @@ import static com.google.common.collect.Iterables.getOnlyElement; import static com.google.common.collect.Range.closedOpen; import static com.google.common.collect.Sets.difference; -import static io.airlift.concurrent.MoreFutures.addSuccessCallback; import static io.trino.SystemSessionProperties.getAdaptivePartialAggregationMinRows; import static io.trino.SystemSessionProperties.getAdaptivePartialAggregationUniqueRowsRatioThreshold; import static io.trino.SystemSessionProperties.getAggregationOperatorUnspillMemoryLimit; @@ -745,9 +744,12 @@ private void registerCoordinatorDynamicFilters(List d difference(consumedFilterIds, dynamicFiltersCollector.getRegisteredDynamicFilterIds())); } - private void addCoordinatorDynamicFilters(Map dynamicTupleDomain) + private Consumer> getCoordinatorDynamicFilterDomainsCollector(Set coordinatorDynamicFilters) { - taskContext.updateDomains(dynamicTupleDomain); + return domains -> taskContext.updateDomains( + domains.entrySet().stream() + .filter(entry -> coordinatorDynamicFilters.contains(entry.getKey())) + .collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue))); } public Optional getIndexSourceContext() @@ -2913,18 +2915,19 @@ private Optional createDynamicFilter( buildSource.getPipelineExecutionStrategy() != GROUPED_EXECUTION, "Dynamic filtering cannot be used with grouped execution"); log.debug("[Join] Dynamic filters: %s", node.getDynamicFilters()); - LocalDynamicFilterConsumer filterConsumer = LocalDynamicFilterConsumer.create(node, buildSource.getTypes(), collectedDynamicFilters); - ListenableFuture> domainsFuture = filterConsumer.getDynamicFilterDomains(); + ImmutableList.Builder>> collectors = ImmutableList.builder(); if (!localDynamicFilters.isEmpty()) { - addSuccessCallback(domainsFuture, context::addLocalDynamicFilters); + collectors.add(context::addLocalDynamicFilters); } if (!coordinatorDynamicFilters.isEmpty()) { - addSuccessCallback( - domainsFuture, - domains -> context.addCoordinatorDynamicFilters(domains.entrySet().stream() - .filter(entry -> coordinatorDynamicFilters.contains(entry.getKey())) - .collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue)))); + collectors.add(context.getCoordinatorDynamicFilterDomainsCollector(coordinatorDynamicFilters)); } + LocalDynamicFilterConsumer filterConsumer = LocalDynamicFilterConsumer.create( + node, + buildSource.getTypes(), + collectedDynamicFilters, + collectors.build()); + return Optional.of(filterConsumer); } @@ -3077,17 +3080,18 @@ public PhysicalOperation visitSemiJoin(SemiJoinNode node, LocalExecutionPlanCont // Add a DynamicFilterSourceOperatorFactory to build operator factories DynamicFilterId filterId = node.getDynamicFilterId().get(); log.debug("[Semi-join] Dynamic filter: %s", filterId); - LocalDynamicFilterConsumer filterConsumer = new LocalDynamicFilterConsumer( - ImmutableMap.of(filterId, buildChannel), - ImmutableMap.of(filterId, buildSource.getTypes().get(buildChannel))); - ListenableFuture> domainsFuture = filterConsumer.getDynamicFilterDomains(); + ImmutableList.Builder>> collectors = ImmutableList.builder(); if (isLocalDynamicFilter) { - addSuccessCallback(domainsFuture, context::addLocalDynamicFilters); + collectors.add(context::addLocalDynamicFilters); } if (isCoordinatorDynamicFilter) { - addSuccessCallback(domainsFuture, context::addCoordinatorDynamicFilters); + collectors.add(context.getCoordinatorDynamicFilterDomainsCollector(ImmutableSet.of(filterId))); } boolean isReplicatedJoin = isBuildSideReplicated(node); + LocalDynamicFilterConsumer filterConsumer = new LocalDynamicFilterConsumer( + ImmutableMap.of(filterId, buildChannel), + ImmutableMap.of(filterId, buildSource.getTypes().get(buildChannel)), + collectors.build()); buildSource = new PhysicalOperation( new DynamicFilterSourceOperatorFactory( operatorId, diff --git a/core/trino-main/src/test/java/io/trino/sql/planner/TestLocalDynamicFilterConsumer.java b/core/trino-main/src/test/java/io/trino/sql/planner/TestLocalDynamicFilterConsumer.java index 219174f66816..dbda2ffbb472 100644 --- a/core/trino-main/src/test/java/io/trino/sql/planner/TestLocalDynamicFilterConsumer.java +++ b/core/trino-main/src/test/java/io/trino/sql/planner/TestLocalDynamicFilterConsumer.java @@ -17,7 +17,6 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; -import com.google.common.util.concurrent.ListenableFuture; import io.trino.spi.predicate.Domain; import io.trino.spi.predicate.TupleDomain; import io.trino.sql.planner.OptimizerConfig.JoinDistributionType; @@ -31,7 +30,9 @@ import java.util.Map; import java.util.Optional; +import java.util.function.Consumer; +import static com.google.common.base.Verify.verify; import static io.trino.SystemSessionProperties.ENABLE_DYNAMIC_FILTERING; import static io.trino.SystemSessionProperties.FORCE_SINGLE_NODE_OUTPUT; import static io.trino.SystemSessionProperties.JOIN_DISTRIBUTION_TYPE; @@ -42,6 +43,7 @@ import static io.trino.spi.type.SmallintType.SMALLINT; import static io.trino.sql.planner.plan.JoinNode.Type.INNER; import static io.trino.testing.assertions.Assert.assertEquals; +import static java.util.Objects.requireNonNull; import static org.testng.Assert.assertFalse; public class TestLocalDynamicFilterConsumer @@ -58,159 +60,158 @@ public TestLocalDynamicFilterConsumer() @Test public void testSimple() - throws Exception { + TestingDynamicFilterCollector collector = new TestingDynamicFilterCollector(); LocalDynamicFilterConsumer filter = new LocalDynamicFilterConsumer( ImmutableMap.of(new DynamicFilterId("123"), 0), - ImmutableMap.of(new DynamicFilterId("123"), INTEGER)); + ImmutableMap.of(new DynamicFilterId("123"), INTEGER), + ImmutableList.of(collector)); filter.setPartitionCount(1); assertEquals(filter.getBuildChannels(), ImmutableMap.of(new DynamicFilterId("123"), 0)); - ListenableFuture> result = filter.getDynamicFilterDomains(); - assertFalse(result.isDone()); + assertFalse(collector.isCollectionComplete()); filter.addPartition(TupleDomain.withColumnDomains(ImmutableMap.of( new DynamicFilterId("123"), Domain.singleValue(INTEGER, 7L)))); - assertEquals(result.get(), ImmutableMap.of( + assertEquals(collector.getCollectedDomains(), ImmutableMap.of( new DynamicFilterId("123"), Domain.singleValue(INTEGER, 7L))); } @Test public void testShortCircuitOnAllTupleDomain() - throws Exception { + TestingDynamicFilterCollector collector = new TestingDynamicFilterCollector(); LocalDynamicFilterConsumer filter = new LocalDynamicFilterConsumer( ImmutableMap.of(new DynamicFilterId("123"), 0), - ImmutableMap.of(new DynamicFilterId("123"), INTEGER)); + ImmutableMap.of(new DynamicFilterId("123"), INTEGER), + ImmutableList.of(collector)); - ListenableFuture> result = filter.getDynamicFilterDomains(); - assertFalse(result.isDone()); + assertFalse(collector.isCollectionComplete()); filter.addPartition(TupleDomain.withColumnDomains(ImmutableMap.of( new DynamicFilterId("123"), Domain.all(INTEGER)))); - assertEquals(result.get(), ImmutableMap.of(new DynamicFilterId("123"), Domain.all(INTEGER))); + assertEquals(collector.getCollectedDomains(), ImmutableMap.of(new DynamicFilterId("123"), Domain.all(INTEGER))); filter.setPartitionCount(2); // adding another partition domain won't change final domain filter.addPartition(TupleDomain.withColumnDomains(ImmutableMap.of( new DynamicFilterId("123"), Domain.singleValue(INTEGER, 1L)))); - assertEquals(result.get(), ImmutableMap.of(new DynamicFilterId("123"), Domain.all(INTEGER))); + assertEquals(collector.getCollectedDomains(), ImmutableMap.of(new DynamicFilterId("123"), Domain.all(INTEGER))); } @Test public void testMultiplePartitions() - throws Exception { + TestingDynamicFilterCollector collector = new TestingDynamicFilterCollector(); LocalDynamicFilterConsumer filter = new LocalDynamicFilterConsumer( ImmutableMap.of(new DynamicFilterId("123"), 0), - ImmutableMap.of(new DynamicFilterId("123"), INTEGER)); + ImmutableMap.of(new DynamicFilterId("123"), INTEGER), + ImmutableList.of(collector)); assertEquals(filter.getBuildChannels(), ImmutableMap.of(new DynamicFilterId("123"), 0)); - ListenableFuture> result = filter.getDynamicFilterDomains(); - assertFalse(result.isDone()); + assertFalse(collector.isCollectionComplete()); filter.addPartition(TupleDomain.withColumnDomains(ImmutableMap.of( new DynamicFilterId("123"), Domain.singleValue(INTEGER, 10L)))); - assertFalse(result.isDone()); + assertFalse(collector.isCollectionComplete()); filter.addPartition(TupleDomain.withColumnDomains(ImmutableMap.of( new DynamicFilterId("123"), Domain.singleValue(INTEGER, 20L)))); - assertFalse(result.isDone()); + assertFalse(collector.isCollectionComplete()); filter.setPartitionCount(2); - assertEquals(result.get(), ImmutableMap.of( + assertEquals(collector.getCollectedDomains(), ImmutableMap.of( new DynamicFilterId("123"), Domain.multipleValues(INTEGER, ImmutableList.of(10L, 20L)))); } @Test public void testAllDomain() - throws Exception { DynamicFilterId filter1 = new DynamicFilterId("123"); DynamicFilterId filter2 = new DynamicFilterId("124"); + TestingDynamicFilterCollector collector = new TestingDynamicFilterCollector(); LocalDynamicFilterConsumer filter = new LocalDynamicFilterConsumer( ImmutableMap.of( filter1, 0, filter2, 1), ImmutableMap.of( filter1, INTEGER, - filter2, INTEGER)); + filter2, INTEGER), + ImmutableList.of(collector)); filter.setPartitionCount(1); - ListenableFuture> result = filter.getDynamicFilterDomains(); - assertFalse(result.isDone()); + assertFalse(collector.isCollectionComplete()); filter.addPartition(TupleDomain.withColumnDomains(ImmutableMap.of( filter1, Domain.all(INTEGER), filter2, Domain.singleValue(INTEGER, 1L)))); - assertEquals(result.get(), ImmutableMap.of(filter1, Domain.all(INTEGER), filter2, Domain.singleValue(INTEGER, 1L))); + assertEquals(collector.getCollectedDomains(), ImmutableMap.of(filter1, Domain.all(INTEGER), filter2, Domain.singleValue(INTEGER, 1L))); } @Test public void testNone() - throws Exception { + TestingDynamicFilterCollector collector = new TestingDynamicFilterCollector(); LocalDynamicFilterConsumer filter = new LocalDynamicFilterConsumer( ImmutableMap.of(new DynamicFilterId("123"), 0), - ImmutableMap.of(new DynamicFilterId("123"), INTEGER)); + ImmutableMap.of(new DynamicFilterId("123"), INTEGER), + ImmutableList.of(collector)); filter.setPartitionCount(1); assertEquals(filter.getBuildChannels(), ImmutableMap.of(new DynamicFilterId("123"), 0)); - ListenableFuture> result = filter.getDynamicFilterDomains(); - assertFalse(result.isDone()); + assertFalse(collector.isCollectionComplete()); filter.addPartition(TupleDomain.none()); - assertEquals(result.get(), ImmutableMap.of( + assertEquals(collector.getCollectedDomains(), ImmutableMap.of( new DynamicFilterId("123"), Domain.none(INTEGER))); } @Test public void testMultipleColumns() - throws Exception { + TestingDynamicFilterCollector collector = new TestingDynamicFilterCollector(); LocalDynamicFilterConsumer filter = new LocalDynamicFilterConsumer( ImmutableMap.of(new DynamicFilterId("123"), 0, new DynamicFilterId("456"), 1), - ImmutableMap.of(new DynamicFilterId("123"), INTEGER, new DynamicFilterId("456"), INTEGER)); + ImmutableMap.of(new DynamicFilterId("123"), INTEGER, new DynamicFilterId("456"), INTEGER), + ImmutableList.of(collector)); filter.setPartitionCount(1); assertEquals(filter.getBuildChannels(), ImmutableMap.of(new DynamicFilterId("123"), 0, new DynamicFilterId("456"), 1)); - ListenableFuture> result = filter.getDynamicFilterDomains(); - assertFalse(result.isDone()); + assertFalse(collector.isCollectionComplete()); filter.addPartition(TupleDomain.withColumnDomains(ImmutableMap.of( new DynamicFilterId("123"), Domain.singleValue(INTEGER, 10L), new DynamicFilterId("456"), Domain.singleValue(INTEGER, 20L)))); - assertEquals(result.get(), ImmutableMap.of( + assertEquals(collector.getCollectedDomains(), ImmutableMap.of( new DynamicFilterId("123"), Domain.singleValue(INTEGER, 10L), new DynamicFilterId("456"), Domain.singleValue(INTEGER, 20L))); } @Test public void testMultiplePartitionsAndColumns() - throws Exception { + TestingDynamicFilterCollector collector = new TestingDynamicFilterCollector(); LocalDynamicFilterConsumer filter = new LocalDynamicFilterConsumer( ImmutableMap.of(new DynamicFilterId("123"), 0, new DynamicFilterId("456"), 1), - ImmutableMap.of(new DynamicFilterId("123"), INTEGER, new DynamicFilterId("456"), BIGINT)); + ImmutableMap.of(new DynamicFilterId("123"), INTEGER, new DynamicFilterId("456"), BIGINT), + ImmutableList.of(collector)); filter.setPartitionCount(2); assertEquals(filter.getBuildChannels(), ImmutableMap.of(new DynamicFilterId("123"), 0, new DynamicFilterId("456"), 1)); - ListenableFuture> result = filter.getDynamicFilterDomains(); - assertFalse(result.isDone()); + assertFalse(collector.isCollectionComplete()); filter.addPartition(TupleDomain.withColumnDomains(ImmutableMap.of( new DynamicFilterId("123"), Domain.singleValue(INTEGER, 10L), new DynamicFilterId("456"), Domain.singleValue(BIGINT, 100L)))); - assertFalse(result.isDone()); + assertFalse(collector.isCollectionComplete()); filter.addPartition(TupleDomain.withColumnDomains(ImmutableMap.of( new DynamicFilterId("123"), Domain.singleValue(INTEGER, 20L), new DynamicFilterId("456"), Domain.singleValue(BIGINT, 200L)))); - assertEquals(result.get(), ImmutableMap.of( + assertEquals(collector.getCollectedDomains(), ImmutableMap.of( new DynamicFilterId("123"), Domain.multipleValues(INTEGER, ImmutableList.of(10L, 20L)), new DynamicFilterId("456"), Domain.multipleValues(BIGINT, ImmutableList.of(100L, 200L)))); } @Test public void testDynamicFilterPruning() - throws Exception { PlanBuilder planBuilder = new PlanBuilder(new PlanNodeIdAllocator(), dummyMetadata(), getQueryRunner().getDefaultSession()); Symbol left1 = planBuilder.symbol("left1", BIGINT); @@ -236,19 +237,45 @@ public void testDynamicFilterPruning() Optional.empty(), Optional.empty(), ImmutableMap.of(filter1, right1, filter2, right2, filter3, right3)); + TestingDynamicFilterCollector collector = new TestingDynamicFilterCollector(); LocalDynamicFilterConsumer consumer = LocalDynamicFilterConsumer.create( joinNode, ImmutableList.of(BIGINT, INTEGER, SMALLINT), - ImmutableSet.of(filter1, filter3)); + ImmutableSet.of(filter1, filter3), + ImmutableList.of(collector)); assertEquals(consumer.getBuildChannels(), ImmutableMap.of(filter1, 0, filter3, 2)); // make sure domain types got propagated correctly - assertFalse(consumer.getDynamicFilterDomains().isDone()); + assertFalse(collector.isCollectionComplete()); consumer.addPartition(TupleDomain.none()); - assertFalse(consumer.getDynamicFilterDomains().isDone()); + assertFalse(collector.isCollectionComplete()); consumer.setPartitionCount(1); assertEquals( - consumer.getDynamicFilterDomains().get(), + collector.getCollectedDomains(), ImmutableMap.of(filter1, Domain.none(BIGINT), filter3, Domain.none(SMALLINT))); } + + private static class TestingDynamicFilterCollector + implements Consumer> + { + private Map collectedDomains; + + @Override + public void accept(Map dynamicFilterDomains) + { + verify(collectedDomains == null, "collectedDomains is already set"); + collectedDomains = dynamicFilterDomains; + } + + public boolean isCollectionComplete() + { + return collectedDomains != null; + } + + public Map getCollectedDomains() + { + requireNonNull(collectedDomains, "collectedDomains is null"); + return collectedDomains; + } + } }