From 3d38a8db1246093782bdb2995dcbc8fbf48b246a Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Wed, 2 Sep 2026 22:26:38 +0200 Subject: [PATCH] feat(parquet): add ParquetExporter (Vortex -> Parquet) Inverse of ParquetImporter, built on Hardwood 1.1's new ColumnWriter/ ParquetFileWriter write API. Flat schemas only: Bool, non-F16 Primitive, Utf8, Binary, and vortex.timestamp (MILLIS/MICROS/NANOS); Struct/List/Map top-level columns throw UnsupportedOperationException. - Bumps dev.hardwood:hardwood-core 1.0.0.Final -> 1.1.0.Beta1 (adds the write API this depends on, plus read-path speedups for ParquetImporter). - ParquetExporter / ExportOptions in the parquet module, mirroring ParquetImporter/ImportOptions' shape. - vortex-reader promoted from test- to production-scope dependency in parquet/pom.xml. - ParquetExporterTest: type-mapping unit tests, flat export + projection + unsupported-type integration tests, and two round-trip tests (Vortex -> Parquet -> Vortex via ParquetImporter) covering flat types (including U8/U32 boundary values) and the timestamp extension. - CLI: the `export` subcommand now dispatches to ParquetExporter when the output path ends `.parquet` (previously CSV-only); usage text and ExportCommandTest updated. - Remote HTTP(S) support, both directions: - ParquetExporter.exportParquet(VortexHandle, Path[, ExportOptions]): exports an already-open handle (local VortexReader or remote VortexHttpReader) without an intervening local copy; the handle is not closed here, the caller keeps ownership. - ParquetImporter.importParquet(URI, Path[, ImportOptions]): imports a Parquet file served over HTTP(S), fetched entirely through targeted Range requests via the new HttpInputFile (implements Hardwood's InputFile), mirroring VortexHttpReader's own range-fetch pattern. No full-file download occurs. - checkNoDuplicateNames/fillRow take a String sourceName instead of a Path, since a remote source has no filesystem path to report in error messages. - CLI: `export`/`import` accept a url source too (Parquet-only in both directions; CSV and stdout streaming from a URL are out of scope). - HttpInputFileTest covers the HEAD/Range request mechanics against a mocked HttpClient. Live-verified end to end against a real ~3M-row public Parquet file (NYC taxi trip data) via the built CLI jar: remote Parquet -> local Vortex -> local Parquet -> reimport, with row counts and min/max stats matching exactly at every step. - docs/reference.md, docs/how-to.md: document the new class/options, the remote overloads, and the CLI's Parquet import/export support. - DocsConsistencyTest: allowlist WriterConfig as a known external receiver cited in reference.md. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01LjU7zWghUiJauRxK58nyUs --- CHANGELOG.md | 9 + .../github/dfa1/vortex/cli/ExportCommand.java | 80 ++- .../github/dfa1/vortex/cli/ImportCommand.java | 71 ++- .../io/github/dfa1/vortex/cli/VortexCli.java | 2 +- .../dfa1/vortex/cli/ExportCommandTest.java | 51 ++ .../dfa1/vortex/cli/ImportCommandTest.java | 12 + docs/how-to.md | 60 +++ docs/reference.md | 53 +- .../integration/DocsConsistencyTest.java | 5 +- parquet/pom.xml | 10 +- .../dfa1/vortex/parquet/ExportOptions.java | 33 ++ .../dfa1/vortex/parquet/HttpInputFile.java | 94 ++++ .../dfa1/vortex/parquet/ParquetExporter.java | 364 ++++++++++++++ .../dfa1/vortex/parquet/ParquetImporter.java | 40 +- .../vortex/parquet/HttpInputFileTest.java | 168 +++++++ .../vortex/parquet/ParquetExporterTest.java | 465 ++++++++++++++++++ .../vortex/parquet/ParquetImporterTest.java | 6 +- pom.xml | 2 +- 18 files changed, 1468 insertions(+), 57 deletions(-) create mode 100644 parquet/src/main/java/io/github/dfa1/vortex/parquet/ExportOptions.java create mode 100644 parquet/src/main/java/io/github/dfa1/vortex/parquet/HttpInputFile.java create mode 100644 parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetExporter.java create mode 100644 parquet/src/test/java/io/github/dfa1/vortex/parquet/HttpInputFileTest.java create mode 100644 parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetExporterTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 141c6af66..2788e4cd0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- `ParquetExporter` (`parquet` module): writes a Vortex file to Parquet, the inverse of `ParquetImporter`. Flat schemas only — `Bool`, non-`F16` `Primitive`, `Utf8`, `Binary`, and `vortex.timestamp` (MILLIS/MICROS/NANOS); `Struct`/`List`/`Map` top-level columns throw `UnsupportedOperationException`. Built on Hardwood 1.1's new `ColumnWriter`/`ParquetFileWriter` write API. The CLI's `export` subcommand now dispatches to it when the output path ends `.parquet` (previously CSV-only). ([#362](https://github.com/dfa1/vortex-java/pull/362)) +- `ParquetExporter.exportParquet(VortexHandle, Path[, ExportOptions])` and `ParquetImporter.importParquet(URI, Path[, ImportOptions])`: both directions now work against a remote source over HTTP(S) — export from an already-open `VortexHttpReader` handle (no intervening local copy), import from a Parquet file served over HTTP(S) (new `HttpInputFile`, fetched entirely through targeted Range requests, mirroring `VortexHttpReader`'s own range-fetch pattern). The CLI's `export`/`import` subcommands accept a `url` source too, Parquet-only in both directions. ([#362](https://github.com/dfa1/vortex-java/pull/362)) + +### Changed + +- `dev.hardwood:hardwood-core` 1.0.0.Final → 1.1.0.Beta1: faster `DELTA_BINARY_PACKED` and dictionary-index decoding plus a fixed-length `LIST` fast path speed up `ParquetImporter`'s read path; its new Parquet write support now backs `ParquetExporter`. ([e66fb6eb](https://github.com/dfa1/vortex-java/commit/e66fb6eb)) + ## [0.13.4] — 2026-09-01 ### Fixed diff --git a/cli/src/main/java/io/github/dfa1/vortex/cli/ExportCommand.java b/cli/src/main/java/io/github/dfa1/vortex/cli/ExportCommand.java index 3c3be74d9..a1a2bc30f 100644 --- a/cli/src/main/java/io/github/dfa1/vortex/cli/ExportCommand.java +++ b/cli/src/main/java/io/github/dfa1/vortex/cli/ExportCommand.java @@ -3,6 +3,8 @@ import io.github.dfa1.vortex.csv.CsvExporter; import io.github.dfa1.vortex.inspect.ByteSize; import io.github.dfa1.vortex.csv.ExportOptions; +import io.github.dfa1.vortex.parquet.ParquetExporter; +import io.github.dfa1.vortex.reader.VortexHandle; import java.io.IOException; import java.io.OutputStreamWriter; @@ -19,31 +21,53 @@ private ExportCommand() { static int run(String[] args) { if (args.length < 2 || args.length > 3) { - System.err.println("usage: export [out.csv | -]"); + System.err.println("usage: export [out.csv | out.parquet | -]"); return ExitStatus.USAGE_ERROR; } - Path inputPath = Path.of(args[1]); + String target = args[1]; + boolean toStdout = args.length == 3 && "-".equals(args[2]); + boolean remote = target.startsWith("http://") || target.startsWith("https://"); + if (remote) { + return runRemote(target, args, toStdout); + } + Path inputPath = Path.of(target); if (!Files.exists(inputPath)) { System.err.println("file not found: " + inputPath); return ExitStatus.FILE_NOT_FOUND; } - boolean toStdout = args.length == 3 && "-".equals(args[2]); Path outputPath = (args.length == 3 && !toStdout) ? Path.of(args[2]) : deriveOutputPath(inputPath); try { - ExportOptions options = ExportOptions.defaults() - .withProgressListener(ProgressBar::render); - if (toStdout) { - Writer stdout = new OutputStreamWriter(System.out, StandardCharsets.UTF_8); - CsvExporter.exportCsv(inputPath, stdout, options); - stdout.flush(); - ProgressBar.clear(); - } else { - CsvExporter.exportCsv(inputPath, outputPath, options); - ProgressBar.clear(); - printResult(inputPath, outputPath); + if (!toStdout && outputPath.getFileName().toString().endsWith(".parquet")) { + return runParquet(inputPath, outputPath); } + return runCsv(inputPath, outputPath, toStdout); + } catch (IOException e) { + ProgressBar.clear(); + System.err.println("error: " + e.getMessage()); + return ExitStatus.ERROR; + } + } + + /// Handles an `http(s)://` source: Parquet output only (an explicit `out.parquet` path is + /// required — CSV export and stdout streaming from a remote source aren't supported yet). + private static int runRemote(String target, String[] args, boolean toStdout) { + if (toStdout || args.length != 3 || !args[2].endsWith(".parquet")) { + System.err.println("usage: export out.parquet (CSV/stdout export from a URL isn't supported yet)"); + return ExitStatus.USAGE_ERROR; + } + Path outputPath = Path.of(args[2]); + try (VortexHandle handle = CliHandles.openTarget(target)) { + if (handle == null) { + return ExitStatus.FILE_NOT_FOUND; + } + io.github.dfa1.vortex.parquet.ExportOptions options = + io.github.dfa1.vortex.parquet.ExportOptions.defaults() + .withProgressListener(ProgressBar::render); + ParquetExporter.exportParquet(handle, outputPath, options); + ProgressBar.clear(); + System.out.printf("written: %s (%s)%n", outputPath, ByteSize.format(Files.size(outputPath))); return ExitStatus.OK; } catch (IOException e) { ProgressBar.clear(); @@ -52,6 +76,34 @@ static int run(String[] args) { } } + private static int runCsv(Path inputPath, Path outputPath, boolean toStdout) throws IOException { + ExportOptions options = ExportOptions.defaults() + .withProgressListener(ProgressBar::render); + if (toStdout) { + Writer stdout = new OutputStreamWriter(System.out, StandardCharsets.UTF_8); + CsvExporter.exportCsv(inputPath, stdout, options); + stdout.flush(); + ProgressBar.clear(); + } else { + CsvExporter.exportCsv(inputPath, outputPath, options); + ProgressBar.clear(); + printResult(inputPath, outputPath); + } + return ExitStatus.OK; + } + + private static int runParquet(Path inputPath, Path outputPath) throws IOException { + io.github.dfa1.vortex.parquet.ExportOptions options = + io.github.dfa1.vortex.parquet.ExportOptions.defaults() + .withProgressListener(ProgressBar::render); + ParquetExporter.exportParquet(inputPath, outputPath, options); + ProgressBar.clear(); + printResult(inputPath, outputPath); + return ExitStatus.OK; + } + + /// Defaults to `.csv` — a Parquet destination must be named explicitly (`out.parquet`), + /// matching [ExportCommand#run]'s extension-on-the-output-path dispatch. private static Path deriveOutputPath(Path inputPath) { String name = inputPath.getFileName().toString(); if (name.endsWith(".vortex")) { diff --git a/cli/src/main/java/io/github/dfa1/vortex/cli/ImportCommand.java b/cli/src/main/java/io/github/dfa1/vortex/cli/ImportCommand.java index 3295acef9..07d1131c9 100644 --- a/cli/src/main/java/io/github/dfa1/vortex/cli/ImportCommand.java +++ b/cli/src/main/java/io/github/dfa1/vortex/cli/ImportCommand.java @@ -6,6 +6,7 @@ import io.github.dfa1.vortex.parquet.ParquetImporter; import java.io.IOException; +import java.net.URI; import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; @@ -14,7 +15,10 @@ @SuppressWarnings("java:S106") final class ImportCommand { - private record ParsedArgs(Path inputPath, Path outputPath, Character delimiter) { + /// `outputTarget` is the raw second positional argument, or `null` when omitted — kept + /// unresolved here because deriving a default depends on whether `inputTarget` turns out to + /// be a local path or a URL, decided in [#run]. + private record ParsedArgs(String inputTarget, String outputTarget, Character delimiter) { } private ImportCommand() { @@ -26,20 +30,28 @@ static int run(String[] args) { parsedArgs = parseArgs(args); } catch (IllegalArgumentException e) { System.err.println(e.getMessage()); - System.err.println("usage: import [--delimiter ] [out.vortex]"); + System.err.println("usage: import [--delimiter ] [out.vortex]"); return ExitStatus.USAGE_ERROR; } - Path inputPath = parsedArgs.inputPath(); - if (!Files.exists(inputPath)) { - System.err.println("file not found: " + inputPath); - return ExitStatus.FILE_NOT_FOUND; - } + String target = parsedArgs.inputTarget(); + boolean remote = target.startsWith("http://") || target.startsWith("https://"); try { + if (remote) { + return runRemote(target, parsedArgs.outputTarget()); + } + Path inputPath = Path.of(target); + if (!Files.exists(inputPath)) { + System.err.println("file not found: " + inputPath); + return ExitStatus.FILE_NOT_FOUND; + } String name = inputPath.getFileName().toString(); + Path outputPath = parsedArgs.outputTarget() != null + ? Path.of(parsedArgs.outputTarget()) + : inputPath.resolveSibling(vortexName(name)); if (name.endsWith(".parquet")) { - return runParquet(inputPath, parsedArgs.outputPath()); + return runParquet(inputPath, outputPath); } else { - return runCsv(inputPath, parsedArgs.outputPath(), parsedArgs.delimiter()); + return runCsv(inputPath, outputPath, parsedArgs.delimiter()); } } catch (IOException e) { ProgressBar.clear(); @@ -48,6 +60,26 @@ static int run(String[] args) { } } + /// Handles an `http(s)://` source: Parquet import only (CSV import from a URL isn't + /// supported yet — CSV has no schema of its own, and type inference over a remote stream + /// needs a design of its own). + private static int runRemote(String parquetUrl, String outputTarget) throws IOException { + if (!parquetUrl.endsWith(".parquet")) { + System.err.println("only Parquet import is supported from a URL"); + return ExitStatus.USAGE_ERROR; + } + Path vortexPath = outputTarget != null + ? Path.of(outputTarget) + : Path.of(vortexName(lastPathSegment(parquetUrl))); + io.github.dfa1.vortex.parquet.ImportOptions options = + io.github.dfa1.vortex.parquet.ImportOptions.defaults() + .withProgressListener(ImportCommand::renderProgress); + ParquetImporter.importParquet(URI.create(parquetUrl), vortexPath, options); + ProgressBar.clear(); + System.out.printf("written: %s (%s)%n", vortexPath, ByteSize.format(Files.size(vortexPath))); + return ExitStatus.OK; + } + private static ParsedArgs parseArgs(String[] args) { if (args.length < 2) { throw new IllegalArgumentException("missing import arguments"); @@ -72,9 +104,8 @@ private static ParsedArgs parseArgs(String[] args) { if (positional.size() < 1 || positional.size() > 2) { throw new IllegalArgumentException("expected input path and optional output path"); } - Path inputPath = Path.of(positional.getFirst()); - Path outputPath = positional.size() == 2 ? Path.of(positional.get(1)) : deriveOutputPath(inputPath); - return new ParsedArgs(inputPath, outputPath, delimiter); + String outputTarget = positional.size() == 2 ? positional.get(1) : null; + return new ParsedArgs(positional.getFirst(), outputTarget, delimiter); } private static int runCsv(Path csvPath, Path vortexPath, Character delimiter) throws IOException { @@ -125,13 +156,23 @@ private static void renderProgress(long done, long total) { } } - private static Path deriveOutputPath(Path inputPath) { - String name = inputPath.getFileName().toString(); + /// Strips a known `.csv`/`.parquet` suffix from `inputFileName` and appends `.vortex`. + private static String vortexName(String inputFileName) { + String name = inputFileName; if (name.endsWith(".csv")) { name = name.substring(0, name.length() - 4); } else if (name.endsWith(".parquet")) { name = name.substring(0, name.length() - 8); } - return inputPath.resolveSibling(name + ".vortex"); + return name + ".vortex"; + } + + /// The last `/`-separated segment of a URL's path, used as the file name a downloaded + /// source is named after (mirroring what a browser would save the URL as). + private static String lastPathSegment(String url) { + String path = URI.create(url).getPath(); + int slash = path.lastIndexOf('/'); + String name = slash < 0 ? path : path.substring(slash + 1); + return name.isEmpty() ? "output.parquet" : name; } } diff --git a/cli/src/main/java/io/github/dfa1/vortex/cli/VortexCli.java b/cli/src/main/java/io/github/dfa1/vortex/cli/VortexCli.java index 0f5ada1f1..631e8e104 100644 --- a/cli/src/main/java/io/github/dfa1/vortex/cli/VortexCli.java +++ b/cli/src/main/java/io/github/dfa1/vortex/cli/VortexCli.java @@ -47,7 +47,7 @@ static void printUsage(PrintStream out) { out.println(" inspect print file structure; url is http(s)://"); out.println(" tui open interactive inspector; url is http(s)://"); out.println(" view open scrollable data grid; url is http(s)://"); - out.println(" export [out.csv|-] write CSV; default output is .csv, `-` for stdout"); + out.println(" export [out.csv|out.parquet|-] write CSV or Parquet; default is .csv, `-` for stdout"); out.println(" import [out.vortex] convert CSV or Parquet to Vortex"); out.println(" schema print dtype (machine-readable)"); out.println(" count print row count"); diff --git a/cli/src/test/java/io/github/dfa1/vortex/cli/ExportCommandTest.java b/cli/src/test/java/io/github/dfa1/vortex/cli/ExportCommandTest.java index a33a5209d..e920424c7 100644 --- a/cli/src/test/java/io/github/dfa1/vortex/cli/ExportCommandTest.java +++ b/cli/src/test/java/io/github/dfa1/vortex/cli/ExportCommandTest.java @@ -49,4 +49,55 @@ void validFile_emitsCsvHeaderAndRows(@TempDir Path tmp) throws IOException { assertThat(result.stdout()).startsWith("id"); assertThat(result.stdout().lines().count()).isEqualTo(4); } + + @Test + void parquetOutputPath_dispatchesToParquetExport(@TempDir Path tmp) throws IOException { + // Given — a `.parquet` destination, dispatching away from the CSV default + Path file = writeSmallVortex(tmp, "export.vortex"); + Path outputPath = tmp.resolve("export.parquet"); + + // When + CliTestSupport.Captured result = capture( + () -> ExportCommand.run(new String[]{"export", file.toString(), outputPath.toString()})); + + // Then + assertThat(result.status()).isEqualTo(ExitStatus.OK); + assertThat(outputPath).exists(); + assertThat(result.stdout()).contains("written:").contains("export.parquet"); + } + + // ── URL source: argument validation only — these fail before any network call, so no + // mocked HTTP client is needed (mirrors ImportCommandTest's equivalent coverage). ── + + @Test + void urlInput_missingOutputPath_returnsUsageError() { + // Given / When — a remote source needs an explicit `out.parquet` path + CliTestSupport.Captured result = capture( + () -> ExportCommand.run(new String[]{"export", "http://example.com/data.vortex"})); + + // Then + assertThat(result.status()).isEqualTo(ExitStatus.USAGE_ERROR); + assertThat(result.stderr()).contains("out.parquet"); + } + + @Test + void urlInput_stdoutRequested_returnsUsageError() { + // Given / When — stdout streaming from a URL isn't supported + CliTestSupport.Captured result = capture( + () -> ExportCommand.run(new String[]{"export", "http://example.com/data.vortex", "-"})); + + // Then + assertThat(result.status()).isEqualTo(ExitStatus.USAGE_ERROR); + } + + @Test + void urlInput_csvOutputRequested_returnsUsageError(@TempDir Path tmp) { + // Given / When — CSV export from a URL isn't supported, only `.parquet` + Path outputPath = tmp.resolve("out.csv"); + CliTestSupport.Captured result = capture(() -> + ExportCommand.run(new String[]{"export", "http://example.com/data.vortex", outputPath.toString()})); + + // Then + assertThat(result.status()).isEqualTo(ExitStatus.USAGE_ERROR); + } } diff --git a/cli/src/test/java/io/github/dfa1/vortex/cli/ImportCommandTest.java b/cli/src/test/java/io/github/dfa1/vortex/cli/ImportCommandTest.java index 7e8c25eca..63a700510 100644 --- a/cli/src/test/java/io/github/dfa1/vortex/cli/ImportCommandTest.java +++ b/cli/src/test/java/io/github/dfa1/vortex/cli/ImportCommandTest.java @@ -61,6 +61,18 @@ void tooManyPositional_returnsUsageError() { assertThat(result.status()).isEqualTo(ExitStatus.USAGE_ERROR); assertThat(result.stderr()).contains("input path"); } + + @Test + void urlInput_nonParquetExtension_returnsUsageError() { + // Given / When — CSV import from a URL isn't supported; caught before any network + // call is made, so this needs no mocked HTTP client + CliTestSupport.Captured result = capture(() -> + ImportCommand.run(new String[]{"import", "http://example.com/data.csv"})); + + // Then + assertThat(result.status()).isEqualTo(ExitStatus.USAGE_ERROR); + assertThat(result.stderr()).contains("only Parquet import is supported from a URL"); + } } @Nested diff --git a/docs/how-to.md b/docs/how-to.md index 6caada936..aacfd293f 100644 --- a/docs/how-to.md +++ b/docs/how-to.md @@ -253,6 +253,13 @@ ImportOptions opts = ImportOptions.defaults() ParquetImporter.importParquet(Path.of("data.parquet"), Path.of("data.vortex"), opts); ``` +From a remote Parquet file over HTTP(S), fetched entirely through targeted Range requests — +no full-file download occurs: + +```java +ParquetImporter.importParquet(URI.create("https://example.com/data.parquet"), Path.of("data.vortex")); +``` + **CLI:** ```bash @@ -261,6 +268,59 @@ java -jar cli/target/vortex-cli-*-all.jar import data.parquet # explicit output path java -jar cli/target/vortex-cli-*-all.jar import data.parquet out.vortex + +# remote source — Parquet only, output path required or derived from the URL's file name +java -jar cli/target/vortex-cli-*-all.jar import https://example.com/data.parquet out.vortex +``` + +--- + +## Convert Vortex to Parquet + +Flat schemas only (`Bool`, non-`F16` `Primitive`, `Utf8`, `Binary`, `vortex.timestamp`); a +`Struct`/`List`/`Map` top-level column throws `UnsupportedOperationException`. + +**API:** + +```java +import io.github.dfa1.vortex.parquet.ParquetExporter; + +ParquetExporter.exportParquet( + Path.of("data.vortex"), + Path.of("data.parquet") +); +``` + +Project specific columns during conversion: + +```java +import io.github.dfa1.vortex.parquet.ExportOptions; + +ExportOptions opts = ExportOptions.defaults() + .withColumns(List.of("trip_distance", "fare_amount")); + +ParquetExporter.exportParquet(Path.of("data.vortex"), Path.of("data.parquet"), opts); +``` + +From an already-open handle — a local `VortexReader` or a remote `VortexHttpReader` — without an +intervening local copy; the handle isn't closed here, the caller keeps ownership of its lifecycle: + +```java +import io.github.dfa1.vortex.reader.VortexHttpReader; + +try (var vortex = VortexHttpReader.open(URI.create("https://example.com/data.vortex"))) { + ParquetExporter.exportParquet(vortex, Path.of("data.parquet")); +} +``` + +**CLI:** + +```bash +# dispatches on the output extension — same `export` subcommand CSV export uses +java -jar cli/target/vortex-cli-*-all.jar export data.vortex out.parquet + +# remote source — Parquet output only, and the output path must be given explicitly +java -jar cli/target/vortex-cli-*-all.jar export https://example.com/data.vortex out.parquet ``` --- diff --git a/docs/reference.md b/docs/reference.md index f32c6be75..fae9a8aaa 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -9,7 +9,7 @@ For task-oriented usage see [how-to.md](how-to.md); for design rationale see [ex - [Scan API](#scan-api) - [Encoding registry](#encoding-registry) - [FSST (`io.github.dfa1.vortex.fsst`)](#fsst-iogithubdfa1vortexfsst) -- [Parquet / CSV import](#parquet--csv-import) +- [Parquet / CSV import and Parquet export](#parquet--csv-import-and-parquet-export) - [CLI](#cli) - [Encoding compatibility](compatibility.md) @@ -379,7 +379,7 @@ A decoder claims its ids via `LayoutDecoder.layoutIds()` (a set — the zoned de (`decodeChild`, `decodeSegment` access, `arena`) and are reachable end-to-end via `VortexReader.open(path, readRegistry, layoutRegistry)`. -## Parquet / CSV import +## Parquet / CSV import and Parquet export ### `ParquetImporter` (`io.github.dfa1.vortex.parquet.ParquetImporter`) @@ -387,10 +387,15 @@ Supports flat schemas and nested `LIST`/`STRUCT` columns, recursively composable `LIST>`); `MAP` and `VARIANT` are not yet supported. Un-annotated `BYTE_ARRAY` maps to `DType.Binary`. -| Method | Notes | -|---------------------------------------------------|----------| -| `importParquet(Path in, Path out)` | Defaults | -| `importParquet(Path in, Path out, ImportOptions)` | Tuned | +| Method | Notes | +|---------------------------------------------------|-------------------------------------| +| `importParquet(Path in, Path out)` | Defaults | +| `importParquet(Path in, Path out, ImportOptions)` | Tuned | +| `importParquet(URI in, Path out)` | Remote source over HTTP(S), defaults | +| `importParquet(URI in, Path out, ImportOptions)` | Remote source over HTTP(S), tuned | + +A `URI` source is fetched entirely through targeted HTTP Range requests (`HttpInputFile`, +internal) — no full-file download occurs. ### `ImportOptions` (`io.github.dfa1.vortex.parquet.ImportOptions`) @@ -404,6 +409,38 @@ Record: `(int chunkSize, List columns, ProgressListener progressListener | `.withWriteOptions(WriteOptions)` | Override write options | | `.withChunkSize(int)` | Override chunk size | +### `ParquetExporter` (`io.github.dfa1.vortex.parquet.ParquetExporter`) + +Supports flat schemas only: `Bool`, non-`F16` `Primitive`, `Utf8`, `Binary`, and the +`vortex.timestamp` extension over MILLIS/MICROS/NANOS resolution. `Struct`/`List`/`Map` top-level +columns throw `UnsupportedOperationException` — the inverse of `ParquetImporter`'s nested +`LIST`/`STRUCT` support does not exist yet. + +| Method | Notes | +|------------------------------------------------------------|-----------------------------------------------------------| +| `exportParquet(Path in, Path out)` | Defaults | +| `exportParquet(Path in, Path out, ExportOptions)` | Tuned | +| `exportParquet(VortexHandle in, Path out)` | Already-open source (local `VortexReader` or remote `VortexHttpReader`), defaults | +| `exportParquet(VortexHandle in, Path out, ExportOptions)`| Already-open source, tuned | + +The `VortexHandle` overloads don't close `in` — the caller opened it and keeps ownership of its +lifecycle, so a remote `VortexHttpReader.open(uri)` can be exported without an intervening local +copy. + +### `ExportOptions` (`io.github.dfa1.vortex.parquet.ExportOptions`) + +Record: `(List columns, ProgressListener progressListener, WriterConfig writerConfig)`. +`WriterConfig` is Hardwood's own writer configuration type +(`dev.hardwood.writer.WriterConfig`) — row group/page targets, compression codec, column +encoding policy. + +| Factory / builder | Notes | +|------------------------------------|-----------------------------------------------------| +| `ExportOptions.defaults()` | No projection, Hardwood's `WriterConfig.defaults()` | +| `.withColumns(List)` | Project columns during export | +| `.withProgressListener(listener)` | Progress callbacks | +| `.withWriterConfig(WriterConfig)` | Override Hardwood writer configuration | + CSV import is CLI-only — types are inferred from the data. --- @@ -428,10 +465,10 @@ java -jar cli/target/vortex-cli-*-all.jar [args] | `schema` | `schema ` | Column names and types | | `count` | `count ` | Total row count | | `stats` | `stats ` | Per-column min/max | -| `export` | `export ` | All columns to CSV on stdout | +| `export` | `export [out.csv\|out.parquet\|-]` | All columns to CSV (default) or Parquet, by output extension; `-` for CSV on stdout. A `url` source requires an explicit `out.parquet` path — CSV/stdout from a URL isn't supported | | `select` | `select [col2 ...]` | Project columns to CSV | | `filter` | `filter ""` | Filter rows to CSV | -| `import` | `import [--delimiter ] [out.vortex]` | Convert CSV or Parquet to Vortex | +| `import` | `import [--delimiter ] [out.vortex]` | Convert CSV or Parquet to Vortex; a `url` source must be `.parquet` — CSV import from a URL isn't supported | ### `filter` expression syntax diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/DocsConsistencyTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/DocsConsistencyTest.java index 8a6fcbbfe..4f022776f 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/DocsConsistencyTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/DocsConsistencyTest.java @@ -58,8 +58,9 @@ class DocsConsistencyTest { "ServiceLoader", "DriverManager", "Class", "Object", "Thread", "LockSupport", "System", "MemorySegment", "Arena", "ValueLayout", "FileChannel", "ByteBuffer", "Runtime", "Math", "Integer", "Long", "Double", - // Hardwood parquet API, cited in the benchmark comparison (explanation.md) - "ColumnReader", "RowReader"); + // Hardwood parquet API, cited in the benchmark comparison (explanation.md) and the + // ParquetExporter/ExportOptions reference (reference.md) + "ColumnReader", "RowReader", "WriterConfig"); private static Path repoRoot; private static List livingDocs; diff --git a/parquet/pom.xml b/parquet/pom.xml index 45de23f75..369920433 100644 --- a/parquet/pom.xml +++ b/parquet/pom.xml @@ -14,6 +14,10 @@ + + io.github.dfa1.vortex + vortex-reader + io.github.dfa1.vortex vortex-writer @@ -29,12 +33,6 @@ zstd-jni - - - io.github.dfa1.vortex - vortex-reader - test - io.github.dfa1.zstd diff --git a/parquet/src/main/java/io/github/dfa1/vortex/parquet/ExportOptions.java b/parquet/src/main/java/io/github/dfa1/vortex/parquet/ExportOptions.java new file mode 100644 index 000000000..a480eda9c --- /dev/null +++ b/parquet/src/main/java/io/github/dfa1/vortex/parquet/ExportOptions.java @@ -0,0 +1,33 @@ +package io.github.dfa1.vortex.parquet; + +import dev.hardwood.writer.WriterConfig; + +import java.util.List; + +/// Options controlling Vortex → Parquet export. +public record ExportOptions( + List columns, + ProgressListener progressListener, + WriterConfig writerConfig +) { + public static ExportOptions defaults() { + return new ExportOptions(List.of(), null, WriterConfig.defaults()); + } + + /// Restrict export to specific top-level columns, in the given order. Empty list = all columns. + public ExportOptions withColumns(List cols) { + return new ExportOptions(List.copyOf(cols), progressListener, writerConfig); + } + + public boolean hasProjection() { + return !columns.isEmpty(); + } + + public ExportOptions withProgressListener(ProgressListener listener) { + return new ExportOptions(columns, listener, writerConfig); + } + + public ExportOptions withWriterConfig(WriterConfig config) { + return new ExportOptions(columns, progressListener, config); + } +} diff --git a/parquet/src/main/java/io/github/dfa1/vortex/parquet/HttpInputFile.java b/parquet/src/main/java/io/github/dfa1/vortex/parquet/HttpInputFile.java new file mode 100644 index 000000000..ace152fda --- /dev/null +++ b/parquet/src/main/java/io/github/dfa1/vortex/parquet/HttpInputFile.java @@ -0,0 +1,94 @@ +package io.github.dfa1.vortex.parquet; + +import dev.hardwood.InputFile; + +import java.io.IOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.ByteBuffer; +import java.util.OptionalLong; + +/// [InputFile] backed by HTTP Range requests, mirroring `VortexHttpReader`'s range-fetch pattern +/// on the reader side of the wire format. Each [#readRange] call fires one targeted `Range` GET; +/// no full-file download occurs. [#length] is discovered once, via `HEAD`, in [#open]. +final class HttpInputFile implements InputFile { + + /// Shared across all instances, mirroring `VortexHttpReader`'s `HttpClient` reuse rationale: + /// the JDK client is heavyweight and designed for reuse. Never closed: lifetime tracks the JVM. + private static final HttpClient DEFAULT_CLIENT = HttpClient.newHttpClient(); + + private final URI uri; + private final HttpClient client; + private long length = -1; + + HttpInputFile(URI uri) { + this(uri, DEFAULT_CLIENT); + } + + HttpInputFile(URI uri, HttpClient client) { + this.uri = uri; + this.client = client; + } + + @Override + public void open() throws IOException { + HttpRequest request = HttpRequest.newBuilder(uri) + .method("HEAD", HttpRequest.BodyPublishers.noBody()) + .build(); + HttpResponse response = send(request, HttpResponse.BodyHandlers.discarding()); + if (response.statusCode() != 200) { + throw new IOException("HTTP " + response.statusCode() + " opening " + uri); + } + OptionalLong contentLength = response.headers().firstValueAsLong("Content-Length"); + if (contentLength.isEmpty()) { + throw new IOException("HEAD response missing Content-Length from " + uri); + } + length = contentLength.getAsLong(); + } + + @Override + public ByteBuffer readRange(long offset, int len) throws IOException { + HttpRequest request = HttpRequest.newBuilder(uri) + .header("Range", "bytes=" + offset + "-" + (offset + len - 1)) + .GET() + .build(); + HttpResponse response = send(request, HttpResponse.BodyHandlers.ofByteArray()); + int status = response.statusCode(); + if (status != 206 && status != 200) { + throw new IOException("HTTP " + status + " fetching range from " + uri); + } + byte[] body = response.body(); + if (body.length != len) { + throw new IOException( + "HTTP range [%d, %d] from %s: expected %d bytes, got %d" + .formatted(offset, offset + len - 1, uri, len, body.length)); + } + return ByteBuffer.wrap(body); + } + + @Override + public long length() { + return length; + } + + @Override + public String name() { + return uri.toString(); + } + + @Override + public void close() { + // The shared HttpClient outlives this InputFile; nothing to release here. + } + + private HttpResponse send(HttpRequest request, HttpResponse.BodyHandler handler) throws IOException { + try { + return client.send(request, handler); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("interrupted fetching " + request.uri(), e); + } + } +} diff --git a/parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetExporter.java b/parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetExporter.java new file mode 100644 index 000000000..39728932b --- /dev/null +++ b/parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetExporter.java @@ -0,0 +1,364 @@ +package io.github.dfa1.vortex.parquet; + +import dev.hardwood.OutputFile; +import dev.hardwood.metadata.LogicalType; +import dev.hardwood.metadata.PhysicalType; +import dev.hardwood.metadata.RepetitionType; +import dev.hardwood.schema.FileSchema; +import dev.hardwood.writer.ColumnBatch; +import dev.hardwood.writer.ColumnWriter; +import dev.hardwood.writer.ParquetFileWriter; +import io.github.dfa1.vortex.core.io.IoBounds; +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.core.model.ExtensionId; +import io.github.dfa1.vortex.core.model.PType; +import io.github.dfa1.vortex.core.model.TimeUnit; +import io.github.dfa1.vortex.core.model.TimestampDtype; +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.VortexHandle; +import io.github.dfa1.vortex.reader.VortexReader; +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.ShortArray; +import io.github.dfa1.vortex.reader.array.VarBinArray; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; + +/// Reads a Vortex file and writes a Parquet file. +/// +/// Supports flat schemas only: every top-level column must be `Bool`, a non-`F16` `Primitive`, +/// `Utf8`, `Binary`, or a `vortex.timestamp` extension over millisecond/microsecond/nanosecond +/// resolution. `Struct`, `List`, `Map`, `F16`, and any extension other than `vortex.timestamp` +/// throw [UnsupportedOperationException] — the inverse direction ([ParquetImporter]) supports +/// nested `LIST`/`STRUCT`; export does not yet. +/// +/// Supported Vortex → Parquet type mapping: +/// - Bool → BOOLEAN +/// - I8/U8, I16/U16 → INT32 (IntType 8/16, signed or unsigned) +/// - I32 → INT32 (no annotation), U32 → INT32 (IntType 32, unsigned) +/// - I64 → INT64 (no annotation), U64 → INT64 (IntType 64, unsigned) +/// - F32 → FLOAT, F64 → DOUBLE +/// - Utf8 → BYTE_ARRAY (STRING), Binary → BYTE_ARRAY (no annotation) +/// - `vortex.timestamp` → INT64 (TIMESTAMP, `isAdjustedToUTC` set from whether the column +/// carries a timezone) +public final class ParquetExporter { + + private ParquetExporter() { + } + + public static void exportParquet(Path vortexPath, Path parquetPath) throws IOException { + exportParquet(vortexPath, parquetPath, ExportOptions.defaults()); + } + + public static void exportParquet(Path vortexPath, Path parquetPath, ExportOptions options) throws IOException { + try (VortexReader reader = VortexReader.open(vortexPath)) { + exportParquet(reader, parquetPath, options); + } + } + + /// Exports an already-open Vortex handle — a local [VortexReader] or a remote + /// [io.github.dfa1.vortex.reader.VortexHttpReader] — to Parquet. The handle is not closed + /// here: the caller opened it and keeps ownership of its lifecycle, the same convention the + /// `csv` module's `CsvExporter` uses for a caller-supplied `Writer`. + /// + /// @param vortex an open handle to the source Vortex data + /// @param parquetPath destination Parquet file + /// @throws IOException if reading `vortex` or writing `parquetPath` fails + public static void exportParquet(VortexHandle vortex, Path parquetPath) throws IOException { + exportParquet(vortex, parquetPath, ExportOptions.defaults()); + } + + /// Exports an already-open Vortex handle to Parquet, tuned by `options`. + /// + /// @param vortex an open handle to the source Vortex data + /// @param parquetPath destination Parquet file + /// @param options export tuning (column projection, progress, Hardwood writer config) + /// @throws IOException if reading `vortex` or writing `parquetPath` fails + public static void exportParquet(VortexHandle vortex, Path parquetPath, ExportOptions options) + throws IOException { + if (!(vortex.dtype() instanceof DType.Struct schema)) { + throw new UnsupportedOperationException("only struct root dtype is supported for Parquet export"); + } + List allNames = schema.fieldNames(); + List allTypes = schema.fieldTypes(); + List names = options.hasProjection() ? options.columns() : namesOf(allNames); + List types = resolveTypes(allNames, allTypes, names); + + FileSchema.Builder builder = FileSchema.builder("schema"); + for (int c = 0; c < names.size(); c++) { + addColumn(builder, names.get(c), types.get(c)); + } + FileSchema parquetSchema = builder.build(); + + long rowsTotal = options.progressListener() != null ? vortex.layout().rowCount() : 0L; + long rowsDone = 0; + + try (ParquetFileWriter writer = ParquetFileWriter.create( + OutputFile.of(parquetPath), parquetSchema, options.writerConfig()); + ScanIterator scan = vortex.scan(ScanOptions.columns(names.toArray(String[]::new)))) { + + ColumnWriter columnWriter = writer.columnWriter(); + while (scan.hasNext()) { + try (Chunk chunk = scan.next()) { + int rowCount = IoBounds.checkCount(chunk.rowCount()); + columnWriter.writeBatch(batch -> { + for (int c = 0; c < names.size(); c++) { + Array array = chunk.column(names.get(c)); + writeColumn(batch, c, types.get(c), array, rowCount); + } + }); + rowsDone += rowCount; + if (options.progressListener() != null) { + options.progressListener().onProgress(rowsDone, rowsTotal); + } + } + } + } + } + + private static List namesOf(List names) { + List result = new ArrayList<>(names.size()); + for (ColumnName name : names) { + result.add(name.value()); + } + return result; + } + + /// Resolves `names` (either every top-level column, or a caller-requested projection) to + /// their declared dtypes, in the order given. + static List resolveTypes(List allNames, List allTypes, List names) { + List result = new ArrayList<>(names.size()); + for (String name : names) { + int idx = allNames.indexOf(ColumnName.of(name)); + if (idx < 0) { + throw new IllegalArgumentException("column not found in Vortex schema: " + name); + } + result.add(allTypes.get(idx)); + } + return result; + } + + static void addColumn(FileSchema.Builder builder, String name, DType type) { + RepetitionType rep = type.nullable() ? RepetitionType.OPTIONAL : RepetitionType.REQUIRED; + switch (type) { + case DType.Bool ignored -> builder.addColumn(name, PhysicalType.BOOLEAN, rep); + case DType.Utf8 ignored -> + builder.addColumn(name, PhysicalType.BYTE_ARRAY, rep, new LogicalType.StringType()); + case DType.Binary ignored -> builder.addColumn(name, PhysicalType.BYTE_ARRAY, rep); + case DType.Primitive p -> addPrimitiveColumn(builder, name, p.ptype(), rep); + case DType.Extension ext -> addTimestampColumn(builder, name, ext, rep); + default -> throw new UnsupportedOperationException( + "unsupported column type for Parquet export (STRUCT/LIST/MAP not yet supported): " + type + + " (column: " + name + ")"); + } + } + + private static void addPrimitiveColumn(FileSchema.Builder builder, String name, PType ptype, RepetitionType rep) { + switch (ptype) { + case I8 -> builder.addColumn(name, PhysicalType.INT32, rep, new LogicalType.IntType(8, true)); + case U8 -> builder.addColumn(name, PhysicalType.INT32, rep, new LogicalType.IntType(8, false)); + case I16 -> builder.addColumn(name, PhysicalType.INT32, rep, new LogicalType.IntType(16, true)); + case U16 -> builder.addColumn(name, PhysicalType.INT32, rep, new LogicalType.IntType(16, false)); + case I32 -> builder.addColumn(name, PhysicalType.INT32, rep); + case U32 -> builder.addColumn(name, PhysicalType.INT32, rep, new LogicalType.IntType(32, false)); + case I64 -> builder.addColumn(name, PhysicalType.INT64, rep); + case U64 -> builder.addColumn(name, PhysicalType.INT64, rep, new LogicalType.IntType(64, false)); + case F32 -> builder.addColumn(name, PhysicalType.FLOAT, rep); + case F64 -> builder.addColumn(name, PhysicalType.DOUBLE, rep); + case F16 -> throw new UnsupportedOperationException( + "F16 columns are not supported for Parquet export (column: " + name + ")"); + } + } + + /// Maps a `vortex.timestamp` extension to Parquet's `TIMESTAMP` logical type. `Seconds` and + /// `Days` have no Parquet `TIMESTAMP` equivalent (the format only defines MILLIS/MICROS/NANOS + /// resolution) and throw; any other extension id throws, since only `vortex.timestamp` is + /// supported for export. + private static void addTimestampColumn(FileSchema.Builder builder, String name, DType.Extension ext, + RepetitionType rep) { + if (!ExtensionId.VORTEX_TIMESTAMP.id().equals(ext.extensionId())) { + throw new UnsupportedOperationException( + "unsupported extension type for Parquet export: " + ext.extensionId() + " (column: " + name + ")"); + } + TimeUnit unit = TimestampDtype.readUnit(ext); + LogicalType.TimeUnit parquetUnit = switch (unit) { + case Milliseconds -> LogicalType.TimeUnit.MILLIS; + case Microseconds -> LogicalType.TimeUnit.MICROS; + case Nanoseconds -> LogicalType.TimeUnit.NANOS; + case Seconds, Days -> throw new UnsupportedOperationException( + "Parquet TIMESTAMP has no " + unit + " resolution (column: " + name + ")"); + }; + boolean isAdjustedToUtc = TimestampDtype.timezone(ext).isPresent(); + builder.addColumn(name, PhysicalType.INT64, rep, new LogicalType.TimestampType(isAdjustedToUtc, parquetUnit)); + } + + /// Writes one column's values for the current batch, addressed by leaf index `idx` + /// (matching the schema order [#addColumn] built it in). A nullable column decodes as a + /// [MaskedArray]; its per-row nulls are read once into a `boolean[]` mask and its inner + /// (unmasked) array supplies the values — the value at a null row is still read rather than + /// skipped (Hardwood ignores it), keeping the read loop branch-free per row. + static void writeColumn(ColumnBatch batch, int idx, DType type, Array array, int rowCount) { + Array target = array; + boolean[] nulls = null; + if (array instanceof MaskedArray masked) { + target = masked.inner(); + nulls = readNulls(masked, rowCount); + } + switch (type) { + case DType.Bool ignored -> writeBooleans(batch, idx, target, rowCount, nulls); + case DType.Utf8 ignored -> writeBytes(batch, idx, target, rowCount, nulls); + case DType.Binary ignored -> writeBytes(batch, idx, target, rowCount, nulls); + case DType.Primitive p -> writePrimitive(batch, idx, p.ptype(), target, rowCount, nulls); + case DType.Extension ignored -> writeLongs(batch, idx, target, rowCount, nulls); + default -> throw new UnsupportedOperationException("unsupported column type for Parquet export: " + type); + } + } + + private static boolean[] readNulls(MaskedArray masked, int rowCount) { + boolean[] nulls = new boolean[rowCount]; + for (int i = 0; i < rowCount; i++) { + nulls[i] = !masked.isValid(i); + } + return nulls; + } + + private static void writePrimitive(ColumnBatch batch, int idx, PType ptype, Array target, int rowCount, + boolean[] nulls) { + switch (ptype) { + case I8, U8, I16, U16, I32, U32 -> { + int[] values = readInts(target, rowCount); + if (nulls == null) { + batch.ints(idx, values); + } else { + batch.ints(idx, values, nulls); + } + } + case I64, U64 -> writeLongs(batch, idx, target, rowCount, nulls); + case F32 -> { + float[] values = readFloats((FloatArray) target, rowCount); + if (nulls == null) { + batch.floats(idx, values); + } else { + batch.floats(idx, values, nulls); + } + } + case F64 -> { + double[] values = readDoubles((DoubleArray) target, rowCount); + if (nulls == null) { + batch.doubles(idx, values); + } else { + batch.doubles(idx, values, nulls); + } + } + case F16 -> throw new UnsupportedOperationException("F16 columns are not supported for Parquet export"); + } + } + + private static void writeBooleans(ColumnBatch batch, int idx, Array target, int rowCount, boolean[] nulls) { + boolean[] values = readBooleans((BoolArray) target, rowCount); + if (nulls == null) { + batch.booleans(idx, values); + } else { + batch.booleans(idx, values, nulls); + } + } + + private static void writeBytes(ColumnBatch batch, int idx, Array target, int rowCount, boolean[] nulls) { + byte[][] values = readBytes((VarBinArray) target, rowCount); + if (nulls == null) { + batch.bytes(idx, values); + } else { + batch.bytes(idx, values, nulls); + } + } + + private static void writeLongs(ColumnBatch batch, int idx, Array target, int rowCount, boolean[] nulls) { + long[] values = readLongs((LongArray) target, rowCount); + if (nulls == null) { + batch.longs(idx, values); + } else { + batch.longs(idx, values, nulls); + } + } + + /// Reads a fixed-width integer array widened to `int`, dispatching on the array's storage + /// width. `IntArray`/`ShortArray`/`ByteArray#getInt` each already return the mathematically + /// correct value for the column's own signedness (sign-extended for a signed narrow type, + /// zero-extended for unsigned), so the raw widened value is exactly what Parquet's matching + /// `IntType` annotation expects — no further conversion needed. + private static int[] readInts(Array arr, int rowCount) { + int[] values = new int[rowCount]; + switch (arr) { + case IntArray ia -> { + for (int i = 0; i < rowCount; i++) { + values[i] = ia.getInt(i); + } + } + case ShortArray sa -> { + for (int i = 0; i < rowCount; i++) { + values[i] = sa.getInt(i); + } + } + case ByteArray ba -> { + for (int i = 0; i < rowCount; i++) { + values[i] = ba.getInt(i); + } + } + default -> throw new IllegalStateException( + "expected an int-widening array, got " + arr.getClass().getSimpleName()); + } + return values; + } + + private static long[] readLongs(LongArray arr, int rowCount) { + long[] values = new long[rowCount]; + for (int i = 0; i < rowCount; i++) { + values[i] = arr.getLong(i); + } + return values; + } + + private static float[] readFloats(FloatArray arr, int rowCount) { + float[] values = new float[rowCount]; + for (int i = 0; i < rowCount; i++) { + values[i] = arr.getFloat(i); + } + return values; + } + + private static double[] readDoubles(DoubleArray arr, int rowCount) { + double[] values = new double[rowCount]; + for (int i = 0; i < rowCount; i++) { + values[i] = arr.getDouble(i); + } + return values; + } + + private static boolean[] readBooleans(BoolArray arr, int rowCount) { + boolean[] values = new boolean[rowCount]; + for (int i = 0; i < rowCount; i++) { + values[i] = arr.getBoolean(i); + } + return values; + } + + private static byte[][] readBytes(VarBinArray arr, int rowCount) { + byte[][] values = new byte[rowCount][]; + for (int i = 0; i < rowCount; i++) { + values[i] = arr.getBytes(i); + } + return values; + } +} diff --git a/parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetImporter.java b/parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetImporter.java index 81dd7e864..1dc8e3a19 100644 --- a/parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetImporter.java +++ b/parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetImporter.java @@ -19,6 +19,7 @@ import io.github.dfa1.vortex.writer.VortexWriter; import java.io.IOException; +import java.net.URI; import java.nio.channels.FileChannel; import java.nio.file.Path; import java.nio.file.StandardOpenOption; @@ -60,7 +61,32 @@ public static void importParquet(Path parquetPath, Path vortexPath) throws IOExc } public static void importParquet(Path parquetPath, Path vortexPath, ImportOptions options) throws IOException { - try (ParquetFileReader parquet = ParquetFileReader.open(InputFile.of(parquetPath))) { + importParquet(InputFile.of(parquetPath), parquetPath.toString(), vortexPath, options); + } + + /// Imports a Parquet file served over HTTP(S), fetched entirely through targeted Range + /// requests (see [HttpInputFile]) — no full-file download occurs. + /// + /// @param parquetUri the `http(s)://` URL of the source Parquet file + /// @param vortexPath destination Vortex file + /// @throws IOException if reading `parquetUri` or writing `vortexPath` fails + public static void importParquet(URI parquetUri, Path vortexPath) throws IOException { + importParquet(parquetUri, vortexPath, ImportOptions.defaults()); + } + + /// Imports a Parquet file served over HTTP(S), tuned by `options`. + /// + /// @param parquetUri the `http(s)://` URL of the source Parquet file + /// @param vortexPath destination Vortex file + /// @param options import tuning (chunk size, column projection, progress, write options) + /// @throws IOException if reading `parquetUri` or writing `vortexPath` fails + public static void importParquet(URI parquetUri, Path vortexPath, ImportOptions options) throws IOException { + importParquet(new HttpInputFile(parquetUri), parquetUri.toString(), vortexPath, options); + } + + private static void importParquet(InputFile input, String sourceName, Path vortexPath, ImportOptions options) + throws IOException { + try (ParquetFileReader parquet = ParquetFileReader.open(input)) { FileSchema fileSchema = parquet.getFileSchema(); List allTopLevel = fileSchema.getRootNode().children(); List topLevel = options.hasProjection() @@ -75,7 +101,7 @@ public static void importParquet(Path parquetPath, Path vortexPath, ImportOption names.add(ColumnName.of(node.name())); types.add(mapDType(node)); } - checkNoDuplicateNames(names, parquetPath); + checkNoDuplicateNames(names, sourceName); DType.Struct schema = new DType.Struct(names, types, false); long totalRows = parquet.getFileMetaData().numRows(); @@ -97,7 +123,7 @@ public static void importParquet(Path parquetPath, Path vortexPath, ImportOption while (rowReader.hasNext()) { rowReader.next(); - fillRow(rowReader, names, types, buffers, nestedBuilders, chunkPos, parquetPath, rowsDone); + fillRow(rowReader, names, types, buffers, nestedBuilders, chunkPos, sourceName, rowsDone); chunkPos++; rowsDone++; @@ -122,7 +148,7 @@ public static void importParquet(Path parquetPath, Path vortexPath, ImportOption } } - static void checkNoDuplicateNames(List names, Path parquetPath) { + static void checkNoDuplicateNames(List names, String sourceName) { Set seen = new HashSet<>(); Set duplicates = new LinkedHashSet<>(); for (ColumnName name : names) { @@ -132,7 +158,7 @@ static void checkNoDuplicateNames(List names, Path parquetPath) { } if (!duplicates.isEmpty()) { throw new IllegalArgumentException( - "Parquet schema has duplicate column name(s): " + duplicates + "; source file: " + parquetPath); + "Parquet schema has duplicate column name(s): " + duplicates + "; source file: " + sourceName); } } @@ -300,7 +326,7 @@ private static Object allocateBuffer(DType type, int chunkSize) { } static void fillRow(RowReader reader, List names, List types, - Object[] buffers, ColumnBuilder[] nestedBuilders, int pos, Path parquetPath, long rowIndex) { + Object[] buffers, ColumnBuilder[] nestedBuilders, int pos, String sourceName, long rowIndex) { for (int c = 0; c < names.size(); c++) { String name = names.get(c).value(); try { @@ -339,7 +365,7 @@ static void fillRow(RowReader reader, List names, List types, throw new IllegalArgumentException( "Parquet column '" + name + "' is declared REQUIRED (non-nullable) but row " + rowIndex + " is null; the file's data violates its own schema. Source file: " - + parquetPath, e); + + sourceName, e); } } } diff --git a/parquet/src/test/java/io/github/dfa1/vortex/parquet/HttpInputFileTest.java b/parquet/src/test/java/io/github/dfa1/vortex/parquet/HttpInputFileTest.java new file mode 100644 index 000000000..4588b6372 --- /dev/null +++ b/parquet/src/test/java/io/github/dfa1/vortex/parquet/HttpInputFileTest.java @@ -0,0 +1,168 @@ +package io.github.dfa1.vortex.parquet; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.io.IOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpHeaders; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import javax.net.ssl.SSLSession; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doReturn; + +/// Verifies [HttpInputFile]'s HTTP Range-request mechanics against a mocked [HttpClient], the +/// same deterministic, network-free approach `VortexHttpReaderTailFetchTest` uses on the reader +/// side of the wire format. +@ExtendWith(MockitoExtension.class) +class HttpInputFileTest { + + @Mock + private HttpClient client; + + private static final URI URI = java.net.URI.create("http://example.com/data.parquet"); + + @Test + void open_headResponseWithContentLength_setsLength() throws Exception { + // Given + doReturn(response(200, Map.of("content-length", List.of("12345")), null)) + .when(client).send(any(), any()); + HttpInputFile sut = new HttpInputFile(URI, client); + + // When + sut.open(); + + // Then + assertThat(sut.length()).isEqualTo(12345L); + } + + @Test + void open_missingContentLength_throws() throws Exception { + // Given — a HEAD response carrying no Content-Length header + doReturn(response(200, Map.of(), null)).when(client).send(any(), any()); + HttpInputFile sut = new HttpInputFile(URI, client); + + // When / Then + assertThatThrownBy(sut::open) + .isInstanceOf(IOException.class) + .hasMessageContaining("Content-Length"); + } + + @Test + void open_non200Status_throws() throws Exception { + // Given + doReturn(response(404, Map.of(), null)).when(client).send(any(), any()); + HttpInputFile sut = new HttpInputFile(URI, client); + + // When / Then + assertThatThrownBy(sut::open) + .isInstanceOf(IOException.class) + .hasMessageContaining("404"); + } + + @Test + void readRange_returnsRequestedBytes() throws Exception { + // Given + byte[] body = "hello".getBytes(StandardCharsets.UTF_8); + doReturn(response(206, Map.of(), body)).when(client).send(any(), any()); + HttpInputFile sut = new HttpInputFile(URI, client); + + // When + ByteBuffer result = sut.readRange(10, body.length); + + // Then + byte[] actual = new byte[result.remaining()]; + result.get(actual); + assertThat(actual).isEqualTo(body); + } + + @Test + void readRange_bodyLengthMismatch_throws() throws Exception { + // Given — server returned fewer bytes than the requested range length + byte[] body = "short".getBytes(StandardCharsets.UTF_8); + doReturn(response(206, Map.of(), body)).when(client).send(any(), any()); + HttpInputFile sut = new HttpInputFile(URI, client); + + // When / Then + assertThatThrownBy(() -> sut.readRange(0, body.length + 1)) + .isInstanceOf(IOException.class) + .hasMessageContaining("expected"); + } + + @Test + void readRange_badStatus_throws() throws Exception { + // Given + doReturn(response(500, Map.of(), new byte[0])).when(client).send(any(), any()); + HttpInputFile sut = new HttpInputFile(URI, client); + + // When / Then + assertThatThrownBy(() -> sut.readRange(0, 1)) + .isInstanceOf(IOException.class) + .hasMessageContaining("500"); + } + + @Test + void name_returnsUriString() { + // Given / When / Then + assertThat(new HttpInputFile(URI, client).name()).isEqualTo(URI.toString()); + } + + // ── helpers ─────────────────────────────────────────────────────────────── + + @SuppressWarnings("unchecked") + private static HttpResponse response(int status, Map> headers, T body) { + return new HttpResponse<>() { + @Override + public int statusCode() { + return status; + } + + @Override + public T body() { + return body; + } + + @Override + public HttpHeaders headers() { + return HttpHeaders.of(headers, (k, v) -> true); + } + + @Override + public HttpRequest request() { + return null; + } + + @Override + public Optional> previousResponse() { + return Optional.empty(); + } + + @Override + public Optional sslSession() { + return Optional.empty(); + } + + @Override + public java.net.URI uri() { + return URI; + } + + @Override + public HttpClient.Version version() { + return HttpClient.Version.HTTP_1_1; + } + }; + } +} diff --git a/parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetExporterTest.java b/parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetExporterTest.java new file mode 100644 index 000000000..5c7670f8c --- /dev/null +++ b/parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetExporterTest.java @@ -0,0 +1,465 @@ +package io.github.dfa1.vortex.parquet; + +import dev.hardwood.InputFile; +import dev.hardwood.metadata.LogicalType; +import dev.hardwood.metadata.PhysicalType; +import dev.hardwood.metadata.RepetitionType; +import dev.hardwood.reader.ParquetFileReader; +import dev.hardwood.reader.RowReader; +import dev.hardwood.schema.FileSchema; +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.core.model.PType; +import io.github.dfa1.vortex.core.model.TimeUnit; +import io.github.dfa1.vortex.core.model.TimestampDtype; +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 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.IntArray; +import io.github.dfa1.vortex.reader.array.LongArray; +import io.github.dfa1.vortex.reader.array.MaskedArray; +import io.github.dfa1.vortex.reader.array.VarBinArray; +import io.github.dfa1.vortex.writer.VortexWriter; +import io.github.dfa1.vortex.writer.WriteOptions; +import org.junit.jupiter.api.Nested; +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.CsvSource; + +import java.nio.channels.FileChannel; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.time.Instant; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class ParquetExporterTest { + + private static FileSchema schemaOf(String name, DType type) { + FileSchema.Builder builder = FileSchema.builder("schema"); + ParquetExporter.addColumn(builder, name, type); + return builder.build(); + } + + private static Path writeVortex(Path tmp, String fileName, List names, List types, + Map chunk) throws Exception { + DType.Struct schema = new DType.Struct(names, types, false); + Path vortex = tmp.resolve(fileName); + try (FileChannel ch = FileChannel.open(vortex, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + VortexWriter writer = VortexWriter.create(ch, schema, WriteOptions.defaults())) { + writer.writeChunk(chunk); + } + return vortex; + } + + @Nested + class TypeMapping { + + @Test + void bool_mapsToBoolean_carryingNullability() { + // Given / When / Then — REQUIRED is non-null, OPTIONAL is nullable + assertThat(schemaOf("b", new DType.Bool(false)).getColumn(0).repetitionType()) + .isEqualTo(RepetitionType.REQUIRED); + assertThat(schemaOf("b", new DType.Bool(true)).getColumn(0).repetitionType()) + .isEqualTo(RepetitionType.OPTIONAL); + assertThat(schemaOf("b", new DType.Bool(false)).getColumn(0).type()).isEqualTo(PhysicalType.BOOLEAN); + } + + @Test + void i32_mapsToInt32_withoutAnnotation() { + // Given + DType type = new DType.Primitive(PType.I32, false); + + // When + var column = schemaOf("i", type).getColumn(0); + + // Then — bare INT32, matching how ParquetImporter also reads an un-annotated INT32 as I32 + assertThat(column.type()).isEqualTo(PhysicalType.INT32); + assertThat(column.logicalType()).isNull(); + } + + @Test + void i64_mapsToInt64_withoutAnnotation() { + // Given + DType type = new DType.Primitive(PType.I64, false); + + // When + var column = schemaOf("l", type).getColumn(0); + + // Then + assertThat(column.type()).isEqualTo(PhysicalType.INT64); + assertThat(column.logicalType()).isNull(); + } + + @ParameterizedTest + @CsvSource({ + "I8, 8, true", + "U8, 8, false", + "I16, 16, true", + "U16, 16, false", + "U32, 32, false", + "U64, 64, false", + }) + void narrowOrUnsignedPrimitive_carriesIntAnnotation(PType ptype, int bitWidth, boolean signed) { + // Given — every PType this codebase carries an IntType annotation for on export, + // the inverse of ParquetImporter's mapInt32/mapInt64 + PhysicalType expectedPhysical = bitWidth == 64 ? PhysicalType.INT64 : PhysicalType.INT32; + + // When + var column = schemaOf("i", new DType.Primitive(ptype, false)).getColumn(0); + + // Then + assertThat(column.type()).isEqualTo(expectedPhysical); + assertThat(column.logicalType()).isEqualTo(new LogicalType.IntType(bitWidth, signed)); + } + + @Test + void f32_mapsToFloat_f64_mapsToDouble() { + // Given / When / Then + assertThat(schemaOf("f", new DType.Primitive(PType.F32, false)).getColumn(0).type()) + .isEqualTo(PhysicalType.FLOAT); + assertThat(schemaOf("d", new DType.Primitive(PType.F64, false)).getColumn(0).type()) + .isEqualTo(PhysicalType.DOUBLE); + } + + @Test + void f16_throws() { + // When / Then — mirrors ParquetImporter, which never produces F16 either + assertThatThrownBy(() -> schemaOf("f", new DType.Primitive(PType.F16, false))) + .isInstanceOf(UnsupportedOperationException.class); + } + + @Test + void utf8_mapsToByteArray_withStringAnnotation() { + // When + var column = schemaOf("s", new DType.Utf8(true)).getColumn(0); + + // Then + assertThat(column.type()).isEqualTo(PhysicalType.BYTE_ARRAY); + assertThat(column.logicalType()).isEqualTo(new LogicalType.StringType()); + assertThat(column.repetitionType()).isEqualTo(RepetitionType.OPTIONAL); + } + + @Test + void binary_mapsToByteArray_withoutAnnotation() { + // When + var column = schemaOf("b", new DType.Binary(false)).getColumn(0); + + // Then + assertThat(column.type()).isEqualTo(PhysicalType.BYTE_ARRAY); + assertThat(column.logicalType()).isNull(); + } + + @ParameterizedTest + @CsvSource({"Milliseconds, MILLIS", "Microseconds, MICROS", "Nanoseconds, NANOS"}) + void timestamp_mapsToInt64Timestamp_carryingUnit(TimeUnit unit, LogicalType.TimeUnit expectedUnit) { + // Given — no timezone recorded, so isAdjustedToUTC is false + DType.Extension ts = TimestampDtype.of(unit, null, true); + + // When + var column = schemaOf("ts", ts).getColumn(0); + + // Then + assertThat(column.type()).isEqualTo(PhysicalType.INT64); + assertThat(column.logicalType()).isEqualTo(new LogicalType.TimestampType(false, expectedUnit)); + assertThat(column.repetitionType()).isEqualTo(RepetitionType.OPTIONAL); + } + + @Test + void timestamp_withTimezone_isAdjustedToUtc() { + // Given + DType.Extension ts = TimestampDtype.of(TimeUnit.Milliseconds, java.time.ZoneOffset.UTC, false); + + // When + var column = schemaOf("ts", ts).getColumn(0); + + // Then + assertThat(column.logicalType()).isEqualTo(new LogicalType.TimestampType(true, LogicalType.TimeUnit.MILLIS)); + } + + @ParameterizedTest + @CsvSource({"Seconds", "Days"}) + void timestamp_secondsOrDaysResolution_throws(TimeUnit unit) { + // Given — Parquet TIMESTAMP has no SECONDS/DAYS resolution + DType.Extension ts = TimestampDtype.of(unit, null, false); + + // When / Then + assertThatThrownBy(() -> schemaOf("ts", ts)).isInstanceOf(UnsupportedOperationException.class); + } + + @Test + void nonTimestampExtension_throws() { + // Given — a made-up extension id this exporter has no mapping for + DType.Extension ext = new DType.Extension("vortex.uuid", new DType.Primitive(PType.I64, false), null, false); + + // When / Then + assertThatThrownBy(() -> schemaOf("u", ext)).isInstanceOf(UnsupportedOperationException.class); + } + + @Test + void struct_throws() { + // Given — nested schemas are out of scope for this exporter + DType.Struct nested = new DType.Struct(List.of(ColumnName.of("x")), List.of(DType.I32), false); + + // When / Then + assertThatThrownBy(() -> schemaOf("s", nested)).isInstanceOf(UnsupportedOperationException.class); + } + + @Test + void list_throws() { + // Given + DType.List list = new DType.List(DType.I32, false); + + // When / Then + assertThatThrownBy(() -> schemaOf("l", list)).isInstanceOf(UnsupportedOperationException.class); + } + } + + @Nested + class Export { + + @Test + void exportsFlatFile_schemaAndValuesRoundTrip(@TempDir Path tmp) throws Exception { + // Given — nullable I64 id, nullable Utf8 name, non-null Bool active + Path vortex = writeVortex(tmp, "in.vortex", + List.of(ColumnName.of("id"), ColumnName.of("name"), ColumnName.of("active")), + List.of(new DType.Primitive(PType.I64, true), new DType.Utf8(true), new DType.Bool(false)), + Map.of( + ColumnName.of("id"), new Long[]{1L, null, 3L}, + ColumnName.of("name"), new String[]{"Ada", "Grace", null}, + ColumnName.of("active"), new boolean[]{true, false, true})); + Path parquet = tmp.resolve("out.parquet"); + + // When + ParquetExporter.exportParquet(vortex, parquet); + + // Then + try (ParquetFileReader reader = ParquetFileReader.open(InputFile.of(parquet))) { + assertThat(reader.getFileMetaData().numRows()).isEqualTo(3L); + try (RowReader rows = reader.buildRowReader().build()) { + assertThat(rows.hasNext()).isTrue(); + rows.next(); + assertThat(rows.getLong("id")).isEqualTo(1L); + assertThat(rows.getString("name")).isEqualTo("Ada"); + assertThat(rows.getBoolean("active")).isTrue(); + + assertThat(rows.hasNext()).isTrue(); + rows.next(); + assertThat(rows.isNull("id")).isTrue(); + assertThat(rows.getString("name")).isEqualTo("Grace"); + assertThat(rows.getBoolean("active")).isFalse(); + + assertThat(rows.hasNext()).isTrue(); + rows.next(); + assertThat(rows.getLong("id")).isEqualTo(3L); + assertThat(rows.isNull("name")).isTrue(); + assertThat(rows.getBoolean("active")).isTrue(); + + assertThat(rows.hasNext()).isFalse(); + } + } + } + + @Test + void exportsTimestampColumn_asInt64Timestamp(@TempDir Path tmp) throws Exception { + // Given — a vortex.timestamp column at millisecond resolution + DType.Extension tsDtype = TimestampDtype.of(TimeUnit.Milliseconds, null, false); + Path vortex = writeVortex(tmp, "in.vortex", + List.of(ColumnName.of("events")), + List.of(tsDtype), + Map.of(ColumnName.of("events"), List.of( + Instant.ofEpochMilli(-1_500L), + Instant.EPOCH, + Instant.ofEpochMilli(1_733_000_000_000L)))); + Path parquet = tmp.resolve("out.parquet"); + + // When + ParquetExporter.exportParquet(vortex, parquet); + + // Then — the raw epoch-millisecond values round-trip exactly + try (ParquetFileReader reader = ParquetFileReader.open(InputFile.of(parquet)); + RowReader rows = reader.buildRowReader().build()) { + rows.next(); + assertThat(rows.getLong("events")).isEqualTo(-1_500L); + rows.next(); + assertThat(rows.getLong("events")).isEqualTo(0L); + rows.next(); + assertThat(rows.getLong("events")).isEqualTo(1_733_000_000_000L); + } + } + + @Test + void projection_exportsOnlyRequestedColumns(@TempDir Path tmp) throws Exception { + // Given + Path vortex = writeVortex(tmp, "in.vortex", + List.of(ColumnName.of("id"), ColumnName.of("name")), + List.of(new DType.Primitive(PType.I64, false), new DType.Utf8(false)), + Map.of(ColumnName.of("id"), new long[]{1L, 2L}, ColumnName.of("name"), new String[]{"a", "b"})); + Path parquet = tmp.resolve("out.parquet"); + ExportOptions options = ExportOptions.defaults().withColumns(List.of("id")); + + // When + ParquetExporter.exportParquet(vortex, parquet, options); + + // Then — only the projected column survives + try (ParquetFileReader reader = ParquetFileReader.open(InputFile.of(parquet))) { + assertThat(reader.getFileSchema().getColumns()).extracting(c -> c.name()).containsExactly("id"); + } + } + + @Test + void projection_unknownColumn_throws(@TempDir Path tmp) throws Exception { + // Given + Path vortex = writeVortex(tmp, "in.vortex", + List.of(ColumnName.of("id")), List.of(new DType.Primitive(PType.I64, false)), + Map.of(ColumnName.of("id"), new long[]{1L})); + Path parquet = tmp.resolve("out.parquet"); + ExportOptions options = ExportOptions.defaults().withColumns(List.of("does_not_exist")); + + // When / Then + assertThatThrownBy(() -> ParquetExporter.exportParquet(vortex, parquet, options)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("does_not_exist"); + } + + @Test + void structColumn_throws(@TempDir Path tmp) throws Exception { + // Given — a top-level Struct column, out of scope for this exporter; the schema + // alone triggers the rejection (at FileSchema-building time, before any chunk is + // read), so the file needs no rows + DType.Struct fieldSchema = new DType.Struct(List.of(ColumnName.of("x")), List.of(DType.I32), false); + DType.Struct schema = new DType.Struct(List.of(ColumnName.of("s")), List.of(fieldSchema), false); + Path vortex = tmp.resolve("in.vortex"); + try (FileChannel ch = FileChannel.open(vortex, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + VortexWriter writer = VortexWriter.create(ch, schema, WriteOptions.defaults())) { + // no writeChunk — an empty file still declares the Struct-typed column + assertThat(writer).isNotNull(); + } + Path parquet = tmp.resolve("out.parquet"); + + // When / Then + assertThatThrownBy(() -> ParquetExporter.exportParquet(vortex, parquet)) + .isInstanceOf(UnsupportedOperationException.class); + } + } + + @Nested + class RoundTrip { + + @Test + void flatTypes_roundTripThroughParquetAndBack(@TempDir Path tmp) throws Exception { + // Given — nullable/non-null primitives (including U8/U32 boundary values, whose raw + // bit pattern must survive both the Parquet IntType(unsigned) annotation on export + // and ParquetImporter's re-decode), Utf8, Binary and Bool + Path original = writeVortex(tmp, "original.vortex", + List.of(ColumnName.of("id"), ColumnName.of("age"), ColumnName.of("score"), + ColumnName.of("name"), ColumnName.of("blob"), ColumnName.of("active"), + ColumnName.of("bigCount")), + List.of(new DType.Primitive(PType.I64, true), + new DType.Primitive(PType.U8, false), + new DType.Primitive(PType.F64, true), + new DType.Utf8(true), + new DType.Binary(false), + new DType.Bool(false), + new DType.Primitive(PType.U32, false)), + Map.of( + ColumnName.of("id"), new Long[]{1L, null, -3L}, + ColumnName.of("age"), new byte[]{0, (byte) 255, 42}, + ColumnName.of("score"), new Double[]{1.5, null, -2.25}, + ColumnName.of("name"), new String[]{"Ada", null, "Grace"}, + ColumnName.of("blob"), new byte[][]{{1, 2}, {}, {9, 9, 9}}, + ColumnName.of("active"), new boolean[]{true, false, true}, + ColumnName.of("bigCount"), new int[]{0, -1, 12345})); + Path parquet = tmp.resolve("out.parquet"); + Path reimported = tmp.resolve("reimported.vortex"); + + // When + ParquetExporter.exportParquet(original, parquet); + ParquetImporter.importParquet(parquet, reimported); + + // Then — every column survives the Vortex -> Parquet -> Vortex trip unchanged + try (VortexReader reader = VortexReader.open(reimported); + ScanIterator iter = reader.scan(ScanOptions.all())) { + assertThat(iter.hasNext()).isTrue(); + try (Chunk chunk = iter.next()) { + MaskedArray id = chunk.column("id"); + LongArray idValues = (LongArray) id.inner(); + assertThat(id.isValid(0)).isTrue(); + assertThat(idValues.getLong(0)).isEqualTo(1L); + assertThat(id.isValid(1)).isFalse(); + assertThat(id.isValid(2)).isTrue(); + assertThat(idValues.getLong(2)).isEqualTo(-3L); + + ByteArray age = chunk.column("age"); + assertThat(age.getInt(0)).isEqualTo(0); + assertThat(age.getInt(1)).isEqualTo(255); + assertThat(age.getInt(2)).isEqualTo(42); + + MaskedArray score = chunk.column("score"); + DoubleArray scoreValues = (DoubleArray) score.inner(); + assertThat(scoreValues.getDouble(0)).isEqualTo(1.5); + assertThat(score.isValid(1)).isFalse(); + assertThat(scoreValues.getDouble(2)).isEqualTo(-2.25); + + MaskedArray name = chunk.column("name"); + VarBinArray nameValues = (VarBinArray) name.inner(); + assertThat(nameValues.getString(0)).isEqualTo("Ada"); + assertThat(name.isValid(1)).isFalse(); + assertThat(nameValues.getString(2)).isEqualTo("Grace"); + + VarBinArray blob = chunk.column("blob"); + assertThat(blob.getBytes(0)).isEqualTo(new byte[]{1, 2}); + assertThat(blob.getBytes(1)).isEqualTo(new byte[]{}); + assertThat(blob.getBytes(2)).isEqualTo(new byte[]{9, 9, 9}); + + BoolArray active = chunk.column("active"); + assertThat(active.getBoolean(0)).isTrue(); + assertThat(active.getBoolean(1)).isFalse(); + assertThat(active.getBoolean(2)).isTrue(); + + // bigCount is U32: row 1's raw bit pattern -1 represents 4294967295 unsigned + IntArray bigCount = chunk.column("bigCount"); + assertThat(bigCount.getInt(0)).isEqualTo(0); + assertThat(bigCount.getInt(1)).isEqualTo(-1); + assertThat(bigCount.getInt(2)).isEqualTo(12345); + } + } + } + + @Test + void timestampColumn_roundTripThroughParquetAndBack(@TempDir Path tmp) throws Exception { + // Given — microsecond-resolution vortex.timestamp, pre-epoch/epoch/future values + DType.Extension tsDtype = TimestampDtype.of(TimeUnit.Microseconds, null, false); + List instants = List.of( + Instant.ofEpochMilli(-1_500L), + Instant.EPOCH, + Instant.ofEpochMilli(1_733_000_000_000L)); + Path original = writeVortex(tmp, "original.vortex", + List.of(ColumnName.of("events")), List.of(tsDtype), + Map.of(ColumnName.of("events"), instants)); + Path parquet = tmp.resolve("out.parquet"); + Path reimported = tmp.resolve("reimported.vortex"); + + // When + ParquetExporter.exportParquet(original, parquet); + ParquetImporter.importParquet(parquet, reimported); + + // Then + try (VortexReader reader = VortexReader.open(reimported); + ScanIterator iter = reader.scan(ScanOptions.all())) { + assertThat(iter.hasNext()).isTrue(); + try (Chunk chunk = iter.next()) { + assertThat(chunk.as("events", Instant.class)).containsExactlyElementsOf(instants); + } + } + } + } +} diff --git a/parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetImporterTest.java b/parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetImporterTest.java index c8ad60529..6f4c99d86 100644 --- a/parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetImporterTest.java +++ b/parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetImporterTest.java @@ -310,7 +310,7 @@ void noDuplicates_doesNotThrow() { List names = List.of(ColumnName.of("a"), ColumnName.of("b"), ColumnName.of("c")); // When / Then - assertThatCode(() -> ParquetImporter.checkNoDuplicateNames(names, Path.of("source.parquet"))) + assertThatCode(() -> ParquetImporter.checkNoDuplicateNames(names, "source.parquet")) .doesNotThrowAnyException(); } @@ -320,7 +320,7 @@ void duplicateName_throwsWithNameAndSourcePath() { // headerless source CSV made a Parquet conversion tool use the first data row as column // names, and two property-type flag columns happened to share the value "A". List names = List.of(ColumnName.of("A"), ColumnName.of("B"), ColumnName.of("A")); - Path source = Path.of("uk-price-paid.parquet"); + String source = "uk-price-paid.parquet"; // When / Then assertThatThrownBy(() -> ParquetImporter.checkNoDuplicateNames(names, source)) @@ -349,7 +349,7 @@ void requiredColumnActuallyNull_throwsClearErrorInsteadOfRawNpe() { List types = List.of(DType.I64); Object[] buffers = {new long[1]}; ColumnBuilder[] nestedBuilders = new ColumnBuilder[1]; - Path source = Path.of("malformed.parquet"); + String source = "malformed.parquet"; given(reader.getLong("id")) .willThrow(new NullPointerException("[malformed.parquet] Column 'id' is null at row 0")); diff --git a/pom.xml b/pom.xml index eb82e7ebd..a8d2726a1 100644 --- a/pom.xml +++ b/pom.xml @@ -64,7 +64,7 @@ 4.4.0 - 1.0.0.Final + 1.1.0.Beta1 1.37 1.5.7-15