<commit_before>package com.ottogroup.bi.soda.crate
import com.ottogroup.bi.soda.crate.ddl.HiveQl
import com.ottogroup.bi.soda.dsl.View
import java.sql.DriverManager
import java.sql.Connection
import java.security.MessageDigest
import org.apache.hadoop.hive.conf.HiveConf
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import scala.collection.mutable.HashSet
import scala.collection.JavaConversions._
import org.apache.hadoop.security.UserGroupInformation
import org.apache.hadoop.fs.FileSystem
import java.security.PrivilegedAction
import scala.Array.canBuildFrom
import com.ottogroup.bi.soda.crate.ddl.HiveQl
import org.apache.hadoop.hive.metastore.api.Partition
import org.apache.hadoop.hive.metastore.IMetaStoreClient
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException
import org.apache.hadoop.hive.metastore.api.Database
import org.apache.hadoop.hive.metastore.api.AlreadyExistsException
import java.io.InvalidObjectException
import com.ottogroup.bi.soda.bottler.api.Settings
import org.apache.hadoop.hive.metastore.api.Function
import org.joda.time.DateTime
import collection.JavaConversions._
import org.apache.hadoop.hive.metastore.api.MetaException
import scala.collection.mutable.HashMap

class DeploySchema(val metastoreClient: IMetaStoreClient, val connection: Connection) {
  val md5 = MessageDigest.getInstance("MD5")
  val existingSchemas = collection.mutable.Set[String]()
  val tablePropSchemaHash = "hash"
  val partitionPropDigestName = "versionDigest"

  // FIXME: add transformation version handling (hash for transformation version + global "digest" for view version
  def digest(string: String): String = md5.digest(string.toCharArray().map(_.toByte)).map("%02X" format _).mkString

  def setTableProperty(dbName: String, tableName: String, key: String, value: String): Unit = {
    val table = metastoreClient.getTable(dbName, tableName)
    table.putToParameters(key, value)
    metastoreClient.alter_table(dbName, tableName, table)
  }

  def setPartitionProperty(dbName: String, tableName: String, part: String, key: String, value: String): Unit = {
    val partition = metastoreClient.getPartition(dbName, tableName, part)
    partition.putToParameters(key, value)
    metastoreClient.alter_partition(dbName, tableName, partition)
  }

  def setPartitionVersion(view: View) = {
    setPartitionProperty(view.dbName, view.n, view.partitionSpec, partitionPropDigestName, view.transformation().versionDigest)
  }

  def getPartitionVersion(view: View): String = {
    try {
      val props = metastoreClient.getPartition(view.dbName, view.n, view.partitionSpec).getParameters()
      if (props.containsKey(partitionPropDigestName))
        props.get(partitionPropDigestName)
      else
        "does not exist"
    } catch {
      case e: Exception => throw e
    }
  }

  def dropAndCreateTableSchema(dbName: String, tableName: String, sql: String): Unit = {
    println("in dropAndCreateSchema " + dbName + "." + tableName + " " + sql)
    val stmt = connection.createStatement()
    if (!metastoreClient.getAllDatabases.contains(dbName)) {
      stmt.execute(s"CREATE DATABASE ${dbName}")
    }
    if (metastoreClient.tableExists(dbName, tableName)) {
      metastoreClient.dropTable(dbName, tableName, false, true)
    }

    stmt.execute(sql)

    setTableProperty(dbName, tableName, tablePropSchemaHash, digest(sql))
    println("!!created table " + sql)
  }

  def schemaExists(dbname: String, tableName: String, sql: String): Boolean = {
    val d = digest(sql)
    if (existingSchemas.contains(d))
      return true;
    if (!metastoreClient.tableExists(dbname, tableName)) {
      false
    } else {
      val table = metastoreClient.getTable(dbname, tableName)
      val props = table.getParameters()
      if (!props.containsKey(tablePropSchemaHash))
        false
      else if (d == props.get(tablePropSchemaHash).toString()) {
        existingSchemas += d
        true
      } else
        false
    }
  }

  def partitionExists(dbname: String, tableName: String, sql: String, partition: String): Boolean = {
    if (!schemaExists(dbname, tableName, sql)) return false
    else
      try {
        metastoreClient.getPartition(dbname, tableName, partition)
      } catch {
        case e: NoSuchObjectException => return false
      }
    true
  }

