Skip to content

Commit 696552b

Browse files
dfa1claude
dfa1
and
claude
committed
feat(cli): add filter subcommand with zone-map pruning + row-level predicate
Parses "col op val" (op: >, >=, <, <=, =, ==). Zone-map RowFilter skips whole chunks; RowPredicate corrects row-level results within surviving chunks. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent ea69d8c commit 696552b

5 files changed

Lines changed: 196 additions & 7 deletions

File tree

‎TODO.md‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@
3030

3131
## Tooling
3232

33-
- [ ] `filter` subcommand — push predicates into scan, skip chunks via zone-maps
3433
- [ ] `head` subcommand — first N rows via `ScanOptions.limit`
3534
- [ ] String min/max in stats — requires utf8 scalar encoding in the proto
3635

‎cli/src/main/java/io/github/dfa1/vortex/cli/FilterCommand.java‎

Lines changed: 169 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,169 @@
1+
package io.github.dfa1.vortex.cli;
2+
3+
import io.github.dfa1.vortex.core.array.BoolArray;
4+
import io.github.dfa1.vortex.core.array.ByteArray;
5+
import io.github.dfa1.vortex.core.array.DoubleArray;
6+
import io.github.dfa1.vortex.core.array.FloatArray;
7+
import io.github.dfa1.vortex.core.array.IntArray;
8+
import io.github.dfa1.vortex.core.array.LongArray;
9+
import io.github.dfa1.vortex.core.array.ShortArray;
10+
import io.github.dfa1.vortex.core.array.VarBinArray;
11+
import io.github.dfa1.vortex.core.array.Array;
12+
import io.github.dfa1.vortex.csv.CsvExporter;
13+
import io.github.dfa1.vortex.csv.ExportOptions;
14+
import io.github.dfa1.vortex.csv.RowPredicate;
15+
import io.github.dfa1.vortex.scan.RowFilter;
16+
import io.github.dfa1.vortex.scan.ScanOptions;
17+
18+
import java.io.IOException;
19+
import java.io.OutputStreamWriter;
20+
import java.io.Writer;
21+
import java.nio.charset.StandardCharsets;
22+
import java.nio.file.Files;
23+
import java.nio.file.Path;
24+
import java.util.Arrays;
25+
import java.util.List;
26+
import java.util.regex.Matcher;
27+
import java.util.regex.Pattern;
28+
29+
final class FilterCommand {
30+
31+
private FilterCommand() {
32+
}
33+
34+
private enum Op { GT, GTE, LT, LTE, EQ }
35+
36+
private record ParsedFilter(String column, Op op, Comparable<?> value) {}
37+
38+
private static final Pattern EXPR = Pattern.compile(
39+
"^(\\w[\\w.]*?)\\s*(>=|<=|==|>|<|=)\\s*(.+)$");
40+
41+
static int run(String[] args) {
42+
if (args.length < 3) {
43+
System.err.println("usage: filter <file.vortex> <expr> (e.g. \"price >= 100\")");
44+
return 1;
45+
}
46+
Path path = Path.of(args[1]);
47+
if (!Files.exists(path)) {
48+
System.err.println("file not found: " + path);
49+
return 2;
50+
}
51+
String expr = String.join(" ", Arrays.asList(args).subList(2, args.length));
52+
ParsedFilter pf;
53+
try {
54+
pf = parseFilter(expr);
55+
} catch (IllegalArgumentException e) {
56+
System.err.println("error: " + e.getMessage());
57+
return 1;
58+
}
59+
ScanOptions scanOptions = new ScanOptions(List.of(), toRowFilter(pf), ScanOptions.NO_LIMIT);
60+
RowPredicate rowPred = toRowPredicate(pf);
61+
try {
62+
Writer stdout = new OutputStreamWriter(System.out, StandardCharsets.UTF_8);
63+
CsvExporter.exportCsvFiltered(path, stdout, ExportOptions.defaults(), scanOptions, rowPred);
64+
stdout.flush();
65+
return 0;
66+
} catch (IOException e) {
67+
System.err.println("error: " + e.getMessage());
68+
return 3;
69+
}
70+
}
71+
72+
private static ParsedFilter parseFilter(String expr) {
73+
Matcher m = EXPR.matcher(expr.trim());
74+
if (!m.matches()) {
75+
throw new IllegalArgumentException("invalid filter expression: \"" + expr
76+
+ "\" expected: col op value (op: >, >=, <, <=, =, ==)");
77+
}
78+
String col = m.group(1);
79+
Op op = switch (m.group(2)) {
80+
case ">" -> Op.GT;
81+
case ">=" -> Op.GTE;
82+
case "<" -> Op.LT;
83+
case "<=" -> Op.LTE;
84+
case "=", "==" -> Op.EQ;
85+
default -> throw new IllegalArgumentException("unknown operator: " + m.group(2));
86+
};
87+
Comparable<?> value = parseValue(m.group(3).trim());
88+
return new ParsedFilter(col, op, value);
89+
}
90+
91+
private static Comparable<?> parseValue(String raw) {
92+
try {
93+
return Long.parseLong(raw);
94+
} catch (NumberFormatException ignored) {
95+
}
96+
try {
97+
return Double.parseDouble(raw);
98+
} catch (NumberFormatException ignored) {
99+
}
100+
if ("true".equalsIgnoreCase(raw)) {
101+
return Boolean.TRUE;
102+
}
103+
if ("false".equalsIgnoreCase(raw)) {
104+
return Boolean.FALSE;
105+
}
106+
return raw;
107+
}
108+
109+
/// Zone-map RowFilter for chunk pruning. Conservative for strict ops (> / <):
110+
/// uses Gte/Lte which may keep one extra chunk at the boundary, but row-level
111+
/// predicate corrects that.
112+
private static RowFilter toRowFilter(ParsedFilter pf) {
113+
return switch (pf.op()) {
114+
case GT, GTE -> RowFilter.gte(pf.column(), pf.value());
115+
case LT, LTE -> RowFilter.lte(pf.column(), pf.value());
116+
case EQ -> RowFilter.eq(pf.column(), pf.value());
117+
};
118+
}
119+
120+
private static RowPredicate toRowPredicate(ParsedFilter pf) {
121+
return (chunk, rowIdx) -> {
122+
Array arr = chunk.column(pf.column());
123+
if (arr == null) {
124+
return false;
125+
}
126+
int cmp = compareValue(arr, rowIdx, pf.value());
127+
return switch (pf.op()) {
128+
case GT -> cmp > 0;
129+
case GTE -> cmp >= 0;
130+
case LT -> cmp < 0;
131+
case LTE -> cmp <= 0;
132+
case EQ -> cmp == 0;
133+
};
134+
};
135+
}
136+
137+
@SuppressWarnings("unchecked")
138+
private static int compareValue(Array arr, long rowIdx, Comparable<?> value) {
139+
return switch (arr) {
140+
case LongArray la -> compareNumeric(la.getLong(rowIdx), value);
141+
case IntArray ia -> compareNumeric(ia.getInt(rowIdx), value);
142+
case ShortArray sa -> compareNumeric(sa.getShort(rowIdx), value);
143+
case ByteArray ba -> compareNumeric(ba.getByte(rowIdx), value);
144+
case DoubleArray da -> compareDouble(da.getDouble(rowIdx), value);
145+
case FloatArray fa -> compareDouble(fa.getFloat(rowIdx), value);
146+
case BoolArray ba -> Boolean.compare(ba.getBoolean(rowIdx), (Boolean) value);
147+
case VarBinArray va -> {
148+
String v = new String(va.getBytes(rowIdx), StandardCharsets.UTF_8);
149+
yield v.compareTo((String) value);
150+
}
151+
default -> throw new IllegalArgumentException(
152+
"filter not supported for column type: " + arr.getClass().getSimpleName());
153+
};
154+
}
155+
156+
private static int compareNumeric(long colVal, Comparable<?> value) {
157+
if (value instanceof Long l) {
158+
return Long.compare(colVal, l);
159+
}
160+
return Double.compare(colVal, (Double) value);
161+
}
162+
163+
private static int compareDouble(double colVal, Comparable<?> value) {
164+
if (value instanceof Long l) {
165+
return Double.compare(colVal, l);
166+
}
167+
return Double.compare(colVal, (Double) value);
168+
}
169+
}

