MongoDB Bulk Operations and Performance Optimization: Advanced Batch Processing for High-Throughput Applications

-- Traditional approach - processing documents one at a time (inefficient pattern)

-- Example: Processing user registration batch - individual operations
INSERT INTO users (name, email, registration_date, status) 
VALUES ('John Doe', '[email protected]', CURRENT_TIMESTAMP, 'pending');

INSERT INTO users (name, email, registration_date, status) 
VALUES ('Jane Smith', '[email protected]', CURRENT_TIMESTAMP, 'pending');

INSERT INTO users (name, email, registration_date, status) 
VALUES ('Bob Johnson', '[email protected]', CURRENT_TIMESTAMP, 'pending');

-- Problems with single-document operations:
-- 1. High network round-trip overhead for each operation
-- 2. Individual index updates and lock acquisitions
-- 3. Inefficient resource utilization and memory allocation
-- 4. Poor scaling characteristics under high load
-- 5. Complex error handling for partial failures
-- 6. Limited transaction scope and atomicity guarantees

-- Example: Updating user statuses individually (performance bottleneck)
UPDATE users SET status = 'active', activated_at = CURRENT_TIMESTAMP 
WHERE email = '[email protected]';

UPDATE users SET status = 'active', activated_at = CURRENT_TIMESTAMP 
WHERE email = '[email protected]';

UPDATE users SET status = 'active', activated_at = CURRENT_TIMESTAMP 
WHERE email = '[email protected]';

-- Individual updates result in:
-- - Multiple database connections and query parsing overhead
-- - Repeated index lookups and document retrieval operations  
-- - Inefficient write operations with individual lock acquisitions
-- - High latency due to network round trips
-- - Difficult error recovery and consistency management
-- - Poor resource utilization with context switching overhead

-- Example: Data cleanup operations (time-consuming individual deletes)
DELETE FROM users WHERE last_login < CURRENT_DATE - INTERVAL '2 years';
-- This approach processes each matching document individually

DELETE FROM user_sessions WHERE created_at < CURRENT_DATE - INTERVAL '30 days';
-- Again, individual document processing

DELETE FROM audit_logs WHERE log_date < CURRENT_DATE - INTERVAL '1 year';
-- More individual processing overhead

-- Single-document limitations:
-- 1. Long-running operations that block other requests
-- 2. Inefficient resource allocation and memory usage
-- 3. Poor progress tracking and monitoring capabilities
-- 4. Difficult to implement proper error handling
-- 5. No batch-level optimization opportunities
-- 6. Complex application logic for managing large datasets
-- 7. Limited ability to prioritize or throttle operations
-- 8. Inefficient use of database connection pooling

-- Traditional PostgreSQL bulk insert attempt (limited capabilities)
BEGIN;
INSERT INTO users (name, email, registration_date, status) VALUES
  ('User 1', '[email protected]', CURRENT_TIMESTAMP, 'pending'),
  ('User 2', '[email protected]', CURRENT_TIMESTAMP, 'pending'),
  ('User 3', '[email protected]', CURRENT_TIMESTAMP, 'pending');
  -- Limited to relatively small batches due to query size restrictions
  -- No advanced error handling or partial success reporting
  -- Limited optimization compared to native bulk operations
COMMIT;

-- PostgreSQL bulk update limitations
UPDATE users SET 
  status = CASE 
    WHEN email = '[email protected]' THEN 'active'
    WHEN email = '[email protected]' THEN 'suspended'
    WHEN email = '[email protected]' THEN 'active'
    ELSE status
  END,
  last_updated = CURRENT_TIMESTAMP
WHERE email IN ('[email protected]', '[email protected]', '[email protected]');

-- Issues with traditional bulk approaches:
-- 1. Complex SQL syntax for conditional updates
-- 2. Limited flexibility for different operations per document
-- 3. No built-in error reporting for individual items
-- 4. Query size limitations for large batches
-- 5. Poor performance characteristics compared to native bulk operations
-- 6. Limited monitoring and progress reporting capabilities

// MongoDB Advanced Bulk Operations - comprehensive batch processing with optimization

const { MongoClient } = require('mongodb');

// Advanced MongoDB Bulk Operations Manager
class MongoDBBulkOperationsManager {
  constructor(db) {
    this.db = db;
    this.performanceMetrics = {
      bulkInserts: { operations: 0, documentsProcessed: 0, totalTime: 0 },
      bulkUpdates: { operations: 0, documentsProcessed: 0, totalTime: 0 },
      bulkDeletes: { operations: 0, documentsProcessed: 0, totalTime: 0 },
      bulkWrites: { operations: 0, documentsProcessed: 0, totalTime: 0 }
    };
    this.errorTracking = new Map();
    this.optimizationSettings = {
      defaultBatchSize: 1000,
      maxBatchSize: 10000,
      enableOrdered: false, // Unordered operations for better performance
      enableBypassValidation: false,
      retryAttempts: 3,
      retryDelayMs: 1000
    };
  }

