Skip to content

[SPARK-6618][SPARK-6669][SQL] Lock Hive metastore client correctly. #5333

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
yhuai wants to merge 7 commits into from
Closed

[SPARK-6618][SPARK-6669][SQL] Lock Hive metastore client correctly. #5333

Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
5416b0f
Just protect client.
yhuai Mar 31, 2015
fee7e9c
A test?
yhuai Mar 31, 2015
1e241af
Protect InsertIntoHive.
yhuai Apr 2, 2015
d6fc32f
Missing `)`.
yhuai Apr 2, 2015
e4a9b0b
Correctly lock on MetastoreCatalog
marmbrus Apr 2, 2015
7667030
Merge pull request #2 from marmbrus/pr/5333
yhuai Apr 2, 2015
59c884f
[SQL] Lock metastore client in analyzeTable
marmbrus Apr 2, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Protect InsertIntoHive.
  • Loading branch information
@yhuai
yhuai committed Apr 2, 2015
commit 1e241af960c75f7477fe2c2e8631ffac8f2413bb
51 changes: 29 additions & 22 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ case class InsertIntoHiveTable(
@transient val sc: HiveContext = sqlContext.asInstanceOf[HiveContext]
Copy link
Contributor

@marmbrus marmbrus Apr 2, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Existing: Calling a HiveContext sc is pretty confusing. Also since this is a convenience cast it should be a def so as to not take up extra space in the object.

@transient lazy val outputClass = newSerializer(table.tableDesc).getSerializedClass
@transient private lazy val hiveContext = new Context(sc.hiveconf)
@transient private lazy val db = Hive.get(sc.hiveconf)
@transient private lazy val catalog = sc.catalog

private def newSerializer(tableDesc: TableDesc): Serializer = {
val serializer = tableDesc.getDeserializerClass.newInstance().asInstanceOf[Serializer]
Expand Down Expand Up @@ -199,38 +199,45 @@ case class InsertIntoHiveTable(
orderedPartitionSpec.put(entry.getName,partitionSpec.get(entry.getName).getOrElse(""))
}
val partVals = MetaStoreUtils.getPvals(table.hiveQlTable.getPartCols, partitionSpec)
db.validatePartitionNameCharacters(partVals)
catalog.synchronized {
catalog.client.validatePartitionNameCharacters(partVals)
}
// inheritTableSpecs is set to true. It should be set to false for a IMPORT query
// which is currently considered as a Hive native command.
val inheritTableSpecs = true
// TODO: Correctly set isSkewedStoreAsSubdir.
val isSkewedStoreAsSubdir = false
if (numDynamicPartitions > 0) {
db.loadDynamicPartitions(
outputPath,
qualifiedTableName,
orderedPartitionSpec,
overwrite,
numDynamicPartitions,
holdDDLTime,
isSkewedStoreAsSubdir
)
catalog.synchronized {
catalog.client.loadDynamicPartitions(
outputPath,
qualifiedTableName,
orderedPartitionSpec,
overwrite,
numDynamicPartitions,
holdDDLTime,
isSkewedStoreAsSubdir
}
} else {
db.loadPartition(
catalog.synchronized {
catalog.client.loadPartition(
outputPath,
qualifiedTableName,
orderedPartitionSpec,
overwrite,
holdDDLTime,
inheritTableSpecs,
isSkewedStoreAsSubdir)
}
}
} else {
catalog.synchronized {
catalog.client.loadTable(
outputPath,
qualifiedTableName,
orderedPartitionSpec,
overwrite,
holdDDLTime,
inheritTableSpecs,
isSkewedStoreAsSubdir)
holdDDLTime)
}
} else {
db.loadTable(
outputPath,
qualifiedTableName,
overwrite,
holdDDLTime)
}

// Invalidate the cache.
Expand Down