pri1712 commented on code in PR #19011:
URL: https://github.com/apache/pinot/pull/19011#discussion_r4062702198
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java:
##########
@@ -1868,8 +1879,80 @@ private static void
handleDistinctCountBitmapOverride(Expression expression) {
}
}
+ /// Attaches a `timeColumn >= (now - retention)` filter to the given query
so records outside the table's retention
+ /// window are excluded, even if their segment has not yet been deleted (see
issue #16689). Applied per-leg for hybrid
+ /// tables so the offline and realtime sides each use their own retention.
No-ops (with a debug log) when the config,
+ /// time column, retention, or schema spec is missing/malformed, rather than
failing the query.
+ @VisibleForTesting
+ static void handleSkipExpiredRecords(@Nullable TableConfig tableConfig,
@Nullable Schema schema,
+ PinotQuery pinotQuery) {
+ if (tableConfig == null || schema == null) {
+ return;
+ }
+ String tableNameWithType = tableConfig.getTableName();
+ SegmentsValidationAndRetentionConfig validationConfig =
tableConfig.getValidationConfig();
+ if (validationConfig == null) {
+ LOGGER.debug("skipExpiredRecords: no validation config for table {},
skipping retention filter",
+ tableNameWithType);
+ return;
+ }
+
+ String timeColumnName = validationConfig.getTimeColumnName();
+ if (timeColumnName == null) {
+ LOGGER.debug("skipExpiredRecords: no time column configured for table
{}, skipping retention filter",
+ tableNameWithType);
+ return;
+ }
+
+ Long retentionMs = getRetentionMs(validationConfig);
+ if (retentionMs == null) {
+ LOGGER.debug("skipExpiredRecords: no valid retention configured for
table {}, skipping retention filter",
+ tableNameWithType);
+ return;
+ }
+ long cutOffMs = System.currentTimeMillis() - retentionMs;
+
+ DateTimeFieldSpec timeFieldSpec =
schema.getSpecForTimeColumn(timeColumnName);
+ if (timeFieldSpec == null) {
+ LOGGER.debug("skipExpiredRecords: time column {} not found in schema for
table {}, skipping retention filter",
+ timeColumnName, tableNameWithType);
+ return;
+ }
+
+ DateTimeFormatSpec formatSpec = timeFieldSpec.getFormatSpec();
+ String cutOffValue = formatSpec.fromMillisToFormat(cutOffMs);
+ Expression cutOffLiteral = formatSpec.getTimeFormat() == TimeFormat.EPOCH
+ ? RequestUtils.getLiteralExpression(Long.parseLong(cutOffValue))
+ : RequestUtils.getLiteralExpression(cutOffValue);
+ Expression retentionFilter =
RequestUtils.getFunctionExpression(FilterKind.GREATER_THAN_OR_EQUAL.name(),
+ RequestUtils.getIdentifierExpression(timeColumnName), cutOffLiteral);
+
+ Expression existingFilter = pinotQuery.getFilterExpression();
+ pinotQuery.setFilterExpression(existingFilter != null
+ ? RequestUtils.getFunctionExpression(FilterKind.AND.name(),
existingFilter, retentionFilter)
+ : retentionFilter);
+ LOGGER.debug("skipExpiredRecords: attached retention filter {} >= {}
(cutOffMs={}) for table {}", timeColumnName,
+ cutOffValue, cutOffMs, tableNameWithType);
+ }
+
+ /// Parses the retention window in millis from the validation config, or
`null` when retention is not configured or is
+ /// malformed (in which case no retention filter is applied rather than
failing the query).
+ @Nullable
+ private static Long getRetentionMs(SegmentsValidationAndRetentionConfig
validationConfig) {
Review Comment:
makes sense, will add it there
--
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]
