QRecordSchema: close schemaLatch in Close

goroutines were leaking waiting on schemaLatch,
if provider hit error then schemaLatch could be left indefinitely open

flow/connectors/bigquery/qrep.go

@@ -24,6 +24,9 @@ func (c *BigQueryConnector) SyncQRepRecords(
 	// Ensure the destination table is available.
 	destTable := config.DestinationTableIdentifier
 	srcSchema := stream.Schema()
+	if srcSchema.Fields == nil {
+		return 0, stream.Err()
+	}
 
 	tblMetadata, err := c.replayTableSchemaDeltasQRep(ctx, config, partition, srcSchema)
 	if err != nil {

flow/connectors/clickhouse/qrep_avro_sync.go

@@ -68,6 +68,9 @@ func (s *ClickHouseAvroSyncMethod) SyncRecords(
 	dstTableName := s.config.DestinationTableIdentifier
 
 	schema := stream.Schema()
+	if schema.Fields == nil {
+		return 0, stream.Err()
+	}
 	s.logger.Info("sync function called and schema acquired",
 		slog.String("dstTable", dstTableName))
 

flow/connectors/elasticsearch/qrep.go

@@ -43,6 +43,9 @@ func (esc *ElasticsearchConnector) SyncQRepRecords(ctx context.Context, config *
 	startTime := time.Now()
 
 	schema := stream.Schema()
+	if schema.Fields == nil {
+		return 0, stream.Err()
+	}
 
 	var bulkIndexFatalError error
 	var bulkIndexErrors []error

flow/connectors/kafka/qrep.go

@@ -28,6 +28,9 @@ func (c *KafkaConnector) SyncQRepRecords(
 	startTime := time.Now()
 	numRecords := atomic.Int64{}
 	schema := stream.Schema()
+	if schema.Fields == nil {
+		return 0, stream.Err()
+	}
 
 	queueCtx, queueErr := context.WithCancelCause(ctx)
 	pool, err := c.createPool(queueCtx, config.Env, config.Script, config.FlowJobName, nil, queueErr)

flow/connectors/postgres/qrep_query_executor.go

@@ -194,7 +194,6 @@ func (qe *QRepQueryExecutor) processFetchedRows(
 			slog.Any("error", err), slog.String("query", query))
 		return 0, fmt.Errorf("[pg_query_executor] failed to execute query in tx: %w", err)
 	}
-
 	defer rows.Close()
 
 	fieldDescriptions := rows.FieldDescriptions()

flow/connectors/pubsub/qrep.go

@@ -25,11 +25,14 @@ func (c *PubSubConnector) SyncQRepRecords(
 	stream *model.QRecordStream,
 ) (int, error) {
 	startTime := time.Now()
-	numRecords := atomic.Int64{}
 	schema := stream.Schema()
+	if schema.Fields == nil {
+		return 0, stream.Err()
+	}
 	topiccache := topicCache{cache: make(map[string]*pubsub.Topic)}
 	publish := make(chan publishResult, 32)
 	waitChan := make(chan struct{})
+	numRecords := atomic.Int64{}
 
 	queueCtx, queueErr := context.WithCancelCause(ctx)
 	pool, err := c.createPool(queueCtx, config.Env, config.Script, config.FlowJobName, &topiccache, publish, queueErr)

flow/connectors/s3/qrep.go

@@ -18,6 +18,9 @@ func (c *S3Connector) SyncQRepRecords(
 	stream *model.QRecordStream,
 ) (int, error) {
 	schema := stream.Schema()
+	if schema.Fields == nil {
+		return 0, stream.Err()
+	}
 
 	dstTableName := config.DestinationTableIdentifier
 	avroSchema, err := getAvroSchema(ctx, config.Env, dstTableName, schema)

flow/connectors/snowflake/qrep_avro_sync.go

@@ -45,6 +45,9 @@ func (s *SnowflakeAvroSyncHandler) SyncRecords(
 	dstTableName := s.config.DestinationTableIdentifier
 
 	schema := stream.Schema()
+	if schema.Fields == nil {
+		return 0, stream.Err()
+	}
 
 	s.logger.Info("sync function called and schema acquired", tableLog)
 

flow/connectors/utils/avro/avro_writer.go

@@ -130,6 +130,9 @@ func (p *peerDBOCFWriter) createOCFWriter(w io.Writer) (*goavro.OCFWriter, error
 func (p *peerDBOCFWriter) writeRecordsToOCFWriter(ctx context.Context, env map[string]string, ocfWriter *goavro.OCFWriter) (int64, error) {
 	logger := shared.LoggerFromCtx(ctx)
 	schema := p.stream.Schema()
+	if schema.Fields == nil {
+		return 0, p.stream.Err()
+	}
 
 	avroConverter, err := model.NewQRecordAvroConverter(
 		ctx,

flow/model/qrecord_stream.go

@@ -55,4 +55,7 @@ func (s *QRecordStream) Close(err error) {
 		s.err = err
 		close(s.Records)
 	}
+	if !s.schemaSet {
+		s.SetSchema(qvalue.QRecordSchema{})
+	}
 }

flow/model/qvalue/qschema.go

@@ -34,10 +34,7 @@ func (q QRecordSchema) EqualNames(other QRecordSchema) bool {
 	}
 
 	for i, field := range q.Fields {
-		// ignore the case of the field name convert to lower case
-		f1 := strings.ToLower(field.Name)
-		f2 := strings.ToLower(other.Fields[i].Name)
-		if f1 != f2 {
+		if !strings.EqualFold(field.Name, other.Fields[i].Name) {
 			return false
 		}
 	}