MongoDB Aggregation Pipeline Optimization: SQL-Style Performance Tuning for Complex Data Analytics

-- Traditional SQL complex analytics - performance challenges at scale
WITH regional_sales AS (
  SELECT 
    r.region_name,
    p.category,
    p.subcategory,
    DATE_TRUNC('month', o.order_date) as month,
    SUM(oi.quantity * oi.unit_price) as gross_revenue,
    SUM(oi.quantity * p.cost_basis) as cost_of_goods,
    COUNT(DISTINCT o.customer_id) as unique_customers,
    COUNT(o.order_id) as total_orders
  FROM orders o
  JOIN order_items oi ON o.order_id = oi.order_id
  JOIN products p ON oi.product_id = p.product_id
  JOIN customers c ON o.customer_id = c.customer_id
  JOIN regions r ON c.region_id = r.region_id
  WHERE o.order_date >= '2024-01-01'
    AND o.status IN ('completed', 'shipped')
  GROUP BY r.region_name, p.category, p.subcategory, DATE_TRUNC('month', o.order_date)
),
monthly_trends AS (
  SELECT 
    region_name,
    category,
    month,
    SUM(gross_revenue) as monthly_revenue,
    SUM(cost_of_goods) as monthly_costs,
    (SUM(gross_revenue) - SUM(cost_of_goods)) as monthly_profit,
    SUM(unique_customers) as monthly_customers,
    SUM(total_orders) as monthly_orders,

    -- Window functions for trend analysis
    LAG(SUM(gross_revenue), 1) OVER (
      PARTITION BY region_name, category 
      ORDER BY month
    ) as previous_month_revenue,

    PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY SUM(gross_revenue)) OVER (
      PARTITION BY region_name, category
    ) as median_monthly_revenue
  FROM regional_sales
  GROUP BY region_name, category, month
)
SELECT 
  region_name,
  category,
  month,
  monthly_revenue,
  monthly_profit,
  monthly_customers,

  -- Growth calculations
  ROUND(
    ((monthly_revenue - previous_month_revenue) / previous_month_revenue) * 100, 2
  ) as revenue_growth_percent,

  -- Performance vs median
  ROUND(
    (monthly_revenue / median_monthly_revenue) * 100, 2
  ) as performance_vs_median,

  -- Customer metrics
  ROUND(monthly_revenue / monthly_customers, 2) as revenue_per_customer,
  ROUND(monthly_orders / monthly_customers, 2) as orders_per_customer

FROM monthly_trends
WHERE month >= '2024-06-01'
ORDER BY region_name, category, month;

-- Problems with traditional approaches:
-- - Complex joins across multiple large tables
-- - Window functions require full data scanning
-- - Memory intensive for large datasets
-- - Limited horizontal scaling capabilities
-- - Rigid schema requirements
-- - Poor performance with nested/dynamic data structures
-- - Difficult to optimize for distributed processing

