-
Notifications
You must be signed in to change notification settings - Fork 29.3k
[SPARK-25768][SQL] fix constant argument expecting UDAFs #22766
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Closed
Closed
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -340,39 +340,39 @@ private[hive] case class HiveUDAFFunction( | |
| resolver.getEvaluator(parameterInfo) | ||
| } | ||
|
|
||
| // The UDAF evaluator used to consume raw input rows and produce partial aggregation results. | ||
| @transient | ||
| private lazy val partial1ModeEvaluator = newEvaluator() | ||
| private case class PartialEvaluator( | ||
| evaluator: GenericUDAFEvaluator, | ||
| objectInspector: ObjectInspector) | ||
|
|
||
| // The UDAF evaluator used to consume raw input rows and produce partial aggregation results. | ||
| // Hive `ObjectInspector` used to inspect partial aggregation results. | ||
| @transient | ||
| private val partialResultInspector = partial1ModeEvaluator.init( | ||
| GenericUDAFEvaluator.Mode.PARTIAL1, | ||
| inputInspectors | ||
| ) | ||
| private lazy val partial1Mode = { | ||
| val evaluator = newEvaluator() | ||
| PartialEvaluator(evaluator, evaluator.init(GenericUDAFEvaluator.Mode.PARTIAL1, inputInspectors)) | ||
| } | ||
|
|
||
| // The UDAF evaluator used to merge partial aggregation results. | ||
| @transient | ||
| private lazy val partial2ModeEvaluator = { | ||
| val evaluator = newEvaluator() | ||
| evaluator.init(GenericUDAFEvaluator.Mode.PARTIAL2, Array(partialResultInspector)) | ||
| evaluator.init(GenericUDAFEvaluator.Mode.PARTIAL2, Array(partial1Mode.objectInspector)) | ||
| evaluator | ||
| } | ||
|
|
||
| // Spark SQL data type of partial aggregation results | ||
| @transient | ||
| private lazy val partialResultDataType = inspectorToDataType(partialResultInspector) | ||
| private lazy val partialResultDataType = inspectorToDataType(partial1Mode.objectInspector) | ||
|
|
||
| // The UDAF evaluator used to compute the final result from a partial aggregation result objects. | ||
| @transient | ||
| private lazy val finalModeEvaluator = newEvaluator() | ||
|
|
||
| // Hive `ObjectInspector` used to inspect the final aggregation result object. | ||
| @transient | ||
| private val returnInspector = finalModeEvaluator.init( | ||
| GenericUDAFEvaluator.Mode.FINAL, | ||
| Array(partialResultInspector) | ||
| ) | ||
| private lazy val finalMode = { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ditto
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ah it's also used in final mode, then maybe |
||
| val evaluator = newEvaluator() | ||
| PartialEvaluator( | ||
| evaluator, | ||
| evaluator.init(GenericUDAFEvaluator.Mode.FINAL, Array(partial1Mode.objectInspector))) | ||
| } | ||
|
|
||
| // Wrapper functions used to wrap Spark SQL input arguments into Hive specific format. | ||
| @transient | ||
|
|
@@ -381,7 +381,7 @@ private[hive] case class HiveUDAFFunction( | |
| // Unwrapper function used to unwrap final aggregation result objects returned by Hive UDAFs into | ||
| // Spark SQL specific format. | ||
| @transient | ||
| private lazy val resultUnwrapper = unwrapperFor(returnInspector) | ||
| private lazy val resultUnwrapper = unwrapperFor(finalMode.objectInspector) | ||
|
|
||
| @transient | ||
| private lazy val cached: Array[AnyRef] = new Array[AnyRef](children.length) | ||
|
|
@@ -391,7 +391,7 @@ private[hive] case class HiveUDAFFunction( | |
|
|
||
| override def nullable: Boolean = true | ||
|
|
||
| override lazy val dataType: DataType = inspectorToDataType(returnInspector) | ||
| override lazy val dataType: DataType = inspectorToDataType(finalMode.objectInspector) | ||
|
|
||
| override def prettyName: String = name | ||
|
|
||
|
|
@@ -401,13 +401,13 @@ private[hive] case class HiveUDAFFunction( | |
| } | ||
|
|
||
| override def createAggregationBuffer(): AggregationBuffer = | ||
| partial1ModeEvaluator.getNewAggregationBuffer | ||
| partial1Mode.evaluator.getNewAggregationBuffer | ||
|
|
||
| @transient | ||
| private lazy val inputProjection = UnsafeProjection.create(children) | ||
|
|
||
| override def update(buffer: AggregationBuffer, input: InternalRow): AggregationBuffer = { | ||
| partial1ModeEvaluator.iterate( | ||
| partial1Mode.evaluator.iterate( | ||
| buffer, wrap(inputProjection(input), inputWrappers, cached, inputDataTypes)) | ||
| buffer | ||
| } | ||
|
|
@@ -417,12 +417,12 @@ private[hive] case class HiveUDAFFunction( | |
| // buffer in the 3rd format mentioned in the ScalaDoc of this class. Originally, Hive converts | ||
| // this `AggregationBuffer`s into this format before shuffling partial aggregation results, and | ||
| // calls `GenericUDAFEvaluator.terminatePartial()` to do the conversion. | ||
| partial2ModeEvaluator.merge(buffer, partial1ModeEvaluator.terminatePartial(input)) | ||
| partial2ModeEvaluator.merge(buffer, partial1Mode.evaluator.terminatePartial(input)) | ||
| buffer | ||
| } | ||
|
|
||
| override def eval(buffer: AggregationBuffer): Any = { | ||
| resultUnwrapper(finalModeEvaluator.terminate(buffer)) | ||
| resultUnwrapper(finalMode.evaluator.terminate(buffer)) | ||
| } | ||
|
|
||
| override def serialize(buffer: AggregationBuffer): Array[Byte] = { | ||
|
|
@@ -439,9 +439,10 @@ private[hive] case class HiveUDAFFunction( | |
|
|
||
| // Helper class used to de/serialize Hive UDAF `AggregationBuffer` objects | ||
| private class AggregationBufferSerDe { | ||
| private val partialResultUnwrapper = unwrapperFor(partialResultInspector) | ||
| private val partialResultUnwrapper = unwrapperFor(partial1Mode.objectInspector) | ||
|
|
||
| private val partialResultWrapper = wrapperFor(partialResultInspector, partialResultDataType) | ||
| private val partialResultWrapper = | ||
| wrapperFor(partial1Mode.objectInspector, partialResultDataType) | ||
|
|
||
| private val projection = UnsafeProjection.create(Array(partialResultDataType)) | ||
|
|
||
|
|
@@ -451,7 +452,7 @@ private[hive] case class HiveUDAFFunction( | |
| // `GenericUDAFEvaluator.terminatePartial()` converts an `AggregationBuffer` into an object | ||
| // that can be inspected by the `ObjectInspector` returned by `GenericUDAFEvaluator.init()`. | ||
| // Then we can unwrap it to a Spark SQL value. | ||
| mutableRow.update(0, partialResultUnwrapper(partial1ModeEvaluator.terminatePartial(buffer))) | ||
| mutableRow.update(0, partialResultUnwrapper(partial1Mode.evaluator.terminatePartial(buffer))) | ||
| val unsafeRow = projection(mutableRow) | ||
| val bytes = ByteBuffer.allocate(unsafeRow.getSizeInBytes) | ||
| unsafeRow.writeTo(bytes) | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
partial1ModeEvaluator