From 0dc0d320f8bfb1eb0f062e2b407e75b64c826f85 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 29 Jul 2025 07:45:49 +0200 Subject: [PATCH 1/6] graph/db+sqldb: queries and logic for batch fetching node data In this commit, we add the queries that will be needed to batch-fetch the data of a set of nodes. The logic for using these new queries is also added but not used yet. --- graph/db/sql_store.go | 155 +++++++++++++++++++++++++++++++++++ sqldb/sqlc/graph.sql.go | 125 ++++++++++++++++++++++++++++ sqldb/sqlc/querier.go | 3 + sqldb/sqlc/queries/graph.sql | 18 ++++ 4 files changed, 301 insertions(+) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index cdbeeab89..cc7e5d4e9 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -69,15 +69,18 @@ type SQLQueries interface { DeleteNode(ctx context.Context, id int64) error GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]sqlc.GraphNodeExtraType, error) + GetNodeExtraTypesBatch(ctx context.Context, ids []int64) ([]sqlc.GraphNodeExtraType, error) UpsertNodeExtraType(ctx context.Context, arg sqlc.UpsertNodeExtraTypeParams) error DeleteExtraNodeType(ctx context.Context, arg sqlc.DeleteExtraNodeTypeParams) error InsertNodeAddress(ctx context.Context, arg sqlc.InsertNodeAddressParams) error GetNodeAddresses(ctx context.Context, nodeID int64) ([]sqlc.GetNodeAddressesRow, error) + GetNodeAddressesBatch(ctx context.Context, ids []int64) ([]sqlc.GraphNodeAddress, error) DeleteNodeAddresses(ctx context.Context, nodeID int64) error InsertNodeFeature(ctx context.Context, arg sqlc.InsertNodeFeatureParams) error GetNodeFeatures(ctx context.Context, nodeID int64) ([]sqlc.GraphNodeFeature, error) + GetNodeFeaturesBatch(ctx context.Context, ids []int64) ([]sqlc.GraphNodeFeature, error) GetNodeFeaturesByPubKey(ctx context.Context, arg sqlc.GetNodeFeaturesByPubKeyParams) ([]int32, error) DeleteNodeFeature(ctx context.Context, arg sqlc.DeleteNodeFeatureParams) error @@ -4672,3 +4675,155 @@ func channelIDToBytes(channelID uint64) []byte { return chanIDB[:] } + +// batchNodeData holds all the related data for a batch of nodes. +type batchNodeData struct { + // features is a map from a DB node ID to the feature bits for that + // node. + features map[int64][]int + + // addresses is a map from a DB node ID to the node's addresses. + addresses map[int64][]nodeAddress + + // extraFields is a map from a DB node ID to the extra signed fields + // for that node. + extraFields map[int64]map[uint64][]byte +} + +// nodeAddress holds the address type, position and address string for a +// node. This is used to batch the fetching of node addresses. +type nodeAddress struct { + addrType dbAddressType + position int32 + address string +} + +// batchLoadNodeData loads all related data for a batch of node IDs using the +// provided SQLQueries interface. It returns a batchNodeData instance containing +// the node features, addresses and extra signed fields. +func batchLoadNodeData(ctx context.Context, cfg *sqldb.PagedQueryConfig, + db SQLQueries, nodeIDs []int64) (*batchNodeData, error) { + + // Batch load the node features. + features, err := batchLoadNodeFeaturesHelper(ctx, cfg, db, nodeIDs) + if err != nil { + return nil, fmt.Errorf("unable to batch load node "+ + "features: %w", err) + } + + // Batch load the node addresses. + addrs, err := batchLoadNodeAddressesHelper(ctx, cfg, db, nodeIDs) + if err != nil { + return nil, fmt.Errorf("unable to batch load node "+ + "addresses: %w", err) + } + + // Batch load the node extra signed fields. + extraTypes, err := batchLoadNodeExtraTypesHelper(ctx, cfg, db, nodeIDs) + if err != nil { + return nil, fmt.Errorf("unable to batch load node extra "+ + "signed fields: %w", err) + } + + return &batchNodeData{ + features: features, + addresses: addrs, + extraFields: extraTypes, + }, nil +} + +// batchLoadNodeFeaturesHelper loads node features for a batch of node IDs +// using ExecutePagedQuery wrapper around the GetNodeFeaturesBatch query. +func batchLoadNodeFeaturesHelper(ctx context.Context, + cfg *sqldb.PagedQueryConfig, db SQLQueries, + nodeIDs []int64) (map[int64][]int, error) { + + features := make(map[int64][]int) + + return features, sqldb.ExecutePagedQuery( + ctx, cfg, nodeIDs, + func(id int64) int64 { + return id + }, + func(ctx context.Context, ids []int64) ([]sqlc.GraphNodeFeature, + error) { + + return db.GetNodeFeaturesBatch(ctx, ids) + }, + func(ctx context.Context, feature sqlc.GraphNodeFeature) error { + features[feature.NodeID] = append( + features[feature.NodeID], + int(feature.FeatureBit), + ) + + return nil + }, + ) +} + +// batchLoadNodeAddressesHelper loads node addresses using ExecutePagedQuery +// wrapper around the GetNodeAddressesBatch query. It returns a map from +// node ID to a slice of nodeAddress structs. +func batchLoadNodeAddressesHelper(ctx context.Context, + cfg *sqldb.PagedQueryConfig, db SQLQueries, + nodeIDs []int64) (map[int64][]nodeAddress, error) { + + addrs := make(map[int64][]nodeAddress) + + return addrs, sqldb.ExecutePagedQuery( + ctx, cfg, nodeIDs, + func(id int64) int64 { + return id + }, + func(ctx context.Context, ids []int64) ([]sqlc.GraphNodeAddress, + error) { + + return db.GetNodeAddressesBatch(ctx, ids) + }, + func(ctx context.Context, addr sqlc.GraphNodeAddress) error { + addrs[addr.NodeID] = append( + addrs[addr.NodeID], nodeAddress{ + addrType: dbAddressType(addr.Type), + position: addr.Position, + address: addr.Address, + }, + ) + + return nil + }, + ) +} + +// batchLoadNodeExtraTypesHelper loads node extra type bytes for a batch of +// node IDs using ExecutePagedQuery wrapper around the GetNodeExtraTypesBatch +// query. +func batchLoadNodeExtraTypesHelper(ctx context.Context, + cfg *sqldb.PagedQueryConfig, db SQLQueries, + nodeIDs []int64) (map[int64]map[uint64][]byte, error) { + + extraFields := make(map[int64]map[uint64][]byte) + + callback := func(ctx context.Context, + field sqlc.GraphNodeExtraType) error { + + if extraFields[field.NodeID] == nil { + extraFields[field.NodeID] = make(map[uint64][]byte) + } + extraFields[field.NodeID][uint64(field.Type)] = field.Value + + return nil + } + + return extraFields, sqldb.ExecutePagedQuery( + ctx, cfg, nodeIDs, + func(id int64) int64 { + return id + }, + func(ctx context.Context, ids []int64) ( + []sqlc.GraphNodeExtraType, error) { + + return db.GetNodeExtraTypesBatch(ctx, ids) + }, + callback, + ) +} diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 4dbdb9aa8..3122d1dab 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -1473,6 +1473,51 @@ func (q *Queries) GetNodeAddresses(ctx context.Context, nodeID int64) ([]GetNode return items, nil } +const getNodeAddressesBatch = `-- name: GetNodeAddressesBatch :many +SELECT node_id, type, position, address +FROM graph_node_addresses +WHERE node_id IN (/*SLICE:ids*/?) +ORDER BY node_id, type, position +` + +func (q *Queries) GetNodeAddressesBatch(ctx context.Context, ids []int64) ([]GraphNodeAddress, error) { + query := getNodeAddressesBatch + 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 []GraphNodeAddress + for rows.Next() { + var i GraphNodeAddress + if err := rows.Scan( + &i.NodeID, + &i.Type, + &i.Position, + &i.Address, + ); 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 getNodeByPubKey = `-- name: GetNodeByPubKey :one SELECT id, version, pub_key, alias, last_update, color, signature FROM graph_nodes @@ -1500,6 +1545,46 @@ func (q *Queries) GetNodeByPubKey(ctx context.Context, arg GetNodeByPubKeyParams return i, err } +const getNodeExtraTypesBatch = `-- name: GetNodeExtraTypesBatch :many +SELECT node_id, type, value +FROM graph_node_extra_types +WHERE node_id IN (/*SLICE:ids*/?) +ORDER BY node_id, type +` + +func (q *Queries) GetNodeExtraTypesBatch(ctx context.Context, ids []int64) ([]GraphNodeExtraType, error) { + query := getNodeExtraTypesBatch + 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 []GraphNodeExtraType + for rows.Next() { + var i GraphNodeExtraType + if err := rows.Scan(&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 getNodeFeatures = `-- name: GetNodeFeatures :many SELECT node_id, feature_bit FROM graph_node_features @@ -1529,6 +1614,46 @@ func (q *Queries) GetNodeFeatures(ctx context.Context, nodeID int64) ([]GraphNod return items, nil } +const getNodeFeaturesBatch = `-- name: GetNodeFeaturesBatch :many +SELECT node_id, feature_bit +FROM graph_node_features +WHERE node_id IN (/*SLICE:ids*/?) +ORDER BY node_id, feature_bit +` + +func (q *Queries) GetNodeFeaturesBatch(ctx context.Context, ids []int64) ([]GraphNodeFeature, error) { + query := getNodeFeaturesBatch + 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 []GraphNodeFeature + for rows.Next() { + var i GraphNodeFeature + if err := rows.Scan(&i.NodeID, &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 getNodeFeaturesByPubKey = `-- name: GetNodeFeaturesByPubKey :many SELECT f.feature_bit FROM graph_nodes n diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 84b9f6cac..e32383032 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -60,8 +60,11 @@ type Querier interface { GetKVInvoicePaymentHashByAddIndex(ctx context.Context, addIndex int64) ([]byte, error) GetMigration(ctx context.Context, version int32) (time.Time, error) GetNodeAddresses(ctx context.Context, nodeID int64) ([]GetNodeAddressesRow, error) + GetNodeAddressesBatch(ctx context.Context, ids []int64) ([]GraphNodeAddress, error) GetNodeByPubKey(ctx context.Context, arg GetNodeByPubKeyParams) (GraphNode, error) + GetNodeExtraTypesBatch(ctx context.Context, ids []int64) ([]GraphNodeExtraType, error) GetNodeFeatures(ctx context.Context, nodeID int64) ([]GraphNodeFeature, error) + GetNodeFeaturesBatch(ctx context.Context, ids []int64) ([]GraphNodeFeature, error) GetNodeFeaturesByPubKey(ctx context.Context, arg GetNodeFeaturesByPubKeyParams) ([]int32, error) GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyParams) (int64, error) GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]GraphNode, error) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index 542fd6004..afa446f0b 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -114,6 +114,12 @@ FROM graph_nodes n WHERE n.pub_key = $1 AND n.version = $2; +-- name: GetNodeFeaturesBatch :many +SELECT node_id, feature_bit +FROM graph_node_features +WHERE node_id IN (sqlc.slice('ids')/*SLICE:ids*/) +ORDER BY node_id, feature_bit; + -- name: DeleteNodeFeature :exec DELETE FROM graph_node_features WHERE node_id = $1 @@ -140,6 +146,12 @@ FROM graph_node_addresses WHERE node_id = $1 ORDER BY type ASC, position ASC; +-- name: GetNodeAddressesBatch :many +SELECT node_id, type, position, address +FROM graph_node_addresses +WHERE node_id IN (sqlc.slice('ids')/*SLICE:ids*/) +ORDER BY node_id, type, position; + -- name: GetNodesByLastUpdateRange :many SELECT * FROM graph_nodes @@ -170,6 +182,12 @@ SELECT * FROM graph_node_extra_types WHERE node_id = $1; +-- name: GetNodeExtraTypesBatch :many +SELECT node_id, type, value +FROM graph_node_extra_types +WHERE node_id IN (sqlc.slice('ids')/*SLICE:ids*/) +ORDER BY node_id, type; + -- name: DeleteExtraNodeType :exec DELETE FROM graph_node_extra_types WHERE node_id = $1 From d05b918c7a91de2b1d73336aad617a6b4e8fb9f4 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 29 Jul 2025 07:52:16 +0200 Subject: [PATCH 2/6] graph/db: refactor buildNode to use batch fetching Here, we add a new `buildNodeWithBatchData` helper method that can be used to construct a `models.LightningNode` object using pre-fetched batch data. The existing `buildNode` method is then adjusted to use this new helper. --- graph/db/sql_store.go | 231 +++++++++++++++++++++++++----------------- 1 file changed, 137 insertions(+), 94 deletions(-) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index cc7e5d4e9..e0192bd39 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -3352,8 +3352,29 @@ func buildCacheableChannelInfo(scid []byte, capacity int64, node1Pub, // buildNode constructs a LightningNode instance from the given database node // record. The node's features, addresses and extra signed fields are also // fetched from the database and set on the node. -func buildNode(ctx context.Context, db SQLQueries, dbNode *sqlc.GraphNode) ( - *models.LightningNode, error) { +func buildNode(ctx context.Context, db SQLQueries, + dbNode *sqlc.GraphNode) (*models.LightningNode, error) { + + // NOTE: buildNode is only used to load the data for a single node, 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 := batchLoadNodeData(ctx, cfg, db, []int64{dbNode.ID}) + if err != nil { + return nil, fmt.Errorf("unable to batch load node data: %w", + err) + } + + return buildNodeWithBatchData(dbNode, data) +} + +// buildNodeWithBatchData builds a models.LightningNode instance +// from the provided sqlc.GraphNode and batchNodeData. If the node does have +// features/addresses/extra fields, then the corresponding fields are expected +// to be present in the batchNodeData. +func buildNodeWithBatchData(dbNode *sqlc.GraphNode, + batchData *batchNodeData) (*models.LightningNode, error) { if dbNode.Version != int16(ProtocolV1) { return nil, fmt.Errorf("unsupported node version: %d", @@ -3387,35 +3408,35 @@ func buildNode(ctx context.Context, db SQLQueries, dbNode *sqlc.GraphNode) ( } } - // Fetch the node's features. - node.Features, err = getNodeFeatures(ctx, db, dbNode.ID) - if err != nil { - return nil, fmt.Errorf("unable to fetch node(%d) "+ - "features: %w", dbNode.ID, err) + // Use preloaded features. + if features, exists := batchData.features[dbNode.ID]; exists { + fv := lnwire.EmptyFeatureVector() + for _, bit := range features { + fv.Set(lnwire.FeatureBit(bit)) + } + node.Features = fv } - // Fetch the node's addresses. - node.Addresses, err = getNodeAddresses(ctx, db, dbNode.ID) - if err != nil { - return nil, fmt.Errorf("unable to fetch node(%d) "+ - "addresses: %w", dbNode.ID, err) + // Use preloaded addresses. + addresses, exists := batchData.addresses[dbNode.ID] + if exists && len(addresses) > 0 { + node.Addresses, err = buildNodeAddresses(addresses) + if err != nil { + return nil, fmt.Errorf("unable to build addresses "+ + "for node(%d): %w", dbNode.ID, err) + } } - // Fetch the node's extra signed fields. - extraTLVMap, err := getNodeExtraSignedFields(ctx, db, dbNode.ID) - if err != nil { - return nil, fmt.Errorf("unable to fetch node(%d) "+ - "extra signed fields: %w", dbNode.ID, err) - } - - recs, err := lnwire.CustomRecords(extraTLVMap).Serialize() - if err != nil { - return nil, fmt.Errorf("unable to serialize extra signed "+ - "fields: %w", err) - } - - if len(recs) != 0 { - node.ExtraOpaqueData = recs + // Use preloaded extra fields. + if extraFields, exists := batchData.extraFields[dbNode.ID]; exists { + recs, err := lnwire.CustomRecords(extraFields).Serialize() + if err != nil { + return nil, fmt.Errorf("unable to serialize extra "+ + "signed fields: %w", err) + } + if len(recs) != 0 { + node.ExtraOpaqueData = recs + } } return node, nil @@ -3440,25 +3461,6 @@ func getNodeFeatures(ctx context.Context, db SQLQueries, return features, nil } -// getNodeExtraSignedFields fetches the extra signed fields for a node with the -// given DB ID. -func getNodeExtraSignedFields(ctx context.Context, db SQLQueries, - nodeID int64) (map[uint64][]byte, error) { - - fields, err := db.GetExtraNodeTypes(ctx, nodeID) - if err != nil { - return nil, fmt.Errorf("unable to get node(%d) extra "+ - "signed fields: %w", nodeID, err) - } - - extraFields := make(map[uint64][]byte) - for _, field := range fields { - extraFields[uint64(field.Type)] = field.Value - } - - return extraFields, nil -} - // upsertNode upserts the node record into the database. If the node already // exists, then the node's information is updated. If the node doesn't exist, // then a new node is created. The node's features, addresses and extra TLV @@ -3714,55 +3716,13 @@ func getNodeAddresses(ctx context.Context, db SQLQueries, id int64) ([]net.Addr, for _, row := range rows { address := row.Address - switch dbAddressType(row.Type) { - case addressTypeIPv4: - tcp, err := net.ResolveTCPAddr("tcp4", address) - if err != nil { - return nil, err - } - tcp.IP = tcp.IP.To4() - - addresses = append(addresses, tcp) - - case addressTypeIPv6: - tcp, err := net.ResolveTCPAddr("tcp6", address) - if err != nil { - return nil, err - } - addresses = append(addresses, tcp) - - case addressTypeTorV3, addressTypeTorV2: - service, portStr, err := net.SplitHostPort(address) - if err != nil { - return nil, fmt.Errorf("unable to "+ - "split tor v3 address: %v", address) - } - - port, err := strconv.Atoi(portStr) - if err != nil { - return nil, err - } - - addresses = append(addresses, &tor.OnionAddr{ - OnionService: service, - Port: port, - }) - - case addressTypeOpaque: - opaque, err := hex.DecodeString(address) - if err != nil { - return nil, fmt.Errorf("unable to "+ - "decode opaque address: %v", address) - } - - addresses = append(addresses, &lnwire.OpaqueAddrs{ - Payload: opaque, - }) - - default: - return nil, fmt.Errorf("unknown address type: %v", - row.Type) + addr, err := parseAddress(dbAddressType(row.Type), address) + if err != nil { + return nil, fmt.Errorf("unable to parse address "+ + "for node(%d): %v: %w", id, address, err) } + + addresses = append(addresses, addr) } // If we have no addresses, then we'll return nil instead of an @@ -4676,6 +4636,89 @@ func channelIDToBytes(channelID uint64) []byte { return chanIDB[:] } +// buildNodeAddresses converts a slice of nodeAddress into a slice of net.Addr. +func buildNodeAddresses(addresses []nodeAddress) ([]net.Addr, error) { + if len(addresses) == 0 { + return nil, nil + } + + result := make([]net.Addr, 0, len(addresses)) + for _, addr := range addresses { + netAddr, err := parseAddress(addr.addrType, addr.address) + if err != nil { + return nil, fmt.Errorf("unable to parse address %s "+ + "of type %d: %w", addr.address, addr.addrType, + err) + } + if netAddr != nil { + result = append(result, netAddr) + } + } + + // If we have no valid addresses, return nil instead of empty slice. + if len(result) == 0 { + return nil, nil + } + + return result, nil +} + +// parseAddress parses the given address string based on the address type +// and returns a net.Addr instance. It supports IPv4, IPv6, Tor v2, Tor v3, +// and opaque addresses. +func parseAddress(addrType dbAddressType, address string) (net.Addr, error) { + switch addrType { + case addressTypeIPv4: + tcp, err := net.ResolveTCPAddr("tcp4", address) + if err != nil { + return nil, err + } + + tcp.IP = tcp.IP.To4() + + return tcp, nil + + case addressTypeIPv6: + tcp, err := net.ResolveTCPAddr("tcp6", address) + if err != nil { + return nil, err + } + + return tcp, nil + + case addressTypeTorV3, addressTypeTorV2: + service, portStr, err := net.SplitHostPort(address) + if err != nil { + return nil, fmt.Errorf("unable to split tor "+ + "address: %v", address) + } + + port, err := strconv.Atoi(portStr) + if err != nil { + return nil, err + } + + return &tor.OnionAddr{ + OnionService: service, + Port: port, + }, nil + + case addressTypeOpaque: + opaque, err := hex.DecodeString(address) + if err != nil { + return nil, fmt.Errorf("unable to decode opaque "+ + "address: %v", address) + } + + return &lnwire.OpaqueAddrs{ + Payload: opaque, + }, nil + + default: + return nil, fmt.Errorf("unknown address type: %v", addrType) + } +} + // batchNodeData holds all the related data for a batch of nodes. type batchNodeData struct { // features is a map from a DB node ID to the feature bits for that From 84b167074253663e0057981f54781d6cb69524b3 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 29 Jul 2025 11:08:14 +0200 Subject: [PATCH 3/6] graph/db: use batch loading for ForEachNode --- graph/db/sql_store.go | 79 +++++++++++++++++++++++++++++-------------- 1 file changed, 54 insertions(+), 25 deletions(-) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index e0192bd39..3bcdd104b 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -807,26 +807,22 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, func (s *SQLStore) ForEachNode(ctx context.Context, cb func(tx NodeRTx) error, reset func()) error { - var lastID int64 = 0 - handleNode := func(db SQLQueries, dbNode sqlc.GraphNode) error { - node, err := buildNode(ctx, db, &dbNode) - if err != nil { - return fmt.Errorf("unable to build node(id=%d): %w", - dbNode.ID, err) - } - - err = cb( - newSQLGraphNodeTx(db, s.cfg.ChainHash, dbNode.ID, node), - ) - if err != nil { - return fmt.Errorf("callback failed for node(id=%d): %w", - dbNode.ID, err) - } - - return nil - } + var lastID int64 return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { + nodeCB := func(dbID int64, node *models.LightningNode) error { + err := cb(newSQLGraphNodeTx( + db, s.cfg.ChainHash, dbID, node, + )) + if err != nil { + return fmt.Errorf("callback failed for "+ + "node(id=%d): %w", dbID, err) + } + lastID = dbID + + return nil + } + for { nodes, err := db.ListNodesPaginated( ctx, sqlc.ListNodesPaginatedParams{ @@ -844,13 +840,12 @@ func (s *SQLStore) ForEachNode(ctx context.Context, break } - for _, dbNode := range nodes { - err = handleNode(db, dbNode) - if err != nil { - return err - } - - lastID = dbNode.ID + err = forEachNodeInBatch( + ctx, s.cfg.PaginationCfg, db, nodes, nodeCB, + ) + if err != nil { + return fmt.Errorf("unable to iterate over "+ + "nodes: %w", err) } } @@ -3442,6 +3437,40 @@ func buildNodeWithBatchData(dbNode *sqlc.GraphNode, return node, nil } +// forEachNodeInBatch fetches all nodes in the provided batch, builds them +// with the preloaded data, and executes the provided callback for each node. +func forEachNodeInBatch(ctx context.Context, cfg *sqldb.PagedQueryConfig, + db SQLQueries, nodes []sqlc.GraphNode, + cb func(dbID int64, node *models.LightningNode) error) error { + + // Extract node IDs for batch loading. + nodeIDs := make([]int64, len(nodes)) + for i, node := range nodes { + nodeIDs[i] = node.ID + } + + // Batch load all related data for this page. + batchData, err := batchLoadNodeData(ctx, cfg, db, nodeIDs) + if err != nil { + return fmt.Errorf("unable to batch load node data: %w", err) + } + + for _, dbNode := range nodes { + node, err := buildNodeWithBatchData(&dbNode, batchData) + if err != nil { + return fmt.Errorf("unable to build node(id=%d): %w", + dbNode.ID, err) + } + + if err := cb(dbNode.ID, node); err != nil { + return fmt.Errorf("callback failed for node(id=%d): %w", + dbNode.ID, err) + } + } + + return nil +} + // getNodeFeatures fetches the feature bits and constructs the feature vector // for a node with the given DB ID. func getNodeFeatures(ctx context.Context, db SQLQueries, From c8c86717f2bd561477ed103bf49b32b9d609b074 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 29 Jul 2025 11:33:29 +0200 Subject: [PATCH 4/6] graph/db: add NodeUpdatesInHorizon benchmark --- graph/db/benchmark_test.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index 81083a97f..18d45be58 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -673,6 +673,15 @@ func BenchmarkGraphReadMethods(b *testing.B) { require.NoError(b, err) }, }, + { + name: "NodeUpdatesInHorizon", + fn: func(b testing.TB, store V1Store) { + _, err := store.NodeUpdatesInHorizon( + time.Unix(0, 0), time.Now(), + ) + require.NoError(b, err) + }, + }, } for _, test := range tests { From 36eb0e3ff6dcac499dc5ff72bc15651b9f033b70 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 29 Jul 2025 11:09:22 +0200 Subject: [PATCH 5/6] graph/db: use batch loading for NodeUpdatesInHorizon --- graph/db/sql_store.go | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 3bcdd104b..5c8c12e8e 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -554,14 +554,16 @@ func (s *SQLStore) NodeUpdatesInHorizon(startTime, return fmt.Errorf("unable to fetch nodes: %w", err) } - for _, dbNode := range dbNodes { - node, err := buildNode(ctx, db, &dbNode) - if err != nil { - return fmt.Errorf("unable to build node: %w", - err) - } + err = forEachNodeInBatch( + ctx, s.cfg.PaginationCfg, db, dbNodes, + func(_ int64, node *models.LightningNode) error { + nodes = append(nodes, *node) - nodes = append(nodes, *node) + return nil + }, + ) + if err != nil { + return fmt.Errorf("unable to build nodes: %w", err) } return nil From 6ca43173a18905f6eadd57a9f45d5f7e4571739f Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 28 Jul 2025 12:37:11 +0200 Subject: [PATCH 6/6] sqldb: change default slices page size --- sqldb/paginate.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/sqldb/paginate.go b/sqldb/paginate.go index 143b456d5..e0b382b40 100644 --- a/sqldb/paginate.go +++ b/sqldb/paginate.go @@ -23,7 +23,9 @@ type PagedQueryConfig struct { // DefaultPagedQueryConfig returns a default configuration func DefaultPagedQueryConfig() *PagedQueryConfig { return &PagedQueryConfig{ - PageSize: 1000, + // TODO(elle): make configurable & have different defaults + // for SQLite and Postgres. + PageSize: 250, } }