Destination Oracle - Remove Normalization

airbyte-cdk/java/airbyte-cdk/core/src/main/java/io/airbyte/cdk/integrations/base/JavaBaseConstants.java

@@ -5,6 +5,7 @@
 package io.airbyte.cdk.integrations.base;
 
 import java.util.List;
+import org.apache.commons.lang3.StringUtils;
 
 public final class JavaBaseConstants {
 
@@ -43,4 +44,8 @@ private JavaBaseConstants() {}
 
   public static final String DEFAULT_AIRBYTE_INTERNAL_NAMESPACE = "airbyte_internal";
 
+  public static String upperQuoted(final String column) {
+    return StringUtils.wrap(column.toUpperCase(), "\"");
+  }
+
 }

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/java/io/airbyte/cdk/integrations/destination/jdbc/AbstractJdbcDestination.java

@@ -77,6 +77,16 @@ protected String getConfigSchemaKey() {
     return "schema";
   }
 
+  /**
+   * If the destination should always disable type dedupe, override this method to return true. We
+   * only type and dedupe if we create final tables.
+   *
+   * @return whether the destination should always disable type dedupe
+   */
+  protected boolean shouldAlwaysDisableTypeDedupe() {
+    return false;
+  }
+
   public AbstractJdbcDestination(final String driverClass,
                                  final NamingConventionTransformer namingResolver,
                                  final SqlOperations sqlOperations) {
@@ -323,7 +333,7 @@ private TyperDeduper getV2TyperDeduper(final JsonNode config, final ConfiguredAi
     final NoopV2TableMigrator v2TableMigrator = new NoopV2TableMigrator();
     final DestinationHandler<? extends MinimumDestinationState> destinationHandler =
         getDestinationHandler(databaseName, database, rawNamespaceOverride.orElse(DEFAULT_AIRBYTE_INTERNAL_NAMESPACE));
-    final boolean disableTypeDedupe = config.has(DISABLE_TYPE_DEDUPE) && config.get(DISABLE_TYPE_DEDUPE).asBoolean(false);
+    final boolean disableTypeDedupe = isTypeDedupeDisabled(config);
     final TyperDeduper typerDeduper;
     if (disableTypeDedupe) {
       typerDeduper = new NoOpTyperDeduperWithV1V2Migrations<>(sqlGenerator, destinationHandler, parsedCatalog, migrator, v2TableMigrator, List.of());
@@ -334,4 +344,8 @@ private TyperDeduper getV2TyperDeduper(final JsonNode config, final ConfiguredAi
     return typerDeduper;
   }
 
+  private boolean isTypeDedupeDisabled(final JsonNode config) {
+    return (config.has(DISABLE_TYPE_DEDUPE) && config.get(DISABLE_TYPE_DEDUPE).asBoolean(false)) || shouldAlwaysDisableTypeDedupe();
+  }
+
 }

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/java/io/airbyte/cdk/integrations/destination/jdbc/typing_deduping/NoOpJdbcDestinationHandler.kt

@@ -0,0 +1,45 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.cdk.integrations.destination.jdbc.typing_deduping
+
+import com.fasterxml.jackson.databind.JsonNode
+import io.airbyte.cdk.db.jdbc.JdbcDatabase
+import io.airbyte.cdk.integrations.destination.jdbc.TableDefinition
+import io.airbyte.integrations.base.destination.typing_deduping.AirbyteType
+import io.airbyte.integrations.base.destination.typing_deduping.Sql
+import io.airbyte.integrations.base.destination.typing_deduping.StreamConfig
+import org.jooq.SQLDialect
+
+class NoOpJdbcDestinationHandler<DestinationState>(
+    databaseName: String,
+    jdbcDatabase: JdbcDatabase,
+    rawTableSchemaName: String,
+    sqlDialect: SQLDialect
+) :
+    JdbcDestinationHandler<DestinationState>(
+        databaseName,
+        jdbcDatabase,
+        rawTableSchemaName,
+        sqlDialect
+    ) {
+    override fun execute(sql: Sql?) {
+        throw NotImplementedError("This JDBC Destination Handler does not support typing deduping")
+    }
+
+    override fun toDestinationState(json: JsonNode?): DestinationState {
+        throw NotImplementedError("This JDBC Destination Handler does not support typing deduping")
+    }
+
+    override fun existingSchemaMatchesStreamConfig(
+        stream: StreamConfig?,
+        existingTable: TableDefinition?
+    ): Boolean {
+        throw NotImplementedError("This JDBC Destination Handler does not support typing deduping")
+    }
+
+    override fun toJdbcTypeName(airbyteType: AirbyteType?): String {
+        throw NotImplementedError("This JDBC Destination Handler does not support typing deduping")
+    }
+}

airbyte-integrations/connectors/destination-clickhouse-strict-encrypt/src/test/resources/expected_spec.json

@@ -3,7 +3,7 @@
   "supportsIncremental": true,
   "supportsNormalization": true,
   "supportsDBT": false,
-  "supported_destination_sync_modes": ["overwrite", "append", "append_dedup"],
+  "supported_destination_sync_modes": ["overwrite", "append"],
   "connectionSpecification": {
     "$schema": "http://json-schema.org/draft-07/schema#",
     "title": "ClickHouse Destination Spec",
@@ -52,6 +52,12 @@
         "type": "string",
         "order": 5
       },
+      "raw_data_schema": {
+        "type": "string",
+        "description": "The schema to write raw tables into (default: airbyte_internal)",
+        "title": "Raw Table Schema Name",
+        "order": 7
+      },
       "tunnel_method": {
         "type": "object",
         "title": "SSH Tunnel Method",

airbyte-integrations/connectors/destination-oracle-strict-encrypt/build.gradle

@@ -4,9 +4,9 @@ plugins {
 }
 
 airbyteJavaConnector {
-    cdkVersionRequired = '0.2.0'
-    features = ['db-destinations']
-    useLocalCdk = false
+    cdkVersionRequired = '0.23.16'
+    features = ['db-destinations', 's3-destinations', 'typing-deduping']
+    useLocalCdk = true
 }
 
 //remove once upgrading the CDK version to 0.4.x or later

airbyte-integrations/connectors/destination-oracle-strict-encrypt/gradle.properties

@@ -0,0 +1 @@
+junitMethodExecutionTimeout = 30 m

airbyte-integrations/connectors/destination-oracle-strict-encrypt/metadata.yaml

@@ -13,11 +13,17 @@ data:
   icon: oracle.svg
   license: ELv2
   name: Oracle
-  normalizationConfig:
-    normalizationIntegrationType: oracle
-    normalizationRepository: airbyte/normalization-oracle
-    normalizationTag: 0.4.1
   releaseStage: alpha
+  releases:
+    breakingChanges:
+      1.0.0:
+        upgradeDeadline: "2024-03-15"
+        message: >
+          This version removes the option to use "normalization" with Oracle. It also changes
+          the schema and database of Airbyte's "raw" tables to be compatible with the new
+          [Destinations V2](https://docs.airbyte.com/release_notes/upgrading_to_destinations_v2/#what-is-destinations-v2)
+          format. These changes will likely require updates to downstream dbt / SQL models.
+          Selecting `Upgrade` will upgrade **all** connections using this destination at their next sync.
   documentationUrl: https://docs.airbyte.com/integrations/destinations/oracle
   supportsDbt: true
   tags:

airbyte-integrations/connectors/destination-oracle-strict-encrypt/src/test-integration/java/io/airbyte/integrations/destination/oracle_strict_encrypt/OracleStrictEncryptDestinationAcceptanceTest.java

@@ -17,13 +17,14 @@
 import io.airbyte.cdk.db.jdbc.DefaultJdbcDatabase;
 import io.airbyte.cdk.db.jdbc.JdbcDatabase;
 import io.airbyte.cdk.db.jdbc.JdbcUtils;
+import io.airbyte.cdk.integrations.base.JavaBaseConstants;
 import io.airbyte.cdk.integrations.destination.StandardNameTransformer;
 import io.airbyte.cdk.integrations.standardtest.destination.DestinationAcceptanceTest;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.string.Strings;
-import io.airbyte.integrations.destination.oracle.OracleDestination;
 import io.airbyte.integrations.destination.oracle.OracleNameTransformer;
 import java.sql.SQLException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
@@ -73,7 +74,7 @@ protected List<JsonNode> retrieveRecords(final TestDestinationEnv env,
     return retrieveRecordsFromTable(namingResolver.getRawTableName(streamName), namespace)
         .stream()
         .map(r -> Jsons.deserialize(
-            r.get(OracleDestination.COLUMN_NAME_DATA.replace("\"", "")).asText()))
+            r.get(JavaBaseConstants.COLUMN_NAME_DATA).asText()))
         .collect(Collectors.toList());
   }
 
@@ -113,16 +114,15 @@ protected List<String> resolveIdentifier(final String identifier) {
 
   private List<JsonNode> retrieveRecordsFromTable(final String tableName, final String schemaName)
       throws SQLException {
-    final String query = String.format("SELECT * FROM %s.%s ORDER BY %s ASC", schemaName, tableName, OracleDestination.COLUMN_NAME_EMITTED_AT);
-
-    try (final DSLContext dslContext = getDslContext(config)) {
-      final List<org.jooq.Record> result = getDatabase(dslContext).query(ctx -> ctx.fetch(query).stream().toList());
-      return result
-          .stream()
-          .map(r -> r.formatJSON(JSON_FORMAT))
-          .map(Jsons::deserialize)
-          .collect(Collectors.toList());
-    }
+    final String query =
+        String.format("SELECT * FROM %s.%s ORDER BY %s ASC", schemaName, tableName, JavaBaseConstants.COLUMN_NAME_AB_EXTRACTED_AT.toUpperCase());
+    final DSLContext dslContext = getDslContext(config);
+    final List<org.jooq.Record> result = getDatabase(dslContext).query(ctx -> ctx.fetch(query).stream().toList());
+    return result
+        .stream()
+        .map(r -> r.formatJSON(JSON_FORMAT))
+        .map(Jsons::deserialize)
+        .collect(Collectors.toList());
   }
 
   private static Database getDatabase(final DSLContext dslContext) {
@@ -151,15 +151,13 @@ protected void setup(final TestDestinationEnv testEnv, final HashSet<String> TES
     db.start();
 
     config = getConfig(db);
+    final DSLContext dslContext = getDslContext(config);
+    final Database database = getDatabase(dslContext);
+    database.query(
+        ctx -> ctx.fetch(String.format("CREATE USER %s IDENTIFIED BY %s", schemaName, schemaName)));
+    database.query(ctx -> ctx.fetch(String.format("GRANT ALL PRIVILEGES TO %s", schemaName)));
 
-    try (final DSLContext dslContext = getDslContext(config)) {
-      final Database database = getDatabase(dslContext);
-      database.query(
-          ctx -> ctx.fetch(String.format("CREATE USER %s IDENTIFIED BY %s", schemaName, schemaName)));
-      database.query(ctx -> ctx.fetch(String.format("GRANT ALL PRIVILEGES TO %s", schemaName)));
-
-      ((ObjectNode) config).put(JdbcUtils.SCHEMA_KEY, dbName);
-    }
+    ((ObjectNode) config).put(JdbcUtils.SCHEMA_KEY, dbName);
   }
 
   @Override
@@ -182,7 +180,8 @@ public void testEncryption() throws SQLException {
                 config.get(JdbcUtils.PORT_KEY).asInt(),
                 config.get("sid").asText()),
             JdbcUtils.parseJdbcParameters("oracle.net.encryption_client=REQUIRED;" +
-                "oracle.net.encryption_types_client=( " + algorithm + " )", ";"));
+                "oracle.net.encryption_types_client=( " + algorithm + " )", ";"),
+            Duration.ofMinutes(5));
     final JdbcDatabase database = new DefaultJdbcDatabase(dataSource);
 
     final String networkServiceBanner =
@@ -208,7 +207,8 @@ public void testCheckProtocol() throws SQLException {
                 config.get(JdbcUtils.PORT_KEY).asInt(),
                 config.get("sid").asText()),
             JdbcUtils.parseJdbcParameters("oracle.net.encryption_client=REQUIRED;" +
-                "oracle.net.encryption_types_client=( " + algorithm + " )", ";"));
+                "oracle.net.encryption_types_client=( " + algorithm + " )", ";"),
+            Duration.ofMinutes(5));
     final JdbcDatabase database = new DefaultJdbcDatabase(dataSource);
 
     final String networkServiceBanner = "SELECT sys_context('USERENV', 'NETWORK_PROTOCOL') as network_protocol FROM dual";

