diff --git a/connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/physical/HeapRmgrDecoder.java b/connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/physical/HeapRmgrDecoder.java index 79e0886d2..3fb9652ff 100644 --- a/connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/physical/HeapRmgrDecoder.java +++ b/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 debug; // (fmt, args) → log line; may be null + public final HeapTupleDecoder.ToastedValueFetcher toastFetcher; public Ctx(PageStateCache cache, boolean walLevelLogical, BiConsumer debug) { + this(cache, walLevelLogical, debug, null); + } + + public Ctx(PageStateCache cache, boolean walLevelLogical, BiConsumer 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 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 decodeMultiInsert(XLogRecord rec, RelationInfo r return out; } - private static Map 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; diff --git a/connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/physical/HeapTupleDecoder.java b/connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/physical/HeapTupleDecoder.java index a5528454a..030bb7b4f 100644 --- a/connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/physical/HeapTupleDecoder.java +++ b/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 decode(byte[] tuple, List columns) { + return decode(tuple, columns, null); + } + + public static Map decode(byte[] tuple, List columns, ToastedValueFetcher toastFetcher) { Map out = new LinkedHashMap<>(); if (tuple == null || tuple.length < SIZE_OF_HEAP_HEADER) { return out; @@ -74,7 +85,7 @@ public static Map decode(byte[] tuple, List 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; + } } diff --git a/connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/physical/PhysicalWalLogMiner.java b/connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/physical/PhysicalWalLogMiner.java index a2c013f54..589e81cf6 100644 --- a/connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/physical/PhysicalWalLogMiner.java +++ b/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 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 decodeHeapLogicalFast(XLogRecord rec, RelationInfo rel) { List 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_), NOT the owning table's OID. The toast relation's name + // is pg_toast_ and can differ from pg_toast_ 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) { diff --git a/connectors-common/postgres-core/src/test/java/io/tapdata/connector/postgres/cdc/physical/HeapTupleDecoderTest.java b/connectors-common/postgres-core/src/test/java/io/tapdata/connector/postgres/cdc/physical/HeapTupleDecoderTest.java index 5ced81039..470171e7f 100644 --- a/connectors-common/postgres-core/src/test/java/io/tapdata/connector/postgres/cdc/physical/HeapTupleDecoderTest.java +++ b/connectors-common/postgres-core/src/test/java/io/tapdata/connector/postgres/cdc/physical/HeapTupleDecoderTest.java @@ -45,19 +45,17 @@ public void testDeformFixedAndVarlena() { new ColumnInfo("b", 2, PgTypeDecoder.INT8, 8, 'd', false), new ColumnInfo("c", 3, PgTypeDecoder.TEXT, -1, 'i', false)); - int tHoff = maxAlign(SIZE_OF_HEAP_TUPLE_HEADER); // 24, no null bitmap + int tHoff = maxAlign(SIZE_OF_HEAP_TUPLE_HEADER); ByteArrayOutputStream o = new ByteArrayOutputStream(); - u16(o, 3); // t_infomask2 = natts - u16(o, 0); // t_infomask = no nulls - o.write(tHoff); // t_hoff - // padding between offset 23 and t_hoff (24) -> 1 byte + u16(o, 3); + u16(o, 0); + o.write(tHoff); for (int i = 0; i < tHoff - SIZE_OF_HEAP_TUPLE_HEADER; i++) { o.write(0); } - u32(o, 42); // a int4 - u32(o, 0); // align padding to 8 for int8 - u64(o, 123456789L); // b int8 - // c text "hi" short varlena: header=(payload+1)<<1|1 + u32(o, 42); + u32(o, 0); + u64(o, 123456789L); o.write(((2 + 1) << 1) | 1); o.write('h'); o.write('i'); @@ -76,19 +74,18 @@ public void testNullBitmapAndDropped() { new ColumnInfo("c", 3, PgTypeDecoder.INT4, 4, 'i', false)); int natts = 3; - int bitmapLen = (natts + 7) / 8; // 1 - int tHoff = maxAlign(SIZE_OF_HEAP_TUPLE_HEADER + bitmapLen); // 24 - // a present, dead null, c present -> bits 1,0,1 -> 0b101 = 0x05 + int bitmapLen = (natts + 7) / 8; + int tHoff = maxAlign(SIZE_OF_HEAP_TUPLE_HEADER + bitmapLen); ByteArrayOutputStream o = new ByteArrayOutputStream(); u16(o, natts); u16(o, HEAP_HASNULL); o.write(tHoff); - o.write(0x05); // null bitmap + o.write(0x05); for (int i = 0; i < tHoff - SIZE_OF_HEAP_TUPLE_HEADER - bitmapLen; i++) { - o.write(0); // padding to t_hoff + o.write(0); } - u32(o, 7); // a - u32(o, 9); // c (dead is null, no bytes) + u32(o, 7); + u32(o, 9); Map m = HeapTupleDecoder.decode(o.toByteArray(), cols); assertEquals(7, m.get("a")); @@ -110,7 +107,7 @@ public void testNullBitmapUsesPhysicalAttnumWhenCatalogHasGap() { u16(o, natts); u16(o, HEAP_HASNULL); o.write(tHoff); - o.write(0x09); // att1 present, att2/att3 null, att4 present + o.write(0x09); for (int i = 0; i < tHoff - SIZE_OF_HEAP_TUPLE_HEADER - bitmapLen; i++) { o.write(0); } @@ -148,4 +145,126 @@ public void testPg15JsonbShortVarlenaFromWalTuple() { assertEquals(12, nestedArrayRow.get("a1")); assertEquals("{\"dfd\":\"sf\",\"kdkdk\":[123123,123,123,\"123\"]}", nestedArrayRow.get("a2")); } + + @Test + public void testExternalToastPointerFetch() { + List cols = Arrays.asList( + new ColumnInfo("payload", 1, PgTypeDecoder.TEXT, -1, 'i', false)); + + int tHoff = maxAlign(SIZE_OF_HEAP_TUPLE_HEADER); + ByteArrayOutputStream o = new ByteArrayOutputStream(); + u16(o, 1); + u16(o, 0); + o.write(tHoff); + for (int i = 0; i < tHoff - SIZE_OF_HEAP_TUPLE_HEADER; i++) { + o.write(0); + } + o.write(0x01); + o.write(18); + u32(o, 11); + u32(o, 7); + u32(o, 77); + u32(o, 991); + + Map m = HeapTupleDecoder.decode(o.toByteArray(), cols, (toastRelId, valueId) -> { + assertEquals(991L, toastRelId); + assertEquals(77L, valueId); + return "payload".getBytes(java.nio.charset.StandardCharsets.UTF_8); + }); + + assertEquals("payload", m.get("payload")); + } + + @Test + public void testExternalToastPointerFetchCompressed() { + assertExternalToastPointerFetchCompressed(0); + } + + @Test + public void testExternalToastPointerFetchCompressedPg14Pglz() { + assertExternalToastPointerFetchCompressed(1); + } + + private static void assertExternalToastPointerFetchCompressed(int method) { + List cols = Arrays.asList( + new ColumnInfo("payload", 1, PgTypeDecoder.TEXT, -1, 'i', false)); + + int tHoff = maxAlign(SIZE_OF_HEAP_TUPLE_HEADER); + ByteArrayOutputStream o = new ByteArrayOutputStream(); + u16(o, 1); + u16(o, 0); + o.write(tHoff); + for (int i = 0; i < tHoff - SIZE_OF_HEAP_TUPLE_HEADER; i++) { + o.write(0); + } + o.write(0x01); + o.write(18); + u32(o, 10); + u32(o, 5 | ((long) method << 30)); + u32(o, 77); + u32(o, 991); + + byte[] toasted = new byte[] { + 0x06, 0x00, 0x00, 0x00, // va_tcinfo: raw size 6 ("ababab") + 0x04, + 'a', 'b', + 0x01, 0x02 + }; + Map m = HeapTupleDecoder.decode(o.toByteArray(), cols, (toastRelId, valueId) -> toasted); + + assertEquals("ababab", m.get("payload")); + } + + @Test + public void testExternalToastPointerFetchUnsupportedCompressionReturnsNull() { + List cols = Arrays.asList( + new ColumnInfo("payload", 1, PgTypeDecoder.TEXT, -1, 'i', false)); + + int tHoff = maxAlign(SIZE_OF_HEAP_TUPLE_HEADER); + ByteArrayOutputStream o = new ByteArrayOutputStream(); + u16(o, 1); + u16(o, 0); + o.write(tHoff); + for (int i = 0; i < tHoff - SIZE_OF_HEAP_TUPLE_HEADER; i++) { + o.write(0); + } + o.write(0x01); + o.write(18); + u32(o, 10); + u32(o, 5 | (2L << 30)); + u32(o, 77); + u32(o, 991); + + Map m = HeapTupleDecoder.decode(o.toByteArray(), cols, (toastRelId, valueId) -> new byte[] { + 0x04, 'a', 'b', 0x01, 0x02 + }); + + assertNull(m.get("payload")); + } + + @Test + public void testInlineCompressedVarlenaPg14Pglz() { + List cols = Arrays.asList( + new ColumnInfo("payload", 1, PgTypeDecoder.TEXT, -1, 'i', false)); + + byte[] compressed = new byte[] { + 0x04, 'a', 'b', 0x01, 0x02 + }; + int tHoff = maxAlign(SIZE_OF_HEAP_TUPLE_HEADER); + int total = 8 + compressed.length; + ByteArrayOutputStream o = new ByteArrayOutputStream(); + u16(o, 1); + u16(o, 0); + o.write(tHoff); + for (int i = 0; i < tHoff - SIZE_OF_HEAP_TUPLE_HEADER; i++) { + o.write(0); + } + u32(o, ((long) total << 2) | 0x02); + u32(o, 6 | (1L << 30)); + o.write(compressed, 0, compressed.length); + + Map m = HeapTupleDecoder.decode(o.toByteArray(), cols); + + assertEquals("ababab", m.get("payload")); + } }