// MongoDB optimized aggregation pipeline - high performance analytics
db.orders.aggregate([
  // Stage 1: Early filtering with index support
  {
    $match: {
      orderDate: { $gte: ISODate('2024-01-01') },
      status: { $in: ['completed', 'shipped'] }
    }
  },

  // Stage 2: Efficient lookup with optimized joins
  {
    $lookup: {
      from: 'customers',
      localField: 'customerId',
      foreignField: '_id',
      as: 'customer',
      pipeline: [
        { $project: { regionId: 1, _id: 0 } } // Project only needed fields
      ]
    }
  },

  // Stage 3: Unwind with preserveNullAndEmptyArrays for performance
  { $unwind: '$customer' },
  { $unwind: '$items' },

  // Stage 4: Second lookup for product data
  {
    $lookup: {
      from: 'products',
      localField: 'items.productId',
      foreignField: '_id',
      as: 'product',
      pipeline: [
        { $project: { category: 1, subcategory: 1, costBasis: 1, _id: 0 } }
      ]
    }
  },

  { $unwind: '$product' },

  // Stage 5: Third lookup for region data
  {
    $lookup: {
      from: 'regions',
      localField: 'customer.regionId',
      foreignField: '_id',
      as: 'region',
      pipeline: [
        { $project: { regionName: 1, _id: 0 } }
      ]
    }
  },

  { $unwind: '$region' },

  // Stage 6: Add computed fields efficiently
  {
    $addFields: {
      month: { 
        $dateFromParts: {
          year: { $year: '$orderDate' },
          month: { $month: '$orderDate' },
          day: 1
        }
      },
      itemRevenue: { $multiply: ['$items.quantity', '$items.unitPrice'] },
      itemCost: { $multiply: ['$items.quantity', '$product.costBasis'] }
    }
  },

  // Stage 7: Group for initial aggregation
  {
    $group: {
      _id: {
        region: '$region.regionName',
        category: '$product.category',
        subcategory: '$product.subcategory',
        month: '$month'
      },
      grossRevenue: { $sum: '$itemRevenue' },
      costOfGoods: { $sum: '$itemCost' },
      uniqueCustomers: { $addToSet: '$customerId' },
      totalOrders: { $sum: 1 }
    }
  },

  // Stage 8: Transform unique customers to count
  {
    $addFields: {
      uniqueCustomerCount: { $size: '$uniqueCustomers' }
    }
  },

  // Stage 9: Project final structure
  {
    $project: {
      region: '$_id.region',
      category: '$_id.category',
      subcategory: '$_id.subcategory',
      month: '$_id.month',
      grossRevenue: 1,
      costOfGoods: 1,
      profit: { $subtract: ['$grossRevenue', '$costOfGoods'] },
      uniqueCustomerCount: 1,
      totalOrders: 1,
      revenuePerCustomer: {
        $round: [
          { $divide: ['$grossRevenue', '$uniqueCustomerCount'] },
          2
        ]
      },
      ordersPerCustomer: {
        $round: [
          { $divide: ['$totalOrders', '$uniqueCustomerCount'] },
          2
        ]
      },
      _id: 0
    }
  },

  // Stage 10: Sort for consistent output
  {
    $sort: {
      region: 1,
      category: 1,
      month: 1
    }
  },

  // Stage 11: Add window functions for trend analysis
  {
    $setWindowFields: {
      partitionBy: { region: '$region', category: '$category' },
      sortBy: { month: 1 },
      output: {
        previousMonthRevenue: {
          $shift: {
            output: '$grossRevenue',
            by: -1
          }
        },
        medianMonthlyRevenue: {
          $median: '$grossRevenue',
          window: {
            documents: ['unbounded preceding', 'unbounded following']
          }
        },
        revenueGrowthPercent: {
          $round: [
            {
              $multiply: [
                {
                  $divide: [
                    { $subtract: ['$grossRevenue', '$previousMonthRevenue'] },
                    '$previousMonthRevenue'
                  ]
                },
                100
              ]
            },
            2
          ]
        }
      }
    }
  },

  // Stage 12: Final filtering for recent months
  {
    $match: {
      month: { $gte: ISODate('2024-06-01') }
    }
  }
], {
  // Pipeline options for optimization
  allowDiskUse: true,        // Allow spilling to disk for large datasets
  maxTimeMS: 300000,         // 5 minute timeout
  hint: { orderDate: 1, status: 1 }, // Suggest index usage
  readConcern: { level: 'majority' }  // Consistency level
});

// Benefits of optimized aggregation pipelines:
// - Early filtering reduces data volume through pipeline
// - Efficient $lookup stages with projected fields
// - Strategic index utilization
// - Memory-efficient processing with disk spilling
// - Native support for complex analytical operations
// - Horizontal scaling across shards
// - Flexible handling of nested/dynamic data
// - Built-in window functions for trend analysis

// Advanced aggregation pipeline optimization patterns
class AggregationOptimizer {
  constructor(db) {
    this.db = db;
    this.performanceMetrics = new Map();
    this.indexRecommendations = [];
  }

  async optimizeEarlyFiltering(collection, pipeline) {
    // Move filtering stages as early as possible
    const optimizedPipeline = [];
    const filterStages = [];
    const nonFilterStages = [];

    // Separate filter stages from other stages
    pipeline.forEach(stage => {
      const stageType = Object.keys(stage)[0];
      if (stageType === '$match' || stageType === '$limit') {
        filterStages.push(stage);
      } else {
        nonFilterStages.push(stage);
      }
    });

    // Early filtering reduces document flow through pipeline
    optimizedPipeline.push(...filterStages);
    optimizedPipeline.push(...nonFilterStages);

    return optimizedPipeline;
  }

