Merge pull request #10148 from ellemouton/graphPerf10

graph/db+sqldb: different defaults for SQLite and Postgres query options

config_builder.go

@@ -1128,9 +1128,8 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
 					continue
 				}
 
-				migFn, ok := getSQLMigration(
+				migFn, ok := d.getSQLMigration(
 					ctx, version, dbs.ChanStateDB.Backend,
-					*d.cfg.ActiveNetParams.GenesisHash,
 				)
 				if !ok {
 					continue

config_prod.go

@@ -5,7 +5,6 @@ package lnd
 import (
 	"context"
 
-	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	graphdb "github.com/lightningnetwork/lnd/graph/db"
 	"github.com/lightningnetwork/lnd/kvdb"
 	"github.com/lightningnetwork/lnd/sqldb"
@@ -30,9 +29,9 @@ func (d *DefaultDatabaseBuilder) getGraphStore(_ *sqldb.BaseDB,
 // NOTE: this is a no-op for the production build since all migrations that are
 // in production will also be in development builds, and so they are not
 // defined behind a build tag.
-func getSQLMigration(ctx context.Context, version int,
+func (d *DefaultDatabaseBuilder) getSQLMigration(ctx context.Context,
-	kvBackend kvdb.Backend,
+	version int, kvBackend kvdb.Backend) (func(tx *sqlc.Queries) error,
-	chain chainhash.Hash) (func(tx *sqlc.Queries) error, bool) {
+	bool) {
 
 	return nil, false
 }

config_test_native_sql.go

@@ -7,9 +7,9 @@ import (
 	"database/sql"
 	"fmt"
 
-	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	graphdb "github.com/lightningnetwork/lnd/graph/db"
 	"github.com/lightningnetwork/lnd/kvdb"
+	"github.com/lightningnetwork/lnd/lncfg"
 	"github.com/lightningnetwork/lnd/sqldb"
 	"github.com/lightningnetwork/lnd/sqldb/sqlc"
 )
@@ -30,10 +30,15 @@ func (d *DefaultDatabaseBuilder) getGraphStore(baseDB *sqldb.BaseDB,
 		},
 	)
 
+	queryConfig := d.cfg.DB.Sqlite.QueryConfig
+	if d.cfg.DB.Backend == lncfg.PostgresBackend {
+		queryConfig = d.cfg.DB.Postgres.QueryConfig
+	}
+
 	return graphdb.NewSQLStore(
 		&graphdb.SQLStoreConfig{
 			ChainHash: *d.cfg.ActiveNetParams.GenesisHash,
-			QueryCfg:  sqldb.DefaultQueryConfig(),
+			QueryCfg:  &queryConfig,
 		},
 		graphExecutor, opts...,
 	)
@@ -44,15 +49,23 @@ func (d *DefaultDatabaseBuilder) getGraphStore(baseDB *sqldb.BaseDB,
 const graphSQLMigration = 9
 
 // getSQLMigration returns a migration function for the given version.
-func getSQLMigration(ctx context.Context, version int,
+func (d *DefaultDatabaseBuilder) getSQLMigration(ctx context.Context,
-	kvBackend kvdb.Backend,
+	version int, kvBackend kvdb.Backend) (func(tx *sqlc.Queries) error,
-	chain chainhash.Hash) (func(tx *sqlc.Queries) error, bool) {
+	bool) {
+
+	cfg := &graphdb.SQLStoreConfig{
+		ChainHash: *d.cfg.ActiveNetParams.GenesisHash,
+		QueryCfg:  &d.cfg.DB.Sqlite.QueryConfig,
+	}
+	if d.cfg.DB.Backend == lncfg.PostgresBackend {
+		cfg.QueryCfg = &d.cfg.DB.Postgres.QueryConfig
+	}
 
 	switch version {
 	case graphSQLMigration:
 		return func(tx *sqlc.Queries) error {
 			err := graphdb.MigrateGraphToSQL(
-				ctx, kvBackend, tx, chain,
+				ctx, cfg, kvBackend, tx,
 			)
 			if err != nil {
 				return fmt.Errorf("failed to migrate graph "+

graph/builder.go

@@ -1037,9 +1037,6 @@ func (b *Builder) AddEdge(ctx context.Context, edge *models.ChannelEdgeInfo,
 // Chain View is updated with the new edge if it is successfully added to the
 // graph. We only persist the channel if we currently dont have it at all in
 // our graph.
-//
-// TODO(elle): this currently also does funding-transaction validation. But this
-// should be moved to the gossiper instead.
 func (b *Builder) addEdge(ctx context.Context, edge *models.ChannelEdgeInfo,
 	op ...batch.SchedulerOption) error {
 

graph/db/benchmark_test.go

@@ -60,10 +60,6 @@ var (
 		WithBatchCommitInterval(500 * time.Millisecond),
 	}
 
-	// testSQLPaginationCfg is used to configure the pagination settings for
-	// the SQL stores we open for testing.
-	testSQLPaginationCfg = sqldb.DefaultQueryConfig()
-
 	// testSqlitePragmaOpts is a set of SQLite pragma options that we apply
 	// to the SQLite databases we open for testing.
 	testSqlitePragmaOpts = []string{
@@ -111,7 +107,8 @@ var (
 		name: "native-sqlite",
 		open: func(b testing.TB) V1Store {
 			return connectNativeSQLite(
-				b, nativeSQLSqlitePath, nativeSQLSqliteFile,
+				b, sqldb.DefaultSQLiteConfig(),
+				nativeSQLSqlitePath, nativeSQLSqliteFile,
 			)
 		},
 	}
@@ -130,15 +127,20 @@ var (
 	nativeSQLPostgresConn = dbConnection{
 		name: "native-postgres",
 		open: func(b testing.TB) V1Store {
-			return connectNativePostgres(b, nativeSQLPostgresDNS)
+			return connectNativePostgres(
+				b, sqldb.DefaultPostgresConfig(),
+				nativeSQLPostgresDNS,
+			)
 		},
 	}
 )
 
 // connectNativePostgres creates a V1Store instance backed by a native Postgres
 // database for testing purposes.
-func connectNativePostgres(t testing.TB, dsn string) V1Store {
+func connectNativePostgres(t testing.TB, cfg *sqldb.QueryConfig,
-	return newSQLStore(t, sqlPostgres(t, dsn))
+	dsn string) V1Store {
+
+	return newSQLStore(t, cfg, sqlPostgres(t, dsn))
 }
 
 // sqlPostgres creates a sqldb.DB instance backed by a native Postgres database
@@ -158,8 +160,10 @@ func sqlPostgres(t testing.TB, dsn string) BatchedSQLQueries {
 
 // connectNativeSQLite creates a V1Store instance backed by a native SQLite
 // database for testing purposes.
-func connectNativeSQLite(t testing.TB, dbPath, file string) V1Store {
+func connectNativeSQLite(t testing.TB, cfg *sqldb.QueryConfig, dbPath,
-	return newSQLStore(t, sqlSQLite(t, dbPath, file))
+	file string) V1Store {
+
+	return newSQLStore(t, cfg, sqlSQLite(t, dbPath, file))
 }
 
 // sqlSQLite creates a sqldb.DB instance backed by a native SQLite database for
@@ -277,11 +281,13 @@ func newSQLExecutor(t testing.TB, db sqldb.DB) BatchedSQLQueries {
 
 // newSQLStore creates a new SQLStore instance for testing using a provided
 // sqldb.DB instance.
-func newSQLStore(t testing.TB, db BatchedSQLQueries) V1Store {
+func newSQLStore(t testing.TB, cfg *sqldb.QueryConfig,
+	db BatchedSQLQueries) V1Store {
+
 	store, err := NewSQLStore(
 		&SQLStoreConfig{
 			ChainHash: dbTestChain,
-			QueryCfg:  testSQLPaginationCfg,
+			QueryCfg:  cfg,
 		},
 		db, testStoreOptions...,
 	)
@@ -443,10 +449,12 @@ func TestPopulateViaMigration(t *testing.T) {
 	log.SetLevel(btclog.LevelDebug)
 
 	var (
+		cfg     = sqldb.DefaultSQLiteConfig()
 		srcKVDB = kvdbSqlite(t, kvdbSqlitePath, kvdbSqliteFile)
 		dstSQL  = sqlSQLite(t, nativeSQLSqlitePath, nativeSQLSqliteFile)
 	)
 	if testPostgres {
+		cfg = sqldb.DefaultPostgresConfig()
 		srcKVDB = kvdbPostgres(t, kvdbPostgresDNS)
 		dstSQL = sqlPostgres(t, nativeSQLPostgresDNS)
 	}
@@ -456,7 +464,10 @@ func TestPopulateViaMigration(t *testing.T) {
 	err := dstSQL.ExecTx(
 		ctx, sqldb.WriteTxOpt(), func(queries SQLQueries) error {
 			return MigrateGraphToSQL(
-				ctx, srcKVDB, queries, dbTestChain,
+				ctx, &SQLStoreConfig{
+					QueryCfg:  cfg,
+					ChainHash: dbTestChain,
+				}, srcKVDB, queries,
 			)
 		}, func() {},
 	)
@@ -763,3 +774,111 @@ func BenchmarkGraphReadMethods(b *testing.B) {
 		}
 	}
 }
+
+// BenchmarkFindOptimalSQLQueryConfig uses the ForEachNode and ForEachChannel
+// methods to find the optimal maximum sqldb QueryConfig values for a given
+// database backend. This is useful for determining the best default values for
+// each backend. The ForEachNode and ForEachChannel methods are used since
+// they make use of both batching and pagination.
+func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) {
+	// NOTE: Set this to true if you want to test with a postgres backend.
+	testPostgres := false
+
+	// NOTE: Set this to true if you want to test with various batch sizes.
+	// By default, page sizes will be tested.
+	testBatching := false
+
+	// Set the various page sizes we want to test.
+	//
+	// NOTE: these are the sqlite paging testing values.
+	testSizes := []int{20, 50, 100, 150, 500}
+
+	configOption := "MaxPageSize"
+	if testBatching {
+		configOption = "MaxBatchSize"
+
+		testSizes = []int{
+			50, 100, 150, 200, 250, 300, 350,
+		}
+	}
+
+	dbName := "sqlite"
+	if testPostgres {
+		dbName = "postgres"
+
+		// Set the various page sizes we want to test.
+		//
+		// NOTE: these are the postgres paging values.
+		testSizes = []int{5000, 7000, 10000, 12000}
+
+		if testBatching {
+			testSizes = []int{
+				1000, 2000, 5000, 7000, 10000,
+			}
+		}
+	}
+
+	for _, size := range testSizes {
+		b.Run(fmt.Sprintf("%s-%s-%d", configOption, dbName, size),
+			func(b *testing.B) {
+				ctx := context.Background()
+
+				cfg := sqldb.DefaultSQLiteConfig()
+				if testPostgres {
+					cfg = sqldb.DefaultPostgresConfig()
+				}
+
+				if testBatching {
+					cfg.MaxBatchSize = size
+				} else {
+					cfg.MaxPageSize = int32(size)
+				}
+
+				store := connectNativeSQLite(
+					b, cfg, nativeSQLSqlitePath,
+					nativeSQLSqliteFile,
+				)
+				if testPostgres {
+					store = connectNativePostgres(
+						b, cfg, nativeSQLPostgresDNS,
+					)
+				}
+
+				// Reset timer to exclude setup time.
+				b.ResetTimer()
+
+				for i := 0; i < b.N; i++ {
+					var (
+						numNodes    = 0
+						numChannels = 0
+					)
+
+					//nolint:ll
+					err := store.ForEachNode(
+						ctx,
+						func(_ *models.LightningNode) error {
+							numNodes++
+
+							return nil
+						}, func() {},
+					)
+					require.NoError(b, err)
+
+					//nolint:ll
+					err = store.ForEachChannel(
+						ctx,
+						func(_ *models.ChannelEdgeInfo,
+							_,
+							_ *models.ChannelEdgePolicy) error {
+
+							numChannels++
+
+							return nil
+						}, func() {},
+					)
+					require.NoError(b, err)
+				}
+			},
+		)
+	}
+}

graph/db/sql_migration.go

@@ -26,8 +26,8 @@ import (
 // NOTE: this is currently not called from any code path. It is called via tests
 // only for now and will be called from the main lnd binary once the
 // migration is fully implemented and tested.
-func MigrateGraphToSQL(ctx context.Context, kvBackend kvdb.Backend,
+func MigrateGraphToSQL(ctx context.Context, cfg *SQLStoreConfig,
-	sqlDB SQLQueries, chain chainhash.Hash) error {
+	kvBackend kvdb.Backend, sqlDB SQLQueries) error {
 
 	log.Infof("Starting migration of the graph store from KV to SQL")
 	t0 := time.Now()
@@ -43,7 +43,8 @@ func MigrateGraphToSQL(ctx context.Context, kvBackend kvdb.Backend,
 	}
 
 	// 1) Migrate all the nodes.
-	if err := migrateNodes(ctx, kvBackend, sqlDB); err != nil {
+	err = migrateNodes(ctx, cfg.QueryCfg, kvBackend, sqlDB)
+	if err != nil {
 		return fmt.Errorf("could not migrate nodes: %w", err)
 	}
 
@@ -53,7 +54,7 @@ func MigrateGraphToSQL(ctx context.Context, kvBackend kvdb.Backend,
 	}
 
 	// 3) Migrate all the channels and channel policies.
-	err = migrateChannelsAndPolicies(ctx, kvBackend, sqlDB, chain)
+	err = migrateChannelsAndPolicies(ctx, cfg, kvBackend, sqlDB)
 	if err != nil {
 		return fmt.Errorf("could not migrate channels and policies: %w",
 			err)
@@ -108,8 +109,8 @@ func checkGraphExists(db kvdb.Backend) (bool, error) {
 // migrateNodes migrates all nodes from the KV backend to the SQL database.
 // This includes doing a sanity check after each migration to ensure that the
 // migrated node matches the original node.
-func migrateNodes(ctx context.Context, kvBackend kvdb.Backend,
+func migrateNodes(ctx context.Context, cfg *sqldb.QueryConfig,
-	sqlDB SQLQueries) error {
+	kvBackend kvdb.Backend, sqlDB SQLQueries) error {
 
 	// Keep track of the number of nodes migrated and the number of
 	// nodes skipped due to errors.
@@ -192,7 +193,7 @@ func migrateNodes(ctx context.Context, kvBackend kvdb.Backend,
 				pub, id, dbNode.ID)
 		}
 
-		migratedNode, err := buildNode(ctx, sqlDB, dbNode)
+		migratedNode, err := buildNode(ctx, cfg, sqlDB, dbNode)
 		if err != nil {
 			return fmt.Errorf("could not build migrated node "+
 				"from dbNode(db id: %d, node pub: %x): %w",
@@ -320,8 +321,8 @@ func migrateSourceNode(ctx context.Context, kvdb kvdb.Backend,
 
 // migrateChannelsAndPolicies migrates all channels and their policies
 // from the KV backend to the SQL database.
-func migrateChannelsAndPolicies(ctx context.Context, kvBackend kvdb.Backend,
+func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig,
-	sqlDB SQLQueries, chain chainhash.Hash) error {
+	kvBackend kvdb.Backend, sqlDB SQLQueries) error {
 
 	var (
 		channelCount       uint64
@@ -373,9 +374,10 @@ func migrateChannelsAndPolicies(ctx context.Context, kvBackend kvdb.Backend,
 		// info, but rather rely on the chain hash LND is running with.
 		// So this is our way of ensuring that LND is running on the
 		// correct network at migration time.
-		if channel.ChainHash != chain {
+		if channel.ChainHash != cfg.ChainHash {
 			return fmt.Errorf("channel %d has chain hash %s, "+
-				"expected %s", scid, channel.ChainHash, chain)
+				"expected %s", scid, channel.ChainHash,
+				cfg.ChainHash)
 		}
 
 		// Sanity check to ensure that the channel has valid extra
@@ -413,7 +415,8 @@ func migrateChannelsAndPolicies(ctx context.Context, kvBackend kvdb.Backend,
 		chunk++
 
 		err = migrateSingleChannel(
-			ctx, sqlDB, channel, policy1, policy2, migChanPolicy,
+			ctx, cfg, sqlDB, channel, policy1, policy2,
+			migChanPolicy,
 		)
 		if err != nil {
 			return fmt.Errorf("could not migrate channel %d: %w",
@@ -448,8 +451,8 @@ func migrateChannelsAndPolicies(ctx context.Context, kvBackend kvdb.Backend,
 	return nil
 }
 
-func migrateSingleChannel(ctx context.Context, sqlDB SQLQueries,
+func migrateSingleChannel(ctx context.Context, cfg *SQLStoreConfig,
-	channel *models.ChannelEdgeInfo,
+	sqlDB SQLQueries, channel *models.ChannelEdgeInfo,
 	policy1, policy2 *models.ChannelEdgePolicy,
 	migChanPolicy func(*models.ChannelEdgePolicy) error) error {
 
@@ -508,7 +511,7 @@ func migrateSingleChannel(ctx context.Context, sqlDB SQLQueries,
 	}
 
 	migChan, migPol1, migPol2, err := getAndBuildChanAndPolicies(
-		ctx, sqlDB, row, channel.ChainHash,
+		ctx, cfg, sqlDB, row,
 	)
 	if err != nil {
 		return fmt.Errorf("could not build migrated channel and "+
@@ -603,6 +606,7 @@ func migratePruneLog(ctx context.Context, kvBackend kvdb.Backend,
 		hash *chainhash.Hash) error {
 
 		count++
+		chunk++
 
 		// Keep track of the prune tip height and hash.
 		if height > pruneTipHeight {
@@ -702,9 +706,9 @@ func migratePruneLog(ctx context.Context, kvBackend kvdb.Backend,
 // getAndBuildChanAndPolicies is a helper that builds the channel edge info
 // and policies from the given row returned by the SQL query
 // GetChannelBySCIDWithPolicies.
-func getAndBuildChanAndPolicies(ctx context.Context, db SQLQueries,
+func getAndBuildChanAndPolicies(ctx context.Context, cfg *SQLStoreConfig,
-	row sqlc.GetChannelBySCIDWithPoliciesRow,
+	db SQLQueries,
-	chain chainhash.Hash) (*models.ChannelEdgeInfo,
+	row sqlc.GetChannelBySCIDWithPoliciesRow) (*models.ChannelEdgeInfo,
 	*models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) {
 
 	node1, node2, err := buildNodeVertices(
@@ -715,7 +719,7 @@ func getAndBuildChanAndPolicies(ctx context.Context, db SQLQueries,
 	}
 
 	edge, err := getAndBuildEdgeInfo(
-		ctx, db, chain, row.GraphChannel, node1, node2,
+		ctx, cfg, db, row.GraphChannel, node1, node2,
 	)
 	if err != nil {
 		return nil, nil, nil, fmt.Errorf("unable to build channel "+
@@ -729,7 +733,8 @@ func getAndBuildChanAndPolicies(ctx context.Context, db SQLQueries,
 	}
 
 	policy1, policy2, err := getAndBuildChanPolicies(
-		ctx, db, dbPol1, dbPol2, edge.ChannelID, node1, node2,
+		ctx, cfg.QueryCfg, db, dbPol1, dbPol2, edge.ChannelID, node1,
+		node2,
 	)
 	if err != nil {
 		return nil, nil, nil, fmt.Errorf("unable to build channel "+
@@ -786,6 +791,7 @@ func migrateClosedSCIDIndex(ctx context.Context, kvBackend kvdb.Backend,
 	)
 	migrateSingleClosedSCID := func(scid lnwire.ShortChannelID) error {
 		count++
+		chunk++
 
 		chanIDB := channelIDToBytes(scid.ToUint64())
 		err := sqlDB.InsertClosedChannel(ctx, chanIDB)
@@ -874,6 +880,7 @@ func migrateZombieIndex(ctx context.Context, kvBackend kvdb.Backend,
 		}
 
 		count++
+		chunk++
 
 		err = sqlDB.UpsertZombieChannel(
 			ctx, sqlc.UpsertZombieChannelParams{

graph/db/sql_migration_test.go

@@ -333,9 +333,7 @@ func TestMigrateGraphToSQL(t *testing.T) {
 			require.True(t, ok)
 
 			// Run the migration.
-			err := MigrateGraphToSQL(
+			err := MigrateGraphToSQL(ctx, sql.cfg, kvDB.db, sql.db)
-				ctx, kvDB.db, sql.db, testChain,
-			)
 			require.NoError(t, err)
 
 			// Validate that the two databases are now in sync.
@@ -753,6 +751,18 @@ func TestMigrationWithChannelDB(t *testing.T) {
 	// bbolt channel.db file, set this to "channel.db".
 	const fileName = "channel.sqlite"
 
+	cfg := &SQLStoreConfig{
+		ChainHash: chain,
+		QueryCfg:  sqldb.DefaultPostgresConfig(),
+	}
+
+	// Determine if we are using a SQLite file or a Bolt DB file.
+	var isSqlite bool
+	if strings.HasSuffix(fileName, ".sqlite") {
+		isSqlite = true
+		cfg.QueryCfg = sqldb.DefaultSQLiteConfig()
+	}
+
 	// Set up logging for the test.
 	UseLogger(btclog.NewSLogger(btclog.NewDefaultHandler(os.Stdout)))
 
@@ -763,7 +773,7 @@ func TestMigrationWithChannelDB(t *testing.T) {
 		err := graphStore.ExecTx(
 			ctx, sqldb.WriteTxOpt(), func(tx SQLQueries) error {
 				return MigrateGraphToSQL(
-					ctx, kvBackend, tx, chain,
+					ctx, cfg, kvBackend, tx,
 				)
 			}, sqldb.NoOpReset,
 		)
@@ -825,12 +835,6 @@ func TestMigrationWithChannelDB(t *testing.T) {
 		},
 	}
 
-	// Determine if we are using a SQLite file or a Bolt DB file.
-	var isSqlite bool
-	if strings.HasSuffix(fileName, ".sqlite") {
-		isSqlite = true
-	}
-
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			chanDBPath := path.Join(test.dbPath, fileName)
@@ -1125,7 +1129,7 @@ func runTestMigration(t *testing.T, populateKV func(t *testing.T, db *KVStore),
 
 	// Run the migration.
 	err := MigrateGraphToSQL(
-		ctx, kvDB.db, sql.db, testChain,
+		ctx, sql.cfg, kvDB.db, sql.db,
 	)
 	require.NoError(t, err)
 
@@ -1312,7 +1316,7 @@ func testMigrateGraphToSQLRapidOnce(t *testing.T, rt *rapid.T,
 	}
 
 	// Run the migration.
-	err := MigrateGraphToSQL(ctx, kvDB.db, sql.db, testChain)
+	err := MigrateGraphToSQL(ctx, sql.cfg, kvDB.db, sql.db)
 	require.NoError(t, err)
 
 	// Create a slice of all nodes.

graph/db/sql_store.go

@@ -255,7 +255,7 @@ func (s *SQLStore) FetchLightningNode(ctx context.Context,
 	var node *models.LightningNode
 	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		var err error
-		_, node, err = getNodeByPubKey(ctx, db, pubKey)
+		_, node, err = getNodeByPubKey(ctx, s.cfg.QueryCfg, db, pubKey)
 
 		return err
 	}, sqldb.NoOpReset)
@@ -483,7 +483,7 @@ func (s *SQLStore) SourceNode(ctx context.Context) (*models.LightningNode,
 				err)
 		}
 
-		_, node, err = getNodeByPubKey(ctx, db, nodePub)
+		_, node, err = getNodeByPubKey(ctx, s.cfg.QueryCfg, db, nodePub)
 
 		return err
 	}, sqldb.NoOpReset)
@@ -779,7 +779,7 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context,
 				}
 
 				_, otherNode, err := getNodeByPubKey(
-					ctx, db, otherNodePub,
+					ctx, s.cfg.QueryCfg, db, otherNodePub,
 				)
 				if err != nil {
 					return fmt.Errorf("unable to fetch "+
@@ -1761,8 +1761,7 @@ func (s *SQLStore) FetchChannelEdgesByID(chanID uint64) (
 		}
 
 		edge, err = getAndBuildEdgeInfo(
-			ctx, db, s.cfg.ChainHash, row.GraphChannel, node1,
+			ctx, s.cfg, db, row.GraphChannel, node1, node2,
-			node2,
 		)
 		if err != nil {
 			return fmt.Errorf("unable to build channel info: %w",
@@ -1776,7 +1775,8 @@ func (s *SQLStore) FetchChannelEdgesByID(chanID uint64) (
 		}
 
 		policy1, policy2, err = getAndBuildChanPolicies(
-			ctx, db, dbPol1, dbPol2, edge.ChannelID, node1, node2,
+			ctx, s.cfg.QueryCfg, db, dbPol1, dbPol2, edge.ChannelID,
+			node1, node2,
 		)
 		if err != nil {
 			return fmt.Errorf("unable to build channel "+
@@ -1833,8 +1833,7 @@ func (s *SQLStore) FetchChannelEdgesByOutpoint(op *wire.OutPoint) (
 		}
 
 		edge, err = getAndBuildEdgeInfo(
-			ctx, db, s.cfg.ChainHash, row.GraphChannel, node1,
+			ctx, s.cfg, db, row.GraphChannel, node1, node2,
-			node2,
 		)
 		if err != nil {
 			return fmt.Errorf("unable to build channel info: %w",
@@ -1848,7 +1847,8 @@ func (s *SQLStore) FetchChannelEdgesByOutpoint(op *wire.OutPoint) (
 		}
 
 		policy1, policy2, err = getAndBuildChanPolicies(
-			ctx, db, dbPol1, dbPol2, edge.ChannelID, node1, node2,
+			ctx, s.cfg.QueryCfg, db, dbPol1, dbPol2, edge.ChannelID,
+			node1, node2,
 		)
 		if err != nil {
 			return fmt.Errorf("unable to build channel "+
@@ -3171,7 +3171,7 @@ func updateChanEdgePolicy(ctx context.Context, tx SQLQueries,
 }
 
 // getNodeByPubKey attempts to look up a target node by its public key.
-func getNodeByPubKey(ctx context.Context, db SQLQueries,
+func getNodeByPubKey(ctx context.Context, cfg *sqldb.QueryConfig, db SQLQueries,
 	pubKey route.Vertex) (int64, *models.LightningNode, error) {
 
 	dbNode, err := db.GetNodeByPubKey(
@@ -3186,7 +3186,7 @@ func getNodeByPubKey(ctx context.Context, db SQLQueries,
 		return 0, nil, fmt.Errorf("unable to fetch node: %w", err)
 	}
 
-	node, err := buildNode(ctx, db, dbNode)
+	node, err := buildNode(ctx, cfg, db, dbNode)
 	if err != nil {
 		return 0, nil, fmt.Errorf("unable to build node: %w", err)
 	}
@@ -3210,14 +3210,9 @@ func buildCacheableChannelInfo(scid []byte, capacity int64, node1Pub,
 // buildNode constructs a LightningNode instance from the given database node
 // record. The node's features, addresses and extra signed fields are also
 // fetched from the database and set on the node.
-func buildNode(ctx context.Context, db SQLQueries,
+func buildNode(ctx context.Context, cfg *sqldb.QueryConfig, db SQLQueries,
 	dbNode sqlc.GraphNode) (*models.LightningNode, error) {
 
-	// NOTE: buildNode is only used to load the data for a single node, and
-	// so no paged queries will be performed. This means that it's ok to
-	// used pass in default config values here.
-	cfg := sqldb.DefaultQueryConfig()
-
 	data, err := batchLoadNodeData(ctx, cfg, db, []int64{dbNode.ID})
 	if err != nil {
 		return nil, fmt.Errorf("unable to batch load node data: %w",
@@ -3945,22 +3940,21 @@ func upsertChanPolicyExtraSignedFields(ctx context.Context, db SQLQueries,
 // getAndBuildEdgeInfo builds a models.ChannelEdgeInfo instance from the
 // provided dbChanRow and also fetches any other required information
 // to construct the edge info.
-func getAndBuildEdgeInfo(ctx context.Context, db SQLQueries,
+func getAndBuildEdgeInfo(ctx context.Context, cfg *SQLStoreConfig,
-	chain chainhash.Hash, dbChan sqlc.GraphChannel, node1,
+	db SQLQueries, dbChan sqlc.GraphChannel, node1,
 	node2 route.Vertex) (*models.ChannelEdgeInfo, error) {
 
-	// NOTE: getAndBuildEdgeInfo is only used to load the data for a single
+	data, err := batchLoadChannelData(
-	// edge, and so no paged queries will be performed. This means that
+		ctx, cfg.QueryCfg, db, []int64{dbChan.ID}, nil,
-	// it's ok to used pass in default config values here.
+	)
-	cfg := sqldb.DefaultQueryConfig()
-
-	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)
+	return buildEdgeInfoWithBatchData(
+		cfg.ChainHash, dbChan, node1, node2, data,
+	)
 }
 
 // buildEdgeInfoWithBatchData builds edge info using pre-loaded batch data.
@@ -4059,9 +4053,9 @@ func buildNodeVertices(node1Pub, node2Pub []byte) (route.Vertex,
 // retrieves all the extra info required to build the complete
 // models.ChannelEdgePolicy types. It returns two policies, which may be nil if
 // the provided sqlc.GraphChannelPolicy records are nil.
-func getAndBuildChanPolicies(ctx context.Context, db SQLQueries,
+func getAndBuildChanPolicies(ctx context.Context, cfg *sqldb.QueryConfig,
-	dbPol1, dbPol2 *sqlc.GraphChannelPolicy, channelID uint64, node1,
+	db SQLQueries, dbPol1, dbPol2 *sqlc.GraphChannelPolicy,
-	node2 route.Vertex) (*models.ChannelEdgePolicy,
+	channelID uint64, node1, node2 route.Vertex) (*models.ChannelEdgePolicy,
 	*models.ChannelEdgePolicy, error) {
 
 	if dbPol1 == nil && dbPol2 == nil {
@@ -4076,12 +4070,6 @@ func getAndBuildChanPolicies(ctx context.Context, db SQLQueries,
 		policyIDs = append(policyIDs, dbPol2.ID)
 	}
 
-	// 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.DefaultQueryConfig()
-
 	batchData, err := batchLoadChannelData(ctx, cfg, db, nil, policyIDs)
 	if err != nil {
 		return nil, nil, fmt.Errorf("unable to batch load channel "+

graph/db/test_postgres.go

@@ -43,7 +43,7 @@ func NewTestDBWithFixture(t testing.TB,
 	store, err := NewSQLStore(
 		&SQLStoreConfig{
 			ChainHash: *chaincfg.MainNetParams.GenesisHash,
-			QueryCfg:  sqldb.DefaultQueryConfig(),
+			QueryCfg:  sqldb.DefaultPostgresConfig(),
 		}, querier,
 	)
 	require.NoError(t, err)

graph/db/test_sqlite.go

@@ -28,7 +28,7 @@ func NewTestDBWithFixture(t testing.TB, _ *sqldb.TestPgFixture) V1Store {
 	store, err := NewSQLStore(
 		&SQLStoreConfig{
 			ChainHash: *chaincfg.MainNetParams.GenesisHash,
-			QueryCfg:  sqldb.DefaultQueryConfig(),
+			QueryCfg:  sqldb.DefaultSQLiteConfig(),
 		}, newBatchQuerier(t),
 	)
 	require.NoError(t, err)

itest/lnd_graph_migration_test.go

@@ -142,10 +142,15 @@ func openNativeSQLGraphDB(ht *lntest.HarnessTest,
 		},
 	)
 
+	queryCfg := sqldb.DefaultSQLiteConfig()
+	if hn.Cfg.DBBackend != node.BackendSqlite {
+		queryCfg = sqldb.DefaultPostgresConfig()
+	}
+
 	store, err := graphdb.NewSQLStore(
 		&graphdb.SQLStoreConfig{
 			ChainHash: *ht.Miner().ActiveNet.GenesisHash,
-			QueryCfg:  sqldb.DefaultQueryConfig(),
+			QueryCfg:  queryCfg,
 		},
 		executor,
 	)

lncfg/db.go

@@ -115,10 +115,12 @@ func DefaultDB() *DB {
 		},
 		Postgres: &sqldb.PostgresConfig{
 			MaxConnections: defaultPostgresMaxConnections,
+			QueryConfig:    *sqldb.DefaultPostgresConfig(),
 		},
 		Sqlite: &sqldb.SqliteConfig{
 			MaxConnections: defaultSqliteMaxConnections,
 			BusyTimeout:    defaultSqliteBusyTimeout,
+			QueryConfig:    *sqldb.DefaultSQLiteConfig(),
 		},
 		UseNativeSQL:           false,
 		SkipNativeSQLMigration: false,
@@ -135,6 +137,9 @@ func (db *DB) Validate() error {
 		}
 
 	case SqliteBackend:
+		if err := db.Sqlite.Validate(); err != nil {
+			return err
+		}
 	case PostgresBackend:
 		if err := db.Postgres.Validate(); err != nil {
 			return err

sample-lnd.conf

@@ -1601,6 +1601,12 @@
 ; Whether to skip executing schema migrations.
 ; db.postgres.skipmigrations=false
 
+; The maximum number of elements to use in a native-SQL batch query IN clause.
+; db.postgres.query.max-batch-size=5000
+
+; The maximum number of records to fetch at a time in a paginated query during
+; native-SQL calls.
+; db.postgres.query.max-page-size=10500
 
 [sqlite]
 
@@ -1617,6 +1623,13 @@
 ; The maximum amount of time to wait to execute a query if the db is locked.
 ; db.sqlite.busytimeout=5s
 
+; The maximum number of elements to use in a native-SQL batch query IN clause.
+; db.sqlite.query.max-batch-size=250
+
+; The maximum number of records to fetch at a time in a paginated query during
+; native-SQL calls.
+; db.sqlite.query.max-page-size=100
+
 ; Raw pragma option pairs to be used when opening the sqlite db. The flag
 ; can be specified multiple times to set multiple options.
 ; Default:

sqldb/config.go

@@ -28,6 +28,16 @@ type SqliteConfig struct {
 	MaxConnections int           `long:"maxconnections" description:"The maximum number of open connections to the database. Set to zero for unlimited."`
 	PragmaOptions  []string      `long:"pragmaoptions" description:"A list of pragma options to set on a database connection. For example, 'auto_vacuum=incremental'. Note that the flag must be specified multiple times if multiple options are to be set."`
 	SkipMigrations bool          `long:"skipmigrations" description:"Skip applying migrations on startup."`
+	QueryConfig    `group:"query" namespace:"query"`
+}
+
+// Validate checks that the SqliteConfig values are valid.
+func (p *SqliteConfig) Validate() error {
+	if err := p.QueryConfig.Validate(true); err != nil {
+		return fmt.Errorf("invalid query config: %w", err)
+	}
+
+	return nil
 }
 
 // PostgresConfig holds the postgres database configuration.
@@ -38,8 +48,10 @@ type PostgresConfig struct {
 	Timeout        time.Duration `long:"timeout" description:"Database connection timeout. Set to zero to disable."`
 	MaxConnections int           `long:"maxconnections" description:"The maximum number of open connections to the database. Set to zero for unlimited."`
 	SkipMigrations bool          `long:"skipmigrations" description:"Skip applying migrations on startup."`
+	QueryConfig    `group:"query" namespace:"query"`
 }
 
+// Validate checks that the PostgresConfig values are valid.
 func (p *PostgresConfig) Validate() error {
 	if p.Dsn == "" {
 		return fmt.Errorf("DSN is required")
@@ -51,5 +63,9 @@ func (p *PostgresConfig) Validate() error {
 		return fmt.Errorf("invalid DSN: %w", err)
 	}
 
+	if err := p.QueryConfig.Validate(false); err != nil {
+		return fmt.Errorf("invalid query config: %w", err)
+	}
+
 	return nil
 }

sqldb/paginate.go

@@ -5,25 +5,88 @@ import (
 	"fmt"
 )
 
+const (
+	// maxSQLiteBatchSize is the maximum number of items that can be
+	// included in a batch query IN clause for SQLite. This was determined
+	// using the TestSQLSliceQueries test.
+	maxSQLiteBatchSize = 32766
+
+	// maxPostgresBatchSize is the maximum number of items that can be
+	// included in a batch query IN clause for Postgres. This was determined
+	// using the TestSQLSliceQueries test.
+	maxPostgresBatchSize = 65535
+
+	// defaultSQLitePageSize is the default page size for SQLite queries.
+	defaultSQLitePageSize = 100
+
+	// defaultPostgresPageSize is the default page size for Postgres
+	// queries.
+	defaultPostgresPageSize = 10500
+
+	// defaultSQLiteBatchSize is the default batch size for SQLite queries.
+	defaultSQLiteBatchSize = 250
+
+	// defaultPostgresBatchSize is the default batch size for Postgres
+	// queries.
+	defaultPostgresBatchSize = 5000
+)
+
 // QueryConfig holds configuration values for SQL queries.
+//
+//nolint:ll
 type QueryConfig struct {
 	// MaxBatchSize is the maximum number of items included in a batch
 	// query IN clauses list.
-	MaxBatchSize int
+	MaxBatchSize int `long:"max-batch-size" description:"The maximum number of items to include in a batch query IN clause. This is used for queries that fetch results based on a list of identifiers."`
 
 	// MaxPageSize is the maximum number of items returned in a single page
 	// of results. This is used for paginated queries.
-	MaxPageSize int32
+	MaxPageSize int32 `long:"max-page-size" description:"The maximum number of items to return in a single page of results. This is used for paginated queries."`
 }
 
-// DefaultQueryConfig returns a default configuration for SQL queries.
+// Validate checks that the QueryConfig values are valid.
-//
+func (c *QueryConfig) Validate(sqlite bool) error {
-// TODO(elle): make configurable & have different defaults for SQLite and
+	if c.MaxBatchSize <= 0 {
-// Postgres.
+		return fmt.Errorf("max batch size must be greater than "+
-func DefaultQueryConfig() *QueryConfig {
+			"zero, got %d", c.MaxBatchSize)
+	}
+	if c.MaxPageSize <= 0 {
+		return fmt.Errorf("max page size must be greater than "+
+			"zero, got %d", c.MaxPageSize)
+	}
+
+	if sqlite {
+		if c.MaxBatchSize > maxSQLiteBatchSize {
+			return fmt.Errorf("max batch size for SQLite cannot "+
+				"exceed %d, got %d", maxSQLiteBatchSize,
+				c.MaxBatchSize)
+		}
+	} else {
+		if c.MaxBatchSize > maxPostgresBatchSize {
+			return fmt.Errorf("max batch size for Postgres cannot "+
+				"exceed %d, got %d", maxPostgresBatchSize,
+				c.MaxBatchSize)
+		}
+	}
+
+	return nil
+}
+
+// DefaultSQLiteConfig returns a default configuration for SQL queries to a
+// SQLite backend.
+func DefaultSQLiteConfig() *QueryConfig {
 	return &QueryConfig{
-		MaxBatchSize: 250,
+		MaxBatchSize: defaultSQLiteBatchSize,
-		MaxPageSize:  10000,
+		MaxPageSize:  defaultSQLitePageSize,
+	}
+}
+
+// DefaultPostgresConfig returns a default configuration for SQL queries to a
+// Postgres backend.
+func DefaultPostgresConfig() *QueryConfig {
+	return &QueryConfig{
+		MaxBatchSize: defaultPostgresBatchSize,
+		MaxPageSize:  defaultPostgresPageSize,
 	}
 }
 

sqldb/paginate_test.go

@@ -22,7 +22,7 @@ func TestExecuteBatchQuery(t *testing.T) {
 
 	t.Run("empty input returns nil", func(t *testing.T) {
 		var (
-			cfg        = DefaultQueryConfig()
+			cfg        = DefaultSQLiteConfig()
 			inputItems []int
 		)
 
@@ -144,7 +144,7 @@ func TestExecuteBatchQuery(t *testing.T) {
 
 	t.Run("query function error is propagated", func(t *testing.T) {
 		var (
-			cfg        = DefaultQueryConfig()
+			cfg        = DefaultSQLiteConfig()
 			inputItems = []int{1, 2, 3}
 		)
 
@@ -174,7 +174,7 @@ func TestExecuteBatchQuery(t *testing.T) {
 
 	t.Run("callback error is propagated", func(t *testing.T) {
 		var (
-			cfg        = DefaultQueryConfig()
+			cfg        = DefaultSQLiteConfig()
 			inputItems = []int{1, 2, 3}
 		)
 
@@ -281,11 +281,17 @@ func TestSQLSliceQueries(t *testing.T) {
 			break
 		}
 
-		x *= 10
+		// If it succeeded, we expect it to be under the maximum that
+		// we expect for this DB.
+		if isSQLite {
+			require.LessOrEqual(t, x, maxSQLiteBatchSize,
+				"SQLite should not exceed 32766 parameters")
+		} else {
+			require.LessOrEqual(t, x, maxPostgresBatchSize,
+				"Postgres should not exceed 65535 parameters")
+		}
 
-		// Just to make sure that the test doesn't carry on too long,
+		x *= 10
-		// we assert that we don't exceed a reasonable limit.
-		require.LessOrEqual(t, x, 100000)
 	}
 
 	// Now that we have found the limit that the raw query can handle, we
@@ -301,7 +307,7 @@ func TestSQLSliceQueries(t *testing.T) {
 
 	err := ExecuteBatchQuery(
 		ctx,
-		DefaultQueryConfig(),
+		DefaultSQLiteConfig(),
 		queryParams,
 		func(s string) string {
 			return s