Skip to content

Commit a29ff05

Browse files
dfa1claude
dfa1
and
claude
committed
feat: add parquet module for Parquet → Vortex import
Reads Parquet files via Hardwood (hardwood-core:1.0.0.CR1) and writes Vortex files via VortexWriter. Flat schemas only; supports BOOLEAN, INT32/INT64 (with IntType annotations for I8/U8/I16/U16/I32/U32/I64/U64), FLOAT, DOUBLE, and BYTE_ARRAY annotated STRING/ENUM/JSON. No Parquet exporter — Hardwood is read-only until post-1.0. Integration test round-trips delta_encoding_optional_column.parquet (apache/parquet-testing): asserts row count, column names, and spot-checks INT64 and STRING column values against the Hardwood reader. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent 4253ea5 commit a29ff05

8 files changed

Lines changed: 499 additions & 1 deletion

File tree

‎TODO.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@
5959

6060
## Tooling
6161

62-
- [ ] hardwood to convert parquet files to vortex
62+
6363
- [ ] Optional `vortex-arrow` bridge module for Arrow ecosystem interop
6464
- Primary API stays `ArrayLong`/`ArrayDouble` (zero-copy, no deps, no Unsafe)
6565
- Bridge wraps typed views into Arrow `BigIntVector`, `Float8Vector`, etc. for users who need

‎integration/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>test</scope>
1919
</dependency>
20+
<dependency>
21+
<groupId>io.github.dfa1.vortex</groupId>
22+
<artifactId>parquet</artifactId>
23+
<scope>test</scope>
24+
</dependency>
25+
<dependency>
26+
<groupId>dev.hardwood</groupId>
27+
<artifactId>hardwood-core</artifactId>
28+
<scope>test</scope>
29+
</dependency>
2030
<dependency>
2131
<groupId>io.github.dfa1.vortex</groupId>
2232
<artifactId>reader</artifactId>

‎integration/src/test/java/io/github/dfa1/vortex/integration/ParquetImportIntegrationTest.java‎