  // High-performance bulk insert operations
  async performBulkInsert(collectionName, documents, options = {}) {
    console.log(`Starting bulk insert of ${documents.length} documents into ${collectionName}`);

    const startTime = Date.now();
    const collection = this.db.collection(collectionName);

    // Configure bulk insert options for optimal performance
    const bulkOptions = {
      ordered: options.ordered !== undefined ? options.ordered : this.optimizationSettings.enableOrdered,
      bypassDocumentValidation: options.bypassValidation || this.optimizationSettings.enableBypassValidation,
      writeConcern: options.writeConcern || { w: 'majority', j: true }
    };

    try {
      // Process documents in optimal batch sizes
      const batchSize = Math.min(
        options.batchSize || this.optimizationSettings.defaultBatchSize,
        this.optimizationSettings.maxBatchSize
      );

      const results = [];
      let totalInserted = 0;
      let totalErrors = 0;

      for (let i = 0; i < documents.length; i += batchSize) {
        const batch = documents.slice(i, i + batchSize);

        try {
          console.log(`Processing batch ${Math.floor(i / batchSize) + 1} of ${Math.ceil(documents.length / batchSize)}`);

          // Add metadata to documents for tracking
          const enrichedBatch = batch.map(doc => ({
            ...doc,
            _bulk_operation_id: `bulk_insert_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`,
            _inserted_at: new Date(),
            _batch_number: Math.floor(i / batchSize) + 1
          }));

          const batchResult = await collection.insertMany(enrichedBatch, bulkOptions);

          results.push({
            batchIndex: Math.floor(i / batchSize),
            insertedCount: batchResult.insertedCount,
            insertedIds: batchResult.insertedIds,
            success: true
          });

          totalInserted += batchResult.insertedCount;

        } catch (error) {
          console.error(`Batch ${Math.floor(i / batchSize) + 1} failed:`, error.message);

          // Handle partial failures in unordered operations
          if (error.result && error.result.insertedCount) {
            totalInserted += error.result.insertedCount;
          }

          totalErrors += batch.length - (error.result?.insertedCount || 0);

          results.push({
            batchIndex: Math.floor(i / batchSize),
            insertedCount: error.result?.insertedCount || 0,
            error: error.message,
            success: false
          });

          // Track errors for analysis
          this.trackBulkOperationError('bulkInsert', error);
        }
      }

      const totalTime = Date.now() - startTime;

      // Update performance metrics
      this.updatePerformanceMetrics('bulkInserts', {
        operations: 1,
        documentsProcessed: totalInserted,
        totalTime: totalTime
      });

      const summary = {
        success: totalErrors === 0,
        totalDocuments: documents.length,
        insertedDocuments: totalInserted,
        failedDocuments: totalErrors,
        executionTimeMs: totalTime,
        throughputDocsPerSecond: Math.round((totalInserted / totalTime) * 1000),
        batchResults: results
      };

      console.log(`Bulk insert completed: ${totalInserted}/${documents.length} documents processed in ${totalTime}ms`);
      return summary;

    } catch (error) {
      console.error('Bulk insert operation failed:', error);
      this.trackBulkOperationError('bulkInsert', error);
      throw error;
    }
  }

  // Advanced bulk update operations with flexible patterns
  async performBulkUpdate(collectionName, updateOperations, options = {}) {
    console.log(`Starting bulk update of ${updateOperations.length} operations on ${collectionName}`);

    const startTime = Date.now();
    const collection = this.db.collection(collectionName);

    try {
      // Initialize ordered or unordered bulk operation
      const bulkOp = options.ordered ? collection.initializeOrderedBulkOp() : 
                                       collection.initializeUnorderedBulkOp();

      let operationCount = 0;

      // Process different types of update operations
      for (const operation of updateOperations) {
        const { filter, update, upsert = false, arrayFilters = null, hint = null } = operation;

        // Add operation metadata for tracking
        const enhancedUpdate = {
          ...update,
          $set: {
            ...update.$set,
            _last_bulk_update: new Date(),
            _bulk_operation_id: `bulk_update_${Date.now()}_${operationCount}`
          }
        };

        // Configure update operation based on type
        const updateConfig = { upsert };
        if (arrayFilters) updateConfig.arrayFilters = arrayFilters;
        if (hint) updateConfig.hint = hint;

        // Add to bulk operation
        if (operation.type === 'updateMany') {
          bulkOp.find(filter).updateMany(enhancedUpdate, updateConfig);
        } else {
          bulkOp.find(filter).updateOne(enhancedUpdate, updateConfig);
        }

        operationCount++;

        // Execute batch when reaching optimal size
        if (operationCount % this.optimizationSettings.defaultBatchSize === 0) {
          console.log(`Executing intermediate batch of ${this.optimizationSettings.defaultBatchSize} operations`);
        }
      }

      // Execute all bulk update operations
      console.log(`Executing ${operationCount} bulk update operations`);
      const result = await bulkOp.execute({
        writeConcern: options.writeConcern || { w: 'majority', j: true }
      });

      const totalTime = Date.now() - startTime;

      // Update performance metrics
      this.updatePerformanceMetrics('bulkUpdates', {
        operations: 1,
        documentsProcessed: result.modifiedCount + result.upsertedCount,
        totalTime: totalTime
      });

      const summary = {
        success: true,
        totalOperations: operationCount,
        matchedDocuments: result.matchedCount,
        modifiedDocuments: result.modifiedCount,
        upsertedDocuments: result.upsertedCount,
        upsertedIds: result.upsertedIds,
        executionTimeMs: totalTime,
        throughputOpsPerSecond: Math.round((operationCount / totalTime) * 1000),
        writeErrors: result.writeErrors || [],
        writeConcernErrors: result.writeConcernErrors || []
      };

      console.log(`Bulk update completed: ${result.modifiedCount} documents modified, ${result.upsertedCount} upserted in ${totalTime}ms`);
      return summary;

    } catch (error) {
      console.error('Bulk update operation failed:', error);
      this.trackBulkOperationError('bulkUpdate', error);

      // Return partial results if available
      if (error.result) {
        const totalTime = Date.now() - startTime;
        return {
          success: false,
          error: error.message,
          partialResult: {
            matchedDocuments: error.result.matchedCount,
            modifiedDocuments: error.result.modifiedCount,
            upsertedDocuments: error.result.upsertedCount,
            executionTimeMs: totalTime
          }
        };
      }
      throw error;
    }
  }

