Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/18483#discussion_r124978673
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/SortBasedAggregationIterator.scala
---
@@ -86,17 +86,6 @@ class SortBasedAggregationIterator(
// The aggregation buffer used by the sort-based aggregation.
private[this] val sortBasedAggregationBuffer: InternalRow = newBuffer
- // This safe projection is used to turn the input row into safe row.
This is necessary
- // because the input row may be produced by unsafe projection in child
operator and all the
- // produced rows share one byte array. However, when we update the
aggregate buffer according to
- // the input row, we may cache some values from input row, e.g. `Max`
will keep the max value from
- // input row via MutableProjection, `CollectList` will keep all values
in an array via
- // ImperativeAggregate framework. These values may get changed
unexpectedly if the underlying
- // unsafe projection update the shared byte array. By applying a safe
projection to the input row,
- // we can cut down the connection from input row to the shared byte
array, and thus it's safe to
- // cache values from input row while updating the aggregation buffer.
- private[this] val safeProj: Projection =
FromUnsafeProjection(valueAttributes.map(_.dataType))
--- End diff --
when we save values to the aggregate buffer(an InternalRow), we will do
copy properly, so this projection is not needed anymore.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
