Fix tests

Author: rejeb

phoenix5-spark/src/it/resources/globalSetup.sql

@@ -50,9 +50,9 @@ UPSERT INTO TEST_DECIMAL VALUES (1, 123.456789)
 CREATE TABLE TEST_SMALL_TINY (ID BIGINT NOT NULL PRIMARY KEY, COL1 SMALLINT, COL2 TINYINT)
 UPSERT INTO TEST_SMALL_TINY VALUES (1, 32767, 127)
 CREATE TABLE DATE_TEST(ID BIGINT NOT NULL PRIMARY KEY, COL1 DATE)
-UPSERT INTO DATE_TEST VALUES(1, CURRENT_DATE())
+UPSERT INTO DATE_TEST VALUES(1, '2021-01-01T00:00:00Z')
 CREATE TABLE TIME_TEST(ID BIGINT NOT NULL PRIMARY KEY, COL1 TIME)
-UPSERT INTO TIME_TEST VALUES(1, CURRENT_TIME())
+UPSERT INTO TIME_TEST VALUES(1, '2021-01-01T00:00:00Z')
 CREATE TABLE "space" ("key" VARCHAR PRIMARY KEY, "first name" VARCHAR)
 UPSERT INTO "space" VALUES ('key1', 'xyz')
 CREATE TABLE "small" ("key" VARCHAR PRIMARY KEY, "first name" VARCHAR, "salary" INTEGER )

phoenix5-spark/src/it/scala/org/apache/phoenix/spark/AbstractPhoenixSparkIT.scala

@@ -14,13 +14,13 @@
 package org.apache.phoenix.spark
 
 import java.sql.{Connection, DriverManager}
-import java.util.Properties
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HConstants;
+import java.util.{Properties, TimeZone}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.hbase.HConstants
 import org.apache.phoenix.query.BaseTest
+import org.apache.phoenix.spark.datasource.v2.writer.PhoenixTestingDataSourceWriter
 import org.apache.phoenix.util.PhoenixRuntime
-import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.ReadOnlyProps
 import org.apache.spark.sql.SparkSession
 import org.scalatest.{BeforeAndAfter, BeforeAndAfterAll, FunSuite, Matchers}
 
