xiangfu0 commented on code in PR #19770:
URL: https://github.com/apache/pinot/pull/19770#discussion_r4201118148
##########
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:
`deleteSegments` re-checks lineage and rejects the **whole batch** with a
failure `PinotResourceManagerResponse` if any single target became
lineage-locked after the candidate selection above
(`rejectIfTargetsLineageLockedSegments`). Since the response is dropped here,
in that race: nothing is deleted, the INFO log above has already claimed the
deletion, and the final WARN stays silent because the already-decremented
`retainedBytes` says we're under the limit.
It self-heals on the next run, but worth checking `isSuccessful()` and
logging the failure so operators can tell eviction didn't actually happen.
##########
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:
`getSegmentsFor(tableNameWithType, false)` includes replaced segments
(`segmentsFrom` of COMPLETED lineage entries still within
`replacedSegmentsRetentionPeriod`). Those segments are delete-blocked by
`SegmentLineageUtils.getDeleteBlockedSegments`, so they count toward
`retainedBytes` but can never be evicted here — meaning live, queryable
segments get deleted to make room for bytes that lineage cleanup is going to
reclaim shortly anyway.
Similarly, while a merge/rollup/replace is IN_PROGRESS, both `segmentsFrom`
and `segmentsTo` are in the ideal state, so the same data is counted twice; the
inflated total can push the table over the limit and evict unrelated live
segments (`testSizeRetentionProtectsBothSidesOfInProgressLineage` encodes
exactly this: `unlocked` — the *newest* segment — is deleted because protected
bytes fill the cap).
I see the javadoc makes "protected bytes still count" deliberate as a raw
storage cap, but the failure mode is permanent deletion of live data triggered
by transient lineage state. Consider `shouldExcludeReplacedSegments=true`
and/or excluding delete-blocked segments from the *accounting* (not just the
victim list) — or skipping the size pass while an IN_PROGRESS lineage entry
exists.
##########
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:
This loop will delete every completed segment in the table — including the
newest — when `retentionSize` is smaller than the newest segment alone
(`testSizeRetentionDeletesNewestSegmentWhenItAloneExceedsLimit` asserts this).
A fat-fingered config (`100M` instead of `100G`) wipes the whole OFFLINE table
on the next retention run. REALTIME gets some floor via the last-LLC-completed
exclusion; consider always retaining the newest completed segment for OFFLINE
too (mirroring that protection), or at least documenting the wipe behavior
prominently.
##########
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:
Hybrid-table interaction: when the hybrid retention strategy is enabled,
time retention only deletes REALTIME segments below the offline time boundary
(data already served from OFFLINE). Size retention has no such guard — it can
evict completed REALTIME segments *newer* than the time boundary, i.e. data
`RealtimeToOfflineSegmentsTask` hasn't moved to the OFFLINE table yet, which is
then lost rather than aged out. If intended, worth stating in the
`retentionSize` javadoc; otherwise consider bounding realtime size-eviction by
the offline time boundary when an offline counterpart exists.
##########
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:
nit: this "configured but not enforceable" condition (and the unknown-size
skip above, which will fire every run for tables with pre-`sizeInBytes`
segments) is exactly what operators would want to alert on, but it's only
visible as a per-run WARN. Consider a controller gauge (à la
`RETENTION_MANAGER_ERROR`) keyed by table so it's monitorable without log
scraping.
--
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]
