Destination Clickhouse - 1.0, remove normalization

airbyte-cdk/java/airbyte-cdk/build.gradle

@@ -1,13 +1,22 @@
+import org.jetbrains.kotlin.gradle.dsl.JvmTarget
+import org.jetbrains.kotlin.gradle.dsl.KotlinVersion
+plugins {
+    id 'org.jetbrains.kotlin.jvm' version '1.9.22'
+}
+
 final var cdkVersion = {
     var props = new Properties()
     file("core/src/main/resources/version.properties").withInputStream(props::load)
     return props.getProperty('version', 'undefined')
 }()
 
+
+
 allprojects {
     apply plugin: 'java-library'
     apply plugin: 'maven-publish'
     apply plugin: 'java-test-fixtures'
+    apply plugin: 'org.jetbrains.kotlin.jvm'
 
     group 'io.airbyte.cdk'
 
@@ -44,6 +53,19 @@ allprojects {
                 }
             }
         }
+
+        compileKotlin {
+            compilerOptions {
+                jvmTarget = JvmTarget.JVM_21
+                languageVersion = KotlinVersion.KOTLIN_1_9
+            }
+        }
+        compileTestKotlin {
+            compilerOptions {
+                jvmTarget = JvmTarget.JVM_21
+                languageVersion = KotlinVersion.KOTLIN_1_9
+            }
+        }
     }
 
     project.configurations {

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

@@ -4,7 +4,13 @@
 
 package io.airbyte.cdk.integrations.base.ssh;
 
-import static io.airbyte.cdk.integrations.base.ssh.SshTunnel.*;
+import static io.airbyte.cdk.integrations.base.ssh.SshTunnel.CONNECTION_OPTIONS_KEY;
+import static io.airbyte.cdk.integrations.base.ssh.SshTunnel.GLOBAL_HEARTBEAT_INTERVAL_DEFAULT_IN_MILLIS;
+import static io.airbyte.cdk.integrations.base.ssh.SshTunnel.GLOBAL_HEARTBEAT_INTERVAL_KEY;
+import static io.airbyte.cdk.integrations.base.ssh.SshTunnel.SESSION_HEARTBEAT_INTERVAL_DEFAULT_IN_MILLIS;
+import static io.airbyte.cdk.integrations.base.ssh.SshTunnel.SESSION_HEARTBEAT_INTERVAL_KEY;
+import static io.airbyte.cdk.integrations.base.ssh.SshTunnel.getInstance;
+import static io.airbyte.cdk.integrations.base.ssh.SshTunnel.sshWrap;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;

airbyte-cdk/java/airbyte-cdk/core/src/main/java/io/airbyte/cdk/integrations/destination/StandardNameTransformer.java

@@ -31,6 +31,7 @@ public String getNamespace(final String namespace) {
   }
 
   @Override
+  // @Deprecated see https://github.com/airbytehq/airbyte/issues/35333
   public String getRawTableName(final String streamName) {
     return convertStreamName("_airbyte_raw_" + streamName);
   }

airbyte-cdk/java/airbyte-cdk/core/src/main/java/io/airbyte/cdk/integrations/util/ConfiguredCatalogUtil.kt

@@ -0,0 +1,20 @@
+package io.airbyte.cdk.integrations.util
+
+import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog
+
+/**
+ * For streams in [catalog] which do not have a namespace specified, explicitly set their namespace
+ * to the [defaultNamespace]
+ */
+ fun addDefaultNamespaceToStreams(catalog: ConfiguredAirbyteCatalog, defaultNamespace: String?) {
+     if (defaultNamespace == null) {
+         return
+     }
+    // TODO: This logic exists in all V2 destinations.
+    // This is sad that if we forget to add this, there will be a null pointer during parseCatalog
+    for (catalogStream in catalog.streams) {
+        if (catalogStream.stream.namespace.isNullOrEmpty()) {
+            catalogStream.stream.namespace = defaultNamespace
+        }
+    }
+}

airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties

@@ -1 +1 @@
-version=0.21.4
+version=0.22.1

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

@@ -5,6 +5,7 @@
 package io.airbyte.cdk.integrations.destination.jdbc;
 
 import static io.airbyte.cdk.integrations.base.errors.messages.ErrorMessage.getErrorMessage;
+import static io.airbyte.cdk.integrations.util.ConfiguredCatalogUtilKt.addDefaultNamespaceToStreams;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.annotations.VisibleForTesting;
@@ -40,7 +41,6 @@
 import io.airbyte.protocol.models.v0.AirbyteConnectionStatus.Status;
 import io.airbyte.protocol.models.v0.AirbyteMessage;
 import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog;
-import io.airbyte.protocol.models.v0.ConfiguredAirbyteStream;
 import java.sql.SQLException;
 import java.util.List;
 import java.util.Map;
@@ -49,7 +49,6 @@
 import java.util.function.Consumer;
 import javax.sql.DataSource;
 import org.apache.commons.lang3.NotImplementedException;
-import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -72,6 +71,10 @@ protected SqlOperations getSqlOperations() {
     return sqlOperations;
   }
 
+  protected String getConfigSchemaKey() {
+    return "schema";
+  }
+
   public AbstractJdbcDestination(final String driverClass,
                                  final NamingConventionTransformer namingResolver,
                                  final SqlOperations sqlOperations) {
@@ -276,44 +279,16 @@ public SerializedAirbyteMessageConsumer getSerializedMessageConsumer(final JsonN
                                                                        final ConfiguredAirbyteCatalog catalog,
                                                                        final Consumer<AirbyteMessage> outputRecordCollector)
       throws Exception {
-    final DataSource dataSource = getDataSource(config);
-    final JdbcDatabase database = getDatabase(dataSource);
+    final JdbcDatabase database = getDatabase(getDataSource(config));
+    final String defaultNamespace;
+    final TyperDeduper typerDeduper;
     if (TypingAndDedupingFlag.isDestinationV2()) {
-      // TODO: This logic exists in all V2 destinations.
-      // This is sad that if we forget to add this, there will be a null pointer during parseCatalog
-      final String defaultNamespace = config.get("schema").asText();
-      for (final ConfiguredAirbyteStream stream : catalog.getStreams()) {
-        if (StringUtils.isEmpty(stream.getStream().getNamespace())) {
-          stream.getStream().setNamespace(defaultNamespace);
-        }
-      }
-      final JdbcSqlGenerator sqlGenerator = getSqlGenerator();
-      final ParsedCatalog parsedCatalog = TypingAndDedupingFlag.getRawNamespaceOverride(RAW_SCHEMA_OVERRIDE)
-          .map(override -> new CatalogParser(sqlGenerator, override))
-          .orElse(new CatalogParser(sqlGenerator))
-          .parseCatalog(catalog);
-      final String databaseName = getDatabaseName(config);
-      final var migrator = new JdbcV1V2Migrator(namingResolver, database, databaseName);
-      final NoopV2TableMigrator v2TableMigrator = new NoopV2TableMigrator();
-      final DestinationHandler<TableDefinition> destinationHandler = getDestinationHandler(databaseName, database);
-      final boolean disableTypeDedupe = config.has(DISABLE_TYPE_DEDUPE) && config.get(DISABLE_TYPE_DEDUPE).asBoolean(false);
-      final TyperDeduper typerDeduper;
-      if (disableTypeDedupe) {
-        typerDeduper = new NoOpTyperDeduperWithV1V2Migrations<>(sqlGenerator, destinationHandler, parsedCatalog, migrator, v2TableMigrator,
-            8);
-      } else {
-        typerDeduper =
-            new DefaultTyperDeduper<>(sqlGenerator, destinationHandler, parsedCatalog, migrator, v2TableMigrator, 8);
-      }
-      return JdbcBufferedConsumerFactory.createAsync(
-          outputRecordCollector,
-          database,
-          sqlOperations,
-          namingResolver,
-          config,
-          catalog,
-          defaultNamespace,
-          typerDeduper);
+      defaultNamespace = config.get(getConfigSchemaKey()).asText();
+      addDefaultNamespaceToStreams(catalog, defaultNamespace);
+      typerDeduper = getV2TyperDeduper(config, catalog, database);
+    } else {
+      defaultNamespace = null;
+      typerDeduper = new NoopTyperDeduper();
     }
     return JdbcBufferedConsumerFactory.createAsync(
         outputRecordCollector,
@@ -322,8 +297,37 @@ public SerializedAirbyteMessageConsumer getSerializedMessageConsumer(final JsonN
         namingResolver,
         config,
         catalog,
-        null,
-        new NoopTyperDeduper());
+        defaultNamespace,
+        typerDeduper);
+  }
+
+  /**
+   * Creates the appropriate TyperDeduper class for the jdbc destination and the user's configuration
+   *
+   * @param config the configuration for the connection
+   * @param catalog the catalog for the connection
+   * @param database a database instance
+   * @return the appropriate TyperDeduper instance for this connection.
+   */
+  private TyperDeduper getV2TyperDeduper(final JsonNode config, final ConfiguredAirbyteCatalog catalog, final JdbcDatabase database) {
+    final JdbcSqlGenerator sqlGenerator = getSqlGenerator();
+    final ParsedCatalog parsedCatalog = TypingAndDedupingFlag.getRawNamespaceOverride(RAW_SCHEMA_OVERRIDE)
+        .map(override -> new CatalogParser(sqlGenerator, override))
+        .orElse(new CatalogParser(sqlGenerator))
+        .parseCatalog(catalog);
+    final String databaseName = getDatabaseName(config);
+    final var migrator = new JdbcV1V2Migrator(namingResolver, database, databaseName);
+    final NoopV2TableMigrator v2TableMigrator = new NoopV2TableMigrator();
+    final DestinationHandler<TableDefinition> destinationHandler = getDestinationHandler(databaseName, database);
+    final boolean disableTypeDedupe = !config.has(DISABLE_TYPE_DEDUPE) || config.get(DISABLE_TYPE_DEDUPE).asBoolean(false);
+    final TyperDeduper typerDeduper;
+    if (disableTypeDedupe) {
+      typerDeduper = new NoOpTyperDeduperWithV1V2Migrations<>(sqlGenerator, destinationHandler, parsedCatalog, migrator, v2TableMigrator);
+    } else {
+      typerDeduper =
+          new DefaultTyperDeduper<>(sqlGenerator, destinationHandler, parsedCatalog, migrator, v2TableMigrator);
+    }
+    return typerDeduper;
   }
 
 }

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

@@ -0,0 +1,66 @@
+package io.airbyte.cdk.integrations.destination.jdbc.typing_deduping
+
+import io.airbyte.cdk.integrations.destination.NamingConventionTransformer
+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.ColumnId
+import io.airbyte.integrations.base.destination.typing_deduping.StreamConfig
+import org.jooq.Condition
+import org.jooq.DataType
+import org.jooq.Field
+import org.jooq.SQLDialect
+import java.util.*
+
+/**
+ * Some Destinations do not support Typing and Deduping but have the updated raw table format
+ * SqlGenerator implementations are only for "final" tables and are a required input for
+ * TyperDeduper classes. This implementation appeases that requirement but does not implement
+ * any "final" table operations.
+ */
+class RawOnlySqlGenerator(private val namingTransformer: NamingConventionTransformer) :
+    JdbcSqlGenerator(namingTransformer) {
+    override fun getStructType(): DataType<*>? {
+        throw NotImplementedError("This Destination does not support final tables")
+    }
+
+    override fun getArrayType(): DataType<*>? {
+        throw NotImplementedError("This Destination does not support final tables")
+    }
+
+    override fun getWidestType(): DataType<*>? {
+        throw NotImplementedError("This Destination does not support final tables")
+    }
+
+    override fun getDialect(): SQLDialect? {
+        throw NotImplementedError("This Destination does not support final tables")
+    }
+
+    override fun extractRawDataFields(
+        columns: LinkedHashMap<ColumnId, AirbyteType>,
+        useExpensiveSaferCasting: Boolean
+    ): List<Field<*>>? {
+        throw NotImplementedError("This Destination does not support final tables")
+    }
+
+    override fun buildAirbyteMetaColumn(columns: LinkedHashMap<ColumnId, AirbyteType>): Field<*>? {
+        throw NotImplementedError("This Destination does not support final tables")
+    }
+
+    override fun cdcDeletedAtNotNullCondition(): Condition? {
+        throw NotImplementedError("This Destination does not support final tables")
+    }
+
+    override fun getRowNumber(
+        primaryKey: List<ColumnId>,
+        cursorField: Optional<ColumnId>
+    ): Field<Int>? {
+        throw NotImplementedError("This Destination does not support final tables")
+    }
+
+    override fun existingSchemaMatchesStreamConfig(
+        stream: StreamConfig,
+        existingTable: TableDefinition
+    ): Boolean {
+        throw NotImplementedError("This Destination does not support final tables")
+    }
+}

airbyte-cdk/java/airbyte-cdk/db-destinations/src/testFixtures/java/io/airbyte/cdk/integrations/standardtest/destination/DestinationAcceptanceTest.java

@@ -789,7 +789,7 @@ public void testIncrementalDedupeSync() throws Exception {
         .map(record -> Jsons.deserialize(record, AirbyteMessage.class))
         .collect(Collectors.toList());
     final JsonNode config = getConfig();
-    runSyncAndVerifyStateOutput(config, firstSyncMessages, configuredCatalog, true);
+    runSyncAndVerifyStateOutput(config, firstSyncMessages, configuredCatalog, supportsNormalization());
 
     final List<AirbyteMessage> secondSyncMessages = Lists.newArrayList(
         new AirbyteMessage()
@@ -820,7 +820,7 @@ public void testIncrementalDedupeSync() throws Exception {
             .withType(Type.STATE)
             .withState(new AirbyteStateMessage().withData(
                 Jsons.jsonNode(ImmutableMap.of("checkpoint", 2)))));
-    runSyncAndVerifyStateOutput(config, secondSyncMessages, configuredCatalog, true);
+    runSyncAndVerifyStateOutput(config, secondSyncMessages, configuredCatalog, false);
 
     final List<AirbyteMessage> expectedMessagesAfterSecondSync = new ArrayList<>();
     expectedMessagesAfterSecondSync.addAll(firstSyncMessages);
@@ -853,22 +853,11 @@ public void testIncrementalDedupeSync() throws Exception {
     final String defaultSchema = getDefaultSchema(config);
     retrieveRawRecordsAndAssertSameMessages(catalog, expectedMessagesAfterSecondSync,
         defaultSchema);
-    final List<AirbyteRecordMessage> actualMessages = retrieveNormalizedRecords(catalog,
-        defaultSchema);
-    assertSameMessages(expectedMessages, actualMessages, true);
-  }
-
-  private String generateBigString(final int addExtraCharacters) {
-    final int length = getMaxRecordValueLimit() + addExtraCharacters;
-    return RANDOM
-        .ints('a', 'z' + 1)
-        .limit(length)
-        .collect(StringBuilder::new, StringBuilder::appendCodePoint, StringBuilder::append)
-        .toString();
-  }
-
-  protected int getGenerateBigStringAddExtraCharacters() {
-    return 0;
+    if (normalizationFromDefinition()) {
+      final List<AirbyteRecordMessage> actualMessages = retrieveNormalizedRecords(catalog,
+          defaultSchema);
+      assertSameMessages(expectedMessages, actualMessages, true);
+    }
   }
 
   /**
@@ -1347,7 +1336,7 @@ private List<AirbyteMessage> runSync(
 
     destination.close();
 
-    if (!runNormalization || (runNormalization && supportsInDestinationNormalization())) {
+    if (!runNormalization || (supportsInDestinationNormalization())) {
       return destinationOutput;
     }
 
@@ -1860,6 +1849,10 @@ public Stream<? extends Arguments> provideArguments(final ExtensionContext conte
 
   }
 
+  private boolean supportsNormalization() {
+    return supportsInDestinationNormalization() || normalizationFromDefinition();
+  }
+
   private static <V0, V1> V0 convertProtocolObject(final V1 v1, final Class<V0> klass) {
     return Jsons.object(Jsons.jsonNode(v1), klass);
   }