Lines changed: 176 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,176 @@
1+
package io.github.dfa1.vortex.integration;
2+
3+
import dev.hardwood.InputFile;
4+
import dev.hardwood.reader.ParquetFileReader;
5+
import dev.hardwood.reader.RowReader;
6+
import io.github.dfa1.vortex.core.DType;
7+
import io.github.dfa1.vortex.core.array.LongArray;
8+
import io.github.dfa1.vortex.core.array.VarBinArray;
9+
import io.github.dfa1.vortex.io.VortexReader;
10+
import io.github.dfa1.vortex.parquet.ParquetImporter;
11+
import io.github.dfa1.vortex.scan.ScanIterator;
12+
import io.github.dfa1.vortex.scan.ScanOptions;
13+
import io.github.dfa1.vortex.scan.ScanResult;
14+
import org.junit.jupiter.api.Test;
15+
import org.junit.jupiter.api.io.TempDir;
16+
17+
import java.net.URI;
18+
import java.nio.file.Files;
19+
import java.nio.file.Path;
20+
import java.nio.file.StandardCopyOption;
21+
import java.util.ArrayList;
22+
import java.util.List;
23+
24+
import static org.assertj.core.api.Assertions.assertThat;
25+
import static org.junit.jupiter.api.Assumptions.assumeTrue;
26+
27+
/// Round-trip: Parquet (via Hardwood) → Vortex (via ParquetImporter) → VortexReader.
28+
///
29+
/// Fixture: delta_encoding_optional_column.parquet from apache/parquet-testing.
30+
/// TPC-DS customer table, 100 rows, INT64 + STRING columns, all nullable.
31+
class ParquetImportIntegrationTest {
32+
33+
private static final String FIXTURE_URL =
34+
"https://raw.githubusercontent.com/apache/parquet-testing/master/data/"
35+
+ "delta_encoding_optional_column.parquet";
36+
37+
private static final int EXPECTED_ROWS = 100;
38+
39+
@Test
40+
void rowCountAndColumnNamesMatch(@TempDir Path tmp) throws Exception {
41+
// Given
42+
assumeNetworkAvailable();
43+
Path parquetFile = download(tmp, "delta_encoding_optional_column.parquet");
44+
Path vortexFile = tmp.resolve("out.vortex");
45+
46+
// When
47+
ParquetImporter.importParquet(parquetFile, vortexFile);
48+
49+
// Then
50+
List<String> parquetColumns = parquetColumnNames(parquetFile);
51+
try (VortexReader reader = VortexReader.open(vortexFile)) {
52+
assertThat(reader.dtype()).isInstanceOf(DType.Struct.class);
53+
DType.Struct schema = (DType.Struct) reader.dtype();
54+
assertThat(schema.fieldNames()).containsExactlyElementsOf(parquetColumns);
55+
56+
long vortexRows = countRows(reader);
57+
assertThat(vortexRows).isEqualTo(EXPECTED_ROWS);
58+
}
59+
}
60+
61+
@Test
62+
void longColumnValuesMatch(@TempDir Path tmp) throws Exception {
63+
// Given
64+
assumeNetworkAvailable();
65+
Path parquetFile = download(tmp, "delta_encoding_optional_column.parquet");
66+
Path vortexFile = tmp.resolve("out.vortex");
67+
68+
// When
69+
ParquetImporter.importParquet(parquetFile, vortexFile);
70+
71+
// Then — c_customer_sk (INT64, nullable): first 3 values are 100, 99, 98
72+
try (VortexReader reader = VortexReader.open(vortexFile);
73+
ScanIterator iter = reader.scan(ScanOptions.all())) {
74+
assertThat(iter.hasNext()).isTrue();
75+
ScanResult first = iter.next();
76+
LongArray col = first.column("c_customer_sk");
77+
assertThat(col.getLong(0)).isEqualTo(100L);
78+
assertThat(col.getLong(1)).isEqualTo(99L);
79+
assertThat(col.getLong(2)).isEqualTo(98L);
80+
}
81+
}
82+
83+
@Test
84+
void stringColumnValuesMatch(@TempDir Path tmp) throws Exception {
85+
// Given
86+
assumeNetworkAvailable();
87+
Path parquetFile = download(tmp, "delta_encoding_optional_column.parquet");
88+
Path vortexFile = tmp.resolve("out.vortex");
89+
90+
// When
91+
ParquetImporter.importParquet(parquetFile, vortexFile);
92+
93+
// Then — c_first_name (STRING, nullable): first 3 values
94+
try (VortexReader reader = VortexReader.open(vortexFile);
95+
ScanIterator iter = reader.scan(ScanOptions.all())) {
96+
assertThat(iter.hasNext()).isTrue();
97+
ScanResult first = iter.next();
98+
VarBinArray col = first.column("c_first_name");
99+
assertThat(col.getString(0)).isEqualTo("Jeannette");
100+
assertThat(col.getString(1)).isEqualTo("Austin");
101+
assertThat(col.getString(2)).isEqualTo("David");
102+
}
103+
}
104+
105+
@Test
106+
void parquetAndVortexRowCountsAreEqual(@TempDir Path tmp) throws Exception {
107+
// Given
108+
assumeNetworkAvailable();
109+
Path parquetFile = download(tmp, "delta_encoding_optional_column.parquet");
110+
Path vortexFile = tmp.resolve("out.vortex");
111+
112+
// When
113+
ParquetImporter.importParquet(parquetFile, vortexFile);
114+
115+
// Then — same row count when reading both with their native reader
116+
long parquetRows = countParquetRows(parquetFile);
117+
long vortexRows;
118+
try (VortexReader reader = VortexReader.open(vortexFile)) {
119+
vortexRows = countRows(reader);
120+
}
121+
assertThat(vortexRows).isEqualTo(parquetRows);
122+
}
123+
124+
// ── helpers ──────────────────────────────────────────────────────────────
125+
126+
private static long countRows(VortexReader reader) {
127+
long total = 0;
128+
try (ScanIterator iter = reader.scan(ScanOptions.all())) {
129+
while (iter.hasNext()) {
130+
total += iter.next().rowCount();
131+
}
132+
}
133+
return total;
134+
}
135+
136+
private static long countParquetRows(Path path) throws Exception {
137+
long total = 0;
138+
try (ParquetFileReader parquet = ParquetFileReader.open(InputFile.of(path));
139+
RowReader rowReader = parquet.rowReader()) {
140+
while (rowReader.hasNext()) {
141+
rowReader.next();
142+
total++;
143+
}
144+
}
145+
return total;
146+
}
147+
148+
private static List<String> parquetColumnNames(Path path) throws Exception {
149+
try (ParquetFileReader parquet = ParquetFileReader.open(InputFile.of(path))) {
150+
List<String> names = new ArrayList<>();
151+
parquet.getFileSchema().getColumns()
152+
.forEach(col -> names.add(col.name()));
153+
return names;
154+
}
155+
}
156+
157+
private static Path download(Path tmp, String name) throws Exception {
158+
Path cached = Path.of("/tmp/parquet-fixtures", name);
159+
if (Files.exists(cached)) {
160+
return cached;
161+
}
162+
Path dest = tmp.resolve(name);
163+
try (var in = URI.create(FIXTURE_URL).toURL().openStream()) {
164+
Files.copy(in, dest, StandardCopyOption.REPLACE_EXISTING);
165+
}
166+
return dest;
167+
}
168+
169+
private static void assumeNetworkAvailable() {
170+
try {
171+
URI.create("https://raw.githubusercontent.com").toURL().openStream().close();
172+
} catch (Exception e) {
173+
assumeTrue(false, "no network");
174+
}
175+
}
176+
}