  async createProjectionOptimizedPipeline(baseCollection, lookupCollections, projections) {
    // Optimize projections and lookups for minimal data transfer
    return [
      // Stage 1: Early projection to reduce document size
      {
        $project: {
          // Only include fields needed for subsequent stages
          ...projections.baseFields,
          // Include fields needed for lookups
          ...projections.lookupKeys
        }
      },

      // Stage 2: Optimized lookups with sub-pipelines
      ...lookupCollections.map(lookup => ({
        $lookup: {
          from: lookup.collection,
          localField: lookup.localField,
          foreignField: lookup.foreignField,
          as: lookup.as,
          pipeline: [
            // Project only needed fields in lookup
            { $project: lookup.projection },
            // Add filters within lookup when possible
            ...(lookup.filters ? [{ $match: lookup.filters }] : [])
          ]
        }
      })),

      // Stage 3: Unwind with null preservation for performance
      ...lookupCollections.map(lookup => ({
        $unwind: {
          path: `$${lookup.as}`,
          preserveNullAndEmptyArrays: lookup.preserveNulls || false
        }
      })),

      // Stage 4: Final projection after all joins
      {
        $project: projections.finalFields
      }
    ];
  }

  async analyzeIndexUsage(collection, pipeline) {
    // Analyze pipeline for index optimization opportunities
    const explanation = await this.db.collection(collection).aggregate(
      pipeline,
      { explain: true }
    ).toArray();

    const indexAnalysis = {
      stagesAnalyzed: [],
      indexesUsed: [],
      indexesRecommended: [],
      performanceIssues: []
    };

    // Analyze each stage for index usage
    explanation.forEach((stage, index) => {
      const stageType = Object.keys(pipeline[index])[0];
      const stageAnalysis = {
        stage: index,
        type: stageType,
        indexUsed: false,
        collectionScanned: false,
        documentsExamined: 0,
        documentsReturned: 0
      };

      if (stage.executionStats) {
        stageAnalysis.indexUsed = stage.executionStats.executionTimeMillisEstimate < 100;
        stageAnalysis.documentsExamined = stage.executionStats.totalDocsExamined;
        stageAnalysis.documentsReturned = stage.executionStats.totalDocsReturned;

        // Identify inefficient stages
        if (stageAnalysis.documentsExamined > stageAnalysis.documentsReturned * 10) {
          indexAnalysis.performanceIssues.push({
            stage: index,
            issue: 'high_document_examination_ratio',
            ratio: stageAnalysis.documentsExamined / stageAnalysis.documentsReturned,
            recommendation: 'Consider adding index for this stage'
          });
        }
      }

      indexAnalysis.stagesAnalyzed.push(stageAnalysis);
    });

    return indexAnalysis;
  }

  async createPerformanceOptimizedPipeline(collection, analyticsQuery) {
    // Create comprehensive performance-optimized pipeline
    const pipeline = [
      // Stage 1: Efficient date range filtering with index
      {
        $match: {
          [analyticsQuery.dateField]: {
            $gte: analyticsQuery.startDate,
            $lte: analyticsQuery.endDate
          },
          // Add compound index filters
          ...analyticsQuery.filters
        }
      },

      // Stage 2: Early sampling for large datasets (if needed)
      ...(analyticsQuery.sampleSize ? [{
        $sample: { size: analyticsQuery.sampleSize }
      }] : []),

      // Stage 3: Efficient faceted search
      {
        $facet: {
          // Main aggregation pipeline
          data: [
            // Lookup with optimized sub-pipeline
            {
              $lookup: {
                from: analyticsQuery.lookupCollection,
                localField: analyticsQuery.localField,
                foreignField: analyticsQuery.foreignField,
                as: 'lookupData',
                pipeline: [
                  { $project: analyticsQuery.lookupProjection },
                  { $limit: 1 } // Limit lookup results when appropriate
                ]
              }
            },

            { $unwind: '$lookupData' },

            // Grouping with efficient accumulators
            {
              $group: {
                _id: analyticsQuery.groupBy,

                // Use $sum for counting instead of $addToSet when possible
                totalCount: { $sum: 1 },
                totalValue: { $sum: analyticsQuery.valueField },
                averageValue: { $avg: analyticsQuery.valueField },

                // Efficient min/max calculations
                minValue: { $min: analyticsQuery.valueField },
                maxValue: { $max: analyticsQuery.valueField },

                // Use $push only when needed for arrays
                ...(analyticsQuery.collectArrays ? {
                  samples: { $push: analyticsQuery.sampleField }
                } : {})
              }
            },

            // Add calculated fields
            {
              $addFields: {
                efficiency: {
                  $round: [
                    { $divide: ['$totalValue', '$totalCount'] },
                    2
                  ]
                },
                valueRange: { $subtract: ['$maxValue', '$minValue'] }
              }
            },

            // Sort for consistent results
            { $sort: { totalValue: -1 } },

            // Limit results to prevent memory issues
            { $limit: analyticsQuery.maxResults || 1000 }
          ],

          // Metadata pipeline for counts and statistics
          metadata: [
            {
              $group: {
                _id: null,
                totalDocuments: { $sum: 1 },
                totalValue: { $sum: analyticsQuery.valueField },
                avgValue: { $avg: analyticsQuery.valueField }
              }
            }
          ]
        }
      },

      // Stage 4: Combine faceted results
      {
        $project: {
          data: 1,
          metadata: { $arrayElemAt: ['$metadata', 0] },
          processingTimestamp: new Date()
        }
      }
    ];

    return pipeline;
  }

