fix: add /register-iceberg endpoint to bypass miniflare R2 list() bug

Miniflare R2 list() doesn't reliably return objects even when uploaded
through the Worker's DATA_BUCKET binding. Add /register-iceberg
endpoint that accepts an explicit metadata.json key and uses only
R2 get()/head() (which work reliably) to discover and register
Iceberg tables. Seed script now uses this instead of query-based
lazy loading.

Author: teamchong

scripts/seed-local-r2.ts

@@ -122,22 +122,28 @@ async function seedIcebergTable(dir: string): Promise<void> {
     console.log(`  PUT ${r2Key} (${(data.length / 1024).toFixed(0)} KB)`);
     await uploadViaWorker(r2Key, data);
   }
-  // Trigger Iceberg discovery via query (should work immediately since R2 list() sees objects)
-  await new Promise(r => setTimeout(r, 500));
-  try {
-    const resp = await fetch(`${BASE_URL}/query`, {
-      method: "POST",
-      headers: { "content-type": "application/json" },
-      body: JSON.stringify({ table: name, filters: [], projections: ["id"], limit: 1 }),
-    });
-    if (resp.ok) {
-      console.log(`  Iceberg registered: ${name}`);
-    } else {
-      const text = await resp.text();
-      console.log(`  Iceberg registration: ${resp.status} (${text.slice(0, 80)})`);
+  // Register Iceberg table explicitly (bypasses R2 list() which is unreliable in miniflare)
+  // Find the metadata.json key we just uploaded
+  const metadataFiles = walkDir(dir)
+    .map(f => `${name}/${relative(dir, f)}`)
+    .filter(k => k.endsWith(".metadata.json"));
+  const metadataKey = metadataFiles[metadataFiles.length - 1]; // latest
+  if (metadataKey) {
+    try {
+      const resp = await fetch(`${BASE_URL}/register-iceberg`, {
+        method: "POST",
+        headers: { "content-type": "application/json" },
+        body: JSON.stringify({ table: name, metadataKey }),
+      });
+      if (resp.ok) {
+        console.log(`  Iceberg registered:`, await resp.json());
+      } else {
+        const text = await resp.text();
+        console.log(`  Iceberg registration: ${resp.status} (${text.slice(0, 80)})`);
+      }
+    } catch (err) {
+      console.log(`  Iceberg registration error: ${String(err).slice(0, 80)}`);
     }
-  } catch (err) {
-    console.log(`  Iceberg registration error: ${String(err).slice(0, 80)}`);
   }
 }
 

src/query-do.ts

@@ -80,6 +80,7 @@ export class QueryDO implements DurableObject {
       case "/tables":       return this.handleListTables();
       case "/meta":         return this.handleGetMeta(request);
       case "/diagnostics":  return this.handleDiagnostics();
+      case "/register-iceberg": return this.handleRegisterIceberg(request);
       default:              return new Response("Not found", { status: 404 });
     }
   }
@@ -999,6 +1000,84 @@ export class QueryDO implements DurableObject {
     return null;
   }
 
