codope commented on code in PR #10272:
URL: https://github.com/apache/hudi/pull/10272#discussion_r1421929085
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/HoodieDatasetBulkInsertHelper.scala:
##########
@@ -186,7 +179,9 @@ object HoodieDatasetBulkInsertHelper
}
try {
- iter.foreach(writer.write)
+ iter.foreach(row => {
+ writer.write(row)
+ })
Review Comment:
how about reverting this change?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala:
##########
@@ -365,6 +365,40 @@ class TestHoodieSparkSqlWriter {
testBulkInsertWithSortMode(BulkInsertSortMode.NONE, populateMetaFields)
}
+@Test
+def testBulkInsertForDropPartitionColumn(): Unit = {
+ //create a new table
+ val tableName = "trips_table"
+ val basePath = "file:///tmp/trips_table"
+ val columns = Seq("ts", "uuid", "rider", "driver", "fare", "city")
+ val data =
+ Seq((1695159649087L, "334e26e9-8355-45cc-97c6-c31daf0df330", "rider-A",
"driver-K", 19.10, "san_francisco"),
+ (1695091554788L, "e96c4396-3fad-413a-a942-4cb36106d721", "rider-C",
"driver-M", 27.70, "san_francisco"),
+ (1695046462179L, "9909a8b1-2d15-4d3d-8ec9-efc48c536a00", "rider-D",
"driver-L", 33.90, "san_francisco"),
+ (1695516137016L, "e3cf430c-889d-4015-bc98-59bdce1e530c", "rider-F",
"driver-P", 34.15, "sao_paulo"),
+ (1695115999911L, "c8abbe79-8d89-47ea-b4ce-4d224bae5bfa", "rider-J",
"driver-T", 17.85, "chennai"));
+
+ var inserts = spark.createDataFrame(data).toDF(columns: _*)
+ inserts.write.format("hudi").
+ option(DataSourceWriteOptions.PARTITIONPATH_FIELD.key(), "city").
+ option(HoodieWriteConfig.TABLE_NAME, tableName).
+ option("hoodie.datasource.write.recordkey.field", "uuid").
+ option("hoodie.datasource.write.precombine.field", "rider").
+ option("hoodie.datasource.write.operation", "bulk_insert").
+ option("hoodie.datasource.write.hive_style_partitioning", "true").
+ option("hoodie.populate.meta.fields", "false").
+ option("hoodie.datasource.write.drop.partition.columns", "true").
+ mode(SaveMode.Overwrite).
+ save(basePath)
+
+ // Ensure the partition column (i.e 'city') can be read back
+ val tripsDF = spark.read.format("hudi").load(basePath)
+ tripsDF.createOrReplaceTempView("trips_table")
+ spark.sql("SELECT uuid, fare, ts, rider, driver, city FROM
trips_table").show()
+
+ // TODO: Peek into the raw parquet file and ensure city column is not
written to the file
Review Comment:
Could you also add test for the TODO comment here?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
