fix: query-do explain/metrics + local-executor file handle leak

query-do.ts:
- fragmentsSkipped used dataset.fragmentMetas.size instead of
  allFragments.length when partition catalog active — inflated count
  by including fragments already pruned by catalog
- explain() reported meta.totalRows as estimatedRows, ignoring
  fragment-level pruning — now uses estimatedRowsAfterPrune

local-executor.ts:
- readPage() didn't close file handle on I/O error — handle leaked
  until process exit. Now closes handle in catch block so subsequent
  calls re-open cleanly.

Author: teamchong

src/local-executor.ts

@@ -310,9 +310,14 @@ export class LocalExecutor implements QueryExecutor {
       columns: projectedColumns,
       async readPage(_col: ColumnMeta, page: PageInfo): Promise<ArrayBuffer> {
         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);
+        try {
+          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);
+        } catch (err) {
+          if (cachedHandle) { await cachedHandle.close().catch(() => {}); cachedHandle = null; }
+          throw err;
+        }
       },
       async close(): Promise<void> {
         if (cachedHandle) { await cachedHandle.close(); cachedHandle = null; }

src/query-do.ts

@@ -763,7 +763,7 @@ export class QueryDO extends DurableObject<Env> {
       pagesScanned: pagesTotal - pagesSkipped,
       estimatedBytes,
       estimatedR2Reads: coalesced.length,
-      estimatedRows: meta.totalRows,
+      estimatedRows: estimatedRowsAfterPrune,
       fragments: totalFragments,
       fragmentsSkipped,
       fragmentsScanned,
@@ -1694,9 +1694,7 @@ export class QueryDO extends DurableObject<Env> {
     const fragments = allFragments.filter(meta =>
       !canSkipFragment(meta, query.filters, query.filterGroups),
     );
-    const fragmentsSkipped = (candidateFragmentIds
-      ? dataset.fragmentMetas.size
-      : allFragments.length) - fragments.length;
+    const fragmentsSkipped = allFragments.length - fragments.length;
     if (fragmentsSkipped > 0) {
       this.log("info", "fragment_prune", {
         total: allFragments.length, skipped: fragmentsSkipped, remaining: fragments.length,