Skip to content

[SPARK-14954][SQL] Add PARTITIONED BY and CLUSTERED BY clause for data source CTAS syntax #12734

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
liancheng wants to merge 5 commits into from
Closed

[SPARK-14954][SQL] Add PARTITIONED BY and CLUSTERED BY clause for data source CTAS syntax #12734

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f51300c
Add PARTITION BY and BUCKET BY clause for "CREATE TABLE ... USING ...…
liancheng Apr 27, 2016
af973d6
Moves test case to MetastoreDataSourcesSuite
liancheng Apr 27, 2016
a193faf
Reverts unnecessary parser changes
liancheng Apr 27, 2016
3df4d47
Merge remote-tracking branch 'upstream/master' into spark-14954
yhuai Apr 27, 2016
442265e
Update tests
yhuai Apr 27, 2016
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
4 changes: 3 additions & 1 deletion sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,9 @@ statement
| createTableHeader ('(' colTypeList ')')? tableProvider
(OPTIONS tablePropertyList)? #createTableUsing
| createTableHeader tableProvider
(OPTIONS tablePropertyList)? AS? query #createTableUsing
(OPTIONS tablePropertyList)?
(PARTITIONED BY partitionColumnNames=identifierList)?
bucketSpec? AS? query #createTableUsing
| createTableHeader ('(' columns=colTypeList ')')?
(COMMENT STRING)?
(PARTITIONED BY '(' partitionColumns=colTypeList ')')?
Expand Down
12 changes: 10 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -289,6 +289,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
}
val options = Option(ctx.tablePropertyList).map(visitTablePropertyList).getOrElse(Map.empty)
val provider = ctx.tableProvider.qualifiedName.getText
val bucketSpec = Option(ctx.bucketSpec()).map(visitBucketSpec)

if (ctx.query != null) {
// Get the backing query.
Expand All @@ -302,9 +303,16 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
} else {
SaveMode.ErrorIfExists
}
CreateTableUsingAsSelect(table, provider, temp, Array.empty, None, mode, options, query)

val partitionColumnNames =
Option(ctx.partitionColumnNames)
.map(visitIdentifierList(_).toArray)
.getOrElse(Array.empty[String])

CreateTableUsingAsSelect(
table, provider, temp, partitionColumnNames, bucketSpec, mode, options, query)
} else {
val struct = Option(ctx.colTypeList).map(createStructType)
val struct = Option(ctx.colTypeList()).map(createStructType)
Copy link
Contributor

@yhuai yhuai Apr 27, 2016 •
edited
Loading

Choose a reason for hiding this comment

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

If the command is not CTAS statement, seems we should throw exceptions if users define any of PARTITIONED BY, SORTED BY, or CLUSTERED BY clause?

Copy link
Contributor

@yhuai yhuai Apr 27, 2016

Choose a reason for hiding this comment

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

One thing that is not very related to this pr. I always find that the keyword CLUSTERED BY is very confusing, because there is a CLUSTER BY keyword (, which is DISTRIBUTE BY + SORT BY). But, we do not need to change it right now.

Copy link
Contributor

@yhuai yhuai Apr 27, 2016

Choose a reason for hiding this comment

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

I am going to add the check for this else branch and add some tests.

Copy link
Contributor

@yhuai yhuai Apr 27, 2016

Choose a reason for hiding this comment

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

oh, sorry. PARTITIONED BY and CLUSTERED BY are both associated with CREATE TABLE USING AS SELECT rule. So, for CREATE TABLE USING, if PARTITIONED BY or CLUSTERED PY is provided, we already throw an exception.

CreateTableUsing(table, struct, provider, temp, options, ifNotExists, managedIfNoPath = false)
}
}
Expand Down
93 changes: 93 additions & 0 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -940,4 +940,97 @@ class MetastoreDataSourcesSuite extends QueryTest with SQLTestUtils with TestHiv
.schema.forall { c => DataTypeParser.parse(c.dataType) == ArrayType(StringType) })
}
}

test("CTAS: persisted partitioned data source table") {
withTempDir { dir =>
withTable("t") {
val path = dir.getCanonicalPath

sql(
s"""CREATE TABLE t USING PARQUET
|OPTIONS (PATH '$path')
|PARTITIONED BY (a)
|AS SELECT 1 AS a, 2 AS b
""".stripMargin
)

val metastoreTable = sharedState.externalCatalog.getTable("default", "t")
assert(metastoreTable.properties("spark.sql.sources.schema.numPartCols").toInt === 1)
assert(!metastoreTable.properties.contains("spark.sql.sources.schema.numBuckets"))
assert(!metastoreTable.properties.contains("spark.sql.sources.schema.numBucketCols"))
assert(!metastoreTable.properties.contains("spark.sql.sources.schema.numSortCols"))

checkAnswer(table("t"), Row(2, 1))
}
}
}

test("CTAS: persisted bucketed data source table") {
withTempDir { dir =>
withTable("t") {
val path = dir.getCanonicalPath

sql(
s"""CREATE TABLE t USING PARQUET
|OPTIONS (PATH '$path')
|CLUSTERED BY (a) SORTED BY (b) INTO 2 BUCKETS
|AS SELECT 1 AS a, 2 AS b
""".stripMargin
)

val metastoreTable = sharedState.externalCatalog.getTable("default", "t")
assert(!metastoreTable.properties.contains("spark.sql.sources.schema.numPartCols"))
assert(metastoreTable.properties("spark.sql.sources.schema.numBuckets").toInt === 2)
assert(metastoreTable.properties("spark.sql.sources.schema.numBucketCols").toInt === 1)
assert(metastoreTable.properties("spark.sql.sources.schema.numSortCols").toInt === 1)

checkAnswer(table("t"), Row(1, 2))
}

withTable("t") {
val path = dir.getCanonicalPath

sql(
s"""CREATE TABLE t USING PARQUET
|OPTIONS (PATH '$path')
|CLUSTERED BY (a) INTO 2 BUCKETS
|AS SELECT 1 AS a, 2 AS b
""".stripMargin
)

val metastoreTable = sharedState.externalCatalog.getTable("default", "t")
assert(!metastoreTable.properties.contains("spark.sql.sources.schema.numPartCols"))
assert(metastoreTable.properties("spark.sql.sources.schema.numBuckets").toInt === 2)
assert(metastoreTable.properties("spark.sql.sources.schema.numBucketCols").toInt === 1)
assert(!metastoreTable.properties.contains("spark.sql.sources.schema.numSortCols"))

checkAnswer(table("t"), Row(1, 2))
}
}
}

test("CTAS: persisted partitioned bucketed data source table") {
withTempDir { dir =>
withTable("t") {
val path = dir.getCanonicalPath

sql(
s"""CREATE TABLE t USING PARQUET
|OPTIONS (PATH '$path')
|PARTITIONED BY (a)
|CLUSTERED BY (b) SORTED BY (c) INTO 2 BUCKETS
|AS SELECT 1 AS a, 2 AS b, 3 AS c
""".stripMargin
)

val metastoreTable = sharedState.externalCatalog.getTable("default", "t")
assert(metastoreTable.properties("spark.sql.sources.schema.numPartCols").toInt === 1)
assert(metastoreTable.properties("spark.sql.sources.schema.numBuckets").toInt === 2)
assert(metastoreTable.properties("spark.sql.sources.schema.numBucketCols").toInt === 1)
assert(metastoreTable.properties("spark.sql.sources.schema.numSortCols").toInt === 1)

checkAnswer(table("t"), Row(2, 3, 1))
}
}
}
}