diff --git a/CHANGELOG.md b/CHANGELOG.md index ae9b68d05..356af0ad3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- New `vortex-mapper` module reads and writes your own objects as rows: `ColumnMapping` declares each column with a getter, and `Rows.stream` maps a projected scan through a `Row` callback. ([#528](https://github.com/dfa1/vortex-java/issues/528)) - `vortex.json` and the `vortex.st.*` geometry columns open without a decoder and read as their storage, as vortex-jni reads them. ([#547](https://github.com/dfa1/vortex-java/pull/547)) - The CLI takes `--allow-unknown` to open files with extension types it has no decoder for, and CSV `ExportOptions.withReadRegistry` picks the registry an export opens the file with. ([#547](https://github.com/dfa1/vortex-java/pull/547)) - Custom extension types: register an `ExtensionDecoder` on `ReadRegistry` and an `ExtensionEncoder` on `WriteRegistry`, with a `validate` hook on both sides, and `Chunk.as` decodes any registered type. ([#530](https://github.com/dfa1/vortex-java/issues/530)) @@ -20,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Breaking:** `ExtensionId` is a sealed `WellKnown`/`Custom` type like `EncodingId`, so third-party extension ids can be named, and `ExtensionId.parse` returns the id instead of an `Optional`. ([#530](https://github.com/dfa1/vortex-java/issues/530)) ### Fixed +- Unsigned integer columns that step evenly only past their type's range, or above the signed range of `U8`/`U16`/`U32`, no longer write a `vortex.sequence` that Rust refuses to read. ([#528](https://github.com/dfa1/vortex-java/issues/528)) - `vortex count`, `schema`, `stats` and `export` report a malformed file as an error instead of crashing with a stack trace. ([#547](https://github.com/dfa1/vortex-java/pull/547)) - `vortex.date` columns in milliseconds, as Rust writes an Arrow `Date64`, decode to the right dates instead of failing or reading the count as days. ([#547](https://github.com/dfa1/vortex-java/pull/547)) - An extension column with no metadata no longer writes a file Rust refuses to read. ([#547](https://github.com/dfa1/vortex-java/pull/547)) diff --git a/CLAUDE.md b/CLAUDE.md index 09422ee36..f0c7a104c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -55,9 +55,13 @@ writer — VortexWriter, WriteRegistry, WriteOptions, ColumnEncoding, Extension writer.encode — EncodingEncoder, EncodeContext, EncodeResult, EncodedBuffer, NullableData + *EncodingEncoder impls, extension encoders +mapper — io.github.dfa1.vortex.mapper: objects in and out as rows through explicit callbacks + (no reflection/annotations) — ColumnMapping (write: name + type + getter per column), + Rows.stream + Row (read: projection-first, cells by projection position) ``` Dependency rule: `writer → core`, `reader → core`. **Writer never depends on reader.** +`mapper` is the only library module that depends on both. `Array` and subtypes are decode outputs — they live in `reader.array`, not `core`. ## Branching diff --git a/bom/pom.xml b/bom/pom.xml index 363808254..740037a73 100644 --- a/bom/pom.xml +++ b/bom/pom.xml @@ -55,6 +55,11 @@ vortex-csv ${project.version} + + io.github.dfa1.vortex + vortex-mapper + ${project.version} + io.github.dfa1.vortex vortex-parquet diff --git a/docs/how-to.md b/docs/how-to.md index b25ca567f..ef7237a54 100644 --- a/docs/how-to.md +++ b/docs/how-to.md @@ -628,6 +628,83 @@ java -jar cli/target/vortex-cli-*-all.jar filter data.vortex "price >= 100" > ou --- +## Write and read Java objects + +The `vortex-mapper` module maps objects to rows through explicit callbacks: no reflection, no +annotations, any class. On the write side, `ColumnMapping` declares each column once, with its +name, its Vortex type (the method name) and a getter. The file schema is that list: + +```java +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.mapper.ColumnMapping; +import io.github.dfa1.vortex.mapper.Rows; +import java.nio.channels.FileChannel; +import java.nio.file.StandardOpenOption; +import java.util.stream.Stream; + +record Trade(long ts, String symbol, double price, int volume, boolean buy, Double fee, byte[] orderId) {} + +ColumnMapping trades = ColumnMapping.builder() + .i64("ts", Trade::ts) + .utf8("symbol", Trade::symbol) + .f64("price", Trade::price) + .u32("volume", Trade::volume) // unsigned: the raw bits of an int + .bool("buy", Trade::buy) + .nullableF64("fee", Trade::fee) // null is a null cell + .binary("order_id", Trade::orderId) + .build(); + +Path path = Path.of("trades.vortex"); +List rows = List.of( + new Trade(1_700_000_000_000L, "AAPL", 189.95, 100, true, 0.5, new byte[]{1}), + new Trade(1_700_000_001_000L, "MSFT", 410.20, -1, false, null, new byte[]{2})); +try (var channel = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.WRITE, + StandardOpenOption.TRUNCATE_EXISTING); + var writer = VortexWriter.create(channel, trades.schema(), WriteOptions.defaults())) { + trades.write(writer, rows); // call again to stream more rows +} +``` + +On the read side, **projection is the model**: list only the columns you need in `ScanOptions`, +and only those are decoded, however wide the file. `Rows.stream` hands each row to a mapper, +which reads the cells by their position in that list: + +```java +record PriceOnly(double price) {} + +try (var reader = VortexReader.open(Path.of("trades.vortex")); + Stream prices = Rows.stream(reader, + ScanOptions.columns(ColumnName.of("price")), + row -> new PriceOnly(row.getDouble(0)))) { + prices.forEach(System.out::println); +} +``` + +Nulls and unsigned values are explicit: check `isNull` before a primitive getter on a nullable +column, and widen unsigned raw bits yourself. Limits and filters on the `ScanOptions` apply as in +a plain scan: + +```java +record Fill(String symbol, long volume, Double fee) {} + +try (var reader = VortexReader.open(Path.of("trades.vortex")); + Stream fills = Rows.stream(reader, + ScanOptions.columns(ColumnName.of("symbol"), ColumnName.of("volume"), ColumnName.of("fee")) + .withLimit(10), + row -> new Fill( + row.getString(0), + Integer.toUnsignedLong(row.getInt(1)), + row.isNull(2) ? null : row.getDouble(2)))) { + fills.forEach(System.out::println); +} +``` + +A getter of the wrong type for its column (`getLong` on an `F64`) and a projected column the file +does not have both fail with `IllegalArgumentException`. Encoding overrides stay on +`WriteOptions#withColumnEncoding`. + +--- + ## Write and read a Map column `DType.Map` has no dedicated write-side value type. Physically, a map column's `vortex.map` diff --git a/docs/reference.md b/docs/reference.md index 849bac9b0..5966538b2 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -10,6 +10,7 @@ For task-oriented usage see [how-to.md](how-to.md); for design rationale see [ex - [Encoding registry](#encoding-registry) - [FSST (`io.github.dfa1.vortex.fsst`)](#fsst-iogithubdfa1vortexfsst) - [Parquet / CSV import and Parquet export](#parquet--csv-import-and-parquet-export) +- [Object mapping](#object-mapping) - [JDBC import](#jdbc-import) - [Calcite SQL adapter](#calcite-sql-adapter) - [CLI](#cli) @@ -556,6 +557,65 @@ Parquet's random-access footer-first format, CSV needs no Range requests and no --- +## Object mapping + +Module `vortex-mapper`: objects in and out as rows, through explicit callbacks (no reflection, no +annotations). See [Write and read Java objects](how-to.md#write-and-read-java-objects). + +### `ColumnMapping` (`io.github.dfa1.vortex.mapper.ColumnMapping`) + +| Method | Description | +|--------------------------------------------|--------------------------------------------------------------------------| +| `ColumnMapping.builder()` | Starts a mapping; columns are declared in file order | +| `schema()` | The `DType.Struct` to pass to `VortexWriter.create` | +| `write(VortexWriter, List)` | Gathers each column from the rows and writes one `writeChunk`; empty list writes nothing | + +Builder methods, each `(String name, getter)`; every one has a `nullableXxx` twin taking a +`Function` whose `null` is a null cell: + +| Method | Column | Getter | +|---------------|---------|--------------------------------| +| `bool` | `Bool` | `Predicate` | +| `i8`, `u8` | `I8`, `U8` | `Function` | +| `i16`, `u16` | `I16`, `U16` | `Function` | +| `i32`, `u32` | `I32`, `U32` | `ToIntFunction` | +| `i64`, `u64` | `I64`, `U64` | `ToLongFunction` | +| `f32` | `F32` | `Function` | +| `f64` | `F64` | `ToDoubleFunction` | +| `utf8` | `Utf8` | `Function` | +| `binary` | `Binary`| `Function` | + +Unsigned columns take the raw bits of the same-width Java type. A non-nullable column rejects a +`null` with `IllegalArgumentException`; a duplicate or invalid column name fails at declaration. + +### `Rows` (`io.github.dfa1.vortex.mapper.Rows`) + +| Method | Description | +|-------------------------------------------------------------------------------|-------------| +| `stream(VortexReader, ScanOptions, Function)` | Lazily maps every row the scan selects; close the stream to release the scan. Fails with `IllegalArgumentException` if a projected column is not in the file | + +### `Row` (`io.github.dfa1.vortex.mapper.Row`) + +One row of a `Rows.stream` scan; cells are addressed by their position in the projection (file +order with no projection). The instance moves along the scan, so do not keep it. + +| Method | Reads | Null cell | +|---------------------------|---------------------------|--------------------| +| `isNull(int)` | any column | `true` | +| `getBoolean(int)` | `Bool` | unspecified | +| `getByte(int)` | `I8`, `U8` (raw bits) | unspecified | +| `getShort(int)` | `I16`, `U16` (raw bits) | unspecified | +| `getInt(int)` | `I32`, `U32` (raw bits) | unspecified | +| `getLong(int)` | `I64`, `U64` (raw bits) | unspecified | +| `getFloat(int)` | `F32` | unspecified | +| `getDouble(int)` | `F64` | unspecified | +| `getString(int)` | `Utf8` | `null` | +| `getBytes(int)` | `Binary`, `Utf8` (a copy) | `null` | + +A getter on a column of another type throws `IllegalArgumentException`. + +--- + ## JDBC import ### `JdbcImporter` (`io.github.dfa1.vortex.jdbc.JdbcImporter`) diff --git a/integration/pom.xml b/integration/pom.xml index 6c915e65e..8a4f069c4 100644 --- a/integration/pom.xml +++ b/integration/pom.xml @@ -51,6 +51,11 @@ vortex-csv test + + io.github.dfa1.vortex + vortex-mapper + test + io.github.dfa1.vortex vortex-jdbc @@ -172,6 +177,11 @@ vortex-csv compile + + io.github.dfa1.vortex + vortex-mapper + compile + diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/JavaWritesRustReadsIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/JavaWritesRustReadsIntegrationTest.java index a27cf3b5e..64305ad5f 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/JavaWritesRustReadsIntegrationTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/JavaWritesRustReadsIntegrationTest.java @@ -577,6 +577,60 @@ void javaWriter_jniReader_sequence_fullFilteredAndRowRange(@TempDir Path tmp) th assertThat(range).containsExactlyElementsOf(slice(Arrays.stream(times).boxed(), 1_501, 70_333)); } + /// Columns that are arithmetic sequences only in wrapping 64-bit arithmetic, or only once a + /// narrow unsigned value is sign-extended. Rust builds a sequence with checked arithmetic in the + /// column's own type, so it refuses to read these as one; the unsigned ones that do step evenly + /// within their type must still be written as a sequence, with an unsigned base. + static Stream sequenceEdges() { + return Stream.of( + // a U8 base above 127, sign-extended, is a base Rust cannot narrow to u8 + Arguments.of(DType.U8, new byte[]{(byte) 200, (byte) 201, (byte) 202}, true), + Arguments.of(DType.U16, new short[]{(short) 60_000, (short) 60_001, (short) 60_002}, true), + Arguments.of(DType.U32, new int[]{(int) 3_000_000_000L, (int) 3_000_000_001L, (int) 3_000_000_002L}, + true), + // descending: the step is negative, which a u64 multiplier cannot hold + Arguments.of(DType.U64, new long[]{10, 7, 4}, false), + // ascending past 2^64: each step overflows u64 + Arguments.of(DType.U64, new long[]{0, -3_000_000_019L, -6_000_000_038L}, false), + // a step of 255 does not fit an i8 + Arguments.of(DType.I8, new byte[]{-128, 127}, false), + // MAX - MIN wraps to -1 + Arguments.of(DType.I64, new long[]{Long.MIN_VALUE, Long.MAX_VALUE}, false)); + } + + @ParameterizedTest(name = "{0} {1}") + @MethodSource("sequenceEdges") + void javaWriter_jniReader_sequenceInColumnTypeArithmetic(DType dtype, Object values, boolean sequence, + @TempDir Path tmp) throws IOException { + // Given + ColumnName column = ColumnName.of("v"); + Path file = tmp.resolve("java_sequence_edge.vtx"); + WriteOptions options = WriteOptions.defaults().withColumnEncoding(column, + io.github.dfa1.vortex.writer.ColumnEncoding.candidates( + io.github.dfa1.vortex.core.model.EncodingId.VORTEX_SEQUENCE)); + try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + var sut = VortexWriter.create(ch, new DType.Struct(List.of(column), List.of(dtype), false), options)) { + sut.writeChunk(Map.of(column, values)); + } + int n = java.lang.reflect.Array.getLength(values); + + // When + List result = readRowRange(file, "v", 0, n); + + // Then + List expected = new ArrayList<>(); + for (int i = 0; i < n; i++) { + expected.add(java.lang.reflect.Array.get(values, i)); + } + // Arrow hands a UInt16 back as a char + assertThat(result).map(v -> v instanceof Character c ? (Object) (short) c.charValue() : v) + .containsExactlyElementsOf(expected); + try (var reader = io.github.dfa1.vortex.reader.VortexReader.open(file)) { + assertThat(reader.footer().arraySpecs().contains(io.github.dfa1.vortex.core.model.EncodingId.VORTEX_SEQUENCE)) + .isEqualTo(sequence); + } + } + @Test void javaWriter_jniReader_floatDictWithCompressedValues_fullFilteredAndRowRange(@TempDir Path tmp) throws IOException { // Given — 500 distinct full-precision prices over 100,000 rows: ALP cannot encode them, so diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/MapperInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/MapperInteropIntegrationTest.java new file mode 100644 index 000000000..55049db71 --- /dev/null +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/MapperInteropIntegrationTest.java @@ -0,0 +1,404 @@ +package io.github.dfa1.vortex.integration; + +import dev.vortex.api.DataSource; +import dev.vortex.api.Expression; +import dev.vortex.api.Partition; +import dev.vortex.api.Scan; +import dev.vortex.api.ScanOptions; +import dev.vortex.api.Session; +import dev.vortex.arrow.ArrowAllocation; +import dev.vortex.jni.NativeLoader; +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.mapper.ColumnMapping; +import io.github.dfa1.vortex.mapper.Row; +import io.github.dfa1.vortex.mapper.Rows; +import io.github.dfa1.vortex.reader.VortexReader; +import io.github.dfa1.vortex.writer.VortexWriter; +import io.github.dfa1.vortex.writer.WriteOptions; +import org.apache.arrow.c.ArrowArray; +import org.apache.arrow.c.ArrowSchema; +import org.apache.arrow.c.Data; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.vector.BigIntVector; +import org.apache.arrow.vector.BitVector; +import org.apache.arrow.vector.FieldVector; +import org.apache.arrow.vector.Float8Vector; +import org.apache.arrow.vector.UInt4Vector; +import org.apache.arrow.vector.VarBinaryVector; +import org.apache.arrow.vector.VarCharVector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.ipc.ArrowReader; +import org.apache.arrow.vector.types.FloatingPointPrecision; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.Schema; +import org.apache.arrow.vector.util.Text; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HexFormat; +import java.util.List; +import java.util.function.BiFunction; +import java.util.function.LongFunction; +import java.util.stream.Stream; + +import static java.nio.file.StandardOpenOption.CREATE; +import static java.nio.file.StandardOpenOption.WRITE; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/// The `mapper` module against vortex-jni, both ways: every [ColumnMapping] column kind written by +/// vortex-java and read back by vortex-jni (full, row-range and filtered) and by [Rows]; and a wide +/// file written by vortex-jni read through small projections, the module's reading model. +class MapperInteropIntegrationTest { + + private static final Session SESSION = Session.create(); + private static final BufferAllocator ALLOCATOR = ArrowAllocation.rootAllocator(); + private static final ColumnName VALUE = ColumnName.of("v"); + private static final int BATCHES = 3; + private static final int BATCH_ROWS = 1_000; + private static final int ROWS = BATCHES * BATCH_ROWS; + private static final long THRESHOLD = 1_234; + + static { + NativeLoader.loadJni(); + } + + /// Reads one cell of a [Row]; a method reference to a typed getter. + @FunctionalInterface + interface Getter { + Object get(Row row, int column); + } + + /// One [ColumnMapping.Builder] method each. Values cross the sign bit and, for the unsigned + /// kinds, sit above the signed range, so a sign-extension or signed/unsigned mix-up reads back + /// different; nullable kinds null every seventh row. + enum Kind { + BOOL(false, r -> r % 3 != 0, (b, v) -> b.bool("v", r -> (Boolean) v.apply(r)), Row::getBoolean), + NULLABLE_BOOL(true, r -> r % 3 != 0, (b, v) -> b.nullableBool("v", r -> (Boolean) v.apply(r)), Row::getBoolean), + I8(false, r -> (byte) (r * 7), (b, v) -> b.i8("v", r -> (Byte) v.apply(r)), Row::getByte), + NULLABLE_I8(true, r -> (byte) (r * 7), (b, v) -> b.nullableI8("v", r -> (Byte) v.apply(r)), Row::getByte), + U8(false, r -> (byte) (r * 7), (b, v) -> b.u8("v", r -> (Byte) v.apply(r)), Row::getByte), + NULLABLE_U8(true, r -> (byte) (r * 7), (b, v) -> b.nullableU8("v", r -> (Byte) v.apply(r)), Row::getByte), + I16(false, r -> (short) (r * 31), (b, v) -> b.i16("v", r -> (Short) v.apply(r)), Row::getShort), + NULLABLE_I16(true, r -> (short) (r * 31), (b, v) -> b.nullableI16("v", r -> (Short) v.apply(r)), Row::getShort), + U16(false, r -> (short) (r * 31), (b, v) -> b.u16("v", r -> (Short) v.apply(r)), Row::getShort), + NULLABLE_U16(true, r -> (short) (r * 31), (b, v) -> b.nullableU16("v", r -> (Short) v.apply(r)), Row::getShort), + I32(false, r -> (int) (r * 1_000_003L), (b, v) -> b.i32("v", r -> (Integer) v.apply(r)), Row::getInt), + NULLABLE_I32(true, r -> (int) (r * 1_000_003L), (b, v) -> b.nullableI32("v", r -> (Integer) v.apply(r)), Row::getInt), + U32(false, r -> (int) (r * 1_000_003L), (b, v) -> b.u32("v", r -> (Integer) v.apply(r)), Row::getInt), + NULLABLE_U32(true, r -> (int) (r * 1_000_003L), (b, v) -> b.nullableU32("v", r -> (Integer) v.apply(r)), Row::getInt), + I64(false, r -> r * -3_000_000_019L, (b, v) -> b.i64("v", r -> (Long) v.apply(r)), Row::getLong), + NULLABLE_I64(true, r -> r * -3_000_000_019L, (b, v) -> b.nullableI64("v", r -> (Long) v.apply(r)), Row::getLong), + U64(false, r -> r * -3_000_000_019L, (b, v) -> b.u64("v", r -> (Long) v.apply(r)), Row::getLong), + NULLABLE_U64(true, r -> r * -3_000_000_019L, (b, v) -> b.nullableU64("v", r -> (Long) v.apply(r)), Row::getLong), + F32(false, r -> r * 0.25f - 100, (b, v) -> b.f32("v", r -> (Float) v.apply(r)), Row::getFloat), + NULLABLE_F32(true, r -> r * 0.25f - 100, (b, v) -> b.nullableF32("v", r -> (Float) v.apply(r)), Row::getFloat), + F64(false, r -> r * 0.5 - 100, (b, v) -> b.f64("v", r -> (Double) v.apply(r)), Row::getDouble), + NULLABLE_F64(true, r -> r * 0.5 - 100, (b, v) -> b.nullableF64("v", r -> (Double) v.apply(r)), Row::getDouble), + // empty and non-ASCII strings among the plain ones: a length or an encoding slip shows + UTF8(false, Kind::string, (b, v) -> b.utf8("v", r -> (String) v.apply(r)), Row::getString), + NULLABLE_UTF8(true, Kind::string, (b, v) -> b.nullableUtf8("v", r -> (String) v.apply(r)), Row::getString), + BINARY(false, Kind::bytes, (b, v) -> b.binary("v", r -> (byte[]) v.apply(r)), Row::getBytes), + NULLABLE_BINARY(true, Kind::bytes, (b, v) -> b.nullableBinary("v", r -> (byte[]) v.apply(r)), Row::getBytes); + + final boolean nullable; + final LongFunction base; + final BiFunction, LongFunction, ColumnMapping.Builder> declare; + final Getter getter; + + Kind(boolean nullable, LongFunction base, + BiFunction, LongFunction, ColumnMapping.Builder> declare, + Getter getter) { + this.nullable = nullable; + this.base = base; + this.declare = declare; + this.getter = getter; + } + + Object value(long row) { + return nullable && row % 7 == 0 ? null : base.apply(row); + } + + private static Object string(long row) { + return row % 5 == 0 ? "" : "r" + row + (row % 11 == 0 ? "é€" : ""); + } + + private static Object bytes(long row) { + return ((String) string(row)).getBytes(StandardCharsets.UTF_8); + } + } + + @ParameterizedTest + @EnumSource(Kind.class) + void mapperWrites_jniReads_fullRowRangeAndFiltered(Kind kind, @TempDir Path tmp) throws IOException { + // Given + Path file = tmp.resolve("mapper.vortex"); + writeWithMapper(file, kind); + + // When + List full = readJni(file, 0, ROWS, null); + List range = readJni(file, 1_500, 2_100, null); + List filtered = readJni(file, 0, ROWS, Expression.binary(Expression.BinaryOp.GTE, + Expression.column("id"), Expression.literal(THRESHOLD))); + + // Then + assertThat(full).containsExactlyElementsOf(expected(kind, 0, ROWS)); + assertThat(range).containsExactlyElementsOf(expected(kind, 1_500, 2_100)); + assertThat(filtered).containsExactlyElementsOf(expected(kind, THRESHOLD, ROWS)); + } + + @ParameterizedTest + @EnumSource(Kind.class) + void mapperWrites_mapperReads(Kind kind, @TempDir Path tmp) throws IOException { + // Given + Path file = tmp.resolve("mapper.vortex"); + writeWithMapper(file, kind); + + // When + List result; + try (var reader = VortexReader.open(file); + Stream values = Rows.stream(reader, io.github.dfa1.vortex.reader.ScanOptions.columns(VALUE), + row -> row.isNull(0) ? null : normalize(kind.getter.get(row, 0)))) { + result = values.toList(); + } + + // Then + assertThat(result).containsExactlyElementsOf(expected(kind, 0, ROWS)); + } + + // ── vortex-jni writes a wide file, the mapper reads projections of it ───────────────────────── + + record Trade(long ts, String symbol, double price, int volume, boolean buy, Double fee, byte[] orderId) { + } + + record PriceOnly(double price) { + } + + record Fill(Double fee, String symbol, long volume) { + } + + @Test + void jniWrites_mapperReadsOneColumnProjection(@TempDir Path tmp) throws IOException { + // Given + Path file = tmp.resolve("trades.vortex"); + List trades = trades(); + writeTradesWithJni(file, trades); + + // When + List result; + try (var reader = VortexReader.open(file); + Stream prices = Rows.stream(reader, + io.github.dfa1.vortex.reader.ScanOptions.columns(ColumnName.of("price")), + row -> new PriceOnly(row.getDouble(0)))) { + result = prices.toList(); + } + + // Then + assertThat(result).containsExactlyElementsOf(trades.stream().map(t -> new PriceOnly(t.price())).toList()); + } + + @Test + void jniWrites_mapperReadsProjectionInItsOwnOrder(@TempDir Path tmp) throws IOException { + // Given: the projection lists fee before symbol, the reverse of the file; the scan hands + // columns back in file order, so a Row indexed by scan order would swap them + Path file = tmp.resolve("trades.vortex"); + List trades = trades(); + writeTradesWithJni(file, trades); + + // When + List result; + try (var reader = VortexReader.open(file); + Stream fills = Rows.stream(reader, + io.github.dfa1.vortex.reader.ScanOptions.columns( + ColumnName.of("fee"), ColumnName.of("symbol"), ColumnName.of("volume")), + row -> new Fill( + row.isNull(0) ? null : row.getDouble(0), + row.getString(1), + Integer.toUnsignedLong(row.getInt(2))))) { + result = fills.toList(); + } + + // Then + assertThat(result).containsExactlyElementsOf(trades.stream() + .map(t -> new Fill(t.fee(), t.symbol(), Integer.toUnsignedLong(t.volume()))).toList()); + } + + @Test + void jniWrites_mapperReadsLimit(@TempDir Path tmp) throws IOException { + // Given + Path file = tmp.resolve("trades.vortex"); + List trades = trades(); + writeTradesWithJni(file, trades); + + // When + List result; + try (var reader = VortexReader.open(file); + Stream ts = Rows.stream(reader, + io.github.dfa1.vortex.reader.ScanOptions.columns(ColumnName.of("ts")).withLimit(1_001), + row -> row.getLong(0))) { + result = ts.toList(); + } + + // Then + assertThat(result).containsExactlyElementsOf(trades.stream().limit(1_001).map(Trade::ts).toList()); + } + + @Test + void stream_rejectsColumnMissingFromFile(@TempDir Path tmp) throws IOException { + // Given + Path file = tmp.resolve("trades.vortex"); + writeTradesWithJni(file, trades()); + + try (var reader = VortexReader.open(file)) { + // When / Then + assertThatThrownBy(() -> Rows.stream(reader, + io.github.dfa1.vortex.reader.ScanOptions.columns(ColumnName.of("nope")), row -> row)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageStartingWith("no column 'nope' in file"); + } + } + + @Test + void getter_rejectsColumnOfAnotherType(@TempDir Path tmp) throws IOException { + // Given + Path file = tmp.resolve("trades.vortex"); + writeTradesWithJni(file, trades()); + + try (var reader = VortexReader.open(file); + Stream prices = Rows.stream(reader, + io.github.dfa1.vortex.reader.ScanOptions.columns(ColumnName.of("price")), + row -> row.getLong(0))) { + // When / Then + assertThatThrownBy(prices::toList) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageStartingWith("column 0 'price' is ") + .hasMessageEndingWith(", which getLong cannot read"); + } + } + + // ── helpers ────────────────────────────────────────────────────────────────────────────────── + + private static void writeWithMapper(Path file, Kind kind) throws IOException { + ColumnMapping mapping = kind.declare.apply( + ColumnMapping.builder().i64("id", Long::longValue), kind::value).build(); + try (var channel = FileChannel.open(file, CREATE, WRITE); + var writer = VortexWriter.create(channel, mapping.schema(), WriteOptions.defaults())) { + for (int batch = 0; batch < BATCHES; batch++) { + List rows = new ArrayList<>(BATCH_ROWS); + for (long row = (long) batch * BATCH_ROWS; row < (long) (batch + 1) * BATCH_ROWS; row++) { + rows.add(row); + } + mapping.write(writer, rows); + } + } + } + + private static List expected(Kind kind, long begin, long end) { + List out = new ArrayList<>(); + for (long row = begin; row < end; row++) { + out.add(normalize(kind.value(row))); + } + return out; + } + + /// One comparable form for both readers: Arrow hands strings as [Text] and `U16` as `char`; + /// byte arrays compare by content. + private static Object normalize(Object value) { + return switch (value) { + case Text text -> text.toString(); + case Character c -> (short) c.charValue(); + case byte[] bytes -> HexFormat.of().formatHex(bytes); + case null, default -> value; + }; + } + + private static List readJni(Path file, long begin, long end, Expression filter) throws IOException { + var builder = ScanOptions.builder() + .projection(Expression.select(new String[]{"v"}, Expression.root())) + .rowRangeBegin(begin) + .rowRangeEnd(end); + if (filter != null) { + builder = builder.filter(filter); + } + List rows = new ArrayList<>(); + Scan scan = DataSource.open(SESSION, file.toAbsolutePath().toUri().toString()).scan(builder.build()); + while (scan.hasNext()) { + Partition partition = scan.next(); + try (ArrowReader reader = partition.scanArrow(ALLOCATOR)) { + while (reader.loadNextBatch()) { + VectorSchemaRoot root = reader.getVectorSchemaRoot(); + FieldVector vector = root.getVector("v"); + for (int i = 0; i < root.getRowCount(); i++) { + rows.add(vector.isNull(i) ? null : normalize(vector.getObject(i))); + } + } + } + } + return rows; + } + + /// Every column kind of the Trade example, with null fees, volumes above `Integer.MAX_VALUE` + /// (raw bits read as negative ints) and empty and non-ASCII symbols. + private static List trades() { + List trades = new ArrayList<>(ROWS); + for (int row = 0; row < ROWS; row++) { + trades.add(new Trade( + 1_700_000_000_000L + row * 1_000L, + row % 5 == 0 ? "" : "SYM" + row % 13 + (row % 11 == 0 ? "é€" : ""), + 100 + row * 0.01, + row * 1_500_007, + row % 2 == 0, + row % 7 == 0 ? null : row * 0.001, + HexFormat.of().parseHex("%08x".formatted(row)))); + } + return trades; + } + + private static void writeTradesWithJni(Path file, List trades) throws IOException { + Schema schema = new Schema(List.of( + Field.notNullable("ts", new ArrowType.Int(64, true)), + Field.notNullable("symbol", ArrowType.Utf8.INSTANCE), + Field.notNullable("price", new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE)), + Field.notNullable("volume", new ArrowType.Int(32, false)), + Field.notNullable("buy", ArrowType.Bool.INSTANCE), + Field.nullable("fee", new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE)), + Field.notNullable("order_id", ArrowType.Binary.INSTANCE))); + try (var writer = dev.vortex.api.VortexWriter.builder(SESSION, file.toAbsolutePath().toUri().toString(), + schema, ALLOCATOR).build()) { + for (int batch = 0; batch < BATCHES; batch++) { + try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, ALLOCATOR)) { + root.allocateNew(); + for (int i = 0; i < BATCH_ROWS; i++) { + Trade t = trades.get(batch * BATCH_ROWS + i); + ((BigIntVector) root.getVector("ts")).setSafe(i, t.ts()); + ((VarCharVector) root.getVector("symbol")).setSafe(i, t.symbol().getBytes(StandardCharsets.UTF_8)); + ((Float8Vector) root.getVector("price")).setSafe(i, t.price()); + ((UInt4Vector) root.getVector("volume")).setSafe(i, t.volume()); + ((BitVector) root.getVector("buy")).setSafe(i, t.buy() ? 1 : 0); + Float8Vector fee = (Float8Vector) root.getVector("fee"); + if (t.fee() == null) { + fee.setNull(i); + } else { + fee.setSafe(i, t.fee()); + } + ((VarBinaryVector) root.getVector("order_id")).setSafe(i, t.orderId()); + } + root.setRowCount(BATCH_ROWS); + try (ArrowArray array = ArrowArray.allocateNew(ALLOCATOR); + ArrowSchema exported = ArrowSchema.allocateNew(ALLOCATOR)) { + Data.exportVectorSchemaRoot(ALLOCATOR, root, null, array, exported); + writer.writeBatch(array.memoryAddress(), exported.memoryAddress()); + } + } + } + } + } +} diff --git a/mapper/pom.xml b/mapper/pom.xml new file mode 100644 index 000000000..314aafb96 --- /dev/null +++ b/mapper/pom.xml @@ -0,0 +1,37 @@ + + + 4.0.0 + + io.github.dfa1.vortex + vortex-java + 0.17.3-SNAPSHOT + + + vortex-mapper + + vortex-mapper + Maps Java objects to and from Vortex rows through explicit column callbacks. + + + + + io.github.dfa1.vortex + vortex-reader + + + io.github.dfa1.vortex + vortex-writer + + + + org.junit.jupiter + junit-jupiter + test + + + org.assertj + assertj-core + test + + + diff --git a/mapper/src/main/java/io/github/dfa1/vortex/mapper/ColumnMapping.java b/mapper/src/main/java/io/github/dfa1/vortex/mapper/ColumnMapping.java new file mode 100644 index 000000000..3433236eb --- /dev/null +++ b/mapper/src/main/java/io/github/dfa1/vortex/mapper/ColumnMapping.java @@ -0,0 +1,452 @@ +package io.github.dfa1.vortex.mapper; + +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.writer.VortexWriter; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.function.Function; +import java.util.function.IntFunction; +import java.util.function.Predicate; +import java.util.function.ToDoubleFunction; +import java.util.function.ToIntFunction; +import java.util.function.ToLongFunction; + +/// Writes objects of type `T` as rows of a Vortex file: each column is declared once, with its name, +/// its Vortex type and the getter that reads it from a `T`, and the file schema is that list in +/// declaration order. +/// +/// ```java +/// ColumnMapping trades = ColumnMapping.builder() +/// .i64("ts", Trade::ts) +/// .utf8("symbol", Trade::symbol) +/// .f64("price", Trade::price) +/// .u32("volume", Trade::volume) +/// .nullableF64("fee", Trade::fee) +/// .build(); +/// try (var writer = VortexWriter.create(channel, trades.schema(), WriteOptions.defaults())) { +/// trades.write(writer, rows); +/// } +/// ``` +/// +/// No reflection and no annotations: any class works, and the column type is the method that +/// declares it. Unsigned columns hold the raw bits of the same-width Java type, as the writer's +/// array API does (`u32` reads an `int`). A non-nullable column rejects a `null` from its getter; +/// the `nullableXxx` variants write it as a null cell. Encoding overrides stay on +/// [io.github.dfa1.vortex.writer.WriteOptions#withColumnEncoding]. +/// +/// @param the row type +public final class ColumnMapping { + + private final DType.Struct schema; + private final List> columns; + + private ColumnMapping(DType.Struct schema, List> columns) { + this.schema = schema; + this.columns = List.copyOf(columns); + } + + /// Starts an empty mapping. + /// + /// @param the row type + /// @return a new builder + public static Builder builder() { + return new Builder<>(); + } + + /// The file schema: one field per declared column, in declaration order. Pass it to + /// [VortexWriter#create(java.nio.channels.WritableByteChannel, DType.Struct, io.github.dfa1.vortex.writer.WriteOptions)]. + /// + /// @return the struct schema + public DType.Struct schema() { + return schema; + } + + /// Writes `rows` as one batch: each column's values are gathered from every row through its + /// getter, then handed to [VortexWriter#writeChunk(Map)], which validates them against the + /// writer's schema. Chunk and zone sizes stay the writer's choice; call this repeatedly to + /// stream a large source. An empty list writes nothing. + /// + /// @param writer a writer created with [#schema()] + /// @param rows the rows to write + /// @throws IOException if the writer fails to write + /// @throws IllegalArgumentException if a non-nullable column's getter returns `null` + public void write(VortexWriter writer, List rows) throws IOException { + if (rows.isEmpty()) { + return; + } + Map chunk = LinkedHashMap.newLinkedHashMap(columns.size()); + for (Column column : columns) { + chunk.put(column.name(), column.values().gather(rows, column.name())); + } + writer.writeChunk(chunk); + } + + private record Column(ColumnName name, Values values) { + } + + /// Gathers one column's values from all rows into the array shape the writer takes for it. + @FunctionalInterface + private interface Values { + Object gather(List rows, ColumnName column); + } + + /// Declares the columns of a [ColumnMapping], in file order. Each method takes the column + /// name (validated by [ColumnName#of(String)]) and the getter for its value. + /// + /// @param the row type + public static final class Builder { + + private final DType.StructBuilder schema = DType.structBuilder(); + private final List> columns = new ArrayList<>(); + + private Builder() { + } + + /// A non-nullable `Bool` column. + /// + /// @param name the column name + /// @param getter reads the value from a row + /// @return this builder + public Builder bool(String name, Predicate getter) { + return add(name, DType.BOOL, (rows, _) -> booleans(rows, getter)); + } + + /// A nullable `Bool` column. + /// + /// @param name the column name + /// @param getter reads the value from a row; `null` is a null cell + /// @return this builder + public Builder nullableBool(String name, Function getter) { + return add(name, DType.BOOL.asNullable(), (rows, _) -> boxed(rows, getter, Boolean[]::new)); + } + + /// A non-nullable `I8` column. + /// + /// @param name the column name + /// @param getter reads the value from a row + /// @return this builder + public Builder i8(String name, Function getter) { + return add(name, DType.I8, (rows, column) -> bytes(rows, getter, column)); + } + + /// A nullable `I8` column. + /// + /// @param name the column name + /// @param getter reads the value from a row; `null` is a null cell + /// @return this builder + public Builder nullableI8(String name, Function getter) { + return add(name, DType.I8.asNullable(), (rows, _) -> boxed(rows, getter, Byte[]::new)); + } + + /// A non-nullable `U8` column, from the raw bits of a `byte`. + /// + /// @param name the column name + /// @param getter reads the value from a row + /// @return this builder + public Builder u8(String name, Function getter) { + return add(name, DType.U8, (rows, column) -> bytes(rows, getter, column)); + } + + /// A nullable `U8` column, from the raw bits of a `byte`. + /// + /// @param name the column name + /// @param getter reads the value from a row; `null` is a null cell + /// @return this builder + public Builder nullableU8(String name, Function getter) { + return add(name, DType.U8.asNullable(), (rows, _) -> boxed(rows, getter, Byte[]::new)); + } + + /// A non-nullable `I16` column. + /// + /// @param name the column name + /// @param getter reads the value from a row + /// @return this builder + public Builder i16(String name, Function getter) { + return add(name, DType.I16, (rows, column) -> shorts(rows, getter, column)); + } + + /// A nullable `I16` column. + /// + /// @param name the column name + /// @param getter reads the value from a row; `null` is a null cell + /// @return this builder + public Builder nullableI16(String name, Function getter) { + return add(name, DType.I16.asNullable(), (rows, _) -> boxed(rows, getter, Short[]::new)); + } + + /// A non-nullable `U16` column, from the raw bits of a `short`. + /// + /// @param name the column name + /// @param getter reads the value from a row + /// @return this builder + public Builder u16(String name, Function getter) { + return add(name, DType.U16, (rows, column) -> shorts(rows, getter, column)); + } + + /// A nullable `U16` column, from the raw bits of a `short`. + /// + /// @param name the column name + /// @param getter reads the value from a row; `null` is a null cell + /// @return this builder + public Builder nullableU16(String name, Function getter) { + return add(name, DType.U16.asNullable(), (rows, _) -> boxed(rows, getter, Short[]::new)); + } + + /// A non-nullable `I32` column. + /// + /// @param name the column name + /// @param getter reads the value from a row + /// @return this builder + public Builder i32(String name, ToIntFunction getter) { + return add(name, DType.I32, (rows, _) -> ints(rows, getter)); + } + + /// A nullable `I32` column. + /// + /// @param name the column name + /// @param getter reads the value from a row; `null` is a null cell + /// @return this builder + public Builder nullableI32(String name, Function getter) { + return add(name, DType.I32.asNullable(), (rows, _) -> boxed(rows, getter, Integer[]::new)); + } + + /// A non-nullable `U32` column, from the raw bits of an `int`. + /// + /// @param name the column name + /// @param getter reads the value from a row + /// @return this builder + public Builder u32(String name, ToIntFunction getter) { + return add(name, DType.U32, (rows, _) -> ints(rows, getter)); + } + + /// A nullable `U32` column, from the raw bits of an `int`. + /// + /// @param name the column name + /// @param getter reads the value from a row; `null` is a null cell + /// @return this builder + public Builder nullableU32(String name, Function getter) { + return add(name, DType.U32.asNullable(), (rows, _) -> boxed(rows, getter, Integer[]::new)); + } + + /// A non-nullable `I64` column. + /// + /// @param name the column name + /// @param getter reads the value from a row + /// @return this builder + public Builder i64(String name, ToLongFunction getter) { + return add(name, DType.I64, (rows, _) -> longs(rows, getter)); + } + + /// A nullable `I64` column. + /// + /// @param name the column name + /// @param getter reads the value from a row; `null` is a null cell + /// @return this builder + public Builder nullableI64(String name, Function getter) { + return add(name, DType.I64.asNullable(), (rows, _) -> boxed(rows, getter, Long[]::new)); + } + + /// A non-nullable `U64` column, from the raw bits of a `long`. + /// + /// @param name the column name + /// @param getter reads the value from a row + /// @return this builder + public Builder u64(String name, ToLongFunction getter) { + return add(name, DType.U64, (rows, _) -> longs(rows, getter)); + } + + /// A nullable `U64` column, from the raw bits of a `long`. + /// + /// @param name the column name + /// @param getter reads the value from a row; `null` is a null cell + /// @return this builder + public Builder nullableU64(String name, Function getter) { + return add(name, DType.U64.asNullable(), (rows, _) -> boxed(rows, getter, Long[]::new)); + } + + /// A non-nullable `F32` column. + /// + /// @param name the column name + /// @param getter reads the value from a row + /// @return this builder + public Builder f32(String name, Function getter) { + return add(name, DType.F32, (rows, column) -> floats(rows, getter, column)); + } + + /// A nullable `F32` column. + /// + /// @param name the column name + /// @param getter reads the value from a row; `null` is a null cell + /// @return this builder + public Builder nullableF32(String name, Function getter) { + return add(name, DType.F32.asNullable(), (rows, _) -> boxed(rows, getter, Float[]::new)); + } + + /// A non-nullable `F64` column. + /// + /// @param name the column name + /// @param getter reads the value from a row + /// @return this builder + public Builder f64(String name, ToDoubleFunction getter) { + return add(name, DType.F64, (rows, _) -> doubles(rows, getter)); + } + + /// A nullable `F64` column. + /// + /// @param name the column name + /// @param getter reads the value from a row; `null` is a null cell + /// @return this builder + public Builder nullableF64(String name, Function getter) { + return add(name, DType.F64.asNullable(), (rows, _) -> boxed(rows, getter, Double[]::new)); + } + + /// A non-nullable `Utf8` column; a `null` from the getter fails the write. + /// + /// @param name the column name + /// @param getter reads the value from a row + /// @return this builder + public Builder utf8(String name, Function getter) { + return add(name, DType.UTF8, (rows, _) -> boxed(rows, getter, String[]::new)); + } + + /// A nullable `Utf8` column. + /// + /// @param name the column name + /// @param getter reads the value from a row; `null` is a null cell + /// @return this builder + public Builder nullableUtf8(String name, Function getter) { + return add(name, DType.UTF8.asNullable(), (rows, _) -> boxed(rows, getter, String[]::new)); + } + + /// A non-nullable `Binary` column; a `null` from the getter fails the write. + /// + /// @param name the column name + /// @param getter reads the value from a row + /// @return this builder + public Builder binary(String name, Function getter) { + return add(name, DType.BINARY, (rows, _) -> boxed(rows, getter, byte[][]::new)); + } + + /// A nullable `Binary` column. + /// + /// @param name the column name + /// @param getter reads the value from a row; `null` is a null cell + /// @return this builder + public Builder nullableBinary(String name, Function getter) { + return add(name, DType.BINARY.asNullable(), (rows, _) -> boxed(rows, getter, byte[][]::new)); + } + + /// Builds the mapping. + /// + /// @return the mapping, with its columns in declaration order + /// @throws IllegalStateException if no column was declared + public ColumnMapping build() { + if (columns.isEmpty()) { + throw new IllegalStateException("a mapping needs at least one column"); + } + return new ColumnMapping<>(schema.build(), columns); + } + + private Builder add(String name, DType dtype, Values values) { + ColumnName column = ColumnName.of(name); + schema.field(column, dtype); + columns.add(new Column<>(column, values)); + return this; + } + } + + // The writer takes primitive arrays for non-nullable columns and boxed arrays (null = null cell) + // for nullable ones; these gather one column into the matching shape. Rows are iterated, not + // indexed, so a LinkedList input stays linear. + + private static boolean[] booleans(List rows, Predicate getter) { + boolean[] out = new boolean[rows.size()]; + int i = 0; + for (T row : rows) { + out[i++] = getter.test(row); + } + return out; + } + + private static int[] ints(List rows, ToIntFunction getter) { + int[] out = new int[rows.size()]; + int i = 0; + for (T row : rows) { + out[i++] = getter.applyAsInt(row); + } + return out; + } + + private static long[] longs(List rows, ToLongFunction getter) { + long[] out = new long[rows.size()]; + int i = 0; + for (T row : rows) { + out[i++] = getter.applyAsLong(row); + } + return out; + } + + private static double[] doubles(List rows, ToDoubleFunction getter) { + double[] out = new double[rows.size()]; + int i = 0; + for (T row : rows) { + out[i++] = getter.applyAsDouble(row); + } + return out; + } + + // byte, short and float have no primitive functional interface in the JDK, so their getters + // return the box; Byte is always cached and the rest are scalar-replaced once inlined. + + private static byte[] bytes(List rows, Function getter, ColumnName column) { + byte[] out = new byte[rows.size()]; + int i = 0; + for (T row : rows) { + out[i] = present(getter.apply(row), column, i); + i++; + } + return out; + } + + private static short[] shorts(List rows, Function getter, ColumnName column) { + short[] out = new short[rows.size()]; + int i = 0; + for (T row : rows) { + out[i] = present(getter.apply(row), column, i); + i++; + } + return out; + } + + private static float[] floats(List rows, Function getter, ColumnName column) { + float[] out = new float[rows.size()]; + int i = 0; + for (T row : rows) { + out[i] = present(getter.apply(row), column, i); + i++; + } + return out; + } + + private static V[] boxed(List rows, Function getter, + IntFunction newArray) { + V[] out = newArray.apply(rows.size()); + int i = 0; + for (T row : rows) { + out[i++] = getter.apply(row); + } + return out; + } + + private static V present(V value, ColumnName column, int row) { + if (value == null) { + throw new IllegalArgumentException("non-nullable column '" + column + "' received null at row " + row); + } + return value; + } +} diff --git a/mapper/src/main/java/io/github/dfa1/vortex/mapper/Row.java b/mapper/src/main/java/io/github/dfa1/vortex/mapper/Row.java new file mode 100644 index 000000000..4d4a46271 --- /dev/null +++ b/mapper/src/main/java/io/github/dfa1/vortex/mapper/Row.java @@ -0,0 +1,189 @@ +package io.github.dfa1.vortex.mapper; + +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.reader.Chunk; +import io.github.dfa1.vortex.reader.array.Array; +import io.github.dfa1.vortex.reader.array.BoolArray; +import io.github.dfa1.vortex.reader.array.ByteArray; +import io.github.dfa1.vortex.reader.array.DoubleArray; +import io.github.dfa1.vortex.reader.array.FloatArray; +import io.github.dfa1.vortex.reader.array.IntArray; +import io.github.dfa1.vortex.reader.array.LongArray; +import io.github.dfa1.vortex.reader.array.MaskedArray; +import io.github.dfa1.vortex.reader.array.NullArray; +import io.github.dfa1.vortex.reader.array.ShortArray; +import io.github.dfa1.vortex.reader.array.VarBinArray; + +import java.util.List; + +/// One row of a scan, handed to the mapper of [Rows#stream]. Columns are addressed by their position +/// in the projection: `getDouble(0)` reads the first column listed in +/// [io.github.dfa1.vortex.reader.ScanOptions#columns(ColumnName...)]. +/// +/// Each getter reads one Java type and fails with [IllegalArgumentException] on a column of another +/// type: `getInt` reads `I32` and `U32`, `getLong` reads `I64` and `U64`, and so on. Unsigned values +/// come back as the raw bits of the same-width type; widen them with [Integer#toUnsignedLong(int)] +/// and friends. `getString` reads `Utf8`; `getBytes` reads `Binary` and `Utf8`. +/// +/// For a null cell, `getString` and `getBytes` return `null`; the primitive getters return an +/// unspecified value, so check [#isNull(int)] first on a nullable column. +/// +/// The same `Row` instance moves along the scan: read what you need inside the mapper and do not +/// keep it. Strings and byte arrays are copies, so values taken out of it stay valid after the scan. +public final class Row { + + private final ColumnName[] names; + private final DType[] dtypes; + private final Array[] values; + private final BoolArray[] validity; + private long index; + + Row(Chunk chunk, List columns) { + int n = columns.size(); + names = new ColumnName[n]; + dtypes = new DType[n]; + values = new Array[n]; + validity = new BoolArray[n]; + for (int i = 0; i < n; i++) { + Chunk.Column column = chunk.columns().get(columns.get(i)); + MaskedArray.Unwrapped unwrapped = MaskedArray.unwrap(column.array()); + names[i] = columns.get(i); + dtypes[i] = column.dtype(); + values[i] = unwrapped.inner(); + validity[i] = unwrapped.validity(); + } + } + + void moveTo(long row) { + index = row; + } + + /// Whether the cell is null. + /// + /// @param column the column's position in the projection + /// @return `true` if the cell holds no value + public boolean isNull(int column) { + // a column that is null in every row of the chunk decodes to a NullArray payload + return values[column] instanceof NullArray + || validity[column] != null && !validity[column].getBoolean(index); + } + + /// Reads a `Bool` cell. + /// + /// @param column the column's position in the projection + /// @return the value + public boolean getBoolean(int column) { + if (values[column] instanceof BoolArray a) { + return a.getBoolean(index); + } + requireNullColumn(column, "getBoolean"); + return false; + } + + /// Reads an `I8` or `U8` cell; a `U8` comes back as its raw bits. + /// + /// @param column the column's position in the projection + /// @return the value + public byte getByte(int column) { + if (values[column] instanceof ByteArray a) { + return a.getByte(index); + } + requireNullColumn(column, "getByte"); + return 0; + } + + /// Reads an `I16` or `U16` cell; a `U16` comes back as its raw bits. + /// + /// @param column the column's position in the projection + /// @return the value + public short getShort(int column) { + if (values[column] instanceof ShortArray a) { + return a.getShort(index); + } + requireNullColumn(column, "getShort"); + return 0; + } + + /// Reads an `I32` or `U32` cell; a `U32` comes back as its raw bits. + /// + /// @param column the column's position in the projection + /// @return the value + public int getInt(int column) { + if (values[column] instanceof IntArray a) { + return a.getInt(index); + } + requireNullColumn(column, "getInt"); + return 0; + } + + /// Reads an `I64` or `U64` cell; a `U64` comes back as its raw bits. + /// + /// @param column the column's position in the projection + /// @return the value + public long getLong(int column) { + if (values[column] instanceof LongArray a) { + return a.getLong(index); + } + requireNullColumn(column, "getLong"); + return 0L; + } + + /// Reads an `F32` cell. + /// + /// @param column the column's position in the projection + /// @return the value + public float getFloat(int column) { + if (values[column] instanceof FloatArray a) { + return a.getFloat(index); + } + requireNullColumn(column, "getFloat"); + return 0f; + } + + /// Reads an `F64` cell. + /// + /// @param column the column's position in the projection + /// @return the value + public double getDouble(int column) { + if (values[column] instanceof DoubleArray a) { + return a.getDouble(index); + } + requireNullColumn(column, "getDouble"); + return 0d; + } + + /// Reads a `Utf8` cell. + /// + /// @param column the column's position in the projection + /// @return the value, or `null` for a null cell + public String getString(int column) { + if (!(dtypes[column] instanceof DType.Utf8)) { + throw mismatch(column, "getString"); + } + return isNull(column) ? null : ((VarBinArray) values[column]).getString(index); + } + + /// Reads a `Binary` or `Utf8` cell as bytes. + /// + /// @param column the column's position in the projection + /// @return a copy of the value, or `null` for a null cell + public byte[] getBytes(int column) { + if (!(dtypes[column] instanceof DType.Binary || dtypes[column] instanceof DType.Utf8)) { + throw mismatch(column, "getBytes"); + } + return isNull(column) ? null : ((VarBinArray) values[column]).getBytes(index); + } + + // An all-null column has no typed payload to read from; any primitive getter is fine on it. + private void requireNullColumn(int column, String getter) { + if (!(values[column] instanceof NullArray)) { + throw mismatch(column, getter); + } + } + + private IllegalArgumentException mismatch(int column, String getter) { + return new IllegalArgumentException("column " + column + " '" + names[column] + "' is " + dtypes[column] + + ", which " + getter + " cannot read"); + } +} diff --git a/mapper/src/main/java/io/github/dfa1/vortex/mapper/Rows.java b/mapper/src/main/java/io/github/dfa1/vortex/mapper/Rows.java new file mode 100644 index 000000000..e638fecbd --- /dev/null +++ b/mapper/src/main/java/io/github/dfa1/vortex/mapper/Rows.java @@ -0,0 +1,83 @@ +package io.github.dfa1.vortex.mapper; + +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.reader.Chunk; +import io.github.dfa1.vortex.reader.ScanIterator; +import io.github.dfa1.vortex.reader.ScanOptions; +import io.github.dfa1.vortex.reader.VortexReader; + +import java.util.ArrayList; +import java.util.List; +import java.util.Spliterator; +import java.util.Spliterators; +import java.util.function.Function; +import java.util.stream.Stream; +import java.util.stream.StreamSupport; + +/// Reads a Vortex file as a stream of objects, one per row, built by a mapper over [Row]. +/// +/// Projection is the reading model: list the columns you need in +/// [ScanOptions#columns(ColumnName...)] and only those are decoded, however wide the file. The +/// mapper addresses them by position in that list: +/// +/// ```java +/// record PriceOnly(double price) {} +/// +/// try (var reader = VortexReader.open(path); +/// Stream prices = Rows.stream(reader, +/// ScanOptions.columns(ColumnName.of("price")), +/// row -> new PriceOnly(row.getDouble(0)))) { +/// ... +/// } +/// ``` +public final class Rows { + + private Rows() { + } + + /// Streams the rows `options` selects, mapped through `mapper`. The scan runs lazily as the + /// stream is consumed, one chunk at a time; close the stream to release the scan (the reader + /// itself stays open). The limit and filter in `options` apply as in [VortexReader#scan]. + /// + /// @param reader the file to read + /// @param options the projection, and optionally a limit and a filter; with no projection the + /// row holds every column in file order + /// @param mapper builds one object from a [Row]; must not keep the row + /// @param the element type + /// @return the mapped rows, in file order + /// @throws IllegalArgumentException if a projected column is not in the file + public static Stream stream(VortexReader reader, ScanOptions options, + Function mapper) { + List projection = options.columns(); + // the scan skips a projected name the file lacks; fail instead of shifting every index after it + if (reader.dtype() instanceof DType.Struct schema) { + for (ColumnName column : projection) { + if (!schema.fieldNames().contains(column)) { + throw new IllegalArgumentException("no column '" + column + "' in file, which has " + + schema.fieldNames()); + } + } + } + ScanIterator scan = reader.scan(options); + return StreamSupport.stream(Spliterators.spliteratorUnknownSize(scan, Spliterator.ORDERED), false) + .flatMap(chunk -> Rows.map(chunk, projection, mapper).stream()) + .onClose(scan::close); + } + + // Maps the whole chunk before closing it, so the scan can move on and no row outlives its arena. + private static List map(Chunk chunk, List projection, + Function mapper) { + try (chunk) { + List columns = projection.isEmpty() ? List.copyOf(chunk.columns().keySet()) : projection; + Row row = new Row(chunk, columns); + long n = chunk.rowCount(); + List out = new ArrayList<>(Math.toIntExact(n)); + for (long i = 0; i < n; i++) { + row.moveTo(i); + out.add(mapper.apply(row)); + } + return out; + } + } +} diff --git a/mapper/src/test/java/io/github/dfa1/vortex/mapper/ColumnMappingTest.java b/mapper/src/test/java/io/github/dfa1/vortex/mapper/ColumnMappingTest.java new file mode 100644 index 000000000..92b6132f7 --- /dev/null +++ b/mapper/src/test/java/io/github/dfa1/vortex/mapper/ColumnMappingTest.java @@ -0,0 +1,146 @@ +package io.github.dfa1.vortex.mapper; + +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.writer.VortexWriter; +import io.github.dfa1.vortex.writer.WriteOptions; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.channels.Channels; +import java.util.Arrays; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class ColumnMappingTest { + + record Trade(long ts, String symbol, double price, int volume, boolean buy, Double fee, byte[] orderId) { + } + + record Narrow(byte i8, short i16, float f32) { + } + + @Test + void schema_followsDeclarationOrderNamesAndTypes() { + // Given + ColumnMapping sut = ColumnMapping.builder() + .i64("ts", Trade::ts) + .utf8("symbol", Trade::symbol) + .f64("price", Trade::price) + .u32("volume", Trade::volume) + .bool("buy", Trade::buy) + .nullableF64("fee", Trade::fee) + .binary("order_id", Trade::orderId) + .build(); + + // When + DType.Struct result = sut.schema(); + + // Then + assertThat(result.fieldNames()).extracting(ColumnName::value) + .containsExactly("ts", "symbol", "price", "volume", "buy", "fee", "order_id"); + assertThat(result.fieldTypes()).containsExactly(DType.I64, DType.UTF8, DType.F64, DType.U32, DType.BOOL, + DType.F64.asNullable(), DType.BINARY); + assertThat(result.nullable()).isFalse(); + } + + @Test + void schema_unsignedAndNullableVariantsKeepTheirOwnType() { + // Given: the signed/unsigned and plain/nullable pairs share a Java type, so only the + // builder method tells them apart; a copy-paste slip between them would go unnoticed + ColumnMapping sut = ColumnMapping.builder() + .i8("a", Narrow::i8) + .u8("b", Narrow::i8) + .nullableU8("c", Narrow::i8) + .i16("d", Narrow::i16) + .u16("e", Narrow::i16) + .nullableU16("f", Narrow::i16) + .f32("g", Narrow::f32) + .nullableF32("h", Narrow::f32) + .build(); + + // When + DType.Struct result = sut.schema(); + + // Then + assertThat(result.fieldTypes()).containsExactly(DType.I8, DType.U8, DType.U8.asNullable(), DType.I16, + DType.U16, DType.U16.asNullable(), DType.F32, DType.F32.asNullable()); + } + + @Test + void build_rejectsDuplicateColumn() { + // Given + ColumnMapping.Builder sut = ColumnMapping.builder().i64("ts", Trade::ts); + + // When / Then + assertThatThrownBy(() -> sut.f64("ts", Trade::price)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("duplicate field name"); + } + + @Test + void build_rejectsNoColumns() { + // Given + ColumnMapping.Builder sut = ColumnMapping.builder(); + + // When / Then + assertThatThrownBy(sut::build) + .isInstanceOf(IllegalStateException.class) + .hasMessage("a mapping needs at least one column"); + } + + @Test + void write_rejectsNullFromNonNullablePrimitiveGetter() throws IOException { + // Given: a boxed getter is the only way a non-nullable byte column can see a null + ColumnMapping sut = ColumnMapping.builder().i8("v", _ -> null).build(); + try (VortexWriter writer = inMemoryWriter(sut)) { + + // When / Then + assertThatThrownBy(() -> sut.write(writer, List.of(new Narrow((byte) 1, (short) 1, 1f)))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("non-nullable column 'v' received null at row 0"); + } + } + + @Test + void write_rejectsNullInNonNullableUtf8() throws IOException { + // Given + ColumnMapping sut = ColumnMapping.builder().utf8("symbol", Trade::symbol).build(); + List rows = Arrays.asList(trade("A"), trade(null)); + try (VortexWriter writer = inMemoryWriter(sut)) { + + // When / Then + assertThatThrownBy(() -> sut.write(writer, rows)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("non-nullable column 'symbol' received null at row 1"); + } + } + + @Test + void write_acceptsNullInNullableUtf8() throws IOException { + // Given + ColumnMapping sut = ColumnMapping.builder().nullableUtf8("symbol", Trade::symbol).build(); + var out = new ByteArrayOutputStream(); + + // When + try (VortexWriter writer = VortexWriter.create(Channels.newChannel(out), sut.schema(), + WriteOptions.defaults())) { + sut.write(writer, Arrays.asList(trade("A"), trade(null))); + } + + // Then + assertThat(out.size()).isPositive(); + } + + private static Trade trade(String symbol) { + return new Trade(0L, symbol, 0d, 0, false, null, new byte[0]); + } + + private static VortexWriter inMemoryWriter(ColumnMapping mapping) { + return VortexWriter.create(Channels.newChannel(new ByteArrayOutputStream()), mapping.schema(), + WriteOptions.defaults()); + } +} diff --git a/pom.xml b/pom.xml index af66b231a..bf3393608 100644 --- a/pom.xml +++ b/pom.xml @@ -44,6 +44,7 @@ core reader writer + mapper csv jdbc parquet @@ -184,6 +185,11 @@ vortex-csv ${project.version} + + io.github.dfa1.vortex + vortex-mapper + ${project.version} + io.github.dfa1.vortex vortex-jdbc diff --git a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/SequenceEncodingEncoder.java b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/SequenceEncodingEncoder.java index e9b7131ad..de051c689 100644 --- a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/SequenceEncodingEncoder.java +++ b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/SequenceEncodingEncoder.java @@ -67,18 +67,24 @@ public EncodeResult encode(DType dtype, Object data, EncodeContext ctx) { return encodeInteger(((DType.Primitive) dtype).ptype(), data); } + /// Rust's `encode_primitive_array`: the step is a checked subtraction and every next value a + /// checked addition, both in the column's own type, so a sequence that only holds by wrapping + /// (a negative step for an unsigned column, a step or a value past the type's range) is not one. + /// Rust refuses to read such a sequence back: its last value is not expressible in the type. private static EncodeResult encodeInteger(PType pt, Object data) { int n = Array.getLength(data); long base = 0; long multiplier = 0; if (n > 0) { base = readLong(pt, data, 0); - multiplier = n > 1 ? readLong(pt, data, 1) - base : 0; - for (int i = 2; i < n; i++) { - long expected = base + i * multiplier; - if (readLong(pt, data, i) != expected) { + multiplier = n > 1 ? step(pt, base, readLong(pt, data, 1)) : 0; + long previous = base; + for (int i = 1; i < n; i++) { + long current = readLong(pt, data, i); + if (!follows(pt, previous, multiplier, current)) { throw new VortexException(EncodingId.VORTEX_SEQUENCE, "not an arithmetic sequence at index " + i); } + previous = current; } } ProtoScalarValue baseScalar = buildIntScalar(pt, base); @@ -111,11 +117,54 @@ private static ProtoScalarValue buildIntScalar(PType pt, long value) { }; } + /// `next - first` in `pt`, or a [VortexException] when the type cannot hold it. Narrow values + /// are exact in a `long`, so only their range needs checking; 64-bit ones need overflow checks. + private static long step(PType pt, long first, long next) { + long step = next - first; + boolean fits = switch (pt) { + case I8 -> step == (byte) step; + case I16 -> step == (short) step; + case I32 -> step == (int) step; + case I64 -> exact(() -> Math.subtractExact(next, first)); + case U8, U16, U32 -> step >= 0; + default -> Long.compareUnsigned(next, first) >= 0; + }; + if (!fits) { + throw new VortexException(EncodingId.VORTEX_SEQUENCE, "step " + first + " -> " + next + " overflows " + pt); + } + return step; + } + + /// Whether `current` is `previous + step` without overflowing `pt`. A narrow sum is exact in a + /// `long` and `current` is in range, so equality suffices there. + private static boolean follows(PType pt, long previous, long step, long current) { + long sum = previous + step; + return switch (pt) { + case I64 -> sum == current && exact(() -> Math.addExact(previous, step)); + case U64 -> sum == current && Long.compareUnsigned(sum, previous) >= 0; + default -> sum == current; + }; + } + + private static boolean exact(java.util.function.LongSupplier arithmetic) { + try { + arithmetic.getAsLong(); + return true; + } catch (ArithmeticException _) { + return false; + } + } + + /// The value at `i`, zero-extended for an unsigned type so it compares and serializes as the + /// number it stands for (a `U8` of 200 is 200, not -56). private static long readLong(PType pt, Object data, int i) { return switch (pt) { - case I8, U8 -> ((byte[]) data)[i]; - case I16, U16 -> ((short[]) data)[i]; - case I32, U32 -> ((int[]) data)[i]; + case I8 -> ((byte[]) data)[i]; + case U8 -> Byte.toUnsignedLong(((byte[]) data)[i]); + case I16 -> ((short[]) data)[i]; + case U16 -> Short.toUnsignedLong(((short[]) data)[i]); + case I32 -> ((int[]) data)[i]; + case U32 -> Integer.toUnsignedLong(((int[]) data)[i]); case I64, U64 -> ((long[]) data)[i]; default -> throw new VortexException(EncodingId.VORTEX_SEQUENCE, "unsupported ptype: " + pt); }; diff --git a/writer/src/test/java/io/github/dfa1/vortex/writer/encode/SequenceEncodingEncoderTest.java b/writer/src/test/java/io/github/dfa1/vortex/writer/encode/SequenceEncodingEncoderTest.java index 9acce2a66..5c0ff6f51 100644 --- a/writer/src/test/java/io/github/dfa1/vortex/writer/encode/SequenceEncodingEncoderTest.java +++ b/writer/src/test/java/io/github/dfa1/vortex/writer/encode/SequenceEncodingEncoderTest.java @@ -229,6 +229,44 @@ void encode_nonArithmeticSequence_throwsVortexException() { .isInstanceOf(VortexException.class); } + @Test + void encode_u8AboveSignedRange_baseIsTheUnsignedValue() throws Exception { + // Given: sign-extended, 200 became a base of -56, which Rust cannot narrow to u8 + byte[] data = {(byte) 200, (byte) 201, (byte) 202}; + + // When + EncodeResult result = ENCODER.encode(DTypes.U8, data, EncodeTestHelper.testCtx()); + + // Then + MemorySegment metaSeg = result.rootNode().metadata(); + ProtoSequenceMetadata meta = ProtoSequenceMetadata.decode(metaSeg, 0, metaSeg.byteSize()); + assertThat(meta.base().uint64_value()).isEqualTo(200L); + assertThat(meta.multiplier().uint64_value()).isEqualTo(1L); + } + + /// Sequences only in wrapping 64-bit arithmetic: Rust's checked arithmetic in the column type + /// rejects each, and reading one back fails with "final value not expressible". + static Stream wrappingSequences() { + return Stream.of( + Arguments.of(DTypes.U64, new long[]{10, 7, 4}), + Arguments.of(DTypes.U64, new long[]{0, -3_000_000_019L, -6_000_000_038L}), + Arguments.of(DTypes.U8, new byte[]{(byte) 2, (byte) 1}), + Arguments.of(DTypes.I8, new byte[]{-128, 127}), + Arguments.of(DTypes.I64, new long[]{Long.MIN_VALUE, Long.MAX_VALUE}), + Arguments.of(DTypes.I64, new long[]{Long.MAX_VALUE - 1, Long.MAX_VALUE, Long.MIN_VALUE})); + } + + @ParameterizedTest(name = "{0} {1}") + @MethodSource("wrappingSequences") + void encode_sequenceOnlyByWrapping_throwsVortexException(DType dtype, Object data) { + // Given + EncodeContext ctx = EncodeTestHelper.testCtx(); + + // When / Then + assertThatThrownBy(() -> ENCODER.encode(dtype, data, ctx)) + .isInstanceOf(VortexException.class); + } + @Test void encode_nonPrimitiveDtype_throwsVortexException() { // Given