This is an automated email from the ASF dual-hosted git repository.
gianm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 8ff36f7ef89 fix: safely publish granularity interval lookups (#20292)
8ff36f7ef89 is described below
commit 8ff36f7ef89bccceb682d6a46d246a16d2b561d0
Author: Frank Chen <[email protected]>
AuthorDate: Thu Sep 10 11:11:33 2026 +0800
fix: safely publish granularity interval lookups (#20292)
---
.../indexer/granularity/BaseGranularitySpec.java | 20 +--
.../granularity/BaseGranularitySpecTest.java | 150 +++++++++++++++++++++
2 files changed, 162 insertions(+), 8 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/indexer/granularity/BaseGranularitySpec.java
b/processing/src/main/java/org/apache/druid/indexer/granularity/BaseGranularitySpec.java
index f2ddc999829..74ca5fe7549 100644
---
a/processing/src/main/java/org/apache/druid/indexer/granularity/BaseGranularitySpec.java
+++
b/processing/src/main/java/org/apache/druid/indexer/granularity/BaseGranularitySpec.java
@@ -21,6 +21,8 @@ package org.apache.druid.indexer.granularity;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Optional;
+import com.google.common.base.Supplier;
+import com.google.common.base.Suppliers;
import com.google.common.collect.Iterators;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.granularity.Granularities;
@@ -84,19 +86,24 @@ public abstract class BaseGranularitySpec implements
GranularitySpec
*/
protected static class LookupIntervalBuckets
{
- private final Iterable<Interval> intervalIterable;
- private final TreeSet<Interval> intervals;
+ private final Supplier<TreeSet<Interval>> intervals;
/**
* @param intervalIterable The intervals to materialize
*/
public LookupIntervalBuckets(Iterable<Interval> intervalIterable)
{
- this.intervalIterable = intervalIterable;
// The tree set will be materialized on demand (see below) to avoid
client code
// blowing up when constructing this data structure and when the
// number of intervals is very large...
- this.intervals = new TreeSet<>(Comparators.intervalsByStartThenEnd());
+ // Memoization serializes initialization and safely publishes only the
complete tree to concurrent readers.
+ this.intervals = Suppliers.memoize(() -> {
+ final TreeSet<Interval> materialized = new
TreeSet<>(Comparators.intervalsByStartThenEnd());
+ if (intervalIterable != null) {
+ Iterators.addAll(materialized, intervalIterable.iterator());
+ }
+ return materialized;
+ });
}
/**
@@ -132,10 +139,7 @@ public abstract class BaseGranularitySpec implements
GranularitySpec
*/
public TreeSet<Interval> materializedIntervals()
{
- if (intervalIterable != null && intervalIterable.iterator().hasNext() &&
intervals.isEmpty()) {
- Iterators.addAll(intervals, intervalIterable.iterator());
- }
- return intervals;
+ return intervals.get();
}
}
}
diff --git
a/processing/src/test/java/org/apache/druid/indexer/granularity/BaseGranularitySpecTest.java
b/processing/src/test/java/org/apache/druid/indexer/granularity/BaseGranularitySpecTest.java
new file mode 100644
index 00000000000..62db35840ff
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/indexer/granularity/BaseGranularitySpecTest.java
@@ -0,0 +1,150 @@
+/*
+ * 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.druid.indexer.granularity;
+
+import com.google.common.base.Optional;
+import org.apache.druid.java.util.common.Intervals;
+import org.joda.time.Interval;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class BaseGranularitySpecTest
+{
+ @Test
+ public void testConcurrentLookupDuringMaterialization() throws Exception
+ {
+ final Interval first = Intervals.of("2026-01-01/2026-01-02");
+ final Interval second = Intervals.of("2026-01-02/2026-01-03");
+ final CountDownLatch partiallyMaterialized = new CountDownLatch(1);
+ final CountDownLatch finishMaterialization = new CountDownLatch(1);
+ final BaseGranularitySpec.LookupIntervalBuckets buckets = new
BaseGranularitySpec.LookupIntervalBuckets(
+ () -> new Iterator<>()
+ {
+ private final Iterator<Interval> delegate = List.of(first,
second).iterator();
+
+ @Override
+ public boolean hasNext()
+ {
+ return delegate.hasNext();
+ }
+
+ @Override
+ public Interval next()
+ {
+ final Interval interval = delegate.next();
+ if (interval.equals(second)) {
+ // The first interval has been inserted, but the second has not
been returned to the builder.
+ partiallyMaterialized.countDown();
+ try {
+ Assertions.assertTrue(finishMaterialization.await(10,
TimeUnit.SECONDS));
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+ return interval;
+ }
+ }
+ );
+ final FutureTask<TreeSet<Interval>> materialization = new
FutureTask<>(buckets::materializedIntervals);
+ final FutureTask<Optional<Interval>> lookup = new FutureTask<>(() ->
buckets.bucketInterval(second.getStart()));
+ final Thread builder = new Thread(materialization, "interval-builder");
+ final Thread reader = new Thread(lookup, "interval-reader");
+ builder.setDaemon(true);
+ reader.setDaemon(true);
+
+ try {
+ builder.start();
+ Assertions.assertTrue(partiallyMaterialized.await(10, TimeUnit.SECONDS));
+ reader.start();
+
+ // Wait for the reader to either contend on initialization or return the
baseline's partial lookup.
+ // This avoids relying on a sleep to assume that the second lookup has
run.
+ final long startNanos = System.nanoTime();
+ while (!lookup.isDone()
+ && reader.getState() != Thread.State.BLOCKED
+ && System.nanoTime() - startNanos < TimeUnit.SECONDS.toNanos(10))
{
+ Thread.sleep(1);
+ }
+ Assertions.assertTrue(lookup.isDone() || reader.getState() ==
Thread.State.BLOCKED, "Reader did not reach lookup");
+ finishMaterialization.countDown();
+
+ Assertions.assertEquals(Optional.of(second), lookup.get(10,
TimeUnit.SECONDS));
+ Assertions.assertEquals(List.of(first, second),
List.copyOf(materialization.get(10, TimeUnit.SECONDS)));
+ Assertions.assertSame(materialization.get(),
buckets.materializedIntervals());
+ }
+ finally {
+ finishMaterialization.countDown();
+ builder.join(10_000);
+ reader.join(10_000);
+ }
+ }
+
+ @Test
+ public void testLazyMaterialization()
+ {
+ final AtomicInteger iterations = new AtomicInteger();
+ final Interval interval = Intervals.of("2026-01-01/2026-01-02");
+ final BaseGranularitySpec.LookupIntervalBuckets buckets = new
BaseGranularitySpec.LookupIntervalBuckets(() -> {
+ iterations.incrementAndGet();
+ return List.of(interval).iterator();
+ });
+
+ Assertions.assertEquals(0, iterations.get());
+ Assertions.assertEquals(Optional.of(interval),
buckets.bucketInterval(interval.getStart()));
+ Assertions.assertEquals(List.of(interval),
List.copyOf(buckets.materializedIntervals()));
+ Assertions.assertEquals(interval, buckets.iterator().next());
+ Assertions.assertEquals(1, iterations.get());
+ }
+
+ @Test
+ public void testEmptyMaterializationIsCached()
+ {
+ final AtomicInteger iterations = new AtomicInteger();
+ final BaseGranularitySpec.LookupIntervalBuckets buckets = new
BaseGranularitySpec.LookupIntervalBuckets(() -> {
+ iterations.incrementAndGet();
+ return Collections.emptyIterator();
+ });
+
+ Assertions.assertEquals(0, iterations.get());
+ final TreeSet<Interval> intervals = buckets.materializedIntervals();
+ Assertions.assertTrue(intervals.isEmpty());
+ Assertions.assertSame(intervals, buckets.materializedIntervals());
+ Assertions.assertEquals(1, iterations.get());
+ }
+
+ @Test
+ public void testNullIntervals()
+ {
+ final BaseGranularitySpec.LookupIntervalBuckets buckets = new
BaseGranularitySpec.LookupIntervalBuckets(null);
+ Assertions.assertTrue(buckets.materializedIntervals().isEmpty());
+ Assertions.assertFalse(buckets.iterator().hasNext());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
