fix: do not allow implicit casting in UDAF function lookup (#5145)

confluentinc · Apr 23, 2020 · 6709010 · 6709010
1 parent d547eca
commit 6709010
Show file tree

Hide file tree

Showing 5 changed files with 43 additions and 6 deletions.
diff --git a/ksqldb-common/src/main/java/io/confluent/ksql/function/TableFunctionFactory.java b/ksqldb-common/src/main/java/io/confluent/ksql/function/TableFunctionFactory.java
@@ -31,7 +31,7 @@ public class TableFunctionFactory {
 
   public TableFunctionFactory(final UdfMetadata metadata) {
     this.metadata = Objects.requireNonNull(metadata, "metadata");
-    this.udtfIndex = new UdfIndex<>(metadata.getName());
+    this.udtfIndex = new UdfIndex<>(metadata.getName(), true);
   }
 
   public UdfMetadata getMetadata() {

diff --git a/ksqldb-common/src/main/java/io/confluent/ksql/function/UdfFactory.java b/ksqldb-common/src/main/java/io/confluent/ksql/function/UdfFactory.java
@@ -33,7 +33,7 @@ public class UdfFactory {
              final UdfMetadata metadata) {
     this.udfClass = Objects.requireNonNull(udfClass, "udfClass can't be null");
     this.metadata = Objects.requireNonNull(metadata, "metadata can't be null");
-    this.udfIndex = new UdfIndex<>(metadata.getName());
+    this.udfIndex = new UdfIndex<>(metadata.getName(), true);
   }
 
   synchronized void addFunction(final KsqlScalarFunction ksqlFunction) {

diff --git a/ksqldb-common/src/main/java/io/confluent/ksql/function/UdfIndex.java b/ksqldb-common/src/main/java/io/confluent/ksql/function/UdfIndex.java
@@ -92,10 +92,12 @@ public class UdfIndex<T extends FunctionSignature> {
   private final String udfName;
   private final Node root = new Node();
   private final Map<List<ParamType>, T> allFunctions;
+  private final boolean supportsImplicitCasts;
 
-  UdfIndex(final String udfName) {
+  UdfIndex(final String udfName, final boolean supportsImplicitCasts) {
     this.udfName = Objects.requireNonNull(udfName, "udfName");
-    allFunctions = new HashMap<>();
+    this.allFunctions = new HashMap<>();
+    this.supportsImplicitCasts = supportsImplicitCasts;
   }
 
   void addFunction(final T function) {
@@ -152,6 +154,8 @@ T getFunction(final List<SqlType> arguments) {
 
     if (fun.isPresent()) {
       return fun.get();
+    } else if (!supportsImplicitCasts) {
+      throw createNoMatchingFunctionException(arguments);
     }
 
     // if none were found (candidates is empty) try again with

diff --git a/ksqldb-common/src/test/java/io/confluent/ksql/function/UdfIndexTest.java b/ksqldb-common/src/test/java/io/confluent/ksql/function/UdfIndexTest.java
@@ -59,7 +59,7 @@ public class UdfIndexTest {
 
   @Before
   public void setUp() {
-    udfIndex = new UdfIndex<>("name");
+    udfIndex = new UdfIndex<>("name", true);
   }
 
   @Test
@@ -741,6 +741,39 @@ public void shouldIncludeAvailableSignaturesIfNotMatchFound() {
         + "other(VARCHAR paramName, INT paramName)"));
   }
 
+  @Test
+  public void shouldSupportMatchAndImplicitCastEnabled() {
+    // Given:
+    givenFunctions(
+            function(EXPECTED, false, DOUBLE)
+    );
+
+    // When:
+    final KsqlFunction fun = udfIndex.getFunction(ImmutableList.of(INTEGER));
+
+    // Then:
+    assertThat(fun.name(), equalTo(EXPECTED));
+  }
+
+  @Test
+  public void shouldThrowIfNoExactMatchAndImplicitCastDisabled() {
+    // Given:
+    udfIndex = new UdfIndex<>("name", false);
+    givenFunctions(
+            function(OTHER, false, DOUBLE)
+    );
+
+    // When:
+    final Exception e = assertThrows(
+            KsqlException.class,
+            () -> udfIndex.getFunction(ImmutableList.of(SqlTypes.INTEGER))
+    );
+
+    // Then:
+    assertThat(e.getMessage(), containsString("Function 'name' does not accept parameters "
+            + "(INTEGER)"));
+  }
+
   private void givenFunctions(final KsqlScalarFunction... functions) {
     Arrays.stream(functions).forEach(udfIndex::addFunction);
   }

diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/function/UdafAggregateFunctionFactory.java b/ksqldb-engine/src/main/java/io/confluent/ksql/function/UdafAggregateFunctionFactory.java
@@ -32,7 +32,7 @@ public class UdafAggregateFunctionFactory extends AggregateFunctionFactory {
       final List<UdafFactoryInvoker> factoryList
   ) {
     super(metadata);
-    udfIndex = new UdfIndex<>(metadata.getName());
+    udfIndex = new UdfIndex<>(metadata.getName(), false);
     factoryList.forEach(udfIndex::addFunction);
   }