Skip to content

Commit ff2ff99

Browse files
dfa1claude
dfa1
and
claude
committed
test(parquet): unit-test ParquetImporter with mapping tests + offline fixture
The importer was only exercised by network-downloading integration tests. Add fast, offline coverage in the parquet module itself: - Deterministic type-mapping tests (mapDType / filterColumns made package-private) covering every supported physical type, all INT32 widths, signed/unsigned INT64, timestamp units, string-like BYTE_ARRAY annotations, and the unsupported-type / unknown-column error paths. - A committed 11 KB fixture (delta_encoding_optional_column.parquet from apache/parquet-testing, Apache-2.0) drives end-to-end import → VortexReader assertions: schema, row count, column values, projection, and multi-chunk split. Adds vortex-reader + aircompressor-v3 as test-scope deps so the imported Vortex output can be decoded in-module. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1 parent a326fbb commit ff2ff99

4 files changed

Lines changed: 310 additions & 2 deletions

File tree

‎parquet/pom.xml‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,18 @@
2929
<artifactId>zstd-jni</artifactId>
3030
</dependency>
3131
<!-- testing -->
32+
<!-- project-internal test deps first: reader verifies the imported Vortex output. -->
33+
<dependency>
34+
<groupId>io.github.dfa1.vortex</groupId>
35+
<artifactId>vortex-reader</artifactId>
36+
<scope>test</scope>
37+
</dependency>
38+
<!-- reader declares aircompressor-v3 optional; pull it in so the test can decode ZSTD chunks -->
39+
<dependency>
40+
<groupId>io.airlift</groupId>
41+
<artifactId>aircompressor-v3</artifactId>
42+
<scope>test</scope>
43+
</dependency>
3244
<dependency>
3345
<groupId>org.junit.jupiter</groupId>
3446
<artifactId>junit-jupiter</artifactId>

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,7 @@ public static void importParquet(Path parquetPath, Path vortexPath, ImportOption
115115
}
116116
}
117117

