Separated logic for update|delete|insert one and many. Implemented insert batch sizing. Updated readme

Author: daniel-percona

README.md

@@ -117,9 +117,11 @@ Environment Variables (Overrides):
   PERCONALOAD_DELETE_PERCENT          % of ops that are DELETE
   PERCONALOAD_AGGREGATE_PERCENT       % of ops that are AGGREGATE
   PERCONALOAD_TRANSACTION_PERCENT     % of ops that are TRANSACTIONAL
+  PERCONALOAD_BULK_INSERT_PERCENT     % of ops that are BULK INSERTS
 
  [Performance Optimization]
   PERCONALOAD_FIND_BATCH_SIZE         Docs returned per cursor batch
+  PERCONALOAD_INSERT_BATCH_SIZE       Number of docs in batch bulk insert
   PERCONALOAD_FIND_LIMIT              Max docs per Find query
   PERCONALOAD_INSERT_CACHE_SIZE       Generator buffer size
   PERCONALOAD_OP_TIMEOUT_MS           Soft timeout per DB op (ms)
@@ -192,13 +194,15 @@ You can override any setting in `config.yaml` using environment variables. This
 | `max_transaction_ops` | `PERCONALOAD_MAX_TRANSACTION_OPS` | Maximum number of operations to group into a single transaction block | `5` |
 | **Operation Ratios** | | (Must sum to ~100) | |
 | `find_percent` | `PERCONALOAD_FIND_PERCENT` | Percentage of Find operations | `50` |
-| `insert_percent` | `PERCONALOAD_INSERT_PERCENT` | Percentage of Insert operations (this is not related to the initial seed inserts) | `20` |
+| `insert_percent` | `PERCONALOAD_INSERT_PERCENT` | Percentage of Insert operations (this is not related to the initial seed inserts) | `10` |
+| `bulk_insert_percent ` | `PERCONALOAD_BULK_INSERT_PERCENT` | Percentage of Bulk Insert operations (this is not related to the initial seed inserts) | `10` |
 | `update_percent` | `PERCONALOAD_UPDATE_PERCENT` | Percentage of Update operations | `10` |
 | `delete_percent` | `PERCONALOAD_DELETE_PERCENT` | Percentage of Delete operations | `10` |
 | `aggregate_percent` | `PERCONALOAD_AGGREGATE_PERCENT` | Percentage of Aggregate operations | `5` |
 | `transaction_percent` | `PERCONALOAD_TRANSACTION_PERCENT` | Percentage of Transactional operations | `5` |
 | **Performance Optimization** | | | |
 | `find_batch_size` | `PERCONALOAD_FIND_BATCH_SIZE` | Documents returned per cursor batch | `100` |
+| `insert_batch_size` | `PERCONALOAD_INSERT_BATCH_SIZE` | Number of documents per insert batch | `100` |
 | `find_limit` | `PERCONALOAD_FIND_LIMIT` | Hard limit on documents per Find query | `10` |
 | `insert_cache_size` | `PERCONALOAD_INSERT_CACHE_SIZE` | Size of the document generation buffer | `1000` |
 | `op_timeout_ms` | `PERCONALOAD_OP_TIMEOUT_MS` | Soft timeout for individual DB operations (ms) | `500` |
@@ -380,7 +384,8 @@ By default, the tool comes preconfigured with the following workload distributio
 | Find	| 50% | 
 | Update	| 20% | 
 | Delete	| 10% | 
-| Insert	| 10% | 
+| Insert	| 5% | 
+| Bulk Inserts | 5% |
 | Aggregate	| 5% | 
 | Transaction	| 5% | 
 
@@ -389,7 +394,7 @@ You can modify any of the values above to run different types of workloads.
 Please note:
 
 * If `use_transactions: false`, the transaction_percent value is ignored.
