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