Skip to content

fix(parquet): report REQUIRED nulls in object columns; clearer DECIMAL errors #638

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dfa1 merged 1 commit into from
Nov 1, 2026
Merged

fix(parquet): report REQUIRED nulls in object columns; clearer DECIMAL errors #638

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- A nullable timestamp column holding nulls is written as `vortex.datetimeparts` with the validity on its days part, as Rust writes it, instead of `vortex.ext` over a masked i64. ([#627](https://github.com/dfa1/vortex-java/issues/627))
- A string or binary column whose FSST codes pass 2 GiB in one chunk, such as large audio blobs, writes and reads back instead of failing with an integer overflow. ([#629](https://github.com/dfa1/vortex-java/issues/629))
- A string or binary chunk past 2 GiB writes and reads back as `vortex.varbin`, `vortex.varbinview` and compact `vortex.zstd` too, and Rust-written `varbin` columns with `u32` offsets past 2 GiB read instead of failing as out of range. ([#633](https://github.com/dfa1/vortex-java/issues/633))
- Parquet import reads a `DECIMAL` column as a Vortex decimal with its precision and scale, in all four physical encodings; `INT32`/`INT64` ones lost their scale and the byte-array ones failed. ([#628](https://github.com/dfa1/vortex-java/issues/628))
- Parquet import reads a `DECIMAL` column as a Vortex decimal with its precision and scale, in all four physical encodings, and names the column of one wider than Vortex's 76 digits; `INT32`/`INT64` ones lost their scale and the byte-array ones failed. ([#628](https://github.com/dfa1/vortex-java/issues/628))
- Parquet import keeps the nulls of a nullable `TIMESTAMP` column instead of failing as if the column were `REQUIRED`. ([#627](https://github.com/dfa1/vortex-java/issues/627))

## [0.19.0] — 2026-11-01
Expand Down
15 changes: 13 additions & 2 deletions .../src/test/java/io/github/dfa1/vortex/integration/ParquetImportInteropIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,8 @@ class ParquetImportInteropIntegrationTest {
private static final ColumnName TS = ColumnName.of("ts");
/// More than two zones (8192 rows), so the filtered read prunes and the range read slices.
private static final int ROWS = 3 * 8192;
/// 10^37: a 12-digit value times it reaches 49 digits, past an i128 (about 1.7e38).
private static final BigInteger WIDE = BigInteger.TEN.pow(37);
private static final long START_MICROS = 1_704_067_200_000_000L; // 2024-01-01T00:00:00Z

static {
Expand Down Expand Up @@ -110,7 +112,7 @@ void nullableTimestampWithNulls_jniReadsImportedFile(@TempDir Path tmp) throws E
"INT64, 0, 18, 4",
"FIXED_LEN_BYTE_ARRAY, 7, 15, 2",
"BYTE_ARRAY, 0, 30, 3",
"BYTE_ARRAY, 0, 50, 3", // past decimal128: Rust's i256 values
"BYTE_ARRAY, 0, 50, 3", // past decimal128: values only an i256 holds
})
void decimalWithNulls_jniReadsImportedFile(PhysicalType type, int length, int precision, int scale,
@TempDir Path tmp) throws Exception {
Expand All @@ -120,7 +122,16 @@ void decimalWithNulls_jniReadsImportedFile(PhysicalType type, int length, int pr
BigInteger[] unscaled = new BigInteger[ROWS];
List<BigDecimal> expected = new ArrayList<>(ROWS);
for (int i = 0; i < ROWS; i++) {
unscaled[i] = isNull.test(i) ? BigInteger.ZERO : BigInteger.valueOf(Math.floorMod(i * 7_919L, 2 * bound) - bound);
// the product wraps 2 * bound many times, so values take both signs
BigInteger value = BigInteger.valueOf(Math.floorMod(i * 7_919_919_919L, 2 * bound) - bound);
if (precision > 38) {
// scaled past what an i128 holds, so Rust reads real i256s
value = value.multiply(WIDE);
}
if (isNull.test(i)) {
value = BigInteger.ZERO;
}
unscaled[i] = value;
expected.add(isNull.test(i) ? null : new BigDecimal(unscaled[i], scale));
}
Path parquet = decimalParquet(tmp, type, length, precision, scale, unscaled, isNull);
Expand Down
22 changes: 17 additions & 5 deletions parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetImporter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import dev.hardwood.schema.ColumnSchema;
import dev.hardwood.schema.FileSchema;
import dev.hardwood.schema.SchemaNode;
import io.github.dfa1.vortex.core.error.VortexException;
import io.github.dfa1.vortex.core.model.ColumnName;
import io.github.dfa1.vortex.core.model.DType;
import io.github.dfa1.vortex.core.model.PType;
Expand All @@ -32,6 +33,7 @@
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

/// Reads a Parquet file and writes a Vortex file.
Expand Down Expand Up @@ -234,7 +236,14 @@ private static DType mapLeafDType(PhysicalType type, LogicalType logical, boolea
// value, big-endian for the byte forms); each maps to Vortex's decimal, as Arrow's
// decimal128/256 does in Rust's Arrow import, never to the bare integer that drops the scale
if (logical instanceof LogicalType.DecimalType(int precision, int scale)) {
return DType.Decimal.of(precision, scale, nullable);
try {
return DType.Decimal.of(precision, scale, nullable);
} catch (VortexException e) {
// Parquet allows a byte-array DECIMAL wider than Vortex's 76 digits (Rust's Arrow
// has no type for it either)
throw new UnsupportedOperationException("DECIMAL(" + precision + ", " + scale + ") column '"
+ name + "' has no Vortex decimal: " + e.getMessage(), e);
}
}
return switch (type) {
case BOOLEAN -> new DType.Bool(nullable);
Expand Down Expand Up @@ -376,7 +385,7 @@ static void fillRow(RowReader reader, List<ColumnName> names, List<DType> types,
boolean isNull = types.get(c).nullable() && reader.isNull(name);
if (types.get(c) instanceof DType.Primitive p && p.ptype() == PType.F16) {
// Same carrier as I16, but the value is a 2-byte FIXED_LEN_BYTE_ARRAY, not an INT32
putHalf(buffers[c], pos, isNull ? null : reader.getBinary(name));
putHalf(buffers[c], pos, isNull ? null : Objects.requireNonNull(reader.getBinary(name)));
continue;
}
switch (buffers[c]) {
Expand All @@ -394,9 +403,12 @@ static void fillRow(RowReader reader, List<ColumnName> names, List<DType> types,
case Short[] arr -> arr[pos] = isNull ? null : (short) reader.getInt(name);
case byte[] arr -> arr[pos] = (byte) reader.getInt(name);
case Byte[] arr -> arr[pos] = isNull ? null : (byte) reader.getInt(name);
case String[] arr -> arr[pos] = isNull ? null : reader.getString(name);
case BigDecimal[] arr -> arr[pos] = isNull ? null : reader.getDecimal(name);
case byte[][] arr -> arr[pos] = isNull ? null : reader.getBinary(name);
case String[] arr -> arr[pos] = isNull ? null : Objects.requireNonNull(reader.getString(name));
// getString/getBinary/getDecimal return null rather than throwing, so a null
// under a REQUIRED column becomes the NPE the catch below reports with its file
// row (a FLOAT16 one used to become a silent +0.0)
case BigDecimal[] arr -> arr[pos] = isNull ? null : Objects.requireNonNull(reader.getDecimal(name));
case byte[][] arr -> arr[pos] = isNull ? null : Objects.requireNonNull(reader.getBinary(name));
default -> throw new UnsupportedOperationException(
"unexpected buffer type: " + buffers[c].getClass().getSimpleName());
}
Expand Down
146 changes: 135 additions & 11 deletions parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetImporterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -375,6 +375,55 @@ void requiredColumnActuallyNull_throwsClearErrorInsteadOfRawNpe() {
.hasMessageContaining("malformed.parquet")
.hasCauseInstanceOf(NullPointerException.class);
}

@ParameterizedTest(name = "{0}")
@CsvSource({"utf8", "binary", "float16"})
void requiredObjectColumnActuallyNull_reportsTheFileRow(String kind) {
// Given — getString/getBinary return null instead of throwing; a FLOAT16 null used to
// become a silent +0.0
DType dtype = switch (kind) {
case "utf8" -> new DType.Utf8(false);
case "binary" -> new DType.Binary(false);
default -> new DType.Primitive(PType.F16, false);
};
Object buffer = switch (kind) {
case "utf8" -> new String[1];
case "binary" -> new byte[1][];
default -> new short[1];
};
List<ColumnName> names = List.of(ColumnName.of("v"));
ColumnBuilder[] nestedBuilders = new ColumnBuilder[1];
if (kind.equals("utf8")) {
given(reader.getString("v")).willReturn(null);
} else {
given(reader.getBinary("v")).willReturn(null);
}

// When / Then
assertThatThrownBy(() -> ParquetImporter.fillRow(reader, names, List.of(dtype), new Object[]{buffer},
nestedBuilders, 0, "malformed.parquet", 42L))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("REQUIRED")
.hasMessageContaining("row 42");
}

@Test
void requiredDecimalActuallyNull_reportsTheFileRow() {
// Given — getDecimal returns null instead of throwing, unlike the primitive getters
List<ColumnName> names = List.of(ColumnName.of("amount"));
List<DType> types = List.of(DType.Decimal.of(15, 2, false));
Object[] buffers = {new BigDecimal[1]};
ColumnBuilder[] nestedBuilders = new ColumnBuilder[1];
given(reader.getDecimal("amount")).willReturn(null);

// When / Then — the file's row 70_000, not the row within the chunk
assertThatThrownBy(() -> ParquetImporter.fillRow(reader, names, types, buffers, nestedBuilders,
0, "malformed.parquet", 70_000L))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("amount")
.hasMessageContaining("REQUIRED")
.hasMessageContaining("row 70000");
}
}

@Nested
Expand Down Expand Up @@ -413,32 +462,107 @@ void decimalColumn_importsAsDecimalWithItsScale(PhysicalType type, int length, i
}
}

@ParameterizedTest(name = "{0}")
@CsvSource({
"INT32, 0, 9, 2",
"INT64, 0, 18, 2",
"FIXED_LEN_BYTE_ARRAY, 7, 15, 2",
"BYTE_ARRAY, 0, 30, 2",
})
void requiredDecimalColumn_importsAsNonNullableDecimal(PhysicalType type, int length, int precision,
int scale, @TempDir Path tmp) throws Exception {
// Given — a REQUIRED DECIMAL column, the non-nullable BigDecimal[] path
List<BigDecimal> values = List.of(new BigDecimal("17.00"), new BigDecimal("-0.05"), new BigDecimal("0.00"));
Path parquet = writeDecimalParquet(tmp, type, length, precision, scale, RepetitionType.REQUIRED, values);
Path vortex = tmp.resolve("out.vortex");

// When
ParquetImporter.importParquet(parquet, vortex);

// Then
try (VortexReader reader = VortexReader.open(vortex);
ScanIterator iter = reader.scan(ScanOptions.all())) {
DType.Struct schema = (DType.Struct) reader.dtype();
assertThat(schema.fieldTypes().getFirst()).isEqualTo(DType.Decimal.of(precision, scale, false));
try (Chunk chunk = iter.next()) {
assertThat(decimals(chunk.column(ColumnName.of("d")), values.size()))
.containsExactlyElementsOf(values);
}
}
}

@Test
void decimalWiderThanVortexAllows_throwsNamingTheColumn() {
// Given — Parquet allows a BYTE_ARRAY DECIMAL past Vortex's 76 digits
ColumnSchema schema = col("amount", PhysicalType.BYTE_ARRAY, RepetitionType.OPTIONAL,
new LogicalType.DecimalType(100, 2));

// When / Then
assertThatThrownBy(() -> ParquetImporter.mapDType(schema))
.isInstanceOf(UnsupportedOperationException.class)
.hasMessageContaining("amount")
.hasMessageContaining("DECIMAL(100, 2)");
}

private static Path writeDecimalParquet(Path tmp, PhysicalType type, int length, int precision, int scale)
throws Exception {
return writeDecimalParquet(tmp, type, length, precision, scale, RepetitionType.OPTIONAL, VALUES);
}

private static Path writeDecimalParquet(Path tmp, PhysicalType type, int length, int precision, int scale,
RepetitionType repetition, List<BigDecimal> values) throws Exception {
FileSchema.Builder builder = FileSchema.builder("schema");
LogicalType decimal = new LogicalType.DecimalType(precision, scale);
if (length > 0) {
builder.addColumn("d", type, RepetitionType.OPTIONAL, length, decimal);
builder.addColumn("d", type, repetition, length, decimal);
} else {
builder.addColumn("d", type, RepetitionType.OPTIONAL, decimal);
builder.addColumn("d", type, repetition, decimal);
}
int n = VALUES.size();
int n = values.size();
boolean[] nulls = new boolean[n];
BigInteger[] unscaled = new BigInteger[n];
for (int i = 0; i < n; i++) {
nulls[i] = VALUES.get(i) == null;
unscaled[i] = nulls[i] ? BigInteger.ZERO : VALUES.get(i).setScale(scale).unscaledValue();
nulls[i] = values.get(i) == null;
unscaled[i] = nulls[i] ? BigInteger.ZERO : values.get(i).setScale(scale).unscaledValue();
}
// hardwood takes a null mask only for an OPTIONAL column, and its mask-less setters otherwise
boolean[] mask = repetition == RepetitionType.OPTIONAL ? nulls : null;
Path parquet = tmp.resolve("decimal.parquet");
try (ParquetFileWriter writer = ParquetFileWriter.create(OutputFile.of(parquet), builder.build())) {
writer.columnWriter().writeBatch(batch -> {
switch (type) {
case INT32 -> batch.ints(0, Arrays.stream(unscaled).mapToInt(BigInteger::intValueExact).toArray(), nulls);
case INT64 -> batch.longs(0, Arrays.stream(unscaled).mapToLong(BigInteger::longValueExact).toArray(), nulls);
case FIXED_LEN_BYTE_ARRAY -> batch.fixed(0, Arrays.stream(unscaled).map(v -> bigEndian(v, length))
.toArray(byte[][]::new), nulls);
case BYTE_ARRAY -> batch.bytes(0, Arrays.stream(unscaled).map(BigInteger::toByteArray)
.toArray(byte[][]::new), nulls);
case INT32 -> {
int[] ints = Arrays.stream(unscaled).mapToInt(BigInteger::intValueExact).toArray();
if (mask == null) {
batch.ints(0, ints);
} else {
batch.ints(0, ints, mask);
}
}
case INT64 -> {
long[] longs = Arrays.stream(unscaled).mapToLong(BigInteger::longValueExact).toArray();
if (mask == null) {
batch.longs(0, longs);
} else {
batch.longs(0, longs, mask);
}
}
case FIXED_LEN_BYTE_ARRAY -> {
byte[][] fixed = Arrays.stream(unscaled).map(v -> bigEndian(v, length)).toArray(byte[][]::new);
if (mask == null) {
batch.fixed(0, fixed);
} else {
batch.fixed(0, fixed, mask);
}
}
case BYTE_ARRAY -> {
byte[][] bytes = Arrays.stream(unscaled).map(BigInteger::toByteArray).toArray(byte[][]::new);
if (mask == null) {
batch.bytes(0, bytes);
} else {
batch.bytes(0, bytes, mask);
}
}
default -> throw new IllegalArgumentException(type.toString());
}
});
Expand Down
Loading