view: stay responsive and fast over HTTP

CHANGELOG.md

@@ -7,7 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- `Chunk.prefetch` and `VortexHandle.prefetch` start the range requests of several columns together over HTTP, merged into few requests, and an HTTP handle keeps the segments it has read. ([#744](https://github.com/dfa1/vortex-java/issues/744))
+
 ### Fixed
+- `vortex view` no longer freezes while it fetches data over HTTP: it shows a loading spinner, `q` quits at once, and it fetches only the columns on screen. ([#744](https://github.com/dfa1/vortex-java/issues/744))
 - A decimal min or max in a flat node's stats is decoded as a `BigDecimal` instead of text, so a filtered scan can prune on it and `columnStats()` reports it. ([#710](https://github.com/dfa1/vortex-java/issues/710))
 - A `vortex.uuid` column's file statistics carry its min and max, as Rust writes them from 0.92, and `WriteSummary` reports them as `UUID`s. ([#730](https://github.com/dfa1/vortex-java/pull/730))
 - Opening a file over HTTP follows redirects (`Redirect.NORMAL`), so hosts that answer with a 302/307 to their storage CDN (e.g. Hugging Face's `/resolve/`) work. ([#742](https://github.com/dfa1/vortex-java/issues/742))

cli/src/main/java/io/github/dfa1/vortex/cli/CliHandles.java

@@ -56,7 +56,8 @@ static Optional<VortexHandle> openOnWorker(IoWorker worker, String target)
     /// @param handle the handle to close
     static void closeOnWorker(IoWorker worker, VortexHandle handle) {
         try {
-            worker.runAndAwait(handle::close);
+            // Bounded: a view quit while a load is stalled on the network must not hang here.
+            worker.runAndAwait(handle::close, IoWorker.TEARDOWN_WAIT);
         } catch (InterruptedException _) {
             Thread.currentThread().interrupt();
         }

cli/src/main/java/io/github/dfa1/vortex/cli/tui/IoWorker.java

@@ -1,8 +1,10 @@
 package io.github.dfa1.vortex.cli.tui;
 
+import java.time.Duration;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /// Single-threaded I/O executor that owns one [io.github.dfa1.vortex.reader.VortexHandle].
@@ -16,6 +18,10 @@
 /// when computing UI state.
 public final class IoWorker implements AutoCloseable {
 
+    /// How long teardown waits for the worker before giving up on a task stalled on the network;
+    /// the worker is a daemon thread, so the process can exit without it.
+    public static final Duration TEARDOWN_WAIT = Duration.ofMillis(250);
+
     private final BlockingQueue<Runnable> queue;
     private final AtomicInteger pending;
     private final Thread thread;
@@ -61,6 +67,25 @@ public void runAndAwait(Runnable task) throws InterruptedException {
         if (closed) {
             return;
         }
+        enqueueWithLatch(task).await();
+    }
+
+    /// Runs a task on the worker thread and waits for it to complete, but at most `timeout`.
+    /// The task stays queued after a timeout and still runs once the worker reaches it. For
+    /// teardown, which must not hang behind a load stalled on the network.
+    ///
+    /// @param task    task to execute
+    /// @param timeout longest time to wait for it
+    /// @return true if the task finished within `timeout`, false if the wait timed out
+    /// @throws InterruptedException if the calling thread is interrupted while waiting
+    public boolean runAndAwait(Runnable task, Duration timeout) throws InterruptedException {
+        if (closed) {
+            return true;
+        }
+        return enqueueWithLatch(task).await(timeout.toNanos(), TimeUnit.NANOSECONDS);
+    }
+
+    private CountDownLatch enqueueWithLatch(Runnable task) {
         // Count down only after pending is decremented, so a caller that reads pending()
         // immediately after this returns never observes the in-flight task still counted.
         CountDownLatch done = new CountDownLatch(1);
@@ -73,7 +98,7 @@ public void runAndAwait(Runnable task) throws InterruptedException {
                 done.countDown();
             }
         });
-        done.await();
+        return done;
     }
 
     /// Number of submitted tasks that have not yet finished.

cli/src/main/java/io/github/dfa1/vortex/cli/tui/LazyGridSource.java

@@ -9,6 +9,8 @@
 import io.github.dfa1.vortex.reader.VortexHandle;
 import io.github.dfa1.vortex.reader.array.Array;
 
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -25,6 +27,8 @@
 /// reader's confined `Arena` is never crossed from the render thread.
 public final class LazyGridSource implements AutoCloseable {
 
+    private static final int COLUMN_MARGIN = 2;
+
     private final VortexHandle handle;
     private final IoWorker worker;
     private final List<String> columns;
@@ -36,9 +40,34 @@ public final class LazyGridSource implements AutoCloseable {
     private ScanIterator iter;
     private int iterNextChunk;
     private Chunk currentChunk;
-    private int currentChunkIdx = -1;
+    private volatile int currentChunkIdx = -1;
     private Array[] currentColumns;
-    private boolean closed;
+    private volatile boolean closed;
+    private volatile Request latest;
+    private volatile Loaded loaded;
+    private volatile Failure failure;
+
+    /// A window the render thread asked [#peekRows(long, int, int, int)] for.
+    private record Request(long start, int count, int colFrom, int colTo) {
+    }
+
+    /// Why the load of `request` failed.
+    private record Failure(Request request, RuntimeException cause) {
+    }
+
+    /// The window the worker last finished formatting; rows are immutable once published and
+    /// hold text only in columns `[colFrom, colTo)`.
+    @SuppressWarnings("java:S6218") // internal data carrier; the rows are never compared.
+    private record Loaded(long start, int colFrom, int colTo, String[][] rows) {
+    }
+
+    /// Result of [#peekRows(long, int, int, int)].
+    ///
+    /// @param rows     one slot per requested row; rows not loaded yet are filled with `…`
+    /// @param complete true when no row is a placeholder, so nothing is left to wait for
+    @SuppressWarnings("java:S6218") // internal data carrier; the rows are never compared.
+    public record Rows(String[][] rows, boolean complete) {
+    }
 
     /// Builds a lazy source over `handle`, walking the layout once up front to
     /// derive chunk boundaries.
@@ -117,50 +146,89 @@ public int currentChunkIndex() {
         return currentChunkIdx;
     }
 
-    /// Returns formatted rows for the absolute-row window `[startAbsRow, startAbsRow + count)`.
+    /// Returns the formatted rows of the absolute-row window `[startAbsRow, startAbsRow + count)`
+    /// that are already loaded, and asks the worker for the rest, without ever waiting for it.
+    ///
+    /// A window may straddle a chunk boundary; the worker handles the chunk transitions. Returned
+    /// rows are independent `String[]` arrays, safe to keep after the underlying chunk is closed,
+    /// and rows past `totalRows()` are empty.
+    ///
+    /// Rows the worker has not delivered yet come back as `…` cells with
+    /// [Rows#complete()] false; the caller redraws later. A request superseded by a newer
+    /// one before the worker reaches it is dropped. Call from one thread only.
     ///
-    /// The window may straddle a chunk boundary; this method handles the chunk
-    /// transitions transparently. Returned rows are independent `String[]`
-    /// arrays — safe to keep after the underlying chunk is closed.
+    /// Only columns `[colFrom, colTo)` are decoded, widened by a small margin so a one-column
+    /// scroll finds its data already loaded; a cell outside that range is `null`.
     ///
     /// @param startAbsRow first absolute row in the window
-    /// @param count       number of rows to read; rows past `totalRows()` are
-    ///                    returned as empty `String[]`
-    /// @return array of length `count`, one row per slot
-    /// @throws InterruptedException if the calling thread is interrupted while
-    ///                              waiting for the worker
-    /// @throws VortexException      if this source has already been closed
-    public String[][] readRows(long startAbsRow, int count) throws InterruptedException {
+    /// @param count       number of rows to read
+    /// @param colFrom     first visible column
+    /// @param colTo       one past the last visible column
+    /// @return the cached rows plus whether any visible cell is still loading
+    /// @throws VortexException if this source is closed or a background load failed
+    public Rows peekRows(long startAbsRow, int count, int colFrom, int colTo) {
         if (closed) {
             throw new VortexException("LazyGridSource closed");
         }
+        Request wanted = new Request(startAbsRow, count,
+                Math.max(0, colFrom - COLUMN_MARGIN), Math.min(columns.size(), colTo + COLUMN_MARGIN));
+        Failure failed = failure;
+        if (failed != null && failed.request.equals(wanted)) {
+            throw failed.cause;
+        }
+        Loaded snapshot = loaded;
+        boolean columnsLoaded = snapshot != null && colFrom >= snapshot.colFrom && colTo <= snapshot.colTo;
         String[][] out = new String[count][];
-        if (count == 0 || startAbsRow >= totalRows) {
-            for (int i = 0; i < count; i++) {
+        boolean complete = true;
+        for (int i = 0; i < count; i++) {
+            long row = startAbsRow + i;
+            if (row >= totalRows) {
                 out[i] = emptyRow();
+            } else if (columnsLoaded && row >= snapshot.start && row - snapshot.start < snapshot.rows.length) {
+                out[i] = snapshot.rows[(int) (row - snapshot.start)];
+            } else {
+                out[i] = loadingRow();
+                complete = false;
             }
-            return out;
         }
-        AtomicReference<RuntimeException> failure = new AtomicReference<>();
-        worker.runAndAwait(() -> {
+        if (!complete) {
+            request(wanted);
+        }
+        return new Rows(out, complete);
+    }
+
+    private void request(Request request) {
+        if (request.equals(latest)) {
+            return;
+        }
+        latest = request;
+        failure = null;
+        worker.submit(() -> {
+            // A load superseded by a newer window, or queued before close, is skipped: each one
+            // costs round trips over HTTP.
+            if (closed || latest != request) {
+                return;
+            }
+            String[][] rows = new String[request.count][];
             try {
-                fillWindow(out, startAbsRow, count);
+                fillWindow(rows, request.start, request.count, request.colFrom, request.colTo);
+                loaded = new Loaded(request.start, request.colFrom, request.colTo, rows);
             } catch (RuntimeException e) {
-                failure.set(e);
+                // Tied to its window, so one failed load (a transient 503) does not poison the rest.
+                if (latest == request) {
+                    failure = new Failure(request, e);
+                }
             }
         });
-        if (failure.get() != null) {
-            throw failure.get();
-        }
-        for (int i = 0; i < count; i++) {
-            if (out[i] == null) {
-                out[i] = emptyRow();
-            }
-        }
-        return out;
     }
 
-    private void fillWindow(String[][] out, long startAbsRow, int count) {
+    private String[] loadingRow() {
+        String[] row = new String[columns.size()];
+        Arrays.fill(row, "…");
+        return row;
+    }
+
+    private void fillWindow(String[][] out, long startAbsRow, int count, int colFrom, int colTo) {
         long row = startAbsRow;
         int slot = 0;
         while (slot < count) {
@@ -170,31 +238,17 @@ private void fillWindow(String[][] out, long startAbsRow, int count) {
             }
             int chunkIdx = findChunk(row);
             loadChunk(chunkIdx);
+            prefetchColumns(colFrom, colTo);
             long chunkStart = chunkStartRows[chunkIdx];
             long chunkEnd = chunkStartRows[chunkIdx + 1];
             long limit = Math.min(chunkEnd, startAbsRow + count);
             while (row < limit && slot < count) {
-                out[slot++] = formatRow(currentColumns, row - chunkStart);
+                out[slot++] = formatRow(row - chunkStart, colFrom, colTo);
                 row++;
             }
         }
     }
 
-    /// Returns a single formatted row at absolute index `absRow`.
-    ///
-    /// @param absRow row index in `[0, totalRows())`
-    /// @return one cell per column
-    /// @throws InterruptedException if the calling thread is interrupted while
-    ///                              waiting for the worker
-    /// @throws VortexException      if `absRow` is out of bounds or this source
-    ///                              has already been closed
-    public String[] row(long absRow) throws InterruptedException {
-        if (absRow < 0 || absRow >= totalRows) {
-            throw new VortexException("row index out of bounds: " + absRow);
-        }
-        return readRows(absRow, 1)[0];
-    }
-
     private int findChunk(long absRow) {
         int lo = 0;
         int hi = chunkCount - 1;
@@ -235,9 +289,18 @@ private void loadChunk(int chunkIdx) {
         currentChunk = chunk;
         currentChunkIdx = chunkIdx;
         currentColumns = new Array[columns.size()];
-        for (int c = 0; c < columns.size(); c++) {
-            currentColumns[c] = chunk.column(ColumnName.of(columns.get(c)));
+    }
+
+    /// Worker-thread-only. Starts the range requests of the columns about to be formatted all
+    /// at once, so over HTTP they cost one round trip together rather than one each.
+    private void prefetchColumns(int colFrom, int colTo) {
+        List<ColumnName> wanted = new ArrayList<>();
+        for (int c = colFrom; c < colTo; c++) {
+            if (currentColumns[c] == null) {
+                wanted.add(ColumnName.of(columns.get(c)));
+            }
         }
+        currentChunk.prefetch(wanted);
     }
 
     private void closeCurrentChunk() {
@@ -249,11 +312,15 @@ private void closeCurrentChunk() {
         }
     }
 
-    private String[] formatRow(Array[] arrays, long inChunk) {
-        int n = columns.size();
-        String[] row = new String[n];
-        for (int c = 0; c < n; c++) {
-            row[c] = GridRender.formatCell(arrays[c], inChunk, columnDtypes.get(c));
+    /// Worker-thread-only. Formats columns `[colFrom, colTo)` of one row; the others stay
+    /// `null`. A column is decoded (one range request over HTTP) the first time it is formatted.
+    private String[] formatRow(long inChunk, int colFrom, int colTo) {
+        String[] row = new String[columns.size()];
+        for (int c = colFrom; c < colTo; c++) {
+            if (currentColumns[c] == null) {
+                currentColumns[c] = currentChunk.column(ColumnName.of(columns.get(c)));
+            }
+            row[c] = GridRender.formatCell(currentColumns[c], inChunk, columnDtypes.get(c));
         }
         return row;
     }
@@ -269,13 +336,15 @@ public void close() {
         }
         closed = true;
         try {
+            // Bounded: a load stalled on the network must not keep `q` from quitting. The cleanup
+            // stays queued and runs when the worker gets to it.
             worker.runAndAwait(() -> {
                 closeCurrentChunk();
                 if (iter != null) {
                     iter.close();
                     iter = null;
                 }
-            });
+            }, IoWorker.TEARDOWN_WAIT);
         } catch (InterruptedException _) {
             Thread.currentThread().interrupt();
         }