add configurable batch size

Author: not-night-but

docs/usage/core-concepts.md

@@ -55,7 +55,8 @@ const queryLeaf = new QueryLeaf(mongoClient, 'mydatabase');
 const results = await queryLeaf.execute('SELECT * FROM users');
 
 // Or use a cursor for more control and memory efficiency
-const cursor = await queryLeaf.executeCursor('SELECT * FROM users');
+// You can optionally specify the batch size to control memory usage
+const cursor = await queryLeaf.executeCursor('SELECT * FROM users', { batchSize: 50 });
 await cursor.forEach(user => {
   console.log(`Processing user: ${user.name}`);
 });

docs/usage/examples.md

@@ -77,11 +77,12 @@ const usersInNY = await queryLeaf.execute(`
 
 ```typescript
 // Use cursor for memory-efficient processing of large result sets
+// Optionally specify a batch size to control how many documents are fetched at once
 const cursor = await queryLeaf.executeCursor(`
   SELECT _id, customer, total, items
   FROM orders
   WHERE status = 'completed'
-`);
+`, { batchSize: 100 }); // Set batch size to 100 documents per batch
 
 try {
   // Process one document at a time without loading everything in memory

packages/lib/README.md

@@ -49,7 +49,8 @@ const results = await queryLeaf.execute('SELECT * FROM users WHERE age > 21');
 console.log(results);
 
 // Get a MongoDB cursor for more control over result processing and memory efficiency
-const cursor = await queryLeaf.executeCursor('SELECT * FROM users WHERE age > 30');
+// You can optionally specify a batch size to control how many documents are fetched at once
+const cursor = await queryLeaf.executeCursor('SELECT * FROM users WHERE age > 30', { batchSize: 50 });
 await cursor.forEach((doc) => {
   console.log(`User: ${doc.name}`);
 });
@@ -106,7 +107,8 @@ When working with large result sets, using MongoDB cursors directly can be more
 
 ```typescript
 // Get a cursor for a SELECT query
-const cursor = await queryLeaf.executeCursor('SELECT * FROM products WHERE price > 100');
+// You can specify a batch size to control memory usage and network behavior
+const cursor = await queryLeaf.executeCursor('SELECT * FROM products WHERE price > 100', { batchSize: 100 });
 
 // Option 1: Convert to array (loads all results into memory)
 const results = await cursor.toArray();
@@ -131,7 +133,7 @@ await cursor.close();
 Features:
 - Returns MongoDB `FindCursor` for normal queries and `AggregationCursor` for aggregations
 - Supports all cursor methods like `forEach()`, `toArray()`, `next()`, `hasNext()`
-- Efficiently handles large result sets with MongoDB's batching system
+- Efficiently handles large result sets with MongoDB's batching system (configurable batch size)
 - Works with all advanced QueryLeaf features (filtering, sorting, aggregations, etc.)
 - Only available for read operations (SELECT queries)
 ```

packages/lib/src/examples/basic-usage.ts

@@ -106,7 +106,10 @@ async function main() {
     let cursor: CursorResult<Document> = null;
     try {
       // Using the executeCursor method to get a MongoDB cursor
-      cursor = await queryLeaf.executeCursor('SELECT * FROM users WHERE active = true');
+      // You can optionally specify a batch size to control memory usage
+      cursor = await queryLeaf.executeCursor('SELECT * FROM users WHERE active = true', {
+        batchSize: 20,
+      });
 
       // Check if we got a cursor back
       if (cursor) {

packages/lib/src/executor/index.ts

@@ -1,4 +1,10 @@
-import { CommandExecutor, Command, ExecutionResult, CursorResult } from '../interfaces';
+import {
+  CommandExecutor,
+  Command,
+  ExecutionResult,
+  CursorResult,
+  CursorOptions,
+} from '../interfaces';
 import { Document, MongoClient, ObjectId } from 'mongodb';
 
 /**
@@ -47,28 +53,32 @@ export class MongoExecutor implements CommandExecutor {
     for (const command of commands) {
       switch (command.type) {
         case 'FIND':
-          const findCursor = database
-            .collection(command.collection)
-            .find(this.convertObjectIds(command.filter || {}));
+          // Prepare find options
+          const findOptions: any = {};
 
           // Apply projection if specified
           if (command.projection) {
-            findCursor.project(command.projection);
+            findOptions.projection = command.projection;
           }
 
           // Apply sorting if specified
           if (command.sort) {
-            findCursor.sort(command.sort);
+            findOptions.sort = command.sort;
           }
 
           // Apply pagination if specified
           if (command.skip) {
-            findCursor.skip(command.skip);
+            findOptions.skip = command.skip;
           }
           if (command.limit && command.limit > 0) {
-            findCursor.limit(command.limit);
+            findOptions.limit = command.limit;
           }
 
+          // Create the cursor with all options at once
+          const findCursor = database
+            .collection(command.collection)
+            .find(this.convertObjectIds(command.filter || {}), findOptions);
+
           // Always return array for the regular execute
           result = await findCursor.toArray();
           break;
@@ -97,7 +107,14 @@ export class MongoExecutor implements CommandExecutor {
         case 'AGGREGATE':
           // Handle aggregation commands
           const pipeline = command.pipeline.map((stage) => this.convertObjectIds(stage));
-          const aggregateCursor = database.collection(command.collection).aggregate(pipeline);
+
+          // Prepare aggregation options
+          const aggregateOptions: any = {};
+
+          // Create the cursor with options
+          const aggregateCursor = database
+            .collection(command.collection)
+            .aggregate(pipeline, aggregateOptions);
 
           // Always return array for the regular execute
           result = await aggregateCursor.toArray();
@@ -114,10 +131,14 @@ export class MongoExecutor implements CommandExecutor {
   /**
    * Execute a series of MongoDB commands and return cursors
    * @param commands Array of commands to execute
+   * @param options Options for cursor execution
    * @returns Cursor for FIND and AGGREGATE commands, null for other commands
    * @typeParam T - The type of documents that will be returned (defaults to Document)
    */
-  async executeCursor<T = Document>(commands: Command[]): Promise<CursorResult<T>> {
+  async executeCursor<T = Document>(
+    commands: Command[],
+    options?: CursorOptions
+  ): Promise<CursorResult<T>> {
     // We assume the client is already connected
     const database = this.client.db(this.dbName);
 
@@ -128,36 +149,55 @@ export class MongoExecutor implements CommandExecutor {
     for (const command of commands) {
       switch (command.type) {
         case 'FIND':
-          const findCursor = database
-            .collection(command.collection)
-            .find(this.convertObjectIds(command.filter || {}));
+          // Prepare find options
+          const findOptions: any = {};
 
           // Apply projection if specified
           if (command.projection) {
-            findCursor.project(command.projection);
+            findOptions.projection = command.projection;
           }
 
           // Apply sorting if specified
           if (command.sort) {
-            findCursor.sort(command.sort);
+            findOptions.sort = command.sort;
           }
 
           // Apply pagination if specified
           if (command.skip) {
-            findCursor.skip(command.skip);
+            findOptions.skip = command.skip;
           }
           if (command.limit && command.limit > 0) {
-            findCursor.limit(command.limit);
+            findOptions.limit = command.limit;
           }
 
+          // Apply batch size from options
+          if (options?.batchSize) {
+            findOptions.batchSize = options.batchSize;
+          }
+
+          // Create the cursor with all options at once
+          const findCursor = database
+            .collection(command.collection)
+            .find(this.convertObjectIds(command.filter || {}), findOptions);
+
           // Return the cursor directly
           result = findCursor;
           break;
 
         case 'AGGREGATE':
           // Handle aggregation commands
           const pipeline = command.pipeline.map((stage) => this.convertObjectIds(stage));
-          result = database.collection(command.collection).aggregate(pipeline);
+
+          // Prepare aggregation options
+          const aggregateOptions: any = {};
+
+          // Apply batch size from options
+          if (options?.batchSize) {
+            aggregateOptions.batchSize = options.batchSize;
+          }
+
+          // Create the cursor with options
+          result = database.collection(command.collection).aggregate(pipeline, aggregateOptions);
           break;
 
         case 'INSERT':

packages/lib/src/index.ts

@@ -6,6 +6,7 @@ import {
   CommandExecutor,
   ExecutionResult,
   CursorResult,
+  CursorOptions,
 } from './interfaces';
 import { Document, MongoClient } from 'mongodb';
 import { SqlParserImpl } from './parser';
@@ -47,13 +48,17 @@ export class QueryLeaf {
   /**
    * Execute a SQL query on MongoDB and return a cursor
    * @param sql SQL query string
+   * @param options Options for cursor execution
    * @returns Cursor for SELECT queries, null for other queries
    * @typeParam T - The type of documents that will be returned (defaults to Document)
    */
-  async executeCursor<T = Document>(sql: string): Promise<CursorResult<T>> {
+  async executeCursor<T = Document>(
+    sql: string,
+    options?: CursorOptions
+  ): Promise<CursorResult<T>> {
     const statement = this.parse(sql);
     const commands = this.compile(statement);
-    return await this.executor.executeCursor(commands);
+    return await this.executor.executeCursor(commands, options);
   }
 
   /**

packages/lib/src/interfaces.ts

@@ -122,6 +122,17 @@ export type CursorResult<T = Document> =
   | AggregationCursor<T> // Cursor from AGGREGATE command
   | null; // No result
 
+/**
+ * Options for cursor execution
+ */
+export interface CursorOptions {
+  /**
+   * Number of documents to fetch per batch
+   * This will be set directly on the query command, not on the cursor after creation
+   */
+  batchSize?: number;
+}
+
 /**
  * MongoDB command executor interface
  */
@@ -138,9 +149,13 @@ export interface CommandExecutor {
   /**
    * Execute MongoDB commands and return cursors for FIND and AGGREGATE commands
    * @param commands Array of commands to execute
+   * @param options Options for cursor execution
    * @returns Cursor for FIND and AGGREGATE commands, null for other commands
    */
-  executeCursor<T = Document>(commands: Command[]): Promise<CursorResult<T>>;
+  executeCursor<T = Document>(
+    commands: Command[],
+    options?: CursorOptions
+  ): Promise<CursorResult<T>>;
 }
 
 /**
@@ -157,9 +172,10 @@ export interface QueryLeaf {
   /**
    * Execute a SQL query and return a cursor for SELECT queries
    * @param sql SQL query string
+   * @param options Options for cursor execution
    * @returns Cursor for SELECT queries, null for other queries
    */
-  executeCursor<T = Document>(sql: string): Promise<CursorResult<T>>;
+  executeCursor<T = Document>(sql: string, options?: CursorOptions): Promise<CursorResult<T>>;
 
   parse(sql: string): SqlStatement;
   compile(statement: SqlStatement): Command[];
@@ -169,7 +185,7 @@ export interface QueryLeaf {
 
 export interface Squongo extends QueryLeaf {
   execute<T = Document>(sql: string): Promise<ExecutionResult<T>>;
-  executeCursor<T = Document>(sql: string): Promise<CursorResult<T>>;
+  executeCursor<T = Document>(sql: string, options?: CursorOptions): Promise<CursorResult<T>>;
   parse(sql: string): SqlStatement;
   compile(statement: SqlStatement): Command[];
   getExecutor(): CommandExecutor;

packages/lib/tests/integration/cursor.integration.test.ts

@@ -335,4 +335,46 @@ describe('MongoDB Cursor Functionality Tests', () => {
     // Clean up
     await cursor?.close();
   });
+  
+  test('should support custom batchSize option', async () => {
+    // This test verifies the batchSize option is used correctly
+    // Arrange
+    const queryLeaf = testSetup.getQueryLeaf();
+    const db = testSetup.getDb();
+    
+    // Generate more data to make batching more apparent
+    const moreDocuments = [];
+    for (let i = 0; i < 50; i++) {
+      moreDocuments.push({ name: `Extra Product ${i}`, value: i });
+    }
+    await db.collection('batch_test').deleteMany({});
+    await db.collection('batch_test').insertMany(moreDocuments);
+    
+    // Define a small batch size
+    const batchSize = 10;
+    
+    // Act - Create a cursor with a specific batch size
+    const cursor = await queryLeaf.executeCursor(
+      'SELECT * FROM batch_test ORDER BY value ASC',
+      { batchSize }
+    );
+    
+    // Assert
+    expect(cursor).not.toBeNull();
+    
+    // Consume the cursor in multiple steps to verify batching
+    const result1 = await cursor!.toArray();
+    
+    // We should have all documents
+    expect(result1.length).toBe(50);
+    
+    // Verify results are in correct order
+    for (let i = 1; i < result1.length; i++) {
+      expect(result1[i].value).toBeGreaterThanOrEqual(result1[i-1].value);
+    }
+    
+    // Clean up
+    await cursor!.close();
+    await db.collection('batch_test').drop();
+  });
 });