Add skeleton for databricks destination (#5629)

Co-authored-by: Liren Tu <tuliren.git@outlook.com>
Co-authored-by: LiRen Tu <tuliren@gmail.com>

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

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

airbyte-integrations/connectors/destination-databricks/Dockerfile

@@ -0,0 +1,11 @@
+FROM airbyte/integration-base-java:dev
+
+WORKDIR /airbyte
+ENV APPLICATION destination-databricks
+
+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-databricks

airbyte-integrations/connectors/destination-databricks/README.md

@@ -0,0 +1,68 @@
+# Destination Databricks
+
+This is the repository for the Databricks destination connector in Java.
+For information about how to use this connector within Airbyte, see [the User Documentation](https://docs.airbyte.io/integrations/destinations/databricks).
+
+## Local development
+
+#### Building via Gradle
+From the Airbyte repository root, run:
+```
+./gradlew :airbyte-integrations:connectors:destination-databricks:build
+```
+
+#### Create credentials
+**If you are a community contributor**, generate the necessary credentials and place them in `secrets/config.json` conforming to the spec file in `src/main/resources/spec.json`.
+Note that the `secrets` directory is git-ignored by default, so there is no danger of accidentally checking in sensitive information.
+
+**If you are an Airbyte core member**, follow the [instructions](https://docs.airbyte.io/connector-development#using-credentials-in-ci) to set up the credentials.
+
+### Locally running the connector docker image
+
+#### Build
+Build the connector image via Gradle:
+```
+./gradlew :airbyte-integrations:connectors:destination-databricks:airbyteDocker
+```
+When building via Gradle, the docker image name and tag, respectively, are the values of the `io.airbyte.name` and `io.airbyte.version` `LABEL`s in
+the Dockerfile.
+
+#### Run
+Then run any of the connector commands as follows:
+```
+docker run --rm airbyte/destination-databricks:dev spec
+docker run --rm -v $(pwd)/secrets:/secrets airbyte/destination-databricks:dev check --config /secrets/config.json
+docker run --rm -v $(pwd)/secrets:/secrets airbyte/destination-databricks:dev discover --config /secrets/config.json
+docker run --rm -v $(pwd)/secrets:/secrets -v $(pwd)/integration_tests:/integration_tests airbyte/destination-databricks:dev read --config /secrets/config.json --catalog /integration_tests/configured_catalog.json
+```
+
+## Testing
+We use `JUnit` for Java tests.
+
+### Unit and Integration Tests
+Place unit tests under `src/test/io/airbyte/integrations/destinations/databricks`.
+
+#### Acceptance Tests
+Airbyte has a standard test suite that all destination connectors must pass. Implement the `TODO`s in
+`src/test-integration/java/io/airbyte/integrations/destinations/databricksDestinationAcceptanceTest.java`.
+
+### Using gradle to run tests
+All commands should be run from airbyte project root.
+To run unit tests:
+```
+./gradlew :airbyte-integrations:connectors:destination-databricks:unitTest
+```
+To run acceptance and custom integration tests:
+```
+./gradlew :airbyte-integrations:connectors:destination-databricks:integrationTest
+```
+
+## Dependency Management
+
+### Publishing a new version of the connector
+You've checked out the repo, implemented a million dollar feature, and you're ready to share your changes with the world. Now what?
+1. Make sure your changes are passing unit and integration tests.
+1. Bump the connector version in `Dockerfile` -- just increment the value of the `LABEL io.airbyte.version` appropriately (we use [SemVer](https://semver.org/)).
+1. Create a Pull Request.
+1. Pat yourself on the back for being an awesome contributor.
+1. Someone from Airbyte will take a look at your PR and iterate with you to merge it into master.

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

@@ -0,0 +1,29 @@
+plugins {
+    id 'application'
+    id 'airbyte-docker'
+    id 'airbyte-integration-test-java'
+}
+
+application {
+    mainClass = 'io.airbyte.integrations.destination.databricks.DatabricksDestination'
+}
+
+dependencies {
+    implementation project(':airbyte-db:lib')
+    implementation project(':airbyte-config:models')
+    implementation project(':airbyte-protocol:models')
+    implementation project(':airbyte-integrations:bases:base-java')
+    implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
+    implementation project(':airbyte-integrations:connectors:destination-jdbc')
+    implementation project(':airbyte-integrations:connectors:destination-s3')
+
+    // parquet
+    implementation group: 'org.apache.hadoop', name: 'hadoop-common', version: '3.3.0'
+    implementation group: 'org.apache.hadoop', name: 'hadoop-aws', version: '3.3.0'
+    implementation group: 'org.apache.hadoop', name: 'hadoop-mapreduce-client-core', version: '3.3.0'
+    implementation group: 'org.apache.parquet', name: 'parquet-avro', version: '1.12.0'
+    implementation group: 'tech.allegro.schema.json2avro', name: 'converter', version: '0.2.10'
+
+    integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
+    integrationTestJavaImplementation project(':airbyte-integrations:connectors:destination-databricks')
+}

airbyte-integrations/connectors/destination-databricks/src/main/java/io/airbyte/integrations/destination/databricks/DatabricksDestination.java

@@ -0,0 +1,91 @@
+/*
+ * 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.databricks;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.airbyte.db.Databases;
+import io.airbyte.db.jdbc.JdbcDatabase;
+import io.airbyte.integrations.base.AirbyteMessageConsumer;
+import io.airbyte.integrations.base.IntegrationRunner;
+import io.airbyte.integrations.destination.ExtendedNameTransformer;
+import io.airbyte.integrations.destination.jdbc.SqlOperations;
+import io.airbyte.integrations.destination.jdbc.copy.CopyConsumerFactory;
+import io.airbyte.integrations.destination.jdbc.copy.CopyDestination;
+import io.airbyte.integrations.destination.jdbc.copy.s3.S3Config;
+import io.airbyte.integrations.destination.jdbc.copy.s3.S3StreamCopier;
+import io.airbyte.protocol.models.AirbyteMessage;
+import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
+import java.util.function.Consumer;
+
+public class DatabricksDestination extends CopyDestination {
+
+  private static final String DRIVER_CLASS = "com.simba.spark.jdbc.Driver";
+
+  public static void main(String[] args) throws Exception {
+    new IntegrationRunner(new DatabricksDestination()).run(args);
+  }
+
+  @Override
+  public AirbyteMessageConsumer getConsumer(JsonNode config, ConfiguredAirbyteCatalog catalog, Consumer<AirbyteMessage> outputRecordCollector) {
+    return CopyConsumerFactory.create(
+        outputRecordCollector,
+        getDatabase(config),
+        getSqlOperations(),
+        getNameTransformer(),
+        S3Config.getS3Config(config),
+        catalog,
+        new DatabricksStreamCopierFactory(),
+        config.get("schema").asText().equals("") ? "default" : config.get("schema").asText()
+    );
+  }
+
+  @Override
+  public void checkPersistence(JsonNode config) {
+    S3StreamCopier.attemptS3WriteAndDelete(S3Config.getS3Config(config));
+  }
+
+  @Override
+  public ExtendedNameTransformer getNameTransformer() {
+    return new DatabricksNameTransformer();
+  }
+
+  @Override
+  public JdbcDatabase getDatabase(JsonNode databricksConfig) {
+    return Databases.createJdbcDatabase(
+        "token",
+        databricksConfig.get("pat").asText(),
+        String.format("jdbc:spark://%s:443/default;transportMode=http;ssl=1;httpPath=%s",
+            databricksConfig.get("serverHostname").asText(),
+            databricksConfig.get("httpPath").asText()),
+        DRIVER_CLASS
+    );
+  }
+
+  @Override
+  public SqlOperations getSqlOperations() {
+    return new DatabricksSqlOperations();
+  }
+
+}

airbyte-integrations/connectors/destination-databricks/src/main/java/io/airbyte/integrations/destination/databricks/DatabricksNameTransformer.java

@@ -0,0 +1,86 @@
+/*
+ * 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.databricks;
+
+import io.airbyte.integrations.destination.ExtendedNameTransformer;
+
+/**
+ * TODO: Replace below MySQL docstring with Databricks equiv.
+ *
+ * Note that MySQL documentation discusses about identifiers case sensitivity using the
+ * lower_case_table_names system variable. As one of their recommendation is: "It is best to adopt a
+ * consistent convention, such as always creating and referring to databases and tables using
+ * lowercase names. This convention is recommended for maximum portability and ease of use.
+ *
+ * Source: https://dev.mysql.com/doc/refman/8.0/en/identifier-case-sensitivity.html"
+ *
+ * As a result, we are here forcing all identifier (table, schema and columns) names to lowercase.
+ */
+public class DatabricksNameTransformer extends ExtendedNameTransformer {
+
+  // These constants must match those in destination_name_transformer.py
+  public static final int MAX_MYSQL_NAME_LENGTH = 64;
+  // DBT appends a suffix to table names
+  public static final int TRUNCATE_DBT_RESERVED_SIZE = 12;
+  // 4 charachters for 1 underscore and 3 suffix (e.g. _ab1)
+  // 4 charachters for 1 underscore and 3 schema hash
+  public static final int TRUNCATE_RESERVED_SIZE = 8;
+  public static final int TRUNCATION_MAX_NAME_LENGTH = MAX_MYSQL_NAME_LENGTH - TRUNCATE_DBT_RESERVED_SIZE - TRUNCATE_RESERVED_SIZE;
+
+  @Override
+  public String getIdentifier(String name) {
+    String identifier = applyDefaultCase(super.getIdentifier(name));
+    return truncateName(identifier, TRUNCATION_MAX_NAME_LENGTH);
+  }
+
+  @Override
+  public String getTmpTableName(String streamName) {
+    String tmpTableName = applyDefaultCase(super.getTmpTableName(streamName));
+    return truncateName(tmpTableName, TRUNCATION_MAX_NAME_LENGTH);
+  }
+
+  @Override
+  public String getRawTableName(String streamName) {
+    String rawTableName = applyDefaultCase(super.getRawTableName(streamName));
+    return truncateName(rawTableName, TRUNCATION_MAX_NAME_LENGTH);
+  }
+
+  static String truncateName(String name, int maxLength) {
+    if (name.length() <= maxLength) {
+      return name;
+    }
+
+    int allowedLength = maxLength - 2;
+    String prefix = name.substring(0, allowedLength / 2);
+    String suffix = name.substring(name.length() - allowedLength / 2);
+    return prefix + "__" + suffix;
+  }
+
+  @Override
+  protected String applyDefaultCase(String input) {
+    return input.toLowerCase();
+  }
+
+}