This is an automated email from the ASF dual-hosted git repository.
KKcorps 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 20cac709644 Add per-table hybrid retention strategy with cluster-level
fallback (#20057)
20cac709644 is described below
commit 20cac7096447a33b809bf5a317b082a09dfbdd51
Author: Manak Bisht <[email protected]>
AuthorDate: Fri Oct 30 13:44:45 2026 +0530
Add per-table hybrid retention strategy with cluster-level fallback (#20057)
Tables can now set their own hybrid retention strategy
(`segmentsConfig.enableHybridTableRetentionStrategy`). If unset (default), the
cluster-level config applies, which defaults to false.
---
.../utils/config/TableConfigSerDeUtilsTest.java | 28 ++++++++++++++
.../helix/core/retention/RetentionManager.java | 15 +++++++-
.../helix/core/retention/RetentionManagerTest.java | 44 ++++++++++++++++++++++
.../SegmentsValidationAndRetentionConfig.java | 11 ++++++
4 files changed, 96 insertions(+), 2 deletions(-)
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeUtilsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeUtilsTest.java
index 4aaa37e4363..5118b4f5936 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeUtilsTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeUtilsTest.java
@@ -374,6 +374,34 @@ public class TableConfigSerDeUtilsTest {
assertEquals(tunerConfigToCompare.getName(), name);
assertEquals(tunerConfigToCompare.getTunerProperties(), props);
}
+ {
+ // With hybrid table retention strategy
+ for (Boolean enableHybridTableRetentionStrategy : new Boolean[]{true,
false}) {
+ TableConfig tableConfig = tableConfigBuilder.build();
+
tableConfig.getValidationConfig().setEnableHybridTableRetentionStrategy(enableHybridTableRetentionStrategy);
+
+ // Serialize then de-serialize
+ TableConfig tableConfigToCompare =
JsonUtils.stringToObject(tableConfig.toJsonString(), TableConfig.class);
+ assertEquals(tableConfigToCompare, tableConfig);
+
assertEquals(tableConfigToCompare.getValidationConfig().getEnableHybridTableRetentionStrategy(),
+ enableHybridTableRetentionStrategy);
+
+ tableConfigToCompare =
TableConfigSerDeUtils.fromZNRecord(TableConfigSerDeUtils.toZNRecord(tableConfig));
+ assertEquals(tableConfigToCompare, tableConfig);
+
assertEquals(tableConfigToCompare.getValidationConfig().getEnableHybridTableRetentionStrategy(),
+ enableHybridTableRetentionStrategy);
+ }
+
+ // Unset by default
+ TableConfig tableConfig = tableConfigBuilder.build();
+
assertNull(tableConfig.getValidationConfig().getEnableHybridTableRetentionStrategy());
+ ObjectNode validationConfigJson = (ObjectNode)
tableConfig.toJsonNode().get(TableConfig.VALIDATION_CONFIG_KEY);
+
assertFalse(validationConfigJson.has("enableHybridTableRetentionStrategy"));
+
+ // Configs without this field deserialize to null
+ TableConfig tableConfigToCompare =
JsonUtils.stringToObject(tableConfig.toJsonString(), TableConfig.class);
+
assertNull(tableConfigToCompare.getValidationConfig().getEnableHybridTableRetentionStrategy());
+ }
}
private void checkSegmentsValidationAndRetentionConfig(TableConfig
tableConfig) {
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 e2a65a92098..c4a7c490b9e 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
@@ -28,6 +28,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -354,13 +355,22 @@ public class RetentionManager extends
ControllerPeriodicTask<Void> {
// Scan all segment ZK metadata and purge segments if necessary
if (TableNameBuilder.isOfflineTableResource(tableNameWithType)) {
+ if (validationConfig.getEnableHybridTableRetentionStrategy() != null) {
+ LOGGER.warn(
+ "enableHybridTableRetentionStrategy is only applicable to realtime
tables,"
+ + "ignoring it for offline table: {}",
+ tableNameWithType);
+ }
manageRetentionForOfflineTable(tableNameWithType, retentionStrategy,
untrackedSegmentsDeletionBatchSize,
untrackedSegmentsRetentionStrategy);
} else {
String rawTableName =
TableNameBuilder.extractRawTableName(tableNameWithType);
TableConfig offlineTableConfig =
_pinotHelixResourceManager.getOfflineTableConfig(rawTableName);
// hybrid table check should be performed before the realtime table
check.
- if (_isHybridTableRetentionStrategyEnabled && offlineTableConfig !=
null) {
+ boolean hybridTableRetentionStrategyEnabled =
+
Objects.requireNonNullElse(validationConfig.getEnableHybridTableRetentionStrategy(),
+ _isHybridTableRetentionStrategyEnabled);
+ if (hybridTableRetentionStrategyEnabled && offlineTableConfig != null) {
// TODO: handle the orphan segment deletion for hybrid table
manageRetentionForHybridTable(tableConfig, offlineTableConfig);
} else {
@@ -392,7 +402,8 @@ public class RetentionManager extends
ControllerPeriodicTask<Void> {
}
}
- private void manageRetentionForRealtimeTable(String realtimeTableName,
RetentionStrategy retentionStrategy,
+ @VisibleForTesting
+ void manageRetentionForRealtimeTable(String realtimeTableName,
RetentionStrategy retentionStrategy,
int untrackedSegmentsDeletionBatchSize, RetentionStrategy
untrackedSegmentsRetentionStrategy) {
List<SegmentZKMetadata> segmentZKMetadataList =
_pinotHelixResourceManager.getSegmentsZKMetadata(realtimeTableName);
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 a442d598587..101969978a7 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
@@ -74,6 +74,7 @@ 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.DataProvider;
import org.testng.annotations.Test;
import static
org.apache.pinot.controller.helix.core.retention.RetentionManager.DEFAULT_UNTRACKED_SEGMENTS_DELETION_BATCH_SIZE;
@@ -515,6 +516,49 @@ public class RetentionManagerTest {
verify(mockPinotHelixResourceManager,
times(1)).deleteSegments(eq(realtimeTableName), anyList());
}
+ @DataProvider(name = "hybridTableRetentionStrategyOverrides")
+ public Boolean[][] hybridTableRetentionStrategyOverrides() {
+ // tableLevel (null falls back to clusterLevel), clusterLevel,
expectHybridPath
+ return new Boolean[][]{
+ {null, false, false},
+ {null, true, true},
+ {true, false, true},
+ {false, true, false},
+ };
+ }
+
+ @Test(dataProvider = "hybridTableRetentionStrategyOverrides")
+ public void
testTableLevelHybridRetentionStrategyOverridesClusterConfig(Boolean tableLevel,
boolean clusterLevel,
+ boolean expectHybridPath) {
+ final int replicaCount = 1;
+ TableConfig realtimeTableConfig = createRealtimeTableConfig1(replicaCount);
+
realtimeTableConfig.getValidationConfig().setEnableHybridTableRetentionStrategy(tableLevel);
+
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ when(resourceManager.getHelixAdmin()).thenReturn(mock(HelixAdmin.class));
+ setupSizeRetentionOfflineCounterpart(resourceManager);
+
+ ControllerConf conf = new ControllerConf();
+ conf.setProperty(ControllerConf.ENABLE_HYBRID_TABLE_RETENTION_STRATEGY,
clusterLevel);
+
+ RetentionManager retentionManager =
spy(createSizeRetentionManager(realtimeTableConfig, List.of(), resourceManager,
+ mock(ControllerMetrics.class), conf, mock(BrokerServiceHelper.class)));
+
+ // Only the retention strategy selection is under test, so stub out both
branches
+ doNothing().when(retentionManager).manageRetentionForHybridTable(any(),
any());
+ doNothing().when(retentionManager).manageRetentionForRealtimeTable(any(),
any(), anyInt(), any());
+
+ retentionManager.processTable(REALTIME_TABLE_NAME);
+
+ if (expectHybridPath) {
+ verify(retentionManager).manageRetentionForHybridTable(any(), any());
+ verify(retentionManager, never()).manageRetentionForRealtimeTable(any(),
any(), anyInt(), any());
+ } else {
+ verify(retentionManager, never()).manageRetentionForHybridTable(any(),
any());
+ verify(retentionManager).manageRetentionForRealtimeTable(any(), any(),
anyInt(), any());
+ }
+ }
+
private TableConfig createOfflineTableConfig() {
return new
TableConfigBuilder(TableType.OFFLINE).setTableName(TEST_TABLE_NAME).setRetentionTimeUnit("DAYS")
.setRetentionTimeValue("365").setNumReplicas(2).build();
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 ae12ddae980..4a165cf50e3 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
@@ -19,6 +19,7 @@
package org.apache.pinot.spi.config.table;
import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
import org.apache.pinot.spi.config.BaseJsonConfig;
import org.apache.pinot.spi.utils.TimeUtils;
@@ -31,6 +32,7 @@ public class SegmentsValidationAndRetentionConfig extends
BaseJsonConfig {
private String _deletedSegmentsRetentionPeriod;
private String _replacedSegmentsRetentionPeriod;
private String _lineageEntryCleanupRetentionPeriod;
+ private Boolean _enableHybridTableRetentionStrategy;
@Deprecated
private String _segmentPushFrequency; // DO NOT REMOVE, this is used in
internal segment generation management
@Deprecated
@@ -184,6 +186,15 @@ public class SegmentsValidationAndRetentionConfig extends
BaseJsonConfig {
_lineageEntryCleanupRetentionPeriod = lineageEntryCleanupRetentionPeriod;
}
+ @Nullable
+ public Boolean getEnableHybridTableRetentionStrategy() {
+ return _enableHybridTableRetentionStrategy;
+ }
+
+ public void setEnableHybridTableRetentionStrategy(Boolean
enableHybridTableRetentionStrategy) {
+ _enableHybridTableRetentionStrategy = enableHybridTableRetentionStrategy;
+ }
+
/// @deprecated Use `segmentIngestionFrequency` from
///
[org.apache.pinot.spi.config.table.ingestion.IngestionConfig#getBatchIngestionConfig()]
@Deprecated(since = "0.8.0", forRemoval = true)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
