graph/db: use new batch helpers for edge loading
Use the new batch helpers to replace the existing logic for loading an edge.
This commit is contained in:
4 files changed
+76
-239
No files matched your search
+76
-90
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user