11028

Build # 11028

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/cfdbe0dcba3fffa7c7d3b3f68dfedfc12f450d07">cfdbe0dcb<a href="https://github.com/apache/carbondata/commit/cfdbe0dcba3fffa7c7d3b3f68dfedfc12f450d07">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/e012e92d6dee22cce5a020d4a55583d2abb4c601">e012e92d6</a>">e012e92d6</a><a href="https://github.com/apache/carbondata/commit/cfdbe0dcba3fffa7c7d3b3f68dfedfc12f450d07"> into 66982f342">66982f342</a>

Run Details

88 of 88 new or added lines in 10 files covered. (100.0%)

62873 of 79313 relevant lines covered (79.27%)

1.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

63.01

/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/QueryTest.scala

/*
 * 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.test.util

import java.util.{Locale, TimeZone}

import scala.collection.JavaConverters._

import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.util._
import org.apache.spark.sql.test.TestQueryExecutor

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.util.CarbonProperties



class QueryTest extends PlanTest {

  val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)

  // Timezone is fixed to America/Los_Angeles for those timezone sensitive tests (timestamp_*)
  TimeZone.setDefault(TimeZone.getTimeZone("America/Los_Angeles"))
  // Add Locale setting
  Locale.setDefault(Locale.US)

  CarbonProperties.getInstance()
    .addProperty(CarbonCommonConstants.SUPPORT_DIRECT_QUERY_ON_DATAMAP, "true")

  /**
   * Runs the plan and makes sure the answer contains all of the keywords, or the
   * none of keywords are listed in the answer
   * @param df the [[DataFrame]] to be executed
   * @param exists true for make sure the keywords are listed in the output, otherwise
   *               to make sure none of the keyword are not listed in the output
   * @param keywords keyword in string array
   */
  def checkExistence(df: DataFrame, exists: Boolean, keywords: String*) {
    val outputs = df.collect().map(_.mkString(" ")).mkString(" ")
    for (key <- keywords) {
      if (exists) {
        assert(outputs.contains(key), s"Failed for $df ($key doesn't exist in result)")
      } else {
        assert(!outputs.contains(key), s"Failed for $df ($key existed in the result)")
      }
    }
  }

  /**
   * Runs the plan and counts the keyword in the answer
   * @param df the [[DataFrame]] to be executed
   * @param count expected count
   * @param keyword keyword to search
   */
  def checkExistenceCount(df: DataFrame, count: Long, keyword: String): Unit = {
    val outputs = df.collect().map(_.mkString).mkString
    assert(outputs.sliding(keyword.length).count(_ == keyword) === count)
  }

  def sqlTest(sqlString: String, expectedAnswer: Seq[Row])(implicit sqlContext: SQLContext) {
    test(sqlString) {
      checkAnswer(sqlContext.sql(sqlString), expectedAnswer)
    }
  }

  /**
   * Runs the plan and makes sure the answer matches the expected result.
   * @param df the [[DataFrame]] to be executed
   * @param expectedAnswer the expected result in a [[Seq]] of [[Row]]s.
   */
  protected def checkAnswer(df: DataFrame, expectedAnswer: Seq[Row]): Unit = {
    QueryTest.checkAnswer(df, expectedAnswer) match {
      case Some(errorMessage) => fail(errorMessage)
      case None =>
    }
  }

  protected def checkAnswer(df: DataFrame, expectedAnswer: Row): Unit = {
    checkAnswer(df, Seq(expectedAnswer))
  }

  protected def checkAnswer(df: DataFrame, expectedAnswer: DataFrame): Unit = {
    checkAnswer(df, expectedAnswer.collect())
  }

  protected def dropTable(tableName: String): Unit = {
    sql(s"DROP TABLE IF EXISTS $tableName")
  }

  protected def dropDataMaps(tableName: String, dataMapNames: String*): Unit = {
    for (dataMapName <- dataMapNames) {
      sql(s"DROP DATAMAP IF EXISTS $dataMapName ON TABLE $tableName")
    }
  }

  def sql(sqlText: String): DataFrame = TestQueryExecutor.INSTANCE.sql(sqlText)

  val sqlContext: SQLContext = TestQueryExecutor.INSTANCE.sqlContext

  lazy val warehouse = TestQueryExecutor.warehouse
  lazy val storeLocation = CarbonProperties.getInstance().
    getProperty(CarbonCommonConstants.STORE_LOCATION)
  val resourcesPath = TestQueryExecutor.resourcesPath
  val metaStoreDB = TestQueryExecutor.metaStoreDB
  val integrationPath = TestQueryExecutor.integrationPath
  val dblocation = TestQueryExecutor.location
  val defaultParallelism = sqlContext.sparkContext.defaultParallelism
}

