xiangfu0 commented on code in PR #19770:
URL: https://github.com/apache/pinot/pull/19770#discussion_r4201516419
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -122,6 +126,123 @@ 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 independently of
time retention. The limit counts one
+ /// copy of every active completed segment, including segments protected by
lineage or realtime recovery. Unknown
+ /// sizes or missing metadata disable this pass; consuming segments are
excluded from both accounting and eviction.
+ @VisibleForTesting
+ protected void manageSizeBasedRetention(TableConfig tableConfig) {
+ String retentionSize =
tableConfig.getValidationConfig().getRetentionSize();
+ if (retentionSize == null) {
+ return;
+ }
+ String tableNameWithType = tableConfig.getTableName();
+ if (tableConfig.getTableType() == TableType.OFFLINE
+ &&
!"APPEND".equalsIgnoreCase(IngestionConfigUtils.getBatchSegmentIngestionType(tableConfig)))
{
+ return;
+ }
+ 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;
+ }
+
+ Set<String> activeSegments = new
HashSet<>(_pinotHelixResourceManager.getSegmentsFor(tableNameWithType, false));
Review Comment:
Fixed in
[2e170fcecf](https://github.com/apache/pinot/commit/2e170fcecfd32620a6ae093e7d5ca705571452f1).
Size accounting now calls `getSegmentsFor(tableNameWithType, true)`, excluding
COMPLETED replacements awaiting lineage cleanup. Any IN_PROGRESS lineage entry
postpones the entire size pass, regardless of the lineage-exclusive deletion
setting. Regression tests verify that replaced sources (including sources with
unknown size) do not inflate the cap and that in-progress replacement cannot
evict unrelated eligible live data.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -122,6 +126,123 @@ 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 independently of
time retention. The limit counts one
+ /// copy of every active completed segment, including segments protected by
lineage or realtime recovery. Unknown
+ /// sizes or missing metadata disable this pass; consuming segments are
excluded from both accounting and eviction.
+ @VisibleForTesting
+ protected void manageSizeBasedRetention(TableConfig tableConfig) {
+ String retentionSize =
tableConfig.getValidationConfig().getRetentionSize();
+ if (retentionSize == null) {
+ return;
+ }
+ String tableNameWithType = tableConfig.getTableName();
+ if (tableConfig.getTableType() == TableType.OFFLINE
+ &&
!"APPEND".equalsIgnoreCase(IngestionConfigUtils.getBatchSegmentIngestionType(tableConfig)))
{
+ return;
+ }
+ 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;
+ }
+
+ Set<String> activeSegments = new
HashSet<>(_pinotHelixResourceManager.getSegmentsFor(tableNameWithType, false));
+ if (activeSegments.isEmpty()) {
+ return;
+ }
+ 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;
+ }
+ // 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;
+ }
+ BigInteger limit = BigInteger.valueOf(retentionSizeBytes);
+ if (retainedBytes.compareTo(limit) <= 0) {
+ return;
+ }
+
+ 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));
+ }
+ // Protect lineage before selecting victims: protected bytes still count
toward the limit.
+ removeLineageLockedSegments(tableNameWithType, candidateNames);
+ Set<String> candidates = new HashSet<>(candidateNames);
+ completedSegments.removeIf(metadata ->
!candidates.contains(metadata.getSegmentName()));
+
completedSegments.sort(Comparator.comparingLong(RetentionManager::getSizeRetentionTimestamp)
+ .thenComparing(SegmentZKMetadata::getSegmentName));
+
+ List<String> segmentsToDelete = new ArrayList<>();
+ for (SegmentZKMetadata metadata : completedSegments) {
+ if (retainedBytes.compareTo(limit) <= 0) {
+ break;
+ }
+ long size = completedSegmentSizes.get(metadata.getSegmentName());
+ if (size == 0) {
+ continue;
+ }
+ segmentsToDelete.add(metadata.getSegmentName());
+ retainedBytes = retainedBytes.subtract(BigInteger.valueOf(size));
+ }
+ if (!segmentsToDelete.isEmpty()) {
+ LOGGER.info("Deleting {} oldest segments from table: {} for retention
size: {}", segmentsToDelete.size(),
+ tableNameWithType, retentionSize);
+ // The public deletion path rechecks live lineage in case it changed
after candidate selection.
+ _pinotHelixResourceManager.deleteSegments(tableNameWithType,
segmentsToDelete);
Review Comment:
Fixed in
[2e170fcecf](https://github.com/apache/pinot/commit/2e170fcecfd32620a6ae093e7d5ca705571452f1).
The size pass checks `PinotResourceManagerResponse.isSuccessful()` before
accepting projected reductions or logging successful deletion. A rejected or
null batch response logs the failure with the original retained bytes and cap,
leaves `SIZE_RETENTION_BLOCKED` at 1, and retries on the next cycle. Regression
tests cover lineage rejection/null response and successful retry clearing the
gauge.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -122,6 +126,123 @@ 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 independently of
time retention. The limit counts one
+ /// copy of every active completed segment, including segments protected by
lineage or realtime recovery. Unknown
+ /// sizes or missing metadata disable this pass; consuming segments are
excluded from both accounting and eviction.
+ @VisibleForTesting
+ protected void manageSizeBasedRetention(TableConfig tableConfig) {
+ String retentionSize =
tableConfig.getValidationConfig().getRetentionSize();
+ if (retentionSize == null) {
+ return;
+ }
+ String tableNameWithType = tableConfig.getTableName();
+ if (tableConfig.getTableType() == TableType.OFFLINE
+ &&
!"APPEND".equalsIgnoreCase(IngestionConfigUtils.getBatchSegmentIngestionType(tableConfig)))
{
+ return;
+ }
+ 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;
+ }
+
+ Set<String> activeSegments = new
HashSet<>(_pinotHelixResourceManager.getSegmentsFor(tableNameWithType, false));
+ if (activeSegments.isEmpty()) {
+ return;
+ }
+ 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;
+ }
+ // 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;
+ }
+ BigInteger limit = BigInteger.valueOf(retentionSizeBytes);
+ if (retainedBytes.compareTo(limit) <= 0) {
+ return;
+ }
+
+ 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));
+ }
+ // Protect lineage before selecting victims: protected bytes still count
toward the limit.
+ removeLineageLockedSegments(tableNameWithType, candidateNames);
+ Set<String> candidates = new HashSet<>(candidateNames);
+ completedSegments.removeIf(metadata ->
!candidates.contains(metadata.getSegmentName()));
+
completedSegments.sort(Comparator.comparingLong(RetentionManager::getSizeRetentionTimestamp)
+ .thenComparing(SegmentZKMetadata::getSegmentName));
+
+ List<String> segmentsToDelete = new ArrayList<>();
Review Comment:
Fixed in
[2e170fcecf](https://github.com/apache/pinot/commit/2e170fcecfd32620a6ae093e7d5ca705571452f1).
OFFLINE size retention always preserves the newest dated segment, defined by
the same end-time/creation-time/push-time fallback and segment-name tie-break
used for eviction. This includes a single segment larger than the cap. If this
floor prevents reaching the cap, retention warns and keeps the blocked gauge at
1. Tests cover an oversized newest segment, the only remaining segment,
timestamp fallbacks, and deterministic ties. The safeguard and best-effort
limit are explicit in the configuration Javadoc and PR description.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -122,6 +126,123 @@ 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 independently of
time retention. The limit counts one
Review Comment:
Fixed in
[2e170fcecf](https://github.com/apache/pinot/commit/2e170fcecfd32620a6ae093e7d5ca705571452f1).
When hybrid retention is enabled and an OFFLINE counterpart exists, REALTIME
size eviction requires a known segment end time strictly below the OFFLINE time
boundary. Segments at/above the boundary and segments with unknown end time are
preserved; an unavailable or invalid boundary postpones eviction. Both
retention paths reuse the existing boundary resolver. Tests cover strict
boundary equality, newer/unknown end times, missing/invalid boundaries, and
ordinary behavior when the strategy is disabled or there is no OFFLINE
counterpart.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -122,6 +126,123 @@ 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 independently of
time retention. The limit counts one
+ /// copy of every active completed segment, including segments protected by
lineage or realtime recovery. Unknown
+ /// sizes or missing metadata disable this pass; consuming segments are
excluded from both accounting and eviction.
+ @VisibleForTesting
+ protected void manageSizeBasedRetention(TableConfig tableConfig) {
+ String retentionSize =
tableConfig.getValidationConfig().getRetentionSize();
+ if (retentionSize == null) {
+ return;
+ }
+ String tableNameWithType = tableConfig.getTableName();
+ if (tableConfig.getTableType() == TableType.OFFLINE
+ &&
!"APPEND".equalsIgnoreCase(IngestionConfigUtils.getBatchSegmentIngestionType(tableConfig)))
{
+ return;
+ }
+ 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;
+ }
+
+ Set<String> activeSegments = new
HashSet<>(_pinotHelixResourceManager.getSegmentsFor(tableNameWithType, false));
+ if (activeSegments.isEmpty()) {
+ return;
+ }
+ 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;
+ }
+ // 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;
+ }
+ BigInteger limit = BigInteger.valueOf(retentionSizeBytes);
+ if (retainedBytes.compareTo(limit) <= 0) {
+ return;
+ }
+
+ 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));
+ }
+ // Protect lineage before selecting victims: protected bytes still count
toward the limit.
+ removeLineageLockedSegments(tableNameWithType, candidateNames);
+ Set<String> candidates = new HashSet<>(candidateNames);
+ completedSegments.removeIf(metadata ->
!candidates.contains(metadata.getSegmentName()));
+
completedSegments.sort(Comparator.comparingLong(RetentionManager::getSizeRetentionTimestamp)
+ .thenComparing(SegmentZKMetadata::getSegmentName));
+
+ List<String> segmentsToDelete = new ArrayList<>();
+ for (SegmentZKMetadata metadata : completedSegments) {
+ if (retainedBytes.compareTo(limit) <= 0) {
+ break;
+ }
+ long size = completedSegmentSizes.get(metadata.getSegmentName());
+ if (size == 0) {
+ continue;
+ }
+ segmentsToDelete.add(metadata.getSegmentName());
+ retainedBytes = retainedBytes.subtract(BigInteger.valueOf(size));
+ }
+ if (!segmentsToDelete.isEmpty()) {
+ LOGGER.info("Deleting {} oldest segments from table: {} for retention
size: {}", segmentsToDelete.size(),
+ tableNameWithType, retentionSize);
+ // The public deletion path rechecks live lineage in case it changed
after candidate selection.
+ _pinotHelixResourceManager.deleteSegments(tableNameWithType,
segmentsToDelete);
+ }
+ if (retainedBytes.compareTo(limit) > 0) {
+ LOGGER.warn("Table: {} remains above retention size: {} due to protected
or undated segments",
Review Comment:
Fixed in
[2e170fcecf](https://github.com/apache/pinot/commit/2e170fcecfd32620a6ae093e7d5ca705571452f1).
Added table-scoped `sizeRetentionBlocked`
(`ControllerGauge.SIZE_RETENTION_BLOCKED`): 1 for unsafe skips, failed
deletion, or an unmet cap; 0 for a disabled/unsupported policy, an
empty/within-cap table, or successful eviction within the cap. The pass
initializes conservatively so read exceptions cannot leave a stale healthy
status, clears on recovery, and removes the gauge when table leadership is lost
or the table is deleted. Regression tests cover blocked states, unknown-size
recovery, deletion retry, disabling the policy, empty tables, and leadership
cleanup.
--
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]
