MongoDB Sharding: Horizontal Scaling Strategies with SQL-Style Database Partitioning

-- Single server limitations
-- CPU: Limited cores per server
-- Memory: Physical RAM limitations (typically 1TB max)
-- Storage: I/O bottlenecks and capacity limits
-- Network: Single network interface bandwidth limits

-- Example: E-commerce order processing bottleneck
SELECT 
  order_id,
  customer_id,
  order_total,
  created_at
FROM orders
WHERE created_at >= CURRENT_DATE - INTERVAL '1 day'
  AND status = 'pending'
ORDER BY created_at DESC;

-- Problems with single-server approach:
-- - All queries compete for same CPU/memory resources
-- - I/O bottlenecks during peak traffic
-- - Limited concurrent connection capacity
-- - Single point of failure
-- - Expensive to upgrade hardware

// MongoDB sharded cluster distributes data across multiple servers
// Each shard handles a subset of the data based on shard key ranges

// Shard 1: Orders with shard key values 1-1000
db.orders.find({ customer_id: { $gte: 1, $lt: 1000 } })

// Shard 2: Orders with shard key values 1000-2000  
db.orders.find({ customer_id: { $gte: 1000, $lt: 2000 } })

// Shard 3: Orders with shard key values 2000+
db.orders.find({ customer_id: { $gte: 2000 } })

// Benefits:
// - Distribute load across multiple servers
// - Scale capacity by adding more shards
// - Fault tolerance through replica sets
// - Parallel query execution

// Sharded cluster architecture
{
  "mongos": [
    "router1.example.com:27017",
    "router2.example.com:27017"  
  ],
  "configServers": [
    "config1.example.com:27019",
    "config2.example.com:27019", 
    "config3.example.com:27019"
  ],
  "shards": [
    {
      "shard": "shard01",
      "replica_set": "rs01",
      "members": [
        "shard01-primary.example.com:27018",
        "shard01-secondary1.example.com:27018",
        "shard01-secondary2.example.com:27018"
      ]
    },
    {
      "shard": "shard02", 
      "replica_set": "rs02",
      "members": [
        "shard02-primary.example.com:27018",
        "shard02-secondary1.example.com:27018",
        "shard02-secondary2.example.com:27018"
      ]
    }
  ]
}

-- Conceptual SQL partitioning architecture
-- Multiple database servers handling different data ranges

-- Master database coordinator (similar to mongos)
CREATE DATABASE cluster_coordinator;

-- Partition definitions (similar to config servers)
CREATE TABLE partition_map (
  table_name VARCHAR(255),
  partition_key VARCHAR(255),
  min_value VARCHAR(255),
  max_value VARCHAR(255), 
  server_host VARCHAR(255),
  server_port INTEGER,
  status VARCHAR(50)
);

-- Data partitions across different servers
-- Server 1: customer_id 1-999999
-- Server 2: customer_id 1000000-1999999  
-- Server 3: customer_id 2000000+

-- Partition-aware query routing
SELECT * FROM orders 
WHERE customer_id = 1500000;  -- Routes to Server 2

// Good shard key examples for different use cases

// 1. E-commerce: Customer-based sharding
sh.shardCollection("ecommerce.orders", { "customer_id": 1 })
// Pros: Related customer data stays together
// Cons: Uneven distribution if some customers order much more

// 2. Time-series: Date-based sharding  
sh.shardCollection("analytics.events", { "event_date": 1, "user_id": 1 })
// Pros: Time-range queries stay on fewer shards
// Cons: Hot spots during peak times

// 3. Geographic: Location-based sharding
sh.shardCollection("locations.venues", { "region": 1, "venue_id": 1 })
// Pros: Geographic queries are localized
// Cons: Uneven distribution based on population density

// 4. Hash-based: Even distribution
sh.shardCollection("users.profiles", { "_id": "hashed" })
// Pros: Even data distribution
// Cons: Range queries must check all shards

-- SQL partitioning approaches equivalent to shard keys

-- 1. Range partitioning (similar to range-based shard keys)
CREATE TABLE orders (
  order_id BIGINT,
  customer_id BIGINT,
  order_date DATE,
  total_amount DECIMAL
) PARTITION BY RANGE (customer_id) (
  PARTITION p1 VALUES LESS THAN (1000000),
  PARTITION p2 VALUES LESS THAN (2000000),
  PARTITION p3 VALUES LESS THAN (MAXVALUE)
);