  // Optimized bulk delete operations
  async performBulkDelete(collectionName, deleteOperations, options = {}) {
    console.log(`Starting bulk delete of ${deleteOperations.length} operations on ${collectionName}`);

    const startTime = Date.now();
    const collection = this.db.collection(collectionName);

    try {
      // Initialize bulk operation
      const bulkOp = options.ordered ? collection.initializeOrderedBulkOp() : 
                                       collection.initializeUnorderedBulkOp();

      let operationCount = 0;

      // Process delete operations
      for (const operation of deleteOperations) {
        const { filter, deleteType = 'deleteMany', hint = null } = operation;

        // Configure delete operation
        const deleteConfig = {};
        if (hint) deleteConfig.hint = hint;

        // Add to bulk operation based on type
        if (deleteType === 'deleteOne') {
          bulkOp.find(filter).deleteOne();
        } else {
          bulkOp.find(filter).delete(); // deleteMany is default
        }

        operationCount++;
      }

      // Execute bulk delete operations
      console.log(`Executing ${operationCount} bulk delete operations`);
      const result = await bulkOp.execute({
        writeConcern: options.writeConcern || { w: 'majority', j: true }
      });

      const totalTime = Date.now() - startTime;

      // Update performance metrics
      this.updatePerformanceMetrics('bulkDeletes', {
        operations: 1,
        documentsProcessed: result.deletedCount,
        totalTime: totalTime
      });

      const summary = {
        success: true,
        totalOperations: operationCount,
        deletedDocuments: result.deletedCount,
        executionTimeMs: totalTime,
        throughputOpsPerSecond: Math.round((operationCount / totalTime) * 1000),
        writeErrors: result.writeErrors || [],
        writeConcernErrors: result.writeConcernErrors || []
      };

      console.log(`Bulk delete completed: ${result.deletedCount} documents deleted in ${totalTime}ms`);
      return summary;

    } catch (error) {
      console.error('Bulk delete operation failed:', error);
      this.trackBulkOperationError('bulkDelete', error);

      if (error.result) {
        const totalTime = Date.now() - startTime;
        return {
          success: false,
          error: error.message,
          partialResult: {
            deletedDocuments: error.result.deletedCount,
            executionTimeMs: totalTime
          }
        };
      }
      throw error;
    }
  }

