chore: remove GenericRowValueTypeEnforcer (#3792)

This change sees `GenericRowValueTypeEnforcer` removed from the code base. It's being removed as the functionality it provides is: 1. Not needed in the production code path as values are always of the right type. 1. Only needed for a few tests to pass, (which have been fixed up). 1. It supports coercing between types which are outside of KSQLs implicit casting rules, e.g. coerce String -> BIGINT. We should support implicitly casting parameters to functions to wider types, as defined by KSQLs implicit casting rules. However, this is not currently the case. See #3791.
confluentinc · Nov 7, 2019 · 57fb964 · 57fb964
1 parent a8cc010
commit 57fb964
Show file tree

Hide file tree

Showing 11 changed files with 53 additions and 737 deletions.
diff --git a/ksql-engine/src/test/java/io/confluent/ksql/codegen/CodeGenRunnerTest.java b/ksql-engine/src/test/java/io/confluent/ksql/codegen/CodeGenRunnerTest.java
@@ -543,9 +543,9 @@ public void shouldHandleArithmeticExpr() {
     @Test
     public void testCastNumericArithmeticExpressions() {
         final Map<Integer, Object> inputValues =
-            ImmutableMap.of(0, 1, 3, 3, 4, 4, 5, 5);
+            ImmutableMap.of(0, 1L, 3, 3.0D, 4, 4.0D, 5, 5);
 
-        // INT64 - INT32
+        // INT - BIGINT
         assertThat(executeExpression(
             "SELECT "
                 + "CAST((col5 - col0) AS INTEGER),"
@@ -555,7 +555,7 @@ public void testCastNumericArithmeticExpressions() {
                 + "FROM codegen_test EMIT CHANGES;",
             inputValues), contains(4, 4L, 4.0, "4"));
 
-        // FLOAT64 - FLOAT64
+        // DOUBLE - DOUBLE
         assertThat(executeExpression(
             "SELECT "
                 + "CAST((col4 - col3) AS INTEGER),"
@@ -565,7 +565,7 @@ public void testCastNumericArithmeticExpressions() {
                 + "FROM codegen_test EMIT CHANGES;",
             inputValues), contains(1, 1L, 1.0, "1.0"));
 
-        // FLOAT64 - INT64
+        // DOUBLE - INT
         assertThat(executeExpression(
             "SELECT "
                 + "CAST((col4 - col0) AS INTEGER),"
@@ -594,7 +594,7 @@ public void shouldHandleMathUdfs() {
         final String query =
             "SELECT FLOOR(col3), CEIL(col3*3), ABS(col0+1.34), ROUND(col3*2)+12 FROM codegen_test EMIT CHANGES;";
 
-        final Map<Integer, Object> inputValues = ImmutableMap.of(0, 15, 3, 1.5);
+        final Map<Integer, Object> inputValues = ImmutableMap.of(0, 15L, 3, 1.5);
 
         // When:
         final List<Object> columns = executeExpression(query, inputValues);
@@ -607,7 +607,7 @@ public void shouldHandleMathUdfs() {
     public void shouldHandleRandomUdf() {
         // Given:
         final String query = "SELECT RANDOM()+10, RANDOM()+col0 FROM codegen_test EMIT CHANGES;";
-        final Map<Integer, Object> inputValues = ImmutableMap.of(0, 15);
+        final Map<Integer, Object> inputValues = ImmutableMap.of(0, 15L);
 
         // When:
         final List<Object> columns = executeExpression(query, inputValues);
@@ -825,7 +825,7 @@ public void shouldHandleFunctionWithNullArgument() {
         final String query =
             "SELECT test_udf(col0, NULL) FROM codegen_test EMIT CHANGES;";
 
-        final Map<Integer, Object> inputValues = ImmutableMap.of(0, 0);
+        final Map<Integer, Object> inputValues = ImmutableMap.of(0, 0L);
         final List<Object> columns = executeExpression(query, inputValues);
         // test
         assertThat(columns, equalTo(Collections.singletonList("doStuffLongString")));
@@ -836,7 +836,7 @@ public void shouldHandleFunctionWithVarargs() {
         final String query =
             "SELECT test_udf(col0, col0, col0, col0, col0) FROM codegen_test EMIT CHANGES;";
 
-        final Map<Integer, Object> inputValues = ImmutableMap.of(0, 0);
+        final Map<Integer, Object> inputValues = ImmutableMap.of(0, 0L);
         final List<Object> columns = executeExpression(query, inputValues);
         // test
         assertThat(columns, equalTo(Collections.singletonList("doStuffLongVarargs")));
@@ -860,7 +860,7 @@ public void shouldChoseFunctionWithCorrectNumberOfArgsWhenNullArgument() {
         final String query =
             "SELECT test_udf(col0, col0, NULL) FROM codegen_test EMIT CHANGES;";
 
-        final Map<Integer, Object> inputValues = ImmutableMap.of(0, 0);
+        final Map<Integer, Object> inputValues = ImmutableMap.of(0, 0L);
         final List<Object> columns = executeExpression(query, inputValues);
         // test
         assertThat(columns, equalTo(Collections.singletonList("doStuffLongLongString")));

diff --git a/...-engine/src/test/java/io/confluent/ksql/integration/SelectValueMapperIntegrationTest.java b/...-engine/src/test/java/io/confluent/ksql/integration/SelectValueMapperIntegrationTest.java
@@ -59,10 +59,10 @@ public void shouldSelectChosenColumns() {
 
     // When:
     final GenericRow transformed = selectMapper.apply(
-        genericRow(1521834663L, "key1", 1L, "hi", "bye", 2.0F, "blah"));
+        genericRow(1521834663L, "key1", 1L, "hi", "bye", 2.0D, "blah"));
 
     // Then:
-    assertThat(transformed, is(genericRow(1L, "bye", 2.0F)));
+    assertThat(transformed, is(genericRow(1L, "bye", 2.0D)));
   }
 
   @Test
@@ -73,10 +73,10 @@ public void shouldApplyUdfsToColumns() {
 
     // When:
     final GenericRow row = selectMapper.apply(
-        genericRow(1521834663L, "key1", 2L, "foo", "whatever", 6.9F, "boo", "hoo"));
+        genericRow(1521834663L, "key1", 2L, "foo", "whatever", 6.9D, "boo", "hoo"));
 
     // Then:
-    assertThat(row, is(genericRow(2L, "foo", "whatever", 7.0F)));
+    assertThat(row, is(genericRow(2L, "foo", "whatever", 7.0D)));
   }
 
   private SelectValueMapper givenSelectMapperFor(final String query) {

diff --git a/ksql-engine/src/test/java/io/confluent/ksql/planner/plan/AggregateNodeTest.java b/ksql-engine/src/test/java/io/confluent/ksql/planner/plan/AggregateNodeTest.java
@@ -143,7 +143,7 @@ public void shouldUseConsistentOrderInPreAggSelectMapper() {
     assertThat("invalid test", valueMappers, hasSize(greaterThanOrEqualTo(1)));
     final ValueMapper preAggSelectMapper = valueMappers.get(0);
     final GenericRow result = (GenericRow) preAggSelectMapper
-        .apply(new GenericRow("rowtime", "rowkey", "0", "1", "2", "3"));
+        .apply(new GenericRow("rowtime", "rowkey", 0L, "1", "2", 3.0D));
     assertThat("should select col0, col1, col2, col3", result.getColumns(),
         contains(0L, "1", "2", 3.0));
   }
@@ -163,7 +163,7 @@ public void shouldUseConsistentOrderInPostAggSelectMapper() {
     assertThat("invalid test", valueMappers, hasSize(greaterThanOrEqualTo(2)));
     final ValueMapper postAggSelect = valueMappers.get(1);
     final GenericRow result = (GenericRow) postAggSelect
-        .apply(new GenericRow("0", "-1", "2", "3", "4"));
+        .apply(new GenericRow(0L, "-1", 2.0D, 3L, 4.0D));
     assertThat("should select col0, agg1, agg2", result.getColumns(), contains(0L, 2.0, 3L, 4.0));
   }
 

diff --git a/ksql-execution/src/main/java/io/confluent/ksql/execution/codegen/CodeGenRunner.java b/ksql-execution/src/main/java/io/confluent/ksql/execution/codegen/CodeGenRunner.java
@@ -25,7 +25,6 @@
 import io.confluent.ksql.execution.expression.tree.TraversalExpressionVisitor;
 import io.confluent.ksql.execution.function.udf.structfieldextractor.FetchFieldFromStruct;
 import io.confluent.ksql.execution.util.ExpressionTypeManager;
-import io.confluent.ksql.execution.util.GenericRowValueTypeEnforcer;
 import io.confluent.ksql.function.FunctionRegistry;
 import io.confluent.ksql.function.KsqlScalarFunction;
 import io.confluent.ksql.function.UdfFactory;
@@ -111,7 +110,6 @@ public ExpressionMetadata buildCodeGenFromParseTree(Expression expression, Strin
           ee,
           spec,
           expressionType,
-          new GenericRowValueTypeEnforcer(schema),
           expression
       );
     } catch (KsqlException | CompileException e) {

diff --git a/ksql-execution/src/main/java/io/confluent/ksql/execution/codegen/CodeGenSpec.java b/ksql-execution/src/main/java/io/confluent/ksql/execution/codegen/CodeGenSpec.java
@@ -20,7 +20,6 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.errorprone.annotations.Immutable;
 import io.confluent.ksql.GenericRow;
-import io.confluent.ksql.execution.util.GenericRowValueTypeEnforcer;
 import io.confluent.ksql.function.udf.Kudf;
 import io.confluent.ksql.name.FunctionName;
 import io.confluent.ksql.schema.ksql.ColumnRef;
@@ -71,16 +70,13 @@ public String getUniqueNameForFunction(FunctionName functionName, int index) {
     return names.get(index);
   }
 
-  public void resolve(
-      GenericRow row, GenericRowValueTypeEnforcer typeEnforcer, Object[] parameters
-  ) {
+  public void resolve(GenericRow row, Object[] parameters) {
     for (int paramIdx = 0; paramIdx < arguments.size(); paramIdx++) {
       ArgumentSpec spec = arguments.get(paramIdx);
 
       if (spec.colIndex().isPresent()) {
         int colIndex = spec.colIndex().getAsInt();
-        parameters[paramIdx] = typeEnforcer
-            .enforceColumnType(colIndex, row.getColumns().get(colIndex));
+        parameters[paramIdx] = row.getColumns().get(colIndex);
       } else {
         int copyOfParamIdxForLambda = paramIdx;
         parameters[paramIdx] = spec.kudf()

diff --git a/ksql-execution/src/main/java/io/confluent/ksql/execution/codegen/ExpressionMetadata.java b/ksql-execution/src/main/java/io/confluent/ksql/execution/codegen/ExpressionMetadata.java
@@ -19,7 +19,6 @@
 import io.confluent.ksql.GenericRow;
 import io.confluent.ksql.execution.codegen.CodeGenSpec.ArgumentSpec;
 import io.confluent.ksql.execution.expression.tree.Expression;
-import io.confluent.ksql.execution.util.GenericRowValueTypeEnforcer;
 import io.confluent.ksql.schema.ksql.types.SqlType;
 import io.confluent.ksql.util.KsqlException;
 import java.lang.reflect.InvocationTargetException;
@@ -32,18 +31,18 @@ public class ExpressionMetadata {
 
   private final IExpressionEvaluator expressionEvaluator;
   private final SqlType expressionType;
-  private final GenericRowValueTypeEnforcer typeEnforcer;
   private final ThreadLocal<Object[]> threadLocalParameters;
   private final Expression expression;
   private final CodeGenSpec spec;
 
   public ExpressionMetadata(
-      IExpressionEvaluator expressionEvaluator, CodeGenSpec spec, SqlType expressionType,
-      GenericRowValueTypeEnforcer typeEnforcer, Expression expression
+      IExpressionEvaluator expressionEvaluator,
+      CodeGenSpec spec,
+      SqlType expressionType,
+      Expression expression
   ) {
     this.expressionEvaluator = Objects.requireNonNull(expressionEvaluator, "expressionEvaluator");
     this.expressionType = Objects.requireNonNull(expressionType, "expressionType");
-    this.typeEnforcer = Objects.requireNonNull(typeEnforcer, "typeEnforcer");
     this.expression = Objects.requireNonNull(expression, "expression");
     this.spec = Objects.requireNonNull(spec, "spec");
     this.threadLocalParameters = ThreadLocal.withInitial(() -> new Object[spec.arguments().size()]);
@@ -80,7 +79,7 @@ public Object evaluate(GenericRow row) {
 
   private Object[] getParameters(GenericRow row) {
     Object[] parameters = this.threadLocalParameters.get();
-    spec.resolve(row, typeEnforcer, parameters);
+    spec.resolve(row, parameters);
     return parameters;
   }
 }
diff --git a/ksql-execution/src/main/java/io/confluent/ksql/execution/sqlpredicate/SqlPredicate.java b/ksql-execution/src/main/java/io/confluent/ksql/execution/sqlpredicate/SqlPredicate.java
@@ -25,7 +25,6 @@
 import io.confluent.ksql.execution.codegen.SqlToJavaVisitor;
 import io.confluent.ksql.execution.expression.tree.Expression;
 import io.confluent.ksql.execution.util.EngineProcessingLogMessageFactory;
-import io.confluent.ksql.execution.util.GenericRowValueTypeEnforcer;
 import io.confluent.ksql.function.FunctionRegistry;
 import io.confluent.ksql.logging.processing.ProcessingLogger;
 import io.confluent.ksql.schema.ksql.LogicalSchema;
@@ -39,16 +38,17 @@ public final class SqlPredicate {
 
   private final Expression filterExpression;
   private final IExpressionEvaluator ee;
-  private final GenericRowValueTypeEnforcer genericRowValueTypeEnforcer;
   private final ProcessingLogger processingLogger;
   private final CodeGenSpec spec;
 
   public SqlPredicate(
-      Expression filterExpression, LogicalSchema schema, KsqlConfig ksqlConfig,
-      FunctionRegistry functionRegistry, ProcessingLogger processingLogger
+      Expression filterExpression,
+      LogicalSchema schema,
+      KsqlConfig ksqlConfig,
+      FunctionRegistry functionRegistry,
+      ProcessingLogger processingLogger
   ) {
     this.filterExpression = requireNonNull(filterExpression, "filterExpression");
-    this.genericRowValueTypeEnforcer = new GenericRowValueTypeEnforcer(schema);
     this.processingLogger = requireNonNull(processingLogger);
 
     CodeGenRunner codeGenRunner = new CodeGenRunner(schema, ksqlConfig, functionRegistry);
@@ -86,7 +86,7 @@ public <K> Predicate<K, GenericRow> getPredicate() {
 
       try {
         Object[] values = new Object[spec.arguments().size()];
-        spec.resolve(row, genericRowValueTypeEnforcer, values);
+        spec.resolve(row, values);
         return (Boolean) ee.evaluate(values);
       } catch (Exception e) {
         logProcessingError(e, row);

diff --git a/...execution/src/main/java/io/confluent/ksql/execution/util/GenericRowValueTypeEnforcer.java b/...execution/src/main/java/io/confluent/ksql/execution/util/GenericRowValueTypeEnforcer.java