@@ -30,6 +30,8 @@ object PhoenixSparkITHelper extends BaseTest {
   def getTestClusterConfig = new Configuration(BaseTest.config);
 
   def doSetup = {
+    // Set-up fixed timezone for DATE and TIMESTAMP tests
+    TimeZone.setDefault(TimeZone.getTimeZone("UTC"))
     // The @ClassRule doesn't seem to be getting picked up, force creation here before setup
     BaseTest.tmpFolder.create()
     BaseTest.setUpTestDriver(ReadOnlyProps.EMPTY_PROPS);
@@ -92,6 +94,11 @@ class AbstractPhoenixSparkIT extends FunSuite with Matchers with BeforeAndAfter
     conn.commit()
   }
 
+  before{
+    // Reset batch counter after each test
+    PhoenixTestingDataSourceWriter.TOTAL_BATCHES_COMMITTED_COUNT = 0
+  }
+
   override def beforeAll() {
     PhoenixSparkITHelper.doSetup
 

phoenix5-spark/src/it/scala/org/apache/phoenix/spark/PhoenixSparkDatasourceV1IT.scala

@@ -24,6 +24,7 @@ import org.apache.spark.sql.types._
 import org.apache.spark.sql.{Row, SaveMode}
 
 import java.sql.DriverManager
+import java.time.{LocalDate, ZoneId}
 import java.util.Date
 import scala.collection.mutable
 import scala.collection.mutable.ListBuffer
@@ -243,30 +244,6 @@ class PhoenixSparkDatasourceV1IT extends AbstractPhoenixSparkIT {
     count shouldEqual 1L
   }
 
-  test("Can use extraOptions to set configs for workers during reads") {
-    // Pass in true, so we will get null when fetching the current row, leading to an NPE
-    var extraOptions = PhoenixTestingInputPartitionReader.RETURN_NULL_CURR_ROW + "=true"
-    var rdd = spark.sqlContext.read
-      .format(PhoenixTestingDataSource.TEST_SOURCE)
-      .options( Map("table" -> "TABLE1", PhoenixDataSource.ZOOKEEPER_URL -> quorumAddress,
-        PhoenixDataSource.PHOENIX_CONFIGS -> extraOptions)).load
-
-    // Expect to get a NullPointerException in the executors
-    var error = intercept[SparkException] {
-      rdd.take(2)(0)(1)
-    }
-    assert(error.getCause.isInstanceOf[NullPointerException])
-
-    // Pass in false, so we will get the expected rows
-    extraOptions = PhoenixTestingInputPartitionReader.RETURN_NULL_CURR_ROW + "=false"
-    rdd = spark.sqlContext.read
-      .format(PhoenixTestingDataSource.TEST_SOURCE)
-      .options( Map("table" -> "TABLE1", PhoenixDataSource.ZOOKEEPER_URL -> quorumAddress,
-        PhoenixDataSource.PHOENIX_CONFIGS -> extraOptions)).load
-    val stringValue = rdd.take(2)(0)(1)
-    stringValue shouldEqual "test_row_1"
-  }
-
   test("Can save to phoenix table from Spark without specifying all the columns") {
     val dataSet = List(Row(1L, "1", 1), Row(2L, "2", 2), Row(3L, "3", 3))
 
@@ -300,47 +277,6 @@ class PhoenixSparkDatasourceV1IT extends AbstractPhoenixSparkIT {
     }
   }
 
-  test("Can use extraOptions to set configs for workers during writes") {
-    val totalRecords = 100
-    val upsertBatchSize = 5
-
-    val records = new mutable.MutableList[Row]
-    for (x <- 1 to totalRecords) {
-      records += Row(x.toLong, x.toString, x)
-    }
-    val dataSet = records.toList
-
-    val schema = StructType(
-      Seq(StructField("ID", LongType, nullable = false),
-        StructField("COL1", StringType),
-        StructField("COL2", IntegerType)))
-
-    // Distribute the dataset into an RDD with just 1 partition so we use only 1 executor.
-    // This makes it easy to deterministically count the batched commits from that executor
-    // since it corresponds to exactly 1 input partition. In case of multiple executors with
-    // an uneven distribution of input partitions, if
-    // (number of records in that partition) % batchSize != 0, some updates would also be committed
-    // via PhoenixDataWriter#commit rather than the batch commits via PhoenixDataWriter#write
-    // and those would thus, not be counted by PhoenixTestingDataWriter.
-    val rowRDD = spark.sparkContext.parallelize(dataSet, 1)
-
-    // Apply the schema to the RDD.
-    val df = spark.sqlContext.createDataFrame(rowRDD, schema)
-    val extraOptions =  PhoenixConfigurationUtil.UPSERT_BATCH_SIZE + "=" + upsertBatchSize.toString
-
-    // Initially, this should be zero
-    PhoenixTestingDataSourceWriter.TOTAL_BATCHES_COMMITTED_COUNT shouldEqual 0
-    df.write
-      .format(PhoenixTestingDataSource.TEST_SOURCE)
-      .options(Map("table" -> "OUTPUT_TEST_TABLE", PhoenixDataSource.ZOOKEEPER_URL -> quorumAddress,
-        PhoenixDataSource.PHOENIX_CONFIGS -> extraOptions))
-      .mode(SaveMode.Overwrite)
-      .save()
-
-    // Verify the number of times batched updates are committed via DataWriters
-    PhoenixTestingDataSourceWriter.TOTAL_BATCHES_COMMITTED_COUNT shouldEqual totalRecords/upsertBatchSize
-  }
-
   test("Can save dates to Phoenix using java.sql.Date") {
     val date = java.sql.Date.valueOf("2016-09-30")
 
@@ -703,14 +639,14 @@ class PhoenixSparkDatasourceV1IT extends AbstractPhoenixSparkIT {
       .options(Map("table" -> "DATE_TEST", PhoenixDataSource.ZOOKEEPER_URL -> quorumAddress))
       .load
     val dt = df.select("COL1").first().getDate(0).getTime
-    val epoch = new Date().getTime
+    val expected = LocalDate.of(2021,1,1).atStartOfDay(ZoneId.systemDefault()).toInstant.toEpochMilli
 
     // NOTE: Spark DateType drops hour, minute, second, as per the java.sql.Date spec
     // Use 'dateAsTimestamp' option to coerce DATE to TIMESTAMP without losing resolution
 
     // Note that Spark also applies the timezone offset to the returned date epoch. Rather than perform timezone
     // gymnastics, just make sure we're within 24H of the epoch generated just now
-    assert(Math.abs(epoch - dt) < 86400000)
+    assert(expected == dt)
 
     df.createOrReplaceTempView("DATE_TEST")
     val df2 = spark.sql("SELECT * FROM DATE_TEST WHERE COL1 > TO_DATE('1990-01-01 00:00:01', 'yyyy-MM-dd HH:mm:ss')")
@@ -773,9 +709,9 @@ class PhoenixSparkDatasourceV1IT extends AbstractPhoenixSparkIT {
       .load
     val dtRes = df.select("COL1").first()
     val ts = dtRes.getTimestamp(0).getTime
-    val epoch = new Date().getTime
+    val expected = LocalDate.of(2021, 1, 1).atStartOfDay(ZoneId.systemDefault()).toInstant.toEpochMilli
 
-    assert(Math.abs(epoch - ts) < 300000)
+    assert(expected == ts)
   }
 
   test("Can load Phoenix Time columns through DataFrame API") {
@@ -784,8 +720,9 @@ class PhoenixSparkDatasourceV1IT extends AbstractPhoenixSparkIT {
       .options(Map("table" -> "TIME_TEST", PhoenixDataSource.ZOOKEEPER_URL -> quorumAddress))
       .load
     val time = df.select("COL1").first().getTimestamp(0).getTime
-    val epoch = new Date().getTime
-    assert(Math.abs(epoch - time) < 86400000)
+    val expected = LocalDate.of(2021, 1, 1).atStartOfDay(ZoneId.systemDefault()).toInstant.toEpochMilli
+
+    assert(expected == time)
   }
 
   test("can read all Phoenix data types") {