‎cli/src/main/java/io/github/dfa1/vortex/cli/VortexCli.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ public static void main(String[] args) {
2323
case "count" -> CountCommand.run(args);
2424
case "select" -> SelectCommand.run(args);
2525
case "stats" -> StatsCommand.run(args);
26+
case "filter" -> FilterCommand.run(args);
2627
default -> {
2728
System.err.println("unknown subcommand: " + args[0]);
2829
printUsage(System.err);
@@ -41,5 +42,6 @@ static void printUsage(PrintStream out) {
4142
out.println(" count <file.vortex> print row count");
4243
out.println(" select <file.vortex> <col> [...] project columns to CSV on stdout");
4344
out.println(" stats <file.vortex> print per-column min/max statistics");
45+
out.println(" filter <file.vortex> <expr> filter rows to CSV (e.g. \"price >= 100\")");
4446
}
4547
}

‎csv/src/main/java/io/github/dfa1/vortex/csv/CsvExporter.java‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -42,12 +42,19 @@ public static void exportCsv(Path vortexPath, Path csvPath, ExportOptions option
4242
CsvWriter csvWriter = CsvWriter.builder()
4343
.fieldSeparator(options.delimiter())
4444
.build(csvPath)) {
45-
export(reader, csvWriter, options);
45+
export(reader, csvWriter, options, ScanOptions.all(), (chunk, rowIdx) -> true);
4646
}
4747
}
4848