airbyte-integrations/connectors/destination-oracle-strict-encrypt/src/test/resources/expected_spec.json

@@ -59,6 +59,126 @@
         "examples": ["airbyte"],
         "default": "airbyte",
         "order": 6
+      },
+      "raw_data_schema": {
+        "type": "string",
+        "description": "The schema to write raw tables into (default: airbyte_internal)",
+        "title": "Raw Table Schema Name",
+        "order": 7
+      },
+      "tunnel_method": {
+        "type": "object",
+        "title": "SSH Tunnel Method",
+        "description": "Whether to initiate an SSH tunnel before connecting to the database, and if so, which kind of authentication to use.",
+        "oneOf": [
+          {
+            "title": "No Tunnel",
+            "required": ["tunnel_method"],
+            "properties": {
+              "tunnel_method": {
+                "description": "No ssh tunnel needed to connect to database",
+                "type": "string",
+                "const": "NO_TUNNEL",
+                "order": 0
+              }
+            }
+          },
+          {
+            "title": "SSH Key Authentication",
+            "required": [
+              "tunnel_method",
+              "tunnel_host",
+              "tunnel_port",
+              "tunnel_user",
+              "ssh_key"
+            ],
+            "properties": {
+              "tunnel_method": {
+                "description": "Connect through a jump server tunnel host using username and ssh key",
+                "type": "string",
+                "const": "SSH_KEY_AUTH",
+                "order": 0
+              },
+              "tunnel_host": {
+                "title": "SSH Tunnel Jump Server Host",
+                "description": "Hostname of the jump server host that allows inbound ssh tunnel.",
+                "type": "string",
+                "order": 1
+              },
+              "tunnel_port": {
+                "title": "SSH Connection Port",
+                "description": "Port on the proxy/jump server that accepts inbound ssh connections.",
+                "type": "integer",
+                "minimum": 0,
+                "maximum": 65536,
+                "default": 22,
+                "examples": ["22"],
+                "order": 2
+              },
+              "tunnel_user": {
+                "title": "SSH Login Username",
+                "description": "OS-level username for logging into the jump server host.",
+                "type": "string",
+                "order": 3
+              },
+              "ssh_key": {
+                "title": "SSH Private Key",
+                "description": "OS-level user account ssh key credentials in RSA PEM format ( created with ssh-keygen -t rsa -m PEM -f myuser_rsa )",
+                "type": "string",
+                "airbyte_secret": true,
+                "multiline": true,
+                "order": 4
+              }
+            }
+          },
+          {
+            "title": "Password Authentication",
+            "required": [
+              "tunnel_method",
+              "tunnel_host",
+              "tunnel_port",
+              "tunnel_user",
+              "tunnel_user_password"
+            ],
+            "properties": {
+              "tunnel_method": {
+                "description": "Connect through a jump server tunnel host using username and password authentication",
+                "type": "string",
+                "const": "SSH_PASSWORD_AUTH",
+                "order": 0
+              },
+              "tunnel_host": {
+                "title": "SSH Tunnel Jump Server Host",
+                "description": "Hostname of the jump server host that allows inbound ssh tunnel.",
+                "type": "string",
+                "order": 1
+              },
+              "tunnel_port": {
+                "title": "SSH Connection Port",
+                "description": "Port on the proxy/jump server that accepts inbound ssh connections.",
+                "type": "integer",
+                "minimum": 0,
+                "maximum": 65536,
+                "default": 22,
+                "examples": ["22"],
+                "order": 2
+              },
+              "tunnel_user": {
+                "title": "SSH Login Username",
+                "description": "OS-level username for logging into the jump server host",
+                "type": "string",
+                "order": 3
+              },
+              "tunnel_user_password": {
+                "title": "Password",
+                "description": "OS-level password for logging into the jump server host",
+                "type": "string",
+                "airbyte_secret": true,
+                "order": 4
+              }
+            }
+          }
+        ]
       }
     }
   }

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

@@ -4,9 +4,9 @@ plugins {
 }
 
 airbyteJavaConnector {
-    cdkVersionRequired = '0.2.0'
-    features = ['db-destinations']
-    useLocalCdk = false
+    cdkVersionRequired = '0.23.16'
+    features = ['db-destinations', 's3-destinations', 'typing-deduping']
+    useLocalCdk = true
 }
 
 //remove once upgrading the CDK version to 0.4.x or later

airbyte-integrations/connectors/destination-oracle/gradle.properties

@@ -0,0 +1 @@
+junitMethodExecutionTimeout = 30 m