Skip to content

[SPARK-22405][SQL] Add new alter table and alter database related ExternalCatalogEvent #19649

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
jerryshao wants to merge 5 commits into from
Closed

[SPARK-22405][SQL] Add new alter table and alter database related ExternalCatalogEvent #19649

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
df2cab9
Add more ExternalCatalogEvent
jerryshao Nov 3, 2017
a3867b7
Fix merging conflict
jerryshao Nov 3, 2017
8eef970
Addess part of the comments
jerryshao Nov 6, 2017
c79c314
Address the comments
jerryshao Nov 7, 2017
6b4fcff
Change to String type as comment suggested
jerryshao Nov 9, 2017
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
27 changes: 24 additions & 3 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,14 @@ abstract class ExternalCatalog
* Note: If the underlying implementation does not support altering a certain field,
* this becomes a no-op.
*/
def alterDatabase(dbDefinition: CatalogDatabase): Unit
final def alterDatabase(dbDefinition: CatalogDatabase): Unit = {
val db = dbDefinition.name
postToAll(AlterDatabasePreEvent(db))
doAlterDatabase(dbDefinition)
postToAll(AlterDatabaseEvent(db))
}

protected def doAlterDatabase(dbDefinition: CatalogDatabase): Unit

def getDatabase(db: String): CatalogDatabase

Expand Down Expand Up @@ -147,7 +154,15 @@ abstract class ExternalCatalog
* Note: If the underlying implementation does not support altering a certain field,
* this becomes a no-op.
*/
def alterTable(tableDefinition: CatalogTable): Unit
final def alterTable(tableDefinition: CatalogTable): Unit = {

@cloud-fan cloud-fan Nov 3, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I'd suggest we leave it for now and watch other alterTableXXX events instead. I feel it's an overkill to have a heavy alterTable method to handling all table metadata updating, I think we will add more and more fine-grained alter table methods in the future, like alterTableStats, alterTableDataSchema, etc. and eventually this alterTable method will go away.

@wzhfy wzhfy Nov 4, 2017 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

This depends on which level of details we wanna collect in the event. Are there any guidelines or documentation of what events spark should monitor?
Besides, partition changes are missing, I think it's necessary to monitor these changes.

@jerryshao jerryshao Nov 6, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@cloud-fan , since now we expose alterTable interface for other components to leverage, if we don't track this, then looks like we missed a piece of ExternalCatalogEvents. I think for now we can add this AlterTableEvent, later on if we removed this method, then we can make this event a no-op (only kept for compatibility), what do you think?

@wzhfy , I was thinking to add partition related events, but I'm not clearly sure why this whole piece is missing and is it necessary to add partition related events? If we have an agreement on such events, I'm OK to add them.

val db = tableDefinition.database
val name = tableDefinition.identifier.table
postToAll(AlterTablePreEvent(db, name))
doAlterTable(tableDefinition)
postToAll(AlterTableEvent(db, name))
}

protected def doAlterTable(tableDefinition: CatalogTable): Unit

/**
* Alter the data schema of a table identified by the provided database and table name. The new
Expand All @@ -158,7 +173,13 @@ abstract class ExternalCatalog
* @param table Name of table to alter schema for
* @param newDataSchema Updated data schema to be used for the table.
*/
def alterTableDataSchema(db: String, table: String, newDataSchema: StructType): Unit
final def alterTableDataSchema(db: String, table: String, newDataSchema: StructType): Unit = {

@cloud-fan cloud-fan Nov 6, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

how about alterTableStats?

postToAll(AlterTableSchemaPreEvent(db, table))

@cloud-fan cloud-fan Nov 3, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

AlterTableDataSchemaXXX

@cloud-fan cloud-fan Nov 3, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

shall we carry the new schema?

@jerryshao jerryshao Nov 6, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

For me I think it is not so necessary to carry the new schema, we can query the catalog by db and table to get this newly set schema.

doAlterTableDataSchema(db, table, newDataSchema)
postToAll(AlterTableSchemaEvent(db, table))
}