  // Mixed bulk operations (insert, update, delete in single batch)
  async performMixedBulkOperations(collectionName, operations, options = {}) {
    console.log(`Starting mixed bulk operations: ${operations.length} operations on ${collectionName}`);

    const startTime = Date.now();
    const collection = this.db.collection(collectionName);

    try {
      const bulkOp = options.ordered ? collection.initializeOrderedBulkOp() : 
                                       collection.initializeUnorderedBulkOp();

      let insertCount = 0;
      let updateCount = 0;
      let deleteCount = 0;

      // Process mixed operations
      for (const operation of operations) {
        const { type, ...opData } = operation;

        switch (type) {
          case 'insert':
            const enrichedDoc = {
              ...opData.document,
              _bulk_operation_id: `bulk_mixed_${Date.now()}_${insertCount}`,
              _inserted_at: new Date()
            };
            bulkOp.insert(enrichedDoc);
            insertCount++;
            break;

          case 'updateOne':
            const updateOneData = {
              ...opData.update,
              $set: {
                ...opData.update.$set,
                _last_bulk_update: new Date(),
                _bulk_operation_id: `bulk_mixed_update_${Date.now()}_${updateCount}`
              }
            };
            bulkOp.find(opData.filter).updateOne(updateOneData, { upsert: opData.upsert || false });
            updateCount++;
            break;

          case 'updateMany':
            const updateManyData = {
              ...opData.update,
              $set: {
                ...opData.update.$set,
                _last_bulk_update: new Date(),
                _bulk_operation_id: `bulk_mixed_update_${Date.now()}_${updateCount}`
              }
            };
            bulkOp.find(opData.filter).updateMany(updateManyData, { upsert: opData.upsert || false });
            updateCount++;
            break;

          case 'deleteOne':
            bulkOp.find(opData.filter).deleteOne();
            deleteCount++;
            break;

          case 'deleteMany':
            bulkOp.find(opData.filter).delete();
            deleteCount++;
            break;

          default:
            console.warn(`Unknown operation type: ${type}`);
        }
      }

      // Execute mixed bulk operations
      console.log(`Executing mixed bulk operations: ${insertCount} inserts, ${updateCount} updates, ${deleteCount} deletes`);
      const result = await bulkOp.execute({
        writeConcern: options.writeConcern || { w: 'majority', j: true }
      });

      const totalTime = Date.now() - startTime;
      const totalDocumentsProcessed = result.insertedCount + result.modifiedCount + result.deletedCount + result.upsertedCount;

      // Update performance metrics
      this.updatePerformanceMetrics('bulkWrites', {
        operations: 1,
        documentsProcessed: totalDocumentsProcessed,
        totalTime: totalTime
      });

      const summary = {
        success: true,
        totalOperations: operations.length,
        operationBreakdown: {
          inserts: insertCount,
          updates: updateCount,
          deletes: deleteCount
        },
        results: {
          insertedDocuments: result.insertedCount,
          insertedIds: result.insertedIds,
          matchedDocuments: result.matchedCount,
          modifiedDocuments: result.modifiedCount,
          deletedDocuments: result.deletedCount,
          upsertedDocuments: result.upsertedCount,
          upsertedIds: result.upsertedIds
        },
        executionTimeMs: totalTime,
        throughputOpsPerSecond: Math.round((operations.length / totalTime) * 1000),
        throughputDocsPerSecond: Math.round((totalDocumentsProcessed / totalTime) * 1000),
        writeErrors: result.writeErrors || [],
        writeConcernErrors: result.writeConcernErrors || []
      };

      console.log(`Mixed bulk operations completed: ${totalDocumentsProcessed} documents processed in ${totalTime}ms`);
      return summary;

    } catch (error) {
      console.error('Mixed bulk operations failed:', error);
      this.trackBulkOperationError('bulkWrite', error);

      if (error.result) {
        const totalTime = Date.now() - startTime;
        const totalDocumentsProcessed = error.result.insertedCount + error.result.modifiedCount + error.result.deletedCount + error.result.upsertedCount;

        return {
          success: false,
          error: error.message,
          partialResult: {
            insertedDocuments: error.result.insertedCount,
            modifiedDocuments: error.result.modifiedCount,
            deletedDocuments: error.result.deletedCount,
            upsertedDocuments: error.result.upsertedCount,
            totalDocumentsProcessed: totalDocumentsProcessed,
            executionTimeMs: totalTime
          }
        };
      }
      throw error;
    }
  }

  // Performance monitoring and optimization
  updatePerformanceMetrics(operationType, metrics) {
    const current = this.performanceMetrics[operationType];
    current.operations += metrics.operations;
    current.documentsProcessed += metrics.documentsProcessed;
    current.totalTime += metrics.totalTime;
  }

  trackBulkOperationError(operationType, error) {
    if (!this.errorTracking.has(operationType)) {
      this.errorTracking.set(operationType, []);
    }

    this.errorTracking.get(operationType).push({
      timestamp: new Date(),
      error: error.message,
      code: error.code,
      details: error.writeErrors || error.result
    });
  }

  getBulkOperationStatistics() {
    const stats = {};

    for (const [operationType, metrics] of Object.entries(this.performanceMetrics)) {
      if (metrics.operations > 0) {
        stats[operationType] = {
          totalOperations: metrics.operations,
          documentsProcessed: metrics.documentsProcessed,
          averageExecutionTimeMs: Math.round(metrics.totalTime / metrics.operations),
          averageThroughputDocsPerSecond: Math.round((metrics.documentsProcessed / metrics.totalTime) * 1000),
          totalExecutionTimeMs: metrics.totalTime
        };
      }
    }

    return stats;
  }

  getErrorStatistics() {
    const errorStats = {};

    for (const [operationType, errors] of this.errorTracking.entries()) {
      errorStats[operationType] = {
        totalErrors: errors.length,
        recentErrors: errors.filter(e => Date.now() - e.timestamp.getTime() < 3600000), // Last hour
        errorBreakdown: this.groupErrorsByCode(errors)
      };
    }

    return errorStats;
  }

  groupErrorsByCode(errors) {
    const breakdown = {};
    errors.forEach(error => {
      const code = error.code || 'Unknown';
      breakdown[code] = (breakdown[code] || 0) + 1;
    });
    return breakdown;
  }

