Skip to content

[SPARK-12286] [SPARK-12290] [SPARK-12294] [SPARK-12284] [SQL] always output UnsafeRow #10511

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.

Sign up for GitHub

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
davies wants to merge 5 commits into from
Closed

[SPARK-12286] [SPARK-12290] [SPARK-12294] [SPARK-12284] [SQL] always output UnsafeRow #10511

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
94fcd01
always output UnsafeRow
Dec 29, 2015
cba3934
output UnsafeRow from Hive
Dec 29, 2015
513597c
support UnsafeRow in InsertIntoHiveTable
Dec 29, 2015
6bc518c
fix python udf
Dec 30, 2015
90405fc
Merge branch 'master' of github.com:apache/spark into unsafe_row
Dec 30, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
output UnsafeRow from Hive
  • Loading branch information
Davies Liu committed Dec 29, 2015
commit cba393448c2d581bd62e31d3181a11e290a2a83d
6 changes: 3 additions & 3 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/Window.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -257,16 +257,16 @@ case class Window(
* @return the final resulting projection.
*/
private[this] def createResultProjection(
expressions: Seq[Expression]): MutableProjection = {
expressions: Seq[Expression]): UnsafeProjection = {
val references = expressions.zipWithIndex.map{ case (e, i) =>
// Results of window expressions will be on the right side of child's output
BoundReference(child.output.size + i, e.dataType, e.nullable)
}
val unboundToRefMap = expressions.zip(references).toMap
val patchedWindowExpression = windowExpression.map(_.transform(unboundToRefMap))
newMutableProjection(
UnsafeProjection.create(
projectList ++ patchedWindowExpression,
child.output)()
child.output)
}

protected override def doExecute(): RDD[InternalRow] = {
Expand Down
16 changes: 11 additions & 5 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTableScan.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,11 +132,17 @@ case class HiveTableScan(
}
}

protected override def doExecute(): RDD[InternalRow] = if (!relation.hiveQlTable.isPartitioned) {
hadoopReader.makeRDDForTable(relation.hiveQlTable)
} else {
hadoopReader.makeRDDForPartitionedTable(
prunePartitions(relation.getHiveQlPartitions(partitionPruningPred)))
protected override def doExecute(): RDD[InternalRow] = {
val rdd = if (!relation.hiveQlTable.isPartitioned) {
hadoopReader.makeRDDForTable(relation.hiveQlTable)
} else {
hadoopReader.makeRDDForPartitionedTable(
prunePartitions(relation.getHiveQlPartitions(partitionPruningPred)))
}
rdd.mapPartitionsInternal { iter =>
val proj = UnsafeProjection.create(schema)
iter.map(proj)
}
}

override def output: Seq[Attribute] = attributes
Expand Down
3 changes: 2 additions & 1 deletion sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/ScriptTransformation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,7 +213,8 @@ case class ScriptTransformation(

child.execute().mapPartitions { iter =>
if (iter.hasNext) {
processIterator(iter)
val proj = UnsafeProjection.create(schema)
processIterator(iter).map(proj)
} else {
// If the input iterator has no rows then do not launch the external script.
Iterator.empty
Expand Down