diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0ce394d33..19ee35e24 100644
--- a/CHANGELOG.md
+++ b/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))
diff --git a/cli/src/main/java/io/github/dfa1/vortex/cli/CliHandles.java b/cli/src/main/java/io/github/dfa1/vortex/cli/CliHandles.java
index 650dd12e1..6ffc8eb44 100644
--- a/cli/src/main/java/io/github/dfa1/vortex/cli/CliHandles.java
+++ b/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();
         }
diff --git a/cli/src/main/java/io/github/dfa1/vortex/cli/tui/IoWorker.java b/cli/src/main/java/io/github/dfa1/vortex/cli/tui/IoWorker.java
index 46c0b44e0..f1193169b 100644
--- a/cli/src/main/java/io/github/dfa1/vortex/cli/tui/IoWorker.java
+++ b/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.
diff --git a/cli/src/main/java/io/github/dfa1/vortex/cli/tui/LazyGridSource.java b/cli/src/main/java/io/github/dfa1/vortex/cli/tui/LazyGridSource.java
index 2456597a2..f45f5dd50 100644
--- a/cli/src/main/java/io/github/dfa1/vortex/cli/tui/LazyGridSource.java
+++ b/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();
         }
diff --git a/cli/src/main/java/io/github/dfa1/vortex/cli/tui/VortexGridTui.java b/cli/src/main/java/io/github/dfa1/vortex/cli/tui/VortexGridTui.java
index d68a188e7..8e387a9c1 100644
--- a/cli/src/main/java/io/github/dfa1/vortex/cli/tui/VortexGridTui.java
+++ b/cli/src/main/java/io/github/dfa1/vortex/cli/tui/VortexGridTui.java
@@ -5,7 +5,9 @@
 import io.github.dfa1.vortex.cli.tui.term.Terminal;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.List;
+import java.util.Optional;
 
 /// Excel-like scrollable grid viewer over a [LazyGridSource].
 ///
