MongoDB Bulk Operations and Batch Processing: High-Performance Data Operations and Enterprise-Scale Processing Optimization

-- Traditional PostgreSQL batch processing - inefficient and resource-intensive approaches

-- Single-record processing with significant overhead and poor performance
CREATE TABLE products_import (
    import_id BIGSERIAL PRIMARY KEY,
    product_id UUID DEFAULT gen_random_uuid(),
    product_name VARCHAR(200) NOT NULL,
    category VARCHAR(100),
    price DECIMAL(10,2) NOT NULL,
    stock_quantity INTEGER NOT NULL DEFAULT 0,
    supplier_id UUID,
    description TEXT,

    -- Import tracking and status management
    import_batch_id VARCHAR(100),
    import_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    import_status VARCHAR(50) DEFAULT 'pending',
    processing_attempts INTEGER DEFAULT 0,

    -- Validation and error tracking
    validation_errors TEXT[],
    processing_error TEXT,
    needs_review BOOLEAN DEFAULT FALSE,

    -- Performance tracking
    processing_start_time TIMESTAMP,
    processing_end_time TIMESTAMP,
    processing_duration_ms INTEGER
);

-- Inefficient single-record insert approach (extremely slow for large datasets)
DO $$
DECLARE
    product_record RECORD;
    processing_start TIMESTAMP;
    processing_end TIMESTAMP;
    error_count INTEGER := 0;
    success_count INTEGER := 0;
    batch_size INTEGER := 1000;
    current_batch INTEGER := 0;
    total_records INTEGER;
BEGIN
    -- Get total record count for progress tracking
    SELECT COUNT(*) INTO total_records FROM raw_product_data;
    RAISE NOTICE 'Processing % total records', total_records;

    -- Process each record individually (inefficient approach)
    FOR product_record IN 
        SELECT * FROM raw_product_data 
        ORDER BY import_order ASC
    LOOP
        processing_start := CURRENT_TIMESTAMP;

        BEGIN
            -- Individual record validation (repeated overhead)
            IF product_record.product_name IS NULL OR LENGTH(product_record.product_name) = 0 THEN
                RAISE EXCEPTION 'Invalid product name';
            END IF;

            IF product_record.price <= 0 THEN
                RAISE EXCEPTION 'Invalid price: %', product_record.price;
            END IF;

            -- Single record insert (high overhead per operation)
            INSERT INTO products_import (
                product_name,
                category,
                price,
                stock_quantity,
                supplier_id,
                description,
                import_batch_id,
                import_status,
                processing_start_time
            ) VALUES (
                product_record.product_name,
                product_record.category,
                product_record.price,
                product_record.stock_quantity,
                product_record.supplier_id,
                product_record.description,
                'batch_' || EXTRACT(EPOCH FROM CURRENT_TIMESTAMP),
                'processing',
                processing_start
            );

            processing_end := CURRENT_TIMESTAMP;

            -- Update processing time (additional overhead)
            UPDATE products_import 
            SET processing_end_time = processing_end,
                processing_duration_ms = EXTRACT(MILLISECONDS FROM processing_end - processing_start),
                import_status = 'completed'
            WHERE product_id = (SELECT product_id FROM products_import 
                              WHERE product_name = product_record.product_name 
                              ORDER BY import_timestamp DESC LIMIT 1);

            success_count := success_count + 1;

        EXCEPTION WHEN OTHERS THEN
            error_count := error_count + 1;

            -- Error logging with additional overhead
            INSERT INTO import_errors (
                import_batch_id,
                error_record_data,
                error_message,
                error_timestamp
            ) VALUES (
                'batch_' || EXTRACT(EPOCH FROM CURRENT_TIMESTAMP),
                row_to_json(product_record),
                SQLERRM,
                CURRENT_TIMESTAMP
            );
        END;

        -- Progress reporting overhead (every record)
        current_batch := current_batch + 1;
        IF current_batch % batch_size = 0 THEN
            RAISE NOTICE 'Processed % of % records (% success, % errors)', 
                current_batch, total_records, success_count, error_count;
        END IF;
    END LOOP;

    RAISE NOTICE 'Processing complete: % success, % errors', success_count, error_count;

END $$;

-- Batch processing with limited effectiveness and complex management
CREATE OR REPLACE FUNCTION process_product_batch(
    batch_id VARCHAR,
    batch_size INTEGER DEFAULT 1000,
    max_batches INTEGER DEFAULT 100
) 
RETURNS TABLE(
    batch_number INTEGER,
    records_processed INTEGER,
    records_success INTEGER,
    records_failed INTEGER,
    processing_time_ms INTEGER,
    total_processing_time_ms BIGINT
) AS $$
DECLARE
    current_batch INTEGER := 1;
    batch_start_time TIMESTAMP;
    batch_end_time TIMESTAMP;
    batch_processing_time INTEGER;
    total_start_time TIMESTAMP := CURRENT_TIMESTAMP;
    records_in_batch INTEGER;
    success_in_batch INTEGER;
    errors_in_batch INTEGER;

