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..f1ea831 100644 --- a/build.zig +++ b/build.zig @@ -3708,6 +3708,84 @@ 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 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", + \\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); + // ─── --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..c4ee930 100644 --- a/src/args.zig +++ b/src/args.zig @@ -1003,7 +1003,11 @@ 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 = 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/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..03265c7 100644 --- a/src/main.zig +++ b/src/main.zig @@ -114,6 +114,108 @@ fn writeStreaming( try out_writer.end(writer); } +/// 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, 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.reader; +} + const progress_interval = loader.progress_interval; const fatal = sqlite_mod.fatal; @@ -307,7 +409,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, stderr_writer) 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 +429,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, stderr_writer) 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 0000000..c2d22ba Binary files /dev/null and b/tests/fixtures/sample.csv.gz differ diff --git a/tests/fixtures/sample.csv.gz_truncated b/tests/fixtures/sample.csv.gz_truncated new file mode 100644 index 0000000..e08934e Binary files /dev/null and b/tests/fixtures/sample.csv.gz_truncated differ