account for peerdb cols in sf, better logging, checking

flow/connectors/bigquery/bigquery.go

@@ -949,10 +949,23 @@ func (c *BigQueryConnector) SetupNormalizedTables(
 		if err == nil { // it exists.
 			// check if the structure of the table is as needed
 			existingSchema := metadata.Schema
-			for i, existingField := range existingSchema {
-				desiredField := schema[i]
+			if len(existingSchema) != len(schema) {
+				return nil, fmt.Errorf("failed to setup normalized table: number of columns on both sides differ")
+			}
+			for _, existingField := range existingSchema {
+				var desiredField *bigquery.FieldSchema
+				for id := range schema {
+					if schema[id].Name == existingField.Name {
+						desiredField = schema[id]
+						break
+					}
+				}
+				if desiredField == nil {
+					return nil, fmt.Errorf("destination column %s not found in source: ", existingField.Name)
+				}
 				if existingField.Name != desiredField.Name || existingField.Type != desiredField.Type {
-					return nil, fmt.Errorf("failed to setup normalized table due to incompatible columns")
+					return nil, fmt.Errorf("failed to setup normalized table due to incompatible columns: %s (destination) and %s (source)",
+						existingField.Name, desiredField.Name)
 				}
 			}
 			// table exists, go to next table

flow/connectors/postgres/postgres.go

@@ -650,7 +650,7 @@ func (c *PostgresConnector) SetupNormalizedTables(req *protos.SetupNormalizedTab
 			tableExistsMapping[tableIdentifier] = true
 			log.Infoln("found existing normalized table, checking if it matches the desired schema")
 			if len(destinationColumns) != len(sourceColumns) {
-				return nil, fmt.Errorf("failed to setup normalized table: schemas on both sides differ")
+				return nil, fmt.Errorf("failed to setup normalized table: number of columns on both sides differ")
 			}
 			for id := range destinationColumns {
 				column := &destinationColumns[id]

flow/connectors/snowflake/snowflake.go

@@ -412,12 +412,12 @@ func (c *SnowflakeConnector) SetupNormalizedTables(
 		if destinationColumns != nil {
 			sourceColumns := req.TableNameSchemaMapping[tableIdentifier].Columns
 			log.Infoln("found existing normalized table, checking if it matches the desired schema")
-			if len(destinationColumns) != len(sourceColumns) {
-				return nil, fmt.Errorf("failed to setup normalized table: schemas on both sides differ")
-			}
 			for id := range destinationColumns {
 				column := &destinationColumns[id]
 				existingName := strings.ToLower(column.Name)
+				if _, found := strings.CutPrefix(existingName, "_peerdb"); found {
+					continue
+				}
 				sourceType, ok := sourceColumns[existingName]
 				if !ok {
 					return nil, fmt.Errorf("failed to setup normalized table:"+