[CALCITE-1913] Make SqlDialect pluggable and remove DatabaseProduct uses

cassandra/src/main/java/org/apache/calcite/adapter/cassandra/CassandraSchema.java

@@ -29,9 +29,9 @@
 import org.apache.calcite.schema.Table;
 import org.apache.calcite.schema.impl.AbstractSchema;
 import org.apache.calcite.schema.impl.MaterializedViewTable;
-import org.apache.calcite.sql.SqlDialect;
 import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.dialect.CalciteSqlDialect;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
 import org.apache.calcite.sql.pretty.SqlPrettyWriter;
@@ -266,7 +266,7 @@ private void addMaterializedViews() {
 
       StringWriter stringWriter = new StringWriter(query.length());
       PrintWriter printWriter = new PrintWriter(stringWriter);
-      SqlWriter writer = new SqlPrettyWriter(SqlDialect.CALCITE, true, printWriter);
+      SqlWriter writer = new SqlPrettyWriter(CalciteSqlDialect.DEFAULT, true, printWriter);
       parsedQuery.unparse(writer, 0, 0);
       query = stringWriter.toString();
 

core/src/main/java/org/apache/calcite/adapter/jdbc/JdbcRules.java

@@ -69,7 +69,6 @@
 import org.apache.calcite.schema.ModifiableTable;
 import org.apache.calcite.sql.SqlAggFunction;
 import org.apache.calcite.sql.SqlDialect;
-import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.util.ImmutableBitSet;
 import org.apache.calcite.util.Util;
 import org.apache.calcite.util.trace.CalciteTrace;
@@ -109,21 +108,6 @@ public static List<RelOptRule> rules(JdbcConvention out) {
         new JdbcValuesRule(out));
   }
 
