This is an automated email from the ASF dual-hosted git repository.
xiangfu0 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 dd2c3cdfac9 Add size-based table retention (#19770)
dd2c3cdfac9 is described below
commit dd2c3cdfac918e18a20ef20cfd2a55eb868d2054
Author: Xiang Fu <[email protected]>
AuthorDate: Sun Oct 11 01:52:49 2026 -0700
Add size-based table retention (#19770)
Add optional segmentsConfig.retentionSize (for example, "100G") for
retention by compressed live-segment bytes, enforced independently for OFFLINE
and REALTIME tables.
Evict oldest eligible segments while preserving the newest OFFLINE segment,
REALTIME recovery segments, and the stopping boundary around all retained
lineage entries. Publish a table-scoped blocked gauge when the configured cap
cannot be met.
Validation: 78 targeted local tests and scoped quality checks passed. All
CI checks passed before merge.
---
.../pinot/common/metrics/ControllerGauge.java | 5 +
.../helix/core/retention/RetentionManager.java | 245 +++++-
.../helix/core/retention/RetentionManagerTest.java | 943 +++++++++++++++++++++
.../segment/local/utils/TableConfigUtils.java | 12 +
.../segment/local/utils/TableConfigUtilsTest.java | 45 +
.../SegmentsValidationAndRetentionConfig.java | 69 +-
.../spi/utils/builder/TableConfigBuilder.java | 9 +
.../spi/utils/builder/TableConfigBuilderTest.java | 41 +
8 files changed, 1340 insertions(+), 29 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerGauge.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerGauge.java
index ae1df54e8fc..75d629dd3d7 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerGauge.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerGauge.java
@@ -208,6 +208,11 @@ public enum ControllerGauge implements
AbstractMetrics.Gauge {
// Metric used to track errors during the periodic table retention management
RETENTION_MANAGER_ERROR("retentionManagerError", false),
+ /// Status of the latest completed table size retention pass: 1 when a
configured policy cannot run safely, deletion
+ /// fails, or protected data keeps the table above its cap; 0 when the
configured policy succeeds within the cap.
+ /// Published once when each configured pass completes; absent when size
retention is disabled or unsupported.
+ SIZE_RETENTION_BLOCKED("sizeRetentionBlocked", false),
+
// Gauge to reflect whether pauseless is enabled or not
PAUSELESS_CONSUMPTION_ENABLED("pauselessConsumptionEnabled", false),
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
index b5ba82d3133..e2a65a92098 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
@@ -23,6 +23,8 @@ import com.google.common.base.Preconditions;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -35,6 +37,7 @@ import org.apache.helix.model.IdealState;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.logging.log4j.util.Strings;
+import org.apache.pinot.common.lineage.LineageEntry;
import org.apache.pinot.common.lineage.SegmentLineage;
import org.apache.pinot.common.lineage.SegmentLineageAccessHelper;
import org.apache.pinot.common.lineage.SegmentLineageUtils;
@@ -47,6 +50,7 @@ import org.apache.pinot.common.utils.URIUtils;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.PinotResourceManagerResponse;
import
org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
import
org.apache.pinot.controller.helix.core.retention.strategy.RetentionStrategy;
import
org.apache.pinot.controller.helix.core.retention.strategy.TimeRetentionStrategy;
@@ -63,6 +67,7 @@ import org.apache.pinot.spi.filesystem.PinotFS;
import org.apache.pinot.spi.filesystem.PinotFSFactory;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.CommonConstants.Segment.Realtime.Status;
+import org.apache.pinot.spi.utils.DataSizeUtils;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
@@ -122,6 +127,198 @@ public class RetentionManager extends
ControllerPeriodicTask<Void> {
// Delete segments based on segment lineage and clean up segment lineage
metadata.
manageSegmentLineageCleanupForTable(tableConfig);
+
+ // Re-read active segments after time and lineage cleanup so
already-deleted bytes do not cause extra eviction.
+ manageSizeBasedRetention(tableConfig);
+ }
+
+ /// Enforces the optional compressed-segment size limit on live data
independently of time retention. Replaced
+ /// segments are excluded from accounting. Eviction proceeds oldest first
and stops before the first active segment
+ /// listed on either side of any lineage entry, including replacement copies
excluded from accounting. The newest
+ /// OFFLINE segment and realtime recovery segments are preserved. Each typed
table uses its own size limit and
+ /// eviction order without consulting the OFFLINE time boundary.
+ /// Unknown sizes or missing metadata disable this pass; consuming segments
are excluded from accounting and eviction.
+ @VisibleForTesting
+ protected void manageSizeBasedRetention(TableConfig tableConfig) {
+ String tableNameWithType = tableConfig.getTableName();
+ String retentionSize =
tableConfig.getValidationConfig().getRetentionSize();
+ if (retentionSize == null || (tableConfig.getTableType() ==
TableType.OFFLINE
+ &&
!"APPEND".equalsIgnoreCase(IngestionConfigUtils.getBatchSegmentIngestionType(tableConfig))))
{
+ _controllerMetrics.removeTableGauge(tableNameWithType,
ControllerGauge.SIZE_RETENTION_BLOCKED);
+ return;
+ }
+ boolean blocked = true;
+ try {
+ blocked = !enforceSizeBasedRetention(tableConfig, retentionSize);
+ } finally {
+ // Publish only the completed pass's result, including a blocked result
if an exception interrupted it.
+ _controllerMetrics.setOrUpdateTableGauge(tableNameWithType,
ControllerGauge.SIZE_RETENTION_BLOCKED,
+ blocked ? 1 : 0);
+ }
+ }
+
+ /// Returns whether the table is within its size cap after this pass; unsafe
skips and failed evictions return false.
+ private boolean enforceSizeBasedRetention(TableConfig tableConfig, String
retentionSize) {
+ String tableNameWithType = tableConfig.getTableName();
+ long retentionSizeBytes;
+ try {
+ retentionSizeBytes = DataSizeUtils.toBytes(retentionSize);
+ Preconditions.checkArgument(retentionSizeBytes > 0, "Retention size must
be positive");
+ } catch (Exception e) {
+ LOGGER.warn("Invalid retention size: {} for table: {}, skip size
retention", retentionSize, tableNameWithType);
+ return false;
+ }
+
+ // Read IdealState before lineage, matching the existing live-segment
filtering order. A replacement starting
+ // after this snapshot cannot add both its source and destination to the
bytes counted by this pass.
+ Set<String> activeSegments = new
HashSet<>(_pinotHelixResourceManager.getSegmentsFor(tableNameWithType, false));
+ SegmentLineage segmentLineage =
+
SegmentLineageAccessHelper.getSegmentLineage(_pinotHelixResourceManager.getPropertyStore(),
tableNameWithType);
+ Map<String, LineageEntry> lineageEntries = segmentLineage == null ?
Map.of() : segmentLineage.getLineageEntries();
+ Set<String> lineageOwnedSegments = new HashSet<>();
+ // Size retention protects both sides of every retained entry, regardless
of state, age, or the deletion flag.
+ lineageEntries.values().forEach(entry -> {
+ lineageOwnedSegments.addAll(entry.getSegmentsFrom());
+ lineageOwnedSegments.addAll(entry.getSegmentsTo());
+ });
+ lineageOwnedSegments.retainAll(activeSegments);
+ // Replaced segments awaiting lineage cleanup are no longer queryable and
must not trigger live-data eviction.
+ SegmentLineageUtils.filterSegmentsBasedOnLineageInPlace(activeSegments,
segmentLineage);
+ if (activeSegments.isEmpty()) {
+ return true;
+ }
+ List<SegmentZKMetadata> metadataList =
_pinotHelixResourceManager.getSegmentsZKMetadata(tableNameWithType);
+ Map<String, SegmentZKMetadata> lineageOwnedMetadata = new HashMap<>();
+ Set<String> segmentsWithMetadata = new HashSet<>();
+ List<SegmentZKMetadata> completedSegments = new ArrayList<>();
+ Map<String, Long> completedSegmentSizes = new HashMap<>();
+ boolean realtime = tableConfig.getTableType() == TableType.REALTIME;
+ long retainedBytes = 0L;
+ for (SegmentZKMetadata metadata : metadataList) {
+ String segmentName = metadata.getSegmentName();
+ if (lineageOwnedSegments.contains(segmentName)) {
+ lineageOwnedMetadata.put(segmentName, metadata);
+ }
+ if (!activeSegments.contains(segmentName)) {
+ continue;
+ }
+ segmentsWithMetadata.add(segmentName);
+ if (realtime && !metadata.getStatus().isCompleted()) {
+ continue;
+ }
+ long size = getSizeRetentionSegmentSize(tableNameWithType, metadata);
+ if (size < 0) {
+ LOGGER.warn("Unknown size for segment: {} in table: {}, skip size
retention", segmentName, tableNameWithType);
+ return false;
+ }
+ // Plain addition assumes realistic table sizes; corrupt totals above
Long.MAX_VALUE may wrap and report healthy.
+ retainedBytes += size;
+ completedSegments.add(metadata);
+ completedSegmentSizes.put(segmentName, size);
+ }
+ if (!segmentsWithMetadata.containsAll(activeSegments)) {
+ LOGGER.warn("Missing active segment metadata for table: {}, skip size
retention", tableNameWithType);
+ return false;
+ }
+ if (retainedBytes <= retentionSizeBytes) {
+ return true;
+ }
+
+ Comparator<SegmentZKMetadata> evictionOrder =
Comparator.comparingLong(RetentionManager::getSizeRetentionTimestamp)
+ .thenComparing(SegmentZKMetadata::getSegmentName);
+ // Shadow copies do not count toward the cap, but still stop eviction at
their position in the ordering.
+ // Without metadata or a timestamp for any active lineage-owned segment,
its position cannot be determined safely.
+ if (lineageOwnedMetadata.size() != lineageOwnedSegments.size() ||
lineageOwnedMetadata.values().stream()
+ .anyMatch(metadata -> getSizeRetentionTimestamp(metadata) < 0)) {
+ LOGGER.warn("Cannot order lineage-owned segments for table: {}, skip
size retention", tableNameWithType);
+ return false;
+ }
+ SegmentZKMetadata lineageBarrier =
lineageOwnedMetadata.values().stream().min(evictionOrder).orElse(null);
+ completedSegments.removeIf(metadata -> getSizeRetentionTimestamp(metadata)
< 0);
+ if (realtime) {
+ Set<String> lastCompletedSegments =
+ new
HashSet<>(_pinotHelixResourceManager.getLastLLCCompletedSegments(metadataList));
+ completedSegments.removeIf(metadata ->
lastCompletedSegments.contains(metadata.getSegmentName()));
+ } else {
+ // A size typo must not remove the entire OFFLINE table, even if its
newest segment exceeds the cap alone.
+ completedSegments.stream().max(evictionOrder)
+ .ifPresent(newest -> completedSegments.removeIf(metadata -> metadata
== newest));
+ }
+ completedSegments.sort(evictionOrder);
+
+ List<String> segmentsToDelete = new ArrayList<>();
+ long projectedRetainedBytes = retainedBytes;
+ for (SegmentZKMetadata metadata : completedSegments) {
+ if (projectedRetainedBytes <= retentionSizeBytes) {
+ break;
+ }
+ // Stop at the lineage boundary even when that segment is excluded by
another protection or from accounting.
+ // This snapshot rule applies regardless of the lineage-exclusive
deletion flag.
+ if (lineageBarrier != null && evictionOrder.compare(metadata,
lineageBarrier) >= 0) {
+ LOGGER.info("Stop size retention for table: {} at lineage-owned
segment: {}", tableNameWithType,
+ lineageBarrier.getSegmentName());
+ break;
+ }
+ long size = completedSegmentSizes.get(metadata.getSegmentName());
+ if (size == 0) {
+ continue;
+ }
+ segmentsToDelete.add(metadata.getSegmentName());
+ projectedRetainedBytes -= size;
+ }
+ if (!segmentsToDelete.isEmpty()) {
+ PinotResourceManagerResponse response;
+ // Coordinate with local lineage writers and reject stale accounting or
a changed stopping point, even when
+ // lineage-exclusive deletion is disabled. Other controllers can still
update lineage after this fresh read.
+ synchronized
(_pinotHelixResourceManager.getLineageUpdaterLock(tableNameWithType)) {
+ SegmentLineage currentLineage =
+
SegmentLineageAccessHelper.getSegmentLineage(_pinotHelixResourceManager.getPropertyStore(),
+ tableNameWithType);
+ Map<String, LineageEntry> currentEntries =
+ currentLineage == null ? Map.of() :
currentLineage.getLineageEntries();
+ if (!lineageEntries.equals(currentEntries)) {
+ LOGGER.warn("Segment lineage changed for table: {}, skip size
retention deletion", tableNameWithType);
+ return false;
+ }
+ response =
_pinotHelixResourceManager.deleteSegments(tableNameWithType, segmentsToDelete);
+ }
+ if (!response.isSuccessful()) {
+ LOGGER.warn("Size retention deletion failed for table: {}, bytes
retained: {}, cap: {}, response: {}",
+ tableNameWithType, retainedBytes, retentionSize, response);
+ return false;
+ }
+ retainedBytes = projectedRetainedBytes;
+ LOGGER.info("Deleted {} oldest segments from table: {} for retention
size: {}", segmentsToDelete.size(),
+ tableNameWithType, retentionSize);
+ }
+ if (retainedBytes > retentionSizeBytes) {
+ LOGGER.warn("Table: {} remains above retention size: {} due to
protected, undated, or lineage-owned segments",
+ tableNameWithType, retentionSize);
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ protected void nonLeaderCleanup(List<String> tableNamesWithType) {
+ for (String tableNameWithType : tableNamesWithType) {
+ _controllerMetrics.removeTableGauge(tableNameWithType,
ControllerGauge.SIZE_RETENTION_BLOCKED);
+ }
+ }
+
+ /// Returns the compressed bytes used by size retention, or a negative value
when the size is unknown.
+ /// Subclasses can supply authoritative sizes for logical segments that
aggregate multiple physical segments.
+ protected long getSizeRetentionSegmentSize(String tableNameWithType,
SegmentZKMetadata metadata) {
+ return metadata.getSizeInBytes();
+ }
+
+ private static long getSizeRetentionTimestamp(SegmentZKMetadata metadata) {
+ long endTime = metadata.getEndTimeMs();
+ if (endTime >= 0) {
+ return endTime;
+ }
+ long creationTime = metadata.getCreationTime();
+ return creationTime >= 0 ? creationTime : metadata.getPushTime();
}
@Override
@@ -239,29 +436,8 @@ public class RetentionManager extends
ControllerPeriodicTask<Void> {
LOGGER.info("Managing retention for hybrid table: {}",
realtimeTableConfig.getTableName());
List<String> segmentsToDelete = new ArrayList<>();
String realtimeTableName = realtimeTableConfig.getTableName();
- String rawTableName =
TableNameBuilder.extractRawTableName(realtimeTableName);
- String offlineTableName =
TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
try {
- ZkHelixPropertyStore<ZNRecord> propertyStore =
_pinotHelixResourceManager.getPropertyStore();
- Schema schema = ZKMetadataProvider.getTableSchema(propertyStore,
offlineTableName);
- Preconditions.checkState(schema != null, "Failed to get schema for
table: " + offlineTableName);
- String timeColumn = null;
- SegmentsValidationAndRetentionConfig validationConfig =
offlineTableConfig.getValidationConfig();
- if (validationConfig != null) {
- timeColumn = validationConfig.getTimeColumnName();
- }
- Preconditions.checkState(StringUtils.isNotEmpty(timeColumn),
- "TimeColumn is null or empty for table: " + offlineTableName);
- DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(timeColumn);
- Preconditions.checkState(dateTimeSpec != null, String.format(
- "Failed to get DateTimeFieldSpec for time column: %s of table: %s",
timeColumn, offlineTableName));
- DateTimeFormatSpec timeFormatSpec = dateTimeSpec.getFormatSpec();
- TimeBoundaryInfo timeBoundaryInfo =
_brokerServiceHelper.getTimeBoundaryInfo(offlineTableConfig);
- Preconditions.checkState(timeBoundaryInfo != null,
- "Failed to get time boundary info for table: " + offlineTableName);
- long timeBoundaryMs =
timeFormatSpec.fromFormatToMillis(timeBoundaryInfo.getTimeValue());
- Preconditions.checkState(timeBoundaryMs > 0,
- "Failed to determine a valid time boundary for table: " +
offlineTableName);
+ long timeBoundaryMs = getHybridTimeBoundaryMs(offlineTableConfig);
// Iterate over all COMPLETED segments of the REALTIME table and check
if they are eligible for deletion.
for (SegmentZKMetadata segmentZKMetadata :
_pinotHelixResourceManager.getSegmentsZKMetadata(realtimeTableName)) {
@@ -287,6 +463,31 @@ public class RetentionManager extends
ControllerPeriodicTask<Void> {
}
}
+ private long getHybridTimeBoundaryMs(TableConfig offlineTableConfig) {
+ String offlineTableName = offlineTableConfig.getTableName();
+ ZkHelixPropertyStore<ZNRecord> propertyStore =
_pinotHelixResourceManager.getPropertyStore();
+ Schema schema = ZKMetadataProvider.getTableSchema(propertyStore,
offlineTableName);
+ Preconditions.checkState(schema != null, "Failed to get schema for table:
" + offlineTableName);
+ String timeColumn = null;
+ SegmentsValidationAndRetentionConfig validationConfig =
offlineTableConfig.getValidationConfig();
+ if (validationConfig != null) {
+ timeColumn = validationConfig.getTimeColumnName();
+ }
+ Preconditions.checkState(StringUtils.isNotEmpty(timeColumn),
+ "TimeColumn is null or empty for table: " + offlineTableName);
+ DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(timeColumn);
+ Preconditions.checkState(dateTimeSpec != null, String.format(
+ "Failed to get DateTimeFieldSpec for time column: %s of table: %s",
timeColumn, offlineTableName));
+ DateTimeFormatSpec timeFormatSpec = dateTimeSpec.getFormatSpec();
+ TimeBoundaryInfo timeBoundaryInfo =
_brokerServiceHelper.getTimeBoundaryInfo(offlineTableConfig);
+ Preconditions.checkState(timeBoundaryInfo != null,
+ "Failed to get time boundary info for table: " + offlineTableName);
+ long timeBoundaryMs =
timeFormatSpec.fromFormatToMillis(timeBoundaryInfo.getTimeValue());
+ Preconditions.checkState(timeBoundaryMs > 0,
+ "Failed to determine a valid time boundary for table: " +
offlineTableName);
+ return timeBoundaryMs;
+ }
+
private boolean shouldDeleteInProgressLLCSegment(String segmentName,
IdealState idealState,
SegmentZKMetadata segmentZKMetadata) {
if (idealState == null) {
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
index c0023bda878..a442d598587 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
@@ -30,6 +30,8 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.io.FileUtils;
import org.apache.helix.AccessOption;
import org.apache.helix.HelixAdmin;
@@ -37,17 +39,24 @@ import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.pinot.common.lineage.LineageEntry;
+import org.apache.pinot.common.lineage.LineageEntryState;
+import org.apache.pinot.common.lineage.SegmentLineage;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.metrics.ControllerGauge;
import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.common.metrics.MetricValueUtils;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.PinotResourceManagerResponse;
import org.apache.pinot.controller.helix.core.PinotTableIdealStateBuilder;
import org.apache.pinot.controller.helix.core.SegmentDeletionManager;
import org.apache.pinot.controller.util.BrokerServiceHelper;
import org.apache.pinot.controller.util.CompletionServiceHelper;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils;
+import org.apache.pinot.core.routing.timeboundary.TimeBoundaryInfo;
import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
@@ -59,8 +68,10 @@ import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.metrics.PinotMetricUtils;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.CommonConstants.Segment.Realtime.Status;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.apache.zookeeper.data.Stat;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -71,6 +82,7 @@ import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.*;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
@@ -915,4 +927,935 @@ public class RetentionManagerTest {
// already asserts the correct segments via TestNG assertions
verify(pinotHelixResourceManager,
times(1)).deleteSegments(eq(OFFLINE_TABLE_NAME), anyList());
}
+
+ @Test
+ public void testSizeOnlyRetentionDeletesOldestSegmentsUntilWithinLimit() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "100B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("newest", 60, now),
+ createSizeRetentionSegment("oldest", 60, now - 2),
+ createSizeRetentionSegment("middle", 60, now - 1));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+
+ // Exercise the complete table pass: a time-retention configuration is
deliberately absent.
+ retentionManager.processTable(OFFLINE_TABLE_NAME);
+
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest", "middle"));
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 0);
+ }
+
+ @Test
+ public void testSizeRetentionKeepsSegmentsAtOrBelowLimit() {
+ for (String limit : List.of("100B", "101B")) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, limit);
+ tableConfig.getValidationConfig().setReplication("3");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 40, now - 1),
+ createSizeRetentionSegment("newest", 60, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ // Stored archive sizes count once, regardless of table replication.
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 0);
+ }
+ }
+
+ @Test
+ public void
testSizeRetentionKeepsNewestOfflineSegmentWhenItAloneExceedsLimit() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "100B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("newest", 120, now),
+ createSizeRetentionSegment("oldest", 40, now - 1));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest"));
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+
+ @Test
+ public void testSizeRetentionNeverDeletesOnlyOfflineSegment() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "1B");
+ List<SegmentZKMetadata> segments =
+ List.of(createSizeRetentionSegment("only", 100,
System.currentTimeMillis()));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+
+ @Test
+ public void testSizeRetentionOrderingWithCreationAndPushTimeFallbacks() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "20B");
+ long now = System.currentTimeMillis();
+ SegmentZKMetadata creationFallback =
createSizeRetentionSegment("creationFallback", 10, -1);
+ creationFallback.setCreationTime(now - 1);
+ SegmentZKMetadata pushFallback =
createSizeRetentionSegment("pushFallback", 10, -1);
+ pushFallback.setCreationTime(-1);
+ pushFallback.setPushTime(now - 4);
+ List<SegmentZKMetadata> segments = List.of(creationFallback,
+ createSizeRetentionSegment("b", 10, now - 3), pushFallback,
+ createSizeRetentionSegment("oldest", 10, now - 5),
createSizeRetentionSegment("a", 10, now - 3));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest", "pushFallback", "a"));
+ }
+
+ @Test
+ public void
testSizeRetentionProtectsIncompleteAndLastCompletedRealtimeSegments() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.REALTIME, "50B");
+ long now = System.currentTimeMillis();
+ SegmentZKMetadata oldest = createSizeRetentionSegment(
+ new LLCSegmentName(TEST_TABLE_NAME, 0, 4, now - 3).getSegmentName(),
20, now - 3);
+ oldest.setStatus(Status.DONE);
+ SegmentZKMetadata lastCompleted = createSizeRetentionSegment(
+ new LLCSegmentName(TEST_TABLE_NAME, 0, 5, now - 1).getSegmentName(),
40, now - 1);
+ lastCompleted.setStatus(Status.DONE);
+ SegmentZKMetadata consuming = createSizeRetentionSegment("consuming", -1,
now - 5);
+ consuming.setStatus(Status.IN_PROGRESS);
+ SegmentZKMetadata committing = createSizeRetentionSegment("committing",
999, now - 4);
+ committing.setStatus(Status.COMMITTING);
+ SegmentZKMetadata uploaded = createSizeRetentionSegment("uploaded", 60,
now - 2);
+ List<SegmentZKMetadata> segments = List.of(lastCompleted, committing,
uploaded, consuming, oldest);
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(REALTIME_TABLE_NAME,
List.of(oldest.getSegmentName(), "uploaded"));
+ }
+
+ @Test
+ public void testSizeRetentionSkipsTableWhenCompletedSegmentSizeIsUnknown() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "50B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 60, now - 2),
+ createSizeRetentionSegment("unknown", -1, now - 1),
createSizeRetentionSegment("newest", 60, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+
+ @Test
+ public void testSizeRetentionIgnoresMetadataOutsideIdealState() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "100B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 40, now - 2),
+ createSizeRetentionSegment("orphan", -1, now - 1),
createSizeRetentionSegment("newest", 40, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager);
+ when(resourceManager.getSegmentsFor(OFFLINE_TABLE_NAME,
false)).thenReturn(List.of("oldest", "newest"));
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ }
+
+ @Test
+ public void
testSizeRetentionStopsAtInProgressLineageEvenWhenExclusiveDeleteIsDisabled() {
+ for (boolean exclusiveDelete : List.of(true, false)) {
+ for (String retentionSize : List.of("50B", "160B")) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, retentionSize);
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("olderLive", 60, now - 3),
+ createSizeRetentionSegment("source", 80, now - 2),
+ createSizeRetentionSegment("destination", -1, now - 1),
+ createSizeRetentionSegment("newerLive", 20, now - 1),
+ createSizeRetentionSegment("newestLive", 60, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ ControllerConf conf = new ControllerConf();
+ conf.setProperty(ControllerConf.LINEAGE_EXCLUSIVE_DELETE_ENABLED,
exclusiveDelete);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics,
+ conf, mock(BrokerServiceHelper.class));
+ setupSizeRetentionLineage(resourceManager, OFFLINE_TABLE_NAME,
"source", "destination",
+ LineageEntryState.IN_PROGRESS);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ // Evict the older prefix, then stop at the source instead of jumping
over it to newer live data.
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("olderLive"));
+ verify(resourceManager,
never()).deleteSegmentsForLineageCleanup(anyString(), anyList());
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME,
retentionSize.equals("160B") ? 0 : 1);
+ }
+ }
+ }
+
+ @Test
+ public void
testSizeRetentionExcludesCompletedReplacedSegmentsFromAccounting() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "100B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("olderLive", 60, now - 3),
+ createSizeRetentionSegment("source", -1, now - 2),
+ createSizeRetentionSegment("destination", 40, now - 1),
createSizeRetentionSegment("newest", 40, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ SegmentLineage lineage = new SegmentLineage(OFFLINE_TABLE_NAME);
+ lineage.addLineageEntry("replacement",
+ new LineageEntry(List.of("source"), List.of("destination"),
LineageEntryState.COMPLETED, now));
+ when(resourceManager.getPropertyStore().get(anyString(), any(Stat.class),
eq(AccessOption.PERSISTENT)))
+ .thenReturn(lineage.toZNRecord());
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).getSegmentsFor(OFFLINE_TABLE_NAME, false);
+ verify(resourceManager, never()).getSegmentsFor(OFFLINE_TABLE_NAME, true);
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("olderLive"));
+ verify(resourceManager.getPropertyStore(), times(2))
+ .get(anyString(), any(Stat.class), eq(AccessOption.PERSISTENT));
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 0);
+ }
+
+ @Test
+ public void testSizeRetentionSeesLineageStartedWhileReadingIdealState() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "50B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("olderLive", 60, now - 3),
+ createSizeRetentionSegment("source", 80, now - 2),
+ createSizeRetentionSegment("destination", -1, now - 1),
createSizeRetentionSegment("newerLive", 20, now - 1),
+ createSizeRetentionSegment("newestLive", 60, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ SegmentLineage lineage = new SegmentLineage(OFFLINE_TABLE_NAME);
+ lineage.addLineageEntry("replacement",
+ new LineageEntry(List.of("source"), List.of("destination"),
LineageEntryState.IN_PROGRESS, now));
+ AtomicReference<ZNRecord> lineageRecord = new AtomicReference<>();
+ ZkHelixPropertyStore<ZNRecord> propertyStore =
resourceManager.getPropertyStore();
+ when(propertyStore.get(anyString(), any(Stat.class),
eq(AccessOption.PERSISTENT)))
+ .thenAnswer(invocation -> lineageRecord.get());
+ when(resourceManager.getSegmentsFor(OFFLINE_TABLE_NAME,
false)).thenAnswer(invocation -> {
+ lineageRecord.set(lineage.toZNRecord());
+ return segments.stream().map(SegmentZKMetadata::getSegmentName).toList();
+ });
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ // The newer lineage snapshot sees replacement bytes exposed by the
just-read ideal state.
+ var readOrder = inOrder(resourceManager, propertyStore);
+ readOrder.verify(resourceManager).getSegmentsFor(OFFLINE_TABLE_NAME,
false);
+ readOrder.verify(propertyStore).get(anyString(), any(Stat.class),
eq(AccessOption.PERSISTENT));
+
readOrder.verify(resourceManager).getSegmentsZKMetadata(OFFLINE_TABLE_NAME);
+ readOrder.verify(propertyStore).get(anyString(), any(Stat.class),
eq(AccessOption.PERSISTENT));
+ readOrder.verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("olderLive"));
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+
+ @Test
+ public void testSizeRetentionStopsBeforeExcludedLineageShadowSegment() {
+ for (LineageEntryState state : LineageEntryState.values()) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "1B");
+ long now = System.currentTimeMillis();
+ String shadow = state == LineageEntryState.COMPLETED ? "source" :
"destination";
+ String live = state == LineageEntryState.COMPLETED ? "destination" :
"source";
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 40, now - 4),
+ createSizeRetentionSegment(shadow, -1, now - 3),
createSizeRetentionSegment("newerLive", 40, now - 2),
+ createSizeRetentionSegment(live, 80, now - 1),
createSizeRetentionSegment("newest", 40, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ setupSizeRetentionLineage(resourceManager, OFFLINE_TABLE_NAME, "source",
"destination", state);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ // Shadow bytes do not count, but their position still prevents deleting
newer queryable data.
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest"));
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+ }
+
+ @Test
+ public void testSizeRetentionDoesNotEvictWhenOldestSegmentIsLineageOwned() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "1B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("source", 80, now - 2),
+ createSizeRetentionSegment("destination", -1, now - 1),
createSizeRetentionSegment("newerLive", 40, now - 1),
+ createSizeRetentionSegment("newest", 40, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ setupSizeRetentionLineage(resourceManager, OFFLINE_TABLE_NAME, "source",
"destination",
+ LineageEntryState.IN_PROGRESS);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+
+ @Test
+ public void testSizeRetentionCannotCrossUnlocatableLineageShadow() {
+ for (boolean missingMetadata : List.of(true, false)) {
+ for (String retentionSize : List.of("50B", "200B")) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, retentionSize);
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments = new
ArrayList<>(List.of(createSizeRetentionSegment("oldest", 40, now - 3),
+ createSizeRetentionSegment("newerLive", 40, now - 2),
+ createSizeRetentionSegment("destination", 40, now - 1),
createSizeRetentionSegment("newest", 40, now)));
+ if (!missingMetadata) {
+ SegmentZKMetadata undatedSource =
createSizeRetentionSegment("source", -1, -1);
+ undatedSource.setCreationTime(-1);
+ undatedSource.setPushTime(-1);
+ segments.add(undatedSource);
+ }
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ when(resourceManager.getSegmentsFor(OFFLINE_TABLE_NAME, false))
+ .thenReturn(List.of("oldest", "newerLive", "source",
"destination", "newest"));
+ setupSizeRetentionLineage(resourceManager, OFFLINE_TABLE_NAME,
"source", "destination",
+ LineageEntryState.COMPLETED);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ // Without the active shadow's timestamp the prefix cannot be located
safely.
+ // An already satisfied cap is healthy.
+ verify(resourceManager, never()).deleteSegments(anyString(),
anyList());
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME,
retentionSize.equals("200B") ? 0 : 1);
+ }
+ }
+ }
+
+ @Test
+ public void
testSizeRetentionLineageBarrierUsesTimestampFallbacksAndNameTieBreak() {
+ for (boolean usePushTime : List.of(true, false)) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "1B");
+ long now = System.currentTimeMillis();
+ SegmentZKMetadata shadow = createSizeRetentionSegment("b", -1, -1);
+ shadow.setCreationTime(usePushTime ? -1 : now);
+ shadow.setPushTime(now);
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("a", 20, now), shadow,
+ createSizeRetentionSegment("z", 20, now),
createSizeRetentionSegment("newest", 20, now + 1));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ setupSizeRetentionLineage(resourceManager, OFFLINE_TABLE_NAME, "b",
"absentDestination",
+ LineageEntryState.COMPLETED);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME, List.of("a"));
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+ }
+
+ @Test
+ public void testSizeRetentionIgnoresLineageWithoutActiveBlockedSegments() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "40B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 40, now - 2),
+ createSizeRetentionSegment("middle", 40, now - 1),
createSizeRetentionSegment("newest", 40, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ setupSizeRetentionLineage(resourceManager, OFFLINE_TABLE_NAME,
"absentSource", "absentDestination",
+ LineageEntryState.IN_PROGRESS);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest", "middle"));
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 0);
+ }
+
+ @Test
+ public void testSizeRetentionRevertedSourceCreatesBarrier() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "40B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 40, now - 3),
+ createSizeRetentionSegment("source", 40, now - 2),
createSizeRetentionSegment("destination", -1, now - 1),
+ createSizeRetentionSegment("newest", 40, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ setupSizeRetentionLineage(resourceManager, OFFLINE_TABLE_NAME, "source",
"destination", LineageEntryState.REVERTED);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest"));
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+
+ @Test
+ public void
testSizeRetentionCompletedDestinationCreatesBarrierWithSourceAbsentOrNewer() {
+ for (boolean sourceAbsent : List.of(true, false)) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "1B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments = new
ArrayList<>(List.of(createSizeRetentionSegment("oldest", 40, now - 3),
+ createSizeRetentionSegment("destination", 40, now - 2),
createSizeRetentionSegment("newerLive", 40, now - 1),
+ createSizeRetentionSegment("newest", 40, now)));
+ if (!sourceAbsent) {
+ segments.add(createSizeRetentionSegment("source", -1, now + 1));
+ }
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ ControllerConf conf = new ControllerConf();
+ conf.setProperty(ControllerConf.LINEAGE_EXCLUSIVE_DELETE_ENABLED, false);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics,
+ conf, mock(BrokerServiceHelper.class));
+ setupSizeRetentionLineage(resourceManager, OFFLINE_TABLE_NAME, "source",
"destination",
+ LineageEntryState.COMPLETED, now - TimeUnit.DAYS.toMillis(400));
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest"));
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+ }
+
+ @Test
+ public void
testSizeRetentionRejectsPrefixWhenLineageMembershipChangesWithExclusiveDeleteDisabled()
{
+ for (LineageEntryState state : LineageEntryState.values()) {
+ for (boolean selectedTarget : List.of(true, false)) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "60B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("lateMarker", 0, now - 3),
+ createSizeRetentionSegment("oldest", 60, now - 2),
createSizeRetentionSegment("middle", 60, now - 1),
+ createSizeRetentionSegment("newest", 60, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ ControllerConf conf = new ControllerConf();
+ conf.setProperty(ControllerConf.LINEAGE_EXCLUSIVE_DELETE_ENABLED,
false);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics,
+ conf, mock(BrokerServiceHelper.class));
+ SegmentLineage lineage = new SegmentLineage(OFFLINE_TABLE_NAME);
+ lineage.addLineageEntry("replacement", new
LineageEntry(List.of("absentSource"),
+ List.of(selectedTarget ? "oldest" : "lateMarker"), state, now));
+ when(resourceManager.getPropertyStore().get(anyString(),
any(Stat.class), eq(AccessOption.PERSISTENT)))
+ .thenReturn(null, lineage.toZNRecord());
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ // An earlier boundary change invalidates the batch even when no
selected target is a new lineage member.
+ verify(resourceManager, never()).deleteSegments(anyString(),
anyList());
+ verify(resourceManager.getPropertyStore(), times(2))
+ .get(anyString(), any(Stat.class), eq(AccessOption.PERSISTENT));
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+ }
+ }
+
+ @Test
+ public void testSizeRetentionReadsFreshLineageAndDeletesUnderUpdaterLock() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "60B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 60, now - 1),
+ createSizeRetentionSegment("newest", 60, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ Object updaterLock =
resourceManager.getLineageUpdaterLock(OFFLINE_TABLE_NAME);
+ AtomicInteger lineageReads = new AtomicInteger();
+ when(resourceManager.getPropertyStore().get(anyString(), any(Stat.class),
eq(AccessOption.PERSISTENT)))
+ .thenAnswer(invocation -> {
+ if (lineageReads.incrementAndGet() == 2) {
+ assertTrue(Thread.holdsLock(updaterLock));
+ }
+ return null;
+ });
+ when(resourceManager.deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest"))).thenAnswer(invocation -> {
+ assertTrue(Thread.holdsLock(updaterLock));
+ return PinotResourceManagerResponse.SUCCESS;
+ });
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ assertEquals(lineageReads.get(), 2);
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest"));
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 0);
+ }
+
+ @Test
+ public void
testSizeRetentionRejectsPrefixWhenLineageStateOrTimestampChanges() {
+ for (boolean stateChanges : List.of(true, false)) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "160B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 60, now - 3),
+ createSizeRetentionSegment("source", 80, now - 2),
createSizeRetentionSegment("destination", 40, now - 1),
+ createSizeRetentionSegment("newest", 60, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ SegmentLineage planningLineage = new SegmentLineage(OFFLINE_TABLE_NAME);
+ planningLineage.addLineageEntry("replacement",
+ new LineageEntry(List.of("source"), List.of("destination"),
LineageEntryState.IN_PROGRESS, now));
+ SegmentLineage latestLineage = new SegmentLineage(OFFLINE_TABLE_NAME);
+ latestLineage.addLineageEntry("replacement", new
LineageEntry(List.of("source"), List.of("destination"),
+ stateChanges ? LineageEntryState.COMPLETED :
LineageEntryState.IN_PROGRESS, stateChanges ? now : now + 1));
+ when(resourceManager.getPropertyStore().get(anyString(),
any(Stat.class), eq(AccessOption.PERSISTENT)))
+ .thenReturn(planningLineage.toZNRecord(),
latestLineage.toZNRecord());
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ // The members are unchanged, but a state transition changes which side
is queryable and how bytes are counted.
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+ }
+
+ @Test
+ public void
testSizeRetentionPublishesBlockedGaugeWhenFreshLineageReadThrows() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "60B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 60, now - 1),
+ createSizeRetentionSegment("newest", 60, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ when(resourceManager.getPropertyStore().get(anyString(), any(Stat.class),
eq(AccessOption.PERSISTENT)))
+ .thenReturn(null).thenThrow(new IllegalStateException("Fresh lineage
read failed"));
+
+ assertThrows(IllegalStateException.class, () ->
retentionManager.manageSizeBasedRetention(tableConfig));
+
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+
+ @Test
+ public void testSizeRetentionRecoveryProtectionCannotHideLineageBarrier() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.REALTIME, "1B");
+ long now = System.currentTimeMillis();
+ String sourceName = new LLCSegmentName(TEST_TABLE_NAME, 0, 5, now -
3).getSegmentName();
+ SegmentZKMetadata source = createSizeRetentionSegment(sourceName, 80, now
- 3);
+ source.setStatus(Status.DONE);
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 40, now - 4), source,
+ createSizeRetentionSegment("destination", -1, now - 2),
createSizeRetentionSegment("newerLive", 40, now - 1),
+ createSizeRetentionSegment("newest", 40, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ setupSizeRetentionLineage(resourceManager, REALTIME_TABLE_NAME,
sourceName, "destination",
+ LineageEntryState.IN_PROGRESS);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(REALTIME_TABLE_NAME,
List.of("oldest"));
+ verifySizeRetentionGauge(metrics, REALTIME_TABLE_NAME, 1);
+ }
+
+ @Test
+ public void testSizeRetentionForHybridStopsAtLineageBarrier() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.REALTIME, "1B");
+ long boundary = System.currentTimeMillis();
+ SegmentZKMetadata source = createSizeRetentionSegment("source", 80, -1);
+ source.setCreationTime(boundary - 2);
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 20, boundary - 3), source,
+ createSizeRetentionSegment("coveredNewer", 20, boundary - 1),
+ createSizeRetentionSegment("newest", 20, boundary + 1),
+ createSizeRetentionSegment("destination", -1, boundary + 2));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ BrokerServiceHelper brokerServiceHelper = mock(BrokerServiceHelper.class);
+ ControllerConf conf = new ControllerConf();
+ conf.setProperty(ControllerConf.ENABLE_HYBRID_TABLE_RETENTION_STRATEGY,
true);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics,
+ conf, brokerServiceHelper);
+ setupSizeRetentionOfflineCounterpart(resourceManager);
+ setupSizeRetentionLineage(resourceManager, REALTIME_TABLE_NAME, "source",
"destination",
+ LineageEntryState.IN_PROGRESS);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(REALTIME_TABLE_NAME,
List.of("oldest"));
+ verify(resourceManager, never()).getOfflineTableConfig(anyString());
+ verify(brokerServiceHelper,
never()).getTimeBoundaryInfo(any(TableConfig.class));
+ verifySizeRetentionGauge(metrics, REALTIME_TABLE_NAME, 1);
+ }
+
+ @Test
+ public void
testSizeRetentionKeepsNewestOfflineSegmentUsingFallbackAndNameTieBreak() {
+ for (boolean usePushTime : List.of(true, false)) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "1B");
+ long now = System.currentTimeMillis();
+ SegmentZKMetadata newest = createSizeRetentionSegment("z", 20, -1);
+ newest.setCreationTime(usePushTime ? -1 : now);
+ newest.setPushTime(now);
+ SegmentZKMetadata sameTimestamp = createSizeRetentionSegment("a", 20,
now);
+ List<SegmentZKMetadata> segments = List.of(newest, sameTimestamp,
+ createSizeRetentionSegment("oldest", 20, now - 1));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest", "a"));
+ }
+ }
+
+ @Test
+ public void
testSizeRetentionGaugeClearsForEmptyTableAndWhenLeadershipIsLost() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "100B");
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, List.of(), resourceManager, metrics);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 0);
+
+ retentionManager.nonLeaderCleanup(List.of(OFFLINE_TABLE_NAME));
+
+ verify(metrics).removeTableGauge(OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED);
+ }
+
+ @Test
+ public void testSizeRetentionGaugeRecoversWhenUnknownSizeBecomesAvailable() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "100B");
+ long now = System.currentTimeMillis();
+ SegmentZKMetadata unknown = createSizeRetentionSegment("unknown", -1, now
- 1);
+ List<SegmentZKMetadata> segments = List.of(unknown,
createSizeRetentionSegment("newest", 40, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ unknown.setSizeInBytes(60);
+ clearInvocations(metrics);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 0);
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ }
+
+ @Test
+ public void testSizeRetentionGaugeRegistryLifecycle() {
+ for (boolean refreshTable : List.of(true, false)) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "100B");
+ IngestionConfig refreshConfig = new IngestionConfig();
+ refreshConfig.setBatchIngestionConfig(new BatchIngestionConfig(null,
"REFRESH", null));
+ tableConfig.setIngestionConfig(refreshTable ? refreshConfig : null);
+ tableConfig.getValidationConfig().setRetentionSize(refreshTable ? "100B"
: null);
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 60, now - 1),
+ createSizeRetentionSegment("newest", 60, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = spy(new
ControllerMetrics("sizeRetentionLifecycle.",
+ PinotMetricUtils.getPinotMetricsRegistry()));
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ try {
+ assertFalse(
+ MetricValueUtils.tableGaugeExists(metrics, OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED));
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verifySizeRetentionGaugeRemoved(metrics);
+ verify(resourceManager, never()).deleteSegments(anyString(),
anyList());
+ clearInvocations(metrics);
+ tableConfig.setIngestionConfig(null);
+ tableConfig.getValidationConfig().setRetentionSize("200B");
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 0);
+ assertSizeRetentionGaugeValue(metrics, 0);
+ clearInvocations(metrics);
+ tableConfig.getValidationConfig().setRetentionSize("100B");
+ when(resourceManager.deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest")))
+ .thenReturn(PinotResourceManagerResponse.failure("Targets became
lineage-locked"),
+ PinotResourceManagerResponse.SUCCESS);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest"));
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ assertSizeRetentionGaugeValue(metrics, 1);
+ clearInvocations(metrics);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager, times(2)).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest"));
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 0);
+ assertSizeRetentionGaugeValue(metrics, 0);
+ clearInvocations(metrics, resourceManager);
+ tableConfig.getValidationConfig().setRetentionSize("invalid");
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ assertSizeRetentionGaugeValue(metrics, 1);
+ clearInvocations(metrics);
+ tableConfig.setIngestionConfig(refreshTable ? refreshConfig : null);
+ tableConfig.getValidationConfig().setRetentionSize(refreshTable ?
"100B" : null);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verifySizeRetentionGaugeRemoved(metrics);
+ verify(resourceManager, never()).deleteSegments(anyString(),
anyList());
+ clearInvocations(metrics);
+ tableConfig.setIngestionConfig(null);
+ tableConfig.getValidationConfig().setRetentionSize("200B");
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 0);
+ assertSizeRetentionGaugeValue(metrics, 0);
+ clearInvocations(metrics);
+
+ retentionManager.nonLeaderCleanup(List.of(OFFLINE_TABLE_NAME));
+
+ verifySizeRetentionGaugeRemoved(metrics);
+ } finally {
+ metrics.removeTableGauge(OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED);
+ }
+ }
+ }
+
+ @Test
+ public void
testSizeRetentionPublishesBlockedGaugeOnceWhenMetadataReadThrows() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "100B");
+ List<SegmentZKMetadata> segments =
+ List.of(createSizeRetentionSegment("only", 60,
System.currentTimeMillis()));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+
when(resourceManager.getSegmentsZKMetadata(OFFLINE_TABLE_NAME)).thenThrow(new
IllegalStateException("Read failed"));
+
+ assertThrows(IllegalStateException.class, () ->
retentionManager.manageSizeBasedRetention(tableConfig));
+
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ }
+
+ @Test
+ public void testSizeRetentionForHybridEvictsOldestWithoutOfflineCoverage() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.REALTIME, "20B");
+ long boundary = System.currentTimeMillis();
+ SegmentZKMetadata invalidEndTime =
createSizeRetentionSegment("invalidEndTime", 20, -1);
+ invalidEndTime.setCreationTime(boundary - 2);
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("before", 20, boundary - 1),
+ createSizeRetentionSegment("at", 20, boundary),
createSizeRetentionSegment("after", 20, boundary + 1),
+ createSizeRetentionSegment("newest", 20, boundary + 2),
invalidEndTime);
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ BrokerServiceHelper brokerServiceHelper = mock(BrokerServiceHelper.class);
+ when(brokerServiceHelper.getTimeBoundaryInfo(any(TableConfig.class)))
+ .thenReturn(new TimeBoundaryInfo("ms", Long.toString(boundary)));
+ ControllerConf conf = new ControllerConf();
+ conf.setProperty(ControllerConf.ENABLE_HYBRID_TABLE_RETENTION_STRATEGY,
true);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics,
+ conf, brokerServiceHelper);
+ setupSizeRetentionOfflineCounterpart(resourceManager);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(REALTIME_TABLE_NAME,
List.of("invalidEndTime", "before", "at", "after"));
+ verify(resourceManager, never()).getOfflineTableConfig(anyString());
+ verify(brokerServiceHelper,
never()).getTimeBoundaryInfo(any(TableConfig.class));
+ verifySizeRetentionGauge(metrics, REALTIME_TABLE_NAME, 0);
+ }
+
+ @Test
+ public void testSizeRetentionDoesNotConsultOfflineTimeBoundary() {
+ for (boolean hybridStrategy : List.of(true, false)) {
+ for (boolean offlineCounterpart : List.of(true, false)) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.REALTIME, "20B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 20, now - 1),
+ createSizeRetentionSegment("newest", 20, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ BrokerServiceHelper brokerServiceHelper =
mock(BrokerServiceHelper.class);
+ when(brokerServiceHelper.getTimeBoundaryInfo(any(TableConfig.class)))
+ .thenThrow(new IllegalStateException("Time boundary unavailable"));
+ ControllerConf conf = new ControllerConf();
+
conf.setProperty(ControllerConf.ENABLE_HYBRID_TABLE_RETENTION_STRATEGY,
hybridStrategy);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics,
+ conf, brokerServiceHelper);
+ if (offlineCounterpart) {
+ setupSizeRetentionOfflineCounterpart(resourceManager);
+ }
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(REALTIME_TABLE_NAME,
List.of("oldest"));
+ verify(resourceManager, never()).getOfflineTableConfig(anyString());
+ verify(brokerServiceHelper,
never()).getTimeBoundaryInfo(any(TableConfig.class));
+ verifySizeRetentionGauge(metrics, REALTIME_TABLE_NAME, 0);
+ }
+ }
+ }
+
+ @Test
+ public void testSizeRetentionRechecksMetadataAfterTimeRetention() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "70B");
+ tableConfig.getValidationConfig().setRetentionTimeUnit("DAYS");
+ tableConfig.getValidationConfig().setRetentionTimeValue("365");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments = new ArrayList<>(List.of(
+ createSizeRetentionSegment("expired", 80, now -
TimeUnit.DAYS.toMillis(400)),
+ createSizeRetentionSegment("recent", 60, now - 1),
createSizeRetentionSegment("newest", 30, now)));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager);
+ when(resourceManager.getSegmentsFor(OFFLINE_TABLE_NAME, false)).thenAnswer(
+ invocation ->
segments.stream().map(SegmentZKMetadata::getSegmentName).toList());
+ doAnswer(invocation -> {
+ List<String> deletedSegments = invocation.getArgument(1);
+ segments.removeIf(segment ->
deletedSegments.contains(segment.getSegmentName()));
+ return PinotResourceManagerResponse.SUCCESS;
+ }).when(resourceManager).deleteSegments(eq(OFFLINE_TABLE_NAME), anyList());
+
+ retentionManager.processTable(OFFLINE_TABLE_NAME);
+
+ var deletionOrder = inOrder(resourceManager);
+ deletionOrder.verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("expired"));
+
deletionOrder.verify(resourceManager).getSegmentsZKMetadata(OFFLINE_TABLE_NAME);
+ deletionOrder.verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("recent"));
+
assertEquals(segments.stream().map(SegmentZKMetadata::getSegmentName).toList(),
List.of("newest"));
+ }
+
+ @Test
+ public void testSizeRetentionSkipsTableWhenActiveMetadataIsMissing() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "50B");
+ List<SegmentZKMetadata> segments =
+ List.of(createSizeRetentionSegment("oversized", 100,
System.currentTimeMillis()));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ when(resourceManager.getSegmentsFor(OFFLINE_TABLE_NAME,
false)).thenReturn(List.of("oversized", "missing"));
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+
+ @Test
+ public void testSizeRetentionKeepsUndatedAndEmptySegments() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "50B");
+ long now = System.currentTimeMillis();
+ SegmentZKMetadata undated = createSizeRetentionSegment("undated", 80, -1);
+ undated.setCreationTime(-1);
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("empty", 0, now - 1), undated,
+ createSizeRetentionSegment("oldest", 60, now - 2),
createSizeRetentionSegment("newest", 60, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest"));
+ }
+
+ @Test
+ public void testSizeRetentionSkipsMalformedStoredConfiguration() {
+ for (String retentionSize : List.of("invalid", "0B", "-1B", "")) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, retentionSize);
+ List<SegmentZKMetadata> segments =
+ List.of(createSizeRetentionSegment("oversized", 100,
System.currentTimeMillis()));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ verifySizeRetentionGauge(metrics, OFFLINE_TABLE_NAME, 1);
+ }
+ }
+
+ private TableConfig createSizeRetentionTableConfig(TableType tableType,
String retentionSize) {
+ TableConfig tableConfig = new
TableConfigBuilder(tableType).setTableName(TEST_TABLE_NAME).build();
+ tableConfig.getValidationConfig().setRetentionSize(retentionSize);
+ return tableConfig;
+ }
+
+ private void verifySizeRetentionGauge(ControllerMetrics metrics, String
tableName, long value) {
+ verify(metrics).setOrUpdateTableGauge(tableName,
ControllerGauge.SIZE_RETENTION_BLOCKED, value);
+ // Publishing only the final result prevents an otherwise healthy pass
from exposing a transient blocked value.
+ verify(metrics, times(1)).setOrUpdateTableGauge(eq(tableName),
eq(ControllerGauge.SIZE_RETENTION_BLOCKED),
+ anyLong());
+ }
+
+ private void assertSizeRetentionGaugeValue(ControllerMetrics metrics, long
value) {
+ assertTrue(MetricValueUtils.tableGaugeExists(metrics, OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED));
+ assertEquals(MetricValueUtils.getTableGaugeValue(metrics,
OFFLINE_TABLE_NAME,
+ ControllerGauge.SIZE_RETENTION_BLOCKED), value);
+ }
+
+ private void verifySizeRetentionGaugeRemoved(ControllerMetrics metrics) {
+ verify(metrics).removeTableGauge(OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED);
+ verify(metrics, never()).setOrUpdateTableGauge(eq(OFFLINE_TABLE_NAME),
+ eq(ControllerGauge.SIZE_RETENTION_BLOCKED), anyLong());
+ assertFalse(MetricValueUtils.tableGaugeExists(metrics, OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED));
+ }
+
+ private SegmentZKMetadata createSizeRetentionSegment(String segmentName,
long sizeInBytes, long endTimeMs) {
+ SegmentZKMetadata metadata = new SegmentZKMetadata(segmentName);
+ metadata.setSizeInBytes(sizeInBytes);
+ metadata.setTimeUnit(TimeUnit.MILLISECONDS);
+ metadata.setEndTime(endTimeMs);
+ metadata.setCreationTime(System.currentTimeMillis());
+ return metadata;
+ }
+
+ private RetentionManager createSizeRetentionManager(TableConfig tableConfig,
List<SegmentZKMetadata> segments,
+ PinotHelixResourceManager resourceManager) {
+ return createSizeRetentionManager(tableConfig, segments, resourceManager,
mock(ControllerMetrics.class));
+ }
+
+ private RetentionManager createSizeRetentionManager(TableConfig tableConfig,
List<SegmentZKMetadata> segments,
+ PinotHelixResourceManager resourceManager, ControllerMetrics metrics) {
+ return createSizeRetentionManager(tableConfig, segments, resourceManager,
metrics, new ControllerConf(),
+ mock(BrokerServiceHelper.class));
+ }
+
+ private RetentionManager createSizeRetentionManager(TableConfig tableConfig,
List<SegmentZKMetadata> segments,
+ PinotHelixResourceManager resourceManager, ControllerMetrics metrics,
ControllerConf conf,
+ BrokerServiceHelper brokerServiceHelper) {
+ String tableName = tableConfig.getTableName();
+ when(resourceManager.getTableConfig(tableName)).thenReturn(tableConfig);
+
when(resourceManager.getSegmentsZKMetadata(tableName)).thenReturn(segments);
+ when(resourceManager.getSegmentsFor(tableName, false))
+
.thenReturn(segments.stream().map(SegmentZKMetadata::getSegmentName).toList());
+ when(resourceManager.deleteSegments(eq(tableName),
anyList())).thenReturn(PinotResourceManagerResponse.SUCCESS);
+
when(resourceManager.getPropertyStore()).thenReturn(mock(ZkHelixPropertyStore.class));
+ when(resourceManager.getLineageUpdaterLock(tableName)).thenReturn(new
Object());
+
when(resourceManager.getLastLLCCompletedSegments(anyList())).thenCallRealMethod();
+
when(resourceManager.getLastLLCCompletedSegments(tableName)).thenCallRealMethod();
+ conf.setUntrackedSegmentDeletionEnabled(false);
+ return createRetentionManager(resourceManager,
mock(LeadControllerManager.class), conf,
+ metrics, brokerServiceHelper);
+ }
+
+ private void setupSizeRetentionOfflineCounterpart(PinotHelixResourceManager
resourceManager) {
+ TableConfig offlineConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "1B");
+
when(resourceManager.getOfflineTableConfig(TEST_TABLE_NAME)).thenReturn(offlineConfig);
+ }
+
+ private void setupSizeRetentionLineage(PinotHelixResourceManager
resourceManager, String tableName,
+ String source, String destination, LineageEntryState state) {
+ setupSizeRetentionLineage(resourceManager, tableName, source, destination,
state, System.currentTimeMillis());
+ }
+
+ private void setupSizeRetentionLineage(PinotHelixResourceManager
resourceManager, String tableName,
+ String source, String destination, LineageEntryState state, long
timestamp) {
+ SegmentLineage lineage = new SegmentLineage(tableName);
+ lineage.addLineageEntry("replacement",
+ new LineageEntry(List.of(source), List.of(destination), state,
timestamp));
+ when(resourceManager.getPropertyStore().get(anyString(), any(Stat.class),
eq(AccessOption.PERSISTENT)))
+ .thenReturn(lineage.toZNRecord());
+ }
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
index c77b42930a4..4e76120982b 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
@@ -301,6 +301,7 @@ public final class TableConfigUtils {
/// Validates retention config. Checks for following things:
/// - Valid segmentPushType
/// - Valid retentionTimeUnit
+ /// - Positive retentionSize in bytes, when configured
private static void validateRetentionConfig(TableConfig tableConfig) {
SegmentsValidationAndRetentionConfig segmentsConfig =
tableConfig.getValidationConfig();
String tableName = tableConfig.getTableName();
@@ -329,6 +330,17 @@ public final class TableConfigUtils {
}
}
+ String retentionSize = segmentsConfig.getRetentionSize();
+ if (retentionSize != null) {
+ try {
+ Preconditions.checkArgument(DataSizeUtils.toBytes(retentionSize) > 0,
"Retention size must be positive");
+ } catch (IllegalArgumentException e) {
+ throw new IllegalStateException(
+ String.format("Table: %s, invalid retention size: %s; must be a
positive size in bytes", tableName,
+ retentionSize), e);
+ }
+ }
+
// Untracked segments retention may not be specified. Ignore validation in
that case.
String untrackedSegmentsRetentionTimeUnitString =
segmentsConfig.getUntrackedSegmentsRetentionTimeUnit();
String untrackedSegmentsRetentionTimeValueString =
segmentsConfig.getUntrackedSegmentsRetentionTimeValue();
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
index ae5af3891de..371b7431096 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
@@ -2564,6 +2564,51 @@ public class TableConfigUtilsTest {
}
}
+ @DataProvider
+ public Object[][] retentionTableTypes() {
+ return new Object[][]{{TableType.OFFLINE}, {TableType.REALTIME}};
+ }
+
+ @Test(dataProvider = "retentionTableTypes")
+ public void testValidateRetentionSize(TableType tableType) {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .build();
+ TableConfigBuilder builder = new
TableConfigBuilder(tableType).setTableName(TABLE_NAME)
+ .setTimeColumnName(TIME_COLUMN);
+ if (tableType == TableType.REALTIME) {
+ builder.setStreamConfigs(getStreamConfigs());
+ }
+
+ // Size retention is optional and does not require time retention.
+ TableConfigUtils.validate(builder.build(), schema);
+ for (String retentionSize : List.of("1", "1B", "100G", "1.5gB")) {
+
TableConfigUtils.validate(builder.setRetentionSize(retentionSize).build(),
schema);
+ }
+
+ // The two retention policies can be configured together.
+
TableConfigUtils.validate(builder.setRetentionTimeUnit("DAYS").setRetentionTimeValue("7").build(),
schema);
+ }
+
+ @Test(dataProvider = "retentionTableTypes")
+ public void testRejectInvalidRetentionSize(TableType tableType) {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .build();
+ TableConfigBuilder builder = new
TableConfigBuilder(tableType).setTableName(TABLE_NAME)
+ .setTimeColumnName(TIME_COLUMN);
+ if (tableType == TableType.REALTIME) {
+ builder.setStreamConfigs(getStreamConfigs());
+ }
+
+ for (String retentionSize : List.of("", " ", "0", "0G", "0.5B", "-1",
"-1G", "100GBB", "abc")) {
+ TableConfig tableConfig =
builder.setRetentionSize(retentionSize).build();
+ IllegalStateException e =
+ expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validate(tableConfig, schema));
+ assertTrue(e.getMessage().contains("invalid retention size: " +
retentionSize), e.getMessage());
+ }
+ }
+
@Test
public void testValidateDedupConfig() {
TableConfig validTableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
index 9027fd3ec57..e503c2986f5 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
@@ -27,6 +27,7 @@ import org.apache.pinot.spi.utils.TimeUtils;
public class SegmentsValidationAndRetentionConfig extends BaseJsonConfig {
private String _retentionTimeUnit;
private String _retentionTimeValue;
+ private String _retentionSize;
private String _deletedSegmentsRetentionPeriod;
private String _replacedSegmentsRetentionPeriod;
private String _lineageEntryCleanupRetentionPeriod;
@@ -88,6 +89,59 @@ public class SegmentsValidationAndRetentionConfig extends
BaseJsonConfig {
_retentionTimeValue = retentionTimeValue;
}
+ /// Returns the size retention limit, expressed as a positive size
understood by
+ /// [DataSizeUtils#toBytes(String)], for example `"100G"`.
+ ///
+ /// The limit sums compressed archive bytes for active, live completed
segments, counted once regardless of
+ /// replication. Live-segment lineage filtering excludes sources of
COMPLETED entries and destinations of other
+ /// entries to avoid counting shadow copies. OFFLINE APPEND and REALTIME
limits are enforced independently, including
+ /// hybrid tables. Each typed table expires its own oldest eligible data
without consulting the OFFLINE routing
+ /// time boundary. REALTIME size retention can therefore expire data newer
than that boundary; hybrid time
+ /// retention continues to use its existing boundary rules.
+ /// For hybrid tables, size the REALTIME cap to comfortably retain all data
newer than the routing boundary.
+ /// Otherwise size retention can delete input before RealtimeToOffline
migrates it, and queries for that time
+ /// range can return no rows from either typed table without an error.
+ ///
+ /// Segments are ordered ascending by the first nonnegative end time,
creation time, or push time, in that priority,
+ /// then by segment name. Eviction removes only an old prefix before the
first active segment listed as a source or
+ /// destination in any retained lineage entry. Every listed segment is
protected until its lineage entry is removed,
+ /// regardless of entry state, age, or the lineage-exclusive deletion
setting. Listed shadow copies remain stop
+ /// positions even when their bytes are excluded. Missing live metadata or
unknown live completed-segment sizes cause
+ /// the pass to skip safely. Missing metadata or timestamps for an active
lineage stop prevent eviction when over cap.
+ ///
+ /// With the default lineage manager, COMPLETED sources wait for
[#getReplacedSegmentsRetentionPeriod()] (4h for
+ /// APPEND, 24h for REFRESH) before deletion. The entry is removed on a
subsequent cleanup pass after sources leave
+ /// IdealState. The retention schedule (6h by default) also delays release
of the boundary. Stale IN_PROGRESS entries
+ /// use [#getLineageEntryCleanupRetentionPeriod()] (1d by default); REVERTED
entries are eligible immediately.
+ /// These are cleanup eligibility windows, not deadlines for meeting the
size cap. Continuous merge or rollup
+ /// activity can keep a lineage boundary present and leave an over-cap table
at `sizeRetentionBlocked=1` for hours
+ /// or longer.
+ ///
+ /// The newest OFFLINE segment, highest-sequence DONE LLC segment per
REALTIME partition group, and undated segments
+ /// are preserved. Consuming segments are neither counted nor removed. These
protections can leave the table above
+ /// its cap.
+ ///
+ /// Size retention rechecks the complete lineage entry snapshot under the
local updater lock before deletion,
+ /// regardless of the lineage-exclusive deletion setting. A changed snapshot
aborts the batch for retry next cycle.
+ /// This deliberately includes timestamp-only changes and entries wholly
newer than the boundary, even when the
+ /// selected prefix is unaffected.
+ /// The check and deletion are not a global transaction across controllers.
+ ///
+ /// This is an asynchronous, best-effort retention policy, not a limit on
decompressed server disk usage,
+ /// memory usage, or ingestion. When time retention is also configured, a
segment can be deleted by either criterion.
+ /// The table-scoped `sizeRetentionBlocked` controller gauge is 1 after an
unsafe, failed, or still-over-cap pass,
+ /// 0 after a successful pass, and absent when the policy is disabled or
unsupported.
+ /// A null value disables size retention.
+ public String getRetentionSize() {
+ return _retentionSize;
+ }
+
+ /// Sets the compressed segment archive size retention limit described by
[#getRetentionSize()].
+ /// Set to null to disable size retention.
+ public void setRetentionSize(String retentionSize) {
+ _retentionSize = retentionSize;
+ }
+
public String getDeletedSegmentsRetentionPeriod() {
return _deletedSegmentsRetentionPeriod;
}
@@ -96,12 +150,12 @@ public class SegmentsValidationAndRetentionConfig extends
BaseJsonConfig {
_deletedSegmentsRetentionPeriod = deletedSegmentsRetentionPeriod;
}
- /// Returns the retention period for segments replaced by a REFRESH
ingestion job. Only applies to tables with
- /// REFRESH ingestion type; for APPEND tables this setting is ignored and
replaced segments are deleted immediately.
+ /// Returns the retention period for source segments replaced by a COMPLETED
lineage entry. Applies to all ingestion
+ /// types, including APPEND and REFRESH.
///
/// When a lineage entry transitions to COMPLETED state, source segments are
preserved for this duration before
- /// being scheduled for deletion, providing a rollback window. Consumers of
this config (e.g. the lineage manager)
- /// treat a null or unparseable value as a 1 day default.
+ /// being scheduled for deletion. The default lineage manager uses 4 hours
for APPEND and other ingestion types,
+ /// and 24 hours for REFRESH, when the value is unset, empty, or unparseable.
///
/// Accepts a human-readable period string (e.g. `"7d"`, `"12h"`) as
understood by
/// `TimeUtils.convertPeriodToMillis`. Setting this value too low (e.g.
`"0d"`) eliminates the rollback
@@ -114,9 +168,10 @@ public class SegmentsValidationAndRetentionConfig extends
BaseJsonConfig {
_replacedSegmentsRetentionPeriod = replacedSegmentsRetentionPeriod;
}
- /// Returns the retention period before stale IN_PROGRESS or REVERTED
lineage entries and their destination segments
- /// are cleaned up. Consumers of this config (e.g. the lineage manager)
treat a null or unparseable value as a
- /// 1 day default.
+ /// Returns the retention period before stale IN_PROGRESS lineage entries
and their destination segments become
+ /// eligible for cleanup. The default lineage manager uses 1 day when the
value is unset, empty, or unparseable.
+ /// REVERTED entries are eligible immediately. This period does not apply to
COMPLETED entries: those are removed
+ /// on a cleanup pass after their source segments leave IdealState.
///
/// Accepts a human-readable period string (e.g. `"7d"`, `"12h"`) as
understood by
/// `TimeUtils.convertPeriodToMillis`.
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
index 8587ea04627..f61f433a4da 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
@@ -74,6 +74,7 @@ public class TableConfigBuilder {
private String _timeType;
private String _retentionTimeUnit;
private String _retentionTimeValue;
+ private String _retentionSize;
private String _deletedSegmentsRetentionPeriod =
DEFAULT_DELETED_SEGMENTS_RETENTION_PERIOD;
private String _replacedSegmentsRetentionPeriod;
private String _lineageEntryCleanupRetentionPeriod;
@@ -203,6 +204,13 @@ public class TableConfigBuilder {
return this;
}
+ /// Sets the compressed segment archive size retention limit described by
+ /// [SegmentsValidationAndRetentionConfig#getRetentionSize()].
+ public TableConfigBuilder setRetentionSize(String retentionSize) {
+ _retentionSize = retentionSize;
+ return this;
+ }
+
public TableConfigBuilder setDeletedSegmentsRetentionPeriod(String
deletedSegmentsRetentionPeriod) {
_deletedSegmentsRetentionPeriod = deletedSegmentsRetentionPeriod;
return this;
@@ -522,6 +530,7 @@ public class TableConfigBuilder {
validationConfig.setTimeType(_timeType);
validationConfig.setRetentionTimeUnit(_retentionTimeUnit);
validationConfig.setRetentionTimeValue(_retentionTimeValue);
+ validationConfig.setRetentionSize(_retentionSize);
validationConfig.setDeletedSegmentsRetentionPeriod(_deletedSegmentsRetentionPeriod);
validationConfig.setReplacedSegmentsRetentionPeriod(_replacedSegmentsRetentionPeriod);
validationConfig.setLineageEntryCleanupRetentionPeriod(_lineageEntryCleanupRetentionPeriod);
diff --git
a/pinot-spi/src/test/java/org/apache/pinot/spi/utils/builder/TableConfigBuilderTest.java
b/pinot-spi/src/test/java/org/apache/pinot/spi/utils/builder/TableConfigBuilderTest.java
index 2f89899fba5..6a64aa08b7a 100644
---
a/pinot-spi/src/test/java/org/apache/pinot/spi/utils/builder/TableConfigBuilderTest.java
+++
b/pinot-spi/src/test/java/org/apache/pinot/spi/utils/builder/TableConfigBuilderTest.java
@@ -19,11 +19,18 @@
package org.apache.pinot.spi.utils.builder;
+import java.io.IOException;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.testng.Assert;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
+
/// Tests for the validations in [TableConfigBuilder]
public class TableConfigBuilderTest {
@@ -31,6 +38,40 @@ public class TableConfigBuilderTest {
private static final String TABLE_NAME = "testTable";
private static final String TIME_COLUMN = "timeColumn";
+ @DataProvider
+ public Object[][] tableTypes() {
+ return new Object[][]{{TableType.OFFLINE}, {TableType.REALTIME}};
+ }
+
+ @Test(dataProvider = "tableTypes")
+ public void testRetentionSizeRoundTrip(TableType tableType)
+ throws IOException {
+ TableConfig tableConfig = new
TableConfigBuilder(tableType).setTableName(TABLE_NAME)
+ .setRetentionSize("100G")
+ .setRetentionTimeUnit("DAYS")
+ .setRetentionTimeValue("7")
+ .build();
+ assertEquals(tableConfig.getValidationConfig().getRetentionSize(), "100G");
+
assertEquals(tableConfig.toJsonNode().get("segmentsConfig").get("retentionSize").asText(),
"100G");
+
+ TableConfig roundTrip =
JsonUtils.stringToObject(tableConfig.toJsonString(), TableConfig.class);
+ assertEquals(roundTrip, tableConfig);
+ assertEquals(roundTrip.getValidationConfig().getRetentionSize(), "100G");
+ assertEquals(roundTrip.getValidationConfig().getRetentionTimeUnit(),
"DAYS");
+ assertEquals(roundTrip.getValidationConfig().getRetentionTimeValue(), "7");
+ assertEquals(new
TableConfig(tableConfig).getValidationConfig().getRetentionSize(), "100G");
+ }
+
+ @Test(dataProvider = "tableTypes")
+ public void testRetentionSizeDefaultsToDisabled(TableType tableType)
+ throws IOException {
+ TableConfig tableConfig = new
TableConfigBuilder(tableType).setTableName(TABLE_NAME).build();
+ assertNull(tableConfig.getValidationConfig().getRetentionSize());
+
assertFalse(tableConfig.toJsonNode().get("segmentsConfig").has("retentionSize"));
+ TableConfig roundTrip =
JsonUtils.stringToObject(tableConfig.toJsonString(), TableConfig.class);
+ assertNull(roundTrip.getValidationConfig().getRetentionSize());
+ }
+
@Test
public void testValidateSkipSegmentPreprocessFlag() {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
