xiangfu0 commented on code in PR #19011:
URL: https://github.com/apache/pinot/pull/19011#discussion_r3698672838
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java:
##########
@@ -464,4 +480,107 @@ protected void onQueryFinish(long requestId) {
protected boolean isQueryCancellationEnabled() {
return _enableQueryCancellation;
}
+
+ /// Appends a where clause to the query to filter out of retention data if
SKIP_OUT_OF_RETENTION_VALUES is
+ /// set to True in query Options.
+ /// @param sqlNodeAndOptions
+ private void applySkipOutOfRetentionValuesIfNeeded(SqlNodeAndOptions
sqlNodeAndOptions) {
+ Map<String, String> options = sqlNodeAndOptions.getOptions();
+ if
(!Boolean.parseBoolean(options.get(QueryOptionKey.SKIP_OUT_OF_RETENTION_VALUES)))
{
+ return;
+ }
+ SqlNode sqlNode = sqlNodeAndOptions.getSqlNode();
+ if (sqlNode == null) {
+ return;
+ }
+
+ // Resolve the inner SqlSelect, unwrapping SqlOrderBy if needed.
+ // SqlWith (CTEs) and other statement types are not supported.
+ SqlSelect sqlSelect;
+ if (sqlNode instanceof SqlSelect) {
+ sqlSelect = (SqlSelect) sqlNode;
+ } else if (sqlNode instanceof SqlOrderBy) {
+ SqlNode query = ((SqlOrderBy) sqlNode).query;
+ if (!(query instanceof SqlSelect)) {
+ return;
+ }
+ sqlSelect = (SqlSelect) query;
+ } else {
+ return;
+ }
+
+ TableNameExtractor tableNameExtractor = new TableNameExtractor();
+ tableNameExtractor.extractTableNames(sqlSelect);
+ Set<String> tableNames = tableNameExtractor.getTableNames();
+ if (tableNames.isEmpty()) {
+ return;
+ }
+
+ // Multi-table queries (JOINs) are not supported: it is ambiguous which
table's retention
+ // applies, and injecting a filter with the wrong table's time column
would produce
+ // incorrect results. Callers must issue separate single-table queries.
+ if (tableNames.size() > 1) {
+ LOGGER.debug("skipOutOfRetentionValues is not supported for multi-table
queries; skipping filter injection");
+ return;
+ }
+
+ String rawTableName =
TableNameBuilder.extractRawTableName(tableNames.iterator().next());
+
+ TableConfig tableConfig =
_tableCache.getTableConfig(TableNameBuilder.OFFLINE.tableNameWithType(rawTableName));
Review Comment:
**[Critical] Resolve the queried physical table before reading retention.**
`extractRawTableName()` drops an explicit `_REALTIME` suffix, then this lookup
always prefers `<raw>_OFFLINE`. For a hybrid table with 30-day OFFLINE and
7-day REALTIME retention, a query against `foo_REALTIME` injects the 30-day
cutoff and silently includes expired rows. This also runs before database and
case canonicalization, so qualified or differently-cased names can miss or
select the wrong config. Preserve the explicit type, resolve the canonical
fully qualified table, and derive retention per physical scan.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java:
##########
@@ -464,4 +480,107 @@ protected void onQueryFinish(long requestId) {
protected boolean isQueryCancellationEnabled() {
return _enableQueryCancellation;
}
+
+ /// Appends a where clause to the query to filter out of retention data if
SKIP_OUT_OF_RETENTION_VALUES is
+ /// set to True in query Options.
+ /// @param sqlNodeAndOptions
+ private void applySkipOutOfRetentionValuesIfNeeded(SqlNodeAndOptions
sqlNodeAndOptions) {
+ Map<String, String> options = sqlNodeAndOptions.getOptions();
+ if
(!Boolean.parseBoolean(options.get(QueryOptionKey.SKIP_OUT_OF_RETENTION_VALUES)))
{
+ return;
+ }
+ SqlNode sqlNode = sqlNodeAndOptions.getSqlNode();
+ if (sqlNode == null) {
+ return;
+ }
+
+ // Resolve the inner SqlSelect, unwrapping SqlOrderBy if needed.
+ // SqlWith (CTEs) and other statement types are not supported.
+ SqlSelect sqlSelect;
+ if (sqlNode instanceof SqlSelect) {
+ sqlSelect = (SqlSelect) sqlNode;
+ } else if (sqlNode instanceof SqlOrderBy) {
+ SqlNode query = ((SqlOrderBy) sqlNode).query;
+ if (!(query instanceof SqlSelect)) {
+ return;
+ }
+ sqlSelect = (SqlSelect) query;
+ } else {
+ return;
+ }
+
+ TableNameExtractor tableNameExtractor = new TableNameExtractor();
+ tableNameExtractor.extractTableNames(sqlSelect);
+ Set<String> tableNames = tableNameExtractor.getTableNames();
+ if (tableNames.isEmpty()) {
+ return;
+ }
+
+ // Multi-table queries (JOINs) are not supported: it is ambiguous which
table's retention
+ // applies, and injecting a filter with the wrong table's time column
would produce
+ // incorrect results. Callers must issue separate single-table queries.
+ if (tableNames.size() > 1) {
+ LOGGER.debug("skipOutOfRetentionValues is not supported for multi-table
queries; skipping filter injection");
+ return;
+ }
+
+ String rawTableName =
TableNameBuilder.extractRawTableName(tableNames.iterator().next());
+
+ TableConfig tableConfig =
_tableCache.getTableConfig(TableNameBuilder.OFFLINE.tableNameWithType(rawTableName));
+ if (tableConfig == null) {
+ tableConfig =
_tableCache.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(rawTableName));
+ }
+ Schema schema = _tableCache.getSchema(rawTableName);
+ if (tableConfig == null || schema == null ||
tableConfig.getValidationConfig() == null) {
+ return;
+ }
+
+ //get timestamp column and retention details
+ SegmentsValidationAndRetentionConfig validationConfig =
tableConfig.getValidationConfig();
+ String timeColumnName = validationConfig.getTimeColumnName();
+ if (StringUtils.isEmpty(timeColumnName) ||
!validationConfig.hasRetention()) {
+ return;
+ }
+ try {
+ long retentionTimeMs = validationConfig.getRetentionTimeMillis();
+ long cutoffMs = System.currentTimeMillis() - retentionTimeMs;
+ DateTimeFieldSpec timeFieldSpec =
schema.getSpecForTimeColumn(timeColumnName);
+ if (timeFieldSpec == null) {
+ return;
+ }
+
+ DateTimeFormatSpec formatSpec = new
DateTimeFormatSpec(timeFieldSpec.getFormat());
+ String formattedCutoffTime = formatSpec.fromMillisToFormat(cutoffMs);
+
+ //add where clause to the AST
+ SqlIdentifier timeColNode = new SqlIdentifier(timeColumnName,
SqlParserPos.ZERO);
+ SqlLiteral cutoffNode;
+ if (timeFieldSpec.getDataType().isNumeric()) {
+ cutoffNode = SqlLiteral.createExactNumeric(formattedCutoffTime,
SqlParserPos.ZERO);
+ } else {
+ cutoffNode = SqlLiteral.createCharString(formattedCutoffTime,
SqlParserPos.ZERO);
Review Comment:
**[Critical] Avoid direct string comparison for variable-width date
formats.** An accepted STRING time format such as `yyyy-M-d` can produce cutoff
`2025-10-1`; an expired stored value `2025-9-30` compares lexicographically
greater and therefore passes this predicate. Schema validation checks component
order but not fixed width. Compare converted epoch values or reject formats
that are not guaranteed to be lexicographically sortable, and add STRING/SDF
boundary tests.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java:
##########
@@ -464,4 +480,107 @@ protected void onQueryFinish(long requestId) {
protected boolean isQueryCancellationEnabled() {
return _enableQueryCancellation;
}
+
+ /// Appends a where clause to the query to filter out of retention data if
SKIP_OUT_OF_RETENTION_VALUES is
+ /// set to True in query Options.
+ /// @param sqlNodeAndOptions
+ private void applySkipOutOfRetentionValuesIfNeeded(SqlNodeAndOptions
sqlNodeAndOptions) {
+ Map<String, String> options = sqlNodeAndOptions.getOptions();
+ if
(!Boolean.parseBoolean(options.get(QueryOptionKey.SKIP_OUT_OF_RETENTION_VALUES)))
{
+ return;
+ }
+ SqlNode sqlNode = sqlNodeAndOptions.getSqlNode();
+ if (sqlNode == null) {
+ return;
+ }
+
+ // Resolve the inner SqlSelect, unwrapping SqlOrderBy if needed.
+ // SqlWith (CTEs) and other statement types are not supported.
+ SqlSelect sqlSelect;
Review Comment:
**[Major] Apply this rewrite at resolved table scans.** CTE and
set-operation roots are silently skipped, while a derived-table query finds the
inner physical table but attaches the predicate to this outer `SqlSelect`. For
example, `SELECT * FROM (SELECT category FROM foo) t` receives an outer
`eventTime` reference that does not exist. Equivalent single-table SQL can
therefore execute unfiltered or fail based only on syntax. Use a planner-level
scan rewrite, or explicitly reject unsupported shapes.
##########
pinot-integration-tests/src/test/resources/log4j2-test.xml:
##########
@@ -30,7 +30,7 @@
<Console name="spammy" target="SYSTEM_OUT">
<PatternLayout pattern="%d{HH:mm:ss.SSS} %p [%c{1}] [%t] %m%n"/>
<Filters>
- <BurstFilter level="ERROR" rate="1" maxBurst="2"/>
+ <BurstFilter level="DEBUG" rate="1" maxBurst="2"/>
Review Comment:
**[Major] Revert this unrelated logging change.** The attached accounting
logger emits WARN and above; with a DEBUG BurstFilter threshold, those
WARN/ERROR events bypass rate limiting across the integration suite. Restore
`ERROR`, or isolate any additional logging required by this test.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java:
##########
@@ -464,4 +480,107 @@ protected void onQueryFinish(long requestId) {
protected boolean isQueryCancellationEnabled() {
return _enableQueryCancellation;
}
+
+ /// Appends a where clause to the query to filter out of retention data if
SKIP_OUT_OF_RETENTION_VALUES is
+ /// set to True in query Options.
+ /// @param sqlNodeAndOptions
+ private void applySkipOutOfRetentionValuesIfNeeded(SqlNodeAndOptions
sqlNodeAndOptions) {
+ Map<String, String> options = sqlNodeAndOptions.getOptions();
+ if
(!Boolean.parseBoolean(options.get(QueryOptionKey.SKIP_OUT_OF_RETENTION_VALUES)))
{
+ return;
+ }
+ SqlNode sqlNode = sqlNodeAndOptions.getSqlNode();
+ if (sqlNode == null) {
+ return;
+ }
+
+ // Resolve the inner SqlSelect, unwrapping SqlOrderBy if needed.
+ // SqlWith (CTEs) and other statement types are not supported.
+ SqlSelect sqlSelect;
+ if (sqlNode instanceof SqlSelect) {
+ sqlSelect = (SqlSelect) sqlNode;
+ } else if (sqlNode instanceof SqlOrderBy) {
+ SqlNode query = ((SqlOrderBy) sqlNode).query;
+ if (!(query instanceof SqlSelect)) {
+ return;
+ }
+ sqlSelect = (SqlSelect) query;
+ } else {
+ return;
+ }
+
+ TableNameExtractor tableNameExtractor = new TableNameExtractor();
+ tableNameExtractor.extractTableNames(sqlSelect);
+ Set<String> tableNames = tableNameExtractor.getTableNames();
+ if (tableNames.isEmpty()) {
+ return;
+ }
+
+ // Multi-table queries (JOINs) are not supported: it is ambiguous which
table's retention
+ // applies, and injecting a filter with the wrong table's time column
would produce
+ // incorrect results. Callers must issue separate single-table queries.
+ if (tableNames.size() > 1) {
+ LOGGER.debug("skipOutOfRetentionValues is not supported for multi-table
queries; skipping filter injection");
+ return;
+ }
+
+ String rawTableName =
TableNameBuilder.extractRawTableName(tableNames.iterator().next());
+
+ TableConfig tableConfig =
_tableCache.getTableConfig(TableNameBuilder.OFFLINE.tableNameWithType(rawTableName));
+ if (tableConfig == null) {
+ tableConfig =
_tableCache.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(rawTableName));
+ }
+ Schema schema = _tableCache.getSchema(rawTableName);
+ if (tableConfig == null || schema == null ||
tableConfig.getValidationConfig() == null) {
+ return;
+ }
+
+ //get timestamp column and retention details
+ SegmentsValidationAndRetentionConfig validationConfig =
tableConfig.getValidationConfig();
+ String timeColumnName = validationConfig.getTimeColumnName();
+ if (StringUtils.isEmpty(timeColumnName) ||
!validationConfig.hasRetention()) {
+ return;
+ }
+ try {
+ long retentionTimeMs = validationConfig.getRetentionTimeMillis();
+ long cutoffMs = System.currentTimeMillis() - retentionTimeMs;
+ DateTimeFieldSpec timeFieldSpec =
schema.getSpecForTimeColumn(timeColumnName);
+ if (timeFieldSpec == null) {
+ return;
+ }
+
+ DateTimeFormatSpec formatSpec = new
DateTimeFormatSpec(timeFieldSpec.getFormat());
+ String formattedCutoffTime = formatSpec.fromMillisToFormat(cutoffMs);
+
+ //add where clause to the AST
+ SqlIdentifier timeColNode = new SqlIdentifier(timeColumnName,
SqlParserPos.ZERO);
+ SqlLiteral cutoffNode;
+ if (timeFieldSpec.getDataType().isNumeric()) {
+ cutoffNode = SqlLiteral.createExactNumeric(formattedCutoffTime,
SqlParserPos.ZERO);
+ } else {
+ cutoffNode = SqlLiteral.createCharString(formattedCutoffTime,
SqlParserPos.ZERO);
+ }
+
+ SqlBasicCall rangeFilter = new SqlBasicCall(
+ SqlStdOperatorTable.GREATER_THAN_OR_EQUAL, List.of(timeColNode,
cutoffNode),
+ SqlParserPos.ZERO);
+
+ SqlNode existingWhere = sqlSelect.getWhere();
+ if (existingWhere != null) {
+ //add an AND if the where exists
+ SqlBasicCall andExpr = new SqlBasicCall(
+ SqlStdOperatorTable.AND,
+ List.of(existingWhere, rangeFilter),
+ SqlParserPos.ZERO
+ );
+ sqlSelect.setWhere(andExpr);
+ } else {
+ sqlSelect.setWhere(rangeFilter);
+ }
+ LOGGER.debug("Injected Calcite AST filter for skipOutOfRetentionValues
on table: {}. Cutoff: {}", rawTableName,
+ formattedCutoffTime);
+ } catch (RuntimeException e) {
Review Comment:
**[Major] Do not silently fail open when the option cannot be honored.**
`hasRetention()` only checks that the two strings are nonempty; a malformed
retention value passes that guard, throws during conversion, and reaches this
catch, after which the original unfiltered query executes with no
client-visible warning. Validate a positive numeric retention value during
table admission and return a query-validation error when an enabled rewrite
cannot be applied.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java:
##########
@@ -464,4 +480,107 @@ protected void onQueryFinish(long requestId) {
protected boolean isQueryCancellationEnabled() {
return _enableQueryCancellation;
}
+
+ /// Appends a where clause to the query to filter out of retention data if
SKIP_OUT_OF_RETENTION_VALUES is
+ /// set to True in query Options.
+ /// @param sqlNodeAndOptions
+ private void applySkipOutOfRetentionValuesIfNeeded(SqlNodeAndOptions
sqlNodeAndOptions) {
+ Map<String, String> options = sqlNodeAndOptions.getOptions();
+ if
(!Boolean.parseBoolean(options.get(QueryOptionKey.SKIP_OUT_OF_RETENTION_VALUES)))
{
+ return;
+ }
+ SqlNode sqlNode = sqlNodeAndOptions.getSqlNode();
+ if (sqlNode == null) {
+ return;
+ }
+
+ // Resolve the inner SqlSelect, unwrapping SqlOrderBy if needed.
+ // SqlWith (CTEs) and other statement types are not supported.
+ SqlSelect sqlSelect;
+ if (sqlNode instanceof SqlSelect) {
+ sqlSelect = (SqlSelect) sqlNode;
+ } else if (sqlNode instanceof SqlOrderBy) {
+ SqlNode query = ((SqlOrderBy) sqlNode).query;
+ if (!(query instanceof SqlSelect)) {
+ return;
+ }
+ sqlSelect = (SqlSelect) query;
+ } else {
+ return;
+ }
+
+ TableNameExtractor tableNameExtractor = new TableNameExtractor();
+ tableNameExtractor.extractTableNames(sqlSelect);
+ Set<String> tableNames = tableNameExtractor.getTableNames();
+ if (tableNames.isEmpty()) {
+ return;
+ }
+
+ // Multi-table queries (JOINs) are not supported: it is ambiguous which
table's retention
+ // applies, and injecting a filter with the wrong table's time column
would produce
+ // incorrect results. Callers must issue separate single-table queries.
+ if (tableNames.size() > 1) {
+ LOGGER.debug("skipOutOfRetentionValues is not supported for multi-table
queries; skipping filter injection");
+ return;
+ }
+
+ String rawTableName =
TableNameBuilder.extractRawTableName(tableNames.iterator().next());
+
+ TableConfig tableConfig =
_tableCache.getTableConfig(TableNameBuilder.OFFLINE.tableNameWithType(rawTableName));
+ if (tableConfig == null) {
+ tableConfig =
_tableCache.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(rawTableName));
+ }
+ Schema schema = _tableCache.getSchema(rawTableName);
+ if (tableConfig == null || schema == null ||
tableConfig.getValidationConfig() == null) {
+ return;
+ }
+
+ //get timestamp column and retention details
+ SegmentsValidationAndRetentionConfig validationConfig =
tableConfig.getValidationConfig();
+ String timeColumnName = validationConfig.getTimeColumnName();
+ if (StringUtils.isEmpty(timeColumnName) ||
!validationConfig.hasRetention()) {
+ return;
+ }
+ try {
+ long retentionTimeMs = validationConfig.getRetentionTimeMillis();
+ long cutoffMs = System.currentTimeMillis() - retentionTimeMs;
+ DateTimeFieldSpec timeFieldSpec =
schema.getSpecForTimeColumn(timeColumnName);
+ if (timeFieldSpec == null) {
+ return;
+ }
+
+ DateTimeFormatSpec formatSpec = new
DateTimeFormatSpec(timeFieldSpec.getFormat());
Review Comment:
**[Major] Reuse the cached format specification.**
`DateTimeFieldSpec.getFormatSpec()` already caches and safely publishes the
parsed `DateTimeFormatSpec`. Constructing another one here reparses the schema
format and allocates formatter state on every opted-in query. Use
`timeFieldSpec.getFormatSpec()`.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SkipOutOfRetentionValuesIntegrationTest.java:
##########
@@ -0,0 +1,171 @@
+/**
+ * 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;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+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.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.apache.pinot.util.TestUtils;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+
+
+ /// Integration test for the `skipOutOfRetentionValues` query option.
+ /// Verifies that when the query option is provided, the broker dynamically
injects a time
+ /// filter based on the table's retention configuration and pushes it down to
the servers.
+public class SkipOutOfRetentionValuesIntegrationTest extends
BaseClusterIntegrationTest {
Review Comment:
**[Major] Use the shared custom integration-test cluster.** This test only
defines ordinary schema, data, and SQL assertions, yet starts a dedicated
ZK/controller/broker/server cluster. Move it under `integration/tests/custom`,
extend `CustomDataQueryClusterIntegrationTest`, add `@Test(suiteName =
"CustomClusterIntegrationTest")`, and include it in the shared custom-cluster
suite.
--
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]
