removed debug logging

flow/activities/flowable.go

@@ -721,5 +721,9 @@ func (a *FlowableActivity) CreateAdditionalTable(
 		TableIdentifierMapping[srcTableIdentifier].GetPostgresTableIdentifier().RelId
 	input.AdditionalTableInfo.TableSchema = tableNameSchemaMapping.TableNameSchemaMapping[srcTableIdentifier]
 
+	log.WithFields(log.Fields{
+		"flowName": input.FlowConnectionConfigs.FlowJobName,
+	}).Infof("finished creating additional table %s\n", dstTableIdentifier)
+
 	return input.AdditionalTableInfo, nil
 }

flow/connectors/postgres/cdc.go

@@ -329,8 +329,6 @@ func (p *postgresCDCSource) processMessage(batch *model.RecordBatch, xld pglogre
 		log.Infof("RelationMessage => RelationID: %d, Namespace: %s, RelationName: %s, Columns: %v",
 			msg.RelationID, msg.Namespace, msg.RelationName, msg.Columns)
 		if p.relationMessageMapping[msg.RelationID] == nil {
-			fmt.Println("tableNameMapping: ", p.tableNameMapping)
-			fmt.Println("srcTableIDNameMapping: ", p.srcTableIDNameMapping)
 			_, ok := p.tableNameMapping[fmt.Sprintf("%s.%s", msg.Namespace, msg.RelationName)]
 			// either it's a table we are aware of from SetupFlow, so it is in the map
 			// or it is a table we are not aware of, in a schema we are aware of.

flow/connectors/postgres/client.go

@@ -225,7 +225,6 @@ func (c *PostgresConnector) createSlotAndPublication(
 
 			stmt = fmt.Sprintf("CREATE PUBLICATION %s FOR TABLE %s", publication, tableNameString)
 		}
-		fmt.Println("PUBLICATION stmt: ", stmt)
 		_, err := c.pool.Exec(c.ctx, stmt)
 		if err != nil {
 			log.Warnf("Error creating publication '%s': %v", publication, err)

flow/workflows/cdc_flow.go

@@ -286,10 +286,6 @@ func CDCFlowWorkflowWithConfig(
 					cfg.SrcTableIdNameMapping[childSyncFlowRes.AdditionalTableInfo.RelId] = srcTableIdentifier
 					// because
 					cfg.TableNameSchemaMapping[dstTableIdentifier] = childSyncFlowRes.AdditionalTableInfo.TableSchema
-
-					fmt.Println("cfg tableNameMapping: ", cfg.TableNameMapping)
-					fmt.Println("cfg SrcTableIdNameMapping: ", cfg.SrcTableIdNameMapping)
-					fmt.Println("cfg TableNameSchemaMapping: ", cfg.TableNameSchemaMapping)
 				}
 			}
 		}

flow/workflows/setup_flow.go

@@ -110,14 +110,14 @@ func (s *SetupFlowExecution) ensurePullability(
 	})
 	tmpMap := make(map[uint32]string)
 
-	srcIdentifiers := maps.Keys(config.TableNameMapping)
-	slices.Sort(srcIdentifiers)
+	srcTblIdentifiers := maps.Keys(config.TableNameMapping)
+	slices.Sort(srcTblIdentifiers)
 
 	// create EnsurePullabilityInput for the srcTableName
 	ensurePullabilityInput := &protos.EnsurePullabilityBatchInput{
 		PeerConnectionConfig:   config.Source,
 		FlowJobName:            s.CDCFlowName,
-		SourceTableIdentifiers: srcIdentifiers,
+		SourceTableIdentifiers: srcTblIdentifiers,
 	}
 
 	future := workflow.ExecuteActivity(ctx, flowable.EnsurePullability, ensurePullabilityInput)