Support MS SQL in Jdbc SQL connector [HZ-2694]

extensions/mapstore/src/test/java/com/hazelcast/mapstore/GenericMapLoaderTest.java

@@ -28,6 +28,7 @@
 import com.hazelcast.test.annotation.QuickTest;
 import com.hazelcast.test.jdbc.H2DatabaseProvider;
 import com.hazelcast.test.jdbc.MySQLDatabaseProvider;
+
 import org.junit.After;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -560,7 +561,7 @@ public void givenTableNameProperty_whenCreateMapLoader_thenUseTableNameWithCusto
     }
 
     private static void createSchema(String schemaName) throws SQLException {
-        executeJdbc("CREATE SCHEMA IF NOT EXISTS " + schemaName + " ");
+        executeJdbc(databaseProvider.createSchemaQuery(schemaName));
     }
 
     private <K> GenericMapLoader<K> createMapLoader() {

extensions/mapstore/src/test/java/com/hazelcast/mapstore/mssql/MSSQLGenericMapStoreTest.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2023 Hazelcast Inc.
+ *
+ * Licensed under the Hazelcast Community License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://hazelcast.com/hazelcast-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.hazelcast.mapstore.mssql;
+
+import com.hazelcast.mapstore.GenericMapStoreTest;
+import com.hazelcast.test.annotation.NightlyTest;
+import com.hazelcast.test.jdbc.MSSQLDatabaseProvider;
+import org.junit.BeforeClass;
+import org.junit.experimental.categories.Category;
+
+
+@Category(NightlyTest.class)
+public class MSSQLGenericMapStoreTest extends GenericMapStoreTest {
+
+    @BeforeClass
+    public static void beforeClass() {
+        initialize(new MSSQLDatabaseProvider());
+    }
+}

hazelcast-sql/src/main/java/com/hazelcast/jet/sql/impl/connector/jdbc/GettersProvider.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2023 Hazelcast Inc.
+ *
+ * Licensed under the Hazelcast Community License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://hazelcast.com/hazelcast-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.hazelcast.jet.sql.impl.connector.jdbc;
+
+import com.hazelcast.function.BiFunctionEx;
+import com.hazelcast.jet.sql.impl.connector.jdbc.mssql.HazelcastMSSQLDialect;
+
+import java.sql.ResultSet;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.util.HashMap;
+import java.util.Map;
+
+@SuppressWarnings("checkstyle:ExecutableStatementCount")
+final class GettersProvider {
+
+    private static final Map<String, BiFunctionEx<ResultSet, Integer, Object>> DEFAULT_GETTERS = new HashMap<>();
+    private static final Map<String, Map<String, BiFunctionEx<ResultSet, Integer, Object>>> GETTERS_BY_DATABASE
+            = new HashMap<>();
+
+    static {
+        DEFAULT_GETTERS.put("BOOLEAN", ResultSet::getBoolean);
+        DEFAULT_GETTERS.put("BOOL", ResultSet::getBoolean);
+
+        DEFAULT_GETTERS.put("TINYINT", ResultSet::getByte);
+
+        DEFAULT_GETTERS.put("SMALLINT", ResultSet::getShort);
+        DEFAULT_GETTERS.put("INT2", ResultSet::getShort);
+
+        DEFAULT_GETTERS.put("INT", ResultSet::getInt);
+        DEFAULT_GETTERS.put("INT4", ResultSet::getInt);
+        DEFAULT_GETTERS.put("INTEGER", ResultSet::getInt);
+
+        DEFAULT_GETTERS.put("INT8", ResultSet::getLong);
+        DEFAULT_GETTERS.put("BIGINT", ResultSet::getLong);
+
+        DEFAULT_GETTERS.put("VARCHAR", ResultSet::getString);
+        DEFAULT_GETTERS.put("CHARACTER VARYING", ResultSet::getString);
+        DEFAULT_GETTERS.put("TEXT", ResultSet::getString);
+
+        DEFAULT_GETTERS.put("REAL", ResultSet::getFloat);
+        DEFAULT_GETTERS.put("FLOAT", ResultSet::getFloat);
+        DEFAULT_GETTERS.put("FLOAT4", ResultSet::getFloat);
+
+        DEFAULT_GETTERS.put("DOUBLE", ResultSet::getDouble);
+        DEFAULT_GETTERS.put("DOUBLE PRECISION", ResultSet::getDouble);
+
+        DEFAULT_GETTERS.put("DECIMAL", ResultSet::getBigDecimal);
+        DEFAULT_GETTERS.put("NUMERIC", ResultSet::getBigDecimal);
+
+        DEFAULT_GETTERS.put("DATE", (rs, columnIndex) -> rs.getObject(columnIndex, LocalDate.class));
+        DEFAULT_GETTERS.put("TIME", (rs, columnIndex) -> rs.getObject(columnIndex, LocalTime.class));
+        DEFAULT_GETTERS.put("TIMESTAMP", (rs, columnIndex) -> rs.getObject(columnIndex, LocalDateTime.class));
+        DEFAULT_GETTERS.put("TIMESTAMP_WITH_TIMEZONE",
+                (rs, columnIndex) -> rs.getObject(columnIndex, OffsetDateTime.class));
+
+        // Override some getters for MS SQL
+        Map<String, BiFunctionEx<ResultSet, Integer, Object>> msSql = new HashMap<>(DEFAULT_GETTERS);
+        msSql.put("FLOAT", ResultSet::getDouble);
+        msSql.put("DATETIME", (rs, columnIndex) -> rs.getObject(columnIndex, LocalDateTime.class));
+        msSql.put("DATETIMEOFFSET", (rs, columnIndex) -> rs.getObject(columnIndex, OffsetDateTime.class));
+        GETTERS_BY_DATABASE.put(HazelcastMSSQLDialect.class.getSimpleName(), msSql);
+    }
+
+    private GettersProvider() {
+    }
+
+    public static Map<String, BiFunctionEx<ResultSet, Integer, Object>> getGetters(String dialect) {
+        return GETTERS_BY_DATABASE.getOrDefault(dialect, DEFAULT_GETTERS);
+    }
+}

hazelcast-sql/src/main/java/com/hazelcast/jet/sql/impl/connector/jdbc/JdbcSqlConnector.java

@@ -28,6 +28,7 @@
 import com.hazelcast.jet.impl.util.Util;
 import com.hazelcast.jet.sql.impl.connector.HazelcastRexNode;
 import com.hazelcast.jet.sql.impl.connector.SqlConnector;
+import com.hazelcast.jet.sql.impl.connector.jdbc.mssql.HazelcastMSSQLDialect;
 import com.hazelcast.jet.sql.impl.connector.jdbc.mysql.HazelcastMySqlDialect;
 import com.hazelcast.spi.impl.NodeEngine;
 import com.hazelcast.sql.impl.QueryException;
@@ -293,6 +294,8 @@ private static SqlDialect resolveDialect(DatabaseMetaData databaseMetaData) thro
         switch (databaseMetaData.getDatabaseProductName().toUpperCase(Locale.ROOT).trim()) {
             case "MYSQL":
                 return new HazelcastMySqlDialect(SqlDialects.createContext(databaseMetaData));
+            case "MICROSOFT SQL SERVER":
+                return new HazelcastMSSQLDialect(SqlDialects.createContext(databaseMetaData));
 
             default:
                 return SqlDialectFactoryImpl.INSTANCE.create(databaseMetaData);
@@ -325,7 +328,8 @@ public Vertex fullScanReader(
                         new SelectProcessorSupplier(
                                 table.getDataConnectionName(),
                                 builder.query(),
-                                builder.parameterPositions()
+                                builder.parameterPositions(),
+                                dialect.getClass().getSimpleName()
                         ))
         );
     }
