[SPARK-4205][SQL] Timestamp and Date with comparisons

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/timetypes.scala

@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import java.sql.{Date, Timestamp}
+import java.text.SimpleDateFormat
+import scala.language.implicitConversions
+
+/**
+ * Subclass of java.sql.Date which provides the usual comparison
+ * operators (as required for catalyst expressions) and which can
+ * be constructed from a string.
+ *
+ * {{{
+ *   scala> val d1 = Date("2014-02-01")
+ *   d1: Date = 2014-02-01
+ *
+ *   scala> val d2 = Date("2014-02-02")
+ *   d2: Date = 2014-02-02
+ * }}}
+ * 
+ * scala> d1 < d2
+ * res1: Boolean = true
+ */
+
+class RichDate(milliseconds: Long) extends Date(milliseconds) {
+  def < (that: Date): Boolean = this.before(that)
+  def > (that: Date): Boolean  = this.after(that)
+  def <= (that: Date): Boolean = (this.before(that) || this.equals(that))
+  def >= (that: Date): Boolean = (this.after(that) || this.equals(that))
+  def === (that: Date): Boolean = this.equals(that)
+  def compare(that: Date): Int = this.getTime.compare(that.getTime)
+  def format(format: String): String = {
+    val sdf = new SimpleDateFormat(format)
+    val d = new Date(this.getTime)
+    sdf.format(d)
+  }
+}
+
+object RichDate {
+  def apply(init: String) = new RichDate(Date.valueOf(init).getTime)
+
+  def unapply(date: Any): Option[RichDate] = Some(RichDate(date.toString)) 
+}
+
+/**
+ * Analogous subclass of java.sql.Timestamp.
+ *
+ * {{{
+ *   scala> val ts1 = Timestamp("2014-03-04 12:34:56.12")
+ *   ts1: Timestamp = 2014-03-04 12:34:56.12
+ *
+ *   scala> val ts2 = Timestamp("2014-03-04 12:34:56.13")
+ *   ts2: Timestamp = 2014-03-04 12:34:56.13
+ *
+ *   scala> ts1 < ts2
+ *   res13: Boolean = true
+ * }}}
+ */
+
+class RichTimestamp(milliseconds: Long) extends Timestamp(milliseconds) {
+  def < (that: Timestamp): Boolean = this.before(that)
+  def > (that: Timestamp): Boolean  = this.after(that)
+  def <= (that: Timestamp): Boolean = (this.before(that) || this.equals(that))
+  def >= (that: Timestamp): Boolean = (this.after(that) || this.equals(that))
+  def === (that: Timestamp): Boolean = this.equals(that)
+   def format(format: String): String = {
+    val sdf = new SimpleDateFormat(format)
+    val ts = new Timestamp(this.getTime)
+    sdf.format(ts)
+  }
+}
+
+object RichTimestamp {
+  def apply(init: String) = new RichTimestamp(Timestamp.valueOf(init).getTime)
+
+  def unapply(timestamp: Any): Option[RichTimestamp] = 
+    Some(RichTimestamp(timestamp.toString)) 
+}
+
+/**
+ * Implicit conversions.
+ */
+
+object TimeConversions {
+
+  implicit def javaDateToRichDate(jdate: Date): RichDate = {
+    new RichDate(jdate.getTime)
+  }
+
+  implicit def javaTimestampToRichTimestamp(jtimestamp: Timestamp): RichTimestamp = {
+    new RichTimestamp(jtimestamp.getTime)
+  }
+
+  implicit def richDateToJavaDate(date: RichDate): Date = {
+    new Date(date.getTime)
+  }
+
+  implicit def richTimestampToJavaTimestamp(timestamp: RichTimestamp): Timestamp = {
+    new Timestamp(timestamp.getTime)
+  }
+
+}

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExpressionEvaluationSuite.scala

