phoenix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jmaho...@apache.org
Subject phoenix git commit: PHOENIX-2469 Problem with arrays in phoenix-spark (Dawid Wysakowicz)
Date Wed, 02 Dec 2015 14:22:07 GMT
Repository: phoenix
Updated Branches:
  refs/heads/4.x-HBase-0.98 06f1cbcdc -> 093cb9969


PHOENIX-2469 Problem with arrays in phoenix-spark (Dawid Wysakowicz)


Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/093cb996
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/093cb996
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/093cb996

Branch: refs/heads/4.x-HBase-0.98
Commit: 093cb9969715afe604a9ed3706290964c8e5562c
Parents: 06f1cbc
Author: Josh Mahonin <jmahonin@interset.com>
Authored: Wed Dec 2 09:17:11 2015 -0500
Committer: Josh Mahonin <jmahonin@interset.com>
Committed: Wed Dec 2 09:21:56 2015 -0500

----------------------------------------------------------------------
 phoenix-spark/pom.xml                           |   2 +-
 phoenix-spark/src/it/resources/setup.sql        |   7 ++
 .../apache/phoenix/spark/PhoenixSparkIT.scala   | 112 ++++++++++++++++++-
 .../phoenix/spark/PhoenixRecordWritable.scala   |  45 +++++---
 4 files changed, 145 insertions(+), 21 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/phoenix/blob/093cb996/phoenix-spark/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-spark/pom.xml b/phoenix-spark/pom.xml
index cb85531..bc77a26 100644
--- a/phoenix-spark/pom.xml
+++ b/phoenix-spark/pom.xml
@@ -34,7 +34,7 @@
   <name>Phoenix - Spark</name>
 
   <properties>
-    <spark.version>1.5.0</spark.version>
+    <spark.version>1.5.2</spark.version>
     <scala.version>2.10.4</scala.version>
     <scala.binary.version>2.10</scala.binary.version>
     <top.dir>${project.basedir}/..</top.dir>

http://git-wip-us.apache.org/repos/asf/phoenix/blob/093cb996/phoenix-spark/src/it/resources/setup.sql
----------------------------------------------------------------------
diff --git a/phoenix-spark/src/it/resources/setup.sql b/phoenix-spark/src/it/resources/setup.sql
index 814d311..137d019 100644
--- a/phoenix-spark/src/it/resources/setup.sql
+++ b/phoenix-spark/src/it/resources/setup.sql
@@ -30,6 +30,13 @@ UPSERT INTO "table3" ("id", "col1") VALUES (1, 'foo')
 UPSERT INTO "table3" ("id", "col1") VALUES (2, 'bar')
 CREATE TABLE ARRAY_TEST_TABLE (ID BIGINT NOT NULL PRIMARY KEY, VCARRAY VARCHAR[])
 UPSERT INTO ARRAY_TEST_TABLE (ID, VCARRAY) VALUES (1, ARRAY['String1', 'String2', 'String3'])
+CREATE TABLE ARRAYBUFFER_TEST_TABLE (ID BIGINT NOT NULL PRIMARY KEY, VCARRAY VARCHAR[], INTARRAY
INTEGER[])
+UPSERT INTO ARRAYBUFFER_TEST_TABLE (ID, VCARRAY, INTARRAY) VALUES (1, ARRAY['String1', 'String2',
'String3'], ARRAY[1, 2, 3])
+CREATE TABLE ARRAY_ANYVAL_TEST_TABLE (ID BIGINT NOT NULL PRIMARY KEY, INTARRAY INTEGER[],
BIGINTARRAY BIGINT[])
+UPSERT INTO ARRAY_ANYVAL_TEST_TABLE (ID, INTARRAY, BIGINTARRAY) VALUES (1, ARRAY[1, 2, 3],
ARRAY[1, 2, 3])
+CREATE TABLE ARRAY_BYTE_TEST_TABLE (ID BIGINT NOT NULL PRIMARY KEY, BYTEARRAY TINYINT[])
+UPSERT INTO ARRAY_BYTE_TEST_TABLE (ID, BYTEARRAY) VALUES (1, ARRAY[1, 2, 3])
+CREATE TABLE VARBINARY_TEST_TABLE (ID BIGINT NOT NULL PRIMARY KEY, BIN BINARY(1), VARBIN
VARBINARY, BINARRAY BINARY(1)[])
 CREATE TABLE DATE_PREDICATE_TEST_TABLE (ID BIGINT NOT NULL, TIMESERIES_KEY TIMESTAMP NOT
NULL CONSTRAINT pk PRIMARY KEY (ID, TIMESERIES_KEY))
 UPSERT INTO DATE_PREDICATE_TEST_TABLE (ID, TIMESERIES_KEY) VALUES (1, CAST(CURRENT_TIME()
AS TIMESTAMP))
 CREATE TABLE OUTPUT_TEST_TABLE (id BIGINT NOT NULL PRIMARY KEY, col1 VARCHAR, col2 INTEGER,
col3 DATE)

