Skip to content

Commit

Permalink
[SPARK-2190][SQL] Specialized ColumnType for Timestamp
Browse files Browse the repository at this point in the history 
JIRA issue: [SPARK-2190](https://issues.apache.org/jira/browse/SPARK-2190)

Added specialized in-memory column type for `Timestamp`. Whitelisted all timestamp related Hive tests except `timestamp_udf`, which is timezone sensitive.

Author: Cheng Lian <lian.cs.zju@gmail.com>

Closes #1440 from liancheng/timestamp-column-type and squashes the following commits:

e682175 [Cheng Lian] Enabled more timezone sensitive Hive tests.
53a358f [Cheng Lian] Fixed failed test suites
01b592d [Cheng Lian] Fixed SimpleDateFormat thread safety issue
2a59343 [Cheng Lian] Removed timezone sensitive Hive timestamp tests
45dd05d [Cheng Lian] Added Timestamp specific in-memory columnar representation
  • Loading branch information
@liancheng @marmbrus
liancheng authored and marmbrus committed Jul 21, 2014
1 parent db56f2d commit cd273a2
Show file tree
Hide file tree
Showing 133 changed files with 287 additions and 77 deletions.
1 change: 1 addition & 0 deletions .gitignore
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ conf/spark-env.sh
conf/streaming-env.sh
conf/log4j.properties
conf/spark-defaults.conf
conf/hive-site.xml
docs/_site
docs/api
target/
Expand Down
23 changes: 23 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.sql.catalyst.expressions

import java.sql.Timestamp
import java.text.{DateFormat, SimpleDateFormat}

import org.apache.spark.sql.catalyst.types._

Expand All @@ -41,6 +42,7 @@ case class Cast(child: Expression, dataType: DataType) extends UnaryExpression {
// UDFToString
private[this] def castToString: Any => Any = child.dataType match {
case BinaryType => buildCast[Array[Byte]](_, new String(_, "UTF-8"))
case TimestampType => buildCast[Timestamp](_, timestampToString)
case _ => buildCast[Any](_, _.toString)
}

Expand Down Expand Up @@ -126,6 +128,18 @@ case class Cast(child: Expression, dataType: DataType) extends UnaryExpression {
ts.getTime / 1000 + ts.getNanos.toDouble / 1000000000
}

// Converts Timestamp to string according to Hive TimestampWritable convention
private[this] def timestampToString(ts: Timestamp): String = {
val timestampString = ts.toString
val formatted = Cast.threadLocalDateFormat.get.format(ts)

if (timestampString.length > 19 && timestampString.substring(19) != ".0") {
formatted + timestampString.substring(19)
} else {
formatted
}
}

private[this] def castToLong: Any => Any = child.dataType match {
case StringType =>
buildCast[String](_, s => try s.toLong catch {
Expand Down Expand Up @@ -249,3 +263,12 @@ case class Cast(child: Expression, dataType: DataType) extends UnaryExpression {
if (evaluated == null) null else cast(evaluated)
}
}

object Cast {
// `SimpleDateFormat` is not thread-safe.
private[sql] val threadLocalDateFormat = new ThreadLocal[DateFormat] {
override def initialValue() = {
new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
}
}
}
22 changes: 11 additions & 11 deletions .../src/test/scala/org/apache/spark/sql/catalyst/expressions/ExpressionEvaluationSuite.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,7 @@ class ExpressionEvaluationSuite extends FunSuite {
checkEvaluation("abc" like regEx, true, new GenericRow(Array[Any]("a%")))
checkEvaluation("abc" like regEx, false, new GenericRow(Array[Any]("b%")))
checkEvaluation("abc" like regEx, false, new GenericRow(Array[Any]("bc%")))

checkEvaluation(Literal(null, StringType) like regEx, null, new GenericRow(Array[Any]("bc%")))
}

Expand Down Expand Up @@ -203,7 +203,7 @@ class ExpressionEvaluationSuite extends FunSuite {

test("data type casting") {

val sts = "1970-01-01 00:00:01.0"
val sts = "1970-01-01 00:00:01.1"
val ts = Timestamp.valueOf(sts)

checkEvaluation("abdef" cast StringType, "abdef")
Expand Down Expand Up @@ -293,7 +293,7 @@ class ExpressionEvaluationSuite extends FunSuite {
// A test for higher precision than millis
checkEvaluation(Cast(Cast(0.00000001, TimestampType), DoubleType), 0.00000001)
}

test("null checking") {
val row = new GenericRow(Array[Any]("^Ba*n", null, true, null))
val c1 = 'a.string.at(0)
Expand All @@ -312,7 +312,7 @@ class ExpressionEvaluationSuite extends FunSuite {

checkEvaluation(IsNull(Literal(null, ShortType)), true)
checkEvaluation(IsNotNull(Literal(null, ShortType)), false)

checkEvaluation(Coalesce(c1 :: c2 :: Nil), "^Ba*n", row)
checkEvaluation(Coalesce(Literal(null, StringType) :: Nil), null, row)
checkEvaluation(Coalesce(Literal(null, StringType) :: c1 :: c2 :: Nil), "^Ba*n", row)
Expand All @@ -323,11 +323,11 @@ class ExpressionEvaluationSuite extends FunSuite {
checkEvaluation(If(Literal(null, BooleanType), c2, c1), "^Ba*n", row)
checkEvaluation(If(Literal(true, BooleanType), c1, c2), "^Ba*n", row)
checkEvaluation(If(Literal(false, BooleanType), c2, c1), "^Ba*n", row)
checkEvaluation(If(Literal(false, BooleanType),
checkEvaluation(If(Literal(false, BooleanType),
Literal("a", StringType), Literal("b", StringType)), "b", row)

checkEvaluation(In(c1, c1 :: c2 :: Nil), true, row)
checkEvaluation(In(Literal("^Ba*n", StringType),
checkEvaluation(In(Literal("^Ba*n", StringType),
Literal("^Ba*n", StringType) :: Nil), true, row)
checkEvaluation(In(Literal("^Ba*n", StringType),
Literal("^Ba*n", StringType) :: c2 :: Nil), true, row)
Expand Down Expand Up @@ -378,7 +378,7 @@ class ExpressionEvaluationSuite extends FunSuite {

test("complex type") {
val row = new GenericRow(Array[Any](
"^Ba*n", // 0
"^Ba*n", // 0
null.asInstanceOf[String], // 1
new GenericRow(Array[Any]("aa", "bb")), // 2
Map("aa"->"bb"), // 3
Expand All @@ -391,18 +391,18 @@ class ExpressionEvaluationSuite extends FunSuite {
val typeMap = MapType(StringType, StringType)
val typeArray = ArrayType(StringType)

checkEvaluation(GetItem(BoundReference(3, AttributeReference("c", typeMap)()),
checkEvaluation(GetItem(BoundReference(3, AttributeReference("c", typeMap)()),
Literal("aa")), "bb", row)
checkEvaluation(GetItem(Literal(null, typeMap), Literal("aa")), null, row)
checkEvaluation(GetItem(Literal(null, typeMap), Literal(null, StringType)), null, row)
checkEvaluation(GetItem(BoundReference(3, AttributeReference("c", typeMap)()),
checkEvaluation(GetItem(BoundReference(3, AttributeReference("c", typeMap)()),
Literal(null, StringType)), null, row)

checkEvaluation(GetItem(BoundReference(4, AttributeReference("c", typeArray)()),
checkEvaluation(GetItem(BoundReference(4, AttributeReference("c", typeArray)()),
Literal(1)), "bb", row)
checkEvaluation(GetItem(Literal(null, typeArray), Literal(1)), null, row)
checkEvaluation(GetItem(Literal(null, typeArray), Literal(null, IntegerType)), null, row)
checkEvaluation(GetItem(BoundReference(4, AttributeReference("c", typeArray)()),
checkEvaluation(GetItem(BoundReference(4, AttributeReference("c", typeArray)()),
Literal(null, IntegerType)), null, row)

checkEvaluation(GetField(BoundReference(2, AttributeReference("c", typeS)()), "a"), "aa", row)
Expand Down
24 changes: 14 additions & 10 deletions sql/core/src/main/scala/org/apache/spark/sql/columnar/ColumnAccessor.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,9 @@ private[sql] class FloatColumnAccessor(buffer: ByteBuffer)
private[sql] class StringColumnAccessor(buffer: ByteBuffer)
extends NativeColumnAccessor(buffer, STRING)

private[sql] class TimestampColumnAccessor(buffer: ByteBuffer)
extends NativeColumnAccessor(buffer, TIMESTAMP)

private[sql] class BinaryColumnAccessor(buffer: ByteBuffer)
extends BasicColumnAccessor[BinaryType.type, Array[Byte]](buffer, BINARY)
with NullableColumnAccessor
Expand All @@ -105,16 +108,17 @@ private[sql] object ColumnAccessor {
val columnTypeId = dup.getInt()

columnTypeId match {
case INT.typeId => new IntColumnAccessor(dup)
case LONG.typeId => new LongColumnAccessor(dup)
case FLOAT.typeId => new FloatColumnAccessor(dup)
case DOUBLE.typeId => new DoubleColumnAccessor(dup)
case BOOLEAN.typeId => new BooleanColumnAccessor(dup)
case BYTE.typeId => new ByteColumnAccessor(dup)
case SHORT.typeId => new ShortColumnAccessor(dup)
case STRING.typeId => new StringColumnAccessor(dup)
case BINARY.typeId => new BinaryColumnAccessor(dup)
case GENERIC.typeId => new GenericColumnAccessor(dup)
case INT.typeId => new IntColumnAccessor(dup)
case LONG.typeId => new LongColumnAccessor(dup)
case FLOAT.typeId => new FloatColumnAccessor(dup)
case DOUBLE.typeId => new DoubleColumnAccessor(dup)
case BOOLEAN.typeId => new BooleanColumnAccessor(dup)
case BYTE.typeId => new ByteColumnAccessor(dup)
case SHORT.typeId => new ShortColumnAccessor(dup)
case STRING.typeId => new StringColumnAccessor(dup)
case TIMESTAMP.typeId => new TimestampColumnAccessor(dup)
case BINARY.typeId => new BinaryColumnAccessor(dup)
case GENERIC.typeId => new GenericColumnAccessor(dup)
}
}
}
3 changes: 3 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/columnar/ColumnBuilder.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,9 @@ private[sql] class FloatColumnBuilder extends NativeColumnBuilder(new FloatColum

private[sql] class StringColumnBuilder extends NativeColumnBuilder(new StringColumnStats, STRING)

private[sql] class TimestampColumnBuilder
extends NativeColumnBuilder(new TimestampColumnStats, TIMESTAMP)

private[sql] class BinaryColumnBuilder extends ComplexColumnBuilder(BINARY)

// TODO (lian) Add support for array, struct and map
Expand Down
37 changes: 34 additions & 3 deletions sql/core/src/main/scala/org/apache/spark/sql/columnar/ColumnStats.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -344,21 +344,52 @@ private[sql] class StringColumnStats extends BasicColumnStats(STRING) {
}

override def contains(row: Row, ordinal: Int) = {
!(upperBound eq null) && {
(upperBound ne null) && {
val field = columnType.getField(row, ordinal)
lowerBound.compareTo(field) <= 0 && field.compareTo(upperBound) <= 0
}
}

override def isAbove(row: Row, ordinal: Int) = {
!(upperBound eq null) && {
(upperBound ne null) && {
val field = columnType.getField(row, ordinal)
field.compareTo(upperBound) < 0
}
}

override def isBelow(row: Row, ordinal: Int) = {
!(lowerBound eq null) && {
(lowerBound ne null) && {
val field = columnType.getField(row, ordinal)
lowerBound.compareTo(field) < 0
}
}
}

private[sql] class TimestampColumnStats extends BasicColumnStats(TIMESTAMP) {
override def initialBounds = (null, null)

override def gatherStats(row: Row, ordinal: Int) {
val field = columnType.getField(row, ordinal)
if ((upperBound eq null) || field.compareTo(upperBound) > 0) _upper = field
if ((lowerBound eq null) || field.compareTo(lowerBound) < 0) _lower = field
}

override def contains(row: Row, ordinal: Int) = {
(upperBound ne null) && {
val field = columnType.getField(row, ordinal)
lowerBound.compareTo(field) <= 0 && field.compareTo(upperBound) <= 0
}
}

override def isAbove(row: Row, ordinal: Int) = {
(lowerBound ne null) && {
val field = columnType.getField(row, ordinal)
field.compareTo(upperBound) < 0
}
}

override def isBelow(row: Row, ordinal: Int) = {
(lowerBound ne null) && {
val field = columnType.getField(row, ordinal)
lowerBound.compareTo(field) < 0
}
Expand Down
47 changes: 35 additions & 12 deletions sql/core/src/main/scala/org/apache/spark/sql/columnar/ColumnType.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@ import java.nio.ByteBuffer

import scala.reflect.runtime.universe.TypeTag

import java.sql.Timestamp

import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.expressions.MutableRow
import org.apache.spark.sql.catalyst.types._
Expand Down Expand Up @@ -221,6 +223,26 @@ private[sql] object STRING extends NativeColumnType(StringType, 7, 8) {
override def getField(row: Row, ordinal: Int) = row.getString(ordinal)
}

private[sql] object TIMESTAMP extends NativeColumnType(TimestampType, 8, 12) {
override def extract(buffer: ByteBuffer) = {
val timestamp = new Timestamp(buffer.getLong())
timestamp.setNanos(buffer.getInt())
timestamp
}

override def append(v: Timestamp, buffer: ByteBuffer) {
buffer.putLong(v.getTime).putInt(v.getNanos)
}

override def getField(row: Row, ordinal: Int) = {
row(ordinal).asInstanceOf[Timestamp]
}

override def setField(row: MutableRow, ordinal: Int, value: Timestamp) {
row(ordinal) = value
}
}

private[sql] sealed abstract class ByteArrayColumnType[T <: DataType](
typeId: Int,
defaultSize: Int)
Expand All @@ -240,7 +262,7 @@ private[sql] sealed abstract class ByteArrayColumnType[T <: DataType](
}
}

private[sql] object BINARY extends ByteArrayColumnType[BinaryType.type](8, 16) {
private[sql] object BINARY extends ByteArrayColumnType[BinaryType.type](9, 16) {
override def setField(row: MutableRow, ordinal: Int, value: Array[Byte]) {
row(ordinal) = value
}
Expand All @@ -251,7 +273,7 @@ private[sql] object BINARY extends ByteArrayColumnType[BinaryType.type](8, 16) {
// Used to process generic objects (all types other than those listed above). Objects should be
// serialized first before appending to the column `ByteBuffer`, and is also extracted as serialized
// byte array.
private[sql] object GENERIC extends ByteArrayColumnType[DataType](9, 16) {
private[sql] object GENERIC extends ByteArrayColumnType[DataType](10, 16) {
override def setField(row: MutableRow, ordinal: Int, value: Array[Byte]) {
row(ordinal) = SparkSqlSerializer.deserialize[Any](value)
}
Expand All @@ -262,16 +284,17 @@ private[sql] object GENERIC extends ByteArrayColumnType[DataType](9, 16) {
private[sql] object ColumnType {
def apply(dataType: DataType): ColumnType[_, _] = {
dataType match {
case IntegerType => INT
case LongType => LONG
case FloatType => FLOAT
case DoubleType => DOUBLE
case BooleanType => BOOLEAN
case ByteType => BYTE
case ShortType => SHORT
case StringType => STRING
case BinaryType => BINARY
case _ => GENERIC
case IntegerType => INT
case LongType => LONG
case FloatType => FLOAT
case DoubleType => DOUBLE
case BooleanType => BOOLEAN
case ByteType => BYTE
case ShortType => SHORT
case StringType => STRING
case BinaryType => BINARY
case TimestampType => TIMESTAMP
case _ => GENERIC
}
}
}
17 changes: 9 additions & 8 deletions sql/core/src/test/scala/org/apache/spark/sql/columnar/ColumnStatsSuite.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,14 +22,15 @@ import org.scalatest.FunSuite
import org.apache.spark.sql.catalyst.types._

class ColumnStatsSuite extends FunSuite {
testColumnStats(classOf[BooleanColumnStats], BOOLEAN)
testColumnStats(classOf[ByteColumnStats], BYTE)
testColumnStats(classOf[ShortColumnStats], SHORT)
testColumnStats(classOf[IntColumnStats], INT)
testColumnStats(classOf[LongColumnStats], LONG)
testColumnStats(classOf[FloatColumnStats], FLOAT)
testColumnStats(classOf[DoubleColumnStats], DOUBLE)
testColumnStats(classOf[StringColumnStats], STRING)
testColumnStats(classOf[BooleanColumnStats], BOOLEAN)
testColumnStats(classOf[ByteColumnStats], BYTE)
testColumnStats(classOf[ShortColumnStats], SHORT)
testColumnStats(classOf[IntColumnStats], INT)
testColumnStats(classOf[LongColumnStats], LONG)
testColumnStats(classOf[FloatColumnStats], FLOAT)
testColumnStats(classOf[DoubleColumnStats], DOUBLE)
testColumnStats(classOf[StringColumnStats], STRING)
testColumnStats(classOf[TimestampColumnStats], TIMESTAMP)

def testColumnStats[T <: NativeType, U <: NativeColumnStats[T]](
columnStatsClass: Class[U],
Expand Down
Loading

0 comments on commit cd273a2

Please sign in to comment.