This is an automated email from the ASF dual-hosted git repository.
kfaraz 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 452b15b2662 perf: Interval tree for managing segment metadata in
memory (#19138)
452b15b2662 is described below
commit 452b15b2662136459b2289d08b6eb89232d27c18
Author: pirvtech <[email protected]>
AuthorDate: Sat Aug 1 09:42:01 2026 -0700
perf: Interval tree for managing segment metadata in memory (#19138)
Segment metadata stored in memory of the Historicals, is used when looking
up segments
that match an interval for query and segment loading purposes. Currently
this is a serial scan
that goes through all segments metadata in ascending start time order to
find matching segments.
This changes introduces an Interval Tree as a more efficient way to store
segment metadata in
memory, to speed up searches for segments, and cut down search times from
O(n) to O(logn).
Changes:
- Add IntervalTreeMap
- Add SegmentTimelineConfig to enable fastIntervalSearch
- Update SegmentManager and VersionedIntervalTimeline to use
IntervalTreeMap when enabled
- Add tests
---
.../druid/msq/exec/MSQCompactionTaskRunTest.java | 4 +
.../druid/java/util/common/guava/Comparators.java | 39 +
.../org/apache/druid/timeline/IntervalTreeMap.java | 880 +++++++++++++++++++++
.../druid/timeline/VersionedIntervalTimeline.java | 125 ++-
.../apache/druid/timeline/IntervalTreeMapTest.java | 550 +++++++++++++
.../VersionedIntervalTimelineSpecificDataTest.java | 22 +-
.../timeline/VersionedIntervalTimelineTest.java | 20 +-
.../VersionedIntervalTimelineTestBase.java | 7 +-
.../org/apache/druid/guice/StorageNodeModule.java | 2 +
.../segment/indexing/SegmentTimelineConfig.java | 49 ++
.../org/apache/druid/server/SegmentManager.java | 22 +-
.../org/apache/druid/sql/guice/SqlModuleTest.java | 6 +
12 files changed, 1693 insertions(+), 33 deletions(-)
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQCompactionTaskRunTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQCompactionTaskRunTest.java
index fa99e238c46..0435f13d6a8 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQCompactionTaskRunTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQCompactionTaskRunTest.java
@@ -86,6 +86,7 @@ import org.apache.druid.segment.DataSegmentsWithSchemas;
import org.apache.druid.segment.IndexSpec;
import org.apache.druid.segment.QueryableIndexSegment;
import org.apache.druid.segment.ReferenceCountedSegmentProvider;
+import org.apache.druid.segment.indexing.SegmentTimelineConfig;
import org.apache.druid.segment.indexing.TuningConfig;
import org.apache.druid.segment.loading.AcquireSegmentAction;
import org.apache.druid.segment.loading.AcquireSegmentResult;
@@ -233,6 +234,8 @@ public class MSQCompactionTaskRunTest extends
CompactionTaskRunBase
((InjectableValues.Std)
objectMapper.getInjectableValues()).addValue(GroupingEngine.class,
groupingEngine);
((InjectableValues.Std)
objectMapper.getInjectableValues()).addValue(QueryToolChestWarehouse.class,
null);
+ SegmentTimelineConfig segmentTimelineConfig =
mock(SegmentTimelineConfig.class);
+
Module modules = Modules.combine(
new DruidGuiceExtensions(),
new LifecycleModule(),
@@ -250,6 +253,7 @@ public class MSQCompactionTaskRunTest extends
CompactionTaskRunBase
.toInstance(new
ForwardingQueryProcessingPool(Execs.singleThreaded("Test-runner-processing-pool"))),
binder ->
binder.bind(ObjectMapper.class).annotatedWith(Json.class).toInstance(objectMapper),
binder ->
binder.bind(SegmentCacheManager.class).toInstance(segmentCacheManager),
+ binder ->
binder.bind(SegmentTimelineConfig.class).toInstance(segmentTimelineConfig),
binder ->
binder.bind(VirtualStorageManager.class).toInstance(MSQTestBase.makeNilVirtualStorageManager()),
binder -> binder.bind(GroupingEngine.class).toInstance(groupingEngine)
);
diff --git
a/processing/src/main/java/org/apache/druid/java/util/common/guava/Comparators.java
b/processing/src/main/java/org/apache/druid/java/util/common/guava/Comparators.java
index 618698c4ac4..04e4f7f0fc0 100644
---
a/processing/src/main/java/org/apache/druid/java/util/common/guava/Comparators.java
+++
b/processing/src/main/java/org/apache/druid/java/util/common/guava/Comparators.java
@@ -119,6 +119,34 @@ public class Comparators
}
};
+ private static final Comparator<Interval> INTERVAL_BY_START = new
Comparator<>()
+ {
+ private final DateTimeComparator dateTimeComp =
DateTimeComparator.getInstance();
+
+ @Override
+ public int compare(Interval lhs, Interval rhs)
+ {
+ if (lhs.getChronology().equals(rhs.getChronology())) {
+ return Long.compare(lhs.getStartMillis(), rhs.getStartMillis());
+ }
+ return dateTimeComp.compare(lhs.getStart(), rhs.getStart());
+ }
+ };
+
+ private static final Comparator<Interval> INTERVAL_BY_END = new
Comparator<>()
+ {
+ private final DateTimeComparator dateTimeComp =
DateTimeComparator.getInstance();
+
+ @Override
+ public int compare(Interval lhs, Interval rhs)
+ {
+ if (lhs.getChronology().equals(rhs.getChronology())) {
+ return Long.compare(lhs.getEndMillis(), rhs.getEndMillis());
+ }
+ return dateTimeComp.compare(lhs.getEnd(), rhs.getEnd());
+ }
+ };
+
@Deprecated
public static Comparator<Interval> intervals()
{
@@ -135,4 +163,15 @@ public class Comparators
return INTERVAL_BY_END_THEN_START;
}
+ public static Comparator<Interval> intervalsByStart()
+ {
+ return INTERVAL_BY_START;
+ }
+
+ public static Comparator<Interval> intervalsByEnd()
+ {
+ return INTERVAL_BY_END;
+ }
+
+
}
diff --git
a/processing/src/main/java/org/apache/druid/timeline/IntervalTreeMap.java
b/processing/src/main/java/org/apache/druid/timeline/IntervalTreeMap.java
new file mode 100644
index 00000000000..5a59966e8f0
--- /dev/null
+++ b/processing/src/main/java/org/apache/druid/timeline/IntervalTreeMap.java
@@ -0,0 +1,880 @@
+/*
+ * 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.timeline;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Predicate;
+import org.joda.time.Interval;
+import org.joda.time.base.BaseInterval;
+
+import javax.validation.constraints.NotNull;
+import java.util.AbstractMap;
+import java.util.AbstractSet;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.NavigableSet;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.function.BiConsumer;
+
+/**
+ * A variation of Interval Trees (https://en.wikipedia.org/wiki/Interval_tree)
+ * Custom optimizations for faster interval search and additional support for
specific joda Interval comparator
+ * arithmetic used in the project. Additionally, it implements NavigableMap
interface with relevant methods for
+ * traversal of the entries.
+ * <p>
+ * <p>
+ * Multiple different intervals can be added to the tree. It can then be
searched to find all intervals matching a given
+ * interval. The user specifies the match condition, such as encompassing the
given interval, overlapping, etc. The
+ * search can return multiple results as multiple intervals in the tree could
match the criteria.
+ * <p>
+ * Using the tree, reduces the search time from O(N) iterating through all the
intervals, to roughly O(log2(N)).
+ * Furthermore, a value can be associated with each interval, which is also
returned in the search result.
+ *
+ * <p>
+ * The tree is a binary search tree sorted by interval start time. The
intervals are stored as nodes in the tree.
+ * Additional state containing the minimum and maximum interval bounds of the
entire subtree under a node is also
+ * stored in each node. This helps speed up the search for matching intervals
by skipping unsuitable subtrees that will
+ * not contain a matching candidate interval.
+ * <p>
+ * To optimize the balancing cost w.r.t the operation time, the tree is not
balanced on every modification operation.
+ * Rather, a configurable imbalance tolerance from the theoretical ideal
height of log2(N) is allowed, breaching which
+ * triggers the rebalance.
+ * <p>
+ * Not thread safe.
+ * <p>
+ */
+public class IntervalTreeMap<T> extends AbstractMap<Interval, T> implements
NavigableMap<Interval, T>
+{
+ // The compartor for comparing the interval start timnes
+ private final Comparator<Interval> startComparator;
+ // The comparator for comparing interval end times
+ private final Comparator<Interval> endComparator;
+
+ @VisibleForTesting
+ private Node<T> root;
+ private int size;
+
+ private int imbalanceTolerance;
+
+ private final EntrySet entrySet = new EntrySet();
+
+ public IntervalTreeMap()
+ {
+ this(Comparator.comparingLong(BaseInterval::getStartMillis),
+ Comparator.comparingLong(BaseInterval::getEndMillis));
+ }
+
+ public IntervalTreeMap(Comparator<Interval> startComparator,
Comparator<Interval> endComparator)
+ {
+ this(startComparator, endComparator, 50);
+ }
+
+ public IntervalTreeMap(Comparator<Interval> startComparator,
Comparator<Interval> endComparator, int imbalanceTolerance)
+ {
+ this.startComparator = startComparator;
+ this.endComparator = endComparator;
+ this.imbalanceTolerance = imbalanceTolerance;
+ }
+
+ /**
+ * Returns the allowed tolerance between the right and left branches of the
tree before triggering a rebalance.
+ * The tolerance is expressed as a percentage deviation from ideal tree
height.
+ * @return The imbalance tolerance
+ */
+ public int getImbalanceTolerance()
+ {
+ return imbalanceTolerance;
+ }
+
+ /**
+ * Set the allowed tolerance between the right and left branches.
+ * The setting will take effect on the next modification operation on the
tree, such as an add or a delete.
+ * @param imbalanceTolerance The tolerance
+ */
+ public void setImbalanceTolerance(int imbalanceTolerance)
+ {
+ this.imbalanceTolerance = imbalanceTolerance;
+ }
+
+ static class Node<T> implements Map.Entry<Interval, T>
+ {
+ Interval interval;
+ T value;
+ int height;
+ // The full interval range of the subtree formed by this Node
+ Interval range;
+ Node<T> parent;
+ Node<T> left;
+ Node<T> right;
+
+ private static final String PRINT_FORMAT = "{%n"
+ + "%sinterval = %s%n"
+ + "%svalue = %s%n"
+ + "%sheight = %d%n"
+ + "%srange = %s%n"
+ + "%sleft = %s%n"
+ + "%sright = %s%n"
+ + "%s}";
+
+ private String print(int level)
+ {
+ String prefix = "\t".repeat(level);
+ String eprefix = "\t".repeat(level - 1);
+ return String.format(Locale.ENGLISH, PRINT_FORMAT,
+ prefix, interval, prefix, value, prefix, height,
+ prefix, range,
+ prefix, (left != null) ? left.print(level + 1) :
null,
+ prefix, (right != null) ? right.print(level + 1)
: null,
+ eprefix
+ );
+ }
+
+ @Override
+ public Interval getKey()
+ {
+ return interval;
+ }
+
+ @Override
+ public T getValue()
+ {
+ return value;
+ }
+
+ @Override
+ public T setValue(T value)
+ {
+ T oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+ }
+
+ @Override
+ public T put(Interval interval, T value)
+ {
+ //root = insert(root, interval, value);
+ T oldValue = insert(null, false, interval, value);
+ checkRebalance();
+ return oldValue;
+ }
+
+ private T insert(Node<T> parent, boolean left, Interval interval, T value)
+ {
+ // Passing parent so that when a new node is created, it can be added to
parent, and we can still use return value
+ // for another purpose, namely returning the old value if the key already
exists in the tree
+ Node<T> node;
+ if (parent == null) {
+ node = root;
+ } else if (left) {
+ node = parent.left;
+ } else {
+ node = parent.right;
+ }
+
+ if (node == null) {
+ node = new Node<>();
+ node.interval = interval;
+ node.value = value;
+ node.height = 0;
+ node.range = interval;
+ if (root == null) {
+ root = node;
+ } else if (left) {
+ setLeftNode(parent, node);
+ } else {
+ setRightNode(parent, node);
+ }
+ ++size;
+ return null;
+ }
+
+ T oldValue;
+
+ int cmp = compareInterval(interval, node.interval);
+
+ // If exact interval already exists, just replace the value and return
+ if (cmp == 0) {
+ oldValue = node.value;
+ node.value = value;
+ return oldValue;
+ }
+
+ if (cmp < 0) {
+ // Go to the left
+ oldValue = insert(node, true, interval, value);
+ } else {
+ // Go to the right
+ oldValue = insert(node, false, interval, value);
+ }
+ recomputeState(node);
+
+ //return node;
+ return oldValue;
+ }
+
+ @Override
+ public T get(Object key)
+ {
+ if (!Interval.class.isAssignableFrom(key.getClass())) {
+ throw new ClassCastException("key must be an instance of Interval");
+ }
+ Interval interval = (Interval) key;
+
+ T value = null;
+ Node<T> node = root;
+ while (node != null) {
+ int cmp = compareInterval(node.getKey(), interval);
+ if (cmp == 0) {
+ value = node.value;
+ break;
+ } else if (cmp > 0) {
+ node = node.left;
+ } else {
+ node = node.right;
+ }
+ }
+ return value;
+ }
+
+ private int compareInterval(Interval interval1, Interval interval2)
+ {
+ int cmp = startComparator.compare(interval1, interval2);
+ if (cmp == 0) {
+ return endComparator.compare(interval1, interval2);
+ }
+ return cmp;
+ }
+
+ public Map<Interval, T> findEncompassing(Interval interval)
+ {
+ return findMatching(i -> i.contains(interval));
+ }
+
+ public Map<Interval, T> findOverlapping(Interval interval)
+ {
+ return findMatching(i -> i.overlaps(interval));
+ }
+
+ /**
+ * Get all entries matching a given condition
+ * @param condition The match condition
+ *
+ * This condition should not only return true when a node matches the
condition but also when a child node range
+ * matches. It is a convenience method for {@link
#forEachMatching(Predicate, Predicate, BiConsumer)} and see the
+ * method's documentation for more information. It calls the method with
rangeCondition set to be same as condition.
+ */
+ public Map<Interval, T> findMatching(Predicate<Interval> condition)
+ {
+ Map<Interval, T> result = new HashMap<>();
+ forEachMatching(condition, result::put);
+ return result;
+ }
+
+ /**
+ * Find entries matching a given condition by doing a full traversal.
+ * @param condition The match condition
+ *
+ * The method traverses through all the nodes of the tree looking for
matches.
+ */
+ public Map<Interval, T> findMatchingFullTraversal(Predicate<Interval>
condition)
+ {
+ Map<Interval, T> result = new HashMap<>();
+ forEachMatchingFullTraversal(condition, result::put);
+ return result;
+ }
+
+ /**
+ * Perform on action for matching nodes.
+ * @param condition The match condition
+ * @param action The action
+ *
+ * This condition should not only return true when a node matches the
condition but also when the child node range
+ * matches. It is a convenience method for {@link
#forEachMatching(Predicate, Predicate, BiConsumer)} and see the
+ * method's documentation for more information. It calls the method with
rangeCondition set to be same as condition.
+ */
+ public void forEachMatching(Predicate<Interval> condition,
BiConsumer<Interval, T> action)
+ {
+ forEachMatching(condition, condition, action);
+ }
+
+ /**
+ * Perform on action for matching nodes by doing a full traversal.
+ * @param condition The match condition
+ * @param action The action
+ *
+ * The method traverses through all the nodes of the tree looking for
matches.
+ */
+ public void forEachMatchingFullTraversal(Predicate<Interval> condition,
BiConsumer<Interval, T> action)
+ {
+ forEachMatching(condition, null, action);
+ }
+
+ /**
+ * Perform an action for matching nodes
+ * @param condition The condition to match for the node
+ * @param rangeCondition The condition to check a child node for, to
determine whether to traverse the subtree
+ * @param action The action to perform
+ *
+ * The rangeCondition is applied on the interval range of the child node and
only if the condition returns true is the
+ * child subtree traversed. Interval range is the min start time to max end
time for all the nodes in the child
+ * subtree. This is a lookup speedup optimization. If rangeCondition is
null, the check is skipped and all the
+ * children are traversed to find matches.
+ *
+ * In some cases such as finding nodes overlapping the given interval or
encompassing the given interval, the same
+ * predicate can be used for condition and rangeCondition. In other
situations a full traversal maybe needed and a
+ * null can be passed in for rangeCondition. There are helper methods for
these.
+ */
+ public void forEachMatching(Predicate<Interval> condition,
Predicate<Interval> rangeCondition, BiConsumer<Interval, T> action)
+ {
+ forEachMatching(root, condition, rangeCondition, action);
+ }
+
+ private void forEachMatching(Node<T> node, Predicate<Interval> condition,
Predicate<Interval> rangeCondition, BiConsumer<Interval, T> action)
+ {
+
+ if (node == null) {
+ return;
+ }
+
+ // Process in-order
+
+ // Search left
+ if ((node.left != null) && ((rangeCondition == null) ||
rangeCondition.apply(node.left.range))) {
+ forEachMatching(node.left, condition, rangeCondition, action);
+ }
+
+ if (condition.apply(node.interval)) {
+ action.accept(node.interval, node.value);
+ }
+
+ // Search right
+ if (node.right != null && ((rangeCondition == null) ||
rangeCondition.apply(node.right.range))) {
+ forEachMatching(node.right, condition, rangeCondition, action);
+ }
+ }
+
+ @Override
+ public T remove(Object key)
+ {
+ return remove((Interval) key);
+ }
+
+ public T remove(Interval interval)
+ {
+ List<T> oldValue = new ArrayList<>(1);
+ root = removeNode(root, interval, oldValue);
+ if (root != null) {
+ root.parent = null;
+ }
+ checkRebalance();
+ return oldValue.size() == 1 ? oldValue.get(0) : null;
+ }
+
+ private Node<T> removeNode(Node<T> node, Interval interval, List<T> oldValue)
+ {
+ // When deleting a node, try to replace it with the right most leaf of the
left sub-tree.
+ // If it is does not exist, i.e., the bottom most right node in the left
subtree only has a left child and does not
+ // have a right child, this node becomes the replacement. Also, in this
scenario, the left child (subtree) of this
+ // node is moved up to its parent as the parent's right child.
+ if (node == null) {
+ return null;
+ }
+
+ int cmp = compareInterval(interval, node.interval);
+
+ if (cmp == 0) {
+ // This is the node to delete
+ --size;
+ oldValue.add(node.value);
+ if ((node.left != null) && (node.right != null)) {
+ // Make the right bottom most child in the left subtree of the node,
the new node at the current level
+ Node<T> left = node.left;
+ Node<T> newNode = unlinkRightLeaf(left);
+ // Make the current left and right children, the left and right
children of the new node respectively.
+ // However, if the new node turns out to be the same as the left node,
it means the left node did not have any
+ // right child. In this case, only set its right child to be the
current node's right child.
+ if (left != newNode) {
+ // A right child exists
+ setLeftNode(newNode, left);
+ }
+ setRightNode(newNode, node.right);
+ recomputeState(newNode);
+ return newNode;
+ } else if (node.left != null) {
+ // Right node is null, make the left node the new node at current level
+ return node.left;
+ } else if (node.right != null) {
+ // Left node is null, make the right node the new node at current level
+ return node.right;
+ }
+ return null;
+ }
+
+ // Current node didn't match, search children
+ if (cmp < 0) {
+ Node<T> left = removeNode(node.left, interval, oldValue);
+ setLeftNode(node, left);
+ } else {
+ Node<T> right = removeNode(node.right, interval, oldValue);
+ setRightNode(node, right);
+ }
+
+ // Update our state as a modification may have happened somewhere in our
subtree
+ recomputeState(node);
+
+ return node;
+ }
+
+ private Node<T> unlinkRightLeaf(Node<T> node)
+ {
+ if (node.right == null) {
+ return node;
+ } else {
+ Node<T> rnode = unlinkRightLeaf(node.right);
+ // If the right node has a left child, make it new right child
+ if (rnode == node.right) {
+ setRightNode(node, rnode.left);
+ rnode.left = null;
+ }
+ recomputeState(node);
+ return rnode;
+ }
+ }
+
+ private void inOrderTraverse(Node<T> node, List<Node<T>> nodes)
+ {
+ if (node == null) {
+ return;
+ }
+ inOrderTraverse(node.left, nodes);
+ nodes.add(node);
+ inOrderTraverse(node.right, nodes);
+ }
+
+ public void rebalance()
+ {
+ // In order traversal followed by repeated binary segmentation of the list
+ List<Node<T>> nodes = new ArrayList<>(size);
+ inOrderTraverse(root, nodes);
+ root = constructTree(nodes, 0, nodes.size());
+ root.parent = null;
+ }
+
+ private Node<T> constructTree(List<Node<T>> nodes, int start, int end)
+ {
+ if (start == end) {
+ return null;
+ }
+ int mid = (start + end - 1) / 2;
+ Node<T> node = nodes.get(mid);
+
+ Node<T> left = constructTree(nodes, start, mid);
+ setLeftNode(node, left);
+
+ Node<T> right = constructTree(nodes, mid + 1, end);
+ setRightNode(node, right);
+
+ recomputeState(node);
+ return node;
+ }
+
+ @Override
+ public Map.Entry<Interval, T> lowerEntry(Interval key)
+ {
+ Node<T> lnode = null;
+ Node<T> node = root;
+ while (node != null) {
+ // Since we want to return a smaller entry even when there is an exact
match, go left in the equality case too
+ if (compareInterval(key, node.getKey()) <= 0) {
+ node = node.left;
+ } else {
+ lnode = node;
+ node = node.right;
+ }
+ }
+ return lnode;
+ }
+
+ @Override
+ public Interval lowerKey(Interval key)
+ {
+ Map.Entry<Interval, T> entry = lowerEntry(key);
+ return entry != null ? entry.getKey() : null;
+ }
+
+ @Override
+ public Map.Entry<Interval, T> floorEntry(Interval key)
+ {
+ Node<T> fnode = null;
+ Node<T> node = root;
+ while (node != null) {
+ int cmp = compareInterval(node.getKey(), key);
+ if (cmp == 0) {
+ fnode = node;
+ break;
+ } else if (cmp > 0) {
+ node = node.left;
+ } else {
+ fnode = node;
+ node = node.right;
+ }
+ }
+ return fnode;
+ }
+
+ @Override
+ public Interval floorKey(Interval key)
+ {
+ Map.Entry<Interval, T> entry = floorEntry(key);
+ return entry != null ? entry.getKey() : null;
+ }
+
+ @Override
+ public Map.Entry<Interval, T> ceilingEntry(Interval key)
+ {
+ Node<T> cnode = null;
+ Node<T> node = root;
+ while (node != null) {
+ int cmp = compareInterval(node.getKey(), key);
+ if (cmp == 0) {
+ cnode = node;
+ break;
+ } else if (cmp > 0) {
+ cnode = node;
+ node = node.left;
+ } else {
+ node = node.right;
+ }
+ }
+ return cnode;
+ }
+
+ @Override
+ public Interval ceilingKey(Interval key)
+ {
+ Entry<Interval, T> entry = ceilingEntry(key);
+ return entry != null ? entry.getKey() : null;
+ }
+
+ @Override
+ public Map.Entry<Interval, T> higherEntry(Interval key)
+ {
+ Node<T> hnode = null;
+ Node<T> node = root;
+ while (node != null) {
+ if (compareInterval(key, node.getKey()) < 0) {
+ hnode = node;
+ node = node.left;
+ } else {
+ node = node.right;
+ }
+ }
+ return hnode;
+ }
+
+ @Override
+ public Interval higherKey(Interval key)
+ {
+ Entry<Interval, T> entry = higherEntry(key);
+ return entry != null ? entry.getKey() : null;
+ }
+
+ @Override
+ public Map.Entry<Interval, T> firstEntry()
+ {
+ return firstEntry(root);
+ }
+
+ @Override
+ public Map.Entry<Interval, T> lastEntry()
+ {
+ if (root == null) {
+ return null;
+ }
+ Node<T> node = root;
+ while (node.right != null) {
+ node = node.right;
+ }
+ return node;
+ }
+
+ @Override
+ public Interval firstKey()
+ {
+ Map.Entry<Interval, T> entry = firstEntry();
+ return entry != null ? entry.getKey() : null;
+ }
+
+ @Override
+ public Interval lastKey()
+ {
+ Map.Entry<Interval, T> entry = lastEntry();
+ return entry != null ? entry.getKey() : null;
+ }
+
+ @Override
+ public Map.Entry<Interval, T> pollFirstEntry()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map.Entry<Interval, T> pollLastEntry()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public NavigableMap<Interval, T> descendingMap()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public NavigableSet<Interval> navigableKeySet()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public NavigableSet<Interval> descendingKeySet()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public NavigableMap<Interval, T> subMap(Interval fromKey, boolean
fromInclusive, Interval toKey, boolean toInclusive)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public NavigableMap<Interval, T> headMap(Interval toKey, boolean inclusive)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public NavigableMap<Interval, T> tailMap(Interval fromKey, boolean inclusive)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Comparator<? super Interval> comparator()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SortedMap<Interval, T> subMap(Interval fromKey, Interval toKey)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SortedMap<Interval, T> headMap(Interval toKey)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SortedMap<Interval, T> tailMap(Interval fromKey)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ private void recomputeState(Node<T> node)
+ {
+ int lheight = (node.left != null) ? node.left.height : -1;
+ int rheight = (node.right != null) ? node.right.height : -1;
+ node.height = Math.max(lheight, rheight) + 1;
+ node.range = computeRange(node.interval, node.left, node.right);
+ }
+
+ @Override
+ public void clear()
+ {
+ root = null;
+ size = 0;
+ }
+
+ @Override
+ public @NotNull Set<Map.Entry<Interval, T>> entrySet()
+ {
+ return entrySet;
+ }
+
+ @Override
+ public int size()
+ {
+ return size;
+ }
+
+ @VisibleForTesting
+ // returns the number of edges from root to leaf along the longest path
+ int height()
+ {
+ return (root != null) ? root.height : -1;
+ }
+
+ class EntrySet extends AbstractSet<Map.Entry<Interval, T>>
+ {
+
+ // Currently this returns a distinct collection when iterating
+ @Override
+ public Iterator<Map.Entry<Interval, T>> iterator()
+ {
+ return new EntrySetIterator();
+ }
+
+ @Override
+ public int size()
+ {
+ return IntervalTreeMap.this.size;
+ }
+
+ class EntrySetIterator implements Iterator<Map.Entry<Interval, T>>
+ {
+
+ Node<T> current = firstEntry(IntervalTreeMap.this.root);
+
+ @Override
+ public boolean hasNext()
+ {
+ return (current != null);
+ }
+
+ @Override
+ public Entry<Interval, T> next()
+ {
+ Entry<Interval, T> entry = current;
+ if (entry == null) {
+ return entry;
+ }
+ // Move current to next node
+ if (current.right != null) {
+ current = firstEntry(current.right);
+ } else {
+ // No more right children, go up one level to the parent.
+ // However, if the current node is right child of parent, keep going
up till you find a parent who is on the
+ // right side
+ Node<T> prev;
+ do {
+ prev = current;
+ current = current.parent;
+ } while ((current != null) && (current.right == prev));
+ }
+ return entry;
+ }
+ }
+
+ }
+
+ /**
+ * Perform a tree rebalance if the imbalance between the left and right
sides of the tree has increased beyond a
+ * tolerated limit, as opposed to rebalancing all the time. This is to done
to strike a balance between performance
+ * degradation arising from an imbalance tree and the processing overheard
of rebalancing each time the contents of
+ * the tree changes.
+ *
+ * The limit is defined using a configurable tolerance percentage in excess
of an ideal balanced tree height for the
+ * number of entries in the tree.
+ */
+ private void checkRebalance()
+ {
+ if (root != null) {
+ int ideal = (int) Math.floor(Math.log10(size + 1) / Math.log10(2));
+ double tolerance = ideal * imbalanceTolerance / 100.0;
+ int threshold = ideal + (int) tolerance;
+ if (root.height > threshold) {
+ rebalance();
+ }
+ }
+ }
+
+ private Node<T> firstEntry(Node<T> node)
+ {
+ if (node == null) {
+ return null;
+ }
+ while (node.left != null) {
+ node = node.left;
+ }
+ return node;
+ }
+
+ private void setLeftNode(Node<T> node, Node<T> left)
+ {
+ if (node.left != left) {
+ node.left = left;
+ if (left != null) {
+ left.parent = node;
+ }
+ }
+ }
+
+ private void setRightNode(Node<T> node, Node<T> right)
+ {
+ if (node.right != right) {
+ node.right = right;
+ if (right != null) {
+ right.parent = node;
+ }
+ }
+ }
+
+ /**
+ * @return The contents of the tree as a string
+ */
+ @Override
+ public String toString()
+ {
+ return (root != null) ? root.print(1) : "";
+ }
+
+ @SafeVarargs
+ private Interval computeRange(Interval interval, Node<T>... nodes)
+ {
+ // Find the intervals that have the minimum start and the maximum end
+ Interval min = interval;
+ Interval max = interval;
+ for (Node<T> node : nodes) {
+ if (node != null) {
+ if (startComparator.compare(node.range, min) < 0) {
+ min = node.range;
+ }
+ if (endComparator.compare(node.range, max) > 0) {
+ max = node.range;
+ }
+ }
+ }
+ // Return an interval with the min and max
+ return interval.withStart(min.getStart()).withEnd(max.getEnd());
+ }
+
+}
diff --git
a/processing/src/main/java/org/apache/druid/timeline/VersionedIntervalTimeline.java
b/processing/src/main/java/org/apache/druid/timeline/VersionedIntervalTimeline.java
index a2f2699c41c..60c434df949 100644
---
a/processing/src/main/java/org/apache/druid/timeline/VersionedIntervalTimeline.java
+++
b/processing/src/main/java/org/apache/druid/timeline/VersionedIntervalTimeline.java
@@ -27,6 +27,7 @@ import com.google.errorprone.annotations.concurrent.GuardedBy;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.UOE;
import org.apache.druid.java.util.common.guava.Comparators;
+import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.timeline.partition.PartitionChunk;
import org.apache.druid.timeline.partition.PartitionHolder;
import org.apache.druid.utils.CollectionUtils;
@@ -74,21 +75,20 @@ import java.util.stream.StreamSupport;
public class VersionedIntervalTimeline<VersionType, ObjectType extends
Overshadowable<ObjectType>>
implements TimelineLookup<VersionType, ObjectType>
{
+ private static final Logger logger = new
Logger(VersionedIntervalTimeline.class);
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
// Below timelines stores only *visible* timelineEntries
// adjusted interval -> timelineEntry
- private final NavigableMap<Interval, TimelineEntry>
completePartitionsTimeline = new TreeMap<>(
- Comparators.intervalsByStartThenEnd()
- );
+ private final NavigableMap<Interval, TimelineEntry>
completePartitionsTimeline;
// IncompletePartitionsTimeline also includes completePartitionsTimeline
// adjusted interval -> timelineEntry
@VisibleForTesting
- final NavigableMap<Interval, TimelineEntry> incompletePartitionsTimeline =
new TreeMap<>(
- Comparators.intervalsByStartThenEnd()
- );
+ final NavigableMap<Interval, TimelineEntry> incompletePartitionsTimeline;
// true interval -> version -> timelineEntry
private final Map<Interval, TreeMap<VersionType, TimelineEntry>>
allTimelineEntries = new HashMap<>();
+ // Only instantiated and used when fastIntervalSearch is enabled
+ private IntervalTreeMap<TreeMap<VersionType, TimelineEntry>>
allTimeIntervals;
private final AtomicInteger numObjects = new AtomicInteger();
private final Comparator<? super VersionType> versionComparator;
@@ -96,15 +96,38 @@ public class VersionedIntervalTimeline<VersionType,
ObjectType extends Overshado
// Set this to true if the client needs to skip tombstones upon lookup (like
the broker)
private final boolean skipObjectsWithNoData;
+ // Set this to true to use an interval tree index for the segment intervals
+ private final boolean fastIntervalSearch;
+
public VersionedIntervalTimeline(Comparator<? super VersionType>
versionComparator)
{
this(versionComparator, false);
}
public VersionedIntervalTimeline(Comparator<? super VersionType>
versionComparator, boolean skipObjectsWithNoData)
+ {
+ this(versionComparator, skipObjectsWithNoData, false);
+ }
+
+ /**
+ * Constructor
+ * @param versionComparator The version comparator
+ * @param skipObjectsWithNoData Skip tombstones during lookup
+ * @param fastIntervalSearch Use the faster segment retrieval index based on
interval tree
+ */
+ public VersionedIntervalTimeline(Comparator<? super VersionType>
versionComparator, boolean skipObjectsWithNoData, boolean fastIntervalSearch)
{
this.versionComparator = versionComparator;
this.skipObjectsWithNoData = skipObjectsWithNoData;
+ this.fastIntervalSearch = fastIntervalSearch;
+ if (fastIntervalSearch) {
+ allTimeIntervals = new IntervalTreeMap<>(Comparators.intervalsByStart(),
Comparators.intervalsByEnd());
+ this.completePartitionsTimeline = new
IntervalTreeMap<>(Comparators.intervalsByStart(), Comparators.intervalsByEnd());
+ this.incompletePartitionsTimeline = new
IntervalTreeMap<>(Comparators.intervalsByStart(), Comparators.intervalsByEnd());
+ } else {
+ this.completePartitionsTimeline = new
TreeMap<>(Comparators.intervalsByStartThenEnd());
+ this.incompletePartitionsTimeline = new
TreeMap<>(Comparators.intervalsByStartThenEnd());
+ }
}
public static <VersionType, ObjectType extends Overshadowable<ObjectType>>
Iterable<ObjectType> getAllObjects(
@@ -210,6 +233,9 @@ public class VersionedIntervalTimeline<VersionType,
ObjectType extends Overshado
TreeMap<VersionType, TimelineEntry> versionEntry = new
TreeMap<>(versionComparator);
versionEntry.put(version, entry);
allTimelineEntries.put(interval, versionEntry);
+ if (fastIntervalSearch) {
+ allTimeIntervals.put(interval, versionEntry);
+ }
numObjects.incrementAndGet();
} else {
entry = exists.get(version);
@@ -269,6 +295,9 @@ public class VersionedIntervalTimeline<VersionType,
ObjectType extends Overshado
versionEntries.remove(version);
if (versionEntries.isEmpty()) {
allTimelineEntries.remove(interval);
+ if (fastIntervalSearch) {
+ allTimeIntervals.remove(interval);
+ }
}
remove(incompletePartitionsTimeline, interval, entry, true);
@@ -289,13 +318,40 @@ public class VersionedIntervalTimeline<VersionType,
ObjectType extends Overshado
{
lock.readLock().lock();
try {
- for (Entry<Interval, TreeMap<VersionType, TimelineEntry>> entry :
allTimelineEntries.entrySet()) {
- if (entry.getKey().equals(interval) ||
entry.getKey().contains(interval)) {
- TimelineEntry foundEntry = entry.getValue().get(version);
- if (foundEntry != null) {
- return foundEntry.getPartitionHolder().getChunk(partitionNum);
+
+ // Speed up search with an exact interval match lookup first
+ TreeMap<VersionType, TimelineEntry> versionEntries =
allTimelineEntries.get(interval);
+ if (versionEntries != null) {
+ TimelineEntry foundEntry = versionEntries.get(version);
+ if (foundEntry != null) {
+ return foundEntry.getPartitionHolder().getChunk(partitionNum);
+ }
+ }
+
+ // If an exact interval match is not found search for an encapsulating
interval
+
+ // If tree search is enabled use it else revert to checking all intervals
+ if (fastIntervalSearch) {
+ Map<Interval, TreeMap<VersionType, TimelineEntry>> possibleMatches =
allTimeIntervals.findEncompassing(interval);
+ for (Entry<Interval, TreeMap<VersionType, TimelineEntry>> entry :
possibleMatches.entrySet()) {
+ Interval possibleInterval = entry.getKey();
+ if (possibleInterval.contains(interval)) {
+ TimelineEntry foundEntry = entry.getValue().get(version);
+ if (foundEntry != null) {
+ return foundEntry.getPartitionHolder().getChunk(partitionNum);
+ }
+ }
+ }
+ } else {
+ for (Entry<Interval, TreeMap<VersionType, TimelineEntry>> entry :
allTimelineEntries.entrySet()) {
+ if (entry.getKey().contains(interval)) {
+ TimelineEntry foundEntry = entry.getValue().get(version);
+ if (foundEntry != null) {
+ return foundEntry.getPartitionHolder().getChunk(partitionNum);
+ }
}
}
+
}
return null;
@@ -747,21 +803,38 @@ public class VersionedIntervalTimeline<VersionType,
ObjectType extends Overshado
timeline = completePartitionsTimeline;
}
- for (Entry<Interval, TimelineEntry> entry : timeline.entrySet()) {
- Interval timelineInterval = entry.getKey();
- TimelineEntry val = entry.getValue();
-
- // exclude empty partition holders (i.e. tombstones) since they do not
add value
- // for higher level code...they have no data rows...
- if ((!skipObjectsWithNoData || val.partitionHolder.hasData()) &&
timelineInterval.overlaps(interval)) {
- retVal.add(
- new TimelineObjectHolder<>(
- timelineInterval,
- val.getTrueInterval(),
- val.getVersion(),
-
PartitionHolder.copyWithOnlyVisibleChunks(val.getPartitionHolder())
- )
- );
+ if (fastIntervalSearch) {
+ IntervalTreeMap<TimelineEntry> tree = (IntervalTreeMap<TimelineEntry>)
timeline;
+ tree.forEachMatching(timelineInterval ->
timelineInterval.overlaps(interval),
+ (timelineInterval, val) -> {
+ if (!skipObjectsWithNoData || val.partitionHolder.hasData()) {
+ retVal.add(
+ new TimelineObjectHolder<>(
+ timelineInterval,
+ val.getTrueInterval(),
+ val.getVersion(),
+
PartitionHolder.copyWithOnlyVisibleChunks(val.getPartitionHolder())
+ )
+ );
+ }
+ });
+ } else {
+ for (Entry<Interval, TimelineEntry> entry : timeline.entrySet()) {
+ Interval timelineInterval = entry.getKey();
+ TimelineEntry val = entry.getValue();
+
+ // exclude empty partition holders (i.e. tombstones) since they do not
add value
+ // for higher level code...they have no data rows...
+ if ((!skipObjectsWithNoData || val.partitionHolder.hasData()) &&
timelineInterval.overlaps(interval)) {
+ retVal.add(
+ new TimelineObjectHolder<>(
+ timelineInterval,
+ val.getTrueInterval(),
+ val.getVersion(),
+
PartitionHolder.copyWithOnlyVisibleChunks(val.getPartitionHolder())
+ )
+ );
+ }
}
}
diff --git
a/processing/src/test/java/org/apache/druid/timeline/IntervalTreeMapTest.java
b/processing/src/test/java/org/apache/druid/timeline/IntervalTreeMapTest.java
new file mode 100644
index 00000000000..d0eefc13ec7
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/timeline/IntervalTreeMapTest.java
@@ -0,0 +1,550 @@
+/*
+ * 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.timeline;
+
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.guava.Comparators;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.joda.time.Interval;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+public class IntervalTreeMapTest
+{
+
+ @Test
+ public void testSize()
+ {
+ IntervalTreeMap<String> tree = setupTree(baseData);
+ Assertions.assertEquals(6, tree.size(), "Size");
+ }
+
+ @Test
+ public void testPut()
+ {
+ IntervalTreeMap<String> tree = setupTree(baseData);
+ compareData(baseData, tree);
+ }
+
+ @Test
+ public void testReplace()
+ {
+ IntervalTreeMap<String> tree = setupTree(baseData);
+ Pair<Interval, String> entry = baseData.get(2);
+ Interval interval = entry.lhs;
+ String value = entry.rhs;
+ String newValue = value + "n";
+ String oldValue = tree.put(interval, newValue);
+ Assertions.assertEquals(oldValue, value, "Old value match");
+ }
+
+ @Test
+ public void testGet()
+ {
+ IntervalTreeMap<String> tree = setupTree(baseData);
+ baseData.forEach(
+ (Pair<Interval, String> item) -> {
+ Interval interval = item.lhs;
+ String evalue = item.rhs;
+ String value = tree.get(interval);
+ Assertions.assertEquals(evalue, value, "value");
+ }
+ );
+ }
+
+ @Test
+ public void testValues()
+ {
+ IntervalTreeMap<String> tree = setupTree(baseData);
+ Set<String> values = new LinkedHashSet<>(tree.values());
+ Set<String> bvalues = baseData.stream().map(entry ->
entry.rhs).collect(Collectors.toCollection(LinkedHashSet::new));
+ Assertions.assertEquals(bvalues, values, "values");
+ }
+
+ @Test
+ public void testMatch()
+ {
+ IntervalTreeMap<String> tree = setupTree(baseData);
+ Map<Interval, String> entries =
tree.findEncompassing(Intervals.of("2025-01-04T00:00:00/P1D"));
+
+ Assertions.assertEquals(1, entries.size());
+ Assertions.assertEquals("v5",
entries.get(Intervals.of("2025-01-04T00:00:00/P1D")), "Match");
+ }
+
+ @Test
+ public void testNoMatch()
+ {
+ IntervalTreeMap<String> tree = setupTree(baseData);
+ Map<Interval, String> entries =
tree.findEncompassing(Intervals.of("2025-01-07T00:00:00/P1D"));
+
+ Assertions.assertEquals(0, entries.size());
+ }
+
+ @Test
+ public void testOverlap()
+ {
+ IntervalTreeMap<String> tree = setupTree(overlapData);
+ Map<Interval, String> entries =
tree.findEncompassing(Intervals.of("2025-01-02T09:00:00/PT1H"));
+
+ Assertions.assertEquals(2, entries.size());
+ Assertions.assertEquals("v4",
entries.get(Intervals.of("2025-01-02T00:00:00/P1D")), "Day match");
+ Assertions.assertEquals("v7",
entries.get(Intervals.of("2025-01-01T00:00:00/P1Y")), "Year match");
+ }
+
+ @Test
+ public void testSparseOverlap()
+ {
+ IntervalTreeMap<String> tree = setupTree(sparseOverlapData);
+ Map<Interval, String> entries =
tree.findEncompassing(Intervals.of("2025-06-03T00:00:00/P1D"));
+
+ Assertions.assertEquals(4, entries.size());
+ Assertions.assertEquals("v1",
entries.get(Intervals.of("2025-05-10T00:00:00/P1M")), "Match 1");
+ Assertions.assertEquals("v7",
entries.get(Intervals.of("2025-06-03T00:00:00/P1D")), "Match 2");
+ Assertions.assertEquals("v13",
entries.get(Intervals.of("2025-06-01T00:00:00/P1M")), "Match 3");
+ Assertions.assertEquals("v14",
entries.get(Intervals.of("2025-01-01T00:00:00/P1Y")), "Match 4");
+ }
+
+ @Test
+ public void testRemove()
+ {
+ IntervalTreeMap<String> tree = setupTree(sparseOverlapData);
+ int size = tree.size();
+
+ // Remove node that does not exist
+ String intervalstr = "2025-03-11T00:00:00/P1M";
+ String oldValue = tree.remove(Intervals.of(intervalstr));
+ Assertions.assertEquals(size, tree.size(), "Size");
+ Assertions.assertNull(oldValue, "Old value");
+ List<Pair<Interval, String>> expectedData = new
ArrayList<>(sparseOverlapData);
+ compareData(expectedData, tree);
+
+ // Remove leaf
+ intervalstr = "2025-06-01T00:00:00/P1M";
+ String value = tree.get(Intervals.of(intervalstr));
+ Assertions.assertNotNull(value, "Value");
+
+ oldValue = tree.remove(Intervals.of(intervalstr));
+ size--;
+ Assertions.assertEquals(size, tree.size(), "Size");
+ Assertions.assertEquals(value, oldValue, "Old value");
+ expectedData = new ArrayList<>(sparseOverlapData);
+ expectedData.remove(Pair.of(Intervals.of(intervalstr), value));
+ compareData(expectedData, tree);
+
+ // Remove node in penultimate level
+ intervalstr = "2025-09-04T00:00:00/P1D";
+ value = tree.get(Intervals.of(intervalstr));
+ Assertions.assertNotNull(value, "Value");
+
+ oldValue = tree.remove(Intervals.of(intervalstr));
+ size--;
+ Assertions.assertEquals(size, tree.size(), "Size");
+ Assertions.assertEquals(value, oldValue, "Old value");
+ expectedData = new ArrayList<>(expectedData);
+ expectedData.remove(Pair.of(Intervals.of(intervalstr), value));
+ compareData(expectedData, tree);
+
+ // Remove node at a higher level
+ intervalstr = "2025-07-12T00:00:00/P1D";
+ value = tree.get(Intervals.of(intervalstr));
+ Assertions.assertNotNull(value, "Value");
+
+ oldValue = tree.remove(Intervals.of(intervalstr));
+ size--;
+ Assertions.assertEquals(size, tree.size(), "Size");
+ Assertions.assertEquals(value, oldValue, "Old value");
+ expectedData = new ArrayList<>(expectedData);
+ expectedData.remove(Pair.of(Intervals.of(intervalstr), value));
+ compareData(expectedData, tree);
+ }
+
+ @Test
+ public void testRemoveRootAndMatch()
+ {
+ IntervalTreeMap<String> tree = setupTree(baseData);
+ tree.remove(Intervals.of("2025-01-03T00:00:00/P1D"));
+ Assertions.assertEquals(5, tree.size(), "Size");
+ Map<Interval, String> entries =
tree.findEncompassing(Intervals.of("2025-01-04T00:00:00/P1D"));
+ Assertions.assertEquals(1, entries.size());
+ Assertions.assertEquals("v5",
entries.get(Intervals.of("2025-01-04T00:00:00/P1D")), "Match");
+ }
+
+ @Test
+ public void testRemoveMultiple()
+ {
+ IntervalTreeMap<String> tree = setupTree(sparseOverlapData);
+ int isize = tree.size();
+ tree.remove(Intervals.of("2025-01-12T00:00:00/P1D"));
+ tree.remove(Intervals.of("2025-06-03T00:00:00/P1D"));
+ tree.remove(Intervals.of("2025-06-01T00:00:00/P1M"));
+ int csize = tree.size();
+ Assertions.assertEquals(3, isize - csize, "Size");
+ }
+
+ @Test
+ public void testClear()
+ {
+ IntervalTreeMap<String> tree = setupTree(baseData);
+ tree.clear();
+ Assertions.assertEquals(0, tree.size(), "Size");
+ }
+
+ @Test
+ public void testLargeLoadTree()
+ {
+ IntervalTreeMap<String> tree = new
IntervalTreeMap<>(Comparators.intervalsByStart(), Comparators.intervalsByEnd());
+ List<Pair<Interval, String>> expectedData = new ArrayList<>();
+ Set<String> existingIntervals = new HashSet<>();
+ Random random = ThreadLocalRandom.current();
+ int total = 100000;
+ int count = 0;
+ while (count < total) {
+ int year = random.nextInt(26) + 2000;
+ int month = random.nextInt(12) + 1;
+ int day = random.nextInt(28) + 1;
+ int hour = random.nextInt(23) + 1;
+ String intervalstr = year + "-" + month + "-" + day + "T" + hour +
":00:00/P" + ((count % 30) + 1) + "D";
+ if (!existingIntervals.contains(intervalstr)) {
+ Interval interval = Intervals.of(intervalstr);
+ String value = "v" + count;
+ tree.put(interval, value);
+ expectedData.add(Pair.of(interval, value));
+ existingIntervals.add(intervalstr);
+ ++count;
+ }
+ }
+ Assertions.assertEquals(total, tree.size(), "Size");
+ compareData(expectedData, tree);
+ }
+
+ private static final Logger log = new Logger(IntervalTreeMapTest.class);
+
+ @Disabled
+ @Test
+ public void testPerf()
+ {
+ IntervalTreeMap<String> tree = new
IntervalTreeMap<>(Comparators.intervalsByStart(), Comparators.intervalsByEnd());
+ List<Pair<Interval, String>> expectedData = new ArrayList<>();
+ Map<Interval, String> mappedData = new HashMap<>();
+ Set<String> existingIntervals = new HashSet<>();
+ Random random = ThreadLocalRandom.current();
+ int total = 10000;
+ int count = 0;
+ while (count < total) {
+ int year = random.nextInt(26) + 2000;
+ int month = random.nextInt(12) + 1;
+ int day = random.nextInt(28) + 1;
+ int hour = random.nextInt(23) + 1;
+ String intervalstr = year + "-" + month + "-" + day + "T" + hour +
":00:00/P" + ((count % 30) + 1) + "D";
+ if (!existingIntervals.contains(intervalstr)) {
+ Interval interval = Intervals.of(intervalstr);
+ String value = "v" + count;
+ tree.put(interval, value);
+ mappedData.put(interval, value);
+ expectedData.add(Pair.of(interval, value));
+ existingIntervals.add(intervalstr);
+ ++count;
+ }
+ }
+ long start = System.currentTimeMillis();
+ for (int i = 0; i < total; i++) {
+ Pair<Interval, String> pair = expectedData.get(i);
+ Interval interval = pair.lhs;
+ for (Map.Entry<Interval, String> entry : mappedData.entrySet()) {
+ if (entry.getKey().contains(interval)) {
+ break;
+ }
+ }
+ }
+ log.info("Seq find time %d", (System.currentTimeMillis() - start));
+ start = System.currentTimeMillis();
+ for (int i = 0; i < total; i++) {
+ Pair<Interval, String> pair = expectedData.get(i);
+ Interval interval = pair.lhs;
+ tree.findEncompassing(interval);
+ }
+ log.info("Tree find time %d", (System.currentTimeMillis() - start));
+ }
+
+ @Test
+ public void testAutoRebalance()
+ {
+ IntervalTreeMap<String> tree = setupTree(sparseOverlapData);
+ Assertions.assertEquals(4, tree.height(), "Height");
+ compareData(sparseOverlapData, tree);
+ }
+
+ @Test
+ public void testManualRebalance()
+ {
+ // Set a high threshold so auto-rebalance does not happen
+ IntervalTreeMap<String> tree = setupTree(sparseOverlapData, t ->
t.setImbalanceTolerance(100));
+ Assertions.assertEquals(4, tree.height(), "Height");
+ compareData(sparseOverlapData, tree);
+ tree.rebalance();
+ Assertions.assertEquals(3, tree.height(), "Height");
+ compareData(sparseOverlapData, tree);
+ }
+
+ @Test
+ public void testIsEmpty()
+ {
+ IntervalTreeMap<String> tree = setupTree(sparseOverlapData);
+ Assertions.assertFalse(tree.isEmpty(), "Not Empty");
+ sparseOverlapData.forEach(t -> tree.remove(t.lhs));
+ Assertions.assertTrue(tree.isEmpty(), "Empty");
+ }
+
+ @Test
+ public void testFirstEntryAndKey()
+ {
+ IntervalTreeMap<String> tree = setupTree(sparseOverlapData);
+ Map.Entry<Interval, String> entry = tree.firstEntry();
+ Interval matchInterval = Intervals.of("2025-01-01T00:00:00/P1D");
+ Assertions.assertEquals(matchInterval, entry.getKey(), "Entry interval");
+ Assertions.assertEquals("v2", entry.getValue(), "Entry value");
+
+ Interval interval = tree.firstKey();
+ Assertions.assertEquals(matchInterval, interval, "Interval key");
+ }
+
+ @Test
+ public void testLastEntryAndKey()
+ {
+ IntervalTreeMap<String> tree = setupTree(sparseOverlapData);
+ Map.Entry<Interval, String> entry = tree.lastEntry();
+ Interval matchInterval = Intervals.of("2025-10-06T00:00:00/P1M");
+ Assertions.assertEquals(matchInterval, entry.getKey(), "Entry interval");
+ Assertions.assertEquals("v12", entry.getValue(), "Entry value");
+
+ Interval interval = tree.lastKey();
+ Assertions.assertEquals(matchInterval, interval, "Interval key");
+ }
+
+ @Test
+ public void testFloorKey()
+ {
+ IntervalTreeMap<String> tree = setupTree(sparseOverlapData);
+
+ // Only one smaller entry
+ Interval floor = tree.floorKey(Intervals.of("2025-01-11T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-01-01T00:00:00/P1Y"), floor,
"Floor key 1");
+
+ // Entry with same start date but different end date
+ floor = tree.lowerKey(Intervals.of("2025-01-01T00:00:00/P1M"));
+ Assertions.assertEquals(Intervals.of("2025-01-01T00:00:00/P1D"), floor,
"Lower key 2");
+
+ // Matching entry
+ floor = tree.floorKey(Intervals.of("2025-01-12T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-01-12T00:00:00/P1D"), floor,
"Floor key 3");
+
+ // Random entry
+ floor = tree.floorKey(Intervals.of("2025-08-01T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-07-12T00:00:00/P1D"), floor,
"Floor key 4");
+
+ // Last entry
+ floor = tree.floorKey(Intervals.of("2025-11-01T00:00:00/P1M"));
+ Assertions.assertEquals(Intervals.of("2025-10-06T00:00:00/P1M"), floor,
"Floor key 5");
+
+ // No smaller entry
+ floor = tree.floorKey(Intervals.of("2024-12-31T00:00:00/P1D"));
+ Assertions.assertNull(floor, "Floor key 6");
+ }
+
+ @Test
+ public void testLowerKey()
+ {
+ IntervalTreeMap<String> tree = setupTree(sparseOverlapData);
+
+ // Only one smaller entry
+ Interval lower = tree.lowerKey(Intervals.of("2025-01-11T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-01-01T00:00:00/P1Y"), lower,
"Lower key 1");
+
+ // Entry with same start date but different end date
+ lower = tree.lowerKey(Intervals.of("2025-01-01T00:00:00/P1M"));
+ Assertions.assertEquals(Intervals.of("2025-01-01T00:00:00/P1D"), lower,
"Lower key 2");
+
+ // Matching entry
+ lower = tree.lowerKey(Intervals.of("2025-01-12T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-01-01T00:00:00/P1Y"), lower,
"Lower key 3");
+
+ // Random entry
+ lower = tree.lowerKey(Intervals.of("2025-08-01T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-07-12T00:00:00/P1D"), lower,
"Lower key 4");
+
+ // Last entry
+ lower = tree.lowerKey(Intervals.of("2025-11-01T00:00:00/P1M"));
+ Assertions.assertEquals(Intervals.of("2025-10-06T00:00:00/P1M"), lower,
"Lower key 5");
+
+ // No smaller entry
+ lower = tree.lowerKey(Intervals.of("2024-12-31T00:00:00/P1D"));
+ Assertions.assertNull(lower, "Lower key 6");
+ }
+
+ @Test
+ public void testCeiinglKey()
+ {
+ IntervalTreeMap<String> tree = setupTree(sparseOverlapData);
+
+ // First entry
+ Interval ceiling =
tree.ceilingKey(Intervals.of("2024-12-31T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-01-01T00:00:00/P1D"), ceiling,
"Ceiling key 1");
+
+ // Entry with same start date but different end date
+ ceiling = tree.ceilingKey(Intervals.of("2025-02-01T00:00:00/PT6H"));
+ Assertions.assertEquals(Intervals.of("2025-02-01T00:00:00/P1D"), ceiling,
"Ceiling key 2");
+
+ // Matching entry
+ ceiling = tree.ceilingKey(Intervals.of("2025-09-04T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-09-04T00:00:00/P1D"), ceiling,
"Ceiling key 3");
+
+ // Random entry
+ ceiling = tree.ceilingKey(Intervals.of("2025-03-31T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-04-02T00:00:00/P1D"), ceiling,
"Ceiling key 4");
+
+ // Only one greater entry
+ ceiling = tree.ceilingKey(Intervals.of("2025-09-28T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-10-06T00:00:00/P1M"), ceiling,
"Ceiling key 5");
+
+ // No greater entry
+ ceiling = tree.ceilingKey(Intervals.of("2025-11-01T00:00:00/P1D"));
+ Assertions.assertNull(ceiling, "Ceiling key 6");
+ }
+
+ @Test
+ public void testHigherKey()
+ {
+ IntervalTreeMap<String> tree = setupTree(sparseOverlapData);
+
+ // First entry
+ Interval higher = tree.higherKey(Intervals.of("2024-12-31T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-01-01T00:00:00/P1D"), higher,
"Higher key 1");
+
+ // Entry with same start date but different end date
+ higher = tree.ceilingKey(Intervals.of("2025-02-01T00:00:00/PT6H"));
+ Assertions.assertEquals(Intervals.of("2025-02-01T00:00:00/P1D"), higher,
"Higher key 2");
+
+ // Matching entry
+ higher = tree.higherKey(Intervals.of("2025-09-04T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-10-06T00:00:00/P1M"), higher,
"Higher key 3");
+
+ // Random entry
+ higher = tree.higherKey(Intervals.of("2025-03-31T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-04-02T00:00:00/P1D"), higher,
"Higher key 4");
+
+ // Only one greater entry
+ higher = tree.higherKey(Intervals.of("2025-09-28T00:00:00/P1D"));
+ Assertions.assertEquals(Intervals.of("2025-10-06T00:00:00/P1M"), higher,
"Higher key 5");
+
+ // No greater entry
+ higher = tree.higherKey(Intervals.of("2025-11-01T00:00:00/P1D"));
+ Assertions.assertNull(higher, "Higher key 6");
+ }
+
+ private void compareData(List<Pair<Interval, String>> inputData,
IntervalTreeMap<String> tree)
+ {
+ //Iterator<Map.Entry<Interval, String>> iterator = tree.inOrderTraverse();
+ Iterator<Map.Entry<Interval, String>> iterator =
tree.entrySet().iterator();
+
+ List<Pair<Interval, String>> expected = inputData.stream()
+ .sorted((p1, p2) ->
Comparators.intervalsByStartThenEnd().compare(p1.lhs, p2.lhs))
+ .collect(Collectors.toList());
+
+ compareEntries(expected.iterator(), iterator);
+ }
+
+ private void compareEntries(Iterator<Pair<Interval, String>> expected,
Iterator<Map.Entry<Interval, String>> actual)
+ {
+ while (actual.hasNext()) {
+ Assertions.assertTrue(expected.hasNext(), "Entry available");
+ Pair<Interval, String> expectedEntry = expected.next();
+ Map.Entry<Interval, String> actualEntry = actual.next();
+ Assertions.assertEquals(expectedEntry.lhs, actualEntry.getKey(),
"Interval match");
+ Assertions.assertEquals(expectedEntry.rhs, actualEntry.getValue(),
"Value match");
+ }
+ Assertions.assertFalse(expected.hasNext(), "No outstanding entries");
+ }
+
+ static List<Pair<Interval, String>> baseData = new ArrayList<>();
+ static List<Pair<Interval, String>> overlapData = new ArrayList<>();
+ static List<Pair<Interval, String>> sparseOverlapData = new ArrayList<>();
+
+ static {
+
+ baseData.add(Pair.of(Intervals.of("2025-01-03T00:00:00/P1D"), "v1"));
+ baseData.add(Pair.of(Intervals.of("2025-01-05T00:00:00/P1D"), "v2"));
+ baseData.add(Pair.of(Intervals.of("2025-01-01T00:00:00/P1D"), "v3"));
+ baseData.add(Pair.of(Intervals.of("2025-01-02T00:00:00/P1D"), "v4"));
+ baseData.add(Pair.of(Intervals.of("2025-01-04T00:00:00/P1D"), "v5"));
+ baseData.add(Pair.of(Intervals.of("2025-01-06T00:00:00/P1D"), "v6"));
+
+ overlapData.addAll(baseData);
+ overlapData.add(Pair.of(Intervals.of("2025-01-01T00:00:00/P1Y"), "v7"));
+
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-05-10T00:00:00/P1M"),
"v1"));
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-01-01T00:00:00/P1D"),
"v2"));
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-02-01T00:00:00/P1D"),
"v3"));
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-01-12T00:00:00/P1D"),
"v4"));
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-07-12T00:00:00/P1D"),
"v5"));
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-02-01T00:00:00/P1M"),
"v6"));
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-06-03T00:00:00/P1D"),
"v7"));
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-08-09T00:00:00/P1D"),
"v8"));
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-08-02T00:00:00/P1M"),
"v9"));
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-09-04T00:00:00/P1D"),
"v10"));
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-04-02T00:00:00/P1D"),
"v11"));
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-10-06T00:00:00/P1M"),
"v12"));
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-06-01T00:00:00/P1M"),
"v13"));
+ sparseOverlapData.add(Pair.of(Intervals.of("2025-01-01T00:00:00/P1Y"),
"v14"));
+
+ }
+
+ private IntervalTreeMap<String> setupTree(List<Pair<Interval, String>>
inputData)
+ {
+ return setupTree(inputData, null);
+ }
+
+ private IntervalTreeMap<String> setupTree(List<Pair<Interval, String>>
inputData, Consumer<IntervalTreeMap<String>> setupFunc)
+ {
+ IntervalTreeMap<String> tree = new
IntervalTreeMap<>(Comparators.intervalsByStart(), Comparators.intervalsByEnd());
+ if (setupFunc != null) {
+ setupFunc.accept(tree);
+ }
+ for (Pair<Interval, String> entry : inputData) {
+ tree.put(entry.lhs, entry.rhs);
+ }
+ return tree;
+ }
+
+}
diff --git
a/processing/src/test/java/org/apache/druid/timeline/VersionedIntervalTimelineSpecificDataTest.java
b/processing/src/test/java/org/apache/druid/timeline/VersionedIntervalTimelineSpecificDataTest.java
index a41eda2fc36..46d57f06bbd 100644
---
a/processing/src/test/java/org/apache/druid/timeline/VersionedIntervalTimelineSpecificDataTest.java
+++
b/processing/src/test/java/org/apache/druid/timeline/VersionedIntervalTimelineSpecificDataTest.java
@@ -32,20 +32,40 @@ import org.joda.time.Interval;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
/**
* This test class is separated from {@link VersionedIntervalTimelineTest}
because it populates specific data for tests
* in {@link #setUp()}.
*/
+@RunWith(Parameterized.class)
public class VersionedIntervalTimelineSpecificDataTest extends
VersionedIntervalTimelineTestBase
{
+ @Parameterized.Parameters
+ public static Collection<Boolean> parameters()
+ {
+ return Arrays.asList(
+ false,
+ true
+ );
+ }
+
+ public VersionedIntervalTimelineSpecificDataTest(boolean fastIntervalSearch)
+ {
+ this.fastIntervalSearch = fastIntervalSearch;
+ }
+
+ private final boolean fastIntervalSearch;
+
@Before
public void setUp()
{
- timeline = makeStringIntegerTimeline();
+ timeline = makeStringIntegerTimeline(fastIntervalSearch);
add("2011-04-01/2011-04-03", "1", 2);
add("2011-04-03/2011-04-06", "1", 3);
diff --git
a/processing/src/test/java/org/apache/druid/timeline/VersionedIntervalTimelineTest.java
b/processing/src/test/java/org/apache/druid/timeline/VersionedIntervalTimelineTest.java
index bd2e61e20cb..ca332773e82 100644
---
a/processing/src/test/java/org/apache/druid/timeline/VersionedIntervalTimelineTest.java
+++
b/processing/src/test/java/org/apache/druid/timeline/VersionedIntervalTimelineTest.java
@@ -34,6 +34,8 @@ import org.joda.time.Interval;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import java.util.Arrays;
import java.util.Collection;
@@ -43,13 +45,29 @@ import java.util.Set;
/**
*/
+@RunWith(Parameterized.class)
public class VersionedIntervalTimelineTest extends
VersionedIntervalTimelineTestBase
{
+ @Parameterized.Parameters
+ public static Collection<Boolean> parameters()
+ {
+ return Arrays.asList(
+ false,
+ true
+ );
+ }
+
+ public VersionedIntervalTimelineTest(boolean fastIntervalSearch)
+ {
+ this.fastIntervalSearch = fastIntervalSearch;
+ }
+
+ private final boolean fastIntervalSearch;
@Before
public void setUp()
{
- timeline = makeStringIntegerTimeline();
+ timeline = makeStringIntegerTimeline(fastIntervalSearch);
}
@Test
diff --git
a/processing/src/test/java/org/apache/druid/timeline/VersionedIntervalTimelineTestBase.java
b/processing/src/test/java/org/apache/druid/timeline/VersionedIntervalTimelineTestBase.java
index 55f5c8ddfa9..fa6986cf28a 100644
---
a/processing/src/test/java/org/apache/druid/timeline/VersionedIntervalTimelineTestBase.java
+++
b/processing/src/test/java/org/apache/druid/timeline/VersionedIntervalTimelineTestBase.java
@@ -110,7 +110,12 @@ public class VersionedIntervalTimelineTestBase
static VersionedIntervalTimeline<String, OvershadowableInteger>
makeStringIntegerTimeline()
{
- return new VersionedIntervalTimeline<>(Ordering.natural());
+ return makeStringIntegerTimeline(false);
+ }
+
+ static VersionedIntervalTimeline<String, OvershadowableInteger>
makeStringIntegerTimeline(boolean fastIntervalSearch)
+ {
+ return new VersionedIntervalTimeline<>(Ordering.natural(), false,
fastIntervalSearch);
}
VersionedIntervalTimeline<String, OvershadowableInteger> timeline;
diff --git a/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java
b/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java
index 8feeb2cc25b..8a36cb81f6e 100644
--- a/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java
+++ b/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java
@@ -34,6 +34,7 @@ import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.query.DruidProcessingConfig;
import org.apache.druid.segment.DefaultColumnFormatConfig;
import org.apache.druid.segment.column.ColumnConfig;
+import org.apache.druid.segment.indexing.SegmentTimelineConfig;
import org.apache.druid.segment.loading.SegmentCacheManager;
import org.apache.druid.segment.loading.SegmentLoaderConfig;
import org.apache.druid.segment.loading.SegmentLocalCacheManager;
@@ -65,6 +66,7 @@ public class StorageNodeModule implements Module
JsonConfigProvider.bind(binder, "druid.server", DruidServerConfig.class);
JsonConfigProvider.bind(binder, "druid.segmentCache",
SegmentLoaderConfig.class);
JsonConfigProvider.bind(binder, "druid.indexing.formats",
DefaultColumnFormatConfig.class);
+ JsonConfigProvider.bind(binder, "druid.segment.timeline",
SegmentTimelineConfig.class);
bindLocationSelectorStrategy(binder);
binder.bind(ServerTypeConfig.class).toProvider(Providers.of(null));
binder.bind(ColumnConfig.class).to(DruidProcessingConfig.class).in(LazySingleton.class);
diff --git
a/server/src/main/java/org/apache/druid/segment/indexing/SegmentTimelineConfig.java
b/server/src/main/java/org/apache/druid/segment/indexing/SegmentTimelineConfig.java
new file mode 100644
index 00000000000..451af0df52f
--- /dev/null
+++
b/server/src/main/java/org/apache/druid/segment/indexing/SegmentTimelineConfig.java
@@ -0,0 +1,49 @@
+/*
+ * 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.segment.indexing;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/**
+ * Configuration settings related to segment timeline management
+ */
+public class SegmentTimelineConfig
+{
+ @JsonProperty
+ private final boolean fastIntervalSearch;
+
+ @JsonCreator
+ public SegmentTimelineConfig(@JsonProperty("fastIntervalSearch") @Nullable
Boolean fastIntervalSearch)
+ {
+ this.fastIntervalSearch = fastIntervalSearch != null && fastIntervalSearch;
+ }
+
+ /**
+ * Whether an indexing mechanism based on an interval tree for organizing
segments in memory is being used, that
+ * leads to faster search
+ */
+ public boolean isFastIntervalSearch()
+ {
+ return fastIntervalSearch;
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/server/SegmentManager.java
b/server/src/main/java/org/apache/druid/server/SegmentManager.java
index 4bea68953ac..73ab935a084 100644
--- a/server/src/main/java/org/apache/druid/server/SegmentManager.java
+++ b/server/src/main/java/org/apache/druid/server/SegmentManager.java
@@ -36,6 +36,7 @@ import org.apache.druid.segment.Segment;
import org.apache.druid.segment.SegmentLazyLoadFailCallback;
import org.apache.druid.segment.SegmentMapFunction;
import org.apache.druid.segment.SegmentReference;
+import org.apache.druid.segment.indexing.SegmentTimelineConfig;
import org.apache.druid.segment.join.table.IndexedTable;
import
org.apache.druid.segment.join.table.ReferenceCountedIndexedTableProvider;
import org.apache.druid.segment.loading.AcquireMode;
@@ -71,12 +72,21 @@ public class SegmentManager
private final SegmentCacheManager cacheManager;
+ private final SegmentTimelineConfig segmentTimelineConfig;
+
private final ConcurrentHashMap<String, DataSourceState> dataSources = new
ConcurrentHashMap<>();
- @Inject
public SegmentManager(SegmentCacheManager cacheManager)
+ {
+ this(cacheManager, new SegmentTimelineConfig(false));
+ }
+
+
+ @Inject
+ public SegmentManager(SegmentCacheManager cacheManager,
SegmentTimelineConfig segmentTimelineConfig)
{
this.cacheManager = cacheManager;
+ this.segmentTimelineConfig = segmentTimelineConfig;
}
@VisibleForTesting
@@ -339,7 +349,7 @@ public class SegmentManager
dataSources.compute(
dataSegment.getDataSource(),
(k, v) -> {
- final DataSourceState dataSourceState = v == null ? new
DataSourceState() : v;
+ final DataSourceState dataSourceState = v == null ? new
DataSourceState(segmentTimelineConfig) : v;
final VersionedIntervalTimeline<String, DataSegment> loadedIntervals
=
dataSourceState.getTimeline();
final PartitionChunk<DataSegment> entry = loadedIntervals.findChunk(
@@ -516,8 +526,7 @@ public class SegmentManager
*/
public static class DataSourceState
{
- private final VersionedIntervalTimeline<String, DataSegment> timeline =
- new VersionedIntervalTimeline<>(Ordering.natural());
+ private final VersionedIntervalTimeline<String, DataSegment> timeline;
private final ConcurrentHashMap<SegmentId,
ReferenceCountedIndexedTableProvider> tablesLookup = new ConcurrentHashMap<>();
private long totalSegmentSize;
@@ -525,6 +534,11 @@ public class SegmentManager
private long rowCount;
private final SegmentRowCountDistribution segmentRowCountDistribution =
new SegmentRowCountDistribution();
+ public DataSourceState(SegmentTimelineConfig segmentTimelineConfig)
+ {
+ timeline = new VersionedIntervalTimeline<>(Ordering.natural(), false,
segmentTimelineConfig.isFastIntervalSearch());
+ }
+
private void addSegment(DataSegment segment, long numOfRows)
{
totalSegmentSize += segment.getSize();
diff --git a/sql/src/test/java/org/apache/druid/sql/guice/SqlModuleTest.java
b/sql/src/test/java/org/apache/druid/sql/guice/SqlModuleTest.java
index 8c86d913d67..e04032a6637 100644
--- a/sql/src/test/java/org/apache/druid/sql/guice/SqlModuleTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/guice/SqlModuleTest.java
@@ -53,6 +53,7 @@ import org.apache.druid.query.QueryToolChestWarehouse;
import org.apache.druid.query.lookup.LookupExtractorFactoryContainerProvider;
import org.apache.druid.rpc.indexing.NoopOverlordClient;
import org.apache.druid.rpc.indexing.OverlordClient;
+import org.apache.druid.segment.indexing.SegmentTimelineConfig;
import org.apache.druid.segment.join.JoinableFactory;
import org.apache.druid.segment.loading.SegmentCacheManager;
import org.apache.druid.segment.metadata.CentralizedDatasourceSchemaConfig;
@@ -118,6 +119,9 @@ public class SqlModuleTest
@Mock
private SegmentCacheManager segmentCacheManager;
+ @Mock
+ private SegmentTimelineConfig segmentTimelineConfig;
+
@Mock
private QueryRunnerFactoryConglomerate conglomerate;
@@ -140,6 +144,7 @@ public class SqlModuleTest
lookupExtractorFactoryContainerProvider,
joinableFactory,
segmentCacheManager,
+ segmentTimelineConfig,
httpClient
);
}
@@ -209,6 +214,7 @@ public class SqlModuleTest
binder.bind(LookupExtractorFactoryContainerProvider.class).toInstance(lookupExtractorFactoryContainerProvider);
binder.bind(JoinableFactory.class).toInstance(joinableFactory);
binder.bind(SegmentCacheManager.class).toInstance(segmentCacheManager);
+
binder.bind(SegmentTimelineConfig.class).toInstance(segmentTimelineConfig);
binder.bind(QuerySchedulerProvider.class).in(LazySingleton.class);
binder.bind(QueryScheduler.class)
.toProvider(QuerySchedulerProvider.class)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