  async benchmarkPipeline(collection, pipeline, options = {}) {
    // Comprehensive pipeline performance benchmarking
    const benchmarkResults = {
      pipelineName: options.name || 'unnamed_pipeline',
      startTime: new Date(),
      stages: [],
      totalExecutionTime: 0,
      documentsProcessed: 0,
      memoryUsage: 0,
      indexesUsed: [],
      recommendations: []
    };

    try {
      // Get execution statistics
      const startTime = Date.now();
      const explanation = await this.db.collection(collection).aggregate(
        pipeline,
        { 
          explain: true,
          allowDiskUse: true,
          ...options
        }
      ).toArray();

      // Analyze execution plan
      explanation.forEach((stageExplan, index) => {
        const stageBenchmark = {
          stageIndex: index,
          stageType: Object.keys(pipeline[index])[0],
          executionTimeMs: stageExplan.executionStats?.executionTimeMillisEstimate || 0,
          documentsIn: stageExplan.executionStats?.totalDocsExamined || 0,
          documentsOut: stageExplan.executionStats?.totalDocsReturned || 0,
          indexUsed: stageExplan.executionStats?.inputStage?.stage === 'IXSCAN',
          memoryUsageBytes: stageExplan.executionStats?.memUsage || 0
        };

        benchmarkResults.stages.push(stageBenchmark);
        benchmarkResults.totalExecutionTime += stageBenchmark.executionTimeMs;
        benchmarkResults.memoryUsage += stageBenchmark.memoryUsageBytes;
      });

      // Run actual pipeline for real-world timing
      const realStartTime = Date.now();
      const results = await this.db.collection(collection).aggregate(
        pipeline,
        { allowDiskUse: true, ...options }
      ).toArray();

      const realExecutionTime = Date.now() - realStartTime;
      benchmarkResults.realExecutionTime = realExecutionTime;
      benchmarkResults.documentsProcessed = results.length;

      // Generate recommendations
      benchmarkResults.recommendations = this.generateOptimizationRecommendations(
        benchmarkResults
      );

    } catch (error) {
      benchmarkResults.error = error.message;
    } finally {
      benchmarkResults.endTime = new Date();
    }

    // Store benchmark results for comparison
    this.performanceMetrics.set(benchmarkResults.pipelineName, benchmarkResults);

    return benchmarkResults;
  }

  generateOptimizationRecommendations(benchmarkResults) {
    const recommendations = [];

    // Check for stages without index usage
    benchmarkResults.stages.forEach((stage, index) => {
      if (!stage.indexUsed && stage.documentsIn > 1000) {
        recommendations.push({
          type: 'index_recommendation',
          stage: index,
          message: `Consider adding index for stage ${index} (${stage.stageType})`,
          priority: 'high',
          potentialImprovement: 'significant'
        });
      }

      if (stage.documentsIn > stage.documentsOut * 100) {
        recommendations.push({
          type: 'filtering_recommendation',
          stage: index,
          message: `Move filtering earlier in pipeline for stage ${index}`,
          priority: 'medium',
          potentialImprovement: 'moderate'
        });
      }
    });

    // Memory usage recommendations
    if (benchmarkResults.memoryUsage > 100 * 1024 * 1024) { // 100MB
      recommendations.push({
        type: 'memory_optimization',
        message: 'High memory usage detected - consider using allowDiskUse: true',
        priority: 'medium',
        potentialImprovement: 'prevents memory errors'
      });
    }

    // Execution time recommendations
    if (benchmarkResults.totalExecutionTime > 30000) { // 30 seconds
      recommendations.push({
        type: 'performance_optimization',
        message: 'Long execution time - review pipeline optimization opportunities',
        priority: 'high',
        potentialImprovement: 'significant'
      });
    }

    return recommendations;
  }

