feat: add ability to DROP custom types (#3281)

ksql-engine/src/main/java/io/confluent/ksql/ddl/commands/CommandFactories.java

@@ -17,6 +17,7 @@
 
 import static io.confluent.ksql.metastore.model.DataSource.DataSourceType;
 
+import io.confluent.ksql.parser.DropType;
 import io.confluent.ksql.parser.tree.CreateStream;
 import io.confluent.ksql.parser.tree.CreateTable;
 import io.confluent.ksql.parser.tree.DdlStatement;
@@ -45,6 +46,7 @@ public class CommandFactories implements DdlCommandFactory {
       .put(DropStream.class, CommandFactories::handleDropStream)
       .put(DropTable.class, CommandFactories::handleDropTable)
       .put(RegisterType.class, CommandFactories::handleRegisterType)
+      .put(DropType.class, CommandFactories::handleDropType)
       .build();
 
   private final ServiceContext serviceContext;
@@ -118,6 +120,10 @@ private RegisterTypeCommand handleRegisterType(final RegisterType statement) {
     return new RegisterTypeCommand(statement);
   }
 
+  private DropTypeCommand handleDropType(final DropType statement) {
+    return new DropTypeCommand(statement.getTypeName());
+  }
+
   private static final class CallInfo {
 
     final String sqlExpression;

ksql-engine/src/main/java/io/confluent/ksql/ddl/commands/DropTypeCommand.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.ddl.commands;
+
+import io.confluent.ksql.metastore.MutableMetaStore;
+import java.util.Objects;
+
+public class DropTypeCommand implements DdlCommand {
+
+  private final String typeName;
+
+  public DropTypeCommand(final String typeName) {
+    this.typeName = Objects.requireNonNull(typeName, "typeName");
+  }
+
+  @Override
+  public DdlCommandResult run(final MutableMetaStore metaStore) {
+    final boolean wasDeleted = metaStore.deleteType(typeName);
+    return wasDeleted
+        ? new DdlCommandResult(true, "Dropped type '" + typeName + "'")
+        : new DdlCommandResult(true, "Type '" + typeName + "' does not exist");
+  }
+}

ksql-engine/src/test/java/io/confluent/ksql/ddl/commands/DropTypeCommandTest.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.ddl.commands;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import io.confluent.ksql.metastore.MutableMetaStore;
+import io.confluent.ksql.parser.DropType;
+import java.util.Optional;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class DropTypeCommandTest {
+
+  private static final DropTypeCommand DROP = new DropTypeCommand("type");
+
+  @Mock
+  private MutableMetaStore metaStore;
+
+  @Test
+  public void shouldDropExistingType() {
+    // Given:
+    when(metaStore.deleteType("type")).thenReturn(true);
+
+    // When:
+    final DdlCommandResult result = DROP.run(metaStore);
+
+    // Then:
+    verify(metaStore).deleteType("type");
+    assertThat("Expected successful execution", result.isSuccess());
+    assertThat(result.getMessage(), is("Dropped type 'type'"));
+  }
+
+  @Test
+  public void shouldDropMissingType() {
+    // Given:
+    when(metaStore.deleteType("type")).thenReturn(false);
+
+    // When:
+    final DdlCommandResult result = DROP.run(metaStore);
+
+    // Then:
+    verify(metaStore).deleteType("type");
+    assertThat("Expected successful execution", result.isSuccess());
+    assertThat(result.getMessage(), is("Type 'type' does not exist"));
+  }
+
+}

ksql-parser/src/main/java/io/confluent/ksql/parser/AstBuilder.java

@@ -17,6 +17,7 @@
 
 import static io.confluent.ksql.metastore.model.DataSource.DataSourceType;
 import static io.confluent.ksql.schema.ksql.TypeContextUtil.getType;
+import static io.confluent.ksql.util.ParserUtil.getIdentifierText;
 import static io.confluent.ksql.util.ParserUtil.getLocation;
 import static io.confluent.ksql.util.ParserUtil.processIntegerNumber;
 import static java.util.Objects.requireNonNull;
@@ -57,6 +58,7 @@
 import io.confluent.ksql.parser.SqlBaseParser.CreateConnectorContext;
 import io.confluent.ksql.parser.SqlBaseParser.DescribeConnectorContext;
 import io.confluent.ksql.parser.SqlBaseParser.DropConnectorContext;
+import io.confluent.ksql.parser.SqlBaseParser.DropTypeContext;
 import io.confluent.ksql.parser.SqlBaseParser.InsertValuesContext;
 import io.confluent.ksql.parser.SqlBaseParser.IntervalClauseContext;
 import io.confluent.ksql.parser.SqlBaseParser.LimitClauseContext;
@@ -616,6 +618,11 @@ public Node visitListConnectors(final ListConnectorsContext ctx) {
       return new ListConnectors(getLocation(ctx), scope);
     }
 
+    @Override
+    public Node visitDropType(final DropTypeContext ctx) {
+      return new DropType(getLocation(ctx), getIdentifierText(ctx.identifier()));
+    }
+
     @Override
     public Node visitTerminateQuery(final SqlBaseParser.TerminateQueryContext context) {
       return new TerminateQuery(getLocation(context), context.qualifiedName().getText());

ksql-parser/src/main/java/io/confluent/ksql/parser/DropType.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.parser;
+
+import com.google.errorprone.annotations.Immutable;
+import io.confluent.ksql.parser.tree.AstVisitor;
+import io.confluent.ksql.parser.tree.ExecutableDdlStatement;
+import io.confluent.ksql.parser.tree.Statement;
+import java.util.Objects;
+import java.util.Optional;
+
+@Immutable
+public class DropType extends Statement implements ExecutableDdlStatement {
+
+  private final String typeName;
+
+  public DropType(final Optional<NodeLocation> location, final String typeName) {
+    super(location);
+    this.typeName = Objects.requireNonNull(typeName, "typeName");
+  }
+
+  public String getTypeName() {
+    return typeName;
+  }
+
+  @Override
+  public <R, C> R accept(final AstVisitor<R, C> visitor, final C context) {
+    return visitor.visitDropType(this, context);
+  }
+
+  @Override
+  public boolean equals(final Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    final DropType that = (DropType) o;
+    return Objects.equals(typeName, that.typeName);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(typeName);
+  }
+
+  @Override
+  public String toString() {
+    return "DropType{"
+        + "typeName='" + typeName + '\''
+        + '}';
+  }
+}

ksql-parser/src/main/java/io/confluent/ksql/parser/tree/AstVisitor.java

@@ -15,6 +15,7 @@
 
 package io.confluent.ksql.parser.tree;
 
+import io.confluent.ksql.parser.DropType;
 import javax.annotation.Nullable;
 
 public abstract class AstVisitor<R, C> {
@@ -178,4 +179,8 @@ protected R visitSetProperty(final SetProperty node, final C context) {
   public R visitRegisterType(final RegisterType node, final C context) {
     return visitStatement(node, context);
   }
+
+  public R visitDropType(final DropType node, final C context) {
+    return visitStatement(node, context);
+  }
 }

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandIdAssigner.java

@@ -16,6 +16,7 @@
 package io.confluent.ksql.rest.server.computation;
 
 import com.google.common.collect.ImmutableMap;
+import io.confluent.ksql.parser.DropType;
 import io.confluent.ksql.parser.tree.CreateStream;
 import io.confluent.ksql.parser.tree.CreateStreamAsSelect;
 import io.confluent.ksql.parser.tree.CreateTable;
@@ -49,6 +50,8 @@ interface CommandIdSupplier {
             command -> getSelectTableCommandId((CreateTableAsSelect) command))
           .put(RegisterType.class,
               command -> getRegisterTypeCommandId((RegisterType) command))
+          .put(DropType.class,
+              command -> getDropTypeCommandId((DropType) command))
           .put(InsertInto.class,
             command -> getInsertIntoCommandId((InsertInto) command))
           .put(TerminateQuery.class,
@@ -101,6 +104,10 @@ private static CommandId getRegisterTypeCommandId(final RegisterType registerTyp
     return new CommandId(CommandId.Type.TYPE, registerType.getName(), Action.CREATE);
   }
 
+  private static CommandId getDropTypeCommandId(final DropType dropType) {
+    return new CommandId(CommandId.Type.TYPE, dropType.getTypeName(), Action.DROP);
+  }
+
   private static CommandId getTerminateCommandId(final TerminateQuery terminateQuery) {
     return new CommandId(
         CommandId.Type.TERMINATE,