boneanxs commented on code in PR #8076:
URL: https://github.com/apache/hudi/pull/8076#discussion_r1201417934
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestInsertTable.scala:
##########
@@ -599,138 +582,250 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
}
}
- test("Test bulk insert") {
+ test("Test bulk insert with insert into for single partitioned table") {
withSQLConf("hoodie.sql.insert.mode" -> "non-strict") {
withRecordType()(withTempDir { tmp =>
Seq("cow", "mor").foreach {tableType =>
- // Test bulk insert for single partition
- val tableName = generateTableName
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | dt string
- |) using hudi
- | tblproperties (
- | type = '$tableType',
- | primaryKey = 'id'
- | )
- | partitioned by (dt)
- | location '${tmp.getCanonicalPath}/$tableName'
- """.stripMargin)
- spark.sql("set hoodie.datasource.write.insert.drop.duplicates =
false")
+ withTable(generateTableName) { tableName =>
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | dt string
+ |) using hudi
+ | tblproperties (
+ | type = '$tableType',
+ | primaryKey = 'id'
+ | )
+ | partitioned by (dt)
+ | location '${tmp.getCanonicalPath}/$tableName'
+ """.stripMargin)
+ spark.sql("set hoodie.datasource.write.insert.drop.duplicates =
false")
- // Enable the bulk insert
- spark.sql("set hoodie.sql.bulk.insert.enable = true")
- spark.sql(s"insert into $tableName values(1, 'a1', 10,
'2021-07-18')")
+ // Enable the bulk insert
+ spark.sql("set hoodie.sql.bulk.insert.enable = true")
+ spark.sql(s"insert into $tableName values(1, 'a1', 10,
'2021-07-18')")
- assertResult(WriteOperationType.BULK_INSERT) {
- getLastCommitMetadata(spark,
s"${tmp.getCanonicalPath}/$tableName").getOperationType
+ assertResult(WriteOperationType.BULK_INSERT) {
+ getLastCommitMetadata(spark,
s"${tmp.getCanonicalPath}/$tableName").getOperationType
+ }
+ checkAnswer(s"select id, name, price, dt from $tableName")(
+ Seq(1, "a1", 10.0, "2021-07-18")
+ )
+
+ // Disable the bulk insert
+ spark.sql("set hoodie.sql.bulk.insert.enable = false")
+ spark.sql(s"insert into $tableName values(2, 'a2', 10,
'2021-07-18')")
+
+ assertResult(WriteOperationType.INSERT) {
+ getLastCommitMetadata(spark,
s"${tmp.getCanonicalPath}/$tableName").getOperationType
+ }
+ checkAnswer(s"select id, name, price, dt from $tableName order by
id")(
+ Seq(1, "a1", 10.0, "2021-07-18"),
+ Seq(2, "a2", 10.0, "2021-07-18")
+ )
}
- checkAnswer(s"select id, name, price, dt from $tableName")(
- Seq(1, "a1", 10.0, "2021-07-18")
- )
+ }
+ })
+ }
+ }
- // Disable the bulk insert
- spark.sql("set hoodie.sql.bulk.insert.enable = false")
- spark.sql(s"insert into $tableName values(2, 'a2', 10,
'2021-07-18')")
+ test("Test bulk insert with insert into for multi partitioned table") {
+ withSQLConf("hoodie.sql.insert.mode" -> "non-strict") {
+ withRecordType()(withTempDir { tmp =>
+ Seq("cow", "mor").foreach { tableType =>
+ withTable(generateTableName) { tableMultiPartition =>
+ spark.sql(
+ s"""
+ |create table $tableMultiPartition (
+ | id int,
+ | name string,
+ | price double,
+ | dt string,
+ | hh string
+ |) using hudi
+ | tblproperties (
+ | type = '$tableType',
+ | primaryKey = 'id'
+ | )
+ | partitioned by (dt, hh)
+ | location '${tmp.getCanonicalPath}/$tableMultiPartition'
+ """.stripMargin)
- assertResult(WriteOperationType.INSERT) {
- getLastCommitMetadata(spark,
s"${tmp.getCanonicalPath}/$tableName").getOperationType
+ // Enable the bulk insert
+ spark.sql("set hoodie.sql.bulk.insert.enable = true")
+ spark.sql(s"insert into $tableMultiPartition values(1, 'a1', 10,
'2021-07-18', '12')")
+
+ checkAnswer(s"select id, name, price, dt, hh from
$tableMultiPartition")(
+ Seq(1, "a1", 10.0, "2021-07-18", "12")
+ )
+ // Disable the bulk insert
+ spark.sql("set hoodie.sql.bulk.insert.enable = false")
+ spark.sql(s"insert into $tableMultiPartition " +
+ s"values(2, 'a2', 10, '2021-07-18','12')")
+
+ checkAnswer(s"select id, name, price, dt, hh from
$tableMultiPartition order by id")(
+ Seq(1, "a1", 10.0, "2021-07-18", "12"),
+ Seq(2, "a2", 10.0, "2021-07-18", "12")
+ )
}
- checkAnswer(s"select id, name, price, dt from $tableName order by
id")(
- Seq(1, "a1", 10.0, "2021-07-18"),
- Seq(2, "a2", 10.0, "2021-07-18")
- )
+ }
+ })
+ }
+ }
- // Test bulk insert for multi-level partition
- val tableMultiPartition = generateTableName
- spark.sql(
- s"""
- |create table $tableMultiPartition (
- | id int,
- | name string,
- | price double,
- | dt string,
- | hh string
- |) using hudi
- | tblproperties (
- | type = '$tableType',
- | primaryKey = 'id'
- | )
- | partitioned by (dt, hh)
- | location '${tmp.getCanonicalPath}/$tableMultiPartition'
- """.stripMargin)
+ test("Test bulk insert with insert into for non partitioned table") {
Review Comment:
`Test bulk insert with insert overwrite table` test `INSERT_OVERWRITE_TABLE`,
`Test bulk insert with insert overwrite partition` test `INSERT_OVERWRITE`
These two tests test all values of `BULKINSERT_OVERWRITE_OPERATION_TYPE`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@hudi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
