Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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", "")
Comment on lines -49 to -50

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why didn't this work before?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🚫 No change needed — answering. It ran too late. By the time Jsons.writeValueAsString(...) returns, Jackson has already encoded the NUL as the six-character escape sequence \u0000 (backslash, u, 0000), so there is no literal NUL code point left in the string for .replace("\u0000", "") to match. The replace silently matched nothing on every record.

Quick demonstration with the Jackson version we build against:

input:  {"a": "x<NUL>y"}          // actual U+0000 in the value
output: {"a":"x\u0000y"}          // six literal characters, zero NUL bytes

And \u0000 in that form is precisely what Postgres' jsonb input function rejects (unsupported Unicode escape sequence / <NULL> cannot be converted to text), which is the failure in the linked issue. Sanitizing the AirbyteValue tree before serialization is what makes it effective, and the new PostgresRecordFormatterTest asserts the absence of both the NUL byte and the textual escape so this can't silently regress to a no-op again.


Devin session

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

shouldn't there be a way to use replace with the string value instead of the byte value?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🙋 There is, but I'd rather not — your call. You're right that .replace("\\u0000", "") on the serialized text would actually match, unlike the old code. Two reasons I went the other way:

  1. It can corrupt valid JSON. If a source string legitimately contains the six characters \u0000 as text (backslash included), Jackson escapes the backslash and emits seven characters: \\u0000. A textual replace of \u0000 matches the tail of that and leaves a dangling \, producing invalid JSON. Operating on the AirbyteValue tree has no such ambiguity — we edit the actual string content, and Jackson escapes whatever remains correctly.

  2. It only covers the raw path. The typed path doesn't serialize the whole record; each jsonb column is serialized independently (PostgresValueCoercer.maptoCsvValue) and the result is written into a CSV row. A post-serialization fixup would have to be repeated there, on strings already mixed with CSV quoting. Sanitizing before serialization is one place that covers both paths and keeps the typed and raw tables consistent.

If you'd still prefer the textual replace on the raw path, I'll switch it — but I'd want to keep the pre-serialization sanitizer for the typed path regardless, so it would be two mechanisms instead of one.


Devin session

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 ->
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
)
Expand Down
Original file line number Diff line number Diff line change
@@ -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<AirbyteValue>? = 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<String, AirbyteValue>? = 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)
}
Original file line number Diff line number Diff line change
@@ -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"))
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)))
Expand Down
1 change: 1 addition & 0 deletions docs/integrations/destinations/postgres.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand Down
Loading