[SPARK-19120] Refresh Metadata Cache After Loading Hive Tables #16500

gatorsmile · 2017-01-08T00:33:49Z

What changes were proposed in this pull request?

        sql("CREATE TABLE tab (a STRING) STORED AS PARQUET")

        // This table fetch is to fill the cache with zero leaf files
        spark.table("tab").show()

        sql(
          s"""
             |LOAD DATA LOCAL INPATH '$newPartitionDir' OVERWRITE
             |INTO TABLE tab
           """.stripMargin)

        spark.table("tab").show()

In the above example, the returned result is empty after table loading. The metadata cache could be out of dated after loading new data into the table, because loading/inserting does not update the cache. So far, the metadata cache is only used for data source tables. Thus, for Hive serde tables, only parquet and orc formats are facing such issues, because the Hive serde tables in the format of parquet/orc could be converted to data source tables when spark.sql.hive.convertMetastoreParquet/spark.sql.hive.convertMetastoreOrc is on.

This PR is to refresh the metadata cache after processing the LOAD DATA command.

In addition, Spark SQL does not convert partitioned Hive tables (orc/parquet) to data source tables in the write path, but the read path is using the metadata cache for both partitioned and non-partitioned Hive tables (orc/parquet). That means, writing the partitioned parquet/orc tables still use InsertIntoHiveTable, instead of InsertIntoHadoopFsRelationCommand. To avoid reading the out-of-dated cache, InsertIntoHiveTable needs to refresh the metadata cache for partitioned tables. Note, it does not need to refresh the cache for non-partitioned parquet/orc tables, because it does not call InsertIntoHiveTable at all. Based on the comments, this PR will keep the existing logics unchanged. That means, we always refresh the table no matter whether the table is partitioned or not.

How was this patch tested?

Added test cases in parquetSuites.scala

gatorsmile · 2017-01-08T00:35:20Z

cc @ericl @cloud-fan @mallman

The actual code changes are just two lines.

gatorsmile · 2017-01-08T00:38:35Z

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala

-    sqlContext.sessionState.catalog.refreshTable(table.catalogTable.identifier)
+    if (partition.nonEmpty) {
+      sqlContext.sessionState.catalog.refreshTable(table.catalogTable.identifier)
+    }


Actually, we can further limit the calls of refreshTable. For example, checking whether the format is parquet or orc.

Why is it safe to restrict this call to the case where partition.nonEmpty?

Is this because hive serde tables do not use the file status cache?

yea I think so, but I don't think it worth to avoid this refreshTable call and add a lot of comments to explain it. This is too subtle.

@cloud-fan @ericl @mallman For non-partitioned parquet/orc tables, we convert them to the data source tables. Thus, it will not call InsertIntoHiveTable.

I know it is a little bit confusing, but I am fine to revert it back

let's revert it first, we should think about cache and refresh more thorough later.

SparkQA · 2017-01-08T02:58:58Z

Test build #71023 has finished for PR 16500 at commit 0f70e91.

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

cloud-fan · 2017-01-10T04:38:05Z

I'm wondering if we need the metadata cache anymore. Now we store partitions in the metastore, and have a cache for leaf files, what's the benefit of metadata cache?

ericl · 2017-01-12T00:40:25Z

I guess the only purpose of the cache now is to associate file-status caches with specific table names. If we removed that, then tables would have to find their file-status cache by path, or we could have a global file cache.

Maybe we also use it to hold the computed schema in some cases? Not sure if that is always provided by the metastore.

cloud-fan · 2017-01-12T02:51:19Z

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