BEGIN
    -- Create batch processing table (overhead)
    CREATE TEMP TABLE IF NOT EXISTS current_batch_data AS
    SELECT * FROM raw_product_data WHERE 1=0;

    WHILE current_batch <= max_batches LOOP
        batch_start_time := CURRENT_TIMESTAMP;

        -- Clear previous batch data
        TRUNCATE current_batch_data;

        -- Load batch data (complex offset/limit approach)
        INSERT INTO current_batch_data
        SELECT *
        FROM raw_product_data
        WHERE processed = FALSE
        ORDER BY import_priority DESC, created_at ASC
        LIMIT batch_size;

        -- Check if batch has data
        SELECT COUNT(*) INTO records_in_batch FROM current_batch_data;
        EXIT WHEN records_in_batch = 0;

        success_in_batch := 0;
        errors_in_batch := 0;

        -- Process batch with individual operations (still inefficient)
        DECLARE
            batch_record RECORD;
        BEGIN
            FOR batch_record IN SELECT * FROM current_batch_data LOOP
                BEGIN
                    -- Validation logic (repeated for every record)
                    PERFORM validate_product_data(
                        batch_record.product_name,
                        batch_record.category,
                        batch_record.price,
                        batch_record.stock_quantity
                    );

                    -- Individual insert (suboptimal)
                    INSERT INTO products_import (
                        product_name,
                        category, 
                        price,
                        stock_quantity,
                        supplier_id,
                        description,
                        import_batch_id,
                        import_status
                    ) VALUES (
                        batch_record.product_name,
                        batch_record.category,
                        batch_record.price,
                        batch_record.stock_quantity,
                        batch_record.supplier_id,
                        batch_record.description,
                        batch_id,
                        'completed'
                    );

                    success_in_batch := success_in_batch + 1;

                EXCEPTION WHEN OTHERS THEN
                    errors_in_batch := errors_in_batch + 1;

                    -- Log error (additional overhead)
                    INSERT INTO batch_processing_errors (
                        batch_id,
                        batch_number,
                        record_data,
                        error_message,
                        error_timestamp
                    ) VALUES (
                        batch_id,
                        current_batch,
                        row_to_json(batch_record),
                        SQLERRM,
                        CURRENT_TIMESTAMP
                    );
                END;
            END LOOP;

        END;

        -- Mark records as processed (additional update overhead)
        UPDATE raw_product_data
        SET processed = TRUE,
            processed_batch = current_batch,
            processed_timestamp = CURRENT_TIMESTAMP
        WHERE id IN (SELECT id FROM current_batch_data);

        batch_end_time := CURRENT_TIMESTAMP;
        batch_processing_time := EXTRACT(MILLISECONDS FROM batch_end_time - batch_start_time);

        -- Return batch results
        batch_number := current_batch;
        records_processed := records_in_batch;
        records_success := success_in_batch;
        records_failed := errors_in_batch;
        processing_time_ms := batch_processing_time;
        total_processing_time_ms := EXTRACT(MILLISECONDS FROM batch_end_time - total_start_time);

        RETURN NEXT;

        current_batch := current_batch + 1;
    END LOOP;

    -- Cleanup
    DROP TABLE IF EXISTS current_batch_data;

END;
$$ LANGUAGE plpgsql;

-- Execute batch processing with limited control and monitoring
SELECT 
    bp.*,
    ROUND(bp.records_processed::NUMERIC / (bp.processing_time_ms / 1000.0), 2) as records_per_second,
    ROUND(bp.records_success::NUMERIC / bp.records_processed * 100, 2) as success_rate_percent
FROM process_product_batch('import_batch_2025', 5000, 50) bp
ORDER BY bp.batch_number;

-- Traditional approach limitations:
-- 1. Individual record processing with high per-operation overhead
-- 2. Limited batch optimization and inefficient resource utilization
-- 3. Complex error handling with poor performance during error conditions
-- 4. No built-in ordering guarantees or transaction-level consistency
-- 5. Difficult to monitor and control processing performance
-- 6. Limited scalability for very large datasets (millions of records)
-- 7. Complex progress tracking and status management overhead
-- 8. No automatic retry or recovery mechanisms for failed batches
-- 9. Inefficient memory usage and connection resource management
-- 10. Poor integration with modern distributed processing patterns

-- Complex bulk update attempt with limited effectiveness
WITH bulk_price_updates AS (
    SELECT 
        product_id,
        category,
        current_price,

        -- Calculate new prices based on complex business logic
        CASE category
            WHEN 'electronics' THEN current_price * 1.15  -- 15% increase
            WHEN 'clothing' THEN 
                CASE 
                    WHEN current_price > 100 THEN current_price * 1.10  -- 10% for high-end
                    ELSE current_price * 1.20  -- 20% for regular
                END
            WHEN 'books' THEN 
                CASE
                    WHEN stock_quantity > 50 THEN current_price * 0.95  -- 5% discount for overstocked
                    WHEN stock_quantity < 5 THEN current_price * 1.25   -- 25% increase for rare
                    ELSE current_price * 1.05  -- 5% standard increase
                END
            ELSE current_price * 1.08  -- 8% default increase
        END as new_price,

        -- Audit trail information
        'bulk_price_update_2025' as update_reason,
        CURRENT_TIMESTAMP as update_timestamp

    FROM products
    WHERE active = TRUE
    AND last_price_update < CURRENT_TIMESTAMP - INTERVAL '6 months'
),

update_validation AS (
    SELECT 
        bpu.*,

        -- Validation checks
        CASE 
            WHEN bpu.new_price <= 0 THEN 'invalid_price_zero_negative'
            WHEN bpu.new_price > bpu.current_price * 3 THEN 'price_increase_too_large'
            WHEN bpu.new_price < bpu.current_price * 0.5 THEN 'price_decrease_too_large'
            ELSE 'valid'
        END as validation_status,

        -- Price change analysis
        bpu.new_price - bpu.current_price as price_change,
        ROUND(((bpu.new_price - bpu.current_price) / bpu.current_price * 100)::NUMERIC, 2) as price_change_percent

    FROM bulk_price_updates bpu
),

validated_updates AS (
    SELECT *
    FROM update_validation
    WHERE validation_status = 'valid'
),

failed_updates AS (
    SELECT *
    FROM update_validation  
    WHERE validation_status != 'valid'
)

-- Execute bulk update (still limited by SQL constraints)
UPDATE products
SET 
    current_price = vu.new_price,
    previous_price = products.current_price,
    last_price_update = vu.update_timestamp,
    price_change_reason = vu.update_reason,
    price_change_amount = vu.price_change,
    price_change_percent = vu.price_change_percent,
    updated_at = CURRENT_TIMESTAMP
FROM validated_updates vu
WHERE products.product_id = vu.product_id;

-- Log failed updates for review
INSERT INTO price_update_errors (
    product_id,
    attempted_price,
    current_price,
    validation_error,
    error_timestamp,
    requires_manual_review
)
SELECT 
    fu.product_id,
    fu.new_price,
    fu.current_price,
    fu.validation_status,
    CURRENT_TIMESTAMP,
    TRUE
FROM failed_updates fu;

-- Limitations of traditional bulk processing:
-- 1. Limited by SQL's capabilities for complex bulk operations
-- 2. No native support for partial success handling in single operations
-- 3. Complex validation and error handling logic
-- 4. Poor performance optimization for very large datasets
-- 5. Difficult to monitor progress of long-running bulk operations
-- 6. No built-in retry mechanisms for transient failures
-- 7. Limited flexibility in operation ordering and dependency management
-- 8. Complex memory management for large batch operations
-- 9. No automatic optimization based on data distribution or system load
-- 10. Difficult integration with distributed systems and microservices

// MongoDB Advanced Bulk Operations and High-Performance Batch Processing System
const { MongoClient, BulkWriteResult } = require('mongodb');

const client = new MongoClient('mongodb://localhost:27017');
const db = client.db('bulk_operations_system');

// Comprehensive MongoDB Bulk Operations Manager
class AdvancedBulkOperationsManager {
  constructor(db, config = {}) {
    this.db = db;
    this.collections = {
      products: db.collection('products'),
      orders: db.collection('orders'),
      customers: db.collection('customers'),
      inventory: db.collection('inventory'),
      bulkOperationLog: db.collection('bulk_operation_log'),
      bulkOperationMetrics: db.collection('bulk_operation_metrics'),
      processingQueue: db.collection('processing_queue')
    };

    // Advanced bulk operations configuration
    this.config = {
      // Batch size optimization
      defaultBatchSize: config.defaultBatchSize || 1000,
      maxBatchSize: config.maxBatchSize || 10000,
      adaptiveBatchSizing: config.adaptiveBatchSizing !== false,

      // Performance optimization
      enableOrderedOperations: config.enableOrderedOperations !== false,
      enableParallelProcessing: config.enableParallelProcessing !== false,
      maxConcurrentBatches: config.maxConcurrentBatches || 5,

      // Error handling and recovery
      enableErrorRecovery: config.enableErrorRecovery !== false,
      maxRetries: config.maxRetries || 3,
      retryDelayMs: config.retryDelayMs || 1000,

      // Monitoring and metrics
      enableMetricsCollection: config.enableMetricsCollection !== false,
      enableProgressTracking: config.enableProgressTracking !== false,
      metricsReportingInterval: config.metricsReportingInterval || 10000,

      // Memory and resource management
      enableMemoryOptimization: config.enableMemoryOptimization !== false,
      maxMemoryUsageMB: config.maxMemoryUsageMB || 1024,
      enableGarbageCollection: config.enableGarbageCollection !== false
    };

    // Operational state management
    this.operationStats = {
      totalOperations: 0,
      successfulOperations: 0,
      failedOperations: 0,
      totalBatches: 0,
      avgBatchProcessingTime: 0,
      totalProcessingTime: 0
    };

    this.activeOperations = new Map();
    this.operationQueue = [];
    this.performanceMetrics = new Map();

    console.log('Advanced Bulk Operations Manager initialized');
  }

  async initializeBulkOperationsSystem() {
    console.log('Initializing comprehensive bulk operations system...');

    try {
      // Setup indexes for performance optimization
      await this.setupPerformanceIndexes();

      // Initialize metrics collection
      await this.initializeMetricsSystem();

      // Setup operation queue for large-scale processing
      await this.initializeProcessingQueue();

      // Configure memory and resource monitoring
      await this.setupResourceMonitoring();

      console.log('Bulk operations system initialized successfully');

    } catch (error) {
      console.error('Error initializing bulk operations system:', error);
      throw error;
    }
  }

  async performAdvancedBulkInsert(collectionName, documents, options = {}) {
    const operation = {
      operationId: `bulk_insert_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`,
      operationType: 'bulk_insert',
      collectionName: collectionName,
      documentsCount: documents.length,
      startTime: new Date(),
      status: 'processing'
    };

    console.log(`Starting bulk insert operation: ${operation.operationId}`);
    console.log(`Inserting ${documents.length} documents into ${collectionName}`);

    try {
      // Register operation for tracking
      this.activeOperations.set(operation.operationId, operation);

      // Validate and prepare documents
      const validatedDocuments = await this.validateAndPrepareDocuments(documents, 'insert');

      // Determine optimal batch configuration
      const batchConfig = await this.optimizeBatchConfiguration(validatedDocuments, options);

      // Execute bulk insert with advanced error handling
      const result = await this.executeBulkInsert(
        this.collections[collectionName], 
        validatedDocuments, 
        batchConfig,
        operation
      );

      // Update operation status
      operation.endTime = new Date();
      operation.status = 'completed';
      operation.result = result;
      operation.processingTime = operation.endTime - operation.startTime;

      // Log operation results
      await this.logBulkOperation(operation);

      // Update performance metrics
      await this.updateOperationMetrics(operation);

      console.log(`Bulk insert completed: ${operation.operationId}`);
      console.log(`Inserted ${result.insertedCount} documents successfully`);

      return result;

    } catch (error) {
      console.error(`Bulk insert failed: ${operation.operationId}`, error);

      // Handle operation failure
      operation.endTime = new Date();
      operation.status = 'failed';
      operation.error = {
        message: error.message,
        stack: error.stack
      };

      await this.handleOperationError(operation, error);
      throw error;

    } finally {
      // Cleanup operation tracking
      this.activeOperations.delete(operation.operationId);
    }
  }

  async executeBulkInsert(collection, documents, batchConfig, operation) {
    const results = {
      insertedCount: 0,
      insertedIds: [],
      errors: [],
      batches: [],
      totalBatches: Math.ceil(documents.length / batchConfig.batchSize)
    };

    console.log(`Executing bulk insert with ${results.totalBatches} batches of size ${batchConfig.batchSize}`);

    // Process documents in optimized batches
    for (let i = 0; i < documents.length; i += batchConfig.batchSize) {
      const batchStart = Date.now();
      const batch = documents.slice(i, i + batchConfig.batchSize);
      const batchNumber = Math.floor(i / batchConfig.batchSize) + 1;

      try {
        console.log(`Processing batch ${batchNumber}/${results.totalBatches} (${batch.length} documents)`);

        // Create bulk write operations for batch
        const bulkOps = batch.map(doc => ({
          insertOne: {
            document: {
              ...doc,
              _bulkOperationId: operation.operationId,
              _batchNumber: batchNumber,
              _insertedAt: new Date()
            }
          }
        }));

        // Execute bulk write with proper options
        const batchResult = await collection.bulkWrite(bulkOps, {
          ordered: batchConfig.ordered,
          bypassDocumentValidation: false,
          ...batchConfig.bulkWriteOptions
        });

        // Process batch results
        const batchProcessingTime = Date.now() - batchStart;
        const batchInfo = {
          batchNumber: batchNumber,
          documentsCount: batch.length,
          insertedCount: batchResult.insertedCount,
          processingTime: batchProcessingTime,
          insertedIds: Object.values(batchResult.insertedIds || {}),
          throughput: batch.length / (batchProcessingTime / 1000)
        };

        results.batches.push(batchInfo);
        results.insertedCount += batchResult.insertedCount;
        results.insertedIds.push(...batchInfo.insertedIds);

        // Update operation progress
        operation.progress = {
          batchesCompleted: batchNumber,
          totalBatches: results.totalBatches,
          documentsProcessed: i + batch.length,
          totalDocuments: documents.length,
          completionPercent: Math.round(((i + batch.length) / documents.length) * 100)
        };

        // Report progress periodically
        if (batchNumber % 10 === 0 || batchNumber === results.totalBatches) {
          console.log(`Progress: ${operation.progress.completionPercent}% (${operation.progress.documentsProcessed}/${operation.progress.totalDocuments})`);
        }

        // Adaptive batch size optimization based on performance
        if (this.config.adaptiveBatchSizing) {
          batchConfig = await this.adaptBatchSize(batchConfig, batchInfo);
        }

        // Memory pressure management
        if (this.config.enableMemoryOptimization) {
          await this.manageMemoryPressure();
        }

      } catch (batchError) {
        console.error(`Batch ${batchNumber} failed:`, batchError);

        // Handle batch-level errors
        const batchErrorInfo = {
          batchNumber: batchNumber,
          documentsCount: batch.length,
          error: {
            message: batchError.message,
            code: batchError.code,
            details: batchError.writeErrors || []
          },
          processingTime: Date.now() - batchStart
        };

        results.errors.push(batchErrorInfo);
        results.batches.push(batchErrorInfo);

        // Determine if operation should continue
        if (batchConfig.ordered && !batchConfig.continueOnError) {
          throw new Error(`Bulk insert failed at batch ${batchNumber}: ${batchError.message}`);
        }

        // Retry failed batch if enabled
        if (this.config.enableErrorRecovery) {
          await this.retryFailedBatch(collection, batch, batchConfig, batchNumber, operation);
        }
      }
    }

    // Calculate final metrics
    results.totalProcessingTime = Date.now() - operation.startTime.getTime();
    results.avgBatchProcessingTime = results.batches
      .filter(b => b.processingTime)
      .reduce((sum, b) => sum + b.processingTime, 0) / results.batches.length;
    results.overallThroughput = results.insertedCount / (results.totalProcessingTime / 1000);
    results.successRate = (results.insertedCount / documents.length) * 100;

    return results;
  }

  async performAdvancedBulkUpdate(collectionName, updates, options = {}) {
    const operation = {
      operationId: `bulk_update_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`,
      operationType: 'bulk_update',
      collectionName: collectionName,
      updatesCount: updates.length,
      startTime: new Date(),
      status: 'processing'
    };

    console.log(`Starting bulk update operation: ${operation.operationId}`);
    console.log(`Updating ${updates.length} documents in ${collectionName}`);

    try {
      // Register operation for tracking
      this.activeOperations.set(operation.operationId, operation);

      // Validate and prepare update operations
      const validatedUpdates = await this.validateAndPrepareUpdates(updates);

      // Optimize batch configuration for updates
      const batchConfig = await this.optimizeBatchConfiguration(validatedUpdates, options);

      // Execute bulk update operations
      const result = await this.executeBulkUpdate(
        this.collections[collectionName],
        validatedUpdates,
        batchConfig,
        operation
      );

      // Complete operation tracking
      operation.endTime = new Date();
      operation.status = 'completed';
      operation.result = result;
      operation.processingTime = operation.endTime - operation.startTime;

      // Log and report results
      await this.logBulkOperation(operation);
      await this.updateOperationMetrics(operation);

      console.log(`Bulk update completed: ${operation.operationId}`);
      console.log(`Updated ${result.modifiedCount} documents successfully`);

      return result;

    } catch (error) {
      console.error(`Bulk update failed: ${operation.operationId}`, error);

      operation.endTime = new Date();
      operation.status = 'failed';
      operation.error = {
        message: error.message,
        stack: error.stack
      };

      await this.handleOperationError(operation, error);
      throw error;

    } finally {
      this.activeOperations.delete(operation.operationId);
    }
  }

  async executeBulkUpdate(collection, updates, batchConfig, operation) {
    const results = {
      matchedCount: 0,
      modifiedCount: 0,
      upsertedCount: 0,
      upsertedIds: [],
      errors: [],
      batches: [],
      totalBatches: Math.ceil(updates.length / batchConfig.batchSize)
    };

    console.log(`Executing bulk update with ${results.totalBatches} batches`);

    // Process updates in optimized batches
    for (let i = 0; i < updates.length; i += batchConfig.batchSize) {
      const batchStart = Date.now();
      const batch = updates.slice(i, i + batchConfig.batchSize);
      const batchNumber = Math.floor(i / batchConfig.batchSize) + 1;

      try {
        console.log(`Processing update batch ${batchNumber}/${results.totalBatches} (${batch.length} operations)`);

        // Create bulk write operations
        const bulkOps = batch.map(update => {
          const updateOp = {
            filter: update.filter,
            update: {
              ...update.update,
              $set: {
                ...update.update.$set,
                _bulkOperationId: operation.operationId,
                _batchNumber: batchNumber,
                _lastUpdated: new Date()
              }
            }
          };

          if (update.upsert) {
            return {
              updateOne: {
                ...updateOp,
                upsert: true
              }
            };
          } else if (update.multi) {
            return {
              updateMany: updateOp
            };
          } else {
            return {
              updateOne: updateOp
            };
          }
        });

        // Execute bulk write
        const batchResult = await collection.bulkWrite(bulkOps, {
          ordered: batchConfig.ordered,
          bypassDocumentValidation: false,
          ...batchConfig.bulkWriteOptions
        });

        // Process batch results
        const batchProcessingTime = Date.now() - batchStart;
        const batchInfo = {
          batchNumber: batchNumber,
          operationsCount: batch.length,
          matchedCount: batchResult.matchedCount || 0,
          modifiedCount: batchResult.modifiedCount || 0,
          upsertedCount: batchResult.upsertedCount || 0,
          processingTime: batchProcessingTime,
          throughput: batch.length / (batchProcessingTime / 1000)
        };

        results.batches.push(batchInfo);
        results.matchedCount += batchInfo.matchedCount;
        results.modifiedCount += batchInfo.modifiedCount;
        results.upsertedCount += batchInfo.upsertedCount;

        if (batchResult.upsertedIds) {
          results.upsertedIds.push(...Object.values(batchResult.upsertedIds));
        }

        // Update progress tracking
        operation.progress = {
          batchesCompleted: batchNumber,
          totalBatches: results.totalBatches,
          operationsProcessed: i + batch.length,
          totalOperations: updates.length,
          completionPercent: Math.round(((i + batch.length) / updates.length) * 100)
        };

        // Progress reporting
        if (batchNumber % 5 === 0 || batchNumber === results.totalBatches) {
          console.log(`Update progress: ${operation.progress.completionPercent}% (${operation.progress.operationsProcessed}/${operation.progress.totalOperations})`);
        }

      } catch (batchError) {
        console.error(`Update batch ${batchNumber} failed:`, batchError);

        const batchErrorInfo = {
          batchNumber: batchNumber,
          operationsCount: batch.length,
          error: {
            message: batchError.message,
            code: batchError.code,
            writeErrors: batchError.writeErrors || []
          },
          processingTime: Date.now() - batchStart
        };

        results.errors.push(batchErrorInfo);
        results.batches.push(batchErrorInfo);

        if (batchConfig.ordered && !batchConfig.continueOnError) {
          throw new Error(`Bulk update failed at batch ${batchNumber}: ${batchError.message}`);
        }
      }
    }

    // Calculate final metrics
    results.totalProcessingTime = Date.now() - operation.startTime.getTime();
    results.avgBatchProcessingTime = results.batches
      .filter(b => b.processingTime)
      .reduce((sum, b) => sum + b.processingTime, 0) / results.batches.length;
    results.overallThroughput = results.modifiedCount / (results.totalProcessingTime / 1000);
    results.successRate = (results.modifiedCount / updates.length) * 100;

    return results;
  }

  async performAdvancedBulkDelete(collectionName, filters, options = {}) {
    const operation = {
      operationId: `bulk_delete_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`,
      operationType: 'bulk_delete',
      collectionName: collectionName,
      filtersCount: filters.length,
      startTime: new Date(),
      status: 'processing'
    };

    console.log(`Starting bulk delete operation: ${operation.operationId}`);
    console.log(`Deleting documents with ${filters.length} filter conditions in ${collectionName}`);

    try {
      this.activeOperations.set(operation.operationId, operation);

      // Validate and prepare delete operations
      const validatedFilters = await this.validateAndPrepareDeletes(filters);

      // Optimize batch configuration for deletes
      const batchConfig = await this.optimizeBatchConfiguration(validatedFilters, options);

      // Execute bulk delete operations
      const result = await this.executeBulkDelete(
        this.collections[collectionName],
        validatedFilters,
        batchConfig,
        operation
      );

      // Complete operation
      operation.endTime = new Date();
      operation.status = 'completed';
      operation.result = result;
      operation.processingTime = operation.endTime - operation.startTime;

      await this.logBulkOperation(operation);
      await this.updateOperationMetrics(operation);

      console.log(`Bulk delete completed: ${operation.operationId}`);
      console.log(`Deleted ${result.deletedCount} documents successfully`);

      return result;

    } catch (error) {
      console.error(`Bulk delete failed: ${operation.operationId}`, error);

      operation.endTime = new Date();
      operation.status = 'failed';
      operation.error = {
        message: error.message,
        stack: error.stack
      };

      await this.handleOperationError(operation, error);
      throw error;

    } finally {
      this.activeOperations.delete(operation.operationId);
    }
  }

  async executeBulkDelete(collection, filters, batchConfig, operation) {
    const results = {
      deletedCount: 0,
      errors: [],
      batches: [],
      totalBatches: Math.ceil(filters.length / batchConfig.batchSize)
    };

    console.log(`Executing bulk delete with ${results.totalBatches} batches`);

    for (let i = 0; i < filters.length; i += batchConfig.batchSize) {
      const batchStart = Date.now();
      const batch = filters.slice(i, i + batchConfig.batchSize);
      const batchNumber = Math.floor(i / batchConfig.batchSize) + 1;

      try {
        console.log(`Processing delete batch ${batchNumber}/${results.totalBatches} (${batch.length} operations)`);

        // Create bulk delete operations
        const bulkOps = batch.map(filter => ({
          deleteMany: {
            filter: filter
          }
        }));

        // Execute bulk write
        const batchResult = await collection.bulkWrite(bulkOps, {
          ordered: batchConfig.ordered,
          ...batchConfig.bulkWriteOptions
        });

        const batchProcessingTime = Date.now() - batchStart;
        const batchInfo = {
          batchNumber: batchNumber,
          operationsCount: batch.length,
          deletedCount: batchResult.deletedCount || 0,
          processingTime: batchProcessingTime,
          throughput: (batchResult.deletedCount || 0) / (batchProcessingTime / 1000)
        };

        results.batches.push(batchInfo);
        results.deletedCount += batchInfo.deletedCount;

        // Update progress
        operation.progress = {
          batchesCompleted: batchNumber,
          totalBatches: results.totalBatches,
          operationsProcessed: i + batch.length,
          totalOperations: filters.length,
          completionPercent: Math.round(((i + batch.length) / filters.length) * 100)
        };

      } catch (batchError) {
        console.error(`Delete batch ${batchNumber} failed:`, batchError);

        const batchErrorInfo = {
          batchNumber: batchNumber,
          operationsCount: batch.length,
          error: {
            message: batchError.message,
            code: batchError.code
          },
          processingTime: Date.now() - batchStart
        };

        results.errors.push(batchErrorInfo);
        results.batches.push(batchErrorInfo);
      }
    }

    results.totalProcessingTime = Date.now() - operation.startTime.getTime();
    results.overallThroughput = results.deletedCount / (results.totalProcessingTime / 1000);

    return results;
  }

  async validateAndPrepareDocuments(documents, operationType) {
    console.log(`Validating and preparing ${documents.length} documents for ${operationType}`);

    const validatedDocuments = [];
    const validationErrors = [];

    for (let i = 0; i < documents.length; i++) {
      const doc = documents[i];

      try {
        // Basic validation
        if (!doc || typeof doc !== 'object') {
          throw new Error('Document must be a valid object');
        }

        // Add operation metadata
        const preparedDoc = {
          ...doc,
          _operationType: operationType,
          _operationTimestamp: new Date(),
          _validatedAt: new Date()
        };

        // Type-specific validation
        if (operationType === 'insert') {
          // Ensure no _id conflicts for inserts
          if (preparedDoc._id) {
            // Keep existing _id but validate it's unique
          }
        }

        validatedDocuments.push(preparedDoc);

      } catch (error) {
        validationErrors.push({
          index: i,
          document: doc,
          error: error.message
        });
      }
    }

    if (validationErrors.length > 0) {
      console.warn(`Found ${validationErrors.length} validation errors out of ${documents.length} documents`);

      // Log validation errors
      await this.collections.bulkOperationLog.insertOne({
        operationType: 'validation',
        validationErrors: validationErrors,
        timestamp: new Date()
      });
    }

    console.log(`Validation complete: ${validatedDocuments.length} valid documents`);
    return validatedDocuments;
  }

  async optimizeBatchConfiguration(data, options) {
    const dataSize = data.length;
    let optimalBatchSize = this.config.defaultBatchSize;

    // Adaptive batch size based on data volume
    if (this.config.adaptiveBatchSizing) {
      if (dataSize > 100000) {
        optimalBatchSize = Math.min(this.config.maxBatchSize, 5000);
      } else if (dataSize > 10000) {
        optimalBatchSize = 2000;
      } else if (dataSize > 1000) {
        optimalBatchSize = 1000;
      } else {
        optimalBatchSize = Math.max(100, dataSize);
      }
    }

    // Consider memory constraints
    if (this.config.enableMemoryOptimization) {
      const estimatedMemoryPerDoc = 1; // KB estimate
      const totalMemoryMB = (dataSize * estimatedMemoryPerDoc) / 1024;

      if (totalMemoryMB > this.config.maxMemoryUsageMB) {
        const memoryAdjustedBatchSize = Math.floor(
          (this.config.maxMemoryUsageMB * 1024) / estimatedMemoryPerDoc
        );
        optimalBatchSize = Math.min(optimalBatchSize, memoryAdjustedBatchSize);
      }
    }

    const batchConfig = {
      batchSize: optimalBatchSize,
      ordered: options.ordered !== false,
      continueOnError: options.continueOnError === true,
      bulkWriteOptions: {
        writeConcern: options.writeConcern || { w: 'majority' },
        ...(options.bulkWriteOptions || {})
      }
    };

    console.log(`Optimized batch configuration: size=${batchConfig.batchSize}, ordered=${batchConfig.ordered}`);
    return batchConfig;
  }

  async logBulkOperation(operation) {
    try {
      await this.collections.bulkOperationLog.insertOne({
        operationId: operation.operationId,
        operationType: operation.operationType,
        collectionName: operation.collectionName,
        status: operation.status,
        startTime: operation.startTime,
        endTime: operation.endTime,
        processingTime: operation.processingTime,
        result: operation.result,
        error: operation.error,
        progress: operation.progress,
        createdAt: new Date()
      });
    } catch (error) {
      console.warn('Error logging bulk operation:', error.message);
    }
  }

  async updateOperationMetrics(operation) {
    try {
      // Update global statistics
      this.operationStats.totalOperations++;
      if (operation.status === 'completed') {
        this.operationStats.successfulOperations++;
      } else {
        this.operationStats.failedOperations++;
      }

      if (operation.result && operation.result.batches) {
        this.operationStats.totalBatches += operation.result.batches.length;

        const avgBatchTime = operation.result.avgBatchProcessingTime;
        if (avgBatchTime) {
          this.operationStats.avgBatchProcessingTime = 
            (this.operationStats.avgBatchProcessingTime + avgBatchTime) / 2;
        }
      }

      // Store detailed metrics
      await this.collections.bulkOperationMetrics.insertOne({
        operationId: operation.operationId,
        operationType: operation.operationType,
        collectionName: operation.collectionName,
        metrics: {
          processingTime: operation.processingTime,
          throughput: operation.result ? operation.result.overallThroughput : null,
          successRate: operation.result ? operation.result.successRate : null,
          batchCount: operation.result ? operation.result.batches.length : null,
          avgBatchTime: operation.result ? operation.result.avgBatchProcessingTime : null
        },
        timestamp: new Date()
      });

    } catch (error) {
      console.warn('Error updating operation metrics:', error.message);
    }
  }

  async generateBulkOperationsReport() {
    console.log('Generating bulk operations performance report...');

    try {
      const report = {
        timestamp: new Date(),
        globalStats: { ...this.operationStats },
        activeOperations: this.activeOperations.size,

        // Recent operations analysis
        recentOperations: await this.collections.bulkOperationLog.find({
          startTime: { $gte: new Date(Date.now() - 24 * 60 * 60 * 1000) }
        }).sort({ startTime: -1 }).limit(50).toArray(),

        // Performance metrics
        performanceMetrics: await this.collections.bulkOperationMetrics.aggregate([
          {
            $match: {
              timestamp: { $gte: new Date(Date.now() - 24 * 60 * 60 * 1000) }
            }
          },
          {
            $group: {
              _id: '$operationType',
              count: { $sum: 1 },
              avgProcessingTime: { $avg: '$metrics.processingTime' },
              avgThroughput: { $avg: '$metrics.throughput' },
              avgSuccessRate: { $avg: '$metrics.successRate' },
              totalBatches: { $sum: '$metrics.batchCount' }
            }
          }
        ]).toArray()
      };

      // Calculate health indicators
      report.healthIndicators = {
        successRate: this.operationStats.totalOperations > 0 ? 
          (this.operationStats.successfulOperations / this.operationStats.totalOperations * 100).toFixed(2) : 0,
        avgProcessingTime: this.operationStats.avgBatchProcessingTime,
        systemLoad: this.activeOperations.size,
        status: this.activeOperations.size > 10 ? 'high_load' : 
                this.operationStats.failedOperations > this.operationStats.successfulOperations ? 'degraded' : 'healthy'
      };

      return report;

    } catch (error) {
      console.error('Error generating bulk operations report:', error);
      return {
        timestamp: new Date(),
        error: error.message,
        globalStats: this.operationStats
      };
    }
  }

  // Additional helper methods for comprehensive bulk operations management
  async setupPerformanceIndexes() {
    console.log('Setting up performance indexes for bulk operations...');

    // Index for operation logging and metrics
    await this.collections.bulkOperationLog.createIndex(
      { operationId: 1, startTime: -1 },
      { background: true }
    );

    await this.collections.bulkOperationMetrics.createIndex(
      { operationType: 1, timestamp: -1 },
      { background: true }
    );
  }

  async adaptBatchSize(currentConfig, batchInfo) {
    // Adaptive batch size optimization based on performance
    if (batchInfo.throughput < 100) { // documents per second
      currentConfig.batchSize = Math.max(100, Math.floor(currentConfig.batchSize * 0.8));
    } else if (batchInfo.throughput > 1000) {
      currentConfig.batchSize = Math.min(this.config.maxBatchSize, Math.floor(currentConfig.batchSize * 1.2));
    }

    return currentConfig;
  }

  async manageMemoryPressure() {
    if (this.config.enableGarbageCollection) {
      if (global.gc) {
        global.gc();
      }
    }
  }
}

// Benefits of MongoDB Advanced Bulk Operations:
// - Native bulk operation support with minimal overhead and maximum throughput
// - Sophisticated error handling with partial success support and retry mechanisms
// - Adaptive batch sizing and performance optimization based on data characteristics
// - Comprehensive operation tracking and monitoring with detailed metrics
// - Memory and resource management for large-scale data processing
// - Built-in transaction-level consistency and ordering guarantees
// - Flexible operation types (insert, update, delete, upsert) with advanced filtering
// - Scalable architecture supporting millions of documents efficiently
// - Integration with MongoDB's native indexing and query optimization
// - SQL-compatible bulk operations through QueryLeaf integration

module.exports = {
  AdvancedBulkOperationsManager
};