Default to ANSI dialect (#125)

* Default to ANSI dialect

* Return flinksql but point it at Ansi rather than mysql

* fix casing

Author: jogrogan

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

@@ -72,12 +72,7 @@ private static SqlDialect dialect(Row row) {
     if (row.DIALECT == null) {
       return SqlDialect.ANSI;
     }
-    switch (row.DIALECT) {
-    case "Flink":
-      return SqlDialect.FLINK;
-    default:
-      return SqlDialect.valueOf(row.DIALECT);
-    }
+    return SqlDialect.valueOf(row.DIALECT.toUpperCase());
   }
 
   @Override

hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/RemoteToEnumerableConverter.java

@@ -59,7 +59,7 @@
 import org.apache.calcite.runtime.SqlFunctions;
 import org.apache.calcite.sql.SqlDialect;
 import org.apache.calcite.sql.SqlKind;
-import org.apache.calcite.sql.dialect.MysqlSqlDialect;
+import org.apache.calcite.sql.dialect.AnsiSqlDialect;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.sql.util.SqlString;
 import org.apache.calcite.util.BuiltInMethod;
@@ -94,7 +94,7 @@ private SqlString generateSql(SqlDialect dialect) {
     RelRoot root = RelRoot.of(getInput(), SqlKind.SELECT);
     try {
       PipelineRel.Implementor plan = DeploymentService.plan(root, Collections.emptyList(), new Properties());
-      return new SqlString(MysqlSqlDialect.DEFAULT, plan.query(connectionProperties)
+      return new SqlString(AnsiSqlDialect.DEFAULT, plan.query(connectionProperties)
           .apply(com.linkedin.hoptimator.SqlDialect.FLINK));  // TODO dialect
     } catch (SQLException e) {
       throw new RuntimeException(e);
@@ -127,7 +127,7 @@ private SqlString generateSql(SqlDialect dialect) {
     final RemoteConvention convention =
         (RemoteConvention) requireNonNull(child.getConvention(),
             () -> "child.getConvention() is null for " + child);
-    SqlString sqlString = generateSql(MysqlSqlDialect.DEFAULT); // TODO hard-coded dialect
+    SqlString sqlString = generateSql(AnsiSqlDialect.DEFAULT); // TODO hard-coded dialect
     String sql = sqlString.getSql();
     if (CalciteSystemProperty.DEBUG.value()) {
       System.out.println("[" + sql + "]");
@@ -141,7 +141,7 @@ private SqlString generateSql(SqlDialect dialect) {
         Expressions.parameter(Modifier.FINAL, ResultSet.class,
             builder.newName("resultSet"));
     final SqlDialect.CalendarPolicy calendarPolicy =
-        MysqlSqlDialect.DEFAULT.getCalendarPolicy();  // TODO hard-coded dialect
+        AnsiSqlDialect.DEFAULT.getCalendarPolicy();  // TODO hard-coded dialect
     final Expression calendar_;
     switch (calendarPolicy) {
     case LOCAL:

hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/ScriptImplementor.java

@@ -35,7 +35,6 @@
 import org.apache.calcite.sql.SqlWriter;
 import org.apache.calcite.sql.SqlWriterConfig;
 import org.apache.calcite.sql.dialect.AnsiSqlDialect;
-import org.apache.calcite.sql.dialect.MysqlSqlDialect;
 import org.apache.calcite.sql.fun.SqlRowOperator;
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.calcite.sql.pretty.SqlPrettyWriter;
@@ -124,11 +123,8 @@ default Function<com.linkedin.hoptimator.SqlDialect, String> seal() {
       final String sql;
       switch (x) {
         case ANSI:
-          sql = sql(AnsiSqlDialect.DEFAULT);
-          break;
         case FLINK:
-          // Flink uses MySQL dialect, more or less
-          sql = sql(MysqlSqlDialect.DEFAULT);
+          sql = sql(AnsiSqlDialect.DEFAULT);
           break;
         default:
           throw new IllegalStateException("unreachable");

hoptimator-venice/src/test/resources/venice-ddl-insert-partial.id

@@ -15,7 +15,7 @@ spec:
     - CREATE TABLE IF NOT EXISTS `VENICE`.`test-store` (`KEY_id` INTEGER, `intField` INTEGER, `stringField` VARCHAR) WITH ('connector'='venice', 'key.fields'='KEY_id', 'key.fields-prefix'='KEY_', 'key.type'='RECORD', 'partial-update-mode'='true', 'storeName'='test-store', 'value.fields-include'='EXCEPT_KEY')
     - CREATE DATABASE IF NOT EXISTS `VENICE` WITH ()
     - CREATE TABLE IF NOT EXISTS `VENICE`.`test-store-1` (`KEY_id` INTEGER, `intField` INTEGER, `stringField` VARCHAR) WITH ('connector'='venice', 'key.fields'='KEY_id', 'key.fields-prefix'='KEY_', 'key.type'='RECORD', 'partial-update-mode'='true', 'storeName'='test-store-1', 'value.fields-include'='EXCEPT_KEY')
-    - INSERT INTO `VENICE`.`test-store-1` (`KEY_id`, `intField`) SELECT `KEY_id`, CAST(`stringField` AS SIGNED) AS `intField` FROM `VENICE`.`test-store`
+    - INSERT INTO `VENICE`.`test-store-1` (`KEY_id`, `intField`) SELECT `KEY_id`, CAST(`stringField` AS INTEGER) AS `intField` FROM `VENICE`.`test-store`
     jarURI: file:///opt/hoptimator-flink-runner.jar
     parallelism: 1
     upgradeMode: stateless