fix(csv): stream import rows to fix OOM on large files (fixes #52)

dfa1 · claude · dfa1 · commit d5280ae2f8b5 · 2026-06-18T08:14:36.000+02:00
Replace readAllRows (full List&lt;String[]&gt; in heap) with two-pass streaming:
- Pass 1: infer schema by streaming the file with O(1) memory
- Pass 2: write in chunks of chunkSize without buffering the full file
Schema provided via withSchema() skips pass 1.

Progress now reports (rowsDone, -1) since total is unknown during
streaming; ImportCommand.renderProgress shows "imported N rows"
instead of a percentage bar when total &lt; 0.

Co-Authored-By: Claude Sonnet 4.6 &lt;noreply@anthropic.com&gt;
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
@@ -113,10 +113,14 @@ vortexPath, formatBytes(inputBytes), formatBytes(vortexBytes),
     }
 
     private static void renderProgress(long done, long total) {
-        int pct = total > 0 ? (int) (done * 100L / total) : 100;
-        int filled = pct * 30 / 100;
-        String bar = "=".repeat(filled) + (filled < 30 ? ">" : "") + " ".repeat(Math.max(0, 29 - filled));
-        System.err.printf("\r  [%s] %3d%%  %,d / %,d rows", bar, pct, done, total);
+        if (total < 0) {
+            System.err.printf("\r  imported %,d rows", done);
+        } else {
+            int pct = total > 0 ? (int) (done * 100L / total) : 100;
+            int filled = pct * 30 / 100;
+            String bar = "=".repeat(filled) + (filled < 30 ? ">" : "") + " ".repeat(Math.max(0, 29 - filled));
+            System.err.printf("\r  [%s] %3d%%  %,d / %,d rows", bar, pct, done, total);
+        }
         System.err.flush();
     }
 
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
@@ -11,6 +11,7 @@
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -20,126 +21,169 @@
 /// Column types are inferred in priority order: long → double → boolean → utf8.
 /// Provide a schema via [ImportOptions#withSchema] to skip inference.
 /// Empty cell values are treated as 0 / false / "" for typed columns.