  async createIndexRecommendations(collection, commonPipelines) {
    // Generate index recommendations based on common pipeline patterns
    const recommendations = [];

    for (const pipeline of commonPipelines) {
      const analysis = await this.analyzeIndexUsage(collection, pipeline.stages);

      pipeline.stages.forEach((stage, index) => {
        const stageType = Object.keys(stage)[0];

        switch (stageType) {
          case '$match':
            const matchFields = Object.keys(stage.$match);
            if (matchFields.length > 0) {
              recommendations.push({
                type: 'compound_index',
                collection: collection,
                fields: matchFields,
                reason: `Optimize $match stage ${index}`,
                estimatedImprovement: 'high'
              });
            }
            break;

          case '$sort':
            const sortFields = Object.keys(stage.$sort);
            recommendations.push({
              type: 'sort_index',
              collection: collection,
              fields: sortFields,
              reason: `Optimize $sort stage ${index}`,
              estimatedImprovement: 'high'
            });
            break;

          case '$group':
            const groupField = stage.$group._id;
            if (typeof groupField === 'string' && groupField.startsWith('$')) {
              recommendations.push({
                type: 'grouping_index',
                collection: collection,
                fields: [groupField.substring(1)],
                reason: `Optimize $group stage ${index}`,
                estimatedImprovement: 'medium'
              });
            }
            break;
        }
      });
    }

    // Deduplicate and prioritize recommendations
    return this.prioritizeIndexRecommendations(recommendations);
  }

  prioritizeIndexRecommendations(recommendations) {
    // Remove duplicates and prioritize by impact
    const uniqueRecommendations = new Map();

    recommendations.forEach(rec => {
      const key = `${rec.collection}_${rec.fields.join('_')}`;
      const existing = uniqueRecommendations.get(key);

      if (!existing || this.getImpactScore(rec) > this.getImpactScore(existing)) {
        uniqueRecommendations.set(key, rec);
      }
    });

    return Array.from(uniqueRecommendations.values())
      .sort((a, b) => this.getImpactScore(b) - this.getImpactScore(a));
  }

  getImpactScore(recommendation) {
    const impactScores = {
      high: 3,
      medium: 2,
      low: 1
    };
    return impactScores[recommendation.estimatedImprovement] || 0;
  }

  async generatePerformanceReport() {
    // Generate comprehensive performance analysis report
    const report = {
      generatedAt: new Date(),
      totalPipelinesAnalyzed: this.performanceMetrics.size,
      performanceSummary: {
        fastPipelines: 0,      // < 1 second
        moderatePipelines: 0,  // 1-10 seconds
        slowPipelines: 0       // > 10 seconds
      },
      topPerformers: [],
      performanceIssues: [],
      indexRecommendations: [],
      overallRecommendations: []
    };

    // Analyze all benchmarked pipelines
    for (const [name, metrics] of this.performanceMetrics.entries()) {
      const executionTime = metrics.realExecutionTime || metrics.totalExecutionTime;

      if (executionTime < 1000) {
        report.performanceSummary.fastPipelines++;
      } else if (executionTime < 10000) {
        report.performanceSummary.moderatePipelines++;
      } else {
        report.performanceSummary.slowPipelines++;
      }

      // Identify top performers and issues
      if (executionTime < 500 && metrics.documentsProcessed > 1000) {
        report.topPerformers.push({
          name: name,
          executionTime: executionTime,
          documentsProcessed: metrics.documentsProcessed,
          efficiency: metrics.documentsProcessed / executionTime
        });
      }

      if (executionTime > 30000 || metrics.memoryUsage > 500 * 1024 * 1024) {
        report.performanceIssues.push({
          name: name,
          executionTime: executionTime,
          memoryUsage: metrics.memoryUsage,
          recommendations: metrics.recommendations
        });
      }
    }

    // Sort top performers by efficiency
    report.topPerformers.sort((a, b) => b.efficiency - a.efficiency);

    // Generate overall recommendations
    if (report.performanceSummary.slowPipelines > 0) {
      report.overallRecommendations.push(
        'Multiple slow pipelines detected - prioritize optimization efforts'
      );
    }

    if (this.indexRecommendations.length > 5) {
      report.overallRecommendations.push(
        'Consider implementing recommended indexes to improve query performance'
      );
    }

    return report;
  }
}

// Advanced memory management and optimization strategies
class AggregationMemoryManager {
  constructor(db) {
    this.db = db;
    this.memoryThresholds = {
      warning: 100 * 1024 * 1024,    // 100MB
      critical: 500 * 1024 * 1024,   // 500MB
      maximum: 1024 * 1024 * 1024    // 1GB
    };
  }