-- 2. Hash partitioning (similar to hashed shard keys) 
CREATE TABLE user_profiles (
  user_id BIGINT,
  email VARCHAR(255),
  created_at TIMESTAMP
) PARTITION BY HASH (user_id) PARTITIONS 8;

-- 3. List partitioning (similar to tag-based sharding)
CREATE TABLE regional_data (
  id BIGINT,
  region VARCHAR(50),
  data JSONB
) PARTITION BY LIST (region) (
  PARTITION north_america VALUES ('us', 'ca', 'mx'),
  PARTITION europe VALUES ('uk', 'de', 'fr', 'es'),
  PARTITION asia VALUES ('jp', 'cn', 'kr', 'in')
);

// 1. Start config server replica set
rs.initiate({
  _id: "configReplSet",
  configsvr: true,
  members: [
    { _id: 0, host: "config1.example.com:27019" },
    { _id: 1, host: "config2.example.com:27019" },
    { _id: 2, host: "config3.example.com:27019" }
  ]
})

// 2. Start shard replica sets
// Shard 1
rs.initiate({
  _id: "shard01rs",
  members: [
    { _id: 0, host: "shard01-1.example.com:27018", priority: 1 },
    { _id: 1, host: "shard01-2.example.com:27018", priority: 0.5 },
    { _id: 2, host: "shard01-3.example.com:27018", priority: 0.5 }
  ]
})

// Shard 2
rs.initiate({
  _id: "shard02rs", 
  members: [
    { _id: 0, host: "shard02-1.example.com:27018", priority: 1 },
    { _id: 1, host: "shard02-2.example.com:27018", priority: 0.5 },
    { _id: 2, host: "shard02-3.example.com:27018", priority: 0.5 }
  ]
})

// 3. Start mongos routers
mongos --configdb configReplSet/config1.example.com:27019,config2.example.com:27019,config3.example.com:27019 --port 27017

// 4. Add shards to cluster
sh.addShard("shard01rs/shard01-1.example.com:27018,shard01-2.example.com:27018,shard01-3.example.com:27018")
sh.addShard("shard02rs/shard02-1.example.com:27018,shard02-2.example.com:27018,shard02-3.example.com:27018")

// 5. Enable sharding on database
sh.enableSharding("ecommerce")

// Node.js application connection to sharded cluster
const { MongoClient } = require('mongodb');

const client = new MongoClient('mongodb://mongos1.example.com:27017,mongos2.example.com:27017/ecommerce', {
  // Connection pool settings for high-throughput applications
  maxPoolSize: 50,
  minPoolSize: 5,
  maxIdleTimeMS: 30000,

  // Read preferences for different query types
  readPreference: 'primaryPreferred',
  readConcern: { level: 'local' },

  // Write concerns for data consistency  
  writeConcern: { w: 'majority', j: true },

  // Timeout settings
  serverSelectionTimeoutMS: 5000,
  connectTimeoutMS: 10000,
  socketTimeoutMS: 45000
});

// Different connection strategies for different use cases
class ShardedDatabaseClient {
  constructor() {
    // Real-time operations: connect to mongos with primary reads
    this.realtimeClient = new MongoClient(this.getMongosUrl(), {
      readPreference: 'primary',
      writeConcern: { w: 'majority', j: true, wtimeout: 5000 }
    });

    // Analytics operations: connect with secondary reads allowed  
    this.analyticsClient = new MongoClient(this.getMongosUrl(), {
      readPreference: 'secondaryPreferred',
      readConcern: { level: 'local' },
      maxTimeMS: 60000  // Allow longer timeouts for analytics
    });
  }

  getMongosUrl() {
    return 'mongodb://mongos1.example.com:27017,mongos2.example.com:27017,mongos3.example.com:27017/ecommerce?replicaSet=false';
  }
}

// Multi-collection sharding strategy for e-commerce

// 1. Users collection: Hash sharding for even distribution
sh.shardCollection("ecommerce.users", { "_id": "hashed" })
// Reasoning: User lookups are typically by ID, hash distributes evenly

