[SPARK-30244][SQL] Emit pre/post events for "Partition" methods in ExternalCatalogWithListener #27030

ayudovin · 2019-12-28T16:08:30Z

What changes were proposed in this pull request?

Add events to CREATE, DROP, RENAME and ALTER partitions in ExternalCatalogWithListener.

Why are the changes needed?

This changes needs for adding hooks for partitions in ExternalCatalogWithListener.

Does this PR introduce any user-facing change?

No

How was this patch tested?

This functionality is covered by unit tests.

ayudovin · 2020-01-04T21:14:06Z

@hvanhovell, Could you please review this pull request?

ayudovin · 2020-01-09T18:16:49Z

cc @dongjoon-hyun

dongjoon-hyun · 2020-01-09T18:27:44Z

ok to test

dongjoon-hyun · 2020-01-09T18:29:28Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/events.scala

+ * Event fired before a partition is created.
+ */
+case class CreatePartitionPreEvent(database: String, name: String,
+                                   parts: Seq[CatalogTablePartition]) extends PartitionEvent


Indentation? You had better follow the existing style used in this file.

dongjoon-hyun · 2020-01-09T18:31:36Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/events.scala

@@ -202,3 +203,62 @@ case class RenameFunctionEvent(
    name: String,
    newName: String)
  extends FunctionEvent
+
+trait PartitionEvent extends DatabaseEvent {


Since Partition belongs to Table, TableEvent is better than DatabaseEvent?

dongjoon-hyun · 2020-01-09T18:32:21Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/events.scala

+  /**
+   * Name of the table that was touched.
+   */
+  val name: String


parts: Seq[CatalogTablePartition] instead of val name: String? If we extends TableEvent, Table name is already inherited in TableEvent.

Does it make sense to use parts: Seq[TablePartitionSpec]?
CatalogTablePartition contains TablePartitionSpec inside.

Create and Alter partition have as parameter Seq[CatalogTablePartition]

Rename and Drop partition have as parameter Seq[TablePartitionSpec]

Got it. I was referring only a partial part of your code. In that case, we don't need anything here in PartitionEvent.

dongjoon-hyun · 2020-01-09T22:00:16Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/events.scala

-                                   parts: Seq[CatalogTablePartition]) extends PartitionEvent
+case class CreatePartitionPreEvent(
+    database: String,
+    name: String) extends PartitionEvent


Ur? What I meant was we don't need to have the partition information at PartitionEvent. We need partition information in this event, don't we?

For the indentation, shall we put extends PartitionEvent into a new line?

Do you mean to create such a event:

case class CreatePartitionPreEvent( database: String, name: String, parts: Seq[CatalogTablePartition]) extends PartitionEvent

but PartitionEvent will be empty ?

shall we put extends PartitionEvent into a new line

yes, I'll put it into a new line.

SparkQA · 2020-01-09T22:43:13Z