  // Optimized data import functionality
  async performOptimizedDataImport(collectionName, dataSource, options = {}) {
    console.log(`Starting optimized data import for ${collectionName}`);

    const importOptions = {
      batchSize: options.batchSize || 5000,
      enableValidation: options.enableValidation !== false,
      createIndexes: options.createIndexes || false,
      dropExistingCollection: options.dropExisting || false,
      parallelBatches: options.parallelBatches || 1
    };

    try {
      const collection = this.db.collection(collectionName);

      // Drop existing collection if requested
      if (importOptions.dropExistingCollection) {
        try {
          await collection.drop();
          console.log(`Existing collection ${collectionName} dropped`);
        } catch (error) {
          console.log(`Collection ${collectionName} did not exist or could not be dropped`);
        }
      }

      // Create indexes before import if specified
      if (importOptions.createIndexes && options.indexes) {
        console.log('Creating indexes before data import...');
        for (const indexSpec of options.indexes) {
          await collection.createIndex(indexSpec.fields, indexSpec.options);
        }
      }

      // Process data in optimized batches
      let totalImported = 0;
      const startTime = Date.now();

      // Assuming dataSource is an array or iterable
      const documents = Array.isArray(dataSource) ? dataSource : await this.convertDataSource(dataSource);

      const result = await this.performBulkInsert(collectionName, documents, {
        batchSize: importOptions.batchSize,
        bypassValidation: !importOptions.enableValidation,
        ordered: false // Unordered for better performance
      });

      console.log(`Data import completed: ${result.insertedDocuments} documents imported in ${result.executionTimeMs}ms`);
      return result;

    } catch (error) {
      console.error(`Data import failed for ${collectionName}:`, error);
      throw error;
    }
  }

  async convertDataSource(dataSource) {
    // Convert various data sources (streams, iterators, etc.) to arrays
    // This is a placeholder - implement based on your specific data source types
    if (typeof dataSource.toArray === 'function') {
      return await dataSource.toArray();
    }

    if (Symbol.iterator in dataSource) {
      return Array.from(dataSource);
    }

    throw new Error('Unsupported data source type');
  }
}

// Example usage: High-performance bulk operations
async function demonstrateBulkOperations() {
  const client = new MongoClient('mongodb://localhost:27017');
  await client.connect();
  const db = client.db('bulk_operations_demo');

  const bulkManager = new MongoDBBulkOperationsManager(db);

  // Demonstrate bulk insert
  const usersToInsert = [];
  for (let i = 0; i < 10000; i++) {
    usersToInsert.push({
      name: `User ${i}`,
      email: `user${i}@example.com`,
      age: Math.floor(Math.random() * 50) + 18,
      department: ['Engineering', 'Sales', 'Marketing', 'HR'][Math.floor(Math.random() * 4)],
      salary: Math.floor(Math.random() * 100000) + 40000,
      join_date: new Date(Date.now() - Math.random() * 365 * 24 * 60 * 60 * 1000)
    });
  }

  const insertResult = await bulkManager.performBulkInsert('users', usersToInsert);
  console.log('Bulk Insert Result:', insertResult);

  // Demonstrate bulk update
  const updateOperations = [
    {
      type: 'updateMany',
      filter: { department: 'Engineering' },
      update: { 
        $set: { department: 'Software Engineering' },
        $inc: { salary: 5000 }
      }
    },
    {
      type: 'updateMany', 
      filter: { age: { $lt: 25 } },
      update: { $set: { employee_type: 'junior' } },
      upsert: false
    }
  ];

  const updateResult = await bulkManager.performBulkUpdate('users', updateOperations);
  console.log('Bulk Update Result:', updateResult);

  // Display performance statistics
  const stats = bulkManager.getBulkOperationStatistics();
  console.log('Performance Statistics:', stats);

  await client.close();
}

// Production-ready MongoDB bulk operations with advanced optimization strategies
class EnterpriseMongoDBBulkManager extends MongoDBBulkOperationsManager {
  constructor(db, enterpriseConfig = {}) {
    super(db);

    this.enterpriseConfig = {
      enableShardingOptimization: enterpriseConfig.enableShardingOptimization || false,
      enableReplicationOptimization: enterpriseConfig.enableReplicationOptimization || false,
      enableCompressionOptimization: enterpriseConfig.enableCompressionOptimization || false,
      maxConcurrentOperations: enterpriseConfig.maxConcurrentOperations || 10,
      enableProgressTracking: enterpriseConfig.enableProgressTracking || true,
      enableResourceMonitoring: enterpriseConfig.enableResourceMonitoring || true
    };

    this.setupEnterpriseOptimizations();
  }

  async performParallelBulkOperations(collectionName, operationBatches, options = {}) {
    console.log(`Starting parallel bulk operations on ${collectionName} with ${operationBatches.length} batches`);

    const concurrency = Math.min(
      options.maxConcurrency || this.enterpriseConfig.maxConcurrentOperations,
      operationBatches.length
    );

    const results = [];
    const startTime = Date.now();

    // Process batches in parallel with controlled concurrency
    for (let i = 0; i < operationBatches.length; i += concurrency) {
      const batchPromises = [];

      for (let j = i; j < Math.min(i + concurrency, operationBatches.length); j++) {
        const batch = operationBatches[j];

        const promise = this.processSingleBatch(collectionName, batch, {
          ...options,
          batchIndex: j
        });

        batchPromises.push(promise);
      }

      // Wait for current set of concurrent batches to complete
      const batchResults = await Promise.allSettled(batchPromises);
      results.push(...batchResults);

      console.log(`Completed ${Math.min(i + concurrency, operationBatches.length)} of ${operationBatches.length} batches`);
    }

    const totalTime = Date.now() - startTime;

    return this.consolidateParallelResults(results, totalTime);
  }