@@ -45,6 +47,8 @@ private static final class Loop {
         private static final int MIN_COL_WIDTH = 4;
         private static final int MAX_COL_WIDTH = 24;
         private static final int ROW_INDEX_PAD = 2;
+        private static final Duration LOADING_TICK = Duration.ofMillis(50);
+        private static final String[] SPINNER = {"⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"};
 
         private final Terminal term;
         private final String source;
@@ -59,6 +63,8 @@ private static final class Loop {
         private long rowOffset;
         private int colOffset;
         private boolean dirty = true;
+        private boolean loading;
+        private int spinnerFrame;
         private String errorMessage;
 
         Loop(Terminal term, String source, LazyGridSource data) {
@@ -81,7 +87,19 @@ void run() throws IOException {
                         render();
                         dirty = false;
                     }
-                    Key key = term.readKey();
+                    Key key;
+                    if (loading) {
+                        // The worker is still filling the window: poll keys so `q` stays live,
+                        // and redraw on a tick to pick up what has arrived.
+                        Optional<Key> polled = term.readKey(LOADING_TICK);
+                        if (polled.isEmpty()) {
+                            dirty = true;
+                            continue;
+                        }
+                        key = polled.get();
+                    } else {
+                        key = term.readKey();
+                    }
                     if (!handle(key)) {
                         return;
                     }
@@ -190,7 +208,7 @@ private void render() throws IOException {
             ensureCursorVisible(termRows, termCols);
 
             int viewportRows = Math.max(0, termRows - 3);
-            String[][] window = fetchWindow(rowOffset, viewportRows);
+            String[][] window = fetchWindow(rowOffset, viewportRows, visibleColsEnd(termCols));
 
             StringBuilder out = new StringBuilder(termCols * termRows);
             out.append(Ansi.CURSOR_HOME);
@@ -203,14 +221,24 @@ private void render() throws IOException {
             term.flush();
         }
 
-        private String[][] fetchWindow(long startAbsRow, int viewportRows) {
+        /// One past the last column drawn, by the same fit rule as [#renderHeader].
+        private int visibleColsEnd(int termCols) {
+            int width = rowNumberWidth;
+            int c = colOffset;
+            while (c < totalCols && width + colWidths[c] + 1 <= termCols) {
+                width += colWidths[c] + 1;
+                c++;
+            }
+            return c;
+        }
+
+        private String[][] fetchWindow(long startAbsRow, int viewportRows, int colEnd) {
             try {
-                return data.readRows(startAbsRow, viewportRows);
-            } catch (InterruptedException _) {
-                Thread.currentThread().interrupt();
-                errorMessage = "interrupted";
-                return placeholderWindow(viewportRows);
+                LazyGridSource.Rows rows = data.peekRows(startAbsRow, viewportRows, colOffset, colEnd);
+                loading = !rows.complete();
+                return rows.rows();
             } catch (RuntimeException e) {
+                loading = false;
                 errorMessage = e.getClass().getSimpleName()
                         + (e.getMessage() != null ? ": " + e.getMessage() : "");
                 return placeholderWindow(viewportRows);
@@ -267,7 +295,7 @@ private void renderTitle(StringBuilder out, int termCols) {
                     : "";
             String title = err != null
                     ? " Vortex View  " + source + "  ERROR: " + err
-                    : " Vortex View  " + source
+                    : " Vortex View  " + loadingLabel() + source
                             + "  rows " + totalRows
                             + "  cols " + totalCols
                             + chunkLabel;
@@ -276,6 +304,16 @@ private void renderTitle(StringBuilder out, int termCols) {
             out.append(Ansi.RESET).append("\r\n");
         }
 
+        /// Leads the title so a long source URL cannot push it off screen; the spinner
+        /// advances once per tick to show the view is alive while a fetch is slow.
+        private String loadingLabel() {
+            if (!loading) {
+                return "";
+            }
+            spinnerFrame = (spinnerFrame + 1) % SPINNER.length;
+            return SPINNER[spinnerFrame] + " loading…  ";
+        }
+
         private void renderHeader(StringBuilder out, int termCols) {
             StringBuilder line = new StringBuilder(termCols);
             appendPadded(line, padRight("#", rowNumberWidth - 1) + " ", rowNumberWidth);
diff --git a/cli/src/test/java/io/github/dfa1/vortex/cli/tui/IoWorkerTest.java b/cli/src/test/java/io/github/dfa1/vortex/cli/tui/IoWorkerTest.java
index 0b3ade61c..c78ea22f6 100644
--- a/cli/src/test/java/io/github/dfa1/vortex/cli/tui/IoWorkerTest.java
+++ b/cli/src/test/java/io/github/dfa1/vortex/cli/tui/IoWorkerTest.java
@@ -2,7 +2,9 @@
 
 import org.junit.jupiter.api.Test;
 
+import java.time.Duration;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
@@ -30,6 +32,30 @@ void runAndAwait_runsTaskOnTheWorkerThread() throws InterruptedException {
         }
     }
 
+    @Test
+    void runAndAwaitWithTimeout_returnsFalseWhileTheWorkerIsBusy_andTrueOnceItIsFree() throws InterruptedException {
+        // Given — the worker is held by a task that outlasts the wait
+        try (IoWorker sut = new IoWorker("test-io")) {
+            CountDownLatch release = new CountDownLatch(1);
+            sut.submit(() -> {
+                try {
+                    release.await();
+                } catch (InterruptedException _) {
+                    Thread.currentThread().interrupt();
+                }
+            });
+
+            // When
+            boolean busy = sut.runAndAwait(() -> { }, Duration.ofMillis(20));
+            release.countDown();
+            boolean result = sut.runAndAwait(() -> { }, Duration.ofSeconds(5));
+
+            // Then — the timed-out task stayed queued, and the next wait saw the worker free again
+            assertThat(busy).isFalse();
+            assertThat(result).isTrue();
+        }
+    }
+
     @Test
     void pending_isZeroAfterTaskCompletes() throws InterruptedException {
         // Given
diff --git a/cli/src/test/java/io/github/dfa1/vortex/cli/tui/LazyGridSourceTest.java b/cli/src/test/java/io/github/dfa1/vortex/cli/tui/LazyGridSourceTest.java
index 119186555..03f61a7d6 100644
--- a/cli/src/test/java/io/github/dfa1/vortex/cli/tui/LazyGridSourceTest.java
+++ b/cli/src/test/java/io/github/dfa1/vortex/cli/tui/LazyGridSourceTest.java
@@ -6,12 +6,15 @@
 import org.junit.jupiter.api.io.TempDir;
 
 import java.nio.file.Path;
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
 
-/// Unit tests for [LazyGridSource] row decoding: column metadata, windowed reads,
-/// single-row access, out-of-range handling, and closed-source rejection. All
+/// Unit tests for [LazyGridSource] row decoding: column metadata, non-blocking windowed reads,
+/// out-of-range handling, superseded and failed loads, and closed-source rejection. All
 /// handle I/O is routed through a real [IoWorker]; the handle is opened on the
 /// worker thread to satisfy the reader's confined-arena threading contract.
 class LazyGridSourceTest {
@@ -25,7 +28,7 @@ void exposesColumnsAndRowCount() throws Exception {
         Path file = TuiTestSupport.writeGridVortex(tmp, "src.vortex", 25);
 
         // When / Then
-        withSource(file, source -> {
+        withSource(file, (source, _) -> {
             assertThat(source.columns()).containsExactly("a", "b", "c");
             assertThat(source.totalRows()).isEqualTo(25L);
             assertThat(source.chunkCount()).isGreaterThanOrEqualTo(1);
@@ -38,8 +41,8 @@ void readsWindowOfRows() throws Exception {
         Path file = TuiTestSupport.writeGridVortex(tmp, "src.vortex", 25);
 
         // When
-        withSource(file, source -> {
-            String[][] window = source.readRows(2, 3);
+        withSource(file, (source, worker) -> {
+            String[][] window = read(source, worker, 2, 3);
 
             // Then
             assertThat(window).hasNumberOfRows(3);
@@ -49,49 +52,118 @@ void readsWindowOfRows() throws Exception {
     }
 
     @Test
-    void padsRowsPastEndWithEmptyArrays() throws Exception {
-        // Given — only 5 rows, but a window of 4 starting at row 3 overruns the end
-        Path file = TuiTestSupport.writeGridVortex(tmp, "src.vortex", 5);
+    void peekRowsReturnsPlaceholdersThenLoadedRowsWithoutBlocking() throws Exception {
+        // Given — nothing cached yet, so the first peek must not wait for the worker
+        Path file = TuiTestSupport.writeGridVortex(tmp, "src.vortex", 25);
 
-        // When
-        withSource(file, source -> {
-            String[][] window = source.readRows(3, 4);
+        withSource(file, (source, worker) -> {
+            // When
+            LazyGridSource.Rows first = source.peekRows(2, 3, 0, 3);
+            awaitWorker(worker);
+            LazyGridSource.Rows result = source.peekRows(2, 3, 0, 3);
 
-            // Then — rows 3,4 present; rows 5,6 are empty (one slot per column)
-            assertThat(window).hasNumberOfRows(4);
-            assertThat(window[0]).containsExactly("3", "103", "203");
-            assertThat(window[2]).hasSize(3).containsOnlyNulls();
+            // Then — placeholders now, real rows once the worker has delivered
+            assertThat(first.complete()).isFalse();
+            assertThat(first.rows()[0]).containsExactly("…", "…", "…");
+            assertThat(result.complete()).isTrue();
+            assertThat(result.rows()[0]).containsExactly("2", "102", "202");
+            assertThat(result.rows()[2]).containsExactly("4", "104", "204");
         });
     }
 
     @Test
-    void rowReturnsSingleFormattedRow() throws Exception {
-        // Given
-        Path file = TuiTestSupport.writeGridVortex(tmp, "src.vortex", 10);
+    void peekRowsSkipsAWindowSupersededBeforeTheWorkerReachesIt() throws Exception {
+        // Given — the worker is busy, so window A is still queued when the user scrolls on to B
+        Path file = TuiTestSupport.writeGridVortex(tmp, "src.vortex", 25);
 
-        // When
-        withSource(file, source -> {
-            String[] row = source.row(7);
+        withSource(file, (source, worker) -> {
+            CountDownLatch release = blockWorker(worker);
+            source.peekRows(0, 3, 0, 3);
+            source.peekRows(10, 3, 0, 3);
+
+            // When
+            release.countDown();
+            awaitWorker(worker);
+
+            // Then — B was loaded and A never was: loading A would only have cost round trips
+            assertThat(source.peekRows(10, 3, 0, 3).complete()).isTrue();
+            assertThat(source.peekRows(0, 3, 0, 3).complete()).isFalse();
+        });
+    }
+
+    @Test
+    void peekRowsReportsColumnsOutsideTheLoadedRangeAsIncomplete() throws Exception {
+        // Given — 4 columns; asking for the first one loads it plus a margin of two
+        Path file = TuiTestSupport.writeMultiTypeVortex(tmp, "typed.vortex");
+
+        withSource(file, (source, worker) -> {
+            source.peekRows(0, 3, 0, 1);
+            awaitWorker(worker);
+
+            // When
+            LazyGridSource.Rows inside = source.peekRows(0, 3, 0, 1);
+            LazyGridSource.Rows outside = source.peekRows(0, 3, 3, 4);
 
             // Then
-            assertThat(row).containsExactly("7", "107", "207");
+            assertThat(inside.complete()).isTrue();
+            assertThat(inside.rows()[0][0]).isEqualTo("0");
+            assertThat(outside.complete()).isFalse();
+            assertThat(outside.rows()[0]).containsOnly("…");
         });
     }
 
     @Test
-    void rowOutOfBoundsThrows() throws Exception {
-        // Given
-        Path file = TuiTestSupport.writeGridVortex(tmp, "src.vortex", 3);
+    void peekRowsFailureBelongsToItsWindow() throws Exception {
+        // Given — a handle closed under the source, so every load fails
+        Path file = TuiTestSupport.writeGridVortex(tmp, "src.vortex", 25);
+        try (IoWorker worker = new IoWorker("src-test-io")) {
+            VortexHandle handle = TuiTestSupport.openOnWorker(worker, file);
+            try (LazyGridSource source = LazyGridSource.open(handle, worker)) {
+                TuiTestSupport.closeOnWorker(worker, handle);
+                source.peekRows(0, 3, 0, 3);
+                awaitWorker(worker);
 
-        // When / Then
-        withSource(file, source ->
-                assertThatThrownBy(() -> source.row(99))
-                        .isInstanceOf(VortexException.class)
-                        .hasMessageContaining("out of bounds"));
+                // When / Then — that window reports the failure, another one does not inherit it
+                assertThatThrownBy(() -> source.peekRows(0, 3, 0, 3)).isInstanceOf(RuntimeException.class);
+                assertThat(source.peekRows(10, 3, 0, 3).complete()).isFalse();
+            }
+        }
     }
 
     @Test
-    void readRowsAfterCloseThrows() throws Exception {
+    void closeDoesNotWaitForALoadStuckOnTheWorker() throws Exception {
+        // Given — a load stalled behind a busy worker, as on a hung HTTP request
+        Path file = TuiTestSupport.writeGridVortex(tmp, "src.vortex", 25);
+        try (IoWorker worker = new IoWorker("src-test-io")) {
+            VortexHandle handle = TuiTestSupport.openOnWorker(worker, file);
+            LazyGridSource source = LazyGridSource.open(handle, worker);
+            CountDownLatch release = blockWorker(worker);
+            source.peekRows(0, 3, 0, 3);
+
+            // When / Then — close gives up after the teardown wait instead of hanging
+            assertTimeoutPreemptively(Duration.ofSeconds(5), source::close);
+            release.countDown();
+        }
+    }
+
+    @Test
+    void padsRowsPastEndWithEmptyArrays() throws Exception {
+        // Given — only 5 rows, but a window of 4 starting at row 3 overruns the end
+        Path file = TuiTestSupport.writeGridVortex(tmp, "src.vortex", 5);
+
+        // When
+        withSource(file, (source, worker) -> {
+            String[][] window = read(source, worker, 3, 4);
+
+            // Then — rows 3,4 present; rows 5,6 are empty (one slot per column)
+            assertThat(window).hasNumberOfRows(4);
+            assertThat(window[0]).containsExactly("3", "103", "203");
+            assertThat(window[2]).hasSize(3).containsOnlyNulls();
+        });
+    }
+
+    @Test
+    void peekRowsAfterCloseThrows() throws Exception {
         // Given
         Path file = TuiTestSupport.writeGridVortex(tmp, "src.vortex", 3);
 
@@ -104,7 +176,7 @@ void readRowsAfterCloseThrows() throws Exception {
                 source.close(); // idempotent
 
                 // Then
-                assertThatThrownBy(() -> source.readRows(0, 1))
+                assertThatThrownBy(() -> source.peekRows(0, 1, 0, 1))
                         .isInstanceOf(VortexException.class)
                         .hasMessageContaining("closed");
             } finally {
@@ -119,9 +191,9 @@ void formatsEachColumnType() throws Exception {
         Path file = TuiTestSupport.writeMultiTypeVortex(tmp, "typed.vortex");
 
         // When
-        withSource(file, source -> {
+        withSource(file, (source, worker) -> {
             assertThat(source.columns()).containsExactly("i", "d", "flag", "name");
-            String[] row = source.row(1);
+            String[] row = read(source, worker, 1, 1)[0];
 
             // Then — long, double, boolean, utf8 each rendered with its formatter
             assertThat(row).containsExactly("1", "1.5", "false", "b");
@@ -135,34 +207,61 @@ void readsAcrossChunkBoundaryThenScrollsBack() throws Exception {
         Path file = TuiTestSupport.writeMultiTypeVortex(tmp, "typed.vortex");
 
         // When
-        withSource(file, source -> {
+        withSource(file, (source, worker) -> {
             assertThat(source.totalRows()).isEqualTo(5L);
             assertThat(source.chunkCount()).isEqualTo(2);
 
             // window straddling the chunk boundary (rows 2,3,4 -> chunk0 tail + chunk1)
-            String[][] straddle = source.readRows(2, 3);
+            String[][] straddle = read(source, worker, 2, 3);
             assertThat(straddle[0]).containsExactly("2", "2.5", "true", "c");
             assertThat(straddle[1]).containsExactly("3", "3.5", "false", "d");
 
             // now scroll back to the head -> reopen scan from chunk 0
-            String[] first = source.row(0);
+            String[] first = read(source, worker, 0, 1)[0];
             assertThat(first).containsExactly("0", "0.5", "true", "a");
         });
     }
 
     @FunctionalInterface
-    private interface SourceAction {
-        void run(LazyGridSource source) throws InterruptedException;
+    private interface WorkerAction {
+        void run(LazyGridSource source, IoWorker worker) throws InterruptedException;
+    }
+
+    /// All columns of the window, once the worker has loaded it.
+    private static String[][] read(LazyGridSource source, IoWorker worker, long start, int count)
+            throws InterruptedException {
+        int columns = source.columns().size();
+        source.peekRows(start, count, 0, columns);
+        awaitWorker(worker);
+        return source.peekRows(start, count, 0, columns).rows();
     }
 
-    private static void withSource(Path file, SourceAction action) throws InterruptedException {
+    private static void withSource(Path file, WorkerAction action) throws InterruptedException {
         try (IoWorker worker = new IoWorker("src-test-io")) {
             VortexHandle handle = TuiTestSupport.openOnWorker(worker, file);
             try (LazyGridSource source = LazyGridSource.open(handle, worker)) {
-                action.run(source);
+                action.run(source, worker);
             } finally {
                 TuiTestSupport.closeOnWorker(worker, handle);
             }
         }
     }
+
+    /// The worker runs tasks in order, so once this returns every task submitted before it is done.
+    private static void awaitWorker(IoWorker worker) throws InterruptedException {
+        worker.runAndAwait(() -> { });
+    }
+
+    /// Occupies the worker until the returned latch is released.
+    private static CountDownLatch blockWorker(IoWorker worker) {
+        CountDownLatch release = new CountDownLatch(1);
+        worker.submit(() -> {
+            try {
+                release.await();
+            } catch (InterruptedException _) {
+                Thread.currentThread().interrupt();
+            }
+        });
+        return release;
+    }
 }
diff --git a/cli/src/test/java/io/github/dfa1/vortex/cli/tui/VortexGridTuiTest.java b/cli/src/test/java/io/github/dfa1/vortex/cli/tui/VortexGridTuiTest.java
index 7f15501eb..4bb3ec220 100644
--- a/cli/src/test/java/io/github/dfa1/vortex/cli/tui/VortexGridTuiTest.java
+++ b/cli/src/test/java/io/github/dfa1/vortex/cli/tui/VortexGridTuiTest.java
@@ -57,6 +57,28 @@ void navigatesAndRendersGrid() throws Exception {
                 .contains("a").contains("b").contains("c");
     }
 
+    @Test
+    void titleShowsLoadingBeforeAnyRowArrives() throws Exception {
+        // Given — the first frame is drawn before the worker has delivered a window;
+        // a long source label must not push the marker off the title line
+        Path file = TuiTestSupport.writeGridVortex(tmp, "grid.vortex", 5);
+        FakeTerminal term = new FakeTerminal(new Terminal.Size(24, 80),
+                List.of(new Key.Char('q')));
+
+        // When
+        try (IoWorker worker = new IoWorker("grid-test-io")) {
+            VortexHandle handle = TuiTestSupport.openOnWorker(worker, file);
+            try (LazyGridSource source = LazyGridSource.open(handle, worker)) {
+                VortexGridTui.run(term, "https://example.com/" + "x".repeat(100), source);
+            } finally {
+                TuiTestSupport.closeOnWorker(worker, handle);
+            }
+        }
+
+        // Then
+        assertThat(term.output()).contains("loading…");
+    }
+
     @Test
     void quitsImmediatelyOnEscape() throws Exception {
         // Given — Escape is one of three quit keys; the loop must exit on the
diff --git a/docs/reference.md b/docs/reference.md
index d2bd79197..c03d39c43 100644
--- a/docs/reference.md
+++ b/docs/reference.md
@@ -83,7 +83,9 @@ shortcut returning a nullable copy), `withNullable(boolean)`, `DType.Struct.fiel
 Interface common to both handle implementations below — everything scan-shaped
 (`dtype()`, `layout()`, `scan(ScanOptions)`, `rawSegment(SegmentSpec)`, …) is declared here so code
 that only reads (e.g. `ParquetExporter.exportParquet(VortexHandle, Path)`) doesn't care whether the
-source is a local file or a remote URL. Implements `Closeable`.
+source is a local file or a remote URL. Implements `Closeable`. `prefetch(Collection<SegmentSpec>)` is
+a read-ahead hint for the segments about to be read: a no-op on `VortexReader`, concurrent merged range
+requests on `VortexHttpReader`. Call it from the thread that opened the handle.
 
 | Implementation    | Backing storage                              |
 |--------------------|-----------------------------------------------|
@@ -116,7 +118,14 @@ Handle to a remote Vortex file read via HTTP Range requests — no full-file dow
 `VortexHandle`. On open, fetches the last 65 KB (`TAIL_SIZE`) in one request to locate the trailer,
 postscript, and metadata blobs, parsing `Content-Range` for the file size instead of a separate
 `HEAD` round trip; each subsequent `rawSegment(SegmentSpec)` call during a scan fires one targeted
-Range request. All fetched bytes share a single confined `Arena`, released on `close()`.
+Range request. `Chunk.prefetch(columns)` (or `VortexHandle.prefetch(specs)`) starts those requests
+together, at most 8 in flight, merging segments up to 1 MiB apart (at most 16 MiB per request, as Rust's
+object-storage `CoalesceConfig`), so a screenful of columns costs about one round trip instead of one each.
+A call queues at most 64 MiB on the heap; segments past that are read on demand. A no-op on a
+memory-mapped file or after `close()`. Every segment read is kept for the life of the handle (no eviction or
+TTL: the file is immutable, and the bytes stay in the `Arena` anyway), so reading it again costs no
+request. All fetched bytes share a single confined `Arena`, released on
+`close()`.
 
 | Method                                                                     | Notes                                                    |
 |------------------------------------------------------------------------------|-----------------------------------------------------------|
@@ -282,6 +291,7 @@ never desync.
 | `columns()`                                 | `SequencedMap<ColumnName, Chunk.Column>`, schema order, unmodifiable |
 | `<T extends Array> column(ColumnName name)` | Typed column lookup; throws `VortexException` if absent  |
 | `as(ColumnName name, Class<T> domainType)` | Extension column → typed `List<T>`, via the registered `ExtensionDecoder` |
+| `prefetch(Collection<ColumnName> names)`    | Hint that these columns are about to be read, so a remote file fetches them concurrently, in merged requests (see `VortexHttpReader`); changes timing only, a no-op on a local file |
 | `slice(long offset, long length)`           | Zero-copy view of a row range; valid while this chunk is open, closing it does nothing |
 | `isClosed()`                                | Whether `close()` has run                                |
 | `close()`                                   | Releases the chunk's arena. Idempotent.                  |
diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/Chunk.java b/reader/src/main/java/io/github/dfa1/vortex/reader/Chunk.java
index ca4fdeed6..61a58bab3 100644
--- a/reader/src/main/java/io/github/dfa1/vortex/reader/Chunk.java
+++ b/reader/src/main/java/io/github/dfa1/vortex/reader/Chunk.java
@@ -9,6 +9,8 @@
 import io.github.dfa1.vortex.reader.extension.ExtensionDecoder;
 
 import java.lang.foreign.Arena;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
@@ -53,6 +55,7 @@ public final class Chunk implements AutoCloseable {
     private final List<ColumnName> names;
     private final Map<ColumnName, Integer> positions;
     private final IntFunction<Column> decoder;
+    private final Consumer<List<Integer>> prefetcher;
     /// Per column, its decoded value once read, else `null`.
     private final Column[] decoded;
     private SequencedMap<ColumnName, Column> columns;
@@ -65,19 +68,21 @@ public final class Chunk implements AutoCloseable {
     Chunk(long rowCount, SequencedMap<ColumnName, Column> columns,
             Arena arena, ReadRegistry registry, Consumer<Chunk> onClose) {
         this(rowCount, List.copyOf(columns.keySet()), positions(List.copyOf(columns.keySet())),
-                List.copyOf(columns.values())::get, arena, registry, onClose);
+                List.copyOf(columns.values())::get, _ -> { }, arena, registry, onClose);
     }
 
     /// A chunk whose column `i`, of `names`, is decoded by `decoder` the first time it is read.
     ///
     /// @param positions each of `names` to its index, shared by the chunks of one scan
-    /// @param arena     the arena to release on close, or `null` for a slice of another chunk
+    /// @param prefetcher starts fetching what the columns at these positions will need, all in one call so a remote file can merge their requests; does nothing for a local file
+    /// @param arena      the arena to release on close, or `null` for a slice of another chunk
     Chunk(long rowCount, List<ColumnName> names, Map<ColumnName, Integer> positions, IntFunction<Column> decoder,
-          Arena arena, ReadRegistry registry, Consumer<Chunk> onClose) {
+          Consumer<List<Integer>> prefetcher, Arena arena, ReadRegistry registry, Consumer<Chunk> onClose) {
         this.rowCount = rowCount;
         this.names = names;
         this.positions = positions;
         this.decoder = Objects.requireNonNull(decoder);
+        this.prefetcher = Objects.requireNonNull(prefetcher);
         this.decoded = new Column[names.size()];
         this.arena = arena;
         this.registry = Objects.requireNonNull(registry);
@@ -106,7 +111,7 @@ public Chunk slice(long offset, long length) {
             Column column = column(i);
             return new Column(ScanIterator.sliceArray(column.array(), offset, length, column.dtype()), column.dtype());
         };
-        return new Chunk(length, names, positions, sliced, null, registry, _ -> { });
+        return new Chunk(length, names, positions, sliced, prefetcher, null, registry, _ -> { });
     }
 
     /// The rows `rows` keeps, of the columns `names`: the chunk a filtered scan returns.
@@ -123,7 +128,9 @@ Chunk select(Mask rows, List<ColumnName> names, Map<ColumnName, Integer> positio
             Column column = column(position(names.get(i)));
             return new Column(SelectedArrays.select(column.array(), rows, arena), column.dtype());
         };
-        return new Chunk(rows.trueCount(), names, positions, selected, arena, registry, onClose);
+        Consumer<List<Integer>> selectedPrefetcher = columns -> prefetcher.accept(
+                columns.stream().map(i -> position(names.get(i))).toList());
+        return new Chunk(rows.trueCount(), names, positions, selected, selectedPrefetcher, arena, registry, onClose);
     }
 
     /// One decoded column: its zero-copy [Array] view and the [DType] it was declared with in the
@@ -178,6 +185,20 @@ public <T extends Array> T column(ColumnName name) {
         return (T) column(position(name)).array();
     }
 
+    /// Hints that the named columns are about to be read, so a remote file can fetch them
+    /// concurrently instead of one round trip per column. Reading a column afterwards works as
+    /// before; the hint changes timing only, and is a no-op on a local file.
+    ///
+    /// @param names columns about to be read
+    /// @throws VortexException if a name is not a column of this chunk
+    public void prefetch(Collection<ColumnName> names) {
+        List<Integer> columns = new ArrayList<>(names.size());
+        for (ColumnName name : names) {
+            columns.add(position(name));
+        }
+        prefetcher.accept(columns);
+    }
+
     private int position(ColumnName name) {
         Integer position = positions.get(name);
         if (position == null) {
diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/ScanIterator.java b/reader/src/main/java/io/github/dfa1/vortex/reader/ScanIterator.java
index 306c39fac..1c221c569 100644
--- a/reader/src/main/java/io/github/dfa1/vortex/reader/ScanIterator.java
+++ b/reader/src/main/java/io/github/dfa1/vortex/reader/ScanIterator.java
@@ -488,7 +488,41 @@ private Chunk chunkFor(ChunkSpec spec, long chunkRows) {
         // columns (a sort key, a filter column) skips decoding the rest of the chunk. The decode
         // runs while the chunk is open, before next() can evict the flats it slices.
         return new Chunk(chunkRows, projectedNames, projectedPositions,
-                i -> decodeColumn(spec, i, chunkRows, arena), arena, file.registry(), this::onChunkClosed);
+                i -> decodeColumn(spec, i, chunkRows, arena),
+                columns -> file.prefetch(segmentSpecs(spec, columns)),
+                arena, file.registry(), this::onChunkClosed);
+    }
+
+    /// The segments a decode of `layout` reads, for [VortexHandle#prefetch(Collection)]. A zoned
+    /// layout reads only its data child; the stats table beside it is read by pruning, not decode.
+    /// An out-of-range index is skipped: the decode itself rejects it.
+    private List<SegmentSpec> segmentSpecs(ChunkSpec spec, List<Integer> columns) {
+        List<SegmentSpec> all = file.footer().segmentSpecs();
+        List<SegmentSpec> found = new ArrayList<>();
+        for (int column : columns) {
+            collectSegmentSpecs(spec.columnLayouts()[column], all, found);
+        }
+        return found;
+    }
+
+    private static void collectSegmentSpecs(Layout layout, List<SegmentSpec> all, List<SegmentSpec> into) {
+        for (int index : layout.segments()) {
+            if (index >= 0 && index < all.size()) {
+                into.add(all.get(index));
+            }
+        }
+        if (layout.isZoned() && !layout.children().isEmpty()) {
+            collectSegmentSpecs(layout.children().getFirst(), all, into);
+            return;
+        }
+        // A chunked layout's stats child is read by pruning, never by decode, like a zoned one's.
+        int firstChild = 0;
+        if (layout.isChunked()) {
+            firstChild = chunkedDataStart(layout);
+        }
+        for (int i = firstChild; i < layout.children().size(); i++) {
+            collectSegmentSpecs(layout.children().get(i), all, into);
+        }
     }
 
     private Chunk.Column decodeColumn(ChunkSpec spec, int i, long chunkRows, Arena arena) {
diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/VortexHandle.java b/reader/src/main/java/io/github/dfa1/vortex/reader/VortexHandle.java
index cc0f52c8e..c33062574 100644
--- a/reader/src/main/java/io/github/dfa1/vortex/reader/VortexHandle.java
+++ b/reader/src/main/java/io/github/dfa1/vortex/reader/VortexHandle.java
@@ -8,6 +8,7 @@
 import java.io.Closeable;
 import java.lang.foreign.MemorySegment;
 import java.lang.foreign.SegmentAllocator;
+import java.util.Collection;
 
 /// Common interface for handles to a Vortex file, regardless of storage backend.
 ///
@@ -60,6 +61,15 @@ public interface VortexHandle extends Closeable {
     ///         `spec.offset() + spec.length()`
     MemorySegment rawSegment(SegmentSpec spec);
 
+    /// Hints that `specs` are about to be read with [#rawSegment(SegmentSpec)], so a remote handle
+    /// can fetch them concurrently instead of one round trip each. Does nothing on a
+    /// memory-mapped handle, where a read is already free. Call it from the thread that reads the
+    /// segments, the one that opened the handle.
+    ///
+    /// @param specs the segments about to be read
+    default void prefetch(Collection<SegmentSpec> specs) {
+    }
+
     /// Starts a scan of the file, yielding one [Chunk] at a time. Close the iterator when done.
     ///
     /// @param options column projection, zone-map filter and row limit
diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/VortexHttpReader.java b/reader/src/main/java/io/github/dfa1/vortex/reader/VortexHttpReader.java
index 59d0903e6..b4ed8c69a 100644
--- a/reader/src/main/java/io/github/dfa1/vortex/reader/VortexHttpReader.java
+++ b/reader/src/main/java/io/github/dfa1/vortex/reader/VortexHttpReader.java
@@ -18,6 +18,18 @@
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Semaphore;
 
 /// Handle to a remote Vortex file read via HTTP Range requests.
 ///
@@ -33,6 +45,20 @@ public final class VortexHttpReader implements VortexHandle {
     /// all metadata blobs for typical Vortex files.
     static final MemorySize TAIL_SIZE = MemorySize.ofKiB(65);
 
+    /// Cap on concurrent prefetch requests: enough to hide a ~200 ms round trip across a
+    /// screenful of columns without opening a connection per segment.
+    private static final int MAX_IN_FLIGHT = 8;
+
+    /// Most heap one [#prefetch(Collection)] call queues: a fetched response waits as a `byte[]`
+    /// until its segments are read, unlike the off-heap arena a read copies it into.
+    private static final long MAX_PREFETCH_BYTES = MemorySize.ofMiB(64).bytes();
+
+    /// Rust's `CoalesceConfig::object_storage()`: merge two segments at most 1 MiB apart.
+    private static final long COALESCE_DISTANCE = MemorySize.ofMiB(1).bytes();
+
+    /// Rust's `CoalesceConfig::object_storage()`: a merged request spans at most 16 MiB.
+    private static final long COALESCE_MAX_SIZE = MemorySize.ofMiB(16).bytes();
+
     /// Shared across all instances. JDK HttpClient is heavyweight and designed for reuse;
     /// per-reader instantiation would create redundant connection pools and selector threads.
     /// Never closed: lifetime tracks the JVM.
@@ -57,6 +83,16 @@ public final class VortexHttpReader implements VortexHandle {
     private final Layout layout;
     private final ReadRegistry registry;
     private final LayoutRegistry layoutRegistry;
+    /// Range requests started by [#prefetch(Collection)] and not yet consumed by [#rawSegment].
+    private final Map<SegmentSpec, Pending> prefetched;
+    /// Every segment already read, so reading it again (scrolling back) costs no request. Never
+    /// evicted and never expired: the file is immutable for the life of this handle, and the bytes
+    /// live in [#arena] until close anyway, so keeping the reference adds no memory. Bounded by the
+    /// file size. Touched only from the arena's owning thread, like [#rawSegment].
+    private final Map<SegmentSpec, MemorySegment> fetched;
+    private final Semaphore inFlight;
+    private final ExecutorService fetcher;
+    private boolean closed;
 
     private VortexHttpReader(
         URI uri, HttpClient client, long fileSize,
@@ -72,6 +108,11 @@ private VortexHttpReader(
         this.layout = layout;
         this.registry = registry;
         this.layoutRegistry = layoutRegistry;
+        this.prefetched = new ConcurrentHashMap<>();
+        this.fetched = new HashMap<>();
+        this.closed = false;
+        this.inFlight = new Semaphore(MAX_IN_FLIGHT);
+        this.fetcher = Executors.newVirtualThreadPerTaskExecutor();
     }
 
     /// Opens a remote Vortex file with the default encoding registry and HTTP client.
@@ -405,18 +446,128 @@ public io.github.dfa1.vortex.reader.array.Array decodeSegment(
     /// @return a read-only [MemorySegment] containing the fetched bytes
     @Override
     public MemorySegment rawSegment(SegmentSpec spec) {
+        MemorySegment cached = fetched.get(spec);
+        if (cached != null) {
+            return cached;
+        }
         long offset = spec.offset();
         long length = spec.length();
         byte[] bytes;
+        long from = 0;
         try {
-            bytes = fetchRange(uri, offset, offset + length - 1, client);
-        } catch (IOException e) {
+            Pending started = prefetched.remove(spec);
+            if (started != null) {
+                bytes = started.bytes().get();
+                from = started.from();
+            } else {
+                bytes = fetchRange(uri, offset, offset + length - 1, client);
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new VortexException("interrupted fetching [%d, %d) from %s".formatted(offset, offset + length, uri));
+        } catch (ExecutionException | IOException e) {
+            Throwable cause = e;
+            if (e instanceof ExecutionException && e.getCause() != null) {
+                cause = e.getCause();
+            }
+            if (cause instanceof VortexException vortex) {
+                throw vortex;
+            }
             throw new VortexException(
-                "failed to fetch [%d, %d) from %s: %s".formatted(offset, offset + length, uri, e.getMessage()));
+                "failed to fetch [%d, %d) from %s: %s".formatted(offset, offset + length, uri, cause.getMessage()));
         }
         MemorySegment seg = arena.allocate(length);
-        MemorySegment.copy(MemorySegment.ofArray(bytes), 0, seg, 0, length);
-        return seg.asReadOnly();
+        MemorySegment.copy(MemorySegment.ofArray(bytes), from, seg, 0, length);
+        MemorySegment readOnly = seg.asReadOnly();
+        fetched.put(spec, readOnly);
+        return readOnly;
+    }
+
+    /// Starts the range requests for `specs` on virtual threads, at most [#MAX_IN_FLIGHT] at a
+    /// time, so the [#rawSegment(SegmentSpec)] calls that follow find their bytes already on the
+    /// way instead of paying one round trip each. Allocation still happens on the calling
+    /// thread, in the reader's confined arena.
+    ///
+    /// Segments close together in the file share one request, as Rust's object-storage
+    /// `CoalesceConfig` does: at most [#COALESCE_DISTANCE] of unwanted bytes between two, and at
+    /// most [#COALESCE_MAX_SIZE] per request. A round trip costs far more than the bytes between
+    /// neighboring segments, which are mostly alignment padding.
+    ///
+    /// A hint only: a call queues at most [#MAX_PREFETCH_BYTES] on the heap, and the segments past
+    /// that are left to be read on demand. Call it from the thread that opened this handle, the
+    /// one that calls [#rawSegment(SegmentSpec)]; it does nothing once the handle is closed.
+    ///
+    /// @param specs the segments about to be read
+    @Override
+    public void prefetch(Collection<SegmentSpec> specs) {
+        if (closed) {
+            return;
+        }
+        List<SegmentSpec> wanted = specs.stream()
+                .filter(s -> !fetched.containsKey(s) && !prefetched.containsKey(s))
+                .toList();
+        long queued = 0;
+        for (List<SegmentSpec> group : coalesce(wanted)) {
+            long start = group.getFirst().offset();
+            long end = start;
+            for (SegmentSpec spec : group) {
+                end = Math.max(end, spec.offset() + spec.length());
+            }
+            if (queued > 0 && queued + (end - start) > MAX_PREFETCH_BYTES) {
+                break;
+            }
+            queued += end - start;
+            fetchGroup(group, start, end);
+        }
+    }
+
+    /// Groups `specs` into the requests that fetch them: sorted by offset, de-duplicated, with
+    /// zero-length segments dropped, and each group at most [#COALESCE_DISTANCE] of gap between
+    /// neighbors and [#COALESCE_MAX_SIZE] long. Package-private for the boundary tests.
+    static List<List<SegmentSpec>> coalesce(Collection<SegmentSpec> specs) {
+        List<SegmentSpec> sorted = specs.stream()
+                .filter(s -> s.length() > 0)
+                .distinct()
+                .sorted(Comparator.comparingLong(SegmentSpec::offset))
+                .toList();
+        List<List<SegmentSpec>> groups = new ArrayList<>();
+        int first = 0;
+        while (first < sorted.size()) {
+            long start = sorted.get(first).offset();
+            long end = start + sorted.get(first).length();
+            int next = first + 1;
+            while (next < sorted.size()) {
+                SegmentSpec candidate = sorted.get(next);
+                long candidateEnd = Math.max(end, candidate.offset() + candidate.length());
+                if (candidate.offset() - end > COALESCE_DISTANCE || candidateEnd - start > COALESCE_MAX_SIZE) {
+                    break;
+                }
+                end = candidateEnd;
+                next++;
+            }
+            groups.add(sorted.subList(first, next));
+            first = next;
+        }
+        return groups;
+    }
+
+    /// One request for `[start, end)`; each of `group` reads its own slice of the response, which is
+    /// held on the heap until the last of them is read.
+    private void fetchGroup(List<SegmentSpec> group, long start, long end) {
+        CompletableFuture<byte[]> whole = CompletableFuture.supplyAsync(() -> {
+            inFlight.acquireUninterruptibly();
+            try {
+                return fetchRange(uri, start, end - 1, client);
+            } catch (IOException e) {
+                throw new VortexException("failed to fetch [%d, %d) from %s: %s"
+                        .formatted(start, end, uri, e.getMessage()));
+            } finally {
+                inFlight.release();
+            }
+        }, fetcher);
+        for (SegmentSpec spec : group) {
+            prefetched.put(spec, new Pending(whole, spec.offset() - start));
+        }
     }
 
     @Override
@@ -436,7 +587,17 @@ public LayoutRegistry layoutRegistry() {
 
     @Override
     public void close() {
+        // The arena first: a close from the wrong thread throws here, before anything is torn down.
+        // The fetcher tasks only produce heap arrays, they never touch the arena.
         arena.close();
+        closed = true;
+        fetcher.shutdownNow();
+        prefetched.clear();
+        fetched.clear();
+    }
+
+    /// A started range request and where one segment's bytes begin in its response.
+    private record Pending(CompletableFuture<byte[]> bytes, long from) {
     }
 
     @SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared.
diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/ChunkLazyDecodeTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/ChunkLazyDecodeTest.java
index a0269af6d..1ec77dd38 100644
--- a/reader/src/test/java/io/github/dfa1/vortex/reader/ChunkLazyDecodeTest.java
+++ b/reader/src/test/java/io/github/dfa1/vortex/reader/ChunkLazyDecodeTest.java
@@ -28,6 +28,33 @@ class ChunkLazyDecodeTest {
 
     private final Arena arena = Arena.ofAuto();
     private final List<Integer> decodes = new ArrayList<>();
+    private final List<List<Integer>> prefetches = new ArrayList<>();
+
+    @Test
+    void prefetch_hintsTheColumnsWithoutDecodingAny() {
+        // Given
+        Chunk sut = chunk();
+
+        // When
+        sut.prefetch(List.of(B, A));
+
+        // Then
+        assertThat(prefetches).containsExactly(List.of(1, 0));
+        assertThat(decodes).isEmpty();
+    }
+
+    @Test
+    void prefetch_unknownName_throws() {
+        // Given
+        Chunk sut = chunk();
+        List<ColumnName> unknown = List.of(ColumnName.of("c"));
+
+        // When / Then
+        assertThatThrownBy(() -> sut.prefetch(unknown))
+                .isInstanceOf(VortexException.class)
+                .hasMessageContaining("unknown column");
+        assertThat(prefetches).isEmpty();
+    }
 
     @Test
     void column_decodesOnlyTheColumnRead_once() {
@@ -106,6 +133,6 @@ private Chunk chunk() {
                 segment.setAtIndex(VortexFormat.LE_LONG, r, i * 100L + r);
             }
             return new Chunk.Column(new MaterializedLongArray(DTypes.I64, ROWS, segment), DTypes.I64);
-        }, Arena.ofConfined(), ReadRegistry.loadAll(), _ -> { });
+        }, prefetches::add, Arena.ofConfined(), ReadRegistry.loadAll(), _ -> { });
     }
 }
diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/VortexHttpReaderCoalesceTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/VortexHttpReaderCoalesceTest.java
new file mode 100644
index 000000000..776da3041
--- /dev/null
+++ b/reader/src/test/java/io/github/dfa1/vortex/reader/VortexHttpReaderCoalesceTest.java
@@ -0,0 +1,95 @@
+package io.github.dfa1.vortex.reader;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/// How [VortexHttpReader#coalesce] groups the segments of one prefetch into requests: neighbors
+/// within Rust's `CoalesceConfig::object_storage()` distance (1 MiB) and size (16 MiB) share a
+/// request, so a screenful of columns costs about one round trip.
+class VortexHttpReaderCoalesceTest {
+
+    private static final long MIB = 1024L * 1024L;
+
+    @Test
+    void coalesce_gapOfExactlyTheDistance_merges() {
+        // Given
+        SegmentSpec first = spec(0, 100);
+        SegmentSpec second = spec(100 + MIB, 100);
+
+        // When
+        var result = VortexHttpReader.coalesce(List.of(first, second));
+
+        // Then
+        assertThat(result).containsExactly(List.of(first, second));
+    }
+
+    @Test
+    void coalesce_gapOneByteOverTheDistance_splits() {
+        // Given
+        SegmentSpec first = spec(0, 100);
+        SegmentSpec second = spec(100 + MIB + 1, 100);
+
+        // When
+        var result = VortexHttpReader.coalesce(List.of(first, second));
+
+        // Then
+        assertThat(result).containsExactly(List.of(first), List.of(second));
+    }
+
+    @Test
+    void coalesce_spanOfExactlyTheMaximum_staysOneRequest() {
+        // Given — two segments filling exactly 16 MiB
+        SegmentSpec first = spec(0, 8 * MIB);
+        SegmentSpec second = spec(8 * MIB, 8 * MIB);
+
+        // When
+        var result = VortexHttpReader.coalesce(List.of(first, second));
+
+        // Then
+        assertThat(result).containsExactly(List.of(first, second));
+    }
+
+    @Test
+    void coalesce_spanOneByteOverTheMaximum_splits() {
+        // Given
+        SegmentSpec first = spec(0, 8 * MIB);
+        SegmentSpec second = spec(8 * MIB, 8 * MIB + 1);
+
+        // When
+        var result = VortexHttpReader.coalesce(List.of(first, second));
+
+        // Then
+        assertThat(result).containsExactly(List.of(first), List.of(second));
+    }
+
+    @Test
+    void coalesce_unsortedDuplicateOverlappingAndEmptySpecs_areNormalized() {
+        // Given — out of order, a duplicate, a segment inside another, and a zero-length one
+        SegmentSpec outer = spec(100, 400);
+        SegmentSpec inner = spec(200, 50);
+        SegmentSpec last = spec(500, 10);
+        SegmentSpec empty = spec(700, 0);
+
+        // When
+        var result = VortexHttpReader.coalesce(List.of(last, inner, outer, outer, empty));
+
+        // Then
+        assertThat(result).containsExactly(List.of(outer, inner, last));
+    }
+
+    @Test
+    void coalesce_noSpecs_isEmpty() {
+        // Given / When
+        var result = VortexHttpReader.coalesce(List.of());
+
+        // Then
+        assertThat(result).isEmpty();
+    }
+
+    private static SegmentSpec spec(long offset, long length) {
+        return new SegmentSpec(offset, length, (byte) 0, CompressionScheme.NONE);
+    }
+}
diff --git a/writer/src/test/java/io/github/dfa1/vortex/writer/WriterHttpPrefetchTest.java b/writer/src/test/java/io/github/dfa1/vortex/writer/WriterHttpPrefetchTest.java
new file mode 100644
index 000000000..158cfa471
--- /dev/null
+++ b/writer/src/test/java/io/github/dfa1/vortex/writer/WriterHttpPrefetchTest.java
@@ -0,0 +1,310 @@
+package io.github.dfa1.vortex.writer;
+
+import io.github.dfa1.vortex.core.error.VortexException;
+import io.github.dfa1.vortex.core.model.ColumnName;
+import io.github.dfa1.vortex.core.model.DType;
+import io.github.dfa1.vortex.core.model.DTypes;
+import io.github.dfa1.vortex.reader.ReadRegistry;
+import io.github.dfa1.vortex.reader.ScanOptions;
+import io.github.dfa1.vortex.reader.VortexHttpReader;
+import io.github.dfa1.vortex.reader.array.LongArray;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpHeaders;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.IntStream;
+import javax.net.ssl.SSLSession;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+
+/// #744: reading the columns of a chunk over HTTP costs one round trip per column. `Chunk#prefetch`
+/// starts those range requests together, so a screenful of columns waits for one round trip
+/// instead of one each.
+///
+/// Lives here because building the fixture needs [VortexWriter].
+@ExtendWith(MockitoExtension.class)
+class WriterHttpPrefetchTest {
+
+    private static final URI FILE_URI = URI.create("http://example.com/prefetch.vortex");
+    private static final int COLUMNS = 6;
+
+    @Mock
+    private HttpClient client;
+
+    @Test
+    void prefetch_fetchesTheColumnsConcurrently_andDecodesTheSameValues(@TempDir Path tmp) throws Exception {
+        // Given a one-chunk file of 6 I64 columns, each its own 4 MB segment, so the 24 MB in all
+        // exceeds one 16 MiB coalesced request and needs two. Each request waits at a barrier for
+        // the other, so the test passes only if both are in flight at once.
+        List<ColumnName> names = columnNames();
+        byte[] bytes = writeFile(tmp, names, 500_000);
+        CyclicBarrier bothInFlight = new CyclicBarrier(2);
+        AtomicBoolean armed = new AtomicBoolean();
+        given(client.send(any(), any())).willAnswer(inv -> {
+            if (armed.get()) {
+                bothInFlight.await(10, TimeUnit.SECONDS);
+            }
+            return rangeResponse(inv.getArgument(0), bytes);
+        });
+
+        try (var reader = VortexHttpReader.open(FILE_URI, ReadRegistry.loadAll(), client);
+             var iter = reader.scan(ScanOptions.all());
+             var chunk = iter.next()) {
+            armed.set(true);
+
+            // When
+            chunk.prefetch(names);
+            Map<ColumnName, long[]> result = new LinkedHashMap<>();
+            for (ColumnName name : names) {
+                LongArray column = chunk.column(name);
+                result.put(name, new long[]{column.getLong(0), column.getLong(1), column.getLong(2)});
+            }
+
+            // Then
+            assertThat(bothInFlight.getNumberWaiting()).isZero();
+            for (int c = 0; c < COLUMNS; c++) {
+                assertThat(result.get(names.get(c))).containsExactly(c * 100L, c * 100L + 1, c * 100L + 2);
+            }
+        }
+    }
+
+    @Test
+    void prefetch_failedRequest_failsEachColumnOfTheGroupThenRefetches(@TempDir Path tmp) throws Exception {
+        // Given — the server answers the coalesced request with a 500
+        List<ColumnName> names = columnNames();
+        byte[] bytes = writeFile(tmp, names);
+        AtomicBoolean failing = new AtomicBoolean();
+        given(client.send(any(), any())).willAnswer(inv -> {
+            if (failing.get()) {
+                return response(500, null, new byte[0]);
+            }
+            return rangeResponse(inv.getArgument(0), bytes);
+        });
+
+        try (var reader = VortexHttpReader.open(FILE_URI, ReadRegistry.loadAll(), client);
+             var iter = reader.scan(ScanOptions.all());
+             var chunk = iter.next()) {
+            failing.set(true);
+            chunk.prefetch(names);
+            ColumnName first = names.getFirst();
+            ColumnName last = names.getLast();
+
+            // When / Then — every member of the group reports the failure as a VortexException
+            assertThatThrownBy(() -> chunk.column(first)).isInstanceOf(VortexException.class);
+            assertThatThrownBy(() -> chunk.column(last)).isInstanceOf(VortexException.class);
+
+            // and once the server recovers, the same column is fetched afresh
+            failing.set(false);
+            LongArray result = chunk.column(first);
+            assertThat(result.getLong(1)).isEqualTo(1L);
+        }
+    }
+
+    @Test
+    void prefetch_neighboringSegments_shareOneRequest(@TempDir Path tmp) throws Exception {
+        // Given — the 6 column segments of a chunk sit next to each other in the file, so one
+        // request covers them all instead of one round trip each
+        List<ColumnName> names = columnNames();
+        byte[] bytes = writeFile(tmp, names);
+        AtomicInteger requests = new AtomicInteger();
+        given(client.send(any(), any())).willAnswer(inv -> {
+            requests.incrementAndGet();
+            return rangeResponse(inv.getArgument(0), bytes);
+        });
+
+        try (var reader = VortexHttpReader.open(FILE_URI, ReadRegistry.loadAll(), client);
+             var iter = reader.scan(ScanOptions.all());
+             var chunk = iter.next()) {
+            requests.set(0);
+
+            // When
+            chunk.prefetch(names);
+            Map<ColumnName, long[]> result = new LinkedHashMap<>();
+            for (ColumnName name : names) {
+                LongArray column = chunk.column(name);
+                result.put(name, new long[]{column.getLong(0), column.getLong(1), column.getLong(2)});
+            }
+
+            // Then — every column still gets its own bytes out of the shared response
+            assertThat(requests.get()).isEqualTo(1);
+            for (int c = 0; c < COLUMNS; c++) {
+                assertThat(result.get(names.get(c))).containsExactly(c * 100L, c * 100L + 1, c * 100L + 2);
+            }
+        }
+    }
+
+    @Test
+    void rereadingAColumn_isServedFromMemory(@TempDir Path tmp) throws Exception {
+        // Given — a column read once, as when the viewer scrolls forward and then back
+        List<ColumnName> names = columnNames();
+        byte[] bytes = writeFile(tmp, names);
+        AtomicInteger requests = new AtomicInteger();
+        given(client.send(any(), any())).willAnswer(inv -> {
+            requests.incrementAndGet();
+            return rangeResponse(inv.getArgument(0), bytes);
+        });
+
+        try (var reader = VortexHttpReader.open(FILE_URI, ReadRegistry.loadAll(), client)) {
+            try (var iter = reader.scan(ScanOptions.all()); var chunk = iter.next()) {
+                chunk.prefetch(names);
+                chunk.column(names.get(2));
+            }
+            requests.set(0);
+
+            // When — a fresh scan reads the same column again
+            long result;
+            try (var iter = reader.scan(ScanOptions.all()); var chunk = iter.next()) {
+                LongArray column = chunk.column(names.get(2));
+                result = column.getLong(1);
+            }
+
+            // Then
+            assertThat(requests.get()).isZero();
+            assertThat(result).isEqualTo(201L);
+        }
+    }
+
+    @Test
+    void withoutPrefetch_columnsAreFetchedOneAtATime(@TempDir Path tmp) throws Exception {
+        // Given
+        List<ColumnName> names = columnNames();
+        byte[] bytes = writeFile(tmp, names);
+        AtomicInteger running = new AtomicInteger();
+        AtomicInteger peak = new AtomicInteger();
+        given(client.send(any(), any())).willAnswer(inv -> {
+            peak.accumulateAndGet(running.incrementAndGet(), Math::max);
+            try {
+                return rangeResponse(inv.getArgument(0), bytes);
+            } finally {
+                running.decrementAndGet();
+            }
+        });
+
+        try (var reader = VortexHttpReader.open(FILE_URI, ReadRegistry.loadAll(), client);
+             var iter = reader.scan(ScanOptions.all());
+             var chunk = iter.next()) {
+            peak.set(0);
+
+            // When
+            for (ColumnName name : names) {
+                chunk.column(name);
+            }
+
+            // Then
+            assertThat(peak.get()).isEqualTo(1);
+        }
+    }
+
+    private static List<ColumnName> columnNames() {
+        return IntStream.range(0, COLUMNS).mapToObj(c -> ColumnName.of("c" + c)).toList();
+    }
+
+    private static byte[] writeFile(Path tmp, List<ColumnName> names) throws Exception {
+        return writeFile(tmp, names, 3);
+    }
+
+    /// Column `c` holds `c * 100 + row`.
+    private static byte[] writeFile(Path tmp, List<ColumnName> names, int rows) throws Exception {
+        DType.Struct schema = new DType.Struct(names, names.stream().<DType>map(_ -> DTypes.I64).toList(), false);
+        WriteOptions opts = WriteOptions.cascading(0).withZoneMaps(false).withGlobalDict(false).withoutEditions();
+        Path file = tmp.resolve("prefetch.vtx");
+        Map<ColumnName, Object> data = new LinkedHashMap<>();
+        for (int c = 0; c < names.size(); c++) {
+            long[] values = new long[rows];
+            for (int r = 0; r < rows; r++) {
+                values[r] = c * 100L + r;
+            }
+            data.put(names.get(c), values);
+        }
+        try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+             var writer = VortexWriter.create(ch, schema, opts)) {
+            writer.writeChunk(data);
+        }
+        return Files.readAllBytes(file);
+    }
+
+    private static HttpResponse<byte[]> rangeResponse(HttpRequest req, byte[] file) {
+        String rangeHeader = req.headers().firstValue("Range").orElseThrow();
+        String[] parts = rangeHeader.substring("bytes=".length()).split("-", 2);
+        int from = Integer.parseInt(parts[0].isEmpty() ? "0" : parts[0]);
+        int to = Math.min(Integer.parseInt(parts[1].isEmpty() ? "0" : parts[1]), file.length - 1);
+        if (parts[0].isEmpty()) {
+            // suffix range "bytes=-N": the last N bytes
+            from = Math.max(0, file.length - Integer.parseInt(parts[1]));
+            to = file.length - 1;
+        }
+        byte[] body = Arrays.copyOfRange(file, from, to + 1);
+        return response(206, "bytes " + from + "-" + to + "/" + file.length, body);
+    }
+
+    private static HttpResponse<byte[]> response(int status, String contentRange, byte[] body) {
+        return new HttpResponse<>() {
+            @Override
+            public int statusCode() {
+                return status;
+            }
+
+            @Override
+            public byte[] body() {
+                return body;
+            }
+
+            @Override
+            public HttpHeaders headers() {
+                Map<String, List<String>> headers = Map.of();
+                if (contentRange != null) {
+                    headers = Map.of("content-range", List.of(contentRange));
+                }
+                return HttpHeaders.of(headers, (k, v) -> true);
+            }
+
+            @Override
+            public HttpRequest request() {
+                return null;
+            }
+
+            @Override
+            public Optional<HttpResponse<byte[]>> previousResponse() {
+                return Optional.empty();
+            }
+
+            @Override
+            public Optional<SSLSession> sslSession() {
+                return Optional.empty();
+            }
+
+            @Override
+            public URI uri() {
+                return FILE_URI;
+            }
+
+            @Override
+            public HttpClient.Version version() {
+                return HttpClient.Version.HTTP_1_1;
+            }
+        };
+    }
+}
diff --git a/writer/src/test/java/io/github/dfa1/vortex/writer/WriterHttpSkipChunkTest.java b/writer/src/test/java/io/github/dfa1/vortex/writer/WriterHttpSkipChunkTest.java
new file mode 100644
index 000000000..1250a5a71
--- /dev/null
+++ b/writer/src/test/java/io/github/dfa1/vortex/writer/WriterHttpSkipChunkTest.java
@@ -0,0 +1,142 @@
+package io.github.dfa1.vortex.writer;
+
+import io.github.dfa1.vortex.core.model.ColumnName;
+import io.github.dfa1.vortex.core.model.DType;
+import io.github.dfa1.vortex.core.model.DTypes;
+import io.github.dfa1.vortex.reader.ReadRegistry;
+import io.github.dfa1.vortex.reader.ScanOptions;
+import io.github.dfa1.vortex.reader.VortexHttpReader;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpHeaders;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.net.ssl.SSLSession;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+
+/// Walking past a chunk with `ScanIterator#next()` fetches nothing over HTTP, because a chunk decodes
+/// a column only when it is read. The viewer relies on it to reach a far chunk (End) by stepping
+/// over the ones in between, so a change that decoded eagerly would make every such jump download
+/// the whole file (#744).
+@ExtendWith(MockitoExtension.class)
+class WriterHttpSkipChunkTest {
+
+    private static final URI FILE_URI = URI.create("http://example.com/skip.vortex");
+    private static final ColumnName V = ColumnName.of("v");
+
+    @Mock
+    private HttpClient client;
+
+    @Test
+    void skippingChunksWithoutReadingAColumn_fetchesNothing(@TempDir Path tmp) throws Exception {
+        // Given — several chunks with zone maps, as a real file has. The writer repartitions batches
+        // into chunks of about 1 MiB, so 4 batches of 300k rows give around 9.
+        DType.Struct schema = new DType.Struct(List.of(V), List.<DType>of(DTypes.I64), false);
+        WriteOptions opts = WriteOptions.cascading(0).withZoneMaps(true).withGlobalDict(false).withoutEditions();
+        Path file = tmp.resolve("skip.vtx");
+        try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+             var writer = VortexWriter.create(ch, schema, opts)) {
+            for (int chunk = 0; chunk < 4; chunk++) {
+                long[] values = new long[300_000];
+                for (int r = 0; r < values.length; r++) {
+                    values[r] = (long) chunk * values.length + r;
+                }
+                writer.writeChunk(Map.of(V, values));
+            }
+        }
+        byte[] bytes = Files.readAllBytes(file);
+        AtomicInteger requests = new AtomicInteger();
+        given(client.send(any(), any())).willAnswer(inv -> {
+            requests.incrementAndGet();
+            return rangeResponse(inv.getArgument(0), bytes);
+        });
+
+        try (var reader = VortexHttpReader.open(FILE_URI, ReadRegistry.loadAll(), client);
+             var iter = reader.scan(ScanOptions.all())) {
+            int afterOpen = requests.get();
+
+            // When — walk past every chunk without reading a column
+            int result = 0;
+            while (iter.hasNext()) {
+                try (var _ = iter.next()) {
+                    result++;
+                }
+            }
+
+            // Then — a chunk decodes its columns when read, so passing one costs no request
+            assertThat(result).isGreaterThan(1);
+            assertThat(requests.get() - afterOpen).isZero();
+        }
+    }
+
+    private static HttpResponse<byte[]> rangeResponse(HttpRequest req, byte[] file) {
+        String[] parts = req.headers().firstValue("Range").orElseThrow().substring("bytes=".length()).split("-", 2);
+        int from = Integer.parseInt(parts[0].isEmpty() ? "0" : parts[0]);
+        int to = Math.min(Integer.parseInt(parts[1].isEmpty() ? "0" : parts[1]), file.length - 1);
+        if (parts[0].isEmpty()) {
+            from = Math.max(0, file.length - Integer.parseInt(parts[1]));
+            to = file.length - 1;
+        }
+        byte[] body = Arrays.copyOfRange(file, from, to + 1);
+        String contentRange = "bytes " + from + "-" + to + "/" + file.length;
+        return new HttpResponse<>() {
+            @Override
+            public int statusCode() {
+                return 206;
+            }
+
+            @Override
+            public byte[] body() {
+                return body;
+            }
+
+            @Override
+            public HttpHeaders headers() {
+                return HttpHeaders.of(Map.of("content-range", List.of(contentRange)), (k, v) -> true);
+            }
+
+            @Override
+            public HttpRequest request() {
+                return null;
+            }
+
+            @Override
+            public Optional<HttpResponse<byte[]>> previousResponse() {
+                return Optional.empty();
+            }
+
+            @Override
+            public Optional<SSLSession> sslSession() {
+                return Optional.empty();
+            }
+
+            @Override
+            public URI uri() {
+                return FILE_URI;
+            }
+
+            @Override
+            public HttpClient.Version version() {
+                return HttpClient.Version.HTTP_1_1;
+            }
+        };
+    }
+}