You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "rajagopr (via GitHub)" <gi...@apache.org> on 2025/03/12 18:54:32 UTC
[PR] Improved retention management for hybrid tables [pinot]
rajagopr opened a new pull request, #15257:
URL: https://github.com/apache/pinot/pull/15257
## Problem
Previously, RetentionManager would delete segments past retention period from a REALTIME table when the REALTIME table is part of a hybrid table setup. At times this resulted in prematurely deleting segments before the RTO task got a chance to move the segments to the OFFLINE table which resulted in data loss.
## Solution
The changes in this PR makes the RetentionManager hybrid table aware. RetentionManager behavior is enhanced such that segments from a hybrid REALTIME table are eligible for deletion when the segment is older than the hybrid table time boundary.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r1999407556
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -118,11 +127,20 @@ private void manageRetentionForTable(TableConfig tableConfig) {
// Scan all segment ZK metadata and purge segments if necessary
if (TableNameBuilder.isOfflineTableResource(tableNameWithType)) {
manageRetentionForOfflineTable(tableNameWithType, retentionStrategy);
+ } else if (isHybridTable(tableConfig)) {
Review Comment:
Introduced the feature flag named `controller.enable.hybrid.table.retention.strategy`
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.
swaminathanmanish commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r1999172245
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -118,11 +127,20 @@ private void manageRetentionForTable(TableConfig tableConfig) {
// Scan all segment ZK metadata and purge segments if necessary
if (TableNameBuilder.isOfflineTableResource(tableNameWithType)) {
manageRetentionForOfflineTable(tableNameWithType, retentionStrategy);
+ } else if (isHybridTable(tableConfig)) {
Review Comment:
Can we add a feature flag/param (timeBoundaryBasedRetention) to opt-in so that we don't change the default behavior?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r2002064702
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -165,6 +187,95 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig) {
+ LOGGER.info("Managing retention for hybrid table: {}", realtimeTableConfig.getTableName());
+ List<String> segmentsToDelete = new ArrayList<>();
+ String realtimeTableName = realtimeTableConfig.getTableName();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ try {
+ TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(offlineTableName);
Review Comment:
Let's pass in both realtime and offline table config to reduce a ZK access
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -119,10 +129,22 @@ private void manageRetentionForTable(TableConfig tableConfig) {
if (TableNameBuilder.isOfflineTableResource(tableNameWithType)) {
manageRetentionForOfflineTable(tableNameWithType, retentionStrategy);
} else {
- manageRetentionForRealtimeTable(tableNameWithType, retentionStrategy);
+ // hybrid table check should be performed before the realtime table check.
+ if (_isHybridTableRetentionStrategyEnabled && isHybridTable(tableConfig)) {
+ manageRetentionForHybridTable(tableConfig);
+ } else {
+ manageRetentionForRealtimeTable(tableNameWithType, retentionStrategy);
+ }
}
}
+ private boolean isHybridTable(TableConfig tableConfig) {
+ String tableName = TableNameBuilder.extractRawTableName(tableConfig.getTableName());
+ TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(tableName);
+ TableConfig realtimeTableConfig = _pinotHelixResourceManager.getRealtimeTableConfig(tableName);
Review Comment:
(minor) No need to check read realtime config again
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -165,6 +187,95 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig) {
+ LOGGER.info("Managing retention for hybrid table: {}", realtimeTableConfig.getTableName());
+ List<String> segmentsToDelete = new ArrayList<>();
+ String realtimeTableName = realtimeTableConfig.getTableName();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ try {
+ TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(offlineTableName);
+ ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore();
+ Schema schema = ZKMetadataProvider.getTableSchema(propertyStore, offlineTableName);
+ if (schema == null) {
+ throw new RuntimeException("Failed to get schema for table: " + offlineTableName);
+ }
+ String timeColumn = null;
+ SegmentsValidationAndRetentionConfig validationConfig = offlineTableConfig.getValidationConfig();
+ if (validationConfig != null) {
+ timeColumn = validationConfig.getTimeColumnName();
+ }
+ if (StringUtils.isEmpty(timeColumn)) {
+ throw new RuntimeException("TimeColumn is null or empty for table: " + offlineTableName);
+ }
+ DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(timeColumn);
+ if (dateTimeSpec == null) {
+ throw new RuntimeException(String.format("Failed to get DateTimeFieldSpec for time column: %s of table: %s",
+ timeColumn, offlineTableName));
+ }
+ DateTimeFormatSpec timeFormatSpec = dateTimeSpec.getFormatSpec();
+ if (timeFormatSpec.getColumnUnit() == null) {
+ throw new RuntimeException(String.format("Time unit must be configured in the field spec for time column: %s of"
+ + " table: %s", timeColumn, offlineTableName));
+ }
+ List<SegmentZKMetadata> offlineSegmentsZKMetadata =
Review Comment:
Reading all ZK metadata for OFFLINE table can be very expensive. Time boundary info can be read from broker API. See `PinotBrokerDebug.getTimeBoundary()`
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r1994344588
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -165,6 +183,85 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig) {
+ List<String> segmentsToDelete = new ArrayList<>();
+ String realtimeTableName = realtimeTableConfig.getTableName();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ try {
+ TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(offlineTableName);
+ ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore();
+ Schema schema = ZKMetadataProvider.getTableSchema(propertyStore, offlineTableName);
+ if (schema == null) {
+ throw new RuntimeException("Failed to get schema for table: " + offlineTableName);
+ }
+ String timeColumn = null;
+ SegmentsValidationAndRetentionConfig validationConfig = offlineTableConfig.getValidationConfig();
+ if (validationConfig != null) {
+ timeColumn = validationConfig.getTimeColumnName();
+ }
+ if (StringUtils.isEmpty(timeColumn)) {
+ throw new RuntimeException("TimeColumn is null or empty for table: " + offlineTableName);
+ }
+ DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(timeColumn);
+ if (dateTimeSpec == null) {
+ throw new RuntimeException(String.format("Failed to get DateTimeFieldSpec for time column: %s of table: %s",
+ timeColumn, offlineTableName));
+ }
+ DateTimeFormatSpec timeFormatSpec = dateTimeSpec.getFormatSpec();
+ if (timeFormatSpec.getColumnUnit() == null) {
+ throw new RuntimeException(String.format("Time unit must be configured in the field spec for time column: %s of"
+ + " table: %s", timeColumn, offlineTableName));
+ }
+ // For HOURLY table with time unit other than DAYS, use (maxEndTime - 1 HOUR) as the time boundary; otherwise, use
+ // (maxEndTime - 1 DAY)
+ boolean isHourlyTable = CommonConstants.Table.PUSH_FREQUENCY_HOURLY.equalsIgnoreCase(
+ IngestionConfigUtils.getBatchSegmentIngestionFrequency(realtimeTableConfig))
+ && timeFormatSpec.getColumnUnit() != TimeUnit.DAYS;
+ long timeOffsetMs = isHourlyTable ? TimeUnit.HOURS.toMillis(1) : TimeUnit.DAYS.toMillis(1);
+ long offlineSegmentsMaxEndTimeMs = -1;
+ // Determine the max end time of all offline segments
+ for (SegmentZKMetadata segmentZKMetadata : _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName)) {
+ // skip empty segments
+ if (segmentZKMetadata.getTotalDocs() == 0) {
+ LOGGER.warn("Skipping empty segment: {} from table: {}", segmentZKMetadata.getSegmentName(),
+ offlineTableName);
+ continue;
+ }
+ offlineSegmentsMaxEndTimeMs = Math.max(offlineSegmentsMaxEndTimeMs, segmentZKMetadata.getEndTimeMs());
+ }
+
+ if (offlineSegmentsMaxEndTimeMs == -1) {
+ throw new RuntimeException("Failed to determine max end time of segments for table: " + offlineTableName);
+ }
+
+ // realtime table segments older than the time boundary will be deleted
+ long timeBoundaryMs = offlineSegmentsMaxEndTimeMs - timeOffsetMs;
+
+ // Iterate over all COMPLETED segments of the REALTIME table and check if they are eligible for deletion.
+ for (SegmentZKMetadata segmentZKMetadata : _pinotHelixResourceManager.getSegmentsZKMetadata(realtimeTableName)) {
+ // The segment should be in COMPLETED state
+ if (segmentZKMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.IN_PROGRESS
+ || segmentZKMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.COMMITTING) {
+ continue;
+ }
+ // The segment should be older than the cutoff time
+ if (segmentZKMetadata.getEndTimeMs() < timeBoundaryMs) {
+ segmentsToDelete.add(segmentZKMetadata.getSegmentName());
+ }
+ }
+ if (!segmentsToDelete.isEmpty()) {
Review Comment:
Done.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r2006085968
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/BrokerServiceHelper.java:
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.util;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.core.routing.TimeBoundaryInfo;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Helper class for interacting with broker APIs.
+ */
+public class BrokerServiceHelper {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BrokerServiceHelper.class);
+ private static final String TIME_BOUNDARY_INFO_API_PATH = "/debug/timeBoundary/%s";
+ private static final int DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
+ private final PinotHelixResourceManager _pinotHelixResourceManager;
+ private final ControllerConf _controllerConf;
+ private final Executor _executorService;
+ private final PoolingHttpClientConnectionManager _connectionManager;
+ private CompletionServiceHelper _completionServiceHelper;
Review Comment:
The `CompletionServiceHelper` is not initialized by default and is constructed everytime. A set() method is provided purely for testing purposes which makes writing tests easier.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.
swaminathanmanish commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r1994032068
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -165,6 +183,85 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig) {
Review Comment:
Will this time boundary logic work for user managed hybrid tables as well?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -165,6 +183,85 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig) {
+ List<String> segmentsToDelete = new ArrayList<>();
+ String realtimeTableName = realtimeTableConfig.getTableName();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ try {
+ TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(offlineTableName);
+ ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore();
+ Schema schema = ZKMetadataProvider.getTableSchema(propertyStore, offlineTableName);
+ if (schema == null) {
+ throw new RuntimeException("Failed to get schema for table: " + offlineTableName);
+ }
+ String timeColumn = null;
+ SegmentsValidationAndRetentionConfig validationConfig = offlineTableConfig.getValidationConfig();
+ if (validationConfig != null) {
+ timeColumn = validationConfig.getTimeColumnName();
+ }
+ if (StringUtils.isEmpty(timeColumn)) {
+ throw new RuntimeException("TimeColumn is null or empty for table: " + offlineTableName);
+ }
+ DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(timeColumn);
+ if (dateTimeSpec == null) {
+ throw new RuntimeException(String.format("Failed to get DateTimeFieldSpec for time column: %s of table: %s",
+ timeColumn, offlineTableName));
+ }
+ DateTimeFormatSpec timeFormatSpec = dateTimeSpec.getFormatSpec();
+ if (timeFormatSpec.getColumnUnit() == null) {
+ throw new RuntimeException(String.format("Time unit must be configured in the field spec for time column: %s of"
+ + " table: %s", timeColumn, offlineTableName));
+ }
+ // For HOURLY table with time unit other than DAYS, use (maxEndTime - 1 HOUR) as the time boundary; otherwise, use
+ // (maxEndTime - 1 DAY)
+ boolean isHourlyTable = CommonConstants.Table.PUSH_FREQUENCY_HOURLY.equalsIgnoreCase(
+ IngestionConfigUtils.getBatchSegmentIngestionFrequency(realtimeTableConfig))
+ && timeFormatSpec.getColumnUnit() != TimeUnit.DAYS;
+ long timeOffsetMs = isHourlyTable ? TimeUnit.HOURS.toMillis(1) : TimeUnit.DAYS.toMillis(1);
+ long offlineSegmentsMaxEndTimeMs = -1;
+ // Determine the max end time of all offline segments
+ for (SegmentZKMetadata segmentZKMetadata : _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName)) {
+ // skip empty segments
+ if (segmentZKMetadata.getTotalDocs() == 0) {
+ LOGGER.warn("Skipping empty segment: {} from table: {}", segmentZKMetadata.getSegmentName(),
+ offlineTableName);
+ continue;
+ }
+ offlineSegmentsMaxEndTimeMs = Math.max(offlineSegmentsMaxEndTimeMs, segmentZKMetadata.getEndTimeMs());
+ }
+
+ if (offlineSegmentsMaxEndTimeMs == -1) {
+ throw new RuntimeException("Failed to determine max end time of segments for table: " + offlineTableName);
+ }
+
+ // realtime table segments older than the time boundary will be deleted
+ long timeBoundaryMs = offlineSegmentsMaxEndTimeMs - timeOffsetMs;
+
+ // Iterate over all COMPLETED segments of the REALTIME table and check if they are eligible for deletion.
+ for (SegmentZKMetadata segmentZKMetadata : _pinotHelixResourceManager.getSegmentsZKMetadata(realtimeTableName)) {
+ // The segment should be in COMPLETED state
+ if (segmentZKMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.IN_PROGRESS
+ || segmentZKMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.COMMITTING) {
+ continue;
+ }
+ // The segment should be older than the cutoff time
+ if (segmentZKMetadata.getEndTimeMs() < timeBoundaryMs) {
+ segmentsToDelete.add(segmentZKMetadata.getSegmentName());
+ }
+ }
+ if (!segmentsToDelete.isEmpty()) {
Review Comment:
Please move log.info above the empty check so that we know how many segments are there (empty or not empty)
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -165,6 +183,85 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig) {
+ List<String> segmentsToDelete = new ArrayList<>();
+ String realtimeTableName = realtimeTableConfig.getTableName();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ try {
+ TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(offlineTableName);
+ ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore();
+ Schema schema = ZKMetadataProvider.getTableSchema(propertyStore, offlineTableName);
+ if (schema == null) {
+ throw new RuntimeException("Failed to get schema for table: " + offlineTableName);
+ }
+ String timeColumn = null;
+ SegmentsValidationAndRetentionConfig validationConfig = offlineTableConfig.getValidationConfig();
+ if (validationConfig != null) {
+ timeColumn = validationConfig.getTimeColumnName();
+ }
+ if (StringUtils.isEmpty(timeColumn)) {
+ throw new RuntimeException("TimeColumn is null or empty for table: " + offlineTableName);
+ }
+ DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(timeColumn);
+ if (dateTimeSpec == null) {
+ throw new RuntimeException(String.format("Failed to get DateTimeFieldSpec for time column: %s of table: %s",
+ timeColumn, offlineTableName));
+ }
+ DateTimeFormatSpec timeFormatSpec = dateTimeSpec.getFormatSpec();
+ if (timeFormatSpec.getColumnUnit() == null) {
+ throw new RuntimeException(String.format("Time unit must be configured in the field spec for time column: %s of"
+ + " table: %s", timeColumn, offlineTableName));
+ }
+ // For HOURLY table with time unit other than DAYS, use (maxEndTime - 1 HOUR) as the time boundary; otherwise, use
+ // (maxEndTime - 1 DAY)
+ boolean isHourlyTable = CommonConstants.Table.PUSH_FREQUENCY_HOURLY.equalsIgnoreCase(
+ IngestionConfigUtils.getBatchSegmentIngestionFrequency(realtimeTableConfig))
+ && timeFormatSpec.getColumnUnit() != TimeUnit.DAYS;
Review Comment:
Curious, do we have any validation to ensure offline & realtime table time ranges do not overlap, since hybrid tables can also be directly created & managed by user without RTO like tasks.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -165,6 +183,85 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig) {
+ List<String> segmentsToDelete = new ArrayList<>();
+ String realtimeTableName = realtimeTableConfig.getTableName();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ try {
+ TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(offlineTableName);
+ ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore();
+ Schema schema = ZKMetadataProvider.getTableSchema(propertyStore, offlineTableName);
+ if (schema == null) {
+ throw new RuntimeException("Failed to get schema for table: " + offlineTableName);
+ }
+ String timeColumn = null;
+ SegmentsValidationAndRetentionConfig validationConfig = offlineTableConfig.getValidationConfig();
+ if (validationConfig != null) {
+ timeColumn = validationConfig.getTimeColumnName();
+ }
+ if (StringUtils.isEmpty(timeColumn)) {
+ throw new RuntimeException("TimeColumn is null or empty for table: " + offlineTableName);
+ }
+ DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(timeColumn);
+ if (dateTimeSpec == null) {
+ throw new RuntimeException(String.format("Failed to get DateTimeFieldSpec for time column: %s of table: %s",
+ timeColumn, offlineTableName));
+ }
+ DateTimeFormatSpec timeFormatSpec = dateTimeSpec.getFormatSpec();
+ if (timeFormatSpec.getColumnUnit() == null) {
+ throw new RuntimeException(String.format("Time unit must be configured in the field spec for time column: %s of"
+ + " table: %s", timeColumn, offlineTableName));
+ }
+ // For HOURLY table with time unit other than DAYS, use (maxEndTime - 1 HOUR) as the time boundary; otherwise, use
+ // (maxEndTime - 1 DAY)
+ boolean isHourlyTable = CommonConstants.Table.PUSH_FREQUENCY_HOURLY.equalsIgnoreCase(
+ IngestionConfigUtils.getBatchSegmentIngestionFrequency(realtimeTableConfig))
Review Comment:
Is this the same logic used by broker as well for hybrid tables? Can we share the time boundary logic
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r2006317564
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/BrokerServiceHelper.java:
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.util;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.core.routing.TimeBoundaryInfo;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Helper class for interacting with broker APIs.
+ */
+public class BrokerServiceHelper {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BrokerServiceHelper.class);
+ private static final String TIME_BOUNDARY_INFO_API_PATH = "/debug/timeBoundary/%s";
+ private static final int DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
+ private final PinotHelixResourceManager _pinotHelixResourceManager;
+ private final ControllerConf _controllerConf;
+ private final Executor _executorService;
+ private final PoolingHttpClientConnectionManager _connectionManager;
+ private CompletionServiceHelper _completionServiceHelper;
Review Comment:
I see your point. Let me fix this.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r2006134057
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -199,6 +225,70 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig, TableConfig offlineTableConfig) {
+ LOGGER.info("Managing retention for hybrid table: {}", realtimeTableConfig.getTableName());
+ List<String> segmentsToDelete = new ArrayList<>();
+ String realtimeTableName = realtimeTableConfig.getTableName();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ try {
+ ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore();
+ Schema schema = ZKMetadataProvider.getTableSchema(propertyStore, offlineTableName);
+ if (schema == null) {
+ throw new RuntimeException("Failed to get schema for table: " + offlineTableName);
+ }
+ String timeColumn = null;
+ SegmentsValidationAndRetentionConfig validationConfig = offlineTableConfig.getValidationConfig();
+ if (validationConfig != null) {
+ timeColumn = validationConfig.getTimeColumnName();
+ }
+ if (StringUtils.isEmpty(timeColumn)) {
+ throw new RuntimeException("TimeColumn is null or empty for table: " + offlineTableName);
+ }
+ DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(timeColumn);
+ if (dateTimeSpec == null) {
+ throw new RuntimeException(String.format("Failed to get DateTimeFieldSpec for time column: %s of table: %s",
+ timeColumn, offlineTableName));
+ }
+ DateTimeFormatSpec timeFormatSpec = dateTimeSpec.getFormatSpec();
+ if (timeFormatSpec.getColumnUnit() == null) {
Review Comment:
The remaining part is correct. We shouldn't check `column unit`
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r2006015698
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -136,10 +150,22 @@ private void manageRetentionForTable(TableConfig tableConfig) {
if (TableNameBuilder.isOfflineTableResource(tableNameWithType)) {
manageRetentionForOfflineTable(tableNameWithType, retentionStrategy, untrackedSegmentsDeletionBatchSize);
} else {
- manageRetentionForRealtimeTable(tableNameWithType, retentionStrategy, untrackedSegmentsDeletionBatchSize);
+ String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType);
+ TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(rawTableName);
+ // hybrid table check should be performed before the realtime table check.
+ if (_isHybridTableRetentionStrategyEnabled && isHybridTable(tableConfig, offlineTableConfig)) {
Review Comment:
(minor) Can reduce one ZK lookup
```suggestion
TableConfig offlineTableConfig;
// hybrid table check should be performed before the realtime table check.
if (_isHybridTableRetentionStrategyEnabled && (offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(rawTableName)) != null) {
```
##########
pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerGauge.java:
##########
@@ -194,7 +194,8 @@ public enum ControllerGauge implements AbstractMetrics.Gauge {
// The number of segments in deepstore that do not have corresponding metadata in ZooKeeper.
// These segments are untracked and should be considered for deletion based on retention policies.
- UNTRACKED_SEGMENTS_COUNT("untrackedSegmentsCount", false);
+ UNTRACKED_SEGMENTS_COUNT("untrackedSegmentsCount", false),
+ RETENTION_MANAGER_ERROR("retentionManagerError", false);
Review Comment:
(minor) Add some javadoc for this
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -199,6 +225,70 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig, TableConfig offlineTableConfig) {
+ LOGGER.info("Managing retention for hybrid table: {}", realtimeTableConfig.getTableName());
+ List<String> segmentsToDelete = new ArrayList<>();
+ String realtimeTableName = realtimeTableConfig.getTableName();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ try {
+ ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore();
+ Schema schema = ZKMetadataProvider.getTableSchema(propertyStore, offlineTableName);
+ if (schema == null) {
+ throw new RuntimeException("Failed to get schema for table: " + offlineTableName);
+ }
+ String timeColumn = null;
+ SegmentsValidationAndRetentionConfig validationConfig = offlineTableConfig.getValidationConfig();
+ if (validationConfig != null) {
+ timeColumn = validationConfig.getTimeColumnName();
+ }
+ if (StringUtils.isEmpty(timeColumn)) {
+ throw new RuntimeException("TimeColumn is null or empty for table: " + offlineTableName);
+ }
+ DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(timeColumn);
+ if (dateTimeSpec == null) {
+ throw new RuntimeException(String.format("Failed to get DateTimeFieldSpec for time column: %s of table: %s",
+ timeColumn, offlineTableName));
+ }
+ DateTimeFormatSpec timeFormatSpec = dateTimeSpec.getFormatSpec();
+ if (timeFormatSpec.getColumnUnit() == null) {
Review Comment:
This shouldn't be needed. In fact it can break use cases with `SIMPLE_DATE_FORMAT`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/BrokerServiceHelper.java:
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.util;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.core.routing.TimeBoundaryInfo;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Helper class for interacting with broker APIs.
+ */
+public class BrokerServiceHelper {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BrokerServiceHelper.class);
+ private static final String TIME_BOUNDARY_INFO_API_PATH = "/debug/timeBoundary/%s";
+ private static final int DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
+ private final PinotHelixResourceManager _pinotHelixResourceManager;
+ private final ControllerConf _controllerConf;
+ private final Executor _executorService;
+ private final PoolingHttpClientConnectionManager _connectionManager;
+ private CompletionServiceHelper _completionServiceHelper;
Review Comment:
Is this safe to be cached and shared? I think we need to recompute the endpoint map every time
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -199,6 +225,70 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig, TableConfig offlineTableConfig) {
+ LOGGER.info("Managing retention for hybrid table: {}", realtimeTableConfig.getTableName());
+ List<String> segmentsToDelete = new ArrayList<>();
+ String realtimeTableName = realtimeTableConfig.getTableName();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ try {
+ ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore();
+ Schema schema = ZKMetadataProvider.getTableSchema(propertyStore, offlineTableName);
+ if (schema == null) {
+ throw new RuntimeException("Failed to get schema for table: " + offlineTableName);
Review Comment:
(minor) We usually throw `IllegalStateException`. You may use `Preconditions.checkState()` for concise. Same for other places
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r2006117235
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -199,6 +225,70 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig, TableConfig offlineTableConfig) {
+ LOGGER.info("Managing retention for hybrid table: {}", realtimeTableConfig.getTableName());
+ List<String> segmentsToDelete = new ArrayList<>();
+ String realtimeTableName = realtimeTableConfig.getTableName();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ try {
+ ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore();
+ Schema schema = ZKMetadataProvider.getTableSchema(propertyStore, offlineTableName);
+ if (schema == null) {
+ throw new RuntimeException("Failed to get schema for table: " + offlineTableName);
+ }
+ String timeColumn = null;
+ SegmentsValidationAndRetentionConfig validationConfig = offlineTableConfig.getValidationConfig();
+ if (validationConfig != null) {
+ timeColumn = validationConfig.getTimeColumnName();
+ }
+ if (StringUtils.isEmpty(timeColumn)) {
+ throw new RuntimeException("TimeColumn is null or empty for table: " + offlineTableName);
+ }
+ DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(timeColumn);
+ if (dateTimeSpec == null) {
+ throw new RuntimeException(String.format("Failed to get DateTimeFieldSpec for time column: %s of table: %s",
+ timeColumn, offlineTableName));
+ }
+ DateTimeFormatSpec timeFormatSpec = dateTimeSpec.getFormatSpec();
+ if (timeFormatSpec.getColumnUnit() == null) {
Review Comment:
The `TimeBoundaryInfo` is providing the `timeValue` in the configured time format. We need to convert this value into a value in milliseconds for comparison with `segmentZKMetadata.getEndTimeMs()`. The method `DateTimeFormatSpec.fromFormatToMillis(String dateTimeValue)` appears to handle the `SIMPLE_DATE_FORMAT`.
Please suggest alternative approach if you see issues with the above approach.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r2006137618
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/BrokerServiceHelper.java:
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.util;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.core.routing.TimeBoundaryInfo;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Helper class for interacting with broker APIs.
+ */
+public class BrokerServiceHelper {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BrokerServiceHelper.class);
+ private static final String TIME_BOUNDARY_INFO_API_PATH = "/debug/timeBoundary/%s";
+ private static final int DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
+ private final PinotHelixResourceManager _pinotHelixResourceManager;
+ private final ControllerConf _controllerConf;
+ private final Executor _executorService;
+ private final PoolingHttpClientConnectionManager _connectionManager;
+ private CompletionServiceHelper _completionServiceHelper;
+
+ public BrokerServiceHelper(PinotHelixResourceManager pinotHelixResourceManager, ControllerConf controllerConf,
+ Executor executor, PoolingHttpClientConnectionManager connectionManager) {
+ _pinotHelixResourceManager = pinotHelixResourceManager;
+ _controllerConf = controllerConf;
+ _executorService = executor;
+ _connectionManager = connectionManager;
+ }
+
+ @VisibleForTesting
+ public void setCompletionServiceHelper(CompletionServiceHelper completionServiceHelper) {
+ _completionServiceHelper = completionServiceHelper;
+ }
+
+ public BiMap<String, String> getBrokerEndpointsForInstance(List<InstanceConfig> instanceConfigs) {
+ String protocol = _controllerConf.getControllerBrokerProtocol();
+ BiMap<String, String> endpointsToInstances = HashBiMap.create(instanceConfigs.size());
+ for (InstanceConfig instanceConfig : instanceConfigs) {
+ String hostName = instanceConfig.getHostName();
+ if (hostName.startsWith(CommonConstants.Helix.PREFIX_OF_BROKER_INSTANCE)) {
+ hostName = hostName.substring(CommonConstants.Helix.BROKER_INSTANCE_PREFIX_LENGTH);
+ }
+ int port =
+ _controllerConf.getControllerBrokerPortOverride() > 0 ? _controllerConf.getControllerBrokerPortOverride()
+ : Integer.parseInt(instanceConfig.getPort());
+ String brokerEndpoint = String.format("%s://%s:%d", protocol, hostName, port);
+ endpointsToInstances.put(brokerEndpoint, instanceConfig.getInstanceName());
+ }
+ return endpointsToInstances;
+ }
+
+ public TimeBoundaryInfo getTimeBoundaryInfo(TableConfig offlineTableConfig) {
+ String offlineTableName = offlineTableConfig.getTableName();
+
+ List<InstanceConfig> instanceConfigs = _pinotHelixResourceManager.getBrokerInstancesConfigsFor(offlineTableName);
+ BiMap<String, String> endpointsToInstances = getBrokerEndpointsForInstance(instanceConfigs);
+
+ CompletionServiceHelper completionServiceHelper;
+ if (_completionServiceHelper == null) {
+ completionServiceHelper = new CompletionServiceHelper(_executorService, _connectionManager, endpointsToInstances);
+ } else {
+ completionServiceHelper = _completionServiceHelper;
+ }
+
+ List<String> timeBoundaryInfoUris = new ArrayList<>(endpointsToInstances.size());
+ for (String endpoint : endpointsToInstances.keySet()) {
+ String timeBoundaryInfoUri = endpoint + String.format(TIME_BOUNDARY_INFO_API_PATH, offlineTableName);
+ timeBoundaryInfoUris.add(timeBoundaryInfoUri);
+ }
+ Map<String, String> reqHeaders = new HashMap<>();
+ CompletionServiceHelper.CompletionServiceResponse serviceResponse =
+ completionServiceHelper.doMultiGetRequest(timeBoundaryInfoUris, offlineTableName, false, reqHeaders,
+ DEFAULT_REQUEST_TIMEOUT_MS, "get time boundary information for table from broker instances");
+ for (Map.Entry<String, String> streamResponse : serviceResponse._httpResponses.entrySet()) {
+ try {
+ return JsonUtils.stringToObject(streamResponse.getValue(), TimeBoundaryInfo.class);
+ } catch (JsonProcessingException e) {
+ LOGGER.error("Failed to parse timeBoundaryInfo response: {}", streamResponse.getKey(), e);
+ }
+ }
+ return null;
Review Comment:
Suggest throwing the exception out instead of using `null` to pass exception. The caller can catch the exception and log the stack trace with more context
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r2006159617
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/BrokerServiceHelper.java:
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.util;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.core.routing.TimeBoundaryInfo;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Helper class for interacting with broker APIs.
+ */
+public class BrokerServiceHelper {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BrokerServiceHelper.class);
+ private static final String TIME_BOUNDARY_INFO_API_PATH = "/debug/timeBoundary/%s";
+ private static final int DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
+ private final PinotHelixResourceManager _pinotHelixResourceManager;
+ private final ControllerConf _controllerConf;
+ private final Executor _executorService;
+ private final PoolingHttpClientConnectionManager _connectionManager;
+ private CompletionServiceHelper _completionServiceHelper;
Review Comment:
yes, the line#95 code path should be exercised only from tests. While not enforced, I have added a comment to the `set()` method that it's meant only for test purposes.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang merged PR #15257:
URL: https://github.com/apache/pinot/pull/15257
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r2006317564
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/BrokerServiceHelper.java:
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.util;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.core.routing.TimeBoundaryInfo;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Helper class for interacting with broker APIs.
+ */
+public class BrokerServiceHelper {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BrokerServiceHelper.class);
+ private static final String TIME_BOUNDARY_INFO_API_PATH = "/debug/timeBoundary/%s";
+ private static final int DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
+ private final PinotHelixResourceManager _pinotHelixResourceManager;
+ private final ControllerConf _controllerConf;
+ private final Executor _executorService;
+ private final PoolingHttpClientConnectionManager _connectionManager;
+ private CompletionServiceHelper _completionServiceHelper;
Review Comment:
I see your point. Let me fix this.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r2006323934
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/BrokerServiceHelper.java:
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.util;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.core.routing.TimeBoundaryInfo;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Helper class for interacting with broker APIs.
+ */
+public class BrokerServiceHelper {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BrokerServiceHelper.class);
+ private static final String TIME_BOUNDARY_INFO_API_PATH = "/debug/timeBoundary/%s";
+ private static final int DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
+ private final PinotHelixResourceManager _pinotHelixResourceManager;
+ private final ControllerConf _controllerConf;
+ private final Executor _executorService;
+ private final PoolingHttpClientConnectionManager _connectionManager;
+ private CompletionServiceHelper _completionServiceHelper;
Review Comment:
I'm using a method-local version of the `CompletionServiceHelper` and the same is not re-used between subsequent calls. The instance specific `CompletionServiceHelper` should be set only for testing purposes. I have added comment in the set method regarding this.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r2006135845
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/BrokerServiceHelper.java:
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.util;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.core.routing.TimeBoundaryInfo;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Helper class for interacting with broker APIs.
+ */
+public class BrokerServiceHelper {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BrokerServiceHelper.class);
+ private static final String TIME_BOUNDARY_INFO_API_PATH = "/debug/timeBoundary/%s";
+ private static final int DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
+ private final PinotHelixResourceManager _pinotHelixResourceManager;
+ private final ControllerConf _controllerConf;
+ private final Executor _executorService;
+ private final PoolingHttpClientConnectionManager _connectionManager;
+ private CompletionServiceHelper _completionServiceHelper;
Review Comment:
On line 95, we are reusing the cached one right?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r1994358106
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -165,6 +183,85 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig) {
Review Comment:
The notion of time boundary would apply to a hybrid table, bet it created by Pinot or manually created by user. Given this, it would be safe to delete segments from the REALTIME table that are older than the offline table time boundary.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -165,6 +183,85 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig) {
+ List<String> segmentsToDelete = new ArrayList<>();
+ String realtimeTableName = realtimeTableConfig.getTableName();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ try {
+ TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(offlineTableName);
+ ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore();
+ Schema schema = ZKMetadataProvider.getTableSchema(propertyStore, offlineTableName);
+ if (schema == null) {
+ throw new RuntimeException("Failed to get schema for table: " + offlineTableName);
+ }
+ String timeColumn = null;
+ SegmentsValidationAndRetentionConfig validationConfig = offlineTableConfig.getValidationConfig();
+ if (validationConfig != null) {
+ timeColumn = validationConfig.getTimeColumnName();
+ }
+ if (StringUtils.isEmpty(timeColumn)) {
+ throw new RuntimeException("TimeColumn is null or empty for table: " + offlineTableName);
+ }
+ DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(timeColumn);
+ if (dateTimeSpec == null) {
+ throw new RuntimeException(String.format("Failed to get DateTimeFieldSpec for time column: %s of table: %s",
+ timeColumn, offlineTableName));
+ }
+ DateTimeFormatSpec timeFormatSpec = dateTimeSpec.getFormatSpec();
+ if (timeFormatSpec.getColumnUnit() == null) {
+ throw new RuntimeException(String.format("Time unit must be configured in the field spec for time column: %s of"
+ + " table: %s", timeColumn, offlineTableName));
+ }
+ // For HOURLY table with time unit other than DAYS, use (maxEndTime - 1 HOUR) as the time boundary; otherwise, use
+ // (maxEndTime - 1 DAY)
+ boolean isHourlyTable = CommonConstants.Table.PUSH_FREQUENCY_HOURLY.equalsIgnoreCase(
+ IngestionConfigUtils.getBatchSegmentIngestionFrequency(realtimeTableConfig))
+ && timeFormatSpec.getColumnUnit() != TimeUnit.DAYS;
Review Comment:
Did you mean to ask that overlapping data should not be there? If so, it's fine for overlapping data to be present in the REALTIME and OFFLINE tables. The time boundary would take care of querying the appropriate table.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -165,6 +183,85 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig) {
+ List<String> segmentsToDelete = new ArrayList<>();
+ String realtimeTableName = realtimeTableConfig.getTableName();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ try {
+ TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(offlineTableName);
+ ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore();
+ Schema schema = ZKMetadataProvider.getTableSchema(propertyStore, offlineTableName);
+ if (schema == null) {
+ throw new RuntimeException("Failed to get schema for table: " + offlineTableName);
+ }
+ String timeColumn = null;
+ SegmentsValidationAndRetentionConfig validationConfig = offlineTableConfig.getValidationConfig();
+ if (validationConfig != null) {
+ timeColumn = validationConfig.getTimeColumnName();
+ }
+ if (StringUtils.isEmpty(timeColumn)) {
+ throw new RuntimeException("TimeColumn is null or empty for table: " + offlineTableName);
+ }
+ DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(timeColumn);
+ if (dateTimeSpec == null) {
+ throw new RuntimeException(String.format("Failed to get DateTimeFieldSpec for time column: %s of table: %s",
+ timeColumn, offlineTableName));
+ }
+ DateTimeFormatSpec timeFormatSpec = dateTimeSpec.getFormatSpec();
+ if (timeFormatSpec.getColumnUnit() == null) {
+ throw new RuntimeException(String.format("Time unit must be configured in the field spec for time column: %s of"
+ + " table: %s", timeColumn, offlineTableName));
+ }
+ // For HOURLY table with time unit other than DAYS, use (maxEndTime - 1 HOUR) as the time boundary; otherwise, use
+ // (maxEndTime - 1 DAY)
+ boolean isHourlyTable = CommonConstants.Table.PUSH_FREQUENCY_HOURLY.equalsIgnoreCase(
+ IngestionConfigUtils.getBatchSegmentIngestionFrequency(realtimeTableConfig))
+ && timeFormatSpec.getColumnUnit() != TimeUnit.DAYS;
+ long timeOffsetMs = isHourlyTable ? TimeUnit.HOURS.toMillis(1) : TimeUnit.DAYS.toMillis(1);
+ long offlineSegmentsMaxEndTimeMs = -1;
+ // Determine the max end time of all offline segments
+ for (SegmentZKMetadata segmentZKMetadata : _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName)) {
+ // skip empty segments
+ if (segmentZKMetadata.getTotalDocs() == 0) {
+ LOGGER.warn("Skipping empty segment: {} from table: {}", segmentZKMetadata.getSegmentName(),
+ offlineTableName);
+ continue;
+ }
+ offlineSegmentsMaxEndTimeMs = Math.max(offlineSegmentsMaxEndTimeMs, segmentZKMetadata.getEndTimeMs());
+ }
+
+ if (offlineSegmentsMaxEndTimeMs == -1) {
+ throw new RuntimeException("Failed to determine max end time of segments for table: " + offlineTableName);
+ }
+
+ // realtime table segments older than the time boundary will be deleted
+ long timeBoundaryMs = offlineSegmentsMaxEndTimeMs - timeOffsetMs;
+
+ // Iterate over all COMPLETED segments of the REALTIME table and check if they are eligible for deletion.
+ for (SegmentZKMetadata segmentZKMetadata : _pinotHelixResourceManager.getSegmentsZKMetadata(realtimeTableName)) {
+ // The segment should be in COMPLETED state
+ if (segmentZKMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.IN_PROGRESS
+ || segmentZKMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.COMMITTING) {
+ continue;
+ }
+ // The segment should be older than the cutoff time
+ if (segmentZKMetadata.getEndTimeMs() < timeBoundaryMs) {
+ segmentsToDelete.add(segmentZKMetadata.getSegmentName());
+ }
+ }
+ if (!segmentsToDelete.isEmpty()) {
Review Comment:
I'm removing this line as this info is explicitly getting printed within the `pinotHelixResourceManager. deleteSegments(...)` method.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -165,6 +183,85 @@ private void manageRetentionForRealtimeTable(String realtimeTableName, Retention
}
}
+ @VisibleForTesting
+ void manageRetentionForHybridTable(TableConfig realtimeTableConfig) {
+ List<String> segmentsToDelete = new ArrayList<>();
+ String realtimeTableName = realtimeTableConfig.getTableName();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
+ String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ try {
+ TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(offlineTableName);
+ ZkHelixPropertyStore<ZNRecord> propertyStore = _pinotHelixResourceManager.getPropertyStore();
+ Schema schema = ZKMetadataProvider.getTableSchema(propertyStore, offlineTableName);
+ if (schema == null) {
+ throw new RuntimeException("Failed to get schema for table: " + offlineTableName);
+ }
+ String timeColumn = null;
+ SegmentsValidationAndRetentionConfig validationConfig = offlineTableConfig.getValidationConfig();
+ if (validationConfig != null) {
+ timeColumn = validationConfig.getTimeColumnName();
+ }
+ if (StringUtils.isEmpty(timeColumn)) {
+ throw new RuntimeException("TimeColumn is null or empty for table: " + offlineTableName);
+ }
+ DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(timeColumn);
+ if (dateTimeSpec == null) {
+ throw new RuntimeException(String.format("Failed to get DateTimeFieldSpec for time column: %s of table: %s",
+ timeColumn, offlineTableName));
+ }
+ DateTimeFormatSpec timeFormatSpec = dateTimeSpec.getFormatSpec();
+ if (timeFormatSpec.getColumnUnit() == null) {
+ throw new RuntimeException(String.format("Time unit must be configured in the field spec for time column: %s of"
+ + " table: %s", timeColumn, offlineTableName));
+ }
+ // For HOURLY table with time unit other than DAYS, use (maxEndTime - 1 HOUR) as the time boundary; otherwise, use
+ // (maxEndTime - 1 DAY)
+ boolean isHourlyTable = CommonConstants.Table.PUSH_FREQUENCY_HOURLY.equalsIgnoreCase(
+ IngestionConfigUtils.getBatchSegmentIngestionFrequency(realtimeTableConfig))
Review Comment:
I had the same thought initially. However, it's a small portion (offsetTime) within the `TimeBoundaryManager` that we are using here and that logic is the same. For queries, the time boundary is calculated by looking into an explicitly set time boundary or calculate like the logic here. I chose to avoid making changes that would affect the query path.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r2006159617
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/BrokerServiceHelper.java:
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.util;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.core.routing.TimeBoundaryInfo;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Helper class for interacting with broker APIs.
+ */
+public class BrokerServiceHelper {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BrokerServiceHelper.class);
+ private static final String TIME_BOUNDARY_INFO_API_PATH = "/debug/timeBoundary/%s";
+ private static final int DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
+ private final PinotHelixResourceManager _pinotHelixResourceManager;
+ private final ControllerConf _controllerConf;
+ private final Executor _executorService;
+ private final PoolingHttpClientConnectionManager _connectionManager;
+ private CompletionServiceHelper _completionServiceHelper;
Review Comment:
yes, the line#95 code path should be exercised only from tests. While not enforced, I have added a comment to the `set()` method that it's meant only for test purposes.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -136,10 +150,22 @@ private void manageRetentionForTable(TableConfig tableConfig) {
if (TableNameBuilder.isOfflineTableResource(tableNameWithType)) {
manageRetentionForOfflineTable(tableNameWithType, retentionStrategy, untrackedSegmentsDeletionBatchSize);
} else {
- manageRetentionForRealtimeTable(tableNameWithType, retentionStrategy, untrackedSegmentsDeletionBatchSize);
+ String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType);
+ TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(rawTableName);
+ // hybrid table check should be performed before the realtime table check.
+ if (_isHybridTableRetentionStrategyEnabled && isHybridTable(tableConfig, offlineTableConfig)) {
Review Comment:
Checkstyle is complaining when code is written in the above style. Looks like it reduces readability.
```
[WARNING] src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:[158,34] (coding) InnerAssignment: Inner assignments should be avoided.
```
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #15257:
URL: https://github.com/apache/pinot/pull/15257#discussion_r2006623788
##########
pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java:
##########
@@ -344,6 +346,9 @@ private static long getRandomInitialDelayInSeconds() {
private static final long DEFAULT_RESOURCE_UTILIZATION_CHECKER_INITIAL_DELAY = 300L; // 5 minutes
private static final long DEFAULT_RESOURCE_UTILIZATION_CHECKER_FREQUENCY = 300L; // 5 minutes
private static final boolean DEFAULT_ENABLE_BATCH_MESSAGE_MODE = false;
+ private static final boolean DEFAULT_ENABLE_HYBRID_TABLE_RETENTION_STRATEGY = false;
+ // Disallow any high level consumer (HLC) table
+ private static final boolean DEFAULT_ALLOW_HLC_TABLES = false;
Review Comment:
Revert
```suggestion
```
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Improved retention management for hybrid tables [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #15257:
URL: https://github.com/apache/pinot/pull/15257#issuecomment-2718912263
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/15257?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 56.01%. Comparing base [(`59551e4`)](https://app.codecov.io/gh/apache/pinot/commit/59551e45224f1535c4863fd577622b37366ccc97?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`d3c47a5`)](https://app.codecov.io/gh/apache/pinot/commit/d3c47a5f24516e6430eda1da078cafe9f15111e3?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 1837 commits behind head on master.
> :exclamation: There is a different number of reports uploaded between BASE (59551e4) and HEAD (d3c47a5). Click for more details.
>
> <details><summary>HEAD has 48 uploads less than BASE</summary>
>
>| Flag | BASE (59551e4) | HEAD (d3c47a5) |
>|------|------|------|
>|integration|7|0|
>|integration2|3|0|
>|temurin|12|2|
>|java-21|7|2|
>|skip-bytebuffers-true|3|1|
>|skip-bytebuffers-false|7|1|
>|unittests|5|2|
>|java-11|5|0|
>|unittests2|3|0|
>|integration1|2|0|
>|custom-integration1|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #15257 +/- ##
============================================
- Coverage 61.75% 56.01% -5.74%
- Complexity 207 685 +478
============================================
Files 2436 2165 -271
Lines 133233 116846 -16387
Branches 20636 18702 -1934
============================================
- Hits 82274 65455 -16819
- Misses 44911 46012 +1101
+ Partials 6048 5379 -669
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/15257/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/15257/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/15257/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/15257/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/15257/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/15257/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/15257/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.01% <100.00%> (-5.61%)` | :arrow_down: |
| [skip-bytebuffers-false](https://app.codecov.io/gh/apache/pinot/pull/15257/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.99% <100.00%> (-5.76%)` | :arrow_down: |
| [skip-bytebuffers-true](https://app.codecov.io/gh/apache/pinot/pull/15257/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.97% <100.00%> (+28.24%)` | :arrow_up: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/15257/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.01% <100.00%> (-5.74%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/15257/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.01% <100.00%> (-5.73%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/15257/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.01% <100.00%> (+9.12%)` | :arrow_up: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/15257/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/15257?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary>🚀 New features to boost your workflow: </summary>
- ❄ [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- 📦 [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org