Longer ?

pgflo · Dec 24, 2024 · 6a7446f · 6a7446f
1 parent 314af54
commit 6a7446f
Show file tree

Hide file tree

Showing 3 changed files with 2 additions and 10 deletions.
diff --git a/internal/scripts/e2e_copy_and_stream.sh b/internal/scripts/e2e_copy_and_stream.sh
@@ -216,7 +216,7 @@ test_pg_flo_cdc() {
   simulate_concurrent_changes
 
   log "Waiting for changes to replicate..."
-  sleep 90
+  sleep 180
   stop_pg_flo_gracefully
   compare_row_counts || return 1
   verify_large_json || return 1

diff --git a/internal/scripts/e2e_copy_only.sh b/internal/scripts/e2e_copy_only.sh
@@ -88,6 +88,7 @@ start_pg_flo_worker() {
     --target-dbname "$TARGET_PG_DB" \
     --target-user "$TARGET_PG_USER" \
     --target-password "$TARGET_PG_PASSWORD" \
+    --batch-size 5000 \
     --target-sync-schema \
     >"$pg_flo_WORKER_LOG" 2>&1 &
   pg_flo_WORKER_PID=$!

diff --git a/pkg/sinks/postgres.go b/pkg/sinks/postgres.go
@@ -154,7 +154,6 @@ func getWhereConditions(message *utils.CDCMessage, useOldValues bool, startingIn
 	var conditions []string
 	var values []interface{}
 	valueIndex := startingIndex
-	// log.Info().Any("message.ReplicationKey", message.ReplicationKey).Msg("getWhereConditions")
 	switch message.ReplicationKey.Type {
 	case utils.ReplicationKeyFull:
 		// For FULL, use all non-null values
@@ -241,14 +240,6 @@ func (s *PostgresSink) handleUpdate(tx pgx.Tx, message *utils.CDCMessage) error
 
 	// Iterate over Data map directly
 	for colName, value := range message.Data {
-		// // Skip toasted columns only if they're marked as toasted AND null
-		// if message.IsColumnToasted(colName) && value == nil {
-		// 	s.logger.Debug().
-		// 		Str("column", colName).
-		// 		Msg("Skipping toasted column")
-		// 	continue
-		// }
-
 		if value == nil {
 			setClauses = append(setClauses, fmt.Sprintf("%s = NULL", colName))
 		} else {