Skip to content

fix: Handle wal2json enum type quoting issue in LOG_BASED replication #736

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
edgarrmondragon merged 6 commits into from
Mar 3, 2026
Merged

fix: Handle wal2json enum type quoting issue in LOG_BASED replication #736

Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
c504f73
fix: Initialize connector before accessing connection_parameters in d…
kashif-se Feb 25, 2026
ab9a4d8
Merge pull request #1 from ksohail22/fix/initialize-connector-before-…
ksohail22 Feb 25, 2026
a3c3647
Merge branch 'MeltanoLabs:main' into main
ksohail22 Feb 26, 2026
81bc0b2
fix: Handle wal2json enum type quoting issue in LOG_BASED replication
kashif-se Feb 26, 2026
c74735d
perf: Compile wal2json enum quote regex as a class-level constant
kashif-se Mar 3, 2026
ef19de9
Merge branch 'main' into fix/handle-wal2json-enum-type-quotingissue-l…
edgarrmondragon Mar 3, 2026
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
30 changes: 25 additions & 5 deletions tap_postgres/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import datetime
import functools
import json
import re
import select
import typing as t
from types import MappingProxyType
Expand Down Expand Up @@ -213,6 +214,8 @@ class PostgresLogBasedStream(SQLStream):

replication_key = "_sdc_lsn"

_WAL2JSON_ENUM_QUOTE_RE = re.compile(r'"type":""([^"]+)""')

connection_parameters: ConnectionParameters

def __init__(
Expand Down Expand Up @@ -349,11 +352,18 @@ def consume(self, message, cursor) -> dict | None:
try:
message_payload = json.loads(message.payload)
except json.JSONDecodeError:
self.logger.warning(
"A message payload of %s could not be converted to JSON",
message.payload,
)
return {}
# wal2json outputs PostgreSQL enum types with unescaped quotes
# e.g., "type":""EnumName"" instead of "type":"EnumName"
# Try to fix this by removing the extra quotes around type values
fixed_payload = self._fix_wal2json_enum_quotes(message.payload)
try:
message_payload = json.loads(fixed_payload)
except json.JSONDecodeError:
self.logger.warning(
"A message payload of %s could not be converted to JSON",
message.payload,
)
Comment on lines +359 to +365
Copy link
Member

@edgarrmondragon edgarrmondragon Feb 26, 2026

Choose a reason for hiding this comment

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

perf: this adds another try/except block to each iteration of the log message loop. Even if the performance hit is measurable, having the sync crash is worse.

I %100 need to add some sort of automatic benchmarks for general and log-based performance here.

return {}

row = {}

Expand Down Expand Up @@ -403,6 +413,16 @@ def consume(self, message, cursor) -> dict | None:

return row

def _fix_wal2json_enum_quotes(self, payload: str) -> str:
"""Fix malformed JSON from wal2json for PostgreSQL enum types.

wal2json outputs enum type names with unescaped quotes, e.g.:
"type":""EnumName""
This is invalid JSON. We fix it by removing the extra quotes:
"type":"EnumName"
"""
return self._WAL2JSON_ENUM_QUOTE_RE.sub(r'"type":"\1"', payload)

def _parse_column_value(self, column, cursor):
# When using log based replication, the wal2json output for columns of
# array types returns a string encoded in sql format, e.g. '{a,b}'
Expand Down