MS SQL Server Destination implementation

airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/d4353156-9217-4cad-8dd7-c108fd4f74cf.json

@@ -0,0 +1,7 @@
+{
+  "destinationDefinitionId": "d4353156-9217-4cad-8dd7-c108fd4f74cf",
+  "name": "MS SQL Server",
+  "dockerRepository": "airbyte/destination-mssql",
+  "dockerImageTag": "0.1.0",
+  "documentationUrl": "https://docs.airbyte.io/integrations/destinations/mssql"
+}

airbyte-config/init/src/main/resources/seed/destination_definitions.yaml

@@ -40,3 +40,8 @@
   dockerRepository: airbyte/destination-mysql
   dockerImageTag: 0.1.3
   documentationUrl: https://docs.airbyte.io/integrations/destinations/mysql
+- destinationDefinitionId: d4353156-9217-4cad-8dd7-c108fd4f74cf
+  name: MS SQL Server
+  dockerRepository: airbyte/destination-mssql
+  dockerImageTag: 0.1.0
+  documentationUrl: https://docs.airbyte.io/integrations/destinations/mssql

airbyte-db/src/main/java/io/airbyte/db/Databases.java

@@ -42,6 +42,10 @@ public static JdbcDatabase createRedshiftDatabase(String username, String passwo
     return createJdbcDatabase(username, password, jdbcConnectionString, "com.amazon.redshift.jdbc.Driver");
   }
 
