diff --git a/airbyte-integrations/connectors/destination-postgres/metadata.yaml b/airbyte-integrations/connectors/destination-postgres/metadata.yaml index 6491a2fe2e15..b617f7925d20 100644 --- a/airbyte-integrations/connectors/destination-postgres/metadata.yaml +++ b/airbyte-integrations/connectors/destination-postgres/metadata.yaml @@ -6,7 +6,7 @@ data: connectorSubtype: database connectorType: destination definitionId: 25c5221d-dce2-4163-ade9-739ef790f503 - dockerImageTag: 3.0.16 + dockerImageTag: 3.0.17-rc.1 dockerRepository: airbyte/destination-postgres documentationUrl: https://docs.airbyte.com/integrations/destinations/postgres githubIssueLabel: destination-postgres diff --git a/airbyte-integrations/connectors/destination-postgres/src/main/kotlin/io/airbyte/integrations/destination/postgres/write/load/PostgresRecordFormatter.kt b/airbyte-integrations/connectors/destination-postgres/src/main/kotlin/io/airbyte/integrations/destination/postgres/write/load/PostgresRecordFormatter.kt index 41e48bd0891a..723135f46f83 100644 --- a/airbyte-integrations/connectors/destination-postgres/src/main/kotlin/io/airbyte/integrations/destination/postgres/write/load/PostgresRecordFormatter.kt +++ b/airbyte-integrations/connectors/destination-postgres/src/main/kotlin/io/airbyte/integrations/destination/postgres/write/load/PostgresRecordFormatter.kt @@ -11,6 +11,7 @@ import io.airbyte.cdk.load.message.Meta import io.airbyte.cdk.load.message.Meta.Companion.COLUMN_NAME_AB_LOADED_AT import io.airbyte.cdk.load.message.Meta.Companion.COLUMN_NAME_DATA import io.airbyte.cdk.load.util.Jsons +import io.airbyte.integrations.destination.postgres.write.transform.sanitizePostgresValue internal val RAW_META_COLUMNS = listOf( @@ -46,8 +47,9 @@ class PostgresRawRecordFormatter( // Do not output null values in the JSON raw output val filteredRecord = record.filter { (k, v) -> v !is NullValue && !RAW_META_COLUMNS.contains(k) } - // Sanitize null bytes from JSON data — PostgreSQL TEXT columns do not support \u0000 - val jsonData = Jsons.writeValueAsString(filteredRecord).replace("\u0000", "") + val sanitizedRecord = + filteredRecord.mapValues { (_, value) -> sanitizePostgresValue(value) } + val jsonData = Jsons.writeValueAsString(sanitizedRecord) // Iterate through columns in the exact order they appear in the table columns.forEach { column -> diff --git a/airbyte-integrations/connectors/destination-postgres/src/main/kotlin/io/airbyte/integrations/destination/postgres/write/transform/PostgresValueCoercer.kt b/airbyte-integrations/connectors/destination-postgres/src/main/kotlin/io/airbyte/integrations/destination/postgres/write/transform/PostgresValueCoercer.kt index bb11cc5638a6..87bce2f3fef6 100644 --- a/airbyte-integrations/connectors/destination-postgres/src/main/kotlin/io/airbyte/integrations/destination/postgres/write/transform/PostgresValueCoercer.kt +++ b/airbyte-integrations/connectors/destination-postgres/src/main/kotlin/io/airbyte/integrations/destination/postgres/write/transform/PostgresValueCoercer.kt @@ -51,6 +51,7 @@ internal const val TIMESTAMP_MAX_EPOCH_SECONDS = 9223371331200L @Singleton class PostgresValueCoercer : ValueCoercer { override fun map(value: EnrichedAirbyteValue): EnrichedAirbyteValue { + value.abValue = sanitizePostgresValue(value.abValue) value.abValue = if (value.type is UnionType || value.type is UnknownType) { // Don't serialize null values - keep them as NullValue @@ -84,19 +85,10 @@ class PostgresValueCoercer : ValueCoercer { } else ValidationResult.Valid } is StringValue -> { - // PostgreSQL doesn't allow null bytes (\u0000) in text fields - // Replace them with empty string to prevent COPY errors - // Using replace() without regex for optimal performance (O(n) vs O(n*m) with regex) - if (abValue.value.contains('\u0000')) { - val sanitizedValue = abValue.value.replace("\u0000", "") - value.abValue = StringValue(sanitizedValue) - } - // Validate string length (conservative check - actual byte size may vary with // encoding) // PostgreSQL uses UTF-8, so we check character count * 4 (max bytes per UTF-8 char) - val currentValue = (value.abValue as StringValue).value - if (currentValue.length * 4 > TEXT_LIMIT_BYTES) { + if (abValue.value.length * 4 > TEXT_LIMIT_BYTES) { ValidationResult.ShouldNullify( AirbyteRecordMessageMetaChange.Reason.DESTINATION_FIELD_SIZE_LIMITATION ) diff --git a/airbyte-integrations/connectors/destination-postgres/src/main/kotlin/io/airbyte/integrations/destination/postgres/write/transform/PostgresValueSanitizer.kt b/airbyte-integrations/connectors/destination-postgres/src/main/kotlin/io/airbyte/integrations/destination/postgres/write/transform/PostgresValueSanitizer.kt new file mode 100644 index 000000000000..43f18d60d6d1 --- /dev/null +++ b/airbyte-integrations/connectors/destination-postgres/src/main/kotlin/io/airbyte/integrations/destination/postgres/write/transform/PostgresValueSanitizer.kt @@ -0,0 +1,57 @@ +/* + * Copyright (c) 2026 Airbyte, Inc., all rights reserved. + */ + +package io.airbyte.integrations.destination.postgres.write.transform + +import io.airbyte.cdk.load.data.AirbyteValue +import io.airbyte.cdk.load.data.ArrayValue +import io.airbyte.cdk.load.data.ObjectValue +import io.airbyte.cdk.load.data.StringValue + +internal fun sanitizePostgresValue(value: AirbyteValue): AirbyteValue = + sanitizePostgresValueWithChange(value).value + +private data class SanitizedValue(val value: AirbyteValue, val changed: Boolean) + +private fun sanitizePostgresValueWithChange(value: AirbyteValue): SanitizedValue = + when (value) { + is StringValue -> { + val sanitizedValue = value.value.replace("\u0000", "") + if (sanitizedValue.length != value.value.length) { + SanitizedValue(StringValue(sanitizedValue), true) + } else { + SanitizedValue(value, false) + } + } + is ArrayValue -> { + var sanitizedValues: MutableList? = null + value.values.forEachIndexed { index, child -> + val sanitized = sanitizePostgresValueWithChange(child) + if (sanitized.changed) { + if (sanitizedValues == null) { + sanitizedValues = value.values.toMutableList() + } + sanitizedValues!![index] = sanitized.value + } + } + sanitizedValues?.let { SanitizedValue(ArrayValue(it), true) } + ?: SanitizedValue(value, false) + } + is ObjectValue -> { + var sanitizedValues: LinkedHashMap? = null + value.values.forEach { (key, child) -> + val sanitized = sanitizePostgresValueWithChange(child) + if (sanitized.changed) { + if (sanitizedValues == null) { + sanitizedValues = LinkedHashMap(value.values) + } + sanitizedValues!![key] = sanitized.value + } + } + // Nested object keys are intentionally not sanitized; only values are destination data. + sanitizedValues?.let { SanitizedValue(ObjectValue(it), true) } + ?: SanitizedValue(value, false) + } + else -> SanitizedValue(value, false) + } diff --git a/airbyte-integrations/connectors/destination-postgres/src/test/kotlin/io/airbyte/integrations/destination/postgres/write/load/PostgresRecordFormatterTest.kt b/airbyte-integrations/connectors/destination-postgres/src/test/kotlin/io/airbyte/integrations/destination/postgres/write/load/PostgresRecordFormatterTest.kt new file mode 100644 index 000000000000..150b7cb44a6f --- /dev/null +++ b/airbyte-integrations/connectors/destination-postgres/src/test/kotlin/io/airbyte/integrations/destination/postgres/write/load/PostgresRecordFormatterTest.kt @@ -0,0 +1,79 @@ +/* + * Copyright (c) 2026 Airbyte, Inc., all rights reserved. + */ + +package io.airbyte.integrations.destination.postgres.write.load + +import io.airbyte.cdk.load.data.ArrayValue +import io.airbyte.cdk.load.data.EnrichedAirbyteValue +import io.airbyte.cdk.load.data.ObjectTypeWithoutSchema +import io.airbyte.cdk.load.data.ObjectValue +import io.airbyte.cdk.load.data.StringValue +import io.airbyte.integrations.destination.postgres.write.transform.PostgresValueCoercer +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Test + +internal class PostgresRecordFormatterTest { + + @Test + fun `schema formatter removes nested null characters before serialization`() { + val value = + ObjectValue( + linkedMapOf( + "entries" to + ArrayValue( + listOf(ObjectValue(linkedMapOf("text" to StringValue("a\u0000b")))) + ) + ) + ) + val enrichedValue = + EnrichedAirbyteValue( + abValue = value, + type = ObjectTypeWithoutSchema, + name = "entries", + changes = mutableListOf(), + airbyteMetaField = null, + ) + val coercer = PostgresValueCoercer() + coercer.map(enrichedValue) + + val serialized = + PostgresSchemaRecordFormatter(listOf("entries")) + .format(mapOf("entries" to enrichedValue.abValue))[0] + .toString() + + assertFalse(serialized.contains('\u0000')) + assertFalse(serialized.contains("\\u0000")) + } + + @Test + fun `raw formatter removes nested null characters before serialization`() { + val value = + ObjectValue( + linkedMapOf( + "entries" to + ArrayValue( + listOf(ObjectValue(linkedMapOf("text" to StringValue("a\u0000b")))) + ) + ) + ) + val enrichedValue = + EnrichedAirbyteValue( + abValue = value, + type = ObjectTypeWithoutSchema, + name = "entries", + changes = mutableListOf(), + airbyteMetaField = null, + ) + val coercer = PostgresValueCoercer() + coercer.map(enrichedValue) + + val serialized = + PostgresRawRecordFormatter(listOf("_airbyte_data")) + .format(mapOf("entries" to enrichedValue.abValue))[0] + .toString() + + assertFalse(serialized.contains('\u0000')) + assertFalse(serialized.contains("\\u0000")) + } +} diff --git a/airbyte-integrations/connectors/destination-postgres/src/test/kotlin/io/airbyte/integrations/destination/postgres/write/transform/PostgresValueCoercerTest.kt b/airbyte-integrations/connectors/destination-postgres/src/test/kotlin/io/airbyte/integrations/destination/postgres/write/transform/PostgresValueCoercerTest.kt index 184a34122cdb..d458d677f847 100644 --- a/airbyte-integrations/connectors/destination-postgres/src/test/kotlin/io/airbyte/integrations/destination/postgres/write/transform/PostgresValueCoercerTest.kt +++ b/airbyte-integrations/connectors/destination-postgres/src/test/kotlin/io/airbyte/integrations/destination/postgres/write/transform/PostgresValueCoercerTest.kt @@ -15,6 +15,7 @@ import io.airbyte.cdk.load.data.NullValue import io.airbyte.cdk.load.data.NumberType import io.airbyte.cdk.load.data.NumberValue import io.airbyte.cdk.load.data.ObjectType +import io.airbyte.cdk.load.data.ObjectTypeWithoutSchema import io.airbyte.cdk.load.data.ObjectValue import io.airbyte.cdk.load.data.StringType import io.airbyte.cdk.load.data.StringValue @@ -77,6 +78,41 @@ internal class PostgresValueCoercerTest { assertEquals(StringType, result.abValue.airbyteType) } + @Test + fun testMapRemovesNestedNullCharacters() { + val objectValue = + ObjectValue( + linkedMapOf( + "objectValue" to StringValue("before\u0000after"), + "arrayValue" to + ArrayValue( + listOf(ObjectValue(linkedMapOf("nested" to StringValue("a\u0000b")))) + ) + ) + ) + val enrichedAirbyteValue = + EnrichedAirbyteValue( + abValue = objectValue, + type = ObjectTypeWithoutSchema, + name = "test", + changes = mutableListOf(), + airbyteMetaField = null, + ) + + coercer.map(enrichedAirbyteValue) + + assertEquals( + ObjectValue( + linkedMapOf( + "objectValue" to StringValue("beforeafter"), + "arrayValue" to + ArrayValue(listOf(ObjectValue(linkedMapOf("nested" to StringValue("ab"))))) + ) + ), + enrichedAirbyteValue.abValue + ) + } + @Test fun testValidateValidArray() { val arrayValue = ArrayValue(listOf(IntegerValue(1), IntegerValue(2), IntegerValue(3))) diff --git a/docs/integrations/destinations/postgres.md b/docs/integrations/destinations/postgres.md index 85255c1f9b58..8d8e2220f3c9 100644 --- a/docs/integrations/destinations/postgres.md +++ b/docs/integrations/destinations/postgres.md @@ -299,6 +299,7 @@ This destination supports [namespaces](https://docs.airbyte.com/platform/using-a | Version | Date | Pull Request | Subject | |:--------|:-----------|:-----------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| 3.0.17-rc.1 | 2026-08-12 | [84321](https://github.com/airbytehq/airbyte/pull/84321) | Prevent sync failures when NUL characters are nested in JSON values. | | 3.0.16 | 2026-03-31 | [75902](https://github.com/airbytehq/airbyte/pull/75902) | Fix silent error swallowing in COPY flush and sanitize null bytes in raw JSON data | | 3.0.15 | 2026-08-07 | [83235](https://github.com/airbytehq/airbyte/pull/83235) | Fail sync on transient DB errors. | | 3.0.14 | 2026-07-30 | [82273](https://github.com/airbytehq/airbyte/pull/82273) | Remove column DROP logic during schema evolution; upgrade CDK to 1.0.20 |