You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "ZiyaZa (via GitHub)" <gi...@apache.org> on 2025/10/09 08:34:43 UTC
[PR] [WIP][SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
ZiyaZa opened a new pull request, #52557:
URL: https://github.com/apache/spark/pull/52557
### What changes were proposed in this pull request?
Currently, if all fields of a struct mentioned in the read schema are missing in a Parquet file, the reader populates the struct with nulls.
This PR modifies the scan behavior so that if the struct exists in the Parquet schema but none of the fields from the read schema are present, we instead pick an arbitrary field from the Parquet file to read and use that to populate NULLs (as well as outer NULLs and array sizes if the struct is nested in another nested type).
This is done by changing the schema requested by the readers. We add an additional field to the requested schema when clipping the Parquet file schema according to the Spark schema. This means that the readers actually read and return more data than requested, which can cause problems. This is only a problem for the `VectorizedParquetRecordReader`, since for the other read code path via parquet-mr, we already have an `UnsafeProjection` for outputting only requested schema fields in `ParquetFileFormat`.
To ensure `VectorizedParquetRecordReader` only returns Spark requested fields, we create the `ColumnarBatch` with vectors that match the requested schema (we get rid of the additional fields by recursively matching `sparkSchema` with `sparkRequestedSchema` and ensuring structs have the same length in both). Then `ParquetColumnVector`s are responsible for allocating dummy vectors to hold the data temporarily while reading, but these are not exposed to the outside.
The heuristic to pick the arbitrary field is as follows: we pick one at the lowest array nesting level (i.e., any scalar field is preferred to `array`, which is preferred to `array<array>`), and prefer narrower scalar fields over wider scalar fields, which are preferred over strings.
### Why are the changes needed?
This is a bug fix, because we were incorrectly assuming non-null struct values to be missing from the file depending on requested fields and returning null values.
### Does this PR introduce _any_ user-facing change?
Yes. We previously assumed structs to be null if all the fields we are trying to read from a Parquet file were missing from that file, even if the file contained other fields that could be used to take definition levels from. See an example from the Jira ticket below:
```python
df_a = sql('SELECT 1 as id, named_struct("a", 1) AS s')
path = "/tmp/missing_col_test"
df_a.write.format("parquet").save(path)
df_b = sql('SELECT 2 as id, named_struct("b", 3) AS s')
spark.read.format("parquet").schema(df_b.schema).load(path).show()
```
This used to return:
```
+---+----+
| id| s|
+---+----+
| 1|NULL|
+---+----+
```
It now returns:
```
+---+------+
| id| s|
+---+------+
| 1|{NULL}|
+---+------+
```
### How was this patch tested?
Added new unit tests, also fixed an old test to expect this new behavior.
### Was this patch authored or co-authored using generative AI tooling?
No.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2441533153
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -309,6 +316,58 @@ public void initBatch(StructType partitionColumns, InternalRow partitionValues)
initBatch(MEMORY_MODE, partitionColumns, partitionValues);
}
+ /**
+ * Keeps the hierarchy and fields of readType, recursively truncating struct fields from the end
+ * of the fields list to match the same number of fields in requestedType. This is used to get rid
+ * of the extra fields that are added to the structs when the fields we wanted to read initially
+ * were missing in the file schema. So this returns a type that we would be reading if everything
+ * was present in the file, matching Spark's expected schema.
+ *
+ * <p> Example: <pre>{@code
+ * readType: array<struct<a:int,b:long,c:int>>
+ * requestedType: array<struct<a:int,b:long>>
+ * returns: array<struct<a:int,b:long>>
+ * }</pre>
+ * We cannot return requestedType here because there might be slight differences, like nullability
Review Comment:
shall we choose a different example? the current example simplify returns `requestedType`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2441532806
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -265,7 +265,13 @@ private void initBatch(
MemoryMode memMode,
StructType partitionColumns,
InternalRow partitionValues) {
- StructType batchSchema = new StructType(sparkSchema.fields());
+ boolean returnNullStructIfAllFieldsMissing = configuration.getBoolean(
+ SQLConf$.MODULE$.LEGACY_PARQUET_RETURN_NULL_STRUCT_IF_ALL_FIELDS_MISSING().key(),
+ (boolean) SQLConf$.MODULE$.LEGACY_PARQUET_RETURN_NULL_STRUCT_IF_ALL_FIELDS_MISSING()
+ .defaultValue().get());
+ StructType batchSchema = !returnNullStructIfAllFieldsMissing
+ ? new StructType(sparkSchema.fields())
+ : (StructType) truncateType(sparkSchema, sparkRequestedSchema);
Review Comment:
the code before is `StructType batchSchema = new StructType(sparkSchema.fields());`, but now we only go to it if legacy config is off?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "Kimahriman (via GitHub)" <gi...@apache.org>.
Kimahriman commented on PR #52557:
URL: https://github.com/apache/spark/pull/52557#issuecomment-3387926488
Wow this has been a problem for us for so long, especially when you read non nullable strict this actually throws an NPE instead of just giving you the wrong data. Thanks for the fix!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2443583458
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -755,10 +763,72 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSparkSession
.add("_4", LongType, nullable = true),
nullable = true)
+ val expectedAnswer = if (!returnNullStructIfAllFieldsMissing) {
+ Row(Row(null, null)) :: Row(Row(null, null)) :: Row(null) :: Nil
+ } else {
+ Row(null) :: Row(null) :: Row(null) :: Nil
+ }
+
withParquetFile(data) { file =>
- checkAnswer(spark.read.schema(readSchema).parquet(file),
- Row(null) :: Row(null) :: Row(null) :: Nil
- )
+ val df = spark.read.schema(readSchema).parquet(file)
+ val scanNode = df.queryExecution.executedPlan.collectLeaves().head
+ VerifyNoAdditionalScanOutputExec(scanNode).execute().first()
Review Comment:
shall we use the existing `ColumnarToRowExec` and then verify the rows? But I'm also fine with this custom physical plan.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "ZiyaZa (via GitHub)" <gi...@apache.org>.
ZiyaZa commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2443887122
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -755,10 +763,72 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSparkSession
.add("_4", LongType, nullable = true),
nullable = true)
+ val expectedAnswer = if (!returnNullStructIfAllFieldsMissing) {
+ Row(Row(null, null)) :: Row(Row(null, null)) :: Row(null) :: Nil
+ } else {
+ Row(null) :: Row(null) :: Row(null) :: Nil
+ }
+
withParquetFile(data) { file =>
- checkAnswer(spark.read.schema(readSchema).parquet(file),
- Row(null) :: Row(null) :: Row(null) :: Nil
- )
+ val df = spark.read.schema(readSchema).parquet(file)
+ val scanNode = df.queryExecution.executedPlan.collectLeaves().head
+ VerifyNoAdditionalScanOutputExec(scanNode).execute().first()
Review Comment:
I tried to use that first, but couldn't get it to work because `ColumnarToRowExec` adds its own `UnsafeProjection` with `child.output`. Let's keep the custom plan node.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2441533594
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala:
##########
@@ -418,11 +428,16 @@ object ParquetReadSupport extends Logging {
parquetRecord.getFields.asScala.groupBy(_.getName.toLowerCase(Locale.ROOT))
lazy val idToParquetFieldMap =
parquetRecord.getFields.asScala.filter(_.getId != null).groupBy(f => f.getId.intValue())
+ var isStructWithMissingFields = true
Review Comment:
```suggestion
var isStructWithMissingAllFields = true
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2441535488
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -265,7 +265,13 @@ private void initBatch(
MemoryMode memMode,
StructType partitionColumns,
InternalRow partitionValues) {
- StructType batchSchema = new StructType(sparkSchema.fields());
+ boolean returnNullStructIfAllFieldsMissing = configuration.getBoolean(
+ SQLConf$.MODULE$.LEGACY_PARQUET_RETURN_NULL_STRUCT_IF_ALL_FIELDS_MISSING().key(),
+ (boolean) SQLConf$.MODULE$.LEGACY_PARQUET_RETURN_NULL_STRUCT_IF_ALL_FIELDS_MISSING()
+ .defaultValue().get());
+ StructType batchSchema = !returnNullStructIfAllFieldsMissing
+ ? new StructType(sparkSchema.fields())
+ : (StructType) truncateType(sparkSchema, sparkRequestedSchema);
Review Comment:
ah I see, we need to call `truncateType` for the legacy code path, to deal with the extra cheap field we picked when all fields are missing. Can we add code comments to explain it?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "ZiyaZa (via GitHub)" <gi...@apache.org>.
ZiyaZa commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r4167405958
##########
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:
Upon quick look it seems there's no close() function or something like that to release the memory, so looks like an issue. But it is not introduced by this PR as vectors were being allocated before as well, recommend filing a Spark ticket for this.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "ZiyaZa (via GitHub)" <gi...@apache.org>.
ZiyaZa commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r4167353994
##########
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:
`ParquetColumnVector.close()` releases all the memory associated with the column and its children.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on PR #52557:
URL: https://github.com/apache/spark/pull/52557#issuecomment-3413874113
@ZiyaZa I am not a big fan of such behavior changes.
Also the PR description is a bit confusing to me:
> we instead pick an arbitrary field from the Parquet file to read and use that to populate NULLs
why just picking one arbitrary field, instead of setting all the fields null?
> The heuristic to pick the arbitrary field is as follows: we pick one at the lowest array nesting level (i.e., any scalar field is preferred to array, which is preferred to array<array>)
Could you provide more details on this one?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2441972398
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -265,7 +265,15 @@ private void initBatch(
MemoryMode memMode,
StructType partitionColumns,
InternalRow partitionValues) {
- StructType batchSchema = new StructType(sparkSchema.fields());
+ boolean returnNullStructIfAllFieldsMissing = configuration.getBoolean(
+ SQLConf$.MODULE$.LEGACY_PARQUET_RETURN_NULL_STRUCT_IF_ALL_FIELDS_MISSING().key(),
+ (boolean) SQLConf$.MODULE$.LEGACY_PARQUET_RETURN_NULL_STRUCT_IF_ALL_FIELDS_MISSING()
+ .defaultValue().get());
+ StructType batchSchema = returnNullStructIfAllFieldsMissing
+ ? new StructType(sparkSchema.fields())
+ // Truncate to match requested schema to make sure extra struct field that we read for
+ // nullability is not included in columnarBatch and exposed outside.
Review Comment:
@ZiyaZa do we have test coverage for it? The tests passed both before and after you flip `returnNullStructIfAllFieldsMissing` here...
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2443585700
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -1655,3 +1725,25 @@ class TaskCommitFailureParquetOutputCommitter(outputPath: Path, context: TaskAtt
sys.error("Intentional exception for testing purposes")
}
}
+
+case class VerifyNoAdditionalScanOutputExec(override val child: SparkPlan) extends UnaryExecNode {
+ override def doExecute(): RDD[InternalRow] = {
+ val childOutputTypes = child.output.map(_.dataType)
+ child.executeColumnar().mapPartitionsInternal { batches =>
+ batches.flatMap { input =>
+ input.rowIterator().asScala
Review Comment:
if we decide to keep this custom physical plan, we can simplify it by checking columnar batches directly:
```
0.until(input.numCols).foreach { index =>
assert(childOutputTypes(index).dataType == input.column(index).dataType)
}
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #52557:
URL: https://github.com/apache/spark/pull/52557#issuecomment-3424139519
thanks, merging to master!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2440775117
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1534,6 +1534,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val PARQUET_READ_ANY_FIELD_FOR_MISSING_STRUCT =
+ buildConf("spark.sql.parquet.readAnyFieldForMissingStruct")
Review Comment:
this should be a legacy config that is off by default, how about `spark.sql.legacy.returnNullStructIfAllFieldsMissing`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "ZiyaZa (via GitHub)" <gi...@apache.org>.
ZiyaZa commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2440909733
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1534,6 +1534,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val PARQUET_READ_ANY_FIELD_FOR_MISSING_STRUCT =
+ buildConf("spark.sql.parquet.readAnyFieldForMissingStruct")
Review Comment:
I named it `spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing` to show that it is for Parquet files.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "ZiyaZa (via GitHub)" <gi...@apache.org>.
ZiyaZa commented on PR #52557:
URL: https://github.com/apache/spark/pull/52557#issuecomment-3416301451
@gengliangwang
> I am not a big fan of such behavior changes.
This is a behavior change that is required to fix a correctness issue. The issue is described in more detail in the linked [JIRA ticket](https://issues.apache.org/jira/browse/SPARK-53535). According to the comment https://github.com/apache/spark/pull/52557#issuecomment-3387926488 above, we could also get NullPointerException if the struct is marked as nullable, because we would wrongly assume all struct values to be null previously.
> why just picking one arbitrary field, instead of setting all the fields null?
We need to understand **for each row** if the struct value is null or it is a struct with all the fields as null (to explain in JSON notation, `null` and `{ 'a': null }` are not the same thing). The only way we can understand this is by looking at a child field of a struct that is present in the file, because Parquet stores nullability information in the definition levels of leaf columns. Based on that definition levels, we can identify in which rows struct is null or non-null.
> Could you provide more details on this one?
Updated the description.
> For breaking changes, usually we should introduce a SQL configuration to control the new/legacy behaviors
Added a flag to control this behavior.
> also we should update https://spark.apache.org/docs/latest/sql-migration-guide.html
Can you please explain how we update that? It looks like that is built from the `sql-migration-guide.md` file, but it currently does not contain anything for the next release if I read it correctly.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on PR #52557:
URL: https://github.com/apache/spark/pull/52557#issuecomment-3413875313
For breaking changes, usually we should introduce a SQL configuration to control the new/legacy behaviors, also we should update https://spark.apache.org/docs/latest/sql-migration-guide.html
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "foliveira-xonai (via GitHub)" <gi...@apache.org>.
foliveira-xonai commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r4167447891
##########
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:
Yes, agree that is not an issue by this PR. Wanted to confirm before opening issue. Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "ZiyaZa (via GitHub)" <gi...@apache.org>.
ZiyaZa commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2441739945
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -265,7 +265,13 @@ private void initBatch(
MemoryMode memMode,
StructType partitionColumns,
InternalRow partitionValues) {
- StructType batchSchema = new StructType(sparkSchema.fields());
+ boolean returnNullStructIfAllFieldsMissing = configuration.getBoolean(
+ SQLConf$.MODULE$.LEGACY_PARQUET_RETURN_NULL_STRUCT_IF_ALL_FIELDS_MISSING().key(),
+ (boolean) SQLConf$.MODULE$.LEGACY_PARQUET_RETURN_NULL_STRUCT_IF_ALL_FIELDS_MISSING()
+ .defaultValue().get());
+ StructType batchSchema = !returnNullStructIfAllFieldsMissing
+ ? new StructType(sparkSchema.fields())
+ : (StructType) truncateType(sparkSchema, sparkRequestedSchema);
Review Comment:
I think I flipped both the condition and the statements when converting the flag from default true to default false legacy flag. In legacy mode, we don't need `truncateType`. It is used to get rid of the additional field we are reading if we are not in the legacy mode.
Added a comment.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "ZiyaZa (via GitHub)" <gi...@apache.org>.
ZiyaZa commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2441976052
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -265,7 +265,15 @@ private void initBatch(
MemoryMode memMode,
StructType partitionColumns,
InternalRow partitionValues) {
- StructType batchSchema = new StructType(sparkSchema.fields());
+ boolean returnNullStructIfAllFieldsMissing = configuration.getBoolean(
+ SQLConf$.MODULE$.LEGACY_PARQUET_RETURN_NULL_STRUCT_IF_ALL_FIELDS_MISSING().key(),
+ (boolean) SQLConf$.MODULE$.LEGACY_PARQUET_RETURN_NULL_STRUCT_IF_ALL_FIELDS_MISSING()
+ .defaultValue().get());
+ StructType batchSchema = returnNullStructIfAllFieldsMissing
+ ? new StructType(sparkSchema.fields())
+ // Truncate to match requested schema to make sure extra struct field that we read for
+ // nullability is not included in columnarBatch and exposed outside.
Review Comment:
Apparently we don't, let me add a test.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on PR #52557:
URL: https://github.com/apache/spark/pull/52557#issuecomment-3416355689
> It looks like that is built from the sql-migration-guide.md file, but it currently does not contain anything for the next release if I read it correctly.
Yes, please create a new section `## Upgrading from Spark SQL 4.0 to 4.1`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2441533153
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -309,6 +316,58 @@ public void initBatch(StructType partitionColumns, InternalRow partitionValues)
initBatch(MEMORY_MODE, partitionColumns, partitionValues);
}
+ /**
+ * Keeps the hierarchy and fields of readType, recursively truncating struct fields from the end
+ * of the fields list to match the same number of fields in requestedType. This is used to get rid
+ * of the extra fields that are added to the structs when the fields we wanted to read initially
+ * were missing in the file schema. So this returns a type that we would be reading if everything
+ * was present in the file, matching Spark's expected schema.
+ *
+ * <p> Example: <pre>{@code
+ * readType: array<struct<a:int,b:long,c:int>>
+ * requestedType: array<struct<a:int,b:long>>
+ * returns: array<struct<a:int,b:long>>
+ * }</pre>
+ * We cannot return requestedType here because there might be slight differences, like nullability
Review Comment:
shall we choose a different example? the current example simplify returns `requestedType`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2441535488
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -265,7 +265,13 @@ private void initBatch(
MemoryMode memMode,
StructType partitionColumns,
InternalRow partitionValues) {
- StructType batchSchema = new StructType(sparkSchema.fields());
+ boolean returnNullStructIfAllFieldsMissing = configuration.getBoolean(
+ SQLConf$.MODULE$.LEGACY_PARQUET_RETURN_NULL_STRUCT_IF_ALL_FIELDS_MISSING().key(),
+ (boolean) SQLConf$.MODULE$.LEGACY_PARQUET_RETURN_NULL_STRUCT_IF_ALL_FIELDS_MISSING()
+ .defaultValue().get());
+ StructType batchSchema = !returnNullStructIfAllFieldsMissing
+ ? new StructType(sparkSchema.fields())
+ : (StructType) truncateType(sparkSchema, sparkRequestedSchema);
Review Comment:
ah I see, we need to call `truncateType` for the legacy code path, to deal with the extra cheap field we picked. Can we add code comments to explain it?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2441534000
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala:
##########
@@ -472,6 +491,64 @@ object ParquetReadSupport extends Logging {
matchCaseInsensitiveField(f)
}
}
+ // Ignore MessageType, because it is the root of the schema, not a struct.
+ if (returnNullStructIfAllFieldsMissing || !isStructWithMissingFields ||
+ parquetRecord.isInstanceOf[MessageType]) {
+ clippedType
+ } else {
+ // Read one arbitrary field to understand when the struct value is null or not null.
+ clippedType :+ findCheapestGroupField(parquetRecord)
+ }
+ }
+
+ /**
+ * Finds the leaf node under a given file schema node that is likely to be cheapest to fetch.
+ * Keeps this leaf node inside the same parent hierarchy. This is used when all struct fields in
+ * the requested schema are missing. Uses a very simple heuristic based on the parquet type.
+ */
+ private def findCheapestGroupField(parentGroupType: GroupType): Type = {
+ def findCheapestGroupFieldRecurse(curType: Type, repLevel: Int = 0): (Type, Int, Int) = {
+ curType match {
+ case groupType: GroupType =>
+ var (bestType, bestRepLevel, bestCost) = (Option.empty[Type], 0, 0)
+ for (field <- groupType.getFields.asScala) {
+ val newRepLevel = repLevel + (if (field.isRepetition(Repetition.REPEATED)) 1 else 0)
+ // Never take a field at a deeper repetition level, since it's likely to have more data.
+ // Don't do safety checks because we should already have done them when traversing the
+ // schema for the first time.
+ if (bestType.isEmpty || newRepLevel <= bestRepLevel) {
+ val (childType, childRepLevel, childCost) =
+ findCheapestGroupFieldRecurse(field, newRepLevel)
+ // Always prefer elements with a lower repetition level, since more nesting of arrays
+ // is likely to result in more data. At the same repetition level, prefer the smaller
+ // type.
+ if (bestType.isEmpty || childRepLevel < bestRepLevel ||
+ (childRepLevel == bestRepLevel && childCost < bestCost)) {
+ // This is the new best path.
+ bestType = Some(childType)
+ bestRepLevel = childRepLevel
+ bestCost = childCost
+ }
+ }
+ }
+ (groupType.withNewFields(bestType.get), bestRepLevel, bestCost)
+ case primitiveType: PrimitiveType =>
+ val cost = primitiveType.getPrimitiveTypeName match {
+ case PrimitiveType.PrimitiveTypeName.BOOLEAN => 1
+ case PrimitiveType.PrimitiveTypeName.INT32 => 4
+ case PrimitiveType.PrimitiveTypeName.INT64 => 8
+ case PrimitiveType.PrimitiveTypeName.FLOAT => 4
+ case PrimitiveType.PrimitiveTypeName.DOUBLE => 8
+ // Strings seem undesirable, since they don't have a fixed size. Give them a high cost.
+ case PrimitiveType.PrimitiveTypeName.BINARY |
+ PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY => 32
+ case PrimitiveType.PrimitiveTypeName.INT96 => 12
Review Comment:
shall we be defensive and add a default case with high cost if there is a new primitive type in the future?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #52557: [SPARK-53535][SQL] Fix missing structs always being assumed as nulls
URL: https://github.com/apache/spark/pull/52557
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "ZiyaZa (via GitHub)" <gi...@apache.org>.
ZiyaZa commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r2443889165
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -1655,3 +1725,25 @@ class TaskCommitFailureParquetOutputCommitter(outputPath: Path, context: TaskAtt
sys.error("Intentional exception for testing purposes")
}
}
+
+case class VerifyNoAdditionalScanOutputExec(override val child: SparkPlan) extends UnaryExecNode {
+ override def doExecute(): RDD[InternalRow] = {
+ val childOutputTypes = child.output.map(_.dataType)
+ child.executeColumnar().mapPartitionsInternal { batches =>
+ batches.flatMap { input =>
+ input.rowIterator().asScala
Review Comment:
Thanks! Simplified it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "foliveira-xonai (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53535][SQL] Fix missing structs always being assumed as nulls [spark]
Posted by "ZiyaZa (via GitHub)" <gi...@apache.org>.
ZiyaZa commented on code in PR #52557:
URL: https://github.com/apache/spark/pull/52557#discussion_r4167353994
##########
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:
`ParquetColumnVector.close()` releases all the memory associated with the column and its children.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org