redshift_catalog_parser_update

airbytehq · May 10, 2024 · 2cb49a4 · 2cb49a4
1 parent 0724d77
commit 2cb49a4
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 10 deletions.
diff --git a/airbyte-integrations/connectors/destination-redshift/build.gradle b/airbyte-integrations/connectors/destination-redshift/build.gradle
@@ -6,7 +6,7 @@ plugins {
 airbyteJavaConnector {
     cdkVersionRequired = '0.34.0'
     features = ['db-destinations', 's3-destinations', 'typing-deduping']
-    useLocalCdk = false
+    useLocalCdk = true
 }
 
 java {

diff --git a/.../main/java/io/airbyte/integrations/destination/redshift/RedshiftStagingS3Destination.java b/.../main/java/io/airbyte/integrations/destination/redshift/RedshiftStagingS3Destination.java
@@ -241,7 +241,7 @@ public SerializedAirbyteMessageConsumer getSerializedMessageConsumer(final JsonN
       catalogParser = new CatalogParser(sqlGenerator, rawNamespace);
     } else {
       rawNamespace = JavaBaseConstants.DEFAULT_AIRBYTE_INTERNAL_NAMESPACE;
-      catalogParser = new CatalogParser(sqlGenerator);
+      catalogParser = new CatalogParser(sqlGenerator, defaultNamespace);
     }
     final RedshiftDestinationHandler redshiftDestinationHandler = new RedshiftDestinationHandler(databaseName, database, rawNamespace);
     parsedCatalog = catalogParser.parseCatalog(catalog);

diff --git a/...o/airbyte/integrations/destination/redshift/typing_deduping/RedshiftSqlGeneratorTest.java b/...o/airbyte/integrations/destination/redshift/typing_deduping/RedshiftSqlGeneratorTest.java
@@ -79,18 +79,19 @@ public void setup() {
     columns.put(redshiftSqlGenerator.buildColumnId("_ab_cdc_deleted_at"), AirbyteProtocolType.TIMESTAMP_WITH_TIMEZONE);
     incrementalDedupStream = new StreamConfig(
         streamId,
-        SyncMode.INCREMENTAL,
         DestinationSyncMode.APPEND_DEDUP,
         primaryKey,
         Optional.of(cursor),
-        columns);
+        columns,
+        0,
+        0,
+        0);
     incrementalAppendStream = new StreamConfig(
         streamId,
-        SyncMode.INCREMENTAL,
         DestinationSyncMode.APPEND,
         primaryKey,
         Optional.of(cursor),
-        columns);
+        columns, 0, 0, 0);
   }
 
   @Test
@@ -132,11 +133,10 @@ public void test2000ColumnSql() {
     }
     final Sql generatedSql = redshiftSqlGenerator.updateTable(new StreamConfig(
         streamId,
-        SyncMode.INCREMENTAL,
         DestinationSyncMode.APPEND_DEDUP,
         primaryKey,
         Optional.of(cursor),
-        columns), "unittest", Optional.of(Instant.parse("2023-02-15T18:35:24.00Z")), false);
+        columns, 0, 0, 0), "unittest", Optional.of(Instant.parse("2023-02-15T18:35:24.00Z")), false);
     // This should not throw an exception.
     assertFalse(generatedSql.transactions().isEmpty());
   }

diff --git a/...grations/destination/redshift/typing_deduping/RedshiftSuperLimitationTransformerTest.java b/...grations/destination/redshift/typing_deduping/RedshiftSuperLimitationTransformerTest.java
@@ -57,11 +57,10 @@ public void setup() {
     final StreamId streamId = new StreamId("test_schema", "users_final", "test_schema", "users_raw", "test_schema", "users_final");
     StreamConfig streamConfig = new StreamConfig(
         streamId,
-        SyncMode.INCREMENTAL,
         DestinationSyncMode.APPEND_DEDUP,
         primaryKey,
         Optional.empty(),
-        columns);
+        columns,0 ,0, 0);
     final ParsedCatalog parsedCatalog = new ParsedCatalog(List.of(streamConfig));
     transformer = new RedshiftSuperLimitationTransformer(parsedCatalog, "test_schema");
   }