This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new bb448114ab2 Add segmentAgeField option to time-based tier selector
(endTime | startTime | creationTime) (#19415)
bb448114ab2 is described below
commit bb448114ab2b93ddc2e27d53ec0ef1b4dee73d9d
Author: Chaitanya Deepthi <[email protected]>
AuthorDate: Tue Sep 1 23:06:49 2026 -0700
Add segmentAgeField option to time-based tier selector (endTime | startTime
| creationTime) (#19415)
---
.../org/apache/pinot/common/tier/TierFactory.java | 4 +-
.../common/tier/TimeBasedTierSegmentSelector.java | 81 +++++++++++++--
.../pinot/common/tier/TierSegmentSelectorTest.java | 109 +++++++++++++++++++++
.../apache/pinot/spi/config/table/TierConfig.java | 19 ++++
4 files changed, 205 insertions(+), 8 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/tier/TierFactory.java
b/pinot-common/src/main/java/org/apache/pinot/common/tier/TierFactory.java
index a0062f969b9..2ef0170cc89 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/tier/TierFactory.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/tier/TierFactory.java
@@ -51,7 +51,9 @@ public final class TierFactory {
LOGGER.debug("Provided segments: {} for tier: {}",
providedSegmentsForTier, tierConfig.getName());
segmentSelector = new FixedTierSegmentSelector(providedSegmentsForTier);
} else if
(segmentSelectorType.equalsIgnoreCase(TierFactory.TIME_SEGMENT_SELECTOR_TYPE)) {
- segmentSelector = new
TimeBasedTierSegmentSelector(tierConfig.getSegmentAge());
+ TimeBasedTierSegmentSelector.AgeField ageField =
+
TimeBasedTierSegmentSelector.AgeField.fromConfig(tierConfig.getSegmentAgeField());
+ segmentSelector = new
TimeBasedTierSegmentSelector(tierConfig.getSegmentAge(), ageField);
} else if
(segmentSelectorType.equalsIgnoreCase(TierFactory.FIXED_SEGMENT_SELECTOR_TYPE))
{
List<String> segments = tierConfig.getSegmentList();
segmentSelector =
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/tier/TimeBasedTierSegmentSelector.java
b/pinot-common/src/main/java/org/apache/pinot/common/tier/TimeBasedTierSegmentSelector.java
index f182e6a17ee..d9e46679e46 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/tier/TimeBasedTierSegmentSelector.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/tier/TimeBasedTierSegmentSelector.java
@@ -19,16 +19,58 @@
package org.apache.pinot.common.tier;
import com.google.common.base.Preconditions;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.spi.utils.TimeUtils;
-/// A [TierSegmentSelector] strategy which selects segments for a tier based
on the age of the segment
+/// A [TierSegmentSelector] strategy which selects segments for a tier based
on the age of the segment.
+///
+/// The age reference is controlled by the tier's `segmentAgeField`:
+/// - `endTime` (default, backward-compatible): uses
`SegmentZKMetadata#getEndTimeMs()`, the segment's
+/// max data timestamp. Suitable when segment age tracks data recency,
e.g. streaming ingest where
+/// endTime is close to wall-clock now.
+/// - `startTime`: uses `SegmentZKMetadata#getStartTimeMs()`, the segment's
min data timestamp.
+/// Suitable when segment age should track the oldest data in the segment.
+/// - `creationTime`: uses `SegmentZKMetadata#getCreationTime()`, when the
segment file was built.
+/// Suitable when segment age should track ingestion recency, e.g. batch
ingest of historical data
+/// where endTime lies far in the past regardless of when the segment was
created.
public class TimeBasedTierSegmentSelector implements TierSegmentSelector {
+
+ /// Which timestamp field on [SegmentZKMetadata] is compared against the age
threshold.
+ public enum AgeField {
+ END_TIME, START_TIME, CREATION_TIME;
+
+ public static AgeField fromConfig(@Nullable String value) {
+ if (StringUtils.isEmpty(value)) {
+ return END_TIME;
+ }
+ String normalized = value.trim();
+ if ("endtime".equalsIgnoreCase(normalized) ||
"end_time".equalsIgnoreCase(normalized)) {
+ return END_TIME;
+ }
+ if ("starttime".equalsIgnoreCase(normalized) ||
"start_time".equalsIgnoreCase(normalized)) {
+ return START_TIME;
+ }
+ if ("creationtime".equalsIgnoreCase(normalized) ||
"creation_time".equalsIgnoreCase(normalized)) {
+ return CREATION_TIME;
+ }
+ throw new IllegalArgumentException(
+ "Unsupported segmentAgeField: '" + value + "'. Expected 'endTime',
'startTime' or 'creationTime'.");
+ }
+ }
+
private final long _segmentAgeMillis;
+ private final AgeField _ageField;
public TimeBasedTierSegmentSelector(String segmentAge) {
+ this(segmentAge, AgeField.END_TIME);
+ }
+
+ public TimeBasedTierSegmentSelector(String segmentAge, AgeField ageField) {
_segmentAgeMillis = TimeUtils.convertPeriodToMillis(segmentAge);
+ _ageField = ageField != null ? ageField : AgeField.END_TIME;
}
@Override
@@ -43,11 +85,30 @@ public class TimeBasedTierSegmentSelector implements
TierSegmentSelector {
return false;
}
- // get segment end time to decide if segment gets selected
- long endTimeMs = segmentZKMetadata.getEndTimeMs();
- Preconditions.checkState(endTimeMs > 0, "Invalid endTimeMs: %s for
segment: %s of table: %s", endTimeMs,
- segmentZKMetadata.getSegmentName(), tableNameWithType);
- return (System.currentTimeMillis() - endTimeMs) > _segmentAgeMillis;
+ long referenceMs;
+ switch (_ageField) {
+ case CREATION_TIME:
+ referenceMs = segmentZKMetadata.getCreationTime();
+ // Segments predating the creationTime field return a non-positive
value; treat as aged so they
+ // qualify for the tier rather than failing evaluation.
+ if (referenceMs <= 0) {
+ return true;
+ }
+ break;
+ case END_TIME:
+ referenceMs = segmentZKMetadata.getEndTimeMs();
+ Preconditions.checkState(referenceMs > 0, "Invalid endTimeMs: %s for
segment: %s of table: %s", referenceMs,
+ segmentZKMetadata.getSegmentName(), tableNameWithType);
+ break;
+ case START_TIME:
+ referenceMs = segmentZKMetadata.getStartTimeMs();
+ Preconditions.checkState(referenceMs > 0, "Invalid startTimeMs: %s for
segment: %s of table: %s", referenceMs,
+ segmentZKMetadata.getSegmentName(), tableNameWithType);
+ break;
+ default:
+ throw new IllegalStateException("Unhandled segmentAgeField: " +
_ageField);
+ }
+ return (System.currentTimeMillis() - referenceMs) > _segmentAgeMillis;
}
/// Gets the age cutoff for segments accepted by this strategy
@@ -55,8 +116,14 @@ public class TimeBasedTierSegmentSelector implements
TierSegmentSelector {
return _segmentAgeMillis;
}
+ /// The [SegmentZKMetadata] field this selector compares against the age
threshold.
+ public AgeField getAgeField() {
+ return _ageField;
+ }
+
@Override
public String toString() {
- return "TimeBasedTierSegmentSelector{_segmentAgeMillis=" +
_segmentAgeMillis + "}";
+ return "TimeBasedTierSegmentSelector{_segmentAgeMillis=" +
_segmentAgeMillis
+ + ", _ageField=" + _ageField + "}";
}
}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/tier/TierSegmentSelectorTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/tier/TierSegmentSelectorTest.java
index 6b3166c8e6f..bb9ef3a0b13 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/tier/TierSegmentSelectorTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/tier/TierSegmentSelectorTest.java
@@ -85,6 +85,115 @@ public class TierSegmentSelectorTest {
Assert.assertFalse(segmentSelector.selectSegment(tableNameWithType,
realtimeSegmentZKMetadata));
}
+ @Test
+ public void testTimeBasedSegmentSelectorWithCreationTimeAgeField() {
+ long now = System.currentTimeMillis();
+ String segmentName = "segment_1";
+ String tableNameWithType = "myTable_OFFLINE";
+
+ // A segment whose *data* is 2 years old (endTime long in the past) but
was created 5 minutes ago —
+ // e.g. batch ingest of historical Iceberg data. The endTime-based default
would match any tier
+ // with any age threshold, defeating tier lifecycle intent. The
creationTime-based selector should
+ // treat this as a 5-minute-old segment.
+ SegmentZKMetadata zk = new SegmentZKMetadata(segmentName);
+ zk.setStartTime(now - TimeUnit.DAYS.toMillis(730));
+ zk.setEndTime(now - TimeUnit.DAYS.toMillis(729));
+ zk.setTimeUnit(TimeUnit.MILLISECONDS);
+ zk.setCreationTime(now - TimeUnit.MINUTES.toMillis(5));
+ zk.setStatus(Status.DONE);
+
+ // Default (END_TIME): historical data → matches ANY sane threshold.
+ TimeBasedTierSegmentSelector byEndTime = new
TimeBasedTierSegmentSelector("30m");
+ Assert.assertTrue(byEndTime.selectSegment(tableNameWithType, zk));
+
+ // creationTime with 30m threshold: segment created 5m ago → NOT matched.
+ TimeBasedTierSegmentSelector byCreation30m = new
TimeBasedTierSegmentSelector("30m",
+ TimeBasedTierSegmentSelector.AgeField.CREATION_TIME);
+ Assert.assertEquals(byCreation30m.getAgeField(),
TimeBasedTierSegmentSelector.AgeField.CREATION_TIME);
+ Assert.assertFalse(byCreation30m.selectSegment(tableNameWithType, zk));
+
+ // creationTime with 1m threshold: created 5m ago > 1m → matched.
+ TimeBasedTierSegmentSelector byCreation1m = new
TimeBasedTierSegmentSelector("1m",
+ TimeBasedTierSegmentSelector.AgeField.CREATION_TIME);
+ Assert.assertTrue(byCreation1m.selectSegment(tableNameWithType, zk));
+
+ // Segments predating the creationTime field (value <= 0) are treated as
aged and qualify for the tier.
+ SegmentZKMetadata legacy = new SegmentZKMetadata("legacy_segment");
+ legacy.setStartTime(now - TimeUnit.DAYS.toMillis(30));
+ legacy.setEndTime(now - TimeUnit.DAYS.toMillis(29));
+ legacy.setTimeUnit(TimeUnit.MILLISECONDS);
+ legacy.setStatus(Status.DONE);
+ // creationTime not set — defaults to -1
+ Assert.assertTrue(byCreation1m.selectSegment(tableNameWithType, legacy));
+
+ // Consuming segments never match regardless of reference field.
+ SegmentZKMetadata consuming = new SegmentZKMetadata("myTable__0__1__" +
now);
+ consuming.setStatus(Status.IN_PROGRESS);
+ consuming.setCreationTime(now - TimeUnit.HOURS.toMillis(1));
+ Assert.assertFalse(byCreation1m.selectSegment("myTable_REALTIME",
consuming));
+ }
+
+ @Test
+ public void testTimeBasedSegmentSelectorWithStartTimeAgeField() {
+ long now = System.currentTimeMillis();
+ String tableNameWithType = "myTable_OFFLINE";
+
+ // Segment covers a 1-day range 8 days ago: startTime 8d ago, endTime 7d
ago.
+ SegmentZKMetadata zk = new SegmentZKMetadata("segment_start");
+ zk.setStartTime(now - TimeUnit.DAYS.toMillis(8));
+ zk.setEndTime(now - TimeUnit.DAYS.toMillis(7));
+ zk.setTimeUnit(TimeUnit.MILLISECONDS);
+ zk.setStatus(Status.DONE);
+
+ // startTime with 7d threshold: startTime is 8d ago > 7d → matched.
+ TimeBasedTierSegmentSelector byStart7d = new
TimeBasedTierSegmentSelector("7d",
+ TimeBasedTierSegmentSelector.AgeField.START_TIME);
+ Assert.assertEquals(byStart7d.getAgeField(),
TimeBasedTierSegmentSelector.AgeField.START_TIME);
+ Assert.assertTrue(byStart7d.selectSegment(tableNameWithType, zk));
+
+ // startTime with 10d threshold: startTime is 8d ago < 10d → NOT matched.
+ TimeBasedTierSegmentSelector byStart10d = new
TimeBasedTierSegmentSelector("10d",
+ TimeBasedTierSegmentSelector.AgeField.START_TIME);
+ Assert.assertFalse(byStart10d.selectSegment(tableNameWithType, zk));
+ }
+
+ @Test
+ public void testAgeFieldParsing() {
+ // Default / null / empty → END_TIME (backward compatible).
+ Assert.assertEquals(TimeBasedTierSegmentSelector.AgeField.fromConfig(null),
+ TimeBasedTierSegmentSelector.AgeField.END_TIME);
+ Assert.assertEquals(TimeBasedTierSegmentSelector.AgeField.fromConfig(""),
+ TimeBasedTierSegmentSelector.AgeField.END_TIME);
+
Assert.assertEquals(TimeBasedTierSegmentSelector.AgeField.fromConfig("endTime"),
+ TimeBasedTierSegmentSelector.AgeField.END_TIME);
+
Assert.assertEquals(TimeBasedTierSegmentSelector.AgeField.fromConfig("END_TIME"),
+ TimeBasedTierSegmentSelector.AgeField.END_TIME);
+
Assert.assertEquals(TimeBasedTierSegmentSelector.AgeField.fromConfig("end_time"),
+ TimeBasedTierSegmentSelector.AgeField.END_TIME);
+
+
Assert.assertEquals(TimeBasedTierSegmentSelector.AgeField.fromConfig("startTime"),
+ TimeBasedTierSegmentSelector.AgeField.START_TIME);
+
Assert.assertEquals(TimeBasedTierSegmentSelector.AgeField.fromConfig("START_TIME"),
+ TimeBasedTierSegmentSelector.AgeField.START_TIME);
+
Assert.assertEquals(TimeBasedTierSegmentSelector.AgeField.fromConfig("start_time"),
+ TimeBasedTierSegmentSelector.AgeField.START_TIME);
+
+
Assert.assertEquals(TimeBasedTierSegmentSelector.AgeField.fromConfig("creationTime"),
+ TimeBasedTierSegmentSelector.AgeField.CREATION_TIME);
+
Assert.assertEquals(TimeBasedTierSegmentSelector.AgeField.fromConfig("CREATION_TIME"),
+ TimeBasedTierSegmentSelector.AgeField.CREATION_TIME);
+
Assert.assertEquals(TimeBasedTierSegmentSelector.AgeField.fromConfig("creation_time"),
+ TimeBasedTierSegmentSelector.AgeField.CREATION_TIME);
+
+ // Unknown value → clear error message so operators spot the typo.
+ try {
+ TimeBasedTierSegmentSelector.AgeField.fromConfig("pushTime");
+ Assert.fail("Expected IllegalArgumentException for unsupported value");
+ } catch (IllegalArgumentException expected) {
+ Assert.assertTrue(expected.getMessage().contains("pushTime"));
+ }
+ }
+
@Test
public void testRealTimeConsumingSegmentShouldNotBeRelocated() {
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TierConfig.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TierConfig.java
index ba39abc5ded..522485e650f 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TierConfig.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TierConfig.java
@@ -38,6 +38,10 @@ public class TierConfig extends BaseJsonConfig {
@JsonPropertyDescription("For 'TIME' segment selector, the period after
which to select segments for this tier")
private final String _segmentAge;
+ @JsonPropertyDescription("For 'TIME' segment selector, the segment ZK
metadata field to compare against for age. "
+ + "Accepts 'endTime' (default, backward-compatible) or 'creationTime'.")
+ private final String _segmentAgeField;
+
@JsonPropertyDescription("For 'FIXED' segment selector, the list of segments
to select for this tier")
private final List<String> _segmentList;
@@ -60,6 +64,7 @@ public class TierConfig extends BaseJsonConfig {
public TierConfig(@JsonProperty(value = "name", required = true) String name,
@JsonProperty(value = "segmentSelectorType", required = true) String
segmentSelectorType,
@JsonProperty("segmentAge") @Nullable String segmentAge,
+ @JsonProperty("segmentAgeField") @Nullable String segmentAgeField,
@JsonProperty("segmentList") @Nullable List<String> segmentList,
@JsonProperty(value = "storageType", required = true) String storageType,
@JsonProperty("serverTag") @Nullable String serverTag,
@@ -72,6 +77,7 @@ public class TierConfig extends BaseJsonConfig {
_name = name;
_segmentSelectorType = segmentSelectorType;
_segmentAge = segmentAge;
+ _segmentAgeField = segmentAgeField;
_segmentList = segmentList;
_storageType = storageType;
_serverTag = serverTag;
@@ -79,6 +85,14 @@ public class TierConfig extends BaseJsonConfig {
_tierBackendProperties = tierBackendProperties;
}
+ /// Backward-compatible constructor without segmentAgeField (defaults to
endTime).
+ public TierConfig(String name, String segmentSelectorType, @Nullable String
segmentAge,
+ @Nullable List<String> segmentList, String storageType, @Nullable String
serverTag,
+ @Nullable String tierBackend, @Nullable Map<String, String>
tierBackendProperties) {
+ this(name, segmentSelectorType, segmentAge, null, segmentList,
storageType, serverTag, tierBackend,
+ tierBackendProperties);
+ }
+
public String getName() {
return _name;
}
@@ -92,6 +106,11 @@ public class TierConfig extends BaseJsonConfig {
return _segmentAge;
}
+ @Nullable
+ public String getSegmentAgeField() {
+ return _segmentAgeField;
+ }
+
@Nullable
public List<String> getSegmentList() {
return _segmentList;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