‎parquet/pom.xml‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
3+
xmlns="http://maven.apache.org/POM/4.0.0"
4+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
5+
<modelVersion>4.0.0</modelVersion>
6+
<parent>
7+
<groupId>io.github.dfa1.vortex</groupId>
8+
<artifactId>vortex-java</artifactId>
9+
<version>0.1.0-SNAPSHOT</version>
10+
</parent>
11+
12+
<artifactId>parquet</artifactId>
13+
14+
<dependencies>
15+
<dependency>
16+
<groupId>io.github.dfa1.vortex</groupId>
17+
<artifactId>reader</artifactId>
18+
</dependency>
19+
<dependency>
20+
<groupId>io.github.dfa1.vortex</groupId>
21+
<artifactId>writer</artifactId>
22+
</dependency>
23+
<dependency>
24+
<groupId>dev.hardwood</groupId>
25+
<artifactId>hardwood-core</artifactId>
26+
</dependency>
27+
<!-- testing -->
28+
<dependency>
29+
<groupId>org.junit.jupiter</groupId>
30+
<artifactId>junit-jupiter</artifactId>
31+
<scope>test</scope>
32+
</dependency>
33+
<dependency>
34+
<groupId>org.assertj</groupId>
35+
<artifactId>assertj-core</artifactId>
36+
<scope>test</scope>
37+
</dependency>
38+
<dependency>
39+
<groupId>org.mockito</groupId>
40+
<artifactId>mockito-junit-jupiter</artifactId>
41+
<scope>test</scope>
42+
</dependency>
43+
</dependencies>
44+
</project>

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

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
package io.github.dfa1.vortex.parquet;
2+
3+
import io.github.dfa1.vortex.writer.WriteOptions;
4+
5+
/// Options controlling Parquet → Vortex import.
6+
public record ImportOptions(
7+
int chunkSize,
8+
ProgressListener progressListener,
9+
WriteOptions writeOptions
10+
) {
11+
public static ImportOptions defaults() {
12+
return new ImportOptions(65_536, null, WriteOptions.cascading(3));
13+
}
14+
15+
public ImportOptions withProgressListener(ProgressListener listener) {
16+
return new ImportOptions(chunkSize, listener, writeOptions);
17+
}
18+
19+
public ImportOptions withWriteOptions(WriteOptions options) {
20+
return new ImportOptions(chunkSize, progressListener, options);
21+
}
22+
23+
public ImportOptions withChunkSize(int size) {
24+
return new ImportOptions(size, progressListener, writeOptions);
25+
}
26+
}

0 commit comments

Comments
 (0)