Merge branch 'main' into fix-metrics

.github/workflows/rust-lint.yml

@@ -25,9 +25,6 @@ jobs:
         with:
           components: clippy
 
-      - name: disable clippy for sqlparser-rs
-        run: sed -i '1s/^/#![allow(clippy::all)]\n/' nexus/sqlparser-rs/src/lib.rs
-
       - name: clippy
         run: cargo clippy -- -D warnings
         working-directory: ./nexus

flow/activities/flowable.go

@@ -20,6 +20,7 @@ import (
 	"github.com/PeerDB-io/peer-flow/shared"
 	"github.com/jackc/pglogrepl"
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgtype"
 	log "github.com/sirupsen/logrus"
 	"go.temporal.io/sdk/activity"
 	"golang.org/x/sync/errgroup"
@@ -719,7 +720,6 @@ func (a *FlowableActivity) DropFlow(ctx context.Context, config *protos.Shutdown
 }
 
 func getPostgresPeerConfigs(ctx context.Context) ([]*protos.Peer, error) {
-	var peerOptions sql.RawBytes
 	catalogPool, catalogErr := catalog.GetCatalogConnectionPoolFromEnv()
 	if catalogErr != nil {
 		return nil, fmt.Errorf("error getting catalog connection pool: %w", catalogErr)
@@ -735,8 +735,9 @@ func getPostgresPeerConfigs(ctx context.Context) ([]*protos.Peer, error) {
 		return nil, err
 	}
 	defer optionRows.Close()
-	var peerName string
+	var peerName pgtype.Text
 	var postgresPeers []*protos.Peer
+	var peerOptions sql.RawBytes
 	for optionRows.Next() {
 		err := optionRows.Scan(&peerName, &peerOptions)
 		if err != nil {
@@ -748,7 +749,7 @@ func getPostgresPeerConfigs(ctx context.Context) ([]*protos.Peer, error) {
 			return nil, unmarshalErr
 		}
 		postgresPeers = append(postgresPeers, &protos.Peer{
-			Name:   peerName,
+			Name:   peerName.String,
 			Type:   protos.DBType_POSTGRES,
 			Config: &protos.Peer_PostgresConfig{PostgresConfig: &pgPeerConfig},
 		})

flow/cmd/handler.go

@@ -13,6 +13,7 @@ import (
 	peerflow "github.com/PeerDB-io/peer-flow/workflows"
 	backoff "github.com/cenkalti/backoff/v4"
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/jackc/pgx/v5/pgxpool"
 	log "github.com/sirupsen/logrus"
 	"go.temporal.io/sdk/client"
@@ -36,14 +37,14 @@ func NewFlowRequestHandler(temporalClient client.Client, pool *pgxpool.Pool, tas
 }
 
 func (h *FlowRequestHandler) getPeerID(ctx context.Context, peerName string) (int32, int32, error) {
-	var id int32
-	var peerType int32
+	var id pgtype.Int4
+	var peerType pgtype.Int4
 	err := h.pool.QueryRow(ctx, "SELECT id,type FROM peers WHERE name = $1", peerName).Scan(&id, &peerType)
 	if err != nil {
 		log.Errorf("unable to query peer id for peer %s: %s", peerName, err.Error())
 		return -1, -1, fmt.Errorf("unable to query peer id for peer %s: %s", peerName, err)
 	}
-	return id, peerType, nil
+	return id.Int32, peerType.Int32, nil
 }
 
 func schemaForTableIdentifier(tableIdentifier string, peerDBType int32) string {
@@ -639,7 +640,7 @@ func (h *FlowRequestHandler) DropPeer(
 		}, fmt.Errorf("failed to obtain peer ID for peer %s: %v", req.PeerName, err)
 	}
 
-	var inMirror int64
+	var inMirror pgtype.Int8
 	queryErr := h.pool.QueryRow(ctx,
 		"SELECT COUNT(*) FROM flows WHERE source_peer=$1 or destination_peer=$2",
 		peerID, peerID).Scan(&inMirror)
@@ -650,7 +651,7 @@ func (h *FlowRequestHandler) DropPeer(
 		}, fmt.Errorf("failed to check for existing mirrors with peer %s", req.PeerName)
 	}
 
-	if inMirror != 0 {
+	if inMirror.Int64 != 0 {
 		return &protos.DropPeerResponse{
 			Ok:           false,
 			ErrorMessage: fmt.Sprintf("Peer %s is currently involved in an ongoing mirror.", req.PeerName),

flow/cmd/peer_data.go

@@ -8,6 +8,7 @@ import (
 	connpostgres "github.com/PeerDB-io/peer-flow/connectors/postgres"
 	"github.com/PeerDB-io/peer-flow/connectors/utils"
 	"github.com/PeerDB-io/peer-flow/generated/protos"
+	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"google.golang.org/protobuf/proto"
 )
@@ -92,13 +93,13 @@ func (h *FlowRequestHandler) GetTablesInSchema(
 	defer rows.Close()
 	var tables []string
 	for rows.Next() {
-		var table string
+		var table pgtype.Text
 		err := rows.Scan(&table)
 		if err != nil {
 			return &protos.SchemaTablesResponse{Tables: nil}, err
 		}
 
-		tables = append(tables, table)
+		tables = append(tables, table.String)
 	}
 	return &protos.SchemaTablesResponse{Tables: tables}, nil
 }
@@ -123,13 +124,13 @@ func (h *FlowRequestHandler) GetAllTables(
 	defer rows.Close()
 	var tables []string
 	for rows.Next() {
-		var table string
+		var table pgtype.Text
 		err := rows.Scan(&table)
 		if err != nil {
 			return &protos.AllTablesResponse{Tables: nil}, err
 		}
 
-		tables = append(tables, table)
+		tables = append(tables, table.String)
 	}
 	return &protos.AllTablesResponse{Tables: tables}, nil
 }
@@ -185,14 +186,14 @@ func (h *FlowRequestHandler) GetColumns(
 	defer rows.Close()
 	var columns []string
 	for rows.Next() {
-		var columnName string
-		var datatype string
-		var isPkey bool
+		var columnName pgtype.Text
+		var datatype pgtype.Text
+		var isPkey pgtype.Bool
 		err := rows.Scan(&columnName, &datatype, &isPkey)
 		if err != nil {
 			return &protos.TableColumnsResponse{Columns: nil}, err
 		}
-		column := fmt.Sprintf("%s:%s:%v", columnName, datatype, isPkey)
+		column := fmt.Sprintf("%s:%s:%v", columnName.String, datatype.String, isPkey.Bool)
 		columns = append(columns, column)
 	}
 	return &protos.TableColumnsResponse{Columns: columns}, nil
@@ -211,6 +212,8 @@ func (h *FlowRequestHandler) GetSlotInfo(
 	if err != nil {
 		return &protos.PeerSlotResponse{SlotData: nil}, err
 	}
+	defer pgConnector.Close()
+
 	slotInfo, err := pgConnector.GetSlotInfo("")
 	if err != nil {
 		return &protos.PeerSlotResponse{SlotData: nil}, err

flow/connectors/external_metadata/store.go

@@ -7,6 +7,7 @@ import (
 	"github.com/PeerDB-io/peer-flow/connectors/utils"
 	cc "github.com/PeerDB-io/peer-flow/connectors/utils/catalog"
 	"github.com/PeerDB-io/peer-flow/generated/protos"
+	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/jackc/pgx/v5/pgxpool"
 	log "github.com/sirupsen/logrus"
 )
@@ -76,14 +77,14 @@ func (p *PostgresMetadataStore) NeedsSetupMetadata() bool {
 	// check if schema exists
 	rows := p.pool.QueryRow(p.ctx, "SELECT count(*) FROM pg_catalog.pg_namespace WHERE nspname = $1", p.schemaName)
 
-	var exists int64
+	var exists pgtype.Int8
 	err := rows.Scan(&exists)
 	if err != nil {
 		log.Errorf("failed to check if schema exists: %v", err)
 		return false
 	}
 
-	if exists > 0 {
+	if exists.Int64 > 0 {
 		return true
 	}
 
@@ -136,7 +137,7 @@ func (p *PostgresMetadataStore) FetchLastOffset(jobName string) (*protos.LastSyn
 		FROM `+p.schemaName+`.`+lastSyncStateTableName+`
 		WHERE job_name = $1
 	`, jobName)
-	var offset int64
+	var offset pgtype.Int8
 	err := rows.Scan(&offset)
 	if err != nil {
 		// if the job doesn't exist, return 0
@@ -152,10 +153,10 @@ func (p *PostgresMetadataStore) FetchLastOffset(jobName string) (*protos.LastSyn
 		return nil, err
 	}
 
-	log.Infof("got last offset for job `%s`: %d", jobName, offset)
+	log.Infof("got last offset for job `%s`: %d", jobName, offset.Int64)
 
 	return &protos.LastSyncState{
-		Checkpoint: offset,
+		Checkpoint: offset.Int64,
 	}, nil
 }
 
@@ -166,7 +167,7 @@ func (p *PostgresMetadataStore) GetLastBatchID(jobName string) (int64, error) {
 		WHERE job_name = $1
 	`, jobName)
 
-	var syncBatchID int64
+	var syncBatchID pgtype.Int8
 	err := rows.Scan(&syncBatchID)
 	if err != nil {
 		// if the job doesn't exist, return 0
@@ -180,9 +181,9 @@ func (p *PostgresMetadataStore) GetLastBatchID(jobName string) (int64, error) {
 		return 0, err
 	}
 
-	log.Infof("got last sync batch ID for job `%s`: %d", jobName, syncBatchID)
+	log.Infof("got last sync batch ID for job `%s`: %d", jobName, syncBatchID.Int64)
 
-	return syncBatchID, nil
+	return syncBatchID.Int64, nil
 }
 
 // update offset for a job

flow/connectors/postgres/cdc.go

@@ -90,14 +90,14 @@ func getChildToParentRelIDMap(ctx context.Context, pool *pgxpool.Pool) (map[uint
 	defer rows.Close()
 
 	childToParentRelIDMap := make(map[uint32]uint32)
-	var parentRelID uint32
-	var childRelID uint32
+	var parentRelID pgtype.Uint32
+	var childRelID pgtype.Uint32
 	for rows.Next() {
 		err := rows.Scan(&parentRelID, &childRelID)
 		if err != nil {
 			return nil, fmt.Errorf("error scanning child to parent relid map: %w", err)
 		}
-		childToParentRelIDMap[childRelID] = parentRelID
+		childToParentRelIDMap[childRelID.Uint32] = parentRelID.Uint32
 	}
 
 	return childToParentRelIDMap, nil