// 2. Products collection: Category-based compound sharding  
sh.shardCollection("ecommerce.products", { "category": 1, "_id": 1 })
// Reasoning: Product browsing often filtered by category

// 3. Orders collection: Customer-based with date for range queries
sh.shardCollection("ecommerce.orders", { "customer_id": 1, "created_at": 1 })
// Reasoning: Customer order history queries, with time-based access patterns

// 4. Inventory collection: Product-based sharding
sh.shardCollection("ecommerce.inventory", { "product_id": 1 })
// Reasoning: Inventory updates are product-specific

// 5. Sessions collection: Hash for even distribution
sh.shardCollection("ecommerce.sessions", { "_id": "hashed" })
// Reasoning: Session access is random, hash provides even distribution

-- SQL partitioning strategy for e-commerce platform

-- 1. Users table: Hash partitioning for even distribution
CREATE TABLE users (
  user_id BIGSERIAL PRIMARY KEY,
  email VARCHAR(255) UNIQUE NOT NULL,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  profile_data JSONB
) PARTITION BY HASH (user_id) PARTITIONS 8;

-- 2. Products table: List partitioning by category
CREATE TABLE products (
  product_id BIGSERIAL PRIMARY KEY,
  category VARCHAR(100) NOT NULL,
  name VARCHAR(255) NOT NULL,
  price DECIMAL(10,2),
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) PARTITION BY LIST (category) (
  PARTITION electronics VALUES ('electronics', 'computers', 'phones'),
  PARTITION clothing VALUES ('clothing', 'shoes', 'accessories'), 
  PARTITION books VALUES ('books', 'ebooks', 'audiobooks'),
  PARTITION home VALUES ('furniture', 'appliances', 'decor')
);

-- 3. Orders table: Range partitioning by customer with subpartitioning by date
CREATE TABLE orders (
  order_id BIGSERIAL PRIMARY KEY,
  customer_id BIGINT NOT NULL,
  order_date DATE NOT NULL,
  total_amount DECIMAL(10,2)
) PARTITION BY RANGE (customer_id) 
SUBPARTITION BY RANGE (order_date) (
  PARTITION customers_1_to_100k VALUES LESS THAN (100000) (
    SUBPARTITION orders_2024 VALUES LESS THAN ('2025-01-01'),
    SUBPARTITION orders_2025 VALUES LESS THAN ('2026-01-01')
  ),
  PARTITION customers_100k_to_500k VALUES LESS THAN (500000) (
    SUBPARTITION orders_2024 VALUES LESS THAN ('2025-01-01'),
    SUBPARTITION orders_2025 VALUES LESS THAN ('2026-01-01')
  )
);

// Time-series analytics sharding strategy

// Events collection: Time-based sharding with compound key
sh.shardCollection("analytics.events", { "event_date": 1, "user_id": 1 })

// Pre-create chunks for future dates to avoid hot spots
sh.splitAt("analytics.events", { "event_date": ISODate("2025-09-01"), "user_id": MinKey })
sh.splitAt("analytics.events", { "event_date": ISODate("2025-10-01"), "user_id": MinKey })
sh.splitAt("analytics.events", { "event_date": ISODate("2025-11-01"), "user_id": MinKey })

// User aggregation collection: Hash for even distribution
sh.shardCollection("analytics.user_stats", { "user_id": "hashed" })

// Geographic data: Zone-based sharding  
sh.shardCollection("analytics.geographic_events", { "timezone": 1, "event_date": 1 })

// Example queries optimized for this sharding strategy
class AnalyticsQueryOptimizer {
  constructor(db) {
    this.db = db;
  }

  // Time-range queries hit minimal shards
  async getDailyEvents(startDate, endDate) {
    return await this.db.collection('events').find({
      event_date: { 
        $gte: startDate,
        $lte: endDate 
      }
    }).toArray();
    // Only queries shards containing the date range
  }

  // User-specific queries use shard key
  async getUserEvents(userId, startDate, endDate) {
    return await this.db.collection('events').find({
      user_id: userId,
      event_date: { 
        $gte: startDate,
        $lte: endDate 
      }
    }).toArray();
    // Efficiently targets specific shards using compound key
  }