+  /** Register an Iceberg table by explicit metadata path (bypasses R2 list()). */
+  private async handleRegisterIceberg(request: Request): Promise<Response> {
+    const { table, metadataKey } = (await request.json()) as { table: string; metadataKey: string };
+    if (!table || !metadataKey) return this.json({ error: "Missing table or metadataKey" }, 400);
+
+    const result = await this.loadIcebergByKey(table, metadataKey);
+    if (!result) return this.json({ error: "Failed to load Iceberg metadata" }, 500);
+    return this.json({ registered: true, table, totalRows: result.totalRows, files: result.parquetFiles.length });
+  }
+
+  /** Load an Iceberg table from an explicit metadata.json key (no R2 list() needed). */
+  private async loadIcebergByKey(tableName: string, metadataKey: string): Promise<IcebergDatasetMeta | null> {
+    const metaObj = await this.env.DATA_BUCKET.get(metadataKey);
+    if (!metaObj) return null;
+
+    const metaJson = await metaObj.text();
+    const icebergMeta = parseIcebergMetadata(metaJson);
+    if (!icebergMeta) return null;
+
+    // Derive prefix from metadataKey (e.g., "bench_iceberg_100k/metadata/v1.metadata.json" → "bench_iceberg_100k/")
+    const prefix = metadataKey.replace(/metadata\/.*$/, "");
+
+    const manifestListKey = `${prefix}${icebergMeta.manifestListPath}`;
+    const manifestListObj = await this.env.DATA_BUCKET.get(manifestListKey);
+    if (!manifestListObj) return null;
+
+    const manifestListBytes = await manifestListObj.arrayBuffer();
+    const parquetPaths = extractParquetPathsFromManifest(manifestListBytes);
+    if (parquetPaths.length === 0) return null;
+
+    // Load each Parquet file's metadata
+    const fragmentMetas = new Map<number, TableMeta>();
+    let totalRows = 0;
+    for (let i = 0; i < parquetPaths.length; i++) {
+      const parquetKey = parquetPaths[i].startsWith(prefix) ? parquetPaths[i] : `${prefix}${parquetPaths[i]}`;
+      const head = await this.env.DATA_BUCKET.head(parquetKey);
+      if (!head) continue;
+
+      const fileSize = BigInt(head.size);
+      const tailObj = await this.env.DATA_BUCKET.get(parquetKey, {
+        range: { offset: Math.max(0, Number(fileSize) - 8), length: Math.min(8, Number(fileSize)) },
+      });
+      if (!tailObj) continue;
+
+      const tailBuf = await tailObj.arrayBuffer();
+      const footerLen = getParquetFooterLength(tailBuf);
+      if (!footerLen) continue;
+
+      const footerObj = await this.env.DATA_BUCKET.get(parquetKey, {
+        range: { offset: Number(fileSize) - footerLen - 8, length: footerLen },
+      });
+      if (!footerObj) continue;
+
+      const parquetFileMeta = parseParquetFooter(await footerObj.arrayBuffer());
+      if (!parquetFileMeta) continue;
+
+      const meta = parquetMetaToTableMeta(parquetFileMeta, parquetKey, fileSize);
+      meta.name = parquetPaths[i];
+      fragmentMetas.set(i, meta);
+      totalRows += meta.totalRows;
+    }
+
+    if (fragmentMetas.size === 0) return null;
+
+    const dataset: IcebergDatasetMeta = {
+      name: tableName, r2Prefix: prefix,
+      schema: icebergMeta.schema, snapshotId: icebergMeta.currentSnapshotId,
+      parquetFiles: parquetPaths, fragmentMetas, totalRows, updatedAt: Date.now(),
+    };
+    this.datasetCache.set(tableName, {
+      name: tableName, r2Prefix: prefix,
+      manifest: { version: 0, fragments: parquetPaths.map((p, idx) => ({ id: idx, filePath: p, physicalRows: 0 })), totalRows, schema: [] },
+      fragmentMetas, totalRows, updatedAt: Date.now(),
+    });
+    this.evictDatasetCache();
+    return dataset;
+  }
+
   /** Discover an Iceberg table in R2 by listing metadata/ for .metadata.json files. */
   private async loadIcebergFromR2(tableName: string): Promise<IcebergDatasetMeta | null> {
     for (const prefix of [`${tableName}/`, `data/${tableName}/`]) {

src/worker.ts

@@ -78,7 +78,8 @@ export default {
       url.pathname === "/query" ||
       url.pathname === "/query/stream" ||
       url.pathname === "/tables" ||
-      url.pathname === "/meta"
+      url.pathname === "/meta" ||
+      url.pathname === "/register-iceberg"
     ) {
       // Use a deterministic name per region so each region gets one Query DO.
       // The CF-Ray header contains the datacenter code (e.g., "SJC", "NRT").