diff --git a/CHANGELOG.md b/CHANGELOG.md index 2788e4cd0..0552554cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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)) +- `CsvImporter.importCsv(URI, Path[, ImportOptions])`: imports a CSV file served over HTTP(S), streaming the response body directly (CSV is read front to back in one pass, so unlike Parquet's random-access format this needs no Range requests). The CLI's `import` subcommand now accepts a `.csv` URL too, and — for both local and remote CSV — a `.parquet` output path chains CSV → temp Vortex → Parquet internally (Vortex stays the hub; Parquet is never a direct CSV-import target), deleting the temp file afterward. ([#363](https://github.com/dfa1/vortex-java/pull/363)) - `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 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 07d1131c9..fe8cc3079 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 @@ -3,6 +3,7 @@ import io.github.dfa1.vortex.csv.CsvImporter; import io.github.dfa1.vortex.inspect.ByteSize; import io.github.dfa1.vortex.csv.ImportOptions; +import io.github.dfa1.vortex.parquet.ParquetExporter; import io.github.dfa1.vortex.parquet.ParquetImporter; import java.io.IOException; @@ -30,14 +31,15 @@ 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|out.parquet]"); return ExitStatus.USAGE_ERROR; } String target = parsedArgs.inputTarget(); boolean remote = target.startsWith("http://") || target.startsWith("https://"); try { if (remote) { - return runRemote(target, parsedArgs.outputTarget()); + return runRemote(target, parsedArgs.outputTarget(), parsedArgs.delimiter()); } Path inputPath = Path.of(target); if (!Files.exists(inputPath)) { @@ -49,6 +51,11 @@ static int run(String[] args) { ? Path.of(parsedArgs.outputTarget()) : inputPath.resolveSibling(vortexName(name)); if (name.endsWith(".parquet")) { + if (isParquetTarget(outputPath)) { + System.err.println("import always converts Parquet to Vortex; " + + "a Parquet source cannot import to a .parquet output"); + return ExitStatus.USAGE_ERROR; + } return runParquet(inputPath, outputPath); } else { return runCsv(inputPath, outputPath, parsedArgs.delimiter()); @@ -60,23 +67,53 @@ 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; + /// Handles an `http(s)://` source, dispatching by the *source* extension (Parquet or CSV; + /// 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)) { + 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"))); + return runRemoteCsv(url, outputPath, delimiter); } - Path vortexPath = outputTarget != null - ? Path.of(outputTarget) - : Path.of(vortexName(lastPathSegment(parquetUrl))); + System.err.println("only Parquet or CSV import is supported from a URL"); + return ExitStatus.USAGE_ERROR; + } + + private static int runRemoteParquet(String parquetUrl, Path vortexPath) throws IOException { 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))); + printSimpleResult(vortexPath); + return ExitStatus.OK; + } + + /// Imports a remote CSV, same as [#runCsv] but with no local input file to size for the + /// 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)) { + chainCsvToParquet(tempVortex -> CsvImporter.importCsv(URI.create(csvUrl), tempVortex, options), + outputPath); + } else { + CsvImporter.importCsv(URI.create(csvUrl), outputPath, options); + } + ProgressBar.clear(); + printSimpleResult(outputPath); return ExitStatus.OK; } @@ -108,15 +145,21 @@ private static ParsedArgs parseArgs(String[] args) { return new ParsedArgs(positional.getFirst(), outputTarget, delimiter); } - private static int runCsv(Path csvPath, Path vortexPath, Character delimiter) throws IOException { - ImportOptions options = ImportOptions.defaults() - .withProgressListener(ImportCommand::renderProgress); - if (delimiter != null) { - options = options.withDelimiter(delimiter); + /// Imports a local CSV file. When `outputPath` ends `.parquet`, the CSV is imported to a + /// temp Vortex file first, then exported to Parquet and the temp file discarded — Vortex is + /// 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)) { + chainCsvToParquet(tempVortex -> CsvImporter.importCsv(csvPath, tempVortex, options), outputPath); + ProgressBar.clear(); + // cascading depth doesn't apply to a Parquet destination — suppressed via 0. + printResult(csvPath, outputPath, 0); + return ExitStatus.OK; } - CsvImporter.importCsv(csvPath, vortexPath, options); + CsvImporter.importCsv(csvPath, outputPath, options); ProgressBar.clear(); - printResult(csvPath, vortexPath, options.writeOptions().allowedCascading()); + printResult(csvPath, outputPath, options.writeOptions().allowedCascading()); return ExitStatus.OK; } @@ -130,6 +173,33 @@ private static int runParquet(Path parquetPath, Path vortexPath) throws IOExcept return ExitStatus.OK; } + private static ImportOptions csvOptions(Character delimiter) { + ImportOptions options = ImportOptions.defaults() + .withProgressListener(ImportCommand::renderProgress); + return delimiter != null ? options.withDelimiter(delimiter) : options; + } + + @FunctionalInterface + private interface CsvToVortex { + void importTo(Path tempVortex) throws IOException; + } + + /// 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"); + try { + importer.importTo(tempVortex); + ParquetExporter.exportParquet(tempVortex, parquetOut); + } finally { + Files.deleteIfExists(tempVortex); + } + } + + private static boolean isParquetTarget(Path path) { + return path.getFileName().toString().endsWith(".parquet"); + } + private static void printResult(Path inputPath, Path vortexPath, int cascadingDepth) throws IOException { long inputBytes = Files.size(inputPath); long vortexBytes = Files.size(vortexPath); @@ -145,6 +215,12 @@ private static void printResult(Path inputPath, Path vortexPath, int cascadingDe sizeChange, cascadingInfo); } + /// Result line for a remote source, which has no cheaply-known local input size to compare + /// against — just the output path and its size. + private static void printSimpleResult(Path outputPath) throws IOException { + System.out.printf("written: %s (%s)%n", outputPath, ByteSize.format(Files.size(outputPath))); + } + /// Progress callback for imports. An indeterminate `total` (`< 0`, e.g. a streamed source with /// no known row count) shows just the running row count; otherwise delegates to the shared bar. private static void renderProgress(long done, long total) { @@ -168,11 +244,12 @@ private static String vortexName(String inputFileName) { } /// 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) { + /// 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) { 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; + return name.isEmpty() ? fallback : 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 631e8e104..ef95b4843 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 @@ -48,7 +48,8 @@ static void printUsage(PrintStream out) { 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|out.parquet|-] write CSV or Parquet; default is .csv, `-` for stdout"); - out.println(" import [out.vortex] convert CSV or Parquet to Vortex"); + out.println(" import [--delimiter ] [out.vortex|out.parquet]"); + out.println(" convert CSV or Parquet (local or url) to Vortex or Parquet"); out.println(" schema print dtype (machine-readable)"); out.println(" count print row count"); out.println(" select [...] project columns to CSV on stdout"); 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 63a700510..7084767c3 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 @@ -63,15 +63,15 @@ void tooManyPositional_returnsUsageError() { } @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 + void urlInput_unsupportedExtension_returnsUsageError() { + // Given / When — only Parquet and CSV sources are supported from a URL; 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"})); + ImportCommand.run(new String[]{"import", "http://example.com/data.json"})); // Then assertThat(result.status()).isEqualTo(ExitStatus.USAGE_ERROR); - assertThat(result.stderr()).contains("only Parquet import is supported from a URL"); + assertThat(result.stderr()).contains("only Parquet or CSV import is supported from a URL"); } } @@ -143,5 +143,46 @@ void csvWithCustomDelimiter_imports(@TempDir Path tmp) throws IOException { assertThat(result.status()).isEqualTo(ExitStatus.OK); assertThat(tmp.resolve("data.tsv.vortex")).exists(); } + + @Test + void csvWithParquetOutputPath_chainsThroughTempVortex(@TempDir Path tmp) throws IOException { + // Given — a `.parquet` destination; Vortex is always the hub, so this chains + // CSV -> temp Vortex -> Parquet internally and discards the temp file + Path csv = tmp.resolve("in.csv"); + Files.writeString(csv, "id,name\n1,Ada\n2,Grace\n", StandardCharsets.UTF_8); + Path out = tmp.resolve("out.parquet"); + + // When + CliTestSupport.Captured result = capture(() -> + ImportCommand.run(new String[]{"import", csv.toString(), out.toString()})); + + // Then — the real Parquet output exists; the temp Vortex file (system temp dir, not + // this directory) leaves nothing behind in the working directory + assertThat(result.status()).isEqualTo(ExitStatus.OK); + assertThat(out).exists(); + assertThat(result.stdout()).contains(out.toString()); + try (var files = Files.list(tmp)) { + assertThat(files.map(p -> p.getFileName().toString())).containsExactlyInAnyOrder("in.csv", "out.parquet"); + } + } + + @Test + void parquetInputWithParquetOutputPath_returnsUsageError(@TempDir Path tmp) throws IOException { + // Given — a Parquet source always produces Vortex; a `.parquet`-named output would + // silently write a Vortex-format file under a misleading name if left unchecked + Path vortex = CliTestSupport.writeSmallVortex(tmp, "src.vortex"); + Path parquetIn = tmp.resolve("src.parquet"); + io.github.dfa1.vortex.parquet.ParquetExporter.exportParquet(vortex, parquetIn); + Path parquetOut = tmp.resolve("out.parquet"); + + // When + CliTestSupport.Captured result = capture(() -> + ImportCommand.run(new String[]{"import", parquetIn.toString(), parquetOut.toString()})); + + // Then + assertThat(result.status()).isEqualTo(ExitStatus.USAGE_ERROR); + assertThat(result.stderr()).contains("Parquet source cannot import to a .parquet output"); + assertThat(parquetOut).doesNotExist(); + } } } 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 c86454401..262c6111b 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 @@ -8,6 +8,11 @@ import io.github.dfa1.vortex.writer.VortexWriter; import java.io.IOException; +import java.io.InputStream; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; import java.nio.channels.FileChannel; import java.nio.file.Path; import java.nio.file.StandardOpenOption; @@ -31,6 +36,14 @@ public final class CsvImporter { private static final long PROGRESS_BATCH = 10_000; + /// 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. + /// + /// Package-private and non-final purely as a unit-test seam: tests substitute a mocked + /// client to drive [#importCsv(URI, Path, ImportOptions)] without real network I/O. + /// Production code never reassigns it. + static HttpClient httpClient = HttpClient.newHttpClient(); + private CsvImporter() { } @@ -56,76 +69,132 @@ public static void importCsv(Path csvPath, Path vortexPath) throws IOException { /// @throws IOException if reading or writing fails /// @throws IllegalArgumentException if the CSV file has no data rows public static void importCsv(Path csvPath, Path vortexPath, ImportOptions options) throws IOException { + try (CsvReader reader = csvReader(csvPath, options)) { + importCsv(reader, vortexPath, options); + } + } + + /// Imports a CSV file served over HTTP(S) to a Vortex file using default options. + /// + /// @param csvUri the `http(s)://` URL of the source CSV file + /// @param vortexPath path to write the output Vortex file + /// @throws IOException if reading or writing fails + public static void importCsv(URI csvUri, Path vortexPath) throws IOException { + importCsv(csvUri, vortexPath, ImportOptions.defaults()); + } + + /// Imports a CSV file served over HTTP(S) to a Vortex file. + /// + /// Unlike Parquet's random-access, footer-first format, CSV is read front to back in one + /// streaming pass, so the response body is consumed directly as it arrives — no Range + /// requests, no local temp file, and no full-file buffering. + /// + /// @param csvUri the `http(s)://` URL of the source CSV file + /// @param vortexPath path to write the output Vortex file + /// @param options import configuration + /// @throws IOException if fetching `csvUri` or writing `vortexPath` fails + /// @throws IllegalArgumentException if the CSV file has no data rows + public static void importCsv(URI csvUri, Path vortexPath, ImportOptions options) throws IOException { + HttpRequest request = HttpRequest.newBuilder(csvUri).GET().build(); + HttpResponse response; + try { + response = httpClient.send(request, HttpResponse.BodyHandlers.ofInputStream()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("interrupted fetching " + csvUri, e); + } + if (response.statusCode() != 200) { + IOException failure = new IOException("HTTP " + response.statusCode() + " fetching " + csvUri); + // Close the still-open body stream before throwing: with the streaming + // ofInputStream() handler, the JDK HttpClient only releases the underlying + // connection once the body is consumed or closed. A failure closing it is + // secondary to the HTTP status that's already failing the call, so it's + // attached as suppressed rather than replacing the real error. + try { + response.body().close(); + } catch (IOException closeFailure) { + failure.addSuppressed(closeFailure); + } + throw failure; + } + try (InputStream body = response.body(); + CsvReader reader = CsvReader.builder() + .fieldSeparator(options.delimiter()) + .ofCsvRecord(body)) { + importCsv(reader, vortexPath, options); + } + } + + private static void importCsv(CsvReader reader, Path vortexPath, ImportOptions options) + throws IOException { int chunkSize = options.chunkSize(); - try (CsvReader reader = csvReader(csvPath, options)) { - // 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. - String[] headers = null; - List firstChunk = new ArrayList<>(chunkSize); - boolean expectHeader = options.hasHeader(); + // 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. + String[] headers = null; + List firstChunk = new ArrayList<>(chunkSize); + boolean expectHeader = options.hasHeader(); - for (CsvRecord csvRecord : reader) { - if (expectHeader) { - headers = csvRecord.getFields().toArray(String[]::new); - expectHeader = false; - } else { - firstChunk.add(csvRecord.getFields().toArray(String[]::new)); - if (firstChunk.size() == chunkSize) { - break; - } + for (CsvRecord csvRecord : reader) { + if (expectHeader) { + headers = csvRecord.getFields().toArray(String[]::new); + expectHeader = false; + } else { + firstChunk.add(csvRecord.getFields().toArray(String[]::new)); + if (firstChunk.size() == chunkSize) { + break; } } + } - if (firstChunk.isEmpty()) { - throw new IllegalArgumentException("CSV file has no data rows"); - } + if (firstChunk.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); - } + // Generate synthetic column names when the file has no header row. + if (headers == null) { + headers = generateHeaders(firstChunk.getFirst().length); + } - DType.Struct schema = options.schema() != null - ? options.schema() - : inferSchemaFromRows(headers, firstChunk); + 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())) { + 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; + long totalRows = 0; + long lastReported = 0; - // Write the buffered first chunk. - writer.writeChunk(buildChunk(schema, firstChunk)); - totalRows += firstChunk.size(); - lastReported = totalRows; - reportProgress(options, totalRows); - firstChunk.clear(); + // Write the buffered first chunk. + writer.writeChunk(buildChunk(schema, firstChunk)); + totalRows += firstChunk.size(); + lastReported = totalRows; + reportProgress(options, totalRows); + firstChunk.clear(); - // 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()) { + // 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) { + if (totalRows - lastReported >= PROGRESS_BATCH) { reportProgress(options, totalRows); + lastReported = totalRows; } } + if (!chunk.isEmpty()) { + writer.writeChunk(buildChunk(schema, chunk)); + } + if (totalRows > lastReported) { + reportProgress(options, totalRows); + } } } diff --git a/csv/src/test/java/io/github/dfa1/vortex/csv/CsvImporterHttpTest.java b/csv/src/test/java/io/github/dfa1/vortex/csv/CsvImporterHttpTest.java new file mode 100644 index 000000000..b99f3c817 --- /dev/null +++ b/csv/src/test/java/io/github/dfa1/vortex/csv/CsvImporterHttpTest.java @@ -0,0 +1,148 @@ +package io.github.dfa1.vortex.csv; + +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.reader.Chunk; +import io.github.dfa1.vortex.reader.ScanIterator; +import io.github.dfa1.vortex.reader.ScanOptions; +import io.github.dfa1.vortex.reader.VortexReader; +import io.github.dfa1.vortex.reader.array.LongArray; +import io.github.dfa1.vortex.reader.array.VarBinArray; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +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.charset.StandardCharsets; +import java.nio.file.Path; +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; + +/// Covers [CsvImporter#importCsv(URI, Path, ImportOptions)] against a mocked [HttpClient], the +/// same deterministic, network-free approach `VortexHttpReaderOpenOverloadTest` uses on the +/// reader side of the wire format. CSV is read front to back in one streaming pass, so — unlike +/// the `parquet` module's Range-request-based `HttpInputFile` — a single plain GET response is +/// all that's needed here. +@ExtendWith(MockitoExtension.class) +class CsvImporterHttpTest { + + @Mock + private HttpClient client; + + private static final URI URI = java.net.URI.create("http://example.com/data.csv"); + + @Test + void importsFromUrl_streamsBodyIntoVortex(@TempDir Path tmp) throws Exception { + // Given + String csv = "id,price,name\n1,1.5,Alice\n2,2.7,Bob\n"; + doReturn(response(200, csv)).when(client).send(any(), any()); + HttpClient original = CsvImporter.httpClient; + CsvImporter.httpClient = client; + Path vortex = tmp.resolve("data.vortex"); + + try { + // When + CsvImporter.importCsv(URI, vortex); + + // Then + try (VortexReader reader = VortexReader.open(vortex)) { + DType.Struct schema = (DType.Struct) reader.dtype(); + assertThat(schema.fieldNames().stream().map(ColumnName::value).toList()) + .containsExactly("id", "price", "name"); + try (ScanIterator iter = reader.scan(ScanOptions.all())) { + assertThat(iter.hasNext()).isTrue(); + try (Chunk chunk = iter.next()) { + assertThat(chunk.rowCount()).isEqualTo(2); + LongArray ids = chunk.column("id"); + assertThat(ids.getLong(0)).isEqualTo(1L); + VarBinArray names = chunk.column("name"); + assertThat(names.getString(0)).isEqualTo("Alice"); + assertThat(names.getString(1)).isEqualTo("Bob"); + } + } + } + } finally { + CsvImporter.httpClient = original; + } + } + + @Test + void nonOkStatus_throws(@TempDir Path tmp) throws Exception { + // Given + doReturn(response(404, "")).when(client).send(any(), any()); + HttpClient original = CsvImporter.httpClient; + CsvImporter.httpClient = client; + Path vortex = tmp.resolve("data.vortex"); + + try { + // When / Then + assertThatThrownBy(() -> CsvImporter.importCsv(URI, vortex)) + .isInstanceOf(IOException.class) + .hasMessageContaining("404"); + } finally { + CsvImporter.httpClient = original; + } + } + + // ── helpers ─────────────────────────────────────────────────────────────── + + @SuppressWarnings("unchecked") + private static HttpResponse response(int status, String body) { + return new HttpResponse<>() { + @Override + public int statusCode() { + return status; + } + + @Override + public InputStream body() { + return new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)); + } + + @Override + public HttpHeaders headers() { + return HttpHeaders.of(Map.of(), (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/docs/how-to.md b/docs/how-to.md index aacfd293f..174d384ba 100644 --- a/docs/how-to.md +++ b/docs/how-to.md @@ -327,11 +327,33 @@ java -jar cli/target/vortex-cli-*-all.jar export https://example.com/data.vortex ## Convert CSV to Vortex -**CLI only** (CSV has no schema — types are inferred): +**API:** + +```java +import io.github.dfa1.vortex.csv.CsvImporter; + +CsvImporter.importCsv(Path.of("data.csv"), Path.of("data.vortex")); +``` + +From a remote CSV file over HTTP(S). CSV is read front to back in one streaming pass, so the +response body is consumed directly — no Range requests, no local temp file: + +```java +CsvImporter.importCsv(URI.create("https://example.com/data.csv"), Path.of("data.vortex")); +``` + +**CLI** (types are inferred from the data): ```bash java -jar cli/target/vortex-cli-*-all.jar import data.csv # writes data.vortex, prints size savings + +# remote source +java -jar cli/target/vortex-cli-*-all.jar import https://example.com/data.csv out.vortex + +# straight to Parquet — chains CSV -> temp Vortex -> Parquet internally, local or remote source +java -jar cli/target/vortex-cli-*-all.jar import data.csv out.parquet +java -jar cli/target/vortex-cli-*-all.jar import https://example.com/data.csv out.parquet ``` --- diff --git a/docs/reference.md b/docs/reference.md index fae9a8aaa..7b1c40230 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -441,7 +441,24 @@ encoding policy. | `.withProgressListener(listener)` | Progress callbacks | | `.withWriterConfig(WriterConfig)` | Override Hardwood writer configuration | -CSV import is CLI-only — types are inferred from the data. +### `CsvImporter` (`io.github.dfa1.vortex.csv.CsvImporter`) + +Column types are inferred from the data (long → double → boolean → utf8, in priority order) +unless a schema is given via `ImportOptions#withSchema`. Schema-driven CLI subcommands +(`schema`, `count`, …) are documented for Vortex/Parquet files only — CSV has no schema of its +own, so it is only ever an `import` *source*: never a destination (there is no CSV export from +`import`), and never a source for the other subcommands, which all expect a Vortex or Parquet +file. + +| Method | Notes | +|-----------------------------------------------|------------------------------------------------------------| +| `importCsv(Path in, Path out)` | Defaults | +| `importCsv(Path in, Path out, ImportOptions)`| Tuned | +| `importCsv(URI in, Path out)` | Remote source over HTTP(S), defaults | +| `importCsv(URI in, Path out, ImportOptions)` | Remote source over HTTP(S), tuned | + +A `URI` source streams the response body directly, front to back, as it arrives — unlike +Parquet's random-access footer-first format, CSV needs no Range requests and no local temp file. --- @@ -468,7 +485,7 @@ java -jar cli/target/vortex-cli-*-all.jar [args] | `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; a `url` source must be `.parquet` — CSV import from a URL isn't supported | +| `import` | `import [--delimiter ] [out.vortex\|out.parquet]` | CSV or Parquet (local or remote) source to Vortex; a `.parquet` output is CSV-only (chains through a temp Vortex file internally) — a Parquet source always produces Vortex, `.parquet` output is rejected | ### `filter` expression syntax