Merge pull request #10154 from ellemouton/graphPerf11

graph/db: use batch validation to improve SQL migration performance

graph/db/benchmark_test.go

@@ -431,9 +431,6 @@ func TestPopulateDBs(t *testing.T) {
 // postgres backend instead of the kvdb-sqlite backend.
 //
 // NOTE: this is a helper test and is not run by default.
-//
-// TODO(elle): this test reveals tht there may be an issue with the postgres
-// migration as it is super slow.
 func TestPopulateViaMigration(t *testing.T) {
 	t.Skipf("Skipping local helper test")
 
@@ -791,13 +788,13 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) {
 	// Set the various page sizes we want to test.
 	//
 	// NOTE: these are the sqlite paging testing values.
-	testSizes := []int{20, 50, 100, 150, 500}
+	testSizes := []uint32{20, 50, 100, 150, 500}
 
 	configOption := "MaxPageSize"
 	if testBatching {
 		configOption = "MaxBatchSize"
 
-		testSizes = []int{
+		testSizes = []uint32{
 			50, 100, 150, 200, 250, 300, 350,
 		}
 	}
@@ -809,10 +806,10 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) {
 		// Set the various page sizes we want to test.
 		//
 		// NOTE: these are the postgres paging values.
-		testSizes = []int{5000, 7000, 10000, 12000}
+		testSizes = []uint32{5000, 7000, 10000, 12000}
 
 		if testBatching {
-			testSizes = []int{
+			testSizes = []uint32{
 				1000, 2000, 5000, 7000, 10000,
 			}
 		}
@@ -831,7 +828,7 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) {
 				if testBatching {
 					cfg.MaxBatchSize = size
 				} else {
-					cfg.MaxPageSize = int32(size)
+					cfg.MaxPageSize = size
 				}
 
 				store := connectNativeSQLite(

graph/db/sql_store.go

@@ -55,6 +55,7 @@ type SQLQueries interface {
 	*/
 	UpsertNode(ctx context.Context, arg sqlc.UpsertNodeParams) (int64, error)
 	GetNodeByPubKey(ctx context.Context, arg sqlc.GetNodeByPubKeyParams) (sqlc.GraphNode, error)
+	GetNodesByIDs(ctx context.Context, ids []int64) ([]sqlc.GraphNode, error)
 	GetNodeIDByPubKey(ctx context.Context, arg sqlc.GetNodeIDByPubKeyParams) (int64, error)
 	GetNodesByLastUpdateRange(ctx context.Context, arg sqlc.GetNodesByLastUpdateRangeParams) ([]sqlc.GraphNode, error)
 	ListNodesPaginated(ctx context.Context, arg sqlc.ListNodesPaginatedParams) ([]sqlc.GraphNode, error)
@@ -97,6 +98,7 @@ type SQLQueries interface {
 	GetChannelsBySCIDRange(ctx context.Context, arg sqlc.GetChannelsBySCIDRangeParams) ([]sqlc.GetChannelsBySCIDRangeRow, error)
 	GetChannelBySCIDWithPolicies(ctx context.Context, arg sqlc.GetChannelBySCIDWithPoliciesParams) (sqlc.GetChannelBySCIDWithPoliciesRow, error)
 	GetChannelsBySCIDWithPolicies(ctx context.Context, arg sqlc.GetChannelsBySCIDWithPoliciesParams) ([]sqlc.GetChannelsBySCIDWithPoliciesRow, error)
+	GetChannelsByIDs(ctx context.Context, ids []int64) ([]sqlc.GetChannelsByIDsRow, error)
 	GetChannelAndNodesBySCID(ctx context.Context, arg sqlc.GetChannelAndNodesBySCIDParams) (sqlc.GetChannelAndNodesBySCIDRow, error)
 	HighestSCID(ctx context.Context, version int16) ([]byte, error)
 	ListChannelsByNodeID(ctx context.Context, arg sqlc.ListChannelsByNodeIDParams) ([]sqlc.ListChannelsByNodeIDRow, error)
@@ -131,6 +133,7 @@ type SQLQueries interface {
 	*/
 	UpsertZombieChannel(ctx context.Context, arg sqlc.UpsertZombieChannelParams) error
 	GetZombieChannel(ctx context.Context, arg sqlc.GetZombieChannelParams) (sqlc.GraphZombieChannel, error)
+	GetZombieChannelsSCIDs(ctx context.Context, arg sqlc.GetZombieChannelsSCIDsParams) ([]sqlc.GraphZombieChannel, error)
 	CountZombieChannels(ctx context.Context, version int16) (int64, error)
 	DeleteZombieChannel(ctx context.Context, arg sqlc.DeleteZombieChannelParams) (sql.Result, error)
 	IsZombieChannel(ctx context.Context, arg sqlc.IsZombieChannelParams) (bool, error)
@@ -140,6 +143,7 @@ type SQLQueries interface {
 	*/
 	GetPruneTip(ctx context.Context) (sqlc.GraphPruneLog, error)
 	GetPruneHashByHeight(ctx context.Context, blockHeight int64) ([]byte, error)
+	GetPruneEntriesForHeights(ctx context.Context, heights []int64) ([]sqlc.GraphPruneLog, error)
 	UpsertPruneLogEntry(ctx context.Context, arg sqlc.UpsertPruneLogEntryParams) error
 	DeletePruneLogEntriesInRange(ctx context.Context, arg sqlc.DeletePruneLogEntriesInRangeParams) error
 
@@ -148,6 +152,7 @@ type SQLQueries interface {
 	*/
 	InsertClosedChannel(ctx context.Context, scid []byte) error
 	IsClosedChannel(ctx context.Context, scid []byte) (bool, error)
+	GetClosedChannelsSCIDs(ctx context.Context, scids [][]byte) ([][]byte, error)
 }
 
 // BatchedSQLQueries is a version of SQLQueries that's capable of batched
@@ -591,15 +596,29 @@ func (s *SQLStore) AddChannelEdge(ctx context.Context,
 			alreadyExists = false
 		},
 		Do: func(tx SQLQueries) error {
-			_, err := insertChannel(ctx, tx, edge)
+			chanIDB := channelIDToBytes(edge.ChannelID)
 
-			// Silence ErrEdgeAlreadyExist so that the batch can
-			// succeed, but propagate the error via local state.
-			if errors.Is(err, ErrEdgeAlreadyExist) {
+			// Make sure that the channel doesn't already exist. We
+			// do this explicitly instead of relying on catching a
+			// unique constraint error because relying on SQL to
+			// throw that error would abort the entire batch of
+			// transactions.
+			_, err := tx.GetChannelBySCID(
+				ctx, sqlc.GetChannelBySCIDParams{
+					Scid:    chanIDB,
+					Version: int16(ProtocolV1),
+				},
+			)
+			if err == nil {
 				alreadyExists = true
 				return nil
+			} else if !errors.Is(err, sql.ErrNoRows) {
+				return fmt.Errorf("unable to fetch channel: %w",
+					err)
 			}
 
+			_, err = insertChannel(ctx, tx, edge)
+
 			return err
 		},
 		OnCommit: func(err error) error {
@@ -3767,24 +3786,6 @@ type dbChanInfo struct {
 func insertChannel(ctx context.Context, db SQLQueries,
 	edge *models.ChannelEdgeInfo) (*dbChanInfo, error) {
 
-	chanIDB := channelIDToBytes(edge.ChannelID)
-
-	// Make sure that the channel doesn't already exist. We do this
-	// explicitly instead of relying on catching a unique constraint error
-	// because relying on SQL to throw that error would abort the entire
-	// batch of transactions.
-	_, err := db.GetChannelBySCID(
-		ctx, sqlc.GetChannelBySCIDParams{
-			Scid:    chanIDB,
-			Version: int16(ProtocolV1),
-		},
-	)
-	if err == nil {
-		return nil, ErrEdgeAlreadyExist
-	} else if !errors.Is(err, sql.ErrNoRows) {
-		return nil, fmt.Errorf("unable to fetch channel: %w", err)
-	}
-
 	// Make sure that at least a "shell" entry for each node is present in
 	// the nodes table.
 	node1DBID, err := maybeCreateShellNode(ctx, db, edge.NodeKey1Bytes)
@@ -3804,7 +3805,7 @@ func insertChannel(ctx context.Context, db SQLQueries,
 
 	createParams := sqlc.CreateChannelParams{
 		Version:     int16(ProtocolV1),
-		Scid:        chanIDB,
+		Scid:        channelIDToBytes(edge.ChannelID),
 		NodeID1:     node1DBID,
 		NodeID2:     node2DBID,
 		Outpoint:    edge.ChannelPoint.String(),
@@ -4522,6 +4523,51 @@ func extractChannelPolicies(row any) (*sqlc.GraphChannelPolicy,
 		}
 
 		return policy1, policy2, nil
+
+	case sqlc.GetChannelsByIDsRow:
+		if r.Policy1ID.Valid {
+			policy1 = &sqlc.GraphChannelPolicy{
+				ID:                      r.Policy1ID.Int64,
+				Version:                 r.Policy1Version.Int16,
+				ChannelID:               r.GraphChannel.ID,
+				NodeID:                  r.Policy1NodeID.Int64,
+				Timelock:                r.Policy1Timelock.Int32,
+				FeePpm:                  r.Policy1FeePpm.Int64,
+				BaseFeeMsat:             r.Policy1BaseFeeMsat.Int64,
+				MinHtlcMsat:             r.Policy1MinHtlcMsat.Int64,
+				MaxHtlcMsat:             r.Policy1MaxHtlcMsat,
+				LastUpdate:              r.Policy1LastUpdate,
+				InboundBaseFeeMsat:      r.Policy1InboundBaseFeeMsat,
+				InboundFeeRateMilliMsat: r.Policy1InboundFeeRateMilliMsat,
+				Disabled:                r.Policy1Disabled,
+				MessageFlags:            r.Policy1MessageFlags,
+				ChannelFlags:            r.Policy1ChannelFlags,
+				Signature:               r.Policy1Signature,
+			}
+		}
+		if r.Policy2ID.Valid {
+			policy2 = &sqlc.GraphChannelPolicy{
+				ID:                      r.Policy2ID.Int64,
+				Version:                 r.Policy2Version.Int16,
+				ChannelID:               r.GraphChannel.ID,
+				NodeID:                  r.Policy2NodeID.Int64,
+				Timelock:                r.Policy2Timelock.Int32,
+				FeePpm:                  r.Policy2FeePpm.Int64,
+				BaseFeeMsat:             r.Policy2BaseFeeMsat.Int64,
+				MinHtlcMsat:             r.Policy2MinHtlcMsat.Int64,
+				MaxHtlcMsat:             r.Policy2MaxHtlcMsat,
+				LastUpdate:              r.Policy2LastUpdate,
+				InboundBaseFeeMsat:      r.Policy2InboundBaseFeeMsat,
+				InboundFeeRateMilliMsat: r.Policy2InboundFeeRateMilliMsat,
+				Disabled:                r.Policy2Disabled,
+				MessageFlags:            r.Policy2MessageFlags,
+				ChannelFlags:            r.Policy2ChannelFlags,
+				Signature:               r.Policy2Signature,
+			}
+		}
+
+		return policy1, policy2, nil
+
 	default:
 		return nil, nil, fmt.Errorf("unexpected row type in "+
 			"extractChannelPolicies: %T", r)

sqldb/paginate.go

@@ -37,11 +37,11 @@ const (
 type QueryConfig struct {
 	// MaxBatchSize is the maximum number of items included in a batch
 	// query IN clauses list.
-	MaxBatchSize int `long:"max-batch-size" description:"The maximum number of items to include in a batch query IN clause. This is used for queries that fetch results based on a list of identifiers."`
+	MaxBatchSize uint32 `long:"max-batch-size" description:"The maximum number of items to include in a batch query IN clause. This is used for queries that fetch results based on a list of identifiers."`
 
 	// MaxPageSize is the maximum number of items returned in a single page
 	// of results. This is used for paginated queries.
-	MaxPageSize int32 `long:"max-page-size" description:"The maximum number of items to return in a single page of results. This is used for paginated queries."`
+	MaxPageSize uint32 `long:"max-page-size" description:"The maximum number of items to return in a single page of results. This is used for paginated queries."`
 }
 
 // Validate checks that the QueryConfig values are valid.
@@ -121,9 +121,9 @@ func ExecuteBatchQuery[I any, T any, R any](ctx context.Context,
 	}
 
 	// Process items in pages.
-	for i := 0; i < len(inputItems); i += cfg.MaxBatchSize {
+	for i := 0; i < len(inputItems); i += int(cfg.MaxBatchSize) {
 		// Calculate the end index for this page.
-		end := i + cfg.MaxBatchSize
+		end := i + int(cfg.MaxBatchSize)
 		if end > len(inputItems) {
 			end = len(inputItems)
 		}
@@ -189,7 +189,7 @@ func ExecutePaginatedQuery[C any, T any](ctx context.Context, cfg *QueryConfig,
 
 	for {
 		// Fetch the next page.
-		items, err := queryFunc(ctx, cursor, cfg.MaxPageSize)
+		items, err := queryFunc(ctx, cursor, int32(cfg.MaxPageSize))
 		if err != nil {
 			return fmt.Errorf("failed to fetch page with "+
 				"cursor %v: %w", cursor, err)
@@ -265,7 +265,7 @@ func ExecuteCollectAndBatchWithSharedDataQuery[C any, T any, I any, D any](
 
 	for {
 		// Step 1: Fetch the next page of items.
-		items, err := pageQueryFunc(ctx, cursor, cfg.MaxPageSize)
+		items, err := pageQueryFunc(ctx, cursor, int32(cfg.MaxPageSize))
 		if err != nil {
 			return fmt.Errorf("failed to fetch page with "+
 				"cursor %v: %w", cursor, err)

sqldb/paginate_test.go

@@ -341,7 +341,7 @@ func TestExecutePaginatedQuery(t *testing.T) {
 
 	tests := []struct {
 		name          string
-		pageSize      int32
+		pageSize      uint32
 		allItems      []testItem
 		initialCursor int64
 		queryError    error
@@ -592,7 +592,7 @@ func TestExecuteCollectAndBatchWithSharedDataQuery(t *testing.T) {
 
 	tests := []struct {
 		name                   string
-		maxPageSize            int32
+		maxPageSize            uint32
 		allRows                []channelRow
 		initialCursor          int64
 		pageQueryError         error

sqldb/sqlc/graph.sql.go

@@ -873,6 +873,179 @@ func (q *Queries) GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds
 	return items, nil
 }
 
+const getChannelsByIDs = `-- name: GetChannelsByIDs :many
+SELECT
+    c.id, c.version, c.scid, c.node_id_1, c.node_id_2, c.outpoint, c.capacity, c.bitcoin_key_1, c.bitcoin_key_2, c.node_1_signature, c.node_2_signature, c.bitcoin_1_signature, c.bitcoin_2_signature,
+
+    -- Minimal node data.
+    n1.id AS node1_id,
+    n1.pub_key AS node1_pub_key,
+    n2.id AS node2_id,
+    n2.pub_key AS node2_pub_key,
+
+    -- Policy 1
+    cp1.id AS policy1_id,
+    cp1.node_id AS policy1_node_id,
+    cp1.version AS policy1_version,
+    cp1.timelock AS policy1_timelock,
+    cp1.fee_ppm AS policy1_fee_ppm,
+    cp1.base_fee_msat AS policy1_base_fee_msat,
+    cp1.min_htlc_msat AS policy1_min_htlc_msat,
+    cp1.max_htlc_msat AS policy1_max_htlc_msat,
+    cp1.last_update AS policy1_last_update,
+    cp1.disabled AS policy1_disabled,
+    cp1.inbound_base_fee_msat AS policy1_inbound_base_fee_msat,
+    cp1.inbound_fee_rate_milli_msat AS policy1_inbound_fee_rate_milli_msat,
+    cp1.message_flags AS policy1_message_flags,
+    cp1.channel_flags AS policy1_channel_flags,
+    cp1.signature AS policy1_signature,
+
+    -- Policy 2
+    cp2.id AS policy2_id,
+    cp2.node_id AS policy2_node_id,
+    cp2.version AS policy2_version,
+    cp2.timelock AS policy2_timelock,
+    cp2.fee_ppm AS policy2_fee_ppm,
+    cp2.base_fee_msat AS policy2_base_fee_msat,
+    cp2.min_htlc_msat AS policy2_min_htlc_msat,
+    cp2.max_htlc_msat AS policy2_max_htlc_msat,
+    cp2.last_update AS policy2_last_update,
+    cp2.disabled AS policy2_disabled,
+    cp2.inbound_base_fee_msat AS policy2_inbound_base_fee_msat,
+    cp2.inbound_fee_rate_milli_msat AS policy2_inbound_fee_rate_milli_msat,
+    cp2.message_flags AS policy2_message_flags,
+    cp2.channel_flags AS policy2_channel_flags,
+    cp2.signature AS policy2_signature
+
+FROM graph_channels c
+    JOIN graph_nodes n1 ON c.node_id_1 = n1.id
+    JOIN graph_nodes n2 ON c.node_id_2 = n2.id
+    LEFT JOIN graph_channel_policies cp1
+        ON cp1.channel_id = c.id AND cp1.node_id = c.node_id_1 AND cp1.version = c.version
+    LEFT JOIN graph_channel_policies cp2
+        ON cp2.channel_id = c.id AND cp2.node_id = c.node_id_2 AND cp2.version = c.version
+WHERE c.id IN (/*SLICE:ids*/?)
+`
+
+type GetChannelsByIDsRow struct {
+	GraphChannel                   GraphChannel
+	Node1ID                        int64
+	Node1PubKey                    []byte
+	Node2ID                        int64
+	Node2PubKey                    []byte
+	Policy1ID                      sql.NullInt64
+	Policy1NodeID                  sql.NullInt64
+	Policy1Version                 sql.NullInt16
+	Policy1Timelock                sql.NullInt32
+	Policy1FeePpm                  sql.NullInt64
+	Policy1BaseFeeMsat             sql.NullInt64
+	Policy1MinHtlcMsat             sql.NullInt64
+	Policy1MaxHtlcMsat             sql.NullInt64
+	Policy1LastUpdate              sql.NullInt64
+	Policy1Disabled                sql.NullBool
+	Policy1InboundBaseFeeMsat      sql.NullInt64
+	Policy1InboundFeeRateMilliMsat sql.NullInt64
+	Policy1MessageFlags            sql.NullInt16
+	Policy1ChannelFlags            sql.NullInt16
+	Policy1Signature               []byte
+	Policy2ID                      sql.NullInt64
+	Policy2NodeID                  sql.NullInt64
+	Policy2Version                 sql.NullInt16
+	Policy2Timelock                sql.NullInt32
+	Policy2FeePpm                  sql.NullInt64
+	Policy2BaseFeeMsat             sql.NullInt64
+	Policy2MinHtlcMsat             sql.NullInt64
+	Policy2MaxHtlcMsat             sql.NullInt64
+	Policy2LastUpdate              sql.NullInt64
+	Policy2Disabled                sql.NullBool
+	Policy2InboundBaseFeeMsat      sql.NullInt64
+	Policy2InboundFeeRateMilliMsat sql.NullInt64
+	Policy2MessageFlags            sql.NullInt16
+	Policy2ChannelFlags            sql.NullInt16
+	Policy2Signature               []byte
+}
+
+func (q *Queries) GetChannelsByIDs(ctx context.Context, ids []int64) ([]GetChannelsByIDsRow, error) {
+	query := getChannelsByIDs
+	var queryParams []interface{}
+	if len(ids) > 0 {
+		for _, v := range ids {
+			queryParams = append(queryParams, v)
+		}
+		query = strings.Replace(query, "/*SLICE:ids*/?", makeQueryParams(len(queryParams), len(ids)), 1)
+	} else {
+		query = strings.Replace(query, "/*SLICE:ids*/?", "NULL", 1)
+	}
+	rows, err := q.db.QueryContext(ctx, query, queryParams...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []GetChannelsByIDsRow
+	for rows.Next() {
+		var i GetChannelsByIDsRow
+		if err := rows.Scan(
+			&i.GraphChannel.ID,
+			&i.GraphChannel.Version,
+			&i.GraphChannel.Scid,
+			&i.GraphChannel.NodeID1,
+			&i.GraphChannel.NodeID2,
+			&i.GraphChannel.Outpoint,
+			&i.GraphChannel.Capacity,
+			&i.GraphChannel.BitcoinKey1,
+			&i.GraphChannel.BitcoinKey2,
+			&i.GraphChannel.Node1Signature,
+			&i.GraphChannel.Node2Signature,
+			&i.GraphChannel.Bitcoin1Signature,
+			&i.GraphChannel.Bitcoin2Signature,
+			&i.Node1ID,
+			&i.Node1PubKey,
+			&i.Node2ID,
+			&i.Node2PubKey,
+			&i.Policy1ID,
+			&i.Policy1NodeID,
+			&i.Policy1Version,
+			&i.Policy1Timelock,
+			&i.Policy1FeePpm,
+			&i.Policy1BaseFeeMsat,
+			&i.Policy1MinHtlcMsat,
+			&i.Policy1MaxHtlcMsat,
+			&i.Policy1LastUpdate,
+			&i.Policy1Disabled,
+			&i.Policy1InboundBaseFeeMsat,
+			&i.Policy1InboundFeeRateMilliMsat,
+			&i.Policy1MessageFlags,
+			&i.Policy1ChannelFlags,
+			&i.Policy1Signature,
+			&i.Policy2ID,
+			&i.Policy2NodeID,
+			&i.Policy2Version,
+			&i.Policy2Timelock,
+			&i.Policy2FeePpm,
+			&i.Policy2BaseFeeMsat,
+			&i.Policy2MinHtlcMsat,
+			&i.Policy2MaxHtlcMsat,
+			&i.Policy2LastUpdate,
+			&i.Policy2Disabled,
+			&i.Policy2InboundBaseFeeMsat,
+			&i.Policy2InboundFeeRateMilliMsat,
+			&i.Policy2MessageFlags,
+			&i.Policy2ChannelFlags,
+			&i.Policy2Signature,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getChannelsByOutpoints = `-- name: GetChannelsByOutpoints :many
 SELECT
     c.id, c.version, c.scid, c.node_id_1, c.node_id_2, c.outpoint, c.capacity, c.bitcoin_key_1, c.bitcoin_key_2, c.node_1_signature, c.node_2_signature, c.bitcoin_1_signature, c.bitcoin_2_signature,
@@ -1429,6 +1602,45 @@ func (q *Queries) GetChannelsBySCIDs(ctx context.Context, arg GetChannelsBySCIDs
 	return items, nil
 }
 
+const getClosedChannelsSCIDs = `-- name: GetClosedChannelsSCIDs :many
+SELECT scid
+FROM graph_closed_scids
+WHERE scid IN (/*SLICE:scids*/?)
+`
+
+func (q *Queries) GetClosedChannelsSCIDs(ctx context.Context, scids [][]byte) ([][]byte, error) {
+	query := getClosedChannelsSCIDs
+	var queryParams []interface{}
+	if len(scids) > 0 {
+		for _, v := range scids {
+			queryParams = append(queryParams, v)
+		}
+		query = strings.Replace(query, "/*SLICE:scids*/?", makeQueryParams(len(queryParams), len(scids)), 1)
+	} else {
+		query = strings.Replace(query, "/*SLICE:scids*/?", "NULL", 1)
+	}
+	rows, err := q.db.QueryContext(ctx, query, queryParams...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items [][]byte
+	for rows.Next() {
+		var scid []byte
+		if err := rows.Scan(&scid); err != nil {
+			return nil, err
+		}
+		items = append(items, scid)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getExtraNodeTypes = `-- name: GetExtraNodeTypes :many
 SELECT node_id, type, value
 FROM graph_node_extra_types
@@ -1729,6 +1941,53 @@ func (q *Queries) GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyPa
 	return id, err
 }
 
+const getNodesByIDs = `-- name: GetNodesByIDs :many
+SELECT id, version, pub_key, alias, last_update, color, signature
+FROM graph_nodes
+WHERE id IN (/*SLICE:ids*/?)
+`
+
+func (q *Queries) GetNodesByIDs(ctx context.Context, ids []int64) ([]GraphNode, error) {
+	query := getNodesByIDs
+	var queryParams []interface{}
+	if len(ids) > 0 {
+		for _, v := range ids {
+			queryParams = append(queryParams, v)
+		}
+		query = strings.Replace(query, "/*SLICE:ids*/?", makeQueryParams(len(queryParams), len(ids)), 1)
+	} else {
+		query = strings.Replace(query, "/*SLICE:ids*/?", "NULL", 1)
+	}
+	rows, err := q.db.QueryContext(ctx, query, queryParams...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []GraphNode
+	for rows.Next() {
+		var i GraphNode
+		if err := rows.Scan(
+			&i.ID,
+			&i.Version,
+			&i.PubKey,
+			&i.Alias,
+			&i.LastUpdate,
+			&i.Color,
+			&i.Signature,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getNodesByLastUpdateRange = `-- name: GetNodesByLastUpdateRange :many
 SELECT id, version, pub_key, alias, last_update, color, signature
 FROM graph_nodes
@@ -1772,6 +2031,46 @@ func (q *Queries) GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByL
 	return items, nil
 }
 
+const getPruneEntriesForHeights = `-- name: GetPruneEntriesForHeights :many
+SELECT block_height, block_hash
+FROM graph_prune_log
+WHERE block_height
+   IN (/*SLICE:heights*/?)
+`
+
+func (q *Queries) GetPruneEntriesForHeights(ctx context.Context, heights []int64) ([]GraphPruneLog, error) {
+	query := getPruneEntriesForHeights
+	var queryParams []interface{}
+	if len(heights) > 0 {
+		for _, v := range heights {
+			queryParams = append(queryParams, v)
+		}
+		query = strings.Replace(query, "/*SLICE:heights*/?", makeQueryParams(len(queryParams), len(heights)), 1)
+	} else {
+		query = strings.Replace(query, "/*SLICE:heights*/?", "NULL", 1)
+	}
+	rows, err := q.db.QueryContext(ctx, query, queryParams...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []GraphPruneLog
+	for rows.Next() {
+		var i GraphPruneLog
+		if err := rows.Scan(&i.BlockHeight, &i.BlockHash); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getPruneHashByHeight = `-- name: GetPruneHashByHeight :one
 SELECT block_hash
 FROM graph_prune_log
@@ -1962,6 +2261,57 @@ func (q *Queries) GetZombieChannel(ctx context.Context, arg GetZombieChannelPara
 	return i, err
 }
 
+const getZombieChannelsSCIDs = `-- name: GetZombieChannelsSCIDs :many
+SELECT scid, version, node_key_1, node_key_2
+FROM graph_zombie_channels
+WHERE version = $1
+  AND scid IN (/*SLICE:scids*/?)
+`
+
+type GetZombieChannelsSCIDsParams struct {
+	Version int16
+	Scids   [][]byte
+}
+
+func (q *Queries) GetZombieChannelsSCIDs(ctx context.Context, arg GetZombieChannelsSCIDsParams) ([]GraphZombieChannel, error) {
+	query := getZombieChannelsSCIDs
+	var queryParams []interface{}
+	queryParams = append(queryParams, arg.Version)
+	if len(arg.Scids) > 0 {
+		for _, v := range arg.Scids {
+			queryParams = append(queryParams, v)
+		}
+		query = strings.Replace(query, "/*SLICE:scids*/?", makeQueryParams(len(queryParams), len(arg.Scids)), 1)
+	} else {
+		query = strings.Replace(query, "/*SLICE:scids*/?", "NULL", 1)
+	}
+	rows, err := q.db.QueryContext(ctx, query, queryParams...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []GraphZombieChannel
+	for rows.Next() {
+		var i GraphZombieChannel
+		if err := rows.Scan(
+			&i.Scid,
+			&i.Version,
+			&i.NodeKey1,
+			&i.NodeKey2,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const highestSCID = `-- name: HighestSCID :one
 SELECT scid
 FROM graph_channels

sqldb/sqlc/querier.go

@@ -42,11 +42,13 @@ type Querier interface {
 	GetChannelFeaturesBatch(ctx context.Context, chanIds []int64) ([]GraphChannelFeature, error)
 	GetChannelPolicyByChannelAndNode(ctx context.Context, arg GetChannelPolicyByChannelAndNodeParams) (GraphChannelPolicy, error)
 	GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds []int64) ([]GetChannelPolicyExtraTypesBatchRow, error)
+	GetChannelsByIDs(ctx context.Context, ids []int64) ([]GetChannelsByIDsRow, error)
 	GetChannelsByOutpoints(ctx context.Context, outpoints []string) ([]GetChannelsByOutpointsRow, error)
 	GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg GetChannelsByPolicyLastUpdateRangeParams) ([]GetChannelsByPolicyLastUpdateRangeRow, error)
 	GetChannelsBySCIDRange(ctx context.Context, arg GetChannelsBySCIDRangeParams) ([]GetChannelsBySCIDRangeRow, error)
 	GetChannelsBySCIDWithPolicies(ctx context.Context, arg GetChannelsBySCIDWithPoliciesParams) ([]GetChannelsBySCIDWithPoliciesRow, error)
 	GetChannelsBySCIDs(ctx context.Context, arg GetChannelsBySCIDsParams) ([]GraphChannel, error)
+	GetClosedChannelsSCIDs(ctx context.Context, scids [][]byte) ([][]byte, error)
 	GetDatabaseVersion(ctx context.Context) (int32, error)
 	GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]GraphNodeExtraType, error)
 	// This method may return more than one invoice if filter using multiple fields
@@ -68,7 +70,9 @@ type Querier interface {
 	GetNodeFeaturesBatch(ctx context.Context, ids []int64) ([]GraphNodeFeature, error)
 	GetNodeFeaturesByPubKey(ctx context.Context, arg GetNodeFeaturesByPubKeyParams) ([]int32, error)
 	GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyParams) (int64, error)
+	GetNodesByIDs(ctx context.Context, ids []int64) ([]GraphNode, error)
 	GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]GraphNode, error)
+	GetPruneEntriesForHeights(ctx context.Context, heights []int64) ([]GraphPruneLog, error)
 	GetPruneHashByHeight(ctx context.Context, blockHeight int64) ([]byte, error)
 	GetPruneTip(ctx context.Context) (GraphPruneLog, error)
 	GetPublicV1ChannelsBySCID(ctx context.Context, arg GetPublicV1ChannelsBySCIDParams) ([]GraphChannel, error)
@@ -80,6 +84,7 @@ type Querier interface {
 	// and so the query for V2 may differ.
 	GetV1DisabledSCIDs(ctx context.Context) ([][]byte, error)
 	GetZombieChannel(ctx context.Context, arg GetZombieChannelParams) (GraphZombieChannel, error)
+	GetZombieChannelsSCIDs(ctx context.Context, arg GetZombieChannelsSCIDsParams) ([]GraphZombieChannel, error)
 	HighestSCID(ctx context.Context, version int16) ([]byte, error)
 	InsertAMPSubInvoice(ctx context.Context, arg InsertAMPSubInvoiceParams) error
 	InsertAMPSubInvoiceHTLC(ctx context.Context, arg InsertAMPSubInvoiceHTLCParams) error

sqldb/sqlc/queries/graph.sql

@@ -21,6 +21,11 @@ WHERE graph_nodes.last_update IS NULL
     OR EXCLUDED.last_update > graph_nodes.last_update
 RETURNING id;
 
+-- name: GetNodesByIDs :many
+SELECT *
+FROM graph_nodes
+WHERE id IN (sqlc.slice('ids')/*SLICE:ids*/);
+
 -- name: GetNodeByPubKey :one
 SELECT *
 FROM graph_nodes
@@ -330,6 +335,59 @@ WHERE
     c.version = @version
   AND c.scid IN (sqlc.slice('scids')/*SLICE:scids*/);
 
+-- name: GetChannelsByIDs :many
+SELECT
+    sqlc.embed(c),
+
+    -- Minimal node data.
+    n1.id AS node1_id,
+    n1.pub_key AS node1_pub_key,
+    n2.id AS node2_id,
+    n2.pub_key AS node2_pub_key,
+
+    -- Policy 1
+    cp1.id AS policy1_id,
+    cp1.node_id AS policy1_node_id,
+    cp1.version AS policy1_version,
+    cp1.timelock AS policy1_timelock,
+    cp1.fee_ppm AS policy1_fee_ppm,
+    cp1.base_fee_msat AS policy1_base_fee_msat,
+    cp1.min_htlc_msat AS policy1_min_htlc_msat,
+    cp1.max_htlc_msat AS policy1_max_htlc_msat,
+    cp1.last_update AS policy1_last_update,
+    cp1.disabled AS policy1_disabled,
+    cp1.inbound_base_fee_msat AS policy1_inbound_base_fee_msat,
+    cp1.inbound_fee_rate_milli_msat AS policy1_inbound_fee_rate_milli_msat,
+    cp1.message_flags AS policy1_message_flags,
+    cp1.channel_flags AS policy1_channel_flags,
+    cp1.signature AS policy1_signature,
+
+    -- Policy 2
+    cp2.id AS policy2_id,
+    cp2.node_id AS policy2_node_id,
+    cp2.version AS policy2_version,
+    cp2.timelock AS policy2_timelock,
+    cp2.fee_ppm AS policy2_fee_ppm,
+    cp2.base_fee_msat AS policy2_base_fee_msat,
+    cp2.min_htlc_msat AS policy2_min_htlc_msat,
+    cp2.max_htlc_msat AS policy2_max_htlc_msat,
+    cp2.last_update AS policy2_last_update,
+    cp2.disabled AS policy2_disabled,
+    cp2.inbound_base_fee_msat AS policy2_inbound_base_fee_msat,
+    cp2.inbound_fee_rate_milli_msat AS policy2_inbound_fee_rate_milli_msat,
+    cp2.message_flags AS policy2_message_flags,
+    cp2.channel_flags AS policy2_channel_flags,
+    cp2.signature AS policy2_signature
+
+FROM graph_channels c
+    JOIN graph_nodes n1 ON c.node_id_1 = n1.id
+    JOIN graph_nodes n2 ON c.node_id_2 = n2.id
+    LEFT JOIN graph_channel_policies cp1
+        ON cp1.channel_id = c.id AND cp1.node_id = c.node_id_1 AND cp1.version = c.version
+    LEFT JOIN graph_channel_policies cp2
+        ON cp2.channel_id = c.id AND cp2.node_id = c.node_id_2 AND cp2.version = c.version
+WHERE c.id IN (sqlc.slice('ids')/*SLICE:ids*/);
+
 -- name: GetChannelsByPolicyLastUpdateRange :many
 SELECT
     sqlc.embed(c),
@@ -850,6 +908,12 @@ DO UPDATE SET
     node_key_1 = COALESCE(EXCLUDED.node_key_1, graph_zombie_channels.node_key_1),
     node_key_2 = COALESCE(EXCLUDED.node_key_2, graph_zombie_channels.node_key_2);
 
+-- name: GetZombieChannelsSCIDs :many
+SELECT *
+FROM graph_zombie_channels
+WHERE version = @version
+  AND scid IN (sqlc.slice('scids')/*SLICE:scids*/);
+
 -- name: DeleteZombieChannel :execresult
 DELETE FROM graph_zombie_channels
 WHERE scid = $1
@@ -894,6 +958,12 @@ FROM graph_prune_log
 ORDER BY block_height DESC
 LIMIT 1;
 
+-- name: GetPruneEntriesForHeights :many
+SELECT block_height, block_hash
+FROM graph_prune_log
+WHERE block_height
+   IN (sqlc.slice('heights')/*SLICE:heights*/);
+
 -- name: GetPruneHashByHeight :one
 SELECT block_hash
 FROM graph_prune_log
@@ -920,3 +990,8 @@ SELECT EXISTS (
     FROM graph_closed_scids
     WHERE scid = $1
 );
+
+-- name: GetClosedChannelsSCIDs :many
+SELECT scid
+FROM graph_closed_scids
+WHERE scid IN (sqlc.slice('scids')/*SLICE:scids*/);