TAP-12456: 【Wynn】pg 自定义字段类型适配

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

@@ -1,5 +1,8 @@
 package io.tapdata.connector.postgres.cdc.physical;
 
+import java.util.Collections;
+import java.util.Map;
+
 /**
  * Physical-storage metadata for a single column, mirroring the fields of
  * pg_attribute that drive heap-tuple deformation: attribute number, on-disk
@@ -18,14 +21,25 @@ public class ColumnInfo {
     /** attalign as a byte boundary: 1 ('c'), 2 ('s'), 4 ('i'), 8 ('d'). */
     public final int typAlign;
     public final boolean dropped;
+    /** For enum columns this is the type OID; for enum[] columns this is the element type OID. */
+    public final long enumTypeOid;
+    /** Maps pg_enum.oid values stored on disk to enum labels. */
+    public final Map<Long, String> enumLabels;
 
     public ColumnInfo(String name, int attnum, long typeOid, int typLen, char typAlign, boolean dropped) {
+        this(name, attnum, typeOid, typLen, typAlign, dropped, 0L, null);
+    }
+
+    public ColumnInfo(String name, int attnum, long typeOid, int typLen, char typAlign, boolean dropped,
+                      long enumTypeOid, Map<Long, String> enumLabels) {
         this.name = name;
         this.attnum = attnum;
         this.typeOid = typeOid;
         this.typLen = typLen;
         this.typAlign = alignOf(typAlign);
         this.dropped = dropped;
+        this.enumTypeOid = enumTypeOid;
+        this.enumLabels = enumLabels == null ? Collections.emptyMap() : Collections.unmodifiableMap(enumLabels);
     }
 
     private static int alignOf(char a) {

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

@@ -101,7 +101,7 @@ private static Object readAttr(WalByteReader r, ColumnInfo col) {
         int attlen = col.typLen;
         if (attlen > 0) {
             r.align(col.typAlign);
-            return PgTypeDecoder.decode(col.typeOid, r.readBytes(attlen));
+            return PgTypeDecoder.decode(col.typeOid, r.readBytes(attlen), col.enumTypeOid, col.enumLabels);
         }
         if (attlen == -1) {
             return readVarlena(r, col);
@@ -130,7 +130,7 @@ private static Object readVarlena(WalByteReader r, ColumnInfo col) {
             }
             int total = (first >> 1) & 0x7F;        // includes the 1-byte header
             r.skip(1);
-            return PgTypeDecoder.decode(col.typeOid, r.readBytes(total - 1));
+            return PgTypeDecoder.decode(col.typeOid, r.readBytes(total - 1), col.enumTypeOid, col.enumLabels);
         }
         // 4-byte header: align first
         r.align(col.typAlign);
@@ -148,9 +148,9 @@ private static Object readVarlena(WalByteReader r, ColumnInfo col) {
             int method = (int) ((tcinfo >> 30) & 0x03);
             byte[] comp = r.readBytes(total - 8);
             byte[] plain = method == 0 ? Pglz.decompress(comp, rawSize) : null;
-            return plain == null ? null : PgTypeDecoder.decode(col.typeOid, plain);
+            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));
+        return PgTypeDecoder.decode(col.typeOid, r.readBytes(total - 4), col.enumTypeOid, col.enumLabels);
     }
 }

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

