spark-test commented on code in PR #59657:
URL: https://github.com/apache/spark/pull/59657#discussion_r4358967426
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala:
##########
@@ -1040,6 +1127,66 @@ class ParquetVectorizedSuite extends ParquetTest with
SharedSparkSession {
}
}
+ private def createVectorizedReader(
+ file: String,
+ readSchema: StructType,
+ offheapEnabled: Boolean): VectorizedParquetRecordReader = {
+ val conf = spark.sessionState.conf
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ hadoopConf.set(ParquetInputFormat.READ_SUPPORT_CLASS,
classOf[ParquetReadSupport].getName)
+ hadoopConf.set(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
readSchema.json)
+ Seq(
+ SQLConf.CASE_SENSITIVE,
+ SQLConf.PARQUET_BINARY_AS_STRING,
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP,
+ SQLConf.PARQUET_INFER_TIMESTAMP_NTZ_ENABLED,
+ SQLConf.LEGACY_PARQUET_NANOS_AS_LONG
+ ).foreach { entry =>
+ hadoopConf.set(entry.key, conf.getConfString(entry.key))
+ }
+
+ val split = new FileSplit(new Path(file), 0, new File(file).length(),
Array.empty[String])
+ val context = new TaskAttemptContextImpl(hadoopConf, new TaskAttemptID())
+ val reader = new VectorizedParquetRecordReader(
+ offheapEnabled,
+ conf.parquetVectorizedReaderBatchSize
+ )
+
+ try {
+ reader.enableReturningBatches()
+ reader.initialize(split, context)
+ } catch {
+ case _: Throwable =>
Review Comment:
**Nit (P3):** If `enableReturningBatches()` or `initialize()` throws, this
handler closes the reader but neither rethrows nor fails, and the helper
returns the closed reader. Both SPARK-60098 tests then call `nextKeyValue()` on
it, which fails later with an unrelated error: an `AssertionError` in
`truncateType` or an NPE in `allocateColumns` when `sparkSchema` was never set,
or an NPE in `loadNextRowGroup()` because `close()` nulled the row-group
reader. A regression in the struct-clipping or shredded-variant setup these
tests exercise would then be reported as that secondary error, and the root
cause would be lost. Matching `Throwable` also swallows fatal errors such as
`OutOfMemoryError` and `InterruptedException`.
Consider rethrowing after cleanup, e.g. `case NonFatal(e) => reader.close();
throw e`.
**Verification:**
- **Regression:** A reader-initialization failure inside the closure tests'
reader-construction helper surfaces as the original initialization exception,
with the reader closed, rather than as a secondary error from using a closed
reader.
- **Behavior:** The missing-struct-fields and shredded-variant closure tests
still pass and still assert that all value, definition-level and
repetition-level vectors are released after reader close, in both on-heap and
off-heap modes.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java:
##########
@@ -120,6 +129,11 @@ public void closeIfFreeable() {
// no-op
}
+ /**
+ * Returns true if this column vector holds no reference to data buffers.
+ */
+ public abstract boolean isFree();
Review Comment:
**Nit (P3):** `isFree()` is only called by the new `checkIsClosed`
assertions in `ParquetVectorizedSuite`, yet it is added as an unmarked `public
abstract` member of this production base class. Other test-only members of
these classes carry `@VisibleForTesting` (`MAX_CAPACITY` here,
`OffHeapColumnVector.valuesNativeAddress()`). The Javadoc also reads as a
whole-vector property, but both implementations check only this vector's own
buffers (`nulls`, data, lengths/offsets) and ignore `childColumns`, reserved
columns and `dictionaryIds`. The tests are fine today, because `close()`
releases those vectors before `releaseMemory()`. However, a future caller could
read the method as a full-release check for a nested or string vector. Suggest
marking it test-only and narrowing the doc:
```suggestion
/**
* Returns true if this vector's own data buffers are released; vectors it
owns are not checked.
*/
@VisibleForTesting
public abstract boolean isFree();
```
**Verification:**
- **Inspection:** The isFree() declaration on the vector base class carries
the test-only marker, and its documentation describes the own-buffers-only
scope that both implementations check.
- **Regression:** The vectorized Parquet closure assertions that rely on
isFree() still pass for on-heap and off-heap readers.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTestAccess.scala:
##########
@@ -244,6 +245,10 @@ object ParquetTestAccess {
f
}
+ def columnVectors(reader: VectorizedParquetRecordReader):
Array[ParquetColumnVector] = {
+ columnVectorsField.get(reader).asInstanceOf[Array[ParquetColumnVector]]
Review Comment:
**Nit (P3):** This casts to (and returns) the package-private
`ParquetColumnVector`, which this object's header rules out: "a package-private
class is looked up by name here, never named in a `classOf` or a cast." The
bridge exists for split-classloader runs (`spark-submit --jars`, e.g. the
benchmarks). There this cast would fail with `IllegalAccessError`, even though
the header now lists the field as bridged. It works today only because
`ParquetVectorizedSuite` runs in a single classloader and already names the
type directly.
Suggest returning `Array[AnyRef]`, as `columnReadersHeld` does for the same
field, and doing the typed cast in the suite's `checkColumnVectorsAreClosed`.
**Verification:**
- **Inspection:** The reflective Parquet test bridge contains no type,
classOf or cast that names a package-private parquet class; such classes appear
only in by-name lookups, consistent with its header.
- **Regression:** The vectorized Parquet closure tests still pass and still
assert release of every reserved vector after reader close.
--
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]