Test build #116410 has finished for PR 27030 at commit 5c242dc.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait PartitionEvent extends DatabaseEvent
case class CreatePartitionPreEvent(database: String, name: String,
case class CreatePartitionEvent(database: String, name: String,
case class DropPartitionPreEvent(database: String, name: String,
case class DropPartitionEvent(database: String, name: String,
case class RenamePartitionPreEvent(database: String, name: String,
case class RenamePartitionEvent(database: String, name: String,
case class AlterPartitionPreEvent(database: String, name: String,
case class AlterPartitionEvent(database: String, name: String,

SparkQA · 2020-01-10T02:05:41Z

Test build #116418 has finished for PR 27030 at commit af04a62.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class CreatePartitionPreEvent(
case class CreatePartitionEvent(
case class DropPartitionPreEvent(
case class DropPartitionEvent(
case class RenamePartitionPreEvent(
case class RenamePartitionEvent(
case class AlterPartitionPreEvent(
case class AlterPartitionEvent(

SparkQA · 2020-01-10T02:54:32Z

Test build #116419 has finished for PR 27030 at commit 0c5edc2.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-01-10T08:05:02Z

Test build #116464 has finished for PR 27030 at commit 3cb79fe.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

…on-events

SparkQA · 2020-01-10T17:48:11Z

Test build #116492 has finished for PR 27030 at commit 4d5ba9b.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2020-01-11T03:41:03Z

...atalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogEventSuite.scala

+    val dbDefinition = createDbDefinition()
+
+    val storage = CatalogStorageFormat.empty.copy(
+      locationUri = Option(uri1))


Is this table storage pointing db path? Maybe, do you want the following?

val path1 = Files.createTempDirectory("db_") val path2 = Files.createTempDirectory(path1, "tbl_") val uri1 = preparePath(path1) val uri2 = preparePath(path2) // CREATE val dbDefinition = createDbDefinition(uri1) val storage = CatalogStorageFormat.empty.copy( locationUri = Option(uri2))

yes, you are right

SparkQA · 2020-01-13T01:01:59Z

Test build #116570 has finished for PR 27030 at commit e1220e1.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2020-01-13T01:57:07Z

...atalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogEventSuite.scala

+    catalog.createTable(tableDefinition, ignoreIfExists = false)
+    checkEvents(CreateTablePreEvent("db5", "tbl1") :: CreateTableEvent("db5", "tbl1") :: Nil)
+
+    catalog.createPartitions("db5", "tbl1", Seq(partition), ignoreIfExists = false)


Ur, is this a valid test case? What happens when we try to create CatalogTypes.emptyTablePartitionSpec?

agree with you, CatalogTypes.emptyTablePartitionSpec is not good option for unit test. I have changed it.

Despite this, I think that empty partition creation should also produce the event.

Does it happen in the real world? Can you give me an SQL example to create the empty partition?

Despite this, I think that empty partition creation should also produce the event.

hm, I cannot find an example, that's why I also think it isn't happening in the real world.

dongjoon-hyun · 2020-01-13T01:59:21Z

...atalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogEventSuite.scala

+    val newPartition = CatalogTablePartition(spec = Map("key1" -> "1", "key2" -> "2"),
+      storage = CatalogStorageFormat.empty)
+
+    catalog.renamePartitions("db5", "tbl1", Seq(partition.spec), Seq(newPartition.spec))


This also sounds like empty spec to key1=1 , key2=2. Is this valid?

now it looks like key=0 to key1=1, key2=2

Could you give me the real SQL statement for that?

So, I could not find the real SQL statement for that, that's why I changed this unit tests.

SparkQA · 2020-01-14T00:15:34Z

Test build #116661 has finished for PR 27030 at commit 43aeb9b.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2020-01-14T22:59:32Z

Retest this please.

SparkQA · 2020-01-15T08:05:01Z

Test build #116758 has finished for PR 27030 at commit 8da2c55.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2020-01-17T04:25:35Z

Please don't forget the above comment.

Could you give me the real SQL statement for that?

ayudovin · 2020-01-17T05:57:43Z

Please don't forget the above comment.

Could you give me the real SQL statement for that?

@dongjoon-hyun, sorry, I have added a comment

ayudovin · 2020-01-23T21:20:07Z

Hi @dongjoon-hyun , Do you have still any suggestions or comments on this pull request?

dongjoon-hyun · 2020-01-23T22:03:05Z

Sorry for being late, @ayudovin . I'll try to review again tonight.

dongjoon-hyun · 2020-01-23T22:03:18Z

Retest this please.

SparkQA · 2020-01-24T02:04:18Z

Test build #117321 has finished for PR 27030 at commit 8da2c55.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2020-01-24T04:27:24Z

...atalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogEventSuite.scala

+
+    catalog.createPartitions("db5", "tbl1", Seq(partition), ignoreIfExists = true)
+    checkEvents(CreatePartitionPreEvent("db5", "tbl1", Seq(partition)) ::
+      CreatePartitionEvent("db5", "tbl1", Seq(partition)) :: Nil)


Ur, is this correct? We already created the same partition at line 240 and we tried with ignoreIfExists=true here. I guess CreatePartitionEvent should not occur.

I agree with you, it looks incorrect, but we can not check if the partition has been created or not created in case of using ignoreIfExists=true

If then, this is not helpful in that case because we cannot distinguish the difference.

dongjoon-hyun · 2020-01-24T04:29:34Z

...atalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogEventSuite.scala

+        RenamePartitionEvent("db5", "tbl1", Seq(partition.spec), Seq(newPartition.spec)) ::
+        Nil)
+
+    // ALTER


ALTER nothing? In this case, I guess no event might be natural.
BTW, if there is no valid SQL query for this, I guess we can remove this test case.

dongjoon-hyun · 2020-01-24T04:30:21Z

...atalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogEventSuite.scala

+    intercept[AnalysisException] {
+      catalog.dropPartitions("db5", "tbl2", Seq(newPartition.spec),
+        ignoreIfNotExists = false, purge = true, retainData = false)
+    }


It's always a good practice to check the error message. Sometime, AnalysisException is changed unexpectedly.

dongjoon-hyun · 2020-01-24T04:33:20Z

Also, I'm worrying the situation where this implementation causes a sabotage on the event message pipeline. Sometimes, the changed number of partition can be huge in the production.

Hi, @hvanhovell . How do you think about this PR? Could you give us some advice?

SparkQA · 2020-01-25T18:54:16Z

Test build #117401 has finished for PR 27030 at commit 30686b4.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2020-01-26T00:31:29Z

@ayudovin . The current status is that we are waiting the original author of this framework, @hvanhovell . We need to check whether this is not added intentionally or not. Also, #27030 (comment) is another concern no this approach because it can be misleading.

ayudovin · 2020-01-26T16:05:41Z

@ayudovin . The current status is that we are waiting the original author of this framework, @hvanhovell . We need to check whether this is not added intentionally or not. Also, #27030 (comment) is another concern no this approach because it can be misleading.

I have got, It makes sense

HyukjinKwon · 2020-02-25T03:41:56Z

ok to test

HyukjinKwon · 2020-02-25T03:42:36Z

WDYT @hvanhovell?

SparkQA · 2020-02-25T08:04:57Z

Test build #118893 has finished for PR 27030 at commit 30686b4.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2020-03-09T19:27:01Z

Hi, @gatorsmile and @hvanhovell .
How do you think about this extention?

wankunde · 2020-05-02T10:44:07Z

@ayudovin @dongjoon-hyun @HyukjinKwon
Can you emit LoadPartitionEvent and LoadDynamicPartitionsEvent ?
Now the user does not know which partitions are loaded, if LoadDynamicPartitionsEvent events are emitted to the listener, the user will know the changed partitions and add a custom callback function.

  def calculateDynamicPartitions(
      loadPath: Path,
      partitionSpec: TablePartitionSpec,
      numDP: Int): Array[TablePartitionSpec] = {
    assert(numDP > 0)
    try {
      val fs = loadPath.getFileSystem(new Configuration())
      fs.listStatus(loadPath).flatMap { fileStatus =>
        val dirName = fileStatus.getPath.getName
        if (dirName.contains("=")) {
          val Array(partitionColumn, partitionValue) = dirName.split("=")
          val newPartitionSpec = partitionSpec + (partitionColumn -> partitionValue)
          if (numDP == 1) {
            Array[TablePartitionSpec](newPartitionSpec)
          } else {
            calculateDynamicPartitions(fileStatus.getPath, newPartitionSpec, numDP - 1)
          }
        } else {
          Array[TablePartitionSpec]()
        }
      }
    } catch {
      case ex: Exception => Array[TablePartitionSpec]()
    }
  }	 
}

AmplabJenkins · 2020-07-16T13:06:22Z

Can one of the admins verify this patch?

obogobo · 2020-09-09T23:56:56Z

would love a status update on this 😄 is it still possible for Spark 3.1.0? (current tag on the ticket)

github-actions · 2020-12-19T00:51:16Z

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!

Adding partition events

5c242dc

ayudovin changed the title ~~[SPARK-30244] - Emit pre/post events for "Partition" methods in ExternalCatalogWithListener~~ [SPARK-30244][Resource-Manager][ Kubernetes] - Emit pre/post events for "Partition" methods in ExternalCatalogWithListener Dec 28, 2019

ayudovin changed the title ~~[SPARK-30244][Resource-Manager][ Kubernetes] - Emit pre/post events for "Partition" methods in ExternalCatalogWithListener~~ [SPARK-30244][Catalyst] - Emit pre/post events for "Partition" methods in ExternalCatalogWithListener Dec 28, 2019

ayudovin changed the title ~~[SPARK-30244][Catalyst] - Emit pre/post events for "Partition" methods in ExternalCatalogWithListener~~ [SPARK-30244][SQL][Catalyst] - Emit pre/post events for "Partition" methods in ExternalCatalogWithListener Dec 28, 2019

dongjoon-hyun changed the title ~~[SPARK-30244][SQL][Catalyst] - Emit pre/post events for "Partition" methods in ExternalCatalogWithListener~~ [SPARK-30244][SQL] Emit pre/post events for "Partition" methods in ExternalCatalogWithListener Jan 9, 2020

dongjoon-hyun added the SQL label Jan 9, 2020

dongjoon-hyun reviewed Jan 9, 2020

View reviewed changes

Applying changes by suggestions

af04a62

dongjoon-hyun reviewed Jan 9, 2020

View reviewed changes

Applying changes by suggestions

0c5edc2

Putting extends PartitionEvent into a new line

3cb79fe

Merge branch 'master' of https://github.com/apache/spark into partiti…

4d5ba9b

…on-events

dongjoon-hyun reviewed Jan 11, 2020

View reviewed changes

Updating unit test

e1220e1

dongjoon-hyun reviewed Jan 13, 2020

View reviewed changes

Updating unit test

43aeb9b

ayudovin requested a review from dongjoon-hyun January 16, 2020 13:08

ayudovin closed this Jan 17, 2020

ayudovin reopened this Jan 17, 2020

dongjoon-hyun reviewed Jan 24, 2020

View reviewed changes

Updating unit test

30686b4

ayudovin requested a review from dongjoon-hyun January 25, 2020 14:42

dongjoon-hyun mentioned this pull request Apr 30, 2020

[SPARK-31616][SQL] Add partition event listener in ExternalCatalogWithListener #28421

Closed

github-actions bot added the Stale label Dec 19, 2020

github-actions bot closed this Dec 20, 2020

[SPARK-30244][SQL] Emit pre/post events for "Partition" methods in ExternalCatalogWithListener #27030

[SPARK-30244][SQL] Emit pre/post events for "Partition" methods in ExternalCatalogWithListener #27030

Conversation

ayudovin commented Dec 28, 2019

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

ayudovin commented Jan 4, 2020

ayudovin commented Jan 9, 2020

dongjoon-hyun commented Jan 9, 2020

Choose a reason for hiding this comment

dongjoon-hyun Jan 9, 2020 • edited

Choose a reason for hiding this comment

dongjoon-hyun Jan 9, 2020 • edited

Choose a reason for hiding this comment

ayudovin Jan 9, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ayudovin Jan 9, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jan 9, 2020

SparkQA commented Jan 10, 2020

SparkQA commented Jan 10, 2020

SparkQA commented Jan 10, 2020

SparkQA commented Jan 10, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jan 13, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jan 14, 2020

dongjoon-hyun commented Jan 14, 2020

SparkQA commented Jan 15, 2020

dongjoon-hyun commented Jan 17, 2020

ayudovin commented Jan 17, 2020 • edited

ayudovin commented Jan 23, 2020

dongjoon-hyun commented Jan 23, 2020

dongjoon-hyun commented Jan 23, 2020

SparkQA commented Jan 24, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dongjoon-hyun commented Jan 24, 2020

SparkQA commented Jan 25, 2020

dongjoon-hyun commented Jan 26, 2020 • edited

ayudovin commented Jan 26, 2020

HyukjinKwon commented Feb 25, 2020

HyukjinKwon commented Feb 25, 2020

SparkQA commented Feb 25, 2020

dongjoon-hyun commented Mar 9, 2020

wankunde commented May 2, 2020 • edited

AmplabJenkins commented Jul 16, 2020

obogobo commented Sep 9, 2020

github-actions bot commented Dec 19, 2020

dongjoon-hyun Jan 9, 2020 •

edited

dongjoon-hyun Jan 9, 2020 •

edited

ayudovin Jan 9, 2020 •

edited

ayudovin Jan 9, 2020 •

edited

ayudovin commented Jan 17, 2020 •

edited

dongjoon-hyun commented Jan 26, 2020 •

edited

wankunde commented May 2, 2020 •

edited