From a5fe7bf6c672062a7aae97170348986c9570dff1 Mon Sep 17 00:00:00 2001 From: "Victor M. Varela" Date: Thu, 6 Aug 2026 10:12:17 +0200 Subject: [PATCH 1/2] feat: transparent gzip input (.gz files and stdin) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add transparent gzip decompression for .gz files and gzip-magic stdin using std.compress.flate.Decompress (gzip container). No new dependencies. - Files: .gz extension detection, auto-format from inner extension (data.csv.gz → csv, table ) - Stdin: peek(2) magic sniff (0x1f 0x8b), non-consuming - Streaming decompress into existing buffered reader - Table name derived from inner name (strip .gz before stem) - Clear error on corrupt gzip stream (fatal + csv_error) - Non-gzip path unchanged (no sniffing overhead beyond 2 bytes) Closes #220 --- README.md | 16 ++++++ build.zig | 70 +++++++++++++++++++++++++ docs/sql-pipe.1.scd | 19 +++++++ src/args.zig | 7 ++- src/format.zig | 15 +++++- src/main.zig | 43 ++++++++++++++- tests/fixtures/sample.csv.gz | Bin 0 -> 74 bytes tests/fixtures/sample.csv.gz_truncated | Bin 0 -> 10 bytes 8 files changed, 166 insertions(+), 4 deletions(-) create mode 100644 tests/fixtures/sample.csv.gz create mode 100644 tests/fixtures/sample.csv.gz_truncated diff --git a/README.md b/README.md index fe8b908..753d5a4 100644 --- a/README.md +++ b/README.md @@ -296,6 +296,22 @@ $ sql-pipe orders.csv customers.csv \ JOIN customers c ON o.cust_id = c.id GROUP BY c.name' ``` +Gzip-compressed inputs are handled transparently (gzip only). A file ending in `.gz` +is decompressed automatically; the format is detected from the inner extension, and +the table name comes from the inner basename: + +```sh +$ sql-pipe data.csv.gz 'SELECT COUNT(*) FROM data' +``` + +Gzipped stdin is detected by its magic bytes, so both decompressed and compressed +pipes work: + +```sh +$ zcat huge.csv.gz | sql-pipe 'SELECT * FROM t' # decompressed by zcat +$ gzip -c data.ndjson | sql-pipe 'SELECT * FROM t' # gzip stdin magic detection +``` + Use `-I` to override auto-detection when the extension is wrong or ambiguous (`.txt`, `.dat`): ```sh diff --git a/build.zig b/build.zig index 62bed45..c284b3e 100644 --- a/build.zig +++ b/build.zig @@ -3708,6 +3708,76 @@ pub fn build(b: *std.Build) void { test_parquet_fuzz_empty.step.dependOn(b.getInstallStep()); test_step.dependOn(&test_parquet_fuzz_empty.step); + // ─── Gzip integration tests ──────────────────────────────────────────────── + + // Integration test: gzip file input (tests/fixtures/sample.csv.gz) + const test_gzip_file = b.addSystemCommand(&.{ + "bash", "-c", + \\result=$(./zig-out/bin/sql-pipe tests/fixtures/sample.csv.gz 'SELECT name FROM sample WHERE age > 27 ORDER BY name') + \\expected=$(printf 'Alice\nCarol') + \\[ "$result" = "$expected" ] + }); + test_gzip_file.step.dependOn(b.getInstallStep()); + test_step.dependOn(&test_gzip_file.step); + + // Integration test: gzip file table name derived from inner name (sample.csv.gz → "sample") + const test_gzip_table_name = b.addSystemCommand(&.{ + "bash", "-c", + \\result=$(./zig-out/bin/sql-pipe tests/fixtures/sample.csv.gz 'SELECT name FROM sample ORDER BY name') + \\expected=$(printf 'Alice\nBob\nCarol') + \\[ "$result" = "$expected" ] + }); + test_gzip_table_name.step.dependOn(b.getInstallStep()); + test_step.dependOn(&test_gzip_table_name.step); + + // Integration test: gzip stdin magic sniffing + // sample.csv.gz is already gzipped; pipe it directly (re-gzipping would double-compress). + const test_gzip_stdin = b.addSystemCommand(&.{ + "bash", "-c", + \\result=$(cat tests/fixtures/sample.csv.gz | ./zig-out/bin/sql-pipe 'SELECT name FROM t WHERE age < 30') + \\expected=$(printf 'Bob\n') + \\[ "$result" = "$expected" ] + }); + test_gzip_stdin.step.dependOn(b.getInstallStep()); + test_step.dependOn(&test_gzip_stdin.step); + + // Integration test: non-gzip data on stdin is not misdetected as corrupt gzip + const test_gzip_non_gzip_stdin = b.addSystemCommand(&.{ + "bash", "-c", + \\msg=$(printf 'hello\n' | ./zig-out/bin/sql-pipe --output-format json 'SELECT 1' 2>&1; echo "EXIT:$?") + \\echo "$msg" | grep -q 'EXIT:0' + }); + test_gzip_non_gzip_stdin.step.dependOn(b.getInstallStep()); + test_step.dependOn(&test_gzip_non_gzip_stdin.step); + + // Integration test: non-gzip stdin unchanged (regular CSV via stdin still works) + const test_gzip_stdin_unchanged = b.addSystemCommand(&.{ + "bash", "-c", + \\result=$(printf 'name,age\nAlice,30\n' | ./zig-out/bin/sql-pipe 'SELECT name FROM t') + \\[ "$result" = "Alice" ] + }); + test_gzip_stdin_unchanged.step.dependOn(b.getInstallStep()); + test_step.dependOn(&test_gzip_stdin_unchanged.step); + + // Integration test: .ndjson.gz inner format detection + const test_gzip_ndjson = b.addSystemCommand(&.{ + "bash", "-c", + \\printf '{"name":"Alice","age":30}\n{"name":"Bob","age":25}\n' | gzip -c > /tmp/test.ndjson.gz + \\result=$(./zig-out/bin/sql-pipe /tmp/test.ndjson.gz 'SELECT name FROM test WHERE age > 27') + \\rm -f /tmp/test.ndjson.gz + \\[ "$result" = "Alice" ] + }); + test_gzip_ndjson.step.dependOn(b.getInstallStep()); + test_step.dependOn(&test_gzip_ndjson.step); + + // Integration test: corrupt/truncated gzip file exits non-zero + const test_gzip_corrupt = b.addSystemCommand(&.{ + "bash", "-c", + \\! ./zig-out/bin/sql-pipe tests/fixtures/sample.csv.gz_truncated 'SELECT 1' >/dev/null 2>&1 + }); + test_gzip_corrupt.step.dependOn(b.getInstallStep()); + test_step.dependOn(&test_gzip_corrupt.step); + // ─── --checksum integration tests (issue #204) ────────────────────────────── // 25 data-driven cases (204a-204y). Scripts run with `set -euo pipefail` so a // failed assertion actually fails the step (a bare `[` + trailing `rm -f` diff --git a/docs/sql-pipe.1.scd b/docs/sql-pipe.1.scd index 75e72cd..7108461 100644 --- a/docs/sql-pipe.1.scd +++ b/docs/sql-pipe.1.scd @@ -36,6 +36,13 @@ DESCRIPTION missing extensions default to CSV. Use *-I* to override auto-detection (e.g., when a TSV file has a *.txt* extension). + Gzip-compressed inputs are supported transparently (gzip only). Files ending + in *.gz* are decompressed automatically; the input format is then detected + from the inner extension (e.g., *data.csv.gz* is read as CSV) and the table + name comes from the inner basename (*data.csv.gz* becomes table *data*). + Gzipped data piped to standard input is detected by its gzip magic bytes + (*\x1f\x8b*), so compressed stdin works without any extra flags. + This tool is useful for quick data transformations, filtering, grouping, aggregations, and multi-file joins without manual SQL database setup. @@ -339,6 +346,18 @@ EXAMPLES $ sql-pipe data.parquet 'SELECT name FROM data WHERE active = true' + Query a gzip-compressed CSV file (format detected from inner extension): + + $ sql-pipe data.csv.gz 'SELECT COUNT(*) FROM data' + + Pipe decompressed stdin (zcat decompresses, stdin reads as plain CSV): + + $ zcat huge.csv.gz | sql-pipe 'SELECT * FROM t' + + Pipe gzip-compressed stdin (detected by magic bytes): + + $ gzip -c data.ndjson | sql-pipe 'SELECT * FROM t' + Override auto-detection when the extension is wrong: $ sql-pipe -I tsv data.txt 'SELECT * FROM data' diff --git a/src/args.zig b/src/args.zig index 672fa2d..f785dba 100644 --- a/src/args.zig +++ b/src/args.zig @@ -1003,7 +1003,12 @@ fn isValidHttpHeader(header: []const u8) bool { } /// Derive a table name from a file path (basename without extension). +/// For .gz files, strips the .gz first so "data.csv.gz" → "data". fn tableNameFromPath(allocator: std.mem.Allocator, path: []const u8) (std.mem.Allocator.Error)![]const u8 { - const stem = std.fs.path.stem(path); + const inner = if (format.InputFormat.isGzipExtension(path)) + format.InputFormat.stripGzExtension(path) + else + path; + const stem = std.fs.path.stem(inner); return allocator.dupe(u8, stem); } diff --git a/src/format.zig b/src/format.zig index afa15d6..b11caeb 100644 --- a/src/format.zig +++ b/src/format.zig @@ -33,13 +33,26 @@ pub const InputFormat = enum { /// Detect input format from file extension. /// Returns null for unrecognized extensions. + /// A ".gz" extension is stripped first, so "data.csv.gz" maps to .csv. pub fn fromExtension(filename: []const u8) ?InputFormat { - const ext = std.fs.path.extension(filename); + const inner = stripGzExtension(filename); + const ext = std.fs.path.extension(inner); if (ext.len == 0) return null; const ext_no_dot = ext[1..]; // skip the leading '.' if (std.mem.eql(u8, ext_no_dot, "yml")) return .yaml; return std.meta.stringToEnum(InputFormat, ext_no_dot); } + + /// Return true when filename ends with ".gz". + pub fn isGzipExtension(filename: []const u8) bool { + return std.mem.endsWith(u8, filename, ".gz"); + } + + /// Return filename without a trailing ".gz", or the original when it isn't a .gz file. + pub fn stripGzExtension(filename: []const u8) []const u8 { + if (!isGzipExtension(filename)) return filename; + return filename[0 .. filename.len - 3]; + } }; // ─── Output format ───────────────────────────────────── diff --git a/src/main.zig b/src/main.zig index ac07ea6..f4a8d78 100644 --- a/src/main.zig +++ b/src/main.zig @@ -114,6 +114,26 @@ fn writeStreaming( try out_writer.end(writer); } +/// Backing storage for a gzip-wrapped reader. Heap-allocated so the returned +/// `*std.Io.Reader` (which points into `decompress.reader`) stays valid for +/// the lifetime of the program. +const GzipReader = struct { + decompress: std.compress.flate.Decompress, + buffer: [std.compress.flate.max_window_len]u8, +}; + +/// Wrap `reader` in a gzip decompressing reader and return its reader. +pub fn makeGzipReader(allocator: std.mem.Allocator, reader: *std.Io.Reader) !*std.Io.Reader { + const gz = try allocator.create(GzipReader); + // ponytail: leaked for program lifetime; input pointer valid only + // within loadPipelineInputs scope. CLI, single-use. + gz.* = .{ + .decompress = std.compress.flate.Decompress.init(reader, .gzip, &gz.buffer), + .buffer = undefined, + }; + return &gz.decompress.reader; +} + const progress_interval = loader.progress_interval; const fatal = sqlite_mod.fatal; @@ -307,7 +327,16 @@ pub fn loadPipelineInputs( fatal("cannot open file '{s}': {s}", stderr_writer, .csv_error, .{ file_input.path, @errorName(err) }); defer std.Io.File.close(file, io); var file_reader = std.Io.File.reader(file, io, &file_buf); - const rows = loadInput(allocator, io, db, file_input.table_name, file_input.format, &file_reader.interface, parsed, stderr_writer); + const is_gz = InputFormat.isGzipExtension(file_input.path); + // For .gz files: wrap in gzip decompressor, auto-detect inner format if not explicit + const effective_reader: *std.Io.Reader = if (is_gz) + makeGzipReader(allocator, &file_reader.interface) catch + fatal("out of memory allocating gzip reader for '{s}'", stderr_writer, .csv_error, .{file_input.path}) + else + &file_reader.interface; + // Format already resolved at parse time (args.zig uses fromExtension which strips .gz) + const effective_format = file_input.format; + const rows = loadInput(allocator, io, db, file_input.table_name, effective_format, effective_reader, parsed, stderr_writer); if (rows == 0) { fatal("empty input file: '{s}'", stderr_writer, .csv_error, .{file_input.path}); } @@ -318,8 +347,18 @@ pub fn loadPipelineInputs( if (parsed.has_stdin) { var stdin_buf: [4096]u8 = undefined; var stdin_reader = std.Io.File.reader(std.Io.File.stdin(), io, &stdin_buf); + // Sniff gzip magic by peeking first 2 bytes (peek fills buffer and + // returns without consuming, so we can wrap in place). + var effective: *std.Io.Reader = &stdin_reader.interface; + const peeked = stdin_reader.interface.peek(2) catch null; + if (peeked) |b| { + if (b.len >= 2 and b[0] == 0x1f and b[1] == 0x8b) { + effective = makeGzipReader(allocator, &stdin_reader.interface) catch + fatal("out of memory allocating gzip reader for stdin", stderr_writer, .csv_error, .{}); + } + } const stdin_table = if (parsed.urls.len > 0 or parsed.files.len > 0) "stdin" else "t"; - const rows = loadInput(allocator, io, db, stdin_table, parsed.input_format, &stdin_reader.interface, parsed, stderr_writer); + const rows = loadInput(allocator, io, db, stdin_table, parsed.input_format, effective, parsed, stderr_writer); total_rows += rows; } diff --git a/tests/fixtures/sample.csv.gz b/tests/fixtures/sample.csv.gz new file mode 100644 index 0000000000000000000000000000000000000000..c2d22badad55679e145ebfcc765e7cb0bd4146dd GIT binary patch literal 74 zcmV-Q0JZ-giwFSJGjwVI1ItUyP1Q+EPt{4zEUDyj%*jkn)iE~E@r!ika>`HAF*4Qh gapZDNEXvQ(F*en4&d5woOwZ>608xd{k2U}R03>}QCjbBd literal 0 HcmV?d00001 diff --git a/tests/fixtures/sample.csv.gz_truncated b/tests/fixtures/sample.csv.gz_truncated new file mode 100644 index 0000000000000000000000000000000000000000..e08934eb8634ab441e18579cc264c8f7dc81b829 GIT binary patch literal 10 Rcmb2|=3uyDT$07W3;+*n0-FE; literal 0 HcmV?d00001 From ed84840e06f72e0941f93cf16fea88829c92f8ff Mon Sep 17 00:00:00 2001 From: "Victor M. Varela" Date: Thu, 6 Aug 2026 13:26:49 +0200 Subject: [PATCH 2/2] fix: gzip input edge cases (empty table name, misleading corrupt error) - args.zig: path literally ".gz" produced an empty table name; fall back to the full path stem when stripping the extension collapses it. - main.zig: wrap the flate reader so error.ReadFailed is translated into a message naming the specific gzip failure (EndOfStream, WrongGzipChecksum, etc.) instead of a generic "failed to parse CSV". The wrapper forwards to flate's vtable methods and mirrors its buffer cursor so the shared-buffer reader keeps working; the previous blind forward hung on empty fill slices. - build.zig: corrupt-gzip integration test now uses a real .gz name (the old fixture ended .gz_truncated and never hit the gzip path) and asserts the new error message. --- build.zig | 12 +++++-- src/args.zig | 9 +++-- src/main.zig | 96 ++++++++++++++++++++++++++++++++++++++++++++++++---- 3 files changed, 103 insertions(+), 14 deletions(-) diff --git a/build.zig b/build.zig index c284b3e..f1ea831 100644 --- a/build.zig +++ b/build.zig @@ -3770,10 +3770,18 @@ pub fn build(b: *std.Build) void { test_gzip_ndjson.step.dependOn(b.getInstallStep()); test_step.dependOn(&test_gzip_ndjson.step); - // Integration test: corrupt/truncated gzip file exits non-zero + // Integration test: corrupt/truncated gzip file exits non-zero with a + // gzip-specific message. The truncated fixture is copied to a real .gz name + // (its own name ends in `.gz_truncated`, which isGzipExtension rejects). const test_gzip_corrupt = b.addSystemCommand(&.{ "bash", "-c", - \\! ./zig-out/bin/sql-pipe tests/fixtures/sample.csv.gz_truncated 'SELECT 1' >/dev/null 2>&1 + \\set -euo pipefail + \\cp tests/fixtures/sample.csv.gz_truncated /tmp/sql-pipe-corrupt.csv.gz + \\trap 'rm -f /tmp/sql-pipe-corrupt.csv.gz' EXIT + \\msg=$(./zig-out/bin/sql-pipe /tmp/sql-pipe-corrupt.csv.gz 'SELECT 1' 2>&1; echo "EXIT:$?") + \\echo "$msg" | grep -q 'EXIT:2' || { echo "expected exit 2, got: $msg"; exit 1; } + \\echo "$msg" | grep -q 'gzip decompression failed: EndOfStream' || { echo "missing gzip error msg: $msg"; exit 1; } + \\ }); test_gzip_corrupt.step.dependOn(b.getInstallStep()); test_step.dependOn(&test_gzip_corrupt.step); diff --git a/src/args.zig b/src/args.zig index f785dba..c4ee930 100644 --- a/src/args.zig +++ b/src/args.zig @@ -1005,10 +1005,9 @@ fn isValidHttpHeader(header: []const u8) bool { /// Derive a table name from a file path (basename without extension). /// For .gz files, strips the .gz first so "data.csv.gz" → "data". fn tableNameFromPath(allocator: std.mem.Allocator, path: []const u8) (std.mem.Allocator.Error)![]const u8 { - const inner = if (format.InputFormat.isGzipExtension(path)) - format.InputFormat.stripGzExtension(path) - else - path; - const stem = std.fs.path.stem(inner); + const inner = format.InputFormat.stripGzExtension(path); + // stripGzExtension can collapse a path named exactly ".gz" to "", whose + // stem is also empty and would produce an invalid empty table name. + const stem = if (std.fs.path.stem(inner).len == 0) std.fs.path.stem(path) else std.fs.path.stem(inner); return allocator.dupe(u8, stem); } diff --git a/src/main.zig b/src/main.zig index f4a8d78..03265c7 100644 --- a/src/main.zig +++ b/src/main.zig @@ -114,24 +114,106 @@ fn writeStreaming( try out_writer.end(writer); } -/// Backing storage for a gzip-wrapped reader. Heap-allocated so the returned -/// `*std.Io.Reader` (which points into `decompress.reader`) stays valid for -/// the lifetime of the program. +/// Wrapper around a gzip-decompressing reader. Heap-allocated so the returned +/// `*std.Io.Reader` stays valid for the lifetime of the program. The wrapper +/// shares the flate decompressor's buffer and mirrors its cursor state, so it +/// behaves exactly like the flate reader; its only job is to translate the raw +/// `error.ReadFailed` into a message naming the specific flate failure (stored +/// in `decompress.err`) instead of a generic read error. +/// +/// The wrapper forwards to flate's *vtable* methods (not the generic +/// `Reader.readVec`/`stream`/`discard` helpers), because the generic helpers +/// short-circuit on the empty `data` slice that `fill` passes and would return +/// 0 without ever refilling the decompressor's buffer, spinning forever. const GzipReader = struct { decompress: std.compress.flate.Decompress, buffer: [std.compress.flate.max_window_len]u8, + stderr_writer: *std.Io.Writer, + reader: std.Io.Reader, + + fn sync(r: *std.Io.Reader) *GzipReader { + const gz: *GzipReader = @fieldParentPtr("reader", r); + r.buffer = gz.decompress.reader.buffer; + r.seek = gz.decompress.reader.seek; + r.end = gz.decompress.reader.end; + return gz; + } + + fn stream(r: *std.Io.Reader, w: *std.Io.Writer, limit: std.Io.Limit) std.Io.Reader.StreamError!usize { + _ = sync(r); + const gz: *GzipReader = @fieldParentPtr("reader", r); + const n = gz.decompress.reader.vtable.stream(&gz.decompress.reader, w, limit) catch |err| switch (err) { + error.WriteFailed => return error.WriteFailed, + error.EndOfStream => return error.EndOfStream, + error.ReadFailed => { + const detail = gz.decompress.err orelse return error.ReadFailed; + fatal("gzip decompression failed: {s}", gz.stderr_writer, .csv_error, .{@errorName(detail)}); + }, + }; + _ = sync(r); + return n; + } + + fn discard(r: *std.Io.Reader, limit: std.Io.Limit) std.Io.Reader.Error!usize { + _ = sync(r); + const gz: *GzipReader = @fieldParentPtr("reader", r); + const n = gz.decompress.reader.vtable.discard(&gz.decompress.reader, limit) catch |err| switch (err) { + error.EndOfStream => return error.EndOfStream, + error.ReadFailed => { + const detail = gz.decompress.err orelse return error.ReadFailed; + fatal("gzip decompression failed: {s}", gz.stderr_writer, .csv_error, .{@errorName(detail)}); + }, + }; + _ = sync(r); + return n; + } + + fn readVec(r: *std.Io.Reader, data: [][]u8) std.Io.Reader.Error!usize { + _ = sync(r); + const gz: *GzipReader = @fieldParentPtr("reader", r); + const n = gz.decompress.reader.vtable.readVec(&gz.decompress.reader, data) catch |err| switch (err) { + error.EndOfStream => return error.EndOfStream, + error.ReadFailed => { + const detail = gz.decompress.err orelse return error.ReadFailed; + fatal("gzip decompression failed: {s}", gz.stderr_writer, .csv_error, .{@errorName(detail)}); + }, + }; + _ = sync(r); + return n; + } + + fn rebase(r: *std.Io.Reader, capacity: usize) std.Io.Reader.RebaseError!void { + _ = sync(r); + const gz: *GzipReader = @fieldParentPtr("reader", r); + try gz.decompress.reader.vtable.rebase(&gz.decompress.reader, capacity); + _ = sync(r); + } + + const vtable: std.Io.Reader.VTable = .{ + .stream = stream, + .discard = discard, + .readVec = readVec, + .rebase = rebase, + }; }; /// Wrap `reader` in a gzip decompressing reader and return its reader. -pub fn makeGzipReader(allocator: std.mem.Allocator, reader: *std.Io.Reader) !*std.Io.Reader { +pub fn makeGzipReader(allocator: std.mem.Allocator, reader: *std.Io.Reader, stderr_writer: *std.Io.Writer) !*std.Io.Reader { const gz = try allocator.create(GzipReader); // ponytail: leaked for program lifetime; input pointer valid only // within loadPipelineInputs scope. CLI, single-use. gz.* = .{ .decompress = std.compress.flate.Decompress.init(reader, .gzip, &gz.buffer), .buffer = undefined, + .stderr_writer = stderr_writer, + .reader = .{ + .vtable = &GzipReader.vtable, + .buffer = &gz.buffer, + .seek = 0, + .end = 0, + }, }; - return &gz.decompress.reader; + return &gz.reader; } const progress_interval = loader.progress_interval; @@ -330,7 +412,7 @@ pub fn loadPipelineInputs( const is_gz = InputFormat.isGzipExtension(file_input.path); // For .gz files: wrap in gzip decompressor, auto-detect inner format if not explicit const effective_reader: *std.Io.Reader = if (is_gz) - makeGzipReader(allocator, &file_reader.interface) catch + makeGzipReader(allocator, &file_reader.interface, stderr_writer) catch fatal("out of memory allocating gzip reader for '{s}'", stderr_writer, .csv_error, .{file_input.path}) else &file_reader.interface; @@ -353,7 +435,7 @@ pub fn loadPipelineInputs( const peeked = stdin_reader.interface.peek(2) catch null; if (peeked) |b| { if (b.len >= 2 and b[0] == 0x1f and b[1] == 0x8b) { - effective = makeGzipReader(allocator, &stdin_reader.interface) catch + effective = makeGzipReader(allocator, &stdin_reader.interface, stderr_writer) catch fatal("out of memory allocating gzip reader for stdin", stderr_writer, .csv_error, .{}); } }