  async createMemoryEfficientPipeline(collection, aggregationConfig) {
    // Design pipeline with memory efficiency in mind
    const memoryOptimizedPipeline = [
      // Stage 1: Early filtering to reduce dataset size
      {
        $match: {
          ...aggregationConfig.filters,
          // Add indexed filters first
          [aggregationConfig.dateField]: {
            $gte: aggregationConfig.startDate,
            $lte: aggregationConfig.endDate
          }
        }
      },

      // Stage 2: Project only necessary fields early
      {
        $project: {
          // Include only fields needed for processing
          ...aggregationConfig.requiredFields,
          // Exclude large text fields unless necessary
          ...(aggregationConfig.excludeFields.reduce((acc, field) => {
            acc[field] = 0;
            return acc;
          }, {}))
        }
      },

      // Stage 3: Use streaming-friendly operations
      {
        $group: {
          _id: aggregationConfig.groupBy,

          // Use memory-efficient accumulators
          count: { $sum: 1 },
          totalValue: { $sum: aggregationConfig.valueField },

          // Avoid $addToSet for large arrays - use $mergeObjects for smaller sets
          ...(aggregationConfig.collectSets && aggregationConfig.expectedSetSize < 1000 ? {
            uniqueValues: { $addToSet: aggregationConfig.setField }
          } : {}),

          // Use $first/$last instead of $push for single values
          firstValue: { $first: aggregationConfig.valueField },
          lastValue: { $last: aggregationConfig.valueField },

          // Calculated fields at group level to avoid later processing
          averageValue: { $avg: aggregationConfig.valueField }
        }
      },

      // Stage 4: Add computed fields efficiently
      {
        $addFields: {
          efficiency: {
            $cond: {
              if: { $gt: ['$count', 0] },
              then: { $divide: ['$totalValue', '$count'] },
              else: 0
            }
          },

          // Avoid complex calculations on large arrays
          setSize: {
            $cond: {
              if: { $isArray: '$uniqueValues' },
              then: { $size: '$uniqueValues' },
              else: 0
            }
          }
        }
      },

      // Stage 5: Sort with limit to prevent large result sets
      { $sort: { totalValue: -1 } },
      { $limit: aggregationConfig.maxResults || 10000 },

      // Stage 6: Final projection to minimize output size
      {
        $project: {
          groupKey: '$_id',
          metrics: {
            count: '$count',
            totalValue: '$totalValue',
            averageValue: '$averageValue',
            efficiency: '$efficiency'
          },
          _id: 0
        }
      }
    ];

    return memoryOptimizedPipeline;
  }

  async processLargeDatasetWithBatching(collection, pipeline, batchConfig) {
    // Process large datasets in batches to manage memory
    const results = [];
    const batchSize = batchConfig.batchSize || 10000;
    const totalBatches = Math.ceil(batchConfig.totalDocuments / batchSize);

    console.log(`Processing ${batchConfig.totalDocuments} documents in ${totalBatches} batches`);

    for (let batch = 0; batch < totalBatches; batch++) {
      const skip = batch * batchSize;

      const batchPipeline = [
        // Add skip and limit for batching
        { $skip: skip },
        { $limit: batchSize },

        // Original pipeline stages
        ...pipeline
      ];

      try {
        const batchResults = await this.db.collection(collection).aggregate(
          batchPipeline,
          {
            allowDiskUse: true,
            maxTimeMS: 60000, // 1 minute per batch
            readConcern: { level: 'available' } // Use available for better performance
          }
        ).toArray();

        results.push(...batchResults);

        console.log(`Completed batch ${batch + 1}/${totalBatches} (${batchResults.length} results)`);

        // Optional: Add delay between batches to reduce load
        if (batchConfig.delayMs && batch < totalBatches - 1) {
          await new Promise(resolve => setTimeout(resolve, batchConfig.delayMs));
        }

      } catch (error) {
        console.error(`Batch ${batch + 1} failed:`, error.message);

        // Optionally continue with remaining batches
        if (batchConfig.continueOnError) {
          continue;
        } else {
          throw error;
        }
      }
    }

    return results;
  }

