Add DROP TRIGGER Support

Author: shrinandthakkar

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDdlExecutor.java

@@ -29,6 +29,7 @@
 import com.linkedin.hoptimator.jdbc.ddl.HoptimatorDdlParserImpl;
 import com.linkedin.hoptimator.jdbc.ddl.SqlCreateMaterializedView;
 import com.linkedin.hoptimator.jdbc.ddl.SqlCreateTrigger;
+import com.linkedin.hoptimator.jdbc.ddl.SqlDropTrigger;
 import com.linkedin.hoptimator.jdbc.ddl.SqlPauseTrigger;
 import com.linkedin.hoptimator.jdbc.ddl.SqlResumeTrigger;
 import com.linkedin.hoptimator.util.DeploymentService;
@@ -342,6 +343,42 @@ public void execute(SqlResumeTrigger resume, CalcitePrepare.Context context) {
     updateTriggerPausedState(resume, resume.name, false);
   }
 
+  /** Executes a {@code DROP TRIGGER} command. */
+  public void execute(SqlDropTrigger drop, CalcitePrepare.Context context) {
+    logger.info("Validating statement: {}", drop);
+    try {
+      ValidationService.validateOrThrow(drop);
+    } catch (SQLException e) {
+      throw new DdlException(drop, e.getMessage(), e);
+    }
+
+    if (drop.name.names.size() > 1) {
+      throw new DdlException(drop, "Triggers cannot belong to a schema or database.");
+    }
+    String name = drop.name.names.get(0);
+
+    Trigger trigger = new Trigger(name, null, new ArrayList<>(), null, new HashMap<>());
+
+    Collection<Deployer> deployers = null;
+    try {
+      logger.info("Deleting trigger {}", name);
+      deployers = DeploymentService.deployers(trigger, connection);
+      DeploymentService.delete(deployers);
+      logger.info("Deleted trigger {}", name);
+      logger.info("DROP TRIGGER {} completed", name);
+    } catch (Exception e) {
+      if (deployers != null) {
+        DeploymentService.restore(deployers);
+      }
+      // Handle IF EXISTS
+      if (drop.ifExists && e.getMessage() != null && e.getMessage().contains("Error getting TableTrigger")) {
+        logger.info("Trigger {} does not exist (IF EXISTS specified)", name);
+        return;
+      }
+      throw new DdlException(drop, e.getMessage(), e);
+    }
+  }
+
   private void updateTriggerPausedState(SqlNode sqlNode, SqlIdentifier triggerName, boolean paused) {
     logger.info("Validating statement: {}", sqlNode);
     try {

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sTriggerDeployer.java

@@ -54,6 +54,16 @@ public void update() throws SQLException {
     super.update();
   }
 
+  @Override
+  public void delete() throws SQLException {
+    String canonicalName = K8sUtils.canonicalizeName(trigger.name());
+    V1alpha1TableTrigger existingTrigger = triggerApi.get(canonicalName);
+    if (existingTrigger == null) {
+      throw new SQLException("Trigger " + trigger.name() + " not found.");
+    }
+    triggerApi.delete(existingTrigger);
+  }
+
   @Override
   protected V1alpha1TableTrigger toK8sObject() throws SQLException {
     String name = K8sUtils.canonicalizeName(trigger.name(), trigger.job().name());

hoptimator-k8s/src/test/resources/k8s-trigger-pause.id

@@ -46,7 +46,12 @@ select name, schema, "TABLE", paused from "k8s".table_triggers where name = 'tes
 
 !ok
 
-delete from "k8s".table_triggers where name = 'test';
-(1 row modified)
+drop trigger test;
+(0 rows modified)
+
+!update
+
+drop trigger if exists test;
+(0 rows modified)
 
 !update