118-
private static DType mapDType(ColumnSchema col) {
118+
static DType mapDType(ColumnSchema col) {
119119
boolean nullable = col.repetitionType() == RepetitionType.OPTIONAL;
120120
return switch (col.type()) {
121121
case BOOLEAN -> new DType.Bool(nullable);
@@ -239,7 +239,7 @@ private static Map<String, Object> buildChunk(List<ColumnSchema> columns, List<D
239239
return chunk;
240240
}
241241

242-
private static List<ColumnSchema> filterColumns(List<ColumnSchema> all, List<String> names) {
242+
static List<ColumnSchema> filterColumns(List<ColumnSchema> all, List<String> names) {
243243
List<ColumnSchema> result = new ArrayList<>(names.size());
244244
for (String name : names) {
245245
boolean found = false;

‎parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetImporterTest.java‎

Lines changed: 296 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,296 @@
1+
package io.github.dfa1.vortex.parquet;
2+
3+
import dev.hardwood.metadata.FieldPath;
4+
import dev.hardwood.metadata.LogicalType;
5+
import dev.hardwood.metadata.PhysicalType;
6+
import dev.hardwood.metadata.RepetitionType;
7+
import dev.hardwood.schema.ColumnSchema;
8+
import io.github.dfa1.vortex.core.DType;
9+
import io.github.dfa1.vortex.core.PType;
10+
import io.github.dfa1.vortex.reader.Chunk;
11+
import io.github.dfa1.vortex.reader.ScanIterator;
12+
import io.github.dfa1.vortex.reader.ScanOptions;
13+
import io.github.dfa1.vortex.reader.VortexReader;
14+
import io.github.dfa1.vortex.reader.array.LongArray;
15+
import io.github.dfa1.vortex.reader.array.VarBinArray;
16+
import org.junit.jupiter.api.Nested;
17+
import org.junit.jupiter.api.Test;
18+
import org.junit.jupiter.api.io.TempDir;
19+
import org.junit.jupiter.params.ParameterizedTest;
20+
import org.junit.jupiter.params.provider.CsvSource;
21+
22+
import java.nio.file.Path;
23+
import java.util.List;
24+
import java.util.concurrent.atomic.AtomicLong;
25+
26+
import static org.assertj.core.api.Assertions.assertThat;
27+
import static org.assertj.core.api.Assertions.assertThatThrownBy;
28+
29+
class ParquetImporterTest {
30+
31+
private static ColumnSchema col(String name, PhysicalType type, RepetitionType rep, LogicalType logical) {
32+
return new ColumnSchema(FieldPath.of(name), type, rep, null, 0, 0, 0, logical);
33+
}
34+
35+
@Nested
36+
class TypeMapping {
37+
38+
@Test
39+
void boolean_mapsToBool_carryingNullability() {
40+
// Given / When / Then — REQUIRED is non-null, OPTIONAL is nullable
41+
assertThat(ParquetImporter.mapDType(col("b", PhysicalType.BOOLEAN, RepetitionType.REQUIRED, null)))
42+
.isEqualTo(new DType.Bool(false));
43+
assertThat(ParquetImporter.mapDType(col("b", PhysicalType.BOOLEAN, RepetitionType.OPTIONAL, null)))
44+
.isEqualTo(new DType.Bool(true));
45+
}
46+
47+
@Test
48+
void int32_withoutAnnotation_mapsToI32() {
49+
// When
50+
DType result = ParquetImporter.mapDType(col("i", PhysicalType.INT32, RepetitionType.REQUIRED, null));
51+
52+
// Then
53+
assertThat(result).isEqualTo(new DType.Primitive(PType.I32, false));
54+
}
55+
56+
@ParameterizedTest
57+
@CsvSource({
58+
"8, true, I8",
59+
"8, false, U8",
60+
"16, true, I16",
61+
"16, false, U16",
62+
"32, true, I32",
63+
"32, false, U32",
64+
})
65+
void int32_withIntAnnotation_mapsToSizedPType(int bitWidth, boolean signed, PType expected) {
66+
// Given — INT32 carrying a width/sign annotation selects the narrow PType
67+
ColumnSchema schema = col("i", PhysicalType.INT32, RepetitionType.REQUIRED,
68+
new LogicalType.IntType(bitWidth, signed));
69+
70+
// When
71+
DType result = ParquetImporter.mapDType(schema);
72+
73+
// Then
74+
assertThat(result).isEqualTo(new DType.Primitive(expected, false));
75+
}
76+
77+
@Test
78+
void int64_signedAndUnsigned_mapToI64AndU64() {
79+
// Given / When / Then
80+
assertThat(ParquetImporter.mapDType(col("l", PhysicalType.INT64, RepetitionType.REQUIRED, null)))
81+
.isEqualTo(new DType.Primitive(PType.I64, false));
82+
assertThat(ParquetImporter.mapDType(col("l", PhysicalType.INT64, RepetitionType.REQUIRED,
83+
new LogicalType.IntType(64, true)))).isEqualTo(new DType.Primitive(PType.I64, false));
84+
assertThat(ParquetImporter.mapDType(col("l", PhysicalType.INT64, RepetitionType.REQUIRED,
85+
new LogicalType.IntType(64, false)))).isEqualTo(new DType.Primitive(PType.U64, false));
86+
}
87+
88+
@ParameterizedTest
89+
@CsvSource({"MILLIS", "MICROS", "NANOS"})
90+
void int64_timestamp_mapsToTimestampExtensionOverI64(LogicalType.TimeUnit unit) {
91+
// Given — a TIMESTAMP-annotated INT64
92+
ColumnSchema schema = col("ts", PhysicalType.INT64, RepetitionType.OPTIONAL,
93+
new LogicalType.TimestampType(true, unit));
94+
95+
// When
96+
DType result = ParquetImporter.mapDType(schema);
97+
98+
// Then — vortex.timestamp extension over nullable I64 storage
99+
assertThat(result).isInstanceOf(DType.Extension.class);
100+
DType.Extension ext = (DType.Extension) result;
101+
assertThat(ext.extensionId()).isEqualTo("vortex.timestamp");
102+
assertThat(ext.storageDType()).isEqualTo(new DType.Primitive(PType.I64, true));
103+
assertThat(ext.nullable()).isTrue();
104+
}
105+
106+
@Test
107+
void float_and_double_mapToF32AndF64() {
108+
// Given / When / Then
109+
assertThat(ParquetImporter.mapDType(col("f", PhysicalType.FLOAT, RepetitionType.REQUIRED, null)))
110+
.isEqualTo(new DType.Primitive(PType.F32, false));
111+
assertThat(ParquetImporter.mapDType(col("d", PhysicalType.DOUBLE, RepetitionType.REQUIRED, null)))
112+
.isEqualTo(new DType.Primitive(PType.F64, false));
113+
}
114+
115+
@Test
116+
void byteArray_stringLikeAnnotations_mapToUtf8() {
117+
// Given — STRING / ENUM / JSON are all logical strings
118+
for (LogicalType logical : List.of(new LogicalType.StringType(),
119+
new LogicalType.EnumType(), new LogicalType.JsonType())) {
120+
// When
121+
DType result = ParquetImporter.mapDType(
122+
col("s", PhysicalType.BYTE_ARRAY, RepetitionType.OPTIONAL, logical));
123+
124+
// Then
125+
assertThat(result).as("logical %s", logical).isEqualTo(new DType.Utf8(true));
126+
}
127+
}
128+
129+
@Test
130+
void byteArray_withoutStringAnnotation_throws() {
131+
// Given — raw BYTE_ARRAY with no string logical type is unsupported
132+
ColumnSchema schema = col("blob", PhysicalType.BYTE_ARRAY, RepetitionType.REQUIRED, null);
133+
134+
// When / Then
135+
assertThatThrownBy(() -> ParquetImporter.mapDType(schema))
136+
.isInstanceOf(UnsupportedOperationException.class)
137+
.hasMessageContaining("blob");
138+
}
139+
140+
@ParameterizedTest
141+
@CsvSource({"INT96", "FIXED_LEN_BYTE_ARRAY"})
142+
void unsupportedPhysicalType_throws(PhysicalType type) {
143+
// Given
144+
ColumnSchema schema = col("x", type, RepetitionType.REQUIRED, null);
145+
146+
// When / Then
147+
assertThatThrownBy(() -> ParquetImporter.mapDType(schema))
148+
.isInstanceOf(UnsupportedOperationException.class)
149+
.hasMessageContaining("unsupported Parquet physical type");
150+
}
151+
}
152+
153+
@Nested
154+
class FilterColumns {
155+
156+
@Test
157+
void keepsRequestedColumnsInRequestedOrder() {
158+
// Given — schema a, b, c; request c, a
159+
List<ColumnSchema> all = List.of(
160+
col("a", PhysicalType.INT32, RepetitionType.REQUIRED, null),
161+
col("b", PhysicalType.INT32, RepetitionType.REQUIRED, null),
162+
col("c", PhysicalType.INT32, RepetitionType.REQUIRED, null));
163+
164+
// When
165+
List<ColumnSchema> result = ParquetImporter.filterColumns(all, List.of("c", "a"));
166+
167+
// Then — projection order wins over schema order
168+
assertThat(result).extracting(ColumnSchema::name).containsExactly("c", "a");
169+
}
170+
171+
@Test
172+
void unknownColumn_throws() {
173+
// Given
174+
List<ColumnSchema> all = List.of(col("a", PhysicalType.INT32, RepetitionType.REQUIRED, null));
175+
176+
// When / Then
177+
assertThatThrownBy(() -> ParquetImporter.filterColumns(all, List.of("missing")))
178+
.isInstanceOf(IllegalArgumentException.class)
179+
.hasMessageContaining("missing");
180+
}
181+
}
182+
183+
@Nested
184+
class Import {
185+
186+
@Test
187+
void importsFixture_schemaAndRowCount(@TempDir Path tmp) throws Exception {
188+
// Given — 100-row TPC-DS customer fixture (INT64 + STRING, all nullable)
189+
Path vortex = tmp.resolve("out.vortex");
190+
191+
// When
192+
ParquetImporter.importParquet(fixture(), vortex);
193+
194+
// Then
195+
try (VortexReader reader = VortexReader.open(vortex)) {
196+
assertThat(reader.dtype()).isInstanceOf(DType.Struct.class);
197+
DType.Struct schema = (DType.Struct) reader.dtype();
198+
assertThat(schema.fieldNames()).contains("c_customer_sk", "c_first_name");
199+
assertThat(countRows(reader)).isEqualTo(100L);
200+
}
201+
}
202+
203+
@Test
204+
void importsFixture_columnValuesRoundTrip(@TempDir Path tmp) throws Exception {
205+
// Given
206+
Path vortex = tmp.resolve("out.vortex");
207+
208+
// When
209+
ParquetImporter.importParquet(fixture(), vortex);
210+
211+
// Then — known first three values of each column
212+
try (VortexReader reader = VortexReader.open(vortex);
213+
ScanIterator iter = reader.scan(ScanOptions.all())) {
214+
assertThat(iter.hasNext()).isTrue();
215+
try (Chunk first = iter.next()) {
216+
LongArray sk = first.column("c_customer_sk");
217+
assertThat(sk.getLong(0)).isEqualTo(100L);
218+
assertThat(sk.getLong(1)).isEqualTo(99L);
219+
assertThat(sk.getLong(2)).isEqualTo(98L);
220+
221+
VarBinArray name = first.column("c_first_name");
222+
assertThat(name.getString(0)).isEqualTo("Jeannette");
223+
assertThat(name.getString(1)).isEqualTo("Austin");
224+
assertThat(name.getString(2)).isEqualTo("David");
225+
}
226+
}
227+
}
228+
229+
@Test
230+
void projection_importsOnlyRequestedColumns(@TempDir Path tmp) throws Exception {
231+
// Given — project a single column out of the fixture
232+
Path vortex = tmp.resolve("out.vortex");
233+
ImportOptions options = ImportOptions.defaults().withColumns(List.of("c_customer_sk"));
234+
235+
// When
236+
ParquetImporter.importParquet(fixture(), vortex, options);
237+
238+
// Then — only the projected column survives
239+
try (VortexReader reader = VortexReader.open(vortex)) {
240+
DType.Struct schema = (DType.Struct) reader.dtype();
241+
assertThat(schema.fieldNames()).containsExactly("c_customer_sk");
242+
assertThat(countRows(reader)).isEqualTo(100L);
243+
}
244+
}
245+
246+
@Test
247+
void smallChunkSize_splitsIntoMultipleChunks(@TempDir Path tmp) throws Exception {
248+
// Given — chunk size 30 forces 4 chunks over 100 rows (exercises trim + chunk flush)
249+
Path vortex = tmp.resolve("out.vortex");
250+
ImportOptions options = ImportOptions.defaults().withChunkSize(30);
251+
252+
// When
253+
ParquetImporter.importParquet(fixture(), vortex, options);
254+
255+
// Then — row count is preserved across the chunk boundaries
256+
try (VortexReader reader = VortexReader.open(vortex);
257+
ScanIterator iter = reader.scan(ScanOptions.all())) {
258+
long chunks = 0;
259+
long rows = 0;
260+
while (iter.hasNext()) {
261+
try (Chunk c = iter.next()) {
262+
chunks++;
263+
rows += c.rowCount();
264+
}
265+
}
266+
assertThat(rows).isEqualTo(100L);
267+
assertThat(chunks).isGreaterThan(1L);
268+
}
269+
}
270+
271+
@Test
272+
void projection_unknownColumn_throws(@TempDir Path tmp) {
273+
// Given
274+
Path vortex = tmp.resolve("out.vortex");
275+
ImportOptions options = ImportOptions.defaults().withColumns(List.of("does_not_exist"));
276+
277+
// When / Then
278+
assertThatThrownBy(() -> ParquetImporter.importParquet(fixture(), vortex, options))
279+
.isInstanceOf(IllegalArgumentException.class)
280+
.hasMessageContaining("does_not_exist");
281+
}
282+
}
283+
284+
private static Path fixture() throws Exception {
285+
return Path.of(ParquetImporterTest.class
286+
.getResource("/fixtures/delta_encoding_optional_column.parquet").toURI());
287+
}
288+
289+
private static long countRows(VortexReader reader) {
290+
AtomicLong total = new AtomicLong();
291+
try (ScanIterator iter = reader.scan(ScanOptions.all())) {
292+
iter.forEachRemaining(c -> total.addAndGet(c.rowCount()));
293+
}
294+
return total.get();
295+
}
296+
}

‎parquet/src/test/resources/fixtures/delta_encoding_optional_column.parquet‎

11.3 KB
Binary file not shown.

0 commit comments

Comments
 (0)