Merge pull request #10113 from ellemouton/graphPerf2

[1] graph/db: add some SQL performance improvements
This commit is contained in:
Elle authored and GitHub committed 2025-07-30 19:27:21 +02:00
commit 4389067989
6 files changed
+277 -80

No files matched your search

+82 -55
View File
@@ -73,7 +73,7 @@ type SQLQueries interface {
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
GetNodeAddressesByPubKey(ctx context.Context, arg sqlc.GetNodeAddressesByPubKeyParams) ([]sqlc.GetNodeAddressesByPubKeyRow, error) GetNodeAddresses(ctx context.Context, nodeID int64) ([]sqlc.GetNodeAddressesRow, 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
@@ -103,6 +103,7 @@ type SQLQueries interface {
HighestSCID(ctx context.Context, version int16) ([]byte, error) HighestSCID(ctx context.Context, version int16) ([]byte, error)
ListChannelsByNodeID(ctx context.Context, arg sqlc.ListChannelsByNodeIDParams) ([]sqlc.ListChannelsByNodeIDRow, error) ListChannelsByNodeID(ctx context.Context, arg sqlc.ListChannelsByNodeIDParams) ([]sqlc.ListChannelsByNodeIDRow, error)
ListChannelsWithPoliciesPaginated(ctx context.Context, arg sqlc.ListChannelsWithPoliciesPaginatedParams) ([]sqlc.ListChannelsWithPoliciesPaginatedRow, error) ListChannelsWithPoliciesPaginated(ctx context.Context, arg sqlc.ListChannelsWithPoliciesPaginatedParams) ([]sqlc.ListChannelsWithPoliciesPaginatedRow, error)
ListChannelsWithPoliciesForCachePaginated(ctx context.Context, arg sqlc.ListChannelsWithPoliciesForCachePaginatedParams) ([]sqlc.ListChannelsWithPoliciesForCachePaginatedRow, error)
ListChannelsPaginated(ctx context.Context, arg sqlc.ListChannelsPaginatedParams) ([]sqlc.ListChannelsPaginatedRow, error) ListChannelsPaginated(ctx context.Context, arg sqlc.ListChannelsPaginatedParams) ([]sqlc.ListChannelsPaginatedRow, error)
GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg sqlc.GetChannelsByPolicyLastUpdateRangeParams) ([]sqlc.GetChannelsByPolicyLastUpdateRangeRow, error) GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg sqlc.GetChannelsByPolicyLastUpdateRangeParams) ([]sqlc.GetChannelsByPolicyLastUpdateRangeRow, error)
GetChannelByOutpointWithPolicies(ctx context.Context, arg sqlc.GetChannelByOutpointWithPoliciesParams) (sqlc.GetChannelByOutpointWithPoliciesRow, error) GetChannelByOutpointWithPolicies(ctx context.Context, arg sqlc.GetChannelByOutpointWithPoliciesParams) (sqlc.GetChannelByOutpointWithPoliciesRow, error)
@@ -320,10 +321,21 @@ func (s *SQLStore) AddrsForNode(ctx context.Context,
known bool known bool
) )
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
var err error // First, check if the node exists and get its DB ID if it
known, addresses, err = getNodeAddresses( // does.
ctx, db, nodePub.SerializeCompressed(), dbID, err := db.GetNodeIDByPubKey(
ctx, sqlc.GetNodeIDByPubKeyParams{
Version: int16(ProtocolV1),
PubKey: nodePub.SerializeCompressed(),
},
) )
if errors.Is(err, sql.ErrNoRows) {
return nil
}
known = true
addresses, err = getNodeAddresses(ctx, db, dbID)
if err != nil { if err != nil {
return fmt.Errorf("unable to fetch node addresses: %w", return fmt.Errorf("unable to fetch node addresses: %w",
err) err)
@@ -1247,8 +1259,8 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
ctx := context.TODO() ctx := context.TODO()
handleChannel := func(db SQLQueries, handleChannel := func(
row sqlc.ListChannelsWithPoliciesPaginatedRow) error { row sqlc.ListChannelsWithPoliciesForCachePaginatedRow) error {
node1, node2, err := buildNodeVertices( node1, node2, err := buildNodeVertices(
row.Node1Pubkey, row.Node2Pubkey, row.Node1Pubkey, row.Node2Pubkey,
@@ -1258,7 +1270,7 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
} }
edge := buildCacheableChannelInfo( edge := buildCacheableChannelInfo(
row.GraphChannel, node1, node2, row.Scid, row.Capacity.Int64, node1, node2,
) )
dbPol1, dbPol2, err := extractChannelPolicies(row) dbPol1, dbPol2, err := extractChannelPolicies(row)
@@ -1299,8 +1311,8 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
lastID := int64(-1) lastID := int64(-1)
for { for {
//nolint:ll //nolint:ll
rows, err := db.ListChannelsWithPoliciesPaginated( rows, err := db.ListChannelsWithPoliciesForCachePaginated(
ctx, sqlc.ListChannelsWithPoliciesPaginatedParams{ ctx, sqlc.ListChannelsWithPoliciesForCachePaginatedParams{
Version: int16(ProtocolV1), Version: int16(ProtocolV1),
ID: lastID, ID: lastID,
Limit: pageSize, Limit: pageSize,
@@ -1315,12 +1327,12 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
} }
for _, row := range rows { for _, row := range rows {
err := handleChannel(db, row) err := handleChannel(row)
if err != nil { if err != nil {
return err return err
} }
lastID = row.GraphChannel.ID lastID = row.ID
} }
} }
@@ -3018,7 +3030,8 @@ func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries,
} }
edge := buildCacheableChannelInfo( edge := buildCacheableChannelInfo(
row.GraphChannel, node1, node2, row.GraphChannel.Scid, row.GraphChannel.Capacity.Int64,
node1, node2,
) )
dbPol1, dbPol2, err := extractChannelPolicies(row) dbPol1, dbPol2, err := extractChannelPolicies(row)
@@ -3321,16 +3334,15 @@ func getNodeByPubKey(ctx context.Context, db SQLQueries,
} }
// buildCacheableChannelInfo builds a models.CachedEdgeInfo instance from the // buildCacheableChannelInfo builds a models.CachedEdgeInfo instance from the
// provided database channel row and the public keys of the two nodes // provided parameters.
// involved in the channel. func buildCacheableChannelInfo(scid []byte, capacity int64, node1Pub,
func buildCacheableChannelInfo(dbChan sqlc.GraphChannel, node1Pub,
node2Pub route.Vertex) *models.CachedEdgeInfo { node2Pub route.Vertex) *models.CachedEdgeInfo {
return &models.CachedEdgeInfo{ return &models.CachedEdgeInfo{
ChannelID: byteOrder.Uint64(dbChan.Scid), ChannelID: byteOrder.Uint64(scid),
NodeKey1Bytes: node1Pub, NodeKey1Bytes: node1Pub,
NodeKey2Bytes: node2Pub, NodeKey2Bytes: node2Pub,
Capacity: btcutil.Amount(dbChan.Capacity.Int64), Capacity: btcutil.Amount(capacity),
} }
} }
@@ -3380,7 +3392,7 @@ func buildNode(ctx context.Context, db SQLQueries, dbNode *sqlc.GraphNode) (
} }
// Fetch the node's addresses. // Fetch the node's addresses.
_, node.Addresses, err = getNodeAddresses(ctx, db, pub[:]) node.Addresses, err = getNodeAddresses(ctx, db, dbNode.ID)
if err != nil { if err != nil {
return nil, fmt.Errorf("unable to fetch node(%d) "+ return nil, fmt.Errorf("unable to fetch node(%d) "+
"addresses: %w", dbNode.ID, err) "addresses: %w", dbNode.ID, err)
@@ -3684,42 +3696,26 @@ func upsertNodeAddresses(ctx context.Context, db SQLQueries, nodeID int64,
return nil return nil
} }
// getNodeAddresses fetches the addresses for a node with the given public key. // getNodeAddresses fetches the addresses for a node with the given DB ID.
func getNodeAddresses(ctx context.Context, db SQLQueries, nodePub []byte) (bool, func getNodeAddresses(ctx context.Context, db SQLQueries, id int64) ([]net.Addr,
[]net.Addr, error) { error) {
// GetNodeAddressesByPubKey ensures that the addresses for a given type // GetNodeAddresses ensures that the addresses for a given type are
// are returned in the same order as they were inserted. // returned in the same order as they were inserted.
rows, err := db.GetNodeAddressesByPubKey( rows, err := db.GetNodeAddresses(ctx, id)
ctx, sqlc.GetNodeAddressesByPubKeyParams{
Version: int16(ProtocolV1),
PubKey: nodePub,
},
)
if err != nil { if err != nil {
return false, nil, err return nil, err
}
// GetNodeAddressesByPubKey uses a left join so there should always be
// at least one row returned if the node exists even if it has no
// addresses.
if len(rows) == 0 {
return false, nil, nil
} }
addresses := make([]net.Addr, 0, len(rows)) addresses := make([]net.Addr, 0, len(rows))
for _, addr := range rows { for _, row := range rows {
if !(addr.Type.Valid && addr.Address.Valid) { address := row.Address
continue
}
address := addr.Address.String switch dbAddressType(row.Type) {
switch dbAddressType(addr.Type.Int16) {
case addressTypeIPv4: case addressTypeIPv4:
tcp, err := net.ResolveTCPAddr("tcp4", address) tcp, err := net.ResolveTCPAddr("tcp4", address)
if err != nil { if err != nil {
return false, nil, nil return nil, err
} }
tcp.IP = tcp.IP.To4() tcp.IP = tcp.IP.To4()
@@ -3728,21 +3724,20 @@ func getNodeAddresses(ctx context.Context, db SQLQueries, nodePub []byte) (bool,
case addressTypeIPv6: case addressTypeIPv6:
tcp, err := net.ResolveTCPAddr("tcp6", address) tcp, err := net.ResolveTCPAddr("tcp6", address)
if err != nil { if err != nil {
return false, nil, nil return nil, err
} }
addresses = append(addresses, tcp) addresses = append(addresses, tcp)
case addressTypeTorV3, addressTypeTorV2: case addressTypeTorV3, addressTypeTorV2:
service, portStr, err := net.SplitHostPort(address) service, portStr, err := net.SplitHostPort(address)
if err != nil { if err != nil {
return false, nil, fmt.Errorf("unable to "+ return nil, fmt.Errorf("unable to "+
"split tor v3 address: %v", "split tor v3 address: %v", address)
addr.Address)
} }
port, err := strconv.Atoi(portStr) port, err := strconv.Atoi(portStr)
if err != nil { if err != nil {
return false, nil, err return nil, err
} }
addresses = append(addresses, &tor.OnionAddr{ addresses = append(addresses, &tor.OnionAddr{
@@ -3753,8 +3748,8 @@ func getNodeAddresses(ctx context.Context, db SQLQueries, nodePub []byte) (bool,
case addressTypeOpaque: case addressTypeOpaque:
opaque, err := hex.DecodeString(address) opaque, err := hex.DecodeString(address)
if err != nil { if err != nil {
return false, nil, fmt.Errorf("unable to "+ return nil, fmt.Errorf("unable to "+
"decode opaque address: %v", addr) "decode opaque address: %v", address)
} }
addresses = append(addresses, &lnwire.OpaqueAddrs{ addresses = append(addresses, &lnwire.OpaqueAddrs{
@@ -3762,8 +3757,8 @@ func getNodeAddresses(ctx context.Context, db SQLQueries, nodePub []byte) (bool,
}) })
default: default:
return false, nil, fmt.Errorf("unknown address "+ return nil, fmt.Errorf("unknown address type: %v",
"type: %v", addr.Type) row.Type)
} }
} }
@@ -3773,7 +3768,7 @@ func getNodeAddresses(ctx context.Context, db SQLQueries, nodePub []byte) (bool,
addresses = nil addresses = nil
} }
return true, addresses, nil return addresses, nil
} }
// upsertNodeExtraSignedFields updates the node's extra signed fields in the // upsertNodeExtraSignedFields updates the node's extra signed fields in the
@@ -4368,6 +4363,38 @@ func extractChannelPolicies(row any) (*sqlc.GraphChannelPolicy,
var policy1, policy2 *sqlc.GraphChannelPolicy var policy1, policy2 *sqlc.GraphChannelPolicy
switch r := row.(type) { switch r := row.(type) {
case sqlc.ListChannelsWithPoliciesForCachePaginatedRow:
if r.Policy1Timelock.Valid {
policy1 = &sqlc.GraphChannelPolicy{
Timelock: r.Policy1Timelock.Int32,
FeePpm: r.Policy1FeePpm.Int64,
BaseFeeMsat: r.Policy1BaseFeeMsat.Int64,
MinHtlcMsat: r.Policy1MinHtlcMsat.Int64,
MaxHtlcMsat: r.Policy1MaxHtlcMsat,
InboundBaseFeeMsat: r.Policy1InboundBaseFeeMsat,
InboundFeeRateMilliMsat: r.Policy1InboundFeeRateMilliMsat,
Disabled: r.Policy1Disabled,
MessageFlags: r.Policy1MessageFlags,
ChannelFlags: r.Policy1ChannelFlags,
}
}
if r.Policy2Timelock.Valid {
policy2 = &sqlc.GraphChannelPolicy{
Timelock: r.Policy2Timelock.Int32,
FeePpm: r.Policy2FeePpm.Int64,
BaseFeeMsat: r.Policy2BaseFeeMsat.Int64,
MinHtlcMsat: r.Policy2MinHtlcMsat.Int64,
MaxHtlcMsat: r.Policy2MaxHtlcMsat,
InboundBaseFeeMsat: r.Policy2InboundBaseFeeMsat,
InboundFeeRateMilliMsat: r.Policy2InboundFeeRateMilliMsat,
Disabled: r.Policy2Disabled,
MessageFlags: r.Policy2MessageFlags,
ChannelFlags: r.Policy2ChannelFlags,
}
}
return policy1, policy2, nil
case sqlc.GetChannelsBySCIDWithPoliciesRow: case sqlc.GetChannelsBySCIDWithPoliciesRow:
if r.Policy1ID.Valid { if r.Policy1ID.Valid {
policy1 = &sqlc.GraphChannelPolicy{ policy1 = &sqlc.GraphChannelPolicy{
+141 -18
View File
@@ -1438,33 +1438,27 @@ func (q *Queries) GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]GraphN
return items, nil return items, nil
} }
const getNodeAddressesByPubKey = `-- name: GetNodeAddressesByPubKey :many const getNodeAddresses = `-- name: GetNodeAddresses :many
SELECT a.type, a.address SELECT type, address
FROM graph_nodes n FROM graph_node_addresses
LEFT JOIN graph_node_addresses a ON a.node_id = n.id WHERE node_id = $1
WHERE n.pub_key = $1 AND n.version = $2 ORDER BY type ASC, position ASC
ORDER BY a.type ASC, a.position ASC
` `
type GetNodeAddressesByPubKeyParams struct { type GetNodeAddressesRow struct {
PubKey []byte Type int16
Version int16 Address string
} }
type GetNodeAddressesByPubKeyRow struct { func (q *Queries) GetNodeAddresses(ctx context.Context, nodeID int64) ([]GetNodeAddressesRow, error) {
Type sql.NullInt16 rows, err := q.db.QueryContext(ctx, getNodeAddresses, nodeID)
Address sql.NullString
}
func (q *Queries) GetNodeAddressesByPubKey(ctx context.Context, arg GetNodeAddressesByPubKeyParams) ([]GetNodeAddressesByPubKeyRow, error) {
rows, err := q.db.QueryContext(ctx, getNodeAddressesByPubKey, arg.PubKey, arg.Version)
if err != nil { if err != nil {
return nil, err return nil, err
} }
defer rows.Close() defer rows.Close()
var items []GetNodeAddressesByPubKeyRow var items []GetNodeAddressesRow
for rows.Next() { for rows.Next() {
var i GetNodeAddressesByPubKeyRow var i GetNodeAddressesRow
if err := rows.Scan(&i.Type, &i.Address); err != nil { if err := rows.Scan(&i.Type, &i.Address); err != nil {
return nil, err return nil, err
} }
@@ -2230,6 +2224,135 @@ func (q *Queries) ListChannelsPaginated(ctx context.Context, arg ListChannelsPag
return items, nil return items, nil
} }
const listChannelsWithPoliciesForCachePaginated = `-- name: ListChannelsWithPoliciesForCachePaginated :many
SELECT
c.id as id,
c.scid as scid,
c.capacity AS capacity,
-- Join node pubkeys
n1.pub_key AS node1_pubkey,
n2.pub_key AS node2_pubkey,
-- Node 1 policy
cp1.timelock AS policy_1_timelock,
cp1.fee_ppm AS policy_1_fee_ppm,
cp1.base_fee_msat AS policy_1_base_fee_msat,
cp1.min_htlc_msat AS policy_1_min_htlc_msat,
cp1.max_htlc_msat AS policy_1_max_htlc_msat,
cp1.disabled AS policy_1_disabled,
cp1.inbound_base_fee_msat AS policy1_inbound_base_fee_msat,
cp1.inbound_fee_rate_milli_msat AS policy1_inbound_fee_rate_milli_msat,
cp1.message_flags AS policy1_message_flags,
cp1.channel_flags AS policy1_channel_flags,
-- Node 2 policy
cp2.timelock AS policy_2_timelock,
cp2.fee_ppm AS policy_2_fee_ppm,
cp2.base_fee_msat AS policy_2_base_fee_msat,
cp2.min_htlc_msat AS policy_2_min_htlc_msat,
cp2.max_htlc_msat AS policy_2_max_htlc_msat,
cp2.disabled AS policy_2_disabled,
cp2.inbound_base_fee_msat AS policy2_inbound_base_fee_msat,
cp2.inbound_fee_rate_milli_msat AS policy2_inbound_fee_rate_milli_msat,
cp2.message_flags AS policy2_message_flags,
cp2.channel_flags AS policy2_channel_flags
FROM graph_channels c
JOIN graph_nodes n1 ON c.node_id_1 = n1.id
JOIN graph_nodes n2 ON c.node_id_2 = n2.id
LEFT JOIN graph_channel_policies cp1
ON cp1.channel_id = c.id AND cp1.node_id = c.node_id_1 AND cp1.version = c.version
LEFT JOIN graph_channel_policies cp2
ON cp2.channel_id = c.id AND cp2.node_id = c.node_id_2 AND cp2.version = c.version
WHERE c.version = $1 AND c.id > $2
ORDER BY c.id
LIMIT $3
`
type ListChannelsWithPoliciesForCachePaginatedParams struct {
Version int16
ID int64
Limit int32
}
type ListChannelsWithPoliciesForCachePaginatedRow struct {
ID int64
Scid []byte
Capacity sql.NullInt64
Node1Pubkey []byte
Node2Pubkey []byte
Policy1Timelock sql.NullInt32
Policy1FeePpm sql.NullInt64
Policy1BaseFeeMsat sql.NullInt64
Policy1MinHtlcMsat sql.NullInt64
Policy1MaxHtlcMsat sql.NullInt64
Policy1Disabled sql.NullBool
Policy1InboundBaseFeeMsat sql.NullInt64
Policy1InboundFeeRateMilliMsat sql.NullInt64
Policy1MessageFlags sql.NullInt16
Policy1ChannelFlags sql.NullInt16
Policy2Timelock sql.NullInt32
Policy2FeePpm sql.NullInt64
Policy2BaseFeeMsat sql.NullInt64
Policy2MinHtlcMsat sql.NullInt64
Policy2MaxHtlcMsat sql.NullInt64
Policy2Disabled sql.NullBool
Policy2InboundBaseFeeMsat sql.NullInt64
Policy2InboundFeeRateMilliMsat sql.NullInt64
Policy2MessageFlags sql.NullInt16
Policy2ChannelFlags sql.NullInt16
}
func (q *Queries) ListChannelsWithPoliciesForCachePaginated(ctx context.Context, arg ListChannelsWithPoliciesForCachePaginatedParams) ([]ListChannelsWithPoliciesForCachePaginatedRow, error) {
rows, err := q.db.QueryContext(ctx, listChannelsWithPoliciesForCachePaginated, arg.Version, arg.ID, arg.Limit)
if err != nil {
return nil, err
}
defer rows.Close()
var items []ListChannelsWithPoliciesForCachePaginatedRow
for rows.Next() {
var i ListChannelsWithPoliciesForCachePaginatedRow
if err := rows.Scan(
&i.ID,
&i.Scid,
&i.Capacity,
&i.Node1Pubkey,
&i.Node2Pubkey,
&i.Policy1Timelock,
&i.Policy1FeePpm,
&i.Policy1BaseFeeMsat,
&i.Policy1MinHtlcMsat,
&i.Policy1MaxHtlcMsat,
&i.Policy1Disabled,
&i.Policy1InboundBaseFeeMsat,
&i.Policy1InboundFeeRateMilliMsat,
&i.Policy1MessageFlags,
&i.Policy1ChannelFlags,
&i.Policy2Timelock,
&i.Policy2FeePpm,
&i.Policy2BaseFeeMsat,
&i.Policy2MinHtlcMsat,
&i.Policy2MaxHtlcMsat,
&i.Policy2Disabled,
&i.Policy2InboundBaseFeeMsat,
&i.Policy2InboundFeeRateMilliMsat,
&i.Policy2MessageFlags,
&i.Policy2ChannelFlags,
); 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 listChannelsWithPoliciesPaginated = `-- name: ListChannelsWithPoliciesPaginated :many const listChannelsWithPoliciesPaginated = `-- name: ListChannelsWithPoliciesPaginated :many
SELECT 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, 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,
@@ -9,6 +9,7 @@ DROP INDEX IF EXISTS graph_channels_node_id_1_idx;
DROP INDEX IF EXISTS graph_channels_node_id_2_idx; DROP INDEX IF EXISTS graph_channels_node_id_2_idx;
DROP INDEX IF EXISTS graph_channels_unique; DROP INDEX IF EXISTS graph_channels_unique;
DROP INDEX IF EXISTS graph_channels_version_outpoint_idx; DROP INDEX IF EXISTS graph_channels_version_outpoint_idx;
DROP INDEX IF EXISTS graph_channels_version_id_idx;
DROP INDEX IF EXISTS graph_channel_features_unique; DROP INDEX IF EXISTS graph_channel_features_unique;
DROP INDEX IF EXISTS graph_channel_extra_types_unique; DROP INDEX IF EXISTS graph_channel_extra_types_unique;
DROP INDEX IF EXISTS graph_channel_policies_unique; DROP INDEX IF EXISTS graph_channel_policies_unique;
@@ -173,6 +173,7 @@ CREATE TABLE IF NOT EXISTS graph_channels (
-- indexes on the node_id_1 and node_id_2 columns. -- indexes on the node_id_1 and node_id_2 columns.
CREATE INDEX IF NOT EXISTS graph_channels_node_id_1_idx ON graph_channels(node_id_1); CREATE INDEX IF NOT EXISTS graph_channels_node_id_1_idx ON graph_channels(node_id_1);
CREATE INDEX IF NOT EXISTS graph_channels_node_id_2_idx ON graph_channels(node_id_2); CREATE INDEX IF NOT EXISTS graph_channels_node_id_2_idx ON graph_channels(node_id_2);
CREATE INDEX IF NOT EXISTS graph_channels_version_id_idx ON graph_channels(version, id);
-- A channel (identified by a short channel id) can only have one active -- A channel (identified by a short channel id) can only have one active
-- channel announcement per protocol version. We also order the index by -- channel announcement per protocol version. We also order the index by
+2 -1
View File
@@ -59,7 +59,7 @@ type Querier interface {
GetInvoiceHTLCs(ctx context.Context, invoiceID int64) ([]InvoiceHtlc, error) GetInvoiceHTLCs(ctx context.Context, invoiceID int64) ([]InvoiceHtlc, error)
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)
GetNodeAddressesByPubKey(ctx context.Context, arg GetNodeAddressesByPubKeyParams) ([]GetNodeAddressesByPubKeyRow, error) GetNodeAddresses(ctx context.Context, nodeID int64) ([]GetNodeAddressesRow, error)
GetNodeByPubKey(ctx context.Context, arg GetNodeByPubKeyParams) (GraphNode, error) GetNodeByPubKey(ctx context.Context, arg GetNodeByPubKeyParams) (GraphNode, error)
GetNodeFeatures(ctx context.Context, nodeID int64) ([]GraphNodeFeature, error) GetNodeFeatures(ctx context.Context, nodeID int64) ([]GraphNodeFeature, error)
GetNodeFeaturesByPubKey(ctx context.Context, arg GetNodeFeaturesByPubKeyParams) ([]int32, error) GetNodeFeaturesByPubKey(ctx context.Context, arg GetNodeFeaturesByPubKeyParams) ([]int32, error)
@@ -95,6 +95,7 @@ type Querier interface {
IsZombieChannel(ctx context.Context, arg IsZombieChannelParams) (bool, error) IsZombieChannel(ctx context.Context, arg IsZombieChannelParams) (bool, error)
ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNodeIDParams) ([]ListChannelsByNodeIDRow, error) ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNodeIDParams) ([]ListChannelsByNodeIDRow, error)
ListChannelsPaginated(ctx context.Context, arg ListChannelsPaginatedParams) ([]ListChannelsPaginatedRow, error) ListChannelsPaginated(ctx context.Context, arg ListChannelsPaginatedParams) ([]ListChannelsPaginatedRow, error)
ListChannelsWithPoliciesForCachePaginated(ctx context.Context, arg ListChannelsWithPoliciesForCachePaginatedParams) ([]ListChannelsWithPoliciesForCachePaginatedRow, error)
ListChannelsWithPoliciesPaginated(ctx context.Context, arg ListChannelsWithPoliciesPaginatedParams) ([]ListChannelsWithPoliciesPaginatedRow, error) ListChannelsWithPoliciesPaginated(ctx context.Context, arg ListChannelsWithPoliciesPaginatedParams) ([]ListChannelsWithPoliciesPaginatedRow, error)
ListNodeIDsAndPubKeys(ctx context.Context, arg ListNodeIDsAndPubKeysParams) ([]ListNodeIDsAndPubKeysRow, error) ListNodeIDsAndPubKeys(ctx context.Context, arg ListNodeIDsAndPubKeysParams) ([]ListNodeIDsAndPubKeysRow, error)
ListNodesPaginated(ctx context.Context, arg ListNodesPaginatedParams) ([]GraphNode, error) ListNodesPaginated(ctx context.Context, arg ListNodesPaginatedParams) ([]GraphNode, error)
+50 -6
View File
@@ -134,12 +134,11 @@ INSERT INTO graph_node_addresses (
$1, $2, $3, $4 $1, $2, $3, $4
); );
-- name: GetNodeAddressesByPubKey :many -- name: GetNodeAddresses :many
SELECT a.type, a.address SELECT type, address
FROM graph_nodes n FROM graph_node_addresses
LEFT JOIN graph_node_addresses a ON a.node_id = n.id WHERE node_id = $1
WHERE n.pub_key = $1 AND n.version = $2 ORDER BY type ASC, position ASC;
ORDER BY a.type ASC, a.position ASC;
-- name: GetNodesByLastUpdateRange :many -- name: GetNodesByLastUpdateRange :many
SELECT * SELECT *
@@ -569,6 +568,51 @@ WHERE c.version = $1 AND c.id > $2
ORDER BY c.id ORDER BY c.id
LIMIT $3; LIMIT $3;
-- name: ListChannelsWithPoliciesForCachePaginated :many
SELECT
c.id as id,
c.scid as scid,
c.capacity AS capacity,
-- Join node pubkeys
n1.pub_key AS node1_pubkey,
n2.pub_key AS node2_pubkey,
-- Node 1 policy
cp1.timelock AS policy_1_timelock,
cp1.fee_ppm AS policy_1_fee_ppm,
cp1.base_fee_msat AS policy_1_base_fee_msat,
cp1.min_htlc_msat AS policy_1_min_htlc_msat,
cp1.max_htlc_msat AS policy_1_max_htlc_msat,
cp1.disabled AS policy_1_disabled,
cp1.inbound_base_fee_msat AS policy1_inbound_base_fee_msat,
cp1.inbound_fee_rate_milli_msat AS policy1_inbound_fee_rate_milli_msat,
cp1.message_flags AS policy1_message_flags,
cp1.channel_flags AS policy1_channel_flags,
-- Node 2 policy
cp2.timelock AS policy_2_timelock,
cp2.fee_ppm AS policy_2_fee_ppm,
cp2.base_fee_msat AS policy_2_base_fee_msat,
cp2.min_htlc_msat AS policy_2_min_htlc_msat,
cp2.max_htlc_msat AS policy_2_max_htlc_msat,
cp2.disabled AS policy_2_disabled,
cp2.inbound_base_fee_msat AS policy2_inbound_base_fee_msat,
cp2.inbound_fee_rate_milli_msat AS policy2_inbound_fee_rate_milli_msat,
cp2.message_flags AS policy2_message_flags,
cp2.channel_flags AS policy2_channel_flags
FROM graph_channels c
JOIN graph_nodes n1 ON c.node_id_1 = n1.id
JOIN graph_nodes n2 ON c.node_id_2 = n2.id
LEFT JOIN graph_channel_policies cp1
ON cp1.channel_id = c.id AND cp1.node_id = c.node_id_1 AND cp1.version = c.version
LEFT JOIN graph_channel_policies cp2
ON cp2.channel_id = c.id AND cp2.node_id = c.node_id_2 AND cp2.version = c.version
WHERE c.version = $1 AND c.id > $2
ORDER BY c.id
LIMIT $3;
-- name: DeleteChannels :exec -- name: DeleteChannels :exec
DELETE FROM graph_channels DELETE FROM graph_channels
WHERE id IN (sqlc.slice('ids')/*SLICE:ids*/); WHERE id IN (sqlc.slice('ids')/*SLICE:ids*/);