diff --git a/ccrawl/client.go b/ccrawl/client.go
index 4f12b87..a2c3ad8 100644
--- a/ccrawl/client.go
+++ b/ccrawl/client.go
@@ -49,9 +49,10 @@ func NewHTTPClient(cfg Config) *HTTPClient {
 	if src == "" {
 		src = SourceHTTPS
 	}
+	tr := pooledTransport()
 	return &HTTPClient{
-		c:          &http.Client{Timeout: cfg.Timeout},
-		download:   &http.Client{},
+		c:          &http.Client{Timeout: cfg.Timeout, Transport: tr},
+		download:   &http.Client{Transport: tr},
 		retries:    max(cfg.Retries, 0),
 		delay:      cfg.Delay,
 		backoff:    backoff,
@@ -62,6 +63,42 @@ func NewHTTPClient(cfg Config) *HTTPClient {
 	}
 }
 
+// pooledTransport returns a transport that keeps enough idle connections for a
+// worker pool. Go's default is two per host, which is fine for a browser and
+// wrong here: every command that fans out talks to one host, so past two
+// workers the rest pay a fresh TLS handshake on every request.
+func pooledTransport() *http.Transport {
+	tr := http.DefaultTransport.(*http.Transport).Clone()
+	tr.MaxIdleConns = 256
+	tr.MaxIdleConnsPerHost = 64
+	tr.IdleConnTimeout = 90 * time.Second
+	return tr
+}
+
+// WithoutDelay returns a copy of h that skips the inter-request delay, sharing
+// the same connection pool, credentials and retry policy.
+//
+// The delay is priced for requests that mean something: a CDX page, a WARC
+// range, a gigabyte file. A columnar scan is thousands of few-kilobyte reads of
+// footers and column indexes, and spacing those two hundred milliseconds apart
+// turns a thirty second query into two minutes. duckdb, which is the default
+// engine for the same queries, applies no delay at all, so throttling the
+// built-in engine only makes the dependency-free path the slow one.
+func (h *HTTPClient) WithoutDelay() *HTTPClient {
+	// Field by field rather than a struct copy, because the mutex must not be
+	// carried over.
+	return &HTTPClient{
+		c:          h.c,
+		download:   h.download,
+		retries:    h.retries,
+		backoff:    h.backoff,
+		backoffMax: h.backoffMax,
+		userAgent:  h.userAgent,
+		source:     h.source,
+		creds:      h.creds,
+	}
+}
+
 // throttle blocks until the configured minimum inter-request delay has elapsed.
 func (h *HTTPClient) throttle(ctx context.Context) error {
 	if h.delay <= 0 {
diff --git a/ccrawl/columnar_native.go b/ccrawl/columnar_native.go
new file mode 100644
index 0000000..3f4b36c
--- /dev/null
+++ b/ccrawl/columnar_native.go
@@ -0,0 +1,792 @@
+package ccrawl
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"sort"
+
+	"strings"
+	"sync"
+
+	"github.com/parquet-go/parquet-go"
+)
+
+// The native columnar engine answers the projection and aggregation subset of
+// the columnar commands without duckdb, by reading the Common Crawl Parquet
+// files directly over ranged HTTP.
+//
+// It is the same trick the URL publisher already uses in production: open the
+// remote file through httpReaderAt, read the footer, and pull only the column
+// chunks the query needs. On top of that it prunes whole row groups whose
+// statistics say they cannot match, so a filter on url_host_tld or
+// url_surtkey touches a fraction of a 530 MiB shard.
+//
+// What it does not do is SQL. `columnar query` stays duckdb only, permanently:
+// a hand written engine that accepts arbitrary SQL is a database, and this is
+// a CLI.
+
+// NativeAggregate selects what a native scan produces.
+type NativeAggregate int
+
+const (
+	// NativeRows emits one record per matching row, projecting Select.
+	NativeRows NativeAggregate = iota
+	// NativeCount emits a single row holding the number of matches.
+	NativeCount
+	// NativeGroupCount emits one row per distinct value of GroupBy, with its
+	// count, ordered by count descending.
+	NativeGroupCount
+)
+
+// NativeScan is one native query: which files to read, what to keep, and what
+// to produce.
+type NativeScan struct {
+	URLs      []string // parquet part URLs, from the crawl's manifest
+	Query     ColumnarQuery
+	Aggregate NativeAggregate
+	Select    []string // output columns, for NativeRows
+	GroupBy   string   // grouping column, for NativeGroupCount
+	Limit     int      // stop after this many output rows (0 = all)
+	Workers   int      // files read concurrently
+}
+
+// nativeColumns are the columns the engine knows how to filter on. Anything
+// else in a query means the query is not expressible natively and belongs to
+// duckdb.
+var nativeColumns = map[string]bool{
+	"url":                        true,
+	"url_surtkey":                true,
+	"url_host_name":              true,
+	"url_host_tld":               true,
+	"url_host_registered_domain": true,
+	"url_path":                   true,
+	"fetch_status":               true,
+	"content_mime_detected":      true,
+	"content_languages":          true,
+	"warc_filename":              true,
+	"warc_record_offset":         true,
+	"warc_record_length":         true,
+}
+
+// NativeExpressible reports whether the native engine can answer this scan. A
+// false here is not a failure: --engine auto falls back to duckdb, which can.
+func NativeExpressible(s NativeScan) bool {
+	for _, c := range s.Select {
+		if !nativeColumns[c] {
+			return false
+		}
+	}
+	if s.GroupBy != "" && !nativeColumns[s.GroupBy] {
+		return false
+	}
+	return true
+}
+
+// predicate is one filter on one column. The kinds are exactly what the
+// columnar flags produce, which is why there is no expression tree here.
+type predicate struct {
+	col      string
+	equals   string   // exact match on a string column
+	prefixes []string // string column starts with any one of these
+	contains string   // string column contains this
+	status   int32    // fetch_status equality, when col is fetch_status
+	isStatus bool     // this predicate compares the int32 fetch_status column
+}
+
+// match evaluates the predicate against one value. A null never matches, which
+// is what SQL equality does too.
+func (p predicate) match(v parquet.Value) bool {
+	if v.IsNull() {
+		return false
+	}
+	if p.isStatus {
+		return v.Int32() == p.status
+	}
+	s := byteString(v.ByteArray())
+	switch {
+	case p.equals != "":
+		return s == p.equals
+	case len(p.prefixes) > 0:
+		for _, pre := range p.prefixes {
+			if strings.HasPrefix(s, pre) {
+				return true
+			}
+		}
+		return false
+	case p.contains != "":
+		return strings.Contains(s, p.contains)
+	}
+	return true
+}
+
+// canPrune reports whether page bounds can rule this predicate out. Only
+// equality and prefixes are ordered against min and max; a substring match says
+// nothing about where the value sorts.
+func (p predicate) canPrune() bool {
+	return p.equals != "" || len(p.prefixes) > 0 || p.isStatus
+}
+
+// excludes reports whether a page whose values all fall in [min, max] can be
+// skipped entirely. A predicate holding several prefixes only excludes the page
+// when every one of them does, since the row needs to match just one.
+func (p predicate) excludes(minV, maxV parquet.Value) bool {
+	if p.isStatus {
+		return p.status < minV.Int32() || p.status > maxV.Int32()
+	}
+	lo, hi := byteString(minV.ByteArray()), byteString(maxV.ByteArray())
+	if p.equals != "" {
+		return p.equals < lo || p.equals > hi
+	}
+	for _, pre := range p.prefixes {
+		// Every string with this prefix sorts inside [prefix, prefix+0xff...].
+		// If that window overlaps the page's bounds at all then a row in here
+		// might start with it, and the page has to be read. Clipping lo to the
+		// prefix length is what makes the lower comparison right: "com,ex" is
+		// not less than "com,example", but a page starting at "com,example"
+		// holds no "com,ex" either.
+		if pre <= hi && pre >= clipTo(lo, len(pre)) {
+			return false
+		}
+	}
+	return len(p.prefixes) > 0
+}
+
+// clipTo cuts s to at most n bytes, so a prefix can be compared against the
+// same number of leading bytes of a bound.
+func clipTo(s string, n int) string {
+	if len(s) <= n {
+		return s
+	}
+	return s[:n]
+}
+
+// predicatesFor turns the flag-shaped query into the filters the scan applies.
+// The surtkey prefixes that SQL() adds for pruning are added here too, and for
+// the same reason: it is the column the files are sorted on, so it is the one
+// that actually skips row groups.
+func predicatesFor(q ColumnarQuery) []predicate {
+	var ps []predicate
+	if q.Domain != "" {
+		ps = append(ps, predicate{col: "url_host_registered_domain", equals: q.Domain})
+		// Both patterns, exactly as the SQL does: the apex host's captures end
+		// the host with ')' and every subdomain's continue with ','. Keeping
+		// them in one predicate rather than two is what stops the apex rows
+		// from being filtered out by the subdomain form.
+		if rev := surtHostKey(q.Domain); rev != "" {
+			ps = append(ps, predicate{col: "url_surtkey", prefixes: []string{rev + ")", rev + ","}})
+		}
+	}
+	if q.Host != "" {
+		ps = append(ps, predicate{col: "url_host_name", equals: q.Host})
+		if rev := surtHostKey(q.Host); rev != "" {
+			ps = append(ps, predicate{col: "url_surtkey", prefixes: []string{rev + ")"}})
+		}
+	}
+	if q.TLD != "" {
+		ps = append(ps, predicate{col: "url_host_tld", equals: q.TLD})
+	}
+	if q.MIME != "" {
+		ps = append(ps, predicate{col: "content_mime_detected", equals: q.MIME})
+	}
+	if q.Lang != "" {
+		ps = append(ps, predicate{col: "content_languages", contains: q.Lang})
+	}
+	if q.PathPrefix != "" {
+		ps = append(ps, predicate{col: "url_path", prefixes: []string{q.PathPrefix}})
+	}
+	if q.Status != 0 {
+		ps = append(ps, predicate{col: "fetch_status", status: int32(q.Status), isStatus: true})
+	}
+	return ps
+}
+
+// nativeResult is what one file's scan contributes.
+type nativeResult struct {
+	count  int64
+	groups map[string]int64
+}
+
+// RunColumnarNative executes the scan across every file, concurrently, and
+// streams results to emit. Rows arrive in whatever order the files finish,
+// which is what duckdb over an unordered file list does too.
+func RunColumnarNative(ctx context.Context, h *HTTPClient, s NativeScan, emit func(map[string]any) error) error {
+	if len(s.URLs) == 0 {
+		return errors.New("no parquet files to scan")
+	}
+	if s.Workers <= 0 {
+		s.Workers = 8
+	}
+
+	plan, err := newScanPlan(s)
+	if err != nil {
+		return err
+	}
+	// Ranged metadata reads are not what the inter-request delay is priced for,
+	// and leaving it on makes this engine four times slower than the duckdb it
+	// is meant to replace.
+	h = h.WithoutDelay()
+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	var (
+		mu      sync.Mutex
+		total   int64
+		groups  = map[string]int64{}
+		emitted int
+		emitErr error
+		done    bool
+	)
+
+	// sink is called with one matching row's projected values, under the lock,
+	// so emit sees rows one at a time even though files are read in parallel.
+	sink := func(vals []parquet.Value) bool {
+		mu.Lock()
+		defer mu.Unlock()
+		if done {
+			return false
+		}
+		row := make(map[string]any, len(plan.selectCols))
+		for i, c := range plan.selectCols {
+			row[c] = goValue(vals[i])
+		}
+		if err := emit(row); err != nil {
+			emitErr, done = err, true
+			cancel()
+			return false
+		}
+		emitted++
+		if s.Limit > 0 && emitted >= s.Limit {
+			done = true
+			cancel()
+			return false
+		}
+		return true
+	}
+
+	jobs := make(chan string)
+	var wg sync.WaitGroup
+	var firstErr error
+	var errOnce sync.Once
+
+	for i := 0; i < s.Workers && i < len(s.URLs); i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for u := range jobs {
+				res, err := scanFile(ctx, h, u, plan, sink)
+				if err != nil {
+					// A cancelled context is how the limit stops the scan, so it
+					// is not an error the user should see.
+					if ctx.Err() == nil {
+						errOnce.Do(func() { firstErr = fmt.Errorf("%s: %w", u, err) })
+						cancel()
+					}
+					return
+				}
+				mu.Lock()
+				total += res.count
+				for k, v := range res.groups {
+					groups[k] += v
+				}
+				mu.Unlock()
+			}
+		}()
+	}
+	for _, u := range s.URLs {
+		select {
+		case jobs <- u:
+		case <-ctx.Done():
+		}
+		if ctx.Err() != nil {
+			break
+		}
+	}
+	close(jobs)
+	wg.Wait()
+
+	if emitErr != nil {
+		return emitErr
+	}
+	if firstErr != nil {
+		return firstErr
+	}
+
+	switch s.Aggregate {
+	case NativeCount:
+		return emit(map[string]any{"n": total})
+	case NativeGroupCount:
+		return emitGroups(groups, s.Limit, emit)
+	}
+	return nil
+}
+
+// emitGroups writes the breakdown ordered by count descending, then by value,
+// so a tie does not reorder between runs.
+func emitGroups(groups map[string]int64, limit int, emit func(map[string]any) error) error {
+	type kv struct {
+		k string
+		n int64
+	}
+	out := make([]kv, 0, len(groups))
+	for k, n := range groups {
+		out = append(out, kv{k, n})
+	}
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].n != out[j].n {
+			return out[i].n > out[j].n
+		}
+		return out[i].k < out[j].k
+	})
+	if limit > 0 && len(out) > limit {
+		out = out[:limit]
+	}
+	for _, e := range out {
+		row := map[string]any{"n": e.n}
+		if e.k == nullGroup {
+			row["value"] = nil
+		} else {
+			row["value"] = e.k
+		}
+		if err := emit(row); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// nullGroup stands in for a NULL group key, which a Go map cannot hold apart
+// from the empty string. Common Crawl has rows with an empty content_languages
+// and rows with none at all, and duckdb reports those as two groups.
+const nullGroup = "\x00null"
+
+// scanPlan is the per-query work shared by every file: which columns to read
+// for filtering, which to read for output, and in what order.
+type scanPlan struct {
+	preds      []predicate
+	filterCols []string // distinct predicate columns, cheapest first
+	predsBy    map[string][]predicate
+	selectCols []string // columns the output needs, in output order
+	aggregate  NativeAggregate
+	groupBy    string
+}
+
+// filterCost orders the filter columns so the one most likely to eliminate rows
+// for the fewest bytes runs first. url_host_tld is a handful of bytes per row
+// and dictionary encodes to almost nothing; url and url_path are the widest
+// columns in the file.
+var filterCost = map[string]int{
+	"url_host_tld":               0,
+	"fetch_status":               1,
+	"content_mime_detected":      2,
+	"url_host_registered_domain": 3,
+	"url_host_name":              4,
+	"content_languages":          5,
+	"url_surtkey":                6,
+	"url_path":                   7,
+	"url":                        8,
+}
+
+func newScanPlan(s NativeScan) (*scanPlan, error) {
+	p := &scanPlan{
+		preds:     predicatesFor(s.Query),
+		predsBy:   map[string][]predicate{},
+		aggregate: s.Aggregate,
+		groupBy:   s.GroupBy,
+	}
+	for _, pr := range p.preds {
+		if !nativeColumns[pr.col] {
+			return nil, fmt.Errorf("native engine cannot filter on %s", pr.col)
+		}
+		if _, seen := p.predsBy[pr.col]; !seen {
+			p.filterCols = append(p.filterCols, pr.col)
+		}
+		p.predsBy[pr.col] = append(p.predsBy[pr.col], pr)
+	}
+	sort.SliceStable(p.filterCols, func(i, j int) bool {
+		return filterCost[p.filterCols[i]] < filterCost[p.filterCols[j]]
+	})
+
+	switch s.Aggregate {
+	case NativeRows:
+		p.selectCols = s.Select
+		if len(p.selectCols) == 0 {
+			return nil, errors.New("native row scan needs columns to select")
+		}
+	case NativeGroupCount:
+		if s.GroupBy == "" {
+			return nil, errors.New("native group scan needs a column to group by")
+		}
+		p.selectCols = []string{s.GroupBy}
+	}
+	return p, nil
+}
+
+// scanFile reads one Parquet part. It walks row groups, prunes the ones whose
+// statistics rule them out, builds a match mask from the filter columns, and
+// only then reads the output columns for the rows that survived.
+func scanFile(ctx context.Context, h *HTTPClient, url string, plan *scanPlan, sink func([]parquet.Value) bool) (nativeResult, error) {
+	var res nativeResult
+	if plan.aggregate == NativeGroupCount {
+		res.groups = map[string]int64{}
+	}
+
+	pf, err := openRemoteParquet(ctx, h, url)
+	if err != nil {
+		return res, err
+	}
+	schema := pf.Schema()
+
+	// Resolve every column this query touches to its leaf index once, and bail
+	// out if the file's schema is not the flat one Common Crawl publishes.
+	idx := map[string]int{}
+	for _, c := range append(append([]string{}, plan.filterCols...), plan.selectCols...) {
+		if _, ok := idx[c]; ok {
+			continue
+		}
+		leaf, ok := schema.Lookup(c)
+		if !ok {
+			return res, fmt.Errorf("column %s is not in this file", c)
+		}
+		if leaf.MaxRepetitionLevel > 0 {
+			return res, fmt.Errorf("column %s is repeated, which the native engine does not read", c)
+		}
+		idx[c] = leaf.ColumnIndex
+	}
+
+	for _, rg := range pf.RowGroups() {
+		if err := ctx.Err(); err != nil {
+			return res, nil
+		}
+		chunks := rg.ColumnChunks()
+		if pruneRowGroup(plan, idx, chunks) {
+			continue
+		}
+		n := int(rg.NumRows())
+		mask, matched, err := buildMask(plan, idx, chunks, n)
+		if err != nil {
+			return res, err
+		}
+		if matched == 0 {
+			continue
+		}
+		switch plan.aggregate {
+		case NativeCount:
+			res.count += int64(matched)
+			continue
+		case NativeGroupCount:
+			if err := countGroups(chunks[idx[plan.groupBy]], mask, res.groups); err != nil {
+				return res, err
+			}
+			continue
+		}
+		if err := emitRows(plan, idx, chunks, mask, matched, sink); err != nil {
+			return res, err
+		}
+		if ctx.Err() != nil {
+			return res, nil
+		}
+	}
+	return res, nil
+}
+
+// pruneRowGroup reports whether the whole row group can be skipped. It reads
+// only the column index, which lives in the footer, so a pruned group costs no
+// data pages at all.
+func pruneRowGroup(plan *scanPlan, idx map[string]int, chunks []parquet.ColumnChunk) bool {
+	for _, col := range plan.filterCols {
+		for _, pr := range plan.predsBy[col] {
+			if !pr.canPrune() {
+				continue
+			}
+			ci, err := chunks[idx[col]].ColumnIndex()
+			if err != nil || ci.NumPages() == 0 {
+				continue // no statistics, so nothing can be ruled out
+			}
+			if chunkExcludes(pr, ci) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// chunkExcludes reports whether every page in the chunk is ruled out by the
+// predicate. Page level bounds are strictly better than one bound for the whole
+// chunk, and this is where the ordering of url_surtkey pays off.
+func chunkExcludes(pr predicate, ci parquet.ColumnIndex) bool {
+	for i := 0; i < ci.NumPages(); i++ {
+		if ci.NullPage(i) {
+			continue
+		}
+		if !pr.excludes(ci.MinValue(i), ci.MaxValue(i)) {
+			return false
+		}
+	}
+	return true
+}
+
+// buildMask reads the filter columns and returns which rows of the group match
+// every predicate. Columns are read cheapest first and the read stops as soon
+// as no row is left alive, so an expensive column is never touched for a group
+// a cheap one already eliminated.
+func buildMask(plan *scanPlan, idx map[string]int, chunks []parquet.ColumnChunk, rows int) ([]bool, int, error) {
+	mask := make([]bool, rows)
+	for i := range mask {
+		mask[i] = true
+	}
+	matched := rows
+
+	for _, col := range plan.filterCols {
+		preds := plan.predsBy[col]
+		matched = 0
+		err := eachValue(chunks[idx[col]], rows, func(row int, v parquet.Value) {
+			if !mask[row] {
+				return
+			}
+			for _, pr := range preds {
+				if !pr.match(v) {
+					mask[row] = false
+					return
+				}
+			}
+			matched++
+		})
+		if err != nil {
+			return nil, 0, err
+		}
+		if matched == 0 {
+			return mask, 0, nil
+		}
+	}
+	return mask, matched, nil
+}
+
+// countGroups tallies the group column over the masked rows.
+func countGroups(chunk parquet.ColumnChunk, mask []bool, groups map[string]int64) error {
+	return eachValue(chunk, len(mask), func(row int, v parquet.Value) {
+		if !mask[row] {
+			return
+		}
+		if v.IsNull() {
+			groups[nullGroup]++
+			return
+		}
+		groups[string(v.ByteArray())]++
+	})
+}
+
+// emitRows reads the projected columns for the masked rows and hands each row
+// to the sink. The columns are read one at a time and the values held until the
+// group is done, because a Parquet file stores columns apart and there is no
+// way to read a row without doing exactly this.
+func emitRows(plan *scanPlan, idx map[string]int, chunks []parquet.ColumnChunk, mask []bool, matched int, sink func([]parquet.Value) bool) error {
+	cols := make([][]parquet.Value, len(plan.selectCols))
+	for i, c := range plan.selectCols {
+		vals := make([]parquet.Value, 0, matched)
+		err := eachValue(chunks[idx[c]], len(mask), func(row int, v parquet.Value) {
+			if mask[row] {
+				vals = append(vals, v.Clone())
+			}
+		})
+		if err != nil {
+			return err
+		}
+		// Every column of a row group holds the same number of rows, so a short
+		// read here means the walk lost alignment and the row about to be built
+		// would pair one column's value with another column's row. That is a
+		// wrong answer rather than a missing one, so it stops the scan.
+		if len(vals) != matched {
+			return fmt.Errorf("column %s yielded %d of %d masked rows", c, len(vals), matched)
+		}
+		cols[i] = vals
+	}
+	row := make([]parquet.Value, len(cols))
+	for r := 0; r < matched; r++ {
+		for i := range cols {
+			row[i] = cols[i][r]
+		}
+		if !sink(row) {
+			return nil
+		}
+	}
+	return nil
+}
+
+// eachValue walks one column chunk and calls fn once per row, in row order.
+// Common Crawl's columns are flat and optional, so one page value is one row
+// and nulls come through as null values, which keeps the row numbering aligned
+// across columns without materialising anything.
+func eachValue(chunk parquet.ColumnChunk, rows int, fn func(row int, v parquet.Value)) error {
+	pages := chunk.Pages()
+	defer func() { _ = pages.Close() }()
+
+	buf := make([]parquet.Value, 1024)
+	row := 0
+	for row < rows {
+		page, err := pages.ReadPage()
+		if err != nil {
+			if errors.Is(err, io.EOF) {
+				return nil
+			}
+			return err
+		}
+		vr := page.Values()
+		for {
+			n, rerr := vr.ReadValues(buf)
+			for i := 0; i < n && row < rows; i++ {
+				fn(row, buf[i])
+				row++
+			}
+			if rerr != nil {
+				if errors.Is(rerr, io.EOF) {
+					break
+				}
+				parquet.Release(page)
+				return rerr
+			}
+		}
+		parquet.Release(page)
+	}
+	return nil
+}
+
+// goValue converts a Parquet value into what the JSON row map carries, matching
+// what duckdb's JSON output produces for the same column.
+func goValue(v parquet.Value) any {
+	if v.IsNull() {
+		return nil
+	}
+	switch v.Kind() {
+	case parquet.Boolean:
+		return v.Boolean()
+	case parquet.Int32:
+		return int64(v.Int32())
+	case parquet.Int64:
+		return v.Int64()
+	case parquet.Float:
+		return float64(v.Float())
+	case parquet.Double:
+		return v.Double()
+	default:
+		return string(v.ByteArray())
+	}
+}
+
+// byteString reads a byte array column's value as a string, so the predicates
+// read as string comparisons rather than bytes.Compare calls.
+func byteString(b []byte) string { return string(b) }
+
+// parquetBlockSize is the span httpReaderAt fetches per request, and
+// parquetCacheBlocks is how many of them one open file keeps. 256 KiB is the
+// measured middle: small enough that opening a part and pruning it costs about
+// 200 KiB rather than megabytes, large enough that reading a real column chunk
+// is not a string of round trips.
+const (
+	parquetBlockSize   = 256 << 10
+	parquetCacheBlocks = 64
+)
+
+// openRemoteParquet opens a Parquet file over ranged HTTP with the options a
+// remote read actually wants. Every one of these was measured against a real
+// Common Crawl part, and together they take the cost of opening one and pruning
+// it from 4.9 MiB down to about 200 KiB:
+//
+//   - the magic bytes at offset 0 are a whole extra block fetched to look at
+//     four bytes, on a file we already know is Parquet;
+//   - the bloom filter headers are never consulted here;
+//   - the eager page index read pulls the index for all thirty odd columns,
+//     where a query touches two or three. Skipping it does not turn pruning
+//     off, because ColumnIndex() then reads that one column's index on demand,
+//     which is exactly the part worth paying for;
+//   - the optimistic read speculatively pulls a megabyte and a half of tail,
+//     which is the single largest cost and buys nothing once the block cache
+//     is doing the same job.
+func openRemoteParquet(ctx context.Context, h *HTTPClient, url string) (*parquet.File, error) {
+	size, err := h.ContentLength(ctx, url)
+	if err != nil {
+		return nil, err
+	}
+	ra := newHTTPReaderAt(ctx, h, url, size, parquetBlockSize, parquetCacheBlocks)
+	return parquet.OpenFile(ra, size,
+		parquet.SkipMagicBytes(true),
+		parquet.SkipBloomFilters(true),
+		parquet.SkipPageIndex(true),
+		parquet.OptimisticRead(false),
+	)
+}
+
+// NativeSchema reads one part's footer and returns the column names and their
+// types, spelled the way duckdb's DESCRIBE spells them so both engines print
+// the same table.
+func NativeSchema(ctx context.Context, h *HTTPClient, url string) ([][2]string, error) {
+	pf, err := openRemoteParquet(ctx, h, url)
+	if err != nil {
+		return nil, err
+	}
+	var out [][2]string
+	for _, f := range pf.Schema().Fields() {
+		out = append(out, [2]string{f.Name(), duckTypeName(f)})
+	}
+	return out, nil
+}
+
+// duckTypeName maps a Parquet leaf type onto the name duckdb reports for it, so
+// `columnar schema` prints the same table whichever engine answered. The
+// logical type is what decides most of it: fetch_status is a physical int32
+// carrying an INT(16) annotation, and duckdb calls that a SMALLINT.
+func duckTypeName(n parquet.Node) string {
+	if !n.Leaf() {
+		return "STRUCT"
+	}
+	if lt := n.Type().LogicalType(); lt != nil {
+		switch {
+		case lt.UTF8 != nil, lt.Enum != nil, lt.Json != nil:
+			return "VARCHAR"
+		case lt.UUID != nil:
+			return "UUID"
+		case lt.Date != nil:
+			return "DATE"
+		case lt.Time != nil:
+			return "TIME"
+		case lt.Timestamp != nil:
+			if lt.Timestamp.IsAdjustedToUTC {
+				return "TIMESTAMP WITH TIME ZONE"
+			}
+			return "TIMESTAMP"
+		case lt.Decimal != nil:
+			return fmt.Sprintf("DECIMAL(%d,%d)", lt.Decimal.Precision, lt.Decimal.Scale)
+		case lt.Integer != nil:
+			return duckIntName(lt.Integer.BitWidth, lt.Integer.IsSigned)
+		}
+	}
+	switch n.Type().Kind() {
+	case parquet.Boolean:
+		return "BOOLEAN"
+	case parquet.Int32:
+		return "INTEGER"
+	case parquet.Int64:
+		return "BIGINT"
+	case parquet.Float:
+		return "FLOAT"
+	case parquet.Double:
+		return "DOUBLE"
+	case parquet.ByteArray, parquet.FixedLenByteArray:
+		return "BLOB"
+	}
+	return strings.ToUpper(n.Type().String())
+}
+
+// duckIntName spells an annotated integer width the way duckdb does.
+func duckIntName(bits int8, signed bool) string {
+	name := map[int8]string{8: "TINYINT", 16: "SMALLINT", 32: "INTEGER", 64: "BIGINT"}[bits]
+	if name == "" {
+		return "BIGINT"
+	}
+	if !signed {
+		return "U" + name
+	}
+	return name
+}
diff --git a/ccrawl/columnar_native_test.go b/ccrawl/columnar_native_test.go
new file mode 100644
index 0000000..63c4083
--- /dev/null
+++ b/ccrawl/columnar_native_test.go
@@ -0,0 +1,403 @@
+package ccrawl
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"sort"
+	"testing"
+
+	"github.com/parquet-go/parquet-go"
+)
+
+// indexRow mirrors the columns of the Common Crawl columnar index the native
+// engine reads. Optional is on every string column because the real files have
+// nulls in most of them, and null handling is the part worth testing.
+type indexRow struct {
+	URLSurtkey  string  `parquet:"url_surtkey,optional,dict"`
+	URL         string  `parquet:"url,optional,dict"`
+	HostName    string  `parquet:"url_host_name,optional,dict"`
+	HostTLD     string  `parquet:"url_host_tld,optional,dict"`
+	HostDomain  string  `parquet:"url_host_registered_domain,optional,dict"`
+	Path        string  `parquet:"url_path,optional,dict"`
+	FetchStatus int32   `parquet:"fetch_status,optional"`
+	MIME        string  `parquet:"content_mime_detected,optional,dict"`
+	Languages   *string `parquet:"content_languages,optional,dict"`
+	WARCFile    string  `parquet:"warc_filename,optional,dict"`
+	Offset      int64   `parquet:"warc_record_offset,optional"`
+	Length      int64   `parquet:"warc_record_length,optional"`
+}
+
+// strp is here so a fixture row can hold an empty language rather than no
+// language, which are two different things in the real index and have to stay
+// two different things through the engine.
+func strp(s string) *string { return &s }
+
+// row builds one index row from a host and path, filling the derived columns
+// the way the real index does so the surtkey pruning has something to prune on.
+func row(host, path, mime string, lang *string, status int32) indexRow {
+	dom := host
+	if n := len(host) - len(".example.com"); n > 0 && host[n:] == ".example.com" {
+		dom = "example.com"
+	}
+	return indexRow{
+		URLSurtkey: surtHostKey(host) + ")" + path, URL: "https://" + host + path,
+		HostName: host, HostTLD: host[len(host)-3:], HostDomain: dom, Path: path,
+		FetchStatus: status, MIME: mime, Languages: lang,
+		WARCFile: "crawl-data/CC-MAIN-2025-05/segments/1/warc/" + host + ".warc.gz",
+		Offset:   1024, Length: 2048,
+	}
+}
+
+// testRows are sorted by surtkey, like the real files, so row group statistics
+// are non-overlapping and pruning has a chance to work.
+func testRows() []indexRow {
+	rows := []indexRow{
+		row("a.example.com", "/one", "text/html", strp("eng"), 200),
+		row("a.example.com", "/two", "text/html", strp("eng,fra"), 200),
+		row("b.example.com", "/three", "application/pdf", strp("eng"), 404),
+		row("example.com", "/", "text/html", strp("vie"), 200),
+		row("example.com", "/about", "text/html", strp("vie,eng"), 200),
+		row("other.org", "/x", "text/html", strp("deu"), 200),
+		// One row with an empty language and one with none at all, which the
+		// breakdown has to report as two groups rather than one.
+		row("other.org", "/y", "text/plain", strp(""), 500),
+		row("shop.example.net", "/p", "text/html", strp("eng"), 200),
+		row("null.example.org", "/n", "text/html", nil, 200),
+	}
+	sort.Slice(rows, func(i, j int) bool { return rows[i].URLSurtkey < rows[j].URLSurtkey })
+	return rows
+}
+
+// writeTestParquet writes the rows into a Parquet file with small row groups,
+// so a single file exercises the multi row group path.
+func writeTestParquet(t *testing.T, rows []indexRow) string {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), "part.parquet")
+	f, err := os.Create(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	w := parquet.NewGenericWriter[indexRow](f, parquet.MaxRowsPerRowGroup(3), parquet.PageBufferSize(1<<10))
+	if _, err := w.Write(rows); err != nil {
+		t.Fatal(err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if err := f.Close(); err != nil {
+		t.Fatal(err)
+	}
+	return path
+}
+
+// serveFile puts the file behind an HTTP server that honors range requests, so
+// the test drives the same httpReaderAt path the real command does.
+func serveFile(t *testing.T, path string) (*HTTPClient, string) {
+	t.Helper()
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.ServeFile(w, r, path)
+	}))
+	t.Cleanup(srv.Close)
+	return NewHTTPClient(Config{}), srv.URL + "/part.parquet"
+}
+
+// scanFixture runs a scan against the fixture file and returns the emitted rows.
+func scanFixture(t *testing.T, s NativeScan) []map[string]any {
+	t.Helper()
+	h, url := serveFile(t, writeTestParquet(t, testRows()))
+	s.URLs = []string{url}
+	var got []map[string]any
+	if err := RunColumnarNative(context.Background(), h, s, func(m map[string]any) error {
+		got = append(got, m)
+		return nil
+	}); err != nil {
+		t.Fatalf("scan: %v", err)
+	}
+	return got
+}
+
+func TestNativeCount(t *testing.T) {
+	cases := []struct {
+		name string
+		q    ColumnarQuery
+		want int64
+	}{
+		{"everything", ColumnarQuery{}, 9},
+		{"tld", ColumnarQuery{TLD: "com"}, 5},
+		{"domain includes apex and subdomains", ColumnarQuery{Domain: "example.com"}, 5},
+		{"host is exact", ColumnarQuery{Host: "example.com"}, 2},
+		{"status", ColumnarQuery{Status: 200}, 7},
+		{"mime", ColumnarQuery{MIME: "application/pdf"}, 1},
+		{"lang substring", ColumnarQuery{Lang: "eng"}, 5},
+		{"path prefix", ColumnarQuery{PathPrefix: "/t"}, 2},
+		{"combined", ColumnarQuery{Domain: "example.com", Status: 200, MIME: "text/html"}, 4},
+		{"matches nothing", ColumnarQuery{Domain: "absent.example"}, 0},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := scanFixture(t, NativeScan{Query: tc.q, Aggregate: NativeCount})
+			if len(got) != 1 {
+				t.Fatalf("want one count row, got %d", len(got))
+			}
+			if got[0]["n"] != tc.want {
+				t.Errorf("count = %v, want %d", got[0]["n"], tc.want)
+			}
+		})
+	}
+}
+
+// The domain filter is the one that has to match both the apex host's captures
+// and every subdomain's, which are two different surtkey shapes. Getting this
+// wrong silently drops the apex, so it gets its own test.
+func TestNativeDomainMatchesApexAndSubdomains(t *testing.T) {
+	got := scanFixture(t, NativeScan{
+		Query:     ColumnarQuery{Domain: "example.com"},
+		Aggregate: NativeRows,
+		Select:    []string{"url_host_name"},
+	})
+	seen := map[string]int{}
+	for _, r := range got {
+		seen[r["url_host_name"].(string)]++
+	}
+	want := map[string]int{"example.com": 2, "a.example.com": 2, "b.example.com": 1}
+	if len(seen) != len(want) {
+		t.Fatalf("hosts = %v, want %v", seen, want)
+	}
+	for h, n := range want {
+		if seen[h] != n {
+			t.Errorf("host %s = %d, want %d", h, seen[h], n)
+		}
+	}
+}
+
+func TestNativeRowsProjectAndLimit(t *testing.T) {
+	got := scanFixture(t, NativeScan{
+		Query:     ColumnarQuery{Host: "a.example.com"},
+		Aggregate: NativeRows,
+		Select:    LocationColumns,
+	})
+	if len(got) != 2 {
+		t.Fatalf("want 2 rows, got %d: %v", len(got), got)
+	}
+	for _, r := range got {
+		if len(r) != len(LocationColumns) {
+			t.Errorf("row has %d columns, want %d: %v", len(r), len(LocationColumns), r)
+		}
+		if r["warc_record_offset"] != int64(1024) || r["warc_record_length"] != int64(2048) {
+			t.Errorf("int columns did not survive: %v", r)
+		}
+		if _, ok := r["url"].(string); !ok {
+			t.Errorf("url is %T, want string", r["url"])
+		}
+	}
+
+	// The limit has to stop the scan, not just truncate what was collected.
+	lim := scanFixture(t, NativeScan{
+		Aggregate: NativeRows, Select: []string{"url"}, Limit: 3, Workers: 1,
+	})
+	if len(lim) != 3 {
+		t.Errorf("limit 3 emitted %d rows", len(lim))
+	}
+}
+
+func TestNativeGroupCount(t *testing.T) {
+	got := scanFixture(t, NativeScan{
+		Query:     ColumnarQuery{},
+		Aggregate: NativeGroupCount,
+		GroupBy:   "content_mime_detected",
+	})
+	if len(got) != 3 {
+		t.Fatalf("want 3 mime groups, got %d: %v", len(got), got)
+	}
+	// Ordered by count descending, so the most common MIME is first.
+	if got[0]["value"] != "text/html" || got[0]["n"] != int64(7) {
+		t.Errorf("top group = %v, want text/html 7", got[0])
+	}
+	for i := 1; i < len(got); i++ {
+		if got[i-1]["n"].(int64) < got[i]["n"].(int64) {
+			t.Errorf("groups are not ordered by count: %v", got)
+		}
+	}
+}
+
+// An empty string and a null are different values, and duckdb reports them as
+// two groups. The engine keeps them apart with a sentinel key, so this checks
+// they stay apart and that the sentinel never leaks into the output.
+func TestNativeGroupCountKeepsNullApartFromEmpty(t *testing.T) {
+	got := scanFixture(t, NativeScan{
+		Aggregate: NativeGroupCount,
+		GroupBy:   "content_languages",
+	})
+	var empty, null, total int64
+	for _, r := range got {
+		total += r["n"].(int64)
+		switch v := r["value"].(type) {
+		case nil:
+			null += r["n"].(int64)
+		case string:
+			if v == nullGroup {
+				t.Fatalf("the null sentinel leaked into the output: %v", got)
+			}
+			if v == "" {
+				empty += r["n"].(int64)
+			}
+		}
+	}
+	if empty != 1 || null != 1 {
+		t.Errorf("empty = %d and null = %d, want one of each: %v", empty, null, got)
+	}
+	if total != 9 {
+		t.Errorf("groups total %d rows, want 9", total)
+	}
+}
+
+func TestNativeGroupCountLimit(t *testing.T) {
+	got := scanFixture(t, NativeScan{
+		Aggregate: NativeGroupCount, GroupBy: "url_host_tld", Limit: 1,
+	})
+	if len(got) != 1 {
+		t.Fatalf("want 1 group, got %d", len(got))
+	}
+	if got[0]["value"] != "com" || got[0]["n"] != int64(5) {
+		t.Errorf("top tld = %v, want com 5", got[0])
+	}
+}
+
+func TestNativeSchema(t *testing.T) {
+	h, url := serveFile(t, writeTestParquet(t, testRows()))
+	cols, err := NativeSchema(context.Background(), h, url)
+	if err != nil {
+		t.Fatal(err)
+	}
+	types := map[string]string{}
+	for _, c := range cols {
+		types[c[0]] = c[1]
+	}
+	if types["url"] != "VARCHAR" {
+		t.Errorf("url is %q, want VARCHAR", types["url"])
+	}
+	if types["fetch_status"] != "INTEGER" {
+		t.Errorf("fetch_status is %q, want INTEGER", types["fetch_status"])
+	}
+	if types["warc_record_offset"] != "BIGINT" {
+		t.Errorf("warc_record_offset is %q, want BIGINT", types["warc_record_offset"])
+	}
+}
+
+// The point of the engine is reading less, so pruning gets tested on the
+// decision itself rather than only on the answers. Counting bytes would prove
+// nothing on a fixture this small, since one range request covers the file.
+func TestNativePrunesRowGroups(t *testing.T) {
+	f, err := os.Open(writeTestParquet(t, testRows()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = f.Close() }()
+	info, err := f.Stat()
+	if err != nil {
+		t.Fatal(err)
+	}
+	pf, err := parquet.OpenFile(f, info.Size())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(pf.RowGroups()) != 3 {
+		t.Fatalf("fixture has %d row groups, want the 3 the counts below assume", len(pf.RowGroups()))
+	}
+
+	// Three row groups of three rows each, in surtkey order:
+	//   0: example.com/, example.com/about, a.example.com/one
+	//   1: a.example.com/two, b.example.com/three, shop.example.net/p
+	//   2: null.example.org/n, other.org/x, other.org/y
+	// The counts below are what min and max on those groups can actually rule
+	// out. A group whose bounds straddle the wanted value survives, which is
+	// the honest answer, so the expectations are exact rather than "some".
+	cases := []struct {
+		name string
+		q    ColumnarQuery
+		want int
+	}{
+		// Every group's tld is below "zzz", so nothing has to be read at all.
+		{"absent tld prunes every group", ColumnarQuery{TLD: "zzz"}, 3},
+		// Group 1 runs from a.example.com to shop.example.net, and the wanted
+		// host sorts inside that, so bounds cannot rule it out.
+		{"absent host prunes what bounds allow", ColumnarQuery{Host: "nope.example.com"}, 2},
+		// Group 2 holds a 200 and a 500, so 418 sorts inside its range.
+		{"absent status prunes what bounds allow", ColumnarQuery{Status: 418}, 2},
+		// Group 1 survives the host name bounds and is then pruned by the
+		// surtkey prefix, which is the pruning this engine is built around.
+		{"a present host prunes the groups it is not in", ColumnarQuery{Host: "other.org"}, 2},
+		// Group 2 is all .org.
+		{"a present tld still prunes", ColumnarQuery{TLD: "com"}, 1},
+		{"no filter prunes nothing", ColumnarQuery{}, 0},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			plan, err := newScanPlan(NativeScan{Query: tc.q, Aggregate: NativeCount})
+			if err != nil {
+				t.Fatal(err)
+			}
+			idx := map[string]int{}
+			for _, c := range plan.filterCols {
+				leaf, ok := pf.Schema().Lookup(c)
+				if !ok {
+					t.Fatalf("column %s is missing from the fixture", c)
+				}
+				idx[c] = leaf.ColumnIndex
+			}
+			pruned := 0
+			for _, rg := range pf.RowGroups() {
+				if pruneRowGroup(plan, idx, rg.ColumnChunks()) {
+					pruned++
+				}
+			}
+			if pruned != tc.want {
+				t.Errorf("pruned %d of %d row groups, want %d", pruned, len(pf.RowGroups()), tc.want)
+			}
+		})
+	}
+}
+
+func TestNativeExpressible(t *testing.T) {
+	if !NativeExpressible(NativeScan{Select: LocationColumns}) {
+		t.Error("the location columns should be expressible")
+	}
+	if !NativeExpressible(NativeScan{Select: DefaultColumnarColumns}) {
+		t.Error("the default columns should be expressible")
+	}
+	if NativeExpressible(NativeScan{Select: []string{"url", "content_charset"}}) {
+		t.Error("a column the engine does not know should not be expressible")
+	}
+	if NativeExpressible(NativeScan{GroupBy: "warc_segment"}) {
+		t.Error("grouping by an unknown column should not be expressible")
+	}
+}
+
+func TestNativeErrors(t *testing.T) {
+	h := NewHTTPClient(Config{})
+	nop := func(map[string]any) error { return nil }
+	if err := RunColumnarNative(context.Background(), h, NativeScan{Aggregate: NativeCount}, nop); err == nil {
+		t.Error("a scan with no files should fail")
+	}
+	scan := NativeScan{URLs: []string{"https://example.invalid/x.parquet"}, Aggregate: NativeRows}
+	if err := RunColumnarNative(context.Background(), h, scan, nop); err == nil {
+		t.Error("a row scan with no columns should fail")
+	}
+}
+
+// An emit that fails has to stop the scan and surface, rather than being
+// swallowed by a worker.
+func TestNativeEmitErrorPropagates(t *testing.T) {
+	h, url := serveFile(t, writeTestParquet(t, testRows()))
+	want := fmt.Errorf("downstream is gone")
+	err := RunColumnarNative(context.Background(), h, NativeScan{
+		URLs: []string{url}, Aggregate: NativeRows, Select: []string{"url"},
+	}, func(map[string]any) error { return want })
+	if err == nil || err.Error() != want.Error() {
+		t.Errorf("err = %v, want %v", err, want)
+	}
+}
diff --git a/ccrawl/journal.go b/ccrawl/journal.go
index c34f524..04fd303 100644
--- a/ccrawl/journal.go
+++ b/ccrawl/journal.go
@@ -294,7 +294,7 @@ func (r *RunReporter) Textf(format string, a ...any) {
 	}
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	fmt.Fprintf(r.out, format, a...)
+	_, _ = fmt.Fprintf(r.out, format, a...)
 }
 
 // record folds an event into the metrics registry.
