diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java index b61abf35f083..2e7e89556a5b 100644 --- a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java +++ b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java @@ -116,7 +116,11 @@ import org.apache.pinot.spi.config.table.FieldConfig; import org.apache.pinot.spi.config.table.QueryConfig; import org.apache.pinot.spi.config.table.RoutingConfig; +import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig; import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.data.DateTimeFieldSpec; +import org.apache.pinot.spi.data.DateTimeFieldSpec.TimeFormat; +import org.apache.pinot.spi.data.DateTimeFormatSpec; import org.apache.pinot.spi.data.LogicalTableConfig; import org.apache.pinot.spi.data.Schema; import org.apache.pinot.spi.env.PinotConfiguration; @@ -708,12 +712,16 @@ protected BrokerResponse doHandleRequest(long requestId, String query, SqlNodeAn BrokerRequest offlineBrokerRequest = null; BrokerRequest realtimeBrokerRequest = null; + boolean skipExpiredRecords = QueryOptionsUtils.isSkipExpiredRecords(serverPinotQuery.getQueryOptions()); if (routeInfo.isHybrid()) { // Hybrid PinotQuery offlinePinotQuery = serverPinotQuery.deepCopy(); offlinePinotQuery.getDataSource().setTableName(offlineTableName); assert timeBoundaryInfo != null; attachTimeBoundary(offlinePinotQuery, timeBoundaryInfo, true); + if (skipExpiredRecords) { + handleSkipExpiredRecords(offlineTableConfig, schema, offlinePinotQuery); + } handleExpressionOverride(offlinePinotQuery, _tableCache.getExpressionOverrideMap(offlineTableName)); handleTimestampIndexOverride(offlinePinotQuery, offlineTableConfig); // Re-optimize after attaching the time boundary filter so that filter optimizers (e.g. NumericalFilterOptimizer, @@ -724,6 +732,9 @@ protected BrokerResponse doHandleRequest(long requestId, String query, SqlNodeAn PinotQuery realtimePinotQuery = serverPinotQuery.deepCopy(); realtimePinotQuery.getDataSource().setTableName(realtimeTableName); attachTimeBoundary(realtimePinotQuery, timeBoundaryInfo, false); + if (skipExpiredRecords) { + handleSkipExpiredRecords(realtimeTableConfig, schema, realtimePinotQuery); + } handleExpressionOverride(realtimePinotQuery, _tableCache.getExpressionOverrideMap(realtimeTableName)); handleTimestampIndexOverride(realtimePinotQuery, realtimeTableConfig); _queryOptimizer.optimize(realtimePinotQuery, schema); @@ -735,6 +746,10 @@ protected BrokerResponse doHandleRequest(long requestId, String query, SqlNodeAn } else if (routeInfo.isOffline()) { // OFFLINE only setTableName(serverBrokerRequest, offlineTableName); + if (skipExpiredRecords) { + handleSkipExpiredRecords(offlineTableConfig, schema, serverPinotQuery); + _queryOptimizer.optimize(serverPinotQuery, schema); + } handleExpressionOverride(serverPinotQuery, _tableCache.getExpressionOverrideMap(offlineTableName)); handleTimestampIndexOverride(serverPinotQuery, offlineTableConfig); offlineBrokerRequest = serverBrokerRequest; @@ -744,6 +759,10 @@ protected BrokerResponse doHandleRequest(long requestId, String query, SqlNodeAn } else { // REALTIME only setTableName(serverBrokerRequest, realtimeTableName); + if (skipExpiredRecords) { + handleSkipExpiredRecords(realtimeTableConfig, schema, serverPinotQuery); + _queryOptimizer.optimize(serverPinotQuery, schema); + } handleExpressionOverride(serverPinotQuery, _tableCache.getExpressionOverrideMap(realtimeTableName)); handleTimestampIndexOverride(serverPinotQuery, realtimeTableConfig); realtimeBrokerRequest = serverBrokerRequest; @@ -1246,7 +1265,6 @@ private CompileResult compileRequest(long requestId, String query, SqlNodeAndOpt if (_enableDistinctCountBitmapOverride) { handleDistinctCountBitmapOverride(serverPinotQuery); } - Schema schema = _tableCache.getSchema(rawTableName); _queryOptimizer.optimize(serverPinotQuery, schema); @@ -1861,6 +1879,78 @@ 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) { + String retentionUnit = validationConfig.getRetentionTimeUnit(); + String retentionValue = validationConfig.getRetentionTimeValue(); + if (StringUtils.isEmpty(retentionUnit) || StringUtils.isEmpty(retentionValue)) { + return null; + } + try { + return TimeUnit.valueOf(retentionUnit.toUpperCase()).toMillis(Long.parseLong(retentionValue)); + } catch (IllegalArgumentException e) { + return null; + } + } + private HandlerContext getHandlerContext(@Nullable QueryConfig offlineTableQueryConfig, @Nullable QueryConfig realtimeTableQueryConfig) { Boolean disableGroovyOverride = null; @@ -2630,6 +2720,8 @@ private ImplicitHybridTableRouteInfo prepareBaseTableHybridRoute(BrokerRequest b TableConfig realtimeTableConfig = baseRouteInfo.getRealtimeTableConfig(); TimeBoundaryInfo timeBoundaryInfo = baseRouteInfo.getTimeBoundaryInfo(); + boolean skipExpiredRecords = + QueryOptionsUtils.isSkipExpiredRecords(baseBrokerRequest.getPinotQuery().getQueryOptions()); if (baseRouteInfo.isHybrid()) { PinotQuery basePinotQuery = baseBrokerRequest.getPinotQuery(); @@ -2638,6 +2730,9 @@ private ImplicitHybridTableRouteInfo prepareBaseTableHybridRoute(BrokerRequest b if (timeBoundaryInfo != null) { attachTimeBoundary(offlinePinotQuery, timeBoundaryInfo, true); } + if (skipExpiredRecords) { + handleSkipExpiredRecords(offlineTableConfig, schema, offlinePinotQuery); + } handleExpressionOverride(offlinePinotQuery, _tableCache.getExpressionOverrideMap(offlineTableName)); handleTimestampIndexOverride(offlinePinotQuery, offlineTableConfig); _queryOptimizer.optimize(offlinePinotQuery, schema); @@ -2648,6 +2743,9 @@ private ImplicitHybridTableRouteInfo prepareBaseTableHybridRoute(BrokerRequest b if (timeBoundaryInfo != null) { attachTimeBoundary(realtimePinotQuery, timeBoundaryInfo, false); } + if (skipExpiredRecords) { + handleSkipExpiredRecords(realtimeTableConfig, schema, realtimePinotQuery); + } handleExpressionOverride(realtimePinotQuery, _tableCache.getExpressionOverrideMap(realtimeTableName)); handleTimestampIndexOverride(realtimePinotQuery, realtimeTableConfig); _queryOptimizer.optimize(realtimePinotQuery, schema); @@ -2657,12 +2755,20 @@ private ImplicitHybridTableRouteInfo prepareBaseTableHybridRoute(BrokerRequest b hybridRoute.setRealtimeBrokerRequest(realtimeBrokerRequest); } else if (baseRouteInfo.isOffline()) { setTableName(baseBrokerRequest, offlineTableName); + if (skipExpiredRecords) { + handleSkipExpiredRecords(offlineTableConfig, schema, baseBrokerRequest.getPinotQuery()); + _queryOptimizer.optimize(baseBrokerRequest.getPinotQuery(), schema); + } handleExpressionOverride(baseBrokerRequest.getPinotQuery(), _tableCache.getExpressionOverrideMap(offlineTableName)); handleTimestampIndexOverride(baseBrokerRequest.getPinotQuery(), offlineTableConfig); hybridRoute.setOfflineBrokerRequest(baseBrokerRequest); } else { setTableName(baseBrokerRequest, realtimeTableName); + if (skipExpiredRecords) { + handleSkipExpiredRecords(realtimeTableConfig, schema, baseBrokerRequest.getPinotQuery()); + _queryOptimizer.optimize(baseBrokerRequest.getPinotQuery(), schema); + } handleExpressionOverride(baseBrokerRequest.getPinotQuery(), _tableCache.getExpressionOverrideMap(realtimeTableName)); handleTimestampIndexOverride(baseBrokerRequest.getPinotQuery(), realtimeTableConfig); diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/SkipExpiredRecordsTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/SkipExpiredRecordsTest.java new file mode 100644 index 000000000000..a8b2f5ebaa5f --- /dev/null +++ b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/SkipExpiredRecordsTest.java @@ -0,0 +1,352 @@ +/** + * 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.broker.requesthandler; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.helix.model.InstanceConfig; +import org.apache.pinot.broker.broker.AllowAllAccessControlFactory; +import org.apache.pinot.broker.queryquota.QueryQuotaManager; +import org.apache.pinot.broker.routing.manager.BrokerRoutingManager; +import org.apache.pinot.common.config.provider.TableCache; +import org.apache.pinot.common.metrics.BrokerMetrics; +import org.apache.pinot.common.request.BrokerRequest; +import org.apache.pinot.common.request.Expression; +import org.apache.pinot.common.request.Function; +import org.apache.pinot.common.response.broker.BrokerResponseNative; +import org.apache.pinot.core.routing.RoutingTable; +import org.apache.pinot.core.routing.SegmentsToQuery; +import org.apache.pinot.core.routing.TableRouteInfo; +import org.apache.pinot.core.transport.ServerInstance; +import org.apache.pinot.spi.accounting.ThreadAccountantUtils; +import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TenantConfig; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.env.PinotConfiguration; +import org.apache.pinot.spi.eventlistener.query.BrokerQueryEventListenerFactory; +import org.apache.pinot.spi.trace.RequestContext; +import org.apache.pinot.spi.utils.CommonConstants.Query.Range; +import org.apache.pinot.sql.FilterKind; +import org.mockito.Mockito; +import org.testng.Assert; +import org.testng.annotations.Test; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/// Tests for the {@code skipExpiredRecords} query option in {@link BaseBrokerRequestHandler}. +///

When the option is set the broker injects a lower-bound time filter derived from the table's +/// retention config before the query reaches any server. Tests verify the injected filter's +/// structure and value, and the skip conditions (option absent, no retention config, missing time +/// column in schema, ORDER BY wrapper, existing WHERE clause). +public class SkipExpiredRecordsTest { + + private static final String RAW_TABLE = "myTable"; + private static final String OFFLINE_TABLE = "myTable_OFFLINE"; + private static final String TIME_COLUMN = "eventTime"; + private static final int RETENTION_DAYS = 30; + private static final long RETENTION_MS = RETENTION_DAYS * 24L * 3600 * 1000; + + /// Allowed delta between expected and actual cutoff to absorb test execution time. + private static final long CUTOFF_TOLERANCE_MS = 5_000L; + + @Test + public void testOptionNotSetNoFilterInjected() + throws Exception { + AtomicReference captured = new AtomicReference<>(); + BaseSingleStageBrokerRequestHandler handler = createHandler(buildTableCache(true, true), captured); + + handler.handleRequest("SELECT * FROM " + RAW_TABLE); + + BrokerRequest req = captured.get(); + Assert.assertNotNull(req); + Assert.assertNull(req.getPinotQuery().getFilterExpression(), + "No filter should be injected when skipExpiredRecords is not set"); + } + + @Test + public void testOptionSetNumericEpochColumnInjectsLowerBoundFilter() + throws Exception { + AtomicReference captured = new AtomicReference<>(); + BaseSingleStageBrokerRequestHandler handler = createHandler(buildTableCache(true, true), captured); + + long beforeMs = System.currentTimeMillis(); + handler.handleRequest("SET skipExpiredRecords='true'; SELECT * FROM " + RAW_TABLE); + long afterMs = System.currentTimeMillis(); + + BrokerRequest req = captured.get(); + Assert.assertNotNull(req); + Expression filter = req.getPinotQuery().getFilterExpression(); + Assert.assertNotNull(filter, "Filter should be injected when option is set"); + Assert.assertTrue(filterContainsColumn(filter, TIME_COLUMN), + "Injected filter must reference the time column"); + + long cutoff = extractLowerBound(filter, TIME_COLUMN); + long expectedMin = beforeMs - RETENTION_MS - CUTOFF_TOLERANCE_MS; + long expectedMax = afterMs - RETENTION_MS + CUTOFF_TOLERANCE_MS; + Assert.assertTrue(cutoff >= expectedMin && cutoff <= expectedMax, + "Cutoff must be approximately now - " + RETENTION_DAYS + " days; got " + cutoff + + " expected in [" + expectedMin + ", " + expectedMax + "]"); + } + + @Test + public void testOptionSetExistingWhereClauseFilterAndedIn() + throws Exception { + AtomicReference captured = new AtomicReference<>(); + BaseSingleStageBrokerRequestHandler handler = createHandler(buildTableCache(true, true), captured); + + handler.handleRequest( + "SET skipExpiredRecords='true'; SELECT * FROM " + RAW_TABLE + " WHERE col = 'foo'"); + + BrokerRequest req = captured.get(); + Assert.assertNotNull(req); + Expression filter = req.getPinotQuery().getFilterExpression(); + Assert.assertNotNull(filter, "Filter should exist"); + + Function func = filter.getFunctionCall(); + Assert.assertNotNull(func, "Filter must be a function call"); + Assert.assertEquals(func.getOperator(), FilterKind.AND.name(), + "Existing WHERE and the injected retention filter must be combined under AND"); + Assert.assertEquals(func.getOperands().size(), 2, + "AND must have exactly two operands"); + Assert.assertTrue(filterContainsColumn(filter, TIME_COLUMN), + "Combined filter must include the time-column retention predicate"); + } + + @Test + public void testOptionSetOrderByQueryFilterInjected() + throws Exception { + AtomicReference captured = new AtomicReference<>(); + BaseSingleStageBrokerRequestHandler handler = createHandler(buildTableCache(true, true), captured); + + // ORDER BY causes the Calcite AST root to be SqlOrderBy, not SqlSelect. + // The broker must unwrap it before injecting the filter. + handler.handleRequest("SET skipExpiredRecords='true'; SELECT * FROM " + RAW_TABLE + + " ORDER BY " + TIME_COLUMN + " DESC LIMIT 10"); + + BrokerRequest req = captured.get(); + Assert.assertNotNull(req); + Expression filter = req.getPinotQuery().getFilterExpression(); + Assert.assertNotNull(filter, "Filter must be injected even for ORDER BY queries"); + Assert.assertTrue(filterContainsColumn(filter, TIME_COLUMN), + "Injected filter must reference the time column"); + } + + @Test + public void testOptionSetNoRetentionConfiguredNoFilterInjected() + throws Exception { + AtomicReference captured = new AtomicReference<>(); + // Table config has no retention values set + BaseSingleStageBrokerRequestHandler handler = createHandler(buildTableCache(false, true), captured); + + handler.handleRequest("SET skipExpiredRecords='true'; SELECT * FROM " + RAW_TABLE); + + BrokerRequest req = captured.get(); + Assert.assertNotNull(req); + Assert.assertNull(req.getPinotQuery().getFilterExpression(), + "No filter should be injected when table has no retention configuration"); + } + + @Test + public void testOptionSetTimeColumnMissingFromSchemaNoFilterInjected() + throws Exception { + AtomicReference captured = new AtomicReference<>(); + // Table config declares a time column and retention, but the schema has no such column + BaseSingleStageBrokerRequestHandler handler = createHandler(buildTableCache(true, false), captured); + + handler.handleRequest("SET skipExpiredRecords='true'; SELECT * FROM " + RAW_TABLE); + + BrokerRequest req = captured.get(); + Assert.assertNotNull(req); + Assert.assertNull(req.getPinotQuery().getFilterExpression(), + "No filter should be injected when the time column is absent from the schema"); + } + + // --------------------------------------------------------------------------- + // Infrastructure helpers + // --------------------------------------------------------------------------- + + /// Builds a mock {@link TableCache} for {@value RAW_TABLE}. + /// + /// @param withRetention when {@code true} the validation config carries a 30-day retention on + /// {@value TIME_COLUMN}; when {@code false} the retention fields are left empty + /// @param withTimeColumnInSchema when {@code true} the schema includes the {@value TIME_COLUMN} + /// DateTime column; when {@code false} only {@code col} is present + private static TableCache buildTableCache(boolean withRetention, boolean withTimeColumnInSchema) { + Schema.SchemaBuilder schemaBuilder = new Schema.SchemaBuilder() + .setSchemaName(RAW_TABLE) + .addSingleValueDimension("col", DataType.STRING); + if (withTimeColumnInSchema) { + schemaBuilder.addDateTime(TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS"); + } + Schema schema = schemaBuilder.build(); + + Map columnMap = withTimeColumnInSchema + ? Map.of(TIME_COLUMN, TIME_COLUMN, "col", "col") + : Map.of("col", "col"); + + SegmentsValidationAndRetentionConfig validationConfig = new SegmentsValidationAndRetentionConfig(); + if (withRetention) { + validationConfig.setTimeColumnName(TIME_COLUMN); + validationConfig.setRetentionTimeUnit("DAYS"); + validationConfig.setRetentionTimeValue(String.valueOf(RETENTION_DAYS)); + } + + TableConfig tableConfig = mock(TableConfig.class); + when(tableConfig.getTenantConfig()).thenReturn(new TenantConfig("tier_BROKER", "tier_SERVER", null)); + when(tableConfig.getValidationConfig()).thenReturn(validationConfig); + + TableCache tableCache = mock(TableCache.class); + when(tableCache.getActualTableName(RAW_TABLE)).thenReturn(RAW_TABLE); + when(tableCache.getSchema(RAW_TABLE)).thenReturn(schema); + when(tableCache.getColumnNameMap(anyString())).thenReturn(columnMap); + // retention lookup uses raw name; routing lookup uses type-suffixed name + when(tableCache.getTableConfig(RAW_TABLE)).thenReturn(tableConfig); + when(tableCache.getTableConfig(OFFLINE_TABLE)).thenReturn(tableConfig); + return tableCache; + } + + private static BaseSingleStageBrokerRequestHandler createHandler(TableCache tableCache, + AtomicReference capturedServerRequest) { + BrokerRoutingManager routingManager = mock(BrokerRoutingManager.class); + when(routingManager.routingExists(OFFLINE_TABLE)).thenReturn(true); + when(routingManager.getQueryTimeoutMs(anyString())).thenReturn(10_000L); + RoutingTable rt = mock(RoutingTable.class); + when(rt.getServerInstanceToSegmentsMap()).thenReturn( + Map.of(new ServerInstance(new InstanceConfig("server01_9000")), + new SegmentsToQuery(List.of("seg01"), List.of()))); + when(routingManager.getRoutingTable(any(), Mockito.anyLong())).thenReturn(rt); + + QueryQuotaManager quotaManager = mock(QueryQuotaManager.class); + when(quotaManager.acquire(anyString())).thenReturn(true); + when(quotaManager.acquireDatabase(anyString())).thenReturn(true); + when(quotaManager.acquireApplication(anyString())).thenReturn(true); + + BrokerMetrics.register(mock(BrokerMetrics.class)); + PinotConfiguration config = new PinotConfiguration(); + BrokerQueryEventListenerFactory.init(config); + + return new BaseSingleStageBrokerRequestHandler(config, "testBroker", new BrokerRequestIdGenerator(), + routingManager, new AllowAllAccessControlFactory(), quotaManager, tableCache, + ThreadAccountantUtils.getNoOpAccountant(), null, null) { + @Override + public void start() { + } + + @Override + public void shutDown() { + } + + @Override + protected BrokerResponseNative processBrokerRequest(long requestId, BrokerRequest originalBrokerRequest, + BrokerRequest serverBrokerRequest, TableRouteInfo route, long timeoutMs, ServerStats serverStats, + RequestContext requestContext) { + capturedServerRequest.set(serverBrokerRequest); + return BrokerResponseNative.empty(); + } + + @Override + protected BrokerResponseNative processMaterializedViewSplitBrokerRequest(long requestId, + long materializedViewRequestId, BrokerRequest originalBrokerRequest, TableRouteInfo baseRoute, + TableRouteInfo materializedViewRoute, long timeoutMs, ServerStats serverStats, + RequestContext requestContext) { + return BrokerResponseNative.empty(); + } + }; + } + + /// Returns true if {@code expr} or any of its descendants references {@code columnName}. + private static boolean filterContainsColumn(Expression expr, String columnName) { + if (expr == null) { + return false; + } + if (expr.getIdentifier() != null && columnName.equals(expr.getIdentifier().getName())) { + return true; + } + if (expr.getFunctionCall() != null) { + for (Expression operand : expr.getFunctionCall().getOperands()) { + if (filterContainsColumn(operand, columnName)) { + return true; + } + } + } + return false; + } + + /// Finds the retention lower-bound predicate on {@code timeColumn} inside {@code filter} and + /// returns the cutoff as an epoch-ms long. + /// + ///

A standalone {@code >=} injected by the broker stays as {@code GREATER_THAN_OR_EQUAL} in + /// the PinotQuery; Pinot only rewrites to a {@code RANGE} when merging two-sided bounds (e.g. + /// {@code BETWEEN}, a time-boundary merge). Both forms are handled here. + private static long extractLowerBound(Expression filter, String timeColumn) { + Assert.assertNotNull(filter, "filter must not be null"); + Function func = filter.getFunctionCall(); + Assert.assertNotNull(func, "filter must be a function call"); + + if (FilterKind.AND.name().equals(func.getOperator())) { + for (Expression operand : func.getOperands()) { + Function fn = operand.getFunctionCall(); + if (fn != null && !fn.getOperands().isEmpty() + && timeColumn.equals(fn.getOperands().get(0).getIdentifier().getName())) { + if (FilterKind.RANGE.name().equals(fn.getOperator())) { + return parseLowerBoundFromRange(fn.getOperands().get(1).getLiteral().getStringValue()); + } + if (FilterKind.GREATER_THAN_OR_EQUAL.name().equals(fn.getOperator())) { + return extractLongLiteral(fn.getOperands().get(1)); + } + } + } + throw new AssertionError( + "No RANGE or GREATER_THAN_OR_EQUAL predicate found on " + timeColumn + " inside AND"); + } + + Assert.assertEquals(func.getOperands().get(0).getIdentifier().getName(), timeColumn, + "Predicate must be on " + timeColumn); + if (FilterKind.RANGE.name().equals(func.getOperator())) { + return parseLowerBoundFromRange(func.getOperands().get(1).getLiteral().getStringValue()); + } + Assert.assertEquals(func.getOperator(), FilterKind.GREATER_THAN_OR_EQUAL.name(), + "Expected GREATER_THAN_OR_EQUAL or RANGE but found: " + func.getOperator()); + return extractLongLiteral(func.getOperands().get(1)); + } + + /// Parses the lower bound from a RANGE string of the form {@code '[value\0*)'}, + /// using {@link Range#DELIMITER} as the separator between lower and upper bound. + private static long parseLowerBoundFromRange(String rangeStr) { + int delimIdx = rangeStr.indexOf(Range.DELIMITER); + Assert.assertTrue(delimIdx > 1, "Range string missing delimiter: " + rangeStr); + return Long.parseLong(rangeStr.substring(1, delimIdx).trim()); + } + + /// Extracts a long value from a numeric literal expression. + private static long extractLongLiteral(Expression expr) { + Assert.assertNotNull(expr.getLiteral(), "Expected a literal expression"); + if (expr.getLiteral().isSetLongValue()) { + return expr.getLiteral().getLongValue(); + } + // Numeric literals compiled from string (e.g. createExactNumeric) may land as stringValue + return Long.parseLong(expr.getLiteral().getStringValue()); + } +} diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java index a48c88cd1468..a7f0756adddf 100644 --- a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java +++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java @@ -331,6 +331,10 @@ public static boolean isSkipUpsertDelete(Map queryOptions) { return Boolean.parseBoolean(queryOptions.get(QueryOptionKey.SKIP_UPSERT_DELETE)); } + public static boolean isSkipExpiredRecords(Map queryOptions) { + return Boolean.parseBoolean(queryOptions.get(QueryOptionKey.SKIP_EXPIRED_RECORDS)); + } + public static boolean isTraceRuleProductions(Map queryOptions) { return Boolean.parseBoolean(queryOptions.get(QueryOptionKey.TRACE_RULE_PRODUCTIONS)); } diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/SkipExpiredRecordsTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/SkipExpiredRecordsTest.java new file mode 100644 index 000000000000..30f24e6550dd --- /dev/null +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/SkipExpiredRecordsTest.java @@ -0,0 +1,154 @@ +/** + * 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.integration.tests.custom; + +import com.fasterxml.jackson.databind.JsonNode; +import java.io.File; +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.apache.avro.file.DataFileWriter; +import org.apache.avro.generic.GenericData; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.data.FieldSpec; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.utils.builder.TableConfigBuilder; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; + + +/// End-to-end integration test for the `skipExpiredRecords` query option (issue #16689). +/// +/// The option is honored by the single-stage engine only: when set, the broker appends a +/// `timeColumn >= (now - retention)` filter so records older than the table's retention window are excluded even though +/// their segment has not been deleted. Rows are split into "fresh" (within retention) and "expired" (well past +/// retention); the test asserts the counts with and without the option. +@Test(suiteName = "CustomClusterIntegrationTest") +public class SkipExpiredRecordsTest extends CustomDataQueryClusterIntegrationTest { + private static final String DEFAULT_TABLE_NAME = "SkipExpiredRecordsTest"; + private static final String ID_COLUMN = "id"; + // Retention is 5 days; a row this far in the past is out of retention. + private static final long EXPIRED_TS = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(30); + // A row at "now" is always within retention. + private static final long FRESH_TS = System.currentTimeMillis(); + private static final int NUM_FRESH_ROWS = 3; + private static final int NUM_EXPIRED_ROWS = 5; + + @Override + public String getTableName() { + return DEFAULT_TABLE_NAME; + } + + @Override + protected long getCountStarResult() { + // waitForAllDocsLoaded queries without the option, so all rows must be present. + return NUM_FRESH_ROWS + NUM_EXPIRED_ROWS; + } + + @Override + public Schema createSchema() { + return new Schema.SchemaBuilder().setSchemaName(getTableName()) + .addSingleValueDimension(ID_COLUMN, FieldSpec.DataType.INT) + .addDateTimeField(TIMESTAMP_FIELD_NAME, FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS") + .build(); + } + + @Override + public TableConfig createOfflineTableConfig() { + return new TableConfigBuilder(TableType.OFFLINE) + .setTableName(getTableName()) + .setTimeColumnName(TIMESTAMP_FIELD_NAME) + .setRetentionTimeUnit(TimeUnit.DAYS.name()) + .setRetentionTimeValue("5") + .build(); + } + + @Override + public List createAvroFiles() + throws Exception { + org.apache.avro.Schema avroSchema = org.apache.avro.Schema.createRecord("skipExpiredRecord", null, null, false); + avroSchema.setFields(List.of( + new org.apache.avro.Schema.Field(ID_COLUMN, org.apache.avro.Schema.create(org.apache.avro.Schema.Type.INT), + null, null), + new org.apache.avro.Schema.Field(TIMESTAMP_FIELD_NAME, + org.apache.avro.Schema.create(org.apache.avro.Schema.Type.LONG), null, null) + )); + + try (AvroFilesAndWriters avroFilesAndWriters = createAvroFilesAndWriters(avroSchema)) { + List> writers = avroFilesAndWriters.getWriters(); + int id = 0; + for (int i = 0; i < NUM_FRESH_ROWS; i++) { + writers.get(id % getNumAvroFiles()).append(newRecord(avroSchema, id++, FRESH_TS)); + } + for (int i = 0; i < NUM_EXPIRED_ROWS; i++) { + writers.get(id % getNumAvroFiles()).append(newRecord(avroSchema, id++, EXPIRED_TS)); + } + return avroFilesAndWriters.getAvroFiles(); + } + } + + private static GenericData.Record newRecord(org.apache.avro.Schema avroSchema, int id, long ts) { + GenericData.Record record = new GenericData.Record(avroSchema); + record.put(ID_COLUMN, id); + record.put(TIMESTAMP_FIELD_NAME, ts); + return record; + } + + @Test + public void testWithoutOptionReturnsAllRows() + throws Exception { + // The option is single-stage-only; scope the whole test to SSE. + setUseMultiStageQueryEngine(false); + JsonNode response = postQuery("SELECT COUNT(*) FROM " + getTableName()); + assertCount(response, NUM_FRESH_ROWS + NUM_EXPIRED_ROWS); + } + + @Test + public void testSkipExpiredRecordsExcludesOutOfRetentionRows() + throws Exception { + setUseMultiStageQueryEngine(false); + JsonNode response = + postQueryWithOptions("SELECT COUNT(*) FROM " + getTableName(), "skipExpiredRecords=true"); + assertCount(response, NUM_FRESH_ROWS); + } + + @Test + public void testSkipExpiredRecordsCombinesWithExistingFilter() + throws Exception { + setUseMultiStageQueryEngine(false); + // id 0 is a fresh row -> kept; the existing predicate AND the retention filter both hold. + JsonNode keptResponse = + postQueryWithOptions("SELECT COUNT(*) FROM " + getTableName() + " WHERE " + ID_COLUMN + " = 0", + "skipExpiredRecords=true"); + assertCount(keptResponse, 1); + + // id = NUM_FRESH_ROWS is the first expired row -> excluded by the retention filter despite matching the predicate. + JsonNode filteredResponse = + postQueryWithOptions("SELECT COUNT(*) FROM " + getTableName() + " WHERE " + ID_COLUMN + " = " + NUM_FRESH_ROWS, + "skipExpiredRecords=true"); + assertCount(filteredResponse, 0); + } + + private void assertCount(JsonNode response, long expectedCount) { + assertEquals(response.path("exceptions").size(), 0, response.toPrettyString()); + assertEquals(getType(response, 0), "LONG"); + assertEquals(getLongCellValue(response, 0, 0), expectedCount); + } +} diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java index 2c76cf1e39f2..2e60ac5b73ff 100644 --- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java +++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java @@ -806,6 +806,8 @@ public static class QueryOptionKey { public static final String MAX_EXECUTION_THREADS = "maxExecutionThreads"; public static final String COLLECT_GC_STATS = "collectGCStats"; public static final String QUERY_HASH = "queryHash"; + /// Query option to skip Expired Records from segments based on your retention time. + public static final String SKIP_EXPIRED_RECORDS = "skipExpiredRecords"; // For group-by queries with order-by clause, the tail groups are trimmed off to reduce the memory footprint. To // ensure the accuracy of the result, {@code max(limit * 5, minTrimSize)} groups are retained. When