You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2021/01/10 05:06:39 UTC
[spark] branch master updated: [SPARK-34055][SQL] Refresh cache in
`ALTER TABLE .. ADD PARTITION`
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new e0e06c1 [SPARK-34055][SQL] Refresh cache in `ALTER TABLE .. ADD PARTITION`
e0e06c1 is described below
commit e0e06c18fd479ac86963ec26ab82b0dc5536bc83
Author: Max Gekk <ma...@gmail.com>
AuthorDate: Sun Jan 10 14:06:17 2021 +0900
[SPARK-34055][SQL] Refresh cache in `ALTER TABLE .. ADD PARTITION`
### What changes were proposed in this pull request?
Invoke `refreshTable()` from `CatalogImpl` which refreshes the cache in v1 `ALTER TABLE .. ADD PARTITION`.
### Why are the changes needed?
This fixes the issues portrayed by the example:
```sql
spark-sql> create table tbl (col int, part int) using parquet partitioned by (part);
spark-sql> insert into tbl partition (part=0) select 0;
spark-sql> cache table tbl;
spark-sql> select * from tbl;
0 0
spark-sql> show table extended like 'tbl' partition(part=0);
default tbl false Partition Values: [part=0]
Location: file:/Users/maximgekk/proj/add-partition-refresh-cache-2/spark-warehouse/tbl/part=0
...
```
Create new partition by copying the existing one:
```
$ cp -r /Users/maximgekk/proj/add-partition-refresh-cache-2/spark-warehouse/tbl/part=0 /Users/maximgekk/proj/add-partition-refresh-cache-2/spark-warehouse/tbl/part=1
```
```sql
spark-sql> alter table tbl add partition (part=1) location '/Users/maximgekk/proj/add-partition-refresh-cache-2/spark-warehouse/tbl/part=1';
spark-sql> select * from tbl;
0 0
```
The last query must return `0 1` since it has been added by `ALTER TABLE .. ADD PARTITION`.
### Does this PR introduce _any_ user-facing change?
Yes. After the changes for the example above:
```sql
...
spark-sql> alter table tbl add partition (part=1) location '/Users/maximgekk/proj/add-partition-refresh-cache-2/spark-warehouse/tbl/part=1';
spark-sql> select * from tbl;
0 0
0 1
```
### How was this patch tested?
By running the affected test suite:
```
$ build/sbt -Phive-2.3 -Phive-thriftserver "test:testOnly *AlterTableAddPartitionSuite"
```
Closes #31101 from MaxGekk/add-partition-refresh-cache-2.
Lead-authored-by: Max Gekk <ma...@gmail.com>
Co-authored-by: Hyukjin Kwon <gu...@gmail.com>
Signed-off-by: HyukjinKwon <gu...@apache.org>
---
.../apache/spark/sql/execution/command/ddl.scala | 1 +
.../command/v1/AlterTableAddPartitionSuite.scala | 33 +++++++++++++++++++++-
2 files changed, 33 insertions(+), 1 deletion(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
index 8195d02..4545d73 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
@@ -485,6 +485,7 @@ case class AlterTableAddPartitionCommand(
catalog.createPartitions(table.identifier, batch, ignoreIfExists = ifNotExists)
}
+ sparkSession.catalog.refreshTable(table.identifier.quotedString)
if (table.stats.nonEmpty) {
if (sparkSession.sessionState.conf.autoSizeUpdateEnabled) {
val addedSize = CommandUtils.calculateMultipleLocationSizes(sparkSession, table.identifier,
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableAddPartitionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableAddPartitionSuite.scala
index b3c118d..bc79919 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableAddPartitionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableAddPartitionSuite.scala
@@ -17,7 +17,11 @@
package org.apache.spark.sql.execution.command.v1
-import org.apache.spark.sql.AnalysisException
+import java.io.File
+
+import org.apache.commons.io.FileUtils
+
+import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.execution.command
/**
@@ -41,6 +45,33 @@ trait AlterTableAddPartitionSuiteBase extends command.AlterTableAddPartitionSuit
"The spec ([p1=]) contains an empty partition column value"))
}
}
+
+ test("SPARK-34055: refresh cache in partition adding") {
+ withTable("t") {
+ sql("CREATE TABLE t (id int, part int) USING parquet PARTITIONED BY (part)")
+ sql("INSERT INTO t PARTITION (part=0) SELECT 0")
+ assert(!spark.catalog.isCached("t"))
+ sql("CACHE TABLE t")
+ assert(spark.catalog.isCached("t"))
+ checkAnswer(sql("SELECT * FROM t"), Seq(Row(0, 0)))
+
+ // Create new partition (part = 1) in the filesystem
+ val information = sql("SHOW TABLE EXTENDED LIKE 't' PARTITION (part = 0)")
+ .select("information")
+ .first().getString(0)
+ val part0Loc = information
+ .split("\\r?\\n")
+ .filter(_.startsWith("Location:"))
+ .head
+ .replace("Location: file:", "")
+ val part1Loc = part0Loc.replace("part=0", "part=1")
+ FileUtils.copyDirectory(new File(part0Loc), new File(part1Loc))
+
+ sql(s"ALTER TABLE t ADD PARTITION (part=1) LOCATION '$part1Loc'")
+ assert(spark.catalog.isCached("t"))
+ checkAnswer(sql("SELECT * FROM t"), Seq(Row(0, 0), Row(0, 1)))
+ }
+ }
}
/**
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org