xiangfu0 commented on code in PR #19770:
URL: https://github.com/apache/pinot/pull/19770#discussion_r4214693032
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -122,6 +128,218 @@ protected void processTable(String tableNameWithType) {
// 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; hybrid
realtime eviction requires offline coverage.
+ /// 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.
Review Comment:
Documented in
[0690fde910](https://github.com/apache/pinot/commit/0690fde910a0dfd2b2231f1db429eb7cc5e67453)
and the PR description. This blocking trade-off is intentional. With the
default lineage manager, COMPLETED sources use
`replacedSegmentsRetentionPeriod` (4h for APPEND/other ingestion types, 24h for
REFRESH), and their entry is removed on a later cleanup pass after the sources
leave IdealState. One clarification: the default 1d
`lineageEntryCleanupRetentionPeriod` ages out stale IN_PROGRESS entries; it is
not an additional COMPLETED-entry window. REVERTED entries are eligible
immediately. The retention cadence (6h by default) also delays release of the
boundary. Continuous merge/rollup can therefore keep the table above cap with
`sizeRetentionBlocked=1` for hours or longer; these are cleanup eligibility
windows, not a deadline for meeting the cap. Also corrected the related config
getter documentation to match the implementation.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -122,6 +128,218 @@ protected void processTable(String tableNameWithType) {
// 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; hybrid
realtime eviction requires offline coverage.
+ /// 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;
+ BigInteger retainedBytes = BigInteger.ZERO;
+ 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;
+ }
+ // Avoid overflowing the table total even if individual segment sizes
are valid longs.
+ retainedBytes = retainedBytes.add(BigInteger.valueOf(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;
+ }
+ BigInteger limit = BigInteger.valueOf(retentionSizeBytes);
+ if (retainedBytes.compareTo(limit) <= 0) {
+ 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);
+ List<String> candidateNames = completedSegments.stream()
+ .filter(metadata -> getSizeRetentionTimestamp(metadata) >= 0)
+ .map(SegmentZKMetadata::getSegmentName)
+ .collect(Collectors.toCollection(ArrayList::new));
+ if (realtime) {
+
candidateNames.removeAll(_pinotHelixResourceManager.getLastLLCCompletedSegments(metadataList));
+ if (_isHybridTableRetentionStrategyEnabled) {
+ TableConfig offlineTableConfig =
+
_pinotHelixResourceManager.getOfflineTableConfig(TableNameBuilder.extractRawTableName(tableNameWithType));
+ if (offlineTableConfig != null) {
+ long timeBoundaryMs;
+ try {
+ timeBoundaryMs = getHybridTimeBoundaryMs(offlineTableConfig);
+ } catch (Exception e) {
+ LOGGER.warn("Cannot determine offline coverage for table: {}, skip
size retention", tableNameWithType, e);
+ return false;
+ }
+ Set<String> coveredSegments = completedSegments.stream()
+ .filter(metadata -> metadata.getEndTimeMs() >= 0 &&
metadata.getEndTimeMs() < timeBoundaryMs)
+
.map(SegmentZKMetadata::getSegmentName).collect(Collectors.toSet());
+ candidateNames.retainAll(coveredSegments);
+ }
+ }
+ } else {
+ // A size typo must not remove the entire OFFLINE table, even if its
newest segment exceeds the cap alone.
+ completedSegments.stream().filter(metadata ->
getSizeRetentionTimestamp(metadata) >= 0)
+
.max(evictionOrder).map(SegmentZKMetadata::getSegmentName).ifPresent(candidateNames::remove);
+ }
+ Set<String> candidates = new HashSet<>(candidateNames);
+ completedSegments.removeIf(metadata ->
!candidates.contains(metadata.getSegmentName()));
+ completedSegments.sort(evictionOrder);
+
+ List<String> segmentsToDelete = new ArrayList<>();
+ BigInteger projectedRetainedBytes = retainedBytes;
+ for (SegmentZKMetadata metadata : completedSegments) {
+ if (projectedRetainedBytes.compareTo(limit) <= 0) {
+ 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 =
projectedRetainedBytes.subtract(BigInteger.valueOf(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)) {
Review Comment:
Confirmed: full-map comparison is deliberately conservative. Changes to an
entry wholly newer than the boundary, or a timestamp-only change, abort the
entire planned prefix for retry on the next pass, with the existing WARN and
blocked gauge.
[0690fde910](https://github.com/apache/pinot/commit/0690fde910a0dfd2b2231f1db429eb7cc5e67453)
makes that choice explicit in the config Javadoc and PR description. The
comparison remains under the local lineage updater lock; the existing
regression tests cover changed membership, state/timestamp changes, read
failures, and lock ownership. The 83-test rebase validation passed.
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/util/ControllerZkHelixUtilsTest.java:
##########
@@ -36,6 +37,12 @@
public class ControllerZkHelixUtilsTest {
+ @AfterMethod(alwaysRun = true)
+ public void resetControllerJobLimits() {
Review Comment:
Addressed by the merge of
[#19773](https://github.com/apache/pinot/pull/19773) and the rebase in
[a456352558](https://github.com/apache/pinot/commit/a45635255840af43692e185af6ef5283783f200d).
The duplicate test-isolation commit was dropped, and
`ControllerZkHelixUtilsTest` is no longer in this PR's diff. The PR description
identifies the upstream fix. After the rebase, all 83 targeted tests passed,
and the ordered ZK-utils/rebalance sequence passed all nine tests in one JVM.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