@@ -551,9 +555,11 @@ private static QueryDataType resolveType(String columnTypeName) {
                 return QueryDataType.TIME;
 
             case "TIMESTAMP":
+            case "DATETIME":
                 return QueryDataType.TIMESTAMP;
 
             case "TIMESTAMP WITH TIME ZONE":
+            case "DATETIMEOFFSET":
                 return QueryDataType.TIMESTAMP_WITH_TZ_OFFSET_DATE_TIME;
 
             default:

hazelcast-sql/src/main/java/com/hazelcast/jet/sql/impl/connector/jdbc/SelectProcessorSupplier.java

@@ -32,16 +32,11 @@
 import javax.annotation.Nullable;
 import java.io.IOException;
 import java.security.Permission;
-import java.sql.PreparedStatement;
 import java.sql.ResultSet;
-import java.sql.ResultSetMetaData;
+import java.sql.PreparedStatement;
 import java.sql.SQLException;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.LocalTime;
-import java.time.OffsetDateTime;
+import java.sql.ResultSetMetaData;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -56,66 +51,31 @@ public class SelectProcessorSupplier
         extends AbstractJdbcSqlConnectorProcessorSupplier
         implements ProcessorSupplier, DataSerializable, SecuredFunction {
 
-    private static final Map<String, BiFunctionEx<ResultSet, Integer, Object>> GETTERS = new HashMap<>();
-
-    static {
-        GETTERS.put("BOOLEAN", ResultSet::getBoolean);
-        GETTERS.put("BOOL", ResultSet::getBoolean);
-        GETTERS.put("BIT", ResultSet::getBoolean);
-
-        GETTERS.put("TINYINT", ResultSet::getByte);
-
-        GETTERS.put("SMALLINT", ResultSet::getShort);
-        GETTERS.put("INT2", ResultSet::getShort);
-
-        GETTERS.put("INT", ResultSet::getInt);
-        GETTERS.put("INT4", ResultSet::getInt);
-        GETTERS.put("INTEGER", ResultSet::getInt);
-
-        GETTERS.put("INT8", ResultSet::getLong);
-        GETTERS.put("BIGINT", ResultSet::getLong);
-
-        GETTERS.put("VARCHAR", ResultSet::getString);
-        GETTERS.put("CHARACTER VARYING", ResultSet::getString);
-        GETTERS.put("TEXT", ResultSet::getString);
-
-        GETTERS.put("REAL", ResultSet::getFloat);
-        GETTERS.put("FLOAT", ResultSet::getFloat);
-        GETTERS.put("FLOAT4", ResultSet::getFloat);
-
-        GETTERS.put("DOUBLE", ResultSet::getDouble);
-        GETTERS.put("DOUBLE PRECISION", ResultSet::getDouble);
-        GETTERS.put("DECIMAL", ResultSet::getBigDecimal);
-        GETTERS.put("NUMERIC", ResultSet::getBigDecimal);
-
-        GETTERS.put("DATE", (rs, columnIndex) -> rs.getObject(columnIndex, LocalDate.class));
-        GETTERS.put("TIME", (rs, columnIndex) -> rs.getObject(columnIndex, LocalTime.class));
-        GETTERS.put("TIMESTAMP", (rs, columnIndex) -> rs.getObject(columnIndex, LocalDateTime.class));
-        GETTERS.put("TIMESTAMP_WITH_TIMEZONE", (rs, columnIndex) -> rs.getObject(columnIndex, OffsetDateTime.class));
-    }
-
     private String query;
     private int[] parameterPositions;
 
     private transient ExpressionEvalContext evalContext;
     private transient volatile BiFunctionEx<ResultSet, Integer, Object>[] valueGetters;
+    private String dialectName;
 
     @SuppressWarnings("unused")
     public SelectProcessorSupplier() {
     }
 
     public SelectProcessorSupplier(@Nonnull String dataConnectionName,
                                    @Nonnull String query,
-                                   @Nonnull int[] parameterPositions) {
+                                   @Nonnull int[] parameterPositions,
+                                   @Nonnull String dialectName) {
         super(dataConnectionName);
         this.query = requireNonNull(query, "query must not be null");
         this.parameterPositions = requireNonNull(parameterPositions, "parameterPositions must not be null");
+        this.dialectName = dialectName;
     }
 
     @Override
     public void init(@Nonnull Context context) throws Exception {
         super.init(context);
-        evalContext = ExpressionEvalContext.from(context);
+        this.evalContext = ExpressionEvalContext.from(context);
     }
 
     @Nonnull
@@ -153,7 +113,6 @@ public Collection<? extends Processor> get(int count) {
         );
 
         return singleton(processor);
-
     }
 
     private BiFunctionEx<ResultSet, Integer, Object>[] prepareValueGettersFromMetadata(ResultSet rs) throws SQLException {
@@ -162,7 +121,8 @@ private BiFunctionEx<ResultSet, Integer, Object>[] prepareValueGettersFromMetada
         BiFunctionEx<ResultSet, Integer, Object>[] valueGetters = new BiFunctionEx[metaData.getColumnCount()];
         for (int j = 0; j < metaData.getColumnCount(); j++) {
             String type = metaData.getColumnTypeName(j + 1).toUpperCase(Locale.ROOT);
-            valueGetters[j] = GETTERS.getOrDefault(
+            Map<String, BiFunctionEx<ResultSet, Integer, Object>> getters = GettersProvider.getGetters(dialectName);
+            valueGetters[j] = getters.getOrDefault(
                     type,
                     (resultSet, n) -> rs.getObject(n)
             );
@@ -181,12 +141,14 @@ public void writeData(ObjectDataOutput out) throws IOException {
         out.writeString(dataConnectionName);
         out.writeString(query);
         out.writeIntArray(parameterPositions);
+        out.writeString(dialectName);
     }
 
     @Override
     public void readData(ObjectDataInput in) throws IOException {
         dataConnectionName = in.readString();
         query = in.readString();
         parameterPositions = in.readIntArray();
+        dialectName = in.readString();
     }
 }

hazelcast-sql/src/main/java/com/hazelcast/jet/sql/impl/connector/jdbc/SupportedDatabases.java

@@ -21,6 +21,7 @@
 import com.hazelcast.logging.Logger;
 import org.apache.calcite.sql.SqlDialect;
 import org.apache.calcite.sql.dialect.H2SqlDialect;
+import org.apache.calcite.sql.dialect.MssqlSqlDialect;
 import org.apache.calcite.sql.dialect.MysqlSqlDialect;
 import org.apache.calcite.sql.dialect.PostgresqlSqlDialect;
 
@@ -42,8 +43,7 @@ final class SupportedDatabases {
         SUPPORTED_DATABASE_NAMES.add("MYSQL");
         SUPPORTED_DATABASE_NAMES.add("POSTGRESQL");
         SUPPORTED_DATABASE_NAMES.add("H2");
-        // Uncomment when officially supported
-        // SUPPORTED_DATABASE_NAMES.add("MICROSOFT SQL SERVER");
+        SUPPORTED_DATABASE_NAMES.add("MICROSOFT SQL SERVER");
     }
 
     private SupportedDatabases() {
@@ -72,10 +72,9 @@ static boolean isNewDatabase(String uppercaseProductName) {
     }
 
     static boolean isDialectSupported(SqlDialect dialect) {
-        return dialect instanceof MysqlSqlDialect ||
-               dialect instanceof PostgresqlSqlDialect ||
-               dialect instanceof H2SqlDialect;
-        // Uncomment when officially supported
-        // dialect instanceof MssqlSqlDialect
+        return dialect instanceof H2SqlDialect ||
+               dialect instanceof MssqlSqlDialect ||
+               dialect instanceof MysqlSqlDialect ||
+               dialect instanceof PostgresqlSqlDialect;
     }
 }

hazelcast-sql/src/main/java/com/hazelcast/jet/sql/impl/connector/jdbc/UpsertBuilder.java

@@ -17,10 +17,12 @@
 package com.hazelcast.jet.sql.impl.connector.jdbc;
 
 import com.hazelcast.jet.sql.impl.connector.jdbc.h2.H2UpsertQueryBuilder;
+import com.hazelcast.jet.sql.impl.connector.jdbc.mssql.MSSQLUpsertQueryBuilder;
 import com.hazelcast.jet.sql.impl.connector.jdbc.mysql.MySQLUpsertQueryBuilder;
 import com.hazelcast.jet.sql.impl.connector.jdbc.postgres.PostgresUpsertQueryBuilder;
 import org.apache.calcite.sql.SqlDialect;
 import org.apache.calcite.sql.dialect.H2SqlDialect;
+import org.apache.calcite.sql.dialect.MssqlSqlDialect;
 import org.apache.calcite.sql.dialect.MysqlSqlDialect;
 import org.apache.calcite.sql.dialect.PostgresqlSqlDialect;
 
@@ -43,6 +45,9 @@ static String getUpsertStatement(JdbcTable jdbcTable, SqlDialect dialect) {
         } else if (dialect instanceof H2SqlDialect) {
             H2UpsertQueryBuilder builder = new H2UpsertQueryBuilder(jdbcTable, dialect);
             query = builder.query();
+        } else if (dialect instanceof MssqlSqlDialect) {
+            MSSQLUpsertQueryBuilder builder = new MSSQLUpsertQueryBuilder(jdbcTable, dialect);
+            query = builder.query();
         }
         return query;
     }