  async processSingleBatch(collectionName, batch, options) {
    // Determine batch type and process accordingly
    if (batch.type === 'insert') {
      return await this.performBulkInsert(collectionName, batch.documents, options);
    } else if (batch.type === 'update') {
      return await this.performBulkUpdate(collectionName, batch.operations, options);
    } else if (batch.type === 'delete') {
      return await this.performBulkDelete(collectionName, batch.operations, options);
    } else if (batch.type === 'mixed') {
      return await this.performMixedBulkOperations(collectionName, batch.operations, options);
    }
  }

  async performShardOptimizedBulkOperations(collectionName, operations, shardKey) {
    console.log(`Performing shard-optimized bulk operations on ${collectionName}`);

    // Group operations by shard key for optimal routing
    const shardGroupedOps = this.groupOperationsByShardKey(operations, shardKey);

    const results = [];

    for (const [shardValue, shardOps] of shardGroupedOps.entries()) {
      console.log(`Processing ${shardOps.length} operations for shard key value: ${shardValue}`);

      const shardResult = await this.performMixedBulkOperations(collectionName, shardOps, {
        ordered: false // Better performance for sharded clusters
      });

      results.push({
        shardKey: shardValue,
        result: shardResult
      });
    }

    return this.consolidateShardResults(results);
  }

  groupOperationsByShardKey(operations, shardKey) {
    const grouped = new Map();

    for (const operation of operations) {
      let keyValue;

      if (operation.type === 'insert') {
        keyValue = operation.document[shardKey];
      } else {
        keyValue = operation.filter[shardKey];
      }

      if (!grouped.has(keyValue)) {
        grouped.set(keyValue, []);
      }

      grouped.get(keyValue).push(operation);
    }

    return grouped;
  }

  async performStreamingBulkOperations(collectionName, dataStream, options = {}) {
    console.log(`Starting streaming bulk operations on ${collectionName}`);

    const batchSize = options.batchSize || 1000;
    const processingOptions = {
      ordered: false,
      ...options
    };

    let batch = [];
    let totalProcessed = 0;
    const results = [];

    return new Promise((resolve, reject) => {
      dataStream.on('data', async (data) => {
        batch.push(data);

        if (batch.length >= batchSize) {
          try {
            const batchResult = await this.performBulkInsert(
              collectionName, 
              batch, 
              processingOptions
            );

            results.push(batchResult);
            totalProcessed += batchResult.insertedDocuments;
            batch = [];

            console.log(`Processed ${totalProcessed} documents so far`);

          } catch (error) {
            reject(error);
          }
        }
      });

      dataStream.on('end', async () => {
        try {
          // Process remaining documents
          if (batch.length > 0) {
            const finalResult = await this.performBulkInsert(
              collectionName, 
              batch, 
              processingOptions
            );
            results.push(finalResult);
            totalProcessed += finalResult.insertedDocuments;
          }

          resolve({
            success: true,
            totalProcessed: totalProcessed,
            batchResults: results
          });

        } catch (error) {
          reject(error);
        }
      });

      dataStream.on('error', reject);
    });
  }
}

-- QueryLeaf bulk operations with SQL-familiar syntax for MongoDB batch processing

-- Bulk insert with SQL VALUES syntax (automatically optimized for MongoDB bulk operations)
INSERT INTO users (name, email, age, department, salary, join_date)
VALUES 
  ('John Doe', '[email protected]', 32, 'Engineering', 85000, CURRENT_DATE),
  ('Jane Smith', '[email protected]', 28, 'Sales', 75000, CURRENT_DATE - INTERVAL '1 month'),
  ('Bob Johnson', '[email protected]', 35, 'Marketing', 70000, CURRENT_DATE - INTERVAL '2 months'),
  ('Alice Brown', '[email protected]', 29, 'HR', 68000, CURRENT_DATE - INTERVAL '3 months'),
  ('Charlie Wilson', '[email protected]', 31, 'Engineering', 90000, CURRENT_DATE - INTERVAL '4 months');

-- QueryLeaf automatically converts this to optimized MongoDB bulk insert:
-- db.users.insertMany([documents...], { ordered: false })

-- Bulk update operations using SQL UPDATE syntax
-- Update all engineers' salaries (automatically uses MongoDB bulk operations)
UPDATE users 
SET salary = salary * 1.1, 
    last_updated = CURRENT_TIMESTAMP,
    promotion_eligible = true
WHERE department = 'Engineering';

-- Update employees based on multiple conditions
UPDATE users 
SET employee_level = CASE 
  WHEN age > 35 AND salary > 80000 THEN 'Senior'
  WHEN age > 30 OR salary > 70000 THEN 'Mid-level'
  ELSE 'Junior'
END,
last_evaluation = CURRENT_DATE
WHERE join_date < CURRENT_DATE - INTERVAL '6 months';

-- QueryLeaf optimizes these as MongoDB bulk update operations:
-- Uses bulkWrite() with updateMany operations for optimal performance

