Tap 12765

connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/physical/HeapRmgrDecoder.java

@@ -37,11 +37,18 @@ public static final class Ctx {
         public final PageStateCache cache;          // null disables tracking
         public final boolean walLevelLogical;       // global bypass when true
         public final BiConsumer<String, Object[]> debug;   // (fmt, args) → log line; may be null
+        public final HeapTupleDecoder.ToastedValueFetcher toastFetcher;
 
         public Ctx(PageStateCache cache, boolean walLevelLogical, BiConsumer<String, Object[]> debug) {
+            this(cache, walLevelLogical, debug, null);
+        }
+
+        public Ctx(PageStateCache cache, boolean walLevelLogical, BiConsumer<String, Object[]> debug,
+                   HeapTupleDecoder.ToastedValueFetcher toastFetcher) {
             this.cache = cache;
             this.walLevelLogical = walLevelLogical;
             this.debug = debug;
+            this.toastFetcher = toastFetcher;
         }
 
         boolean trackingEnabled(RelationInfo rel) {
@@ -116,7 +123,7 @@ private static NormalRedo decodeInsert(XLogRecord rec, RelationInfo rel, Ctx ctx
             cp.put(offnum, tuple);
         }
         NormalRedo r = base(rec, rel, OperationEnum.INSERT);
-        r.setRedoRecord(decodeTuple(tuple, rel, ctx, "insert"));
+        r.setRedoRecord(HeapTupleDecoder.decode(tuple, rel.columns, ctx == null ? null : ctx.toastFetcher));
         return r;
     }
 
@@ -150,7 +157,7 @@ private static NormalRedo decodeDelete(XLogRecord rec, RelationInfo rel, Ctx ctx
         // overwrites the offset before any later record should read it.
         NormalRedo r = base(rec, rel, OperationEnum.DELETE);
         if (oldTuple != null) {
-            r.setUndoRecord(decodeTuple(oldTuple, rel, ctx, "delete-old"));
+            r.setUndoRecord(HeapTupleDecoder.decode(oldTuple, rel.columns, ctx == null ? null : ctx.toastFetcher));
         }
         return r;
     }
@@ -215,10 +222,10 @@ private static NormalRedo decodeUpdate(XLogRecord rec, RelationInfo rel, Ctx ctx
         }
         NormalRedo r = base(rec, rel, OperationEnum.UPDATE);
         if (oldTuple != null) {
-            r.setUndoRecord(decodeTuple(oldTuple, rel, ctx, "update-old"));
+            r.setUndoRecord(HeapTupleDecoder.decode(oldTuple, rel.columns, ctx == null ? null : ctx.toastFetcher));
         }
         if (newTuple != null) {
-            r.setRedoRecord(decodeTuple(newTuple, rel, ctx, "update-new"));
+            r.setRedoRecord(HeapTupleDecoder.decode(newTuple, rel.columns, ctx == null ? null : ctx.toastFetcher));
         }
         return r;
     }
@@ -261,7 +268,7 @@ private static List<NormalRedo> decodeMultiInsert(XLogRecord rec, RelationInfo r
                 continue;
             }
             NormalRedo nr = base(rec, rel, OperationEnum.INSERT);
