MS SQL Server Destination implementation

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

@@ -41,6 +41,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/bases/standard-destination-test/src/main/java/io/airbyte/integrations/standardtest/destination/TestDestination.java

@@ -419,6 +419,11 @@ public void testIncrementalDedupeSync() throws Exception {
       return;
     }
 
+    if (!implementsBasicNormalization()) {
+      LOGGER.info("Destination does not implement normalization");
+      return;
+    }
+
     final AirbyteCatalog catalog =
         Jsons.deserialize(MoreResources.readResource(DataArgumentsProvider.EXCHANGE_RATE_CONFIG.catalogFile), AirbyteCatalog.class);
     final ConfiguredAirbyteCatalog configuredCatalog = CatalogHelpers.toDefaultConfiguredCatalog(catalog);

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,107 @@
+/*
+ * 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.db.Databases;
+import io.airbyte.db.jdbc.JdbcDatabase;
+import io.airbyte.integrations.base.Destination;
+import io.airbyte.integrations.base.IntegrationRunner;
+import io.airbyte.integrations.destination.jdbc.AbstractJdbcDestination;
+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
+  protected JdbcDatabase getDatabase(JsonNode config) {
+    return Databases.createJdbcDatabase(
+        config.get("username").asText(),
+        config.get("password").asText(),
+        String.format("jdbc:sqlserver://%s:%s",
+            config.get("host").asText(),
+            config.get("port").asInt()),
+        DRIVER_CLASS);
+  }
+
+  @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;user=%s;password=%s;?",
+        config.get("host").asText(),
+        config.get("port").asText(),
+        config.get("database").asText(),
+        config.get("username").asText(),
+        config.get("password").asText()));
+
+    if (config.has("ssl") && config.get("ssl").asBoolean()) {
+      additionalParameters.add("encrypt=true");
+      if (config.has("trustServerCertificate") && config.get("trustServerCertificate").asBoolean()) {
+        additionalParameters.add("trustServerCertificate=true");
+      } else {
+        additionalParameters.add("trustServerCertificate=false");
+        additionalParameters.add("trustStore=" + config.get("trustStoreName").asText());
+        additionalParameters.add("trustStorePassword=" + config.get("trustStorePassword").asText());
+        if (config.has("hostNameInCertificate")) {
+          additionalParameters.add("hostNameInCertificate=" + config.get("hostNameInCertificate").asText());
+        }
+      }
+    }
+
+    if (!additionalParameters.isEmpty()) {
+      additionalParameters.forEach(x -> jdbcUrl.append(x).append(";"));
+    }
+
+    final ImmutableMap.Builder<Object, Object> configBuilder = ImmutableMap.builder()
+        .put("jdbc_url", jdbcUrl.toString())
+        .put("schema", schema);
+
+    return Jsons.jsonNode(configBuilder.build());
+  }
+
+  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;");
+  }
+
+}

airbyte-integrations/connectors/destination-mssql/src/main/resources/spec.json

@@ -0,0 +1,72 @@
+{
+  "documentationUrl": "https://docs.airbyte.io/integrations/destinations/postgres",
+  "supportsIncremental": true,
+  "supported_destination_sync_modes": ["overwrite", "append"],
+  "connectionSpecification": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "title": "Postgres Destination Spec",
+    "type": "object",
+    "required": ["host", "port", "username", "database", "schema"],
+    "additionalProperties": false,
+    "properties": {
+      "host": {
+        "title": "Host",
+        "description": "Hostname of the database.",
+        "type": "string",
+        "order": 0
+      },
+      "port": {
+        "title": "Port",
+        "description": "Port of the database.",
+        "type": "integer",
+        "minimum": 0,
+        "maximum": 65536,
+        "default": 5432,
+        "examples": ["5432"],
+        "order": 1
+      },
+      "database": {
+        "title": "DB Name",
+        "description": "Name of the database.",
+        "type": "string",
+        "order": 2
+      },
+      "schema": {
+        "title": "Default Schema",
+        "description": "The default schema tables are written to if the source does not specify a namespace. The usual value for this field is \"public\".",
+        "type": "string",
+        "examples": ["public"],
+        "default": "public",
+        "order": 3
+      },
+      "username": {
+        "title": "User",
+        "description": "Username to use to access the database.",
+        "type": "string",
+        "order": 4
+      },
+      "password": {
+        "title": "Password",
+        "description": "Password associated with the username.",
+        "type": "string",
+        "airbyte_secret": true,
+        "order": 5
+      },
+      "ssl": {
+        "title": "SSL Connection",
+        "description": "Encrypt data using SSL.",
+        "type": "boolean",
+        "default": false,
+        "order": 6
+      },
+      "basic_normalization": {
+        "title": "Basic Normalization",
+        "type": "boolean",
+        "default": false,
+        "description": "Whether or not to normalize the data in the destination. See <a href=\"https://docs.airbyte.io/architecture/basic-normalization\">basic normalization</a> for more details.",
+        "examples": [true, false],
+        "order": 7
+      }
+    }
+  }
+}