fix: allow insertion of null values in migrations tool (#8297)

* fix: intert null values in migrations tool

* add an integration tests

* cyclomatic complexity

* limit 5

ksqldb-tools/src/main/java/io/confluent/ksql/tools/migrations/util/CommandParser.java

@@ -27,6 +27,7 @@
 import io.confluent.ksql.execution.expression.tree.Expression;
 import io.confluent.ksql.execution.expression.tree.IntegerLiteral;
 import io.confluent.ksql.execution.expression.tree.LongLiteral;
+import io.confluent.ksql.execution.expression.tree.NullLiteral;
 import io.confluent.ksql.execution.expression.tree.StringLiteral;
 import io.confluent.ksql.metastore.TypeRegistry;
 import io.confluent.ksql.name.ColumnName;
@@ -167,7 +168,9 @@ private static void validateToken(final String token, final int index) {
   /**
   * Converts a generic expression into the proper Java type.
   */
+  // CHECKSTYLE_RULES.OFF: CyclomaticComplexity
   public static Object toFieldType(final Expression expressionValue) {
+    // CHECKSTYLE_RULES.ON: CyclomaticComplexity
     if (expressionValue instanceof StringLiteral) {
       return ((StringLiteral) expressionValue).getValue();
     } else if (expressionValue instanceof IntegerLiteral) {
@@ -180,6 +183,8 @@ public static Object toFieldType(final Expression expressionValue) {
       return ((BooleanLiteral) expressionValue).getValue();
     } else if (expressionValue instanceof DecimalLiteral) {
       return ((DecimalLiteral) expressionValue).getValue();
+    } else if (expressionValue instanceof NullLiteral) {
+      return null;
     } else if (expressionValue instanceof CreateArrayExpression) {
       return ((CreateArrayExpression) expressionValue)
           .getValues()

ksqldb-tools/src/test/java/io/confluent/ksql/tools/migrations/MigrationsTest.java

@@ -243,6 +243,7 @@ private void shouldApplyMigrations() throws Exception {
             "INSERT INTO FOO (A) VALUES ('GOOD''BYE');" +
             "INSERT INTO ${streamName} (A) VALUES ('${onlyDefinedInFile1}--ha\nha');" +
             "INSERT INTO FOO (A) VALUES ('');" +
+            "INSERT INTO FOO (A) VALUES (NULL);" +
             "DEFINE variable = 'cool';" +
             "SeT 'ksql.output.topic.name.prefix' = '${variable}';" +
             "CREATE STREAM `bar` AS SELECT CONCAT(A, 'woo''hoo') AS A FROM FOO;" +
@@ -342,8 +343,8 @@ private static void verifyMigrationsApplied() {
 
     // verify foo
     final List<StreamedRow> foo = assertThatEventually(
-        () -> makeKsqlQuery("SELECT * FROM FOO EMIT CHANGES LIMIT 4;"),
-        hasSize(6)); // first row is a header, last row is a message saying "Limit Reached"
+        () -> makeKsqlQuery("SELECT * FROM FOO EMIT CHANGES LIMIT 5;"),
+        hasSize(7)); // first row is a header, last row is a message saying "Limit Reached"
     assertThat(foo.get(1).getRow().get().getColumns().size(), is(3));
     assertThat(foo.get(1).getRow().get().getColumns().get(0), is("HELLO"));
     assertThat(foo.get(1).getRow().get().getColumns().get(1), is(50));
@@ -353,6 +354,7 @@ private static void verifyMigrationsApplied() {
     assertNull(foo.get(2).getRow().get().getColumns().get(2));
     assertThat(foo.get(3).getRow().get().getColumns().get(0), is("${onlyDefinedInFile1}--ha\nha"));
     assertThat(foo.get(4).getRow().get().getColumns().get(0), is(""));
+    assertNull(foo.get(5).getRow().get().getColumns().get(0));
 
     // verify bar
     final List<StreamedRow> bar = assertThatEventually(

ksqldb-tools/src/test/java/io/confluent/ksql/tools/migrations/util/CommandParserTest.java

@@ -20,6 +20,7 @@
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
@@ -59,6 +60,22 @@ public void shouldParseInsertValuesStatement() {
     assertThat(toFieldType(insertValues.getValues().get(0)), is(55));
   }
 
+  @Test
+  public void shouldParseInsertNullValuesStatement() {
+    // When:
+    List<SqlCommand> commands = parse("INSERT INTO FOO VALUES (NULL);");
+
+    // Then:
+    assertThat(commands.size(), is(1));
+    assertThat(commands.get(0), instanceOf(SqlInsertValues.class));
+    final SqlInsertValues insertValues = (SqlInsertValues) commands.get(0);
+
+    assertThat(insertValues.getSourceName(), is("`FOO`"));
+    assertThat(insertValues.getColumns(), is(Collections.emptyList()));
+    assertThat(insertValues.getValues().size(), is(1));
+    assertNull(toFieldType(insertValues.getValues().get(0)));
+  }
+
   @Test
   public void shouldParseInsertValuesStatementWithExplicitFields() {
     // When: