WIP

FerretDB · Mar 4, 2024 · b228093 · b228093
1 parent 1bf9901
commit b228093
Show file tree

Hide file tree

Showing 12 changed files with 63 additions and 41 deletions.
diff --git a/internal/backends/postgresql/backend.go b/internal/backends/postgresql/backend.go
@@ -37,15 +37,16 @@ type backend struct {
 //
 //nolint:vet // for readability
 type NewBackendParams struct {
-	URI string
-	L   *zap.Logger
-	P   *state.Provider
-	_   struct{} // prevent unkeyed literals
+	URI       string
+	L         *zap.Logger
+	P         *state.Provider
+	BatchSize int
+	_         struct{} // prevent unkeyed literals
 }
 
 // NewBackend creates a new Backend.
 func NewBackend(params *NewBackendParams) (backends.Backend, error) {
-	r, err := metadata.NewRegistry(params.URI, params.L, params.P)
+	r, err := metadata.NewRegistry(params.URI, params.BatchSize, params.L, params.P)
 	if err != nil {
 		return nil, err
 	}

diff --git a/internal/backends/postgresql/collection.go b/internal/backends/postgresql/collection.go
@@ -134,8 +134,7 @@ func (c *collection) InsertAll(ctx context.Context, params *backends.InsertAllPa
 	}
 
 	err = pool.InTransaction(ctx, p, func(tx pgx.Tx) error {
-		// TODO https://github.com/FerretDB/FerretDB/issues/3708
-		const batchSize = 100
+		batchSize := c.r.BatchSize()
 
 		var batch []*types.Document
 		docs := params.Docs

diff --git a/internal/backends/postgresql/metadata/registry.go b/internal/backends/postgresql/metadata/registry.go
@@ -81,20 +81,22 @@ type Registry struct {
 	// One global lock should be replaced by more granular locks – one per database or even one per collection.
 	// But that requires some redesign.
 	// TODO https://github.com/FerretDB/FerretDB/issues/2755
-	rw    sync.RWMutex
-	colls map[string]map[string]*Collection // database name -> collection name -> collection
+	rw        sync.RWMutex
+	colls     map[string]map[string]*Collection // database name -> collection name -> collection
+	batchSize int
 }
 
 // NewRegistry creates a registry for PostgreSQL databases with a given base URI.
-func NewRegistry(u string, l *zap.Logger, sp *state.Provider) (*Registry, error) {
+func NewRegistry(u string, batchSize int, l *zap.Logger, sp *state.Provider) (*Registry, error) {
 	p, err := pool.New(u, l, sp)
 	if err != nil {
 		return nil, err
 	}
 
 	r := &Registry{
-		p: p,
-		l: l,
+		p:         p,
+		l:         l,
+		batchSize: batchSize,
 	}
 
 	return r, nil
@@ -1032,6 +1034,14 @@ func (r *Registry) Collect(ch chan<- prometheus.Metric) {
 	}
 }
 
+// BatchSize returns number of maximum size of query parameters.
+func (r *Registry) BatchSize() int {
+	r.rw.RLock()
+	batchSize := r.batchSize
+	r.rw.RUnlock()
+	return batchSize
+}
+
 // check interfaces
 var (
 	_ prometheus.Collector = (*Registry)(nil)

diff --git a/internal/backends/postgresql/metadata/registry_test.go b/internal/backends/postgresql/metadata/registry_test.go
@@ -91,7 +91,7 @@ func createDatabase(t *testing.T, ctx context.Context) (*Registry, *pgxpool.Pool
 	sp, err := state.NewProvider("")
 	require.NoError(t, err)
 
-	r, err := NewRegistry(u, testutil.Logger(t), sp)
+	r, err := NewRegistry(u, 100, testutil.Logger(t), sp)
 	require.NoError(t, err)
 	t.Cleanup(r.Close)
 
@@ -140,7 +140,7 @@ func TestAuth(t *testing.T) {
 			sp, err := state.NewProvider("")
 			require.NoError(t, err)
 
-			r, err := NewRegistry(tc.uri, testutil.Logger(t), sp)
+			r, err := NewRegistry(tc.uri, 100, testutil.Logger(t), sp)
 			require.NoError(t, err)
 			t.Cleanup(r.Close)
 

diff --git a/internal/backends/sqlite/backend.go b/internal/backends/sqlite/backend.go
@@ -37,15 +37,16 @@ type backend struct {
 //
 //nolint:vet // for readability
 type NewBackendParams struct {
-	URI string
-	L   *zap.Logger
-	P   *state.Provider
-	_   struct{} // prevent unkeyed literals
+	URI       string
+	L         *zap.Logger
+	P         *state.Provider
+	BatchSize int
+	_         struct{} // prevent unkeyed literals
 }
 
 // NewBackend creates a new Backend.
 func NewBackend(params *NewBackendParams) (backends.Backend, error) {
-	r, err := metadata.NewRegistry(params.URI, params.L, params.P)
+	r, err := metadata.NewRegistry(params.URI, params.BatchSize, params.L, params.P)
 	if err != nil {
 		return nil, err
 	}

diff --git a/internal/backends/sqlite/collection.go b/internal/backends/sqlite/collection.go
@@ -109,8 +109,7 @@ func (c *collection) InsertAll(ctx context.Context, params *backends.InsertAllPa
 	meta := c.r.CollectionGet(ctx, c.dbName, c.name)
 
 	err := db.InTransaction(ctx, func(tx *fsql.Tx) error {
-		// TODO https://github.com/FerretDB/FerretDB/issues/3708
-		const batchSize = 100
+		batchSize := c.r.BatchSize()
 
 		var batch []*types.Document
 		docs := params.Docs

diff --git a/internal/backends/sqlite/metadata/registry.go b/internal/backends/sqlite/metadata/registry.go
@@ -64,21 +64,23 @@ type Registry struct {
 	// One global lock should be replaced by more granular locks – one per database or even one per collection.
 	// But that requires some redesign.
 	// TODO https://github.com/FerretDB/FerretDB/issues/2755
-	rw    sync.RWMutex
-	colls map[string]map[string]*Collection // database name -> collection name -> collection
+	rw        sync.RWMutex
+	colls     map[string]map[string]*Collection // database name -> collection name -> collection
+	batchSize int
 }
 
 // NewRegistry creates a registry for SQLite databases in the directory specified by SQLite URI.
-func NewRegistry(u string, l *zap.Logger, sp *state.Provider) (*Registry, error) {
+func NewRegistry(u string, batchSize int, l *zap.Logger, sp *state.Provider) (*Registry, error) {
 	p, initDBs, err := pool.New(u, l, sp)
 	if err != nil {
 		return nil, err
 	}
 
 	r := &Registry{
-		p:     p,
-		l:     l,
-		colls: map[string]map[string]*Collection{},
+		p:         p,
+		l:         l,
+		colls:     map[string]map[string]*Collection{},
+		batchSize: batchSize,
 	}
 
 	for name, db := range initDBs {
@@ -640,6 +642,14 @@ func (r *Registry) Collect(ch chan<- prometheus.Metric) {
 	}
 }
 
+// BatchSize returns number of maximum size of query parameters.
+func (r *Registry) BatchSize() int {
+	r.rw.RLock()
+	batchSize := r.batchSize
+	r.rw.RUnlock()
+	return batchSize
+}
+
 // check interfaces
 var (
 	_ prometheus.Collector = (*Registry)(nil)

diff --git a/internal/backends/sqlite/metadata/registry_test.go b/internal/backends/sqlite/metadata/registry_test.go
@@ -75,7 +75,7 @@ func TestCreateDrop(t *testing.T) {
 	sp, err := state.NewProvider("")
 	require.NoError(t, err)
 
-	r, err := NewRegistry(testutil.TestSQLiteURI(t, ""), testutil.Logger(t), sp)
+	r, err := NewRegistry(testutil.TestSQLiteURI(t, ""), 100, testutil.Logger(t), sp)
 	require.NoError(t, err)
 	t.Cleanup(r.Close)
 
@@ -113,7 +113,7 @@ func TestCreateDropStress(t *testing.T) {
 	} {
 		t.Run(testName, func(t *testing.T) {
 			uri := testutil.TestSQLiteURI(t, "") + params
-			r, err := NewRegistry(uri, testutil.Logger(t), sp)
+			r, err := NewRegistry(uri, 100, testutil.Logger(t), sp)
 			require.NoError(t, err)
 			t.Cleanup(r.Close)
 
@@ -158,7 +158,7 @@ func TestCreateSameStress(t *testing.T) {
 	} {
 		t.Run(testName, func(t *testing.T) {
 			uri := testutil.TestSQLiteURI(t, "") + params
-			r, err := NewRegistry(uri, testutil.Logger(t), sp)
+			r, err := NewRegistry(uri, 100, testutil.Logger(t), sp)
 			require.NoError(t, err)
 			t.Cleanup(r.Close)
 
@@ -227,7 +227,7 @@ func TestDropSameStress(t *testing.T) {
 	} {
 		t.Run(testName, func(t *testing.T) {
 			uri := testutil.TestSQLiteURI(t, "") + params
-			r, err := NewRegistry(uri, testutil.Logger(t), sp)
+			r, err := NewRegistry(uri, 100, testutil.Logger(t), sp)
 			require.NoError(t, err)
 			t.Cleanup(r.Close)
 
@@ -280,7 +280,7 @@ func TestCreateDropSameStress(t *testing.T) {
 	} {
 		t.Run(testName, func(t *testing.T) {
 			uri := testutil.TestSQLiteURI(t, "") + params
-			r, err := NewRegistry(uri, testutil.Logger(t), sp)
+			r, err := NewRegistry(uri, 100, testutil.Logger(t), sp)
 			require.NoError(t, err)
 			t.Cleanup(r.Close)
 
@@ -329,7 +329,7 @@ func TestIndexesCreateDrop(t *testing.T) {
 	sp, err := state.NewProvider("")
 	require.NoError(t, err)
 
-	r, err := NewRegistry(testutil.TestSQLiteURI(t, ""), testutil.Logger(t), sp)
+	r, err := NewRegistry(testutil.TestSQLiteURI(t, ""), 100, testutil.Logger(t), sp)
 	require.NoError(t, err)
 	t.Cleanup(r.Close)
 

diff --git a/internal/backends/sqlite/sqlite_test.go b/internal/backends/sqlite/sqlite_test.go
@@ -31,7 +31,7 @@ func TestCollectionsStats(t *testing.T) {
 	sp, err := state.NewProvider("")
 	require.NoError(t, err)
 
-	r, err := metadata.NewRegistry(testutil.TestSQLiteURI(t, ""), testutil.Logger(t), sp)
+	r, err := metadata.NewRegistry(testutil.TestSQLiteURI(t, ""), 100, testutil.Logger(t), sp)
 	require.NoError(t, err)
 	t.Cleanup(r.Close)
 

diff --git a/internal/handler/registry/hana.go b/internal/handler/registry/hana.go
@@ -48,7 +48,7 @@ func init() {
 			CappedCleanupPercentage: opts.CappedCleanupPercentage,
 			CappedCleanupInterval:   opts.CappedCleanupInterval,
 			EnableNewAuth:           opts.EnableNewAuth,
-			BatchSize: opts.BatchSize
+			BatchSize:               opts.BatchSize,
 		}
 
 		h, err := handler.New(handlerOpts)

diff --git a/internal/handler/registry/postgresql.go b/internal/handler/registry/postgresql.go
@@ -25,9 +25,10 @@ import (
 func init() {
 	registry["postgresql"] = func(opts *NewHandlerOpts) (*handler.Handler, CloseBackendFunc, error) {
 		b, err := postgresql.NewBackend(&postgresql.NewBackendParams{
-			URI: opts.PostgreSQLURL,
-			L:   opts.Logger.Named("postgresql"),
-			P:   opts.StateProvider,
+			URI:       opts.PostgreSQLURL,
+			L:         opts.Logger.Named("postgresql"),
+			P:         opts.StateProvider,
+			BatchSize: opts.BatchSize,
 		})
 		if err != nil {
 			return nil, nil, err

diff --git a/internal/handler/registry/sqlite.go b/internal/handler/registry/sqlite.go
@@ -25,9 +25,10 @@ import (
 func init() {
 	registry["sqlite"] = func(opts *NewHandlerOpts) (*handler.Handler, CloseBackendFunc, error) {
 		b, err := sqlite.NewBackend(&sqlite.NewBackendParams{
-			URI: opts.SQLiteURL,
-			L:   opts.Logger.Named("sqlite"),
-			P:   opts.StateProvider,
+			URI:       opts.SQLiteURL,
+			L:         opts.Logger.Named("sqlite"),
+			P:         opts.StateProvider,
+			BatchSize: opts.BatchSize,
 		})
 		if err != nil {
 			return nil, nil, err