From fa7fa9dda578bb80171200502c6a14376f6900e8 Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Thu, 3 Sep 2026 18:44:59 +0200 Subject: [PATCH] refactor(cli): FileName/FileFormat domain primitive; fix new-code Sonar findings FileFormat (CSV/PARQUET/VORTEX) + FileName wrap the extension parsing and swapping import/export both need, replacing a scattered set of endsWith(".xxx") checks and hand-counted substring(0, name.length() - N) suffix strips duplicated across ImportCommand and ExportCommand. Also fixes the 17 SonarCloud findings in the new-code period (sinceLeakPeriod=true), triaged individually: - ImportCommand: the .parquet literal duplication (S1192) is resolved as a side effect of the FileName refactor above. Two ternaries moved inside their single Path.of(...) call (S9358). The CSV-to-Parquet temp file now gets owner-only POSIX permissions (rw-------) via FileAttribute, since the system temp directory is commonly world-writable (S5443, CRITICAL) -- falls back to the plain overload on Windows, where FileAttribute-based permissions aren't supported. - CsvImporter: split the 17-cognitive-complexity private importCsv helper into readFirstChunk/writeFirstChunk/streamRemainingChunks, each well under the 15 threshold (S3776). - ParquetExporter: switched `case DType.Xxx ignored ->` to the JDK unnamed-variable `_`, matching the convention CsvImporter's own switches already use elsewhere in this codebase -- `_` is exempt from the unused-variable rule by design, `ignored` isn't (S1481 x7). - ParquetExporterTest: extracted an inline `new DType.Primitive(...)` out of an assertThatThrownBy lambda into a `// Given` variable, so the lambda contains exactly one invocation that can throw (S5778); three isEqualTo(0) assertions switched to isZero() (S5838 x3). - ZstdEncodingEncoderTest: same S5778 shape (EncodeTestHelper.testCtx() invocation extracted out of the lambda), from a slightly earlier commit still inside the leak period. Review caught two real regressions in the FileName refactor itself, both fixed: - FileName#withFormat stripped ANY known extension before appending the target's, including one already equal to the target -- so deriving a default output name from an input already in that format (e.g. `import data.vortex` with no output arg) returned the input's own name back. import/export would then open that path for writing while still reading it as the source, truncating it. Fixed by only stripping a *different* known extension; added a regression test (withFormat_alreadyTargetFormat_appendsRatherThanReturningSameName). - CsvImporter's FirstChunk record had its rows list mutated (.clear()) after construction, fighting the immutable-value-object expectation records set up. Restructured so importCsv stops referencing the FirstChunk (and thus its buffered rows) after handing it to writeFirstChunk, instead of mutating it -- the streaming loop that follows still doesn't retain the buffered rows, without needing a mutation to make that true. Also dropped the redundant chunkSize parameter threaded through two helpers when it was always just options.chunkSize(). Also converted two FileNameTest cases to // Given / // When / // Then (they'd collapsed to a single comment with stacked assertions). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01LjU7zWghUiJauRxK58nyUs --- .../github/dfa1/vortex/cli/ExportCommand.java | 16 +-- .../io/github/dfa1/vortex/cli/FileFormat.java | 40 +++++++ .../io/github/dfa1/vortex/cli/FileName.java | 35 ++++++ .../github/dfa1/vortex/cli/ImportCommand.java | 65 +++++------ .../github/dfa1/vortex/cli/FileNameTest.java | 93 ++++++++++++++++ .../github/dfa1/vortex/csv/CsvImporter.java | 104 +++++++++++------- .../dfa1/vortex/parquet/ParquetExporter.java | 14 +-- .../vortex/parquet/ParquetExporterTest.java | 14 ++- .../encode/ZstdEncodingEncoderTest.java | 3 +- 9 files changed, 287 insertions(+), 97 deletions(-) create mode 100644 cli/src/main/java/io/github/dfa1/vortex/cli/FileFormat.java create mode 100644 cli/src/main/java/io/github/dfa1/vortex/cli/FileName.java create mode 100644 cli/src/test/java/io/github/dfa1/vortex/cli/FileNameTest.java 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 a1a2bc30..28fdfa24 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 @@ -37,9 +37,9 @@ static int run(String[] args) { } Path outputPath = (args.length == 3 && !toStdout) ? Path.of(args[2]) - : deriveOutputPath(inputPath); + : inputPath.resolveSibling(FileName.of(inputPath).withFormat(FileFormat.CSV)); try { - if (!toStdout && outputPath.getFileName().toString().endsWith(".parquet")) { + if (!toStdout && FileName.of(outputPath).is(FileFormat.PARQUET)) { return runParquet(inputPath, outputPath); } return runCsv(inputPath, outputPath, toStdout); @@ -53,7 +53,7 @@ static int run(String[] args) { /// 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")) { + if (toStdout || args.length != 3 || !new FileName(args[2]).is(FileFormat.PARQUET)) { System.err.println("usage: export out.parquet (CSV/stdout export from a URL isn't supported yet)"); return ExitStatus.USAGE_ERROR; } @@ -102,16 +102,6 @@ private static int runParquet(Path inputPath, Path outputPath) throws IOExceptio 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")) { - name = name.substring(0, name.length() - 7); - } - return inputPath.resolveSibling(name + ".csv"); - } - private static void printResult(Path inputPath, Path outputPath) throws IOException { long inputBytes = Files.size(inputPath); long outputBytes = Files.size(outputPath); diff --git a/cli/src/main/java/io/github/dfa1/vortex/cli/FileFormat.java b/cli/src/main/java/io/github/dfa1/vortex/cli/FileFormat.java new file mode 100644 index 00000000..2df693a5 --- /dev/null +++ b/cli/src/main/java/io/github/dfa1/vortex/cli/FileFormat.java @@ -0,0 +1,40 @@ +package io.github.dfa1.vortex.cli; + +import java.util.Optional; + +/// The file formats `import`/`export` recognize by extension. The single source of truth for +/// what an extension means — no other file in this module spells out `.csv`/`.parquet`/`.vortex` +/// or their lengths as string literals. +enum FileFormat { + + CSV(".csv"), + PARQUET(".parquet"), + VORTEX(".vortex"); + + private final String extension; + + FileFormat(String extension) { + this.extension = extension; + } + + String extension() { + return extension; + } + + boolean matches(String fileName) { + return fileName.endsWith(extension); + } + + /// Resolves `fileName`'s format from its extension. + /// + /// @param fileName a file name or URL path, e.g. `"data.parquet"` + /// @return the matching format, or empty if none of `.csv`/`.parquet`/`.vortex` matches + static Optional of(String fileName) { + for (FileFormat format : values()) { + if (format.matches(fileName)) { + return Optional.of(format); + } + } + return Optional.empty(); + } +} diff --git a/cli/src/main/java/io/github/dfa1/vortex/cli/FileName.java b/cli/src/main/java/io/github/dfa1/vortex/cli/FileName.java new file mode 100644 index 00000000..d35be433 --- /dev/null +++ b/cli/src/main/java/io/github/dfa1/vortex/cli/FileName.java @@ -0,0 +1,35 @@ +package io.github.dfa1.vortex.cli; + +import java.nio.file.Path; + +/// A file or URL path's last segment, typed around its [FileFormat]. Centralizes the +/// extension parsing/swapping `import`/`export` both need — replacing a scattered set of +/// `endsWith(".xxx")` checks and hand-counted `substring(0, name.length() - N)` suffix strips. +/// +/// @param value the raw name, e.g. `"data.parquet"` or a URL's last `/`-segment +record FileName(String value) { + + /// The name of `path`'s final component, e.g. `FileName.of(Path.of("a/data.csv"))` is + /// `FileName("data.csv")`. + static FileName of(Path path) { + return new FileName(path.getFileName().toString()); + } + + /// Whether this name ends in `format`'s extension. + boolean is(FileFormat format) { + return format.matches(value); + } + + /// Swaps this name's extension for `target`'s, stripping any *different* known extension + /// first — `"data.csv".withFormat(VORTEX)` and `"data.parquet".withFormat(VORTEX)` both give + /// `"data.vortex"`. A name with no known extension, or one already in `target`'s format, is + /// never stripped — only appended to — so the result always differs from `value`: a caller + /// deriving a default output name from an input name can never get back the input's own name. + String withFormat(FileFormat target) { + String stem = FileFormat.of(value) + .filter(current -> current != target) + .map(current -> value.substring(0, value.length() - current.extension().length())) + .orElse(value); + return stem + target.extension(); + } +} 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 fe8cc307..cca948c8 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 @@ -8,8 +8,11 @@ import java.io.IOException; import java.net.URI; +import java.nio.file.FileSystems; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.attribute.FileAttribute; +import java.nio.file.attribute.PosixFilePermissions; import java.util.ArrayList; import java.util.List; @@ -46,12 +49,12 @@ static int run(String[] args) { System.err.println("file not found: " + inputPath); return ExitStatus.FILE_NOT_FOUND; } - String name = inputPath.getFileName().toString(); + FileName name = FileName.of(inputPath); Path outputPath = parsedArgs.outputTarget() != null ? Path.of(parsedArgs.outputTarget()) - : inputPath.resolveSibling(vortexName(name)); - if (name.endsWith(".parquet")) { - if (isParquetTarget(outputPath)) { + : inputPath.resolveSibling(name.withFormat(FileFormat.VORTEX)); + if (name.is(FileFormat.PARQUET)) { + if (FileName.of(outputPath).is(FileFormat.PARQUET)) { System.err.println("import always converts Parquet to Vortex; " + "a Parquet source cannot import to a .parquet output"); return ExitStatus.USAGE_ERROR; @@ -71,21 +74,22 @@ static int run(String[] args) { /// nothing else is supported from a URL). The output target may independently be `.vortex` /// or `.parquet` — see [#runCsv] / [#runRemoteCsv] for the CSV-to-Parquet chain. private static int runRemote(String url, String outputTarget, Character delimiter) throws IOException { - if (url.endsWith(".parquet")) { - Path vortexPath = outputTarget != null - ? Path.of(outputTarget) - : Path.of(vortexName(lastPathSegment(url, "output.parquet"))); - if (isParquetTarget(vortexPath)) { + FileName source = new FileName(url); + if (source.is(FileFormat.PARQUET)) { + Path vortexPath = Path.of(outputTarget != null + ? outputTarget + : lastPathSegment(url, "output.parquet").withFormat(FileFormat.VORTEX)); + if (FileName.of(vortexPath).is(FileFormat.PARQUET)) { System.err.println("import always converts Parquet to Vortex; " + "a Parquet source cannot import to a .parquet output"); return ExitStatus.USAGE_ERROR; } return runRemoteParquet(url, vortexPath); } - if (url.endsWith(".csv")) { - Path outputPath = outputTarget != null - ? Path.of(outputTarget) - : Path.of(vortexName(lastPathSegment(url, "output.csv"))); + if (source.is(FileFormat.CSV)) { + Path outputPath = Path.of(outputTarget != null + ? outputTarget + : lastPathSegment(url, "output.csv").withFormat(FileFormat.VORTEX)); return runRemoteCsv(url, outputPath, delimiter); } System.err.println("only Parquet or CSV import is supported from a URL"); @@ -106,7 +110,7 @@ private static int runRemoteParquet(String parquetUrl, Path vortexPath) throws I /// progress/result print, so the result line reports only the output size. private static int runRemoteCsv(String csvUrl, Path outputPath, Character delimiter) throws IOException { ImportOptions options = csvOptions(delimiter); - if (isParquetTarget(outputPath)) { + if (FileName.of(outputPath).is(FileFormat.PARQUET)) { chainCsvToParquet(tempVortex -> CsvImporter.importCsv(URI.create(csvUrl), tempVortex, options), outputPath); } else { @@ -150,7 +154,7 @@ private static ParsedArgs parseArgs(String[] args) { /// always the hub, Parquet is never a direct CSV-import target. private static int runCsv(Path csvPath, Path outputPath, Character delimiter) throws IOException { ImportOptions options = csvOptions(delimiter); - if (isParquetTarget(outputPath)) { + if (FileName.of(outputPath).is(FileFormat.PARQUET)) { chainCsvToParquet(tempVortex -> CsvImporter.importCsv(csvPath, tempVortex, options), outputPath); ProgressBar.clear(); // cascading depth doesn't apply to a Parquet destination — suppressed via 0. @@ -187,7 +191,7 @@ private interface CsvToVortex { /// Imports CSV to a temp Vortex file via `importer`, exports that to `parquetOut`, then /// discards the temp file — the CSV-to-Parquet chain shared by [#runCsv] and [#runRemoteCsv]. private static void chainCsvToParquet(CsvToVortex importer, Path parquetOut) throws IOException { - Path tempVortex = Files.createTempFile("vortex-cli-import-", ".vortex"); + Path tempVortex = createTempVortex(); try { importer.importTo(tempVortex); ParquetExporter.exportParquet(tempVortex, parquetOut); @@ -196,8 +200,18 @@ private static void chainCsvToParquet(CsvToVortex importer, Path parquetOut) thr } } - private static boolean isParquetTarget(Path path) { - return path.getFileName().toString().endsWith(".parquet"); + /// Creates the CSV-to-Parquet chain's scratch file, owner-only readable/writable on POSIX + /// systems (`rw-------`) — the system temp directory is commonly world-writable, so a + /// predictable or loosely-permissioned temp name is a symlink/race target for another local + /// user. `FileAttribute`-based permissions aren't supported on Windows, whose per-user temp + /// directory doesn't share this exposure, so this falls back to the plain overload there. + private static Path createTempVortex() throws IOException { + String suffix = FileFormat.VORTEX.extension(); + if (FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { + FileAttribute ownerOnly = PosixFilePermissions.asFileAttribute(PosixFilePermissions.fromString("rw-------")); + return Files.createTempFile("vortex-cli-import-", suffix, ownerOnly); + } + return Files.createTempFile("vortex-cli-import-", suffix); } private static void printResult(Path inputPath, Path vortexPath, int cascadingDepth) throws IOException { @@ -232,24 +246,13 @@ private static void renderProgress(long done, long total) { } } - /// 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 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). Falls back to /// `fallback` when the path has no final segment (e.g. `https://host` with no path). - private static String lastPathSegment(String url, String fallback) { + private static FileName lastPathSegment(String url, String fallback) { String path = URI.create(url).getPath(); int slash = path.lastIndexOf('/'); String name = slash < 0 ? path : path.substring(slash + 1); - return name.isEmpty() ? fallback : name; + return new FileName(name.isEmpty() ? fallback : name); } } diff --git a/cli/src/test/java/io/github/dfa1/vortex/cli/FileNameTest.java b/cli/src/test/java/io/github/dfa1/vortex/cli/FileNameTest.java new file mode 100644 index 00000000..6c06b33b --- /dev/null +++ b/cli/src/test/java/io/github/dfa1/vortex/cli/FileNameTest.java @@ -0,0 +1,93 @@ +package io.github.dfa1.vortex.cli; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.nio.file.Path; + +import static org.assertj.core.api.Assertions.assertThat; + +class FileNameTest { + + @ParameterizedTest + @CsvSource({ + "data.csv, CSV, true", + "data.csv, PARQUET, false", + "data.parquet, PARQUET, true", + "data.vortex, VORTEX, true", + }) + void is_matchesKnownExtension(String name, FileFormat format, boolean expected) { + // Given + FileName fileName = new FileName(name); + + // When + boolean result = fileName.is(format); + + // Then + assertThat(result).isEqualTo(expected); + } + + @ParameterizedTest + @CsvSource({ + "data.csv, VORTEX, data.vortex", + "data.parquet, VORTEX, data.vortex", + "data.vortex, CSV, data.csv", + }) + void withFormat_swapsDifferentKnownExtension(String name, FileFormat target, String expected) { + // Given + FileName fileName = new FileName(name); + + // When + String result = fileName.withFormat(target); + + // Then — the stem is kept, only the trailing known extension changes + assertThat(result).isEqualTo(expected); + } + + @Test + void withFormat_noKnownExtension_appendsTarget() { + // Given — a name that matches none of CSV/PARQUET/VORTEX (e.g. a .tsv file) + FileName name = new FileName("data.tsv"); + + // When + String result = name.withFormat(FileFormat.VORTEX); + + // Then — nothing is stripped, the target extension is just appended + assertThat(result).isEqualTo("data.tsv.vortex"); + } + + @ParameterizedTest + @CsvSource({ + "data.csv, CSV", + "data.parquet, PARQUET", + "data.vortex, VORTEX", + }) + void withFormat_alreadyTargetFormat_appendsRatherThanReturningSameName(String name, FileFormat target) { + // Given — a name already in the target format. `ImportCommand`/`ExportCommand` derive a + // default output name this way, e.g. `withFormat(VORTEX)` on an import source that + // happens to already be named "data.vortex" — the caller must never get its own input + // name back, or it would open that path for writing while still reading it as the + // source (regression: an earlier version of #withFormat stripped a known extension + // whenever the name had one, even when it equaled the target, so "data.vortex" mapped + // straight back to itself instead of "data.vortex.vortex"). + FileName fileName = new FileName(name); + + // When + String result = fileName.withFormat(target); + + // Then + assertThat(result).isNotEqualTo(name); + assertThat(result).isEqualTo(name + target.extension()); + } + + @Test + void of_takesPathsFinalComponent() { + // Given / When + FileName name = FileName.of(Path.of("a", "b", "data.parquet")); + + // Then + assertThat(name.value()).isEqualTo("data.parquet"); + assertThat(name.is(FileFormat.PARQUET)).isTrue(); + } +} diff --git a/csv/src/main/java/io/github/dfa1/vortex/csv/CsvImporter.java b/csv/src/main/java/io/github/dfa1/vortex/csv/CsvImporter.java index 262c6111..b30fddd6 100644 --- a/csv/src/main/java/io/github/dfa1/vortex/csv/CsvImporter.java +++ b/csv/src/main/java/io/github/dfa1/vortex/csv/CsvImporter.java @@ -125,14 +125,37 @@ public static void importCsv(URI csvUri, Path vortexPath, ImportOptions options) } } + /// The buffered first chunk of data rows (used for schema inference) plus the resolved + /// header names and schema, produced by [#readFirstChunk]. + private record FirstChunk(List rows, DType.Struct schema) { + } + private static void importCsv(CsvReader reader, Path vortexPath, ImportOptions options) throws IOException { - int chunkSize = options.chunkSize(); + FirstChunk firstChunk = readFirstChunk(reader, options); + DType.Struct schema = firstChunk.schema(); + + try (FileChannel channel = FileChannel.open( + vortexPath, StandardOpenOption.CREATE, StandardOpenOption.WRITE, + StandardOpenOption.TRUNCATE_EXISTING); + VortexWriter writer = VortexWriter.create(channel, schema, options.writeOptions())) { + // firstChunk isn't touched again after this call, so its buffered rows (up to + // chunkSize row arrays) become collectible before the streaming loop below runs, + // rather than staying reachable for the rest of a potentially large import. + long totalRows = writeFirstChunk(writer, firstChunk, options); + streamRemainingChunks(reader, writer, schema, options, totalRows); + } + } - // Read header row (if present) and buffer the first chunk of data rows. - // Both happen in a single pass so the reader position advances correctly. + /// Reads the header row (if present) and buffers the first [ImportOptions#chunkSize()] data + /// rows in the same pass, so the reader position advances correctly; infers the schema from + /// those buffered rows unless [ImportOptions#schema()] already gives one. + /// + /// @throws IllegalArgumentException if the CSV file has no data rows + private static FirstChunk readFirstChunk(CsvReader reader, ImportOptions options) { + int chunkSize = options.chunkSize(); String[] headers = null; - List firstChunk = new ArrayList<>(chunkSize); + List rows = new ArrayList<>(chunkSize); boolean expectHeader = options.hasHeader(); for (CsvRecord csvRecord : reader) { @@ -140,62 +163,65 @@ private static void importCsv(CsvReader reader, Path vortexPath, Impo headers = csvRecord.getFields().toArray(String[]::new); expectHeader = false; } else { - firstChunk.add(csvRecord.getFields().toArray(String[]::new)); - if (firstChunk.size() == chunkSize) { + rows.add(csvRecord.getFields().toArray(String[]::new)); + if (rows.size() == chunkSize) { break; } } } - if (firstChunk.isEmpty()) { + if (rows.isEmpty()) { throw new IllegalArgumentException("CSV file has no data rows"); } // Generate synthetic column names when the file has no header row. if (headers == null) { - headers = generateHeaders(firstChunk.getFirst().length); + headers = generateHeaders(rows.getFirst().length); } DType.Struct schema = options.schema() != null ? options.schema() - : inferSchemaFromRows(headers, firstChunk); - - try (FileChannel channel = FileChannel.open( - vortexPath, StandardOpenOption.CREATE, StandardOpenOption.WRITE, - StandardOpenOption.TRUNCATE_EXISTING); - VortexWriter writer = VortexWriter.create(channel, schema, options.writeOptions())) { - - long totalRows = 0; - long lastReported = 0; + : inferSchemaFromRows(headers, rows); + return new FirstChunk(rows, schema); + } - // Write the buffered first chunk. - writer.writeChunk(buildChunk(schema, firstChunk)); - totalRows += firstChunk.size(); - lastReported = totalRows; - reportProgress(options, totalRows); - firstChunk.clear(); + /// Writes the buffered first chunk and reports progress once. + /// + /// @return the row count written so far (the first chunk's size) + private static long writeFirstChunk(VortexWriter writer, FirstChunk firstChunk, ImportOptions options) + throws IOException { + writer.writeChunk(buildChunk(firstChunk.schema(), firstChunk.rows())); + long totalRows = firstChunk.rows().size(); + reportProgress(options, totalRows); + return totalRows; + } - // Stream the rest of the file through the still-open reader. - List chunk = new ArrayList<>(chunkSize); - for (CsvRecord csvRecord : reader) { - chunk.add(csvRecord.getFields().toArray(String[]::new)); - totalRows++; - if (chunk.size() == chunkSize) { - writer.writeChunk(buildChunk(schema, chunk)); - chunk.clear(); - } - if (totalRows - lastReported >= PROGRESS_BATCH) { - reportProgress(options, totalRows); - lastReported = totalRows; - } - } - if (!chunk.isEmpty()) { + /// Streams the rest of the file through the still-open `reader`, writing a chunk every + /// [ImportOptions#chunkSize()] rows and reporting progress every [#PROGRESS_BATCH] rows. + private static void streamRemainingChunks(CsvReader reader, VortexWriter writer, + DType.Struct schema, ImportOptions options, long rowsSoFar) throws IOException { + int chunkSize = options.chunkSize(); + long totalRows = rowsSoFar; + long lastReported = totalRows; + List chunk = new ArrayList<>(chunkSize); + for (CsvRecord csvRecord : reader) { + chunk.add(csvRecord.getFields().toArray(String[]::new)); + totalRows++; + if (chunk.size() == chunkSize) { writer.writeChunk(buildChunk(schema, chunk)); + chunk.clear(); } - if (totalRows > lastReported) { + if (totalRows - lastReported >= PROGRESS_BATCH) { reportProgress(options, totalRows); + lastReported = totalRows; } } + if (!chunk.isEmpty()) { + writer.writeChunk(buildChunk(schema, chunk)); + } + if (totalRows > lastReported) { + reportProgress(options, totalRows); + } } private static DType.Struct inferSchemaFromRows(String[] headers, List rows) { 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 index 39728932..d293d695 100644 --- a/parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetExporter.java +++ b/parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetExporter.java @@ -153,10 +153,10 @@ static List resolveTypes(List allNames, List allTypes, 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 -> + case DType.Bool _ -> builder.addColumn(name, PhysicalType.BOOLEAN, rep); + case DType.Utf8 _ -> builder.addColumn(name, PhysicalType.BYTE_ARRAY, rep, new LogicalType.StringType()); - case DType.Binary ignored -> builder.addColumn(name, PhysicalType.BYTE_ARRAY, rep); + case DType.Binary _ -> 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( @@ -217,11 +217,11 @@ static void writeColumn(ColumnBatch batch, int idx, DType type, Array array, int 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.Bool _ -> writeBooleans(batch, idx, target, rowCount, nulls); + case DType.Utf8 _ -> writeBytes(batch, idx, target, rowCount, nulls); + case DType.Binary _ -> 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); + case DType.Extension _ -> writeLongs(batch, idx, target, rowCount, nulls); default -> throw new UnsupportedOperationException("unsupported column type for Parquet export: " + type); } } 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 index 5c7670f8..efbf3a4c 100644 --- a/parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetExporterTest.java +++ b/parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetExporterTest.java @@ -132,9 +132,11 @@ void f32_mapsToFloat_f64_mapsToDouble() { @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); + // Given — mirrors ParquetImporter, which never produces F16 either + DType f16 = new DType.Primitive(PType.F16, false); + + // When / Then + assertThatThrownBy(() -> schemaOf("f", f16)).isInstanceOf(UnsupportedOperationException.class); } @Test @@ -290,7 +292,7 @@ void exportsTimestampColumn_asInt64Timestamp(@TempDir Path tmp) throws Exception rows.next(); assertThat(rows.getLong("events")).isEqualTo(-1_500L); rows.next(); - assertThat(rows.getLong("events")).isEqualTo(0L); + assertThat(rows.getLong("events")).isZero(); rows.next(); assertThat(rows.getLong("events")).isEqualTo(1_733_000_000_000L); } @@ -399,7 +401,7 @@ void flatTypes_roundTripThroughParquetAndBack(@TempDir Path tmp) throws Exceptio assertThat(idValues.getLong(2)).isEqualTo(-3L); ByteArray age = chunk.column("age"); - assertThat(age.getInt(0)).isEqualTo(0); + assertThat(age.getInt(0)).isZero(); assertThat(age.getInt(1)).isEqualTo(255); assertThat(age.getInt(2)).isEqualTo(42); @@ -427,7 +429,7 @@ void flatTypes_roundTripThroughParquetAndBack(@TempDir Path tmp) throws Exceptio // 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(0)).isZero(); assertThat(bigCount.getInt(1)).isEqualTo(-1); assertThat(bigCount.getInt(2)).isEqualTo(12345); } diff --git a/writer/src/test/java/io/github/dfa1/vortex/writer/encode/ZstdEncodingEncoderTest.java b/writer/src/test/java/io/github/dfa1/vortex/writer/encode/ZstdEncodingEncoderTest.java index bba3b727..2f812c76 100644 --- a/writer/src/test/java/io/github/dfa1/vortex/writer/encode/ZstdEncodingEncoderTest.java +++ b/writer/src/test/java/io/github/dfa1/vortex/writer/encode/ZstdEncodingEncoderTest.java @@ -147,9 +147,10 @@ void encode_nonNullableBinaryWithNull_throwsVortexException() { // reject it rather than silently emit a nullable layout the dtype does not declare. byte[][] data = {{0x01}, null, {0x02}}; DType binary = new DType.Binary(false); + EncodeContext ctx = EncodeTestHelper.testCtx(); // When / Then - assertThatThrownBy(() -> ENCODER.encode(binary, data, EncodeTestHelper.testCtx())) + assertThatThrownBy(() -> ENCODER.encode(binary, data, ctx)) .isInstanceOf(VortexException.class); }