Skip to content

[SPARK-19723][SQL]create datasource table with an non-existent location should work #17055

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
windpiger wants to merge 15 commits into from
Closed

[SPARK-19723][SQL]create datasource table with an non-existent location should work #17055

Show file tree
Hide file tree
Changes from 13 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
89eb03a
[SPARK-19723][SQL]create datasource table with an non-existent locati…
windpiger Feb 24, 2017
0e09b92
merge with master
windpiger Mar 2, 2017
d4378dd
optimize some code
windpiger Mar 6, 2017
4c82dfb
fix code style
windpiger Mar 6, 2017
3ea3e9d
fix code style
windpiger Mar 6, 2017
b26cce1
Merge branch 'master' of github.com:apache/spark into CTDataSourcePat…
windpiger Mar 6, 2017
6f7045c
Merge branch 'master' into CTDataSourcePathNotExists
windpiger Mar 9, 2017
d53f7ab
Merge branch 'master' into CTDataSourcePathNotExists
windpiger Mar 9, 2017
3ae5b2a
merge with master
windpiger Mar 9, 2017
bc7f08d
fix a code
windpiger Mar 9, 2017
5699059
fix code style
windpiger Mar 9, 2017
207448d
fix code style
windpiger Mar 9, 2017
4024d7c
fix code style
windpiger Mar 9, 2017
e1c891e
fix some code review
windpiger Mar 11, 2017
ada275c
fix some code review
windpiger Mar 11, 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
3 changes: 2 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/command/createDataSourceTables.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,8 @@ case class CreateDataSourceTableCommand(table: CatalogTable, ignoreIfExists: Boo
className = table.provider.get,
bucketSpec = table.bucketSpec,
options = table.storage.properties ++ pathOption,
catalogTable = Some(tableWithDefaultOptions)).resolveRelation()
// As discussed in SPARK-19583, we don't check if the location is existed
catalogTable = Some(tableWithDefaultOptions)).resolveRelation(checkFilesExist = false)

val partitionColumnNames = if (table.schema.nonEmpty) {
table.partitionColumnNames
Expand Down
113 changes: 72 additions & 41 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -230,7 +230,7 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
}

private def getDBPath(dbName: String): URI = {
val warehousePath = makeQualifiedPath(s"${spark.sessionState.conf.warehousePath}")
val warehousePath = makeQualifiedPath(spark.sessionState.conf.warehousePath)
new Path(CatalogUtils.URIToString(warehousePath), s"$dbName.db").toUri
}

Expand Down Expand Up @@ -1899,7 +1899,7 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
}
}

