diff --git a/.github/workflows/release-all-tools.yml b/.github/workflows/release-all-tools.yml
index 1e9c70f54..2825b0a90 100644
--- a/.github/workflows/release-all-tools.yml
+++ b/.github/workflows/release-all-tools.yml
@@ -3,6 +3,7 @@ name: Combined release tools
on:
pull_request:
paths:
+ - 'matrix-parquet/**'
- 'releaseAll.sh'
- 'scripts/release-all/**'
- 'matrix-bom/verifyBomApi.sh'
@@ -73,3 +74,26 @@ jobs:
run: |
BOM_VERIFY_REPO="$GITHUB_WORKSPACE/build/releaseAll/bom-verify" bash matrix-bom/verifyBomApi.sh
groovy scripts/release-all/test-manifest-verifier.groovy . build/releaseAll/bom-verify
+
+ parquet-interop:
+ runs-on: ubuntu-latest
+ timeout-minutes: 30
+ steps:
+ - uses: actions/checkout@v4
+ - uses: actions/setup-java@v4
+ with:
+ distribution: temurin
+ java-version: '21'
+ - uses: actions/setup-python@v5
+ with:
+ python-version: '3.11'
+ - uses: gradle/actions/setup-gradle@v4
+ - run: python -m pip install pyarrow==21.0.0 huggingface_hub==2.2.0
+ - run: python matrix-parquet/src/test/python/generate_optimized.py
+ - run: python matrix-parquet/src/test/python/generate_hub_urls.py
+ - run: git diff --exit-code -- matrix-parquet/src/test/resources/optimized
+ - run: ./gradlew :matrix-parquet:codenarcMain :matrix-parquet:codenarcTest :matrix-parquet:spotlessCheck :matrix-parquet:test :matrix-parquet:compileConsumerApi -PrunSlowTests=true
+ env:
+ PARQUET_PYTHON: python
+ - run: python matrix-parquet/src/test/python/verify_matrix_output.py
+ - run: ./gradlew :matrix-parquet:test --tests ParquetInteropTest --rerun-tasks
diff --git a/docs/cookbook/matrix-parquet.md b/docs/cookbook/matrix-parquet.md
index a7c6ba5b3..41ffdf6b8 100644
--- a/docs/cookbook/matrix-parquet.md
+++ b/docs/cookbook/matrix-parquet.md
@@ -158,3 +158,29 @@ For files with no Matrix metadata at all, Parquet `DECIMAL` columns are inferred
---
[Back to index](cookbook.md)
+
+
+## Optimized Parquet, projection and bounded reads (0.8.0)
+
+```groovy
+import se.alipsa.matrix.parquet.MatrixParquetReader
+import se.alipsa.matrix.parquet.Predicates
+
+def subset = MatrixParquetReader.builder().columns(['id', 'text'])
+ .filter(Predicates.ge('id', 1000L)).limit(100)
+ .read(new File('downloaded-optimized.parquet'))
+
+MatrixParquetReader.builder().maxBatchRows(500).batches(new File('downloaded-optimized.parquet'))
+ .withCloseable { batches ->
+ while (batches.hasNext()) {
+ def batch = batches.next()
+ // Process this Matrix before requesting another batch.
+ }
+ }
+```
+
+CDC pages decode as standard Parquet; page indexes help eligible predicates skip pages. These APIs
+include exact logical-schema predicate conversion and retain selected types for empty results.
+See the [complete module guide](../../matrix-parquet/readme.md) for all parameter defaults,
+HTTP/Hub access, sharded datasets, writer tuning, fixture reproduction and the external CDC rewrite.
+Read the [0.8.0 migration table](../../matrix-parquet/release.md) before upgrading existing calls.
diff --git a/docs/tutorial/11-matrix-parquet.md b/docs/tutorial/11-matrix-parquet.md
index a11261085..2850a1e6f 100644
--- a/docs/tutorial/11-matrix-parquet.md
+++ b/docs/tutorial/11-matrix-parquet.md
@@ -261,3 +261,29 @@ The module uses Apache Parquet and Hadoop libraries directly:
In the next section, we'll explore the matrix-bigquery module, which provides functionality for interacting with Google BigQuery.
Go to [previous section](10-matrix-bom.md) | Go to [next section](12-matrix-bigquery.md) | Back to [outline](outline.md)
+
+
+## Optimized Parquet, projection and bounded reads (0.8.0)
+
+```groovy
+import se.alipsa.matrix.parquet.MatrixParquetReader
+import se.alipsa.matrix.parquet.Predicates
+
+def subset = MatrixParquetReader.builder().columns(['id', 'text'])
+ .filter(Predicates.ge('id', 1000L)).limit(100)
+ .read(new File('downloaded-optimized.parquet'))
+
+MatrixParquetReader.builder().maxBatchRows(500).batches(new File('downloaded-optimized.parquet'))
+ .withCloseable { batches ->
+ while (batches.hasNext()) {
+ def batch = batches.next()
+ // Process this Matrix before requesting another batch.
+ }
+ }
+```
+
+CDC pages decode as standard Parquet; page indexes help eligible predicates skip pages. These APIs
+include exact logical-schema predicate conversion and retain selected types for empty results.
+See the [complete module guide](../../matrix-parquet/readme.md) for all parameter defaults,
+HTTP/Hub access, sharded datasets, writer tuning, fixture reproduction and the external CDC rewrite.
+Read the [0.8.0 migration table](../../matrix-parquet/release.md) before upgrading existing calls.
diff --git a/matrix-bom/bom.xml b/matrix-bom/bom.xml
index c34d6cc9a..5202867c5 100644
--- a/matrix-bom/bom.xml
+++ b/matrix-bom/bom.xml
@@ -46,7 +46,7 @@
2.3.3-SNAPSHOT0.1.1-SNAPSHOT0.1.3-SNAPSHOT
- 0.7.1-SNAPSHOT
+ 0.8.0-SNAPSHOT0.7.0-SNAPSHOT2.5.1-SNAPSHOT2.5.1-SNAPSHOT
diff --git a/matrix-parquet/benchmarks/optimized-parquet.md b/matrix-parquet/benchmarks/optimized-parquet.md
new file mode 100644
index 000000000..f7591f476
--- /dev/null
+++ b/matrix-parquet/benchmarks/optimized-parquet.md
@@ -0,0 +1,145 @@
+# Optimized Parquet measurements
+
+Measured 2026-10-17 on macOS arm64 (Apple M5 Max), G1 GC, OpenJDK 21.0.12.1, 4 GiB test heap, parquet-java 1.18.1,
+SNAPPY, default 128 MiB row-group target and 1 MiB pages. Each workload has 1,000,000 rows,
+an Integer ID and six String columns. Wide text repeats 1,000 labels per column; categorical
+repeats 12. Mixed adds LocalDate and Map(label, exact BigDecimal) columns. Generated inputs
+live in build/, outside git. The compared Group loop uses shared corrected value conversion.
+
+| Workload | File bytes | Group median ms | Columnar median ms | Lower median time | Group retained bytes | Columnar retained bytes |
+|---|---:|---:|---:|---:|---:|---:|
+| wide | 4,898,180 | 7,990 | 767 | 90.4% | 381,674,424 | 45,751,352 |
+| categorical | 4,173,171 | 7,940 | 775 | 90.2% | 342,270,976 | 45,364,064 |
+| mixed/nested | 7,934,477 | 18,569 | 2,462 | 86.7% | 790,769,704 | 406,370,144 |
+
+Medians exclude run 0; three warmed runs are included. Group warmed elapsed ranges were
+7,368–10,150 / 6,845–11,093 / 18,334–21,881 ms. Columnar ranges were 765–782 / 763–783 /
+2,332–2,486 ms. One codec-spike build overlapped the wide Group run and contributes to its
+variance; these are local development measurements, not universal speed guarantees. The
+original pre-change reader baseline was separately captured before production edits at
+5,633 / 5,413 / 5,433 / 5,424 ms on the same million-row wide shape (file 4,898,111 bytes).
+
+Current-thread allocated bytes were approximately 13.83 GB versus 1.32 GB (wide), 13.78 GB
+versus 1.31 GB (categorical), and 38.53 GB versus 3.52 GB (mixed). Retained deltas are sampled
+after explicit full GC with the result live. They measure retained result heap, not peak RSS
+or a rigorously sampled peak-live-heap metric. In particular the categorical retained result
+uses 86.7% less heap; do not relabel that measurement as peak heap. No OS cache purge was
+performed. Run 0 represents first application access, not a guaranteed cold filesystem cache.
+Timing thresholds are absent from correctness CI.
+
+## Confirmation with GC live-heap sampling
+
+A second million-row run added GC notifications. The sampler records the maximum sum of heap
+pool usage **after GC** during each read, including the final full GC with the result live.
+This is sampled peak live heap, including benchmark inputs/buffers, not instantaneous heap
+occupancy between GCs or process RSS. Medians below exclude run 0 (three warmed runs).
+
+| Workload | Group ms | Columnar ms | Group sampled live peak bytes | Columnar sampled live peak bytes |
+|---|---:|---:|---:|---:|
+| wide | 7,280 | 764 | 1,009,640,184 | 584,545,248 |
+| categorical | 7,158 | 797 | 963,066,680 | 541,390,488 |
+| mixed/nested | 18,936 | 2,238 | 1,998,185,928 | 1,343,803,744 |
+
+Wide elapsed time is 89.5% lower; categorical sampled live heap is 43.8% lower; mixed elapsed
+time is 88.2% lower. Thus the initial 20% elapsed / 25% live-heap / at-most-5% regression targets
+are met on these workloads with this stated sampling method. Warmed elapsed ranges were
+7,230–7,295 / 6,922–7,263 / 18,605–19,304 ms for Group and 752–786 / 776–800 / 2,213–2,306 ms
+for columnar. Allocations were approximately 14.20/1.32, 14.15/1.31 and 40.16/3.40 GB respectively.
+These local measurements remain evidence, not performance guarantees or CI timing assertions.
+
+## Reproduction
+
+Create an init file outside the repository:
+
+```groovy
+// /tmp/parquet-benchmark.init.gradle
+gradle.projectsEvaluated {
+ gradle.rootProject.project(':matrix-parquet').tasks.named('test') {
+ systemProperty 'parquet.baseline', 'true'
+ maxHeapSize = '4g'
+ }
+}
+```
+
+```bash
+./gradlew :matrix-parquet:test --tests '*ParquetReadBenchmarkTest' \
+ -I /tmp/parquet-benchmark.init.gradle --no-configuration-cache --console=plain
+```
+
+The BENCH records include workload, implementation, run, rows, file bytes, elapsed milliseconds,
+thread allocation and retained heap. Avoid concurrent builds for isolated timing measurements.
+
+## Correctness and I/O evidence
+
+`ParquetPruningTest` inspects actual offset/column indexes and multiple pages, compares indexed
+and nonindexed byte counts with row-group/statistics/dictionary/bloom pruning disabled, and
+checks dictionary String identity. `ParquetDatasetAndHttpTest` counts transferred range bytes
+for narrow projection and selective filtering, checks fallback and ETag changes, and compares
+ordered sequential/parallel shard results. These assertions use bytes/results, not wall clocks.
+
+The no-Hadoop spike (`./gradlew :matrix-parquet:codecWithoutHadoop`) found that SNAPPY, ZSTD,
+GZIP and UNCOMPRESSED writer initialization all require
+`org/apache/hadoop/io/compress/zlib/ZlibCompressor$CompressionLevel`. Therefore Hadoop remains
+on the runtime classpath even though local InputFile/OutputFile use PlainParquetConfiguration.
+
+The public Java consumer probe (`./gradlew :matrix-parquet:compileConsumerApi`) removes
+parquet-java and Hadoop from its compile classpath and successfully compiles projection,
+Predicates, limits and writer tuning. This verifies that new public predicates do not require
+consumers to add parquet-column explicitly.
+
+## Bounded dataset parallelism
+
+Four shards of 250,000 rows each (Integer ID and 1,000 categorical text labels), SNAPPY,
+maxBatchRows=65,536, all 1,000,000 rows materialized in deterministic order:
+
+| Workers | Warm ms (runs 1–3) | Median ms | Median sampled live peak bytes |
+|---|---|---:|---:|
+| 1 | 370 / 341 / 308 | 341 | 66,905,816 |
+| 4 | 247 / 242 / 239 | 242 | 57,235,736 |
+
+This local workload gains about 29% median throughput time with four workers. Heap samples vary
+with collection timing; the measured queue stays bounded to one batch per worker, not a claim
+that parallel reads always use less memory. The benchmark verifies first/last IDs and row count.
+Run `*ParquetDatasetBenchmarkTest` with the same init setup but system property
+`parquet.datasetBenchmark=true` instead of `parquet.baseline`.
+
+Observed deterministic I/O checks on the controlled fixtures: page-index reads 2,914 bytes
+versus 82,700 without page pruning; bloom-filter reads 16,737 versus 40,373 with statistics
+unable to exclude the interior key and dictionaries/pages disabled; selective HTTP transferred
+88,476 versus a 906,195-byte full object. HTTP data uses generated UUID strings, so exact byte
+counts vary; assertions require reduced transfer and exact returned rows, not these constants.
+
+## Identical Arrow layouts: full scan, projection and predicate
+
+PyArrow 21.0.0 generated 1,000,000 sorted int64 IDs and six text columns (1,000 repeated labels,
+about 140 characters/label), four 250,000-row groups, SNAPPY, 64 KiB target pages, default
+dictionary encoding and default CDC sizes. Only CDC/index settings differ. No OS cache purge;
+run 0 is warmup, medians include runs 1–3. Heap is the same GC live sample defined above.
+
+| Layout | Mode | Returned rows | File bytes | Read bytes (footer included) | Median ms | Median live peak bytes |
+|---|---|---:|---:|---:|---:|---:|
+| ordinary | full | 1,000,000 | 6,066,382 | 6,066,378 | 727 | 110,855,320 |
+| ordinary | id projection | 1,000,000 | 6,066,382 | 5,085,954 | 214 | 55,266,648 |
+| ordinary | id >= 999900 | 100 | 6,066,382 | 1,279,020 | 99 | 32,019,888 |
+| indexed | full | 1,000,000 | 6,069,245 | 6,021,834 | 741 | 117,800,072 |
+| indexed | id projection | 1,000,000 | 6,069,245 | 5,082,882 | 226 | 56,010,848 |
+| indexed | id >= 999900 | 100 | 6,069,245 | 552,869 | 7 | 27,740,176 |
+| CDC + indexed | full | 1,000,000 | 7,250,428 | 6,968,991 | 693 | 111,151,624 |
+| CDC + indexed | id projection | 1,000,000 | 7,250,428 | 5,105,311 | 209 | 57,589,672 |
+| CDC + indexed | id >= 999900 | 100 | 7,250,428 | 542,383 | 3 | 27,766,232 |
+
+The selective read benefits from row-group pruning in every layout and page pruning where
+indexes exist. These short timings demonstrate this workload, not a guaranteed CDC speedup;
+CDC has a larger file here. The independent page-index test isolates page pruning alone.
+
+Reproduce with pinned PyArrow and the same init script as above, replacing its system property
+with `parquet.layoutBenchmark=true`:
+
+```bash
+/tmp/matrix-parquet-fixtures/bin/python matrix-parquet/src/test/python/generate_benchmarks.py
+./gradlew :matrix-parquet:test --tests '*ParquetLayoutBenchmarkTest' \
+ -I /tmp/parquet-layout.init.gradle --no-configuration-cache --console=plain
+```
+
+The generator writes only build/layout-benchmarks. The optional test checks returned counts and
+last IDs and emits per-run time, exact input bytes and sampled live heap for every combination.
diff --git a/matrix-parquet/build.gradle b/matrix-parquet/build.gradle
index 177374de0..094566f83 100644
--- a/matrix-parquet/build.gradle
+++ b/matrix-parquet/build.gradle
@@ -7,7 +7,7 @@ plugins {
}
group = 'se.alipsa.matrix'
-version = '0.7.1-SNAPSHOT'
+version = '0.8.0-SNAPSHOT'
description = "Groovy library for importing and exporting parquet files to and from a Matrix"
JavaCompile javaCompile = compileJava {
@@ -124,3 +124,48 @@ nexusReleasePlugin {
password = project.ext.properties.sonatypePassword
mavenPublication = publishing.publications.maven
}
+
+// Generate the default-size CDC case outside the repository only for the slow lane.
+boolean runCdcSlowTests = providers.gradleProperty('runSlowTests').orNull == 'true'
+File parquetProjectDirectory = projectDir
+tasks.register('generateDefaultCdcFixture') {
+ onlyIf { runCdcSlowTests }
+ doLast {
+ String python = System.getenv('PARQUET_PYTHON') ?: 'python3'
+ try {
+ def probe = new ProcessBuilder(python, '-c', 'import pyarrow; assert pyarrow.__version__ == "21.0.0"').inheritIO().start()
+ if (probe.waitFor() != 0) {
+ logger.lifecycle('SKIP default CDC fixture: pinned PyArrow 21.0.0 is unavailable')
+ return
+ }
+ } catch (IOException ignored) {
+ logger.lifecycle('SKIP default CDC fixture: Python is unavailable')
+ return
+ }
+ def generation = new ProcessBuilder(python, 'src/test/python/generate_optimized.py',
+ 'build/default-cdc', '--default-cdc').directory(parquetProjectDirectory).inheritIO().start()
+ if (generation.waitFor() != 0) { throw new GradleException('CDC fixture generation failed') }
+ }
+}
+tasks.named('test') { dependsOn 'generateDefaultCdcFixture' }
+
+// Diagnostic spike: report runtime classes still required by upstream codec/writer code.
+tasks.register('codecWithoutHadoop', Test) {
+ testClassesDirs = sourceSets.test.output.classesDirs
+ classpath = sourceSets.test.runtimeClasspath.filter { !it.name.startsWith('hadoop-') }
+ useJUnitPlatform()
+ include '**/ParquetCodecProbeTest.class'
+ systemProperty 'parquet.codecSpike', 'true'
+ testLogging.showStandardStreams = true
+}
+
+tasks.register('compileConsumerApi', JavaCompile) {
+ dependsOn classes
+ source = fileTree('src/consumerTest/java') { include '**/*.java' }
+ classpath = sourceSets.test.compileClasspath.filter {
+ !it.name.startsWith('parquet-') && !it.name.startsWith('hadoop-')
+ } + sourceSets.main.output
+ destinationDirectory = layout.buildDirectory.dir('consumer-api')
+ options.release = 21
+}
+tasks.named('check') { dependsOn 'compileConsumerApi' }
diff --git a/matrix-parquet/readme.md b/matrix-parquet/readme.md
index 9ddfcd525..9d32c5c9c 100644
--- a/matrix-parquet/readme.md
+++ b/matrix-parquet/readme.md
@@ -1,340 +1,244 @@
-# Matrix-Parquet
+# Matrix Parquet
-This module enables import and export of [Apache Parquet](https://parquet.apache.org/) files to and from Matrix objects using `MatrixParquetWriter` and `MatrixParquetReader`.
+Read and write Apache Parquet from Groovy/Java Matrix applications on JDK 21. This development
+version targets **0.8.0** and uses parquet-java 1.18.1. Include `matrix-core` and Groovy alongside
+`matrix-parquet` in your application. To test this checkout, run `./gradlew :matrix-parquet:test`.
-## At a Glance
+## Optimized Parquet and selective reads
-| Goal | Entry point |
-|------|-------------|
-| Read a file | `MatrixParquetReader.builder().read(file)` |
-| Write a file | `MatrixParquetWriter.builder(matrix).write(file)` |
-| Read via SPI | `Matrix.read(file)` |
-| Write via SPI | `matrix.write(file)` |
-| Discover read options | `Matrix.listReadOptions('parquet')` / `ParquetReadOptions.describe()` |
-| Discover write options | `Matrix.listWriteOptions('parquet')` / `ParquetWriteOptions.describe()` |
+Hugging Face ordinary and optimized downloads are standard Parquet. Content-defined chunking
+(CDC) changes writer page boundaries for deduplication; it requires no special reader. Page
+indexes are independent and let compatible predicates skip pages. The checked-in PyArrow 21
+fixtures exercise all four CDC/index combinations with SNAPPY/ZSTD and V1/V2 pages.
-## API Surface
-
-**Direct API (recommended)**
-- `MatrixParquetReader.builder()` — fluent reader with `.matrixName()`, `.zoneId()`, `.read(File/Path/URL/InputStream/byte[])`
-- `MatrixParquetWriter.builder(matrix)` — fluent writer with `.precision()`, `.scale()`, `.decimalMeta()`, `.roundingMode()`, `.compressionCodec()`, `.zoneId()`, `.inferPrecisionAndScale()`, `.write(File/Path/OutputStream)`, `.writeBytes()`
-
-**SPI (generic Matrix API)**
-- `Matrix.read(options, file)` — options map with keys `matrixName`, `zoneId`
-- `matrix.write(options, file)` — options map with keys `inferPrecisionAndScale`, `precision`, `scale`, `decimalMeta`, `roundingMode`, `compressionCodec`, `zoneId`
-
-**Typed options classes**
-- `ParquetReadOptions` — typed options for reading; use `ParquetReadOptions.describe()` for runtime discovery
-- `ParquetWriteOptions` — typed options for writing; use `ParquetWriteOptions.describe()` for runtime discovery
-
-## Default Behavior
-
-**Reading**
-- Matrix name for file reads: `matrixName` option wins; otherwise the file basename without extension (e.g. `data.parquet` → matrix named `data`)
-- Matrix name for stream/byte-array/URL reads: `matrixName` option wins; otherwise the Parquet schema/message name when available
-- Timezone: system default; override with `zoneId` option
-
-**Writing**
-- Parquet message type name: `matrix.matrixName` when present, otherwise `MatrixSchema`
-- BigDecimal: precision and scale are inferred from the data by default (`inferPrecisionAndScale = true`)
-- Explicit DECIMAL scale conversion: `roundingMode = UNNECESSARY` by default, so a discarded non-zero fraction fails instead of being silently rounded
-- Compression: `SNAPPY` by default; override with the `compressionCodec` option (e.g. `GZIP`, `ZSTD`, `UNCOMPRESSED`)
-- Timezone: system default; override with `zoneId` option
-- When the target is an existing directory, the Matrix name is converted to one safe filename inside it. Path separators and unsafe characters are replaced, trailing dots/spaces are removed, blank or dots-only names become `matrix.parquet`, and an existing `.parquet` suffix is preserved. Pass an explicit file (including a non-existent file path) to retain the exact filename.
-- Nested Map columns: stored as MAP when value types are homogeneous, as STRUCT when heterogeneous
-- Index columns: written to `matrix.indexColumns` metadata as a JSON string array so column names containing commas round-trip; readers also accept the legacy comma-delimited metadata form
-
-## Installation
-
-Add the following to your Gradle build script:
-```groovy
-implementation 'org.apache.groovy:groovy:5.0.6'
-implementation 'se.alipsa.matrix:matrix-core:3.9.0'
-implementation 'se.alipsa.matrix:matrix-parquet:0.7.0'
-```
-
-## Basic Usage
-
-### Using Matrix.read() / matrix.write()
-
-If `matrix-parquet` is on the classpath, `.parquet` files are also available through the generic Matrix SPI API:
-
-```groovy
-import se.alipsa.matrix.core.Matrix
-
-Matrix data = Matrix.read(new File('data.parquet'))
-Matrix stockholm = Matrix.read([matrixName: 'events', zoneId: 'Europe/Stockholm'], new File('events.parquet'))
-
-data.write([inferPrecisionAndScale: true], new File('copy.parquet'))
-data.write([precision: 38, scale: 18, zoneId: 'Europe/Stockholm'], new File('money.parquet'))
-data.write([decimalMeta: [amount: [8, 2]]], new File('money-meta.parquet'))
-
-println Matrix.listReadOptions('parquet')
-println Matrix.listWriteOptions('parquet')
-```
-
-### Reading a Parquet file
```groovy
+import java.time.LocalDate
import se.alipsa.matrix.core.Matrix
import se.alipsa.matrix.parquet.MatrixParquetReader
-
-File file = new File("data.parquet")
-Matrix matrix = MatrixParquetReader.read(file)
-
-// Or with a custom matrix name
-Matrix matrix = MatrixParquetReader.read(file, "myData")
-
-// Read from a Path
-Matrix matrix = MatrixParquetReader.read(file.toPath())
-
-// Read from an InputStream
-new FileInputStream(file).withCloseable { is ->
- Matrix matrix = MatrixParquetReader.read(is)
-}
-
-// Read from a URL (e.g. file:// or https://)
-URL url = file.toURI().toURL()
-Matrix matrix = MatrixParquetReader.read(url)
-
-// Read from a file path string
-Matrix matrix = MatrixParquetReader.readFile("/path/to/data.parquet")
-```
-
-### Writing a Matrix to a Parquet file
-```groovy
-import se.alipsa.matrix.core.Matrix
import se.alipsa.matrix.parquet.MatrixParquetWriter
-import se.alipsa.matrix.datasets.Dataset
+import se.alipsa.matrix.parquet.Predicates
-Matrix data = Dataset.cars().withMatrixName('cars')
-File file = new File("cars.parquet")
-MatrixParquetWriter.write(data, file)
+File file = new File('events.parquet')
+Matrix events = Matrix.builder('events').columns(
+ id: [1L, 2L, 3L], date: [LocalDate.of(2026, 1, 1)] * 3,
+ text: ['first', 'second', 'third']).build()
+MatrixParquetWriter.builder(events).pageSize(65536).write(file)
-// Or write to a directory (filename derived from matrix name)
-File dir = new File("output/")
-MatrixParquetWriter.write(data, dir) // Creates output/cars.parquet
+Matrix selected = MatrixParquetReader.builder().columns(['text', 'id'])
+ .filter(Predicates.ge('id', 2L)).limit(10).read(file)
+assert selected.column('id') == [2L, 3L]
```
-### Builder API (recommended)
+Projection accepts top-level fields in output order. Filter-only columns are read internally.
+`Predicates.eq/ne/lt/le/gt/ge` have Number, String, Boolean, LocalDate, LocalDateTime, Instant,
+LocalTime and byte[] overloads. Use `isNull`, `isNotNull`, `and`, `or` and `not` for composition.
+Values must match the file's logical schema. Integer overflow, fractional integer values,
+lossy decimal rescaling, and timestamp/time precision loss are rejected before reading pages.
+Repeated/nested predicates are rejected. Missing indexes fall back to scans; corrupt data fails.
+
+String ordering uses unsigned UTF-8 bytes, matching Parquet. Unannotated BINARY predicates take
+byte[] even when legacy metadata or `binaryAsString` makes output text. Comparisons with null are false; null tests are explicit. `not` is Boolean negation, including
+for null rows. NaN comparisons are false except `ne`; numeric infinities retain their ordering.
+Safe physical comparisons use row-group, dictionary, bloom and page-index pruning. Binary
+DECIMAL comparisons and expressions without safe pushdown use exact residual evaluation.
+LocalDateTime predicates on adjusted timestamps use `zoneId`; Instant predicates on naive
+local timestamps are rejected. External TIME fields retain the existing `java.sql.Time`
+millisecond output contract; filtering evaluates the original time precision.
-The builder API provides a fluent, discoverable interface for reading and writing:
+## Batches and shards
```groovy
-import se.alipsa.matrix.core.Matrix
import se.alipsa.matrix.parquet.MatrixParquetReader
-import se.alipsa.matrix.parquet.MatrixParquetWriter
-// Writing
-MatrixParquetWriter.builder(matrix)
- .precision(38)
- .scale(18)
- .zoneId('Europe/Stockholm')
- .write(new File('output.parquet'))
-
-// Writing to byte array
-byte[] bytes = MatrixParquetWriter.builder(matrix)
- .inferPrecisionAndScale(true)
- .writeBytes()
-
-// Writing to OutputStream
-outputStream.withCloseable { os ->
- MatrixParquetWriter.builder(matrix).write(os)
-}
-
-// Reading
-Matrix data = MatrixParquetReader.builder()
- .matrixName('myData')
- .zoneId('Europe/Stockholm')
- .read(new File('data.parquet'))
-
-// Reading from byte array, InputStream, URL, or Path also supported
-Matrix data = MatrixParquetReader.builder()
- .read(parquetBytes)
+MatrixParquetReader.builder().limit(1000).maxBatchRows(250)
+ .batches(new File('events.parquet')).withCloseable { batches ->
+ while (batches.hasNext()) {
+ def batch = batches.next()
+ assert batch.rowCount() <= 250
+ }
+ }
+
+// Explicit lists retain caller order; directories and explicit globs sort normalized paths.
+def dataset = MatrixParquetReader.builder().parallelism(2).sourceFileColumn('_source')
+ .read([new File('train-00001.parquet'), new File('train-00000.parquet')])
+def fromDirectory = MatrixParquetReader.read(new File('train'))
+def fromGlob = MatrixParquetReader.readGlob('train/train-*.parquet')
```
-## Supported Data Types
-
-| Java/Groovy Type | Parquet Type | Notes |
-|--------------------|--------------------------------|---------------------------------------------------|
-| Integer | INT32 | |
-| Long | INT64 | |
-| BigInteger | BINARY (DECIMAL, scale=0) | Precision auto-inferred from data; no `Long` range limit |
-| Float | FLOAT | |
-| Double | DOUBLE | |
-| BigDecimal | FIXED_LEN_BYTE_ARRAY (DECIMAL) | See [BigDecimal Precision](#bigdecimal-precision) |
-| Boolean | BOOLEAN | |
-| LocalDate | INT32 (DATE) | |
-| java.sql.Date | INT32 (DATE) | |
-| java.sql.Time | INT32 (TIME_MILLIS) | Millisecond precision; external TIME(MICROS/NANOS) is downgraded to milliseconds |
-| LocalDateTime | INT64 (TIMESTAMP_MICROS) | Microsecond precision |
-| java.sql.Timestamp | INT64 (TIMESTAMP_MICROS) | Microsecond precision |
-| java.util.Date | INT64 | Epoch milliseconds |
-| List | LIST | See [Nested Structures](#nested-structures) |
-| Map | MAP or STRUCT | See [Nested Structures](#nested-structures) |
-| Other | BINARY (STRING) | Stored as toString() |
-
-## BigDecimal Precision
-
-BigDecimal values require special handling to preserve precision. There are three approaches:
-
-### 1. Automatic Precision Inference (Default)
-```groovy
-// Automatically infers precision and scale from the data
-MatrixParquetWriter.write(data, file)
-// Or explicitly:
-MatrixParquetWriter.write(data, file, true)
-```
+A batch normally contains one selected row group. `maxBatchRows` splits large groups; individual
+cells and compressed row-group pages still need heap. Close iterators on early exit; exhaustion
+and errors close them automatically. Limits count accepted rows across batches/files. Footer
+IDs remain valid when filters prune their groups; such groups produce no batch. There is no
+ordinal row seeking API.
-### 2. Uniform Precision for All BigDecimal Columns
-```groovy
-// Use precision=38, scale=18 for all BigDecimal columns
-MatrixParquetWriter.write(data, file, 38, 18)
-```
+Datasets validate every footer before returning rows and require identical logical fields,
+order, decimal precision/scale, timestamp units/UTC flags and Matrix type metadata. Duplicate
+canonical files and empty selections fail. Directory reads include direct `.parquet` children;
+globs are explicit and can select deeper paths. Compatible Matrix indices are restored when every shard has the same index metadata and
+projection retains its columns. A source column
+contains canonical absolute paths and must not collide with a physical column. Parallel workers
+use independent streams and converters, reuse each parsed footer, prefetch at most one batch per
+worker, and preserve file/group order. `rowGroups` is for one file; use
+`rowGroupsByFile: [(file.canonicalPath): [1, 0]]` for explicit dataset group selections.
-Explicit precision/scale writes reject a value that would lose a non-zero fraction by default. To opt in to a deliberate rounding policy, use the builder or the typed options map:
+## HTTP and Hub URIs
```groovy
-import java.math.RoundingMode
-
-// Direct API
-MatrixParquetWriter.write(data, file, 8, 2, RoundingMode.HALF_UP)
-
-// Builder API
-MatrixParquetWriter.builder(data)
- .precision(8)
- .scale(2)
- .roundingMode(RoundingMode.HALF_UP)
- .write(file)
+import se.alipsa.matrix.parquet.MatrixParquetReader
+import se.alipsa.matrix.parquet.Predicates
-data.write([precision: 8, scale: 2, roundingMode: 'HALF_UP'], file)
+def remote = MatrixParquetReader.builder().remoteReadMode('AUTO').columns(['id'])
+ .filter(Predicates.ge('id', 1000L)).limit(100)
+ .readUrl('https://huggingface.co/datasets/owner/repo/resolve/main/train.parquet')
+def hub = MatrixParquetReader.builder().remoteReadMode('AUTO').limit(20)
+ .readUrl('hf://datasets/owner/repo@refs/convert/parquet/train/0000.parquet')
```
-`roundingMode` accepts any `RoundingMode` enum name, case-insensitively. Use `UNNECESSARY` (the default) for lossless conversion; a rounded result must still fit the declared precision.
+Replace owner/repo/path with a real dataset. DOWNLOAD preserves full-download behavior. AUTO
+verifies `206 Content-Range` support then uses ranges, or downloads when the initial probe is
+unsupported. RANGE requires range support. Once decoding starts, transport/object-change errors
+propagate. Range caching is bounded by `rangeCacheBytes`; individual requests are capped separately at
+8 MiB. Bulk reads request the needed span rather than splitting it into 64 KiB requests. Small
+reads prefetch at most 64 KiB. Validated redirect targets are reused, and fully consumed range
+responses permit HTTP keep-alive reuse. ETag/Last-Modified validators and total length detect
+changes when available; same-length changes without validators cannot be detected reliably.
+Redirects are bounded and HTTPS downgrades rejected. Connection failures before response headers
+are retried once; malformed responses and body failures propagate. Streams are spooled to
+temporary files; byte arrays use a seekable memory input.
+
+Hub dataset URI parsing is pinned to huggingface_hub v2.2.0. Revisions follow the repository:
+`hf://datasets/owner/repo@revision/path`. Encode ordinary revision slashes as `%2F`; special
+`refs/convert/` and `refs/pr/N` forms may remain unescaped. `@` inside a file path is literal.
+Dot and parent path/revision segments are rejected. HF URI paths are literal (already-percent-encoded path text is escaped again). `HF_TOKEN` is
+optional and read only for explicit hf:// calls; authorization is attached only to HTTPS
+`huggingface.co` on its default/443 port and never to CDN redirect origins. This is HTTP range
+access, with no Xet download/deduplication client.
+
+## Read options
+
+Use fluent builder methods, `MatrixParquetReader.builder(typedOptions)`, or SPI maps:
+`Matrix.read([columns: ['id'], limit: 10L], file)`. Unknown options are rejected. Null map values preserve defaults, including nullable AUTO options.
+`ParquetReadOptions.describe()` lists these options at runtime.
+
+| Option | Default | Meaning |
+|---|---|---|
+| matrixName | null | Explicit name; otherwise filename for local files, schema name for memory/streams/URLs |
+| zoneId | system zone (null) | Adjusted timestamp conversion; ZoneId or zone name |
+| columns | null/all | Nonempty unique ordered top-level output names |
+| filter | null | Immutable Matrix-owned ParquetPredicate |
+| binaryAsString | null/AUTO | Legacy Matrix unannotated binary is text; external/new binary is bytes; true reads external unannotated binary as UTF-8 recursively, false reads raw bytes; annotations and explicit new binary metadata take precedence |
+| useColumnIndexFilter | true | Permit page pruning; false is useful for result/I/O comparisons |
+| limit | null/unlimited | Nonnegative matching-row bound; zero retains schema |
+| rowGroups | null/all | Unique, zero-based original footer IDs, caller order |
+| rowGroupsByFile | empty/all | Dataset-only canonical file path to ordered group ID list |
+| maxBatchRows | null | Positive accepted-row batch cap; default one row group |
+| sourceFileColumn | null | Optional canonical local file/shard path column, added after projection; unsupported for memory/stream/URL sources |
+| parallelism | 1 | Positive maximum dataset workers across files and row groups |
+| remoteReadMode | DOWNLOAD | DOWNLOAD, AUTO or RANGE |
+| connectTimeoutMillis | 10000 | Positive HTTP connection timeout |
+| readTimeoutMillis | 60000 | Positive HTTP read timeout |
+| rangeCacheBytes | 8388608 | Positive per-stream cache ceiling; independent request cap is 8 MiB |
+
+## Writer options and exact types
-### 3. Per-Column Precision Specification
```groovy
-// Specify precision and scale for specific columns
-MatrixParquetWriter.write(data, file, [
- salary: [10, 2], // precision=10, scale=2
- rate: [8, 4] // precision=8, scale=4
-])
-```
-
-**Note:** If the specified precision is too small for a value, an `IllegalArgumentException` will be thrown with a descriptive message indicating the required vs. allowed precision.
-
-## Nested Structures
-
-The module supports nested Groovy collections:
+import se.alipsa.matrix.core.Matrix
+import se.alipsa.matrix.parquet.MatrixParquetWriter
-```groovy
-def data = Matrix.builder('products').data(
- id: [1, 2],
- tags: [['electronics', 'sale'], ['furniture']], // List
- attributes: [[color: 'red', size: 'L'], [color: 'blue']], // Map as STRUCT
- reviews: [[[rating: 5, user: 'A']], [[rating: 4, user: 'B']]] // List of Maps
-).types([Integer, List, Map, List]).build()
-
-MatrixParquetWriter.write(data, file)
-Matrix restored = MatrixParquetReader.read(file)
+def values = Matrix.builder('prices').columns(id: [1L, 2L], price: [1.25G, 2.50G]).build()
+MatrixParquetWriter.builder(values).writerVersion('V2').compressionCodec('ZSTD')
+ .rowGroupSize(16L * 1024 * 1024).pageSize(65536)
+ .bloomFilterEnabled([id: true]).bloomFilterNDV([id: 1000L])
+ .decimalMeta([price: [8, 2] as int[]]).write(new File('prices.parquet'))
```
-### Nested Structure Limitations
-- Maps with heterogeneous value types are stored as STRUCT (preserves key names as field names)
-- Maps with homogeneous value types are stored as MAP (standard Parquet map)
-- Deeply nested structures are supported but may impact performance
-- Bean/POJO objects are converted to Maps using reflection
-
-## Timezone Handling
-
-By default, `LocalDateTime` values are converted to/from UTC using the system default timezone. To use a specific timezone:
-
-```groovy
-import java.time.ZoneId
-
-// Writing with a specific timezone
-MatrixParquetWriter.write(data, file, ZoneId.of("America/New_York"))
-
-// Reading with a specific timezone
-Matrix matrix = MatrixParquetReader.read(file, ZoneId.of("America/New_York"))
-
-// Reading with name and timezone
-Matrix matrix = MatrixParquetReader.read(file, "myData", ZoneId.of("Europe/London"))
+File, Path, byte-array and OutputStream writes share writer tuning. Size settings are targets,
+not hard memory limits. V2 changes value encoders as well as page layout; V1 remains the default
+for compatibility with older consumers. Bloom NDVs affect memory/false-positive tradeoffs and do
+not enable a filter by themselves. Column settings use physical primitive paths (for example
+`tags.list.element`); unknown/nonprimitive paths fail. Disabled bloom columns remain disabled.
+
+| Option | Default | Meaning |
+|---|---|---|
+| compressionCodec | SNAPPY | Codec name/CompressionCodecName; tested SNAPPY, ZSTD, GZIP, UNCOMPRESSED |
+| inferPrecisionAndScale | true | Infer exact decimal settings when no explicit settings apply |
+| precision / scale | null / null | Uniform positive precision and nonnegative scale, provided together |
+| decimalMeta | empty | Logical field path to `[precision, scale]`; overrides uniform settings |
+| roundingMode | UNNECESSARY | Reject lossy decimal rescaling; opt in to e.g. HALF_UP deliberately |
+| zoneId | system zone (null) | LocalDateTime to adjusted timestamp conversion |
+| rowGroupSize | 134217728 | Positive target row-group bytes |
+| pageSize | 1048576 | Positive target data-page bytes |
+| dictionaryEncoding | true | Dictionary encoding with upstream fallback |
+| writerVersion | V1 | V1 or V2 writer encodings and page layout |
+| columnIndexTruncateLength | 64 | Positive binary page-bound truncation bytes |
+| statisticsTruncateLength | 2147483647 | Positive row-group binary statistics truncation bytes |
+| bloomFilterEnabled | empty, disabled | Primitive path to Boolean override |
+| bloomFilterNDV | empty | Primitive path to positive expected distinct-value count |
+| sizeStatisticsEnabled | true | Collect size statistics |
+
+Files are written to a temporary sibling and atomically replace the destination after successful
+close. Failed writes preserve an existing destination; filesystems without atomic replacement
+support fail safely with `AtomicMoveNotSupportedException`. New files use normal process/umask
+permissions; existing POSIX permissions are retained. Existing symbolic links are resolved before
+writing, including dangling links, so the link is preserved and its target is created or replaced.
+Abrupt JVM termination can leave `.matrix-parquet-.tmp` siblings beside the resolved output;
+these are unpublished temporary output and can be removed once no writer is using them.
+
+JSON and ENUM annotations decode as UTF-8 String; BSON remains binary.
+Strings are STRING-annotated recursively; byte[] stays raw BINARY. Byte/Short use signed width
+annotations, UUID uses FIXED[16], LocalTime uses TIME(NANOS), and Instant uses adjusted
+TIMESTAMP(MICROS). Instants with sub-microsecond precision or int64-microsecond overflow fail;
+Instant.MIN/MAX are unsupported. Character and enums become String (`name()` for enums).
+ZonedDateTime/OffsetDateTime/custom objects require explicit conversion to Instant/String/Map.
+Date/Timestamp subclasses must share a single encoding policy; mixed units fail before output.
+
+Object and abstract/interface declarations scan every non-null value, check assignability and
+infer one encoding; canonical metadata replaces the supertype. Concrete declarations require
+instance-compatible values and a single runtime encoding policy. Mixed numeric classes need
+explicit normalization. All-null Object becomes String; all-null Number/Temporal requires an
+explicit supported concrete type. Containers use canonical List/Map metadata. All supported
+metadata names are allowlisted; unknown external class names fall back to schema inference.
+
+Every resolved BigDecimal uses exact DECIMAL FIXED storage. Settings precedence is per-field,
+uniform, then inference. Nested logical paths use `.element`, `.key`, `.value`, and struct field
+names (for example `nested.element.amount`). With inference disabled every decimal needs settings.
+All-null BigDecimal infers `(1, 0)`. There is no DOUBLE fallback. Explicit decimal conversion obeys
+roundingMode. See [migration notes](release.md) for changed 0.7.x behavior.
+
+## CDC output, tests and benchmarks
+
+Native CDC writing is deferred: [parquet-java #3818](https://github.com/apache/parquet-java/pull/3818)
+was open/unmerged when checked on 2026-10-17. Tuning pages/row groups is not CDC. For CDC output,
+rewrite with PyArrow and preserve metadata:
+
+```bash
+python3 -m venv /tmp/matrix-parquet-fixtures
+/tmp/matrix-parquet-fixtures/bin/python -m pip install pyarrow==21.0.0
+/tmp/matrix-parquet-fixtures/bin/python - <<'PY'
+import pyarrow.parquet as pq
+table = pq.read_table('prices.parquet')
+pq.write_table(table, 'prices-cdc.parquet', compression='zstd',
+ use_content_defined_chunking=True, write_page_index=True)
+assert pq.read_table('prices-cdc.parquet').equals(table, check_metadata=True)
+PY
```
-**Important:** For consistent round-trip behavior, use the same timezone for writing and reading. If files are shared across systems with different default timezones, explicitly specify the timezone.
-
-`java.sql.Time` is different: it is always treated as a local time of day in the JVM default timezone, independently of `zoneId`, and preserves milliseconds. External `TIME(MICROS)` and `TIME(NANOS)` values are read as `Time` with their sub-millisecond fraction truncated; the Parquet `isAdjustedToUTC` flag does not cause a zone shift.
-
-## Compression
-
-Parquet files are compressed with `SNAPPY` by default. Override per write:
-
-```groovy
-import org.apache.parquet.hadoop.metadata.CompressionCodecName
-
-// Builder API
-MatrixParquetWriter.builder(data)
- .compressionCodec(CompressionCodecName.GZIP)
- .write(file)
-
-// String form (also accepted)
-MatrixParquetWriter.builder(data)
- .compressionCodec('ZSTD')
- .write(file)
-
-// SPI options map
-data.write([compressionCodec: 'UNCOMPRESSED'], file)
+This loads the whole Arrow table and adds a rewrite pass. Python is not a runtime dependency.
+Hadoop filesystem configuration is avoided for local reads/writes; upstream codec initialization
+still requires Hadoop classes, so the dependency and JDK 21 limit remain.
+
+```bash
+./gradlew :matrix-parquet:codenarcMain :matrix-parquet:codenarcTest
+./gradlew :matrix-parquet:spotlessCheck
+./gradlew :matrix-parquet:test :matrix-parquet:compileConsumerApi
+PARQUET_PYTHON=/tmp/matrix-parquet-fixtures/bin/python ./gradlew :matrix-parquet:test -PrunSlowTests=true
+/tmp/matrix-parquet-fixtures/bin/python matrix-parquet/src/test/python/verify_matrix_output.py
+./gradlew :matrix-parquet:test --tests ParquetInteropTest --rerun-tasks
+./gradlew test
```
-Supported codec names: `UNCOMPRESSED`, `SNAPPY`, `GZIP`, `ZSTD`, `LZ4_RAW` (any value of `org.apache.parquet.hadoop.metadata.CompressionCodecName`). Compression is transparent on read — `MatrixParquetReader` auto-detects the codec from the file footer, no reader-side configuration is needed.
-
-## Known Limitations
-
-### Memory Usage
-Matrix is an **in-memory data structure**. The entire Parquet file is loaded into memory when reading. Ensure you have sufficient RAM available for:
-- The raw data size
-- Java object overhead (typically 2-3x the raw data size)
-- Working memory for processing
-
-For very large datasets that don't fit in memory, consider:
-- Processing files in chunks externally before loading
-- Using dedicated big data tools (Spark, Hadoop) for initial filtering
-- Increasing JVM heap size (`-Xmx`)
-
-### JDK Compatibility
-This module requires **JDK 21** and is limited to JDK 21 maximum due to Hadoop 3.4.x compatibility constraints. JDK 22+ is not supported.
-
-### Timestamp Precision
-- `LocalDateTime` and `Timestamp` are stored with **microsecond** precision
-- `Time` is stored with **millisecond** precision
-- Nanosecond precision is not preserved
-
-### Type Metadata
-Column type information is stored in Parquet file metadata. When reading files not created by `MatrixParquetWriter`:
-- Types are inferred from Parquet schema logical type annotations
-- Some type information may be lost (e.g., `java.sql.Timestamp` vs `LocalDateTime`)
-- `BigInteger` columns read without Matrix metadata (e.g. external files) are inferred as `BigDecimal` with scale 0, since both the Parquet schema and the inference logic only see a generic `DECIMAL` logical annotation
-- `BigInteger` values nested inside `List`/`Map` columns always get a declared precision of 38 regardless of their actual digit count (top-level column precision inference does not extend into nested structures), and round-trip back as numerically-equal `BigDecimal` values rather than `BigInteger` (nested elements carry no expected-type metadata, unlike top-level columns). This does not cause data loss or truncation, but external Parquet readers that validate `DECIMAL` precision against declared metadata rather than actual byte length may reject or misinterpret values with more than 38 digits
-
-## Technical Notes
-
-### Thread Safety
-Both `MatrixParquetWriter` and `MatrixParquetReader` use thread-safe caching for:
-- PropertyDescriptor lookups (for struct handling)
-- Class.forName() results (for type parsing)
-
-The timezone parameter uses thread-local storage, making concurrent writes/reads with different timezones safe.
-
-### Performance Tips
-1. Use `inferPrecisionAndScale=true` (default) for automatic BigDecimal handling
-2. For large matrices with struct columns, the first write may be slower due to reflection caching
-3. Subsequent operations on the same struct types benefit from cached metadata
-
-## 3rd Party Libraries
-
-| Library | Purpose | License |
-|---------|---------|---------|
-| org.apache.parquet:parquet-column | Parquet columnar format | Apache 2.0 |
-| org.apache.parquet:parquet-hadoop | Hadoop integration | Apache 2.0 |
-| org.apache.hadoop:hadoop-common | Hadoop utilities | Apache 2.0 |
-| org.apache.hadoop:hadoop-mapreduce-client-core | MapReduce client | Apache 2.0 |
+Fixture reproduction: run `matrix-parquet/src/test/python/generate_optimized.py` from the repository root with
+pinned PyArrow. Checked-in Parquet fixtures total less than 4 MiB (each less than 1 MiB).
+Default-size CDC is generated in `build/default-cdc` only in the slow lane; missing Python/Arrow
+produces an explicit skip. CI provisions pinned Arrow and checks fixture reproducibility.
+See [measurements and reproduction](benchmarks/optimized-parquet.md) for the opt-in benchmark.
diff --git a/matrix-parquet/release.md b/matrix-parquet/release.md
index 59fd65be9..25e394e0c 100644
--- a/matrix-parquet/release.md
+++ b/matrix-parquet/release.md
@@ -1,5 +1,66 @@
# Matrix-parquet Release History
+## v0.8.0 (unreleased)
+
+Add CDC/page-index interoperability fixtures; a column-oriented row-group reader; typed predicates,
+projection, limits and closeable batches; strict sharded datasets with bounded parallel decoding;
+HTTP range reads; Hub dataset URI resolution; and shared writer tuning. Local files and byte arrays
+use InputFile directly. New files carry `matrix.parquet.version=2`.
+
+Complete defaults, supported types and end-to-end examples are in [the module guide](readme.md).
+
+| Behavior | 0.8.0 result | Compatibility/opt-out |
+|---|---|---|
+| New Matrix strings | STRING-annotated BINARY recursively | No opt-out; standard annotation fixes interoperability |
+| Legacy Matrix plain BINARY text | Remains String, including nested text/map keys | Automatic; metadata carrying the new format marker overrides legacy inference |
+| 0.7.x byte[] columns | Fail with a contextual unrecoverable-legacy-binary error | The pinned 0.7.x writer fails with InvalidSchemaException for byte[]; legacy `[B` text payloads cannot recover bytes; regenerate from source |
+| JSON / ENUM BINARY | UTF-8 String, including dictionary/null-only cases | Preserves pre-0.8 text behavior; BSON remains byte[] |
+| External raw BINARY | byte[] | `binaryAsString(true)` restores prior UTF-8 behavior |
+| Unadjusted timestamps | Local wall-clock time preserved | No bug-preserving flag; applications needing the former shift can explicitly convert the local value through UTC into the desired zone |
+| Integer-backed DECIMAL | BigDecimal with declared scale | No bug-preserving flag; explicit `unscaledValue()` obtains prior raw numeric value |
+| UUID annotation | UUID | No text-decoding compatibility flag; explicit UUID/string/byte conversion at application boundary |
+| FLOAT16 / INT96 unsupported cases | Contextual unsupported-type error | No UTF-8 fallback; convert these types explicitly before reading |
+| Byte / Short fallback | Signed INT32 with INT(8/16), restored as Byte/Short | Legacy unannotated text stays String; explicitly convert to String before writing to retain text output |
+| LocalTime fallback | TIME(NANOS), restored as LocalTime using new-format metadata | Legacy text stays String; explicit String conversion retains text output |
+| Instant fallback | TIMESTAMP(MICROS, adjusted=true), restored as Instant using new-format metadata | Legacy text stays String; reject sub-microsecond precision loss and int64-micros overflow; explicitly convert to String for text output |
+| UUID writer fallback | FIXED[16] + UUID, restored as UUID | Legacy text stays String; explicit String conversion retains text output |
+| Character fallback | STRING, returned as String | Normalize emitted metadata to String; Character identity is not preserved |
+| Enum fallback | STRING using enum name(), returned as String | Normalize metadata to String; convert explicitly for former overridden toString() output |
+| ZonedDateTime / OffsetDateTime fallback | Explicit unsupported-type error | Convert to Instant to retain instant, or String to retain zone/offset text; no implicit zone loss |
+| Object column, single supported policy type | That type's encoding and canonical metadata | Reads back typed as the inferred canonical class instead of Object; no explicit types required |
+| Abstract/interface declared supertype (Number, CharSequence, Comparable, Serializable, Temporal) | Validate assignability, infer one supported policy type and emit its canonical name | Reads back as the resolved class instead of the declared supertype; normalize heterogeneous values explicitly |
+| Concrete declared type with incompatible runtime values | Validation error before writing | Box primitives then require every non-null value to be an instance; no silent schema/metadata mismatch |
+| BigDecimal, declared or inferred through a supertype | Exact FIXED_LEN_BYTE_ARRAY DECIMAL with resolved precision/scale | Remove DOUBLE fallback; infer by default, or require explicit decimal settings when inference is disabled; no lossy compatibility mode |
+| All-null Number/Temporal declaration | Validation error before writing | New write failure: retype explicitly to a supported concrete type; no nominal String metadata |
+| Concrete Date column mixing Date/Timestamp or sql.Date/Timestamp | Validation error before writing | Runtime encoding policies differ; normalize every value to one supported temporal class before writing |
+| All-null Object column | STRING/BINARY with canonical String metadata | Retain current text schema behavior; reads back as String-typed column |
+| Mixed/unsupported Object column or other custom types | Contextual validation error before writing | Name the first offending row/class; explicitly normalize values or map to String |
+
+Unknown metadata class names are never loaded and fall back to physical/logical schema inference.
+Incompatible metadata declarations also fall back rather than creating falsely typed columns.
+Legacy Matrix text compatibility applies recursively; new raw binary metadata remains authoritative.
+The checked-in original 0.7.x string fixture predates these writer changes. Legacy byte[] recovery
+claims were narrowed after an end-to-end check found InvalidSchemaException in this checkout.
+
+External INT96/FLOAT16 raise contextual unsupported-type errors. External TIME still returns
+java.sql.Time at millisecond precision; exact predicate filtering preserves original precision.
+Mixed Date/Timestamp policies are now rejected before writing, preventing the prior 1000× unit
+corruption. New Object/supertype values are resolved consistently for schema and metadata.
+Unknown SPI options are now rejected. Dataset reads restore common compatible Matrix indices, require matching type
+metadata, and reject empty selections and duplicate paths.
+
+Review fixes also cover empty Arrow dictionaries; consistent static URL/Hub and directory overloads;
+local source columns; daemon dataset workers; null option-map defaults; path-segment validation;
+projection around legacy `[B` columns; one validated type-policy scan per column; and atomic output
+replacement on success. New files use normal process/umask permissions; existing POSIX permissions
+are retained, and writes through existing symbolic links update the resolved target while preserving
+the link. Local file writes require same-directory atomic replacement support; unsupported network
+or FUSE filesystems raise `AtomicMoveNotSupportedException` and leave existing output unchanged.
+HTTP bulk requests reuse resolved targets and a separately bounded cache.
+
+Native CDC writing remains deferred to upstream parquet-java #3818; the Python rewrite preserves
+custom metadata. Hadoop stays as a runtime dependency because codec initialization still loads it.
+
## v0.7.0, 2026-09-22
- SPI `inferPrecisionAndScale` parsing is now strict: Boolean values and trimmed, case-insensitive `true`/`false` strings are accepted; former truthy inputs are rejected.
- Breaking change: explicit DECIMAL scale conversion now defaults to `RoundingMode.UNNECESSARY`, rejecting lossy conversion with a contextual `IllegalArgumentException`. Set `roundingMode: 'HALF_UP'` (or another named `RoundingMode`) when rounding is intentional.
diff --git a/matrix-parquet/req/optimized-parquet-plan.md b/matrix-parquet/req/optimized-parquet-plan.md
new file mode 100644
index 000000000..5429af520
--- /dev/null
+++ b/matrix-parquet/req/optimized-parquet-plan.md
@@ -0,0 +1,786 @@
+# Hugging Face optimized Parquet: analysis and enhancement plan
+
+Analysis date: 2026-10-17. Scope: compatibility and baseline measurements (P0), columnar and bounded local reads,
+typed predicates and writer tuning (P1), sharded datasets and HTTP ranges (P2),
+optional Hub resolution and upstream CDC tracking (P3). Implementation and verification evidence is recorded at the end of this document.
+The findings below describe the pre-implementation baseline.
+
+## Findings
+
+### What “optimized Parquet” means
+
+Hugging Face describes optimized Parquet as ordinary Parquet written with content-defined
+chunking (CDC) and page indexes. These are writer/layout features, not a separate container
+format, codec, file extension, or reader mode. PyArrow exposes
+`use_content_defined_chunking=True` and `write_page_index=True`; its CDC option requires
+PyArrow 21 or later. [Hugging Face documentation](https://huggingface.co/docs/hub/en/datasets-libraries),
+[PyArrow integration](https://huggingface.co/docs/hub/en/datasets-pyarrow).
+
+CDC chooses boundaries from logical column content before encoding/compression. Readers
+decode the resulting ordinary pages without reproducing the chunking algorithm. Xet can
+reuse matching stored chunks across revisions; reading a downloaded local file does not
+itself invoke Xet or give matrix-parquet incremental download capability.
+[Hugging Face CDC explanation](https://huggingface.co/blog/parquet-cdc).
+
+Page indexes consist of `ColumnIndex` (page statistics) and `OffsetIndex` (page locations
+and first-row positions), associated with column chunks. They support skipping candidate
+pages and aligning reads across columns. Indexes are optional, and page boundaries need
+not align across columns. Filtering must still reject nonmatching records within candidate
+pages. [Parquet page-index specification](https://parquet.apache.org/docs/file-format/pageindex/).
+
+PyArrow can move page statistics from page headers to the index when index writing is
+enabled. Thus tests must include real PyArrow output rather than assuming that Java-written
+round trips cover this layout. CDC and page indexes are independent switches; test all four
+combinations. Do not assume fixed page sizes or infer CDC from a page-size distribution.
+[PyArrow writer reference](https://arrow.apache.org/docs/python/generated/pyarrow.parquet.write_table.html).
+
+### Pre-implementation repository behavior
+
+The authoritative dependencies are **parquet-java 1.18.1**, Hadoop **3.5.0** in
+`gradle/libs.versions.toml`; the module is **0.7.1-SNAPSHOT**. The older v0.6.0 plan is
+historical and should not determine dependency decisions here. Keep JDK 21.
+
+| Area | Evidence in current code | Assessment |
+|---|---|---|
+| Full-file decoding | `MatrixParquetReader.read(File, ...)` reads the footer and uses `ParquetReader` / `GroupReadSupport` | CDC/indexed files should decode through the existing dependency; unproven until external fixtures run |
+| Selective reads | `ParquetReadOptions` and `ReaderBuilder` only expose name/time zone; core loop materializes every row and field | No caller-supplied predicate, projection, or bounded read; no useful page pruning for the current full scan |
+| Remote reads | URL overloads open a stream; `readFromInputStream` copies it entirely to a temporary file | Entire file transfers before reading; page pruning cannot reduce network traffic |
+| Metadata | Reader uses Matrix-specific type/index keys when present, otherwise schema inference | External files need tests without Matrix metadata; Matrix indexes are unrelated to Parquet page indexes |
+| Writes | File and memory writers configure schema, codec and Matrix metadata | Index production is delegated to upstream defaults; no CDC configuration |
+| Coverage | Six Groovy test classes and `mtcars.parquet` fixture; no CDC/index-specific suite | Cross-implementation and page-pruning coverage missing |
+
+parquet-java 1.18.1 already exposes `withFilter`, `useColumnIndexFilter(boolean)`, and
+record filtering. Column-index filtering is enabled by default by its read-options builder.
+The gap is supplying a predicate and preserving correct materialization, rather than merely
+turning an upstream flag on.
+[Reader API](https://github.com/apache/parquet-java/blob/apache-parquet-1.18.1/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetReader.java),
+[read defaults](https://github.com/apache/parquet-java/blob/apache-parquet-1.18.1/parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java).
+
+The upstream writer enables statistics by default and serializes column/offset indexes.
+Consequently, “add page-index writing” is not yet an established missing feature: first
+inspect actual matrix-parquet output for supported columns. There is no CDC option in the
+examined 1.18.1 `ParquetWriter.Builder` or `ParquetProperties.Builder`. Do not introduce a
+nonfunctional `optimizedParquet` or `contentDefinedChunking` option.
+[Writer API](https://github.com/apache/parquet-java/blob/apache-parquet-1.18.1/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java),
+[writer properties](https://github.com/apache/parquet-java/blob/apache-parquet-1.18.1/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java),
+[index serialization](https://github.com/apache/parquet-java/blob/apache-parquet-1.18.1/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java).
+
+### Public API and dependency boundary
+
+`parquet-column` and `parquet-hadoop` are `implementation` dependencies. Exposing
+`FilterPredicate` would force downstream consumers to add parquet-java to their compile
+classpath or require changing publication scope to `api`. Use Matrix-owned predicate and
+writer-version types instead; translate them internally. Keep parquet-java implementation
+private for the new API and add downstream Gradle/Maven compile smoke tests. Existing public
+codec signatures are a separate legacy exposure to characterize, not a reason to introduce
+more upstream types.
+
+### Adjacent schema risks
+
+These affect Hugging Face datasets independently of optimization and must be tracked
+separately from CDC compatibility:
+
+- Unannotated `BINARY`/`FIXED_LEN_BYTE_ARRAY` currently defaults to `String`, and
+ `readPrimitive` decodes it as UTF-8. Arbitrary image/audio bytes therefore lose fidelity.
+- DECIMAL annotations on `INT32`/`INT64` are inferred as `BigDecimal`, but the corresponding
+ primitive read branches return raw integers without applying decimal scale.
+- TIMESTAMP decoding ignores `isAdjustedToUTC=false` and shifts local wall-clock values
+ through the configured/system zone. Test naive PyArrow timestamps explicitly before declaring
+ their annotation confirmed by fixtures.
+- UUID and FLOAT16 annotations on FIXED_LEN_BYTE_ARRAY reach the UTF-8 fallback. INT96
+ reaches `group.getString`; the exact exception remains to be characterized by a fixture. Neither
+ path constitutes valid decoding.
+- `readList` expects the standard three-level `list` structure. Include PyArrow compliant
+ lists, nested structs and null/empty values; do not claim support for all historical LIST
+ representations without additional work.
+
+Prefer a small representative text dataset as the initial compatibility target. Include
+binary/media and integer-backed decimal fixtures to make broader limitations explicit.
+
+## Implementation sequence
+
+### 0. Set the release and migration contract (P0 prerequisite)
+
+- 0.1 [x] Before production changes, set `matrix-parquet/build.gradle` to `0.8.0-SNAPSHOT`; target
+ release **0.8.0**. Keep P2/P3 follow-ups separately releasable rather than promising their
+ delivery in 0.8.0. Add a 0.8.0 migration section to release notes with the behavior table below,
+ complete option defaults and runnable examples.
+
+- 0.2 [x] Capture representative 0.7.x-produced fixtures before changing the writer, including
+ top-level/nested strings, string map keys and Matrix temporal/decimal metadata. Include a 0.7.x
+ byte[] failure fixture; do not assume every old file is recoverable. Include them in the 4 MiB
+ fixture budget. Add before/after characterization tests plus external fixtures for behavior
+ 0.7.x did not write, asserting values and runtime types. The Group reader retained for
+ differential tests must receive phase-2 fixes; separately preserve original expected 0.7.x
+ results for migration characterization.
+
+ | Behavior | 0.8.0 result | Compatibility/opt-out |
+ |---|---|---|
+ | New Matrix strings | STRING-annotated BINARY recursively | No opt-out; standard annotation fixes interoperability |
+ | Legacy Matrix plain BINARY text | Remains String, including nested text/map keys | Automatic; metadata carrying the new format marker overrides legacy inference |
+ | 0.7.x byte[] columns | Fail with a contextual unrecoverable-legacy-binary error | The pinned original writer fails with InvalidSchemaException for byte[]; legacy `[B` text cannot recover bytes; regenerate from source |
+ | External raw BINARY | byte[] | `binaryAsString(true)` restores prior UTF-8 behavior |
+ | Unadjusted timestamps | Local wall-clock time preserved | No bug-preserving flag; applications needing the former shift can explicitly convert the local value through UTC into the desired zone |
+ | Integer-backed DECIMAL | BigDecimal with declared scale | No bug-preserving flag; explicit `unscaledValue()` obtains prior raw numeric value |
+ | UUID annotation | UUID | No text-decoding compatibility flag; explicit UUID/string/byte conversion at application boundary |
+ | FLOAT16 / INT96 unsupported cases | Contextual unsupported-type error | No UTF-8 fallback; supported-type contract fixed before release |
+ | Byte / Short fallback | Signed INT32 with INT(8/16), restored as Byte/Short | Legacy unannotated text stays String; explicitly convert to String before writing to retain text output |
+ | LocalTime fallback | TIME(NANOS), restored as LocalTime using new-format metadata | Legacy text stays String; explicit String conversion retains text output |
+ | Instant fallback | TIMESTAMP(MICROS, adjusted=true), restored as Instant using new-format metadata | Legacy text stays String; reject sub-microsecond precision loss and int64-micros overflow; explicitly convert to String for text output |
+ | UUID writer fallback | FIXED[16] + UUID, restored as UUID | Legacy text stays String; explicit String conversion retains text output |
+ | Character fallback | STRING, returned as String | Normalize emitted metadata to String; Character identity is not preserved |
+ | Enum fallback | STRING using enum name(), returned as String | Normalize metadata to String; convert explicitly for former overridden toString() output |
+ | ZonedDateTime / OffsetDateTime fallback | Explicit unsupported-type error | Convert to Instant to retain instant, or String to retain zone/offset text; no implicit zone loss |
+ | Object column, single supported policy type | That type's encoding and canonical metadata | Reads back typed as the inferred canonical class instead of Object; no explicit types required |
+ | Abstract/interface declared supertype (Number, CharSequence, Comparable, Serializable, Temporal) | Validate assignability, infer one supported policy type and emit its canonical name | Reads back as the resolved class instead of the declared supertype; normalize heterogeneous values explicitly |
+ | Concrete declared type with incompatible runtime values | Validation error before writing | Box primitives then require every non-null value to be an instance; no silent schema/metadata mismatch |
+ | BigDecimal, declared or inferred through a supertype | Exact FIXED_LEN_BYTE_ARRAY DECIMAL with resolved precision/scale | Remove DOUBLE fallback; infer by default, or require explicit decimal settings when inference is disabled; no lossy compatibility mode |
+ | All-null Number/Temporal declaration | Validation error before writing | New write failure: retype explicitly to a supported concrete type; no nominal String metadata |
+ | Concrete Date column mixing Date/Timestamp or sql.Date/Timestamp | Validation error before writing | Runtime encoding policies differ; normalize every value to one supported temporal class before writing |
+ | All-null Object column | STRING/BINARY with canonical String metadata | Retain current text schema behavior; reads back as String-typed column |
+ | Mixed/unsupported Object column or other custom types | Contextual validation error before writing | Name the first offending row/class; explicitly normalize values or map to String |
+
+### 1. Establish compatibility and writer-index facts (P0, first deliverable)
+
+- 1.1 [x] Add `OptimizedParquetTest.groovy` and checked-in deterministic fixtures
+ under `src/test/resources/optimized/`. Add a generator script plus a manifest containing
+ generator/PyArrow version, options, schema, row counts, expected values and SHA-256 checksums.
+ Python is a fixture-generation tool, not a runtime or normal Gradle-test dependency.
+
+- 1.2 [x] Generate paired ordinary/CDC-only/index-only/CDC+index files from identical values.
+ Include SNAPPY and ZSTD, dictionary enabled/disabled, data pages v1/v2, multiple pages per
+ column and multiple row groups. Use deterministic varied long strings so the CDC fixtures
+ actually cross chunk boundaries; verify physical multi-page structure. Set a **4 MiB aggregate
+ checked-in binary-fixture budget**, with a 1 MiB per-file cap. Use CDC `{'min_chunk_size': 4096,
+ 'max_chunk_size': 16384, 'norm_level': 0}` for small fixtures, and verify the resulting
+ boundaries/pages. PyArrow 21 accepts this dictionary with both size keys required, confirmed in
+ its [versioned implementation](https://github.com/apache/arrow/blob/apache-arrow-21.0.0/python/pyarrow/_parquet.pyx).
+ Add an opt-in default-CDC generation/interoperability test under `./gradlew :matrix-parquet:test
+ -PrunSlowTests=true`, using a build-directory fixture. Skip with an explicit reason if Python or
+ pinned PyArrow is absent; do not silently install packages. Add the Python/PyArrow lane to
+ `.github/workflows/release-all-tools.yml` (including its pull_request runs), invoking the opt-in
+ command so this case actually runs. Fail generation errors when dependencies are available and
+ enforce the checked-in budget. Pairwise coverage replaces a Cartesian product. Include
+ ordered/unordered columns, null-only pages, null/empty nested lists and structs, Unicode,
+ decimals and timestamps.
+
+- 1.3 [x] Assert exact values, types, order and null semantics through file, path, bytes, stream,
+ builder and SPI entry points. Use a local HTTP server for URL paths; regular tests must not
+ depend on Hugging Face availability or credentials. Cover empty files and external metadata
+ without Matrix-specific keys.
+
+- 1.4 [x] Inspect index references and deserialize `ColumnIndex`/`OffsetIndex` using
+ `ParquetFileReader`. Check page counts, monotonic offsets, row positions and valid bounds.
+ Inspect both Java-produced file and byte-array output. Record which types produce indexes and
+ any upstream omissions. Do not infer index presence from a flag.
+
+- 1.5 [x] Repair any demonstrated CDC/index decoding failure with a targeted regression test. If
+ all supported-schema fixtures pass, record that no CDC decoder change is necessary. Do not
+ upgrade parquet-java solely because the files use CDC.
+
+- 1.6 [x] Before reader changes, establish a repeatable full-scan baseline using generated
+ approximately 1M-row wide-text, low-cardinality categorical and mixed/nested datasets. Keep
+ large inputs outside Git. Record generation commands/seeds, JVM/heap/GC, CPU, warmup and
+ repeated-run medians, rows/s, elapsed time, allocation/GC, peak heap and bytes read. Separate
+ cold and warm I/O and compare ordinary/indexed/CDC+indexed layouts. Reuse these exact inputs
+ throughout P1/P2; include narrow projection and selective predicates as soon as those APIs
+ exist.
+
+### 2. Fix demonstrated external-schema fidelity gaps (P0 for broad HF support)
+
+Order within 2.0: **2.0 → 2.0c (shared descriptor) → 2.0a/2.0b/2.0d
+(policies implemented on that descriptor) → 2.0e (integration verification)**.
+Task identifiers remain stable for review references; their numeric order does not override
+these dependencies. Do not implement policies on the old first-value inference as an interim step.
+
+- 2.0 [x] Before 2.1, write String as BINARY + `LogicalTypeAnnotation.stringType()` through the
+ shared primitive schema builder, including list elements, struct fields, map values and required
+ map keys. Verify the annotations with PyArrow. Preserve existing files: when Matrix metadata
+ identifies a legacy file, unannotated BINARY remains String recursively, including nested values
+ whose expected type is null. New binary-capable output must include a versioned Matrix
+ schema/encoding marker to distinguish raw binary from legacy nested text; do not apply legacy
+ rules blindly to newly written media bytes. Add nullable Boolean `binaryAsString` (default
+ null/AUTO): null uses legacy-Matrix text compatibility and raw bytes for external/new-format
+ files; true forces unannotated BINARY to UTF-8 strings recursively; false forces raw bytes
+ recursively. Annotations take precedence; explicitly declared Matrix String fields remain
+ strings. Define FIXED binary separately: never override DECIMAL/UUID/FLOAT16 annotations.
+ Propagate through options, builder and SPI, and test old nested strings/map keys plus new mixed
+ binary/text fields.
+
+- 2.0a [x] Require raw byte[] writing as BINARY without STRING annotation, with the new format
+ marker, at top level and inside nested structures. Add an explicit byte[] branch using Parquet
+ Binary bytes; reject unsupported schema/value combinations with contextual errors instead of the
+ `value.toString()` fallback only after implementing the explicit type policies in 2.0b–2.0d. Land
+ 2.0a–2.0e together so fallback removal has no implicit scope. Resolve metadata array class names
+ through a fixed allowlist mapping of names to supported writer-produced types, including `[B` to
+ byte[]. Do not load arbitrary classes named by untrusted file metadata. If a resolver still
+ needs `Class.forName(name, false, loader)`, call it only for allowlisted names. Unknown names
+ fall back to per-field schema inference, without failing the whole read; malformed metadata must
+ not shift type/column alignment. Test unknown/custom class names and arrays. The legacy `[B`
+ corruption rule below takes precedence over both allowlist resolution and schema fallback. Add
+ file/bytes/stream/SPI round-trip tests for null, empty and arbitrary non-UTF-8 bytes, plus
+ PyArrow verification of exact payloads. Characterize original byte[] write failure and the `[B`
+ class-resolution failure. Use explicitly synthetic legacy text metadata if the original writer
+ cannot produce a valid file; fixing class lookup cannot recover a textual payload. Fail explicitly on legacy Matrix byte[] metadata rather than returning the
+ textual payload as if it were original binary. State this in migration notes and test it;
+ marker-aware new byte[] files must read normally.
+
+- 2.0b [x] Implement per-type encodings and explicit rejections. Replace remaining implicit
+ `toString()` cases with the policies in 0.2. Implement Byte/Short as signed INT32 +
+ `intType(8/16, true)`, with range validation and annotated nested-width inference. Encode LocalTime as
+ TIME(NANOS) to preserve nanoseconds, and Instant as adjusted TIMESTAMP(MICROS). Reject Instant
+ values with sub-microsecond precision rather than silently truncating, and validate int64
+ epoch-micros overflow using checked arithmetic, including negative/pre-epoch instants. Test
+ historical dates beyond the int64-nanos range, exact microseconds, fractional-microsecond
+ rejection and extreme Instant bounds. MICROS has a wider range than NANOS but still cannot
+ represent Instant.MIN/MAX; do not promise those. No NANOS write option is required in this
+ release. New-format metadata restores these exact Java types; document external schema-inference
+ types separately and preserve existing java.sql.Time/LocalDateTime contracts. Encode UUID as 16
+ bytes with `uuidType()` in the specification's byte order. Character becomes STRING; enums become
+ STRING using `name()`, with canonical String metadata at every supported nesting level. Reject
+ ZonedDateTime, OffsetDateTime and remaining custom classes before writing rows; messages describe
+ explicit Instant/String conversion or column retyping.
+
+- 2.0c [x] Unify column type resolution and validation. Resolve every column through one validated
+ schema/type descriptor used for physical schema, value encoding, decimal settings and emitted
+ metadata. For Object and abstract/interface supertypes (Number, CharSequence, Comparable,
+ Serializable, Temporal), scan all non-null values, require assignability to the boxed declaration
+ and one supported policy type, then emit that encoding's canonical name. Never decide from only
+ the first sample. For concrete declarations, including non-final concrete classes, require each
+ non-null value to be an instance after primitive boxing normalization; additionally validate the
+ single encoding policy across subclasses. A non-final modifier alone does not turn a concrete
+ declaration into an inference-only supertype. Report the first offending zero-based row, declared
+ type, runtime class and expected policy before output mutation. Resolve supported concrete
+ all-null columns from their declaration; Object or abstract/interface all-null columns use
+ canonical String only if compatible with the declaration, otherwise require an explicit supported
+ type. Thus all-null Number/Temporal columns need explicit retyping rather than false String
+ metadata. For supported concrete declarations, the single encoding policy comes from the runtime
+ values, while metadata retains the declared supported type unless an explicit canonicalizing
+ policy (Character/Enum) applies. Empty columns use the declared policy. Mixed Date/Timestamp
+ values fail even though Timestamp is assignable to Date, because MILLIS/MICROS policies differ;
+ never dispatch incompatible units per value. Add a legacy characterization test for Date first,
+ Timestamp later, recording the current 1000× timestamp corruption, and sql.Date first, Timestamp
+ later, recording the current write failure. New tests require rejection before output mutation in
+ either order, plus homogeneous temporal-subclass round trips with declared Date metadata. Test
+ Object/supertype inference, concrete mismatch, boxed normalization and all-null rules separately.
+ Test Number/CharSequence/Comparable/Serializable/Temporal declarations, and reject a concrete
+ BigDecimal column containing Integer values before writing.
+
+- 2.0d [x] Route every resolved BigDecimal column, declared or resolved from Object/Number/other
+ supertypes, through shared `inferDecimalPrecisionAndScale` and column-name `decimalMeta` handling
+ before schema construction. Use explicit per-column settings first, explicit global
+ precision/scale next, then inference if enabled. With `inferPrecisionAndScale(false)`, require
+ complete explicit settings for each resolved BigDecimal field; reject missing settings rather than
+ falling back to DOUBLE. Apply existing rounding/precision validation to all these paths. Always
+ emit FIXED_LEN_BYTE_ARRAY + DECIMAL and canonical BigDecimal metadata together; remove the DOUBLE
+ fallback for declared BigDecimal too. Handle nested decimals with the same resolved-field policy;
+ document nested metadata addressing and reject unresolved settings. Use a documented minimum
+ precision/scale (1, 0) for all-null declared BigDecimal when inferring. Test untyped Groovy
+ literals `[1.5, 2.25]`, high-precision values that DOUBLE cannot represent, negative scales,
+ Number-declared decimals, per-column overrides and inference disabled with and without explicit
+ settings. Assert exact BigDecimal values, runtime classes, declared Matrix types, physical DECIMAL
+ metadata and all-null behavior across entry points. General declared-type validation tests belong
+ to 2.0c. Canonical metadata must never claim BigDecimal over a DOUBLE or INT32 encoding.
+
+- 2.0e [x] Verify metadata coverage, migration contracts and cross-entry-point behavior. Write the
+ canonical type name instead of Object or the abstract/interface declaration. Canonicalized
+ Character/Enum values follow the String policy; do not silently widen heterogeneous numeric types.
+ Reject mixed or unsupported policy types with the first offending row index (document zero-based),
+ actual class and expected policy. All-null Object columns retain STRING/BINARY with canonical
+ String metadata. Validate nested inferred schemas consistently before writing. Do not discover
+ unsupported types halfway through overwriting an output file. Include null/all-null columns and
+ boxed primitive normalization in schema validation. Add ordinary/optimized and file/bytes/SPI
+ round-trip tests for each supported policy, nested variants where supported, and contextual
+ rejection tests for each rejected class. Characterize legacy text separately: it remains String,
+ not reconstructed Java objects. Assert writer-emitted matrix.columnTypes names are a subset of the
+ resolver allowlist across all supported types and canonicalized columns, including Byte/Short and
+ containers. Never emit java.lang.Object or arbitrary enum/custom class names in new files; old
+ unknown names continue to use the schema fallback. Test the untyped builder example with
+ Integer/String data, homogeneous nullable Object columns, all-null Object columns and mixed
+ `[1, 'x']` rejection before output mutation. Assert inferred canonical metadata names and resulting
+ Matrix types, including Object columns of each supported policy type. Extend migration notes with
+ each 0.2 row. For each of 2.0a–2.0e, run the ordered module checks in 6.3 and record the
+ successful commands/results before checking off that task; do not use a single completion checkbox
+ for all five areas.
+
+- 2.1 [x] Add byte-preserving inference/conversion for raw binary fields, including binary inside
+ nested media structs. Retain string decoding for STRING annotations and DECIMAL handling and the
+ 2.0 legacy/override rules. Document the compatibility impact of changing external raw-binary
+ columns from String to byte[]. Binary writer/reader round-trip support in 2.0a is mandatory, not
+ conditional.
+
+- 2.2 [x] Apply DECIMAL scale to INT32/INT64 physical values, including nested values; test
+ negative, null, scale-zero and fractional values against PyArrow expectations.
+
+- 2.3 [x] Respect TIMESTAMP `isAdjustedToUTC`: adjusted timestamps use the configured zone;
+ unadjusted timestamps retain local wall-clock fields via UTC epoch arithmetic
+ (`LocalDateTime.ofEpochSecond(instant.epochSecond, instant.nano, ZoneOffset.UTC)`). Cover
+ MILLIS/MICROS/NANOS, pre-epoch values, nested timestamps, DST boundaries and
+ UTC/Europe-Stockholm/America-New_York. Generate naive and timezone-aware PyArrow fields and
+ assert their footer annotations; preserve adjusted Matrix round trips. Specify Date/Timestamp
+ metadata overrides explicitly rather than silently assigning an instant to local timestamps.
+
+- 2.4 [x] Add UUID, FLOAT16 and INT96 fixtures. Decode UUID to `UUID`; define FLOAT16 widening to
+ `Float` with special-value tests, or fail explicitly if unsupported in the first delivery.
+ Characterize INT96 failure and implement documented legacy timestamp semantics or a contextual
+ unsupported-type error. Never silently decode these as text. Cover compliant, two-level and
+ legacy LIST layouts with explicit support/error contracts.
+
+- 2.5 [x] Keep these tests paired across ordinary and optimized layouts so failures are attributed
+ to schema conversion rather than CDC. Report unsupported logical types explicitly; broaden
+ beyond these cases only when a fixture demonstrates need.
+
+### 3. Expose selective local reads using upstream page filtering (P1)
+
+- 3.1 [x] Build a shared `InputFile` core using `LocalInputFile` and `PlainParquetConfiguration`;
+ use `LocalOutputFile`/`OutputFile` and plain configuration on file writes too. Preserve naming,
+ time-zone and ownership behavior across legacy, builder and SPI paths. Add an in-memory seekable
+ `InputFile` for byte arrays, replacing their temp-file copy. Test seeks, EOF and independent
+ streams without full-buffer copies. Use `ParquetFileReader.open(InputFile, ParquetReadOptions)`
+ and an explicit per-row-group loop, with `getRowGroups()`,
+ `readFilteredRowGroup(int)`/`readRowGroup(int)` and `MessageColumnIO.getRecordReader(pages,
+ materializer, filter)`. Set requested schema before filtering. Keep original footer group IDs
+ stable even if pruning removes groups from an internal list; spike and test this mapping before
+ exposing selectors. Close each `PageReadStore` on advancement/failure. Do not build the new core
+ around `ParquetReader`. Its 1.18.1 API does expose current-row-group index, but does not provide
+ direct indexed group selection; explicit groups fit batching and independent workers better.
+ Follow the upstream [internal reader loop](https://github.com/apache/parquet-java/blob/apache-parquet-1.18.1/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordReader.java),
+ without depending on that package-private class. Keep accept/rollback bookkeeping in the Matrix
+ materializer using the hooks in 3.2; the loop respects skipped/null records without duplicating
+ column-list truncation. Spike this integration before removing Hadoop paths. Test
+ SNAPPY/ZSTD/GZIP/uncompressed read/write with hadoop-common/mapreduce excluded, and inspect
+ linkage/dependency reports. Keep required codec classes until this proves removal safe;
+ distinguish avoiding filesystem initialization from removing dependencies. Do not raise the JDK
+ target as part of this change.
+
+- 3.2 [x] Replace `GroupReadSupport` with a custom `ReadSupport`/`RecordMaterializer` whose
+ converters append to typed column lists. Precompute schema/converter mappings and avoid per-cell
+ name lookup, per-row Group/list/Object[] creation and row-to-column transposes. Spike Matrix
+ construction from columns to verify ownership/copy behavior. Preserve nested list/map/struct
+ nulls, empty collections and parent-row alignment. Any active row filter, including ordinary
+ pushed-down record filtering, requires transactional output: converters may append values before
+ acceptance is known. Snapshot each output/filter-column list size at row start. Accept in Matrix
+ `RecordMaterializer.getCurrentRecord()`, which the filtering wrapper calls only after its
+ pushed-down predicate passes: evaluate any residual predicate there before accepting and return
+ null on rejection. Return a non-null lightweight record token for acceptance, including all-null
+ rows. Roll back to the snapshots in `skipCurrentRecord()`, reached through parquet-java's
+ skipped/null-record handling, and reset nested assembly state. Keep rollback idempotent. Mid-row
+ exceptions must invoke the same materializer-owned abort/rollback operation before propagating;
+ snapshots also protect this path. Test the actual hook ordering with pushed-down and residual
+ rejection. Test rejected first/middle/last rows and entire rejected groups, ensuring every
+ column has the same accepted-row count. Use staging where nested assembly requires it, without
+ allocating a generic row buffer for every unfiltered flat row. Differential-test against the old
+ reader before adding new projection/filter options; preserve schema fixes from phase 2 in shared
+ conversions. A ReadSupport adapter may initialize the converter tree, but the explicit
+ ParquetFileReader loop owns group selection and requested schema.
+
+- 3.3 [x] Implement dictionary-aware string reuse in converters: decode each dictionary ID once
+ per active dictionary and reuse the String reference, supporting dictionary changes and
+ plain-page fallback. No global `String.intern()` or unbounded cross-file cache. Verify identity
+ reuse on categorical data and exact values after dictionary fallback.
+
+- 3.4 [x] Add the proposed options below, including `toMap`, `fromMap`, descriptors, validation,
+ builder methods and public GroovyDoc. Use Matrix-owned immutable predicate types, translated
+ internally; no upstream predicate classes or closures in the new public contract. Reject unknown
+ columns, duplicates and incompatible predicate types before reading. Start projection with
+ top-level column names.
+
+ | Proposed option | Type | Default | Meaning |
+ |---|---|---|---|
+ | `columns` | `List` | `null` | All columns; supplied nonempty list selects output columns in requested order |
+ | `filter` | Matrix-owned `ParquetPredicate` | `null` | Logical-schema predicate with exact record filtering |
+ | `useColumnIndexFilter` | `boolean` | `true` | Allow upstream page pruning when available; false is a comparison/diagnostic path |
+
+- 3.5 [x] Pass translated upstream predicates via `FilterCompat.get(predicate)`. The core loop
+ sets the union of output columns and all pushdown/residual filter columns via
+ `ParquetFileReader.setRequestedSchema(MessageType)` before filtering/group reads. ReadSupport
+ only initializes the matching converter/materializer tree; it does not own a second projection
+ implementation. Keep exact record filtering enabled. Read filter-only columns internally and
+ exclude them from the resulting Matrix. Add `Predicates.eq/ne/lt/le/gt/ge`, `isNull/isNotNull`,
+ `and/or/not` with typed value overloads for Number, String, Boolean, LocalDate, LocalDateTime,
+ Instant, LocalTime and byte[]. Convert dates, timestamp/time units, UTC adjustment, decimals
+ (including integer and fixed binary physical forms) using the actual file schema. Reject
+ overflow, lossy rescaling, incompatible local/instant comparisons and unsupported repeated-field
+ predicates. Implement exact logical residual evaluation when safe pushdown is unavailable; do
+ not incorrectly compare signed decimal bytes as generic lexicographic binary. Partition
+ pushdown/residual expressions safely, especially under OR/NOT. Document null/NaN semantics and
+ test them against full evaluation.
+
+- 3.6 [x] Remap `matrix.columnTypes` by original field name for projected/reordered columns.
+ Restore Matrix index metadata only when all indexed columns survive; otherwise omit that Matrix
+ index. Empty matches must still retain the selected schema and types.
+
+- 3.7 [x] Test result equivalence with index filtering on/off and indexes absent. Cover AND/OR,
+ null predicates, unordered data, truncated string bounds, boundary values and filter-only
+ columns. Treat unavailable indexes as a normal scan fallback; propagate genuine corruption
+ rather than silently returning partial results.
+
+- 3.8 [x] Prove page skipping using a counting seekable input and controlled multi-page fixture.
+ Disable row-group/dictionary/bloom pruning in this focused test so an entire row-group skip
+ cannot masquerade as page-index use. Require identical results and fewer data-page bytes read;
+ avoid timing assertions and exact read-call counts.
+
+- 3.9 [x] Add `limit(long)` (default unlimited/null, nonnegative; zero returns typed empty
+ output), `rowGroups(List)` (default all, zero-based, unique IDs in caller-specified
+ order), and `batches(...)` returning a Matrix-owned closeable iterator of Matrix values,
+ normally one batch per selected row group. Limit counts matching rows after filtering and
+ applies across batches. Footer IDs in range remain valid when statistics, dictionary or bloom
+ pruning excludes them: contribute no rows and skip the empty batch. Only out-of-range or
+ duplicate IDs are rejected. Preserve caller-selected order and test valid pruned IDs. Close
+ resources on exhaustion, explicit close and failure; document try-with-resources/withCloseable.
+ Iterator state stays on one input and does not prefetch unbounded groups. One row group can
+ exceed heap, so also support `maxBatchRows` (positive, default null) to split it; document that
+ byte-array cells/one huge row still constrain memory. Skip empty filtered batches and retain
+ schema for empty materialized results. Test early stop and cleanup.
+
+- 3.10 [x] Measure 3.2/3.3 against 1.6 before removing the old path. Initial acceptance target: at
+ least 20% lower median full-scan elapsed time on wide text and 25% lower peak live heap on
+ categorical data, with no greater than 5% median regression on mixed/nested data. Record
+ repeated-run variance; revise targets explicitly if evidence shows they are unrealistic. Timing
+ thresholds belong in benchmark reports, not flaky CI assertions.
+
+### 4. Add selective HTTP access (P2, depends on 3)
+
+- 4.1 [x] Implement a seekable HTTP `InputFile` / `SeekableInputStream` with bounded range
+ caching. Discover stable length and verify actual `206`/`Content-Range` behavior; fetch
+ footer/index/data ranges on demand. Reuse the same remote-object identity for metadata and rows,
+ detecting changes via validators where available.
+
+- 4.2 [x] Preserve ordinary URL reads through fallback to the existing temporary-file path when a
+ server lacks range support. Proposed `remoteReadMode` enum/string default `DOWNLOAD` preserves
+ current behavior; `AUTO` attempts ranges then downloads safely, and `RANGE` requires verified
+ range support. Add positive `connectTimeoutMillis` and `readTimeoutMillis`, proposed defaults
+ 10000/60000, plus positive `rangeCacheBytes`, proposed default 8388608. Propagate these through
+ builder/options/SPI.
+
+- 4.3 [x] Test redirects typical of Hub resolve URLs, servers ignoring Range, bad range responses,
+ EOF, retries with bounded attempts, resource cleanup and object changes. Do not forward
+ authorization across unrelated redirect origins. Keep credentials out of logs. No new Hub
+ authentication or Xet-client integration is required for public URLs.
+
+- 4.4 [x] Use a local range-capable HTTP server and selective projection/predicate fixture to
+ prove fewer transferred bytes than a full download. Count footer/index overhead too. Document
+ that this is HTTP range access, not Xet deduplication or a streaming Matrix.
+
+### 5. Confirm writing behavior; defer native CDC to a separate decision (P1/P3)
+
+- 5.1 [x] Based on 1.4, document existing page-index output. If an eligible-column omission is
+ reproducible, fix shared writer configuration across file/memory/stream paths; extract common
+ configuration rather than copying more builder chains. Add page-size tuning through the
+ explicitly planned options in 5.4; benchmarks inform guidance, not whether the options are
+ implemented.
+
+- 5.2 [x] Keep native CDC writing deferred unless a supported parquet-java release exposes it or a
+ separately scoped upstream implementation is accepted. No private upstream patch, mandatory
+ Python bridge, or misleading “optimized” switch in this enhancement. Page-size/row-group tuning
+ alone is not CDC.
+
+- 5.3 [x] For users needing CDC output now, document the external PyArrow rewrite below, verify
+ values and custom metadata survive, and state the extra I/O and Python requirement. A future
+ native CDC proposal must cover per-column logical hashing, nested levels, dictionary effects,
+ deterministic output and measured reuse after inserts/deletes.
+
+- 5.4 [x] Expose the safe tuning subset below through typed options, builder, SPI and shared
+ writer configuration. Use a Matrix-owned writer-version enum/string. Validate positive
+ sizes/NDVs, primitive column paths and names against the resolved schema; reject unknown map
+ keys/columns and invalid versions. Keep current defaults explicitly.
+
+ | Option | Type | Proposed default | Meaning |
+ |---|---|---|---|
+ | `rowGroupSize` | long | 134217728 | Target row-group bytes |
+ | `pageSize` | int | 1048576 | Target encoded data-page bytes |
+ | `dictionaryEncoding` | boolean | true | Enable dictionary encoding with upstream fallback |
+ | `writerVersion` | Matrix enum/string | V1 | Select V1/V2 writer encodings and page layout |
+ | `columnIndexTruncateLength` | int | 64 | Binary bounds truncation for page indexes |
+ | `statisticsTruncateLength` | int | 2147483647 | Binary bounds truncation for row-group statistics |
+ | `bloomFilterEnabled` | Map | empty | Per-column overrides, globally disabled |
+ | `bloomFilterNDV` | Map | empty | Per-column expected distinct values; omission retains upstream sizing |
+ | `sizeStatisticsEnabled` | boolean | true | Collect upstream size statistics |
+
+ V2 selects a different encoder factory as well as page layout; do not describe this as just a
+ page-header toggle. Keep V1 default and document that older consumers may not support all V2
+ encodings; name specific consumers only after compatibility testing. Describe target-size versus
+ hard-limit semantics, NDV memory tradeoffs and disabled-column interactions. Test physical
+ metadata/page versions and logical equivalence across every output path. Add indexed equality
+ tests proving bloom-filter pruning independently from page/dictionary/row-group-statistics
+ pruning. Match any changed defaults to versioned upstream constants before implementation.
+
+- 5.5 [x] Track native CDC in P3: investigate [parquet-java
+ #3818](https://github.com/apache/parquet-java/pull/3818), referenced by
+ [Arrow #51684](https://github.com/apache/arrow/issues/51684) as a Java CDC implementation. The
+ Java PR remains open and unmerged as verified on 2026-10-17; 1.18.1 exposes no supported
+ CDC builder API. Recheck before choosing a future dependency. Link the applicable issue/PR and supported release here,
+ or draft a concrete upstream issue covering hooks/page boundaries, nested hashing and
+ interoperability. Ask for authorization before posting externally; a reviewable issue draft is
+ sufficient until then. Do not silently leave deferral untracked.
+
+### 6. Documentation, measurements and completion gates
+
+- 6.1 [x] Update module README, `docs/cookbook/matrix-parquet.md`,
+ `docs/tutorial/11-matrix-parquet.md` and release notes. Distinguish tested full-file
+ compatibility, selective local reading, selective HTTP reading and CDC writing. Include complete
+ examples, every new parameter/default, supported types and fallback behavior. Do not advertise
+ arbitrary ordinal random access from a filter API; `withFileRange` is a byte range, not a row
+ interval. Ordinal seeking is a separate feature.
+
+- 6.2 [x] Record a reproducible benchmark for full scan, narrow projection and selective
+ predicates on ordinary/indexed/CDC+indexed files: rows, file/page/row-group sizes, compression,
+ returned rows, bytes read/transferred, wall time and peak memory. Keep timing benchmarks outside
+ normal correctness tests; compare full-scan results to 1.6 and the measured targets in 3.10.
+
+- 6.3 [x] For each implementation phase run, in order: `./gradlew :matrix-parquet:codenarcMain`,
+ `./gradlew :matrix-parquet:spotlessCheck`, `./gradlew :matrix-parquet:test`. Run `./gradlew
+ test` before completing implementation. Record successful commands and outcomes here or in the
+ PR before checking tasks off. Include `codenarcTest` for changed tests.
+
+### 7. Sharded datasets and bounded parallel decoding (P2)
+
+- 7.1 [x] Add `read(List)`, directory input and explicit `readGlob(String)` using caller
+ order for lists and sorted normalized-path order for directories/globs; reject duplicate
+ normalized paths. No implicit glob interpretation in ordinary file strings. Directory reads
+ select direct `.parquet` children by default, not recursive trees. Reject empty selections with
+ a clear error. Check all schemas before collecting results: strict logical types, field
+ names/order, decimal precision/scale and timestamp units/UTC flags by default. Do not silently
+ coerce incompatible shards. Test Matrix metadata policy separately from physical schema
+ compatibility.
+
+- 7.2 [x] Add `sourceFileColumn` (String, default null; reject collisions) containing each shard's
+ normalized absolute local path. Apply projection/predicates independently per file schema and a
+ single matching-row limit across the dataset. Define dataset row-group selectors as explicit
+ per-file selections rather than ambiguous global group numbers. Extend closeable batches to
+ shards while preserving deterministic file/group order.
+
+- 7.3 [x] Add `parallelism` (positive int, default 1) and bounded prefetch (default one work item
+ per worker). Decode independent files/groups on separate readers; never share a seekable stream,
+ converter or mutable dictionary. Reuse each file's already-parsed footer through
+ `ParquetFileReader.open(InputFile, ParquetMetadata, ParquetReadOptions, SeekableInputStream)`
+ with a separate stream per worker. Verify the source identity/length matches the footer snapshot
+ and close streams on constructor/open failures. Test one footer parse per file rather than per
+ worker; do not share mutable pruning state between workers. Preserve order, cap queued results,
+ cancel remaining work on limit/close/failure and propagate errors with source context. Measure
+ throughput/heap against sequential decoding and test resource cleanup and deterministic results;
+ avoid decoding whole datasets before returning the first batch.
+
+### 8. Optional Hub URL resolution (P3)
+
+- 8.1 [x] Specify and test `hf://datasets//@/` resolution to an HTTPS
+ `/datasets///resolve//` URL; revision defaults to `main`. Match
+ huggingface_hub parsing/quoting, including percent-encoded revision slashes
+ (`refs%2Fconvert%2Fparquet`) and supported special unescaped refs. Pin compatibility to
+ huggingface_hub **v2.2.0** `parse_hf_uri` and URL quoting, after comparing v1.0.0 `resolve_path`,
+ `unquote`, special-ref regex and `safe_revision`. The new convert-ref regex allows dots/dashes;
+ deep-path @ is literal. Record differences and regenerate offline differential reference cases
+ with the adopted tagged client. Preserve unescaped refs/convert/parquet and refs/pr/N; percent-encode
+ other slash-containing revisions according to the selected client. Add differential URL tests
+ for @, nested paths, encoded and special refs, Unicode and avoiding double decoding/encoding.
+ [Client implementation](https://github.com/huggingface/huggingface_hub/blob/v2.2.0/src/huggingface_hub/hf_file_system.py).
+ Reuse phase 4 transport without adding Xet claims. Optional HF_TOKEN is sent only to approved
+ `https://huggingface.co` origins, never signed CDN redirect origins or logs; test redirect
+ behavior with mocked origins. Verify current Hub URL/auth documentation before implementation.
+ Existing public HTTPS reads need no token.
+
+Execution order: **P0: 0–2; P1: 3 and 5.1/5.4; P2: 7 and 4; P3: 8 and 5.5**.
+Documentation and completion checks in 6 apply to every delivery. Phase 5.2/5.3 explains
+CDC deferral and the external rewrite. Native CDC writing is not a prerequisite for reading
+optimized downloads correctly. The execution record below describes the delivered implementation.
+
+## Usage examples and fixture reproduction
+
+API for a downloaded optimized file (compatibility verified by phase 1):
+
+```groovy
+import se.alipsa.matrix.parquet.MatrixParquetReader
+
+def data = MatrixParquetReader.read(new File('downloaded-optimized.parquet'))
+```
+
+Phase 3 API; values convert through the file schema:
+
+```groovy
+import se.alipsa.matrix.parquet.Predicates
+import se.alipsa.matrix.parquet.MatrixParquetReader
+
+def subset = MatrixParquetReader.builder()
+ .columns(['id', 'text'])
+ .filter(Predicates.ge('id', 1000L))
+ .limit(100)
+ .useColumnIndexFilter(true)
+ .read(new File('downloaded-optimized.parquet'))
+```
+
+Additional P1/P2 usage:
+
+```groovy
+import java.time.LocalDate
+import se.alipsa.matrix.parquet.MatrixParquetReader
+import se.alipsa.matrix.parquet.Predicates
+import se.alipsa.matrix.parquet.MatrixParquetWriter
+
+// Logical DATE comparisons require no epoch-day conversion in application code.
+def preview = MatrixParquetReader.builder()
+ .filter(Predicates.ge('date', LocalDate.of(2026, 1, 1)))
+ .limit(20)
+ .read(new File('events.parquet'))
+
+// Close explicitly even if consumption stops early; limit counts matches globally.
+MatrixParquetReader.builder().limit(1000).maxBatchRows(250)
+ .batches(new File('events.parquet')).withCloseable { batches ->
+ while (batches.hasNext()) {
+ def batch = batches.next()
+ // Process this Matrix before requesting the next bounded batch.
+ }
+ }
+
+MatrixParquetWriter.builder(preview)
+ .rowGroupSize(16L * 1024 * 1024)
+ .pageSize(64 * 1024)
+ .dictionaryEncoding(true)
+ .writerVersion('V2')
+ .write(new File('preview.parquet'))
+
+// Shard order follows this supplied list; directory/glob APIs sort paths.
+def shards = MatrixParquetReader.builder().sourceFileColumn('_source').parallelism(2)
+ .read([new File('train-00000.parquet'), new File('train-00001.parquet')])
+```
+
+For explicit lists, preserve caller order (and reject duplicate normalized paths); for
+folder/glob selections use sorted normalized paths. Carry this distinction into 7.1.
+
+Minimal end-to-end external rewrite for CDC output. Use a dedicated environment and pin
+the generator version in the fixture manifest; 21.0.0 is a suitable minimum baseline:
+
+```bash
+python3 -m venv /tmp/matrix-parquet-fixtures
+/tmp/matrix-parquet-fixtures/bin/python -m pip install pyarrow==21.0.0
+/tmp/matrix-parquet-fixtures/bin/python - <<'PY'
+import pyarrow.parquet as pq
+
+# input.parquet is an existing matrix-parquet output file.
+table = pq.read_table('input.parquet')
+pq.write_table(
+ table, 'optimized.parquet', compression='snappy',
+ use_content_defined_chunking=True, write_page_index=True,
+)
+assert pq.read_table('optimized.parquet').equals(table, check_metadata=True)
+PY
+```
+
+Here `table` is the full Arrow table, the second argument is the output path,
+`compression` explicitly chooses SNAPPY (also PyArrow's default), CDC defaults to false
+unless enabled, and page-index writing defaults to false unless enabled. Other settings
+use the pinned PyArrow version's defaults. This example loads the table into memory;
+the eventual fixture generator must explicitly set row-group/page/encoding options and
+emit all four CDC/index combinations plus its manifest. It must not require Hub uploads.
+
+
+## Implementation and verification record — 2026-10-17
+
+All numbered deliverables above are implemented and verified in this branch. Native CDC writing
+is intentionally deferred as specified by 5.2; its upstream tracking and external rewrite are
+complete. The module remains on parquet-java 1.18.1, Hadoop 3.5.0 and JDK 21.
+
+### Final ordered gates
+
+These successful commands verify the delivered tree, including each separate 2.0a–2.0e task:
+
+```bash
+./gradlew :matrix-parquet:codenarcMain :matrix-parquet:codenarcTest --console=plain
+./gradlew :matrix-parquet:spotlessCheck --console=plain
+PARQUET_PYTHON=/tmp/matrix-parquet-fixtures/bin/python \
+ ./gradlew :matrix-parquet:test :matrix-parquet:compileConsumerApi -PrunSlowTests=true --console=plain
+./gradlew test -Pheadless=true --console=plain
+```
+
+CodeNarc and Spotless passed. The Parquet suite reported **122 tests, 117 passed, zero failures,
+five skipped**. The skips are opt-in benchmarks and original-writer fixture capture; default-size
+CDC was generated and tested. The Java consumer compiles without parquet-java/Hadoop on its
+compile classpath. The full repository regression task passed; unaffected tests may be up-to-date.
+
+| Task | Implementation and independently asserted coverage |
+|---|---|
+| 0.1–1.5 | 0.8.0 migration table; original 0.7.x strings/temporal/decimal fixtures; eight Arrow layout cases and manifest; physical indexes/pages; all primary entry points |
+| 2.0 | Recursive STRING annotations, version marker, automatic legacy nested text compatibility and binaryAsString options |
+| 2.0a | Raw binary writes and exact reads through bytes/files/streams/SPI; fixed metadata allowlist; unknown/incompatible metadata fallback; explicit legacy `[B` rejection |
+| 2.0b | Byte/Short/LocalTime/Instant/UUID/Character/Enum policies; precision/overflow checks; contextual unsupported-type rejection |
+| 2.0c | Shared ColumnTypePolicy descriptor; complete scans, abstract/interface inference, concrete assignability, all-null rules and incompatible Date subclass rejection in either order |
+| 2.0d | Exact inferred/declared/nested BigDecimal DECIMAL routing; per-field/global precedence, disabled-inference errors, high precision, negative scale and all-null cases |
+| 2.0e | Allowlist coverage for all emitted canonical types; every supported policy through bytes, streams and SPI files; validation before output mutation and migration notes |
+| 2.1–2.5 | External raw media bytes, integer decimals, naive/adjusted timestamps at every unit, UUID and explicit FLOAT16/INT96 errors, legacy LIST variants |
+| 3.1–3.9 | InputFile/ParquetFileReader core, column converters, accept/skip rollback, dictionary reuse, logical predicates, projection/index remapping, physical pruning, limits and closeable bounded batches |
+| 4.1–4.4 | Range input, fallback, cache/timeouts, bounded redirects/retries, origin-scoped tokens, malformed/truncated ranges, object changes and fewer transferred bytes |
+| 5.1–5.5 | Physical output indexes across file/memory/stream paths, shared tuning, bloom-only pruning, metadata-preserving Arrow CDC rewrite and upstream CDC tracking |
+| 6.1, 6.3 | Complete README/defaults/examples, tutorial/cookbook/release updates; ordered gates and full repository suite |
+| 1.6, 3.10, 6.2, 7.3 | Original wide baseline, million-row Group/columnar comparisons, sampled live heap, three Arrow layouts with full/projection/filter modes and sequential/parallel dataset measurements |
+| 7.1–7.3 | Ordered list/directory/glob shards, strict schemas/metadata, source column, per-file groups, reused footers, bounded independent workers, limit and early-close cleanup |
+| 8.1 | Tagged huggingface_hub 2.2.0 offline differential cases after v1.0.0 comparison; revision/path quoting and redirect credential tests |
+
+### Reproduction and additional successful checks
+
+```bash
+/tmp/matrix-parquet-fixtures/bin/python matrix-parquet/src/test/python/generate_optimized.py
+/tmp/matrix-hf-fixtures/bin/python matrix-parquet/src/test/python/generate_hub_urls.py
+/tmp/matrix-parquet-fixtures/bin/python matrix-parquet/src/test/python/verify_matrix_output.py
+./gradlew :matrix-parquet:test --tests ParquetInteropTest --console=plain
+./gradlew :matrix-parquet:codecWithoutHadoop --console=plain
+groovy scripts/release-all/test-workflow.groovy .
+```
+
+Fixture regeneration is byte-for-byte deterministic on the pinned environments. Checked-in
+Parquet binaries total **737,019 bytes**, within the 4 MiB budget and 1 MiB individual cap.
+Python environments use pyarrow==21.0.0 and huggingface_hub==2.2.0 respectively. Python verification
+confirmed SNAPPY/ZSTD annotated strings, binary payloads and exact decimals; CDC rewrites retain
+values and Matrix metadata and are reread by Java. CI runs these checks in
+.github/workflows/release-all-tools.yml. Fixture provenance and client-version differences are
+recorded in src/test/resources/optimized/README.md.
+
+Original-writer capture was run successfully with:
+
+```bash
+./gradlew :matrix-parquet:test --tests ParquetLegacyCaptureTest \
+ -I /tmp/parquet-legacy.init.gradle --no-configuration-cache --console=plain
+```
+
+That init script sets test system property parquet.captureLegacy=true. The capture compiles the
+writer/value helper at b2095159664b34ce426685c387adf0bccdb33e01 together in an isolated loader.
+The byte[] hypothesis was corrected by this end-to-end experiment: this pinned writer throws
+InvalidSchemaException and cannot produce a valid byte[] fixture. The checked-in `[B` text footer
+is explicitly synthetic; no original bytes are claimed recoverable. Date/Timestamp corruption is
+preserved in an actual original-writer fixture and new mixed policies are rejected before output.
+
+The codec-exclusion spike showed that even UNCOMPRESSED writer initialization loads Hadoop's
+ZlibCompressor.CompressionLevel. Local filesystem/configuration setup is removed from the new
+read/write core, but runtime Hadoop dependencies remain until upstream codec linkage permits
+removal. No JDK target increase or dependency upgrade is claimed.
+
+### Measurements
+
+See [the benchmark report](../benchmarks/optimized-parquet.md) for hardware/JVM/heap/GC, input
+generation, warmup/variance, allocations, sampled live-heap definitions, exact byte counts and
+reproduction commands. The original wide baseline was captured before production edits; the
+three-workload differential comparison retains the original Group loop with shared corrected
+schema conversions. Inputs are generated outside Git, and no claim of an OS-cold cache is made.
+
+The million-row comparison measured 89.5% lower wide median time, 43.8% lower categorical sampled
+live heap and 88.2% lower mixed/nested time, meeting the initial targets with the documented
+sampling method. Identical Arrow layouts exercise full scans, ID projection and 100-row selection;
+page indexes reduced selective reads from 1,279,020 to 552,869 bytes (542,383 with CDC). An isolated
+page-index test reads 2,914 versus 82,700 bytes with all other pruning disabled. Independent bloom
+pruning reads 16,737 versus 40,373 bytes. HTTP ranges transfer 88,476 bytes of an approximately 906,000-byte object,
+including footer overhead. Four-worker shard decoding measured a 242 ms warm median versus 341 ms
+sequentially. These are local observations, not timing assertions or general speed guarantees.
+
+Benchmark commands (init-file setup is fully documented in the report):
+
+```bash
+./gradlew :matrix-parquet:test --tests '*ParquetReadBenchmarkTest' \
+ -I /tmp/parquet-baseline.init.gradle --no-configuration-cache --console=plain
+./gradlew :matrix-parquet:test --tests '*ParquetDatasetBenchmarkTest' \
+ -I /tmp/parquet-dataset.init.gradle --no-configuration-cache --console=plain
+/tmp/matrix-parquet-fixtures/bin/python matrix-parquet/src/test/python/generate_benchmarks.py
+./gradlew :matrix-parquet:test --tests '*ParquetLayoutBenchmarkTest' \
+ -I /tmp/parquet-layout.init.gradle --no-configuration-cache --console=plain
+```
+
+
+The subsequent PR review fixes and their verification are recorded in
+[PR #481 review fixes](optimized-parquet-review.md).
diff --git a/matrix-parquet/req/optimized-parquet-review.md b/matrix-parquet/req/optimized-parquet-review.md
new file mode 100644
index 000000000..a0f21a03e
--- /dev/null
+++ b/matrix-parquet/req/optimized-parquet-review.md
@@ -0,0 +1,122 @@
+# 9. PR #481 review fixes
+
+The numbered tasks correspond to the thirteen findings in the review. The fixes were developed
+on `feature/optimized-parquet-review`, based on the original PR commit eee2b16e.
+
+- 9.1 [x] Handle empty dictionaries without a descending Groovy range. Add the exact three-row
+ all-null PyArrow string fixture, generator provenance and a read regression.
+- 9.2 [x] Decode JSON and ENUM annotations as UTF-8 String, including dictionary conversion;
+ retain BSON bytes. Test PyArrow JSON and Java ENUM/BSON with binary overrides and update the
+ migration table.
+- 9.3 [x] Separate the 8 MiB request cap from the configured cache ceiling. Bulk reads use the
+ requested span; reuse validated redirect targets, refresh rejected signed targets through the
+ original URL, and leave fully consumed successful range connections eligible for keep-alive.
+ Measure actual request counts with a local redirect/range server.
+- 9.4 [x] Route static URL/URL-string overloads through the builder transport, including named
+ and zoned hf:// overloads. Test bounded timeouts, downgrade rejection and Hub parsing.
+- 9.5 [x] Write local files to temporary siblings and atomically replace the destination only
+ after successful close. Preserve existing POSIX permissions and clean temporary files on
+ failure. Test rounding and precision failures after a valid row and compare original bytes.
+- 9.6 [x] Support sourceFileColumn for local single-file reads and batches. Restore common
+ compatible indices for both collected and batched dataset reads, including truncated batches;
+ omit indices when projection removes their columns or shards disagree on index metadata.
+- 9.7 [x] Resolve each top-level column's validated type policy once and carry it in WriteSchema
+ through decimal/schema construction and metadata emission. Remove obsolete duplicate inference
+ helpers and report full nested logical paths; test nested error context.
+- 9.8 [x] Treat null option-map values as absent consistently, preserving Boolean/default/map
+ values and nullable AUTO semantics. Reject null bloom entries with contextual validation.
+- 9.9 [x] Create dataset worker threads through a daemon ThreadFactory. Assert their daemon
+ status while an iterator remains open; explicit close is still required for prompt cleanup.
+- 9.10 [x] Route named/zoned File/Path/string overloads through the shared directory-capable
+ builder and test directory results, names, time-zone values and index restoration.
+- 9.11 [x] Reject literal dot/parent path segments and decoded revision dot/parent segments;
+ retain the pinned client URL cases. Test nested paths and encoded revision traversal.
+- 9.12 [x] Reject unrecoverable legacy [B metadata only for required output/filter columns.
+ Test successful projection around the damaged column and rejection when a filter needs it.
+- 9.13 [x] Correct the reader's memory/batch/cache comments and writer option documentation,
+ including literal V1 defaults; update source/index/transport behavior in the module guide.
+
+## Verification
+
+The required ordered module gates passed:
+
+```bash
+./gradlew :matrix-parquet:codenarcMain :matrix-parquet:codenarcTest --console=plain
+./gradlew :matrix-parquet:spotlessCheck --console=plain
+PARQUET_PYTHON=/tmp/matrix-parquet-fixtures/bin/python \
+ ./gradlew :matrix-parquet:test :matrix-parquet:compileConsumerApi -PrunSlowTests=true --console=plain
+./gradlew test -Pheadless=true --console=plain
+```
+
+The Parquet suite reported **133 tests, 128 passed, zero failed, five intentional opt-in
+benchmark/capture skips**. Default-size CDC generation/read and the downstream consumer compile
+probe passed. The full repository task passed; unaffected tasks may be up-to-date.
+
+Additional successful checks:
+
+```bash
+/tmp/matrix-parquet-fixtures/bin/python matrix-parquet/src/test/python/generate_optimized.py
+/tmp/matrix-parquet-fixtures/bin/python matrix-parquet/src/test/python/verify_matrix_output.py
+```
+
+Fixture regeneration is deterministic. Checked-in Parquet binaries total **738,113 bytes**,
+below the 4 MiB aggregate and 1 MiB individual limits. Arrow verifies SNAPPY/ZSTD strings,
+binary payloads, exact decimals and metadata-preserving CDC rewrites.
+
+HTTP request-count assertions show one 1 MiB data request after probing, one redirect resolution
+per input, and no extra request when the default cache serves the repeated block. A 1 KiB cache
+still permits the same single 1 MiB bulk request. The test checks byte-array and ByteBuffer
+readFully paths. Successful bodies remain eligible for JDK keep-alive pooling; socket-level
+connection reuse and live Hub latency were not measured. Existing selective-transfer, page-index
+and bloom-filter byte-reduction tests also pass.
+
+## 10. Follow-up output and range-buffer review
+
+- 10.1 [x] Create temporary siblings with ordinary file creation permissions, so new output
+ follows the process umask. Compare new Parquet permissions with an ordinary sibling file on
+ POSIX, while retaining the existing-destination permission regression.
+- 10.2 [x] Resolve existing output paths before creating the sibling and replacing the target.
+ Test that relative symbolic links survive both failed and successful writes, and that the
+ resolved file retains its contents on failure and receives the new rows on success.
+- 10.3 [x] Document the same-directory atomic replacement requirement and
+ `AtomicMoveNotSupportedException` in the release notes and module guide. Keep safe failure
+ rather than introduce a replacement fallback with weaker failure guarantees.
+- 10.4 [x] Read directly into array-backed ByteBuffers using their array offset and current
+ position. Retain the bounded scratch buffer for direct buffers. Exercise sliced buffers with
+ nonzero positions, sentinel bytes outside the slice, direct buffers, EOF and empty reads in
+ the existing HTTP request-count regression.
+
+Follow-up verification passed, in the required order:
+
+```bash
+./gradlew :matrix-parquet:codenarcMain :matrix-parquet:codenarcTest :matrix-parquet:spotlessCheck --console=plain
+PARQUET_PYTHON=/tmp/matrix-parquet-fixtures/bin/python \
+ ./gradlew :matrix-parquet:test :matrix-parquet:compileConsumerApi -PrunSlowTests=true --console=plain
+./gradlew test -Pheadless=true --console=plain
+```
+
+The Parquet suite reported **135 tests, 130 passed, zero failed, five intentional skips**,
+including successful POSIX permission/symlink tests, Python interoperability and consumer API
+compilation. The full repository task passed; unaffected tasks may be up-to-date.
+
+## 11. Dangling symlinks and interrupted-write documentation
+
+- 11.1 [x] Resolve symbolic links even when their destination does not exist, including relative
+ and absolute targets and link chains. Reject cycles before creating temporary output. Test
+ successful target creation and failed writes preserving dangling links and absent targets.
+- 11.2 [x] Explain that abrupt JVM termination may leave `.matrix-parquet-.tmp` siblings
+ beside the resolved output and that unused leftovers can be removed.
+
+Verification passed in order:
+
+```bash
+./gradlew :matrix-parquet:codenarcMain :matrix-parquet:codenarcTest --console=plain
+./gradlew :matrix-parquet:spotlessCheck --console=plain
+PARQUET_PYTHON=/tmp/matrix-parquet-fixtures/bin/python \
+ ./gradlew :matrix-parquet:test :matrix-parquet:compileConsumerApi -PrunSlowTests=true --console=plain
+./gradlew test -Pheadless=true --console=plain
+```
+
+The Parquet suite reported **137 tests, 132 passed, zero failed, five intentional skips**.
+The POSIX dangling-link and cycle tests ran successfully, as did Python interoperability,
+consumer API compilation and the full repository suite; unaffected tasks may be up-to-date.
diff --git a/matrix-parquet/src/consumerTest/java/ParquetConsumer.java b/matrix-parquet/src/consumerTest/java/ParquetConsumer.java
new file mode 100644
index 000000000..23e71f202
--- /dev/null
+++ b/matrix-parquet/src/consumerTest/java/ParquetConsumer.java
@@ -0,0 +1,20 @@
+import java.io.File;
+import java.util.List;
+import se.alipsa.matrix.core.Matrix;
+import se.alipsa.matrix.parquet.MatrixParquetReader;
+import se.alipsa.matrix.parquet.MatrixParquetWriter;
+import se.alipsa.matrix.parquet.Predicates;
+
+/** Compile-only check: consumers need no parquet-java or Hadoop on their compile classpath. */
+public final class ParquetConsumer {
+ /** Selects an ID preview without upstream predicate types. */
+ public static Matrix select(File file) {
+ return MatrixParquetReader.builder().columns(List.of("id"))
+ .filter(Predicates.ge("id", 1000L)).limit(10).read(file);
+ }
+
+ /** Writes using Matrix-owned tuning methods. */
+ public static void save(Matrix matrix, File file) {
+ MatrixParquetWriter.builder(matrix).pageSize(4096).writerVersion("V2").write(file);
+ }
+}
diff --git a/matrix-parquet/src/main/groovy/se/alipsa/matrix/parquet/ColumnMaterializer.groovy b/matrix-parquet/src/main/groovy/se/alipsa/matrix/parquet/ColumnMaterializer.groovy
new file mode 100644
index 000000000..6e92eb0ca
--- /dev/null
+++ b/matrix-parquet/src/main/groovy/se/alipsa/matrix/parquet/ColumnMaterializer.groovy
@@ -0,0 +1,175 @@
+package se.alipsa.matrix.parquet
+
+import groovy.transform.PackageScope
+
+import org.apache.parquet.column.Dictionary
+import org.apache.parquet.io.api.Binary
+import org.apache.parquet.io.api.Converter
+import org.apache.parquet.io.api.GroupConverter
+import org.apache.parquet.io.api.PrimitiveConverter
+import org.apache.parquet.io.api.RecordMaterializer
+import org.apache.parquet.schema.GroupType
+import org.apache.parquet.schema.LogicalTypeAnnotation
+import org.apache.parquet.schema.MessageType
+import org.apache.parquet.schema.PrimitiveType
+import org.apache.parquet.schema.Type
+
+/** Column-oriented materialization; record acceptance and rollback remain in upstream hooks. */
+@PackageScope
+final class ColumnMaterializer extends RecordMaterializer {
+
+ final List columns
+ final Map current = [:]
+ private final GroupConverter root
+ private final int[] sizes
+ private final ParquetPredicate residual
+
+ ColumnMaterializer(MessageType schema, List types, boolean binaryAsString, ParquetPredicate residual) {
+ columns = schema.fields.collect { [] }
+ sizes = new int[columns.size()]
+ this.residual = residual
+ Converter[] children = new Converter[schema.fieldCount]
+ schema.fields.eachWithIndex { Type field, int index ->
+ Closure logicalSink = residual != null && residual.columns().contains(field.name) ? { Object value -> current[field.name] = value } : null
+ children[index] = converter(field, types[index], binaryAsString,
+ { Object value -> columns[index].add(value) }, logicalSink)
+ }
+ root = new GroupConverter() {
+ @Override
+ Converter getConverter(int index) { children[index] }
+
+ @Override
+ void start() {
+ current.clear()
+ columns.eachWithIndex { List column, int index -> sizes[index] = column.size() }
+ }
+
+ @Override
+ void end() {
+ columns.eachWithIndex { List column, int index ->
+ if (column.size() == sizes[index]) { column.add(null) }
+ }
+ }
+ }
+ }
+
+ @Override
+ GroupConverter getRootConverter() { root }
+
+ @Override
+ Boolean getCurrentRecord() {
+ residual == null || residual.test(current) ? Boolean.TRUE : null
+ }
+
+ @Override
+ void skipCurrentRecord() {
+ columns.eachWithIndex { List column, int index ->
+ if (column.size() > sizes[index]) { column.subList(sizes[index], column.size()).clear() }
+ }
+ }
+
+ private static Converter converter(Type type, Class expected, boolean binaryAsString, Closure sink, Closure logicalSink = null) {
+ if (type.primitive) {
+ return new ValueConverter(type.asPrimitiveType(), expected, binaryAsString, sink, logicalSink)
+ }
+ GroupType group = type.asGroupType()
+ Map values = [:]
+ Converter[] children = new Converter[group.fieldCount]
+ group.fields.eachWithIndex { Type child, int index ->
+ children[index] = converter(child, null, binaryAsString) { Object value ->
+ if (child.repetition == Type.Repetition.REPEATED) {
+ ((List) values.computeIfAbsent(child.name) { [] }).add(value)
+ } else {
+ values[child.name] = value
+ }
+ }
+ }
+ new GroupConverter() {
+ @Override
+ Converter getConverter(int index) { children[index] }
+
+ @Override
+ void start() { values.clear() }
+
+ @Override
+ void end() {
+ def logical = group.logicalTypeAnnotation
+ if (logical instanceof LogicalTypeAnnotation.ListLogicalTypeAnnotation) {
+ Type repeated = group.getType(0)
+ List entries = (List) (values[repeated.name] ?: [])
+ boolean wrapped = !repeated.primitive && repeated.asGroupType().fieldCount == 1 &&
+ !(repeated.name in ['array', "${group.name}_tuple".toString()])
+ sink.call(wrapped ? entries.collect { entry -> ((Map) entry)[repeated.asGroupType().getType(0).name] } : entries)
+ } else if (logical instanceof LogicalTypeAnnotation.MapLogicalTypeAnnotation) {
+ Map result = [:]
+ ((List) (values[group.getType(0).name] ?: [])).each { entry -> Map pair = (Map) entry; result[pair.key] = pair.value }
+ sink.call(result)
+ } else {
+ Map result = [:]
+ group.fields.each { Type child -> result[child.name] = values[child.name] }
+ sink.call(result)
+ }
+ }
+ }
+ }
+
+ private static class ValueConverter extends PrimitiveConverter {
+ private final PrimitiveType field
+ private final Class expected
+ private final boolean binaryAsString
+ private final Closure sink
+ private final Closure logicalSink
+ private Object[] dictionary
+ private Object[] logicalDictionary
+
+ ValueConverter(PrimitiveType field, Class expected, boolean binaryAsString, Closure sink, Closure logicalSink = null) {
+ this.field = field
+ this.expected = expected
+ this.binaryAsString = binaryAsString
+ this.sink = sink
+ this.logicalSink = logicalSink
+ }
+
+ private void accept(Object value) {
+ sink.call(MatrixParquetReader.decodePrimitive(value, field, expected, binaryAsString))
+ if (logicalSink != null) { logicalSink.call(logicalValue(value)) }
+ }
+
+ private Object logicalValue(Object value) {
+ Class logicalExpected = field.logicalTypeAnnotation instanceof LogicalTypeAnnotation.TimeLogicalTypeAnnotation ? java.time.LocalTime :
+ field.logicalTypeAnnotation instanceof LogicalTypeAnnotation.TimestampLogicalTypeAnnotation &&
+ ((LogicalTypeAnnotation.TimestampLogicalTypeAnnotation) field.logicalTypeAnnotation).isAdjustedToUTC() ? java.time.Instant :
+ field.logicalTypeAnnotation == null && field.primitiveTypeName == PrimitiveType.PrimitiveTypeName.BINARY ? byte[] : null
+ MatrixParquetReader.decodePrimitive(value, field, logicalExpected, false)
+ }
+
+ @Override
+ void addInt(int value) { accept(value) }
+ @Override
+ void addLong(long value) { accept(value) }
+ @Override
+ void addFloat(float value) { accept(value) }
+ @Override
+ void addDouble(double value) { accept(value) }
+ @Override
+ void addBoolean(boolean value) { accept(value) }
+ @Override
+ void addBinary(Binary value) { accept(value) }
+ @Override
+ boolean hasDictionarySupport() {
+ MatrixParquetReader.isTextLogicalType(field.logicalTypeAnnotation) || expected == String
+ }
+ @Override
+ void setDictionary(Dictionary source) {
+ dictionary = new Object[source.maxId + 1]
+ if (logicalSink != null) { logicalDictionary = new Object[source.maxId + 1] }
+ for (int id = 0; id <= source.maxId; id++) {
+ Binary value = source.decodeToBinary(id)
+ dictionary[id] = MatrixParquetReader.decodePrimitive(value, field, expected, binaryAsString)
+ if (logicalSink != null) { logicalDictionary[id] = logicalValue(value) }
+ }
+ }
+ @Override
+ void addValueFromDictionary(int id) { sink.call(dictionary[id]); logicalSink?.call(logicalDictionary[id]) }
+ }
+}
diff --git a/matrix-parquet/src/main/groovy/se/alipsa/matrix/parquet/ColumnTypePolicy.groovy b/matrix-parquet/src/main/groovy/se/alipsa/matrix/parquet/ColumnTypePolicy.groovy
new file mode 100644
index 000000000..a3549f13c
--- /dev/null
+++ b/matrix-parquet/src/main/groovy/se/alipsa/matrix/parquet/ColumnTypePolicy.groovy
@@ -0,0 +1,105 @@
+package se.alipsa.matrix.parquet
+
+import groovy.transform.PackageScope
+
+import java.lang.reflect.Modifier
+import java.sql.Time
+import java.sql.Timestamp
+import java.time.Instant
+import java.time.LocalDate
+import java.time.LocalDateTime
+import java.time.LocalTime
+
+/** Validated encoding and metadata types shared by schema construction and metadata emission. */
+@PackageScope
+final class ColumnTypePolicy {
+
+ private static final int NANOS_PER_MICRO = 1000
+ static final List SUPPORTED = [String, Integer, Long, Float, Double, Boolean, Byte, Short,
+ BigInteger, BigDecimal, Character, byte[], LocalDate, LocalDateTime, LocalTime, Instant,
+ UUID, Date, java.sql.Date, Time, Timestamp, List, Map]
+ static final Map METADATA_TYPES = SUPPORTED.collectEntries { Class type -> [(type.name): type] }
+
+ final Class encoding
+ final Class metadata
+
+ private ColumnTypePolicy(Class encoding, Class metadata) {
+ this.encoding = encoding
+ this.metadata = metadata
+ }
+
+ static ColumnTypePolicy resolve(String name, List> values, Class declaration) {
+ Class declared = box(declaration ?: Object)
+ boolean infer = declared == Object || declared.interface || Modifier.isAbstract(declared.modifiers)
+ Class policy = null
+ values.eachWithIndex { value, int row ->
+ if (value == null) {
+ return
+ }
+ if (!declared.isInstance(value)) {
+ throw new IllegalArgumentException("Column '$name', row $row: declared ${declared.name}, got ${value.getClass().name}")
+ }
+ if (value instanceof Instant) {
+ if (value.nano % NANOS_PER_MICRO != 0) { throw new IllegalArgumentException("Column '$name', row $row: Instant loses sub-microsecond precision") }
+ try { Math.addExact(Math.multiplyExact(value.epochSecond, 1_000_000L), value.nano.intdiv(NANOS_PER_MICRO)) }
+ catch (ArithmeticException failure) { throw new IllegalArgumentException("Column '$name', row $row: Instant overflows epoch micros", failure) }
+ }
+ Class current = policyType(value.getClass())
+ if (current == null || (policy != null && policy != current)) {
+ throw new IllegalArgumentException("Column '$name', row $row: unsupported/mixed ${value.getClass().name}; expected ${policy?.name ?: 'supported encoding'}")
+ }
+ policy = current
+ }
+ if (policy == null) {
+ if (infer && !(declared in [List, Map])) {
+ if (!declared.isAssignableFrom(String)) {
+ throw new IllegalArgumentException("All-null column '$name' declared ${declared.name} requires explicit retyping")
+ }
+ policy = String
+ } else {
+ policy = policyType(declared)
+ }
+ }
+ if (policy == null) {
+ throw new IllegalArgumentException("Unsupported type ${declared.name} for column '$name'; convert explicitly")
+ }
+ Class metadata = infer || policy in [String, List, Map] ? policy : declared
+ if (!METADATA_TYPES.containsKey(metadata.name)) {
+ throw new IllegalArgumentException("Unsupported declared type ${declared.name} for column '$name'")
+ }
+ new ColumnTypePolicy(policy, metadata)
+ }
+
+ static Class policyType(Class type) {
+ if (type == Character || Enum.isAssignableFrom(type) || CharSequence.isAssignableFrom(type)) {
+ return String
+ }
+ if (Collection.isAssignableFrom(type)) {
+ return List
+ }
+ if (Map.isAssignableFrom(type)) {
+ return Map
+ }
+ SUPPORTED.contains(type) ? type : null
+ }
+
+ static Class box(Class type) {
+ if (!type.primitive) {
+ return type
+ }
+ [ (int): Integer, (long): Long, (float): Float, (double): Double,
+ (boolean): Boolean, (byte): Byte, (short): Short, (char): Character ][type]
+ }
+
+ static int[] decimal(List> values) {
+ int scale = 0
+ int left = 1
+ values.each { value ->
+ if (value instanceof BigDecimal) {
+ scale = Math.max(scale, value.scale())
+ left = Math.max(left, value.precision() - value.scale())
+ }
+ }
+ [left + scale, scale] as int[]
+ }
+}
diff --git a/matrix-parquet/src/main/groovy/se/alipsa/matrix/parquet/DatasetBatches.groovy b/matrix-parquet/src/main/groovy/se/alipsa/matrix/parquet/DatasetBatches.groovy
new file mode 100644
index 000000000..0ce753e36
--- /dev/null
+++ b/matrix-parquet/src/main/groovy/se/alipsa/matrix/parquet/DatasetBatches.groovy
@@ -0,0 +1,200 @@
+package se.alipsa.matrix.parquet
+
+import org.apache.parquet.conf.PlainParquetConfiguration
+import org.apache.parquet.hadoop.ParquetFileReader
+import org.apache.parquet.hadoop.metadata.ParquetMetadata
+import org.apache.parquet.io.LocalInputFile
+
+import se.alipsa.matrix.core.Matrix
+
+import java.util.concurrent.Callable
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+import java.util.concurrent.Future
+
+/** Ordered shard batches with strict schema checks and at most one prefetched batch per worker. */
+final class DatasetBatches implements MatrixBatchIterator {
+ private final List files
+ private final List footers = []
+ private final List
*
* @param matrix the matrix to write
- * @param schema the Parquet MessageType schema
+ * @param resolved the validated physical schema and column policies
* @return byte array containing Parquet data
*/
- private static byte[] writeBytesInternal(Matrix matrix, MessageType schema,
- CompressionCodecName compressionCodec = CompressionCodecName.SNAPPY, RoundingMode roundingMode = RoundingMode.UNNECESSARY) {
- def conf = new Configuration()
+ private static byte[] writeBytesInternal(Matrix matrix, WriteSchema resolved,
+ CompressionCodecName compressionCodec = CompressionCodecName.SNAPPY, RoundingMode roundingMode = RoundingMode.UNNECESSARY, ParquetWriteOptions options = new ParquetWriteOptions()) {
+ MessageType schema = resolved.schema
+ def conf = new org.apache.parquet.conf.PlainParquetConfiguration()
InMemoryOutputFile outputFile = new InMemoryOutputFile()
- def writer = ExampleParquetWriter.builder(outputFile)
+ def writerBuilder = ExampleParquetWriter.builder(outputFile)
.withConf(conf)
.withType(schema)
.withCompressionCodec(compressionCodec)
- .withExtraMetaData(buildExtraMeta(matrix))
- .build()
+ .withExtraMetaData(buildExtraMeta(matrix, resolved))
+ def writer = configureWriter(writerBuilder, schema, options).build()
writer.withCloseable { parquetWriter -> writeRows(parquetWriter, matrix, schema, roundingMode) }
return outputFile.getBytes()
}
@@ -656,18 +715,25 @@ class MatrixParquetWriter {
*
* @param matrix the matrix
* @param explicitDecimalMeta map of column name to [precision, scale] array for BigDecimal columns.
- * @return the Parquet MessageType schema.
+ * @return validated physical schema and metadata policies.
*/
- private static MessageType buildSchema(Matrix matrix, Map explicitDecimalMeta) {
- Map decimalMeta = withBigIntegerPrecision(matrix, explicitDecimalMeta)
+ private static WriteSchema buildSchema(Matrix matrix, Map explicitDecimalMeta, DecimalSchemaPolicy decimals = null) {
+ Map types = matrix.columnNames().collectEntries { String name ->
+ [(name): ColumnTypePolicy.resolve(name, matrix.column(name), matrix.type(name))]
+ }
+ Map decimalMeta = withBigIntegerPrecision(matrix, explicitDecimalMeta, types)
+ DecimalSchemaPolicy policy = decimals ?: new DecimalSchemaPolicy(explicitDecimalMeta ?: [:])
def builder = Types.buildMessage()
matrix.columnNames().each { col ->
def type = matrix.type(col)
def meta = decimalMeta?.get(col)
log.debug("Building schema for column '$col' of type $type with decimal meta: $meta")
- builder.addField(buildParquetType(col, matrix.column(col), type, meta))
+ builder.addField(buildParquetType(col, matrix.column(col), type, meta, false, policy.child(col), types[col]))
}
- return builder.named(matrix.matrixName ?: 'MatrixSchema')
+ Set unknown = new HashSet<>(policy.settings.keySet())
+ unknown.removeAll(policy.used)
+ if (!unknown.empty) { throw new IllegalArgumentException("Unknown/non-decimal metadata paths: $unknown") }
+ new WriteSchema(builder.named(matrix.matrixName ?: 'MatrixSchema'), types)
}
/**
@@ -675,10 +741,10 @@ class MatrixParquetWriter {
* Explicit caller-supplied metadata wins after validating its scale is 0
* and that its precision is large enough to hold every value in the column.
*/
- private static Map withBigIntegerPrecision(Matrix matrix, Map explicitDecimalMeta) {
+ private static Map withBigIntegerPrecision(Matrix matrix, Map explicitDecimalMeta, Map types) {
Map merged = [:] + (explicitDecimalMeta ?: [:])
matrix.columnNames().each { col ->
- if (matrix.type(col) != BigInteger) {
+ if (types[col].encoding != BigInteger) {
return
}
int[] existing = merged[col]
@@ -687,35 +753,19 @@ class MatrixParquetWriter {
throw new IllegalArgumentException(
"decimalMeta['$col'] scale must be 0 for BigInteger columns but was ${existing[1]}")
}
- int requiredPrecision = inferBigIntegerPrecision(matrix, col)
+ int requiredPrecision = inferBigIntegerPrecision(matrix.column(col))
if (existing[0] < requiredPrecision) {
throw new IllegalArgumentException(
"decimalMeta['$col'] precision (${existing[0]}) is too small to hold the data in this " +
"column (requires at least $requiredPrecision digits)")
}
} else {
- merged[col] = [inferBigIntegerPrecision(matrix, col), 0] as int[]
+ merged[col] = [inferBigIntegerPrecision(matrix.column(col)), 0] as int[]
}
}
merged
}
- /**
- * Infers the minimum decimal precision needed to represent every BigInteger
- * value in the given column as Parquet DECIMAL(precision, 0).
- */
- private static int inferBigIntegerPrecision(Matrix matrix, String col) {
- int maxDigits = 1
- (0..
- def value = matrix[i, col]
- if (value instanceof BigInteger) {
- int digits = value.abs().toString().length()
- maxDigits = Math.max(maxDigits, digits)
- }
- }
- maxDigits
- }
-
private static int inferBigIntegerPrecision(List> values) {
int maxDigits = 1
values.each { value ->
@@ -731,16 +781,12 @@ class MatrixParquetWriter {
* Builds the Parquet schema based on the matrix, inferring decimal precision/scale if requested.
* This is for backward compatibility and delegates to the explicit map builder.
* @param matrix the matrix
- * @param inferPrecisionAndScale if true, infers decimal precision and scale. If false, BigDecimal columns default to
- * double storage (which leads to some loss of precision).
- * @return the Parquet MessageType schema.
+ * @param inferPrecisionAndScale if true, infers decimal precision and scale. If false, BigDecimal fields require
+ * explicit precision/scale settings.
+ * @return validated physical schema and metadata policies.
*/
- private static MessageType buildSchema(Matrix matrix, boolean inferPrecisionAndScale = false) {
- Map decimalMeta = null
- if (inferPrecisionAndScale) {
- decimalMeta = inferDecimalPrecisionAndScale(matrix)
- }
- return buildSchema(matrix, decimalMeta)
+ private static WriteSchema buildSchema(Matrix matrix, boolean inferPrecisionAndScale = false) {
+ buildSchema(matrix, [:], new DecimalSchemaPolicy([:], inferPrecisionAndScale))
}
/**
@@ -752,35 +798,35 @@ class MatrixParquetWriter {
* @param decimalMeta optional decimal precision/scale metadata for {@link BigDecimal} columns
* @return a Parquet {@link Type} describing the column
*/
- private static Type buildParquetType(String name, List> values, Class clazz, int[] decimalMeta = null, boolean preferStructForMaps = false) {
- if (isListType(clazz, values)) {
+ private static Type buildParquetType(String name, List> values, Class clazz, int[] decimalMeta = null, boolean preferStructForMaps = false, DecimalSchemaPolicy decimals = new DecimalSchemaPolicy([:]), ColumnTypePolicy resolved = null) {
+ resolved = resolved ?: ColumnTypePolicy.resolve(decimals.path ?: name, values, clazz)
+ if (resolved.encoding == List) {
List> nestedValues = extractListElements(values)
- Type elementType = buildParquetType(FIELD_ELEMENT, nestedValues, inferClass(nestedValues, Object), null, true)
+ Type elementType = buildParquetType(FIELD_ELEMENT, nestedValues, Object, null, true, decimals.child(FIELD_ELEMENT))
if (elementType == null) {
elementType = buildPrimitiveType(FIELD_ELEMENT, String, null, false)
}
return Types.optionalList().element(elementType).named(name)
}
- if (isMapType(clazz, values)) {
+ if (resolved.encoding == Map) {
if (preferStructForMaps) {
- return buildStructType(name, values, clazz)
+ return buildStructType(name, values, clazz, decimals)
}
- return buildMapType(name, values)
+ return buildMapType(name, values, decimals)
}
- if (isStructType(clazz, values)) {
- return buildStructType(name, values, clazz)
- }
-
- Class> effectiveClass = inferClass(values, clazz)
+ Class> effectiveClass = resolved.encoding
int[] effectiveDecimalMeta = decimalMeta
+ if (effectiveClass == BigDecimal) { effectiveDecimalMeta = decimals.resolve(values, effectiveDecimalMeta) }
+ if (effectiveClass == BigInteger && decimals.settings.containsKey(decimals.path)) { decimals.used.add(decimals.path) }
if (effectiveClass == BigInteger && effectiveDecimalMeta == null) {
effectiveDecimalMeta = [inferBigIntegerPrecision(values), 0] as int[]
}
return buildPrimitiveType(name, effectiveClass, effectiveDecimalMeta, false)
}
+ @SuppressWarnings(['AbcMetric', 'CyclomaticComplexity']) // Exhaustive physical/logical type dispatch.
private static PrimitiveType buildPrimitiveType(String name, Class clazz, int[] decimalMeta = null, boolean required = false) {
if (clazz == BigInteger) {
if (decimalMeta == null || decimalMeta.length != DECIMAL_META_LENGTH || decimalMeta[0] <= 0 || decimalMeta[1] != 0) {
@@ -804,15 +850,31 @@ class MatrixParquetWriter {
.as(LogicalTypeAnnotation.decimalType(scale, precision))
return builder.named(name)
}
- // fallback to double (discouraged, but allowed for backward compat)
- def fallbackBuilder = required ? Types.required(PrimitiveTypeName.DOUBLE) : Types.optional(PrimitiveTypeName.DOUBLE)
- return fallbackBuilder.named(name)
+ throw new IllegalArgumentException("BigDecimal field '$name' requires precision/scale")
+ }
+
+ if (clazz == UUID) {
+ def uuidBuilder = required ? Types.required(PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY) : Types.optional(PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY)
+ return uuidBuilder.length(16).as(LogicalTypeAnnotation.uuidType()).named(name)
}
def primitive
def logical = null
switch (clazz) {
+ case Byte, Short -> {
+ primitive = PrimitiveTypeName.INT32
+ logical = LogicalTypeAnnotation.intType(clazz == Byte ? 8 : 16, true)
+ }
+ case byte[] -> primitive = PrimitiveTypeName.BINARY
+ case String, Character -> {
+ primitive = PrimitiveTypeName.BINARY
+ logical = LogicalTypeAnnotation.stringType()
+ }
+ case LocalTime -> {
+ primitive = PrimitiveTypeName.INT64
+ logical = LogicalTypeAnnotation.timeType(false, LogicalTypeAnnotation.TimeUnit.NANOS)
+ }
case Integer, int -> primitive = PrimitiveTypeName.INT32
case Long, long -> primitive = PrimitiveTypeName.INT64
case Float, float -> primitive = PrimitiveTypeName.FLOAT
@@ -822,7 +884,7 @@ class MatrixParquetWriter {
primitive = PrimitiveTypeName.INT32
logical = LogicalTypeAnnotation.dateType()
}
- case LocalDateTime, Timestamp -> {
+ case LocalDateTime, Timestamp, Instant -> {
primitive = PrimitiveTypeName.INT64
logical = LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MICROS)
}
@@ -834,7 +896,7 @@ class MatrixParquetWriter {
primitive = PrimitiveTypeName.INT64
logical = LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)
}
- default -> primitive = PrimitiveTypeName.BINARY
+ default -> throw new IllegalArgumentException("Unsupported primitive $clazz for field $name")
}
def builder = required ? Types.required(primitive) : Types.optional(primitive)
@@ -844,33 +906,33 @@ class MatrixParquetWriter {
return builder.named(name)
}
- private static Type buildMapType(String name, List> values) {
+ private static Type buildMapType(String name, List> values, DecimalSchemaPolicy decimals) {
def mapBuilder = Types.optionalMap()
List