  async createStreamingAggregation(collection, pipeline, outputHandler) {
    // Create streaming aggregation for real-time processing
    const cursor = this.db.collection(collection).aggregate(pipeline, {
      allowDiskUse: true,
      batchSize: 1000, // Small batch size for streaming
      readConcern: { level: 'available' }
    });

    const streamingStats = {
      documentsProcessed: 0,
      startTime: new Date(),
      memoryPeakUsage: 0,
      batchesProcessed: 0
    };

    try {
      while (await cursor.hasNext()) {
        const document = await cursor.next();

        // Process document through handler
        await outputHandler(document, streamingStats);

        streamingStats.documentsProcessed++;

        // Monitor memory usage (approximate)
        if (streamingStats.documentsProcessed % 1000 === 0) {
          const memoryUsage = process.memoryUsage();
          streamingStats.memoryPeakUsage = Math.max(
            streamingStats.memoryPeakUsage,
            memoryUsage.heapUsed
          );

          console.log(`Processed ${streamingStats.documentsProcessed} documents, Memory: ${Math.round(memoryUsage.heapUsed / 1024 / 1024)}MB`);
        }
      }

    } finally {
      await cursor.close();
      streamingStats.endTime = new Date();
      streamingStats.totalProcessingTime = streamingStats.endTime - streamingStats.startTime;
    }

    return streamingStats;
  }

  async optimizePipelineForLargeArrays(collection, pipeline, arrayOptimizations) {
    // Optimize pipelines that work with large arrays
    const optimizedPipeline = [];

    pipeline.forEach((stage, index) => {
      const stageType = Object.keys(stage)[0];

      switch (stageType) {
        case '$unwind':
          // Add preserveNullAndEmptyArrays and includeArrayIndex for efficiency
          optimizedPipeline.push({
            $unwind: {
              path: stage.$unwind.path || stage.$unwind,
              preserveNullAndEmptyArrays: true,
              includeArrayIndex: `${stage.$unwind.path || stage.$unwind}_index`
            }
          });
          break;

        case '$group':
          // Optimize group operations for array handling
          const groupStage = { ...stage };

          // Replace $addToSet with $mergeObjects for better performance when possible
          Object.keys(groupStage.$group).forEach(key => {
            if (key !== '_id') {
              const accumulator = groupStage.$group[key];

              if (accumulator.$addToSet && arrayOptimizations.convertAddToSetToMerge) {
                // Convert to more efficient operation when possible
                groupStage.$group[key] = { $push: accumulator.$addToSet };
              }
            }
          });

          optimizedPipeline.push(groupStage);
          break;

        case '$project':
          // Optimize array operations in projection
          const projectStage = { ...stage };

          Object.keys(projectStage.$project).forEach(key => {
            const projection = projectStage.$project[key];

            // Replace array operations with more efficient alternatives
            if (projection && typeof projection === 'object' && projection.$size) {
              // $size can be expensive on very large arrays
              if (arrayOptimizations.approximateArraySizes) {
                projectStage.$project[`${key}_approx`] = {
                  $cond: {
                    if: { $isArray: projection.$size },
                    then: { $min: [{ $size: projection.$size }, 10000] }, // Cap at 10k
                    else: 0
                  }
                };
              }
            }
          });

          optimizedPipeline.push(projectStage);
          break;

        default:
          optimizedPipeline.push(stage);
      }
    });

    // Add array-specific optimizations
    if (arrayOptimizations.limitArrayProcessing) {
      // Add $limit stages after $unwind to prevent processing too many array elements
      optimizedPipeline.forEach((stage, index) => {
        if (stage.$unwind && index < optimizedPipeline.length - 1) {
          optimizedPipeline.splice(index + 1, 0, {
            $limit: arrayOptimizations.maxArrayElements || 100000
          });
        }
      });
    }

    return optimizedPipeline;
  }