@@ -63,9 +63,18 @@ private PgTypeDecoder() {
     private static final LocalDateTime PG_EPOCH_TS = LocalDateTime.of(2000, 1, 1, 0, 0);
 
     public static Object decode(long oid, byte[] v) {
+        return decode(oid, v, 0L, null);
+    }
+
+    public static Object decode(long oid, byte[] v, long enumTypeOid, Map<Long, String> enumLabels) {
         if (v == null) {
             return null;
         }
+        if (oid == enumTypeOid && enumLabels != null && !enumLabels.isEmpty() && v.length == 4) {
+            long enumValueOid = le32(v) & 0xFFFFFFFFL;
+            String label = enumLabels.get(enumValueOid);
+            return label != null ? label : enumValueOid;
+        }
         if (oid == BOOL) {
             return v.length > 0 && v[0] != 0;
         } else if (oid == INT2) {
@@ -141,7 +150,7 @@ public static Object decode(long oid, byte[] v) {
             return new String(v, StandardCharsets.UTF_8);
         }
         // Try array type (int[], varchar[], text[], timestamp[], etc.)
-        Object arrResult = decodeArray(v);
+        Object arrResult = decodeArray(v, enumTypeOid, enumLabels);
         if (arrResult != null) {
             return arrResult;
         }
@@ -160,7 +169,7 @@ public static Object decode(long oid, byte[] v) {
      *     — fixed-length (typlen &gt; 0): raw bytes, typlen per element
      *     — variable-length (typlen &lt; 0): full varlena datum per element
      */
-    private static Object decodeArray(byte[] v) {
+    private static Object decodeArray(byte[] v, long enumTypeOid, Map<Long, String> enumLabels) {
         if (v == null || v.length < 12) return null;
         int ndim = (int) le32(v, 0);
         if (ndim < 1 || ndim > 6) return null;
@@ -172,12 +181,12 @@ private static Object decodeArray(byte[] v) {
         //  but ndim=1 and dataoffset=0 happen to mask the shift for small-OID types)
         int elemOff = 8;
         long elemOid = le32(v, elemOff);
-        if (elemOid < 1 || elemOid > 10000) {
+        if (!isSupportedArrayElement(elemOid, enumTypeOid)) {
             elemOff = 9;
             if (v.length > elemOff + 3) {
                 elemOid = le32(v, elemOff);
             }
-            if (elemOid < 1 || elemOid > 10000) {
+            if (!isSupportedArrayElement(elemOid, enumTypeOid)) {
                 return null;
             }
         }
@@ -206,6 +215,9 @@ private static Object decodeArray(byte[] v) {
 
         // Determine element format
         Integer typlen = TYPE_LEN.get(elemOid);
+        if (typlen == null && elemOid == enumTypeOid && enumLabels != null && !enumLabels.isEmpty()) {
+            typlen = 4;
+        }
         boolean isFixed = typlen != null && typlen > 0;
 
         List<Object> values = new ArrayList<>(total);
@@ -229,12 +241,16 @@ private static Object decodeArray(byte[] v) {
                 elemBytes = datum.value;
             }
 
-            values.add(decode(elemOid, elemBytes));
+            values.add(decode(elemOid, elemBytes, enumTypeOid, enumLabels));
         }
 
         return ndim == 1 ? values : nestArray(values, dims, 0, 0).value;
     }
 
+    private static boolean isSupportedArrayElement(long elemOid, long enumTypeOid) {
+        return elemOid >= 1 && (elemOid <= 10000 || TYPE_LEN.containsKey(elemOid) || elemOid == enumTypeOid);
+    }
+
     private static VarlenaDatum readArrayVarlena(byte[] data, int offset) {
         if (offset >= data.length) return null;
         int first = data[offset] & 0xFF;

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

@@ -1734,7 +1734,7 @@ private RelationInfo resolveRelForConsumer(long relNumber) {
                 tapLogger.info("[WAL-DEBUG] applying {} pending pg_attribute change(s) to {}.{} before DML decode; beforeColumns={}",
                         pending.size(), rel.schema, rel.table, columnLayout(rel));
             }
-            rel = RelationCatalog.applyPendingChanges(rel, pending);
+            rel = catalog.applyPendingChangesWithTypeInfo(rel, pending);
             catalog.cache(relNumber, rel);
             if (isWalDebugEnabled()) {
                 tapLogger.info("[WAL-DEBUG] applied pending pg_attribute changes to {}.{}; afterColumns={}",

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

@@ -4,6 +4,7 @@
 import io.tapdata.entity.logger.Log;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -23,14 +24,19 @@ public class RelationCatalog {
     private final Log log;
     private final Map<Long, RelationInfo> cache = new ConcurrentHashMap<>();
     private final Map<Long, Boolean> negative = new ConcurrentHashMap<>();
+    private final Map<Long, EnumTypeInfo> enumTypeCache = new ConcurrentHashMap<>();
 
     private static final String REL_BY_FILENODE =
             "SELECT c.oid, n.nspname, c.relname, c.relreplident FROM pg_class c " +
                     "JOIN pg_namespace n ON c.relnamespace = n.oid " +
                     "WHERE c.relfilenode = %d AND c.relkind IN ('r','m','p') LIMIT 1";
     private static final String COLUMNS =
-            "SELECT attname, attnum, atttypid, attlen, attalign, attisdropped FROM pg_attribute " +
-                    "WHERE attrelid = %d AND attnum > 0 ORDER BY attnum";
+            "SELECT a.attname, a.attnum, a.atttypid, a.attlen, a.attalign, a.attisdropped, " +
+                    "CASE WHEN t.typtype = 'e' THEN t.oid WHEN et.typtype = 'e' THEN et.oid ELSE 0 END AS enumtypid " +
+                    "FROM pg_attribute a " +
+                    "JOIN pg_type t ON a.atttypid = t.oid " +
+                    "LEFT JOIN pg_type et ON t.typelem = et.oid " +
+                    "WHERE a.attrelid = %d AND a.attnum > 0 ORDER BY a.attnum";
     private static final String KEYS =
             "SELECT a.attname FROM pg_index i " +
                     "JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) " +
@@ -64,6 +70,7 @@ public RelationInfo lookup(long relNumber) {
     public void invalidate() {
         cache.clear();
         negative.clear();
+        enumTypeCache.clear();
     }
 
     public void cache(long relNumber, RelationInfo rel) {
@@ -96,7 +103,7 @@ public boolean applyPgAttributeChange(String schema, String table, String op,
             List<ColumnInfo> newColumns = new ArrayList<>(rel.columns);
             switch (op) {
                 case "INSERT": {
-                    ColumnInfo col = new ColumnInfo(attname, attnum, atttypid, attlen, attalign, attisdropped);
+                    ColumnInfo col = columnInfo(attname, attnum, atttypid, attlen, attalign, attisdropped);
                     int pos = 0;
                     while (pos < newColumns.size() && newColumns.get(pos).attnum < attnum) {
                         pos++;
@@ -111,7 +118,7 @@ public boolean applyPgAttributeChange(String schema, String table, String op,
                 case "UPDATE": {
                     for (int i = 0; i < newColumns.size(); i++) {
                         if (newColumns.get(i).attnum == attnum) {
-                            newColumns.set(i, new ColumnInfo(attname, attnum, atttypid, attlen, attalign, attisdropped));
+                            newColumns.set(i, columnInfo(attname, attnum, atttypid, attlen, attalign, attisdropped));
                             break;
                         }
                     }
@@ -181,6 +188,47 @@ public static RelationInfo applyPendingChanges(RelationInfo rel, java.util.List<
                 new ArrayList<>(rel.keyColumns), rel.replicaIdentityFull);
     }
 
+    public RelationInfo applyPendingChangesWithTypeInfo(RelationInfo rel, java.util.List<PgAttributeChange> pending) {
+        if (pending == null || pending.isEmpty()) {
+            return rel;
+        }
+        List<ColumnInfo> columns = new ArrayList<>(rel.columns);
+        for (PgAttributeChange c : pending) {
+            if (c.attnum <= 0) {
+                continue;
+            }
+            switch (c.op) {
+                case "INSERT": {
+                    ColumnInfo col = columnInfo(c.attname, c.attnum, c.atttypid, c.attlen, c.attalign, c.attisdropped);
+                    int pos = 0;
+                    while (pos < columns.size() && columns.get(pos).attnum < c.attnum) {
+                        pos++;
+                    }
+                    if (pos < columns.size() && columns.get(pos).attnum == c.attnum) {
+                        columns.set(pos, col);
+                    } else {
+                        columns.add(pos, col);
+                    }
+                    break;
+                }
+                case "UPDATE": {
+                    for (int i = 0; i < columns.size(); i++) {
+                        if (columns.get(i).attnum == c.attnum) {
+                            columns.set(i, columnInfo(c.attname, c.attnum, c.atttypid, c.attlen, c.attalign, c.attisdropped));
+                            break;
+                        }
+                    }
+                    break;
+                }
+                case "DELETE":
+                    columns.removeIf(col -> col.attnum == c.attnum);
+                    break;
+            }
+        }
+        return new RelationInfo(rel.schema, rel.table, columns,
+                new ArrayList<>(rel.keyColumns), rel.replicaIdentityFull);
+    }
+
     /**
      * Captures a single pg_attribute row change decoded from WAL, stored until
      * the affected table's RelationInfo is loaded into the cache.
@@ -227,13 +275,15 @@ private RelationInfo load(long relNumber) {
             jdbcContext.query(String.format(COLUMNS, oid[0]), rs -> {
                 while (rs.next()) {
                     String align = rs.getString("attalign");
-                    columns.add(new ColumnInfo(
+                    long enumTypeOid = rs.getLong("enumtypid");
+                    columns.add(columnInfo(
                             rs.getString("attname"),
                             rs.getInt("attnum"),
                             rs.getLong("atttypid"),
                             rs.getInt("attlen"),
                             align == null || align.isEmpty() ? 'c' : align.charAt(0),
-                            rs.getBoolean("attisdropped")));
+                            rs.getBoolean("attisdropped"),
+                            enumTypeOid));
                 }
             });
             List<String> keys = new ArrayList<>();
@@ -248,4 +298,66 @@ private RelationInfo load(long relNumber) {
             return null;
         }
     }
+
+    private ColumnInfo columnInfo(String name, int attnum, long typeOid, int typLen, char typAlign, boolean dropped) {
+        return columnInfo(name, attnum, typeOid, typLen, typAlign, dropped, resolveEnumTypeOid(typeOid));
+    }
+
+    private ColumnInfo columnInfo(String name, int attnum, long typeOid, int typLen, char typAlign, boolean dropped,
+                                  long enumTypeOid) {
+        if (enumTypeOid <= 0) {
+            return new ColumnInfo(name, attnum, typeOid, typLen, typAlign, dropped);
+        }
+        EnumTypeInfo enumInfo = enumTypeInfo(enumTypeOid);
+        if (enumInfo == null || enumInfo.labels.isEmpty()) {
+            return new ColumnInfo(name, attnum, typeOid, typLen, typAlign, dropped);
+        }
+        return new ColumnInfo(name, attnum, typeOid, typLen, typAlign, dropped, enumTypeOid, enumInfo.labels);
+    }
+
+    private long resolveEnumTypeOid(long typeOid) {
+        long[] out = {0L};
+        try {
+            jdbcContext.queryWithNext(
+                    "SELECT CASE WHEN t.typtype = 'e' THEN t.oid WHEN et.typtype = 'e' THEN et.oid ELSE 0 END AS enumtypid "
+                            + "FROM pg_type t LEFT JOIN pg_type et ON t.typelem = et.oid WHERE t.oid = " + typeOid,
+                    rs -> out[0] = rs.getLong("enumtypid"));
+        } catch (Throwable e) {
+            return 0L;
+        }
+        return out[0];
+    }
+
+    private EnumTypeInfo enumTypeInfo(long enumTypeOid) {
+        if (enumTypeOid <= 0) {
+            return null;
+        }
+        EnumTypeInfo cached = enumTypeCache.get(enumTypeOid);
+        if (cached != null) {
+            return cached;
+        }
+        Map<Long, String> labels = new HashMap<>();
+        try {
+            jdbcContext.query(
+                    "SELECT oid, enumlabel FROM pg_enum WHERE enumtypid = " + enumTypeOid + " ORDER BY enumsortorder, oid",
+                    rs -> {
+                        while (rs.next()) {
+                            labels.put(rs.getLong("oid"), rs.getString("enumlabel"));
+                        }
+                    });
+        } catch (Throwable e) {
+            return null;
+        }
+        EnumTypeInfo loaded = new EnumTypeInfo(labels);
+        enumTypeCache.put(enumTypeOid, loaded);
+        return loaded;
+    }
+
+    private static final class EnumTypeInfo {
+        final Map<Long, String> labels;
+
+        EnumTypeInfo(Map<Long, String> labels) {
+            this.labels = labels;
+        }
+    }
 }

connectors-common/postgres-core/src/test/java/io/tapdata/connector/postgres/cdc/physical/PgTypeDecoderTest.java

@@ -7,6 +7,8 @@
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -44,6 +46,33 @@ public void testNamePadStripped() {
         assertEquals("id", PgTypeDecoder.decode(PgTypeDecoder.NAME, buf));
     }
 
+    @Test
+    public void testDecodeEnumLabel() {
+        Map<Long, String> labels = new HashMap<>();
+        labels.put(70001L, "APPROVED");
+
+        assertEquals("APPROVED", PgTypeDecoder.decode(50001L, le32(70001L), 50001L, labels));
+    }
+
+    @Test
+    public void testDecodeEnumArrayWithUserTypeOid() {
+        Map<Long, String> labels = new HashMap<>();
+        labels.put(70001L, "STAGE");
+        labels.put(70002L, "WEB");
+
+        java.io.ByteArrayOutputStream o = new java.io.ByteArrayOutputStream();
+        write32(o, 1);      // ndim
+        write32(o, 0);      // dataoffset
+        write32(o, 50001);  // elemtype: user-defined enum OID
+        write32(o, 2);      // dimension length
+        write32(o, 1);      // lower bound
+        write32(o, 70001);  // enum value OID -> STAGE
+        write32(o, 70002);  // enum value OID -> WEB
+
+        Object result = PgTypeDecoder.decode(50002L, o.toByteArray(), 50001L, labels);
+        assertEquals(Arrays.asList("STAGE", "WEB"), result);
+    }
+
     @Test
     public void testDate() {
         // 2000-01-02 -> 1 day after PG epoch