MaxGekk commented on a change in pull request #11293:
URL: https://github.com/apache/spark/pull/11293#discussion_r541560603
##########
File path:
sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
##########
@@ -234,167 +235,184 @@ private[hive] class HiveClientImpl(
if (getDatabaseOption(databaseName).isDefined) {
state.setCurrentDatabase(databaseName)
} else {
- throw new NoSuchDatabaseException
+ throw new NoSuchDatabaseException(databaseName)
}
}
- override def createDatabase(database: HiveDatabase): Unit = withHiveState {
+ override def createDatabase(
+ database: CatalogDatabase,
+ ignoreIfExists: Boolean): Unit = withHiveState {
client.createDatabase(
- new Database(
+ new HiveDatabase(
database.name,
- "",
- new File(database.location).toURI.toString,
- new java.util.HashMap),
- true)
+ database.description,
+ database.locationUri,
+ database.properties.asJava),
+ ignoreIfExists)
}
- override def getDatabaseOption(name: String): Option[HiveDatabase] =
withHiveState {
+ override def dropDatabase(
+ name: String,
+ ignoreIfNotExists: Boolean,
+ cascade: Boolean): Unit = withHiveState {
+ client.dropDatabase(name, true, ignoreIfNotExists, cascade)
+ }
+
+ override def alterDatabase(database: CatalogDatabase): Unit = withHiveState {
+ client.alterDatabase(
+ database.name,
+ new HiveDatabase(
+ database.name,
+ database.description,
+ database.locationUri,
+ database.properties.asJava))
+ }
+
+ override def getDatabaseOption(name: String): Option[CatalogDatabase] =
withHiveState {
Option(client.getDatabase(name)).map { d =>
- HiveDatabase(
+ CatalogDatabase(
name = d.getName,
- location = d.getLocationUri)
+ description = d.getDescription,
+ locationUri = d.getLocationUri,
+ properties = d.getParameters.asScala.toMap)
}
}
+ override def listDatabases(pattern: String): Seq[String] = withHiveState {
+ client.getDatabasesByPattern(pattern).asScala.toSeq
+ }
+
override def getTableOption(
dbName: String,
- tableName: String): Option[HiveTable] = withHiveState {
-
+ tableName: String): Option[CatalogTable] = withHiveState {
logDebug(s"Looking up $dbName.$tableName")
-
- val hiveTable = Option(client.getTable(dbName, tableName, false))
- val converted = hiveTable.map { h =>
-
- HiveTable(
- name = h.getTableName,
+ Option(client.getTable(dbName, tableName, false)).map { h =>
+ CatalogTable(
specifiedDatabase = Option(h.getDbName),
- schema = h.getCols.asScala.map(f => HiveColumn(f.getName, f.getType,
f.getComment)),
- partitionColumns = h.getPartCols.asScala.map(f =>
- HiveColumn(f.getName, f.getType, f.getComment)),
- properties = h.getParameters.asScala.toMap,
- serdeProperties =
h.getTTable.getSd.getSerdeInfo.getParameters.asScala.toMap,
+ name = h.getTableName,
tableType = h.getTableType match {
- case HTableType.MANAGED_TABLE => ManagedTable
- case HTableType.EXTERNAL_TABLE => ExternalTable
- case HTableType.VIRTUAL_VIEW => VirtualView
- case HTableType.INDEX_TABLE => IndexTable
+ case HiveTableType.EXTERNAL_TABLE => CatalogTableType.EXTERNAL_TABLE
+ case HiveTableType.MANAGED_TABLE => CatalogTableType.MANAGED_TABLE
+ case HiveTableType.INDEX_TABLE => CatalogTableType.INDEX_TABLE
+ case HiveTableType.VIRTUAL_VIEW => CatalogTableType.VIRTUAL_VIEW
},
- location = shim.getDataLocation(h),
- inputFormat = Option(h.getInputFormatClass).map(_.getName),
- outputFormat = Option(h.getOutputFormatClass).map(_.getName),
- serde = Option(h.getSerializationLib),
- viewText = Option(h.getViewExpandedText)).withClient(this)
+ schema = h.getCols.asScala.map(fromHiveColumn),
+ partitionColumns = h.getPartCols.asScala.map(fromHiveColumn),
+ sortColumns = Seq(),
+ numBuckets = h.getNumBuckets,
+ createTime = h.getTTable.getCreateTime.toLong * 1000,
+ lastAccessTime = h.getLastAccessTime.toLong * 1000,
+ storage = CatalogStorageFormat(
+ locationUri = shim.getDataLocation(h),
+ inputFormat = Option(h.getInputFormatClass).map(_.getName),
+ outputFormat = Option(h.getOutputFormatClass).map(_.getName),
+ serde = Option(h.getSerializationLib),
+ serdeProperties =
h.getTTable.getSd.getSerdeInfo.getParameters.asScala.toMap
+ ),
+ properties = h.getParameters.asScala.toMap,
+ viewOriginalText = Option(h.getViewOriginalText),
+ viewText = Option(h.getViewExpandedText))
}
- converted
}
- private def toInputFormat(name: String) =
- Utils.classForName(name).asInstanceOf[Class[_ <:
org.apache.hadoop.mapred.InputFormat[_, _]]]
-
- private def toOutputFormat(name: String) =
- Utils.classForName(name)
- .asInstanceOf[Class[_ <:
org.apache.hadoop.hive.ql.io.HiveOutputFormat[_, _]]]
-
- private def toQlTable(table: HiveTable): metadata.Table = {
- val qlTable = new metadata.Table(table.database, table.name)
-
- qlTable.setFields(table.schema.map(c => new FieldSchema(c.name,
c.hiveType, c.comment)).asJava)
- qlTable.setPartCols(
- table.partitionColumns.map(c => new FieldSchema(c.name, c.hiveType,
c.comment)).asJava)
- table.properties.foreach { case (k, v) => qlTable.setProperty(k, v) }
- table.serdeProperties.foreach { case (k, v) => qlTable.setSerdeParam(k, v)
}
-
- // set owner
- qlTable.setOwner(conf.getUser)
- // set create time
- qlTable.setCreateTime((System.currentTimeMillis() /
1000).asInstanceOf[Int])
-
- table.location.foreach { loc => shim.setDataLocation(qlTable, loc) }
- table.inputFormat.map(toInputFormat).foreach(qlTable.setInputFormatClass)
-
table.outputFormat.map(toOutputFormat).foreach(qlTable.setOutputFormatClass)
- table.serde.foreach(qlTable.setSerializationLib)
-
- qlTable
+ override def createView(view: CatalogTable): Unit = withHiveState {
+ client.createTable(toHiveViewTable(view))
}
- private def toViewTable(view: HiveTable): metadata.Table = {
- // TODO: this is duplicated with `toQlTable` except the table type stuff.
- val tbl = new metadata.Table(view.database, view.name)
- tbl.setTableType(HTableType.VIRTUAL_VIEW)
- tbl.setSerializationLib(null)
- tbl.clearSerDeInfo()
-
- // TODO: we will save the same SQL string to original and expanded text,
which is different
- // from Hive.
- tbl.setViewOriginalText(view.viewText.get)
- tbl.setViewExpandedText(view.viewText.get)
-
- tbl.setFields(view.schema.map(c => new FieldSchema(c.name, c.hiveType,
c.comment)).asJava)
- view.properties.foreach { case (k, v) => tbl.setProperty(k, v) }
-
- // set owner
- tbl.setOwner(conf.getUser)
- // set create time
- tbl.setCreateTime((System.currentTimeMillis() / 1000).asInstanceOf[Int])
-
- tbl
+ override def alertView(view: CatalogTable): Unit = withHiveState {
+ client.alterTable(view.qualifiedName, toHiveViewTable(view))
}
- override def createView(view: HiveTable): Unit = withHiveState {
- client.createTable(toViewTable(view))
+ override def createTable(table: CatalogTable, ignoreIfExists: Boolean): Unit
= withHiveState {
+ client.createTable(toHiveTable(table), ignoreIfExists)
}
- override def alertView(view: HiveTable): Unit = withHiveState {
- client.alterTable(view.qualifiedName, toViewTable(view))
+ override def dropTable(
+ dbName: String,
+ tableName: String,
+ ignoreIfNotExists: Boolean): Unit = withHiveState {
+ client.dropTable(dbName, tableName, true, ignoreIfNotExists)
}
- override def createTable(table: HiveTable): Unit = withHiveState {
- val qlTable = toQlTable(table)
- client.createTable(qlTable)
+ override def alterTable(tableName: String, table: CatalogTable): Unit =
withHiveState {
+ val hiveTable = toHiveTable(table)
+ // Do not use `table.qualifiedName` here because this may be a rename
+ val qualifiedTableName = s"${table.database}.$tableName"
+ client.alterTable(qualifiedTableName, hiveTable)
}
- override def alterTable(table: HiveTable): Unit = withHiveState {
- val qlTable = toQlTable(table)
- client.alterTable(table.qualifiedName, qlTable)
+ override def createPartitions(
+ db: String,
+ table: String,
+ parts: Seq[CatalogTablePartition],
+ ignoreIfExists: Boolean): Unit = withHiveState {
+ val addPartitionDesc = new AddPartitionDesc(db, table, ignoreIfExists)
+ parts.foreach { s =>
+ addPartitionDesc.addPartition(s.spec.asJava,
s.storage.locationUri.orNull)
+ }
+ client.createPartitions(addPartitionDesc)
+ }
+
+ override def dropPartitions(
+ db: String,
+ table: String,
+ specs: Seq[Catalog.TablePartitionSpec]): Unit = withHiveState {
+ // TODO: figure out how to drop multiple partitions in one call
Review comment:
This TODO still exists in source code. Actually, Hive MetaStore Client
API has a method for deleting multiple partitions:
```
public List<Partition> dropPartitions(String dbName,
String tblName,
List<ObjectPair<Integer,byte[]>>
partExprs,
boolean deleteData,
boolean ifExists) throws
NoSuchObjectException,
MetaException,
org.apache.thrift.TException
```
Potentially, we could optimize it in OSS or in ... cc @cloud-fan
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
