perf: cache file handles in FragmentSource — eliminate open/close per page read

readPage opened and closed a file handle for every page read, causing
potentially 1000+ open/close syscalls per query on multi-page files.
Now lazily caches the handle on first read and closes via the new
FragmentSource.close() method, called by ScanOperator.close() which
is invoked by drainPipeline's finally block.

Author: teamchong

src/local-executor.ts

@@ -301,17 +301,17 @@ export class LocalExecutor implements QueryExecutor {
       };
     }
     const fsMod = await import("node:fs/promises");
+    let cachedHandle: import("node:fs/promises").FileHandle | null = null;
     return {
       columns: projectedColumns,
       async readPage(_col: ColumnMeta, page: PageInfo): Promise<ArrayBuffer> {
-        const handle = await fsMod.open(table, "r");
-        try {
-          const buf = Buffer.alloc(page.byteLength);
-          await handle.read(buf, 0, page.byteLength, Number(page.byteOffset));
-          return buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength);
-        } finally {
-          await handle.close();
-        }
+        if (!cachedHandle) cachedHandle = await fsMod.open(table, "r");
+        const buf = Buffer.alloc(page.byteLength);
+        await cachedHandle.read(buf, 0, page.byteLength, Number(page.byteOffset));
+        return buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength);
+      },
+      async close(): Promise<void> {
+        if (cachedHandle) { await cachedHandle.close(); cachedHandle = null; }
       },
     };
   }
@@ -899,17 +899,17 @@ export class LocalExecutor implements QueryExecutor {
       const projectedColumns = meta.columns.filter(c => neededCols.has(c.name));
       const filePath = meta.r2Key;
 
+      let cachedHandle: import("node:fs/promises").FileHandle | null = null;
       fragments.push({
         columns: projectedColumns,
         async readPage(_col: ColumnMeta, page: PageInfo): Promise<ArrayBuffer> {
-          const handle = await fsMod.open(filePath, "r");
-          try {
-            const buf = Buffer.alloc(page.byteLength);
-            await handle.read(buf, 0, page.byteLength, Number(page.byteOffset));
-            return buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength);
-          } finally {
-            await handle.close();
-          }
+          if (!cachedHandle) cachedHandle = await fsMod.open(filePath, "r");
+          const buf = Buffer.alloc(page.byteLength);
+          await cachedHandle.read(buf, 0, page.byteLength, Number(page.byteOffset));
+          return buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength);
+        },
+        async close(): Promise<void> {
+          if (cachedHandle) { await cachedHandle.close(); cachedHandle = null; }
         },
       });
     }

src/operators.ts

@@ -90,6 +90,8 @@ export interface FragmentSource {
   columns: ColumnMeta[];
   /** Read page data from the file. */
   readPage(col: ColumnMeta, page: PageInfo): Promise<ArrayBuffer>;
+  /** Optional: release resources (e.g., file handles) when scanning is complete. */
+  close?(): Promise<void>;
 }
 
 /** Check if a page can be skipped by checking ALL filter columns' min/max stats.
@@ -383,7 +385,9 @@ export class ScanOperator implements Operator {
     return materializeRows(batch);
   }
 
-  async close(): Promise<void> { /* no-op */ }
+  async close(): Promise<void> {
+    for (const frag of this.fragments) await frag.close?.();
+  }
 }
 
 // ---------------------------------------------------------------------------