diff --git a/cli/misc.go b/cli/misc.go
index 2c63e96..7f3a94d 100644
--- a/cli/misc.go
+++ b/cli/misc.go
@@ -53,7 +53,9 @@ func sampleHash(s string) float64 {
 
 func itoa64(n int64) string { return strconv.FormatInt(n, 10) }
 
-// str renders a DuckDB JSON value as a string.
+// str renders a result value as a string. Numbers arrive as float64 from
+// duckdb, which goes through encoding/json, and as int64 from the native
+// engine, which reads the Parquet column directly.
 func str(v any) string {
 	if v == nil {
 		return ""
@@ -73,8 +75,19 @@ func str(v any) string {
 	}
 }
 
+// toInt64 reads an integer out of a result value. duckdb's JSON gives every
+// number as a float64, but the native engine hands back the Parquet column's
+// own type, so a warc offset arrives as an int64 and used to fall through to
+// zero here. A location with a zero offset and length is a record nobody can
+// fetch, and nothing about it looks wrong until the fetch comes back empty.
 func toInt64(v any) int64 {
 	switch t := v.(type) {
+	case int64:
+		return t
+	case int32:
+		return int64(t)
+	case int:
+		return int64(t)
 	case float64:
 		return int64(t)
 	case string:
diff --git a/cli/misc_test.go b/cli/misc_test.go
index 358fe0f..abc3e48 100644
--- a/cli/misc_test.go
+++ b/cli/misc_test.go
@@ -19,6 +19,32 @@ func TestParseLocationLine(t *testing.T) {
 	}
 }
 
+// Both engines feed the same row map to the same output code, and they do not
+// agree on the Go type of a number: duckdb goes through encoding/json and gives
+// float64, the native engine gives the Parquet column's own int64. Missing one
+// of those silently zeroed every warc offset the native engine produced.
+func TestToInt64AcceptsBothEngines(t *testing.T) {
+	cases := map[string]struct {
+		in   any
+		want int64
+	}{
+		"duckdb json number": {float64(1234), 1234},
+		"native int64":       {int64(1234), 1234},
+		"native int32":       {int32(1234), 1234},
+		"plain int":          {1234, 1234},
+		"cdx string":         {"1234", 1234},
+		"missing":            {nil, 0},
+		"not a number":       {"abc", 0},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			if got := toInt64(tc.in); got != tc.want {
+				t.Errorf("toInt64(%#v) = %d, want %d", tc.in, got, tc.want)
+			}
+		})
+	}
+}
+
 func TestHumanBytes(t *testing.T) {
 	cases := map[int64]string{
 		512:             "512 B",
diff --git a/cli/table.go b/cli/table.go
index aa66e93..2296497 100644
--- a/cli/table.go
+++ b/cli/table.go
@@ -42,7 +42,7 @@ func (tf *tableFlags) bind(f *kit.FlagSet) {
 	f.IntVar(&tf.status, "status", 0, "fetch_status (e.g. 200)")
 	f.StringVar(&tf.pathPrefix, "path-prefix", "", "url_path prefix")
 	f.StringVar(&tf.subset, "subset", "warc", "warc|crawldiagnostics|robotstxt")
-	f.StringVar(&tf.engine, "engine", "auto", "auto|duckdb|print")
+	f.StringVar(&tf.engine, "engine", "auto", "auto|duckdb|native|print")
 	f.BoolVar(&tf.print, "print", false, "print the SQL and exit")
 }
 
@@ -54,9 +54,14 @@ func newTableCmd() kit.Command {
 		Long: `Query Common Crawl's columnar (Parquet) index, the fastest way to answer bulk
 questions like "every PDF on .gov domains" without touching a single WARC.
 
-By default the SQL runs against a local duckdb binary over the public Parquet
-files. With --engine print (or no duckdb installed) the ready-to-run SQL is
-printed so you can paste it into Athena, Spark, Trino, or DuckDB yourself.
+urls, locations, count, langs, mimes and schema are answered by the built-in
+native engine, which reads the Parquet files directly over ranged HTTP and skips
+the row groups whose statistics rule them out. Nothing needs installing for it.
+
+"columnar query" and "columnar sql --run" take arbitrary SQL, so they run
+against a local duckdb binary. With --engine print the ready-to-run SQL is
+printed so you can paste it into Athena, Spark, Trino, or DuckDB yourself. Use
+--engine duckdb to send a filter query through duckdb too.
 
 Examples:
   ccrawl columnar urls --domain example.com --status 200 -o url
@@ -199,8 +204,24 @@ func (t *tableCmd) runBreakdown(ctx context.Context, app *App, id string) error
 	if app.Limit > 0 {
 		sql += fmt.Sprintf("\nLIMIT %d", app.Limit)
 	}
-	return runColumnarSQL(ctx, app, sql, &t.tf, func(row map[string]any) error {
-		return app.Out.Emit(Row{Cols: []string{col, "count"}, Vals: []string{str(row[col]), str(row["n"])}, Value: row})
+	scan := nativeScan(q, app)
+	scan.Aggregate = ccrawl.NativeGroupCount
+	scan.GroupBy = col
+	// The native engine returns the group under "value" rather than under the
+	// column's own name, since one field name keeps the sink generic. Both
+	// spellings are read here, and the row is rebuilt under the column's own
+	// name, so the JSON is the same whichever engine ran.
+	return runColumnarScan(ctx, app, sql, scan, &t.tf, func(row map[string]any) error {
+		v, ok := row[col]
+		if !ok {
+			v = row["value"]
+		}
+		n := row["n"]
+		return app.Out.Emit(Row{
+			Cols:  []string{col, "count"},
+			Vals:  []string{str(v), str(n)},
+			Value: map[string]any{col: v, "n": n},
+		})
 	})
 }
 
@@ -218,18 +239,63 @@ func (t *tableCmd) runSchema(ctx context.Context, app *App, id string) error {
 	// -json mode, which yields no JSON rows; the subquery makes the output
 	// consistent across duckdb versions.
 	sql := fmt.Sprintf("SELECT column_name, column_type FROM (DESCRIBE SELECT * FROM read_parquet('%s', hive_partitioning=1) LIMIT 1)", src)
-	return runColumnarSQL(ctx, app, sql, &t.tf, func(row map[string]any) error {
+	emit := func(name, typ string) error {
 		return app.Out.Emit(Row{
 			Cols:  []string{"column", "type"},
-			Vals:  []string{str(row["column_name"]), str(row["column_type"])},
-			Value: row,
+			Vals:  []string{name, typ},
+			Value: map[string]any{"column_name": name, "column_type": typ},
 		})
+	}
+	switch pickEngine(&t.tf, true) {
+	case enginePrint:
+		_, _ = fmt.Fprintln(cmdOut, sql)
+		return nil
+	case engineNative:
+		// The schema is the same in every part, so reading one footer answers it.
+		urls, err := ccrawl.ColumnarParquetURLs(ctx, app.HTTP, app.Cache, id, t.tf.subset, app.Cfg.Source)
+		if err != nil {
+			return err
+		}
+		if len(urls) == 0 {
+			return noResults("no parquet files for this crawl and subset")
+		}
+		cols, err := ccrawl.NativeSchema(ctx, app.HTTP, urls[0])
+		if err != nil {
+			return err
+		}
+		// crawl and subset are hive partitions in the path rather than columns
+		// in the file, so duckdb reports them and the footer does not. They are
+		// appended by hand to keep the two engines printing the same table.
+		cols = append(cols, [2]string{"crawl", "VARCHAR"}, [2]string{"subset", "VARCHAR"})
+		for _, c := range cols {
+			if err := emit(c[0], c[1]); err != nil {
+				return err
+			}
+		}
+		return app.Out.Flush()
+	}
+	return runColumnarSQL(ctx, app, sql, &t.tf, func(row map[string]any) error {
+		return emit(str(row["column_name"]), str(row["column_type"]))
 	})
 }
 
-// runColumnar renders the SQL from q and dispatches to the engine.
+// runColumnar renders the SQL from q, builds the equivalent native scan, and
+// dispatches to whichever engine is going to run.
 func runColumnar(ctx context.Context, app *App, q ccrawl.ColumnarQuery, tf *tableFlags, emit func(map[string]any) error) error {
-	return runColumnarSQL(ctx, app, q.SQL(app.Cfg.Source), tf, emit)
+	scan := nativeScan(q, app)
+	scan.Aggregate = ccrawl.NativeRows
+	scan.Select = q.Select
+	if len(q.Select) == 1 && strings.HasPrefix(q.Select[0], "count(") {
+		scan.Aggregate = ccrawl.NativeCount
+		scan.Select = nil
+	}
+	return runColumnarScan(ctx, app, q.SQL(app.Cfg.Source), scan, tf, emit)
+}
+
+// nativeScan carries the parts of a columnar query the native engine shares
+// with the SQL, leaving the caller to say what it wants produced.
+func nativeScan(q ccrawl.ColumnarQuery, app *App) ccrawl.NativeScan {
+	return ccrawl.NativeScan{Query: q, Limit: app.Limit, Workers: app.Workers}
 }
 
 // resolveGlobForDuckDB rewrites the quoted `*.parquet` glob in sql into a
@@ -252,17 +318,73 @@ func resolveGlobForDuckDB(ctx context.Context, app *App, tf *tableFlags, sql str
 	return strings.ReplaceAll(sql, glob, ccrawl.ParquetListLiteral(urls)), nil
 }
 
-func runColumnarSQL(ctx context.Context, app *App, sql string, tf *tableFlags, emit func(map[string]any) error) error {
-	engine := tf.engine
-	if tf.print || engine == "print" {
+// chosen is which engine a command is about to use.
+type chosen int
+
+const (
+	engineDuckDB chosen = iota
+	engineNative
+	enginePrint
+)
+
+// pickEngine resolves --engine and --print into the engine that will actually
+// run. Auto takes native whenever native can answer the query, installed duckdb
+// or not, so the same command behaves the same way on every box. What is left
+// is the arbitrary SQL, which goes to duckdb where duckdb exists and is printed
+// where it does not. expressible says whether the native engine could answer
+// this one at all.
+func pickEngine(tf *tableFlags, expressible bool) chosen {
+	switch {
+	case tf.print || tf.engine == "print":
+		return enginePrint
+	case tf.engine == "native":
+		return engineNative
+	case tf.engine == "duckdb":
+		return engineDuckDB
+	case expressible:
+		return engineNative
+	case ccrawl.DuckDBAvailable():
+		return engineDuckDB
+	default:
+		return enginePrint
+	}
+}
+
+// runColumnarScan runs a query that both engines can answer. scan is the native
+// form and sql is the duckdb form; they have to agree, which is why they are
+// built side by side by the caller.
+func runColumnarScan(ctx context.Context, app *App, sql string, scan ccrawl.NativeScan, tf *tableFlags, emit func(map[string]any) error) error {
+	switch pickEngine(tf, ccrawl.NativeExpressible(scan)) {
+	case enginePrint:
+		if !tf.print && tf.engine == "auto" {
+			_, _ = fmt.Fprintln(cmdErr, "no duckdb binary found and this query needs SQL; printing it instead")
+		}
 		_, _ = fmt.Fprintln(cmdOut, sql)
 		return nil
+	case engineNative:
+		if !ccrawl.NativeExpressible(scan) {
+			return usageErr("the native engine cannot answer this query; use --engine duckdb")
+		}
+		return runColumnarNative(ctx, app, scan, tf, emit)
 	}
-	if engine == "auto" && !ccrawl.DuckDBAvailable() {
-		_, _ = fmt.Fprintln(cmdErr, "no duckdb binary found; printing SQL (install duckdb or use --engine duckdb)")
+	return runDuckDB(ctx, app, sql, tf, emit)
+}
+
+func runColumnarSQL(ctx context.Context, app *App, sql string, tf *tableFlags, emit func(map[string]any) error) error {
+	switch pickEngine(tf, false) {
+	case enginePrint:
+		if !tf.print && tf.engine == "auto" {
+			_, _ = fmt.Fprintln(cmdErr, "no duckdb binary found; printing SQL (install duckdb or use --engine duckdb)")
+		}
 		_, _ = fmt.Fprintln(cmdOut, sql)
 		return nil
+	case engineNative:
+		return usageErr("the native engine does not run SQL; use --engine duckdb")
 	}
+	return runDuckDB(ctx, app, sql, tf, emit)
+}
+
+func runDuckDB(ctx context.Context, app *App, sql string, tf *tableFlags, emit func(map[string]any) error) error {
 	// The printed SQL carries the `*.parquet` glob, which Athena and Spark expand
 	// themselves. duckdb cannot list the bucket, so for the duckdb run we swap the
 	// glob for the explicit file list from the crawl's manifest.
@@ -285,3 +407,30 @@ func runColumnarSQL(ctx context.Context, app *App, sql string, tf *tableFlags, e
 	}
 	return nil
 }
+
+func runColumnarNative(ctx context.Context, app *App, scan ccrawl.NativeScan, tf *tableFlags, emit func(map[string]any) error) error {
+	id, err := app.Crawl(ctx)
+	if err != nil {
+		return err
+	}
+	scan.URLs, err = ccrawl.ColumnarParquetURLs(ctx, app.HTTP, app.Cache, id, tf.subset, app.Cfg.Source)
+	if err != nil {
+		return err
+	}
+	// A count over the whole index reports 0 rather than nothing, so the empty
+	// result only applies to the queries that list rows.
+	n := 0
+	if err := ccrawl.RunColumnarNative(ctx, app.HTTP, scan, func(row map[string]any) error {
+		n++
+		return emit(row)
+	}); err != nil {
+		return err
+	}
+	if err := app.Out.Flush(); err != nil {
+		return err
+	}
+	if n == 0 {
+		return noResults("query returned no rows")
+	}
+	return nil
+}
diff --git a/cli/table_test.go b/cli/table_test.go
new file mode 100644
index 0000000..7090a07
--- /dev/null
+++ b/cli/table_test.go
@@ -0,0 +1,76 @@
+package cli
+
+import (
+	"testing"
+
+	"github.com/tamnd/ccrawl-cli/ccrawl"
+)
+
+// The engine picker is where a wrong branch means a query silently runs
+// somewhere the user did not ask for, so every combination gets pinned. The
+// duckdb-available cases are the ones that depend on the machine, so they are
+// split out below rather than guessed at.
+func TestPickEngineExplicit(t *testing.T) {
+	cases := []struct {
+		name        string
+		tf          tableFlags
+		expressible bool
+		want        chosen
+	}{
+		{"print flag beats everything", tableFlags{engine: "native", print: true}, true, enginePrint},
+		{"engine print", tableFlags{engine: "print"}, true, enginePrint},
+		{"engine native", tableFlags{engine: "native"}, true, engineNative},
+		// Asking for native on a query it cannot answer still picks native, so
+		// the caller can refuse rather than quietly running duckdb instead.
+		{"engine native on SQL", tableFlags{engine: "native"}, false, engineNative},
+		{"engine duckdb", tableFlags{engine: "duckdb"}, true, engineDuckDB},
+		{"engine duckdb on SQL", tableFlags{engine: "duckdb"}, false, engineDuckDB},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := pickEngine(&tc.tf, tc.expressible); got != tc.want {
+				t.Errorf("pickEngine = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+// Auto has to land on native for anything native can answer, whether or not
+// duckdb happens to be installed, or the same command would do different things
+// on two boxes. Only the arbitrary SQL is allowed to depend on the machine.
+func TestPickEngineAuto(t *testing.T) {
+	tf := tableFlags{engine: "auto"}
+	if got := pickEngine(&tf, true); got != engineNative {
+		t.Errorf("auto on an expressible query = %v, want native", got)
+	}
+	want := enginePrint
+	if ccrawl.DuckDBAvailable() {
+		want = engineDuckDB
+	}
+	if got := pickEngine(&tf, false); got != want {
+		t.Errorf("auto on SQL = %v, want %v", got, want)
+	}
+}
+
+// The columns each subcommand selects have to be ones the native engine knows,
+// or auto would fall through to printing SQL on a machine with no duckdb, which
+// is exactly the case the engine exists for.
+func TestColumnarSelectionsAreNativelyExpressible(t *testing.T) {
+	cases := []struct {
+		name string
+		scan ccrawl.NativeScan
+	}{
+		{"urls", ccrawl.NativeScan{Select: []string{"url", "fetch_status", "content_mime_detected", "content_languages"}}},
+		{"locations", ccrawl.NativeScan{Select: ccrawl.LocationColumns}},
+		{"count", ccrawl.NativeScan{Aggregate: ccrawl.NativeCount}},
+		{"langs", ccrawl.NativeScan{Aggregate: ccrawl.NativeGroupCount, GroupBy: "content_languages"}},
+		{"mimes", ccrawl.NativeScan{Aggregate: ccrawl.NativeGroupCount, GroupBy: "content_mime_detected"}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if !ccrawl.NativeExpressible(tc.scan) {
+				t.Errorf("%s is not natively expressible", tc.name)
+			}
+		})
+	}
+}
diff --git a/docs/content/reference/_index.md b/docs/content/reference/_index.md
index 4c1b517..c8ba442 100644
--- a/docs/content/reference/_index.md
+++ b/docs/content/reference/_index.md
@@ -5,4 +5,4 @@ description: "Every command, flag, output format, and setting, plus a troublesho
 weight: 30
 ---
 
-The complete surface, for when you know what you want and just need the exact spelling: the [CLI](/reference/cli/) command tree and flags, the [Markdown pipelines](/reference/markdown/) and their schemas, [configuration](/reference/configuration/) and environment, the [run journal](/reference/run-journal/) a long run writes, [output formats](/reference/output/), [exit codes](/reference/exit-codes/), and [troubleshooting](/reference/troubleshooting/).
+The complete surface, for when you know what you want and just need the exact spelling: the [CLI](/reference/cli/) command tree and flags, the [Markdown pipelines](/reference/markdown/) and their schemas, the [columnar engines](/reference/columnar-engines/) behind `ccrawl columnar`, [configuration](/reference/configuration/) and environment, the [run journal](/reference/run-journal/) a long run writes, [output formats](/reference/output/), [exit codes](/reference/exit-codes/), and [troubleshooting](/reference/troubleshooting/).
diff --git a/docs/content/reference/cli.md b/docs/content/reference/cli.md
index e9035e2..cb6c6a6 100644
--- a/docs/content/reference/cli.md
+++ b/docs/content/reference/cli.md
@@ -212,7 +212,7 @@ Aliases: `table`, `athena`.
 | `columnar schema` | The columns of the index |
 
 Filters: `--domain`, `--host`, `--tld`, `--mime`, `--status`, `--lang`, `--path-prefix`, `--subset`.
-Engine: `--engine` (`auto|duckdb|print`).
+Engine: `--engine` (`auto|duckdb|native|print`). See [the columnar engines](/reference/columnar-engines/) for what each one can answer and how fast.
 
 ---
 
diff --git a/docs/content/reference/columnar-engines.md b/docs/content/reference/columnar-engines.md
new file mode 100644
index 0000000..2135c3e
--- /dev/null
+++ b/docs/content/reference/columnar-engines.md
@@ -0,0 +1,55 @@
+---
+title: "Columnar engines"
+linkTitle: "Columnar engines"
+description: "The two engines behind ccrawl columnar, what each one can answer, and which one runs when."
+weight: 17
+---
+
+`ccrawl columnar` answers bulk questions against Common Crawl's Parquet index. There are two engines behind it, and `--engine` picks between them.
+
+| Engine | What it is | Needs |
+|---|---|---|
+| `duckdb` | The SQL is handed to a local duckdb binary reading the Parquet over HTTPS | duckdb on PATH |
+| `native` | ccrawl reads the Parquet files itself over ranged HTTP | nothing |
+| `print` | The SQL is printed for you to run in Athena, Spark, Trino, or duckdb yourself | nothing |
+| `auto` | native for anything native can answer, duckdb for the rest | nothing |
+
+The default is `auto`. It takes the native engine for every query the native engine can answer, whether or not duckdb is installed, so the same command does the same thing on every box. What is left is the arbitrary SQL, and that goes to duckdb where duckdb exists and gets printed where it does not. `--engine duckdb` forces the old path on a box that has both.
+
+## What the native engine can answer
+
+`urls`, `locations`, `count`, `langs`, `mimes` and `schema`, with any combination of the filter flags: `--domain`, `--host`, `--tld`, `--mime`, `--lang`, `--status`, `--path-prefix` and `--subset`.
+
+`columnar query` and `columnar sql` take arbitrary SQL, so they need duckdb and always will. A hand written engine that accepts arbitrary SQL is a database, and this is a command line tool. Asking for `--engine native` on one of those is an error rather than a silent fallback, so a script cannot end up quietly running somewhere else.
+
+## How it reads less
+
+A crawl's `warc` subset is thousands of Parquet parts. Reading all of them to count the captures on one TLD would be absurd, so the native engine does what a query planner does.
+
+It opens each part over ranged HTTP and reads only the footer. From the footer it reads the page index for the columns the filters mention, and compares each page's minimum and maximum against the filter. A part whose pages all fall outside the filter is dropped without a single data page being read. Because the files are sorted by `url_surtkey`, a `--domain` or `--host` filter also gets a prefix predicate on that column, which is the one that prunes hardest.
+
+What survives is read a column at a time, cheapest column first, and each column narrows the set of rows still alive. A row group where the first filter kills everything never has the expensive columns touched at all. Only then are the output columns read, and only for the rows that matched.
+
+The upshot on a real crawl: opening a part and deciding it holds nothing costs about 200 KiB out of an 8.7 MiB file.
+
+## Speed
+
+Counting the Vietnamese TLD across the 300 parts of one crawl's `robotstxt` subset, on a home connection:
+
+| Engine | Wall clock |
+|---|---|
+| native, `-j 16` | 15s |
+| native, default `-j 8` | 22s |
+| duckdb | 39s |
+
+Being faster than duckdb is not the point and will not hold for every query. duckdb is a vectorized engine and will win where a query actually has to read a lot of rows. The point is that the no-dependency path is not a consolation prize.
+
+The native engine does not apply the `--rate` inter-request delay. That delay is priced for requests that mean something, a CDX page or a WARC range or a whole file, and a columnar scan is thousands of few-kilobyte reads of footers and column indexes. duckdb, which answers the same queries by default, applies no delay either.
+
+## Differences to know about
+
+Rows come back in whichever order the parts finish, which is also what duckdb over an unordered file list does. Add your own sort if you need one.
+
+`langs` and `mimes` order by count descending and break ties by value, so repeated runs agree with each other.
+
+An empty string and a missing value are two different groups in a breakdown, under both engines.