test("insert data to a data source table which has a not existed location should succeed") {
test("insert data to a data source table which has a non-existing location should succeed") {
withTable("t") {
withTempDir { dir =>
spark.sql(
Expand Down Expand Up @@ -1939,7 +1939,7 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
}
}

test("insert into a data source table with no existed partition location should succeed") {
test("insert into a data source table with a non-existing partition location should succeed") {
withTable("t") {
withTempDir { dir =>
spark.sql(
Expand All @@ -1966,7 +1966,7 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
}
}

test("read data from a data source table which has a not existed location should succeed") {
test("read data from a data source table which has a non-existing location should succeed") {
withTable("t") {
withTempDir { dir =>
spark.sql(
Expand Down Expand Up @@ -1994,7 +1994,7 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
}
}

test("read data from a data source table with no existed partition location should succeed") {
test("read data from a data source table with non-existing partition location should succeed") {
withTable("t") {
withTempDir { dir =>
spark.sql(
Expand All @@ -2016,48 +2016,79 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
}
}

test("create datasource table with a non-existing location") {
withTable("t", "t1") {
withTempPath { dir =>
spark.sql(s"CREATE TABLE t(a int, b int) USING parquet LOCATION '$dir'")

val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))

spark.sql("INSERT INTO TABLE t SELECT 1, 2")
assert(dir.exists())

checkAnswer(spark.table("t"), Row(1, 2))
}
// partition table
withTempPath { dir =>
spark.sql(
s"""
|CREATE TABLE t1(a int, b int) USING parquet PARTITIONED BY(a) LOCATION '$dir'
""".stripMargin)
Copy link
Member

@gatorsmile gatorsmile Mar 10, 2017

Choose a reason for hiding this comment

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

Nit: one line?


val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t1"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))

spark.sql("INSERT INTO TABLE t1 PARTITION(a=1) SELECT 2")

val partDir = new File(dir, "a=1")
assert(partDir.exists())

checkAnswer(spark.table("t1"), Row(2, 1))
}
}
}

Seq(true, false).foreach { shouldDelete =>
val tcName = if (shouldDelete) "non-existent" else "existed"
val tcName = if (shouldDelete) "non-existing" else "existed"
test(s"CTAS for external data source table with a $tcName location") {
withTable("t", "t1") {
withTempDir {
dir =>
if (shouldDelete) {
dir.delete()
}
spark.sql(
s"""
|CREATE TABLE t
|USING parquet
|LOCATION '$dir'
|AS SELECT 3 as a, 4 as b, 1 as c, 2 as d
""".stripMargin)
val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))
withTempDir { dir =>
if (shouldDelete) {
dir.delete()
}
Copy link
Member

@gatorsmile gatorsmile Mar 10, 2017

Choose a reason for hiding this comment

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

if (shouldDelete) dir.delete()

Copy link
Contributor Author

@windpiger windpiger Mar 11, 2017

Choose a reason for hiding this comment

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

ok thanks~~

spark.sql(
s"""
|CREATE TABLE t
|USING parquet
|LOCATION '$dir'
|AS SELECT 3 as a, 4 as b, 1 as c, 2 as d
""".stripMargin)
Copy link
Member

@gatorsmile gatorsmile Mar 6, 2017

Choose a reason for hiding this comment

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

The test case is already pretty long. Please reduce the sql statement to a single line.

val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))

checkAnswer(spark.table("t"), Row(3, 4, 1, 2))
checkAnswer(spark.table("t"), Row(3, 4, 1, 2))
}
// partition table
withTempDir {
dir =>
if (shouldDelete) {
dir.delete()
}
spark.sql(
s"""
|CREATE TABLE t1
|USING parquet
|PARTITIONED BY(a, b)
|LOCATION '$dir'
|AS SELECT 3 as a, 4 as b, 1 as c, 2 as d
""".stripMargin)
val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t1"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))

val partDir = new File(dir, "a=3")
assert(partDir.exists())

checkAnswer(spark.table("t1"), Row(1, 2, 3, 4))
withTempDir { dir =>
if (shouldDelete) {
dir.delete()
}
spark.sql(
s"""
|CREATE TABLE t1
|USING parquet
|PARTITIONED BY(a, b)
|LOCATION '$dir'
|AS SELECT 3 as a, 4 as b, 1 as c, 2 as d
""".stripMargin)
val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t1"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))

val partDir = new File(dir, "a=3")
assert(partDir.exists())

checkAnswer(spark.table("t1"), Row(1, 2, 3, 4))
}
}
}
Expand Down
144 changes: 88 additions & 56 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1663,95 +1663,127 @@ class HiveDDLSuite
}
}

test("create hive table with a non-existing location") {
withTable("t", "t1") {
withTempPath { dir =>
spark.sql(s"CREATE TABLE t(a int, b int) USING hive LOCATION '$dir'")

val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))

spark.sql("INSERT INTO TABLE t SELECT 1, 2")
assert(dir.exists())

checkAnswer(spark.table("t"), Row(1, 2))
}
// partition table
withTempPath { dir =>
spark.sql(
s"""
|CREATE TABLE t1(a int, b int)
|USING hive
|PARTITIONED BY(a)
|LOCATION '$dir'
""".stripMargin)

val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t1"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))

spark.sql("INSERT INTO TABLE t1 PARTITION(a=1) SELECT 2")

val partDir = new File(dir, "a=1")
assert(partDir.exists())

checkAnswer(spark.table("t1"), Row(2, 1))
}
}
}

Seq(true, false).foreach { shouldDelete =>
val tcName = if (shouldDelete) "non-existent" else "existed"
val tcName = if (shouldDelete) "non-existing" else "existed"
test(s"CTAS for external data source table with a $tcName location") {
Copy link
Contributor

@cloud-fan cloud-fan Mar 10, 2017

Choose a reason for hiding this comment

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

now it's duplicated with https://github.com/apache/spark/pull/17055/files#diff-2d45592564aa2eea4e85ffd000d8a14eR2054

Copy link
Contributor Author

@windpiger windpiger Mar 11, 2017 •
edited
Loading

Choose a reason for hiding this comment

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

oh, it is. after we also qualified the location path for datasource table , the code for equal two location are the same with HiveExternalCatalog. thanks~

withTable("t", "t1") {
withTempDir {
dir =>
withTempDir { dir =>
if (shouldDelete) {
dir.delete()
}
spark.sql(
s"""
|CREATE TABLE t
|USING parquet
|LOCATION '$dir'
|AS SELECT 3 as a, 4 as b, 1 as c, 2 as d
""".stripMargin)

val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))

checkAnswer(spark.table("t"), Row(3, 4, 1, 2))
}
// partition table
withTempDir { dir =>
if (shouldDelete) {
dir.delete()
}
spark.sql(
s"""
|CREATE TABLE t1
|USING parquet
|PARTITIONED BY(a, b)
|LOCATION '$dir'
|AS SELECT 3 as a, 4 as b, 1 as c, 2 as d
""".stripMargin)

val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t1"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))

val partDir = new File(dir, "a=3")
assert(partDir.exists())

checkAnswer(spark.table("t1"), Row(1, 2, 3, 4))
}
}
}

test(s"CTAS for external hive table with a $tcName location") {
withTable("t", "t1") {
withSQLConf("hive.exec.dynamic.partition.mode" -> "nonstrict") {
withTempDir { dir =>
if (shouldDelete) {
dir.delete()
}
spark.sql(
s"""
|CREATE TABLE t
|USING parquet
|USING hive
|LOCATION '$dir'
|AS SELECT 3 as a, 4 as b, 1 as c, 2 as d
""".stripMargin)

val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))

checkAnswer(spark.table("t"), Row(3, 4, 1, 2))
}
// partition table
withTempDir {
dir =>
}
// partition table
withTempDir { dir =>
if (shouldDelete) {
dir.delete()
}
spark.sql(
s"""
|CREATE TABLE t1
|USING parquet
|USING hive
|PARTITIONED BY(a, b)
|LOCATION '$dir'
|AS SELECT 3 as a, 4 as b, 1 as c, 2 as d
""".stripMargin)

val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t1"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))

val partDir = new File(dir, "a=3")
assert(partDir.exists())

checkAnswer(spark.table("t1"), Row(1, 2, 3, 4))
}
}
}

test(s"CTAS for external hive table with a $tcName location") {
withTable("t", "t1") {
withSQLConf("hive.exec.dynamic.partition.mode" -> "nonstrict") {
withTempDir {
dir =>
if (shouldDelete) {
dir.delete()
}
spark.sql(
s"""
|CREATE TABLE t
|USING hive
|LOCATION '$dir'
|AS SELECT 3 as a, 4 as b, 1 as c, 2 as d
""".stripMargin)
val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))

checkAnswer(spark.table("t"), Row(3, 4, 1, 2))
}
// partition table
withTempDir {
dir =>
if (shouldDelete) {
dir.delete()
}
spark.sql(
s"""
|CREATE TABLE t1
|USING hive
|PARTITIONED BY(a, b)
|LOCATION '$dir'
|AS SELECT 3 as a, 4 as b, 1 as c, 2 as d
""".stripMargin)
val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t1"))
assert(table.location == makeQualifiedPath(dir.getAbsolutePath))

val partDir = new File(dir, "a=3")
assert(partDir.exists())

checkAnswer(spark.table("t1"), Row(1, 2, 3, 4))
}
}
}
Expand Down