From 8ad5f633bc213e52e5e68fe0f104005e5924185a Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 29 Jul 2025 12:46:58 +0200 Subject: [PATCH 1/5] sqldb: add channel data batch queries Also add the calling logic for these queries. This logic is not yet used. --- graph/db/sql_store.go | 163 +++++++++++++++++++++++++++++++++++ sqldb/sqlc/graph.sql.go | 134 ++++++++++++++++++++++++++++ sqldb/sqlc/querier.go | 3 + sqldb/sqlc/queries/graph.sql | 26 ++++++ 4 files changed, 326 insertions(+) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 5c8c12e8e..ca7b8cdcd 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -115,7 +115,9 @@ type SQLQueries interface { DeleteChannels(ctx context.Context, ids []int64) error CreateChannelExtraType(ctx context.Context, arg sqlc.CreateChannelExtraTypeParams) error + GetChannelExtrasBatch(ctx context.Context, chanIds []int64) ([]sqlc.GraphChannelExtraType, error) InsertChannelFeature(ctx context.Context, arg sqlc.InsertChannelFeatureParams) error + GetChannelFeaturesBatch(ctx context.Context, chanIds []int64) ([]sqlc.GraphChannelFeature, error) /* Channel Policy table queries. @@ -126,6 +128,7 @@ type SQLQueries interface { InsertChanPolicyExtraType(ctx context.Context, arg sqlc.InsertChanPolicyExtraTypeParams) error GetChannelPolicyExtraTypes(ctx context.Context, arg sqlc.GetChannelPolicyExtraTypesParams) ([]sqlc.GetChannelPolicyExtraTypesRow, error) + GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds []int64) ([]sqlc.GetChannelPolicyExtraTypesBatchRow, error) DeleteChannelPolicyExtraTypes(ctx context.Context, channelPolicyID int64) error /* @@ -4901,3 +4904,163 @@ func batchLoadNodeExtraTypesHelper(ctx context.Context, callback, ) } + +// batchChannelData holds all the related data for a batch of channels. +type batchChannelData struct { + // chanFeatures is a map from DB channel ID to a slice of feature bits. + chanfeatures map[int64][]int + + // chanExtras is a map from DB channel ID to a map of TLV type to + // extra signed field bytes. + chanExtraTypes map[int64]map[uint64][]byte + + // policyExtras is a map from DB channel policy ID to a map of TLV type + // to extra signed field bytes. + policyExtras map[int64]map[uint64][]byte +} + +// batchLoadChannelData loads all related data for batches of channels and +// policies. +func batchLoadChannelData(ctx context.Context, cfg *sqldb.PagedQueryConfig, + db SQLQueries, channelIDs []int64, + policyIDs []int64) (*batchChannelData, error) { + + batchData := &batchChannelData{ + chanfeatures: make(map[int64][]int), + chanExtraTypes: make(map[int64]map[uint64][]byte), + policyExtras: make(map[int64]map[uint64][]byte), + } + + // Batch load channel features and extras + var err error + if len(channelIDs) > 0 { + batchData.chanfeatures, err = batchLoadChannelFeaturesHelper( + ctx, cfg, db, channelIDs, + ) + if err != nil { + return nil, fmt.Errorf("unable to batch load "+ + "channel features: %w", err) + } + + batchData.chanExtraTypes, err = batchLoadChannelExtrasHelper( + ctx, cfg, db, channelIDs, + ) + if err != nil { + return nil, fmt.Errorf("unable to batch load "+ + "channel extras: %w", err) + } + } + + if len(policyIDs) > 0 { + policyExtras, err := batchLoadChannelPolicyExtrasHelper( + ctx, cfg, db, policyIDs, + ) + if err != nil { + return nil, fmt.Errorf("unable to batch load "+ + "policy extras: %w", err) + } + batchData.policyExtras = policyExtras + } + + return batchData, nil +} + +// batchLoadChannelFeaturesHelper loads channel features for a batch of +// channel IDs using ExecutePagedQuery wrapper around the +// GetChannelFeaturesBatch query. It returns a map from DB channel ID to a +// slice of feature bits. +func batchLoadChannelFeaturesHelper(ctx context.Context, + cfg *sqldb.PagedQueryConfig, db SQLQueries, + channelIDs []int64) (map[int64][]int, error) { + + features := make(map[int64][]int) + + return features, sqldb.ExecutePagedQuery( + ctx, cfg, channelIDs, + func(id int64) int64 { + return id + }, + func(ctx context.Context, + ids []int64) ([]sqlc.GraphChannelFeature, error) { + + return db.GetChannelFeaturesBatch(ctx, ids) + }, + func(ctx context.Context, + feature sqlc.GraphChannelFeature) error { + + features[feature.ChannelID] = append( + features[feature.ChannelID], + int(feature.FeatureBit), + ) + + return nil + }, + ) +} + +// batchLoadChannelExtrasHelper loads channel extra types for a batch of +// channel IDs using ExecutePagedQuery wrapper around the GetChannelExtrasBatch +// query. It returns a map from DB channel ID to a map of TLV type to extra +// signed field bytes. +func batchLoadChannelExtrasHelper(ctx context.Context, + cfg *sqldb.PagedQueryConfig, db SQLQueries, + channelIDs []int64) (map[int64]map[uint64][]byte, error) { + + extras := make(map[int64]map[uint64][]byte) + + cb := func(ctx context.Context, + extra sqlc.GraphChannelExtraType) error { + + if extras[extra.ChannelID] == nil { + extras[extra.ChannelID] = make(map[uint64][]byte) + } + extras[extra.ChannelID][uint64(extra.Type)] = extra.Value + + return nil + } + + return extras, sqldb.ExecutePagedQuery( + ctx, cfg, channelIDs, + func(id int64) int64 { + return id + }, + func(ctx context.Context, + ids []int64) ([]sqlc.GraphChannelExtraType, error) { + + return db.GetChannelExtrasBatch(ctx, ids) + }, cb, + ) +} + +// batchLoadChannelPolicyExtrasHelper loads channel policy extra types for a +// batch of policy IDs using ExecutePagedQuery wrapper around the +// GetChannelPolicyExtraTypesBatch query. It returns a map from DB policy ID to +// a map of TLV type to extra signed field bytes. +func batchLoadChannelPolicyExtrasHelper(ctx context.Context, + cfg *sqldb.PagedQueryConfig, db SQLQueries, + policyIDs []int64) (map[int64]map[uint64][]byte, error) { + + extras := make(map[int64]map[uint64][]byte) + + return extras, sqldb.ExecutePagedQuery( + ctx, cfg, policyIDs, + func(id int64) int64 { + return id + }, + func(ctx context.Context, ids []int64) ( + []sqlc.GetChannelPolicyExtraTypesBatchRow, error) { + + return db.GetChannelPolicyExtraTypesBatch(ctx, ids) + }, + func(ctx context.Context, + row sqlc.GetChannelPolicyExtraTypesBatchRow) error { + + if extras[row.PolicyID] == nil { + extras[row.PolicyID] = make(map[uint64][]byte) + } + extras[row.PolicyID][uint64(row.Type)] = row.Value + + return nil + }, + ) +} diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 3122d1dab..9f847720e 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -701,6 +701,49 @@ func (q *Queries) GetChannelBySCIDWithPolicies(ctx context.Context, arg GetChann return i, err } +const getChannelExtrasBatch = `-- name: GetChannelExtrasBatch :many +SELECT + channel_id, + type, + value +FROM graph_channel_extra_types +WHERE channel_id IN (/*SLICE:chan_ids*/?) +ORDER BY channel_id, type +` + +func (q *Queries) GetChannelExtrasBatch(ctx context.Context, chanIds []int64) ([]GraphChannelExtraType, error) { + query := getChannelExtrasBatch + var queryParams []interface{} + if len(chanIds) > 0 { + for _, v := range chanIds { + queryParams = append(queryParams, v) + } + query = strings.Replace(query, "/*SLICE:chan_ids*/?", makeQueryParams(len(queryParams), len(chanIds)), 1) + } else { + query = strings.Replace(query, "/*SLICE:chan_ids*/?", "NULL", 1) + } + rows, err := q.db.QueryContext(ctx, query, queryParams...) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphChannelExtraType + for rows.Next() { + var i GraphChannelExtraType + if err := rows.Scan(&i.ChannelID, &i.Type, &i.Value); 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 getChannelFeaturesAndExtras = `-- name: GetChannelFeaturesAndExtras :many SELECT cf.channel_id, @@ -760,6 +803,48 @@ func (q *Queries) GetChannelFeaturesAndExtras(ctx context.Context, channelID int return items, nil } +const getChannelFeaturesBatch = `-- name: GetChannelFeaturesBatch :many +SELECT + channel_id, + feature_bit +FROM graph_channel_features +WHERE channel_id IN (/*SLICE:chan_ids*/?) +ORDER BY channel_id, feature_bit +` + +func (q *Queries) GetChannelFeaturesBatch(ctx context.Context, chanIds []int64) ([]GraphChannelFeature, error) { + query := getChannelFeaturesBatch + var queryParams []interface{} + if len(chanIds) > 0 { + for _, v := range chanIds { + queryParams = append(queryParams, v) + } + query = strings.Replace(query, "/*SLICE:chan_ids*/?", makeQueryParams(len(queryParams), len(chanIds)), 1) + } else { + query = strings.Replace(query, "/*SLICE:chan_ids*/?", "NULL", 1) + } + rows, err := q.db.QueryContext(ctx, query, queryParams...) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphChannelFeature + for rows.Next() { + var i GraphChannelFeature + if err := rows.Scan(&i.ChannelID, &i.FeatureBit); 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 getChannelPolicyByChannelAndNode = `-- name: GetChannelPolicyByChannelAndNode :one SELECT id, version, channel_id, node_id, timelock, fee_ppm, base_fee_msat, min_htlc_msat, max_htlc_msat, last_update, disabled, inbound_base_fee_msat, inbound_fee_rate_milli_msat, message_flags, channel_flags, signature FROM graph_channel_policies @@ -853,6 +938,55 @@ func (q *Queries) GetChannelPolicyExtraTypes(ctx context.Context, arg GetChannel return items, nil } +const getChannelPolicyExtraTypesBatch = `-- name: GetChannelPolicyExtraTypesBatch :many +SELECT + channel_policy_id as policy_id, + type, + value +FROM graph_channel_policy_extra_types +WHERE channel_policy_id IN (/*SLICE:policy_ids*/?) +ORDER BY channel_policy_id, type +` + +type GetChannelPolicyExtraTypesBatchRow struct { + PolicyID int64 + Type int64 + Value []byte +} + +func (q *Queries) GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds []int64) ([]GetChannelPolicyExtraTypesBatchRow, error) { + query := getChannelPolicyExtraTypesBatch + var queryParams []interface{} + if len(policyIds) > 0 { + for _, v := range policyIds { + queryParams = append(queryParams, v) + } + query = strings.Replace(query, "/*SLICE:policy_ids*/?", makeQueryParams(len(queryParams), len(policyIds)), 1) + } else { + query = strings.Replace(query, "/*SLICE:policy_ids*/?", "NULL", 1) + } + rows, err := q.db.QueryContext(ctx, query, queryParams...) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetChannelPolicyExtraTypesBatchRow + for rows.Next() { + var i GetChannelPolicyExtraTypesBatchRow + if err := rows.Scan(&i.PolicyID, &i.Type, &i.Value); 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, diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index e32383032..7f2faeb3b 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -38,9 +38,12 @@ type Querier interface { GetChannelByOutpointWithPolicies(ctx context.Context, arg GetChannelByOutpointWithPoliciesParams) (GetChannelByOutpointWithPoliciesRow, error) GetChannelBySCID(ctx context.Context, arg GetChannelBySCIDParams) (GraphChannel, error) GetChannelBySCIDWithPolicies(ctx context.Context, arg GetChannelBySCIDWithPoliciesParams) (GetChannelBySCIDWithPoliciesRow, error) + GetChannelExtrasBatch(ctx context.Context, chanIds []int64) ([]GraphChannelExtraType, error) GetChannelFeaturesAndExtras(ctx context.Context, channelID int64) ([]GetChannelFeaturesAndExtrasRow, error) + GetChannelFeaturesBatch(ctx context.Context, chanIds []int64) ([]GraphChannelFeature, error) GetChannelPolicyByChannelAndNode(ctx context.Context, arg GetChannelPolicyByChannelAndNodeParams) (GraphChannelPolicy, error) GetChannelPolicyExtraTypes(ctx context.Context, arg GetChannelPolicyExtraTypesParams) ([]GetChannelPolicyExtraTypesRow, error) + GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds []int64) ([]GetChannelPolicyExtraTypesBatchRow, 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) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index afa446f0b..cd7dfb467 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -647,6 +647,14 @@ INSERT INTO graph_channel_features ( $1, $2 ); +-- name: GetChannelFeaturesBatch :many +SELECT + channel_id, + feature_bit +FROM graph_channel_features +WHERE channel_id IN (sqlc.slice('chan_ids')/*SLICE:chan_ids*/) +ORDER BY channel_id, feature_bit; + /* ───────────────────────────────────────────── graph_channel_extra_types table queries ───────────────────────────────────────────── @@ -658,6 +666,15 @@ INSERT INTO graph_channel_extra_types ( ) VALUES ($1, $2, $3); +-- name: GetChannelExtrasBatch :many +SELECT + channel_id, + type, + value +FROM graph_channel_extra_types +WHERE channel_id IN (sqlc.slice('chan_ids')/*SLICE:chan_ids*/) +ORDER BY channel_id, type; + /* ───────────────────────────────────────────── graph_channel_policies table queries ───────────────────────────────────────────── @@ -760,6 +777,15 @@ INSERT INTO graph_channel_policy_extra_types ( ) VALUES ($1, $2, $3); +-- name: GetChannelPolicyExtraTypesBatch :many +SELECT + channel_policy_id as policy_id, + type, + value +FROM graph_channel_policy_extra_types +WHERE channel_policy_id IN (sqlc.slice('policy_ids')/*SLICE:policy_ids*/) +ORDER BY channel_policy_id, type; + -- name: GetChannelPolicyExtraTypes :many SELECT cp.id AS policy_id, From f39edae6e34d331bf66be219580f86bfe1be77b2 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 29 Jul 2025 12:59:11 +0200 Subject: [PATCH 2/5] graph/db: use new batch helpers for edge loading Use the new batch helpers to replace the existing logic for loading an edge. --- graph/db/sql_store.go | 166 ++++++++++++++++------------------- sqldb/sqlc/graph.sql.go | 114 ------------------------ sqldb/sqlc/querier.go | 2 - sqldb/sqlc/queries/graph.sql | 33 ------- 4 files changed, 76 insertions(+), 239 deletions(-) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index ca7b8cdcd..329bff0b0 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -102,7 +102,6 @@ type SQLQueries interface { GetChannelBySCIDWithPolicies(ctx context.Context, arg sqlc.GetChannelBySCIDWithPoliciesParams) (sqlc.GetChannelBySCIDWithPoliciesRow, error) GetChannelsBySCIDWithPolicies(ctx context.Context, arg sqlc.GetChannelsBySCIDWithPoliciesParams) ([]sqlc.GetChannelsBySCIDWithPoliciesRow, error) GetChannelAndNodesBySCID(ctx context.Context, arg sqlc.GetChannelAndNodesBySCIDParams) (sqlc.GetChannelAndNodesBySCIDRow, error) - GetChannelFeaturesAndExtras(ctx context.Context, channelID int64) ([]sqlc.GetChannelFeaturesAndExtrasRow, error) HighestSCID(ctx context.Context, version int16) ([]byte, error) ListChannelsByNodeID(ctx context.Context, arg sqlc.ListChannelsByNodeIDParams) ([]sqlc.ListChannelsByNodeIDRow, error) ListChannelsWithPoliciesPaginated(ctx context.Context, arg sqlc.ListChannelsWithPoliciesPaginatedParams) ([]sqlc.ListChannelsWithPoliciesPaginatedRow, error) @@ -127,7 +126,6 @@ type SQLQueries interface { GetV1DisabledSCIDs(ctx context.Context) ([][]byte, error) InsertChanPolicyExtraType(ctx context.Context, arg sqlc.InsertChanPolicyExtraTypeParams) error - GetChannelPolicyExtraTypes(ctx context.Context, arg sqlc.GetChannelPolicyExtraTypesParams) ([]sqlc.GetChannelPolicyExtraTypesRow, error) GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds []int64) ([]sqlc.GetChannelPolicyExtraTypesBatchRow, error) DeleteChannelPolicyExtraTypes(ctx context.Context, channelPolicyID int64) error @@ -4091,16 +4089,44 @@ func getAndBuildEdgeInfo(ctx context.Context, db SQLQueries, chain chainhash.Hash, dbChanID int64, dbChan sqlc.GraphChannel, node1, node2 route.Vertex) (*models.ChannelEdgeInfo, error) { + // NOTE: getAndBuildEdgeInfo is only used to load the data for a single + // edge, and so no paged queries will be performed. This means that + // it's ok to used pass in default config values here. + cfg := sqldb.DefaultPagedQueryConfig() + + data, err := batchLoadChannelData(ctx, cfg, db, []int64{dbChan.ID}, nil) + if err != nil { + return nil, fmt.Errorf("unable to batch load channel data: %w", + err) + } + + return buildEdgeInfoWithBatchData(chain, dbChan, node1, node2, data) +} + +// buildEdgeInfoWithBatchData builds edge info using pre-loaded batch data. +func buildEdgeInfoWithBatchData(chain chainhash.Hash, + dbChan sqlc.GraphChannel, node1, node2 route.Vertex, + batchData *batchChannelData) (*models.ChannelEdgeInfo, error) { + if dbChan.Version != int16(ProtocolV1) { return nil, fmt.Errorf("unsupported channel version: %d", dbChan.Version) } - fv, extras, err := getChanFeaturesAndExtras( - ctx, db, dbChanID, - ) - if err != nil { - return nil, err + // Use pre-loaded features and extras types. + fv := lnwire.EmptyFeatureVector() + if features, exists := batchData.chanfeatures[dbChan.ID]; exists { + for _, bit := range features { + fv.Set(lnwire.FeatureBit(bit)) + } + } + + var extras map[uint64][]byte + channelExtras, exists := batchData.chanExtraTypes[dbChan.ID] + if exists { + extras = channelExtras + } else { + extras = make(map[uint64][]byte) } op, err := wire.NewOutPointFromString(dbChan.Outpoint) @@ -4169,46 +4195,6 @@ func buildNodeVertices(node1Pub, node2Pub []byte) (route.Vertex, return node1Vertex, node2Vertex, nil } -// getChanFeaturesAndExtras fetches the channel features and extra TLV types -// for a channel with the given ID. -func getChanFeaturesAndExtras(ctx context.Context, db SQLQueries, - id int64) (*lnwire.FeatureVector, map[uint64][]byte, error) { - - rows, err := db.GetChannelFeaturesAndExtras(ctx, id) - if err != nil { - return nil, nil, fmt.Errorf("unable to fetch channel "+ - "features and extras: %w", err) - } - - var ( - fv = lnwire.EmptyFeatureVector() - extras = make(map[uint64][]byte) - ) - for _, row := range rows { - if row.IsFeature { - fv.Set(lnwire.FeatureBit(row.FeatureBit)) - - continue - } - - tlvType, ok := row.ExtraKey.(int64) - if !ok { - return nil, nil, fmt.Errorf("unexpected type for "+ - "TLV type: %T", row.ExtraKey) - } - - valueBytes, ok := row.Value.([]byte) - if !ok { - return nil, nil, fmt.Errorf("unexpected type for "+ - "Value: %T", row.Value) - } - - extras[uint64(tlvType)] = valueBytes - } - - return fv, extras, nil -} - // getAndBuildChanPolicies uses the given sqlc.GraphChannelPolicy and also // retrieves all the extra info required to build the complete // models.ChannelEdgePolicy types. It returns two policies, which may be nil if @@ -4222,58 +4208,38 @@ func getAndBuildChanPolicies(ctx context.Context, db SQLQueries, return nil, nil, nil } - var ( - policy1ID int64 - policy2ID int64 - ) + var policyIDs = make([]int64, 0, 2) if dbPol1 != nil { - policy1ID = dbPol1.ID + policyIDs = append(policyIDs, dbPol1.ID) } if dbPol2 != nil { - policy2ID = dbPol2.ID + policyIDs = append(policyIDs, dbPol2.ID) } - rows, err := db.GetChannelPolicyExtraTypes( - ctx, sqlc.GetChannelPolicyExtraTypesParams{ - ID: policy1ID, - ID_2: policy2ID, - }, + + // NOTE: getAndBuildChanPolicies is only used to load the data for + // a maximum of two policies, and so no paged queries will be + // performed (unless the page size is one). So it's ok to use + // the default config values here. + cfg := sqldb.DefaultPagedQueryConfig() + + batchData, err := batchLoadChannelData(ctx, cfg, db, nil, policyIDs) + if err != nil { + return nil, nil, fmt.Errorf("unable to batch load channel "+ + "data: %w", err) + } + + pol1, err := buildChanPolicyWithBatchData( + dbPol1, channelID, node2, batchData, ) if err != nil { - return nil, nil, err + return nil, nil, fmt.Errorf("unable to build policy1: %w", err) } - var ( - dbPol1Extras = make(map[uint64][]byte) - dbPol2Extras = make(map[uint64][]byte) + pol2, err := buildChanPolicyWithBatchData( + dbPol2, channelID, node1, batchData, ) - for _, row := range rows { - switch row.PolicyID { - case policy1ID: - dbPol1Extras[uint64(row.Type)] = row.Value - case policy2ID: - dbPol2Extras[uint64(row.Type)] = row.Value - default: - return nil, nil, fmt.Errorf("unexpected policy ID %d "+ - "in row: %v", row.PolicyID, row) - } - } - - var pol1, pol2 *models.ChannelEdgePolicy - if dbPol1 != nil { - pol1, err = buildChanPolicy( - *dbPol1, channelID, dbPol1Extras, node2, - ) - if err != nil { - return nil, nil, err - } - } - if dbPol2 != nil { - pol2, err = buildChanPolicy( - *dbPol2, channelID, dbPol2Extras, node1, - ) - if err != nil { - return nil, nil, err - } + if err != nil { + return nil, nil, fmt.Errorf("unable to build policy2: %w", err) } return pol1, pol2, nil @@ -4905,6 +4871,26 @@ func batchLoadNodeExtraTypesHelper(ctx context.Context, ) } +// buildChanPolicyWithBatchData builds a models.ChannelEdgePolicy instance from +// the provided sqlc.GraphChannelPolicy and the provided batchChannelData. +func buildChanPolicyWithBatchData(dbPol *sqlc.GraphChannelPolicy, + channelID uint64, toNode route.Vertex, + batchData *batchChannelData) (*models.ChannelEdgePolicy, error) { + + if dbPol == nil { + return nil, nil + } + + var dbPol1Extras map[uint64][]byte + if extras, exists := batchData.policyExtras[dbPol.ID]; exists { + dbPol1Extras = extras + } else { + dbPol1Extras = make(map[uint64][]byte) + } + + return buildChanPolicy(*dbPol, channelID, dbPol1Extras, toNode) +} + // batchChannelData holds all the related data for a batch of channels. type batchChannelData struct { // chanFeatures is a map from DB channel ID to a slice of feature bits. diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 9f847720e..fa6d50803 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -744,65 +744,6 @@ func (q *Queries) GetChannelExtrasBatch(ctx context.Context, chanIds []int64) ([ return items, nil } -const getChannelFeaturesAndExtras = `-- name: GetChannelFeaturesAndExtras :many -SELECT - cf.channel_id, - true AS is_feature, - cf.feature_bit AS feature_bit, - NULL AS extra_key, - NULL AS value -FROM graph_channel_features cf -WHERE cf.channel_id = $1 - -UNION ALL - -SELECT - cet.channel_id, - false AS is_feature, - 0 AS feature_bit, - cet.type AS extra_key, - cet.value AS value -FROM graph_channel_extra_types cet -WHERE cet.channel_id = $1 -` - -type GetChannelFeaturesAndExtrasRow struct { - ChannelID int64 - IsFeature bool - FeatureBit int32 - ExtraKey interface{} - Value interface{} -} - -func (q *Queries) GetChannelFeaturesAndExtras(ctx context.Context, channelID int64) ([]GetChannelFeaturesAndExtrasRow, error) { - rows, err := q.db.QueryContext(ctx, getChannelFeaturesAndExtras, channelID) - if err != nil { - return nil, err - } - defer rows.Close() - var items []GetChannelFeaturesAndExtrasRow - for rows.Next() { - var i GetChannelFeaturesAndExtrasRow - if err := rows.Scan( - &i.ChannelID, - &i.IsFeature, - &i.FeatureBit, - &i.ExtraKey, - &i.Value, - ); 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 getChannelFeaturesBatch = `-- name: GetChannelFeaturesBatch :many SELECT channel_id, @@ -883,61 +824,6 @@ func (q *Queries) GetChannelPolicyByChannelAndNode(ctx context.Context, arg GetC return i, err } -const getChannelPolicyExtraTypes = `-- name: GetChannelPolicyExtraTypes :many -SELECT - cp.id AS policy_id, - cp.channel_id, - cp.node_id, - cpet.type, - cpet.value -FROM graph_channel_policies cp -JOIN graph_channel_policy_extra_types cpet -ON cp.id = cpet.channel_policy_id -WHERE cp.id = $1 OR cp.id = $2 -` - -type GetChannelPolicyExtraTypesParams struct { - ID int64 - ID_2 int64 -} - -type GetChannelPolicyExtraTypesRow struct { - PolicyID int64 - ChannelID int64 - NodeID int64 - Type int64 - Value []byte -} - -func (q *Queries) GetChannelPolicyExtraTypes(ctx context.Context, arg GetChannelPolicyExtraTypesParams) ([]GetChannelPolicyExtraTypesRow, error) { - rows, err := q.db.QueryContext(ctx, getChannelPolicyExtraTypes, arg.ID, arg.ID_2) - if err != nil { - return nil, err - } - defer rows.Close() - var items []GetChannelPolicyExtraTypesRow - for rows.Next() { - var i GetChannelPolicyExtraTypesRow - if err := rows.Scan( - &i.PolicyID, - &i.ChannelID, - &i.NodeID, - &i.Type, - &i.Value, - ); 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 getChannelPolicyExtraTypesBatch = `-- name: GetChannelPolicyExtraTypesBatch :many SELECT channel_policy_id as policy_id, diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 7f2faeb3b..f7fad44ab 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -39,10 +39,8 @@ type Querier interface { GetChannelBySCID(ctx context.Context, arg GetChannelBySCIDParams) (GraphChannel, error) GetChannelBySCIDWithPolicies(ctx context.Context, arg GetChannelBySCIDWithPoliciesParams) (GetChannelBySCIDWithPoliciesRow, error) GetChannelExtrasBatch(ctx context.Context, chanIds []int64) ([]GraphChannelExtraType, error) - GetChannelFeaturesAndExtras(ctx context.Context, channelID int64) ([]GetChannelFeaturesAndExtrasRow, error) GetChannelFeaturesBatch(ctx context.Context, chanIds []int64) ([]GraphChannelFeature, error) GetChannelPolicyByChannelAndNode(ctx context.Context, arg GetChannelPolicyByChannelAndNodeParams) (GraphChannelPolicy, error) - GetChannelPolicyExtraTypes(ctx context.Context, arg GetChannelPolicyExtraTypesParams) ([]GetChannelPolicyExtraTypesRow, error) GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds []int64) ([]GetChannelPolicyExtraTypesBatchRow, error) GetChannelsByOutpoints(ctx context.Context, outpoints []string) ([]GetChannelsByOutpointsRow, error) GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg GetChannelsByPolicyLastUpdateRangeParams) ([]GetChannelsByPolicyLastUpdateRangeRow, error) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index cd7dfb467..ae4979413 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -275,27 +275,6 @@ FROM graph_channels c WHERE c.scid = $1 AND c.version = $2; --- name: GetChannelFeaturesAndExtras :many -SELECT - cf.channel_id, - true AS is_feature, - cf.feature_bit AS feature_bit, - NULL AS extra_key, - NULL AS value -FROM graph_channel_features cf -WHERE cf.channel_id = $1 - -UNION ALL - -SELECT - cet.channel_id, - false AS is_feature, - 0 AS feature_bit, - cet.type AS extra_key, - cet.value AS value -FROM graph_channel_extra_types cet -WHERE cet.channel_id = $1; - -- name: GetSCIDByOutpoint :one SELECT scid from graph_channels WHERE outpoint = $1 AND version = $2; @@ -786,18 +765,6 @@ FROM graph_channel_policy_extra_types WHERE channel_policy_id IN (sqlc.slice('policy_ids')/*SLICE:policy_ids*/) ORDER BY channel_policy_id, type; --- name: GetChannelPolicyExtraTypes :many -SELECT - cp.id AS policy_id, - cp.channel_id, - cp.node_id, - cpet.type, - cpet.value -FROM graph_channel_policies cp -JOIN graph_channel_policy_extra_types cpet -ON cp.id = cpet.channel_policy_id -WHERE cp.id = $1 OR cp.id = $2; - -- name: GetV1DisabledSCIDs :many SELECT c.scid FROM graph_channels c From d5da637a38747e012c779a64835e9c9f5c68c964 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 29 Jul 2025 13:20:09 +0200 Subject: [PATCH 3/5] graph/db: remove unused param --- graph/db/sql_migration.go | 3 +-- graph/db/sql_store.go | 42 ++++++++++++++++++--------------------- 2 files changed, 20 insertions(+), 25 deletions(-) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 2f1792b1f..e1ff256a2 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -713,8 +713,7 @@ func getAndBuildChanAndPolicies(ctx context.Context, db SQLQueries, } edge, err := getAndBuildEdgeInfo( - ctx, db, chain, row.GraphChannel.ID, row.GraphChannel, node1, - node2, + ctx, db, chain, row.GraphChannel, node1, node2, ) if err != nil { return nil, nil, nil, fmt.Errorf("unable to build channel "+ diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 329bff0b0..9a3ae4fd0 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1054,9 +1054,8 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, } channel, err := getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1.PubKeyBytes, - node2.PubKeyBytes, + ctx, db, s.cfg.ChainHash, row.GraphChannel, + node1.PubKeyBytes, node2.PubKeyBytes, ) if err != nil { return fmt.Errorf("unable to build channel "+ @@ -1162,8 +1161,7 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, e, err := getAndBuildEdgeInfo( ctx, db, s.cfg.ChainHash, - row.GraphChannel.ID, row.GraphChannel, - node1, node2, + row.GraphChannel, node1, node2, ) if err != nil { return fmt.Errorf("unable to build "+ @@ -1368,8 +1366,8 @@ func (s *SQLStore) ForEachChannel(ctx context.Context, } edge, err := getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1, node2, + ctx, db, s.cfg.ChainHash, row.GraphChannel, node1, + node2, ) if err != nil { return fmt.Errorf("unable to build channel info: %w", @@ -1757,8 +1755,8 @@ func (s *SQLStore) DeleteChannelEdges(strictZombiePruning, markZombie bool, } info, err := getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1, node2, + ctx, db, s.cfg.ChainHash, row.GraphChannel, + node1, node2, ) if err != nil { return err @@ -1910,8 +1908,8 @@ func (s *SQLStore) FetchChannelEdgesByID(chanID uint64) ( } edge, err = getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1, node2, + ctx, db, s.cfg.ChainHash, row.GraphChannel, node1, + node2, ) if err != nil { return fmt.Errorf("unable to build channel info: %w", @@ -1982,8 +1980,8 @@ func (s *SQLStore) FetchChannelEdgesByOutpoint(op *wire.OutPoint) ( } edge, err = getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1, node2, + ctx, db, s.cfg.ChainHash, row.GraphChannel, node1, + node2, ) if err != nil { return fmt.Errorf("unable to build channel info: %w", @@ -2216,9 +2214,8 @@ func (s *SQLStore) FetchChanInfos(chanIDs []uint64) ([]ChannelEdge, error) { } edge, err := getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1.PubKeyBytes, - node2.PubKeyBytes, + ctx, db, s.cfg.ChainHash, row.GraphChannel, + node1.PubKeyBytes, node2.PubKeyBytes, ) if err != nil { return fmt.Errorf("unable to build "+ @@ -2489,8 +2486,8 @@ func (s *SQLStore) PruneGraph(spentOutputs []*wire.OutPoint, } info, err := getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1, node2, + ctx, db, s.cfg.ChainHash, row.GraphChannel, + node1, node2, ) if err != nil { return err @@ -2788,8 +2785,8 @@ func (s *SQLStore) DisconnectBlockAtHeight(height uint32) ( } channel, err := getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1, node2, + ctx, db, s.cfg.ChainHash, row.GraphChannel, + node1, node2, ) if err != nil { return err @@ -3175,8 +3172,7 @@ func forEachNodeChannel(ctx context.Context, db SQLQueries, } edge, err := getAndBuildEdgeInfo( - ctx, db, chain, row.GraphChannel.ID, row.GraphChannel, - node1, node2, + ctx, db, chain, row.GraphChannel, node1, node2, ) if err != nil { return fmt.Errorf("unable to build channel info: %w", @@ -4086,7 +4082,7 @@ func upsertChanPolicyExtraSignedFields(ctx context.Context, db SQLQueries, // provided dbChanRow and also fetches any other required information // to construct the edge info. func getAndBuildEdgeInfo(ctx context.Context, db SQLQueries, - chain chainhash.Hash, dbChanID int64, dbChan sqlc.GraphChannel, node1, + chain chainhash.Hash, dbChan sqlc.GraphChannel, node1, node2 route.Vertex) (*models.ChannelEdgeInfo, error) { // NOTE: getAndBuildEdgeInfo is only used to load the data for a single From a6b077064f849475f8fcc9fa057cd88bbfdcfe0f Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 29 Jul 2025 13:24:53 +0200 Subject: [PATCH 4/5] graph/db: use batch loading for ForEachChannel --- graph/db/sql_store.go | 75 +++++++++++++++++++++++++++++++++++++++---- 1 file changed, 68 insertions(+), 7 deletions(-) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 9a3ae4fd0..d7f93e65d 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1354,7 +1354,7 @@ func (s *SQLStore) ForEachChannel(ctx context.Context, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, reset func()) error { - handleChannel := func(db SQLQueries, + handleChannel := func(db SQLQueries, batchData *batchChannelData, row sqlc.ListChannelsWithPoliciesPaginatedRow) error { node1, node2, err := buildNodeVertices( @@ -1365,9 +1365,9 @@ func (s *SQLStore) ForEachChannel(ctx context.Context, err) } - edge, err := getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel, node1, - node2, + edge, err := buildEdgeInfoWithBatchData( + s.cfg.ChainHash, row.GraphChannel, node1, node2, + batchData, ) if err != nil { return fmt.Errorf("unable to build channel info: %w", @@ -1380,8 +1380,8 @@ func (s *SQLStore) ForEachChannel(ctx context.Context, "policies: %w", err) } - p1, p2, err := getAndBuildChanPolicies( - ctx, db, dbPol1, dbPol2, edge.ChannelID, node1, node2, + p1, p2, err := buildChanPoliciesWithBatchData( + dbPol1, dbPol2, edge.ChannelID, node1, node2, batchData, ) if err != nil { return fmt.Errorf("unable to build channel "+ @@ -1416,8 +1416,44 @@ func (s *SQLStore) ForEachChannel(ctx context.Context, break } + // Collect the channel & policy IDs that we want to + // do a batch collection for. + var ( + channelIDs = make([]int64, len(rows)) + policyIDs = make([]int64, 0, len(rows)*2) + ) + for i, row := range rows { + channelIDs[i] = row.GraphChannel.ID + + // Extract policy IDs from the row + dbPol1, dbPol2, err := extractChannelPolicies( + row, + ) + if err != nil { + return fmt.Errorf("unable to extract "+ + "channel policies: %w", err) + } + + if dbPol1 != nil { + policyIDs = append(policyIDs, dbPol1.ID) + } + + if dbPol2 != nil { + policyIDs = append(policyIDs, dbPol2.ID) + } + } + + batchData, err := batchLoadChannelData( + ctx, s.cfg.PaginationCfg, db, channelIDs, + policyIDs, + ) + if err != nil { + return fmt.Errorf("unable to batch load "+ + "channel data: %w", err) + } + for _, row := range rows { - err := handleChannel(db, row) + err := handleChannel(db, batchData, row) if err != nil { return err } @@ -4867,6 +4903,31 @@ func batchLoadNodeExtraTypesHelper(ctx context.Context, ) } +// buildChanPoliciesWithBatchData builds two models.ChannelEdgePolicy instances +// from the provided sqlc.GraphChannelPolicy records and the +// provided batchChannelData. +func buildChanPoliciesWithBatchData(dbPol1, dbPol2 *sqlc.GraphChannelPolicy, + channelID uint64, node1, node2 route.Vertex, + batchData *batchChannelData) (*models.ChannelEdgePolicy, + *models.ChannelEdgePolicy, error) { + + pol1, err := buildChanPolicyWithBatchData( + dbPol1, channelID, node2, batchData, + ) + if err != nil { + return nil, nil, fmt.Errorf("unable to build policy1: %w", err) + } + + pol2, err := buildChanPolicyWithBatchData( + dbPol2, channelID, node1, batchData, + ) + if err != nil { + return nil, nil, fmt.Errorf("unable to build policy2: %w", err) + } + + return pol1, pol2, nil +} + // buildChanPolicyWithBatchData builds a models.ChannelEdgePolicy instance from // the provided sqlc.GraphChannelPolicy and the provided batchChannelData. func buildChanPolicyWithBatchData(dbPol *sqlc.GraphChannelPolicy, From c1ba7a97a86725019793741e3864f4f7325074ec Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 29 Jul 2025 13:29:40 +0200 Subject: [PATCH 5/5] graph/db: increase default page size --- graph/db/sql_store.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index d7f93e65d..f996bfbfa 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -33,7 +33,10 @@ import ( // pageSize is the limit for the number of records that can be returned // in a paginated query. This can be tuned after some benchmarks. -const pageSize = 2000 +// +// TODO(elle): make this configurable & have different defaults for SQLite and +// Postgres. +const pageSize = 10000 // ProtocolVersion is an enum that defines the gossip protocol version of a // message.