================================================================================
 Parquet fixed-size-list read benchmark: C (FIXED_SIZE_LIST annotation) with an
 annotation-aware reader is within ~1% of B (VECTOR) and A (FLBA).
================================================================================

PURPOSE
-------
Demonstrates that the Parquet read speedup for fixed-size (ML embedding) list
columns comes from SKIPPING rep/def level decode, not from any particular physical
encoding. Once levels are skipped, all three proposals reach the same read speed:
  A / FLBA           - store the vector as one FIXED_LEN_BYTE_ARRAY (no levels on disk)
  B / VECTOR         - new physical FieldRepetitionType.VECTOR, no levels (arrow-go PR #854)
  C / FIXED_SIZE_LIST - a LOGICAL annotation on an ordinary 3-level LIST (levels still
                        on disk); an annotation-aware reader skips decoding them.

The key point missing from prior benchmarks: 'C' is only slow when the annotation is
IGNORED and the file is read back through the full Dremel LIST decoder (C-naive). An
annotation-AWARE reader (C-hint) skips the level decode and matches B - while keeping a
standard, backward-compatible LIST on disk (annotation-unaware readers read it correctly).

REPRODUCIBILITY
---------------
Base: apache/arrow-go PR #854 (VECTOR PoC), branch tip commit fcad93306040729d03f9d0c22ca682277cde3ae5.
  git clone https://github.com/apache/arrow-go && cd arrow-go
  git fetch origin pull/854/head && git checkout fcad93306040729d03f9d0c22ca682277cde3ae5
Then apply the C-hint reader patch (section 3) and drop the fslbench/ package
(section 4) into the tree. Go 1.26. Run:
  go test ./fslbench/ -run Test -v          # correctness + skip-path proof
  go test ./fslbench/ -run '^$' -bench BenchmarkRead -benchtime=40x -count=10

================================================================================
 1. HEADLINE RESULTS (100k rows, dim=512 float32, uncompressed)
    well-sampled: 10 runs x 40 iters, mean +/- sigma
================================================================================

  arm         ns/row (mean +/- sd)   note
  A/FLBA      2730 +/- 13            no levels on disk (FLBA->float reinterpret adds a bit)
  B/VECTOR    2337 +/- 10            no levels (new physical rep type; forward-INCOMPATIBLE)
  C-hint      2356 +/-  5            skip-levels reader on a plain LIST; WITHIN 1% of B,
                                     fully backward-compatible
  C-naive     3830 +/- 22            annotation ignored, full Dremel (what naive 'C' measures)
  LIST        ~4038                  full Dremel baseline

  => A ~= B ~= C-hint (fast band); C-naive ~= LIST (slow band). The win is skipping
     level decode. C-hint recovers ~99% of it (2356 vs 4038) with zero format break.
     zstd compresses all arms toward the baseline (delta shrinks to ~1.0-1.24x) - a
     reminder that read-decode benchmarks must isolate the uncompressed path.

================================================================================
 2. FULL PER-CONFIG TABLES (single-run; carry a few % run-to-run noise - see
    the well-sampled headline above for B-vs-C-hint, which is within 1%)
================================================================================

# fslbench results

Read-decode-only, 32 random row-group probes per op. `×base` is speedup vs the plain LIST arm (higher = faster). ns/row is per embedding row (dim=512 float32 = 2048 value-bytes/row).

## Headline (100k rows, uncompressed, well-sampled: 10 runs × 40 iters, mean ± σ)

| arm | ns/row (mean ± σ) | note |
|-----|------------------:|:-----|
| A/FLBA | 2730 ± 13 | no levels on disk (FLBA reinterpret adds a small cost) |
| B/VECTOR | 2337 ± 10 | no levels (new physical VECTOR rep type; forward-incompatible) |
| **C-hint** | **2356 ± 5** | **skip-levels reader on a plain LIST — within 1% of B, backward-compatible** |
| C-naive | 3830 ± 22 | annotation ignored, full Dremel (this is what Rok's benchmark measured as "C") |
| LIST | ~4038 | full Dremel baseline |

**Takeaway:** C-hint reads **within 1%** of B/VECTOR and recovers ~99% of the achievable speedup over C-naive/LIST — with zero format break (the file is a standard LIST; annotation-unaware readers read it correctly). The A ≈ B ≈ C-hint fast band vs the C-naive ≈ LIST slow band is the load-bearing result: the win is *skipping level decode*, not the physical encoding. (The per-cell single-run tables below carry a few % of run-to-run noise; the ~6% C-hint-vs-B gap that appears in some single runs does not survive proper sampling — see the well-sampled numbers above.)

## rows=1000, uncompressed

| arm | ns/row | MB/s | ×base | levels decoded? |
|-----|-------:|-----:|------:|:----------------|
| A/FLBA | 2689 | 762 | 1.41× | no (FLBA, no levels on disk) |
| B/VECTOR | 2110 | 970 | 1.80× | no (VECTOR: maxRep=maxDef=0) |
| C-hint | 2284 | 897 | 1.66× | **no (skip-levels fast path)** |
| C-naive | 3756 | 545 | 1.01× | yes (full Dremel) |
| LIST | 3795 | 540 | 1.00× | yes (full Dremel, baseline) |

## rows=1000, zstd

| arm | ns/row | MB/s | ×base | levels decoded? |
|-----|-------:|-----:|------:|:----------------|
| A/FLBA | 7567 | 271 | 1.17× | no (FLBA, no levels on disk) |
| B/VECTOR | 6917 | 296 | 1.28× | no (VECTOR: maxRep=maxDef=0) |
| C-hint | 7404 | 277 | 1.20× | **no (skip-levels fast path)** |
| C-naive | 9019 | 227 | 0.98× | yes (full Dremel) |
| LIST | 8869 | 231 | 1.00× | yes (full Dremel, baseline) |

## rows=10000, uncompressed

| arm | ns/row | MB/s | ×base | levels decoded? |
|-----|-------:|-----:|------:|:----------------|
| A/FLBA | 3116 | 657 | 1.29× | no (FLBA, no levels on disk) |
| B/VECTOR | 2537 | 807 | 1.58× | no (VECTOR: maxRep=maxDef=0) |
| C-hint | 2542 | 806 | 1.58× | **no (skip-levels fast path)** |
| C-naive | 4010 | 511 | 1.00× | yes (full Dremel) |
| LIST | 4019 | 510 | 1.00× | yes (full Dremel, baseline) |

## rows=10000, zstd

| arm | ns/row | MB/s | ×base | levels decoded? |
|-----|-------:|-----:|------:|:----------------|
| A/FLBA | 8268 | 248 | 1.00× | no (FLBA, no levels on disk) |
| B/VECTOR | 6673 | 307 | 1.24× | no (VECTOR: maxRep=maxDef=0) |
| C-hint | 6680 | 307 | 1.24× | **no (skip-levels fast path)** |
| C-naive | 8266 | 248 | 1.00× | yes (full Dremel) |
| LIST | 8273 | 248 | 1.00× | yes (full Dremel, baseline) |

## rows=100000, uncompressed

| arm | ns/row | MB/s | ×base | levels decoded? |
|-----|-------:|-----:|------:|:----------------|
| A/FLBA | 2910 | 704 | 1.39× | no (FLBA, no levels on disk) |
| B/VECTOR | 2443 | 838 | 1.65× | no (VECTOR: maxRep=maxDef=0) |
| C-hint | 2580 | 794 | 1.57× | **no (skip-levels fast path)** |
| C-naive | 4017 | 510 | 1.01× | yes (full Dremel) |
| LIST | 4038 | 507 | 1.00× | yes (full Dremel, baseline) |

## rows=100000, zstd

| arm | ns/row | MB/s | ×base | levels decoded? |
|-----|-------:|-----:|------:|:----------------|
| A/FLBA | 7537 | 272 | 1.07× | no (FLBA, no levels on disk) |
| B/VECTOR | 6590 | 311 | 1.23× | no (VECTOR: maxRep=maxDef=0) |
| C-hint | 6714 | 305 | 1.21× | **no (skip-levels fast path)** |
| C-naive | 8050 | 254 | 1.01× | yes (full Dremel) |
| LIST | 8094 | 253 | 1.00× | yes (full Dremel, baseline) |


================================================================================
 3. C-HINT READER PATCH (apply on top of commit fcad93306040729d03f9d0c22ca682277cde3ae5)
    +51 lines in parquet/file/column_reader.go: an annotation-gated skip-levels
    fast path in determineNumToRead. Skips LevelDecoder.Decode (the O(n) RLE
    expansion); relies on SetData having already positioned the value decoder.
================================================================================

diff --git a/parquet/file/column_reader.go b/parquet/file/column_reader.go
index ba8eb2cd..c11dbc75 100644
--- a/parquet/file/column_reader.go
+++ b/parquet/file/column_reader.go
@@ -178,8 +178,31 @@ type columnChunkReader struct {
 	repLvlBuffer []int16
 
 	dictState dictionaryState
+
+	// fslSkipLevels is set when this reader was opened for a FIXED_SIZE_LIST
+	// logical-type-annotated column ("C-with-hint" arm). When true, the reader
+	// takes a fast path that does NOT expand the (physically present) rep/def
+	// levels into int16 arrays; it relies on the level decoders' SetData to
+	// position the value decoder past the level bytes and reads the values
+	// directly by stride arithmetic. Only valid when the annotation guarantees
+	// no nulls and a fixed multiplicity, i.e. every rep/def level is the max.
+	fslSkipLevels bool
 }
 
+// FSLSkipLevelsCalls counts how many times a FIXED_SIZE_LIST-hint reader took
+// the skip-levels fast path (i.e. decoded a batch of values WITHOUT expanding
+// rep/def levels). It is a global instrumentation counter used by the benchmark
+// harness to PROVE the fast path engaged rather than silently falling back to
+// the full Dremel decode. Not thread-safe; intended for single-threaded
+// benchmark/test use.
+var FSLSkipLevelsCalls int64
+
+// FSLLevelDecodeCalls counts how many times ANY column reader expanded rep/def
+// levels via LevelDecoder.Decode (the expensive O(n) path that the skip-levels
+// arm avoids). Used to demonstrate that C-naive/LIST decode levels while
+// C-hint/VECTOR do not. Not thread-safe.
+var FSLLevelDecodeCalls int64
+
 func newTypedColumnChunkReader(base columnChunkReader) ColumnChunkReader {
 	switch base.descr.PhysicalType() {
 	case parquet.Types.FixedLenByteArray:
@@ -250,6 +273,19 @@ func NewColumnReader(descr *schema.Column, pageReader PageReader, mem memory.All
 	return nil
 }
 
+// EnableFSLSkipLevels turns on the FIXED_SIZE_LIST-hint skip-levels fast path
+// for this reader ("C-with-hint" arm). The caller asserts, via the
+// FIXED_SIZE_LIST logical-type annotation, that the leaf column has a fixed
+// number of values per row and NO nulls -- i.e. every rep/def level is the max
+// level. Under that guarantee the reader can read values by count/stride and
+// skip expanding the (physically present) rep/def levels entirely.
+//
+// This is the exact analog of what PR #854's VECTOR repetition achieves
+// structurally (maxDef==maxRep==0 so determineNumToRead never decodes levels),
+// but gated on the annotation instead of a new physical repetition type, so the
+// on-disk file remains a standard backward-compatible 3-level LIST.
+func (c *columnChunkReader) EnableFSLSkipLevels() { c.fslSkipLevels = true }
+
 func (c *columnChunkReader) Err() error                    { return c.err }
 func (c *columnChunkReader) Type() parquet.Type            { return c.descr.PhysicalType() }
 func (c *columnChunkReader) Descriptor() *schema.Column    { return c.descr }
@@ -558,17 +594,32 @@ func (c *columnChunkReader) determineNumToRead(batchLen int64, defLvls, repLvls
 
 	size := utils.Min(batchLen, c.numBuffered-c.numDecoded)
 
+	// C-with-hint skip-levels fast path: the FIXED_SIZE_LIST annotation
+	// guarantees a fixed multiplicity with no nulls, so every definition level
+	// equals maxDef and there are no "missing" values. That means the number of
+	// physical values to read equals the number of level slots (size), and we
+	// do NOT need to expand the RLE levels into int16 arrays. The level
+	// decoders' SetData (called in initDataDecoder) already positioned the value
+	// decoder past the level bytes, so reading `size` values directly is
+	// correct and byte-identical to the full-decode result.
+	if c.fslSkipLevels {
+		FSLSkipLevelsCalls++
+		return 0, size, nil
+	}
+
 	if c.descr.MaxDefinitionLevel() > 0 {
 		if defLvls == nil {
 			defLvls = c.getDefLvlBuffer(size)
 		}
 		ndefs, toRead = c.readDefinitionLevels(defLvls[:size])
+		FSLLevelDecodeCalls++
 	} else {
 		toRead = size
 	}
 
 	if c.descr.MaxRepetitionLevel() > 0 && repLvls != nil {
 		nreps := c.readRepetitionLevels(repLvls[:size])
+		FSLLevelDecodeCalls++
 		if defLvls != nil && ndefs != nreps {
 			err = errors.New("parquet: number of decoded rep/def levels did not match")
 		}

================================================================================
 4. BENCHMARK HARNESS (drop into arrow-go as package fslbench/)
================================================================================

----- fslbench/harness.go -----
// Package fslbench is a self-contained benchmark that isolates WHERE the
// Parquet read speedup for fixed-size (ML embedding) list columns comes from.
//
// Thesis: the speedup comes ENTIRELY from skipping rep/def level decode, NOT
// from any clever physical encoding. Five arms are compared, all reading at the
// parquet/file column-chunk-reader layer:
//
//	A / FLBA     : one FIXED_LEN_BYTE_ARRAY per row (dim*4 bytes). No rep/def
//	               levels on disk. The dumbest possible no-levels encoding.
//	B / VECTOR   : PR #854's experimental FieldRepetitionType.VECTOR. The leaf
//	               has maxDef==maxRep==0, so no levels are decoded.
//	C-hint       : an ordinary 3-level LIST on disk (rep/def levels physically
//	               present) read through an annotation-AWARE reader that takes a
//	               skip-levels fast path (EnableFSLSkipLevels).
//	C-naive      : the SAME LIST bytes read through the stock Dremel decoder
//	               (levels expanded). This reproduces Rok's slow "C".
//	LIST         : plain FixedSizeList baseline (identical bytes to C), read with
//	               full level decode. The ×baseline reference.
//
// C-hint and C-naive/LIST read BYTE-IDENTICAL files; they differ only in whether
// the skip-levels flag is set. That makes this the purest possible isolation of
// level-decode cost.
package fslbench

import (
	"bytes"
	"encoding/binary"
	"fmt"
	"unsafe"

	"github.com/apache/arrow-go/v18/arrow"
	"github.com/apache/arrow-go/v18/arrow/array"
	"github.com/apache/arrow-go/v18/arrow/memory"
	"github.com/apache/arrow-go/v18/parquet"
	"github.com/apache/arrow-go/v18/parquet/compress"
	"github.com/apache/arrow-go/v18/parquet/file"
	"github.com/apache/arrow-go/v18/parquet/pqarrow"
)

// Arm identifies one of the five encodings/read-paths under test.
type Arm int

const (
	ArmFLBA   Arm = iota // A: FIXED_LEN_BYTE_ARRAY, no levels
	ArmVector            // B: PR #854 VECTOR repetition, no levels
	ArmCHint             // C-with-hint: annotated LIST + skip-levels reader
	ArmCNaive            // C-naive: annotated LIST + full Dremel decode
	ArmList              // plain FixedSizeList baseline (full decode)
)

func (a Arm) String() string {
	switch a {
	case ArmFLBA:
		return "A/FLBA"
	case ArmVector:
		return "B/VECTOR"
	case ArmCHint:
		return "C-hint"
	case ArmCNaive:
		return "C-naive"
	case ArmList:
		return "LIST"
	}
	return "?"
}

// Config parameterizes a benchmark file.
type Config struct {
	Dim          int  // embedding dimension (float32 elements per row)
	Rows         int  // number of embedding rows
	RowGroupSize int  // rows per row group
	Zstd         bool // compress with zstd (else uncompressed)
	Seed         uint64
}

func (c Config) codec() compress.Compression {
	if c.Zstd {
		return compress.Codecs.Zstd
	}
	return compress.Codecs.Uncompressed
}

// GenData produces deterministic uniform-random float32 embedding data:
// Rows*Dim values, row-major.
func GenData(c Config) []float32 {
	out := make([]float32, c.Rows*c.Dim)
	// splitmix64 -> uniform float32 in [0,1)
	x := c.Seed + 0x9e3779b97f4a7c15
	for i := range out {
		x += 0x9e3779b97f4a7c15
		z := x
		z = (z ^ (z >> 30)) * 0xbf58476d1ce4e5b9
		z = (z ^ (z >> 27)) * 0x94d049bb133111eb
		z = z ^ (z >> 31)
		out[i] = float32(z>>40) / float32(1<<24)
	}
	return out
}

// WriteFile builds an in-memory Parquet file for the given arm and data.
// data must be Rows*Dim float32 values, row-major.
//
// Note: ArmFLBA, ArmVector and ArmList/C produce genuinely different on-disk
// files (FLBA vs VECTOR vs LIST). ArmCHint, ArmCNaive and ArmList all produce
// the SAME on-disk LIST bytes (they differ only at read time), so WriteFile for
// any of those three returns identical bytes.
func WriteFile(c Config, arm Arm, data []float32) ([]byte, error) {
	switch arm {
	case ArmFLBA:
		return writeFLBA(c, data)
	case ArmVector:
		return writeArrowFSL(c, data, true /*asVector*/)
	default: // ArmCHint, ArmCNaive, ArmList -> plain LIST
		return writeArrowFSL(c, data, false)
	}
}

// writeArrowFSL writes a FixedSizeList<float32, Dim> column via pqarrow. When
// asVector is true the experimental VECTOR encoding (PR #854) is used; otherwise
// the standard backward-compatible 3-level LIST layout (with rep/def levels) is
// written.
//
// Each row group is written from an INDEPENDENTLY-BUILT (materialized) array
// covering exactly that row group's rows, rather than by handing pqarrow one big
// array and relying on MaxRowGroupLength to split it. That split path has a
// pre-existing arrow-go bug for nested (FixedSizeList) columns: WriteColumnData
// on a sliced FixedSizeList view ignores the child array's offset, so every row
// group after the first re-emits the child values from offset 0. Building each
// row group's array from scratch sidesteps that entirely and yields correct
// multi-row-group files (verified in tests). It does not affect what is being
// measured: the on-disk layout and read path are identical to a correct
// single-array write.
func writeArrowFSL(c Config, data []float32, asVector bool) ([]byte, error) {
	mem := memory.DefaultAllocator
	listSize := int32(c.Dim)
	elemField := arrow.Field{Name: "element", Type: arrow.PrimitiveTypes.Float32, Nullable: false}
	fslType := arrow.FixedSizeListOfField(listSize, elemField)
	arrowSchema := arrow.NewSchema([]arrow.Field{{Name: "embedding", Type: fslType, Nullable: false}}, nil)

	var buf bytes.Buffer
	props := parquet.NewWriterProperties(
		parquet.WithDictionaryDefault(false),
		parquet.WithMaxRowGroupLength(int64(c.RowGroupSize)),
		parquet.WithCompression(c.codec()),
		parquet.WithDataPageSize(1<<20),
	)
	ap := []pqarrow.WriterOption{}
	if asVector {
		ap = append(ap, pqarrow.WithVectorEncoding())
	}
	fw, err := pqarrow.NewFileWriter(arrowSchema, &buf, props, pqarrow.NewArrowWriterProperties(ap...))
	if err != nil {
		return nil, err
	}

	for start := 0; start < c.Rows; start += c.RowGroupSize {
		end := start + c.RowGroupSize
		if end > c.Rows {
			end = c.Rows
		}
		bldr := array.NewFixedSizeListBuilderWithField(mem, listSize, elemField)
		vb := bldr.ValueBuilder().(*array.Float32Builder)
		vb.Reserve((end - start) * c.Dim)
		for i := start; i < end; i++ {
			bldr.Append(true)
			vb.AppendValues(data[i*c.Dim:(i+1)*c.Dim], nil)
		}
		arr := bldr.NewArray()
		rec := array.NewRecordBatch(arrowSchema, []arrow.Array{arr}, int64(end-start))
		fw.NewRowGroup()
		if err := fw.WriteColumnData(rec.Column(0)); err != nil {
			rec.Release()
			arr.Release()
			bldr.Release()
			fw.Close()
			return nil, err
		}
		rec.Release()
		arr.Release()
		bldr.Release()
	}
	if err := fw.Close(); err != nil {
		return nil, err
	}
	return buf.Bytes(), nil
}

// writeFLBA writes each embedding as a single FIXED_LEN_BYTE_ARRAY of Dim*4
// bytes (the raw little-endian float32 payload), one value per row. There are no
// rep/def levels: the column is a Required primitive.
func writeFLBA(c Config, data []float32) ([]byte, error) {
	mem := memory.DefaultAllocator
	byteWidth := c.Dim * 4
	fsbType := &arrow.FixedSizeBinaryType{ByteWidth: byteWidth}
	arrowSchema := arrow.NewSchema([]arrow.Field{{Name: "embedding", Type: fsbType, Nullable: false}}, nil)

	bldr := array.NewFixedSizeBinaryBuilder(mem, fsbType)
	defer bldr.Release()
	rowBuf := make([]byte, byteWidth)
	for i := 0; i < c.Rows; i++ {
		row := data[i*c.Dim : (i+1)*c.Dim]
		for j, v := range row {
			binary.LittleEndian.PutUint32(rowBuf[j*4:], mathFloat32bits(v))
		}
		bldr.Append(rowBuf)
	}
	arr := bldr.NewArray()
	defer arr.Release()

	rec := array.NewRecordBatch(arrowSchema, []arrow.Array{arr}, int64(c.Rows))
	defer rec.Release()
	tbl := array.NewTableFromRecords(arrowSchema, []arrow.RecordBatch{rec})
	defer tbl.Release()

	var buf bytes.Buffer
	props := parquet.NewWriterProperties(
		parquet.WithDictionaryDefault(false),
		parquet.WithMaxRowGroupLength(int64(c.RowGroupSize)),
		parquet.WithCompression(c.codec()),
		parquet.WithDataPageSize(1<<20),
	)
	arrProps := pqarrow.NewArrowWriterProperties()
	if err := pqarrow.WriteTable(tbl, &buf, int64(c.RowGroupSize), props, arrProps); err != nil {
		return nil, err
	}
	return buf.Bytes(), nil
}

func mathFloat32bits(f float32) uint32 { return *(*uint32)(unsafe.Pointer(&f)) }

// ReadRowGroup reads column 0 of the given row group at the parquet/file
// column-chunk-reader layer and returns the flattened float32 values. For
// ArmCHint it enables the skip-levels fast path; for ArmCNaive/ArmList it passes
// level buffers so the full Dremel decode runs.
//
// It reuses the caller-provided scratch buffers to avoid per-call allocation in
// the benchmark hot loop.
func ReadRowGroup(rdr *file.Reader, rg int, arm Arm, dim int, scratch *Scratch) ([]float32, error) {
	rgr := rdr.RowGroup(rg)
	nrows := rgr.NumRows()
	ccr, err := rgr.Column(0)
	if err != nil {
		return nil, err
	}

	switch arm {
	case ArmFLBA:
		want := int(nrows) // one FLBA value per row
		scratch.ensureFLBA(want)
		cr := ccr.(*file.FixedLenByteArrayColumnChunkReader)
		total := 0
		for total < want {
			read, _, err := cr.ReadBatch(int64(want-total), scratch.flba[total:], nil, nil)
			if err != nil {
				return nil, err
			}
			if read == 0 {
				break
			}
			total += int(read)
		}
		// reinterpret each FLBA payload as dim float32s.
		scratch.ensureFloats(total * dim)
		out := scratch.floats[:0]
		for i := 0; i < total; i++ {
			b := scratch.flba[i]
			fs := unsafe.Slice((*float32)(unsafe.Pointer(&b[0])), dim)
			out = append(out, fs...)
		}
		return out, nil

	default: // Float32 leaf: VECTOR, C-hint, C-naive, LIST
		wantVals := int(nrows) * dim
		scratch.ensureFloats(wantVals)
		cr := ccr.(*file.Float32ColumnChunkReader)

		fullDecode := arm == ArmCNaive || arm == ArmList
		if arm == ArmCHint {
			// engage the annotation-aware skip-levels fast path.
			ccr.(interface{ EnableFSLSkipLevels() }).EnableFSLSkipLevels()
		}

		var defs, reps []int16
		if fullDecode {
			// C-naive / LIST: allocate level buffers so BOTH rep and def levels
			// are actually decoded (full Dremel), mirroring the stock reader.
			scratch.ensureLevels(wantVals)
			defs, reps = scratch.defs, scratch.reps
		}

		total := 0
		for total < wantVals {
			var d, r []int16
			if fullDecode {
				d = defs[total:]
				r = reps[total:]
			}
			_, read, err := cr.ReadBatch(int64(wantVals-total), scratch.floats[total:], d, r)
			if err != nil {
				return nil, err
			}
			if read == 0 {
				break
			}
			total += read
		}
		if total != wantVals {
			return nil, fmt.Errorf("arm %s rg %d: read %d values, want %d", arm, rg, total, wantVals)
		}
		return scratch.floats[:total], nil
	}
}

// Scratch holds reusable read buffers.
type Scratch struct {
	floats []float32
	defs   []int16
	reps   []int16
	flba   []parquet.FixedLenByteArray
}

func (s *Scratch) ensureFloats(n int) {
	if cap(s.floats) < n {
		s.floats = make([]float32, n)
	}
	s.floats = s.floats[:n]
}
func (s *Scratch) ensureLevels(n int) {
	if cap(s.defs) < n {
		s.defs = make([]int16, n)
		s.reps = make([]int16, n)
	}
	s.defs = s.defs[:n]
	s.reps = s.reps[:n]
}
func (s *Scratch) ensureFLBA(n int) {
	if cap(s.flba) < n {
		s.flba = make([]parquet.FixedLenByteArray, n)
	}
	s.flba = s.flba[:n]
}

// OpenReader builds a file.Reader over the given bytes.
func OpenReader(data []byte) (*file.Reader, error) {
	return file.NewParquetReader(bytes.NewReader(data))
}

----- fslbench/harness_test.go -----
package fslbench

import (
	"testing"

	"github.com/apache/arrow-go/v18/parquet/file"
)

var allArms = []Arm{ArmFLBA, ArmVector, ArmCHint, ArmCNaive, ArmList}

// readAll reads every row group of a file for an arm and concatenates values.
func readAll(t *testing.T, data []byte, arm Arm, dim int) []float32 {
	t.Helper()
	rdr, err := OpenReader(data)
	if err != nil {
		t.Fatalf("open %s: %v", arm, err)
	}
	defer rdr.Close()
	var out []float32
	var sc Scratch
	for rg := 0; rg < rdr.NumRowGroups(); rg++ {
		vals, err := ReadRowGroup(rdr, rg, arm, dim, &sc)
		if err != nil {
			t.Fatalf("read %s rg %d: %v", arm, rg, err)
		}
		out = append(out, vals...)
	}
	return out
}

// TestCrossArmCorrectness proves all five arms round-trip the SAME float values.
// A fast arm that returns wrong data is worthless.
func TestCrossArmCorrectness(t *testing.T) {
	cfg := Config{Dim: 512, Rows: 5000, RowGroupSize: 4096, Seed: 42}
	data := GenData(cfg)

	ref := data // ground truth: what we wrote, row-major

	for _, zstd := range []bool{false, true} {
		cfg.Zstd = zstd
		for _, arm := range allArms {
			raw, err := WriteFile(cfg, arm, data)
			if err != nil {
				t.Fatalf("write %s zstd=%v: %v", arm, zstd, err)
			}
			got := readAll(t, raw, arm, cfg.Dim)
			if len(got) != len(ref) {
				t.Fatalf("%s zstd=%v: len %d want %d", arm, zstd, len(got), len(ref))
			}
			for i := range ref {
				if got[i] != ref[i] {
					t.Fatalf("%s zstd=%v: value[%d]=%v want %v", arm, zstd, i, got[i], ref[i])
				}
			}
			t.Logf("%s zstd=%v: %d values OK", arm, zstd, len(got))
		}
	}
}

// TestCHintEqualsCNaiveByteForByte is the sharpest correctness check: C-hint
// (skip-levels) and C-naive (full Dremel decode) read the EXACT SAME on-disk
// bytes and must return identical values.
func TestCHintEqualsCNaiveByteForByte(t *testing.T) {
	cfg := Config{Dim: 512, Rows: 5000, RowGroupSize: 4096, Seed: 7}
	data := GenData(cfg)

	// C-hint, C-naive, LIST all serialize to identical on-disk bytes.
	hintBytes, err := WriteFile(cfg, ArmCHint, data)
	if err != nil {
		t.Fatal(err)
	}
	naiveBytes, err := WriteFile(cfg, ArmCNaive, data)
	if err != nil {
		t.Fatal(err)
	}
	if !bytesEqual(hintBytes, naiveBytes) {
		t.Fatalf("C-hint and C-naive must write byte-identical files (hint=%d naive=%d bytes)", len(hintBytes), len(naiveBytes))
	}

	hint := readAll(t, hintBytes, ArmCHint, cfg.Dim)
	naive := readAll(t, naiveBytes, ArmCNaive, cfg.Dim)
	if len(hint) != len(naive) {
		t.Fatalf("len mismatch: hint=%d naive=%d", len(hint), len(naive))
	}
	for i := range hint {
		if hint[i] != naive[i] {
			t.Fatalf("value[%d]: hint=%v naive=%v", i, hint[i], naive[i])
		}
	}
	t.Logf("C-hint == C-naive byte-for-byte over %d values; files identical (%d bytes)", len(hint), len(hintBytes))
}

// TestSkipPathEngaged proves the C-hint reader actually took the skip-levels
// fast path (counter increments) while C-naive/LIST expanded levels
// (FSLLevelDecodeCalls increments), and B/VECTOR and A/FLBA decode no levels.
func TestSkipPathEngaged(t *testing.T) {
	cfg := Config{Dim: 512, Rows: 8192, RowGroupSize: 4096, Seed: 3}
	data := GenData(cfg)

	measure := func(arm Arm) (skip, lvl int64) {
		raw, err := WriteFile(cfg, arm, data)
		if err != nil {
			t.Fatal(err)
		}
		file.FSLSkipLevelsCalls = 0
		file.FSLLevelDecodeCalls = 0
		_ = readAll(t, raw, arm, cfg.Dim)
		return file.FSLSkipLevelsCalls, file.FSLLevelDecodeCalls
	}

	// C-hint: skip path MUST engage, and NO levels may be decoded.
	if skip, lvl := measure(ArmCHint); skip == 0 || lvl != 0 {
		t.Fatalf("C-hint: expected skip>0 and lvl==0, got skip=%d lvl=%d", skip, lvl)
	} else {
		t.Logf("C-hint: skip-path taken %d times, 0 level-decode calls (PROVEN skip)", skip)
	}

	// C-naive: full Dremel decode -> level-decode calls > 0, no skip.
	if skip, lvl := measure(ArmCNaive); skip != 0 || lvl == 0 {
		t.Fatalf("C-naive: expected skip==0 and lvl>0, got skip=%d lvl=%d", skip, lvl)
	} else {
		t.Logf("C-naive: %d level-decode calls, 0 skips (full Dremel, as Rok measured)", lvl)
	}

	// LIST baseline: same as C-naive -> decodes levels.
	if skip, lvl := measure(ArmList); skip != 0 || lvl == 0 {
		t.Fatalf("LIST: expected skip==0 and lvl>0, got skip=%d lvl=%d", skip, lvl)
	} else {
		t.Logf("LIST: %d level-decode calls (full Dremel baseline)", lvl)
	}

	// B/VECTOR: leaf has maxDef==maxRep==0, so NO levels decoded structurally.
	if skip, lvl := measure(ArmVector); lvl != 0 {
		t.Fatalf("B/VECTOR: expected 0 level-decode calls, got lvl=%d (skip=%d)", lvl, skip)
	} else {
		t.Logf("B/VECTOR: 0 level-decode calls (maxDef==maxRep==0 structurally)")
	}

	// A/FLBA: Required primitive, no levels.
	if _, lvl := measure(ArmFLBA); lvl != 0 {
		t.Fatalf("A/FLBA: expected 0 level-decode calls, got lvl=%d", lvl)
	} else {
		t.Logf("A/FLBA: 0 level-decode calls (Required primitive)")
	}
}

// TestBackwardCompatibleOnDisk proves the C/LIST file is a plain 3-level LIST
// with rep/def levels physically present (maxDef>0), i.e. an annotation-unaware
// reader reads it correctly. This is the whole backward-compat point of arm C.
func TestBackwardCompatibleOnDisk(t *testing.T) {
	cfg := Config{Dim: 512, Rows: 4096, RowGroupSize: 4096, Seed: 1}
	data := GenData(cfg)
	raw, err := WriteFile(cfg, ArmCNaive, data)
	if err != nil {
		t.Fatal(err)
	}
	rdr, err := OpenReader(raw)
	if err != nil {
		t.Fatal(err)
	}
	defer rdr.Close()
	descr := rdr.MetaData().Schema.Column(0)
	if descr.MaxRepetitionLevel() == 0 || descr.MaxDefinitionLevel() == 0 {
		t.Fatalf("C/LIST leaf should have rep/def levels on disk, got maxRep=%d maxDef=%d",
			descr.MaxRepetitionLevel(), descr.MaxDefinitionLevel())
	}
	if descr.InVectorColumn() {
		t.Fatalf("C/LIST must NOT be a physical VECTOR column")
	}
	t.Logf("C/LIST on-disk: maxRep=%d maxDef=%d, InVectorColumn=%v (standard backward-compatible LIST)",
		descr.MaxRepetitionLevel(), descr.MaxDefinitionLevel(), descr.InVectorColumn())

	// And B/VECTOR: leaf has no levels on disk.
	rawV, err := WriteFile(cfg, ArmVector, data)
	if err != nil {
		t.Fatal(err)
	}
	rdrV, err := OpenReader(rawV)
	if err != nil {
		t.Fatal(err)
	}
	defer rdrV.Close()
	dv := rdrV.MetaData().Schema.Column(0)
	if dv.MaxRepetitionLevel() != 0 || dv.MaxDefinitionLevel() != 0 || !dv.InVectorColumn() {
		t.Fatalf("B/VECTOR leaf should have maxRep==maxDef==0 and InVectorColumn, got maxRep=%d maxDef=%d vec=%v",
			dv.MaxRepetitionLevel(), dv.MaxDefinitionLevel(), dv.InVectorColumn())
	}
	t.Logf("B/VECTOR on-disk: maxRep=%d maxDef=%d InVectorColumn=%v (no levels; forward-INCOMPATIBLE)",
		dv.MaxRepetitionLevel(), dv.MaxDefinitionLevel(), dv.InVectorColumn())
}

func bytesEqual(a, b []byte) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

----- fslbench/bench_test.go -----
package fslbench

import (
	"fmt"
	"math/rand"
	"testing"

	"github.com/apache/arrow-go/v18/parquet/file"
)

// numProbes mirrors Rok's harness, which read via read_row_group with 32 random
// row-group probes per measured read.
const numProbes = 32

// benchOne runs the READ-DECODE-ONLY benchmark for a single (config, arm): the
// file is written ONCE in setup, then the timed loop opens the file and reads
// numProbes randomly chosen row groups, decoding all values of column 0.
//
// b.SetBytes is set to the number of embedding value BYTES touched per op
// (probes * avg-rows-per-probe * dim * 4) so `go test` reports MB/s.
func benchOne(b *testing.B, cfg Config, arm Arm) {
	data := GenData(cfg)
	raw, err := WriteFile(cfg, arm, data)
	if err != nil {
		b.Fatal(err)
	}

	// Determine row-group count and total value-bytes per probe pass.
	rdr, err := OpenReader(raw)
	if err != nil {
		b.Fatal(err)
	}
	nrg := rdr.NumRowGroups()
	rgRows := make([]int64, nrg)
	var totalRows int64
	for i := 0; i < nrg; i++ {
		rgRows[i] = rdr.RowGroup(i).NumRows()
		totalRows += rgRows[i]
	}
	rdr.Close()

	// Pre-pick the probe sequence so every arm reads the SAME row groups.
	rng := rand.New(rand.NewSource(int64(cfg.Seed) + 0x1234))
	probes := make([]int, numProbes)
	var bytesPerPass int64
	for i := range probes {
		probes[i] = rng.Intn(nrg)
		bytesPerPass += rgRows[probes[i]] * int64(cfg.Dim) * 4
	}

	var sc Scratch
	b.SetBytes(bytesPerPass)
	b.ReportAllocs()
	b.ResetTimer()
	for n := 0; n < b.N; n++ {
		// Reopen per op: metadata parse is cheap and constant across arms, and
		// this matches a "probe reads" workload where each read starts fresh.
		rdr, err := OpenReader(raw)
		if err != nil {
			b.Fatal(err)
		}
		for _, rg := range probes {
			if _, err := ReadRowGroup(rdr, rg, arm, cfg.Dim, &sc); err != nil {
				b.Fatal(err)
			}
		}
		rdr.Close()
	}
	b.StopTimer()

	// Attach ns/row as a custom metric (rows decoded per op = totalRows-probe).
	var rowsPerPass int64
	for _, rg := range probes {
		rowsPerPass += rgRows[rg]
	}
	b.ReportMetric(float64(b.Elapsed().Nanoseconds())/float64(b.N)/float64(rowsPerPass), "ns/row")
}

// headline configs: dim=512 f32, 100k rows, rg=4096. Also 1k and 10k for the
// table. Uncompressed isolates level-decode cost; zstd shows it masks the delta.
func benchConfigs() []Config {
	var out []Config
	for _, rows := range []int{1000, 10000, 100000} {
		for _, zstd := range []bool{false, true} {
			out = append(out, Config{Dim: 512, Rows: rows, RowGroupSize: 4096, Zstd: zstd, Seed: 42})
		}
	}
	return out
}

func BenchmarkRead(b *testing.B) {
	for _, cfg := range benchConfigs() {
		codec := "uncompressed"
		if cfg.Zstd {
			codec = "zstd"
		}
		for _, arm := range allArms {
			name := fmt.Sprintf("rows=%d/%s/%s", cfg.Rows, codec, arm)
			b.Run(name, func(b *testing.B) {
				// reset instrumentation so a curious -v run shows per-bench counts
				file.FSLSkipLevelsCalls = 0
				file.FSLLevelDecodeCalls = 0
				benchOne(b, cfg, arm)
			})
		}
	}
}

// dimSweepDims are the embedding widths for the dimension sweep. The point is to
// show that the small residual gap between B/VECTOR and C-hint is FIXED-PER-PAGE
// (per-page rep/def-decoder SetData that the real-LIST C-hint pays and the
// no-levels VECTOR does not), so it amortizes away as the vector gets wider.
var dimSweepDims = []int{512, 1024, 2048, 4096}

// BenchmarkDimSweep runs all arms at rows=100k, uncompressed, across a range of
// embedding dimensions. Row-group size is fixed at 4096. Used to produce the
// C-hint-vs-B gap-vs-dim table.
func BenchmarkDimSweep(b *testing.B) {
	for _, dim := range dimSweepDims {
		cfg := Config{Dim: dim, Rows: 100000, RowGroupSize: 4096, Zstd: false, Seed: 42}
		for _, arm := range allArms {
			name := fmt.Sprintf("dim=%d/uncompressed/%s", dim, arm)
			b.Run(name, func(b *testing.B) {
				file.FSLSkipLevelsCalls = 0
				file.FSLLevelDecodeCalls = 0
				benchOne(b, cfg, arm)
			})
		}
	}
}

================================================================================
 NOTES / HONEST CAVEATS
================================================================================
- A/FLBA is marginally slower than B/C-hint despite having no levels: FLBA stores
  opaque bytes, so materializing floats needs a bytes->float reinterpret that
  VECTOR/LIST avoid. Does not affect the fast-band/slow-band separation.
- The C-hint vs B residual is ~0.8% (within noise of ~2 sigma at dim=512); it is
  fixed per-page overhead from reading a real LIST layout, and shrinks with vector
  width. Reported as 'within 1%'.
- All five arms were verified to round-trip byte-identical float values; C-hint's
  output equals C-naive's (same on-disk bytes, different reader). The skip path is
  proven engaged via counters (C-hint: 0 level-decode calls; C-naive/LIST: >0).