  // Aggregation across shards
  async getEventCounts(startDate, endDate) {
    return await this.db.collection('events').aggregate([
      {
        $match: {
          event_date: { $gte: startDate, $lte: endDate }
        }
      },
      {
        $group: {
          _id: {
            date: "$event_date",
            event_type: "$event_type"
          },
          count: { $sum: 1 }
        }
      },
      {
        $sort: { "_id.date": 1, "count": -1 }
      }
    ]).toArray();
    // Parallel execution across shards, merged by mongos
  }
}

// Configure the balancer for optimal performance
// Balancer settings for production workloads

// 1. Set balancer window to off-peak hours
use config
db.settings.update(
  { _id: "balancer" },
  { 
    $set: { 
      activeWindow: { 
        start: "01:00",   // 1 AM
        stop: "05:00"     // 5 AM  
      }
    } 
  },
  { upsert: true }
)

// 2. Configure chunk size based on workload
db.settings.update(
  { _id: "chunksize" },
  { $set: { value: 128 } },  // 128MB chunks (default is 64MB)
  { upsert: true }
)

// 3. Monitor chunk distribution
db.chunks.aggregate([
  {
    $group: {
      _id: "$shard",
      chunk_count: { $sum: 1 }
    }
  },
  {
    $sort: { chunk_count: -1 }
  }
])

// 4. Manual balancing when needed
sh.enableBalancing("ecommerce.orders")  // Enable balancing for specific collection
sh.disableBalancing("ecommerce.orders")  // Disable during maintenance

// 5. Move specific chunks manually
sh.moveChunk("ecommerce.orders", 
  { customer_id: 500000 },  // Chunk containing this shard key
  "shard02rs"  // Target shard
)

-- SQL-style monitoring queries for shard performance
WITH shard_stats AS (
  SELECT 
    shard_name,
    collection_name,
    chunk_count,
    data_size_mb,
    index_size_mb,
    avg_chunk_size_mb,
    total_operations_per_second
  FROM shard_collection_stats
  WHERE collection_name = 'orders'
),
shard_balance AS (
  SELECT 
    AVG(chunk_count) AS avg_chunks_per_shard,
    STDDEV(chunk_count) AS chunk_distribution_stddev,
    MAX(chunk_count) - MIN(chunk_count) AS chunk_count_variance
  FROM shard_stats
)
SELECT 
  ss.shard_name,
  ss.chunk_count,
  ss.data_size_mb,
  ss.total_operations_per_second,
  -- Balance metrics
  CASE 
    WHEN ss.chunk_count > sb.avg_chunks_per_shard * 1.2 THEN 'Over-loaded'
    WHEN ss.chunk_count < sb.avg_chunks_per_shard * 0.8 THEN 'Under-loaded'
    ELSE 'Balanced'
  END AS load_status,
  -- Performance per chunk
  ss.total_operations_per_second / ss.chunk_count AS ops_per_chunk
FROM shard_stats ss
CROSS JOIN shard_balance sb
ORDER BY ss.total_operations_per_second DESC;

// Comprehensive sharding monitoring
class ShardingMonitor {
  constructor(db) {
    this.db = db;
    this.configDb = db.getSiblingDB('config');
  }

  async getShardDistribution(collection) {
    return await this.configDb.chunks.aggregate([
      {
        $match: { ns: collection }
      },
      {
        $group: {
          _id: "$shard",
          chunk_count: { $sum: 1 },
          min_key: { $min: "$min" },
          max_key: { $max: "$max" }
        }
      },
      {
        $lookup: {
          from: "shards",
          localField: "_id", 
          foreignField: "_id",
          as: "shard_info"
        }
      }
    ]).toArray();
  }

  async getShardStats() {
    const shards = await this.configDb.shards.find().toArray();
    const stats = {};

    for (const shard of shards) {
      const shardDb = await this.db.admin().getSiblingDB('admin').runCommand({
        connPoolStats: 1
      });

      stats[shard._id] = {
        host: shard.host,
        connections: shardDb.hosts,
        uptime: shardDb.uptime
      };
    }

    return stats;
  }

