xiangfu0 commented on code in PR #19770:
URL: https://github.com/apache/pinot/pull/19770#discussion_r4212895127
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -122,6 +128,187 @@ 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, and an in-progress replacement
postpones the entire pass. 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);
+ if (segmentLineage != null &&
segmentLineage.getLineageEntries().values().stream()
+ .anyMatch(entry -> entry.getState() == LineageEntryState.IN_PROGRESS))
{
+ LOGGER.warn("In-progress segment lineage for table: {}, skip size
retention", tableNameWithType);
+ return false;
+ }
+ // 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);
+ 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 (!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);
+ 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);
+ }
+ // Use the same lineage snapshot for accounting and candidate protection;
deletion still performs a fresh check.
+ if (_controllerConf.isLineageExclusiveDeleteEnabled() && segmentLineage !=
null) {
+
candidateNames.removeAll(SegmentLineageUtils.getDeleteBlockedSegments(segmentLineage));
Review Comment:
nit (take it or leave it): against the same snapshot this removal is
provably a no-op — `getDeleteBlockedSegments` returns IN_PROGRESS `from∪to`
(but any IN_PROGRESS entry already skipped the whole pass above) plus COMPLETED
`from` (but those were already filtered out of `activeSegments` by
`filterSegmentsBasedOnLineageInPlace`, so they never reach `candidateNames`);
REVERTED locks nothing. Fine to keep as defense-in-depth given a subclass
overriding `getSizeRetentionSegmentSize` might surprise us, but worth either a
word in the comment saying it's belt-and-braces, or dropping it so future
readers don't hunt for the case it covers.
--
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]
