This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 999b978f8b6 Add per-table hybrid retention strategy with cluster-level
fallback (#20116)
999b978f8b6 is described below
commit 999b978f8b68157b4e6050aac04e0d2c61c85fb9
Author: Manak Bisht <[email protected]>
AuthorDate: Fri Nov 6 01:36:50 2026 +0530
Add per-table hybrid retention strategy with cluster-level fallback (#20116)
---
.../utils/config/TableConfigSerDeUtilsTest.java | 64 ++++++++++++++++++++++
.../helix/core/retention/RetentionManager.java | 12 +++-
.../helix/core/retention/RetentionManagerTest.java | 60 ++++++++++++++++++++
.../SegmentsValidationAndRetentionConfig.java | 29 ++++++++++
4 files changed, 162 insertions(+), 3 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 32ab13854b3..1c15da00a7e 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
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.common.utils.config;
+import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.collect.Lists;
import java.io.IOException;
@@ -376,6 +377,69 @@ public class TableConfigSerDeUtilsTest {
assertEquals(tunerConfigToCompare.getName(), name);
assertEquals(tunerConfigToCompare.getTunerProperties(), props);
}
+ {
+ // With hybrid table retention strategy
+ String hybridTableRetentionStrategy = "hybridTableRetentionStrategy";
+ for (Enablement hybridTableRetentionStrategyValue : Enablement.values())
{
+ TableConfig tableConfig = tableConfigBuilder.build();
+
tableConfig.getValidationConfig().setHybridTableRetentionStrategy(hybridTableRetentionStrategyValue);
+
+ // Only serialized when not DEFAULT
+ JsonNode validationConfigJson =
tableConfig.toJsonNode().get(TableConfig.VALIDATION_CONFIG_KEY);
+ if (hybridTableRetentionStrategyValue == Enablement.DEFAULT) {
+ assertFalse(validationConfigJson.has(hybridTableRetentionStrategy));
+ } else {
+
assertEquals(validationConfigJson.get(hybridTableRetentionStrategy).asText(),
+ hybridTableRetentionStrategyValue.name());
+ }
+
+ // Serialize then de-serialize
+ TableConfig tableConfigToCompare =
JsonUtils.stringToObject(tableConfig.toJsonString(), TableConfig.class);
+ assertEquals(tableConfigToCompare, tableConfig);
+
assertEquals(tableConfigToCompare.getValidationConfig().getHybridTableRetentionStrategy(),
+ hybridTableRetentionStrategyValue);
+
+ tableConfigToCompare =
TableConfigSerDeUtils.fromZNRecord(TableConfigSerDeUtils.toZNRecord(tableConfig));
+ assertEquals(tableConfigToCompare, tableConfig);
+
assertEquals(tableConfigToCompare.getValidationConfig().getHybridTableRetentionStrategy(),
+ hybridTableRetentionStrategyValue);
+ }
+
+ // DEFAULT and not serialized when unset
+ TableConfig tableConfig = tableConfigBuilder.build();
+
assertEquals(tableConfig.getValidationConfig().getHybridTableRetentionStrategy(),
Enablement.DEFAULT);
+
assertFalse(tableConfig.toJsonNode().get(TableConfig.VALIDATION_CONFIG_KEY).has(hybridTableRetentionStrategy));
+
+ // Explicit null deserializes to DEFAULT
+ ObjectNode tableConfigJson = (ObjectNode) tableConfig.toJsonNode();
+ ((ObjectNode)
tableConfigJson.get(TableConfig.VALIDATION_CONFIG_KEY)).putNull(hybridTableRetentionStrategy);
+ TableConfig tableConfigToCompare =
JsonUtils.jsonNodeToObject(tableConfigJson, TableConfig.class);
+
assertEquals(tableConfigToCompare.getValidationConfig().getHybridTableRetentionStrategy(),
Enablement.DEFAULT);
+ assertEquals(tableConfigToCompare, tableConfig);
+
+ // Omitted field deserializes to DEFAULT
+ TableConfig explicitlySetTableConfig = tableConfigBuilder.build();
+
explicitlySetTableConfig.getValidationConfig().setHybridTableRetentionStrategy(Enablement.ENABLE);
+ tableConfigJson = (ObjectNode) explicitlySetTableConfig.toJsonNode();
+ ObjectNode validationConfigJson = (ObjectNode)
tableConfigJson.get(TableConfig.VALIDATION_CONFIG_KEY);
+ assertTrue(validationConfigJson.has(hybridTableRetentionStrategy));
+ validationConfigJson.remove(hybridTableRetentionStrategy);
+ tableConfigToCompare = JsonUtils.jsonNodeToObject(tableConfigJson,
TableConfig.class);
+
assertEquals(tableConfigToCompare.getValidationConfig().getHybridTableRetentionStrategy(),
Enablement.DEFAULT);
+ assertEquals(tableConfigToCompare, tableConfig);
+
+ // Explicit DEFAULT deserializes to DEFAULT and is not serialized
+ tableConfigJson = (ObjectNode) tableConfig.toJsonNode();
+ ((ObjectNode)
tableConfigJson.get(TableConfig.VALIDATION_CONFIG_KEY)).put(hybridTableRetentionStrategy,
+ Enablement.DEFAULT.name());
+ tableConfigToCompare = JsonUtils.jsonNodeToObject(tableConfigJson,
TableConfig.class);
+
assertEquals(tableConfigToCompare.getValidationConfig().getHybridTableRetentionStrategy(),
Enablement.DEFAULT);
+ assertFalse(
+
tableConfigToCompare.toJsonNode().get(TableConfig.VALIDATION_CONFIG_KEY).has(hybridTableRetentionStrategy));
+ assertEquals(tableConfigToCompare, tableConfig);
+ tableConfigToCompare =
TableConfigSerDeUtils.fromZNRecord(TableConfigSerDeUtils.toZNRecord(tableConfigToCompare));
+ assertEquals(tableConfigToCompare, tableConfig);
+ }
}
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 72d47f8b82e..5b3f4d4d291 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
@@ -68,6 +68,7 @@ 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.Enablement;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
@@ -354,6 +355,10 @@ public class RetentionManager extends
ControllerPeriodicTask<Void> {
// Scan all segment ZK metadata and purge segments if necessary
if (TableNameBuilder.isOfflineTableResource(tableNameWithType)) {
+ if (validationConfig.getHybridTableRetentionStrategy() !=
Enablement.DEFAULT) {
+ LOGGER.warn("hybridTableRetentionStrategy is only applicable to
realtime tables, "
+ + "ignoring it for offline table: {}", tableNameWithType);
+ }
manageRetentionForOfflineTable(tableNameWithType, retentionStrategy,
untrackedSegmentsDeletionBatchSize,
untrackedSegmentsRetentionStrategy);
} else {
@@ -373,11 +378,12 @@ public class RetentionManager extends
ControllerPeriodicTask<Void> {
/// Returns whether the hybrid table retention strategy should be applied to
the given REALTIME table. Subclasses can
/// override this to decide per table.
///
- /// The default returns the value of
[ControllerConf#ENABLE_HYBRID_TABLE_RETENTION_STRATEGY], which can be updated at
- /// runtime through cluster config.
+ /// By default, the table-level `hybridTableRetentionStrategy` takes
precedence; when it is `DEFAULT`, the value of
+ /// [ControllerConf#ENABLE_HYBRID_TABLE_RETENTION_STRATEGY] is used.
protected boolean shouldEnableHybridTableRetentionStrategy(TableConfig
realtimeTableConfig,
TableConfig offlineTableConfig) {
- return _isHybridTableRetentionStrategyEnabled;
+ return
realtimeTableConfig.getValidationConfig().getHybridTableRetentionStrategy()
+ .isEnabled(_isHybridTableRetentionStrategyEnabled);
}
private void manageRetentionForOfflineTable(String offlineTableName,
RetentionStrategy retentionStrategy,
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 9c6feb64638..95e5d84a45e 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
@@ -32,6 +32,7 @@ import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
import org.apache.helix.AccessOption;
import org.apache.helix.HelixAdmin;
@@ -69,11 +70,13 @@ 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.Enablement;
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.DataProvider;
import org.testng.annotations.Test;
import static
org.apache.pinot.controller.helix.core.retention.RetentionManager.DEFAULT_UNTRACKED_SEGMENTS_DELETION_BATCH_SIZE;
@@ -561,6 +564,63 @@ public class RetentionManagerTest {
return retentionManager;
}
+ @DataProvider(name = "hybridTableRetentionStrategyOverrides")
+ public Object[][] hybridTableRetentionStrategyOverrides() {
+ // tableLevel (null means omitted; null and DEFAULT fall back to
clusterLevel), clusterLevel, hasOfflineTable,
+ // expectHybridPath
+ return new Object[][]{
+ {null, false, true, false},
+ {null, true, true, true},
+ {Enablement.DEFAULT, false, true, false},
+ {Enablement.DEFAULT, true, true, true},
+ {Enablement.ENABLE, false, true, true},
+ {Enablement.ENABLE, true, true, true},
+ {Enablement.DISABLE, false, true, false},
+ {Enablement.DISABLE, true, true, false},
+ // Without an OFFLINE counterpart, realtime retention is used
regardless of the table and cluster level configs
+ {null, true, false, false},
+ {Enablement.DEFAULT, true, false, false},
+ {Enablement.ENABLE, false, false, false},
+ {Enablement.ENABLE, true, false, false},
+ };
+ }
+
+ @Test(dataProvider = "hybridTableRetentionStrategyOverrides")
+ public void
testTableLevelHybridRetentionStrategyOverridesClusterConfig(@Nullable
Enablement tableLevel,
+ boolean clusterLevel, boolean hasOfflineTable, boolean expectHybridPath)
{
+ int replicaCount = 1;
+ TableConfig realtimeTableConfig = createRealtimeTableConfig1(replicaCount);
+ if (tableLevel != null) {
+
realtimeTableConfig.getValidationConfig().setHybridTableRetentionStrategy(tableLevel);
+ }
+
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ when(resourceManager.getHelixAdmin()).thenReturn(mock(HelixAdmin.class));
+ if (hasOfflineTable) {
+ 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..47fc81cb7a2 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
@@ -18,8 +18,12 @@
*/
package org.apache.pinot.spi.config.table;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
import org.apache.pinot.spi.config.BaseJsonConfig;
+import org.apache.pinot.spi.utils.Enablement;
import org.apache.pinot.spi.utils.TimeUtils;
@@ -31,6 +35,10 @@ public class SegmentsValidationAndRetentionConfig extends
BaseJsonConfig {
private String _deletedSegmentsRetentionPeriod;
private String _replacedSegmentsRetentionPeriod;
private String _lineageEntryCleanupRetentionPeriod;
+ // Stored as null when DEFAULT so that it is only serialized when explicitly
set to ENABLE or DISABLE.
+ @JsonProperty("hybridTableRetentionStrategy")
+ @Nullable
+ private Enablement _hybridTableRetentionStrategy;
@Deprecated
private String _segmentPushFrequency; // DO NOT REMOVE, this is used in
internal segment generation management
@Deprecated
@@ -184,6 +192,27 @@ public class SegmentsValidationAndRetentionConfig extends
BaseJsonConfig {
_lineageEntryCleanupRetentionPeriod = lineageEntryCleanupRetentionPeriod;
}
+ /// Whether to use the hybrid table retention strategy as the time-based
retention policy for this table. Only
+ /// applicable to realtime tables. DEFAULT falls back to the
controller-level config
+ /// `controller.enable.hybrid.table.retention.strategy`.
+ ///
+ /// When enabled and an OFFLINE counterpart exists, realtime segments are
purged once the OFFLINE time boundary moves
+ /// past them instead of by `retentionTimeValue`. Valid `retentionTimeUnit`
and `retentionTimeValue` are still
+ /// required for time-based retention to run. When disabled, or when there
is no OFFLINE counterpart, ordinary
+ /// realtime time-based retention applies. Size-based retention
([#getRetentionSize()]) runs independently and may
+ /// still delete realtime data newer than the OFFLINE time boundary.
+ // Ignored for serialization so that DEFAULT is omitted, see the field
+ @JsonIgnore
+ public Enablement getHybridTableRetentionStrategy() {
+ return _hybridTableRetentionStrategy == null ? Enablement.DEFAULT :
_hybridTableRetentionStrategy;
+ }
+
+ @JsonProperty("hybridTableRetentionStrategy")
+ public void setHybridTableRetentionStrategy(@Nullable Enablement
hybridTableRetentionStrategy) {
+ _hybridTableRetentionStrategy =
+ hybridTableRetentionStrategy == Enablement.DEFAULT ? null :
hybridTableRetentionStrategy;
+ }
+
/// @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]