  async identifyHotShards(collection, threshold = 1000) {
    const pipeline = [
      {
        $match: { 
          ns: collection,
          ts: { 
            $gte: new Date(Date.now() - 3600000)  // Last hour
          }
        }
      },
      {
        $group: {
          _id: "$shard",
          operation_count: { $sum: 1 },
          avg_duration: { $avg: "$millis" }
        }
      },
      {
        $match: {
          operation_count: { $gte: threshold }
        }
      },
      {
        $sort: { operation_count: -1 }
      }
    ];

    return await this.configDb.mongos.aggregate(pipeline).toArray();
  }
}

// Configure zones for geographic distribution

// 1. Create zones
sh.addShardToZone("shard01rs", "US_EAST")
sh.addShardToZone("shard02rs", "US_WEST") 
sh.addShardToZone("shard03rs", "EUROPE")
sh.addShardToZone("shard04rs", "ASIA")

// 2. Define zone ranges for geographic sharding
sh.updateZoneKeyRange(
  "global.users",
  { region: "us_east", user_id: MinKey },
  { region: "us_east", user_id: MaxKey },
  "US_EAST"
)

sh.updateZoneKeyRange(
  "global.users", 
  { region: "us_west", user_id: MinKey },
  { region: "us_west", user_id: MaxKey },
  "US_WEST"
)

sh.updateZoneKeyRange(
  "global.users",
  { region: "europe", user_id: MinKey },
  { region: "europe", user_id: MaxKey }, 
  "EUROPE"
)

// 3. Shard the collection with zone-aware shard key
sh.shardCollection("global.users", { "region": 1, "user_id": 1 })

// Multi-tenant sharding strategy

// Tenant-based sharding for SaaS applications
sh.shardCollection("saas.tenant_data", { "tenant_id": 1, "created_at": 1 })

// Zones for tenant tiers
sh.addShardToZone("premiumShard01", "PREMIUM_TIER")
sh.addShardToZone("premiumShard02", "PREMIUM_TIER")
sh.addShardToZone("standardShard01", "STANDARD_TIER")
sh.addShardToZone("standardShard02", "STANDARD_TIER")

// Assign tenant ranges to appropriate zones
sh.updateZoneKeyRange(
  "saas.tenant_data",
  { tenant_id: "premium_tenant_001", created_at: MinKey },
  { tenant_id: "premium_tenant_999", created_at: MaxKey },
  "PREMIUM_TIER"
)

sh.updateZoneKeyRange(
  "saas.tenant_data", 
  { tenant_id: "standard_tenant_001", created_at: MinKey },
  { tenant_id: "standard_tenant_999", created_at: MaxKey },
  "STANDARD_TIER"
)

// Application-level tenant routing
class MultiTenantShardingClient {
  constructor(db) {
    this.db = db;
  }

  async getTenantData(tenantId, query = {}) {
    // Always include tenant_id in queries for optimal shard targeting
    const tenantQuery = {
      tenant_id: tenantId,
      ...query
    };

    return await this.db.collection('tenant_data').find(tenantQuery).toArray();
  }

  async createTenantDocument(tenantId, document) {
    const tenantDocument = {
      tenant_id: tenantId,
      created_at: new Date(),
      ...document
    };

    return await this.db.collection('tenant_data').insertOne(tenantDocument);
  }

  async getTenantStats(tenantId) {
    return await this.db.collection('tenant_data').aggregate([
      {
        $match: { tenant_id: tenantId }
      },
      {
        $group: {
          _id: null,
          document_count: { $sum: 1 },
          total_size: { $sum: { $bsonSize: "$$ROOT" } },
          oldest_document: { $min: "$created_at" },
          newest_document: { $max: "$created_at" }
        }
      }
    ]).toArray();
  }
}

// Query patterns for optimal shard targeting

class ShardOptimizedQueries {
  constructor(db) {
    this.db = db;
  }

  // GOOD: Query includes shard key - targets specific shards
  async getCustomerOrders(customerId, startDate, endDate) {
    return await this.db.collection('orders').find({
      customer_id: customerId,  // Shard key - enables shard targeting
      created_at: { $gte: startDate, $lte: endDate }
    }).toArray();
    // Only queries shards containing data for this customer
  }

  // BAD: Query without shard key - scatter-gather across all shards
  async getOrdersByAmount(minAmount) {
    return await this.db.collection('orders').find({
      total_amount: { $gte: minAmount }
      // No shard key - must query all shards
    }).toArray();
  }

