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(JdbcSqlTestSupport.databaseProvider.createSchemaQuery(schemaName));
     }
 
     private <K> GenericMapLoader<K> createMapLoader() {

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

@@ -0,0 +1,42 @@
+/*
+ * 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.nio.serialization.genericrecord.GenericRecord;
+import com.hazelcast.test.annotation.NightlyTest;
+import com.hazelcast.test.jdbc.MSSQLDatabaseProvider;
+import com.hazelcast.test.jdbc.MySQLDatabaseProvider;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.util.Properties;
+
+import static com.hazelcast.mapstore.GenericMapLoader.DATA_CONNECTION_REF_PROPERTY;
+import static com.hazelcast.mapstore.GenericMapLoader.EXTERNAL_NAME_PROPERTY;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assume.assumeFalse;
+
+@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/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;
@@ -292,6 +293,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);
@@ -549,9 +552,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/mssql/HazelcastMSSQLDialect.java

@@ -0,0 +1,65 @@
+/*
+ * 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.mssql;
+
+import com.hazelcast.jet.sql.impl.validate.operators.string.HazelcastConcatOperator;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.SqlWriter.Frame;
+import org.apache.calcite.sql.SqlWriter.FrameTypeEnum;
+import org.apache.calcite.sql.dialect.MssqlSqlDialect;
+
+/**
+ * Custom dialect for MSSQL which allows correct unparsing of MSSQL specific operators, like CONCAT
+ */
+public class HazelcastMSSQLDialect extends MssqlSqlDialect {
+
+    /**
+     * Creates a HazelcastMSSQLDialect.
+     */
+    public HazelcastMSSQLDialect(Context context) {
+        super(context);
+    }
+
+    @Override
+    public void unparseCall(SqlWriter writer, SqlCall call, int leftPrec, int rightPrec) {
+        switch (call.getKind()) {
+            case OTHER:
+                if (call.getOperator() instanceof HazelcastConcatOperator) {
+                    unparseConcat(writer, call);
+                    break;
+                } else {
+                    super.unparseCall(writer, call, leftPrec, rightPrec);
+                    break;
+                }
+
+            default:
+                super.unparseCall(writer, call, leftPrec, rightPrec);
+        }
+    }
+
+    private void unparseConcat(SqlWriter writer, SqlCall call) {
+        writer.print("CONCAT");
+        Frame frame = writer.startList(FrameTypeEnum.PARENTHESES, "(", ")");
+        for (SqlNode operand : call.getOperandList()) {
+            writer.sep(",");
+            operand.unparse(writer, 0, 0);
+        }
+        writer.endList(frame);
+    }
+}

hazelcast-sql/src/test/java/com/hazelcast/jet/sql/impl/connector/jdbc/AllTypesInsertJdbcSqlConnectorTest.java

@@ -39,6 +39,7 @@
 import java.time.ZoneOffset;
 import java.util.Collection;
 
+import static org.assertj.core.util.Lists.newArrayList;
 import static java.util.Arrays.asList;
 
 @RunWith(HazelcastParametrizedRunner.class)
@@ -107,6 +108,7 @@ public void insertRowWithAllTypes() throws Exception {
         execute("INSERT INTO " + mappingName + " VALUES(1, ?)", javaValue);
 
         assertJdbcRowsAnyOrder(tableName,
+                newArrayList(Integer.class, jdbcValue.getClass()),
                 new Row(0, jdbcValue),
                 new Row(1, jdbcValue)
         );

hazelcast-sql/src/test/java/com/hazelcast/jet/sql/impl/connector/jdbc/JdbcSqlTestSupport.java

@@ -185,6 +185,11 @@ protected static void assertJdbcRowsAnyOrder(String tableName, Row... rows) {
         assertThat(actualRows).containsExactlyInAnyOrderElementsOf(Arrays.asList(rows));
     }
 
+    protected static void assertJdbcRowsAnyOrder(String tableName, List<Class<?>> columnType,Row... rows) {
+        List<Row> actualRows = jdbcRowsTable(tableName, columnType);
+        assertThat(actualRows).containsExactlyInAnyOrderElementsOf(Arrays.asList(rows));
+    }
+
     protected static void assertJdbcQueryRowsAnyOrder(String query, Row... rows) {
         List<Row> actualRows = jdbcRows(query);
         assertThat(actualRows).containsExactlyInAnyOrderElementsOf(Arrays.asList(rows));
@@ -193,12 +198,19 @@ protected static void assertJdbcQueryRowsAnyOrder(String query, Row... rows) {
     protected static List<Row> jdbcRowsTable(String tableName) {
         return jdbcRows("SELECT * FROM " + tableName);
     }
+    protected static List<Row> jdbcRowsTable(String tableName, List<Class<?>> columnType) {
+        return jdbcRows("SELECT * FROM " + tableName, columnType);
+    }
 
     @Nonnull
     protected static List<Row> jdbcRows(String query) {
         return jdbcRows(query, dbConnectionUrl);
     }
 
+    protected static List<Row> jdbcRows(String query, List<Class<?>> columnType) {
+        return jdbcRows(query, dbConnectionUrl, columnType);
+    }
+
     public static List<Row> jdbcRows(String query, String connectionUrl) {
         List<Row> rows = new ArrayList<>();
         try (Connection conn = DriverManager.getConnection(connectionUrl);
@@ -218,4 +230,23 @@ public static List<Row> jdbcRows(String query, String connectionUrl) {
             throw new RuntimeException(e);
         }
     }
+    public static List<Row> jdbcRows(String query, String connectionUrl, List<Class<?>> columnType) {
+        List<Row> rows = new ArrayList<>();
+        try (Connection conn = DriverManager.getConnection(connectionUrl);
+             Statement stmt = conn.createStatement()
+        ) {
+            stmt.execute(query);
+            ResultSet resultSet = stmt.getResultSet();
+            while (resultSet.next()) {
+                Object[] values = new Object[resultSet.getMetaData().getColumnCount()];
+                for (int i = 0; i < values.length; i++) {
+                    values[i] = resultSet.getObject(i + 1, columnType.get(i));
+                }
+                rows.add(new Row(values));
+            }
+            return rows;
+        } catch (SQLException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }

hazelcast-sql/src/test/java/com/hazelcast/jet/sql/impl/connector/jdbc/mssql/MSSQLAllTypesInsertJdbcSqlConnectorTest.java

@@ -0,0 +1,58 @@
+/*
+ * 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.mssql;
+
+import com.hazelcast.jet.sql.impl.connector.jdbc.AllTypesInsertJdbcSqlConnectorTest;
+import com.hazelcast.test.annotation.NightlyTest;
+import com.hazelcast.test.jdbc.MSSQLDatabaseProvider;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.experimental.categories.Category;
+
+import static org.assertj.core.api.Assumptions.assumeThat;
+
+@Category(NightlyTest.class)
+public class MSSQLAllTypesInsertJdbcSqlConnectorTest extends AllTypesInsertJdbcSqlConnectorTest {
+
+    @BeforeClass
+    public static void beforeClass() {
+        initialize(new MSSQLDatabaseProvider());
+    }
+
+
+    @Before
+    public void setUp() throws Exception {
+        // MSSQL doesn't support BOOLEAN but BIT can be used instead and, uses FLOAT as 8-byte floating point type
+        // For Float in MSSQL see https://learn.microsoft.com/en-us/sql/t-sql/data-types/float-and-real-transact-sql?view=sql-server-ver16
+        // For BIT in MSSQL see https://learn.microsoft.com/en-us/sql/t-sql/data-types/bit-transact-sql?view=sql-server-ver16
+        // For TIMESTAMP in MSSQL see https://learn.microsoft.com/en-us/previous-versions/sql/sql-server-2005/ms182776(v=sql.90)?redirectedfrom=MSDN
+        // For DATETIMEOFFSET in MSSQL see https://learn.microsoft.com/en-us/sql/t-sql/data-types/datetimeoffset-transact-sql?view=sql-server-ver16
+        if (type.equals("BOOLEAN")) {
+            type = "BIT";
+        }
+        if (type.equals("DOUBLE")) {
+            type = "FLOAT";
+        }
+        if (type.equals("TIMESTAMP")) {
+            type = "DATETIME";
+        }
+        if (type.equals("TIMESTAMP WITH TIME ZONE")) {
+            type = "DATETIMEOFFSET";
+        }
+    }
+
+}

hazelcast-sql/src/test/java/com/hazelcast/jet/sql/impl/connector/jdbc/mssql/MSSQLSelectJdbcSqlConnectorTest.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.jet.sql.impl.connector.jdbc.mssql;
+
+import com.hazelcast.jet.sql.impl.connector.jdbc.SelectJdbcSqlConnectorTest;
+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 MSSQLSelectJdbcSqlConnectorTest extends SelectJdbcSqlConnectorTest {
+
+    @BeforeClass
+    public static void beforeClass() {
+        initialize(new MSSQLDatabaseProvider());
+    }
+
+}

hazelcast-sql/src/test/java/com/hazelcast/jet/sql/impl/connector/jdbc/mssql/MSSQLUpdateJdbcSqlConnectorTest.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.jet.sql.impl.connector.jdbc.mssql;
+
+import com.hazelcast.jet.sql.impl.connector.jdbc.UpdateJdbcSqlConnectorTest;
+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 MSSQLUpdateJdbcSqlConnectorTest extends UpdateJdbcSqlConnectorTest {
+
+    @BeforeClass
+    public static void beforeClass() {
+        initialize(new MSSQLDatabaseProvider());
+    }
+}
+

hazelcast/src/test/java/com/hazelcast/test/jdbc/MSSQLDatabaseProvider.java

@@ -45,4 +45,10 @@ public void shutdown() {
             container = null;
         }
     }
+
+    @Override
+    public String createSchemaQuery(String schemaName) {
+        return "IF NOT EXISTS (SELECT 0 FROM information_schema.schemata WHERE schema_name = '" + schemaName
+                + "') BEGIN EXEC sp_executesql N'CREATE SCHEMA " + schemaName + "'; END";
+    }
 }

hazelcast/src/test/java/com/hazelcast/test/jdbc/TestDatabaseProvider.java

@@ -85,4 +85,8 @@ default String quote(String[] parts) {
                      .map(part -> '\"' + part.replaceAll("\"", "\"\"") + '\"')
                      .collect(joining("."));
     };
+
+    default String createSchemaQuery(String schemaName) {
+        return "CREATE SCHEMA IF NOT EXISTS " + schemaName;
+    }
 }