-* If there are no aggregation queries defined in queries.json, the aggregate_percent value is also ignored.
+* If there are no aggregation queries defined in queries.json, the aggregate_percent value is also ignored. 
 * Aggregate operations will only generate activity if at least one query with "operation": "aggregate" is defined in your active JSON query files.
 * The maximum number of operations within a transaction is defined in the config file via `max_transaction_ops` or the env var `PERCONALOAD_MAX_TRANSACTION_OPS`. The number of operations per transaction will be randomized, with the max number being set as explained above. 
 * Multi-Collection Load: If multiple collections are defined in your collections_path, each worker will randomly select a collection for every operation. This includes operations within a transaction, allowing for cross-collection atomic updates.
@@ -418,12 +423,15 @@ These settings affect the efficiency of individual database operations and memor
 * **`find_batch_size`**: The number of documents returned per batch in a cursor.
     * *Tip:* Higher values reduce network round-trips but increase memory usage per worker.
     * *Default:* `10`
+* **`insert_batch_size`**: The number of documents to be inserted by bulk inserts.
+    * *Default:* `10`   
 * **`find_limit`**: The hard limit on documents returned for `find` operations.
     * *Default:* `5`
 * **`insert_cache_size`**: The buffer size for the document generator channel.
     * *Tip:* This decouples document generation from database insertion. A larger buffer ensures workers rarely wait for data generation logic.
     * *Default:* `1000`
-* **`Upserts`**: Any updateOne or updateMany operation in your query JSON files can include "upsert": true. This will cause MongoDB to create the document if no match is found for the filter.    
+* **`upserts`**: Any updateOne or updateMany operation in your query JSON files can include "upsert": true. This will cause MongoDB to create the document if no match is found for the filter.  
+
 
 #### Timeouts & Reliability
 Control how plgm reacts to network lag or database pressure.

cmd/plgm/main.go