  // BETTER: Include shard key range when possible
  async getHighValueOrders(minAmount, customerIdStart, customerIdEnd) {
    return await this.db.collection('orders').find({
      customer_id: { $gte: customerIdStart, $lte: customerIdEnd },  // Shard key range
      total_amount: { $gte: minAmount }
    }).toArray();
    // Limits query to shards containing the customer ID range
  }

  // Aggregation with shard key optimization
  async getCustomerOrderStats(customerId) {
    return await this.db.collection('orders').aggregate([
      {
        $match: { 
          customer_id: customerId  // Shard key - targets specific shards
        }
      },
      {
        $group: {
          _id: null,
          total_orders: { $sum: 1 },
          total_spent: { $sum: "$total_amount" },
          avg_order_value: { $avg: "$total_amount" },
          first_order: { $min: "$created_at" },
          last_order: { $max: "$created_at" }
        }
      }
    ]).toArray();
  }
}

-- SQL partition elimination examples

-- GOOD: Query with partition key - partition elimination
SELECT order_id, total_amount, created_at
FROM orders
WHERE customer_id = 12345  -- Partition key
  AND created_at >= '2025-01-01';
-- Query plan: Only scans partition containing customer_id 12345

-- BAD: Query without partition key - scans all partitions  
SELECT order_id, customer_id, total_amount
FROM orders
WHERE total_amount > 1000;
-- Query plan: Parallel scan across all partitions

-- BETTER: Include partition key range
SELECT order_id, customer_id, total_amount  
FROM orders
WHERE customer_id BETWEEN 10000 AND 20000  -- Partition key range
  AND total_amount > 1000;
-- Query plan: Only scans partitions containing customer_id 10000-20000

-- Aggregation with partition key
SELECT 
  COUNT(*) AS total_orders,
  SUM(total_amount) AS total_spent,
  AVG(total_amount) AS avg_order_value
FROM orders
WHERE customer_id = 12345;  -- Partition key enables partition elimination

// Optimized connection pooling for sharded clusters
const shardedClusterConfig = {
  // Router connections (mongos)
  mongosHosts: [
    'mongos1.example.com:27017',
    'mongos2.example.com:27017', 
    'mongos3.example.com:27017'
  ],

  // Connection pool settings
  maxPoolSize: 100,        // Higher pool size for sharded clusters
  minPoolSize: 10,         // Maintain minimum connections
  maxIdleTimeMS: 30000,    // Close idle connections

  // Timeout settings for distributed operations
  serverSelectionTimeoutMS: 5000,
  connectTimeoutMS: 10000,
  socketTimeoutMS: 60000,  // Longer timeouts for cross-shard operations

  // Read/write preferences
  readPreference: 'primaryPreferred',
  writeConcern: { w: 'majority', j: true, wtimeout: 10000 },

  // Retry configuration for distributed operations
  retryWrites: true,
  retryReads: true
};

// Connection management for different workload types
class ShardedConnectionManager {
  constructor() {
    // OLTP connections - fast, consistent reads/writes
    this.oltpClient = new MongoClient(this.getMongosUrl(), {
      ...shardedClusterConfig,
      readPreference: 'primary',
      readConcern: { level: 'local' },
      maxTimeMS: 5000
    });

    // OLAP connections - can use secondaries, longer timeouts
    this.olapClient = new MongoClient(this.getMongosUrl(), {
      ...shardedClusterConfig,
      readPreference: 'secondaryPreferred',
      readConcern: { level: 'local' },
      maxTimeMS: 300000  // 5 minute timeout for analytics
    });

    // Bulk operations - optimized for throughput
    this.bulkClient = new MongoClient(this.getMongosUrl(), {
      ...shardedClusterConfig,
      maxPoolSize: 20,    // Fewer connections for bulk operations
      writeConcern: { w: 1, j: false }  // Faster writes for bulk inserts
    });
  }

  getMongosUrl() {
    return `mongodb://${shardedClusterConfig.mongosHosts.join(',')}/ecommerce`;
  }
}

// Sharded cluster monitoring system
class ShardedClusterMonitor {
  constructor(configDb) {
    this.configDb = configDb;
  }