4949
/// Export to a caller-owned [Writer]; the writer is flushed but not closed.
5050
public static void exportCsv(Path vortexPath, Writer out, ExportOptions options) throws IOException {
51+
exportCsvFiltered(vortexPath, out, options, ScanOptions.all(), (chunk, rowIdx) -> true);
52+
}
53+
54+
/// Like [#exportCsv(Path, Writer, ExportOptions)] but with zone-map chunk pruning
55+
/// ([ScanOptions#rowFilter]) and a row-level predicate applied after decoding.
56+
public static void exportCsvFiltered(Path vortexPath, Writer out, ExportOptions options,
57+
ScanOptions scanOptions, RowPredicate predicate) throws IOException {
5158
Writer shielded = new FilterWriter(out) {
5259
@Override
5360
public void close() {
@@ -58,11 +65,12 @@ public void close() {
5865
CsvWriter csvWriter = CsvWriter.builder()
5966
.fieldSeparator(options.delimiter())
6067
.build(shielded)) {
61-
export(reader, csvWriter, options);
68+
export(reader, csvWriter, options, scanOptions, predicate);
6269
}
6370
}
6471

65-
private static void export(VortexReader reader, CsvWriter csvWriter, ExportOptions options) throws IOException {
72+
private static void export(VortexReader reader, CsvWriter csvWriter, ExportOptions options,
73+
ScanOptions scanOptions, RowPredicate predicate) throws IOException {
6674
if (!(reader.dtype() instanceof DType.Struct schema)) {
6775
throw new VortexException("only struct root dtype supported for CSV export");
6876
}
@@ -73,9 +81,6 @@ private static void export(VortexReader reader, CsvWriter csvWriter, ExportOptio
7381
csvWriter.writeRecord(colNames);
7482
}
7583

76-
ScanOptions scanOptions = options.hasProjection()
77-
? new ScanOptions(colNames, null, ScanOptions.NO_LIMIT)
78-
: ScanOptions.all();
7984
String[] row = new String[colCount];
8085
try (ScanIterator iter = reader.scan(scanOptions)) {
8186
while (iter.hasNext()) {
@@ -86,6 +91,9 @@ private static void export(VortexReader reader, CsvWriter csvWriter, ExportOptio
8691
}
8792
long rowCount = chunk.rowCount();
8893
for (long r = 0; r < rowCount; r++) {
94+
if (!predicate.test(chunk, r)) {
95+
continue;
96+
}
8997
for (int c = 0; c < colCount; c++) {
9098
row[c] = cellValue(arrays[c], r);
9199
}

‎csv/src/main/java/io/github/dfa1/vortex/csv/RowPredicate.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
package io.github.dfa1.vortex.csv;
2+
3+
import io.github.dfa1.vortex.scan.ScanResult;
4+
5+
/// Row-level predicate evaluated against decoded chunk data.
6+
/// Used in conjunction with zone-map pruning: zone-maps skip whole chunks,
7+
/// this predicate filters individual rows within surviving chunks.
8+
@FunctionalInterface
9+
public interface RowPredicate {
10+
boolean test(ScanResult chunk, long rowIndex);
11+
}

0 commit comments

Comments
 (0)