Merge pull request #10115 from ellemouton/graphPerf3
[2] graph/db: batch-fetch node data
This commit is contained in:
6 files changed
+513
-127
No files matched your search
@@ -673,6 +673,15 @@ func BenchmarkGraphReadMethods(b *testing.B) {
|
|||||||
require.NoError(b, err)
|
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 {
|
for _, test := range tests {
|
||||||
|
|||||||
+355
-126
@@ -69,15 +69,18 @@ type SQLQueries interface {
|
|||||||
DeleteNode(ctx context.Context, id int64) error
|
DeleteNode(ctx context.Context, id int64) error
|
||||||
|
|
||||||
GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]sqlc.GraphNodeExtraType, 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
|
UpsertNodeExtraType(ctx context.Context, arg sqlc.UpsertNodeExtraTypeParams) error
|
||||||
DeleteExtraNodeType(ctx context.Context, arg sqlc.DeleteExtraNodeTypeParams) error
|
DeleteExtraNodeType(ctx context.Context, arg sqlc.DeleteExtraNodeTypeParams) error
|
||||||
|
|
||||||
InsertNodeAddress(ctx context.Context, arg sqlc.InsertNodeAddressParams) error
|
InsertNodeAddress(ctx context.Context, arg sqlc.InsertNodeAddressParams) error
|
||||||
GetNodeAddresses(ctx context.Context, nodeID int64) ([]sqlc.GetNodeAddressesRow, 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
|
DeleteNodeAddresses(ctx context.Context, nodeID int64) error
|
||||||
|
|
||||||
InsertNodeFeature(ctx context.Context, arg sqlc.InsertNodeFeatureParams) error
|
InsertNodeFeature(ctx context.Context, arg sqlc.InsertNodeFeatureParams) error
|
||||||
GetNodeFeatures(ctx context.Context, nodeID int64) ([]sqlc.GraphNodeFeature, 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)
|
GetNodeFeaturesByPubKey(ctx context.Context, arg sqlc.GetNodeFeaturesByPubKeyParams) ([]int32, error)
|
||||||
DeleteNodeFeature(ctx context.Context, arg sqlc.DeleteNodeFeatureParams) error
|
DeleteNodeFeature(ctx context.Context, arg sqlc.DeleteNodeFeatureParams) error
|
||||||
|
|
||||||
@@ -551,14 +554,16 @@ func (s *SQLStore) NodeUpdatesInHorizon(startTime,
|
|||||||
return fmt.Errorf("unable to fetch nodes: %w", err)
|
return fmt.Errorf("unable to fetch nodes: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, dbNode := range dbNodes {
|
err = forEachNodeInBatch(
|
||||||
node, err := buildNode(ctx, db, &dbNode)
|
ctx, s.cfg.PaginationCfg, db, dbNodes,
|
||||||
if err != nil {
|
func(_ int64, node *models.LightningNode) error {
|
||||||
return fmt.Errorf("unable to build node: %w",
|
nodes = append(nodes, *node)
|
||||||
err)
|
|
||||||
}
|
|
||||||
|
|
||||||
nodes = append(nodes, *node)
|
return nil
|
||||||
|
},
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("unable to build nodes: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
@@ -804,26 +809,22 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context,
|
|||||||
func (s *SQLStore) ForEachNode(ctx context.Context,
|
func (s *SQLStore) ForEachNode(ctx context.Context,
|
||||||
cb func(tx NodeRTx) error, reset func()) error {
|
cb func(tx NodeRTx) error, reset func()) error {
|
||||||
|
|
||||||
var lastID int64 = 0
|
var lastID int64
|
||||||
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
|
|
||||||
}
|
|
||||||
|
|
||||||
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
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 {
|
for {
|
||||||
nodes, err := db.ListNodesPaginated(
|
nodes, err := db.ListNodesPaginated(
|
||||||
ctx, sqlc.ListNodesPaginatedParams{
|
ctx, sqlc.ListNodesPaginatedParams{
|
||||||
@@ -841,13 +842,12 @@ func (s *SQLStore) ForEachNode(ctx context.Context,
|
|||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, dbNode := range nodes {
|
err = forEachNodeInBatch(
|
||||||
err = handleNode(db, dbNode)
|
ctx, s.cfg.PaginationCfg, db, nodes, nodeCB,
|
||||||
if err != nil {
|
)
|
||||||
return err
|
if err != nil {
|
||||||
}
|
return fmt.Errorf("unable to iterate over "+
|
||||||
|
"nodes: %w", err)
|
||||||
lastID = dbNode.ID
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -3349,8 +3349,29 @@ func buildCacheableChannelInfo(scid []byte, capacity int64, node1Pub,
|
|||||||
// buildNode constructs a LightningNode instance from the given database node
|
// buildNode constructs a LightningNode instance from the given database node
|
||||||
// record. The node's features, addresses and extra signed fields are also
|
// record. The node's features, addresses and extra signed fields are also
|
||||||
// fetched from the database and set on the node.
|
// fetched from the database and set on the node.
|
||||||
func buildNode(ctx context.Context, db SQLQueries, dbNode *sqlc.GraphNode) (
|
func buildNode(ctx context.Context, db SQLQueries,
|
||||||
*models.LightningNode, error) {
|
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) {
|
if dbNode.Version != int16(ProtocolV1) {
|
||||||
return nil, fmt.Errorf("unsupported node version: %d",
|
return nil, fmt.Errorf("unsupported node version: %d",
|
||||||
@@ -3384,40 +3405,74 @@ func buildNode(ctx context.Context, db SQLQueries, dbNode *sqlc.GraphNode) (
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Fetch the node's features.
|
// Use preloaded features.
|
||||||
node.Features, err = getNodeFeatures(ctx, db, dbNode.ID)
|
if features, exists := batchData.features[dbNode.ID]; exists {
|
||||||
if err != nil {
|
fv := lnwire.EmptyFeatureVector()
|
||||||
return nil, fmt.Errorf("unable to fetch node(%d) "+
|
for _, bit := range features {
|
||||||
"features: %w", dbNode.ID, err)
|
fv.Set(lnwire.FeatureBit(bit))
|
||||||
|
}
|
||||||
|
node.Features = fv
|
||||||
}
|
}
|
||||||
|
|
||||||
// Fetch the node's addresses.
|
// Use preloaded addresses.
|
||||||
node.Addresses, err = getNodeAddresses(ctx, db, dbNode.ID)
|
addresses, exists := batchData.addresses[dbNode.ID]
|
||||||
if err != nil {
|
if exists && len(addresses) > 0 {
|
||||||
return nil, fmt.Errorf("unable to fetch node(%d) "+
|
node.Addresses, err = buildNodeAddresses(addresses)
|
||||||
"addresses: %w", dbNode.ID, err)
|
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.
|
// Use preloaded extra fields.
|
||||||
extraTLVMap, err := getNodeExtraSignedFields(ctx, db, dbNode.ID)
|
if extraFields, exists := batchData.extraFields[dbNode.ID]; exists {
|
||||||
if err != nil {
|
recs, err := lnwire.CustomRecords(extraFields).Serialize()
|
||||||
return nil, fmt.Errorf("unable to fetch node(%d) "+
|
if err != nil {
|
||||||
"extra signed fields: %w", dbNode.ID, err)
|
return nil, fmt.Errorf("unable to serialize extra "+
|
||||||
}
|
"signed fields: %w", err)
|
||||||
|
}
|
||||||
recs, err := lnwire.CustomRecords(extraTLVMap).Serialize()
|
if len(recs) != 0 {
|
||||||
if err != nil {
|
node.ExtraOpaqueData = recs
|
||||||
return nil, fmt.Errorf("unable to serialize extra signed "+
|
}
|
||||||
"fields: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if len(recs) != 0 {
|
|
||||||
node.ExtraOpaqueData = recs
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return node, nil
|
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
|
// getNodeFeatures fetches the feature bits and constructs the feature vector
|
||||||
// for a node with the given DB ID.
|
// for a node with the given DB ID.
|
||||||
func getNodeFeatures(ctx context.Context, db SQLQueries,
|
func getNodeFeatures(ctx context.Context, db SQLQueries,
|
||||||
@@ -3437,25 +3492,6 @@ func getNodeFeatures(ctx context.Context, db SQLQueries,
|
|||||||
return features, nil
|
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
|
// 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,
|
// 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
|
// then a new node is created. The node's features, addresses and extra TLV
|
||||||
@@ -3711,55 +3747,13 @@ func getNodeAddresses(ctx context.Context, db SQLQueries, id int64) ([]net.Addr,
|
|||||||
for _, row := range rows {
|
for _, row := range rows {
|
||||||
address := row.Address
|
address := row.Address
|
||||||
|
|
||||||
switch dbAddressType(row.Type) {
|
addr, err := parseAddress(dbAddressType(row.Type), address)
|
||||||
case addressTypeIPv4:
|
if err != nil {
|
||||||
tcp, err := net.ResolveTCPAddr("tcp4", address)
|
return nil, fmt.Errorf("unable to parse address "+
|
||||||
if err != nil {
|
"for node(%d): %v: %w", id, address, err)
|
||||||
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)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
addresses = append(addresses, addr)
|
||||||
}
|
}
|
||||||
|
|
||||||
// If we have no addresses, then we'll return nil instead of an
|
// If we have no addresses, then we'll return nil instead of an
|
||||||
@@ -4672,3 +4666,238 @@ func channelIDToBytes(channelID uint64) []byte {
|
|||||||
|
|
||||||
return chanIDB[:]
|
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
|
||||||
|
// 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,
|
||||||
|
)
|
||||||
|
}
|
||||||
+3
-1
@@ -23,7 +23,9 @@ type PagedQueryConfig struct {
|
|||||||
// DefaultPagedQueryConfig returns a default configuration
|
// DefaultPagedQueryConfig returns a default configuration
|
||||||
func DefaultPagedQueryConfig() *PagedQueryConfig {
|
func DefaultPagedQueryConfig() *PagedQueryConfig {
|
||||||
return &PagedQueryConfig{
|
return &PagedQueryConfig{
|
||||||
PageSize: 1000,
|
// TODO(elle): make configurable & have different defaults
|
||||||
|
// for SQLite and Postgres.
|
||||||
|
PageSize: 250,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1473,6 +1473,51 @@ func (q *Queries) GetNodeAddresses(ctx context.Context, nodeID int64) ([]GetNode
|
|||||||
return items, nil
|
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
|
const getNodeByPubKey = `-- name: GetNodeByPubKey :one
|
||||||
SELECT id, version, pub_key, alias, last_update, color, signature
|
SELECT id, version, pub_key, alias, last_update, color, signature
|
||||||
FROM graph_nodes
|
FROM graph_nodes
|
||||||
@@ -1500,6 +1545,46 @@ func (q *Queries) GetNodeByPubKey(ctx context.Context, arg GetNodeByPubKeyParams
|
|||||||
return i, err
|
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
|
const getNodeFeatures = `-- name: GetNodeFeatures :many
|
||||||
SELECT node_id, feature_bit
|
SELECT node_id, feature_bit
|
||||||
FROM graph_node_features
|
FROM graph_node_features
|
||||||
@@ -1529,6 +1614,46 @@ func (q *Queries) GetNodeFeatures(ctx context.Context, nodeID int64) ([]GraphNod
|
|||||||
return items, nil
|
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
|
const getNodeFeaturesByPubKey = `-- name: GetNodeFeaturesByPubKey :many
|
||||||
SELECT f.feature_bit
|
SELECT f.feature_bit
|
||||||
FROM graph_nodes n
|
FROM graph_nodes n
|
||||||
|
|||||||
@@ -60,8 +60,11 @@ type Querier interface {
|
|||||||
GetKVInvoicePaymentHashByAddIndex(ctx context.Context, addIndex int64) ([]byte, error)
|
GetKVInvoicePaymentHashByAddIndex(ctx context.Context, addIndex int64) ([]byte, error)
|
||||||
GetMigration(ctx context.Context, version int32) (time.Time, error)
|
GetMigration(ctx context.Context, version int32) (time.Time, error)
|
||||||
GetNodeAddresses(ctx context.Context, nodeID int64) ([]GetNodeAddressesRow, 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)
|
GetNodeByPubKey(ctx context.Context, arg GetNodeByPubKeyParams) (GraphNode, error)
|
||||||
|
GetNodeExtraTypesBatch(ctx context.Context, ids []int64) ([]GraphNodeExtraType, error)
|
||||||
GetNodeFeatures(ctx context.Context, nodeID int64) ([]GraphNodeFeature, 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)
|
GetNodeFeaturesByPubKey(ctx context.Context, arg GetNodeFeaturesByPubKeyParams) ([]int32, error)
|
||||||
GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyParams) (int64, error)
|
GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyParams) (int64, error)
|
||||||
GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]GraphNode, error)
|
GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]GraphNode, error)
|
||||||
|
|||||||
@@ -114,6 +114,12 @@ FROM graph_nodes n
|
|||||||
WHERE n.pub_key = $1
|
WHERE n.pub_key = $1
|
||||||
AND n.version = $2;
|
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
|
-- name: DeleteNodeFeature :exec
|
||||||
DELETE FROM graph_node_features
|
DELETE FROM graph_node_features
|
||||||
WHERE node_id = $1
|
WHERE node_id = $1
|
||||||
@@ -140,6 +146,12 @@ FROM graph_node_addresses
|
|||||||
WHERE node_id = $1
|
WHERE node_id = $1
|
||||||
ORDER BY type ASC, position ASC;
|
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
|
-- name: GetNodesByLastUpdateRange :many
|
||||||
SELECT *
|
SELECT *
|
||||||
FROM graph_nodes
|
FROM graph_nodes
|
||||||
@@ -170,6 +182,12 @@ SELECT *
|
|||||||
FROM graph_node_extra_types
|
FROM graph_node_extra_types
|
||||||
WHERE node_id = $1;
|
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
|
-- name: DeleteExtraNodeType :exec
|
||||||
DELETE FROM graph_node_extra_types
|
DELETE FROM graph_node_extra_types
|
||||||
WHERE node_id = $1
|
WHERE node_id = $1
|
||||||
|
|||||||
Reference in new issue
Block a user