Skip to content

Commit 84b4a75

Browse files
dfa1claude
dfa1
and
claude
committed
perf: add ParquetVsVortexReadBenchmark (NYC taxi 2024-01, 3 cols)
Also adds column projection to ImportOptions/ParquetImporter so unsupported columns (TIMESTAMP etc.) can be skipped when converting to Vortex. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent e585c47 commit 84b4a75

4 files changed

Lines changed: 233 additions & 6 deletions

File tree

‎parquet/src/main/java/io/github/dfa1/vortex/parquet/ImportOptions.java‎

Lines changed: 16 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,25 +2,37 @@
22

33
import io.github.dfa1.vortex.writer.WriteOptions;
44

5+
import java.util.List;
6+
57
/// Options controlling Parquet → Vortex import.
68
public record ImportOptions(
79
int chunkSize,
10+
List<String> columns,
811
ProgressListener progressListener,
912
WriteOptions writeOptions
1013
) {
1114
public static ImportOptions defaults() {
12-
return new ImportOptions(65_536, null, WriteOptions.cascading(3));
15+
return new ImportOptions(65_536, List.of(), null, WriteOptions.cascading(3));
16+
}
17+
18+
/// Restrict import to specific columns. Empty list = all columns.
19+
public ImportOptions withColumns(List<String> cols) {
20+
return new ImportOptions(chunkSize, List.copyOf(cols), progressListener, writeOptions);
21+
}
22+
23+
public boolean hasProjection() {
24+
return !columns.isEmpty();
1325
}
1426

1527
public ImportOptions withProgressListener(ProgressListener listener) {
16-
return new ImportOptions(chunkSize, listener, writeOptions);
28+
return new ImportOptions(chunkSize, columns, listener, writeOptions);
1729
}
1830

1931
public ImportOptions withWriteOptions(WriteOptions options) {
20-
return new ImportOptions(chunkSize, progressListener, options);
32+
return new ImportOptions(chunkSize, columns, progressListener, options);
2133
}
2234

2335
public ImportOptions withChunkSize(int size) {
24-
return new ImportOptions(size, progressListener, writeOptions);
36+
return new ImportOptions(size, columns, progressListener, writeOptions);
2537
}
2638
}

‎parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetImporter.java‎

Lines changed: 28 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
import dev.hardwood.metadata.RepetitionType;
66
import dev.hardwood.reader.ParquetFileReader;
77
import dev.hardwood.reader.RowReader;
8+
import dev.hardwood.schema.ColumnProjection;
89
import dev.hardwood.schema.ColumnSchema;
910
import dev.hardwood.schema.FileSchema;
1011
import io.github.dfa1.vortex.core.DType;
@@ -56,7 +57,10 @@ public static void importParquet(Path parquetPath, Path vortexPath, ImportOption
5657
"nested Parquet schemas not yet supported; schema: " + fileSchema);
5758
}
5859

59-
List<ColumnSchema> columns = fileSchema.getColumns();
60+
List<ColumnSchema> allColumns = fileSchema.getColumns();
61+
List<ColumnSchema> columns = options.hasProjection()
62+
? filterColumns(allColumns, options.columns())
63+
: allColumns;
6064
int colCount = columns.size();
6165

