Add insert-into command for testing purposes (#14)

* Publish to LinkedInJFrog

* Publish hoptimator-cli and hoptimator-operator

* Add insert-into command to CLI for testing purposes

* Case insensitive regex h/t @ehoner

* Fix integration test workflow

* Fix sample kafka topic spec

Author: ryannedolan

.github/workflows/integration-tests.yml

@@ -35,10 +35,10 @@ jobs:
       run: make deploy-dev-environment
     - name: Deploy Hoptimator
       run: make deploy
-    - name: Wait for Readiness
-      run: kubectl wait pod hoptimator --for condition=Ready --timeout=10m
     - name: Deploy Samples
       run: make deploy-samples
+    - name: Wait for Readiness
+      run: kubectl wait pod hoptimator --for condition=Ready --timeout=10m
     - name: Wait for Flink Jobs
       run: |
         i=0
@@ -48,7 +48,7 @@ jobs:
           i=$(($i+1))
           echo "No stable Flink jobs after $i tries..."
         done
-    - name: Integration Tests
+    - name: Run Integration Tests
       run: make integration-tests
     - name: Capture Cluster State
       if: always()

deploy/samples/kafkatopics.yaml

@@ -0,0 +1,9 @@
+apiVersion: hoptimator.linkedin.com/v1alpha1
+kind: KafkaTopic
+metadata:
+  name: test-sink
+spec:
+  topicName: test-sink
+  clientOverrides:
+    bootstrap.servers: my-cluster-kafka-bootstrap.kafka.svc:9092
+    

etc/integration-tests.sql

@@ -13,3 +13,7 @@ SELECT * FROM INVENTORY."products_on_hand" LIMIT 1;
 -- MySQL CDC -> Kafka
 SELECT * FROM RAWKAFKA."products" LIMIT 1;
 
+-- test insert into command
+!insert into RAWKAFKA."test-sink" SELECT AGE AS PAYLOAD, NAME AS KEY FROM DATAGEN.PERSON
+SELECT * FROM RAWKAFKA."test-sink" LIMIT 5;
+

etc/readiness-probe.sql

@@ -4,3 +4,4 @@
 SELECT * FROM DATAGEN.PERSON;
 SELECT * FROM DATAGEN.COMPANY;
 SELECT * FROM INVENTORY."products_on_hand" LIMIT 1;
+SELECT * FROM RAWKAFKA."test-sink" LIMIT 0;

hoptimator-cli/src/main/java/com/linkedin/hoptimator/HoptimatorCliApp.java

@@ -20,6 +20,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Iterator;
 import java.util.Scanner;
 import java.util.Properties;
 import java.io.IOException;
@@ -47,6 +48,7 @@ protected int run(String[] args) throws IOException {
     commandHandlers.add(new YamlCommandHandler());
     commandHandlers.add(new PipelineCommandHandler());
     commandHandlers.add(new IntroCommandHandler());
+    commandHandlers.add(new InsertCommandHandler());
     sqlline.updateCommandHandlers(commandHandlers);
     return sqlline.begin(args, null, true).ordinal();
   }
@@ -264,6 +266,98 @@ public boolean echoToFile() {
     }
   }
 
+  private class InsertCommandHandler implements CommandHandler {
+
+    @Override
+    public String getName() {
+      return "insert";
+    }
+
+    @Override
+    public List<String> getNames() {
+      return Collections.singletonList(getName());
+    }
+
+    @Override
+    public String getHelpText() {
+      return "Run an ephemeral pipeline with an existing sink.";
+    }
+
+    @Override
+    public String matches(String line) {
+      String sql = line;
+      if (sql.startsWith(SqlLine.COMMAND_PREFIX)) {
+        sql = sql.substring(1);
+      }
+
+      if (sql.startsWith("insert into")) {
+        sql = sql.substring("insert into".length() + 1);
+        return sql;
+      }
+
+      return null;
+    }
+
+    @Override
+    public void execute(String line, DispatchCallback dispatchCallback) {
+      String sql = line;
+      if (sql.startsWith(SqlLine.COMMAND_PREFIX)) {
+        sql = sql.substring(1);
+      }
+
+      if (sql.startsWith("insert into")) {
+        sql = sql.substring("insert into".length() + 1);
+      }
+
+      String connectionUrl = sqlline.getConnectionMetadata().getUrl();
+      try {
+        String[] parts = sql.split("(?i)SELECT"); // case insensitive
+        if (parts.length != 2) {
+          throw new IllegalArgumentException("Expected ... SELECT ...");
+        }
+        String[] parts2 = parts[0].split("\\.");
+        if (parts2.length != 2) {
+          throw new IllegalArgumentException("Expected ... DATABASE.TABLE ...");
+        }
+        // TODO unquote correctly
+        String database = parts2[0].replaceAll("[\\\"']", "").trim();
+        String table = parts2[1].replaceAll("[\\\"']", "").trim();
+        String query = parts[1];
+
+        HoptimatorPlanner planner = HoptimatorPlanner.fromModelFile(connectionUrl, new Properties());
+        PipelineRel plan = planner.pipeline("SELECT " + query);
+        PipelineRel.Implementor impl = new PipelineRel.Implementor(plan);
+        HopTable sink = planner.database(database).makeTable(table, impl.rowType());
+        String pipelineSql = impl.insertInto(sink) + "\nSELECT 'SUCCESS';";
+        FlinkIterable iterable = new FlinkIterable(pipelineSql);
+        Iterator<String> iter = iterable.<String>field(0).iterator();
+        if (iter.hasNext()) {
+          dispatchCallback.setToSuccess();
+        } else {
+          throw new IllegalArgumentException("No result from:\n" + pipelineSql);
+        }
+        while (iter.hasNext()) {
+          sqlline.output(iter.next());
+        }
+      } catch (Exception e) {
+        sqlline.error(e.toString());
+        e.printStackTrace();
+        dispatchCallback.setToFailure();
+      }
+    }
+
+    @Override
+    public List<Completer> getParameterCompleters() {
+      return Collections.emptyList();
+    }
+
+    @Override
+    public boolean echoToFile() {
+      return false;
+    }
+  }
+
+
   private class IntroCommandHandler implements CommandHandler {
 
     @Override

hoptimator-planner/src/main/java/com/linkedin/hoptimator/planner/PipelineRel.java

@@ -78,7 +78,8 @@ public String query() {
 
     /** Script ending in INSERT INTO ... */
     public String insertInto(HopTable sink) {
-      return script.insert(sink.database(), sink.name(), relNode).sql(OUTPUT_DIALECT);
+      return script.database(sink.database()).with(sink)
+        .insert(sink.database(), sink.name(), relNode).sql(OUTPUT_DIALECT);
     }
 
     /** Add any resources, SQL, DDL etc required to access the table. */