Destination redshift: escape string literal at runtime (#34544)

airbytehq · Jan 26, 2024 · 0f7f77f · 0f7f77f
1 parent 96b9f5a
commit 0f7f77f
Show file tree

Hide file tree

Showing 4 changed files with 23 additions and 13 deletions.
diff --git a/airbyte-integrations/connectors/destination-redshift/metadata.yaml b/airbyte-integrations/connectors/destination-redshift/metadata.yaml
@@ -5,7 +5,7 @@ data:
   connectorSubtype: database
   connectorType: destination
   definitionId: f7a7d195-377f-cf5b-70a5-be6b819019dc
-  dockerImageTag: 2.1.2
+  dockerImageTag: 2.1.3
   dockerRepository: airbyte/destination-redshift
   documentationUrl: https://docs.airbyte.com/integrations/destinations/redshift
   githubIssueLabel: destination-redshift

diff --git a/...n/java/io/airbyte/integrations/destination/redshift/operations/RedshiftSqlOperations.java b/...n/java/io/airbyte/integrations/destination/redshift/operations/RedshiftSqlOperations.java
@@ -31,6 +31,7 @@
 import org.jooq.InsertValuesStep4;
 import org.jooq.Record;
 import org.jooq.SQLDialect;
+import org.jooq.conf.ParamType;
 import org.jooq.impl.DefaultDataType;
 import org.jooq.impl.SQLDataType;
 import org.slf4j.Logger;
@@ -133,11 +134,17 @@ protected void insertRecordsInternalV2(final JdbcDatabase database,
           for (final PartialAirbyteMessage record : batch) {
             insert = insert.values(
                 val(UUID.randomUUID().toString()),
-                function("JSON_PARSE", String.class, val(record.getSerialized())),
+                function("JSON_PARSE", String.class, val(escapeStringLiteral(record.getSerialized()))),
                 val(Instant.ofEpochMilli(record.getRecord().getEmittedAt()).atOffset(ZoneOffset.UTC)),
                 val((OffsetDateTime) null));
           }
-          insert.execute();
+          // Intentionally don't use insert.execute().
+          // jooq will try to use a parameterized query if there are not too many query params.
+          // This means that for small queries, we would need to not-escape backslashes,
+          // but for large queries we _would_ need to escape them.
+          // Instead, force jooq to always inline params.
+          // This allows us to always escape backslashes.
+          connection.createStatement().execute(insert.getSQL(ParamType.INLINED));
           LOGGER.info("Executed batch size: {}, {}, {}", batch.size(), schemaName, tableName);
         }
       });
@@ -147,4 +154,15 @@ protected void insertRecordsInternalV2(final JdbcDatabase database,
     }
   }
 
+  public static String escapeStringLiteral(final String str) {
+    if (str == null) {
+      return null;
+    } else {
+      // jooq handles most things
+      // but we need to manually escape backslashes because postgres and redshift have
+      // different backslash handling.
+      return str.replace("\\", "\\\\");
+    }
+  }
+
 }
diff --git a/...ntegrations/destination/redshift/typing_deduping/RedshiftSqlGeneratorIntegrationTest.java b/...ntegrations/destination/redshift/typing_deduping/RedshiftSqlGeneratorIntegrationTest.java
@@ -5,6 +5,7 @@
 package io.airbyte.integrations.destination.redshift.typing_deduping;
 
 import static io.airbyte.cdk.db.jdbc.DateTimeConverter.putJavaSQLTime;
+import static io.airbyte.integrations.destination.redshift.operations.RedshiftSqlOperations.escapeStringLiteral;
 import static org.junit.jupiter.api.Assertions.assertAll;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -205,14 +206,4 @@ public void testCreateTableIncremental() throws Exception {
     // TODO assert on table clustering, etc.
   }
 
-  private static String escapeStringLiteral(final String str) {
-    if (str == null) {
-      return null;
-    } else {
-      // jooq handles most things
-      // but we need to manually escape backslashes for some reason
-      return str.replace("\\", "\\\\");
-    }
-  }
-
 }
diff --git a/docs/integrations/destinations/redshift.md b/docs/integrations/destinations/redshift.md
@@ -237,6 +237,7 @@ Each stream will be output into its own raw table in Redshift. Each table will c
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                                                                                          |
 |:--------|:-----------|:-----------------------------------------------------------|:-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 2.1.3   | 2024-01-26 | [34544](https://github.com/airbytehq/airbyte/pull/34544)   | Proper string-escaping in raw tables                                                                                                                                                                             |
 | 2.1.2   | 2024-01-24 | [34451](https://github.com/airbytehq/airbyte/pull/34451)   | Improve logging for unparseable input                                                                                                                                                                            |
 | 2.1.1   | 2024-01-24 | [34458](https://github.com/airbytehq/airbyte/pull/34458)   | Improve error reporting                                                                                                                                                                                          |
 | 2.1.0   | 2024-01-24 | [34467](https://github.com/airbytehq/airbyte/pull/34467)   | Upgrade CDK to 0.14.0                                                                                                                                                                                            |