tapdata · jarad0628 · Mar 20, 2025 · Feb 6, 2025 · Feb 8, 2025 · Feb 10, 2025
@@ -157,7 +157,11 @@ public static boolean ignoreCreateIndex(TapIndex exists, TapIndex created) {
     }
 
     public static String buildIndexName(String table) {
-        return "TAPIDX_" + table.substring(Math.max(table.length() - 10, 0)) + UUID.randomUUID().toString().replaceAll("-", "").substring(20);
+        return "IDX_" + table.substring(Math.max(table.length() - 10, 0)) + UUID.randomUUID().toString().replaceAll("-", "").substring(20);
+    }
+
+    public static String buildForeignKeyName(String table) {
+        return "FK_" + table.substring(Math.max(table.length() - 10, 0)) + UUID.randomUUID().toString().replaceAll("-", "").substring(20);
     }
 
     public static <T> List<List<T>> splitToPieces(List<T> data, int eachPieceSize) {

@@ -6,6 +6,7 @@
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
+import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
@@ -190,7 +191,7 @@ public static String removeHeadTail(String str, String remove, Boolean upperCase
         if (EmptyKit.isBlank(str) || EmptyKit.isBlank(remove)) {
             return str;
         }
-        if (str.startsWith(remove) && str.endsWith(remove) && str.length() > 2 * remove.length()) {
+        if (str.startsWith(remove) && str.endsWith(remove) && str.length() >= 2 * remove.length()) {
             return str.substring(remove.length(), str.length() - remove.length());
         }
         if (EmptyKit.isNull(upperCase)) {
@@ -342,4 +343,27 @@ public static String trimTailBlank(Object str) {
         if (null == str) return null;
         return ("_" + str).trim().substring(1);
     }
+
+    public static String escape(String name, String escapes) {
+        String res = name;
+        for (int i = 0; i < escapes.length(); i++) {
+            char escape = escapes.charAt(i);
+            res = escape(res, escape);
+        }
+        return res;
+    }
+
+    public static String escape(String name, char escape) {
+        return name.replace(escape + "", "" + escape + escape);
+    }
+
+    private static final Pattern REGEX_SPECIAL_CHARS = Pattern.compile("[\\\\^$|*+?.,()\\[\\]{}]");
+
+    public static String escapeRegex(String input) {
+        if (input == null || input.isEmpty()) {
+            return input;
+        }
+        Matcher matcher = REGEX_SPECIAL_CHARS.matcher(input);
+        return matcher.replaceAll("\\\\$0");
+    }
 }
@@ -583,7 +583,7 @@ private void tableUnlock() throws SQLException {
     }
 
     private String quote(String dbOrTableName) {
-        return "`" + dbOrTableName + "`";
+        return "`" + dbOrTableName.replace("`", "``") + "`";
     }
 
     private String quote(TableId id) {

@@ -63,7 +63,7 @@ public PostgresChangeRecordEmitter(OffsetContext offset, Clock clock, PostgresCo
         this.connectorConfig = connectorConfig;
         this.connection = connection;
 
-        this.tableId = PostgresSchema.parse(message.getTable());
+        this.tableId = message.getTableId();
         this.unchangedToastColumnMarkerMissing = !connectorConfig.plugin().hasUnchangedToastColumnMarker();
         this.nullToastedValuesMissingFromOld = !connectorConfig.plugin().sendsNullToastedValuesInOld();
         Objects.requireNonNull(tableId);

@@ -213,7 +213,7 @@ private void refreshToastableColumnsMap(PostgresConnection connection, TableId t
         tableIdToToastableColumns.put(tableId, Collections.unmodifiableList(toastableColumns));
     }
 
-    protected static TableId parse(String table) {
+    public static TableId parse(String table) {
         TableId tableId = TableId.parse(table, false);
         if (tableId == null) {
             return null;

@@ -5,23 +5,8 @@
  */
 package io.debezium.connector.postgresql;
 
-import java.sql.SQLException;
-import java.util.Map;
-import java.util.Objects;
-import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicReference;
-
-import org.apache.kafka.connect.errors.ConnectException;
-import org.postgresql.core.BaseConnection;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import io.debezium.connector.postgresql.connection.Lsn;
-import io.debezium.connector.postgresql.connection.PostgresConnection;
-import io.debezium.connector.postgresql.connection.ReplicationConnection;
+import io.debezium.connector.postgresql.connection.*;
 import io.debezium.connector.postgresql.connection.ReplicationMessage.Operation;
-import io.debezium.connector.postgresql.connection.ReplicationStream;
-import io.debezium.connector.postgresql.connection.WalPositionLocator;
 import io.debezium.connector.postgresql.spi.Snapshotter;
 import io.debezium.heartbeat.Heartbeat;
 import io.debezium.pipeline.ErrorHandler;
@@ -30,9 +15,18 @@
 import io.debezium.relational.TableId;
 import io.debezium.util.Clock;
 import io.debezium.util.DelayStrategy;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.postgresql.core.BaseConnection;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.SQLException;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
- *
  * @author Horia Chiorean (hchiorea@redhat.com), Jiri Pechanec
  */
 public class PostgresStreamingChangeEventSource implements StreamingChangeEventSource {
@@ -104,8 +98,7 @@ public void execute(ChangeEventSourceContext context) throws InterruptedExceptio
                 LOGGER.info("Retrieved latest position from stored offset '{}'", lsn);
                 walPosition = new WalPositionLocator(offsetContext.lastCommitLsn(), lsn);
                 replicationStream.compareAndSet(null, replicationConnection.startStreaming(lsn, walPosition));
-            }
-            else {
+            } else {
                 LOGGER.info("No previous LSN found in Kafka, streaming from the latest xlogpos or flushed LSN...");
                 walPosition = new WalPositionLocator();
                 replicationStream.compareAndSet(null, replicationConnection.startStreaming(walPosition));
@@ -133,8 +126,7 @@ public void execute(ChangeEventSourceContext context) throws InterruptedExceptio
                     if (!isInPreSnapshotCatchUpStreaming()) {
                         connection.commit();
                     }
-                }
-                catch (Exception e) {
+                } catch (Exception e) {
                     LOGGER.info("Commit failed while preparing for reconnect", e);
                 }
                 walPosition.enableFiltering();
@@ -145,11 +137,9 @@ public void execute(ChangeEventSourceContext context) throws InterruptedExceptio
                 stream.startKeepAlive(Executors.newSingleThreadExecutor());
             }
             processMessages(context, stream);
-        }
-        catch (Throwable e) {
+        } catch (Throwable e) {
             errorHandler.setProducerThrowable(e);
-        }
-        finally {
+        } finally {
             if (replicationConnection != null) {
                 LOGGER.debug("stopping streaming...");
                 // stop the keep alive thread, this also shuts down the
@@ -166,8 +156,7 @@ public void execute(ChangeEventSourceContext context) throws InterruptedExceptio
                         connection.commit();
                     }
                     replicationConnection.close();
-                }
-                catch (Exception e) {
+                } catch (Exception e) {
                     LOGGER.debug("Exception while closing the connection", e);
                 }
                 replicationStream.set(null);
@@ -205,8 +194,7 @@ private void processMessages(ChangeEventSourceContext context, final Replication
                             taskContext.getSlotXmin(connection));
                     if (message.getOperation() == Operation.BEGIN) {
                         dispatcher.dispatchTransactionStartedEvent(Long.toString(message.getTransactionId()), offsetContext);
-                    }
-                    else if (message.getOperation() == Operation.COMMIT) {
+                    } else if (message.getOperation() == Operation.COMMIT) {
                         commitMessage(lsn);
                         dispatcher.dispatchTransactionCommittedEvent(offsetContext);
                     }
@@ -216,7 +204,7 @@ else if (message.getOperation() == Operation.COMMIT) {
                 else {
                     TableId tableId = null;
                     if (message.getOperation() != Operation.NOOP) {
-                        tableId = PostgresSchema.parse(message.getTable());
+                        tableId = message.getTableId();
                         Objects.requireNonNull(tableId);
                     }
 
@@ -239,8 +227,7 @@ else if (message.getOperation() == Operation.COMMIT) {
 
             if (receivedMessage) {
                 noMessageIterations = 0;
-            }
-            else {
+            } else {
                 if (offsetContext.hasCompletelyProcessedPosition()) {
                     dispatcher.dispatchHeartbeatEvent(offsetContext);
                 }
@@ -275,8 +262,7 @@ private void searchWalPosition(ChangeEventSourceContext context, final Replicati
 
             if (receivedMessage) {
                 noMessageIterations = 0;
-            }
-            else {
+            } else {
                 noMessageIterations++;
                 if (noMessageIterations >= THROTTLE_NO_MESSAGE_BEFORE_PAUSE) {
                     noMessageIterations = 0;
@@ -307,23 +293,21 @@ private void commitMessage(final Lsn lsn) throws SQLException, InterruptedExcept
      * The purpose of this method is to detect this situation and log a warning
      * every {@link #GROWING_WAL_WARNING_LOG_INTERVAL} filtered events.
      *
-     * @param dispatched
-     *            Whether an event was sent to the broker or not
+     * @param dispatched Whether an event was sent to the broker or not
      */
     private void maybeWarnAboutGrowingWalBacklog(boolean dispatched) {
         if (dispatched) {
             numberOfEventsSinceLastEventSentOrWalGrowingWarning = 0;
-        }
-        else {
+        } else {
             numberOfEventsSinceLastEventSentOrWalGrowingWarning++;
         }
 
         if (numberOfEventsSinceLastEventSentOrWalGrowingWarning > GROWING_WAL_WARNING_LOG_INTERVAL && !dispatcher.heartbeatsEnabled()) {
             LOGGER.warn("Received {} events which were all filtered out, so no offset could be committed. "
-                    + "This prevents the replication slot from acknowledging the processed WAL offsets, "
-                    + "causing a growing backlog of non-removeable WAL segments on the database server. "
-                    + "Consider to either adjust your filter configuration or enable heartbeat events "
-                    + "(via the {} option) to avoid this situation.",
+                            + "This prevents the replication slot from acknowledging the processed WAL offsets, "
+                            + "causing a growing backlog of non-removeable WAL segments on the database server. "
+                            + "Consider to either adjust your filter configuration or enable heartbeat events "
+                            + "(via the {} option) to avoid this situation.",
                     numberOfEventsSinceLastEventSentOrWalGrowingWarning, Heartbeat.HEARTBEAT_INTERVAL_PROPERTY_NAME);
 
             numberOfEventsSinceLastEventSentOrWalGrowingWarning = 0;
@@ -344,12 +328,10 @@ public void commitOffset(Map<String, ?> offset) {
                 }
                 // tell the server the point up to which we've processed data, so it can be free to recycle WAL segments
                 replicationStream.flushLsn(lsn);
-            }
-            else {
+            } else {
                 LOGGER.debug("Streaming has already stopped, ignoring commit callback...");
             }
-        }
-        catch (SQLException e) {
+        } catch (SQLException e) {
             throw new ConnectException(e);
         }
     }
@@ -358,7 +340,7 @@ public void commitOffset(Map<String, ?> offset) {
      * Returns whether the current streaming phase is running a catch up streaming
      * phase that runs before a snapshot. This is useful for transaction
      * management.
-     *
+     * <p>
      * During pre-snapshot catch up streaming, we open the snapshot transaction
      * early and hold the transaction open throughout the pre snapshot catch up
      * streaming phase so that we know where to stop streaming and can start the

@@ -12,6 +12,8 @@
 import java.time.OffsetTime;
 import java.util.List;
 
+import io.debezium.connector.postgresql.PostgresSchema;
+import io.debezium.relational.TableId;
 import org.postgresql.geometric.PGbox;
 import org.postgresql.geometric.PGcircle;
 import org.postgresql.geometric.PGline;
@@ -156,6 +158,10 @@ public interface ColumnValue<T> {
      */
     public String getTable();
 
+    default TableId getTableId() {
+        return PostgresSchema.parse(getTable());
+    }
+
     /**
      * @return Set of original values of table columns, null for INSERT
      */

@@ -5,6 +5,7 @@
  */
 package io.debezium.connector.postgresql.connection.pgoutput;
 
+import java.io.ByteArrayOutputStream;
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.sql.DatabaseMetaData;
@@ -250,8 +251,8 @@ private void handleCommitMessage(ByteBuffer buffer, ReplicationMessageProcessor
      */
     private void handleRelationMessage(ByteBuffer buffer, TypeRegistry typeRegistry) throws SQLException {
         int relationId = buffer.getInt();
-        String schemaName = readString(buffer);
-        String tableName = readString(buffer);
+        String schemaName = readStringV2(buffer);
+        String tableName = readStringV2(buffer);
         int replicaIdentityId = buffer.get();
         short columnCount = buffer.getShort();
 
@@ -276,7 +277,7 @@ private void handleRelationMessage(ByteBuffer buffer, TypeRegistry typeRegistry)
         Set<String> columnNames = new HashSet<>();
         for (short i = 0; i < columnCount; ++i) {
             byte flags = buffer.get();
-            String columnName = Strings.unquoteIdentifierPart(readString(buffer));
+            String columnName = Strings.unquoteIdentifierPart(readStringV2(buffer));
             int columnType = buffer.getInt();
             int attypmod = buffer.getInt();
 
@@ -379,7 +380,7 @@ private void decodeInsert(ByteBuffer buffer, TypeRegistry typeRegistry, Replicat
             List<Column> columns = resolveColumnsFromStreamTupleData(buffer, typeRegistry, table);
             processor.process(new PgOutputReplicationMessage(
                     Operation.INSERT,
-                    table.id().toDoubleQuotedString(),
+                    table.id(),
                     commitTimestamp,
                     transactionId,
                     null,
@@ -426,7 +427,7 @@ private void decodeUpdate(ByteBuffer buffer, TypeRegistry typeRegistry, Replicat
             List<Column> columns = resolveColumnsFromStreamTupleData(buffer, typeRegistry, table);
             processor.process(new PgOutputReplicationMessage(
                     Operation.UPDATE,
-                    table.id().toDoubleQuotedString(),
+                    table.id(),
                     commitTimestamp,
                     transactionId,
                     oldColumns,
@@ -459,7 +460,7 @@ private void decodeDelete(ByteBuffer buffer, TypeRegistry typeRegistry, Replicat
             List<Column> columns = resolveColumnsFromStreamTupleData(buffer, typeRegistry, table);
             processor.process(new PgOutputReplicationMessage(
                     Operation.DELETE,
-                    table.id().toDoubleQuotedString(),
+                    table.id(),
                     commitTimestamp,
                     transactionId,
                     columns,
@@ -512,7 +513,7 @@ private void decodeTruncate(ByteBuffer buffer, TypeRegistry typeRegistry, Replic
             boolean lastTableInTruncate = (i + 1) == noOfResolvedTables;
             processor.process(new PgOutputTruncateReplicationMessage(
                     Operation.TRUNCATE,
-                    table.id().toDoubleQuotedString(),
+                    table.id(),
                     commitTimestamp,
                     transactionId,
                     lastTableInTruncate));
@@ -596,6 +597,15 @@ private static String readString(ByteBuffer buffer) {
         return sb.toString();
     }
 
+    private static String readStringV2(ByteBuffer buffer) {
+        ByteArrayOutputStream buf = new ByteArrayOutputStream();
+        byte b = 0;
+        while ((b = buffer.get()) != 0) {
+            buf.write(b);
+        }
+        return buf.toString();
+    }
+
     /**
      * Reads the replication stream where the column stream specifies a length followed by the value.
      *