+///
+/// Import is streaming: rows are read and written in chunks of [ImportOptions#chunkSize]
+/// without loading the entire file into memory. Schema inference requires a first
+/// sequential pass over the file; writing is a second pass. When a schema is provided
+/// via [ImportOptions#withSchema] only one pass is needed.
 public final class CsvImporter {
 
     private CsvImporter() {
     }
 
+    /// Imports a CSV file to a Vortex file using default options.
+    ///
+    /// @param csvPath    path to the source CSV file
+    /// @param vortexPath path to write the output Vortex file
+    /// @throws IOException if reading or writing fails
     public static void importCsv(Path csvPath, Path vortexPath) throws IOException {
         importCsv(csvPath, vortexPath, ImportOptions.defaults());
     }
 
+    /// Imports a CSV file to a Vortex file.
+    ///
+    /// Rows are streamed in chunks — the file is never fully loaded into memory.
+    /// If no schema is provided, a first streaming pass infers column types; a
+    /// second pass writes the data. Progress is reported via
+    /// {@link ProgressListener#onProgress(long, long)} with `rowsTotal = -1`
+    /// (total unknown) on each completed chunk.
+    ///
+    /// @param csvPath    path to the source CSV file
+    /// @param vortexPath path to write the output Vortex file
+    /// @param options    import configuration
+    /// @throws IOException              if reading or writing fails
+    /// @throws IllegalArgumentException if the CSV file is empty
     public static void importCsv(Path csvPath, Path vortexPath, ImportOptions options) throws IOException {
-        List<String[]> rows = readAllRows(csvPath, options);
-        if (rows.isEmpty()) {
-            throw new IllegalArgumentException("CSV file is empty");
-        }
-
-        String[] headers;
-        int dataStart;
-        if (options.hasHeader()) {
-            headers = rows.getFirst();
-            dataStart = 1;
-        } else {
-            headers = generateHeaders(rows.getFirst().length);
-            dataStart = 0;
-        }
-
-        List<String[]> dataRows = rows.subList(dataStart, rows.size());
-        int colCount = headers.length;
+        String[] headers = readHeader(csvPath, options);
 
-        DType.Struct schema;
-        if (options.schema() != null) {
-            schema = options.schema();
-        } else {
-            schema = inferSchema(headers, dataRows, colCount);
-        }
+        DType.Struct schema = options.schema() != null
+                ? options.schema()
+                : inferSchemaStreaming(csvPath, options, headers);
 
         try (FileChannel channel = FileChannel.open(
                 vortexPath, StandardOpenOption.CREATE, StandardOpenOption.WRITE,
                 StandardOpenOption.TRUNCATE_EXISTING);
              VortexWriter writer = VortexWriter.create(channel, schema, options.writeOptions())) {
-            int chunkSize = options.chunkSize();
-            int total = dataRows.size();
-            for (int start = 0; start < total; start += chunkSize) {
-                int end = Math.min(start + chunkSize, total);
-                writer.writeChunk(buildChunk(schema, dataRows.subList(start, end)));
-                if (options.progressListener() != null) {
-                    options.progressListener().onProgress(end, total);
-                }
-            }
+            writeChunksStreaming(csvPath, options, schema, writer);
         }
     }
 
-    private static List<String[]> readAllRows(Path path, ImportOptions options) throws IOException {
-        List<String[]> rows = new ArrayList<>();
-        try (CsvReader<CsvRecord> reader = CsvReader.builder()
-                                                   .fieldSeparator(options.delimiter())
-                                                   .ofCsvRecord(path)) {
-            for (CsvRecord record : reader) {
-                rows.add(record.getFields().toArray(String[]::new));
+    private static String[] readHeader(Path path, ImportOptions options) throws IOException {
+        try (CsvReader<CsvRecord> reader = csvReader(path, options)) {
+            CsvRecord first = reader.stream().findFirst().orElse(null);
+            if (first == null) {
+                throw new IllegalArgumentException("CSV file is empty");
             }
+            String[] fields = first.getFields().toArray(String[]::new);
+            return options.hasHeader() ? fields : generateHeaders(fields.length);
         }
-        return rows;
     }
 
-    private static String[] generateHeaders(int colCount) {
-        String[] headers = new String[colCount];
-        for (int i = 0; i < colCount; i++) {
-            headers[i] = "col" + i;
+    private static DType.Struct inferSchemaStreaming(Path path, ImportOptions options, String[] headers)
+            throws IOException {
+        int colCount = headers.length;
+        boolean[] canBeLong = new boolean[colCount];
+        boolean[] canBeDouble = new boolean[colCount];
+        boolean[] canBeBool = new boolean[colCount];
+        Arrays.fill(canBeLong, true);
+        Arrays.fill(canBeDouble, true);
+        Arrays.fill(canBeBool, true);
+
+        try (CsvReader<CsvRecord> reader = csvReader(path, options)) {
+            boolean skipFirst = options.hasHeader();
+            for (CsvRecord record : reader) {
+                if (skipFirst) {
+                    skipFirst = false;
+                    continue;
+                }
+                String[] fields = record.getFields().toArray(String[]::new);
+                for (int c = 0; c < colCount; c++) {
+                    String val = safeGet(fields, c);
+                    if (val.isEmpty()) {
+                        continue;
+                    }
+                    if (canBeLong[c]) {
+                        try {
+                            Long.parseLong(val);
+                        } catch (NumberFormatException e) {
+                            canBeLong[c] = false;
+                        }
+                    }
+                    if (canBeDouble[c]) {
+                        try {
+                            Double.parseDouble(val);
+                        } catch (NumberFormatException e) {
+                            canBeDouble[c] = false;
+                        }
+                    }
+                    if (canBeBool[c]) {
+                        if (!val.equalsIgnoreCase("true") && !val.equalsIgnoreCase("false")) {
+                            canBeBool[c] = false;
+                        }
+                    }
+                }
+            }
         }
-        return headers;
-    }
 
-    private static DType.Struct inferSchema(String[] headers, List<String[]> rows, int colCount) {
         List<String> names = List.of(headers);
         List<DType> types = new ArrayList<>(colCount);
         for (int c = 0; c < colCount; c++) {
-            types.add(inferColumnType(rows, c));
+            types.add(resolveType(canBeLong[c], canBeDouble[c], canBeBool[c]));
         }
         return new DType.Struct(names, types, false);
     }
 
-    /// Infers the narrowest type for a column using a single pass.
-    ///
-    /// Priority: long → double → bool → utf8. Each flag starts true and can only
-    /// transition to false. Empty cells are skipped (compatible with any type).
-    /// An all-empty column is inferred as long (all flags remain true).
-    ///
-    /// Integer values always infer as i64 (not i32/i16): CSV has no type annotations,
-    /// so the widest safe integer is chosen. Use [ImportOptions#withSchema] to force i32/i16.
-    /// Floating-point values always infer as f64 (not f32) for the same reason.
-    private static DType inferColumnType(List<String[]> rows, int colIdx) {
-        boolean canBeLong = true;
-        boolean canBeDouble = true;
-        boolean canBeBool = true;
-
-        for (String[] row : rows) {
-            String val = safeGet(row, colIdx);
-            if (val.isEmpty()) {
-                continue;
-            }
-            if (canBeLong) {
-                try {
-                    Long.parseLong(val);
-                } catch (NumberFormatException e) {
-                    canBeLong = false;
-                }
-            }
-            if (canBeDouble) {
-                try {
-                    Double.parseDouble(val);
-                } catch (NumberFormatException e) {
-                    canBeDouble = false;
+    private static void writeChunksStreaming(Path path, ImportOptions options, DType.Struct schema,
+                                             VortexWriter writer) throws IOException {
+        int chunkSize = options.chunkSize();
+        List<String[]> chunk = new ArrayList<>(chunkSize);
+        long totalRows = 0;
+
+        try (CsvReader<CsvRecord> reader = csvReader(path, options)) {
+            boolean skipFirst = options.hasHeader();
+            for (CsvRecord record : reader) {
+                if (skipFirst) {
+                    skipFirst = false;
+                    continue;
                 }
-            }
-            if (canBeBool) {
-                if (!val.equalsIgnoreCase("true") && !val.equalsIgnoreCase("false")) {
-                    canBeBool = false;
+                chunk.add(record.getFields().toArray(String[]::new));
+                if (chunk.size() == chunkSize) {
+                    writer.writeChunk(buildChunk(schema, chunk));
+                    totalRows += chunk.size();
+                    reportProgress(options, totalRows);
+                    chunk.clear();
                 }
             }
         }
 
+        if (!chunk.isEmpty()) {
+            writer.writeChunk(buildChunk(schema, chunk));
+            totalRows += chunk.size();
+            reportProgress(options, totalRows);
+        }
+    }
+
+    private static void reportProgress(ImportOptions options, long totalRows) {
+        if (options.progressListener() != null) {
+            options.progressListener().onProgress(totalRows, -1);
+        }
+    }
+
+    private static CsvReader<CsvRecord> csvReader(Path path, ImportOptions options) throws IOException {
+        return CsvReader.builder()
+                        .fieldSeparator(options.delimiter())
+                        .ofCsvRecord(path);
+    }
+
+    private static String[] generateHeaders(int colCount) {
+        String[] headers = new String[colCount];
+        for (int i = 0; i < colCount; i++) {
+            headers[i] = "col" + i;
+        }
+        return headers;
+    }
+
+    private static DType resolveType(boolean canBeLong, boolean canBeDouble, boolean canBeBool) {
         if (canBeLong) {
             return new DType.Primitive(PType.I64, false);
         }
@@ -152,7 +196,7 @@ private static DType inferColumnType(List<String[]> rows, int colIdx) {
         return new DType.Utf8(false);
     }
 
-    private static Map<String, Object> buildChunk(DType.Struct schema, List<String[]> rows) {
+    static Map<String, Object> buildChunk(DType.Struct schema, List<String[]> rows) {
         int n = rows.size();
         Map<String, Object> chunk = new LinkedHashMap<>();
         for (int c = 0; c < schema.fieldNames().size(); c++) {
diff --git a/csv/src/main/java/io/github/dfa1/vortex/csv/ProgressListener.java b/csv/src/main/java/io/github/dfa1/vortex/csv/ProgressListener.java
@@ -3,5 +3,10 @@
 /// Callback invoked during CSV import or export to report row-level progress.
 @FunctionalInterface
 public interface ProgressListener {
+
+    /// Called after each chunk is processed.
+    ///
+    /// @param rowsDone  number of data rows processed so far
+    /// @param rowsTotal total row count, or `-1` if the total is not known (streaming import)
     void onProgress(long rowsDone, long rowsTotal);
 }
