From 0c0e09f567deb775ee378f5385a16884f68b332d Mon Sep 17 00:00:00 2001
From: Daoyuan Wang <daoyuan.wang@intel.com>
Date: Thu, 9 Oct 2014 14:59:03 -0700
Subject: [PATCH] [SPARK-3412][SQL]add missing row api
chenghao-intel assigned this to me, check PR #2284 for previous discussion
Author: Daoyuan Wang <daoyuan.wang@intel.com>
Closes #2529 from adrian-wang/rowapi and squashes the following commits:
c6594b2 [Daoyuan Wang] using boxed
7b7e6e3 [Daoyuan Wang] update pattern match
7a39456 [Daoyuan Wang] rename file and refresh getAs[T]
4c18c29 [Daoyuan Wang] remove setAs[T] and null judge
1614493 [Daoyuan Wang] add missing row api
---
.../sql/catalyst/expressions/Projection.scala | 15 ++++++++++++++
.../spark/sql/catalyst/expressions/Row.scala | 20 ++++++++++---------
...ificRow.scala => SpecificMutableRow.scala} | 8 ++++++--
3 files changed, 32 insertions(+), 11 deletions(-)
rename sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/{SpecificRow.scala => SpecificMutableRow.scala} (97%)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Projection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Projection.scala
index ef1d12531f..204904ecf0 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Projection.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Projection.scala
@@ -137,6 +137,9 @@ class JoinedRow extends Row {
def getString(i: Int): String =
if (i < row1.size) row1.getString(i) else row2.getString(i - row1.size)
+ override def getAs[T](i: Int): T =
+ if (i < row1.size) row1.getAs[T](i) else row2.getAs[T](i - row1.size)
+
def copy() = {
val totalSize = row1.size + row2.size
val copiedValues = new Array[Any](totalSize)
@@ -226,6 +229,9 @@ class JoinedRow2 extends Row {
def getString(i: Int): String =
if (i < row1.size) row1.getString(i) else row2.getString(i - row1.size)
+ override def getAs[T](i: Int): T =
+ if (i < row1.size) row1.getAs[T](i) else row2.getAs[T](i - row1.size)
+
def copy() = {
val totalSize = row1.size + row2.size
val copiedValues = new Array[Any](totalSize)
@@ -309,6 +315,9 @@ class JoinedRow3 extends Row {
def getString(i: Int): String =
if (i < row1.size) row1.getString(i) else row2.getString(i - row1.size)
+ override def getAs[T](i: Int): T =
+ if (i < row1.size) row1.getAs[T](i) else row2.getAs[T](i - row1.size)
+
def copy() = {
val totalSize = row1.size + row2.size
val copiedValues = new Array[Any](totalSize)
@@ -392,6 +401,9 @@ class JoinedRow4 extends Row {
def getString(i: Int): String =
if (i < row1.size) row1.getString(i) else row2.getString(i - row1.size)
+ override def getAs[T](i: Int): T =
+ if (i < row1.size) row1.getAs[T](i) else row2.getAs[T](i - row1.size)
+
def copy() = {
val totalSize = row1.size + row2.size
val copiedValues = new Array[Any](totalSize)
@@ -475,6 +487,9 @@ class JoinedRow5 extends Row {
def getString(i: Int): String =
if (i < row1.size) row1.getString(i) else row2.getString(i - row1.size)
+ override def getAs[T](i: Int): T =
+ if (i < row1.size) row1.getAs[T](i) else row2.getAs[T](i - row1.size)
+
def copy() = {
val totalSize = row1.size + row2.size
val copiedValues = new Array[Any](totalSize)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Row.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Row.scala
index d68a4fabea..d00ec39774 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Row.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Row.scala
@@ -64,6 +64,7 @@ trait Row extends Seq[Any] with Serializable {
def getShort(i: Int): Short
def getByte(i: Int): Byte
def getString(i: Int): String
+ def getAs[T](i: Int): T = apply(i).asInstanceOf[T]
override def toString() =
s"[${this.mkString(",")}]"
@@ -118,6 +119,7 @@ object EmptyRow extends Row {
def getShort(i: Int): Short = throw new UnsupportedOperationException
def getByte(i: Int): Byte = throw new UnsupportedOperationException
def getString(i: Int): String = throw new UnsupportedOperationException
+ override def getAs[T](i: Int): T = throw new UnsupportedOperationException
def copy() = this
}
@@ -217,19 +219,19 @@ class GenericMutableRow(size: Int) extends GenericRow(size) with MutableRow {
/** No-arg constructor for serialization. */
def this() = this(0)
- override def setBoolean(ordinal: Int,value: Boolean): Unit = { values(ordinal) = value }
- override def setByte(ordinal: Int,value: Byte): Unit = { values(ordinal) = value }
- override def setDouble(ordinal: Int,value: Double): Unit = { values(ordinal) = value }
- override def setFloat(ordinal: Int,value: Float): Unit = { values(ordinal) = value }
- override def setInt(ordinal: Int,value: Int): Unit = { values(ordinal) = value }
- override def setLong(ordinal: Int,value: Long): Unit = { values(ordinal) = value }
- override def setString(ordinal: Int,value: String): Unit = { values(ordinal) = value }
+ override def setBoolean(ordinal: Int, value: Boolean): Unit = { values(ordinal) = value }
+ override def setByte(ordinal: Int, value: Byte): Unit = { values(ordinal) = value }
+ override def setDouble(ordinal: Int, value: Double): Unit = { values(ordinal) = value }
+ override def setFloat(ordinal: Int, value: Float): Unit = { values(ordinal) = value }
+ override def setInt(ordinal: Int, value: Int): Unit = { values(ordinal) = value }
+ override def setLong(ordinal: Int, value: Long): Unit = { values(ordinal) = value }
+ override def setString(ordinal: Int, value: String): Unit = { values(ordinal) = value }
override def setNullAt(i: Int): Unit = { values(i) = null }
- override def setShort(ordinal: Int,value: Short): Unit = { values(ordinal) = value }
+ override def setShort(ordinal: Int, value: Short): Unit = { values(ordinal) = value }
- override def update(ordinal: Int,value: Any): Unit = { values(ordinal) = value }
+ override def update(ordinal: Int, value: Any): Unit = { values(ordinal) = value }
override def copy() = new GenericRow(values.clone())
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SpecificRow.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SpecificMutableRow.scala
similarity index 97%
rename from sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SpecificRow.scala
rename to sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SpecificMutableRow.scala
index 9cbab3d5d0..570379c533 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SpecificRow.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SpecificMutableRow.scala
@@ -233,9 +233,9 @@ final class SpecificMutableRow(val values: Array[MutableValue]) extends MutableR
override def iterator: Iterator[Any] = values.map(_.boxed).iterator
- def setString(ordinal: Int, value: String) = update(ordinal, value)
+ override def setString(ordinal: Int, value: String) = update(ordinal, value)
- def getString(ordinal: Int) = apply(ordinal).asInstanceOf[String]
+ override def getString(ordinal: Int) = apply(ordinal).asInstanceOf[String]
override def setInt(ordinal: Int, value: Int): Unit = {
val currentValue = values(ordinal).asInstanceOf[MutableInt]
@@ -306,4 +306,8 @@ final class SpecificMutableRow(val values: Array[MutableValue]) extends MutableR
override def getByte(i: Int): Byte = {
values(i).asInstanceOf[MutableByte].value
}
+
+ override def getAs[T](i: Int): T = {
+ values(i).boxed.asInstanceOf[T]
+ }
}
--
GitLab