[SPARK-17970][SQL] store partition spec in metastore for data source table #15515

cloud-fan · 2016-10-17T15:46:39Z

What changes were proposed in this pull request?

We should follow hive table and also store partition spec in metastore for data source table.
This brings 2 benefits:

It's more flexible to manage the table data files, as users can use ADD PARTITION, DROP PARTITION and RENAME PARTITION
We don't need to cache all file status for data source table anymore.

How was this patch tested?

existing tests.

to answer a query

partitions' files from a table file catalog

BasicFileCatalog (or vice-versa)

instead of once per partition

moving a protected method down and making it private

tables

* [SPARK-16980][SQL] Load only catalog table partition metadata required to answer a query * Add a new catalyst optimizer rule to SQL core for pruning unneeded partitions' files from a table file catalog * Include the type of file catalog in the FileSourceScanExec metadata * TODO: Consider renaming FileCatalog to better differentiate it from BasicFileCatalog (or vice-versa) * try out parquet case insensitive fallback * Refactor the FileSourceScanExec.metadata val to make it prettier * fix and add test for input files * rename test * Refactor `TableFileCatalog.listFiles` to call `listDataLeafFiles` once instead of once per partition * fix it * more test cases * also fix a bug with zero partitions selected * feature flag * add comments * extend and fix flakiness in test * Enhance `ParquetMetastoreSuite` with mixed-case partition columns * Tidy up a little by removing some unused imports, an unused method and moving a protected method down and making it private * Put partition count in `FileSourceScanExec.metadata` for partitioned tables * Fix some errors in my revision of `ParquetSourceSuite` * Thu Oct 13 17:18:14 PDT 2016 * more generic * Thu Oct 13 18:09:42 PDT 2016 * Thu Oct 13 18:09:55 PDT 2016 * Thu Oct 13 18:22:31 PDT 2016

) * Thu Oct 13 19:02:36 PDT 2016 * Thu Oct 13 19:03:06 PDT 2016

to answer a query

partition data from a HadoopFsRelation's file catalog

instead of once per partition

SparkQA · 2016-10-27T00:16:38Z

Test build #67604 has finished for PR 15515 at commit 9a6fff6.

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

SparkQA · 2016-10-27T03:12:56Z

Test build #67610 has finished for PR 15515 at commit 8c80555.

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

SparkQA · 2016-10-27T04:41:58Z

Test build #67615 has finished for PR 15515 at commit b6776cc.

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

ericl · 2016-10-27T05:06:00Z

[info] - SPARK-10562: partition by column with mixed case name *** FAILED *** (605 milliseconds)
[info]   java.lang.reflect.InvocationTargetException:
[info]   at sun.reflect.GeneratedMethodAccessor156.invoke(Unknown Source)
[info]   at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
[info]   at java.lang.reflect.Method.invoke(Method.java:497)
[info]   at org.apache.spark.sql.hive.client.Shim_v0_13.getPartitionsByFilter(HiveShim.scala:588)

Odd that the last commit could cause this, maybe it's a flake? jenkins test this please

cloud-fan · 2016-10-27T05:21:49Z

retest this please

SparkQA · 2016-10-27T07:08:11Z

Test build #67625 has finished for PR 15515 at commit b6776cc.

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

cloud-fan · 2016-10-27T07:31:03Z

retest this please

SparkQA · 2016-10-27T09:51:12Z

Test build #67632 has finished for PR 15515 at commit b6776cc.

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

yhuai · 2016-10-27T20:01:02Z

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala

+            df.sparkSession.sqlContext.conf.manageFilesourcePartitions) {
+          // Need to recover partitions into the metastore so our saved data is visible.
+          val recoverPartitionCmd = AlterTableRecoverPartitionsCommand(tableDesc.identifier)
+          Union(createCmd, recoverPartitionCmd)


Let's add a special node for running a sequence of commands. We are relying on the implementation of Union at here. Let's address this in a follow-up PR.

yhuai · 2016-10-27T20:28:02Z

sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeColumnCommand.scala

      case logicalRel: LogicalRelation if logicalRel.catalogTable.isDefined =>
-        updateStats(logicalRel.catalogTable.get, logicalRel.relation.sizeInBytes)
+        updateStats(logicalRel.catalogTable.get,
+          AnalyzeTableCommand.calculateTotalSize(sessionState, logicalRel.catalogTable.get))


How's the cost of AnalyzeTableCommand.calculateTotalSize? Also, why is sizeInBytes not the latest size?

yhuai · 2016-10-27T20:32:35Z

sql/core/src/main/scala/org/apache/spark/sql/execution/command/createDataSourceTables.scala

+          sparkSession.sqlContext.conf.manageFilesourcePartitions =>
+        // Need to recover partitions into the metastore so our saved data is visible.
+        sparkSession.sessionState.executePlan(
+          AlterTableRecoverPartitionsCommand(table.identifier)).toRdd


Seems we have a similar logic at https://github.com/apache/spark/pull/15515/files#diff-94fbd986b04087223f53697d4b6cab24R396. Are we recovering partitions twice?

I just double checked with create table test_sel2 USING parquet PARTITIONED BY (fieldone, fieldtwo) AS SELECT id as fieldzero, id as fieldone, id as fieldtwo from range(100) and it is using a different path, so the recovery is not duplicated.

yhuai · 2016-10-27T20:34:01Z

sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala

+    // This is always the case for Hive format tables, but is not true for Datasource tables created
+    // before Spark 2.1 unless they are converted via `msck repair table`.
+    spark.sessionState.catalog.alterTable(table.copy(partitionProviderIsHive = true))
+    catalog.refreshTable(tableName)


Let's also update the doc.

Yeah, this is a pretty major change for 2.1. Shall we do it in a followup once the patches for 2.1 are finalized?

yhuai · 2016-10-27T20:36:09Z

Looks good. I left a few questions. Let me know if you want to address them in follow-up prs.

ericl · 2016-10-27T20:47:51Z

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala

+            df.sparkSession.sqlContext.conf.manageFilesourcePartitions) {
+          // Need to recover partitions into the metastore so our saved data is visible.
+          val recoverPartitionCmd = AlterTableRecoverPartitionsCommand(tableDesc.identifier)
+          Union(createCmd, recoverPartitionCmd)


Makes sense. I'll file a ticket after this is merged

ericl · 2016-10-27T20:48:47Z

sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeColumnCommand.scala

      case logicalRel: LogicalRelation if logicalRel.catalogTable.isDefined =>
-        updateStats(logicalRel.catalogTable.get, logicalRel.relation.sizeInBytes)
+        updateStats(logicalRel.catalogTable.get,
+          AnalyzeTableCommand.calculateTotalSize(sessionState, logicalRel.catalogTable.get))


The reason is that the relation's size is no longer computed when it is resolved, so we have to force a table scan here to get an updated size.

Weird that github reordered my comment above actually ^

ericl · 2016-10-27T21:15:15Z

sql/core/src/main/scala/org/apache/spark/sql/execution/command/createDataSourceTables.scala

+          sparkSession.sqlContext.conf.manageFilesourcePartitions =>
+        // Need to recover partitions into the metastore so our saved data is visible.
+        sparkSession.sessionState.executePlan(
+          AlterTableRecoverPartitionsCommand(table.identifier)).toRdd


I just double checked with create table test_sel2 USING parquet PARTITIONED BY (fieldone, fieldtwo) AS SELECT id as fieldzero, id as fieldone, id as fieldtwo from range(100) and it is using a different path, so the recovery is not duplicated.

ericl · 2016-10-27T21:16:07Z

sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala

+    // This is always the case for Hive format tables, but is not true for Datasource tables created
+    // before Spark 2.1 unless they are converted via `msck repair table`.
+    spark.sessionState.catalog.alterTable(table.copy(partitionProviderIsHive = true))
+    catalog.refreshTable(tableName)


Yeah, this is a pretty major change for 2.1. Shall we do it in a followup once the patches for 2.1 are finalized?

ericl · 2016-10-27T21:16:46Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala

+
+      if (l.catalogTable.isDefined && l.catalogTable.get.partitionColumnNames.nonEmpty &&
+          l.catalogTable.get.partitionProviderIsHive) {
+        // TODO(ekl) we should be more efficient here and only recover the newly added partitions


I have a follow-up pr for this. Will cc you.

yhuai · 2016-10-27T21:21:34Z

Cool. I am merging this pr to unblock other tasks.

gatorsmile · 2016-12-18T01:50:05Z

sql/hive/src/test/scala/org/apache/spark/sql/hive/PartitionProviderCompatibilitySuite.scala

+    }
+  }
+
+  test("when partition management is disabled, we preserve the old behavior even for new tables") {


I just checked the old behavior. It is different from the existing behavior in our Spark 2.0 build. Let me do a quick fix to resolve it.

Michael Allman and others added 30 commits October 13, 2016 16:58

[SPARK-16980][SQL] Load only catalog table partition metadata required

c2eacb7

to answer a query

Add a new catalyst optimizer rule to SQL core for pruning unneeded

1f611c4

partitions' files from a table file catalog

Include the type of file catalog in the FileSourceScanExec metadata

8b24ead

TODO: Consider renaming FileCatalog to better differentiate it from

f82f0d2

BasicFileCatalog (or vice-versa)

Refactor the FileSourceScanExec.metadata val to make it prettier

198dd94

try out parquet case insensitive fallback

1f0d5d8

fix and add test for input files

59de5ca

rename test

3b51624

Refactor TableFileCatalog.listFiles to call listDataLeafFiles once

acc84f0

instead of once per partition

fix it

f94863d

more test cases

022d5b9

also fix a bug with zero partitions selected

8bd27be

feature flag

0958bcd

add comments

291cee7

extend and fix flakiness in test

627572e

Enhance ParquetMetastoreSuite with mixed-case partition columns

6d8e7ea

Tidy up a little by removing some unused imports, an unused method and

21caa93

moving a protected method down and making it private

Put partition count in FileSourceScanExec.metadata for partitioned

d7795cd

tables

Fix some errors in my revision of ParquetSourceSuite

765f93c

Actually register the hive catalog metrics, also revert broken tests (#6

71049d1

) * Thu Oct 13 19:02:36 PDT 2016 * Thu Oct 13 19:03:06 PDT 2016

Fri Oct 14 14:04:01 PDT 2016

6a63afd

[SPARK-16980][SQL] Load only catalog table partition metadata required

6b02b3c

to answer a query

Add a new catalyst optimizer rule to SQL core for pruning unnecessary

e816919

partition data from a HadoopFsRelation's file catalog

Include the type of file catalog in the FileSourceScanExec metadata

8cca6dc

try out parquet case insensitive fallback

7acc3f1

Refactor the FileSourceScanExec.metadata val to make it prettier

cf7d1f1

fix and add test for input files

c75855c

rename test

821372f

Refactor TableFileCatalog.listFiles to call listDataLeafFiles once

d0b893b

instead of once per partition

fix tree-node suite

8c80555

ericl force-pushed the partition branch from b336de4 to 8c80555 Compare October 27, 2016 00:49

minor cleanup

b6776cc

yhuai reviewed Oct 27, 2016

View reviewed changes

ericl approved these changes Oct 27, 2016

View reviewed changes

asfgit closed this in ccb1154 Oct 27, 2016

ericl mentioned this pull request Oct 29, 2016

[SPARK-18146] [SQL] Avoid using Union to chain together create table and repair partition commands #15665

Closed

gatorsmile reviewed Dec 18, 2016

View reviewed changes

MaxGekk mentioned this pull request Dec 5, 2020

[SPARK-33670][SQL] Verify the partition provider is Hive in v1 SHOW TABLE EXTENDED #30618

Closed

MaxGekk mentioned this pull request Dec 7, 2020

[SPARK-33670][SQL][3.0] Verify the partition provider is Hive in v1 SHOW TABLE EXTENDED #30640

Closed

MaxGekk mentioned this pull request Dec 7, 2020

[SPARK-33670][SQL][2.4] Verify the partition provider is Hive in v1 SHOW TABLE EXTENDED #30641

Closed

sunchao mentioned this pull request Jul 20, 2021

[SPARK-36136][SQL][TESTS] Refactor PruneFileSourcePartitionsSuite etc to a different package #33350

Closed

[SPARK-17970][SQL] store partition spec in metastore for data source table #15515

[SPARK-17970][SQL] store partition spec in metastore for data source table #15515

Uh oh!

Conversation

cloud-fan commented Oct 17, 2016

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

SparkQA commented Oct 27, 2016

Uh oh!

SparkQA commented Oct 27, 2016

Uh oh!

SparkQA commented Oct 27, 2016

Uh oh!

ericl commented Oct 27, 2016

Uh oh!

cloud-fan commented Oct 27, 2016

Uh oh!

SparkQA commented Oct 27, 2016

Uh oh!

cloud-fan commented Oct 27, 2016

Uh oh!

SparkQA commented Oct 27, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yhuai commented Oct 27, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yhuai commented Oct 27, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants