Operator conversion deny list (#13766)

### Description

This change adds a new config property `druid.sql.planner.operatorConversion.denyList`, which allows a user to specify
any operator conversions that they wish to disallow. A user may want to do this for a number of reasons, including security concerns. The default value of this property is the empty list `[]`, which does not disallow any operator conversions.

An example usage of this property is `druid.sql.planner.operatorConversion.denyList=["extern"]`, which disallows the usage of the `extern` operator conversion. If the property is configured this way, and a user of the Druid cluster tries to submit a query that uses the `extern` function, such as the example given [here](https://druid.apache.org/docs/latest/multi-stage-query/examples.html#insert-with-no-rollup), a response with http response code `400` is returned with en error body similar to the following:

```
{
  "taskId": "4ec5b0b6-fa9b-4c3a-827d-2308294e9985",
  "state": "FAILED",
  "error": {
    "error": "Plan validation failed",
    "errorMessage": "org.apache.calcite.runtime.CalciteContextException: From line 28, column 5 to line 32, column 5: No match found for function signature EXTERN(<CHARACTER>, <CHARACTER>, <CHARACTER>)",
    "errorClass": "org.apache.calcite.tools.ValidationException",
    "host": null
  }
}
```

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

@@ -54,6 +54,7 @@ import org.apache.druid.sql.calcite.planner.DruidOperatorTable;
 import org.apache.druid.sql.calcite.planner.DruidPlanner;
 import org.apache.druid.sql.calcite.planner.PlannerConfig;
 import org.apache.druid.sql.calcite.planner.PlannerFactory;
+import org.apache.druid.sql.calcite.planner.PlannerOperatorConfig;
 import org.apache.druid.sql.calcite.planner.PlannerResult;
 import org.apache.druid.sql.calcite.run.SqlEngine;
 import org.apache.druid.sql.calcite.schema.DruidSchemaCatalog;
@@ -530,7 +531,7 @@ public class SqlBenchmark
           new ApproxCountDistinctSqlAggregator(new HllSketchApproxCountDistinctSqlAggregator());
       aggregators.add(new CountSqlAggregator(countDistinctSqlAggregator));
       aggregators.add(countDistinctSqlAggregator);
-      return new DruidOperatorTable(aggregators, extractionOperators);
+      return new DruidOperatorTable(aggregators, extractionOperators, PlannerOperatorConfig.newInstance(null));
     }
     catch (Exception e) {
       throw new RuntimeException(e);

docs/configuration/index.md

@@ -1921,6 +1921,7 @@ The Druid SQL server is configured through the following properties on the Broke
 |`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.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.planner.operator.denyList`|The list of operators that should be denied.|`[]` (no operators are disallowed)|
 |`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|
 
 > Previous versions of Druid had properties named `druid.sql.planner.maxQueryCount` and `druid.sql.planner.maxSemiJoinRowsInMemory`.

sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlannerModule.java

@@ -40,6 +40,7 @@ public class CalcitePlannerModule implements Module
     // We're actually binding the class to the config prefix, not the other way around.
     JsonConfigProvider.bind(binder, "druid.sql.planner", PlannerConfig.class);
     JsonConfigProvider.bind(binder, "druid.sql.planner", SegmentMetadataCacheConfig.class);
+    JsonConfigProvider.bind(binder, PlannerOperatorConfig.CONFIG_PATH, PlannerOperatorConfig.class);
     binder.bind(PlannerFactory.class).in(LazySingleton.class);
     binder.bind(DruidOperatorTable.class).in(LazySingleton.class);
     Multibinder.newSetBinder(binder, ExtensionCalciteRuleProvider.class);

sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidOperatorTable.java

@@ -21,6 +21,7 @@ package org.apache.druid.sql.calcite.planner;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
 import com.google.inject.Inject;
 import org.apache.calcite.sql.SqlAggFunction;
 import org.apache.calcite.sql.SqlFunctionCategory;
@@ -32,6 +33,7 @@ import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.validate.SqlNameMatcher;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.sql.calcite.aggregation.SqlAggregator;
 import org.apache.druid.sql.calcite.aggregation.builtin.ArrayConcatSqlAggregator;
 import org.apache.druid.sql.calcite.aggregation.builtin.ArraySqlAggregator;
@@ -133,6 +135,7 @@ import java.util.stream.Collectors;
 
 public class DruidOperatorTable implements SqlOperatorTable
 {
+  private static final Logger LOG = new Logger(DruidOperatorTable.class);
   // COUNT and APPROX_COUNT_DISTINCT are not here because they are added by SqlAggregationModule.
   private static final List<SqlAggregator> STANDARD_AGGREGATORS =
       ImmutableList.<SqlAggregator>builder()
@@ -415,11 +418,13 @@ public class DruidOperatorTable implements SqlOperatorTable
   @Inject
   public DruidOperatorTable(
       final Set<SqlAggregator> aggregators,
-      final Set<SqlOperatorConversion> operatorConversions
+      final Set<SqlOperatorConversion> operatorConversions,
+      final PlannerOperatorConfig plannerOperatorConfig
   )
   {
     this.aggregators = new HashMap<>();
     this.operatorConversions = new HashMap<>();
+    Set<String> operationConversionDenySet = ImmutableSet.copyOf(plannerOperatorConfig.getDenyList());
 
     for (SqlAggregator aggregator : aggregators) {
       final OperatorKey operatorKey = OperatorKey.of(aggregator.calciteFunction());
@@ -437,6 +442,12 @@ public class DruidOperatorTable implements SqlOperatorTable
 
     for (SqlOperatorConversion operatorConversion : operatorConversions) {
       final OperatorKey operatorKey = OperatorKey.of(operatorConversion.calciteOperator());
+      if (operationConversionDenySet.contains(operatorKey.name)) {
+        LOG.info(
+            "Operator %s is not available as it is in the deny list.",
+            operatorKey.name);
+        continue;
+      }
       if (this.aggregators.containsKey(operatorKey)
           || this.operatorConversions.put(operatorKey, operatorConversion) != null) {
         throw new ISE("Cannot have two operators with key [%s]", operatorKey);

sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerOperatorConfig.java

@@ -0,0 +1,76 @@
+/*
+ * 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.druid.sql.calcite.planner;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.ImmutableList;
+
+import javax.validation.constraints.NotNull;
+import java.util.List;
+import java.util.Objects;
+
+public class PlannerOperatorConfig
+{
+  public static final String CONFIG_PATH = "druid.sql.planner.operator";
+  @JsonProperty
+  private List<String> denyList;
+
+
+  @NotNull
+  public List<String> getDenyList()
+  {
+    return denyList == null ? ImmutableList.of() : denyList;
+  }
+
+  @Override
+  public boolean equals(final Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    final PlannerOperatorConfig that = (PlannerOperatorConfig) o;
+    return denyList.equals(that.denyList);
+  }
+
+  @Override
+  public int hashCode()
+  {
+
+    return Objects.hash(denyList);
+  }
+
+  @Override
+  public String toString()
+  {
+    return "PlannerOperatorConfig{" +
+           "denyList=" + denyList +
+           '}';
+  }
+
+  public static PlannerOperatorConfig newInstance(List<String> denyList)
+  {
+    PlannerOperatorConfig config = new PlannerOperatorConfig();
+    config.denyList = denyList;
+    return config;
+  }
+}

sql/src/test/java/org/apache/druid/sql/calcite/planner/DruidOperatorTableTest.java

@@ -0,0 +1,113 @@
+/*
+ * 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.druid.sql.calcite.planner;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import org.apache.druid.catalog.model.TableDefnRegistry;
+import org.apache.druid.sql.calcite.expression.SqlOperatorConversion;
+import org.apache.druid.sql.calcite.external.ExternalOperatorConversion;
+import org.apache.druid.sql.calcite.external.HttpOperatorConversion;
+import org.apache.druid.sql.calcite.external.InlineOperatorConversion;
+import org.apache.druid.sql.calcite.external.LocalOperatorConversion;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class DruidOperatorTableTest
+{
+  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+  private final TableDefnRegistry tableDefnRegistry = new TableDefnRegistry(OBJECT_MAPPER);
+
+  private DruidOperatorTable operatorTable;
+
+  @Test
+  public void test_operatorTable_with_operatorConversionDenyList_deniedConversionsUnavailable()
+  {
+    final ExternalOperatorConversion externalOperatorConversion =
+        new ExternalOperatorConversion(OBJECT_MAPPER);
+    final HttpOperatorConversion httpOperatorConversion =
+        new HttpOperatorConversion(tableDefnRegistry);
+    final InlineOperatorConversion inlineOperatorConversion =
+        new InlineOperatorConversion(tableDefnRegistry);
+    final LocalOperatorConversion localOperatorConversion =
+        new LocalOperatorConversion(tableDefnRegistry);
+    operatorTable = new DruidOperatorTable(
+        ImmutableSet.of(),
+        ImmutableSet.of(
+            externalOperatorConversion,
+            httpOperatorConversion,
+            inlineOperatorConversion,
+            localOperatorConversion
+        ),
+        PlannerOperatorConfig.newInstance(ImmutableList.of("extern", "http", "inline", "localfiles"))
+    );
+
+    SqlOperatorConversion operatorConversion =
+        operatorTable.lookupOperatorConversion(externalOperatorConversion.calciteOperator());
+    Assert.assertNull(operatorConversion);
+
+    operatorConversion = operatorTable.lookupOperatorConversion(httpOperatorConversion.calciteOperator());
+    Assert.assertNull(operatorConversion);
+
+    operatorConversion = operatorTable.lookupOperatorConversion(inlineOperatorConversion.calciteOperator());
+    Assert.assertNull(operatorConversion);
+
+    operatorConversion = operatorTable.lookupOperatorConversion(localOperatorConversion.calciteOperator());
+    Assert.assertNull(operatorConversion);
+  }
+
+  @Test
+  public void test_operatorTable_with_emptyOperatorConversionDenyList_conversionsAavailable()
+  {
+    final ExternalOperatorConversion externalOperatorConversion =
+        new ExternalOperatorConversion(OBJECT_MAPPER);
+    final HttpOperatorConversion httpOperatorConversion =
+        new HttpOperatorConversion(tableDefnRegistry);
+    final InlineOperatorConversion inlineOperatorConversion =
+        new InlineOperatorConversion(tableDefnRegistry);
+    final LocalOperatorConversion localOperatorConversion =
+        new LocalOperatorConversion(tableDefnRegistry);
+    operatorTable = new DruidOperatorTable(
+        ImmutableSet.of(),
+        ImmutableSet.of(
+            externalOperatorConversion,
+            httpOperatorConversion,
+            inlineOperatorConversion,
+            localOperatorConversion
+        ),
+        PlannerOperatorConfig.newInstance(null)
+    );
+
+    SqlOperatorConversion operatorConversion =
+        operatorTable.lookupOperatorConversion(externalOperatorConversion.calciteOperator());
+    Assert.assertEquals(externalOperatorConversion, operatorConversion);
+
+    operatorConversion = operatorTable.lookupOperatorConversion(httpOperatorConversion.calciteOperator());
+    Assert.assertEquals(httpOperatorConversion, operatorConversion);
+
+    operatorConversion = operatorTable.lookupOperatorConversion(inlineOperatorConversion.calciteOperator());
+    Assert.assertEquals(inlineOperatorConversion, operatorConversion);
+
+    operatorConversion = operatorTable.lookupOperatorConversion(localOperatorConversion.calciteOperator());
+    Assert.assertEquals(localOperatorConversion, operatorConversion);
+  }
+}

sql/src/test/java/org/apache/druid/sql/calcite/planner/DruidRexExecutorTest.java

@@ -83,7 +83,8 @@ public class DruidRexExecutorTest extends InitializedNullHandlingTest
       "SELECT 1", // The actual query isn't important for this test
       new DruidOperatorTable(
           Collections.emptySet(),
-          ImmutableSet.of(new DirectOperatorConversion(OPERATOR, "hyper_unique"))
+          ImmutableSet.of(new DirectOperatorConversion(OPERATOR, "hyper_unique")),
+          PlannerOperatorConfig.newInstance(null)
       ),
       CalciteTests.createExprMacroTable(),
       CalciteTests.getJsonMapper(),

sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerOperatorConfigTest.java

@@ -0,0 +1,34 @@
+/*
+ * 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.druid.sql.calcite.planner;
+
+import nl.jqno.equalsverifier.EqualsVerifier;
+import org.junit.Test;
+
+public class PlannerOperatorConfigTest
+{
+  @Test
+  public void testEquals()
+  {
+    EqualsVerifier.simple().forClass(PlannerOperatorConfig.class)
+                  .usingGetClass()
+                  .verify();
+  }
+}