add Byte and Short.

Author: viirya

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SqlParser.scala

@@ -343,6 +343,8 @@ class SqlParser extends AbstractSparkSQLParser {
     val bigIntValue = BigDecimal(value)
 
     bigIntValue match {
+      case v if bigIntValue.isValidByte => v.toByteExact
+      case v if bigIntValue.isValidShort => v.toShortExact
       case v if bigIntValue.isValidInt => v.toIntExact
       case v if bigIntValue.isValidLong => v.toLongExact
       case v => v

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/complexTypes.scala

@@ -55,7 +55,11 @@ case class GetItem(child: Expression, ordinal: Expression) extends Expression {
           // TODO: consider using Array[_] for ArrayType child to avoid
           // boxing of primitives
           val baseValue = value.asInstanceOf[Seq[_]]
-          val o = key.asInstanceOf[Int]
+          val o = key match {
+            case k if k.isInstanceOf[Byte] => k.asInstanceOf[Byte]
+            case k if k.isInstanceOf[Short] => k.asInstanceOf[Short]
+            case k => k.asInstanceOf[Int]
+          }
           if (o >= baseValue.size || o < 0) {
             null
           } else {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala

@@ -47,6 +47,8 @@ object Literal {
 object IntegerLiteral {
   def unapply(a: Any): Option[Int] = a match {
     case Literal(a: Int, IntegerType) => Some(a)
+    case Literal(a: Byte, ByteType) => Some(a.toInt)
+    case Literal(a: Short, ShortType) => Some(a.toInt)
     case _ => None
   }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringOperations.scala

@@ -257,9 +257,16 @@ case class Substring(str: Expression, pos: Expression, len: Expression) extends
     if ((string == null) || (po == null) || (ln == null)) {
       null
     } else {
-      val start = po.asInstanceOf[Int]
-      val length = ln.asInstanceOf[Int] 
-
+      val start = po match {
+            case k if k.isInstanceOf[Byte] => k.asInstanceOf[Byte]
+            case k if k.isInstanceOf[Short] => k.asInstanceOf[Short]
+            case k => k.asInstanceOf[Int]
+      }
+      val length = ln match {
+            case k if k.isInstanceOf[Byte] => k.asInstanceOf[Byte]
+            case k if k.isInstanceOf[Short] => k.asInstanceOf[Short]
+            case k => k.asInstanceOf[Int]
+      }
       string match {
         case ba: Array[Byte] => slice(ba, start, length)
         case other => slice(other.toString, start, length)