  async monitorAggregationPerformance(collection, pipeline, options = {}) {
    // Comprehensive performance monitoring for aggregations
    const performanceMonitor = {
      startTime: new Date(),
      memorySnapshots: [],
      stageTimings: [],
      resourceUsage: {
        cpuStart: process.cpuUsage(),
        memoryStart: process.memoryUsage()
      }
    };

    // Function to take memory snapshots
    const takeMemorySnapshot = () => {
      const memoryUsage = process.memoryUsage();
      performanceMonitor.memorySnapshots.push({
        timestamp: new Date(),
        heapUsed: memoryUsage.heapUsed,
        heapTotal: memoryUsage.heapTotal,
        external: memoryUsage.external,
        rss: memoryUsage.rss
      });
    };

    // Take initial snapshot
    takeMemorySnapshot();

    try {
      let results;

      if (options.explain) {
        // Get execution plan with timing
        results = await this.db.collection(collection).aggregate(
          pipeline,
          { 
            explain: true,
            allowDiskUse: options.allowDiskUse || true,
            maxTimeMS: options.maxTimeMS || 300000
          }
        ).toArray();

        // Analyze execution plan
        results.forEach((stageExplan, index) => {
          performanceMonitor.stageTimings.push({
            stage: index,
            type: Object.keys(pipeline[index])[0],
            executionTimeMs: stageExplan.executionStats?.executionTimeMillisEstimate || 0,
            documentsIn: stageExplan.executionStats?.totalDocsExamined || 0,
            documentsOut: stageExplan.executionStats?.totalDocsReturned || 0
          });
        });

      } else {
        // Execute actual pipeline with monitoring
        const monitoringInterval = setInterval(takeMemorySnapshot, 5000); // Every 5 seconds

        try {
          results = await this.db.collection(collection).aggregate(
            pipeline,
            {
              allowDiskUse: options.allowDiskUse || true,
              maxTimeMS: options.maxTimeMS || 300000,
              batchSize: options.batchSize || 1000
            }
          ).toArray();

        } finally {
          clearInterval(monitoringInterval);
        }
      }

      // Take final snapshot
      takeMemorySnapshot();

      // Calculate performance metrics
      const endTime = new Date();
      const totalTime = endTime - performanceMonitor.startTime;
      const finalCpuUsage = process.cpuUsage(performanceMonitor.resourceUsage.cpuStart);
      const finalMemoryUsage = process.memoryUsage();

      performanceMonitor.summary = {
        totalExecutionTime: totalTime,
        documentsReturned: results.length,
        avgMemoryUsage: performanceMonitor.memorySnapshots.reduce(
          (sum, snapshot) => sum + snapshot.heapUsed, 0
        ) / performanceMonitor.memorySnapshots.length,
        peakMemoryUsage: Math.max(
          ...performanceMonitor.memorySnapshots.map(s => s.heapUsed)
        ),
        cpuUserTime: finalCpuUsage.user / 1000, // Convert to milliseconds
        cpuSystemTime: finalCpuUsage.system / 1000,
        memoryDifference: finalMemoryUsage.heapUsed - performanceMonitor.resourceUsage.memoryStart.heapUsed
      };

      return {
        results: results,
        performanceData: performanceMonitor
      };

    } catch (error) {
      performanceMonitor.error = error.message;
      throw error;
    }
  }

  async optimizeForShardedCollection(collection, pipeline, shardingConfig) {
    // Optimize pipeline for sharded collections
    const shardOptimizedPipeline = [];

    // Add shard key filtering early if possible
    if (shardingConfig.shardKey && shardingConfig.shardKeyValues) {
      shardOptimizedPipeline.push({
        $match: {
          [shardingConfig.shardKey]: {
            $in: shardingConfig.shardKeyValues
          }
        }
      });
    }

    pipeline.forEach((stage, index) => {
      const stageType = Object.keys(stage)[0];

      switch (stageType) {
        case '$group':
          // Ensure group operations can be parallelized across shards
          const groupStage = { ...stage };

          // Add shard key to group _id when possible for better parallelization
          if (typeof groupStage.$group._id === 'object' && shardingConfig.includeShardKeyInGroup) {
            groupStage.$group._id[shardingConfig.shardKey] = `$${shardingConfig.shardKey}`;
          }

          shardOptimizedPipeline.push(groupStage);
          break;

        case '$sort':
          // Optimize sort for sharded collections
          const sortStage = { ...stage };

          // Include shard key in sort to prevent scatter-gather when possible
          if (shardingConfig.includeShardKeyInSort) {
            sortStage.$sort = {
              [shardingConfig.shardKey]: 1,
              ...sortStage.$sort
            };
          }

          shardOptimizedPipeline.push(sortStage);
          break;

        case '$lookup':
          // Optimize lookups for sharded collections
          const lookupStage = { ...stage };

          // Add hint to use shard key when doing lookups
          if (shardingConfig.optimizeLookups) {
            lookupStage.$lookup.pipeline = lookupStage.$lookup.pipeline || [];
            lookupStage.$lookup.pipeline.unshift({
              $match: {
                // Add efficient filters in lookup pipeline
              }
            });
          }

          shardOptimizedPipeline.push(lookupStage);
          break;

        default:
          shardOptimizedPipeline.push(stage);
      }
    });

    return shardOptimizedPipeline;
  }
}