Skip to content

[Core] Fix deadLock between XSQLSessionCatalog and workingDataSource #68

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
wenfang6 wants to merge 2 commits into from
Closed

[Core] Fix deadLock between XSQLSessionCatalog and workingDataSource #68

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9227022
fix deadLock of XSQLSessionCatalog and workingDataSource
Jan 5, 2020
f7018e1
optimize code
Jan 20, 2020
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
41 changes: 18 additions & 23 deletions sql/xsql/src/main/scala/org/apache/spark/sql/xsql/XSQLSessionCatalog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -211,11 +211,9 @@ private[xsql] class XSQLSessionCatalog(
}

def setCurrentDatabase(dsName: String, dbName: String): Unit = {
synchronized {
setWorkingDataSource(Option(dsName)) {
super.setCurrentDatabase(dbName)
externalCatalog.setCurrentDatabase(dbName)
}
setWorkingDataSource(Option(dsName)) {
super.setCurrentDatabase(dbName)
externalCatalog.setCurrentDatabase(dbName)
}
}

Expand Down Expand Up @@ -407,18 +405,17 @@ private[xsql] class XSQLSessionCatalog(
// | Methods that interact with temporary and metastore tables |
// -------------------------------------------------------------

override def renameTable(oldName: TableIdentifier, newName: TableIdentifier): Unit =
synchronized {
val oldDs = getDataSourceName(oldName)
val newDs = getDataSourceName(newName)
if (oldDs != newDs) {
throw new AnalysisException(
s"RENAME TABLE source and destination datasources do not match: '$oldDs' != '$newDs'")
}
setWorkingDataSource(oldName.dataSource) {
super.renameTable(oldName, newName)
}
override def renameTable(oldName: TableIdentifier, newName: TableIdentifier): Unit = {
val oldDs = getDataSourceName(oldName)
val newDs = getDataSourceName(newName)
if (oldDs != newDs) {
throw new AnalysisException(
s"RENAME TABLE source and destination datasources do not match: '$oldDs' != '$newDs'")
}
setWorkingDataSource(oldName.dataSource) {
super.renameTable(oldName, newName)
}
}

def truncateTable(
name: TableIdentifier,
Expand All @@ -433,17 +430,15 @@ private[xsql] class XSQLSessionCatalog(
override def dropTable(
name: TableIdentifier,
ignoreIfNotExists: Boolean,
purge: Boolean): Unit = synchronized {
purge: Boolean): Unit = {
setWorkingDataSource(name.dataSource) {
super.dropTable(name, ignoreIfNotExists, purge)
}
}

override def lookupRelation(name: TableIdentifier): LogicalPlan = {
synchronized {
setWorkingDataSource(name.dataSource) {
super.lookupRelation(name)
}
setWorkingDataSource(name.dataSource) {
super.lookupRelation(name)
}
}

Expand All @@ -459,7 +454,7 @@ private[xsql] class XSQLSessionCatalog(
}
}

override def refreshTable(name: TableIdentifier): Unit = synchronized {
override def refreshTable(name: TableIdentifier): Unit = {
setWorkingDataSource(name.dataSource) {
super.refreshTable(name)
}
Expand Down Expand Up @@ -795,7 +790,7 @@ private[xsql] class XSQLSessionCatalog(
/**
* Look up the [[ExpressionInfo]] associated with the specified function, assuming it exists.
*/
override def lookupFunctionInfo(name: FunctionIdentifier): ExpressionInfo = synchronized {
override def lookupFunctionInfo(name: FunctionIdentifier): ExpressionInfo = {
setWorkingDataSource(Option(getDataSourceName(name))) {
super.lookupFunctionInfo(name)
}
Expand Down