-- Bulk delete operations
-- Clean up old inactive users
DELETE FROM users 
WHERE last_login < CURRENT_DATE - INTERVAL '2 years' 
  AND status = 'inactive';

-- Remove test data
DELETE FROM users 
WHERE email LIKE '%test%' OR email LIKE '%example%';

-- QueryLeaf converts to optimized bulk delete operations

-- Advanced bulk processing with data transformation and aggregation
WITH user_statistics AS (
  SELECT 
    department,
    COUNT(*) as employee_count,
    AVG(salary) as avg_salary,
    MAX(salary) as max_salary,
    MIN(join_date) as earliest_hire
  FROM users 
  GROUP BY department
),

salary_adjustments AS (
  SELECT 
    u._id,
    u.name,
    u.department,
    u.salary,
    us.avg_salary,

    -- Calculate adjustment based on department average
    CASE 
      WHEN u.salary < us.avg_salary * 0.8 THEN u.salary * 1.15  -- 15% increase
      WHEN u.salary < us.avg_salary * 0.9 THEN u.salary * 1.10  -- 10% increase  
      WHEN u.salary > us.avg_salary * 1.2 THEN u.salary * 1.02  -- 2% increase
      ELSE u.salary * 1.05  -- 5% standard increase
    END as new_salary,

    CURRENT_DATE as adjustment_date

  FROM users u
  JOIN user_statistics us ON u.department = us.department
  WHERE u.status = 'active'
)

-- Bulk update with calculated values (QueryLeaf optimizes this as bulk operation)
UPDATE users 
SET salary = sa.new_salary,
    last_salary_review = sa.adjustment_date,
    salary_review_reason = CONCAT('Department average adjustment - Previous: $', 
                                 CAST(sa.salary AS VARCHAR), 
                                 ', New: $', 
                                 CAST(sa.new_salary AS VARCHAR))
FROM salary_adjustments sa
WHERE users._id = sa._id;

-- Bulk data processing with conditional operations
-- Process employee performance reviews in batches
WITH performance_data AS (
  SELECT 
    _id,
    name,
    department,
    performance_score,

    -- Calculate performance category
    CASE 
      WHEN performance_score >= 90 THEN 'exceptional'
      WHEN performance_score >= 80 THEN 'exceeds_expectations'  
      WHEN performance_score >= 70 THEN 'meets_expectations'
      WHEN performance_score >= 60 THEN 'needs_improvement'
      ELSE 'unsatisfactory'
    END as performance_category,

    -- Calculate bonus eligibility
    CASE 
      WHEN performance_score >= 85 AND department IN ('Sales', 'Engineering') THEN true
      WHEN performance_score >= 90 THEN true
      ELSE false
    END as bonus_eligible,

    -- Calculate development plan requirement
    CASE 
      WHEN performance_score < 70 THEN true
      ELSE false  
    END as requires_development_plan

  FROM employees 
  WHERE review_period = '2025-Q3'
),

bonus_calculations AS (
  SELECT 
    pd._id,
    pd.bonus_eligible,

    -- Calculate bonus amount
    CASE 
      WHEN pd.performance_score >= 95 THEN u.salary * 0.15  -- 15% bonus
      WHEN pd.performance_score >= 90 THEN u.salary * 0.12  -- 12% bonus  
      WHEN pd.performance_score >= 85 THEN u.salary * 0.10  -- 10% bonus
      ELSE 0
    END as bonus_amount

  FROM performance_data pd
  JOIN users u ON pd._id = u._id
  WHERE pd.bonus_eligible = true
)

-- Execute bulk updates for performance review results
UPDATE users 
SET performance_category = pd.performance_category,
    bonus_eligible = pd.bonus_eligible,
    bonus_amount = COALESCE(bc.bonus_amount, 0),
    requires_development_plan = pd.requires_development_plan,
    last_performance_review = CURRENT_DATE,
    review_status = 'completed'
FROM performance_data pd
LEFT JOIN bonus_calculations bc ON pd._id = bc._id  
WHERE users._id = pd._id;

-- Advanced batch processing with data validation and error handling
-- Bulk data import with validation
INSERT INTO products (sku, name, category, price, stock_quantity, supplier_id, created_at)
SELECT 
  import_sku,
  import_name,
  import_category,
  CAST(import_price AS DECIMAL(10,2)),
  CAST(import_stock AS INTEGER),
  supplier_lookup.supplier_id,
  CURRENT_TIMESTAMP

FROM product_import_staging pis
JOIN suppliers supplier_lookup ON pis.supplier_name = supplier_lookup.name

-- Validation conditions
WHERE import_sku IS NOT NULL
  AND import_name IS NOT NULL  
  AND import_category IN ('Electronics', 'Clothing', 'Books', 'Home', 'Sports')
  AND import_price::DECIMAL(10,2) > 0
  AND import_stock::INTEGER >= 0
  AND supplier_lookup.supplier_id IS NOT NULL

  -- Duplicate check
  AND NOT EXISTS (
    SELECT 1 FROM products p 
    WHERE p.sku = pis.import_sku
  );