+  test("Non-partitioned table readable after load") {
+    withTable("tab") {
+      withTempDir { src =>
+        val newPartitionDir = new File(src, "data").getCanonicalPath


why don't we just use src as the table path?

cloud-fan · 2017-01-12T02:52:09Z

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

+    }
+  }
+
+  test("Partitioned table readable after insert") {


this and the next test are not needed if we revert https://github.com/apache/spark/pull/16500/files#diff-d579db9a8f27e0bbef37720ab14ec3f6L395

cloud-fan · 2017-01-12T02:53:15Z

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

+  test("Explicitly added partitions should be readable after load") {
+    withTable("test_added_partitions") {
+      withTempDir { src =>
+        val newPartitionDir = new File(src, "partition").getCanonicalPath


why don't we just use src as partition path?

gatorsmile · 2017-01-13T02:47:21Z

Let me think whether we can improve the existing verification mechanism for both caches in the test cases. It can help us to know what the caches actually contain.

…HiveTable

SparkQA · 2017-01-13T06:48:42Z

Test build #71295 has started for PR 16500 at commit 11507cc.

SparkQA · 2017-01-13T06:52:40Z

Test build #71296 has started for PR 16500 at commit 203e36c.

cloud-fan · 2017-01-13T10:34:37Z

retest this please

SparkQA · 2017-01-13T13:49:43Z

Test build #71313 has finished for PR 16500 at commit 203e36c.

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

gatorsmile · 2017-01-14T00:23:40Z

oops... Wrong branch... Need to revert it.

…HiveTable

SparkQA · 2017-01-14T02:47:18Z

Test build #71357 has finished for PR 16500 at commit d2d751b.

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

SparkQA · 2017-01-14T02:51:20Z

Test build #71354 has finished for PR 16500 at commit c27a9af.

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

cloud-fan · 2017-01-14T04:00:26Z

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

+
+        checkAnswer(
+          spark.table("test_added_partitions"),
+          Seq(("0", 1), ("1", 1)).toDF("a", "b"))


we usually write checkAnswer(df, Row("0", 1) :: Row("1", 1) :: Nil)

or use Seq[Row] like you already did: https://github.com/apache/spark/pull/16500/files#diff-3822214ea47830564619c00c4fe7eb0aR664

: ) That is copied from the other test cases. Let me correct it and the others in this test suite.

SparkQA · 2017-01-14T23:09:54Z

Test build #71382 has finished for PR 16500 at commit 14da2b6.

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

```Scala sql("CREATE TABLE tab (a STRING) STORED AS PARQUET") // This table fetch is to fill the cache with zero leaf files spark.table("tab").show() sql( s""" |LOAD DATA LOCAL INPATH '$newPartitionDir' OVERWRITE |INTO TABLE tab """.stripMargin) spark.table("tab").show() ``` In the above example, the returned result is empty after table loading. The metadata cache could be out of dated after loading new data into the table, because loading/inserting does not update the cache. So far, the metadata cache is only used for data source tables. Thus, for Hive serde tables, only `parquet` and `orc` formats are facing such issues, because the Hive serde tables in the format of parquet/orc could be converted to data source tables when `spark.sql.hive.convertMetastoreParquet`/`spark.sql.hive.convertMetastoreOrc` is on. This PR is to refresh the metadata cache after processing the `LOAD DATA` command. In addition, Spark SQL does not convert **partitioned** Hive tables (orc/parquet) to data source tables in the write path, but the read path is using the metadata cache for both **partitioned** and non-partitioned Hive tables (orc/parquet). That means, writing the partitioned parquet/orc tables still use `InsertIntoHiveTable`, instead of `InsertIntoHadoopFsRelationCommand`. To avoid reading the out-of-dated cache, `InsertIntoHiveTable` needs to refresh the metadata cache for partitioned tables. Note, it does not need to refresh the cache for non-partitioned parquet/orc tables, because it does not call `InsertIntoHiveTable` at all. Based on the comments, this PR will keep the existing logics unchanged. That means, we always refresh the table no matter whether the table is partitioned or not. Added test cases in parquetSuites.scala Author: gatorsmile <gatorsmile@gmail.com> Closes #16500 from gatorsmile/refreshInsertIntoHiveTable. (cherry picked from commit de62ddf) Signed-off-by: Wenchen Fan <wenchen@databricks.com>

cloud-fan · 2017-01-15T12:43:51Z

thanks, merging to master/2.1!

### What changes were proposed in this pull request? ```Scala sql("CREATE TABLE tab (a STRING) STORED AS PARQUET") // This table fetch is to fill the cache with zero leaf files spark.table("tab").show() sql( s""" |LOAD DATA LOCAL INPATH '$newPartitionDir' OVERWRITE |INTO TABLE tab """.stripMargin) spark.table("tab").show() ``` In the above example, the returned result is empty after table loading. The metadata cache could be out of dated after loading new data into the table, because loading/inserting does not update the cache. So far, the metadata cache is only used for data source tables. Thus, for Hive serde tables, only `parquet` and `orc` formats are facing such issues, because the Hive serde tables in the format of parquet/orc could be converted to data source tables when `spark.sql.hive.convertMetastoreParquet`/`spark.sql.hive.convertMetastoreOrc` is on. This PR is to refresh the metadata cache after processing the `LOAD DATA` command. In addition, Spark SQL does not convert **partitioned** Hive tables (orc/parquet) to data source tables in the write path, but the read path is using the metadata cache for both **partitioned** and non-partitioned Hive tables (orc/parquet). That means, writing the partitioned parquet/orc tables still use `InsertIntoHiveTable`, instead of `InsertIntoHadoopFsRelationCommand`. To avoid reading the out-of-dated cache, `InsertIntoHiveTable` needs to refresh the metadata cache for partitioned tables. Note, it does not need to refresh the cache for non-partitioned parquet/orc tables, because it does not call `InsertIntoHiveTable` at all. Based on the comments, this PR will keep the existing logics unchanged. That means, we always refresh the table no matter whether the table is partitioned or not. ### How was this patch tested? Added test cases in parquetSuites.scala Author: gatorsmile <gatorsmile@gmail.com> Closes apache#16500 from gatorsmile/refreshInsertIntoHiveTable.

gatorsmile added 4 commits January 7, 2017 10:59

fix.

ea91cb0

fix.

b7013c2

fix.

27fab56

more test cases.

0f70e91

gatorsmile commented Jan 8, 2017

View reviewed changes

gatorsmile changed the title ~~[SPARK-19120] [SPARK-19121] Refresh Metadata Cache After Load Partitioned Hive Tables~~ [SPARK-19120] [SPARK-19121] Refresh Metadata Cache After Loading Hive Tables Jan 8, 2017

cloud-fan reviewed Jan 12, 2017

View reviewed changes

rename

46d41eb

gatorsmile changed the title ~~[SPARK-19120] [SPARK-19121] Refresh Metadata Cache After Loading Hive Tables~~ [SPARK-19120] Refresh Metadata Cache After Loading Hive Tables Jan 13, 2017

gatorsmile added 2 commits January 12, 2017 22:42

address comments.

11507cc

Merge remote-tracking branch 'upstream/master' into refreshInsertInto…

203e36c

…HiveTable

Merge remote-tracking branch 'upstream/master' into refreshInsertInto…

d2d751b

…HiveTable

gatorsmile force-pushed the refreshInsertIntoHiveTable branch from c27a9af to d2d751b Compare January 14, 2017 00:29

cloud-fan reviewed Jan 14, 2017

View reviewed changes

address comments.

14da2b6

asfgit closed this in de62ddf Jan 15, 2017

[SPARK-19120] Refresh Metadata Cache After Loading Hive Tables #16500

[SPARK-19120] Refresh Metadata Cache After Loading Hive Tables #16500

Uh oh!

Conversation

gatorsmile commented Jan 8, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

gatorsmile commented Jan 8, 2017

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!

cloud-fan Jan 12, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gatorsmile Jan 13, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

SparkQA commented Jan 8, 2017

Uh oh!

cloud-fan commented Jan 10, 2017

Uh oh!

ericl commented Jan 12, 2017

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!

gatorsmile commented Jan 13, 2017

Uh oh!

SparkQA commented Jan 13, 2017

Uh oh!

SparkQA commented Jan 13, 2017

Uh oh!

cloud-fan commented Jan 13, 2017

Uh oh!

SparkQA commented Jan 13, 2017

Uh oh!

gatorsmile commented Jan 14, 2017

Uh oh!

SparkQA commented Jan 14, 2017

Uh oh!

SparkQA commented Jan 14, 2017

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!

SparkQA commented Jan 14, 2017

Uh oh!

cloud-fan commented Jan 15, 2017

Uh oh!

Uh oh!

gatorsmile commented Jan 8, 2017 •

edited

Loading

cloud-fan Jan 12, 2017 •

edited

Loading

gatorsmile Jan 13, 2017 •

edited

Loading