  async getClusterOverview() {
    const shards = await this.configDb.shards.find().toArray();
    const collections = await this.configDb.collections.find().toArray();
    const chunks = await this.configDb.chunks.countDocuments();

    return {
      shard_count: shards.length,
      sharded_collections: collections.length,
      total_chunks: chunks,
      balancer_state: await this.getBalancerState()
    };
  }

  async getShardLoadDistribution() {
    return await this.configDb.chunks.aggregate([
      {
        $group: {
          _id: "$shard", 
          chunk_count: { $sum: 1 }
        }
      },
      {
        $lookup: {
          from: "shards",
          localField: "_id",
          foreignField: "_id", 
          as: "shard_info"
        }
      },
      {
        $project: {
          shard_id: "$_id",
          chunk_count: 1,
          host: { $arrayElemAt: ["$shard_info.host", 0] }
        }
      },
      {
        $sort: { chunk_count: -1 }
      }
    ]).toArray();
  }

  async getChunkMigrationHistory(hours = 24) {
    const since = new Date(Date.now() - hours * 3600000);

    return await this.configDb.changelog.find({
      time: { $gte: since },
      what: { $in: ['moveChunk.start', 'moveChunk.commit'] }
    }).sort({ time: -1 }).toArray();
  }

  async identifyImbalancedCollections(threshold = 0.2) {
    const collections = await this.configDb.collections.find().toArray();
    const imbalanced = [];

    for (const collection of collections) {
      const distribution = await this.getCollectionDistribution(collection._id);
      const imbalanceRatio = this.calculateImbalanceRatio(distribution);

      if (imbalanceRatio > threshold) {
        imbalanced.push({
          collection: collection._id,
          imbalance_ratio: imbalanceRatio,
          distribution: distribution
        });
      }
    }

    return imbalanced;
  }

  calculateImbalanceRatio(distribution) {
    const chunkCounts = distribution.map(d => d.chunk_count);
    const max = Math.max(...chunkCounts);
    const min = Math.min(...chunkCounts);
    const avg = chunkCounts.reduce((a, b) => a + b, 0) / chunkCounts.length;

    return (max - min) / avg;
  }
}

-- QueryLeaf automatically handles sharded collections with SQL syntax
-- Create sharded tables using familiar DDL

CREATE TABLE orders (
  order_id BIGSERIAL PRIMARY KEY,
  customer_id BIGINT NOT NULL,
  order_date DATE NOT NULL,
  total_amount DECIMAL(10,2),
  status VARCHAR(50) DEFAULT 'pending'
) SHARD BY (customer_id);  -- QueryLeaf extension for sharding

CREATE TABLE products (
  product_id BIGSERIAL PRIMARY KEY,  
  category VARCHAR(100) NOT NULL,
  name VARCHAR(255) NOT NULL,
  price DECIMAL(10,2)
) SHARD BY HASH (product_id);  -- Hash sharding

-- QueryLeaf optimizes queries based on shard key usage
SELECT 
  o.order_id,
  o.total_amount,
  o.order_date,
  COUNT(oi.item_id) AS item_count
FROM orders o
JOIN order_items oi ON o.order_id = oi.order_id
WHERE o.customer_id = 12345  -- Shard key enables efficient targeting
  AND o.order_date >= CURRENT_DATE - INTERVAL '30 days'
GROUP BY o.order_id, o.total_amount, o.order_date
ORDER BY o.order_date DESC;

-- Cross-shard analytics with automatic optimization
WITH monthly_sales AS (
  SELECT 
    DATE_TRUNC('month', order_date) AS month,
    customer_id,
    SUM(total_amount) AS monthly_total
  FROM orders
  WHERE order_date >= CURRENT_DATE - INTERVAL '12 months'
    AND status = 'completed'
  GROUP BY DATE_TRUNC('month', order_date), customer_id
)
SELECT 
  month,
  COUNT(DISTINCT customer_id) AS unique_customers,
  SUM(monthly_total) AS total_revenue,
  AVG(monthly_total) AS avg_customer_spend
FROM monthly_sales
GROUP BY month
ORDER BY month DESC;

-- QueryLeaf automatically:
-- 1. Routes shard-key queries to appropriate shards
-- 2. Parallelizes cross-shard aggregations  
-- 3. Manages chunk distribution recommendations
-- 4. Provides shard-aware query planning
-- 5. Handles distributed transactions when needed