This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang 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 8d140dcfba3 Fix rlsFiltersApplied reporting true for multi-stage
queries when no RLS filter is applied (#20089)
8d140dcfba3 is described below
commit 8d140dcfba31efb69766ceb622e2f68005de5bf0
Author: Dylan Wylie <[email protected]>
AuthorDate: Sat Nov 7 00:00:47 2026 +0000
Fix rlsFiltersApplied reporting true for multi-stage queries when no RLS
filter is applied (#20089)
---
.../MultiStageBrokerRequestHandler.java | 45 +++++++++++++---------
.../MultiStageBrokerRequestHandlerTest.java | 38 +++++++++++++++++-
2 files changed, 63 insertions(+), 20 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
index 1d1f9fd9809..81dbfe2ddac 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
@@ -41,7 +41,6 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import javax.ws.rs.WebApplicationException;
@@ -470,8 +469,7 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
QueryEnvironment.CompiledQuery compiledQuery = compileQuery(requestId,
query, sqlNodeAndOptions, requestContext,
httpHeaders, queryTimer)) {
validatePhysicalTablesWithMultiClusterRouting(compiledQuery.getTableNames(),
compiledQuery.getOptions());
- AtomicBoolean rlsFiltersApplied = new AtomicBoolean(false);
- checkAuthorization(requesterIdentity, requestContext, httpHeaders,
compiledQuery, rlsFiltersApplied);
+ boolean rlsFiltersApplied = checkAuthorization(requesterIdentity,
requestContext, httpHeaders, compiledQuery);
// Apply broker-default query options before branching to
EXPLAIN/execute so both paths see the same options.
applyBrokerDefaultQueryOptions(compiledQuery.getOptions());
@@ -481,7 +479,7 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
return explain(compiledQuery, requestId, requestContext, queryTimer,
idSetSubqueries);
} else {
return query(compiledQuery, requestId, requesterIdentity,
requestContext, httpHeaders, queryTimer,
- rlsFiltersApplied.get(), queryWasLogged, workloadName);
+ rlsFiltersApplied, queryWasLogged, workloadName);
}
}
}
@@ -518,8 +516,8 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
}
}
- private void checkAuthorization(RequesterIdentity requesterIdentity,
RequestContext requestContext,
- HttpHeaders httpHeaders, QueryEnvironment.CompiledQuery compiledQuery,
AtomicBoolean rlsFiltersApplied) {
+ private boolean checkAuthorization(RequesterIdentity requesterIdentity,
RequestContext requestContext,
+ HttpHeaders httpHeaders, QueryEnvironment.CompiledQuery compiledQuery) {
Set<String> tables = compiledQuery.getTableNames();
if (tables != null && !tables.isEmpty()) {
TableAuthorizationResult tableAuthorizationResult =
@@ -527,21 +525,30 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
if (!tableAuthorizationResult.hasAccess()) {
throwTableAccessError(tableAuthorizationResult);
}
- if (_enableRowColumnLevelAuth) {
- AccessControl accessControl = _accessControlFactory.create();
- for (String tableName : tables) {
- accessControl.getRowColFilters(requesterIdentity,
tableName).getRLSFilters()
- .ifPresent(rowFilters -> {
- rlsFiltersApplied.set(true);
- String combinedFilters =
- rowFilters.stream().map(filter -> "( " + filter + "
)").collect(Collectors.joining(" AND "));
- String key = RlsUtils.buildRlsFilterKey(tableName);
- compiledQuery.getOptions().put(key, combinedFilters);
- _brokerMetrics.addMeteredTableValue(tableName,
BrokerMeter.RLS_FILTERS_APPLIED, 1);
- });
- }
+ return _enableRowColumnLevelAuth && applyRlsFilters(requesterIdentity,
tables, compiledQuery.getOptions());
+ }
+ return false;
+ }
+
+ /// Returns `true` if an RLS filter was added for at least one table.
+ @VisibleForTesting
+ boolean applyRlsFilters(@Nullable RequesterIdentity requesterIdentity,
Set<String> tables,
+ Map<String, String> queryOptions) {
+ AccessControl accessControl = _accessControlFactory.create();
+ boolean applied = false;
+ for (String tableName : tables) {
+ List<String> rowFilters =
accessControl.getRowColFilters(requesterIdentity, tableName).getRLSFilters()
+ .orElse(null);
+ if (rowFilters == null || rowFilters.isEmpty()) {
+ continue;
}
+ String combinedFilters =
+ rowFilters.stream().map(filter -> "( " + filter + "
)").collect(Collectors.joining(" AND "));
+ queryOptions.put(RlsUtils.buildRlsFilterKey(tableName), combinedFilters);
+ _brokerMetrics.addMeteredTableValue(tableName,
BrokerMeter.RLS_FILTERS_APPLIED, 1);
+ applied = true;
}
+ return applied;
}
private ImmutableQueryEnvironment.Config getQueryEnvConf(HttpHeaders
httpHeaders, Map<String, String> queryOptions,
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java
index b89ae2fc01b..a7cf7bc1699 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java
@@ -27,6 +27,7 @@ import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nullable;
import javax.ws.rs.core.HttpHeaders;
import org.apache.pinot.broker.api.AccessControl;
+import org.apache.pinot.broker.broker.AccessControlFactory;
import org.apache.pinot.broker.broker.AllowAllAccessControlFactory;
import org.apache.pinot.broker.queryquota.QueryQuotaManager;
import org.apache.pinot.common.config.provider.TableCache;
@@ -56,6 +57,7 @@ import
org.apache.pinot.query.runtime.plan.MultiStageQueryStats;
import org.apache.pinot.query.runtime.plan.pipeline.PipelineBreakerOperator;
import org.apache.pinot.query.service.dispatch.QueryDispatcher;
import org.apache.pinot.spi.accounting.ThreadAccountantUtils;
+import org.apache.pinot.spi.auth.TableRowColAccessResultImpl;
import org.apache.pinot.spi.auth.broker.RequesterIdentity;
import org.apache.pinot.spi.env.PinotConfiguration;
import
org.apache.pinot.spi.eventlistener.query.BrokerQueryEventListenerFactory;
@@ -73,6 +75,7 @@ import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.NetUtils;
import org.apache.pinot.sql.parsers.CalciteSqlParser;
import org.apache.pinot.sql.parsers.SqlNodeAndOptions;
+import org.apache.pinot.sql.parsers.rewriter.RlsUtils;
import org.apache.pinot.util.TestUtils;
import org.mockito.ArgumentCaptor;
import org.mockito.MockedConstruction;
@@ -83,6 +86,7 @@ import org.testng.annotations.Test;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
@@ -470,6 +474,14 @@ public class MultiStageBrokerRequestHandlerTest extends
QueryEnvironmentTestBase
private static MultiStageBrokerRequestHandler newHandler(PinotConfiguration
config, RoutingManager routingManager,
FailureDetector failureDetector, @Nullable MultiClusterRoutingContext
multiClusterRoutingContext)
throws Exception {
+ return newHandler(config, routingManager, failureDetector,
multiClusterRoutingContext,
+ new AllowAllAccessControlFactory());
+ }
+
+ private static MultiStageBrokerRequestHandler newHandler(PinotConfiguration
config, RoutingManager routingManager,
+ FailureDetector failureDetector, @Nullable MultiClusterRoutingContext
multiClusterRoutingContext,
+ AccessControlFactory accessControlFactory)
+ throws Exception {
config.setProperty(MultiStageQueryRunner.KEY_OF_QUERY_RUNNER_HOSTNAME,
"localhost");
config.setProperty(MultiStageQueryRunner.KEY_OF_QUERY_RUNNER_PORT,
Integer.toString(NetUtils.findOpenPort()));
BrokerQueryEventListenerFactory.init(config);
@@ -481,7 +493,7 @@ public class MultiStageBrokerRequestHandlerTest extends
QueryEnvironmentTestBase
when(queryQuotaManager.acquireApplication(anyString())).thenReturn(true);
return new MultiStageBrokerRequestHandler(config, "testBrokerId", new
BrokerRequestIdGenerator(),
- routingManager, new AllowAllAccessControlFactory(), queryQuotaManager,
+ routingManager, accessControlFactory, queryQuotaManager,
mock(TableCache.class), mock(MultiStageQueryThrottler.class),
failureDetector,
ThreadAccountantUtils.getNoOpAccountant(), multiClusterRoutingContext,
mock(WorkerManager.class),
mock(WorkerManager.class)) {
@@ -496,6 +508,30 @@ public class MultiStageBrokerRequestHandlerTest extends
QueryEnvironmentTestBase
};
}
+ /// An access control can return an empty RLS filter list for a table
without row filters. No predicate is attached
+ /// for it, so the response must not report that RLS filters were applied.
+ @Test
+ public void testApplyRlsFiltersIgnoresEmptyFilterList()
+ throws Exception {
+ AccessControl accessControl = mock(AccessControl.class);
+ when(accessControl.getRowColFilters(any(), eq("empty"))).thenReturn(new
TableRowColAccessResultImpl(List.of()));
+ when(accessControl.getRowColFilters(any(),
eq("absent"))).thenReturn(TableRowColAccessResultImpl.unrestricted());
+ when(accessControl.getRowColFilters(any(), eq("filtered"))).thenReturn(
+ new TableRowColAccessResultImpl(List.of("col1 = 'a'", "col2 = 'b'")));
+ AccessControlFactory accessControlFactory =
mock(AccessControlFactory.class);
+ when(accessControlFactory.create()).thenReturn(accessControl);
+ MultiStageBrokerRequestHandler handler = newHandler(new
PinotConfiguration(), mock(RoutingManager.class),
+ mock(FailureDetector.class), null, accessControlFactory);
+
+ Map<String, String> queryOptions = new HashMap<>();
+ Assert.assertFalse(handler.applyRlsFilters(null, Set.of("empty",
"absent"), queryOptions));
+ assertEquals(queryOptions, Map.of());
+
+ Assert.assertTrue(handler.applyRlsFilters(null, Set.of("empty", "absent",
"filtered"), queryOptions));
+ assertEquals(queryOptions,
+ Map.of(RlsUtils.buildRlsFilterKey("filtered"), "( col1 = 'a' ) AND (
col2 = 'b' )"));
+ }
+
/// The failure detector keeps an instance unhealthy while any retrier
returns UNHEALTHY for it. A multi-stage worker
/// that is not an enabled server of the routing manager, for example a
server of a remote cluster, must recover
/// once the dispatch channel to it is READY, with the retriers of all
request handlers registered.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