@@ -68,9 +68,11 @@ func main() {
 		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PERCONALOAD_DELETE_PERCENT", "% of ops that are DELETE")
 		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PERCONALOAD_AGGREGATE_PERCENT", "% of ops that are AGGREGATE")
 		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PERCONALOAD_TRANSACTION_PERCENT", "% of ops that are TRANSACTIONAL")
+		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PERCONALOAD_BULK_INSERT_PERCENT", "% of ops that are BULK INSERTS")
 
 		fmt.Fprintf(os.Stderr, "\n [Performance Optimization]\n")
 		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PERCONALOAD_FIND_BATCH_SIZE", "Docs returned per cursor batch")
+		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PERCONALOAD_INSERT_BATCH_SIZE", "Number of docs in batch bulk insert")
 		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PERCONALOAD_FIND_LIMIT", "Max docs per Find query")
 		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PERCONALOAD_INSERT_CACHE_SIZE", "Generator buffer size")
 		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PERCONALOAD_OP_TIMEOUT_MS", "Soft timeout per DB op (ms)")

config.yaml

@@ -85,7 +85,8 @@ max_transaction_ops: 5
 find_percent: 50
 update_percent: 20
 delete_percent: 10
-insert_percent: 10
+insert_percent: 5
+bulk_insert_percent: 5
 aggregate_percent: 5
 transaction_percent: 5
 
@@ -104,6 +105,7 @@ use_findone_for_limit_one: true # Optimization: use FindOne() if limit is 1.
 
 # --- Write Operations ---
 insert_cache_size: 1000         # Size of the internal buffer for generated documents.
+insert_batch_size: 10           # Number of documents per InsertMany call
 
 # --- Monitoring ---
 status_refresh_rate_sec: 1      # How often (seconds) to print the "AVG Ops/Sec" log line.

internal/config/config.go

@@ -28,6 +28,8 @@ type AppConfig struct {
 	InsertPercent      int    `yaml:"insert_percent"`
 	AggregatePercent   int    `yaml:"aggregate_percent"`
 	TransactionPercent int    `yaml:"transaction_percent"`
+	BulkInsertPercent  int    `yaml:"bulk_insert_percent"`
+	InsertBatchSize    int    `yaml:"insert_batch_size"`
 	UseTransactions    bool   `yaml:"use_transactions"`
 	MaxTransactionOps  int    `yaml:"max_transaction_ops"`
 	DebugMode          bool   `yaml:"debug_mode"`
@@ -82,6 +84,9 @@ func applyDefaults(cfg *AppConfig) {
 	if cfg.FindBatchSize <= 0 {
 		cfg.FindBatchSize = 10
 	}
+	if cfg.InsertBatchSize <= 0 {
+		cfg.InsertBatchSize = 10
+	}
 	if cfg.FindLimit <= 0 {
 		cfg.FindLimit = 5
 	}
@@ -249,6 +254,16 @@ func applyEnvOverrides(cfg *AppConfig) {
 			cfg.StatusRefreshRateSec = n
 		}
 	}
+	if p := os.Getenv("PERCONALOAD_BULK_INSERT_PERCENT"); p != "" {
+		if n, err := strconv.Atoi(p); err == nil && n >= 0 {
+			cfg.BulkInsertPercent = n
+		}
+	}
+	if v := os.Getenv("PERCONALOAD_INSERT_BATCH_SIZE"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			cfg.InsertBatchSize = n
+		}
+	}
 }
 
 func normalizePercentages(cfg *AppConfig) {
@@ -257,7 +272,7 @@ func normalizePercentages(cfg *AppConfig) {
 		cfg.TransactionPercent = 0
 	}
 
-	total := cfg.FindPercent + cfg.UpdatePercent + cfg.DeletePercent + cfg.InsertPercent + cfg.AggregatePercent + cfg.TransactionPercent
+	total := cfg.FindPercent + cfg.UpdatePercent + cfg.DeletePercent + cfg.InsertPercent + cfg.AggregatePercent + cfg.TransactionPercent + cfg.BulkInsertPercent
 	if total <= 0 {
 		cfg.FindPercent = 100
 		return
@@ -271,8 +286,9 @@ func normalizePercentages(cfg *AppConfig) {
 		cfg.InsertPercent = int(float64(cfg.InsertPercent) * factor)
 		cfg.AggregatePercent = int(float64(cfg.AggregatePercent) * factor)
 		cfg.TransactionPercent = int(float64(cfg.TransactionPercent) * factor)
+		cfg.BulkInsertPercent = int(float64(cfg.BulkInsertPercent) * factor)
 
-		finalTotal := cfg.FindPercent + cfg.UpdatePercent + cfg.DeletePercent + cfg.InsertPercent + cfg.AggregatePercent + cfg.TransactionPercent
+		finalTotal := cfg.FindPercent + cfg.UpdatePercent + cfg.DeletePercent + cfg.InsertPercent + cfg.AggregatePercent + cfg.TransactionPercent + cfg.BulkInsertPercent
 		if finalTotal != 100 {
 			cfg.FindPercent += (100 - finalTotal)
 		}

internal/mongo/runner.go

@@ -42,7 +42,8 @@ type workloadConfig struct {
 
 var InsertDocumentCache chan map[string]interface{}
 
-var operationTypes = []string{"find", "update", "delete", "insert", "aggregate", "transaction"}
+// base operation types for selection logic
+var operationTypes = []string{"find", "update", "delete", "insert", "insertMany", "aggregate", "transaction"}
 
 func selectOperation(percentages map[string]int, rng *rand.Rand) string {
 	if percentages == nil {
@@ -147,6 +148,19 @@ func generateInsertQuery(col config.CollectionDefinition, rng *rand.Rand, cfg *c
 	}
 }
 
+func generateInsertManyQuery(col config.CollectionDefinition, rng *rand.Rand, cfg *config.AppConfig) []interface{} {
+	count := cfg.InsertBatchSize
+	docs := make([]interface{}, count)
+	for i := 0; i < count; i++ {
+		select {
+		case docs[i] = <-InsertDocumentCache:
+		default:
+			docs[i] = workloads.GenerateDocument(col, cfg)
+		}
+	}
+	return docs
+}
+
 func insertDocumentProducer(ctx context.Context, col config.CollectionDefinition, cacheSize int, cfg *config.AppConfig) {
 	for {
 		select {
@@ -176,33 +190,36 @@ func runTransaction(ctx context.Context, id int, wCfg workloadConfig, rng *rand.
 	_, err = session.WithTransaction(ctx, func(sessCtx context.Context) (interface{}, error) {
 		numOps := rng.Intn(wCfg.appConfig.MaxTransactionOps) + 1
 		for i := 0; i < numOps; i++ {
-			// Select random collection for this step of the transaction
 			currentCol := wCfg.collections[rng.Intn(len(wCfg.collections))]
-
 			innerOp := selectOperation(wCfg.percentages, rng)
 			if innerOp == "aggregate" || innerOp == "transaction" {
 				innerOp = "find"
 			}
 
 			var q config.QueryDefinition
+			var insertManyDocs []interface{}
 			var run bool
 
-			if innerOp == "insert" {
+			switch innerOp {
+			case "insert":
 				q = generateInsertQuery(currentCol, rng, wCfg.appConfig)
 				run = true
-			} else {
+			case "insertMany":
+				insertManyDocs = generateInsertManyQuery(currentCol, rng, wCfg.appConfig)
+				run = true
+			default:
 				q, run = selectRandomQueryByType(sessCtx, wCfg.database, innerOp, wCfg.queryMap, currentCol, wCfg.debug, rng, wCfg.primaryFilterField, wCfg.appConfig)
 			}
 
 			if !run {
 				continue
 			}
 
-			coll := wCfg.database.Collection(q.Collection)
+			coll := wCfg.database.Collection(currentCol.Name)
 			filter := cloneMap(q.Filter)
 			processRecursive(filter, rng)
 
-			switch q.Operation {
+			switch innerOp {
 			case "find":
 				cursor, err := coll.Find(sessCtx, filter, options.Find().SetLimit(1))
 				if err == nil {
@@ -216,10 +233,14 @@ func runTransaction(ctx context.Context, id int, wCfg workloadConfig, rng *rand.
 			case "updateMany":
 				opts := options.UpdateMany().SetUpsert(q.Upsert)
 				_, err = coll.UpdateMany(sessCtx, filter, q.Update, opts)
-			case "deleteOne", "deleteMany":
+			case "deleteOne":
 				_, err = coll.DeleteOne(sessCtx, filter)
+			case "deleteMany":
+				_, err = coll.DeleteMany(sessCtx, filter)
 			case "insert":
 				_, err = coll.InsertOne(sessCtx, q.Filter)
+			case "insertMany":
+				_, err = coll.InsertMany(sessCtx, insertManyDocs)
 			}
 
 			if err != nil {
@@ -250,9 +271,7 @@ func independentWorker(ctx context.Context, id int, wg *sync.WaitGroup, wCfg wor
 		default:
 		}
 
-		// Pick random collection for this operation
 		currentCol := wCfg.collections[rng.Intn(len(wCfg.collections))]
-
 		opType := selectOperation(wCfg.percentages, rng)
 
 		if opType == "transaction" {
@@ -264,12 +283,16 @@ func independentWorker(ctx context.Context, id int, wg *sync.WaitGroup, wCfg wor
 		}
 
 		var q config.QueryDefinition
+		var insertManyDocs []interface{}
 		var run bool
 
 		switch opType {
 		case "insert":
 			q = generateInsertQuery(currentCol, rng, wCfg.appConfig)
 			run = true
+		case "insertMany":
+			insertManyDocs = generateInsertManyQuery(currentCol, rng, wCfg.appConfig)
+			run = true
 		case "find", "updateOne", "updateMany", "deleteOne", "deleteMany", "aggregate":
 			q, run = selectRandomQueryByType(dbOpCtx, wCfg.database, opType, wCfg.queryMap, currentCol, wCfg.debug, rng, wCfg.primaryFilterField, wCfg.appConfig)
 		default:
@@ -281,9 +304,7 @@ func independentWorker(ctx context.Context, id int, wg *sync.WaitGroup, wCfg wor
 			continue
 		}
 
-		db := wCfg.database
-		coll := db.Collection(q.Collection)
-
+		coll := wCfg.database.Collection(currentCol.Name)
 		var filter map[string]interface{}
 		var pipeline []interface{}
 
@@ -292,16 +313,16 @@ func independentWorker(ctx context.Context, id int, wg *sync.WaitGroup, wCfg wor
 				pipeline = cloned
 				processRecursive(pipeline, rng)
 			}
-		} else {
+		} else if opType != "insertMany" {
 			filter = cloneMap(q.Filter)
 			processRecursive(filter, rng)
 		}
 
 		start := time.Now()
 
-		switch q.Operation {
+		switch opType {
 		case "find":
-			limit := int64(q.Limit)
+			limit := q.Limit
 			if limit <= 0 {
 				limit = wCfg.findLimit
 			}
@@ -327,26 +348,28 @@ func independentWorker(ctx context.Context, id int, wg *sync.WaitGroup, wCfg wor
 				_ = cursor.Close(dbOpCtx)
 			}
 		case "updateOne":
-			// Use UpdateOne() for single document updates in v2
 			opts := options.UpdateOne().SetUpsert(q.Upsert)
 			_, err := coll.UpdateOne(dbOpCtx, filter, q.Update, opts)
 			if err != nil && wCfg.debug {
 				log.Printf("[Worker %d] UpdateOne error: %v", id, err)
 			}
 		case "updateMany":
-			// Use UpdateMany() for multiple document updates in v2
 			opts := options.UpdateMany().SetUpsert(q.Upsert)
 			_, err := coll.UpdateMany(dbOpCtx, filter, q.Update, opts)
 			if err != nil && wCfg.debug {
 				log.Printf("[Worker %d] UpdateMany error: %v", id, err)
 			}
-		case "deleteOne", "deleteMany":
+		case "deleteOne":
 			coll.DeleteOne(dbOpCtx, filter)
+		case "deleteMany":
+			coll.DeleteMany(dbOpCtx, filter)
 		case "insert":
 			coll.InsertOne(dbOpCtx, q.Filter)
+		case "insertMany":
+			coll.InsertMany(dbOpCtx, insertManyDocs)
 		}
 
-		wCfg.collector.Track(q.Operation, time.Since(start))
+		wCfg.collector.Track(opType, time.Since(start))
 	}
 }
 
@@ -436,6 +459,7 @@ func RunWorkload(ctx context.Context, db *mongo.Database, collections []config.C
 			"update":      cfg.UpdatePercent,
 			"delete":      cfg.DeletePercent,
 			"insert":      cfg.InsertPercent,
+			"insertMany":  cfg.BulkInsertPercent,
 			"aggregate":   cfg.AggregatePercent,
 			"transaction": cfg.TransactionPercent,
 		},
@@ -455,7 +479,6 @@ func runContinuousWorkload(ctx context.Context, wCfg workloadConfig) error {
 	workloadCtx, cancel := context.WithTimeout(ctx, wCfg.duration)
 	defer cancel()
 
-	// Handle initial document producer for multiple collections
 	for _, col := range wCfg.collections {
 		go insertDocumentProducer(workloadCtx, col, wCfg.maxInsertCache, wCfg.appConfig)
 	}
@@ -488,7 +511,7 @@ func runAllQueriesOnce(ctx context.Context, db *mongo.Database, queries []config
 	wg.Add(1)
 	go queryWorkerOnce(ctx, 1, tasks, &wg)
 	for i, q := range queries {
-		if q.Operation == "insert" {
+		if q.Operation == "insert" || q.Operation == "insertMany" {
 			continue
 		}
 		tasks <- &queryTask{