protected def doAlterTableDataSchema(db: String, table: String, newDataSchema: StructType): Unit

/** Alter the statistics of a table. If `stats` is None, then remove all existing statistics. */
def alterTableStats(db: String, table: String, stats: Option[CatalogStatistics]): Unit
Expand Down
6 changes: 3 additions & 3 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/InMemoryCatalog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,7 +152,7 @@ class InMemoryCatalog(
}
}

override def alterDatabase(dbDefinition: CatalogDatabase): Unit = synchronized {
override def doAlterDatabase(dbDefinition: CatalogDatabase): Unit = synchronized {
requireDbExists(dbDefinition.name)
catalog(dbDefinition.name).db = dbDefinition
}
Expand Down Expand Up @@ -294,7 +294,7 @@ class InMemoryCatalog(
catalog(db).tables.remove(oldName)
}

override def alterTable(tableDefinition: CatalogTable): Unit = synchronized {
override def doAlterTable(tableDefinition: CatalogTable): Unit = synchronized {
assert(tableDefinition.identifier.database.isDefined)
val db = tableDefinition.identifier.database.get
requireTableExists(db, tableDefinition.identifier.table)
Expand All @@ -303,7 +303,7 @@ class InMemoryCatalog(
catalog(db).tables(tableDefinition.identifier.table).table = newTableDefinition
}

override def alterTableDataSchema(
override def doAlterTableDataSchema(
db: String,
table: String,
newDataSchema: StructType): Unit = synchronized {
Expand Down
34 changes: 33 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/events.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@
package org.apache.spark.sql.catalyst.catalog

import org.apache.spark.scheduler.SparkListenerEvent
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.types.StructType

@viirya viirya Nov 4, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I didn't see StructType and TablePartitionSpec are used below.


/**
* Event emitted by the external catalog when it is modified. Events are either fired before or
Expand Down Expand Up @@ -61,6 +63,16 @@ case class DropDatabasePreEvent(database: String) extends DatabaseEvent
*/
case class DropDatabaseEvent(database: String) extends DatabaseEvent

/**
* Event fired before a database is altered.
*/
case class AlterDatabasePreEvent(database: String) extends DatabaseEvent

@dongjoon-hyun dongjoon-hyun Nov 8, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Hi, @jerryshao .
We are adding AlterTableEvent and AlterDatabaseEvent. Can we have a more specific PR title instead of Add more ExternalCatalogEvent?

@jerryshao jerryshao Nov 9, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Sure, I will update the title.


/**
* Event fired after a database is altered.
*/
case class AlterDatabaseEvent(database: String) extends DatabaseEvent

/**
* Event fired when a table is created, dropped or renamed.
*/
Expand Down Expand Up @@ -110,7 +122,27 @@ case class RenameTableEvent(
extends TableEvent

/**
* Event fired when a function is created, dropped or renamed.
* Event fired before a table is altered.
*/
case class AlterTablePreEvent(database: String, name: String) extends TableEvent

/**
* Event fired after a table is altered.
*/
case class AlterTableEvent(database: String, name: String) extends TableEvent

@cloud-fan cloud-fan Nov 6, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Maybe we can have one event for all alterTableXXX, and add one more parameter to indicate which part of the table is altered.


/**
* Event fired before table schema is altered.
*/
case class AlterTableSchemaPreEvent(database: String, name: String) extends TableEvent

@hvanhovell hvanhovell Nov 3, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Do we need a separate event for alter schema?


/**
* Event fired after table schema is altered.
*/
case class AlterTableSchemaEvent(database: String, name: String) extends TableEvent

/**
* Event fired when a function is created, dropped, altered or renamed.
*/
trait FunctionEvent extends DatabaseEvent {
/**
Expand Down
18 changes: 18 additions & 0 deletions ...lyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogEventSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,11 @@ class ExternalCatalogEventSuite extends SparkFunSuite {
}
checkEvents(CreateDatabasePreEvent("db5") :: Nil)

// ALTER
val newDbDefinition = dbDefinition.copy(description = "test")
catalog.alterDatabase(newDbDefinition)
checkEvents(AlterDatabasePreEvent("db5") :: AlterDatabaseEvent("db5") :: Nil)

// DROP
intercept[AnalysisException] {
catalog.dropDatabase("db4", ignoreIfNotExists = false, cascade = false)
Expand Down Expand Up @@ -104,6 +109,8 @@ class ExternalCatalogEventSuite extends SparkFunSuite {
tableType = CatalogTableType.MANAGED,
storage = storage,
schema = new StructType().add("id", "long"))
val tableDefWithSparkVersion =

@jiangxb1987 jiangxb1987 Nov 3, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Where do we need this val?

@jerryshao jerryshao Nov 6, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Sorry this was from my original code, will update it.

tableDefinition.copy(createVersion = org.apache.spark.SPARK_VERSION)

catalog.createDatabase(dbDefinition, ignoreIfExists = false)
checkEvents(CreateDatabasePreEvent("db5") :: CreateDatabaseEvent("db5") :: Nil)
Expand All @@ -119,6 +126,17 @@ class ExternalCatalogEventSuite extends SparkFunSuite {
}
checkEvents(CreateTablePreEvent("db5", "tbl1") :: Nil)

// ALTER
val newTableDefinition = tableDefinition.copy(tableType = CatalogTableType.EXTERNAL)
catalog.alterTable(newTableDefinition)
checkEvents(AlterTablePreEvent("db5", "tbl1") :: AlterTableEvent("db5", "tbl1") :: Nil)

// ALTER schema
val newSchema = new StructType().add("id", "long", nullable = false)
catalog.alterTableDataSchema("db5", "tbl1", newSchema)
checkEvents(
AlterTableSchemaPreEvent("db5", "tbl1") :: AlterTableSchemaEvent("db5", "tbl1") :: Nil)

// RENAME
catalog.renameTable("db5", "tbl1", "tbl2")
checkEvents(
Expand Down
10 changes: 6 additions & 4 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,7 +177,7 @@ private[spark] class HiveExternalCatalog(conf: SparkConf, hadoopConf: Configurat
*
* Note: As of now, this only supports altering database properties!
*/
override def alterDatabase(dbDefinition: CatalogDatabase): Unit = withClient {
override def doAlterDatabase(dbDefinition: CatalogDatabase): Unit = withClient {
val existingDb = getDatabase(dbDefinition.name)
if (existingDb.properties == dbDefinition.properties) {
logWarning(s"Request to alter database ${dbDefinition.name} is a no-op because " +
Expand Down Expand Up @@ -540,7 +540,7 @@ private[spark] class HiveExternalCatalog(conf: SparkConf, hadoopConf: Configurat
* Note: As of now, this doesn't support altering table schema, partition column names and bucket
* specification. We will ignore them even if users do specify different values for these fields.
*/
override def alterTable(tableDefinition: CatalogTable): Unit = withClient {
override def doAlterTable(tableDefinition: CatalogTable): Unit = withClient {
assert(tableDefinition.identifier.database.isDefined)
val db = tableDefinition.identifier.database.get
requireTableExists(db, tableDefinition.identifier.table)
Expand Down Expand Up @@ -619,8 +619,10 @@ private[spark] class HiveExternalCatalog(conf: SparkConf, hadoopConf: Configurat
}
}

override def alterTableDataSchema(
db: String, table: String, newDataSchema: StructType): Unit = withClient {
override def doAlterTableDataSchema(

@jiangxb1987 jiangxb1987 Nov 3, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Please add description of this method.

db: String,
table: String,
newDataSchema: StructType): Unit = withClient {
requireTableExists(db, table)
val oldTable = getTable(db, table)
verifyDataSchema(oldTable.identifier, oldTable.tableType, newDataSchema)
Expand Down