object QueryTest {

  def checkAnswer(df: DataFrame, expectedAnswer: java.util.List[Row]): String = {
    checkAnswer(df, expectedAnswer.asScala) match {
      case Some(errorMessage) => errorMessage
      case None => null
    }
  }

  /**
   * Runs the plan and makes sure the answer matches the expected result.
   * If there was exception during the execution or the contents of the DataFrame does not
   * match the expected result, an error message will be returned. Otherwise, a [[None]] will
   * be returned.
   * @param df the [[DataFrame]] to be executed
   * @param expectedAnswer the expected result in a [[Seq]] of [[Row]]s.
   */
  def checkAnswer(df: DataFrame, expectedAnswer: Seq[Row]): Option[String] = {
    val isSorted = df.logicalPlan.collect { case s: logical.Sort => s }.nonEmpty
    def prepareAnswer(answer: Seq[Row]): Seq[Row] = {
      // Converts data to types that we can do equality comparison using Scala collections.
      // For BigDecimal type, the Scala type has a better definition of equality test (similar to
      // Java's java.math.BigDecimal.compareTo).
      // For binary arrays, we convert it to Seq to avoid of calling java.util.Arrays.equals for
      // equality test.
      val converted: Seq[Row] = answer.map { s =>
        Row.fromSeq(s.toSeq.map {
          case d: java.math.BigDecimal => BigDecimal(d)
          case b: Array[Byte] => b.toSeq
          case d : Double =>
            if (!d.isInfinite && !d.isNaN) {
              var bd = BigDecimal(d)
              bd = bd.setScale(5, BigDecimal.RoundingMode.UP)
              bd.doubleValue()
            }
            else {
              d
            }
          case o => o
        })
      }
      if (!isSorted) converted.sortBy(_.toString()) else converted
    }
    val sparkAnswer = try df.collect().toSeq catch {
      case e: Exception =>
        val errorMessage =
          s"""
             |Exception thrown while executing query:
             |${df.queryExecution}
             |== Exception ==
             |$e
             |${org.apache.spark.sql.catalyst.util.stackTraceToString(e)}
          """.stripMargin
        return Some(errorMessage)
    }

    if (prepareAnswer(expectedAnswer) != prepareAnswer(sparkAnswer)) {
      val errorMessage =
        s"""
           |Results do not match for query:
           |${df.queryExecution}
           |== Results ==
           |${
          sideBySide(
            s"== Correct Answer - ${expectedAnswer.size} ==" +:
              prepareAnswer(expectedAnswer).map(_.toString()),
            s"== Spark Answer - ${sparkAnswer.size} ==" +:
              prepareAnswer(sparkAnswer).map(_.toString())).mkString("\n")
        }
      """.stripMargin
      return Some(errorMessage)
    }

    return None
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.spark.sql.test.util
19
20	import java.util.{Locale, TimeZone}
21
22	import scala.collection.JavaConverters._
23
24	import org.apache.spark.sql.{DataFrame, Row, SQLContext}
25	import org.apache.spark.sql.catalyst.plans._
26	import org.apache.spark.sql.catalyst.util._
27	import org.apache.spark.sql.test.TestQueryExecutor
28
29	import org.apache.carbondata.common.logging.LogServiceFactory
30	import org.apache.carbondata.core.constants.CarbonCommonConstants
31	import org.apache.carbondata.core.util.CarbonProperties
32
33
34
35	class QueryTest extends PlanTest {	1✔
36
37	val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	1✔
38
39	// Timezone is fixed to America/Los_Angeles for those timezone sensitive tests (timestamp_*)
40	TimeZone.setDefault(TimeZone.getTimeZone("America/Los_Angeles"))	1✔
41	// Add Locale setting
42	Locale.setDefault(Locale.US)	1✔
43
44	CarbonProperties.getInstance()	1✔
45	.addProperty(CarbonCommonConstants.SUPPORT_DIRECT_QUERY_ON_DATAMAP, "true")	1✔
46
47	/**
48	* Runs the plan and makes sure the answer contains all of the keywords, or the
49	* none of keywords are listed in the answer
50	* @param df the [[DataFrame]] to be executed
51	* @param exists true for make sure the keywords are listed in the output, otherwise
52	* to make sure none of the keyword are not listed in the output
53	* @param keywords keyword in string array
54	*/
55	def checkExistence(df: DataFrame, exists: Boolean, keywords: String*) {
56	val outputs = df.collect().map(_.mkString(" ")).mkString(" ")	1✔
57	for (key <- keywords) {	1✔
58	if (exists) {	1✔
59	assert(outputs.contains(key), s"Failed for $df ($key doesn't exist in result)")	1✔
60	} else {
61	assert(!outputs.contains(key), s"Failed for $df ($key existed in the result)")	1✔
62	}
63	}
64	}
65
66	/**
67	* Runs the plan and counts the keyword in the answer
68	* @param df the [[DataFrame]] to be executed
69	* @param count expected count
70	* @param keyword keyword to search
71	*/
72	def checkExistenceCount(df: DataFrame, count: Long, keyword: String): Unit = {
73	val outputs = df.collect().map(_.mkString).mkString	×
74	assert(outputs.sliding(keyword.length).count(_ == keyword) === count)	×
75	}
76
77	def sqlTest(sqlString: String, expectedAnswer: Seq[Row])(implicit sqlContext: SQLContext) {
78	test(sqlString) {	×
79	checkAnswer(sqlContext.sql(sqlString), expectedAnswer)	×
80	}
81	}
82
83	/**
84	* Runs the plan and makes sure the answer matches the expected result.
85	* @param df the [[DataFrame]] to be executed
86	* @param expectedAnswer the expected result in a [[Seq]] of [[Row]]s.
87	*/
88	protected def checkAnswer(df: DataFrame, expectedAnswer: Seq[Row]): Unit = {
89	QueryTest.checkAnswer(df, expectedAnswer) match {	1✔
90	case Some(errorMessage) => fail(errorMessage)	1✔
91	case None =>	1✔
92	}
93	}
94
95	protected def checkAnswer(df: DataFrame, expectedAnswer: Row): Unit = {
96	checkAnswer(df, Seq(expectedAnswer))	1✔
97	}
98
99	protected def checkAnswer(df: DataFrame, expectedAnswer: DataFrame): Unit = {
100	checkAnswer(df, expectedAnswer.collect())	1✔
101	}
102
103	protected def dropTable(tableName: String): Unit = {
104	sql(s"DROP TABLE IF EXISTS $tableName")	1✔
105	}
106
107	protected def dropDataMaps(tableName: String, dataMapNames: String*): Unit = {
108	for (dataMapName <- dataMapNames) {	1✔
109	sql(s"DROP DATAMAP IF EXISTS $dataMapName ON TABLE $tableName")	1✔
110	}
111	}
112
113	def sql(sqlText: String): DataFrame = TestQueryExecutor.INSTANCE.sql(sqlText)	1✔
114
115	val sqlContext: SQLContext = TestQueryExecutor.INSTANCE.sqlContext	1✔
116
117	lazy val warehouse = TestQueryExecutor.warehouse	1✔
118	lazy val storeLocation = CarbonProperties.getInstance().	1✔
119	getProperty(CarbonCommonConstants.STORE_LOCATION)	1✔
120	val resourcesPath = TestQueryExecutor.resourcesPath	1✔
121	val metaStoreDB = TestQueryExecutor.metaStoreDB	1✔
122	val integrationPath = TestQueryExecutor.integrationPath	1✔
123	val dblocation = TestQueryExecutor.location	1✔
124	val defaultParallelism = sqlContext.sparkContext.defaultParallelism	1✔
125	}
126
127	object QueryTest {	×
128
129	def checkAnswer(df: DataFrame, expectedAnswer: java.util.List[Row]): String = {
130	checkAnswer(df, expectedAnswer.asScala) match {	×
131	case Some(errorMessage) => errorMessage	×
132	case None => null	×
133	}
134	}
135
136	/**
137	* Runs the plan and makes sure the answer matches the expected result.
138	* If there was exception during the execution or the contents of the DataFrame does not
139	* match the expected result, an error message will be returned. Otherwise, a [[None]] will
140	* be returned.
141	* @param df the [[DataFrame]] to be executed
142	* @param expectedAnswer the expected result in a [[Seq]] of [[Row]]s.
143	*/
144	def checkAnswer(df: DataFrame, expectedAnswer: Seq[Row]): Option[String] = {
145	val isSorted = df.logicalPlan.collect { case s: logical.Sort => s }.nonEmpty	1✔
146	def prepareAnswer(answer: Seq[Row]): Seq[Row] = {	1✔
147	// Converts data to types that we can do equality comparison using Scala collections.
148	// For BigDecimal type, the Scala type has a better definition of equality test (similar to
149	// Java's java.math.BigDecimal.compareTo).
150	// For binary arrays, we convert it to Seq to avoid of calling java.util.Arrays.equals for
151	// equality test.
152	val converted: Seq[Row] = answer.map { s =>	1✔
153	Row.fromSeq(s.toSeq.map {	1✔
154	case d: java.math.BigDecimal => BigDecimal(d)	1✔
155	case b: Array[Byte] => b.toSeq	1✔
156	case d : Double =>	1✔
157	if (!d.isInfinite && !d.isNaN) {	1✔
158	var bd = BigDecimal(d)	1✔
159	bd = bd.setScale(5, BigDecimal.RoundingMode.UP)	1✔
160	bd.doubleValue()	1✔
161	}
162	else {
163	d	×
164	}
165	case o => o	1✔
166	})
167	}
168	if (!isSorted) converted.sortBy(_.toString()) else converted	1✔
169	}
170	val sparkAnswer = try df.collect().toSeq catch {	1✔
171	case e: Exception =>
172	val errorMessage =	×
173	s"""	×
174	\|Exception thrown while executing query:
175	\|${df.queryExecution}	×
176	\|== Exception ==
177	\|$e	×
178	\|${org.apache.spark.sql.catalyst.util.stackTraceToString(e)}	×
179	""".stripMargin	×
180	return Some(errorMessage)	×
181	}
182
183	if (prepareAnswer(expectedAnswer) != prepareAnswer(sparkAnswer)) {	1✔
184	val errorMessage =	×
185	s"""	×
186	\|Results do not match for query:
187	\|${df.queryExecution}	×
188	\|== Results ==
189	\|${
190	sideBySide(	×
191	s"== Correct Answer - ${expectedAnswer.size} ==" +:	×
192	prepareAnswer(expectedAnswer).map(_.toString()),	×
193	s"== Spark Answer - ${sparkAnswer.size} ==" +:	×
194	prepareAnswer(sparkAnswer).map(_.toString())).mkString("\n")	×
195	}	×
196	""".stripMargin	×
197	return Some(errorMessage)	×
198	}
199
200	return None	1✔
201	}
202	}	1✔

apache / carbondata / 11028

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous