WIP

FerretDB · AlekSi · Mar 19, 2024 · Feb 27, 2024 · Mar 4, 2024 · Mar 4, 2024
commit 2c51cabf9741b1016ef174df4b18a226ae38eedf
diff --git a/Taskfile.yml b/Taskfile.yml
@@ -381,6 +381,7 @@ tasks:
         --handler=sqlite
         --sqlite-url=file:tmp/sqlite/
         --test-records-dir=tmp/records
+        --test-batch-size=1000
 
   run-mysql:
     desc: "Run FerretDB with `mysql` backend"
@@ -407,6 +408,7 @@ tasks:
         --mode=diff-normal
         --handler=hana
         --test-records-dir=tmp/records
+        --test-batch-size=100
 
   run-secured:
     desc: "Run FerretDB with `postgresql` backend (TLS, auth required)"
@@ -424,6 +426,7 @@ tasks:
         --handler=pg
         --postgresql-url='postgres://127.0.0.1:5433/ferretdb?search_path='
         --test-records-dir=tmp/records
+        --test-batch-size=100
 
   run-proxy:
     desc: "Run FerretDB in diff-proxy mode"
@@ -437,6 +440,7 @@ tasks:
         --handler=pg
         --postgresql-url='postgres://username@127.0.0.1:5432/ferretdb?search_path='
         --test-records-dir=tmp/records
+        --test-batch-size=100
 
   run-sqlite-proxy:
     desc: "Run FerretDB with `sqlite` handler in diff-proxy mode"
@@ -451,6 +455,7 @@ tasks:
         --handler=sqlite
         --sqlite-url=file:tmp/sqlite/
         --test-records-dir=tmp/records
+        --test-batch-size=1000
 
   run-proxy-secured:
     desc: "Run FerretDB in diff-proxy mode (TLS, auth required)"
@@ -471,6 +476,7 @@ tasks:
         --handler=pg
         --postgresql-url='postgres://username@127.0.0.1:5433/ferretdb?search_path='
         --test-records-dir=tmp/records
+        --test-batch-size=100
 
   lint:
     desc: "Run linters"

@@ -101,7 +101,7 @@ var cli struct {
 		} `embed:"" prefix:"capped-cleanup-"`
 
 		EnableNewAuth bool `default:"false" help:"Experimental: enable new authentication."`
-		BatchSize     int  `default:"100" help:"number of maximum size of query parameters"`
+		BatchSize     int  `default:"100"	help:"number of maximum size of query parameters"`
 
 		Telemetry struct {
 			URL            string        `default:"https://beacon.ferretdb.com/" help:"Telemetry: reporting URL."`

@@ -37,9 +37,10 @@ type backend struct {
 //
 //nolint:vet // for readability
 type NewBackendParams struct {
-	URI string
-	L   *zap.Logger
-	P   *state.Provider
+	URI       string
+	L         *zap.Logger
+	P         *state.Provider
+	BatchSize int
 }
 
 // NewBackend creates a new Backend.
@@ -50,6 +51,7 @@ func NewBackend(params *NewBackendParams) (backends.Backend, error) {
 	}
 
 	hdb := fsql.WrapDB(db, "hana", params.L)
+	hdb.BatchSize = params.BatchSize
 
 	return backends.BackendContract(&backend{
 		hdb: hdb,

@@ -134,7 +134,7 @@ func (c *collection) InsertAll(ctx context.Context, params *backends.InsertAllPa
 	}
 
 	err = pool.InTransaction(ctx, p, func(tx pgx.Tx) error {
-		batchSize := c.r.BatchSize()
+		batchSize := c.r.BatchSize
 
 		var batch []*types.Document
 		docs := params.Docs

@@ -73,17 +73,17 @@ var specialCharacters = regexp.MustCompile("[^a-z][^a-z0-9_]*")
 //
 //nolint:vet // for readability
 type Registry struct {
-	p *pool.Pool
-	l *zap.Logger
+	p         *pool.Pool
+	l         *zap.Logger
+	BatchSize int
 
 	// rw protects colls but also acts like a global lock for the whole registry.
 	// The latter effectively replaces transactions (see the postgresql backend package description for more info).
 	// 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
-	batchSize int
+	rw    sync.RWMutex
+	colls map[string]map[string]*Collection // database name -> collection name -> collection
 }
 
 // NewRegistry creates a registry for PostgreSQL databases with a given base URI.
@@ -96,7 +96,7 @@ func NewRegistry(u string, batchSize int, l *zap.Logger, sp *state.Provider) (*R
 	r := &Registry{
 		p:         p,
 		l:         l,
-		batchSize: batchSize,
+		BatchSize: batchSize,
 	}
 
 	return r, nil
@@ -1034,14 +1034,6 @@ 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)

@@ -109,7 +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 {
-		batchSize := c.r.BatchSize()
+		batchSize := c.r.BatchSize
 
 		var batch []*types.Document
 		docs := params.Docs

@@ -56,17 +56,17 @@ const (
 //
 // Exported methods are safe for concurrent use. Unexported methods are not.
 type Registry struct {
-	p *pool.Pool
-	l *zap.Logger
+	p         *pool.Pool
+	l         *zap.Logger
+	BatchSize int
 
 	// rw protects colls but also acts like a global lock for the whole registry.
 	// The latter effectively replaces transactions (see the sqlite backend package description for more info).
 	// 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
-	batchSize int
+	rw    sync.RWMutex
+	colls map[string]map[string]*Collection // database name -> collection name -> collection
 }
 
 // NewRegistry creates a registry for SQLite databases in the directory specified by SQLite URI.
@@ -79,8 +79,8 @@ func NewRegistry(u string, batchSize int, l *zap.Logger, sp *state.Provider) (*R
 	r := &Registry{
 		p:         p,
 		l:         l,
+		BatchSize: batchSize,
 		colls:     map[string]map[string]*Collection{},
-		batchSize: batchSize,
 	}
 
 	for name, db := range initDBs {
@@ -642,14 +642,6 @@ 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)

@@ -85,7 +85,6 @@ func (h *Handler) MsgInsert(ctx context.Context, msg *wire.OpMsg) (*wire.OpMsg,
 
 	var done bool
 	for !done {
-
 		docs := make([]*types.Document, 0, h.BatchSize)
 		docsIndexes := make([]int, 0, h.BatchSize)
 

@@ -27,9 +27,10 @@ func init() {
 		opts.Logger.Warn("HANA handler is in alpha. It is not supported yet.")
 
 		b, err := hana.NewBackend(&hana.NewBackendParams{
-			URI: opts.HANAURL,
-			L:   opts.Logger.Named("hana"),
-			P:   opts.StateProvider,
+			URI:       opts.HANAURL,
+			L:         opts.Logger.Named("hana"),
+			P:         opts.StateProvider,
+			BatchSize: opts.BatchSize,
 		})
 		if err != nil {
 			return nil, nil, err

@@ -35,9 +35,10 @@ import (
 type DB struct {
 	*metricsCollector
 
-	sqlDB *sql.DB
-	l     *zap.Logger
-	token *resource.Token
+	sqlDB     *sql.DB
+	l         *zap.Logger
+	token     *resource.Token
+	BatchSize int
 }
 
 // WrapDB creates a new DB.