http://git-wip-us.apache.org/repos/asf/phoenix/blob/093cb996/phoenix-spark/src/it/scala/org/apache/phoenix/spark/PhoenixSparkIT.scala
----------------------------------------------------------------------
diff --git a/phoenix-spark/src/it/scala/org/apache/phoenix/spark/PhoenixSparkIT.scala b/phoenix-spark/src/it/scala/org/apache/phoenix/spark/PhoenixSparkIT.scala
index a34670e..2bc7850 100644
--- a/phoenix-spark/src/it/scala/org/apache/phoenix/spark/PhoenixSparkIT.scala
+++ b/phoenix-spark/src/it/scala/org/apache/phoenix/spark/PhoenixSparkIT.scala
@@ -16,19 +16,17 @@ package org.apache.phoenix.spark
 import java.sql.{Connection, DriverManager}
 import java.util.Date
 
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.hbase.{HBaseConfiguration, HConstants, HBaseTestingUtility}
+import org.apache.hadoop.hbase.{HConstants}
 import org.apache.phoenix.end2end.BaseHBaseManagedTimeIT
 import org.apache.phoenix.query.BaseTest
-import org.apache.phoenix.schema.{TableNotFoundException, ColumnNotFoundException}
+import org.apache.phoenix.schema.{ColumnNotFoundException}
 import org.apache.phoenix.schema.types.PVarchar
 import org.apache.phoenix.util.{SchemaUtil, ColumnInfo}
-import org.apache.spark.sql.{Row, SaveMode, execution, SQLContext}
+import org.apache.spark.sql.{Row, SaveMode, SQLContext}
 import org.apache.spark.sql.types._
 import org.apache.spark.{SparkConf, SparkContext}
 import org.joda.time.DateTime
 import org.scalatest._
-import org.apache.phoenix.spark._
 
 import scala.collection.mutable.ListBuffer
 
@@ -469,7 +467,7 @@ class PhoenixSparkIT extends FunSuite with Matchers with BeforeAndAfterAll
{
 
     df.saveToPhoenix("TABLE2", zkUrl = Some(quorumAddress))
   }
