You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ms...@apache.org on 2026/03/17 19:32:40 UTC
(lucene) branch main updated: Add Temporal Merge Policy for time-series data (#15620)
This is an automated email from the ASF dual-hosted git repository.
msfroh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/lucene.git
The following commit(s) were added to refs/heads/main by this push:
new 1596b7604b6 Add Temporal Merge Policy for time-series data (#15620)
1596b7604b6 is described below
commit 1596b7604b67ed47314aa648f14b91dc75ce8134
Author: Churro <88...@users.noreply.github.com>
AuthorDate: Tue Mar 17 12:32:33 2026 -0700
Add Temporal Merge Policy for time-series data (#15620)
Introduces TemporalMergePolicy, a new merge policy designed for time-series workloads where documents
contain a timestamp field. The policy groups segments into time windows and merges segments within the same
window, but never merges segments across different time windows. This preserves temporal locality and
improves query performance for time-range queries.
---------
Co-authored-by: Shubham Sharma <sh...@gmail.com>
Co-authored-by: Michael Froh <ms...@apache.org>
---
lucene/CHANGES.txt | 31 +-
.../apache/lucene/index/TemporalMergePolicy.java | 989 +++++++++++++++++++++
.../lucene/index/TestTemporalMergePolicy.java | 349 ++++++++
.../TestTemporalMergePolicyAcrossBuckets.java | 308 +++++++
4 files changed, 1662 insertions(+), 15 deletions(-)
diff --git a/lucene/CHANGES.txt b/lucene/CHANGES.txt
index b599dafaf4a..a2bb435fe27 100644
--- a/lucene/CHANGES.txt
+++ b/lucene/CHANGES.txt
@@ -76,7 +76,6 @@ API Changes
New Features
---------------------
-
* GITHUB#15505: Upgrade snowball to 2d2e312df56f2ede014a4ffb3e91e6dea43c24be. New stemmer: PolishStemmer (and
PolishSnowballAnalyzer in the stempel package) (Justas Sakalauskas, Dawid Weiss)
@@ -354,24 +353,26 @@ New Features
`Lucene104HnswScalarQuantizedVectorsFormat` replaces the now legacy `Lucene99HnswScalarQuantizedVectorsFormat`
(Trevor McCulloch)
- * GITHUB#15271, GITHUB#15353: Extend `Lucene104ScalarQuantizedVectorsFormat` and `Lucene104HnswScalarQuantizedVectorsFormat` to
- allow asymmetric quantization. The initially supported bits are single bit with 4 bit queries. This is a replacement
- for the now legacy `Lucene102HnswBinaryQuantizedVectorsFormat` and `Lucene102BinaryQuantizedVectorsFormat`. To use
- the new format for asymmetric binary quantization,
- `Lucene104HnswScalarQuantizedVectorsFormat(ScalarEncoding.SINGLE_BIT_QUERY_NIBBLE, int, int)`
- (Ben Trent)
-
- * GITHUB#15564: Further extend `Lucene104ScalarQuantizedVectorsFormat` and `Lucene104HnswScalarQuantizedVectorsFormat` to
- allow asymmetric quantization with 2 bits and 4 bit queries. This can provide increased recall over single bit
- with little additional storage cost and latency.
+* GITHUB#15271, GITHUB#15353: Extend `Lucene104ScalarQuantizedVectorsFormat` and `Lucene104HnswScalarQuantizedVectorsFormat` to
+ allow asymmetric quantization. The initially supported bits are single bit with 4 bit queries. This is a replacement
+ for the now legacy `Lucene102HnswBinaryQuantizedVectorsFormat` and `Lucene102BinaryQuantizedVectorsFormat`. To use
+ the new format for asymmetric binary quantization,
+ `Lucene104HnswScalarQuantizedVectorsFormat(ScalarEncoding.SINGLE_BIT_QUERY_NIBBLE, int, int)`
(Ben Trent)
- * GITHUB#15415: Add fallback support to Lucene104ScalarQuantizedVectorsFormat getFloatVectorValues when there are
- no full-precision vectors present (Pulkit Gupta)
+* GITHUB#15564: Further extend `Lucene104ScalarQuantizedVectorsFormat` and `Lucene104HnswScalarQuantizedVectorsFormat` to
+ allow asymmetric quantization with 2 bits and 4 bit queries. This can provide increased recall over single bit
+ with little additional storage cost and latency.
+ (Ben Trent)
+
+* GITHUB#15415: Add fallback support to Lucene104ScalarQuantizedVectorsFormat getFloatVectorValues when there are
+ no full-precision vectors present (Pulkit Gupta)
+
+* GITHUB#15468: Add support for `@SuppressAssertingFormats` annotation for fine-grained control over `AssertingCodec` formats (Prudhvi Godithi)
- * GITHUB#15468: Add support for `@SuppressAssertingFormats` annotation for fine-grained control over `AssertingCodec` formats (Prudhvi Godithi)
+* GITHUB#15518: Add support for post-collection faceting to the new faceting API in the sandbox module. (Egor Potemkin)
- * GITHUB#15518: Add support for post-collection faceting to the new faceting API in the sandbox module. (Egor Potemkin)
+* GITHUB#15620: Add Temporal Merge Policy for time-series data
* GITHUB#14792: Introduced OffHeapQuantizedFloatVectorValues class to access float vectors when only quantized byte vectors are available in the index. (Pulkit Gupta)
diff --git a/lucene/core/src/java/org/apache/lucene/index/TemporalMergePolicy.java b/lucene/core/src/java/org/apache/lucene/index/TemporalMergePolicy.java
new file mode 100644
index 00000000000..4facca397b3
--- /dev/null
+++ b/lucene/core/src/java/org/apache/lucene/index/TemporalMergePolicy.java
@@ -0,0 +1,989 @@
+/*
+ * 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.lucene.index;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Logger;
+import org.apache.lucene.codecs.PointsFormat;
+import org.apache.lucene.codecs.PointsReader;
+import org.apache.lucene.document.LongPoint;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.IOContext;
+
+/**
+ * A merge policy that groups segments by time windows and merges segments within the same window,
+ * This policy is designed for time-series data where documents contain a timestamp field indexed as
+ * a {@link org.apache.lucene.document.LongPoint}.
+ *
+ * <p>This policy organizes segments into time buckets based on the maximum timestamp in each
+ * segment. Recent data goes into small time windows (e.g., 1 hour), while older data is grouped
+ * into exponentially larger windows (e.g., 4 hours, 16 hours, etc.). Segments within the same time
+ * window are merged together when they meet the configured thresholds, but segments from different
+ * time windows are never merged together, preserving temporal locality.
+ *
+ * <p><b>When to use this policy:</b>
+ *
+ * <ul>
+ * <li>Time-series data where queries typically filter by time ranges
+ * <li>Data with a timestamp field that can be used for bucketing
+ * <li>Workloads where older data is queried less frequently than recent data
+ * <li>Use cases where you want to avoid mixing old and new data in the same segment
+ * </ul>
+ *
+ * <p><b>Configuration:</b>
+ *
+ * <pre class="prettyprint">
+ * TemporalMergePolicy policy = new TemporalMergePolicy()
+ * .setTemporalField("timestamp") // Required: name of the timestamp field
+ * .setBaseTimeSeconds(3600) // Base window size: 1 hour
+ * .setMinThreshold(4) // Merge when 4+ segments in a window
+ * .setMaxThreshold(8) // Merge at most 8 segments at once
+ * .setCompactionRatio(1.2); // Size ratio threshold for merging
+ * // By default, exponential buckets are enabled. Use .disableExponentialBuckets() to disable.
+ *
+ * IndexWriterConfig config = new IndexWriterConfig(analyzer);
+ * config.setMergePolicy(policy);
+ * </pre>
+ *
+ * <p><b>Time bucketing:</b> By default, window sizes grow exponentially: {@code baseTime}, {@code
+ * baseTime * minThreshold}, {@code baseTime * minThreshold^2}, etc. This ensures that recent data
+ * is in small, frequently-merged windows while older data is in larger, less-frequently-merged
+ * windows. Call {@link #disableExponentialBuckets()} to use fixed-size windows instead, where all
+ * windows have the same size ({@code baseTime}).
+ *
+ * <p><b>Compaction ratio:</b> The {@link #setCompactionRatio} parameter controls when merges are
+ * triggered. A merge is considered when the total document count across candidate segments exceeds
+ * {@code largestSegment * compactionRatio}. Lower values (e.g., 1.2) trigger merges more
+ * aggressively, while higher values (e.g., 2.0) allow more segments to accumulate before merging.
+ * Set to 1.0 for most aggressive merging.
+ *
+ * <p><b>NOTE:</b> This policy requires a timestamp field indexed as a {@link
+ * org.apache.lucene.document.LongPoint}. The timestamp can be in seconds, milliseconds, or
+ * microseconds (auto-detected based on value magnitude).
+ *
+ * <p><b>NOTE:</b> Segments from different time windows are never merged together, even during
+ * {@link IndexWriter#forceMerge(int)}. If you call {@code forceMerge(1)} but have segments in
+ * multiple time windows, you will end up with one segment per time window.
+ *
+ * <p><b>NOTE:</b> Very old segments (older than {@link #setMaxAgeSeconds}) are not merged to avoid
+ * unnecessary I/O on cold data.
+ *
+ * @lucene.experimental
+ */
+public class TemporalMergePolicy extends MergePolicy {
+
+ private static final Logger log = Logger.getLogger(TemporalMergePolicy.class.getName());
+
+ // Configuration parameters
+ private String temporalField = "";
+ private long baseTimeSeconds = 3600; // 1 hour default
+ private int minThreshold = 4; // Minimum segments to trigger merge within a window
+ private boolean useExponentialBuckets = true; // Use exponential window sizing
+ private long maxWindowSizeSeconds = TimeUnit.DAYS.toSeconds(365); // Maximum window size (1 year)
+ private long maxAgeSeconds =
+ Long.MAX_VALUE; // Don't compact segments older than this (default: no limit)
+ private int maxThreshold = 8; // Maximum segments to merge from a window at once
+ private double compactionRatio =
+ 1.2d; // Ratio between the total docs and biggest segment to trigger merge
+ private double forceMergeDeletesPctAllowed = 10.0d;
+ private Map<SegmentCommitInfo, SegmentDateRange> segmentDateRangeOverrides;
+
+ /** Sole constructor, setting all settings to their defaults. */
+ public TemporalMergePolicy() {}
+
+ /**
+ * Sets the name of the timestamp field used for temporal bucketing. This field must be indexed as
+ * a {@link LongPoint} and contain timestamp values in seconds, milliseconds, or microseconds
+ * (auto-detected based on value magnitude).
+ *
+ * <p><b>This parameter is required</b> and must be set before the policy can schedule any merges.
+ * The merge policy will extract the minimum and maximum timestamps from each segment to determine
+ * which time window the segment belongs to.
+ *
+ * <p>Default is empty (no temporal field configured, policy is inactive).
+ */
+ public TemporalMergePolicy setTemporalField(String temporalField) {
+ if (temporalField == null || temporalField.isBlank()) {
+ throw new IllegalArgumentException("temporalField cannot be blank");
+ }
+ this.temporalField = temporalField;
+ return this;
+ }
+
+ /**
+ * Returns the current temporal field name.
+ *
+ * @see #setTemporalField
+ */
+ public String getTemporalField() {
+ return temporalField;
+ }
+
+ /**
+ * Sets the base time window size in seconds. This determines the size of the smallest (most
+ * recent) time buckets.
+ *
+ * <p>By default, window sizes grow exponentially: {@code baseTime}, {@code baseTime *
+ * minThreshold}, {@code baseTime * minThreshold^2}, etc. If you call {@link
+ * #disableExponentialBuckets()}, all windows will have the same size equal to {@code baseTime}.
+ *
+ * <p>Smaller values create finer-grained time windows, which can improve query performance for
+ * time-range queries but may result in more segments. Larger values reduce the number of time
+ * windows but may mix data from a wider time range in the same segment.
+ *
+ * <p>Default is 3600 seconds (1 hour).
+ */
+ public TemporalMergePolicy setBaseTimeInSeconds(long baseTimeInSeconds) {
+ if (baseTimeInSeconds <= 0) {
+ throw new IllegalArgumentException("baseTimeSeconds must be positive");
+ }
+ this.baseTimeSeconds = baseTimeInSeconds;
+ return this;
+ }
+
+ /**
+ * Returns the current base time window size in seconds.
+ *
+ * @see #setBaseTimeInSeconds
+ */
+ public long getBaseTimeInSeconds() {
+ return baseTimeSeconds;
+ }
+
+ /**
+ * Sets the minimum number of segments required in a time window to trigger a merge. Higher values
+ * reduce merge frequency and I/O but allow more segments to accumulate. Lower values keep segment
+ * counts lower but increase write amplification.
+ *
+ * <p>This threshold is also used as the growth factor for exponential bucketing (which is enabled
+ * by default). For example, with {@code minThreshold=4}, window sizes will be: {@code baseTime},
+ * {@code baseTime * 4}, {@code baseTime * 16}, etc.
+ *
+ * <p>Must be at least 2 and cannot exceed {@link #setMaxThreshold}. Default is 4.
+ */
+ public TemporalMergePolicy setMinThreshold(int minThreshold) {
+ if (minThreshold < 2) {
+ throw new IllegalArgumentException("minThreshold must be at least 2");
+ }
+ if (minThreshold > maxThreshold) {
+ throw new IllegalArgumentException(
+ "minThreshold cannot exceed maxThreshold (" + maxThreshold + ")");
+ }
+ this.minThreshold = minThreshold;
+ return this;
+ }
+
+ /**
+ * Returns the current minimum threshold for merging.
+ *
+ * @see #setMinThreshold
+ */
+ public int getMinThreshold() {
+ return minThreshold;
+ }
+
+ /**
+ * Disables exponentially growing time windows. By default, older data is grouped into
+ * progressively larger time buckets: {@code baseTime}, {@code baseTime * minThreshold}, {@code
+ * baseTime * minThreshold^2}, etc.
+ *
+ * <p>Calling this method changes the behavior so that all time windows have a fixed size equal to
+ * {@code baseTime}, which can be useful for workloads with uniform query patterns across all time
+ * ranges.
+ *
+ * <p>Exponential bucketing (the default) is recommended for typical time-series use cases where
+ * recent data is accessed more frequently than older data.
+ */
+ public TemporalMergePolicy disableExponentialBuckets() {
+ this.useExponentialBuckets = false;
+ return this;
+ }
+
+ /**
+ * Returns whether exponential bucketing is enabled.
+ *
+ * @see #disableExponentialBuckets()
+ */
+ public boolean getUseExponentialBuckets() {
+ return useExponentialBuckets;
+ }
+
+ /**
+ * Sets the maximum number of segments to merge at once within a time window. Larger values allow
+ * more aggressive merging (reducing segment count faster) but increase the cost of individual
+ * merge operations.
+ *
+ * <p>Must be at least equal to {@link #setMinThreshold}. When a time window accumulates more
+ * segments than this threshold, the policy will schedule multiple smaller merges rather than one
+ * large merge.
+ *
+ * <p>Default is 8.
+ */
+ public TemporalMergePolicy setMaxThreshold(int maxThreshold) {
+ if (maxThreshold < minThreshold) {
+ throw new IllegalArgumentException(
+ "maxThreshold must be >= minThreshold (" + minThreshold + ")");
+ }
+ this.maxThreshold = maxThreshold;
+ return this;
+ }
+
+ /**
+ * Returns the current maximum threshold for merging.
+ *
+ * @see #setMaxThreshold
+ */
+ public int getMaxThreshold() {
+ return maxThreshold;
+ }
+
+ /**
+ * Sets the compaction ratio that controls when merges are triggered based on segment size
+ * distribution. A merge is considered when the total document count of candidate segments exceeds
+ * {@code largestSegment * compactionRatio}.
+ *
+ * <p>Lower values (e.g., 1.2) trigger merges more aggressively, even when segment sizes are
+ * relatively balanced. Higher values (e.g., 2.0 or higher) wait for more size imbalance before
+ * merging, allowing more segments to accumulate but reducing write amplification.
+ *
+ * <p>Setting this to exactly 1.0 enables the most aggressive merging mode, where merges occur
+ * whenever the minimum threshold is met, regardless of segment size distribution.
+ *
+ * <p>This parameter works together with {@link #setMinThreshold}: a time window must have both
+ * (1) at least {@code minThreshold} segments, and (2) satisfy the compaction ratio, before a
+ * merge is triggered.
+ *
+ * <p>Default is 1.2.
+ */
+ public TemporalMergePolicy setCompactionRatio(double compactionRatio) {
+ if (compactionRatio < 1.0d) {
+ throw new IllegalArgumentException("compactionRatio must be >= 1.0");
+ }
+ this.compactionRatio = compactionRatio;
+ return this;
+ }
+
+ /**
+ * Returns the current compaction ratio.
+ *
+ * @see #setCompactionRatio
+ */
+ public double getCompactionRatio() {
+ return compactionRatio;
+ }
+
+ /**
+ * Sets the maximum size for exponentially growing time windows. When exponential bucketing is
+ * enabled (the default), window sizes grow exponentially but are capped at this value.
+ *
+ * <p>This prevents extremely large time windows for very old data, which could mix data from
+ * vastly different time periods. Once window size reaches this limit, all older data uses
+ * fixed-size windows of this duration.
+ *
+ * <p>Default is 31536000 seconds (365 days).
+ */
+ public TemporalMergePolicy setMaxWindowSizeSeconds(long maxWindowSizeSeconds) {
+ if (maxWindowSizeSeconds <= 0) {
+ throw new IllegalArgumentException("maxWindowSizeSeconds must be positive");
+ }
+ this.maxWindowSizeSeconds = maxWindowSizeSeconds;
+ return this;
+ }
+
+ /**
+ * Returns the current maximum window size in seconds.
+ *
+ * @see #setMaxWindowSizeSeconds
+ */
+ public long getMaxWindowSizeSeconds() {
+ return maxWindowSizeSeconds;
+ }
+
+ /**
+ * Sets the maximum age threshold for merging segments. Segments containing data older than this
+ * threshold (based on current time minus the segment's maximum timestamp) will not be merged.
+ *
+ * <p>This is useful for preventing unnecessary I/O on cold, historical data that is rarely
+ * queried. These old segments are placed in a special "old data" bucket and skipped during merge
+ * selection.
+ *
+ * <p>Default is {@link Long#MAX_VALUE} (no age limit, all segments are merge candidates).
+ */
+ public TemporalMergePolicy setMaxAgeSeconds(long maxAgeSeconds) {
+ if (maxAgeSeconds <= 0) {
+ throw new IllegalArgumentException("maxAgeSeconds must be positive");
+ }
+ this.maxAgeSeconds = maxAgeSeconds;
+ return this;
+ }
+
+ /**
+ * Returns the current maximum age threshold in seconds.
+ *
+ * @see #setMaxAgeSeconds
+ */
+ public long getMaxAgeSeconds() {
+ return maxAgeSeconds;
+ }
+
+ /**
+ * When {@link IndexWriter#forceMergeDeletes()} is called, only merge segments whose delete
+ * percentage exceeds this threshold. Lower values merge more aggressively to reclaim space from
+ * deleted documents, but increase I/O and write amplification.
+ *
+ * <p>The delete percentage is calculated as: {@code (deleted docs / total docs) * 100}.
+ *
+ * <p>Default is 10.0 (merge segments with more than 10% deleted documents).
+ */
+ public TemporalMergePolicy setForceMergeDeletesPctAllowed(double pct) {
+ if (pct < 0d) {
+ throw new IllegalArgumentException("forceMergeDeletesPctAllowed must be >= 0");
+ }
+ this.forceMergeDeletesPctAllowed = pct;
+ return this;
+ }
+
+ /**
+ * Returns the current force merge deletes percentage threshold.
+ *
+ * @see #setForceMergeDeletesPctAllowed
+ */
+ public double getForceMergeDeletesPctAllowed() {
+ return forceMergeDeletesPctAllowed;
+ }
+
+ // Visible for testing.
+ void setSegmentDateRangeOverrides(Map<SegmentCommitInfo, SegmentDateRange> overrides) {
+ this.segmentDateRangeOverrides = overrides;
+ }
+
+ @Override
+ public MergeSpecification findMerges(
+ MergeTrigger trigger, SegmentInfos segments, MergeContext context) throws IOException {
+ if (temporalField == null || temporalField.isBlank() || segments.size() == 0) {
+ // No date field configured, nothing to do
+ return null;
+ }
+
+ // Identify which segments are already being processed by the MergeScheduler, to avoid already
+ // merging errors
+ Set<SegmentCommitInfo> alreadyMerging =
+ context != null ? context.getMergingSegments() : Collections.emptySet();
+
+ // Resolve date ranges for all segments (ideally this will be cached in the future)
+ Map<SegmentCommitInfo, SegmentDateRange> allRanges = resolveSegmentDateRanges(segments);
+
+ if (allRanges.isEmpty()) {
+ return null;
+ }
+
+ // Group only segments that are NOT already merging into time windows
+ Map<Long, List<SegmentCommitInfo>> windowBuckets = new TreeMap<>();
+ long now = System.currentTimeMillis();
+
+ for (SegmentCommitInfo info : segments) {
+ // Skip segments that are already being merged
+ if (alreadyMerging.contains(info)) {
+ continue;
+ }
+
+ SegmentDateRange range = allRanges.get(info);
+ if (range != null) {
+ assignToBucket(windowBuckets, now, info, range);
+ }
+ }
+
+ if (windowBuckets.isEmpty()) {
+ return null;
+ }
+
+ // Find merge candidates within time windows (only considering available segments)
+ return findMergeCandidates(windowBuckets, allRanges);
+ }
+
+ @Override
+ public MergeSpecification findForcedMerges(
+ SegmentInfos segmentInfos,
+ int maxSegmentCount,
+ Map<SegmentCommitInfo, Boolean> segmentsToMerge,
+ MergeContext mergeContext)
+ throws IOException {
+ if (maxSegmentCount < 1) {
+ throw new IllegalArgumentException("maxSegmentCount must be >= 1");
+ }
+
+ Map<SegmentCommitInfo, SegmentDateRange> segmentDateRanges =
+ resolveSegmentDateRanges(segmentInfos);
+ if (segmentDateRanges.isEmpty()) {
+ return null;
+ }
+
+ boolean mergeAll = segmentsToMerge == null || segmentsToMerge.isEmpty();
+ Set<SegmentCommitInfo> merging =
+ mergeContext != null ? mergeContext.getMergingSegments() : Collections.emptySet();
+
+ Map<SegmentCommitInfo, SegmentDateRange> eligibleRanges = new HashMap<>();
+ boolean forceMergeRunning = false;
+ for (SegmentCommitInfo info : segmentInfos) {
+ boolean include = mergeAll || segmentsToMerge.containsKey(info);
+ if (include) {
+ if (merging.contains(info)) {
+ forceMergeRunning = true;
+ break;
+ }
+
+ SegmentDateRange range = segmentDateRanges.get(info);
+ if (range != null) {
+ eligibleRanges.put(info, range);
+ }
+ }
+ }
+
+ if (forceMergeRunning) {
+ return null;
+ }
+
+ if (eligibleRanges.isEmpty()) {
+ return null;
+ }
+
+ Map<Long, List<SegmentCommitInfo>> windowBuckets = groupSegmentsByTimeWindow(eligibleRanges);
+ int windowCount = windowBuckets.size();
+
+ if (windowCount == 0) {
+ return null;
+ }
+
+ // Calculate target segments per window based on maxSegmentCount
+ // IMPORTANT: We NEVER merge across time buckets, even in forceMerge
+ // If maxSegmentCount < windowCount, we'll end up with more segments than requested
+ int segmentsPerWindow; // Default: merge each window to 1 segment
+ MergeSpecification spec = null;
+ int eligibleWindows;
+
+ // First pass: count all windows (all windows are eligible for forced merge)
+ eligibleWindows = windowBuckets.size();
+
+ // Recalculate segmentsPerWindow based on eligible windows
+ segmentsPerWindow = Math.max(1, maxSegmentCount / eligibleWindows);
+ // Cap at maxThreshold to ensure we still merge aggressively even with few windows
+ // Without this cap, a single window with 27 segments and maxSegmentCount=32
+ // would result in segmentsPerWindow=32, causing no merges to be scheduled
+ segmentsPerWindow = Math.min(segmentsPerWindow, maxThreshold);
+
+ // Second pass: schedule merges for all windows
+ for (Map.Entry<Long, List<SegmentCommitInfo>> entry : windowBuckets.entrySet()) {
+ long windowStart = entry.getKey();
+ List<SegmentCommitInfo> bucketSegments = entry.getValue();
+
+ // Skip old data bucket (-1 sentinel) - don't merge very old data
+ if (windowStart == -1) {
+ continue;
+ }
+
+ if (bucketSegments.size() < 2) {
+ continue;
+ }
+
+ MergeSpecification bucketSpec = buildForcedMerges(bucketSegments, segmentsPerWindow);
+ if (bucketSpec == null) {
+ continue;
+ }
+
+ if (spec == null) {
+ spec = new MergeSpecification();
+ }
+ for (OneMerge merge : bucketSpec.merges) {
+ spec.add(merge);
+ }
+ }
+
+ return spec;
+ }
+
+ @Override
+ public MergeSpecification findForcedDeletesMerges(
+ SegmentInfos segmentInfos, MergeContext mergeContext) throws IOException {
+ if (mergeContext == null) {
+ return null;
+ }
+
+ Map<SegmentCommitInfo, SegmentDateRange> segmentDateRanges =
+ resolveSegmentDateRanges(segmentInfos);
+ if (segmentDateRanges.isEmpty()) {
+ return null;
+ }
+
+ Set<SegmentCommitInfo> merging = mergeContext.getMergingSegments();
+ Map<SegmentCommitInfo, SegmentDateRange> candidateRanges = new HashMap<>();
+ Map<SegmentCommitInfo, Double> deleteRatios = new HashMap<>();
+
+ for (SegmentCommitInfo info : segmentInfos) {
+ if (merging.contains(info)) {
+ continue;
+ }
+
+ int delCount = mergeContext.numDeletesToMerge(info);
+ if (delCount <= 0) {
+ continue;
+ }
+ int maxDoc = Math.max(1, info.info.maxDoc());
+ double pctDeletes = (100.0d * delCount) / maxDoc;
+ if (pctDeletes > forceMergeDeletesPctAllowed) {
+ SegmentDateRange range = segmentDateRanges.get(info);
+ if (range != null) {
+ candidateRanges.put(info, range);
+ deleteRatios.put(info, pctDeletes);
+ }
+ }
+ }
+
+ if (candidateRanges.isEmpty()) {
+ return null;
+ }
+
+ Map<Long, List<SegmentCommitInfo>> windowBuckets = groupSegmentsByTimeWindow(candidateRanges);
+ MergeSpecification spec = null;
+
+ for (Map.Entry<Long, List<SegmentCommitInfo>> entry : windowBuckets.entrySet()) {
+ List<SegmentCommitInfo> bucketSegments = entry.getValue();
+ if (bucketSegments.size() < 2) {
+ continue;
+ }
+
+ bucketSegments.sort(
+ (a, b) ->
+ Double.compare(
+ deleteRatios.getOrDefault(b, 0.0d), deleteRatios.getOrDefault(a, 0.0d)));
+
+ MergeSpecification bucketSpec = buildSequentialMerges(bucketSegments);
+ if (bucketSpec == null) {
+ continue;
+ }
+
+ if (spec == null) {
+ spec = new MergeSpecification();
+ }
+
+ for (OneMerge merge : bucketSpec.merges) {
+ spec.add(merge);
+ }
+ }
+
+ return spec;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("[" + getClass().getSimpleName() + ": ");
+ sb.append("temporalField=").append(temporalField).append(", ");
+ sb.append("baseTimeSeconds=").append(baseTimeSeconds).append(", ");
+ sb.append("minThreshold=").append(minThreshold).append(", ");
+ sb.append("maxThreshold=").append(maxThreshold).append(", ");
+ sb.append("useExponentialBuckets=").append(useExponentialBuckets).append(", ");
+ sb.append("maxWindowSizeSeconds=").append(maxWindowSizeSeconds).append(", ");
+ sb.append("maxAgeSeconds=").append(maxAgeSeconds).append(", ");
+ sb.append("compactionRatio=").append(compactionRatio).append(", ");
+ sb.append("forceMergeDeletesPctAllowed=").append(forceMergeDeletesPctAllowed);
+ sb.append("]");
+ return sb.toString();
+ }
+
+ /**
+ * Extract min/max date ranges from all segments by reading point values directly from segment
+ * files.
+ *
+ * <p>This method uses the Codec API to read segment files directly, which allows it to access
+ * both committed and uncommitted (newly flushed) segments. This is done on purpose instead of
+ * using DirectoryReader.open(), which only sees committed segments and would miss segments that
+ * have been flushed but not yet committed during indexing.
+ *
+ * <p>The method handles both regular segment files and compound file format (.cfs) segments.
+ *
+ * @param segments the segment infos to extract date ranges from
+ * @return a map from segment info to its date range (min/max timestamps in milliseconds)
+ */
+ private Map<SegmentCommitInfo, SegmentDateRange> extractSegmentDateRanges(SegmentInfos segments) {
+ Map<SegmentCommitInfo, SegmentDateRange> ranges = new HashMap<>();
+
+ if (segments.size() == 0) {
+ return ranges;
+ }
+
+ for (SegmentCommitInfo segmentInfo : segments) {
+ try {
+ SegmentDateRange range = extractDateRangeFromSegment(segmentInfo);
+ if (range != null) {
+ ranges.put(segmentInfo, range);
+ }
+ } catch (IOException e) {
+ log.warning(
+ "Failed to read point values from segment "
+ + segmentInfo.info.name
+ + ": "
+ + e.getMessage());
+ }
+ }
+ return ranges;
+ }
+
+ /**
+ * Extract date range (min/max timestamps) from a single segment by reading point values.
+ *
+ * @param segmentInfo the segment to read from
+ * @return the date range, or null if the temporal field is not present or has no values
+ * @throws IOException if there's an error reading the segment
+ */
+ private SegmentDateRange extractDateRangeFromSegment(SegmentCommitInfo segmentInfo)
+ throws IOException {
+ SegmentInfo si = segmentInfo.info;
+
+ // Track compound directory separately to ensure we never close si.dir
+ Directory compoundDir = null;
+ try {
+ Directory readerDir =
+ si.getUseCompoundFile()
+ ? (compoundDir = si.getCodec().compoundFormat().getCompoundReader(si.dir, si))
+ : si.dir;
+
+ FieldInfos fieldInfos =
+ si.getCodec().fieldInfosFormat().read(readerDir, si, "", IOContext.READONCE);
+
+ // Validate that the temporal field exists and is a point field
+ FieldInfo fieldInfo = fieldInfos.fieldInfo(temporalField);
+ if (fieldInfo == null) {
+ return null;
+ }
+
+ if (fieldInfo.getPointDimensionCount() == 0) {
+ log.warning(
+ "Segment "
+ + si.name
+ + ": temporal field '"
+ + temporalField
+ + "' is not indexed as a point field (found: "
+ + fieldInfo
+ + "). "
+ + "Skipping this segment for date-tiered merging. This may occur with legacy segments "
+ + "or after schema changes.");
+ return null;
+ }
+
+ // Read point values using the codec
+ PointsFormat pointsFormat = si.getCodec().pointsFormat();
+ try (PointsReader pointsReader =
+ pointsFormat.fieldsReader(
+ new SegmentReadState(readerDir, si, fieldInfos, IOContext.READONCE))) {
+
+ PointValues pointValues = pointsReader.getValues(temporalField);
+ if (pointValues == null) {
+ return null;
+ }
+
+ byte[] minPackedValue = pointValues.getMinPackedValue();
+ byte[] maxPackedValue = pointValues.getMaxPackedValue();
+
+ if (minPackedValue == null || maxPackedValue == null) {
+ return null;
+ }
+
+ // Decode the packed values as longs, we have to make an assumption here
+ // since we don't have the schema :/
+ long minDate = LongPoint.decodeDimension(minPackedValue, 0);
+ long maxDate = LongPoint.decodeDimension(maxPackedValue, 0);
+
+ // Convert to milliseconds based on detected unit
+ long divisor = getTemporalFieldDivisor(maxDate);
+ long minDateMillis, maxDateMillis;
+ if (divisor < 0) {
+ long multiplier = -divisor;
+ minDateMillis = minDate * multiplier;
+ maxDateMillis = maxDate * multiplier;
+ } else {
+ minDateMillis = minDate / divisor;
+ maxDateMillis = maxDate / divisor;
+ }
+
+ return new SegmentDateRange(minDateMillis, maxDateMillis);
+ }
+ } finally {
+ // Close compound directory if we opened it (never close si.dir)
+ if (compoundDir != null) {
+ compoundDir.close();
+ }
+ }
+ }
+
+ // Auto-detect the divisor to convert the temporal field's raw value to milliseconds.
+ private long getTemporalFieldDivisor(long maxValue) {
+ if (maxValue > 100_000_000_000_000L) {
+ // Values > 10^14 are microseconds
+ return 1_000;
+ } else if (maxValue > 100_000_000_000L) {
+ // Values > 10^11 are milliseconds
+ return 1;
+ } else {
+ // Values <= 10^11 are seconds
+ return -1000;
+ }
+ }
+
+ // Group segments into time windows based on their maximum date.
+ private Map<Long, List<SegmentCommitInfo>> groupSegmentsByTimeWindow(
+ Map<SegmentCommitInfo, SegmentDateRange> segmentDateRanges) {
+
+ // Map from window identifier (bucket start timestamp) to segments
+ Map<Long, List<SegmentCommitInfo>> buckets = new TreeMap<>();
+
+ long now = System.currentTimeMillis();
+
+ for (Map.Entry<SegmentCommitInfo, SegmentDateRange> entry : segmentDateRanges.entrySet()) {
+ SegmentCommitInfo segment = entry.getKey();
+ SegmentDateRange dateRange = entry.getValue();
+ assignToBucket(buckets, now, segment, dateRange);
+ }
+ return buckets;
+ }
+
+ private void assignToBucket(
+ Map<Long, List<SegmentCommitInfo>> buckets,
+ long now,
+ SegmentCommitInfo segment,
+ SegmentDateRange dateRange) {
+
+ long maxDateSeconds = dateRange.maxDate / 1000;
+ long bucket = getBucketForTimestamp(maxDateSeconds, now / 1000);
+
+ buckets.computeIfAbsent(bucket, _ -> new ArrayList<>()).add(segment);
+ }
+
+ private long getBucketForTimestamp(long timestampSeconds, long nowSeconds) {
+ long ageSeconds = nowSeconds - timestampSeconds;
+ if (ageSeconds < 0) {
+ // Future timestamp, put in most recent bucket
+ ageSeconds = 0;
+ }
+
+ // If data is older than maxAgeSeconds, put it all in one monster "old data" bucket
+ // Use -1 as the sentinel value since bucket IDs are timestamps (always >= 0)
+ if (ageSeconds > maxAgeSeconds) {
+ return -1;
+ }
+
+ if (useExponentialBuckets == false) {
+ return (timestampSeconds / baseTimeSeconds) * baseTimeSeconds;
+ }
+
+ // Determine bucket (0 = baseTime, 1 = baseTime*minThreshold, 2 = baseTime*minThreshold^2,...
+ long bucketSizeSeconds = baseTimeSeconds;
+
+ while (ageSeconds >= bucketSizeSeconds * minThreshold
+ && bucketSizeSeconds < maxWindowSizeSeconds) {
+ bucketSizeSeconds *= minThreshold;
+ }
+
+ if (bucketSizeSeconds > maxWindowSizeSeconds) {
+ bucketSizeSeconds = maxWindowSizeSeconds;
+ }
+
+ return (timestampSeconds / bucketSizeSeconds) * bucketSizeSeconds;
+ }
+
+ /**
+ * Find merge candidates within time windows. Merges segments when a window has >= minThreshold
+ * segments. Skips windows that are older than maxAgeSeconds.
+ */
+ MergeSpecification findMergeCandidates(
+ Map<Long, List<SegmentCommitInfo>> buckets,
+ Map<SegmentCommitInfo, SegmentDateRange> segmentDateRanges) {
+ MergeSpecification spec = null;
+
+ for (Map.Entry<Long, List<SegmentCommitInfo>> entry : buckets.entrySet()) {
+ Long windowStart = entry.getKey();
+ List<SegmentCommitInfo> segmentsInWindow = entry.getValue();
+
+ // Skip old data bucket (-1 sentinel) - don't merge very old data
+ if (windowStart == -1) {
+ continue;
+ }
+
+ if (segmentsInWindow.size() < minThreshold) {
+ continue;
+ }
+
+ List<List<SegmentCommitInfo>> mergesForWindow =
+ planWindowMerges(windowStart, segmentsInWindow, segmentDateRanges);
+
+ if (mergesForWindow.isEmpty()) {
+ continue;
+ }
+
+ if (spec == null) {
+ spec = new MergeSpecification();
+ }
+
+ for (List<SegmentCommitInfo> mergeSegments : mergesForWindow) {
+ OneMerge merge = new OneMerge(mergeSegments);
+ spec.add(merge);
+ }
+ }
+
+ return spec;
+ }
+
+ /**
+ * Builds merges for a single window, enforcing the compaction ratio and thresholds. Segments are
+ * sorted by their newest timestamp to ensure we merge recent data first.
+ */
+ private List<List<SegmentCommitInfo>> planWindowMerges(
+ long windowStart,
+ List<SegmentCommitInfo> segmentsInWindow,
+ Map<SegmentCommitInfo, SegmentDateRange> segmentDateRanges) {
+
+ List<SegmentCommitInfo> ordered = new ArrayList<>(segmentsInWindow);
+ ordered.sort(Comparator.comparingLong(sci -> segmentDateRanges.get(sci).maxDate).reversed());
+
+ List<List<SegmentCommitInfo>> planned = new ArrayList<>();
+ int cursor = 0;
+
+ while (ordered.size() - cursor >= minThreshold) {
+ long totalDocs = 0;
+ long largestDocs = 0;
+ int end = cursor;
+ boolean emittedMerge = false;
+
+ while (end < ordered.size() && end - cursor < maxThreshold) {
+ SegmentCommitInfo candidate = ordered.get(end);
+ long docCount = candidate.info.maxDoc();
+ totalDocs += docCount;
+ largestDocs = Math.max(largestDocs, docCount);
+ end++;
+
+ int candidateSize = end - cursor;
+ if (candidateSize < minThreshold) {
+ continue;
+ }
+
+ boolean reachedMax = candidateSize == maxThreshold;
+ boolean exhaustedSegments = end == ordered.size();
+
+ // In aggressive mode (ratio <= 1.0), merge when:
+ // 1. We reach maxThreshold (preferred), OR
+ // 2. We've exhausted all segments and have at least minThreshold
+ if (compactionRatio <= 1.0d) {
+ if (reachedMax || exhaustedSegments) {
+ planned.add(new ArrayList<>(ordered.subList(cursor, end)));
+ cursor = end;
+ emittedMerge = true;
+ break;
+ }
+ } else {
+ // In normal mode, merge when ratio is satisfied OR we reach maxThreshold
+ boolean ratioSatisfied = totalDocs >= Math.ceil(largestDocs * compactionRatio);
+ if (ratioSatisfied || reachedMax) {
+ planned.add(new ArrayList<>(ordered.subList(cursor, end)));
+ cursor = end;
+ emittedMerge = true;
+ break;
+ }
+ }
+ }
+
+ if (emittedMerge == false) {
+ break;
+ }
+ }
+
+ return planned;
+ }
+
+ private MergeSpecification buildForcedMerges(
+ List<SegmentCommitInfo> candidates, int maxSegmentCount) {
+ if (candidates.size() < 2) {
+ return null;
+ }
+
+ MergeSpecification spec = null;
+ int remaining = candidates.size();
+ int offset = 0;
+
+ while (remaining > maxSegmentCount && (candidates.size() - offset) >= 2) {
+ int neededReduction = remaining - maxSegmentCount;
+ int mergeInputs = Math.min(maxThreshold, Math.max(2, neededReduction + 1));
+ mergeInputs = Math.min(mergeInputs, candidates.size() - offset);
+
+ List<SegmentCommitInfo> batch = candidates.subList(offset, offset + mergeInputs);
+
+ if (spec == null) {
+ spec = new MergeSpecification();
+ }
+ spec.add(new OneMerge(new ArrayList<>(batch)));
+ offset += mergeInputs;
+ remaining -= (mergeInputs - 1);
+ }
+
+ return spec;
+ }
+
+ private MergeSpecification buildSequentialMerges(List<SegmentCommitInfo> candidates) {
+ if (candidates.size() < 2) {
+ return null;
+ }
+
+ MergeSpecification spec = null;
+ List<SegmentCommitInfo> batch = new ArrayList<>(maxThreshold);
+
+ for (SegmentCommitInfo info : candidates) {
+ batch.add(info);
+ if (batch.size() == maxThreshold) {
+ if (spec == null) {
+ spec = new MergeSpecification();
+ }
+ spec.add(new OneMerge(new ArrayList<>(batch)));
+ batch.clear();
+ }
+ }
+
+ if (batch.size() > 1) {
+ if (spec == null) {
+ spec = new MergeSpecification();
+ }
+ spec.add(new OneMerge(new ArrayList<>(batch)));
+ }
+
+ return spec;
+ }
+
+ private Map<SegmentCommitInfo, SegmentDateRange> resolveSegmentDateRanges(SegmentInfos segments) {
+ if (segmentDateRangeOverrides != null) {
+ return segmentDateRangeOverrides;
+ }
+ return extractSegmentDateRanges(segments);
+ }
+
+ /** Helper class to store min/max date range for a segment. */
+ record SegmentDateRange(long minDate, long maxDate) {}
+}
diff --git a/lucene/core/src/test/org/apache/lucene/index/TestTemporalMergePolicy.java b/lucene/core/src/test/org/apache/lucene/index/TestTemporalMergePolicy.java
new file mode 100644
index 00000000000..3d001c7db61
--- /dev/null
+++ b/lucene/core/src/test/org/apache/lucene/index/TestTemporalMergePolicy.java
@@ -0,0 +1,349 @@
+/*
+ * 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.lucene.index;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import org.apache.lucene.codecs.Codec;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.LongPoint;
+import org.apache.lucene.document.NumericDocValuesField;
+import org.apache.lucene.store.ByteBuffersDirectory;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.tests.util.LuceneTestCase;
+import org.apache.lucene.util.InfoStream;
+import org.apache.lucene.util.StringHelper;
+import org.apache.lucene.util.Version;
+
+public class TestTemporalMergePolicy extends LuceneTestCase {
+
+ public void testSkipsMergeWhenCompactionRatioNotMet() {
+ TemporalMergePolicy policy = buildPolicy();
+ policy.setMinThreshold(2).setMaxThreshold(4).setCompactionRatio(2.0d);
+
+ SegmentCommitInfo large = newSegment("large", 1000);
+ SegmentCommitInfo smallOne = newSegment("small1", 100);
+ SegmentCommitInfo smallTwo = newSegment("small2", 100);
+
+ Map<SegmentCommitInfo, TemporalMergePolicy.SegmentDateRange> ranges = new HashMap<>();
+ ranges.put(large, new TemporalMergePolicy.SegmentDateRange(0, 3000));
+ ranges.put(smallOne, new TemporalMergePolicy.SegmentDateRange(0, 2000));
+ ranges.put(smallTwo, new TemporalMergePolicy.SegmentDateRange(0, 1000));
+
+ Map<Long, List<SegmentCommitInfo>> buckets = new TreeMap<>();
+ buckets.put(0L, Arrays.asList(large, smallOne, smallTwo));
+
+ MergePolicy.MergeSpecification spec = policy.findMergeCandidates(buckets, ranges);
+ assertNull("Merge should wait for more small segments to accumulate", spec);
+ }
+
+ public void testMergesSubsetWhenRatioSatisfied() {
+ TemporalMergePolicy policy = buildPolicy();
+ policy.setMinThreshold(2).setMaxThreshold(3).setCompactionRatio(1.6d);
+
+ SegmentCommitInfo segA = newSegment("segA", 400);
+ SegmentCommitInfo segB = newSegment("segB", 200);
+ SegmentCommitInfo segC = newSegment("segC", 200);
+ SegmentCommitInfo segD = newSegment("segD", 200);
+ SegmentCommitInfo segE = newSegment("segE", 200);
+
+ List<SegmentCommitInfo> segments = Arrays.asList(segA, segB, segC, segD, segE);
+ Map<SegmentCommitInfo, TemporalMergePolicy.SegmentDateRange> ranges = new HashMap<>();
+ long timestamp = 5000;
+ for (SegmentCommitInfo segment : segments) {
+ ranges.put(segment, new TemporalMergePolicy.SegmentDateRange(timestamp - 100, timestamp));
+ timestamp -= 100;
+ }
+
+ Map<Long, List<SegmentCommitInfo>> buckets = new TreeMap<>();
+ buckets.put(0L, segments);
+
+ MergePolicy.MergeSpecification spec = policy.findMergeCandidates(buckets, ranges);
+ assertNotNull(spec);
+ assertEquals(2, spec.merges.size());
+
+ assertEquals(3, spec.merges.get(0).segments.size());
+ assertEquals(2, spec.merges.get(1).segments.size());
+ }
+
+ public void testForcedMergeReducesSegments() throws Exception {
+ TemporalMergePolicy policy = buildPolicy().disableExponentialBuckets().setBaseTimeInSeconds(10);
+ SegmentInfos infos = new SegmentInfos(Version.LATEST.major);
+ SegmentCommitInfo segA = newSegment("forceA", 400);
+ SegmentCommitInfo segB = newSegment("forceB", 300);
+ SegmentCommitInfo segC = newSegment("forceC", 200);
+
+ infos.add(segA);
+ infos.add(segB);
+ infos.add(segC);
+
+ Map<SegmentCommitInfo, Boolean> segmentsToMerge = new HashMap<>();
+ segmentsToMerge.put(segA, Boolean.TRUE);
+ segmentsToMerge.put(segB, Boolean.TRUE);
+ segmentsToMerge.put(segC, Boolean.TRUE);
+
+ Map<SegmentCommitInfo, TemporalMergePolicy.SegmentDateRange> ranges = new HashMap<>();
+ ranges.put(segA, range(10000));
+ ranges.put(segB, range(11000));
+ ranges.put(segC, range(12000));
+ policy.setSegmentDateRangeOverrides(ranges);
+
+ MergePolicy.MergeSpecification spec =
+ policy.findForcedMerges(infos, 1, segmentsToMerge, new StubMergeContext());
+ assertNotNull(spec);
+ assertFalse(spec.merges.isEmpty());
+ assertEquals(1, spec.merges.size());
+ assertEquals(3, spec.merges.getFirst().segments.size());
+ }
+
+ public void testForcedDeletesMergesRespectThreshold() throws Exception {
+ TemporalMergePolicy policy =
+ buildPolicy()
+ .setForceMergeDeletesPctAllowed(5.0d)
+ .disableExponentialBuckets()
+ .setBaseTimeInSeconds(10);
+
+ SegmentInfos infos = new SegmentInfos(Version.LATEST.major);
+ SegmentCommitInfo segA = newSegment("delA", 100);
+ SegmentCommitInfo segB = newSegment("delB", 100);
+ SegmentCommitInfo segC = newSegment("delC", 100);
+ infos.add(segA);
+ infos.add(segB);
+ infos.add(segC);
+
+ Map<SegmentCommitInfo, Integer> deletes = new HashMap<>();
+ deletes.put(segA, 10);
+ deletes.put(segB, 1);
+ deletes.put(segC, 20);
+
+ Map<SegmentCommitInfo, TemporalMergePolicy.SegmentDateRange> ranges = new HashMap<>();
+ ranges.put(segA, range(10000));
+ ranges.put(segB, range(11000));
+ ranges.put(segC, range(12000));
+ policy.setSegmentDateRangeOverrides(ranges);
+
+ MergePolicy.MergeSpecification spec =
+ policy.findForcedDeletesMerges(infos, new StubMergeContext(deletes));
+
+ assertNotNull(spec);
+ assertEquals(1, spec.merges.size());
+ assertEquals(2, spec.merges.getFirst().segments.size());
+ }
+
+ public void testForcedMergeKeepsWindowsDisjoint() throws Exception {
+ TemporalMergePolicy policy = buildPolicy().disableExponentialBuckets().setBaseTimeInSeconds(10);
+
+ SegmentInfos infos = new SegmentInfos(Version.LATEST.major);
+ SegmentCommitInfo segA1 = newSegment("bucketA1", 100);
+ SegmentCommitInfo segA2 = newSegment("bucketA2", 100);
+ SegmentCommitInfo segB1 = newSegment("bucketB1", 100);
+ SegmentCommitInfo segB2 = newSegment("bucketB2", 100);
+
+ infos.add(segA1);
+ infos.add(segA2);
+ infos.add(segB1);
+ infos.add(segB2);
+
+ Map<SegmentCommitInfo, Boolean> segmentsToMerge = new HashMap<>();
+ segmentsToMerge.put(segA1, Boolean.TRUE);
+ segmentsToMerge.put(segA2, Boolean.TRUE);
+ segmentsToMerge.put(segB1, Boolean.TRUE);
+ segmentsToMerge.put(segB2, Boolean.TRUE);
+
+ Map<SegmentCommitInfo, TemporalMergePolicy.SegmentDateRange> ranges = new HashMap<>();
+ ranges.put(segA1, range(5000));
+ ranges.put(segA2, range(6000));
+ ranges.put(segB1, range(25000));
+ ranges.put(segB2, range(26000));
+ policy.setSegmentDateRangeOverrides(ranges);
+
+ MergePolicy.MergeSpecification spec =
+ policy.findForcedMerges(infos, 1, segmentsToMerge, new StubMergeContext());
+
+ assertNotNull(spec);
+ assertEquals("one merge per window", 2, spec.merges.size());
+
+ Set<Set<SegmentCommitInfo>> actual = new HashSet<>();
+ for (MergePolicy.OneMerge merge : spec.merges) {
+ actual.add(new HashSet<>(merge.segments));
+ }
+
+ Set<Set<SegmentCommitInfo>> expected = new HashSet<>();
+ expected.add(new HashSet<>(Arrays.asList(segA1, segA2)));
+ expected.add(new HashSet<>(Arrays.asList(segB1, segB2)));
+
+ assertEquals(expected, actual);
+ }
+
+ public void testForcedDeletesKeepWindowsDisjoint() throws Exception {
+ TemporalMergePolicy policy =
+ buildPolicy()
+ .setForceMergeDeletesPctAllowed(5.0d)
+ .disableExponentialBuckets()
+ .setBaseTimeInSeconds(10);
+
+ SegmentInfos infos = new SegmentInfos(Version.LATEST.major);
+ SegmentCommitInfo segA1 = newSegment("delA1", 100);
+ SegmentCommitInfo segA2 = newSegment("delA2", 100);
+ SegmentCommitInfo segB1 = newSegment("delB1", 100);
+
+ infos.add(segA1);
+ infos.add(segA2);
+ infos.add(segB1);
+
+ Map<SegmentCommitInfo, Integer> deletes = new HashMap<>();
+ deletes.put(segA1, 20);
+ deletes.put(segA2, 15);
+ deletes.put(segB1, 12);
+
+ Map<SegmentCommitInfo, TemporalMergePolicy.SegmentDateRange> ranges = new HashMap<>();
+ ranges.put(segA1, range(5000));
+ ranges.put(segA2, range(6000));
+ ranges.put(segB1, range(25000));
+ policy.setSegmentDateRangeOverrides(ranges);
+
+ MergePolicy.MergeSpecification spec =
+ policy.findForcedDeletesMerges(infos, new StubMergeContext(deletes));
+
+ assertNotNull(spec);
+ assertEquals(1, spec.merges.size());
+ List<SegmentCommitInfo> mergedSegments = spec.merges.getFirst().segments;
+ assertEquals(2, mergedSegments.size());
+ assertTrue(mergedSegments.contains(segA1));
+ assertTrue(mergedSegments.contains(segA2));
+ }
+
+ private TemporalMergePolicy buildPolicy() {
+ return new TemporalMergePolicy().setTemporalField("ts");
+ }
+
+ private SegmentCommitInfo newSegment(String name, int maxDoc) {
+ String segmentName =
+ name.startsWith("_") ? name.toLowerCase(Locale.ROOT) : "_" + name.toLowerCase(Locale.ROOT);
+ SegmentInfo info =
+ new SegmentInfo(
+ new ByteBuffersDirectory(),
+ Version.LATEST,
+ Version.LATEST,
+ segmentName,
+ maxDoc,
+ false,
+ true,
+ Codec.getDefault(),
+ Collections.emptyMap(),
+ StringHelper.randomId(),
+ Collections.emptyMap(),
+ null);
+ info.setFiles(Collections.singleton(IndexFileNames.segmentFileName(segmentName, "", "si")));
+ return new SegmentCommitInfo(info, 0, 0, -1, -1, -1, StringHelper.randomId());
+ }
+
+ public void testMergesWorkWithUncommittedSegments() throws IOException {
+ Directory dir = new ByteBuffersDirectory();
+
+ TemporalMergePolicy policy =
+ new TemporalMergePolicy()
+ .setTemporalField("timestamp_tl")
+ .setBaseTimeInSeconds(3600)
+ .setMinThreshold(3)
+ .setMaxThreshold(4)
+ .setCompactionRatio(1.0);
+
+ IndexWriterConfig config = new IndexWriterConfig();
+ config.setMergePolicy(policy);
+ config.setMaxBufferedDocs(10);
+
+ IndexWriter writer = new IndexWriter(dir, config);
+
+ long nowMillis = 1769540949799L;
+ long nowMicros = nowMillis * 1000;
+
+ for (int segment = 0; segment < 4; segment++) {
+ for (int doc = 0; doc < 10; doc++) {
+ Document d = new Document();
+ long timestamp = nowMicros + (doc * 1000000L);
+ d.add(new LongPoint("timestamp_tl", timestamp));
+ d.add(new NumericDocValuesField("timestamp_tl", timestamp));
+ writer.addDocument(d);
+ }
+ writer.flush(); // Flush to create segment but DON'T commit
+ }
+
+ // Force a merge check - this will use TemporalMergePolicy.findMerges()
+ // which calls extractSegmentDateRanges() to read uncommitted segments
+ writer.maybeMerge();
+
+ writer.forceMerge(1, true);
+
+ writer.commit();
+
+ DirectoryReader reader = DirectoryReader.open(dir);
+ int finalSegmentCount = reader.leaves().size();
+
+ reader.close();
+ writer.close();
+ dir.close();
+
+ assertTrue(
+ "Temporal merging should work on uncommitted segments (Codec approach)",
+ finalSegmentCount < 4);
+ }
+
+ private TemporalMergePolicy.SegmentDateRange range(long maxMillis) {
+ return new TemporalMergePolicy.SegmentDateRange(0, maxMillis);
+ }
+
+ private record StubMergeContext(
+ Map<SegmentCommitInfo, Integer> deleteCounts, Set<SegmentCommitInfo> mergingSegments)
+ implements MergePolicy.MergeContext {
+ StubMergeContext() {
+ this(Collections.emptyMap(), Collections.emptySet());
+ }
+
+ StubMergeContext(Map<SegmentCommitInfo, Integer> deleteCounts) {
+ this(deleteCounts, Collections.emptySet());
+ }
+
+ @Override
+ public int numDeletesToMerge(SegmentCommitInfo info) {
+ return deleteCounts.getOrDefault(info, 0);
+ }
+
+ @Override
+ public int numDeletedDocs(SegmentCommitInfo info) {
+ return numDeletesToMerge(info);
+ }
+
+ @Override
+ public InfoStream getInfoStream() {
+ return InfoStream.NO_OUTPUT;
+ }
+
+ @Override
+ public Set<SegmentCommitInfo> getMergingSegments() {
+ return mergingSegments;
+ }
+ }
+}
diff --git a/lucene/core/src/test/org/apache/lucene/index/TestTemporalMergePolicyAcrossBuckets.java b/lucene/core/src/test/org/apache/lucene/index/TestTemporalMergePolicyAcrossBuckets.java
new file mode 100644
index 00000000000..8f3a259ab5d
--- /dev/null
+++ b/lucene/core/src/test/org/apache/lucene/index/TestTemporalMergePolicyAcrossBuckets.java
@@ -0,0 +1,308 @@
+/*
+ * 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.lucene.index;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.LongPoint;
+import org.apache.lucene.document.NumericDocValuesField;
+import org.apache.lucene.store.ByteBuffersDirectory;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.tests.util.LuceneTestCase;
+
+/**
+ * Comprehensive test to ensure TemporalMergePolicy NEVER merges segments from different time
+ * buckets, even under aggressive force merge scenarios.
+ *
+ * <p>This test creates multiple segments across different year buckets and verifies that they
+ * remain in separate buckets after force merge operations.
+ */
+public class TestTemporalMergePolicyAcrossBuckets extends LuceneTestCase {
+
+ public void testMultipleYearBucketsNeverMerge() throws IOException {
+ Directory dir = new ByteBuffersDirectory();
+
+ TemporalMergePolicy policy =
+ new TemporalMergePolicy()
+ .setTemporalField("timestamp_tl")
+ .setBaseTimeInSeconds(TimeUnit.DAYS.toSeconds(365))
+ .setMaxAgeSeconds(TimeUnit.DAYS.toSeconds(365 * 10))
+ .setMinThreshold(2)
+ .setMaxThreshold(10)
+ .disableExponentialBuckets()
+ .setCompactionRatio(1.0);
+
+ IndexWriterConfig config = new IndexWriterConfig();
+ config.setMergePolicy(policy);
+ config.setMaxBufferedDocs(1000);
+
+ IndexWriter writer = new IndexWriter(dir, config);
+
+ // Create segments for 5 different years: 2020, 2021, 2022, 2023, 2024
+ // Each year gets 3 segments with unique timestamps
+ long year2020 = 1577836800000L; // 2020-01-01 00:00:00 UTC
+ long year2021 = 1609459200000L; // 2021-01-01 00:00:00 UTC
+ long year2022 = 1640995200000L; // 2022-01-01 00:00:00 UTC
+ long year2023 = 1672531200000L; // 2023-01-01 00:00:00 UTC
+ long year2024 = 1704067200000L; // 2024-01-01 00:00:00 UTC
+
+ long[] years = {year2020, year2021, year2022, year2023, year2024};
+
+ for (long yearTimestamp : years) {
+ for (int segInYear = 0; segInYear < 3; segInYear++) {
+ for (int doc = 0; doc < 10; doc++) {
+ Document d = new Document();
+ long timestamp = yearTimestamp + (segInYear * 1000000L) + (doc * 1000L);
+ d.add(new LongPoint("timestamp_tl", timestamp * 1000));
+ d.add(new NumericDocValuesField("timestamp_tl", timestamp * 1000));
+ writer.addDocument(d);
+ }
+ writer.flush();
+ }
+ }
+
+ writer.forceMerge(1, true);
+ writer.commit();
+
+ DirectoryReader reader = DirectoryReader.open(dir);
+ int finalSegmentCount = reader.leaves().size();
+
+ for (int i = 0; i < reader.leaves().size(); i++) {
+ LeafReaderContext ctx = reader.leaves().get(i);
+ PointValues pointValues;
+ pointValues = ctx.reader().getPointValues("timestamp_tl");
+
+ if (pointValues != null) {
+ byte[] minValue = pointValues.getMinPackedValue();
+ byte[] maxValue = pointValues.getMaxPackedValue();
+
+ long minTimestamp = LongPoint.decodeDimension(minValue, 0) / 1000;
+ long maxTimestamp = LongPoint.decodeDimension(maxValue, 0) / 1000;
+
+ long minYearBucket = (minTimestamp / 1000) / TimeUnit.DAYS.toSeconds(365);
+ long maxYearBucket = (maxTimestamp / 1000) / TimeUnit.DAYS.toSeconds(365);
+
+ // Verify segment doesn't span multiple year buckets
+ // Allow up to 1 year difference (since we're using 1-year buckets)
+ long bucketSpan = Math.abs(maxYearBucket - minYearBucket);
+ assertTrue(
+ "Segment "
+ + i
+ + " spans "
+ + bucketSpan
+ + " year buckets (should be ≤ 1). "
+ + "This indicates cross-bucket merging occurred!",
+ bucketSpan <= 1);
+ }
+ }
+
+ reader.close();
+ writer.close();
+ dir.close();
+
+ // With 5 year buckets and 3 segments per bucket, after merging:
+ // - Within each bucket: 3 segments should merge into 1 (minThreshold=2)
+ // - Across buckets: NEVER merge
+ // Expected: 5 segments (one per year bucket)
+
+ // Could be more if segments within a bucket didn't meet merge criteria
+ assertTrue(
+ "Expected at least 5 segments (one per year bucket), but got "
+ + finalSegmentCount
+ + ". This indicates segments from different year buckets were merged together!",
+ finalSegmentCount >= 5);
+ }
+
+ public void testTenYearBucketsRemainSeparate() throws IOException {
+ Directory dir = new ByteBuffersDirectory();
+
+ TemporalMergePolicy policy =
+ new TemporalMergePolicy()
+ .setTemporalField("timestamp_tl")
+ .setBaseTimeInSeconds(TimeUnit.DAYS.toSeconds(365)) // 1-year buckets
+ .setMaxAgeSeconds(TimeUnit.DAYS.toSeconds(365 * 10)) // 10 years
+ .setMinThreshold(2)
+ .setMaxThreshold(10)
+ .disableExponentialBuckets()
+ .setCompactionRatio(1.0);
+
+ IndexWriterConfig config = new IndexWriterConfig();
+ config.setMergePolicy(policy);
+ config.setMaxBufferedDocs(1000);
+
+ IndexWriter writer = new IndexWriter(dir, config);
+
+ long baseYear = 1577836800000L; // 2020-01-01 00:00:00 UTC
+ long oneYearMillis = TimeUnit.DAYS.toMillis(365);
+
+ for (int year = 0; year < 10; year++) {
+ long yearTimestamp = baseYear + (year * oneYearMillis);
+
+ for (int seg = 0; seg < 2; seg++) {
+ for (int doc = 0; doc < 10; doc++) {
+ Document d = new Document();
+ long timestamp = yearTimestamp + (seg * 1000000L) + (doc * 1000L);
+ d.add(new LongPoint("timestamp_tl", timestamp * 1000));
+ d.add(new NumericDocValuesField("timestamp_tl", timestamp * 1000));
+ writer.addDocument(d);
+ }
+ writer.flush();
+ }
+ }
+
+ writer.forceMerge(1, true);
+ writer.commit();
+
+ DirectoryReader reader = DirectoryReader.open(dir);
+ int finalSegmentCount = reader.leaves().size();
+
+ reader.close();
+ writer.close();
+ dir.close();
+
+ // With 10 year buckets and minThreshold=2, each bucket should merge its 2 segments into 1
+ // Expected: exactly 10 segments (one per year bucket)
+ assertTrue(
+ "Expected at least 10 segments (one per year bucket), but got " + finalSegmentCount,
+ finalSegmentCount >= 10);
+ }
+
+ public void testOldDataBucketDoesNotMergeAcrossYearBoundaries() throws IOException {
+ Directory dir = new ByteBuffersDirectory();
+
+ TemporalMergePolicy policy =
+ new TemporalMergePolicy()
+ .setTemporalField("timestamp_tl")
+ .setBaseTimeInSeconds(TimeUnit.DAYS.toSeconds(365))
+ .setMaxAgeSeconds(TimeUnit.DAYS.toSeconds(365 * 10))
+ .setMinThreshold(2)
+ .setMaxThreshold(10)
+ .disableExponentialBuckets()
+ .setCompactionRatio(1.0);
+
+ IndexWriterConfig config = new IndexWriterConfig();
+ config.setMergePolicy(policy);
+ config.setMaxBufferedDocs(100);
+
+ IndexWriter writer = new IndexWriter(dir, config);
+
+ // Segment A: Contains data from 1970 to 2024 (max_timestamp = 2024)
+ // Should be assigned to 2024 bucket
+ long timestamp2024 = 1735689600000L; // 2025-01-01 00:00:00 UTC in millis
+ long timestamp1970 = 0L;
+
+ for (int i = 0; i < 10; i++) {
+ Document d = new Document();
+ long timestamp = timestamp1970 + (i * 1000L); // Old data
+ d.add(new LongPoint("timestamp_tl", timestamp * 1000));
+ d.add(new NumericDocValuesField("timestamp_tl", timestamp * 1000));
+ writer.addDocument(d);
+ }
+ // Add one recent doc to set max_timestamp = 2024
+ Document d = new Document();
+ d.add(new LongPoint("timestamp_tl", timestamp2024 * 1000));
+ d.add(new NumericDocValuesField("timestamp_tl", timestamp2024 * 1000));
+ writer.addDocument(d);
+ writer.flush();
+
+ // Segment B: Contains data from 1970 to 2023 (max_timestamp = 2023)
+ // Should be assigned to 2023 bucket
+ long timestamp2023 = 1704067200000L; // 2024-01-01 00:00:00 UTC in millis
+
+ for (int i = 0; i < 10; i++) {
+ Document d2 = new Document();
+ long timestamp = timestamp1970 + (i * 2000L); // Old data
+ d2.add(new LongPoint("timestamp_tl", timestamp * 1000));
+ d2.add(new NumericDocValuesField("timestamp_tl", timestamp * 1000));
+ writer.addDocument(d2);
+ }
+ // Add one recent doc to set max_timestamp = 2023
+ Document d2 = new Document();
+ d2.add(new LongPoint("timestamp_tl", timestamp2023 * 1000));
+ d2.add(new NumericDocValuesField("timestamp_tl", timestamp2023 * 1000));
+ writer.addDocument(d2);
+ writer.flush();
+
+ // Trigger merging - this should NOT merge segments A and B together
+ // because they're in different year buckets
+ writer.forceMerge(1, true);
+ writer.commit();
+
+ // Check final segment count
+ DirectoryReader reader = DirectoryReader.open(dir);
+ int finalSegmentCount = reader.leaves().size();
+
+ reader.close();
+ writer.close();
+ dir.close();
+
+ // Assert that segments from different buckets don't merge
+ assertEquals(
+ "Segments from different year buckets should NOT be merged together", 2, finalSegmentCount);
+ }
+
+ public void testOldDataBucketWithMultipleOldSegments() throws IOException {
+ Directory dir = new ByteBuffersDirectory();
+
+ TemporalMergePolicy policy =
+ new TemporalMergePolicy()
+ .setTemporalField("timestamp_tl")
+ .setBaseTimeInSeconds(TimeUnit.DAYS.toSeconds(365))
+ .setMaxAgeSeconds(TimeUnit.DAYS.toSeconds(365 * 3))
+ .setMinThreshold(2)
+ .setMaxThreshold(10)
+ .disableExponentialBuckets()
+ .setCompactionRatio(1.0);
+
+ IndexWriterConfig config = new IndexWriterConfig();
+ config.setMergePolicy(policy);
+ config.setMaxBufferedDocs(100);
+
+ IndexWriter writer = new IndexWriter(dir, config);
+
+ long now = 1769540949799L;
+ long fiveYearsAgo = now - TimeUnit.DAYS.toMillis(365 * 5);
+
+ // Create 3 segments, all containing data older than 3 years
+ // These should all go into bucket -1 "old data bucket" which is NOT merged
+ for (int seg = 0; seg < 3; seg++) {
+ for (int i = 0; i < 10; i++) {
+ Document d = new Document();
+ long timestamp = fiveYearsAgo + (seg * 1000000L) + (i * 1000L);
+ d.add(new LongPoint("timestamp_tl", timestamp * 1000));
+ d.add(new NumericDocValuesField("timestamp_tl", timestamp * 1000));
+ writer.addDocument(d);
+ }
+ writer.flush();
+ }
+
+ writer.forceMerge(1, true);
+ writer.commit();
+
+ DirectoryReader reader = DirectoryReader.open(dir);
+ int finalSegmentCount = reader.leaves().size();
+
+ reader.close();
+ writer.close();
+ dir.close();
+
+ // Old data segments (older than maxAgeSeconds) should NOT merge - they get skipped
+ assertEquals(
+ "Old data segments (all >3 years) should NOT be merged (skipped)", 3, finalSegmentCount);
+ }
+}