6266
List<String> names = new ArrayList<>(colCount);
@@ -68,11 +72,15 @@ public static void importParquet(Path parquetPath, Path vortexPath, ImportOption
6872
DType.Struct schema = new DType.Struct(names, types, false);
6973
long totalRows = parquet.getFileMetaData().numRows();
7074

75+
ColumnProjection projection = options.hasProjection()
76+
? ColumnProjection.columns(options.columns().toArray(String[]::new))
77+
: ColumnProjection.all();
78+
7179
try (FileChannel channel = FileChannel.open(
7280
vortexPath,
7381
StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING);
7482
VortexWriter writer = VortexWriter.create(channel, schema, options.writeOptions());
75-
RowReader rowReader = parquet.rowReader()) {
83+
RowReader rowReader = parquet.buildRowReader().projection(projection).build()) {
7684

7785
int chunkSize = options.chunkSize();
7886
Object[] buffers = allocateBuffers(columns, chunkSize);
@@ -206,6 +214,24 @@ private static Map<String, Object> buildChunk(List<ColumnSchema> columns, Object
206214
return chunk;
207215
}
208216

217+
private static List<ColumnSchema> filterColumns(List<ColumnSchema> all, List<String> names) {
218+
List<ColumnSchema> result = new ArrayList<>(names.size());
219+
for (String name : names) {
220+
boolean found = false;
221+
for (ColumnSchema col : all) {
222+
if (col.name().equals(name)) {
223+
result.add(col);
224+
found = true;
225+
break;
226+
}
227+
}
228+
if (!found) {
229+
throw new IllegalArgumentException("column not found in Parquet schema: " + name);
230+
}
231+
}
232+
return result;
233+
}
234+
209235
private static Object trimBuffer(Object buffer, int size) {
210236
return switch (buffer) {
211237
case boolean[] arr -> size == arr.length ? arr : Arrays.copyOf(arr, size);

‎performance/pom.xml‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,16 @@
1717
<artifactId>writer</artifactId>
1818
<scope>compile</scope>
1919
</dependency>
20+
<dependency>
21+
<groupId>io.github.dfa1.vortex</groupId>
22+
<artifactId>parquet</artifactId>
23+
<scope>compile</scope>
24+
</dependency>
25+
<dependency>
26+
<groupId>dev.hardwood</groupId>
27+
<artifactId>hardwood-core</artifactId>
28+
<scope>compile</scope>
29+
</dependency>
2030
<dependency>
2131
<groupId>io.github.dfa1.vortex</groupId>
2232
<artifactId>reader</artifactId>

‎performance/src/main/java/io/github/dfa1/vortex/performance/ParquetVsVortexReadBenchmark.java‎

Lines changed: 179 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,179 @@
1+
package io.github.dfa1.vortex.performance;
2+
3+
import dev.hardwood.InputFile;
4+
import dev.hardwood.reader.ParquetFileReader;
5+
import dev.hardwood.reader.RowReader;
6+
import dev.hardwood.schema.ColumnProjection;
7+
import io.github.dfa1.vortex.core.array.DoubleArray;
8+
import io.github.dfa1.vortex.core.array.IntArray;
9+
import io.github.dfa1.vortex.encoding.EncodingRegistry;
10+
import io.github.dfa1.vortex.io.VortexReader;
11+
import io.github.dfa1.vortex.parquet.ImportOptions;
12+
import io.github.dfa1.vortex.parquet.ParquetImporter;
13+
import io.github.dfa1.vortex.scan.ScanResult;
14+
import org.openjdk.jmh.annotations.Benchmark;
15+
import org.openjdk.jmh.annotations.BenchmarkMode;
16+
import org.openjdk.jmh.annotations.Fork;
17+
import org.openjdk.jmh.annotations.Level;
18+
import org.openjdk.jmh.annotations.Measurement;
19+
import org.openjdk.jmh.annotations.Mode;
20+
import org.openjdk.jmh.annotations.OutputTimeUnit;
21+
import org.openjdk.jmh.annotations.Scope;
22+
import org.openjdk.jmh.annotations.Setup;
23+
import org.openjdk.jmh.annotations.State;
24+
import org.openjdk.jmh.annotations.TearDown;
25+
import org.openjdk.jmh.annotations.Warmup;
26+
27+
import java.io.IOException;
28+
import java.io.InputStream;
29+
import java.net.URI;
30+
import java.net.http.HttpClient;
31+
import java.net.http.HttpRequest;
32+
import java.net.http.HttpResponse;
33+
import java.nio.file.Files;
34+
import java.nio.file.Path;
35+
import java.util.List;
36+
import java.util.concurrent.TimeUnit;
37+
38+
/// Benchmark: Hardwood Parquet read vs Java Vortex read on the same real-world dataset.
39+
///
40+
/// Dataset: NYC Yellow Taxi 2024-01 (~3M rows), three columns:
41+
/// trip_distance (DOUBLE), fare_amount (DOUBLE), PULocationID (INT32)
42+
///
43+
/// Setup downloads the Parquet file once per trial (cached in /tmp), converts it to
44+
/// Vortex with column projection, then both benchmarks scan and sum trip_distance.
45+
///
46+
/// Override the Parquet source with: -Dbench.parquet=/path/to/file.parquet
47+
///
48+
/// Run: java -jar performance/target/benchmarks.jar ParquetVsVortexReadBenchmark
49+
@State(Scope.Benchmark)
50+
@BenchmarkMode(Mode.Throughput)
51+
@OutputTimeUnit(TimeUnit.SECONDS)
52+
@Warmup(iterations = 3, time = 3)
53+
@Measurement(iterations = 5, time = 5)
54+
@Fork(value = 1, jvmArgsAppend = {
55+
"--add-opens", "java.base/java.nio=ALL-UNNAMED",
56+
"--enable-native-access=ALL-UNNAMED",
57+
"--sun-misc-unsafe-memory-access=allow"
58+
})
59+
public class ParquetVsVortexReadBenchmark {
60+
61+
private static final String PARQUET_URL =
62+
"https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2024-01.parquet";
63+
private static final Path CACHE_PATH =
64+
Path.of(System.getProperty("java.io.tmpdir"), "yellow_tripdata_2024-01.parquet");
65+
private static final List<String> COLUMNS = List.of("trip_distance", "fare_amount", "PULocationID");
66+
67+
private static final Object SETUP_LOCK = new Object();
68+
private static Path sharedParquetFile;
69+
private static Path sharedVortexFile;
70+
private static int refCount;
71+
72+
private Path parquetFile;
73+
private Path vortexFile;
74+
private EncodingRegistry registry;
75+
76+
@Setup(Level.Trial)
77+
public void setup() throws Exception {
78+
registry = EncodingRegistry.loadAll();
79+
synchronized (SETUP_LOCK) {
80+
if (sharedParquetFile == null) {
81+
String override = System.getProperty("bench.parquet");
82+
if (override != null && !override.isEmpty()) {
83+
sharedParquetFile = Path.of(override);
84+
System.out.printf("[ParquetVsVortexReadBenchmark] using override: %s%n", sharedParquetFile);
85+
} else {
86+
sharedParquetFile = downloadIfAbsent(CACHE_PATH, PARQUET_URL);
87+
System.out.printf("[ParquetVsVortexReadBenchmark] parquet: %s (%.1f MB)%n",
88+
sharedParquetFile, Files.size(sharedParquetFile) / 1_048_576.0);
89+
}
90+
sharedVortexFile = Files.createTempFile("taxi-bench", ".vortex");
91+
System.out.print("[ParquetVsVortexReadBenchmark] converting to vortex...");
92+
ImportOptions opts = ImportOptions.defaults().withColumns(COLUMNS);
93+
ParquetImporter.importParquet(sharedParquetFile, sharedVortexFile, opts);
94+
System.out.printf(" done (%.1f MB)%n", Files.size(sharedVortexFile) / 1_048_576.0);
95+
}
96+
refCount++;
97+
parquetFile = sharedParquetFile;
98+
vortexFile = sharedVortexFile;
99+
}
100+
}
101+
102+
@TearDown(Level.Trial)
103+
public void tearDown() throws IOException {
104+
synchronized (SETUP_LOCK) {
105+
refCount--;
106+
if (refCount == 0) {
107+
Files.deleteIfExists(sharedVortexFile);
108+
sharedVortexFile = null;
109+
sharedParquetFile = null;
110+
}
111+
}
112+
}
113+
114+
/// Hardwood: scan all rows, sum trip_distance.
115+
@Benchmark
116+
public double parquetRead() throws IOException {
117+
double sum = 0.0;
118+
ColumnProjection projection = ColumnProjection.columns("trip_distance", "fare_amount", "PULocationID");
119+
try (ParquetFileReader reader = ParquetFileReader.open(InputFile.of(parquetFile));
120+
RowReader rows = reader.buildRowReader().projection(projection).build()) {
121+
while (rows.hasNext()) {
122+
rows.next();
123+
sum += rows.getDouble("trip_distance");
124+
}
125+
}
126+
return sum;
127+
}
128+
129+
/// Vortex: scan all rows, sum trip_distance via fold.
130+
@Benchmark
131+
public double vortexRead() throws IOException {
132+
double sum = 0.0;
133+
try (VortexReader vr = VortexReader.open(vortexFile, registry)) {
134+
var iter = vr.scan(io.github.dfa1.vortex.scan.ScanOptions.columns("trip_distance"));
135+
while (iter.hasNext()) {
136+
ScanResult r = iter.next();
137+
DoubleArray col = r.column("trip_distance");
138+
sum += col.fold(0.0, Double::sum);
139+
}
140+
}
141+
return sum;
142+
}
143+
144+
/// Vortex: scan fare_amount + PULocationID, sum both (exercises multi-column projection).
145+
@Benchmark
146+
public double vortexReadMultiColumn() throws IOException {
147+
double fareSum = 0.0;
148+
long idSum = 0L;
149+
try (VortexReader vr = VortexReader.open(vortexFile, registry)) {
150+
var iter = vr.scan(io.github.dfa1.vortex.scan.ScanOptions.columns("fare_amount", "PULocationID"));
151+
while (iter.hasNext()) {
152+
ScanResult r = iter.next();
153+
DoubleArray fare = r.column("fare_amount");
154+
IntArray loc = r.column("PULocationID");
155+
fareSum += fare.fold(0.0, Double::sum);
156+
idSum += loc.fold(0, Integer::sum);
157+
}
158+
}
159+
return fareSum + idSum;
160+
}
161+
162+
private static Path downloadIfAbsent(Path dest, String url) throws Exception {
163+
if (Files.exists(dest)) {
164+
System.out.printf("[ParquetVsVortexReadBenchmark] cache hit: %s%n", dest);
165+
return dest;
166+
}
167+
System.out.printf("[ParquetVsVortexReadBenchmark] downloading %s...%n", url);
168+
HttpClient client = HttpClient.newBuilder().followRedirects(HttpClient.Redirect.NORMAL).build();
169+
HttpRequest request = HttpRequest.newBuilder().uri(URI.create(url)).build();
170+
HttpResponse<InputStream> response = client.send(request, HttpResponse.BodyHandlers.ofInputStream());
171+
if (response.statusCode() != 200) {
172+
throw new IOException("HTTP " + response.statusCode() + " for " + url);
173+
}
174+
try (InputStream in = response.body()) {
175+
Files.copy(in, dest);
176+
}
177+
return dest;
178+
}
179+
}

0 commit comments

Comments
 (0)