+  public static Database createSqlServerDatabase(String username, String password, String jdbcConnectionString) {
+    return createDatabase(username, password, jdbcConnectionString, "com.microsoft.sqlserver.jdbc.SQLServerDriver", SQLDialect.DEFAULT);
+  }
+
   public static Database createDatabase(final String username,
                                         final String password,
                                         final String jdbcConnectionString,

airbyte-integrations/connectors/destination-mssql/.dockerignore

@@ -0,0 +1,3 @@
+*
+!Dockerfile
+!build

airbyte-integrations/connectors/destination-mssql/Dockerfile

@@ -0,0 +1,12 @@
+FROM airbyte/integration-base-java:dev
+
+WORKDIR /airbyte
+
+ENV APPLICATION destination-mssql
+
+COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
+
+RUN tar xf ${APPLICATION}.tar --strip-components=1
+
+LABEL io.airbyte.version=0.1.0
+LABEL io.airbyte.name=airbyte/destination-mssql

airbyte-integrations/connectors/destination-mssql/build.gradle

@@ -0,0 +1,26 @@
+plugins {
+    id 'application'
+    id 'airbyte-docker'
+    id 'airbyte-integration-test-java'
+}
+
+application {
+    mainClass = 'io.airbyte.integrations.destination.mssql.MSSQLDestination'
+}
+
+dependencies {
+    implementation project(':airbyte-db')
+    implementation project(':airbyte-integrations:bases:base-java')
+    implementation project(':airbyte-protocol:models')
+    implementation project(':airbyte-integrations:connectors:destination-jdbc')
+
+    implementation 'com.microsoft.sqlserver:mssql-jdbc:8.4.1.jre14'
+
+    testImplementation 'org.apache.commons:commons-lang3:3.11'
+    testImplementation "org.testcontainers:mssqlserver:1.15.3"
+
+    integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
+
+    implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
+    integrationTestJavaImplementation files(project(':airbyte-integrations:bases:base-normalization').airbyteDocker.outputs)
+}

airbyte-integrations/connectors/destination-mssql/src/main/java/io/airbyte/integrations/destination/mssql/MSSQLDestination.java

@@ -0,0 +1,118 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.integrations.destination.mssql;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.integrations.base.Destination;
+import io.airbyte.integrations.base.IntegrationRunner;
+import io.airbyte.integrations.destination.jdbc.AbstractJdbcDestination;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MSSQLDestination extends AbstractJdbcDestination implements Destination {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(MSSQLDestination.class);
+
+  public static final String DRIVER_CLASS = "com.microsoft.sqlserver.jdbc.SQLServerDriver";
+
+  public MSSQLDestination() {
+    super(DRIVER_CLASS, new MSSQLNameTransformer(), new SqlServerOperations());
+  }
+
+  @Override
+  public JsonNode toJdbcConfig(JsonNode config) {
+    final String schema = Optional.ofNullable(config.get("schema")).map(JsonNode::asText).orElse("public");
+
+    List<String> additionalParameters = new ArrayList<>();
+
+    final StringBuilder jdbcUrl = new StringBuilder(String.format("jdbc:sqlserver://%s:%s;databaseName=%s;",
+        config.get("host").asText(),
+        config.get("port").asText(),
+        config.get("database").asText()));
+
+    if (config.has("ssl_method")) {
+      readSsl(config, additionalParameters);
+    }
+
+    if (!additionalParameters.isEmpty()) {
+      jdbcUrl.append(String.join(";", additionalParameters));
+    }
+
+    final ImmutableMap.Builder<Object, Object> configBuilder = ImmutableMap.builder()
+        .put("jdbc_url", jdbcUrl.toString())
+        .put("username", config.get("username").asText())
+        .put("password", config.get("password").asText())
+        .put("schema", schema);
+
+    return Jsons.jsonNode(configBuilder.build());
+  }
+
+  private void readSsl(JsonNode config, List<String> additionalParameters) {
+    switch (config.get("ssl_method").asText()) {
+      case "unencrypted":
+        additionalParameters.add("encrypt=false");
+        break;
+      case "encrypted_trust_server_certificate":
+        additionalParameters.add("encrypt=true");
+        additionalParameters.add("trustServerCertificate=true");
+        break;
+      case "encrypted_verify_certificate":
+        additionalParameters.add("encrypt=true");
+
+        // trust store location code found at https://stackoverflow.com/a/56570588
+        String trustStoreLocation = Optional.ofNullable(System.getProperty("javax.net.ssl.trustStore"))
+            .orElseGet(() -> System.getProperty("java.home") + "/lib/security/cacerts");
+        File trustStoreFile = new File(trustStoreLocation);
+        if (!trustStoreFile.exists()) {
+          throw new RuntimeException("Unable to locate the Java TrustStore: the system property javax.net.ssl.trustStore is undefined or "
+              + trustStoreLocation + " does not exist.");
+        }
+        String trustStorePassword = System.getProperty("javax.net.ssl.trustStorePassword");
+
+        additionalParameters.add("trustStore=" + trustStoreLocation);
+        if (trustStorePassword != null && !trustStorePassword.isEmpty()) {
+          additionalParameters.add("trustStorePassword=" + config.get("trustStorePassword").asText());
+        }
+        if (config.has("hostNameInCertificate")) {
+          additionalParameters.add("hostNameInCertificate=" + config.get("hostNameInCertificate").asText());
+        }
+        break;
+    }
+  }
+
+  public static void main(String[] args) throws Exception {
+    final Destination destination = new MSSQLDestination();
+    LOGGER.info("starting destination: {}", MSSQLDestination.class);
+    new IntegrationRunner(destination).run(args);
+    LOGGER.info("completed destination: {}", MSSQLDestination.class);
+  }
+
+}

airbyte-integrations/connectors/destination-mssql/src/main/java/io/airbyte/integrations/destination/mssql/MSSQLNameTransformer.java

@@ -0,0 +1,36 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.integrations.destination.mssql;
+
+import io.airbyte.integrations.destination.ExtendedNameTransformer;
+
+public class MSSQLNameTransformer extends ExtendedNameTransformer {
+
+  @Override
+  protected String applyDefaultCase(String input) {
+    return input.toUpperCase();
+  }
+
+}

airbyte-integrations/connectors/destination-mssql/src/main/java/io/airbyte/integrations/destination/mssql/SqlServerOperations.java

@@ -0,0 +1,106 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.integrations.destination.mssql;
+
+import io.airbyte.db.jdbc.JdbcDatabase;
+import io.airbyte.integrations.base.JavaBaseConstants;
+import io.airbyte.integrations.destination.jdbc.SqlOperations;
+import io.airbyte.integrations.destination.jdbc.SqlOperationsUtils;
+import io.airbyte.protocol.models.AirbyteRecordMessage;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class SqlServerOperations implements SqlOperations {
+
+  @Override
+  public void createSchemaIfNotExists(JdbcDatabase database, String schemaName) throws Exception {
+    final String query = String.format("IF NOT EXISTS ( SELECT * FROM sys.schemas WHERE name = '%s') EXEC('CREATE SCHEMA [%s]')",
+        schemaName,
+        schemaName);
+    database.execute(query);
+  }
+
+  @Override
+  public void createTableIfNotExists(JdbcDatabase database, String schemaName, String tableName) throws Exception {
+    database.execute(createTableQuery(schemaName, tableName));
+  }
+
+  @Override
+  public String createTableQuery(String schemaName, String tableName) {
+    return String.format(
+        "IF NOT EXISTS (SELECT * FROM sys.tables t JOIN sys.schemas s ON t.schema_id = s.schema_id "
+            + "WHERE s.name = '%s' AND t.name = '%s') "
+            + "CREATE TABLE %s.%s ( \n"
+            + "%s VARCHAR(64) PRIMARY KEY,\n"
+            + "%s VARCHAR(MAX),\n"
+            + "%s DATETIMEOFFSET(7) DEFAULT SYSDATETIMEOFFSET()\n"
+            + ");\n",
+        schemaName, tableName, schemaName, tableName, JavaBaseConstants.COLUMN_NAME_AB_ID, JavaBaseConstants.COLUMN_NAME_DATA,
+        JavaBaseConstants.COLUMN_NAME_EMITTED_AT);
+  }
+
+  @Override
+  public void dropTableIfExists(JdbcDatabase database, String schemaName, String tableName) throws Exception {
+    final String query = String.format(
+        "IF EXISTS (SELECT * FROM sys.tables t JOIN sys.schemas s ON t.schema_id = s.schema_id "
+            + "WHERE s.name = '%s' AND t.name = '%s') "
+            + "DROP TABLE %s.%s",
+        schemaName, tableName, schemaName, tableName);
+    database.execute(query);
+  }
+
+  @Override
+  public String truncateTableQuery(String schemaName, String tableName) {
+    return String.format("TRUNCATE TABLE %s.%s\n", schemaName, tableName);
+  }
+
+  @Override
+  public void insertRecords(JdbcDatabase database, Stream<AirbyteRecordMessage> recordsStream, String schemaName, String tempTableName)
+      throws Exception {
+    final List<AirbyteRecordMessage> records = recordsStream.collect(Collectors.toList());
+
+    final String insertQueryComponent = String.format(
+        "INSERT INTO %s.%s (%s, %s, %s) VALUES\n",
+        schemaName,
+        tempTableName,
+        JavaBaseConstants.COLUMN_NAME_AB_ID,
+        JavaBaseConstants.COLUMN_NAME_DATA,
+        JavaBaseConstants.COLUMN_NAME_EMITTED_AT);
+    final String recordQueryComponent = "(?, ?, ?),\n";
+    SqlOperationsUtils.insertRawRecordsInSingleQuery(insertQueryComponent, recordQueryComponent, database, records);
+  }
+
+  @Override
+  public String copyTableQuery(String schemaName, String sourceTableName, String destinationTableName) {
+    return String.format("INSERT INTO %s.%s SELECT * FROM %s.%s;\n", schemaName, destinationTableName, schemaName, sourceTableName);
+  }
+
+  @Override
+  public void executeTransaction(JdbcDatabase database, String queries) throws Exception {
+    database.execute("BEGIN TRAN;\n" + queries + "COMMIT TRAN;");
+  }
+
+}