foliveira-xonai commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r4167296763
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnVector.java:
##########
@@ -111,35 +100,41 @@ final class ParquetColumnVector {
if (column.variantFileType().isDefined()) {
ParquetColumn fileContentCol = column.variantFileType().get();
- WritableColumnVector fileContent = memoryMode == MemoryMode.OFF_HEAP
- ? new OffHeapColumnVector(capacity, fileContentCol.sparkType())
- : new OnHeapColumnVector(capacity, fileContentCol.sparkType());
- ParquetColumnVector contentVector = new
ParquetColumnVector(fileContentCol,
- fileContent, capacity, memoryMode, missingColumns, false, null);
+ WritableColumnVector fileContent = vector.reserveNewColumn(
+ capacity, fileContentCol.sparkType());
+ ParquetColumnVector contentVector = new
ParquetColumnVector(fileContentCol, fileContent,
+ capacity, missingColumns, /* isTopLevel= */ false, /* defaultValue= */
null);
children.add(contentVector);
variantSchema =
SparkShreddingUtils.buildVariantSchema(fileContentCol.sparkType());
fieldsToExtract =
SparkShreddingUtils.getFieldsToExtract(column.sparkType(), variantSchema);
repetitionLevels = contentVector.repetitionLevels;
definitionLevels = contentVector.definitionLevels;
} else if (isPrimitive) {
if (column.repetitionLevel() > 0) {
- repetitionLevels = allocateLevelsVector(capacity, memoryMode);
+ repetitionLevels = vector.reserveNewColumn(capacity,
DataTypes.IntegerType);
}
// We don't need to create and store definition levels if the column is
top-level.
if (!isTopLevel) {
- definitionLevels = allocateLevelsVector(capacity, memoryMode);
+ definitionLevels = vector.reserveNewColumn(capacity,
DataTypes.IntegerType);
}
} else {
- JavaUtils.checkArgument(column.children().size() ==
vector.getNumChildren(),
- "The number of column children is different from the number of vector
children");
+ // If a child is not present in the allocated vectors, it means we don't
care about this
+ // child's data, we just want to read its levels to help assemble some
parent struct. So we
+ // create a dummy vector below to hold the child's data. There can only
be one such child.
+ JavaUtils.checkArgument(column.children().size() ==
vector.getNumChildren() ||
+ column.children().size() == vector.getNumChildren() + 1,
+ "The number of column children is not equal to the number of vector
children or that + 1");
boolean allChildrenAreMissing = true;
for (int i = 0; i < column.children().size(); i++) {
- ParquetColumnVector childCv = new
ParquetColumnVector(column.children().apply(i),
- vector.getChild(i), capacity, memoryMode, missingColumns, false,
null);
+ ParquetColumn childColumn = column.children().apply(i);
+ WritableColumnVector childVector = i < vector.getNumChildren()
+ ? vector.getChild(i)
+ : vector.reserveNewColumn(capacity, childColumn.sparkType());
Review Comment:
Where is this column being released? Am I missing something or is this a
memory leak for `OffHeapColumnVector`? Same in `repetitionLevels` and
`definitionLevels`
--
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]
