Skip to content
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

unnecessary serialize/deserialize for cases where no conversion was performed #329

Open
wants to merge 4 commits into
base: branch-0.9
Choose a base branch
from
Open

unnecessary serialize/deserialize for cases where no conversion was performed #329

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0b2a7a2
unnecessary serialize/deserialize for cases where no conversion was p…
chiragaggarwal May 6, 2014
7bfc9e6
unnecessary serialize/deserialize for cases where no conversion was p…
chiragaggarwal May 23, 2014
eea72cd
Incorporated the feedback
chiragaggarwal May 29, 2014
6efb4d0
Since partValues does not change, so moving it out of iter.map
chiragaggarwal May 29, 2014
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
52 changes: 31 additions & 21 deletions src/main/scala/shark/execution/HadoopTableReader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ import org.apache.hadoop.hive.serde2.`lazy`.LazyStruct
import org.apache.hadoop.hive.serde2.objectinspector.{StructObjectInspector, ObjectInspectorConverters}
import org.apache.hadoop.io.Writable
import org.apache.hadoop.mapred.{FileInputFormat, InputFormat, JobConf}
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorConverters.IdentityConverter

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.{EmptyRDD, HadoopRDD, RDD, UnionRDD}
Expand Down Expand Up @@ -194,30 +195,39 @@ class HadoopTableReader(@transient _tableDesc: TableDesc, @transient _localHConf
partSerDe.getObjectInspector(), tblConvertedOI)
val rowWithPartArr = new Array[Object](2)
// Map each tuple to a row object
iter.map { value =>
val deserializedRow = {

// If partition schema does not match table schema, update the row to match
val convertedRow = partTblObjectInspectorConverter.convert(partSerDe.deserialize(value))

// If conversion was performed, convertedRow will be a standard Object, but if
// conversion wasn't necessary, it will still be lazy. We can't have both across
// partitions, so we serialize and deserialize again to make it lazy.
if (tableSerDe.isInstanceOf[OrcSerde]) {
convertedRow
} else {
convertedRow match {
case _: LazyStruct => convertedRow
case _: HiveColumnarStruct => convertedRow
case _ => tableSerDe.deserialize(
tableSerDe.asInstanceOf[Serializer].serialize(
convertedRow, tblConvertedOI))

// this is done per partition, and no necessary put it in the iterations (in iter.map).
rowWithPartArr.update(1, partValues)
if (partTblObjectInspectorConverter.isInstanceOf[IdentityConverter]) {
iter.map { value =>
rowWithPartArr.update(0, partSerDe.deserialize(value))
rowWithPartArr.asInstanceOf[Object]
}
}
else {
iter.map { value =>
val deserializedRow = {

// If partition schema does not match table schema, update the row to match
val convertedRow = partTblObjectInspectorConverter.convert(partSerDe.deserialize(value))

// If conversion was performed, convertedRow will be a standard Object, but if
// conversion wasn't necessary, it will still be lazy. We can't have both across
// partitions, so we serialize and deserialize again to make it lazy.
if (tableSerDe.isInstanceOf[OrcSerde]) {
convertedRow
} else {
convertedRow match {
case _: LazyStruct => convertedRow
case _: HiveColumnarStruct => convertedRow
case _ =>
tableSerDe.deserialize( tableSerDe.asInstanceOf[Serializer].serialize( convertedRow, tblConvertedOI))
}
}
}
rowWithPartArr.update(0, deserializedRow)
rowWithPartArr.asInstanceOf[Object]
}
rowWithPartArr.update(0, deserializedRow)
rowWithPartArr.update(1, partValues)
rowWithPartArr.asInstanceOf[Object]
}
}
}.toSeq
Expand Down