-            nr.setRedoRecord(decodeTuple(tuples[i], rel, ctx, "multi-insert"));
+            nr.setRedoRecord(HeapTupleDecoder.decode(tuples[i], rel.columns, ctx == null ? null : ctx.toastFetcher));
             out.add(nr);
         }
         ctx.log("[WAL-DEBUG] MULTI_INSERT rel={} blk={} ntuples={} offnums={} dataLen={}",
@@ -277,19 +284,6 @@ private static List<NormalRedo> decodeMultiInsert(XLogRecord rec, RelationInfo r
         return out;
     }
 
-    private static Map<String, Object> decodeTuple(byte[] tuple, RelationInfo rel, Ctx ctx, String image) {
-        if (tuple != null && tuple.length >= SIZE_OF_HEAP_HEADER) {
-            int infomask2 = u16(tuple, 0);
-            int infomask = u16(tuple, 2);
-            int tHoff = tuple[4] & 0xFF;
-            int natts = infomask2 & HEAP_NATTS_MASK;
-            ctx.log("[WAL-DEBUG] TUPLE-DECODE image={} rel={} tupleLen={} natts={} infomask=0x{} tHoff={} relColumns={}",
-                    new Object[]{image, relTag(rel), tuple.length, natts, Integer.toHexString(infomask),
-                            tHoff, columnLayout(rel)});
-        }
-        return HeapTupleDecoder.decode(tuple, rel.columns);
-    }
-
     private static int u16(byte[] bytes, int offset) {
         if (bytes == null || bytes.length < offset + 2) {
             return 0;

connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/physical/HeapTupleDecoder.java

@@ -1,5 +1,6 @@
 package io.tapdata.connector.postgres.cdc.physical;
 
+import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -24,11 +25,21 @@ public final class HeapTupleDecoder {
     private HeapTupleDecoder() {
     }
 
+    @FunctionalInterface
+    public interface ToastedValueFetcher {
+        byte[] fetch(long toastRelId, long valueId);
+    }
+
     /* on-disk TOAST pointer: VARHDRSZ_EXTERNAL(2) + sizeof(varatt_external)(16). */
     private static final int VARTAG_ONDISK = 18;
     private static final int EXTERNAL_ONDISK_SIZE = 2 + 16;
+    private static final int VARHDRSZ = 4;
 
     public static Map<String, Object> decode(byte[] tuple, List<ColumnInfo> columns) {
+        return decode(tuple, columns, null);
+    }
+
+    public static Map<String, Object> decode(byte[] tuple, List<ColumnInfo> columns, ToastedValueFetcher toastFetcher) {
         Map<String, Object> out = new LinkedHashMap<>();
         if (tuple == null || tuple.length < SIZE_OF_HEAP_HEADER) {
             return out;
@@ -74,7 +85,7 @@ public static Map<String, Object> decode(byte[] tuple, List<ColumnInfo> columns)
             }
             Object value;
             try {
-                value = readAttr(r, col);
+                value = readAttr(r, col, toastFetcher);
             } catch (RuntimeException ex) {
                 // Malformed/garbage tuple bytes (typically a stale cache page or
                 // an unrecoverable delta-only update): abort the rest of the row
@@ -98,14 +109,14 @@ private static boolean bitSet(byte[] bitmap, int i) {
         return (bitmap[i >> 3] & (1 << (i & 7))) != 0;
     }
 
-    private static Object readAttr(WalByteReader r, ColumnInfo col) {
+    private static Object readAttr(WalByteReader r, ColumnInfo col, ToastedValueFetcher toastFetcher) {
         int attlen = col.typLen;
         if (attlen > 0) {
             r.align(col.typAlign);
             return PgTypeDecoder.decode(col.typeOid, r.readBytes(attlen), col.enumTypeOid, col.enumLabels);
         }
         if (attlen == -1) {
-            return readVarlena(r, col);
+            return readVarlena(r, col, toastFetcher);
         }
         if (attlen == -2) {
             // cstring: null-terminated, no alignment beyond byte
@@ -119,15 +130,49 @@ private static Object readAttr(WalByteReader r, ColumnInfo col) {
         throw new IllegalStateException("unsupported attlen " + attlen + " for column " + col.name);
     }
 
-    private static Object readVarlena(WalByteReader r, ColumnInfo col) {
+    private static Object readVarlena(WalByteReader r, ColumnInfo col, ToastedValueFetcher toastFetcher) {
         int first = r.peekUInt8();
         if ((first & 0x01) == 0x01) {
             if (first == 0x01) {
                 // 1B external TOAST pointer; value lives in the TOAST relation
                 int tag = r.peekUInt8(1);
-                int size = tag == VARTAG_ONDISK ? EXTERNAL_ONDISK_SIZE : 2;
-                r.skip(size);
-                return null;            // external value not present in this record
+                if (tag != VARTAG_ONDISK) {
+                    r.skip(2);
+                    return null;
+                }
+                r.skip(2);
+                int rawSize = r.readInt32();
+                long extInfo = r.readUInt32();
+                long valueId = r.readUInt32();
+                long toastRelId = r.readUInt32();
+                byte[] toasted;
+                try {
+                    toasted = toastFetcher == null ? null : toastFetcher.fetch(toastRelId, valueId);
+                } catch (RuntimeException ex) {
+                    toasted = null;
+                }
+                if (toasted == null) {
+                    return null;
+                }
+                int extSize = (int) (extInfo & 0x3FFFFFFFL);
+                boolean compressed = extSize < rawSize - VARHDRSZ;
+                if (compressed) {
+                    int method = (int) ((extInfo >> 30) & 0x03);
+                    if (!isPglzCompressionMethod(method)) {
+                        return null;
+                    }
+                    // A TOASTed compressed value stores VARDATA in the toast
+                    // relation, which begins with the 4-byte va_tcinfo word
+                    // (raw size) followed by the pglz stream. Skip the tcinfo
+                    // before inflating, mirroring the inline 4B-header branch.
+                    if (toasted.length <= 4) {
+                        return null;
+                    }
+                    byte[] pglzData = Arrays.copyOfRange(toasted, 4, toasted.length);
+                    byte[] plain = Pglz.decompress(pglzData, rawSize - VARHDRSZ);
+                    return plain == null ? null : PgTypeDecoder.decode(col.typeOid, plain);
+                }
+                return PgTypeDecoder.decode(col.typeOid, toasted);
             }
             int total = (first >> 1) & 0x7F;        // includes the 1-byte header
             r.skip(1);
@@ -148,10 +193,14 @@ private static Object readVarlena(WalByteReader r, ColumnInfo col) {
             int rawSize = (int) (tcinfo & 0x3FFFFFFF);
             int method = (int) ((tcinfo >> 30) & 0x03);
             byte[] comp = r.readBytes(total - 8);
-            byte[] plain = method == 0 ? Pglz.decompress(comp, rawSize) : null;
+            byte[] plain = isPglzCompressionMethod(method) ? Pglz.decompress(comp, rawSize) : null;
             return plain == null ? null : PgTypeDecoder.decode(col.typeOid, plain, col.enumTypeOid, col.enumLabels);
         }
         r.skip(4);
         return PgTypeDecoder.decode(col.typeOid, r.readBytes(total - 4), col.enumTypeOid, col.enumLabels);
     }
+
+    private static boolean isPglzCompressionMethod(int method) {
+        return method == 0 || method == 1;
+    }
 }

connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/physical/PhysicalWalLogMiner.java

@@ -214,10 +214,13 @@ public class PhysicalWalLogMiner extends AbstractWalLogMiner {
     /*
      * Side-effect-free context used only by pool threads for wal_level=logical.
      * cache=null and walLevelLogical=true guarantee HeapRmgrDecoder will not
-     * read or mutate PageStateCache while pre-decoding DML.
+     * read or mutate PageStateCache while pre-decoding DML. Not static because
+     * it carries the instance toastFetcher: logical-level WAL stores large
+     * values (jsonb etc.) as TOAST pointers, and fast pre-decode must resolve
+     * them through pg_toast just like the consumer-side decodeCtx does.
+     * Initialized in resetCachesForRecovery alongside decodeCtx.
      */
-    private static final HeapRmgrDecoder.Ctx LOGICAL_FAST_DECODE_CTX =
-            new HeapRmgrDecoder.Ctx(null, true, null);
+    private HeapRmgrDecoder.Ctx logicalFastDecodeCtx;
     /* Running diagnostics for before-image coverage under wal_level=replica:
      * how many emitted UPDATE/DELETE carried a null before-image (cache miss),
      * throttled to one warning per interval. Consumer-thread only. */
@@ -385,7 +388,11 @@ public void startMiner(Supplier<Boolean> isAlive) throws Throwable {
             throw new UncheckedIOException("Cannot create spill directory " + spillDir, e);
         }
         decodeCtx = new HeapRmgrDecoder.Ctx(pageCache, walLevelLogical,
-                isWalDebugEnabled() ? tapLogger::info : null);
+                isWalDebugEnabled() ? tapLogger::info : null,
+                this::fetchToastValue);
+        logicalFastDecodeCtx = new HeapRmgrDecoder.Ctx(null, true,
+                isWalDebugEnabled() ? tapLogger::info : null,
+                this::fetchToastValue);
         buildDdlWatch();
         // Seed strategy (standby-friendly). PageStateCache is seeded by each
         // page's first post-checkpoint FPI, which is what unlocks UPDATE/DELETE
@@ -1631,7 +1638,7 @@ private long parseOffsetLsn(String offset) {
      *
      * Fast path: when wal_level=logical, DML WAL carries enough tuple bytes to
      * decode without PageStateCache. The worker may pre-decode those records
-     * with LOGICAL_FAST_DECODE_CTX, which has no cache and no debug callback.
+     * with logicalFastDecodeCtx, which has no cache and no page-state tracking.
      *
      * Slow path: replica-level WAL and any DDL-sensitive record are decoded on
      * the consumer thread in WAL order. Workers must not mutate page cache,
@@ -1750,7 +1757,7 @@ private RelationInfo resolveRelForConsumer(long relNumber) {
     private List<NormalRedo> decodeHeapLogicalFast(XLogRecord rec, RelationInfo rel) {
         List<NormalRedo> redos;
         try {
-            redos = HeapRmgrDecoder.decode(rec, rel, LOGICAL_FAST_DECODE_CTX);
+            redos = HeapRmgrDecoder.decode(rec, rel, logicalFastDecodeCtx);
         } catch (RuntimeException ex) {
             tapLogger.warn("skip logical fast heap record at lsn={} rel={}.{} due to decode error: {}",
                     lsnStr(rec.lsn), rel.schema, rel.table, ex.getMessage());
@@ -3220,15 +3227,10 @@ private void buildDdlWatch() {
         // Give the catalog its own page overlay so FPI-less pg_attribute UPDATEs
         // (DROP/RENAME/ALTER COLUMN) reconstruct their before/after tuple images
         // from an earlier FPI on the same page, mirroring the user-table cache.
-        //
-        // This must stay enabled even when the user-table WAL level is logical:
-        // pg_attribute UPDATE records may still omit the old tuple bytes. DDL
-        // recognition relies on the catalog overlay to recover the before-image
-        // and mark the xid as DDL before later DML in the same transaction is
-        // decoded/emitted.
         catalogPageCache = new PageStateCache(getPageCacheCapacity());
         catalogDecodeCtx = new HeapRmgrDecoder.Ctx(catalogPageCache, false,
-                isWalDebugEnabled() ? tapLogger::info : null);
+                isWalDebugEnabled() ? tapLogger::info : null,
+                this::fetchToastValue);
         // Baseline column layout for every monitored table; later pg_attribute
         // changes are diffed against this to derive the concrete field DDL.
         try {
@@ -4255,6 +4257,57 @@ private AncestorCatchupStalledException(String message) {
         }
     }
 
+    private byte[] fetchToastValue(long toastRelId, long valueId) {
+        if (toastRelId <= 0 || valueId <= 0) {
+            return null;
+        }
+        // The TOAST pointer stores the TOAST table's own OID (pg_class.oid of
+        // pg_toast_<n>), NOT the owning table's OID. The toast relation's name
+        // is pg_toast_<owningOid> and can differ from pg_toast_<toastRelId> once
+        // OIDs and relfilenodes diverge (e.g. after a table rebuild), so resolve
+        // the real relation name through pg_class instead of string-concatenating
+        // the pointer value into a table name.
+        String[] toastTable = {null};
+        ErrorKit.ignoreAnyError(() -> postgresJdbcContext.query(
+                "SELECT relname FROM pg_class WHERE oid = " + toastRelId,
+                rs -> {
+                    if (rs.next()) {
+                        toastTable[0] = rs.getString(1);
+                    }
+                }));
+        if (toastTable[0] == null) {
+            tapLogger.warn("TAP-12765 fetchToastValue: no pg_class row for oid {} (valueId={}); returning null",
+                    toastRelId, valueId);
+            return null;
+        }
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        try {
+            postgresJdbcContext.query(
+                    "SELECT chunk_data FROM pg_toast." + toastTable[0]
+                            + " WHERE chunk_id = " + valueId + " ORDER BY chunk_seq",
+                    rs -> {
+                        while (rs.next()) {
+                            byte[] chunk = rs.getBytes(1);
+                            if (chunk != null && chunk.length > 0) {
+                                out.write(chunk, 0, chunk.length);
+                            }
+                        }
+                    });
+        } catch (Throwable e) {
+            tapLogger.warn("TAP-12765 fetchToastValue: query pg_toast.{} chunk_id={} failed: {}",
+                    toastTable[0], valueId, e.getMessage());
+            return null;
+        }
+        if (out.size() == 0) {
+            tapLogger.warn("TAP-12765 fetchToastValue: pg_toast.{} chunk_id={} returned 0 bytes",
+                    toastTable[0], valueId);
+            return null;
+        }
+        tapLogger.info("TAP-12765 fetchToastValue: pg_toast.{} chunk_id={} fetched {} bytes",
+                toastTable[0], valueId, out.size());
+        return out.toByteArray();
+    }
+
     /* Best-effort LSN string -> long; 0 on blank/parse failure so callers can
      * treat it as "unknown" without throwing during startup. */
     private static long lsnAsLong(String lsn) {