From 5631be30340c8831e0755e6627a58d83cdac6b21 Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Fri, 13 Nov 2026 21:28:29 +0100 Subject: [PATCH] Add vortex rewrite: re-encode a file with the current writer `vortex rewrite [options] ` scans every chunk of the source and writes it again through VortexWriter, so the output is what today's writer picks, not what the source's writer picked. - Options: --compact, --no-zone-maps, --no-global-dict, --edition (also --edition=). An edition too old for the data fails with the writer guard's message. - First version: a flat struct of primitive, bool and utf8 columns, nullable or not. Any other column (nested, decimal, binary, extension, f16) is rejected before the output exists. Nullable columns reach the writer as a primitive array plus validity (NullableData), utf8 as raw bytes. - The output must not exist, so a rewrite cannot overwrite its source; any failure removes the partial output. - Over HTTP each chunk prefetches all its columns together: the 105 MB OpenPowerlifting sample rewrites in 16 s from a URL instead of 51 s. - docs/how-to.md has a Re-encode a Vortex file section; vortex-cli's version is managed in the root POM so the integration module can use it. Cross-checked against vortex-jni: a jni-written file is rewritten, then jni reads both back whole and under filters that prune zones. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 1 + .../dfa1/vortex/cli/RewriteCommand.java | 106 +++++++ .../io/github/dfa1/vortex/cli/Rewriter.java | 190 ++++++++++++ .../io/github/dfa1/vortex/cli/VortexCli.java | 2 + .../dfa1/vortex/cli/RewriteCommandTest.java | 288 ++++++++++++++++++ docs/how-to.md | 34 +++ docs/reference.md | 1 + integration/pom.xml | 5 + .../cli/RewriteInteropIntegrationTest.java | 161 ++++++++++ pom.xml | 5 + 10 files changed, 793 insertions(+) create mode 100644 cli/src/main/java/io/github/dfa1/vortex/cli/RewriteCommand.java create mode 100644 cli/src/main/java/io/github/dfa1/vortex/cli/Rewriter.java create mode 100644 cli/src/test/java/io/github/dfa1/vortex/cli/RewriteCommandTest.java create mode 100644 integration/src/test/java/io/github/dfa1/vortex/cli/RewriteInteropIntegrationTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 959934503..6c57d5077 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- `vortex rewrite` re-encodes a file with the current writer. ([#749](https://github.com/dfa1/vortex-java/pull/749)) - `Chunk.prefetch` and `VortexHandle.prefetch` start the range requests of several columns together over HTTP, merged into few requests, and an HTTP handle keeps the segments it has read. ([#744](https://github.com/dfa1/vortex-java/issues/744)) ### Fixed diff --git a/cli/src/main/java/io/github/dfa1/vortex/cli/RewriteCommand.java b/cli/src/main/java/io/github/dfa1/vortex/cli/RewriteCommand.java new file mode 100644 index 000000000..0c1287455 --- /dev/null +++ b/cli/src/main/java/io/github/dfa1/vortex/cli/RewriteCommand.java @@ -0,0 +1,106 @@ +package io.github.dfa1.vortex.cli; + +import io.github.dfa1.vortex.core.error.VortexException; +import io.github.dfa1.vortex.core.model.Edition; +import io.github.dfa1.vortex.core.model.Editions; +import io.github.dfa1.vortex.inspect.ByteSize; +import io.github.dfa1.vortex.reader.VortexHandle; +import io.github.dfa1.vortex.writer.WriteOptions; + +import java.io.IOException; +import java.nio.file.FileAlreadyExistsException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ForkJoinPool; +import java.util.stream.Collectors; + +@SuppressWarnings("java:S106") // CLI command: stdout is the intended output channel +final class RewriteCommand { + + private static final String USAGE = "usage: rewrite [--compact] [--no-zone-maps] [--no-global-dict]" + + " [--edition ] "; + + private RewriteCommand() { + } + + static int run(String[] args) { + WriteOptions options = WriteOptions.defaults().withExecutor(ForkJoinPool.commonPool()); + List positional = new ArrayList<>(); + for (int i = 1; i < args.length; i++) { + switch (args[i]) { + case "--compact" -> options = options.withCompact(true); + case "--no-zone-maps" -> options = options.withZoneMaps(false); + case "--no-global-dict" -> options = options.withGlobalDict(false); + case "--edition" -> { + Optional edition = i + 1 < args.length ? edition(args[i + 1]) : Optional.empty(); + if (edition.isEmpty()) { + return unknownEdition(); + } + options = options.withEdition(edition.get()); + i++; + } + case String flag when flag.startsWith("--edition=") -> { + Optional edition = edition(flag.substring("--edition=".length())); + if (edition.isEmpty()) { + return unknownEdition(); + } + options = options.withEdition(edition.get()); + } + default -> positional.add(args[i]); + } + } + if (positional.size() != 2) { + System.err.println(USAGE); + return ExitStatus.USAGE_ERROR; + } + Path output = Path.of(positional.get(1)); + if (Files.exists(output)) { + System.err.println("refusing to overwrite: " + output); + return ExitStatus.USAGE_ERROR; + } + try (VortexHandle handle = CliHandles.openTarget(positional.get(0))) { + if (handle == null) { + return ExitStatus.FILE_NOT_FOUND; + } + long rows = Rewriter.rewrite(handle, output, options, ProgressBar::render); + ProgressBar.clear(); + System.out.printf("written: %s (%,d rows, %s)%n", output, rows, ByteSize.format(Files.size(output))); + return ExitStatus.OK; + } catch (FileAlreadyExistsException e) { + // another process created the output after the check above: it is theirs, so leave it + ProgressBar.clear(); + System.err.println("refusing to overwrite: " + output); + return ExitStatus.ERROR; + } catch (IOException | RuntimeException e) { + ProgressBar.clear(); + System.err.println("error: " + CliHandles.describe(e)); + if (!(e instanceof VortexException)) { + e.printStackTrace(System.err); + } + deleteQuietly(output); + return ExitStatus.ERROR; + } + } + + private static int unknownEdition() { + System.err.println("--edition needs one of: " + Editions.ALL.stream() + .map(e -> e.id().toString()).collect(Collectors.joining(", "))); + return ExitStatus.USAGE_ERROR; + } + + private static Optional edition(String id) { + return Editions.ALL.stream().filter(e -> e.id().toString().equals(id)).findFirst(); + } + + /// A failed rewrite leaves a partial file, which a retry would then refuse to overwrite. + private static void deleteQuietly(Path output) { + try { + Files.deleteIfExists(output); + } catch (IOException _) { + // best effort: the original error is the one worth reporting + } + } +} diff --git a/cli/src/main/java/io/github/dfa1/vortex/cli/Rewriter.java b/cli/src/main/java/io/github/dfa1/vortex/cli/Rewriter.java new file mode 100644 index 000000000..0c0d66dbe --- /dev/null +++ b/cli/src/main/java/io/github/dfa1/vortex/cli/Rewriter.java @@ -0,0 +1,190 @@ +package io.github.dfa1.vortex.cli; + +import io.github.dfa1.vortex.core.error.VortexException; +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.csv.ProgressListener; +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.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 io.github.dfa1.vortex.writer.VortexWriter; +import io.github.dfa1.vortex.writer.WriteOptions; +import io.github.dfa1.vortex.writer.encode.NullableData; + +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.function.LongConsumer; + +/// Re-encodes a file: scans every chunk of a source and writes it again through the +/// [VortexWriter], so the output is whatever the current writer and `options` pick, not what the +/// source's writer picked. +/// +/// The first version covers a flat struct of primitive, bool and utf8 columns, nullable or not. +/// Anything else (nested, decimal, binary, extension, f16) is rejected up front, before the +/// output file is created, rather than written wrongly. +final class Rewriter { + + private Rewriter() { + } + + /// Rewrites `source` into `output`. + /// + /// @param source the file to read + /// @param output the file to create; must not exist + /// @param options how to write it + /// @param progress receives the rows written so far, and the total + /// @return the number of rows written + /// @throws IOException if the output cannot be written + /// @throws VortexException if the source has a column type this version cannot rewrite + static long rewrite(VortexHandle source, Path output, WriteOptions options, ProgressListener progress) + throws IOException { + if (!(source.dtype() instanceof DType.Struct schema)) { + throw new VortexException("rewrite needs a struct file, got " + source.dtype()); + } + for (int i = 0; i < schema.fieldNames().size(); i++) { + requireSupported(schema.fieldNames().get(i), schema.fieldTypes().get(i)); + } + long total = source.layout().rowCount(); + long written = 0; + try (FileChannel channel = FileChannel.open(output, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE); + VortexWriter writer = VortexWriter.create(channel, schema, options); + ScanIterator scan = source.scan(ScanOptions.all())) { + while (scan.hasNext()) { + try (Chunk chunk = scan.next()) { + // over HTTP a column is fetched when read: start all of them together, merged + chunk.prefetch(schema.fieldNames()); + Map columns = new LinkedHashMap<>(); + for (Map.Entry column : chunk.columns().entrySet()) { + columns.put(column.getKey(), carrier(column.getKey(), column.getValue())); + } + writer.writeChunk(columns); + written += chunk.rowCount(); + progress.onProgress(written, total); + } + } + } + if (written != total) { + throw new VortexException("rewrote " + written + " rows of " + total + ": the source's layout and chunks disagree"); + } + return written; + } + + private static void requireSupported(ColumnName name, DType dtype) { + boolean supported = switch (dtype) { + case DType.Primitive p -> p.ptype() != PType.F16; + case DType.Bool _, DType.Utf8 _ -> true; + default -> false; + }; + if (!supported) { + throw new VortexException("rewrite does not support column '" + name.value() + "' of type " + dtype); + } + } + + /// The writer's input for one column: a primitive array when the column is not nullable, + /// otherwise a [NullableData] of the values (zero at each null, as the writer requires) and a + /// validity per row. Utf8 is the raw bytes, `null` per null row: going through `String` would + /// replace any byte sequence that is not valid UTF-8, which the format does not forbid. + private static Object carrier(ColumnName name, Chunk.Column column) { + try { + return carrier(column.array(), column.dtype()); + } catch (ClassCastException e) { + throw new VortexException("rewrite cannot read column '" + name.value() + "': unexpected array " + + column.array().getClass().getSimpleName() + " for " + column.dtype(), e); + } + } + + private static Object carrier(Array array, DType dtype) { + if (array.length() > Integer.MAX_VALUE) { + throw new VortexException("rewrite cannot hold a chunk of " + array.length() + " rows"); + } + MaskedArray.Unwrapped unwrapped = MaskedArray.unwrap(array); + BoolArray validity = unwrapped.validity(); + int rows = (int) array.length(); + Object values = switch (dtype) { + case DType.Primitive p -> primitive(unwrapped.inner(), p.ptype(), validity, rows); + case DType.Bool _ -> bools(unwrapped.inner(), validity, rows); + default -> utf8(unwrapped.inner(), validity, rows); + }; + if (!dtype.nullable() || dtype instanceof DType.Utf8) { + return values; + } + boolean[] present = new boolean[rows]; + fill(rows, i -> present[(int) i] = valid(validity, i)); + return new NullableData(values, present); + } + + private static Object primitive(Array values, PType ptype, BoolArray validity, int rows) { + switch (ptype) { + case I8, U8 -> { + byte[] out = new byte[rows]; + fill(rows, i -> out[(int) i] = valid(validity, i) ? ((ByteArray) values).getByte(i) : 0); + return out; + } + case I16, U16 -> { + short[] out = new short[rows]; + fill(rows, i -> out[(int) i] = valid(validity, i) ? ((ShortArray) values).getShort(i) : 0); + return out; + } + case I32, U32 -> { + int[] out = new int[rows]; + fill(rows, i -> out[(int) i] = valid(validity, i) ? ((IntArray) values).getInt(i) : 0); + return out; + } + case I64, U64 -> { + long[] out = new long[rows]; + fill(rows, i -> out[(int) i] = valid(validity, i) ? ((LongArray) values).getLong(i) : 0); + return out; + } + case F32 -> { + float[] out = new float[rows]; + fill(rows, i -> out[(int) i] = valid(validity, i) ? ((FloatArray) values).getFloat(i) : 0); + return out; + } + case F64 -> { + double[] out = new double[rows]; + fill(rows, i -> out[(int) i] = valid(validity, i) ? ((DoubleArray) values).getDouble(i) : 0); + return out; + } + default -> throw new VortexException("rewrite does not support " + ptype); + } + } + + private static boolean[] bools(Array values, BoolArray validity, int rows) { + boolean[] out = new boolean[rows]; + fill(rows, i -> out[(int) i] = valid(validity, i) && ((BoolArray) values).getBoolean(i)); + return out; + } + + private static byte[][] utf8(Array values, BoolArray validity, int rows) { + byte[][] out = new byte[rows][]; + VarBinArray bytes = (VarBinArray) values; + fill(rows, i -> out[(int) i] = valid(validity, i) ? bytes.getBytes(i) : null); + return out; + } + + private static boolean valid(BoolArray validity, long i) { + return validity == null || validity.getBoolean(i); + } + + private static void fill(int rows, LongConsumer body) { + for (long i = 0; i < rows; i++) { + body.accept(i); + } + } +} 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 0d1bc867d..3ef0ca701 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 @@ -68,6 +68,7 @@ private static int dispatch(String[] args) { case "select" -> SelectCommand.run(args); case "stats" -> StatsCommand.run(args); case "filter" -> FilterCommand.run(args); + case "rewrite" -> RewriteCommand.run(args); default -> { System.err.println("unknown subcommand: " + args[0]); printUsage(System.err); @@ -92,6 +93,7 @@ static void printUsage(PrintStream out) { out.println(" select [...] [--where ] project columns to CSV on stdout, optionally filtered"); out.println(" stats print per-column min/max statistics"); out.println(" filter filter rows to CSV (e.g. \"price >= 100\")"); + out.println(" rewrite [options] re-encode a file (--compact --no-zone-maps --no-global-dict --edition )"); out.println("Any subcommand also takes --timing: print the elapsed time to stderr, JVM start-up excluded."); out.println("Any subcommand also takes --allow-unknown: open a file whose extension types have no decoder,"); out.println(" reading those columns as their storage."); diff --git a/cli/src/test/java/io/github/dfa1/vortex/cli/RewriteCommandTest.java b/cli/src/test/java/io/github/dfa1/vortex/cli/RewriteCommandTest.java new file mode 100644 index 000000000..ebb1f35c7 --- /dev/null +++ b/cli/src/test/java/io/github/dfa1/vortex/cli/RewriteCommandTest.java @@ -0,0 +1,288 @@ +package io.github.dfa1.vortex.cli; + +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.core.model.DTypes; +import io.github.dfa1.vortex.core.model.PType; +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.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.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.stream.Stream; + +import static io.github.dfa1.vortex.cli.CliTestSupport.capture; +import static org.assertj.core.api.Assertions.assertThat; + +class RewriteCommandTest { + + private static final ColumnName ID = ColumnName.of("id"); + private static final ColumnName QTY = ColumnName.of("qty"); + private static final ColumnName PRICE = ColumnName.of("price"); + private static final ColumnName NAME = ColumnName.of("name"); + private static final ColumnName FLAG = ColumnName.of("flag"); + + @Test + void wrongArity_returnsUsageError() { + // Given / When + CliTestSupport.Captured result = capture(() -> RewriteCommand.run(new String[]{"rewrite", "only-one"})); + + // Then + assertThat(result.status()).isEqualTo(ExitStatus.USAGE_ERROR); + assertThat(result.stderr()).contains("usage:"); + } + + @Test + void existingOutput_isNotOverwritten(@TempDir Path tmp) throws IOException { + // Given — overwriting would destroy the source if the two paths were the same file + Path source = writeSource(tmp, 10); + + // When + CliTestSupport.Captured result = capture(() -> + RewriteCommand.run(new String[]{"rewrite", source.toString(), source.toString()})); + + // Then + assertThat(result.status()).isEqualTo(ExitStatus.USAGE_ERROR); + assertThat(result.stderr()).contains("refusing to overwrite"); + assertThat(Files.size(source)).isPositive(); + } + + @Test + void missingSource_returnsFileNotFound(@TempDir Path tmp) { + // Given + Path missing = tmp.resolve("nope.vortex"); + + // When + CliTestSupport.Captured result = capture(() -> + RewriteCommand.run(new String[]{"rewrite", missing.toString(), tmp.resolve("out.vortex").toString()})); + + // Then + assertThat(result.status()).isEqualTo(ExitStatus.FILE_NOT_FOUND); + } + + static Stream> optionSets() { + return Stream.of( + List.of(), + List.of("--compact"), + List.of("--no-zone-maps"), + List.of("--no-global-dict"), + List.of("--edition", "core2025.10.0"), + List.of("--edition=core2025.10.0"), + List.of("--compact", "--no-zone-maps", "--no-global-dict", "--edition", "core2026.08.0")); + } + + @ParameterizedTest + @MethodSource("optionSets") + void rewrite_keepsEveryValueAndNull_whateverTheOptions(List options, @TempDir Path tmp) throws IOException { + // Given — nullable and non-nullable columns of every supported family, with nulls in the + // nullable ones, so a dropped validity or a shifted row shows up as a value mismatch + int rows = 5_000; + Path source = writeSource(tmp, rows); + Path output = tmp.resolve("out.vortex"); + List command = new ArrayList<>(List.of("rewrite")); + command.addAll(options); + command.addAll(List.of(source.toString(), output.toString())); + String[] args = command.toArray(String[]::new); + + // When + CliTestSupport.Captured result = capture(() -> RewriteCommand.run(args)); + + // Then + assertThat(result.status()).as(result.stderr()).isEqualTo(ExitStatus.OK); + assertThat(result.stdout()).contains("written:").containsPattern("5\\D000 rows"); + assertThat(longs(output, ID)).isEqualTo(longs(source, ID)); + assertThat(longs(output, QTY)).isEqualTo(longs(source, QTY)); + assertThat(doubles(output, PRICE)).isEqualTo(doubles(source, PRICE)); + assertThat(strings(output, NAME)).isEqualTo(strings(source, NAME)); + assertThat(bools(output, FLAG)).isEqualTo(bools(source, FLAG)); + assertThat(strings(output, NAME)).containsNull(); + assertThat(longs(output, QTY)).containsNull(); + } + + @Test + void rewrite_keepsMultiByteTextAndNulls(@TempDir Path tmp) throws IOException { + // Given — 2-, 3- and 4-byte sequences, an empty string and a null: text is copied as bytes, so + // none may be cut or re-encoded. (Invalid UTF-8 cannot be a source here: the writer rejects it.) + DType.Struct schema = new DType.Struct(List.of(NAME), List.of(new DType.Utf8(true)), false); + String[] values = {"héllo", "日本語", "😀 ok", "", null, "plain"}; + Path source = tmp.resolve("text.vortex"); + try (FileChannel ch = FileChannel.open(source, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + VortexWriter writer = VortexWriter.create(ch, schema, WriteOptions.defaults())) { + writer.writeChunk(Map.of(NAME, values)); + } + Path output = tmp.resolve("out.vortex"); + + // When + CliTestSupport.Captured result = capture(() -> + RewriteCommand.run(new String[]{"rewrite", source.toString(), output.toString()})); + + // Then + assertThat(result.status()).as(result.stderr()).isEqualTo(ExitStatus.OK); + assertThat(strings(output, NAME)).containsExactly(values); + } + + @ParameterizedTest + @ValueSource(strings = {"--edition core1999.01.0", "--edition nonsense", "--edition=nonsense", "--edition --compact"}) + void unknownEdition_returnsUsageErrorListingTheKnownOnes(String edition, @TempDir Path tmp) throws IOException { + // Given + Path source = writeSource(tmp, 10); + List command = new ArrayList<>(List.of("rewrite")); + command.addAll(List.of(edition.split(" "))); + command.addAll(List.of(source.toString(), tmp.resolve("out.vortex").toString())); + + // When + CliTestSupport.Captured result = capture(() -> RewriteCommand.run(command.toArray(String[]::new))); + + // Then + assertThat(result.status()).isEqualTo(ExitStatus.USAGE_ERROR); + assertThat(result.stderr()).contains("--edition needs one of:").contains("core2026.08.0"); + assertThat(tmp.resolve("out.vortex")).doesNotExist(); + } + + @Test + void editionTooOldForTheData_failsAndLeavesNoOutput(@TempDir Path tmp) throws IOException { + // Given — a nullable utf8 column is written as vortex.masked, which joins core2025.10.0 + Path source = writeSource(tmp, 10); + Path output = tmp.resolve("out.vortex"); + + // When + CliTestSupport.Captured result = capture(() -> RewriteCommand.run(new String[]{ + "rewrite", "--edition", "core2025.05.0", source.toString(), output.toString()})); + + // Then — the edition guard's own message says which edition to enable, and no partial file stays + assertThat(result.status()).isEqualTo(ExitStatus.ERROR); + assertThat(result.stderr()).contains("vortex.masked").contains("core2025.10.0"); + assertThat(output).doesNotExist(); + } + + @Test + void editionWithoutValue_returnsUsageError() { + // Given / When + CliTestSupport.Captured result = capture(() -> RewriteCommand.run(new String[]{"rewrite", "a", "--edition"})); + + // Then + assertThat(result.status()).isEqualTo(ExitStatus.USAGE_ERROR); + } + + @Test + void unsupportedColumn_failsBeforeCreatingTheOutput(@TempDir Path tmp) throws IOException { + // Given — a decimal column: rewriting it as anything else would silently change the data + DType.Struct schema = new DType.Struct(List.of(ID), List.of(new DType.Decimal((byte) 10, (byte) 2, false)), false); + Path source = tmp.resolve("decimal.vortex"); + try (FileChannel ch = FileChannel.open(source, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + VortexWriter writer = VortexWriter.create(ch, schema, WriteOptions.defaults())) { + writer.writeChunk(Map.of(ID, new java.math.BigDecimal[]{java.math.BigDecimal.ONE})); + } + Path output = tmp.resolve("out.vortex"); + + // When + CliTestSupport.Captured result = capture(() -> + RewriteCommand.run(new String[]{"rewrite", source.toString(), output.toString()})); + + // Then + assertThat(result.status()).isEqualTo(ExitStatus.ERROR); + assertThat(result.stderr()).contains("does not support column 'id'"); + assertThat(output).doesNotExist(); + } + + // ── helpers ─────────────────────────────────────────────────────────────── + + private static Path writeSource(Path dir, int rows) throws IOException { + DType.Struct schema = new DType.Struct( + List.of(ID, QTY, PRICE, NAME, FLAG), + List.of(DTypes.I64, new DType.Primitive(PType.I32, true), DTypes.F64, new DType.Utf8(true), DTypes.BOOL), + false); + long[] ids = new long[rows]; + Integer[] qty = new Integer[rows]; + double[] price = new double[rows]; + String[] name = new String[rows]; + boolean[] flag = new boolean[rows]; + for (int i = 0; i < rows; i++) { + ids[i] = 1_000L * i; + qty[i] = i % 7 == 0 ? null : i % 100; + price[i] = i / 4.0; + name[i] = i % 5 == 0 ? null : "name-" + (i % 50); + flag[i] = i % 3 == 0; + } + Path file = dir.resolve("source.vortex"); + try (FileChannel ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + VortexWriter writer = VortexWriter.create(ch, schema, WriteOptions.defaults())) { + writer.writeChunk(Map.of(ID, ids, QTY, qty, PRICE, price, NAME, name, FLAG, flag)); + } + return file; + } + + private static List longs(Path file, ColumnName column) throws IOException { + List out = new ArrayList<>(); + scan(file, column, (array, validity, i) -> { + if (array instanceof LongArray a) { + out.add(valid(validity, i) ? a.getLong(i) : null); + } else { + out.add(valid(validity, i) ? (long) ((IntArray) array).getInt(i) : null); + } + }); + return out; + } + + private static List doubles(Path file, ColumnName column) throws IOException { + List out = new ArrayList<>(); + scan(file, column, (array, validity, i) -> out.add(valid(validity, i) ? ((DoubleArray) array).getDouble(i) : null)); + return out; + } + + private static List strings(Path file, ColumnName column) throws IOException { + List out = new ArrayList<>(); + scan(file, column, (array, validity, i) -> out.add(valid(validity, i) ? ((VarBinArray) array).getString(i) : null)); + return out; + } + + private static List bools(Path file, ColumnName column) throws IOException { + List out = new ArrayList<>(); + scan(file, column, (array, validity, i) -> out.add(valid(validity, i) ? ((BoolArray) array).getBoolean(i) : null)); + return out; + } + + private static boolean valid(BoolArray validity, long i) { + return validity == null || validity.getBoolean(i); + } + + private static void scan(Path file, ColumnName column, RowVisitor visitor) throws IOException { + try (VortexReader reader = VortexReader.open(file); + ScanIterator scan = reader.scan(ScanOptions.all())) { + while (scan.hasNext()) { + try (Chunk chunk = scan.next()) { + MaskedArray.Unwrapped unwrapped = MaskedArray.unwrap(chunk.column(column)); + for (long i = 0; i < chunk.rowCount(); i++) { + visitor.visit(unwrapped.inner(), unwrapped.validity(), i); + } + } + } + } + } + + @FunctionalInterface + private interface RowVisitor { + void visit(io.github.dfa1.vortex.reader.array.Array array, BoolArray validity, long row); + } +} diff --git a/docs/how-to.md b/docs/how-to.md index 9354ffec5..485b4e090 100644 --- a/docs/how-to.md +++ b/docs/how-to.md @@ -508,6 +508,40 @@ forces the encodings of one column. --- +## Re-encode a Vortex file + +`rewrite` reads every chunk of a file and writes it again with the current writer, so the +output is what today's writer picks, not what the writer that made the file picked. Use it to +repack a file with `--compact`, to target an older edition, or to pick up a newer writer's +encodings: + +```bash +java -jar cli/target/vortex-cli-*-all.jar rewrite --compact in.vortex out.vortex + +# straight from a URL +java -jar cli/target/vortex-cli-*-all.jar rewrite https://example.com/data.vortex out.vortex +``` + +| Option | Effect | +|---|---| +| `--compact` | Rust's compact preset (Zstandard for text, Pco for numbers), as for `import` | +| `--no-zone-maps` | write no per-zone statistics, so filtered scans cannot prune | +| `--no-global-dict` | no shared dictionary across chunks | +| `--edition ` | write only the encodings of that edition, e.g. `core2025.10.0`; an edition too old for the data fails and names the one to enable | + +On the OpenPowerlifting sample (4.1M rows) a default rewrite wrote 161.6 MB and `--compact` +104.8 MB. The rewrite of that file from a local copy took 8 s. + +Limits of this first version: + +- The file must be a flat struct of primitive (not `f16`), bool and utf8 columns, nullable or + not. Any other column (nested, decimal, binary, extension) fails before the output is created. +- The output must not exist, so a rewrite can never overwrite its source. A failed run deletes its + partial output. +- A URL source is read whole, chunk by chunk. + +--- + ## Import from a JDBC source **API:** diff --git a/docs/reference.md b/docs/reference.md index 6c3195a24..11e78bd96 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -812,6 +812,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 ...] [--where "" ...]` | Project columns to CSV; each `--where` is a `filter` expression and all must hold (the filtered columns need not be printed) | | `filter` | `filter ""` | Filter rows to CSV | +| `rewrite` | `rewrite [--compact] [--no-zone-maps] [--no-global-dict] [--edition ] ` | Re-encode a file with the current writer (`` as in `Editions`, e.g. `core2026.08.0`); flat struct of primitive, bool and utf8 columns | | `import` | `import [--delimiter ] [--compact] [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; `--compact` is Rust's compact preset (Zstandard for text, Pco for numbers), see [Convert CSV to Vortex](how-to.md#convert-csv-to-vortex) | Any subcommand also takes `--timing`, anywhere on the command line, including before the subcommand: it prints diff --git a/integration/pom.xml b/integration/pom.xml index 792e15ea3..91fbb2505 100644 --- a/integration/pom.xml +++ b/integration/pom.xml @@ -57,6 +57,11 @@ vortex-parquet test + + io.github.dfa1.vortex + vortex-cli + test + io.github.dfa1.vortex vortex-csv diff --git a/integration/src/test/java/io/github/dfa1/vortex/cli/RewriteInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/cli/RewriteInteropIntegrationTest.java new file mode 100644 index 000000000..3cb687ed7 --- /dev/null +++ b/integration/src/test/java/io/github/dfa1/vortex/cli/RewriteInteropIntegrationTest.java @@ -0,0 +1,161 @@ +package io.github.dfa1.vortex.cli; + +import dev.vortex.api.DataSource; +import dev.vortex.api.Expression; +import dev.vortex.api.Partition; +import dev.vortex.api.Scan; +import dev.vortex.api.ScanOptions; +import dev.vortex.api.Session; +import dev.vortex.api.VortexWriter; +import dev.vortex.arrow.ArrowAllocation; +import dev.vortex.jni.NativeLoader; +import org.apache.arrow.c.ArrowArray; +import org.apache.arrow.c.ArrowSchema; +import org.apache.arrow.c.Data; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.vector.BigIntVector; +import org.apache.arrow.vector.BitVector; +import org.apache.arrow.vector.FieldVector; +import org.apache.arrow.vector.Float8Vector; +import org.apache.arrow.vector.IntVector; +import org.apache.arrow.vector.VarCharVector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.ipc.ArrowReader; +import org.apache.arrow.vector.types.FloatingPointPrecision; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.Schema; +import org.apache.arrow.vector.util.Text; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/// `vortex rewrite` of a vortex-jni file: vortex-jni reads the rewritten file with the same rows, +/// nulls included, whole and under a filter that prunes zones and slices the other columns. A +/// Java-only test shares the writer's own assumptions; this is what says the rewritten file is +/// still a file Rust reads as the source (#743 follow-up). +class RewriteInteropIntegrationTest { + + private static final Session SESSION = Session.create(); + private static final BufferAllocator ALLOCATOR = ArrowAllocation.rootAllocator(); + private static final int ROWS = 30_000; + private static final String[] COLUMNS = {"id", "qty", "price", "name", "flag"}; + + static { + NativeLoader.loadJni(); + } + + @TempDir + Path tmp; + + @Test + void rewrittenFile_readsBackTheSameThroughVortexJni() throws Exception { + // Given — a file written by vortex-jni, with nulls in the nullable columns and enough rows + // for several zones + Path source = writeWithJni(); + Path output = tmp.resolve("rewritten.vortex"); + + // When + int status = RewriteCommand.run(new String[]{"rewrite", source.toString(), output.toString()}); + int compactStatus = RewriteCommand.run(new String[]{"rewrite", "--compact", source.toString(), + tmp.resolve("compact.vortex").toString()}); + + // Then + assertThat(status).isEqualTo(ExitStatus.OK); + assertThat(compactStatus).isEqualTo(ExitStatus.OK); + List> expected = readJni(source, null); + assertThat(expected).hasSize(ROWS); + assertThat(readJni(output, null)).isEqualTo(expected); + assertThat(readJni(tmp.resolve("compact.vortex"), null)).isEqualTo(expected); + Expression late = Expression.binary(Expression.BinaryOp.GTE, Expression.column("id"), + Expression.literal(20_000L)); + assertThat(readJni(output, late)).isEqualTo(readJni(source, late)); + Expression byQty = Expression.binary(Expression.BinaryOp.GTE, Expression.column("qty"), + Expression.literal(60)); + assertThat(readJni(output, byQty)).isEqualTo(readJni(source, byQty)); + } + + private Path writeWithJni() throws IOException { + Schema arrow = new Schema(List.of( + Field.notNullable("id", new ArrowType.Int(64, true)), + Field.nullable("qty", new ArrowType.Int(32, true)), + Field.notNullable("price", new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE)), + Field.nullable("name", ArrowType.Utf8.INSTANCE), + Field.notNullable("flag", ArrowType.Bool.INSTANCE))); + Path file = tmp.resolve("source.vortex"); + try (VortexWriter writer = VortexWriter.builder(SESSION, file.toAbsolutePath().toUri().toString(), arrow, + ALLOCATOR).build(); + VectorSchemaRoot root = VectorSchemaRoot.create(arrow, ALLOCATOR)) { + root.allocateNew(); + BigIntVector id = (BigIntVector) root.getVector("id"); + IntVector qty = (IntVector) root.getVector("qty"); + Float8Vector price = (Float8Vector) root.getVector("price"); + VarCharVector name = (VarCharVector) root.getVector("name"); + BitVector flag = (BitVector) root.getVector("flag"); + for (int i = 0; i < ROWS; i++) { + id.setSafe(i, i); + if (i % 7 == 0) { + qty.setNull(i); + } else { + qty.setSafe(i, i % 100); + } + price.setSafe(i, i / 4.0); + if (i % 5 == 0) { + name.setNull(i); + } else { + name.setSafe(i, ("name-" + i % 50).getBytes(StandardCharsets.UTF_8)); + } + flag.setSafe(i, i % 3 == 0 ? 1 : 0); + } + root.setRowCount(ROWS); + try (ArrowArray array = ArrowArray.allocateNew(ALLOCATOR); ArrowSchema as = ArrowSchema.allocateNew(ALLOCATOR)) { + Data.exportVectorSchemaRoot(ALLOCATOR, root, null, array, as); + writer.writeBatch(array.memoryAddress(), as.memoryAddress()); + } + } + return file; + } + + private static List> readJni(Path file, Expression filter) throws IOException { + var options = ScanOptions.builder().projection(Expression.select(COLUMNS, Expression.root())); + if (filter != null) { + options.filter(filter); + } + List> rows = new ArrayList<>(); + Scan scan = DataSource.open(SESSION, file.toAbsolutePath().toUri().toString()).scan(options.build()); + while (scan.hasNext()) { + Partition partition = scan.next(); + try (ArrowReader reader = partition.scanArrow(ALLOCATOR)) { + while (reader.loadNextBatch()) { + VectorSchemaRoot root = reader.getVectorSchemaRoot(); + for (int i = 0; i < root.getRowCount(); i++) { + List row = new ArrayList<>(); + for (String column : COLUMNS) { + row.add(value(root.getVector(column), i)); + } + rows.add(row); + } + } + } + } + return rows; + } + + private static Object value(FieldVector vector, int i) { + if (vector.isNull(i)) { + return null; + } + Object value = vector.getObject(i); + if (value instanceof Text text) { + return text.toString(); + } + return value; + } +} diff --git a/pom.xml b/pom.xml index 3c0758927..7a4290d69 100644 --- a/pom.xml +++ b/pom.xml @@ -194,6 +194,11 @@ vortex-jdbc ${project.version} + + io.github.dfa1.vortex + vortex-cli + ${project.version} + io.github.dfa1.vortex vortex-parquet