You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "szehon-ho (via GitHub)" <gi...@apache.org> on 2025/09/10 23:53:24 UTC
[PR] [SPARK-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
szehon-ho opened a new pull request, #52299:
URL: https://github.com/apache/spark/pull/52299
### What changes were proposed in this pull request?
InMemoryTable does not handle schema evolution properly. In particular, it always uses the current schema to read the in-memory rows. For newly-added columns that don't exist in the row, the InternalRow sometimes return some junk value.
This fixes InMemoryDataSource to properly handle schema evolution. Like modern table format, it persists the 'writeSchema' of the row batch, and then at read time resolves it with the 'readSchema'.
### Why are the changes needed?
While writing tests for MERGE INTO schema evolution, realized that in these cases InMemoryTable is returning junk values, unlike a real V2 data source.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Add a new test in AlterTableTests
### 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-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #52299:
URL: https://github.com/apache/spark/pull/52299#discussion_r2338235671
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala:
##########
@@ -71,28 +71,35 @@ class InMemoryTable(
}
override def withData(data: Array[BufferedRows]): InMemoryTable = {
- withData(data, schema)
+ withData(data, schema(), newData = true)
}
override def withData(
data: Array[BufferedRows],
- writeSchema: StructType): InMemoryTable = {
+ writeSchema: StructType,
+ newData: Boolean): InMemoryTable = {
Review Comment:
because its used by alterTable, we have a flag newData=false to preserve the original 'writeSchema' for the BufferedRows
--
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-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #52299:
URL: https://github.com/apache/spark/pull/52299#discussion_r2345529454
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala:
##########
@@ -184,27 +184,28 @@ class InMemoryRowLevelOperationTable(
private object TestDeltaBatchWrite extends RowLevelOperationBatchWrite with DeltaBatchWrite{
override def createBatchWriterFactory(info: PhysicalWriteInfo): DeltaWriterFactory = {
- DeltaBufferedRowsWriterFactory
+ new DeltaBufferedRowsWriterFactory(CatalogV2Util.v2ColumnsToStructType(columns()))
Review Comment:
Note: Array[Column] is not serializable, so hence passing StructType to these BatchWriterFactory and BatchWriters
--
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-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #52299:
URL: https://github.com/apache/spark/pull/52299#discussion_r2345529454
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala:
##########
@@ -184,27 +184,28 @@ class InMemoryRowLevelOperationTable(
private object TestDeltaBatchWrite extends RowLevelOperationBatchWrite with DeltaBatchWrite{
override def createBatchWriterFactory(info: PhysicalWriteInfo): DeltaWriterFactory = {
- DeltaBufferedRowsWriterFactory
+ new DeltaBufferedRowsWriterFactory(CatalogV2Util.v2ColumnsToStructType(columns()))
Review Comment:
Array[Column] is not serializable, so hence passing StructType to these BatchWriterFactory and BatchWriters
--
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-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #52299:
URL: https://github.com/apache/spark/pull/52299#discussion_r2338235671
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala:
##########
@@ -71,28 +71,35 @@ class InMemoryTable(
}
override def withData(data: Array[BufferedRows]): InMemoryTable = {
- withData(data, schema)
+ withData(data, schema(), newData = true)
}
override def withData(
data: Array[BufferedRows],
- writeSchema: StructType): InMemoryTable = {
+ writeSchema: StructType,
+ newData: Boolean): InMemoryTable = {
Review Comment:
because its used by alterTable, we should preserve the original 'writeSchema' for the BufferedRows
--
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-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #52299:
URL: https://github.com/apache/spark/pull/52299#discussion_r2345528597
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala:
##########
@@ -71,28 +71,35 @@ class InMemoryTable(
}
override def withData(data: Array[BufferedRows]): InMemoryTable = {
- withData(data, schema)
+ withData(data, schema(), newData = true)
}
override def withData(
data: Array[BufferedRows],
- writeSchema: StructType): InMemoryTable = {
+ writeSchema: StructType,
+ newData: Boolean): InMemoryTable = {
Review Comment:
I end up, making another method for alterTable to not pollute existing logic with if-else. That version preserves the 'writeSchema' on the BufferedRows.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSessionCatalogSuite.scala:
##########
@@ -133,7 +133,9 @@ class InMemoryTableSessionCatalog extends TestV2SessionCatalogBase[InMemoryTable
}
val newTable = new InMemoryTable(table.name, schema, table.partitioning, properties)
- .withData(table.data)
+ .withData(table.data,
Review Comment:
done
--
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-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52299:
URL: https://github.com/apache/spark/pull/52299#discussion_r2347962456
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -329,20 +335,48 @@ abstract class InMemoryBaseTable(
val key = getKey(row, writeSchema)
dataMap += dataMap.get(key)
.map { splits =>
- val newSplits = if (splits.last.rows.size >= numRowsPerSplit) {
- splits :+ new BufferedRows(key)
+ val newSplits = if ((splits.last.rows.size >= numRowsPerSplit)
+ || (splits.last.schema != writeSchema)) {
+ splits :+ new BufferedRows(key, writeSchema)
} else {
splits
}
newSplits.last.withRow(row)
key -> newSplits
}
- .getOrElse(key -> Seq(new BufferedRows(key).withRow(row)))
+ .getOrElse(key -> Seq(
+ new BufferedRows(key, writeSchema).withRow(row)))
addPartitionKey(key)
})
this
}
+ def alterTableWithData(data: Array[BufferedRows],
+ newSchema: StructType): InMemoryBaseTable = {
Review Comment:
style nit:
```suggestion
def alterTableWithData(
data: Array[BufferedRows],
newSchema: StructType): InMemoryBaseTable = {
```
--
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-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52299:
URL: https://github.com/apache/spark/pull/52299#discussion_r2341042704
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -718,7 +723,9 @@ object InMemoryBaseTable {
}
}
-class BufferedRows(val key: Seq[Any] = Seq.empty) extends WriterCommitMessage
+class BufferedRows(val key: Seq[Any],
+ val schema: StructType)
Review Comment:
shall we add a classdoc? This class is a bit complicated now.
--
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-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52299:
URL: https://github.com/apache/spark/pull/52299#discussion_r2347962967
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -329,20 +335,48 @@ abstract class InMemoryBaseTable(
val key = getKey(row, writeSchema)
dataMap += dataMap.get(key)
.map { splits =>
- val newSplits = if (splits.last.rows.size >= numRowsPerSplit) {
- splits :+ new BufferedRows(key)
+ val newSplits = if ((splits.last.rows.size >= numRowsPerSplit)
+ || (splits.last.schema != writeSchema)) {
+ splits :+ new BufferedRows(key, writeSchema)
} else {
splits
}
newSplits.last.withRow(row)
key -> newSplits
}
- .getOrElse(key -> Seq(new BufferedRows(key).withRow(row)))
+ .getOrElse(key -> Seq(
+ new BufferedRows(key, writeSchema).withRow(row)))
addPartitionKey(key)
})
this
}
+ def alterTableWithData(data: Array[BufferedRows],
+ newSchema: StructType): InMemoryBaseTable = {
Review Comment:
You can fix them in your next PR.
--
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-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52299:
URL: https://github.com/apache/spark/pull/52299#discussion_r2341038091
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSessionCatalogSuite.scala:
##########
@@ -133,7 +133,9 @@ class InMemoryTableSessionCatalog extends TestV2SessionCatalogBase[InMemoryTable
}
val newTable = new InMemoryTable(table.name, schema, table.partitioning, properties)
- .withData(table.data)
+ .withData(table.data,
Review Comment:
shall we add a overload of `withData` that takes `Array[Column]`?
--
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-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #52299:
URL: https://github.com/apache/spark/pull/52299#discussion_r2338235671
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala:
##########
@@ -71,28 +71,35 @@ class InMemoryTable(
}
override def withData(data: Array[BufferedRows]): InMemoryTable = {
- withData(data, schema)
+ withData(data, schema(), newData = true)
}
override def withData(
data: Array[BufferedRows],
- writeSchema: StructType): InMemoryTable = {
+ writeSchema: StructType,
+ newData: Boolean): InMemoryTable = {
Review Comment:
this method is used by alterTable, we have a flag newData=false to preserve the original 'writeSchema' for the BufferedRows in the alterTable case instead of using the new schema.
--
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-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #52299: [SPARK-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields
URL: https://github.com/apache/spark/pull/52299
--
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-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #52299:
URL: https://github.com/apache/spark/pull/52299#issuecomment-3290622175
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-53546][SQL][TESTS] Fix InMemoryDataSource to return default value or null for new fields [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #52299:
URL: https://github.com/apache/spark/pull/52299#discussion_r2338235671
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala:
##########
@@ -71,28 +71,35 @@ class InMemoryTable(
}
override def withData(data: Array[BufferedRows]): InMemoryTable = {
- withData(data, schema)
+ withData(data, schema(), newData = true)
}
override def withData(
data: Array[BufferedRows],
- writeSchema: StructType): InMemoryTable = {
+ writeSchema: StructType,
+ newData: Boolean): InMemoryTable = {
Review Comment:
this method is used by alterTable and regular write, so we have a flag newData=false to preserve the original 'writeSchema' for the BufferedRows in the alterTable case instead of using the new schema. In the write case, we mark the new rows with the new schema.
--
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