path: root/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/VersionsSuite.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.hive.client

import java.io.File

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.util.VersionInfo

import org.apache.spark.{SparkConf, SparkFunSuite}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, EqualTo, Literal, NamedExpression}
import org.apache.spark.sql.catalyst.util.quietly
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.tags.ExtendedHiveTest
import org.apache.spark.util.Utils

/**
 * A simple set of tests that call the methods of a [[HiveClient]], loading different version
 * of hive from maven central.  These tests are simple in that they are mostly just testing to make
 * sure that reflective calls are not throwing NoSuchMethod error, but the actually functionality
 * is not fully tested.
 */
@ExtendedHiveTest
class VersionsSuite extends SparkFunSuite with Logging {

  private val sparkConf = new SparkConf()

  // In order to speed up test execution during development or in Jenkins, you can specify the path
  // of an existing Ivy cache:
  private val ivyPath: Option[String] = {
    sys.env.get("SPARK_VERSIONS_SUITE_IVY_PATH").orElse(
      Some(new File(sys.props("java.io.tmpdir"), "hive-ivy-cache").getAbsolutePath))
  }

  private def buildConf() = {
    lazy val warehousePath = Utils.createTempDir()
    lazy val metastorePath = Utils.createTempDir()
    metastorePath.delete()
    Map(
      "javax.jdo.option.ConnectionURL" -> s"jdbc:derby:;databaseName=$metastorePath;create=true",
      "hive.metastore.warehouse.dir" -> warehousePath.toString)
  }

  test("success sanity check") {
    val badClient = IsolatedClientLoader.forVersion(
      hiveMetastoreVersion = HiveContext.hiveExecutionVersion,
      hadoopVersion = VersionInfo.getVersion,
      sparkConf = sparkConf,
      hadoopConf = new Configuration(),
      config = buildConf(),
      ivyPath = ivyPath).createClient()
    val db = new CatalogDatabase("default", "desc", "loc", Map())
    badClient.createDatabase(db, ignoreIfExists = true)
  }

  test("hadoop configuration preserved") {
    val hadoopConf = new Configuration();
    hadoopConf.set("test", "success")
    val client = IsolatedClientLoader.forVersion(
      hiveMetastoreVersion = HiveContext.hiveExecutionVersion,
      hadoopVersion = VersionInfo.getVersion,
      sparkConf = sparkConf,
      hadoopConf = hadoopConf,
      config = buildConf(),
      ivyPath = ivyPath).createClient()
    assert("success" === client.getConf("test", null))
  }

  private def getNestedMessages(e: Throwable): String = {
    var causes = ""
    var lastException = e
    while (lastException != null) {
      causes += lastException.toString + "\n"
      lastException = lastException.getCause
    }
    causes
  }

  private val emptyDir = Utils.createTempDir().getCanonicalPath

  private def partSpec = {
    val hashMap = new java.util.LinkedHashMap[String, String]
    hashMap.put("key", "1")
    hashMap
  }

  // Its actually pretty easy to mess things up and have all of your tests "pass" by accidentally
  // connecting to an auto-populated, in-process metastore.  Let's make sure we are getting the
  // versions right by forcing a known compatibility failure.
  // TODO: currently only works on mysql where we manually create the schema...
  ignore("failure sanity check") {
    val e = intercept[Throwable] {
      val badClient = quietly {
        IsolatedClientLoader.forVersion(
          hiveMetastoreVersion = "13",
          hadoopVersion = VersionInfo.getVersion,
          sparkConf = sparkConf,
          hadoopConf = new Configuration(),
          config = buildConf(),
          ivyPath = ivyPath).createClient()
      }
    }
    assert(getNestedMessages(e) contains "Unknown column 'A0.OWNER_NAME' in 'field list'")
  }

  private val versions = Seq("12", "13", "14", "1.0.0", "1.1.0", "1.2.0")

  private var client: HiveClient = null

  versions.foreach { version =>
    test(s"$version: create client") {
      client = null
      System.gc() // Hack to avoid SEGV on some JVM versions.
      client =
        IsolatedClientLoader.forVersion(
          hiveMetastoreVersion = version,
          hadoopVersion = VersionInfo.getVersion,
          sparkConf = sparkConf,
          hadoopConf = new Configuration(),
          config = buildConf(),
          ivyPath = ivyPath).createClient()
    }

    test(s"$version: createDatabase") {
      val db = CatalogDatabase("default", "desc", "loc", Map())
      client.createDatabase(db, ignoreIfExists = true)
    }

    test(s"$version: createTable") {
      val table =
        CatalogTable(
          identifier = TableIdentifier("src", Some("default")),
          tableType = CatalogTableType.MANAGED_TABLE,
          schema = Seq(CatalogColumn("key", "int")),
          storage = CatalogStorageFormat(
            locationUri = None,
            inputFormat = Some(classOf[org.apache.hadoop.mapred.TextInputFormat].getName),
            outputFormat = Some(
              classOf[org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat[_, _]].getName),
            serde = Some(classOf[org.apache.hadoop.hive.serde2.`lazy`.LazySimpleSerDe].getName()),
            serdeProperties = Map.empty
          ))

      client.createTable(table, ignoreIfExists = false)
    }

    test(s"$version: getTable") {
      client.getTable("default", "src")
    }

    test(s"$version: listTables") {
      assert(client.listTables("default") === Seq("src"))
    }

    test(s"$version: getDatabase") {
      client.getDatabase("default")
    }

    test(s"$version: alterTable") {
      client.alterTable(client.getTable("default", "src"))
    }

    test(s"$version: set command") {
      client.runSqlHive("SET spark.sql.test.key=1")
    }

    test(s"$version: create partitioned table DDL") {
      client.runSqlHive("CREATE TABLE src_part (value INT) PARTITIONED BY (key INT)")
      client.runSqlHive("ALTER TABLE src_part ADD PARTITION (key = '1')")
    }

    test(s"$version: getPartitions") {
      client.getAllPartitions(client.getTable("default", "src_part"))
    }

    test(s"$version: getPartitionsByFilter") {
      client.getPartitionsByFilter(client.getTable("default", "src_part"), Seq(EqualTo(
        AttributeReference("key", IntegerType, false)(NamedExpression.newExprId),
        Literal(1))))
    }

    test(s"$version: loadPartition") {
      client.loadPartition(
        emptyDir,
        "default.src_part",
        partSpec,
        false,
        false,
        false,
        false)
    }

    test(s"$version: loadTable") {
      client.loadTable(
        emptyDir,
        "src",
        false,
        false)
    }

    test(s"$version: loadDynamicPartitions") {
      client.loadDynamicPartitions(
        emptyDir,
        "default.src_part",
        partSpec,
        false,
        1,
        false,
        false)
    }

    test(s"$version: create index and reset") {
      client.runSqlHive("CREATE TABLE indexed_table (key INT)")
      client.runSqlHive("CREATE INDEX index_1 ON TABLE indexed_table(key) " +
        "as 'COMPACT' WITH DEFERRED REBUILD")
      client.reset()
    }
  }
}