feat: support multiple planner output dialects (#31)

* feat: support multiple planner output dialects

* fix: minor

Author: HoganChu98

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

@@ -1,5 +1,6 @@
 package com.linkedin.hoptimator;
 
+import org.apache.calcite.sql.dialect.MysqlSqlDialect;
 import sqlline.SqlLine;
 import sqlline.CommandHandler;
 import sqlline.DispatchCallback;
@@ -247,7 +248,7 @@ public void execute(String line, DispatchCallback dispatchCallback) {
         sqlline.output("SQL:");
         HopTable outputTable = new HopTable("PIPELINE", "SINK", plan.getRowType(),
           Collections.singletonMap("connector", "dummy"));
-        sqlline.output(impl.insertInto(outputTable));
+        sqlline.output(impl.insertInto(outputTable).sql(MysqlSqlDialect.DEFAULT));
         dispatchCallback.setToSuccess();
       } catch (Exception e) {
         sqlline.error(e.toString());
@@ -345,7 +346,7 @@ public void execute(String line, DispatchCallback dispatchCallback) {
         HoptimatorPlanner planner = HoptimatorPlanner.fromModelFile(connectionUrl, new Properties());
         PipelineRel plan = planner.pipeline(query);
         PipelineRel.Implementor impl = new PipelineRel.Implementor(plan);
-        String pipelineSql = impl.query();
+        String pipelineSql = impl.query().sql(MysqlSqlDialect.DEFAULT);
         FlinkIterable iterable = new FlinkIterable(pipelineSql);
         Iterator<String> iter = iterable.<String>field(0, 1).iterator();
         switch(checkType) {
@@ -454,7 +455,7 @@ public void execute(String line, DispatchCallback dispatchCallback) {
         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';";
+        String pipelineSql = impl.insertInto(sink).sql(MysqlSqlDialect.DEFAULT) + "\nSELECT 'SUCCESS';";
         FlinkIterable iterable = new FlinkIterable(pipelineSql);
         Iterator<String> iter = iterable.<String>field(0).iterator();
         if (iter.hasNext()) {

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

@@ -21,6 +21,7 @@
 import org.apache.calcite.rel.convert.ConverterImpl;
 import org.apache.calcite.rel.convert.ConverterRule;
 import org.apache.calcite.runtime.Hook;
+import org.apache.calcite.sql.dialect.MysqlSqlDialect;
 import org.apache.calcite.util.BuiltInMethod;
 
 import java.util.Collections;
@@ -64,7 +65,7 @@ public Result implement(EnumerableRelImplementor implementor, Prefer pref) {
       BlockBuilder builder = new BlockBuilder();
       PhysType physType = PhysTypeImpl.of(implementor.getTypeFactory(), rowType, pref.preferCustom());
       PipelineRel.Implementor impl = new PipelineRel.Implementor(getInput());
-      String sql = impl.query();
+      String sql = impl.query().sql(MysqlSqlDialect.DEFAULT);
       Hook.QUERY_PLAN.run(sql);         // for script validation in tests
       HoptimatorHook.QUERY_PLAN.run(sql);  // ditto
       Expression iter = builder.append("iter", Expressions.new_(FlinkIterable.class, Expressions.constant(sql),

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

@@ -6,15 +6,14 @@
 import org.apache.calcite.rel.type.RelProtoDataType;
 import org.apache.calcite.rel.type.RelDataTypeImpl;
 import org.apache.calcite.sql.SqlDialect;
-import org.apache.calcite.sql.dialect.MysqlSqlDialect;
+import org.apache.calcite.sql.dialect.AnsiSqlDialect;
 
 import com.linkedin.hoptimator.catalog.Resource;
 import com.linkedin.hoptimator.catalog.HopTable;
 import com.linkedin.hoptimator.catalog.ScriptImplementor;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 
 /**
  * Calling convention which implements an SQL-based streaming data pipeline.
@@ -36,8 +35,6 @@
 public interface PipelineRel extends RelNode {
 
   Convention CONVENTION = new Convention.Impl("PIPELINE", PipelineRel.class);
-  SqlDialect OUTPUT_DIALECT = MysqlSqlDialect.DEFAULT; // closely resembles Flink SQL
-  // TODO support alternative output dialects
 
   void implement(Implementor implementor);
 
@@ -72,14 +69,14 @@ private void visit(RelNode input) {
     }
 
     /** Script ending in SELECT... */
-    public String query() {
-      return script.query(relNode).sql(OUTPUT_DIALECT);
+    public ScriptImplementor query() {
+      return script.query(relNode);
     }
 
     /** Script ending in INSERT INTO ... */
-    public String insertInto(HopTable sink) {
+    public ScriptImplementor insertInto(HopTable sink) {
       return script.database(sink.database()).with(sink)
-        .insert(sink.database(), sink.name(), relNode).sql(OUTPUT_DIALECT);
+        .insert(sink.database(), sink.name(), relNode);
     }
 
     /** Add any resources, SQL, DDL etc required to access the table. */
@@ -88,11 +85,15 @@ public void implement(HopTable table) {
       table.resources().forEach(x -> resource(x));
     }
 
-    /** Combine SQL and any Resources into a Pipeline */
     public Pipeline pipeline(HopTable sink) {
+      return pipeline(sink, AnsiSqlDialect.DEFAULT);
+    }
+
+    /** Combine SQL and any Resources into a Pipeline */
+    public Pipeline pipeline(HopTable sink, SqlDialect sqlDialect) {
       List<Resource> resourcesAndJob = new ArrayList<>();
       resourcesAndJob.addAll(resources);
-      resourcesAndJob.add(new SqlJob(insertInto(sink)));
+      resourcesAndJob.add(new SqlJob(insertInto(sink).sql(sqlDialect)));
       return new Pipeline(resourcesAndJob, rowType());
     }
   }