Merge branch 'main' into always-init-childrelid-map

.github/workflows/flow.yml

@@ -178,3 +178,4 @@ jobs:
           PEERDB_CATALOG_DATABASE: postgres
           PEERDB_QUEUE_FORCE_TOPIC_CREATION: "true"
           ELASTICSEARCH_TEST_ADDRESS: http://localhost:9200
+          CI_PG_VERSION: ${{ matrix.postgres-version }}

flow/connectors/postgres/client.go

@@ -657,9 +657,16 @@ func (c *PostgresConnector) getCurrentLSN(ctx context.Context) (pglogrepl.LSN, e
 	var result pgtype.Text
 	err := row.Scan(&result)
 	if err != nil {
-		return 0, fmt.Errorf("error while running query: %w", err)
+		return 0, fmt.Errorf("error while running query for current LSN: %w", err)
 	}
-	return pglogrepl.ParseLSN(result.String)
+	if !result.Valid || result.String == "" {
+		return 0, errors.New("error while getting current LSN: no LSN available")
+	}
+	lsn, err := pglogrepl.ParseLSN(result.String)
+	if err != nil {
+		return 0, fmt.Errorf("error while parsing LSN %s: %w", result.String, err)
+	}
+	return lsn, nil
 }
 
 func (c *PostgresConnector) getDefaultPublicationName(jobName string) string {

flow/connectors/postgres/qvalue_convert.go

@@ -584,18 +584,18 @@ func customTypeToQKind(typeName string) qvalue.QValueKind {
 // in tstzrange.
 // convertTimeRangeBound removes the +0000 UTC part
 func convertTimeRangeBound(timeBound interface{}) (string, error) {
+	if timeBound, isInfinite := timeBound.(pgtype.InfinityModifier); isInfinite {
+		return timeBound.String(), nil
+	}
+
 	layout := "2006-01-02 15:04:05 -0700 MST"
 	postgresFormat := "2006-01-02 15:04:05"
-	var convertedTime string
 	if timeBound != nil {
 		lowerParsed, err := time.Parse(layout, fmt.Sprint(timeBound))
 		if err != nil {
-			return "", fmt.Errorf("unexpected lower bound value in tstzrange. Error: %v", err)
+			return "", fmt.Errorf("unexpected bound value in tstzrange. Error: %v", err)
 		}
-		convertedTime = lowerParsed.Format(postgresFormat)
-	} else {
-		convertedTime = ""
+		return lowerParsed.Format(postgresFormat), nil
 	}
-
-	return convertedTime, nil
+	return "", nil
 }

flow/e2e/bigquery/bigquery.go

@@ -103,8 +103,7 @@ func SetupSuite(t *testing.T) PeerFlowE2ETestSuiteBQ {
 		t.Fatalf("Failed to create helper: %v", err)
 	}
 
-	err = bqHelper.RecreateDataset()
-	if err != nil {
+	if err := bqHelper.RecreateDataset(); err != nil {
 		t.Fatalf("Failed to recreate dataset: %v", err)
 	}
 

flow/e2e/bigquery/peer_flow_bq_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"os"
 	"testing"
 	"time"
 
@@ -20,6 +21,10 @@ import (
 )
 
 func TestPeerFlowE2ETestSuiteBQ(t *testing.T) {
+	if val, ok := os.LookupEnv("CI_PG_VERSION"); ok && val != "16" {
+		t.Skip("Only running in PG16 to reduce flakiness from high concurrency")
+	}
+
 	e2eshared.RunSuite(t, SetupSuite)
 }
 

flow/e2e/snowflake/peer_flow_sf_test.go

@@ -3,6 +3,7 @@ package e2e_snowflake
 import (
 	"context"
 	"fmt"
+	"os"
 	"testing"
 	"time"
 
@@ -17,6 +18,10 @@ import (
 )
 
 func TestPeerFlowE2ETestSuiteSF(t *testing.T) {
+	if val, ok := os.LookupEnv("CI_PG_VERSION"); ok && val != "17" {
+		t.Skip("Only running in PG17 to reduce flakiness from high concurrency")
+	}
+
 	e2eshared.RunSuite(t, SetupSuite)
 }