-  
+
   test("Ensure Dataframe supports LIKE and IN filters (PHOENIX-2328)") {
     val sqlContext = new SQLContext(sc)
     val df = sqlContext.load("org.apache.phoenix.spark", Map("table" -> "TABLE1",
@@ -526,4 +524,106 @@ class PhoenixSparkIT extends FunSuite with Matchers with BeforeAndAfterAll
{
     assert(df.select("COL1").first().getShort(0).toInt == 32767)
     assert(df.select("COL2").first().getByte(0).toInt == 127)
   }
+
+  test("Can save arrays from custom dataframes back to phoenix") {
+    val dataSet = List(Row(2L, Array("String1", "String2", "String3"), Array(1, 2, 3)))
+
+    val sqlContext = new SQLContext(sc)
+
+    val schema = StructType(
+      Seq(StructField("ID", LongType, nullable = false),
+        StructField("VCARRAY", ArrayType(StringType)),
+        StructField("INTARRAY", ArrayType(IntegerType))))
+
+    val rowRDD = sc.parallelize(dataSet)
+
+    // Apply the schema to the RDD.
+    val df = sqlContext.createDataFrame(rowRDD, schema)
+
+    df.write
+      .format("org.apache.phoenix.spark")
+      .options(Map("table" -> "ARRAYBUFFER_TEST_TABLE", "zkUrl" -> quorumAddress))
+      .mode(SaveMode.Overwrite)
+      .save()
+
+    // Load the results back
+    val stmt = conn.createStatement()
+    val rs = stmt.executeQuery("SELECT VCARRAY, INTARRAY FROM ARRAYBUFFER_TEST_TABLE WHERE
ID = 2")
+    rs.next()
+    val stringArray = rs.getArray(1).getArray().asInstanceOf[Array[String]]
+    val intArray = rs.getArray(2).getArray().asInstanceOf[Array[Int]]
+
+    // Verify the arrays are equal
+    stringArray shouldEqual dataSet(0).getAs[Array[String]](1)
+    intArray shouldEqual dataSet(0).getAs[Array[Int]](2)
+  }
+
+  test("Can save arrays of AnyVal type back to phoenix") {
+    val dataSet = List((2L, Array(1, 2, 3), Array(1L, 2L, 3L)))
+
+    sc
+      .parallelize(dataSet)
+      .saveToPhoenix(
+        "ARRAY_ANYVAL_TEST_TABLE",
+        Seq("ID", "INTARRAY", "BIGINTARRAY"),
+        zkUrl = Some(quorumAddress)
+      )
+
+    // Load the results back
+    val stmt = conn.createStatement()
+    val rs = stmt.executeQuery("SELECT INTARRAY, BIGINTARRAY FROM ARRAY_ANYVAL_TEST_TABLE
WHERE ID = 2")
+    rs.next()
+    val intArray = rs.getArray(1).getArray().asInstanceOf[Array[Int]]
+    val longArray = rs.getArray(2).getArray().asInstanceOf[Array[Long]]
+
+    // Verify the arrays are equal
+    intArray shouldEqual dataSet(0)._2
+    longArray shouldEqual dataSet(0)._3
+  }
+
+  test("Can save arrays of Byte type back to phoenix") {
+    val dataSet = List((2L, Array(1.toByte, 2.toByte, 3.toByte)))
+
+    sc
+      .parallelize(dataSet)
+      .saveToPhoenix(
+        "ARRAY_BYTE_TEST_TABLE",
+        Seq("ID", "BYTEARRAY"),
+        zkUrl = Some(quorumAddress)
+      )
+
+    // Load the results back
+    val stmt = conn.createStatement()
+    val rs = stmt.executeQuery("SELECT BYTEARRAY FROM ARRAY_BYTE_TEST_TABLE WHERE ID = 2")
+    rs.next()
+    val byteArray = rs.getArray(1).getArray().asInstanceOf[Array[Byte]]
+
+    // Verify the arrays are equal
+    byteArray shouldEqual dataSet(0)._2
+  }
+
+  test("Can save binary types back to phoenix") {
+    val dataSet = List((2L, Array[Byte](1), Array[Byte](1,2,3), Array[Array[Byte]](Array[Byte](1),
Array[Byte](2))))
+
+    sc
+      .parallelize(dataSet)
+      .saveToPhoenix(
+        "VARBINARY_TEST_TABLE",
+        Seq("ID", "BIN", "VARBIN", "BINARRAY"),
+        zkUrl = Some(quorumAddress)
+      )
+
+    // Load the results back
+    val stmt = conn.createStatement()
+    val rs = stmt.executeQuery("SELECT BIN, VARBIN, BINARRAY FROM VARBINARY_TEST_TABLE WHERE
ID = 2")
+    rs.next()
+    val byte = rs.getBytes("BIN")
+    val varByte = rs.getBytes("VARBIN")
+    val varByteArray = rs.getArray("BINARRAY").getArray().asInstanceOf[Array[Array[Byte]]]
+
+    // Verify the arrays are equal
+    byte shouldEqual dataSet(0)._2
+    varByte shouldEqual dataSet(0)._3
+    varByteArray shouldEqual dataSet(0)._4
+  }
 }

http://git-wip-us.apache.org/repos/asf/phoenix/blob/093cb996/phoenix-spark/src/main/scala/org/apache/phoenix/spark/PhoenixRecordWritable.scala
----------------------------------------------------------------------
diff --git a/phoenix-spark/src/main/scala/org/apache/phoenix/spark/PhoenixRecordWritable.scala
b/phoenix-spark/src/main/scala/org/apache/phoenix/spark/PhoenixRecordWritable.scala
index f11f9cc..c35cc54 100644
--- a/phoenix-spark/src/main/scala/org/apache/phoenix/spark/PhoenixRecordWritable.scala
+++ b/phoenix-spark/src/main/scala/org/apache/phoenix/spark/PhoenixRecordWritable.scala
@@ -14,13 +14,11 @@
 package org.apache.phoenix.spark
 
 import java.sql.{PreparedStatement, ResultSet}
-import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.mapreduce.lib.db.DBWritable
-import org.apache.phoenix.schema.types.{PDataType, PDate, PhoenixArray}
+import org.apache.phoenix.schema.types._
 import org.apache.phoenix.util.ColumnInfo
 import org.joda.time.DateTime
-import scala.collection.{immutable, mutable}
-import scala.collection.JavaConversions._
+import scala.collection.{mutable, immutable}
 
 
 class PhoenixRecordWritable(columnMetaDataList: List[ColumnInfo]) extends DBWritable {
@@ -52,16 +50,35 @@ class PhoenixRecordWritable(columnMetaDataList: List[ColumnInfo]) extends
DBWrit
             case _ => (v, c.getPDataType)
           }
 
-          // Save as array or object
-          finalObj match {
-            case obj: Array[AnyRef] => {
-              // Create a java.sql.Array, need to lookup the base sql type name
-              val sqlArray = statement.getConnection.createArrayOf(
-                PDataType.arrayBaseType(finalType).getSqlTypeName,
-                obj
-              )
-              statement.setArray(i + 1, sqlArray)
-            }
+
+          // Helper method to create an SQL array for a specific PDatatype, and set it on
the statement
+          def setArrayInStatement(obj: Array[AnyRef]): Unit = {
+            // Create a java.sql.Array, need to lookup the base sql type name
+            val sqlArray = statement.getConnection.createArrayOf(
+              PDataType.arrayBaseType(finalType).getSqlTypeName,
+              obj
+            )
+            statement.setArray(i + 1, sqlArray)
+          }
+
+          // Determine whether to save as an array or object
+          (finalObj, finalType) match {
+            case (obj: Array[AnyRef], _) => setArrayInStatement(obj)
+            case (obj: mutable.ArrayBuffer[AnyVal], _) => setArrayInStatement(obj.map(_.asInstanceOf[AnyRef]).toArray)
+            case (obj: mutable.ArrayBuffer[AnyRef], _) => setArrayInStatement(obj.toArray)
+            case (obj: mutable.WrappedArray[AnyVal], _) => setArrayInStatement(obj.map(_.asInstanceOf[AnyRef]).toArray)
+            case (obj: mutable.WrappedArray[AnyRef], _) => setArrayInStatement(obj.toArray)
+            case (obj: Array[Int], _) => setArrayInStatement(obj.map(_.asInstanceOf[AnyRef]))
+            case (obj: Array[Long], _) => setArrayInStatement(obj.map(_.asInstanceOf[AnyRef]))
+            case (obj: Array[Char], _) => setArrayInStatement(obj.map(_.asInstanceOf[AnyRef]))
+            case (obj: Array[Short], _) => setArrayInStatement(obj.map(_.asInstanceOf[AnyRef]))
+            case (obj: Array[Float], _) => setArrayInStatement(obj.map(_.asInstanceOf[AnyRef]))
+            case (obj: Array[Double], _) => setArrayInStatement(obj.map(_.asInstanceOf[AnyRef]))
+            // PVarbinary and PBinary come in as Array[Byte] but they're SQL objects
+            case (obj: Array[Byte], _ : PVarbinary) => statement.setObject(i + 1, obj)
+            case (obj: Array[Byte], _ : PBinary) => statement.setObject(i + 1, obj)
+            // Otherwise set as array type
+            case (obj: Array[Byte], _) => setArrayInStatement(obj.map(_.asInstanceOf[AnyRef]))
             case _ => statement.setObject(i + 1, finalObj)
           }
         } else {


Mime
View raw message