@@ -819,4 +819,32 @@ class ExpressionEvaluationSuite extends FunSuite {
     checkEvaluation(c1 ^ c2, 3, row)
     checkEvaluation(~c1, -2, row)
   }
+
+  test("comparison operators for RichDate and RichTimestamp") {
+    import org.scalatest.Assertions.{convertToEqualizer => EQ}
+    assert(EQ(RichDate("2014-11-05") < RichDate("2014-11-06")).===(true))
+    assert(EQ(RichDate("2014-11-05") <= RichDate("2013-11-06")).===(false))
+    assert(EQ(RichTimestamp("2014-11-05 12:34:56.5432") > RichTimestamp("2014-11-05 00:00:00"))
+	   .===(true))
+    assert(EQ(RichTimestamp("2014-11-05 12:34:56") >= RichTimestamp("2014-11-06 00:00:00"))
+	   .===(false))
+  }
+
+  test("format methods for RichDate and RichTimestamp") {
+    val s1:String = RichDate("2014-11-22").format("MMMM d yyyy")
+    val s2:String = RichTimestamp("2014-11-22 12:34:56").format("MMMM d HH:mm") 
+    assert(s1 == "November 22 2014")
+    assert(s2 == "November 22 12:34")
+  }
+
+  test("implicit conversions for RichDate and RichTimestamp") {
+    import org.apache.spark.sql.catalyst.expressions.TimeConversions._
+    val d1 = RichDate("2014-01-01")
+    val d2 = javaDateToRichDate(richDateToJavaDate(d1))
+    assert(d1 === d2 )
+    val t1 = RichTimestamp("2014-01-01 12:34:56.789")
+    val t2 = javaTimestampToRichTimestamp(richTimestampToJavaTimestamp(t1))
+    assert(t1 === t2)
+  }
+
 }

sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala

@@ -502,4 +502,21 @@ class SQLContext(@transient val sparkContext: SparkContext)
 
     new SchemaRDD(this, LogicalRDD(schema.toAttributes, rowRdd)(self))
   }
+
+  /**
+   * In DSL expressions date and time can be used as aliases for RichDate
+   * and RichTimestamp:
+   * {{{
+   * val res = sqlrdd.where('date > date("2014-01-01")).select('date, 'high, 'close)
+   * }}}
+   */
+  import org.apache.spark.sql.catalyst.expressions.{RichDate, RichTimestamp}
+  val date = RichDate
+  val timestamp = RichTimestamp
+
+  /**
+   * Row Fields can be extracted asInstanceOf[RichDate] or asInstanceOf[RichTimestamp]
+   */
+  type RichDate = org.apache.spark.sql.catalyst.expressions.RichDate
+  type RichTimestamp = org.apache.spark.sql.catalyst.expressions.RichTimestamp
 }

sql/core/src/main/scala/org/apache/spark/sql/package.scala

@@ -470,4 +470,37 @@ package object sql {
    */
   @DeveloperApi
   type MetadataBuilder = catalyst.util.MetadataBuilder
+
+  /**
+   * :: DeveloperApi ::
+   *
+   * A Date class which support the standard comparison operators, for
+   * use in DSL expressions.  Implicit conversions to java.sql.Date
+   * are provided.  The class intializer accepts a String, e.g.
+   *
+   * {{{
+   * val d = RichDate("2014-01-01")
+   * }}}
+   * 
+   * @group dataType
+   */
+  @DeveloperApi
+  val RichDate = catalyst.expressions.RichDate
+
+  /**
+   * :: DeveloperApi ::
+   *
+   * A Timestamp class which support the standard comparison
+   * operators, for use in DSL expressions.  Implicit conversions to
+   * java.sql.timestamp are provided.  The class intializer accepts a
+   * String, e.g.
+   *
+   * {{{
+   * val ts = RichTimestamp("2014-01-01 12:34:56.78")
+   * }}}
+   * 
+   * @group timeClasses
+   */
+  @DeveloperApi
+  val RichTimestamp = catalyst.expressions.RichTimestamp
 }