🎉 New destination: Yugabytedb (#18039)

* add yugabytedb destination connector

* add dest def + add changelog

* auto-bump connector version

Co-authored-by: marcosmarxm <[email protected]>
Co-authored-by: Marcos Marx <[email protected]>
Co-authored-by: Octavia Squidington III <[email protected]>

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

@@ -329,3 +329,10 @@
   documentationUrl: https://docs.airbyte.com/integrations/destinations/tidb
   icon: tidb.svg
   releaseStage: alpha
+- name: YugabyteDB
+  destinationDefinitionId: 2300fdcf-a532-419f-9f24-a014336e7966
+  dockerRepository: airbyte/destination-yugabytedb
+  dockerImageTag: 0.1.0
+  documentationUrl: https://docs.airbyte.com/integrations/destinations/yugabytedb
+  icon: yugabytedb.svg
+  releaseStage: alpha

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

@@ -6145,3 +6145,72 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
+- dockerImage: "airbyte/destination-yugabytedb:0.1.0"
+  spec:
+    documentationUrl: "https://docs.airbyte.io/integrations/destinations/yugabytedb"
+    connectionSpecification:
+      $schema: "http://json-schema.org/draft-07/schema#"
+      title: "Yugabytedb destination spec"
+      type: "object"
+      required:
+      - "host"
+      - "port"
+      - "username"
+      - "database"
+      - "schema"
+      additionalProperties: true
+      properties:
+        host:
+          title: "Host"
+          description: "The Hostname of the database."
+          type: "string"
+          order: 0
+        port:
+          title: "Port"
+          description: "The Port of the database."
+          type: "integer"
+          minimum: 0
+          maximum: 65536
+          default: 3306
+          examples:
+          - "3306"
+          order: 1
+        database:
+          title: "Database"
+          description: "Name of the database."
+          type: "string"
+          order: 2
+        username:
+          title: "Username"
+          description: "The Username which is used to access the database."
+          type: "string"
+          order: 3
+        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
+        password:
+          title: "Password"
+          description: "The Password associated with the username."
+          type: "string"
+          airbyte_secret: true
+          order: 4
+        jdbc_url_params:
+          description: "Additional properties to pass to the JDBC URL string when\
+            \ connecting to the database formatted as 'key=value' pairs separated\
+            \ by the symbol '&'. (example: key1=value1&key2=value2&key3=value3)."
+          title: "JDBC URL Params"
+          type: "string"
+          order: 5
+    supportsIncremental: true
+    supportsNormalization: false
+    supportsDBT: false
+    supported_destination_sync_modes:
+    - "overwrite"
+    - "append"

airbyte-db/db-lib/src/main/java/io/airbyte/db/factory/DatabaseDriver.java

@@ -18,7 +18,8 @@ public enum DatabaseDriver {
   ORACLE("oracle.jdbc.OracleDriver", "jdbc:oracle:thin:@%s:%d/%s"),
   POSTGRESQL("org.postgresql.Driver", "jdbc:postgresql://%s:%d/%s"),
   REDSHIFT("com.amazon.redshift.jdbc.Driver", "jdbc:redshift://%s:%d/%s"),
-  SNOWFLAKE("net.snowflake.client.jdbc.SnowflakeDriver", "jdbc:snowflake://%s/");
+  SNOWFLAKE("net.snowflake.client.jdbc.SnowflakeDriver", "jdbc:snowflake://%s/"),
+  YUGABYTEDB("com.yugabyte.Driver", "jdbc:yugabytedb://%s:%d/%s");
 
   private final String driverClassName;
   private final String urlFormatString;

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

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

airbyte-integrations/connectors/destination-yugabytedb/Dockerfile

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

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

@@ -0,0 +1,68 @@
+# Destination Yugabytedb
+
+This is the repository for the Yugabytedb destination connector in Java.
+For information about how to use this connector within Airbyte, see [the User Documentation](https://docs.airbyte.io/integrations/destinations/yugabytedb).
+
+## Local development
+
+#### Building via Gradle
+From the Airbyte repository root, run:
+```
+./gradlew :airbyte-integrations:connectors:destination-yugabytedb: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-yugabytedb: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-yugabytedb:dev spec
+docker run --rm -v $(pwd)/secrets:/secrets airbyte/destination-yugabytedb:dev check --config /secrets/config.json
+docker run --rm -v $(pwd)/secrets:/secrets airbyte/destination-yugabytedb:dev discover --config /secrets/config.json
+docker run --rm -v $(pwd)/secrets:/secrets -v $(pwd)/integration_tests:/integration_tests airbyte/destination-yugabytedb: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/yugabytedb`.
+
+#### 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/yugabytedbDestinationAcceptanceTest.java`.
+
+### Using gradle to run tests
+All commands should be run from airbyte project root.
+To run unit tests:
+```
+./gradlew :airbyte-integrations:connectors:destination-yugabytedb:unitTest
+```
+To run acceptance and custom integration tests:
+```
+./gradlew :airbyte-integrations:connectors:destination-yugabytedb: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-yugabytedb/build.gradle

@@ -0,0 +1,32 @@
+plugins {
+    id 'application'
+    id 'airbyte-docker'
+    id 'airbyte-integration-test-java'
+}
+
+application {
+    mainClass = 'io.airbyte.integrations.destination.yugabytedb.YugabytedbDestination'
+}
+
+dependencies {
+    implementation project(':airbyte-config:config-models')
+    implementation project(':airbyte-protocol: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-db:db-lib')
+
+    implementation 'com.yugabyte:jdbc-yugabytedb:42.3.5-yb-1'
+
+    testImplementation project(':airbyte-integrations:bases:standard-destination-test')
+
+    testImplementation "org.assertj:assertj-core:3.21.0"
+    testImplementation "org.junit.jupiter:junit-jupiter:5.8.1"
+    testImplementation "org.testcontainers:junit-jupiter:1.17.5"
+    testImplementation "org.testcontainers:jdbc:1.17.5"
+
+
+
+    integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
+    integrationTestJavaImplementation project(':airbyte-integrations:connectors:destination-yugabytedb')
+}

airbyte-integrations/connectors/destination-yugabytedb/docker-compose.yml

@@ -0,0 +1,31 @@
+version: '3'
+
+
+# Note: add mount points at /mnt/master and /mnt/tserver for persistence
+
+services:
+  yb-master:
+      image: yugabytedb/yugabyte:latest
+      container_name: yb-master-n1
+      command: [ "/home/yugabyte/bin/yb-master",
+                "--fs_data_dirs=/mnt/master",
+                "--master_addresses=yb-master-n1:7100",
+                "--rpc_bind_addresses=yb-master-n1:7100",
+                "--replication_factor=1"]
+      ports:
+      - "7000:7000"
+
+  yb-tserver:
+      image: yugabytedb/yugabyte:latest
+      container_name: yb-tserver-n1
+      command: [ "/home/yugabyte/bin/yb-tserver",
+                "--fs_data_dirs=/mnt/tserver",
+                "--start_pgsql_proxy",
+                "--rpc_bind_addresses=yb-tserver-n1:9100",
+                "--tserver_master_addrs=yb-master-n1:7100"]
+      ports:
+      - "9042:9042"
+      - "5433:5433"
+      - "9000:9000"
+      depends_on:
+      - yb-master

airbyte-integrations/connectors/destination-yugabytedb/src/main/java/io/airbyte/integrations/destination/yugabytedb/YugabytedbDestination.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.yugabytedb;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.db.factory.DatabaseDriver;
+import io.airbyte.db.jdbc.JdbcUtils;
+import io.airbyte.integrations.base.IntegrationRunner;
+import io.airbyte.integrations.destination.jdbc.AbstractJdbcDestination;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class YugabytedbDestination extends AbstractJdbcDestination {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(YugabytedbDestination.class);
+
+  public static final String DRIVER_CLASS = DatabaseDriver.YUGABYTEDB.getDriverClassName();
+
+  public YugabytedbDestination() {
+    super(DRIVER_CLASS, new YugabytedbNamingTransformer(), new YugabytedbSqlOperations());
+  }
+
+  public static void main(String[] args) throws Exception {
+    LOGGER.info("starting destination: {}", YugabytedbDestination.class);
+    new IntegrationRunner(new YugabytedbDestination()).run(args);
+    LOGGER.info("completed destination: {}", YugabytedbDestination.class);
+  }
+
+  @Override
+  protected Map<String, String> getDefaultConnectionProperties(JsonNode config) {
+    return Collections.emptyMap();
+  }
+
+  @Override
+  public JsonNode toJdbcConfig(JsonNode config) {
+    String schema =
+        Optional.ofNullable(config.get(JdbcUtils.SCHEMA_KEY)).map(JsonNode::asText).orElse("public");
+
+    String jdbcUrl = "jdbc:yugabytedb://" + config.get(JdbcUtils.HOST_KEY).asText() + ":"
+        + config.get(JdbcUtils.PORT_KEY).asText() + "/"
+        + config.get(JdbcUtils.DATABASE_KEY).asText();
+
+    ImmutableMap.Builder<Object, Object> configBuilder = ImmutableMap.builder()
+        .put(JdbcUtils.USERNAME_KEY, config.get(JdbcUtils.USERNAME_KEY).asText())
+        .put(JdbcUtils.JDBC_URL_KEY, jdbcUrl)
+        .put(JdbcUtils.SCHEMA_KEY, schema);
+
+    if (config.has(JdbcUtils.PASSWORD_KEY)) {
+      configBuilder.put(JdbcUtils.PASSWORD_KEY, config.get(JdbcUtils.PASSWORD_KEY).asText());
+    }
+
+    if (config.has(JdbcUtils.JDBC_URL_PARAMS_KEY)) {
+      configBuilder.put(JdbcUtils.JDBC_URL_PARAMS_KEY, config.get(JdbcUtils.JDBC_URL_PARAMS_KEY).asText());
+    }
+
+    return Jsons.jsonNode(configBuilder.build());
+  }
+
+}

airbyte-integrations/connectors/destination-yugabytedb/src/main/java/io/airbyte/integrations/destination/yugabytedb/YugabytedbNamingTransformer.java

@@ -0,0 +1,16 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.yugabytedb;
+
+import io.airbyte.integrations.destination.ExtendedNameTransformer;
+
+public class YugabytedbNamingTransformer extends ExtendedNameTransformer {
+
+  @Override
+  public String applyDefaultCase(final String input) {
+    return input.toLowerCase();
+  }
+
+}

airbyte-integrations/connectors/destination-yugabytedb/src/main/java/io/airbyte/integrations/destination/yugabytedb/YugabytedbSqlOperations.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.yugabytedb;
+
+import com.yugabyte.copy.CopyManager;
+import com.yugabyte.core.BaseConnection;
+import io.airbyte.db.jdbc.JdbcDatabase;
+import io.airbyte.integrations.destination.jdbc.JdbcSqlOperations;
+import io.airbyte.protocol.models.AirbyteRecordMessage;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.List;
+
+public class YugabytedbSqlOperations extends JdbcSqlOperations {
+
+  @Override
+  protected void insertRecordsInternal(JdbcDatabase database,
+                                       List<AirbyteRecordMessage> records,
+                                       String schemaName,
+                                       String tableName)
+      throws Exception {
+
+    if (records.isEmpty()) {
+      return;
+    }
+
+    File tempFile = null;
+    try {
+      tempFile = Files.createTempFile(tableName + "-", ".tmp").toFile();
+      writeBatchToFile(tempFile, records);
+
+      File finalTempFile = tempFile;
+      database.execute(connection -> {
+
+        var copyManager = new CopyManager(connection.unwrap(BaseConnection.class));
+        var sql = String.format("COPY %s.%s FROM STDIN DELIMITER ',' CSV", schemaName, tableName);
+
+        try (var bufferedReader = new BufferedReader(new FileReader(finalTempFile, StandardCharsets.UTF_8))) {
+          copyManager.copyIn(sql, bufferedReader);
+        } catch (IOException e) {
+          throw new UncheckedIOException(e);
+        }
+      });
+    } finally {
+      if (tempFile != null) {
+        Files.delete(tempFile.toPath());
+      }
+    }
+  }
+
+}