  def createPartition(view: View): Partition = {
    if (!schemaExists(view.dbName, view.n, HiveQl.ddl(view))) {
      dropAndCreateTableSchema(view.dbName, view.n, HiveQl.ddl(view))
    }
    try {
      val now = new DateTime().getMillis.toInt
      val sd = metastoreClient.getTable(view.dbName, view.n).getSd
      sd.setLocation(view.fullPath)
      val part = new Partition(view.partitionValues, view.dbName, view.n, now, now, sd, HashMap[String, String]())
      //val part = metastoreClient.appendPartition(view.dbName, view.n, view.partitionSpec)
      metastoreClient.add_partitions(List(part), true, false)
      metastoreClient.getPartition(view.dbName, view.n, view.partitionSpec)
    } catch {
      case e: AlreadyExistsException => metastoreClient.getPartition(view.dbName, view.n, view.partitionSpec)
      case e: InvalidObjectException =>
        println(view.partitionSpec); throw (e)
      case e: MetaException => println(view.partitionSpec); throw (e)
    }<commit_msg>remove println<commit_after>package com.ottogroup.bi.soda.crate
import com.ottogroup.bi.soda.crate.ddl.HiveQl
import com.ottogroup.bi.soda.dsl.View
import java.sql.DriverManager
import java.sql.Connection
import java.security.MessageDigest
import org.apache.hadoop.hive.conf.HiveConf
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import scala.collection.mutable.HashSet
import scala.collection.JavaConversions._
import org.apache.hadoop.security.UserGroupInformation
import org.apache.hadoop.fs.FileSystem
import java.security.PrivilegedAction
import scala.Array.canBuildFrom
import com.ottogroup.bi.soda.crate.ddl.HiveQl
import org.apache.hadoop.hive.metastore.api.Partition
import org.apache.hadoop.hive.metastore.IMetaStoreClient
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException
import org.apache.hadoop.hive.metastore.api.Database
import org.apache.hadoop.hive.metastore.api.AlreadyExistsException
import java.io.InvalidObjectException
import com.ottogroup.bi.soda.bottler.api.Settings
import org.apache.hadoop.hive.metastore.api.Function
import org.joda.time.DateTime
import collection.JavaConversions._
import org.apache.hadoop.hive.metastore.api.MetaException
import scala.collection.mutable.HashMap

class DeploySchema(val metastoreClient: IMetaStoreClient, val connection: Connection) {
  val md5 = MessageDigest.getInstance("MD5")
  val existingSchemas = collection.mutable.Set[String]()
  val tablePropSchemaHash = "hash"
  val partitionPropDigestName = "versionDigest"

  // FIXME: add transformation version handling (hash for transformation version + global "digest" for view version
  def digest(string: String): String = md5.digest(string.toCharArray().map(_.toByte)).map("%02X" format _).mkString

  def setTableProperty(dbName: String, tableName: String, key: String, value: String): Unit = {
    val table = metastoreClient.getTable(dbName, tableName)
    table.putToParameters(key, value)
    metastoreClient.alter_table(dbName, tableName, table)
  }

  def setPartitionProperty(dbName: String, tableName: String, part: String, key: String, value: String): Unit = {
    val partition = metastoreClient.getPartition(dbName, tableName, part)
    partition.putToParameters(key, value)
    metastoreClient.alter_partition(dbName, tableName, partition)
  }

  def setPartitionVersion(view: View) = {
    setPartitionProperty(view.dbName, view.n, view.partitionSpec, partitionPropDigestName, view.transformation().versionDigest)
  }

  def getPartitionVersion(view: View): String = {
    try {
      val props = metastoreClient.getPartition(view.dbName, view.n, view.partitionSpec).getParameters()
      if (props.containsKey(partitionPropDigestName))
        props.get(partitionPropDigestName)
      else
        "does not exist"
    } catch {
      case e: Exception => throw e
    }
  }

  def dropAndCreateTableSchema(dbName: String, tableName: String, sql: String): Unit = {
    println("in dropAndCreateSchema " + dbName + "." + tableName + " " + sql)
    val stmt = connection.createStatement()
    if (!metastoreClient.getAllDatabases.contains(dbName)) {
      stmt.execute(s"CREATE DATABASE ${dbName}")
    }
    if (metastoreClient.tableExists(dbName, tableName)) {
      metastoreClient.dropTable(dbName, tableName, false, true)
    }

    stmt.execute(sql)

    setTableProperty(dbName, tableName, tablePropSchemaHash, digest(sql))
    println("!!created table " + sql)
  }

  def schemaExists(dbname: String, tableName: String, sql: String): Boolean = {
    val d = digest(sql)
    if (existingSchemas.contains(d))
      return true;
    if (!metastoreClient.tableExists(dbname, tableName)) {
      false
    } else {
      val table = metastoreClient.getTable(dbname, tableName)
      val props = table.getParameters()
      if (!props.containsKey(tablePropSchemaHash))
        false
      else if (d == props.get(tablePropSchemaHash).toString()) {
        existingSchemas += d
        true
      } else
        false
    }
  }

  def partitionExists(dbname: String, tableName: String, sql: String, partition: String): Boolean = {
    if (!schemaExists(dbname, tableName, sql)) return false
    else
      try {
        metastoreClient.getPartition(dbname, tableName, partition)
      } catch {
        case e: NoSuchObjectException => return false
      }
    true
  }

  def createPartition(view: View): Partition = {
    if (!schemaExists(view.dbName, view.n, HiveQl.ddl(view))) {
      dropAndCreateTableSchema(view.dbName, view.n, HiveQl.ddl(view))
    }
    try {
      val now = new DateTime().getMillis.toInt
      val sd = metastoreClient.getTable(view.dbName, view.n).getSd
 
      sd.setLocation(view.fullPath)
      val part = new Partition(view.partitionValues, view.dbName, view.n, now, now, sd, HashMap[String, String]())
      //val part = metastoreClient.appendPartition(view.dbName, view.n, view.partitionSpec)
      metastoreClient.add_partitions(List(part), true, false)
      metastoreClient.getPartition(view.dbName, view.n, view.partitionSpec)
    } catch {
      case e: AlreadyExistsException => metastoreClient.getPartition(view.dbName, view.n, view.partitionSpec)
      case e: InvalidObjectException =>
        println(view.partitionSpec); throw (e)
      case e: MetaException => println(view.partitionSpec); throw (e)
    }
{'fraction_non_alphanumeric': 0.10765661252900233, 'fraction_numerical': 0.0009280742459396752, 'mean_word_length': 4.205797101449275, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6959572', 'n_tokens_mistral': 3337, 'n_tokens_neox': 3291, 'n_words': 671}