🎉 New Destination: Teradata Vantage

airbyte-commons-worker/src/main/java/io/airbyte/workers/normalization/DefaultNormalizationRunner.java

@@ -238,4 +238,4 @@ DestinationType getDestinationType() {
     return destinationType;
   }
 
-}
+}

airbyte-commons-worker/src/main/java/io/airbyte/workers/normalization/NormalizationRunnerFactory.java

@@ -68,4 +68,4 @@ public static ImmutablePair<String, DestinationType> getNormalizationInfoForConn
     }
   }
 
-}
+}

airbyte-db/db-lib/build.gradle

@@ -51,6 +51,9 @@ dependencies {
 
     // MongoDB
     implementation 'org.mongodb:mongodb-driver-sync:4.3.0'
+
+    // Teradata
+	implementation 'com.teradata.jdbc:terajdbc4:17.20.00.12'
 
     // MySQL
     implementation 'mysql:mysql-connector-java:8.0.30'

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

@@ -19,7 +19,8 @@ public enum DatabaseDriver {
   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/"),
-  YUGABYTEDB("com.yugabyte.Driver", "jdbc:yugabytedb://%s:%d/%s");
+  YUGABYTEDB("com.yugabyte.Driver", "jdbc:yugabytedb://%s:%d/%s"),
+  TERADATA("com.teradata.jdbc.TeraDriver", "jdbc:teradata://%s/");
 
   private final String driverClassName;
   private final String urlFormatString;

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

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

airbyte-integrations/connectors/destination-teradata/Dockerfile

@@ -0,0 +1,18 @@
+FROM airbyte/integration-base-java:dev AS build
+
+WORKDIR /airbyte
+ENV APPLICATION destination-teradata
+
+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-teradata
+
+COPY --from=build /airbyte /airbyte
+
+LABEL io.airbyte.version=0.1.0
+LABEL io.airbyte.name=airbyte/destination-teradata

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

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

@@ -0,0 +1,25 @@
+plugins {
+    id 'application'
+    id 'airbyte-integration-test-java'
+}
+
+application {
+    mainClass = 'io.airbyte.integrations.destination.teradata.TeradataDestination'
+applicationDefaultJvmArgs = ['-XX:+ExitOnOutOfMemoryError', '-XX:MaxRAMPercentage=75.0']   
+}
+
+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-db:db-lib')
+    implementation project(':airbyte-integrations:bases:base-java')
+    implementation project(':airbyte-protocol:protocol-models')
+    implementation project(':airbyte-integrations:connectors:destination-jdbc')
+	implementation 'com.teradata.jdbc:terajdbc4:17.20.00.12'
+
+    integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
+    integrationTestJavaImplementation project(':airbyte-integrations:connectors:destination-teradata')
+}
+

airbyte-integrations/connectors/destination-teradata/sample_files/configured_catalog.json

@@ -0,0 +1,25 @@
+{
+  "streams": [
+    {
+      "destination_sync_mode": "append",
+      "sync_mode": "full_refresh",
+      "stream": {
+        "name": "airbyte_acceptance_table",
+        "supported_sync_modes": ["full_refresh"],
+        "source_defined_cursor": false,
+        "json_schema": {
+          "type": "object",
+          "required": ["name"],
+          "properties": {
+            "name": {
+              "type": "string"
+            },
+            "age": {
+              "type": "number"
+            }
+          }
+        }
+      }
+    }
+  ]
+}

airbyte-integrations/connectors/destination-teradata/src/main/java/io/airbyte/integrations/destination/teradata/TeradataDestination.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.teradata;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.db.jdbc.JdbcUtils;
+import io.airbyte.integrations.base.Destination;
+import io.airbyte.integrations.base.IntegrationRunner;
+import io.airbyte.integrations.destination.ExtendedNameTransformer;
+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 TeradataDestination extends AbstractJdbcDestination implements Destination {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(TeradataDestination.class);
+  /**
+   * Teradata JDBC driver
+   */
+  public static final String DRIVER_CLASS = "com.teradata.jdbc.TeraDriver";
+  /**
+   * Default schema name
+   */
+  public static final String DEFAULT_SCHEMA_NAME = "public";
+
+  public static void main(String[] args) throws Exception {
+    new IntegrationRunner(new TeradataDestination()).run(args);
+  }
+
+  public TeradataDestination() {
+    super(DRIVER_CLASS, new ExtendedNameTransformer(), new TeradataSqlOperations());
+  }
+
+  @Override
+  protected Map<String, String> getDefaultConnectionProperties(final JsonNode config) {
+    return Collections.emptyMap();
+  }
+
+  @Override
+  public JsonNode toJdbcConfig(final JsonNode config) {
+    final String schema = Optional.ofNullable(config.get(JdbcUtils.SCHEMA_KEY)).map(JsonNode::asText).orElse(DEFAULT_SCHEMA_NAME);
+
+    final String jdbcUrl = String.format("jdbc:teradata://%s/",
+        config.get(JdbcUtils.HOST_KEY).asText());
+
+    final 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-teradata/src/main/java/io/airbyte/integrations/destination/teradata/TeradataSqlOperations.java

@@ -0,0 +1,151 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.teradata;
+
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.db.jdbc.JdbcDatabase;
+import io.airbyte.integrations.base.AirbyteTraceMessageUtility;
+import io.airbyte.integrations.base.JavaBaseConstants;
+import io.airbyte.integrations.destination.jdbc.JdbcSqlOperations;
+import io.airbyte.protocol.models.AirbyteRecordMessage;
+import java.sql.*;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.util.List;
+import java.util.UUID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TeradataSqlOperations extends JdbcSqlOperations {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(TeradataSqlOperations.class);
+
+  @Override
+  public void insertRecordsInternal(final JdbcDatabase database,
+                                    final List<AirbyteRecordMessage> records,
+                                    final String schemaName,
+                                    final String tmpTableName)
+      throws SQLException {
+    if (records.isEmpty()) {
+      return;
+    }
+    final String insertQueryComponent = String.format("INSERT INTO %s.%s (%s, %s, %s) VALUES", schemaName, tmpTableName,
+        JavaBaseConstants.COLUMN_NAME_AB_ID,
+        JavaBaseConstants.COLUMN_NAME_DATA,
+        JavaBaseConstants.COLUMN_NAME_EMITTED_AT);
+    final String recordQueryComponent = " (?, ?, ?)";
+
+    database.execute(con -> {
+      try {
+
+        PreparedStatement pstmt = con.prepareStatement(insertQueryComponent + recordQueryComponent);
+        for (final AirbyteRecordMessage record : records) {
+          final String uuid = UUID.randomUUID().toString();
+          final String jsonData = Jsons.serialize(formatData(record.getData()));
+          final Timestamp emittedAt = Timestamp.from(Instant.ofEpochMilli(record.getEmittedAt()));
+          pstmt.setString(1, uuid);
+          LOGGER.debug("uuid: " + uuid);
+          LOGGER.debug("jsonData: " + jsonData);
+          LOGGER.debug("jsonData: " + emittedAt);
+          pstmt.setString(2, jsonData);
+          pstmt.setTimestamp(3, emittedAt);
+          pstmt.addBatch();
+        }
+        int updateCounts[] = pstmt.executeBatch();
+      } catch (final SQLException e) {
+        AirbyteTraceMessageUtility.emitSystemErrorTrace(e,
+            "Connector failed while inserting records to staging table");
+        throw new RuntimeException(e);
+      } catch (final Exception e) {
+        AirbyteTraceMessageUtility.emitSystemErrorTrace(e,
+            "Connector failed while inserting records to staging table");
+        throw new RuntimeException(e);
+      } finally {
+
+      }
+    });
+  }
+
+  @Override
+  public void createSchemaIfNotExists(final JdbcDatabase database, final String schemaName) throws Exception {
+    try {
+      database.execute(String.format("CREATE DATABASE \"%s\" AS PERM = 1e9;", schemaName));
+    } catch (SQLException e) {
+      if (e.getMessage().contains("already exists")) {
+        LOGGER.warn("Database " + schemaName + " already exists.");
+      }
+    }
+
+  }
+
+  @Override
+  public void createTableIfNotExists(final JdbcDatabase database, final String schemaName, final String tableName)
+      throws SQLException {
+    try {
+      database.execute(createTableQuery(database, schemaName, tableName));
+    } catch (SQLException e) {
+      if (e.getMessage().contains("already exists")) {
+        LOGGER.warn("Table " + schemaName + "." + tableName + " already exists.");
+      }
+    }
+  }
+
+  @Override
+  public String createTableQuery(final JdbcDatabase database, final String schemaName, final String tableName) {
+    return String.format(
+        "CREATE SET TABLE %s.%s, FALLBACK ( \n" + "%s VARCHAR(256), \n" + "%s JSON, \n" + "%s TIMESTAMP(6) \n"
+            + ");\n",
+        schemaName, tableName, JavaBaseConstants.COLUMN_NAME_AB_ID, JavaBaseConstants.COLUMN_NAME_DATA,
+        JavaBaseConstants.COLUMN_NAME_EMITTED_AT);
+  }
+
+  @Override
+  public void dropTableIfExists(final JdbcDatabase database, final String schemaName, final String tableName)
+      throws SQLException {
+    try {
+      database.execute(dropTableIfExistsQuery(schemaName, tableName));
+    } catch (SQLException e) {
+      AirbyteTraceMessageUtility.emitSystemErrorTrace(e,
+          "Connector failed while dropping table " + schemaName + "." + tableName);
+    }
+  }
+
+  @Override
+  public String truncateTableQuery(final JdbcDatabase database, final String schemaName, final String tableName) {
+    try {
+      return String.format("DELETE %s.%s ALL;\n", schemaName, tableName);
+    } catch (Exception e) {
+      AirbyteTraceMessageUtility.emitSystemErrorTrace(e,
+          "Connector failed while truncating table " + schemaName + "." + tableName);
+    }
+    return "";
+  }
+
+  private String dropTableIfExistsQuery(final String schemaName, final String tableName) {
+    try {
+      return String.format("DROP TABLE  %s.%s;\n", schemaName, tableName);
+    } catch (Exception e) {
+      AirbyteTraceMessageUtility.emitSystemErrorTrace(e,
+          "Connector failed while dropping table " + schemaName + "." + tableName);
+    }
+    return "";
+  }
+
+  @Override
+  public void executeTransaction(final JdbcDatabase database, final List<String> queries) throws Exception {
+    final StringBuilder appendedQueries = new StringBuilder();
+    try {
+      for (final String query : queries) {
+        appendedQueries.append(query);
+      }
+      database.execute(appendedQueries.toString());
+    } catch (SQLException e) {
+      AirbyteTraceMessageUtility.emitSystemErrorTrace(e,
+          "Connector failed while executing queries : " + appendedQueries.toString());
+    }
+  }
+
+}