-- Bulk inventory adjustments with audit trail
WITH inventory_adjustments AS (
  SELECT 
    product_id,
    adjustment_quantity,
    adjustment_reason,
    adjustment_type, -- 'increase', 'decrease', 'recount'
    CURRENT_TIMESTAMP as adjustment_timestamp,
    'system' as adjusted_by
  FROM inventory_adjustment_queue
  WHERE processed = false
),

stock_calculations AS (
  SELECT 
    ia.product_id,
    p.stock_quantity as current_stock,

    CASE ia.adjustment_type
      WHEN 'increase' THEN p.stock_quantity + ia.adjustment_quantity
      WHEN 'decrease' THEN GREATEST(p.stock_quantity - ia.adjustment_quantity, 0)
      WHEN 'recount' THEN ia.adjustment_quantity
      ELSE p.stock_quantity
    END as new_stock_quantity,

    ia.adjustment_reason,
    ia.adjustment_timestamp,
    ia.adjusted_by

  FROM inventory_adjustments ia
  JOIN products p ON ia.product_id = p._id
)

-- Bulk update product stock levels
UPDATE products 
SET stock_quantity = sc.new_stock_quantity,
    last_stock_update = sc.adjustment_timestamp,
    stock_updated_by = sc.adjusted_by
FROM stock_calculations sc
WHERE products._id = sc.product_id;

-- Insert audit records for inventory changes
INSERT INTO inventory_audit_log (
  product_id,
  previous_stock,
  new_stock,
  adjustment_reason,
  adjustment_timestamp,
  adjusted_by
)
SELECT 
  sc.product_id,
  sc.current_stock,
  sc.new_stock_quantity,
  sc.adjustment_reason,
  sc.adjustment_timestamp,
  sc.adjusted_by
FROM stock_calculations sc;

-- Mark adjustment queue items as processed
UPDATE inventory_adjustment_queue 
SET processed = true,
    processed_at = CURRENT_TIMESTAMP
WHERE processed = false;

-- High-performance bulk operations with monitoring
-- Query for bulk operation performance analysis
WITH operation_metrics AS (
  SELECT 
    DATE_TRUNC('hour', operation_timestamp) as hour_bucket,
    operation_type, -- 'bulk_insert', 'bulk_update', 'bulk_delete'
    collection_name,

    -- Performance metrics
    COUNT(*) as operations_count,
    SUM(documents_processed) as total_documents,
    AVG(execution_time_ms) as avg_execution_time_ms,
    MAX(execution_time_ms) as max_execution_time_ms,
    MIN(execution_time_ms) as min_execution_time_ms,

    -- Throughput calculations
    AVG(throughput_docs_per_second) as avg_throughput_docs_per_sec,
    MAX(throughput_docs_per_second) as max_throughput_docs_per_sec,

    -- Error tracking
    COUNT(*) FILTER (WHERE success = false) as failed_operations,
    COUNT(*) FILTER (WHERE success = true) as successful_operations,

    -- Resource utilization
    AVG(memory_usage_mb) as avg_memory_usage_mb,
    AVG(cpu_utilization_percent) as avg_cpu_utilization

  FROM bulk_operation_log
  WHERE operation_timestamp >= CURRENT_TIMESTAMP - INTERVAL '24 hours'
  GROUP BY DATE_TRUNC('hour', operation_timestamp), operation_type, collection_name
)

SELECT 
  hour_bucket,
  operation_type,
  collection_name,
  operations_count,
  total_documents,

  -- Performance summary
  ROUND(avg_execution_time_ms, 2) as avg_execution_time_ms,
  ROUND(avg_throughput_docs_per_sec, 0) as avg_throughput_docs_per_sec,
  max_throughput_docs_per_sec,

  -- Success rate
  successful_operations,
  failed_operations,
  ROUND((successful_operations::DECIMAL / (successful_operations + failed_operations)) * 100, 2) as success_rate_percent,

  -- Resource efficiency  
  ROUND(avg_memory_usage_mb, 1) as avg_memory_usage_mb,
  ROUND(avg_cpu_utilization, 1) as avg_cpu_utilization_percent,

  -- Performance assessment
  CASE 
    WHEN avg_execution_time_ms < 100 AND success_rate_percent > 99 THEN 'excellent'
    WHEN avg_execution_time_ms < 500 AND success_rate_percent > 95 THEN 'good'
    WHEN avg_execution_time_ms < 1000 AND success_rate_percent > 90 THEN 'acceptable'
    ELSE 'needs_optimization'
  END as performance_rating

FROM operation_metrics
ORDER BY hour_bucket DESC, total_documents DESC;

-- QueryLeaf provides comprehensive bulk operation support:
-- 1. Automatic conversion of SQL batch operations to MongoDB bulk operations
-- 2. Optimal batching strategies based on operation types and data characteristics
-- 3. Advanced error handling with partial success reporting
-- 4. Performance monitoring and optimization recommendations
-- 5. Support for complex data transformations during bulk processing
-- 6. Intelligent resource utilization and concurrency management
-- 7. Integration with MongoDB's native bulk operation optimizations
-- 8. Familiar SQL syntax for complex batch processing workflows