diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 2f1792b1f..e1ff256a2 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -713,8 +713,7 @@ func getAndBuildChanAndPolicies(ctx context.Context, db SQLQueries, } edge, err := getAndBuildEdgeInfo( - ctx, db, chain, row.GraphChannel.ID, row.GraphChannel, node1, - node2, + ctx, db, chain, row.GraphChannel, node1, node2, ) if err != nil { return nil, nil, nil, fmt.Errorf("unable to build channel "+ diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 5c8c12e8e..f996bfbfa 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -33,7 +33,10 @@ import ( // pageSize is the limit for the number of records that can be returned // in a paginated query. This can be tuned after some benchmarks. -const pageSize = 2000 +// +// TODO(elle): make this configurable & have different defaults for SQLite and +// Postgres. +const pageSize = 10000 // ProtocolVersion is an enum that defines the gossip protocol version of a // message. @@ -102,7 +105,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) @@ -115,7 +117,9 @@ type SQLQueries interface { DeleteChannels(ctx context.Context, ids []int64) error CreateChannelExtraType(ctx context.Context, arg sqlc.CreateChannelExtraTypeParams) error + GetChannelExtrasBatch(ctx context.Context, chanIds []int64) ([]sqlc.GraphChannelExtraType, error) InsertChannelFeature(ctx context.Context, arg sqlc.InsertChannelFeatureParams) error + GetChannelFeaturesBatch(ctx context.Context, chanIds []int64) ([]sqlc.GraphChannelFeature, error) /* Channel Policy table queries. @@ -125,7 +129,7 @@ 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 /* @@ -1053,9 +1057,8 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, } channel, err := getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1.PubKeyBytes, - node2.PubKeyBytes, + ctx, db, s.cfg.ChainHash, row.GraphChannel, + node1.PubKeyBytes, node2.PubKeyBytes, ) if err != nil { return fmt.Errorf("unable to build channel "+ @@ -1161,8 +1164,7 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, e, err := getAndBuildEdgeInfo( ctx, db, s.cfg.ChainHash, - row.GraphChannel.ID, row.GraphChannel, - node1, node2, + row.GraphChannel, node1, node2, ) if err != nil { return fmt.Errorf("unable to build "+ @@ -1355,7 +1357,7 @@ func (s *SQLStore) ForEachChannel(ctx context.Context, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, reset func()) error { - handleChannel := func(db SQLQueries, + handleChannel := func(db SQLQueries, batchData *batchChannelData, row sqlc.ListChannelsWithPoliciesPaginatedRow) error { node1, node2, err := buildNodeVertices( @@ -1366,9 +1368,9 @@ func (s *SQLStore) ForEachChannel(ctx context.Context, err) } - edge, err := getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1, node2, + edge, err := buildEdgeInfoWithBatchData( + s.cfg.ChainHash, row.GraphChannel, node1, node2, + batchData, ) if err != nil { return fmt.Errorf("unable to build channel info: %w", @@ -1381,8 +1383,8 @@ func (s *SQLStore) ForEachChannel(ctx context.Context, "policies: %w", err) } - p1, p2, err := getAndBuildChanPolicies( - ctx, db, dbPol1, dbPol2, edge.ChannelID, node1, node2, + p1, p2, err := buildChanPoliciesWithBatchData( + dbPol1, dbPol2, edge.ChannelID, node1, node2, batchData, ) if err != nil { return fmt.Errorf("unable to build channel "+ @@ -1417,8 +1419,44 @@ func (s *SQLStore) ForEachChannel(ctx context.Context, break } + // Collect the channel & policy IDs that we want to + // do a batch collection for. + var ( + channelIDs = make([]int64, len(rows)) + policyIDs = make([]int64, 0, len(rows)*2) + ) + for i, row := range rows { + channelIDs[i] = row.GraphChannel.ID + + // Extract policy IDs from the row + dbPol1, dbPol2, err := extractChannelPolicies( + row, + ) + if err != nil { + return fmt.Errorf("unable to extract "+ + "channel policies: %w", err) + } + + if dbPol1 != nil { + policyIDs = append(policyIDs, dbPol1.ID) + } + + if dbPol2 != nil { + policyIDs = append(policyIDs, dbPol2.ID) + } + } + + batchData, err := batchLoadChannelData( + ctx, s.cfg.PaginationCfg, db, channelIDs, + policyIDs, + ) + if err != nil { + return fmt.Errorf("unable to batch load "+ + "channel data: %w", err) + } + for _, row := range rows { - err := handleChannel(db, row) + err := handleChannel(db, batchData, row) if err != nil { return err } @@ -1756,8 +1794,8 @@ func (s *SQLStore) DeleteChannelEdges(strictZombiePruning, markZombie bool, } info, err := getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1, node2, + ctx, db, s.cfg.ChainHash, row.GraphChannel, + node1, node2, ) if err != nil { return err @@ -1909,8 +1947,8 @@ func (s *SQLStore) FetchChannelEdgesByID(chanID uint64) ( } edge, err = getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1, node2, + ctx, db, s.cfg.ChainHash, row.GraphChannel, node1, + node2, ) if err != nil { return fmt.Errorf("unable to build channel info: %w", @@ -1981,8 +2019,8 @@ func (s *SQLStore) FetchChannelEdgesByOutpoint(op *wire.OutPoint) ( } edge, err = getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1, node2, + ctx, db, s.cfg.ChainHash, row.GraphChannel, node1, + node2, ) if err != nil { return fmt.Errorf("unable to build channel info: %w", @@ -2215,9 +2253,8 @@ func (s *SQLStore) FetchChanInfos(chanIDs []uint64) ([]ChannelEdge, error) { } edge, err := getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1.PubKeyBytes, - node2.PubKeyBytes, + ctx, db, s.cfg.ChainHash, row.GraphChannel, + node1.PubKeyBytes, node2.PubKeyBytes, ) if err != nil { return fmt.Errorf("unable to build "+ @@ -2488,8 +2525,8 @@ func (s *SQLStore) PruneGraph(spentOutputs []*wire.OutPoint, } info, err := getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1, node2, + ctx, db, s.cfg.ChainHash, row.GraphChannel, + node1, node2, ) if err != nil { return err @@ -2787,8 +2824,8 @@ func (s *SQLStore) DisconnectBlockAtHeight(height uint32) ( } channel, err := getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel.ID, - row.GraphChannel, node1, node2, + ctx, db, s.cfg.ChainHash, row.GraphChannel, + node1, node2, ) if err != nil { return err @@ -3174,8 +3211,7 @@ func forEachNodeChannel(ctx context.Context, db SQLQueries, } edge, err := getAndBuildEdgeInfo( - ctx, db, chain, row.GraphChannel.ID, row.GraphChannel, - node1, node2, + ctx, db, chain, row.GraphChannel, node1, node2, ) if err != nil { return fmt.Errorf("unable to build channel info: %w", @@ -4085,19 +4121,47 @@ func upsertChanPolicyExtraSignedFields(ctx context.Context, db SQLQueries, // provided dbChanRow and also fetches any other required information // to construct the edge info. func getAndBuildEdgeInfo(ctx context.Context, db SQLQueries, - chain chainhash.Hash, dbChanID int64, dbChan sqlc.GraphChannel, node1, + chain chainhash.Hash, dbChan sqlc.GraphChannel, node1, node2 route.Vertex) (*models.ChannelEdgeInfo, error) { + // NOTE: getAndBuildEdgeInfo is only used to load the data for a single + // 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) @@ -4166,46 +4230,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 @@ -4219,58 +4243,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 @@ -4901,3 +4905,208 @@ func batchLoadNodeExtraTypesHelper(ctx context.Context, callback, ) } + +// buildChanPoliciesWithBatchData builds two models.ChannelEdgePolicy instances +// from the provided sqlc.GraphChannelPolicy records and the +// provided batchChannelData. +func buildChanPoliciesWithBatchData(dbPol1, dbPol2 *sqlc.GraphChannelPolicy, + channelID uint64, node1, node2 route.Vertex, + batchData *batchChannelData) (*models.ChannelEdgePolicy, + *models.ChannelEdgePolicy, error) { + + pol1, err := buildChanPolicyWithBatchData( + dbPol1, channelID, node2, batchData, + ) + if err != nil { + return nil, nil, fmt.Errorf("unable to build policy1: %w", err) + } + + pol2, err := buildChanPolicyWithBatchData( + dbPol2, channelID, node1, batchData, + ) + if err != nil { + return nil, nil, fmt.Errorf("unable to build policy2: %w", err) + } + + return pol1, pol2, nil +} + +// buildChanPolicyWithBatchData builds a models.ChannelEdgePolicy instance from +// the provided sqlc.GraphChannelPolicy and the provided batchChannelData. +func buildChanPolicyWithBatchData(dbPol *sqlc.GraphChannelPolicy, + 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. + chanfeatures map[int64][]int + + // chanExtras is a map from DB channel ID to a map of TLV type to + // extra signed field bytes. + chanExtraTypes map[int64]map[uint64][]byte + + // policyExtras is a map from DB channel policy ID to a map of TLV type + // to extra signed field bytes. + policyExtras map[int64]map[uint64][]byte +} + +// batchLoadChannelData loads all related data for batches of channels and +// policies. +func batchLoadChannelData(ctx context.Context, cfg *sqldb.PagedQueryConfig, + db SQLQueries, channelIDs []int64, + policyIDs []int64) (*batchChannelData, error) { + + batchData := &batchChannelData{ + chanfeatures: make(map[int64][]int), + chanExtraTypes: make(map[int64]map[uint64][]byte), + policyExtras: make(map[int64]map[uint64][]byte), + } + + // Batch load channel features and extras + var err error + if len(channelIDs) > 0 { + batchData.chanfeatures, err = batchLoadChannelFeaturesHelper( + ctx, cfg, db, channelIDs, + ) + if err != nil { + return nil, fmt.Errorf("unable to batch load "+ + "channel features: %w", err) + } + + batchData.chanExtraTypes, err = batchLoadChannelExtrasHelper( + ctx, cfg, db, channelIDs, + ) + if err != nil { + return nil, fmt.Errorf("unable to batch load "+ + "channel extras: %w", err) + } + } + + if len(policyIDs) > 0 { + policyExtras, err := batchLoadChannelPolicyExtrasHelper( + ctx, cfg, db, policyIDs, + ) + if err != nil { + return nil, fmt.Errorf("unable to batch load "+ + "policy extras: %w", err) + } + batchData.policyExtras = policyExtras + } + + return batchData, nil +} + +// batchLoadChannelFeaturesHelper loads channel features for a batch of +// channel IDs using ExecutePagedQuery wrapper around the +// GetChannelFeaturesBatch query. It returns a map from DB channel ID to a +// slice of feature bits. +func batchLoadChannelFeaturesHelper(ctx context.Context, + cfg *sqldb.PagedQueryConfig, db SQLQueries, + channelIDs []int64) (map[int64][]int, error) { + + features := make(map[int64][]int) + + return features, sqldb.ExecutePagedQuery( + ctx, cfg, channelIDs, + func(id int64) int64 { + return id + }, + func(ctx context.Context, + ids []int64) ([]sqlc.GraphChannelFeature, error) { + + return db.GetChannelFeaturesBatch(ctx, ids) + }, + func(ctx context.Context, + feature sqlc.GraphChannelFeature) error { + + features[feature.ChannelID] = append( + features[feature.ChannelID], + int(feature.FeatureBit), + ) + + return nil + }, + ) +} + +// batchLoadChannelExtrasHelper loads channel extra types for a batch of +// channel IDs using ExecutePagedQuery wrapper around the GetChannelExtrasBatch +// query. It returns a map from DB channel ID to a map of TLV type to extra +// signed field bytes. +func batchLoadChannelExtrasHelper(ctx context.Context, + cfg *sqldb.PagedQueryConfig, db SQLQueries, + channelIDs []int64) (map[int64]map[uint64][]byte, error) { + + extras := make(map[int64]map[uint64][]byte) + + cb := func(ctx context.Context, + extra sqlc.GraphChannelExtraType) error { + + if extras[extra.ChannelID] == nil { + extras[extra.ChannelID] = make(map[uint64][]byte) + } + extras[extra.ChannelID][uint64(extra.Type)] = extra.Value + + return nil + } + + return extras, sqldb.ExecutePagedQuery( + ctx, cfg, channelIDs, + func(id int64) int64 { + return id + }, + func(ctx context.Context, + ids []int64) ([]sqlc.GraphChannelExtraType, error) { + + return db.GetChannelExtrasBatch(ctx, ids) + }, cb, + ) +} + +// batchLoadChannelPolicyExtrasHelper loads channel policy extra types for a +// batch of policy IDs using ExecutePagedQuery wrapper around the +// GetChannelPolicyExtraTypesBatch query. It returns a map from DB policy ID to +// a map of TLV type to extra signed field bytes. +func batchLoadChannelPolicyExtrasHelper(ctx context.Context, + cfg *sqldb.PagedQueryConfig, db SQLQueries, + policyIDs []int64) (map[int64]map[uint64][]byte, error) { + + extras := make(map[int64]map[uint64][]byte) + + return extras, sqldb.ExecutePagedQuery( + ctx, cfg, policyIDs, + func(id int64) int64 { + return id + }, + func(ctx context.Context, ids []int64) ( + []sqlc.GetChannelPolicyExtraTypesBatchRow, error) { + + return db.GetChannelPolicyExtraTypesBatch(ctx, ids) + }, + func(ctx context.Context, + row sqlc.GetChannelPolicyExtraTypesBatchRow) error { + + if extras[row.PolicyID] == nil { + extras[row.PolicyID] = make(map[uint64][]byte) + } + extras[row.PolicyID][uint64(row.Type)] = row.Value + + return nil + }, + ) +} diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 3122d1dab..fa6d50803 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -701,52 +701,78 @@ func (q *Queries) GetChannelBySCIDWithPolicies(ctx context.Context, arg GetChann return i, err } -const getChannelFeaturesAndExtras = `-- name: GetChannelFeaturesAndExtras :many +const getChannelExtrasBatch = `-- name: GetChannelExtrasBatch :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 + channel_id, + type, + value +FROM graph_channel_extra_types +WHERE channel_id IN (/*SLICE:chan_ids*/?) +ORDER BY channel_id, type ` -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) +func (q *Queries) GetChannelExtrasBatch(ctx context.Context, chanIds []int64) ([]GraphChannelExtraType, error) { + query := getChannelExtrasBatch + var queryParams []interface{} + if len(chanIds) > 0 { + for _, v := range chanIds { + queryParams = append(queryParams, v) + } + query = strings.Replace(query, "/*SLICE:chan_ids*/?", makeQueryParams(len(queryParams), len(chanIds)), 1) + } else { + query = strings.Replace(query, "/*SLICE:chan_ids*/?", "NULL", 1) + } + rows, err := q.db.QueryContext(ctx, query, queryParams...) if err != nil { return nil, err } defer rows.Close() - var items []GetChannelFeaturesAndExtrasRow + var items []GraphChannelExtraType for rows.Next() { - var i GetChannelFeaturesAndExtrasRow - if err := rows.Scan( - &i.ChannelID, - &i.IsFeature, - &i.FeatureBit, - &i.ExtraKey, - &i.Value, - ); err != nil { + var i GraphChannelExtraType + if err := rows.Scan(&i.ChannelID, &i.Type, &i.Value); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getChannelFeaturesBatch = `-- name: GetChannelFeaturesBatch :many +SELECT + channel_id, + feature_bit +FROM graph_channel_features +WHERE channel_id IN (/*SLICE:chan_ids*/?) +ORDER BY channel_id, feature_bit +` + +func (q *Queries) GetChannelFeaturesBatch(ctx context.Context, chanIds []int64) ([]GraphChannelFeature, error) { + query := getChannelFeaturesBatch + var queryParams []interface{} + if len(chanIds) > 0 { + for _, v := range chanIds { + queryParams = append(queryParams, v) + } + query = strings.Replace(query, "/*SLICE:chan_ids*/?", makeQueryParams(len(queryParams), len(chanIds)), 1) + } else { + query = strings.Replace(query, "/*SLICE:chan_ids*/?", "NULL", 1) + } + rows, err := q.db.QueryContext(ctx, query, queryParams...) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphChannelFeature + for rows.Next() { + var i GraphChannelFeature + if err := rows.Scan(&i.ChannelID, &i.FeatureBit); err != nil { return nil, err } items = append(items, i) @@ -798,48 +824,42 @@ func (q *Queries) GetChannelPolicyByChannelAndNode(ctx context.Context, arg GetC return i, err } -const getChannelPolicyExtraTypes = `-- name: GetChannelPolicyExtraTypes :many +const getChannelPolicyExtraTypesBatch = `-- name: GetChannelPolicyExtraTypesBatch :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 + channel_policy_id as policy_id, + type, + value +FROM graph_channel_policy_extra_types +WHERE channel_policy_id IN (/*SLICE:policy_ids*/?) +ORDER BY channel_policy_id, type ` -type GetChannelPolicyExtraTypesParams struct { - ID int64 - ID_2 int64 +type GetChannelPolicyExtraTypesBatchRow struct { + PolicyID int64 + Type int64 + Value []byte } -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) +func (q *Queries) GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds []int64) ([]GetChannelPolicyExtraTypesBatchRow, error) { + query := getChannelPolicyExtraTypesBatch + var queryParams []interface{} + if len(policyIds) > 0 { + for _, v := range policyIds { + queryParams = append(queryParams, v) + } + query = strings.Replace(query, "/*SLICE:policy_ids*/?", makeQueryParams(len(queryParams), len(policyIds)), 1) + } else { + query = strings.Replace(query, "/*SLICE:policy_ids*/?", "NULL", 1) + } + rows, err := q.db.QueryContext(ctx, query, queryParams...) if err != nil { return nil, err } defer rows.Close() - var items []GetChannelPolicyExtraTypesRow + var items []GetChannelPolicyExtraTypesBatchRow for rows.Next() { - var i GetChannelPolicyExtraTypesRow - if err := rows.Scan( - &i.PolicyID, - &i.ChannelID, - &i.NodeID, - &i.Type, - &i.Value, - ); err != nil { + var i GetChannelPolicyExtraTypesBatchRow + if err := rows.Scan(&i.PolicyID, &i.Type, &i.Value); err != nil { return nil, err } items = append(items, i) diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index e32383032..f7fad44ab 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -38,9 +38,10 @@ type Querier interface { GetChannelByOutpointWithPolicies(ctx context.Context, arg GetChannelByOutpointWithPoliciesParams) (GetChannelByOutpointWithPoliciesRow, error) GetChannelBySCID(ctx context.Context, arg GetChannelBySCIDParams) (GraphChannel, error) GetChannelBySCIDWithPolicies(ctx context.Context, arg GetChannelBySCIDWithPoliciesParams) (GetChannelBySCIDWithPoliciesRow, error) - GetChannelFeaturesAndExtras(ctx context.Context, channelID int64) ([]GetChannelFeaturesAndExtrasRow, error) + GetChannelExtrasBatch(ctx context.Context, chanIds []int64) ([]GraphChannelExtraType, error) + GetChannelFeaturesBatch(ctx context.Context, chanIds []int64) ([]GraphChannelFeature, error) GetChannelPolicyByChannelAndNode(ctx context.Context, arg GetChannelPolicyByChannelAndNodeParams) (GraphChannelPolicy, error) - GetChannelPolicyExtraTypes(ctx context.Context, arg GetChannelPolicyExtraTypesParams) ([]GetChannelPolicyExtraTypesRow, error) + GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds []int64) ([]GetChannelPolicyExtraTypesBatchRow, error) GetChannelsByOutpoints(ctx context.Context, outpoints []string) ([]GetChannelsByOutpointsRow, error) GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg GetChannelsByPolicyLastUpdateRangeParams) ([]GetChannelsByPolicyLastUpdateRangeRow, error) GetChannelsBySCIDRange(ctx context.Context, arg GetChannelsBySCIDRangeParams) ([]GetChannelsBySCIDRangeRow, error) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index afa446f0b..ae4979413 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -275,27 +275,6 @@ FROM graph_channels c WHERE c.scid = $1 AND c.version = $2; --- name: GetChannelFeaturesAndExtras :many -SELECT - cf.channel_id, - true AS is_feature, - cf.feature_bit AS feature_bit, - NULL AS extra_key, - NULL AS value -FROM graph_channel_features cf -WHERE cf.channel_id = $1 - -UNION ALL - -SELECT - cet.channel_id, - false AS is_feature, - 0 AS feature_bit, - cet.type AS extra_key, - cet.value AS value -FROM graph_channel_extra_types cet -WHERE cet.channel_id = $1; - -- name: GetSCIDByOutpoint :one SELECT scid from graph_channels WHERE outpoint = $1 AND version = $2; @@ -647,6 +626,14 @@ INSERT INTO graph_channel_features ( $1, $2 ); +-- name: GetChannelFeaturesBatch :many +SELECT + channel_id, + feature_bit +FROM graph_channel_features +WHERE channel_id IN (sqlc.slice('chan_ids')/*SLICE:chan_ids*/) +ORDER BY channel_id, feature_bit; + /* ───────────────────────────────────────────── graph_channel_extra_types table queries ───────────────────────────────────────────── @@ -658,6 +645,15 @@ INSERT INTO graph_channel_extra_types ( ) VALUES ($1, $2, $3); +-- name: GetChannelExtrasBatch :many +SELECT + channel_id, + type, + value +FROM graph_channel_extra_types +WHERE channel_id IN (sqlc.slice('chan_ids')/*SLICE:chan_ids*/) +ORDER BY channel_id, type; + /* ───────────────────────────────────────────── graph_channel_policies table queries ───────────────────────────────────────────── @@ -760,17 +756,14 @@ INSERT INTO graph_channel_policy_extra_types ( ) VALUES ($1, $2, $3); --- name: GetChannelPolicyExtraTypes :many +-- name: GetChannelPolicyExtraTypesBatch :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; + channel_policy_id as policy_id, + type, + value +FROM graph_channel_policy_extra_types +WHERE channel_policy_id IN (sqlc.slice('policy_ids')/*SLICE:policy_ids*/) +ORDER BY channel_policy_id, type; -- name: GetV1DisabledSCIDs :many SELECT c.scid