This is an automated email from the ASF dual-hosted git repository.
yashmayya pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 1d48614087f Don't let RLS filters clobber the leaf's
serverReturnFinalResult flag (#19418)
1d48614087f is described below
commit 1d48614087f0ceee0fc3a118f9e5b8ce07e24a62
Author: Yash Mayya <[email protected]>
AuthorDate: Tue Sep 1 22:26:55 2026 -0400
Don't let RLS filters clobber the leaf's serverReturnFinalResult flag
(#19418)
---
.../runtime/plan/server/ServerPlanRequestUtils.java | 7 ++++---
.../pinot/query/runtime/queries/QueryRunnerTest.java | 16 ++++++++++++++++
.../pinot/query/runtime/queries/QueryRunnerTestBase.java | 8 ++++++++
3 files changed, 28 insertions(+), 3 deletions(-)
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/server/ServerPlanRequestUtils.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/server/ServerPlanRequestUtils.java
index 4cdd84594bf..8c35aa5442f 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/server/ServerPlanRequestUtils.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/server/ServerPlanRequestUtils.java
@@ -28,7 +28,6 @@ import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.function.BiConsumer;
import javax.annotation.Nullable;
-import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.request.BrokerRequest;
@@ -110,8 +109,10 @@ public class ServerPlanRequestUtils {
PinotQuery pinotQuery = serverContext.getPinotQuery();
pinotQuery.setExplain(explain);
- if (MapUtils.isNotEmpty(rowFilters)) {
- pinotQuery.setQueryOptions(rowFilters);
+ if (rowFilters != null) {
+ // Merge, never replace: the plan visitor may already have stamped
SERVER_RETURN_FINAL_RESULT[_KEY_UNPARTITIONED]
+ // here, and updateQueryOptions() below does not restore them.
+ rowFilters.forEach(pinotQuery::putToQueryOptions);
}
List<InstanceRequest> instanceRequests;
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
index 951d7ae15ed..234c99cea74 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
@@ -48,6 +48,7 @@ import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
import org.apache.pinot.spi.utils.CommonConstants.MultiStageQueryRunner;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.apache.pinot.sql.parsers.rewriter.RlsUtils;
import org.assertj.core.api.Assertions;
import org.intellij.lang.annotations.Language;
import org.testng.Assert;
@@ -372,6 +373,21 @@ public class QueryRunnerTest extends QueryRunnerTestBase {
}
}
+ /// RLS filters are stamped onto the leaf's query options, which is also the
only place the planner records that the
+ /// leaf must finalize its aggregates (`is_partitioned_by_group_by_keys`
makes the aggregate `AggType.DIRECT`, so no
+ /// stage above it can finalize anything). Stamping must merge, not replace:
dropping the flag makes the leaf emit a
+ /// raw `IntOpenHashSet` into a column typed `INT`. Table b lives on a
single server, so one row per group.
+ @Test
+ public void testDirectAggregateWithRowLevelSecurityFilter() {
+ String sql = "SELECT /*+
aggOptions(is_partitioned_by_group_by_keys='true') */ col1, DISTINCTCOUNT(col3)
FROM b "
+ + "GROUP BY col1 ORDER BY col1";
+ Map<String, String> rlsFilters = Map.of(RlsUtils.buildRlsFilterKey("b"),
"col3 > 1");
+ QueryDispatcher.QueryResult queryResult = queryRunner(sql, false,
rlsFilters);
+ Assert.assertNull(queryResult.getProcessingException(), "Query failed: " +
queryResult.getProcessingException());
+ // The RLS filter keeps only the two rows with col3 = 42.
+ compareRowEquals(queryResult.getResultTable(), List.of(new Object[]{"bar",
1}, new Object[]{"bob", 1}), true);
+ }
+
@DataProvider(name = "testDataWithSqlToFinalRowCount")
protected Object[][] provideTestSqlAndRowCount() {
//@formatter:off
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTestBase.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTestBase.java
index 6d0b455402d..fc5db95f8f2 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTestBase.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTestBase.java
@@ -114,6 +114,13 @@ public abstract class QueryRunnerTestBase extends
QueryTestSet {
/// Dispatch query to each pinot-server. The logic should mimic
QueryDispatcher.submit() but does not actually make
/// ser/de dispatches.
protected QueryDispatcher.QueryResult queryRunner(String sql, boolean trace)
{
+ return queryRunner(sql, trace, Map.of());
+ }
+
+ /// Same as [#queryRunner(String, boolean)], but adds metadata the broker
stamps outside the SQL text, e.g. the
+ /// `rlsFilters-<table>` entries carrying row-level-security filters.
+ protected QueryDispatcher.QueryResult queryRunner(String sql, boolean trace,
+ Map<String, String> extraRequestMetadata) {
long startTimeMs = System.currentTimeMillis();
long requestId = REQUEST_ID_GEN.getAndIncrement();
SqlNodeAndOptions sqlNodeAndOptions =
CalciteSqlParser.compileToSqlNodeAndOptions(sql);
@@ -141,6 +148,7 @@ public abstract class QueryRunnerTestBase extends
QueryTestSet {
requestMetadataMap.put(QueryOptionKey.TIMEOUT_MS,
Long.toString(timeoutMs));
requestMetadataMap.put(QueryOptionKey.EXTRA_PASSIVE_TIMEOUT_MS,
Long.toString(extraPassiveTimeoutMs));
requestMetadataMap.putIfAbsent(QueryOptionKey.ENABLE_NULL_HANDLING,
"true");
+ requestMetadataMap.putAll(extraRequestMetadata);
// Putting trace testing here as extra options as it doesn't go along with
the rest of the items.
if (trace) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
