feat: User-aware SchemaProviders for SQL.

benchmarks/src/test/java/org/apache/druid/benchmark/query/InPlanningBenchmark.java

@@ -53,7 +53,7 @@
 import org.apache.druid.sql.calcite.planner.PlannerFactory;
 import org.apache.druid.sql.calcite.planner.PlannerResult;
 import org.apache.druid.sql.calcite.run.SqlEngine;
-import org.apache.druid.sql.calcite.schema.DruidSchemaCatalog;
+import org.apache.druid.sql.calcite.schema.DruidSchemaCatalogProvider;
 import org.apache.druid.sql.calcite.util.CalciteTests;
 import org.apache.druid.sql.hook.DruidHookDispatcher;
 import org.apache.druid.timeline.DataSegment;
@@ -186,13 +186,17 @@ public void setup() throws JsonProcessingException
     );
     closer.register(walker);
     final ObjectMapper jsonMapper = CalciteTests.getJsonMapper();
-    final DruidSchemaCatalog rootSchema =
-        CalciteTests.createMockRootSchema(conglomerate, walker, plannerConfig, AuthTestUtils.TEST_AUTHORIZER_MAPPER);
+    final DruidSchemaCatalogProvider rootSchemaProvider = CalciteTests.createMockRootSchemaProvider(
+        conglomerate,
+        walker,
+        plannerConfig,
+        AuthTestUtils.TEST_AUTHORIZER_MAPPER
+    );
 
     engine = CalciteTests.createMockSqlEngine(walker, conglomerate);
 
     plannerFactory = new PlannerFactory(
-        rootSchema,
+        rootSchemaProvider,
         CalciteTests.createOperatorTable(),
         CalciteTests.createExprMacroTable(),
         plannerConfig,

benchmarks/src/test/java/org/apache/druid/benchmark/query/SqlBaseBenchmark.java

@@ -100,7 +100,7 @@
 import org.apache.druid.sql.calcite.planner.PlannerFactory;
 import org.apache.druid.sql.calcite.planner.PlannerResult;
 import org.apache.druid.sql.calcite.run.SqlEngine;
-import org.apache.druid.sql.calcite.schema.DruidSchemaCatalog;
+import org.apache.druid.sql.calcite.schema.DruidSchemaCatalogProvider;
 import org.apache.druid.sql.calcite.util.CalciteTests;
 import org.apache.druid.sql.calcite.util.LookylooModule;
 import org.apache.druid.sql.calcite.util.QueryFrameworkUtils;
@@ -480,8 +480,8 @@ public static Pair<PlannerFactory, SqlEngine> createSqlSystem(
     ObjectMapper injected = injector.getInstance(Key.get(ObjectMapper.class, Json.class));
     injected.registerModules(new HllSketchModule().getJacksonModules());
 
-    final DruidSchemaCatalog rootSchema =
-        QueryFrameworkUtils.createMockRootSchema(
+    final DruidSchemaCatalogProvider schemaProvider =
+        QueryFrameworkUtils.createMockRootSchemaProvider(
             injector,
             conglomerate,
             walker,
@@ -492,7 +492,7 @@ public static Pair<PlannerFactory, SqlEngine> createSqlSystem(
     final SqlEngine engine = CalciteTests.createMockSqlEngine(walker, conglomerate);
 
     final PlannerFactory plannerFactory = new PlannerFactory(
-        rootSchema,
+        schemaProvider,
         createOperatorTable(injector),
         injector.getInstance(ExprMacroTable.class),
         plannerConfig,

benchmarks/src/test/java/org/apache/druid/benchmark/query/SqlVsNativeBenchmark.java

@@ -48,7 +48,7 @@
 import org.apache.druid.sql.calcite.planner.PlannerFactory;
 import org.apache.druid.sql.calcite.planner.PlannerResult;
 import org.apache.druid.sql.calcite.run.SqlEngine;
-import org.apache.druid.sql.calcite.schema.DruidSchemaCatalog;
+import org.apache.druid.sql.calcite.schema.DruidSchemaCatalogProvider;
 import org.apache.druid.sql.calcite.util.CalciteTests;
 import org.apache.druid.sql.hook.DruidHookDispatcher;
 import org.apache.druid.timeline.DataSegment;
@@ -115,11 +115,15 @@ public void setup()
     final PlannerConfig plannerConfig = new PlannerConfig();
 
     this.walker = closer.register(SpecificSegmentsQuerySegmentWalker.createWalker(conglomerate).add(dataSegment, index));
-    final DruidSchemaCatalog rootSchema =
-        CalciteTests.createMockRootSchema(conglomerate, walker, plannerConfig, AuthTestUtils.TEST_AUTHORIZER_MAPPER);
+    final DruidSchemaCatalogProvider rootSchemaProvider = CalciteTests.createMockRootSchemaProvider(
+        conglomerate,
+        walker,
+        plannerConfig,
+        AuthTestUtils.TEST_AUTHORIZER_MAPPER
+    );
     engine = CalciteTests.createMockSqlEngine(walker, conglomerate);
     plannerFactory = new PlannerFactory(
-        rootSchema,
+        rootSchemaProvider,
         CalciteTests.createOperatorTable(),
         CalciteTests.createExprMacroTable(),
         plannerConfig,

benchmarks/src/test/java/org/apache/druid/sql/calcite/schema/SysSegmentsTableBenchmark.java

@@ -35,6 +35,7 @@
 import org.apache.druid.client.InternalQueryConfig;
 import org.apache.druid.client.TimelineServerView;
 import org.apache.druid.client.coordinator.NoopCoordinatorClient;
+import org.apache.druid.error.NotYetImplemented;
 import org.apache.druid.jackson.DefaultObjectMapper;
 import org.apache.druid.java.util.common.CloseableIterators;
 import org.apache.druid.java.util.common.Intervals;
@@ -50,8 +51,6 @@
 import org.apache.druid.server.security.Authorizer;
 import org.apache.druid.server.security.AuthorizerMapper;
 import org.apache.druid.server.security.Escalator;
-import org.apache.druid.sql.calcite.planner.CatalogResolver;
-import org.apache.druid.sql.calcite.planner.PlannerContext;
 import org.apache.druid.timeline.DataSegment;
 import org.apache.druid.timeline.SegmentId;
 import org.apache.druid.timeline.SegmentStatusInCluster;
@@ -202,9 +201,6 @@ public ListenableFuture<CloseableIterator<SegmentStatusInCluster>> fetchAllUsedS
         new NoopServiceEmitter()
     );
 
-    final DruidSchema druidSchema =
-        new DruidSchema(new EmptyBrokerSegmentMetadataCache(), null, CatalogResolver.NULL_RESOLVER);
-
     final AuthorizerMapper authorizerMapper = new AuthorizerMapper(null)
     {
       @Override
@@ -214,11 +210,16 @@ public Authorizer getAuthorizer(String name)
       }
     };
 
-    segmentsTable = new SystemSchema.SegmentsTable(druidSchema, metadataView, new DefaultObjectMapper(), authorizerMapper);
+    segmentsTable = new SystemSchema.SegmentsTable(
+        new EmptyBrokerSegmentMetadataCache(),
+        metadataView,
+        new DefaultObjectMapper(),
+        authorizerMapper,
+        new AuthenticationResult("benchmark", "benchmark", null, null)
+    );
 
     filtersByQuery = buildFilters();
 
-    final AuthenticationResult authenticationResult = new AuthenticationResult("benchmark", "benchmark", null, null);
     dataContext = new DataContext()
     {
       @Override
@@ -242,7 +243,7 @@ public QueryProvider getQueryProvider()
       @Override
       public Object get(String name)
       {
-        return PlannerContext.DATA_CTX_AUTHENTICATION_RESULT.equals(name) ? authenticationResult : null;
+        throw NotYetImplemented.ex(null, "Not expected to be called");
       }
     };
   }

docs/configuration/index.md

@@ -1899,6 +1899,7 @@ The Druid SQL server is configured through the following properties on the Broke
 |`druid.sql.planner.metadataSegmentCacheEnable`|Whether to keep a cache of published segments on Broker that can be used to serve queries against `sys.segments`. If true, broker polls coordinator in background to get segments from metadata store and maintains a local cache. If false, coordinator's REST API will be invoked when broker needs published segments info.|true|
 |`druid.sql.planner.metadataSegmentPollPeriod`|How often to poll coordinator for published segments list if `druid.sql.planner.metadataSegmentCacheEnable` is set to true. Poll period is in milliseconds. |60000|
 |`druid.sql.planner.authorizeSystemTablesDirectly`|If true, Druid authorizes queries against any of the system schema tables (`sys` in SQL) as `SYSTEM_TABLE` resources which require `READ` access, in addition to permissions based content filtering.|false|
+|`druid.sql.planner.authorizeTableVisibility`|Whether [READ DATASOURCE](../multi-stage-query/security.md) permissions are required for table visibility in the validator. When this is set, users that query unauthorized tables see a "not found" error rather than "forbidden". Additionally, when this is set, INSERT and REPLACE require both READ and WRITE access to the target table. (If this property is not set, they require only WRITE.) Regardless of the value of this property, READ access is required for tables to show up in the INFORMATION_SCHEMA.|true|
 |`druid.sql.planner.useNativeQueryExplain`|If true, `EXPLAIN PLAN FOR` will return the explain plan as a JSON representation of equivalent native query(s), else it will return the original version of explain plan generated by Calcite. It can be overridden per query with `useNativeQueryExplain` context key.|true|
 |`druid.sql.planner.maxNumericInFilters`|Max limit for the amount of numeric values that can be compared for a string type dimension when the entire SQL WHERE clause of a query translates to an [OR](../querying/filters.md#or) of [Bound filter](../querying/filters.md#bound-filter). By default, Druid does not restrict the amount of numeric Bound Filters on String columns, although this situation may block other queries from running. Set this property to a smaller value to prevent Druid from running queries that have prohibitively long segment processing times. The optimal limit requires some trial and error; we recommend starting with 100.  Users who submit a query that exceeds the limit of `maxNumericInFilters` should instead rewrite their queries to use strings in the `WHERE` clause instead of numbers. For example, `WHERE someString IN (‘123’, ‘456’)`. If this value is disabled, `maxNumericInFilters` set through query context is ignored.|`-1` (disabled)|
 |`druid.sql.approxCountDistinct.function`|Implementation to use for the [`APPROX_COUNT_DISTINCT` function](../querying/sql-aggregations.md). Without extensions loaded, the only valid value is `APPROX_COUNT_DISTINCT_BUILTIN` (a HyperLogLog, or HLL, based implementation). If the [DataSketches extension](../development/extensions-core/datasketches-extension.md) is loaded, this can also be `APPROX_COUNT_DISTINCT_DS_HLL` (alternative HLL implementation) or `APPROX_COUNT_DISTINCT_DS_THETA`.<br /><br />Theta sketches use significantly more memory than HLL sketches, so you should prefer one of the two HLL implementations.|`APPROX_COUNT_DISTINCT_BUILTIN`|

docs/multi-stage-query/security.md

@@ -30,8 +30,10 @@ data. The permission needed depends on what the user is trying to do.
 To submit a query:
 
 - SELECT from a Druid datasource requires the READ DATASOURCE permission on that datasource.
-- [INSERT](reference.md#insert) or [REPLACE](reference.md#replace) into a Druid datasource requires the WRITE DATASOURCE
-  permission on that datasource.
+- [INSERT](reference.md#insert) or [REPLACE](reference.md#replace) into a Druid datasource require the WRITE DATASOURCE
+  and READ DATASOURCE permissions on the target datasource. (Special case: if
+  [`druid.sql.planner.authorizeTableVisibility = false`](../configuration/index.md#sql) is set, only WRITE DATASOURCE
+  is required.)
 - [EXTERN](reference.md#extern-function) and the input-source-specific table functions require READ permission on a
   resource named "EXTERNAL" with type "EXTERNAL". Users without the correct
   permission encounter a 403 error when trying to run queries that include `EXTERN`.

embedded-tests/src/test/java/org/apache/druid/testing/embedded/auth/AbstractAuthConfigurationTest.java

@@ -38,7 +38,6 @@
 import org.apache.druid.query.QueryContexts;
 import org.apache.druid.query.http.SqlTaskStatus;
 import org.apache.druid.segment.TestHelper;
-import org.apache.druid.server.security.Access;
 import org.apache.druid.server.security.Action;
 import org.apache.druid.server.security.Resource;
 import org.apache.druid.server.security.ResourceAction;
@@ -338,30 +337,32 @@ public void test_systemSchemaAccess_datasourceOnlyUser()
         getServerUrl(broker) + "/druid/v2/datasources/auth_test"
     );
 
-    // as user that can only read auth_test
-    final String expectedMsg = "{\"Access-Check-Result\":\"" + Access.DEFAULT_ERROR_MESSAGE + "\"}";
-    verifySystemSchemaQueryIsForbidden(
+    // As a user that can only read auth_test. This cluster runs with the default
+    // druid.sql.planner.authorizeTableVisibility = true, so the sys tables are not visible to this user at all and
+    // naming one is a validation error rather than an authorization error. (With authorizeTableVisibility = false
+    // these would instead be forbidden.)
+    verifySystemSchemaQueryIsNotFound(
         datasourceOnlyUserClient,
         SYS_SCHEMA_SEGMENTS_QUERY,
-        expectedMsg
+        "segments"
     );
 
-    verifySystemSchemaQueryIsForbidden(
+    verifySystemSchemaQueryIsNotFound(
         datasourceOnlyUserClient,
         SYS_SCHEMA_SERVERS_QUERY,
-        expectedMsg
+        "servers"
     );
 
-    verifySystemSchemaQueryIsForbidden(
+    verifySystemSchemaQueryIsNotFound(
         datasourceOnlyUserClient,
         SYS_SCHEMA_SERVER_SEGMENTS_QUERY,
-        expectedMsg
+        "server_segments"
     );
 
-    verifySystemSchemaQueryIsForbidden(
+    verifySystemSchemaQueryIsNotFound(
         datasourceOnlyUserClient,
         SYS_SCHEMA_TASKS_QUERY,
-        expectedMsg
+        "tasks"
     );
   }
 
@@ -874,6 +875,28 @@ private void verifySystemSchemaQueryIsForbidden(
     Assertions.assertEquals(responseHolder.getContent(), expectedErrorMessage);
   }
 
+  /**
+   * Verifies that a sys table is not visible to the user at all, i.e. that naming it is a validation error rather
+   * than an authorization error. This is the behavior when {@code druid.sql.planner.authorizeTableVisibility} is
+   * true (the default) and the user lacks READ on the corresponding SYSTEM_TABLE resource.
+   */
+  private void verifySystemSchemaQueryIsNotFound(
+      HttpClient client,
+      String query,
+      String expectedMissingTable
+  )
+  {
+    final StatusResponseHolder responseHolder =
+        makeSQLQueryRequest(client, query, HttpResponseStatus.BAD_REQUEST);
+    Assertions.assertEquals(HttpResponseStatus.BAD_REQUEST, responseHolder.getStatus());
+
+    final String content = responseHolder.getContent();
+    Assertions.assertTrue(
+        content.contains("Object '" + expectedMissingTable + "' not found within 'sys'"),
+        StringUtils.format("Expected [%s] to be not found within 'sys', but got[%s]", expectedMissingTable, content)
+    );
+  }
+
   protected String getBrokerAvacticaUrl()
   {
     return "jdbc:avatica:remote:url=" + getServerUrl(broker) + DruidAvaticaJsonHandler.AVATICA_PATH;

embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/JdbcQueryTest.java

@@ -99,8 +99,8 @@ public void testJdbcMetadata()
           schemas.add(schema);
         }
         LOG.info("'druid' catalog schemas %s", schemas);
-        // maybe more schemas than this, but at least should have these
-        Assertions.assertTrue(schemas.containsAll(ImmutableList.of("INFORMATION_SCHEMA", "druid", "lookup", "sys")));
+        // Minimum set of schemas.
+        Assertions.assertTrue(schemas.containsAll(ImmutableList.of("INFORMATION_SCHEMA", "druid", "sys")));
 
         Set<String> druidTables = new HashSet<>();
         ResultSet tablesMetadata = metadata.getTables("druid", "druid", null, null);

extensions-contrib/grpc-query/src/test/java/org/apache/druid/grpc/BasicAuthTest.java

@@ -40,6 +40,7 @@
 import org.apache.druid.sql.calcite.BaseCalciteQueryTest;
 import org.apache.druid.sql.calcite.util.CalciteTests;
 import org.apache.druid.sql.calcite.util.SqlTestFramework;
+import org.hamcrest.CoreMatchers;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
@@ -48,6 +49,7 @@
 import java.io.IOException;
 import java.util.Map;
 
+import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
@@ -234,8 +236,12 @@ public void testUnauthorized()
                                        .build();
 
     try (TestClient client = new TestClient(TestClient.DEFAULT_HOST, "regular", "pwd")) {
-      StatusRuntimeException e = assertThrows(StatusRuntimeException.class, () -> client.getQueryClient().submitQuery(request));
-      assertEquals(Status.PERMISSION_DENIED, e.getStatus());
+      QueryResponse response = client.getQueryClient().submitQuery(request);
+      assertEquals(QueryStatus.RUNTIME_ERROR, response.getStatus());
+      assertThat(
+          response.getErrorMessage(),
+          CoreMatchers.startsWith("Object 'forbiddenDatasource' not found")
+      );
     }
   }
 }

multi-stage-query/src/main/java/org/apache/druid/msq/sql/MSQTaskSqlEngine.java

@@ -206,11 +206,7 @@ public QueryMaker buildQueryMakerForInsert(
     validateInsert(
         relRoot,
         destination instanceof TableDestination
-        ? plannerContext.getPlannerToolbox()
-                        .rootSchema()
-                        .getNamedSchema(plannerContext.getPlannerToolbox().druidSchemaName())
-                        .getSchema()
-                        .getTable(((TableDestination) destination).getTableName())
+        ? plannerContext.getDruidTable(((TableDestination) destination).getTableName())
         : null,
         plannerContext
     );

multi-stage-query/src/test/java/org/apache/druid/msq/dart/controller/http/DartSqlResourceTest.java

@@ -72,14 +72,13 @@
 import org.apache.druid.server.mocks.MockHttpServletResponse;
 import org.apache.druid.server.security.AuthConfig;
 import org.apache.druid.server.security.AuthenticationResult;
-import org.apache.druid.server.security.ForbiddenException;
 import org.apache.druid.sql.SqlLifecycleManager;
 import org.apache.druid.sql.SqlToolbox;
 import org.apache.druid.sql.calcite.planner.CalciteRulesManager;
 import org.apache.druid.sql.calcite.planner.CatalogResolver;
 import org.apache.druid.sql.calcite.planner.PlannerConfig;
 import org.apache.druid.sql.calcite.planner.PlannerFactory;
-import org.apache.druid.sql.calcite.schema.DruidSchemaCatalog;
+import org.apache.druid.sql.calcite.schema.DruidSchemaCatalogProvider;
 import org.apache.druid.sql.calcite.schema.NoopDruidSchemaManager;
 import org.apache.druid.sql.calcite.util.CalciteTests;
 import org.apache.druid.sql.calcite.util.QueryFrameworkUtils;
@@ -186,7 +185,7 @@ void setUp()
   {
     mockCloser = MockitoAnnotations.openMocks(this);
 
-    final DruidSchemaCatalog rootSchema = QueryFrameworkUtils.createMockRootSchema(
+    final DruidSchemaCatalogProvider schemaProvider = QueryFrameworkUtils.createMockRootSchemaProvider(
         CalciteTests.INJECTOR,
         queryFramework().conglomerate(),
         queryFramework().walker(),
@@ -198,7 +197,7 @@ void setUp()
     );
 
     final PlannerFactory plannerFactory = new PlannerFactory(
-        rootSchema,
+        schemaProvider,
         queryFramework().operatorTable(),
         queryFramework().macroTable(),
         PLANNER_CONFIG_DEFAULT,
@@ -560,7 +559,7 @@ public void test_doPost_informationSchema()
     Assertions.assertNull(sqlResource.doPost(sqlQuery, httpServletRequest));
     Assertions.assertEquals(Response.Status.OK.getStatusCode(), asyncResponse.getStatus());
     Assertions.assertEquals(
-        "[[\"INFORMATION_SCHEMA\"],[\"druid\"],[\"lookup\"],[\"sys\"],[\"view\"]]\n",
+        "[[\"INFORMATION_SCHEMA\"],[\"druid\"],[\"lookup\"],[\"sys\"]]\n",
         StringUtils.fromUtf8(asyncResponse.baos.toByteArray())
     );
   }
@@ -596,7 +595,7 @@ public void test_doPost_sysTableJoinedToDatasource()
   }
 
   @Test
-  public void test_doPost_regularUser_forbidden()
+  public void test_doPost_regularUser_unauthorizedTable()
   {
     final MockAsyncContext asyncContext = new MockAsyncContext();
     final MockHttpServletResponse asyncResponse = new MockHttpServletResponse();
@@ -617,9 +616,20 @@ public void test_doPost_regularUser_forbidden()
         Collections.emptyList()
     );
 
-    Assertions.assertThrows(
-        ForbiddenException.class,
-        () -> sqlResource.doPost(sqlQuery, httpServletRequest)
+    // 400 Bad Request: the table is not visible to this user, so it cannot be resolved.
+    final Response response = sqlResource.doPost(sqlQuery, httpServletRequest);
+    Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), response.getStatus());
+
+    final Map<String, Object> e = objectMapper.convertValue(
+        response.getEntity(),
+        JacksonUtils.TYPE_REFERENCE_MAP_STRING_OBJECT
+    );
+
+    Assertions.assertEquals("invalidInput", e.get("errorCode"));
+    Assertions.assertEquals("INVALID_INPUT", e.get("category"));
+    assertThat(
+        (String) e.get("errorMessage"),
+        CoreMatchers.startsWith("Object 'forbiddenDatasource' not found")
     );
   }