HyukjinKwon commented on code in PR #59329:
URL: https://github.com/apache/spark/pull/59329#discussion_r4237879075
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -231,6 +231,12 @@ public void initialize(
@Override
public void close() throws IOException {
+ if (columnVectors != null) {
+ for (ParquetColumnVector columnVector : columnVectors) {
+ columnVector.close();
Review Comment:
**Non-blocking (P2):** `initBatch` assigns `columnVectors = new
ParquetColumnVector[n]` before the constructor loop fills it. If a
`ParquetColumnVector` constructor throws partway through, the later slots stay
null. Examples are `INVALID_VARIANT_SHREDDING_SCHEMA` from `buildVariantSchema`
on a malformed shredded file, and the `IllegalArgumentException` for an
unsupported existence default. `ParquetFileFormat.buildVectorizedIterator` then
calls `iter.close()` in its SPARK-23457 catch block, and this loop throws
`NullPointerException` on the null slot. The NPE replaces the original error
before `throw e` runs. It also skips `columnarBatch.close()` and
`super.close()`, so on exactly the failure path this cleanup exists for, the
batch and the Parquet input stream leak.
Skipping null entries, and running the rest of the cleanup in a `finally`
block as `LateMaterializationParquetRecordReader.close()` does, would keep both
the original error and the existing cleanup.
**Verification:**
- **Regression:** When initBatch fails partway through (for example, an
unsupported existence-default value or an invalid variant shredding schema),
the original error is surfaced and no NullPointerException is raised from close.
- **Behavior:** Normal reads still release the batch and file reader exactly
once, and repeated close() calls remain safe.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnVector.java:
##########
@@ -39,6 +39,11 @@ final class ParquetColumnVector {
private final List<ParquetColumnVector> children;
private final WritableColumnVector vector;
+ // Cache of number of children in vector for use in `close`. Call to `close`
is not guaranteed to
+ // happen before `ColumnarBatch.close` which dereferences
`WritableColumnVector.childColumns`
Review Comment:
**Nit (P3):** `WritableColumnVector.close()` sets `childColumns` to null
rather than dereferencing it. It is the later `getNumChildren()` call that
dereferences it and throws. Also, `VectorizedParquetRecordReader.close()` runs
this before `columnarBatch.close()`. The caller that closes the batch's vectors
first is `LateMaterializationParquetRecordReader.close()`, via
`closeAll(persistentBatchColumns)` before `super.close()`. Naming that caller
would make it clear why the count is cached.
**Verification:**
- **Inspection:** The comment accurately describes why the child count is
cached: the late-materialization close ordering, and
WritableColumnVector.close() nulling childColumns.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnVector.java:
##########
@@ -231,6 +238,30 @@ void reset() {
}
}
+ void close() {
+ // Calls `ColumnVector.close` for scenarios using `reserveNewColumn`.
+ if (column.variantFileType().isDefined()) {
+ ParquetColumnVector child = children.get(0);
Review Comment:
**Blocking (P1):** When a requested variant column is missing from the file
and has no default, the constructor returns early at `vector.setMissing();
return;`, before the branch that adds this file-content child.
`column.variantFileType()` is still defined in that case.
`clipParquetGroupFields` synthesizes the missing field as a value/metadata
group, and `ParquetToSparkSchemaConverter` sets `variantFileType` because
`spark.sql.variant.allowReadingShredded` is on by default. So `children` is
empty here, and `children.get(0)` throws `IndexOutOfBoundsException` when the
reader is closed at the end of the split.
This is reachable with ordinary reads, for example `spark.read.schema("id
INT, v VARIANT").parquet(<files written with only id>)`, a Parquet table that
gained a `VARIANT` column after older files were written, or a variant field
missing inside a struct that is present. Those reads return NULLs today and
fail the task with this change. Because the exception escapes before
`columnarBatch.close()` and `super.close()`, the batch memory and the file
stream also leak.
`assemble()` already avoids this by checking `variantSchema != null`, which
is only set after the child is added. Using the same condition here would make
close() a no-op for the missing column.
**Verification:**
- **Regression:** Reading a file that lacks a requested variant column
(top-level and nested in a present struct) through the vectorized reader
succeeds and returns NULLs, in both on-heap and off-heap column vector modes.
- **Behavior:** Reading present shredded and unshredded variant columns
returns the same results as before.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