-  static final ImmutableList<SqlKind> AGG_FUNCS;
-  static final ImmutableList<SqlKind> MYSQL_AGG_FUNCS;
-
-  static {
-    ImmutableList.Builder<SqlKind> builder = ImmutableList.builder();
-    builder.add(SqlKind.COUNT);
-    builder.add(SqlKind.SUM);
-    builder.add(SqlKind.SUM0);
-    builder.add(SqlKind.MIN);
-    builder.add(SqlKind.MAX);
-    AGG_FUNCS = builder.build();
-    builder.add(SqlKind.SINGLE_VALUE);
-    MYSQL_AGG_FUNCS = builder.build();
-  }
-
   /** Abstract base class for rule that converts to JDBC. */
   abstract static class JdbcConverterRule extends ConverterRule {
     protected final JdbcConvention out;
@@ -477,12 +461,7 @@ public RelNode convert(RelNode rel) {
   /** Returns whether this JDBC data source can implement a given aggregate
    * function. */
   private static boolean canImplement(SqlAggFunction aggregation, SqlDialect sqlDialect) {
-    switch (sqlDialect.getDatabaseProduct()) {
-    case MYSQL:
-      return MYSQL_AGG_FUNCS.contains(aggregation.getKind());
-    default:
-      return AGG_FUNCS.contains(aggregation.getKind());
-    }
+    return sqlDialect.supportsAggregateFunction(aggregation.getKind());
   }
 
   /** Aggregate operator implemented in JDBC convention. */

core/src/main/java/org/apache/calcite/adapter/jdbc/JdbcSchema.java

@@ -32,6 +32,8 @@
 import org.apache.calcite.schema.Schemas;
 import org.apache.calcite.schema.Table;
 import org.apache.calcite.sql.SqlDialect;
+import org.apache.calcite.sql.SqlDialectFactory;
+import org.apache.calcite.sql.SqlDialectFactoryImpl;
 import org.apache.calcite.sql.type.SqlTypeFactoryImpl;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.util.Util;
@@ -102,9 +104,19 @@ public static JdbcSchema create(
       DataSource dataSource,
       String catalog,
       String schema) {
+    return create(parentSchema, name, dataSource, new SqlDialectFactoryImpl(), catalog, schema);
+  }
+
+  public static JdbcSchema create(
+      SchemaPlus parentSchema,
+      String name,
+      DataSource dataSource,
+      SqlDialectFactory dialectFactory,
+      String catalog,
+      String schema) {
     final Expression expression =
         Schemas.subSchemaExpression(parentSchema, name, JdbcSchema.class);
-    final SqlDialect dialect = createDialect(dataSource);
+    final SqlDialect dialect = createDialect(dialectFactory, dataSource);
     final JdbcConvention convention =
         JdbcConvention.of(dialect, expression, name);
     return new JdbcSchema(dataSource, dialect, convention, catalog, schema);
@@ -140,13 +152,33 @@ public static JdbcSchema create(
     }
     String jdbcCatalog = (String) operand.get("jdbcCatalog");
     String jdbcSchema = (String) operand.get("jdbcSchema");
-    return JdbcSchema.create(
-        parentSchema, name, dataSource, jdbcCatalog, jdbcSchema);
+    String sqlDialectFactory = (String) operand.get("sqlDialectFactory");
+
+    if (sqlDialectFactory == null || sqlDialectFactory.isEmpty()) {
+      return JdbcSchema.create(
+          parentSchema, name, dataSource, jdbcCatalog, jdbcSchema);
+    } else {
+      SqlDialectFactory factory = AvaticaUtils.instantiatePlugin(
+          SqlDialectFactory.class, sqlDialectFactory);
+      return JdbcSchema.create(
+          parentSchema, name, dataSource, factory, jdbcCatalog, jdbcSchema);
+    }
   }
 
-  /** Returns a suitable SQL dialect for the given data source. */
+  /**
+   * Returns a suitable SQL dialect for the given data source..
+   *
+   * @param dataSource The data source
+   * @deprecated Use {@link #createDialect(SqlDialectFactory, DataSource)} instead
+   */
+  @Deprecated
   public static SqlDialect createDialect(DataSource dataSource) {
-    return JdbcUtils.DialectPool.INSTANCE.get(dataSource);
+    return createDialect(new SqlDialectFactoryImpl(), dataSource);
+  }
+
+  /** Returns a suitable SQL dialect for the given data source. */
+  public static SqlDialect createDialect(SqlDialectFactory dialectFactory, DataSource dataSource) {
+    return JdbcUtils.DialectPool.INSTANCE.get(dialectFactory, dataSource);
   }
 
   /** Creates a JDBC data source with the given specification. */

core/src/main/java/org/apache/calcite/adapter/jdbc/JdbcToEnumerableConverter.java

@@ -100,7 +100,7 @@ public Result implement(EnumerableRelImplementor implementor, Prefer pref) {
     final ParameterExpression resultSet_ =
         Expressions.parameter(Modifier.FINAL, ResultSet.class,
             builder.newName("resultSet"));
-    CalendarPolicy calendarPolicy = CalendarPolicy.of(jdbcConvention.dialect);
+    SqlDialect.CalendarPolicy calendarPolicy = jdbcConvention.dialect.getCalendarPolicy();
     final Expression calendar_;
     switch (calendarPolicy) {
     case LOCAL:
@@ -179,7 +179,7 @@ private UnaryExpression getTimeZoneExpression(
   private void generateGet(EnumerableRelImplementor implementor,
       PhysType physType, BlockBuilder builder, ParameterExpression resultSet_,
       int i, Expression target, Expression calendar_,
-      CalendarPolicy calendarPolicy) {
+      SqlDialect.CalendarPolicy calendarPolicy) {
     final Primitive primitive = Primitive.ofBoxOr(physType.fieldClass(i));
     final RelDataType fieldType =
         physType.getRowType().getFieldList().get(i).getType();
@@ -299,27 +299,6 @@ private String generateSql(SqlDialect dialect) {
         jdbcImplementor.visitChild(0, getInput());
     return result.asStatement().toSqlString(dialect).getSql();
   }
-
-  /** Whether this JDBC driver needs you to pass a Calendar object to methods
-   * such as {@link ResultSet#getTimestamp(int, java.util.Calendar)}. */
-  private enum CalendarPolicy {
-    NONE,
-    NULL,
-    LOCAL,
-    DIRECT,
-    SHIFT;
-
-    static CalendarPolicy of(SqlDialect dialect) {
-      switch (dialect.getDatabaseProduct()) {
-      case MYSQL:
-        return SHIFT;
-      case HSQLDB:
-      default:
-        // NULL works for hsqldb-2.3; nothing worked for hsqldb-1.8.
-        return NULL;
-      }
-    }
-  }
 }
 
 // End JdbcToEnumerableConverter.java

core/src/main/java/org/apache/calcite/adapter/jdbc/JdbcUtils.java

@@ -21,6 +21,7 @@
 import org.apache.calcite.linq4j.function.Function0;
 import org.apache.calcite.linq4j.function.Function1;
 import org.apache.calcite.sql.SqlDialect;
+import org.apache.calcite.sql.SqlDialectFactory;
 import org.apache.calcite.util.ImmutableNullableList;
 import org.apache.calcite.util.Pair;
 
@@ -58,28 +59,36 @@ private JdbcUtils() {
 
   /** Pool of dialects. */
   static class DialectPool {
-    final Map<DataSource, SqlDialect> map0 = new IdentityHashMap<>();
+    final Map<DataSource, Map<SqlDialectFactory, SqlDialect>> map0 = new IdentityHashMap<>();
     final Map<List, SqlDialect> map = new HashMap<>();
 
     public static final DialectPool INSTANCE = new DialectPool();
 
-    SqlDialect get(DataSource dataSource) {
-      final SqlDialect sqlDialect = map0.get(dataSource);
-      if (sqlDialect != null) {
-        return sqlDialect;
+    // TODO: Discuss why we need a pool. If we do, I'd like to improve performance
+    synchronized SqlDialect get(SqlDialectFactory dialectFactory, DataSource dataSource) {
+      Map<SqlDialectFactory, SqlDialect> dialectMap = map0.get(dataSource);
+      if (dialectMap != null) {
+        final SqlDialect sqlDialect = dialectMap.get(dialectFactory);
+        if (sqlDialect != null) {
+          return sqlDialect;
+        }
       }
       Connection connection = null;
       try {
         connection = dataSource.getConnection();
         DatabaseMetaData metaData = connection.getMetaData();
         String productName = metaData.getDatabaseProductName();
         String productVersion = metaData.getDatabaseProductVersion();
-        List key = ImmutableList.of(productName, productVersion);
+        List key = ImmutableList.of(productName, productVersion, dialectFactory);
         SqlDialect dialect = map.get(key);
         if (dialect == null) {
-          dialect = SqlDialect.create(metaData);
+          dialect = dialectFactory.create(metaData);
           map.put(key, dialect);
-          map0.put(dataSource, dialect);
+          if (dialectMap == null) {
+            dialectMap = new IdentityHashMap<>();
+            map0.put(dataSource, dialectMap);
+          }
+          dialectMap.put(dialectFactory, dialect);
         }
         connection.close();
         connection = null;

core/src/main/java/org/apache/calcite/model/JsonJdbcSchema.java

@@ -32,6 +32,13 @@ public class JsonJdbcSchema extends JsonSchema {
    */
   public String jdbcDriver;
 
+  /** The FQN of the {@link org.apache.calcite.sql.SqlDialectFactory} implementation.
+   *
+   * <p>Optional. If not specified, uses whichever class the JDBC
+   * {@link java.sql.DriverManager} chooses.
+   */
+  public String sqlDialectFactory;
+
   /** JDBC connect string, for example "jdbc:mysql://localhost/foodmart".
    *
    * <p>Optional.

core/src/main/java/org/apache/calcite/model/ModelHandler.java

@@ -37,6 +37,7 @@
 import org.apache.calcite.schema.impl.TableFunctionImpl;
 import org.apache.calcite.schema.impl.TableMacroImpl;
 import org.apache.calcite.schema.impl.ViewTable;
+import org.apache.calcite.sql.SqlDialectFactory;
 import org.apache.calcite.util.Pair;
 import org.apache.calcite.util.Util;
 
@@ -320,9 +321,18 @@ public void visit(JsonJdbcSchema jsonSchema) {
             jsonSchema.jdbcDriver,
             jsonSchema.jdbcUser,
             jsonSchema.jdbcPassword);
-    JdbcSchema schema =
-        JdbcSchema.create(parentSchema, jsonSchema.name, dataSource,
-            jsonSchema.jdbcCatalog, jsonSchema.jdbcSchema);
+    final JdbcSchema schema;
+    if (jsonSchema.sqlDialectFactory == null || jsonSchema.sqlDialectFactory.isEmpty()) {
+      schema =
+          JdbcSchema.create(parentSchema, jsonSchema.name, dataSource,
+              jsonSchema.jdbcCatalog, jsonSchema.jdbcSchema);
+    } else {
+      SqlDialectFactory factory = AvaticaUtils.instantiatePlugin(
+          SqlDialectFactory.class, jsonSchema.sqlDialectFactory);
+      schema =
+          JdbcSchema.create(parentSchema, jsonSchema.name, dataSource,
+              factory, jsonSchema.jdbcCatalog, jsonSchema.jdbcSchema);
+    }
     final SchemaPlus schemaPlus = parentSchema.add(jsonSchema.name, schema);
     populateSchema(jsonSchema, schemaPlus);
   }

core/src/main/java/org/apache/calcite/prepare/CalcitePrepareImpl.java

@@ -1120,7 +1120,6 @@ public RelNode get() {
 
     private PreparedResult prepare_(Supplier<RelNode> fn,
         RelDataType resultType) {
-      queryString = null;
       Class runtimeContextClass = Object.class;
       init(runtimeContextClass);
 

core/src/main/java/org/apache/calcite/prepare/Prepare.java

@@ -87,7 +87,6 @@ public abstract class Prepare {
 
   protected final CalcitePrepare.Context context;
   protected final CatalogReader catalogReader;
-  protected String queryString = null;
   /**
    * Convention via which results should be returned by execution.
    */
@@ -241,8 +240,6 @@ public PreparedResult prepareSql(
       Class runtimeContextClass,
       SqlValidator validator,
       boolean needsValidation) {
-    queryString = sqlQuery.toString();
-
     init(runtimeContextClass);
 
     final SqlToRelConverter.ConfigBuilder builder =

core/src/main/java/org/apache/calcite/rel/rel2sql/RelToSqlConverter.java

@@ -198,8 +198,8 @@ public Result visit(Aggregate e) {
     for (AggregateCall aggCall : e.getAggCallList()) {
       SqlNode aggCallSqlNode = builder.context.toSql(aggCall);
       if (aggCall.getAggregation() instanceof SqlSingleValueAggFunction) {
-        aggCallSqlNode =
-            rewriteSingleValueExpr(aggCallSqlNode, dialect);
+        aggCallSqlNode = dialect.
+            rewriteSingleValueExpr(aggCallSqlNode);
       }
       addSelect(selectList, aggCallSqlNode, e.getRowType());
     }
@@ -397,12 +397,15 @@ public Result visit(Match e) {
     final List<SqlNode> orderBySqlList = new ArrayList<>();
     if (e.getOrderKeys() != null) {
       for (RelFieldCollation fc : e.getOrderKeys().getFieldCollations()) {
-        if (fc.nullDirection != RelFieldCollation.NullDirection.UNSPECIFIED
-            && dialect.getDatabaseProduct() == SqlDialect.DatabaseProduct.MYSQL) {
-          orderBySqlList.add(
-              ISNULL_FUNCTION.createCall(POS, context.field(fc.getFieldIndex())));
-          fc = new RelFieldCollation(fc.getFieldIndex(), fc.getDirection(),
-              RelFieldCollation.NullDirection.UNSPECIFIED);
+        if (fc.nullDirection != RelFieldCollation.NullDirection.UNSPECIFIED) {
+          boolean first = fc.nullDirection == RelFieldCollation.NullDirection.FIRST;
+          SqlNode nullDirectionNode = dialect.emulateNullDirection(
+              context.field(fc.getFieldIndex()), first);
+          if (nullDirectionNode != null) {
+            orderBySqlList.add(nullDirectionNode);
+            fc = new RelFieldCollation(fc.getFieldIndex(), fc.getDirection(),
+                RelFieldCollation.NullDirection.UNSPECIFIED);
+          }
         }
         orderBySqlList.add(context.toSql(fc));
       }