The local grep service: one newest-first file walk, sidecar-or-scan per file (LLP 0265 T4)

src/core/cache/iceberg/store.js

@@ -448,6 +448,52 @@ async function loadDeletedPositions(metadata, resolver, dataFileMap) {
   return out
 }
 
+/**
+ * List the table's live data files with their identity-partition values and
+ * committed position-delete positions, for readers that walk files directly
+ * rather than scanning the table as one stream. The grep service is the
+ * consumer: its two tiers are per FILE (a sidecar-indexed file is searched
+ * through `parquetFind`, an unindexed one is brute-scanned), so it needs the
+ * file list, each file's partition `date` for the newest-first walk, and the
+ * deleted positions, because a raw file read does not apply position deletes
+ * and would otherwise resurrect purged rows (LLP 0104).
+ *
+ * Files whose manifest entry is deleted (status 2) or not parquet are
+ * excluded. A missing or snapshot-less table returns `[]`, matching the
+ * "empty table" degradation of `dataSourceForTable`. A metadata load that
+ * FAILS propagates, for the same reason: unreadable table metadata means an
+ * unknown row set, and a reader that swallows it reports "no matches" over a
+ * partition it never read. `dataSourceForTable` lets that error out, so
+ * `hyp query sql` fails loudly; grep search must not answer zero where SQL
+ * raises.
+ *
+ * @param {string} tablePath
+ * @returns {Promise<{ filePath: string, partition: Record<string, unknown>, recordCount: number, deletedPositions: Set<bigint> | undefined }[]>}
+ */
+export async function listLiveDataFiles(tablePath) {
+  if (!tableExists(tablePath)) return []
+  const { resolver, lister } = await getLocalIO()
+  const url = tableUrlForDir(tablePath)
+  const { metadata } = await loadLatestFileCatalogMetadata({ tableUrl: url, resolver, lister })
+  if (metadata['current-snapshot-id'] === undefined || !metadata.snapshots?.length) return []
+  const dataFileMap = await findDataFileEntries(metadata, resolver)
+  if (dataFileMap.size === 0) return []
+  const deleted = await loadDeletedPositions(metadata, resolver, dataFileMap)
+  /** @type {{ filePath: string, partition: Record<string, unknown>, recordCount: number, deletedPositions: Set<bigint> | undefined }[]} */
+  const out = []
+  for (const [filePath, { partition, entry }] of dataFileMap) {
+    const file = entry.data_file
+    if (String(file.file_format).toLowerCase() !== 'parquet') continue
+    out.push({
+      filePath,
+      partition,
+      recordCount: Number(file.record_count ?? 0),
+      deletedPositions: deleted.get(filePath),
+    })
+  }
+  return out
+}
+
 /**
  * Streaming counterpart to `readRowsFromTable`. Yields rows one at a
  * time so callers (in particular `QueryStorageService.readRows`) never

src/core/query/sql.js

@@ -483,14 +483,21 @@ export async function executeQuerySql(args) {
 }
 
 /**
+ * The query seam's freshness move: flush each referenced partition's
+ * pending spool (debounced under `auto`, forced under `always`, skipped
+ * under `never`) and report the staleness the debounce accepted.
+ * Exported so the grep service applies the identical policy; two read
+ * surfaces with different flush rules would answer differently about
+ * the same seconds-old row.
+ *
  * @param {{
  *   partitions: Array<{ tablePath?: string }>,
  *   storage: ExtendedQueryStorageService,
  *   refresh: RefreshMode,
  *   messages: string[],
  * }} args
  */
-async function settlePendingCacheForQuery(args) {
+export async function settlePendingCacheForQuery(args) {
   const now = Date.now()
   for (const partition of args.partitions) {
     if (!partition.tablePath) continue

src/core/query/visibility.js

@@ -68,6 +68,31 @@ export function callerSeesEverything(callerRank) {
   return callerRank >= MAX_CLASS_RANK
 }
 
+/**
+ * The one row-level withholding rule: a row whose `cwd` resolves to a class
+ * that outranks the caller's on the restrictiveness lattice is withheld.
+ * `withLocalOnlyVisibility` below applies it on the SQL read path; the grep
+ * service applies it per scanned row on its file walk (which cannot route
+ * through an `AsyncDataSource` wrapper). One exported predicate rather than
+ * two copies, so the two read surfaces cannot drift on what "local-only"
+ * hides. A cwd-less value returns false: those rows are `full`-class by
+ * construction on cwd-bearing datasets (see the wrapper's contract below).
+ *
+ * A corrupt machine-local list makes `resolve` throw
+ * (LocalOnlyListUnreadableError); callers let it propagate so the read
+ * fails loudly rather than silently resolving to "nothing withheld".
+ *
+ * @ref LLP 0105 [implements]: caller class >= row class on the lattice, the shared predicate form
+ * @param {UsagePolicyResolver} resolver
+ * @param {number} callerRank
+ * @param {unknown} cwd
+ * @returns {boolean}
+ */
+export function cwdWithheldFromCaller(resolver, callerRank, cwd) {
+  if (typeof cwd !== 'string' || cwd === '') return false
+  return CLASS_RANK[resolver.resolve(cwd).class] > callerRank
+}
+
 /**
  * Decorate one dataset's data source so every row the engine pulls honors
  * LLP 0105's invariant: content may only surface in a context at least as
@@ -153,7 +178,7 @@ export function withLocalOnlyVisibility(source, opts) {
               // (LocalOnlyListUnreadableError); let it propagate so the query
               // fails loudly rather than silently resolving to "nothing
               // withheld", matching the export seam's fail-safe polarity.
-              if (CLASS_RANK[resolver.resolve(cwd).class] > callerRank) {
+              if (cwdWithheldFromCaller(resolver, callerRank, cwd)) {
                 report.withheldRows += 1
                 continue
               }