You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/01/18 19:24:36 UTC
[PR] feat: add complex type support to native Parquet writer [datafusion-comet]
andygrove opened a new pull request, #3214:
URL: https://github.com/apache/datafusion-comet/pull/3214
## Summary
- Enables support for complex types (arrays, maps, structs) in Comet's native Parquet writer
- Removes the blocking check that previously prevented complex types
- Adds comprehensive test coverage for complex types
## Changes
- Remove complex type blocking check in `CometDataWritingCommand.scala`
- Add 12 new tests for complex types in `CometParquetWriterSuite.scala`:
- Basic complex types (array, struct, map)
- Nested complex types (array of structs, struct containing array, map with struct values, deeply nested)
- Nullable complex types with nulls at various nesting levels
- Complex types containing decimal and temporal types
- Empty arrays and maps
- Fuzz testing with randomly generated complex type schemas
- Update documentation to reflect complex type support
## Test plan
- [x] All 15 tests in `CometParquetWriterSuite` pass
- [x] Tests verify round-trip compatibility (write with Comet, read with Spark/Comet)
- [x] Fuzz testing with randomly generated schemas
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: add complex type support to native Parquet writer [datafusion-comet]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #3214:
URL: https://github.com/apache/datafusion-comet/pull/3214#issuecomment-3765671013
## [Codecov](https://app.codecov.io/gh/apache/datafusion-comet/pull/3214?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 9.88%. Comparing base ([`f09f8af`](https://app.codecov.io/gh/apache/datafusion-comet/commit/f09f8af64c6599255e116a376f4f008f2fd63b43?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`12a76fd`](https://app.codecov.io/gh/apache/datafusion-comet/commit/12a76fdf9605da12c27f116eaa0822e14fa1a2bb?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 855 commits behind head on main.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #3214 +/- ##
============================================
- Coverage 56.12% 9.88% -46.25%
+ Complexity 976 234 -742
============================================
Files 119 170 +51
Lines 11743 15686 +3943
Branches 2251 2569 +318
============================================
- Hits 6591 1550 -5041
- Misses 4012 13959 +9947
+ Partials 1140 177 -963
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/datafusion-comet/pull/3214?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: add complex type support to native Parquet writer [datafusion-comet]
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #3214:
URL: https://github.com/apache/datafusion-comet/pull/3214#discussion_r2704352049
##########
spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala:
##########
@@ -228,4 +230,299 @@ class CometParquetWriterSuite extends CometTestBase {
}
}
}
+
+ // ===== Complex Type Tests =====
+
+ private def writeComplexTypeData(
+ inputDf: DataFrame,
+ outputPath: String,
+ expectedRows: Int): Unit = {
+ withTempPath { inputDir =>
+ val inputPath = new File(inputDir, "input.parquet").getAbsolutePath
+
+ // First write the input data without Comet to get proper Arrow arrays when reading
+ withSQLConf(
+ CometConf.COMET_NATIVE_PARQUET_WRITE_ENABLED.key -> "false",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Denver") {
+ inputDf.write.parquet(inputPath)
+ }
+
+ // Now read and write with Comet native writer
+ // Use auto scan mode so native_iceberg_compat is used (which supports complex types)
+ // instead of native_comet. This overrides the COMET_PARQUET_SCAN_IMPL env var set by CI.
+ withSQLConf(
+ CometConf.COMET_NATIVE_PARQUET_WRITE_ENABLED.key -> "true",
+ CometConf.COMET_NATIVE_SCAN_IMPL.key -> "auto",
+ CometConf.COMET_SCAN_ALLOW_INCOMPATIBLE.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Halifax",
+ CometConf.getOperatorAllowIncompatConfigKey(classOf[DataWritingCommandExec]) -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true") {
+
+ val parquetDf = spark.read.parquet(inputPath)
+ parquetDf.write.parquet(outputPath)
+
+ // Verify round-trip: read with Spark and Comet, compare results
+ var sparkDf: DataFrame = null
+ var cometDf: DataFrame = null
+ withSQLConf(CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false") {
+ sparkDf = spark.read.parquet(outputPath)
+ }
+ withSQLConf(CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "true") {
+ cometDf = spark.read.parquet(outputPath)
+ }
+
+ assert(sparkDf.count() == expectedRows, s"Expected $expectedRows rows")
+ checkAnswer(sparkDf, cometDf)
Review Comment:
Adding the following codes confirms that the configuration in L267/L270 did not take effect.
```
println("parquetDf :\n" + parquetDf.queryExecution.executedPlan)
println("cometDf :\n" + cometDf.queryExecution.executedPlan)
println("sparkDf :\n" + sparkDf.queryExecution.executedPlan)
```
got output:
```
parquetDf :
*(1) CometColumnarToRow
+- CometScan [native_iceberg_compat] parquet [id#13,values#14] Batched: true, DataFilters: [], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/private/var/folders/4p/9gtjq1s10fd6frkv9kzy0y740000gn/T/spark-8c..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<id:int,values:array<int>>
cometDf :
*(1) CometColumnarToRow
+- CometScan [native_iceberg_compat] parquet [id#23,values#24] Batched: true, DataFilters: [], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/private/var/folders/4p/9gtjq1s10fd6frkv9kzy0y740000gn/T/spark-17..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<id:int,values:array<int>>
sparkDf :
*(1) CometColumnarToRow
+- CometScan [native_iceberg_compat] parquet [id#19,values#20] Batched: true, DataFilters: [], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/private/var/folders/4p/9gtjq1s10fd6frkv9kzy0y740000gn/T/spark-17..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<id:int,values:array<int>>
```
suggestion for change:
```
val inputDf = spark.read.parquet(inputPath)
inputDf.write.parquet(outputPath)
val inputRows = inputDf.collect()
var outputRows: Seq[Row] = null
withSQLConf(CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "true") {
val cometDf = spark.read.parquet(outputPath)
outputRows = cometDf.collect()
}
// Verify round-trip: read with Spark and Comet, compare results
withSQLConf(CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false") {
val sparkDf = spark.read.parquet(outputPath)
assert(sparkDf.count() == expectedRows, s"Expected $expectedRows rows")
checkAnswer(sparkDf, inputRows)
checkAnswer(sparkDf, outputRows)
}
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: add complex type support to native Parquet writer [datafusion-comet]
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #3214:
URL: https://github.com/apache/datafusion-comet/pull/3214#discussion_r2704353956
##########
spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala:
##########
@@ -228,4 +230,299 @@ class CometParquetWriterSuite extends CometTestBase {
}
}
}
+
+ // ===== Complex Type Tests =====
+
+ private def writeComplexTypeData(
+ inputDf: DataFrame,
+ outputPath: String,
+ expectedRows: Int): Unit = {
+ withTempPath { inputDir =>
+ val inputPath = new File(inputDir, "input.parquet").getAbsolutePath
+
+ // First write the input data without Comet to get proper Arrow arrays when reading
+ withSQLConf(
+ CometConf.COMET_NATIVE_PARQUET_WRITE_ENABLED.key -> "false",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Denver") {
+ inputDf.write.parquet(inputPath)
+ }
+
+ // Now read and write with Comet native writer
+ // Use auto scan mode so native_iceberg_compat is used (which supports complex types)
+ // instead of native_comet. This overrides the COMET_PARQUET_SCAN_IMPL env var set by CI.
+ withSQLConf(
+ CometConf.COMET_NATIVE_PARQUET_WRITE_ENABLED.key -> "true",
+ CometConf.COMET_NATIVE_SCAN_IMPL.key -> "auto",
+ CometConf.COMET_SCAN_ALLOW_INCOMPATIBLE.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Halifax",
+ CometConf.getOperatorAllowIncompatConfigKey(classOf[DataWritingCommandExec]) -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true") {
+
+ val parquetDf = spark.read.parquet(inputPath)
+ parquetDf.write.parquet(outputPath)
+
+ // Verify round-trip: read with Spark and Comet, compare results
+ var sparkDf: DataFrame = null
+ var cometDf: DataFrame = null
+ withSQLConf(CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false") {
+ sparkDf = spark.read.parquet(outputPath)
+ }
+ withSQLConf(CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "true") {
+ cometDf = spark.read.parquet(outputPath)
+ }
+
+ assert(sparkDf.count() == expectedRows, s"Expected $expectedRows rows")
+ checkAnswer(sparkDf, cometDf)
+ checkAnswer(parquetDf, sparkDf)
+ }
+ }
+ }
+
+ test("parquet write with array type") {
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ val df = Seq((1, Seq(1, 2, 3)), (2, Seq(4, 5)), (3, Seq[Int]()), (4, Seq(6, 7, 8, 9)))
+ .toDF("id", "values")
+
+ writeComplexTypeData(df, outputPath, 4)
+ }
+ }
+
+ test("parquet write with struct type") {
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ val df =
+ Seq((1, ("Alice", 30)), (2, ("Bob", 25)), (3, ("Charlie", 35))).toDF("id", "person")
+
+ writeComplexTypeData(df, outputPath, 3)
+ }
+ }
+
+ test("parquet write with map type") {
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ val df = Seq(
+ (1, Map("a" -> 1, "b" -> 2)),
+ (2, Map("c" -> 3)),
+ (3, Map[String, Int]()),
+ (4, Map("d" -> 4, "e" -> 5, "f" -> 6))).toDF("id", "properties")
+
+ writeComplexTypeData(df, outputPath, 4)
+ }
+ }
+
+ test("parquet write with array of structs") {
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ val df = Seq(
+ (1, Seq(("Alice", 30), ("Bob", 25))),
+ (2, Seq(("Charlie", 35))),
+ (3, Seq[(String, Int)]())).toDF("id", "people")
+
+ writeComplexTypeData(df, outputPath, 3)
+ }
+ }
+
+ test("parquet write with struct containing array") {
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ val df = spark.sql("""
+ SELECT
+ 1 as id,
+ named_struct('name', 'Team A', 'scores', array(95, 87, 92)) as team
+ UNION ALL SELECT
+ 2 as id,
+ named_struct('name', 'Team B', 'scores', array(88, 91)) as team
+ UNION ALL SELECT
+ 3 as id,
+ named_struct('name', 'Team C', 'scores', array(100)) as team
+ """)
+
+ writeComplexTypeData(df, outputPath, 3)
+ }
+ }
+
+ test("parquet write with map with struct values") {
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ val df = spark.sql("""
+ SELECT
+ 1 as id,
+ map('emp1', named_struct('name', 'Alice', 'age', 30),
+ 'emp2', named_struct('name', 'Bob', 'age', 25)) as employees
+ UNION ALL SELECT
+ 2 as id,
+ map('emp3', named_struct('name', 'Charlie', 'age', 35)) as employees
+ """)
+
+ writeComplexTypeData(df, outputPath, 2)
+ }
+ }
+
+ test("parquet write with deeply nested types") {
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ // Create deeply nested structure: array of maps containing arrays
+ val df = spark.sql("""
+ SELECT
+ 1 as id,
+ array(
+ map('key1', array(1, 2, 3), 'key2', array(4, 5)),
+ map('key3', array(6, 7, 8, 9))
+ ) as nested_data
+ UNION ALL SELECT
+ 2 as id,
+ array(
+ map('key4', array(10, 11))
+ ) as nested_data
+ """)
+
+ writeComplexTypeData(df, outputPath, 2)
+ }
+ }
+
+ test("parquet write with nullable complex types") {
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ // Test nulls at various levels
+ val df = spark.sql("""
+ SELECT
+ 1 as id,
+ array(1, null, 3) as arr_with_nulls,
+ named_struct('a', 1, 'b', cast(null as int)) as struct_with_nulls,
+ map('x', 1, 'y', cast(null as int)) as map_with_nulls
+ UNION ALL SELECT
+ 2 as id,
+ cast(null as array<int>) as arr_with_nulls,
+ cast(null as struct<a:int, b:int>) as struct_with_nulls,
+ cast(null as map<string, int>) as map_with_nulls
+ UNION ALL SELECT
+ 3 as id,
+ array(4, 5, 6) as arr_with_nulls,
+ named_struct('a', 7, 'b', 8) as struct_with_nulls,
+ map('z', 9) as map_with_nulls
+ """)
+
+ writeComplexTypeData(df, outputPath, 3)
+ }
+ }
+
+ test("parquet write with decimal types within complex types") {
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ val df = spark.sql("""
+ SELECT
+ 1 as id,
+ array(cast(1.23 as decimal(10,2)), cast(4.56 as decimal(10,2))) as decimal_arr,
+ named_struct('amount', cast(99.99 as decimal(10,2))) as decimal_struct,
+ map('price', cast(19.99 as decimal(10,2))) as decimal_map
+ UNION ALL SELECT
+ 2 as id,
+ array(cast(7.89 as decimal(10,2))) as decimal_arr,
+ named_struct('amount', cast(0.01 as decimal(10,2))) as decimal_struct,
+ map('price', cast(0.50 as decimal(10,2))) as decimal_map
+ """)
+
+ writeComplexTypeData(df, outputPath, 2)
+ }
+ }
+
+ test("parquet write with temporal types within complex types") {
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ val df = spark.sql("""
+ SELECT
+ 1 as id,
+ array(date'2024-01-15', date'2024-02-20') as date_arr,
+ named_struct('ts', timestamp'2024-01-15 10:30:00') as ts_struct,
+ map('event', timestamp'2024-03-01 14:00:00') as ts_map
+ UNION ALL SELECT
+ 2 as id,
+ array(date'2024-06-30') as date_arr,
+ named_struct('ts', timestamp'2024-07-04 12:00:00') as ts_struct,
+ map('event', timestamp'2024-12-25 00:00:00') as ts_map
+ """)
+
+ writeComplexTypeData(df, outputPath, 2)
+ }
+ }
+
+ test("parquet write with empty arrays and maps") {
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ val df = Seq(
+ (1, Seq[Int](), Map[String, Int]()),
+ (2, Seq(1, 2), Map("a" -> 1)),
+ (3, Seq[Int](), Map[String, Int]())).toDF("id", "arr", "mp")
+
+ writeComplexTypeData(df, outputPath, 3)
+ }
+ }
+
+ test("parquet write complex types fuzz test") {
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ withTempPath { inputDir =>
+ val inputPath = new File(inputDir, "input.parquet").getAbsolutePath
+
+ // Generate test data with complex types enabled
+ val schema = FuzzDataGenerator.generateSchema(
+ SchemaGenOptions(generateArray = true, generateStruct = true, generateMap = true))
+ val df = FuzzDataGenerator.generateDataFrame(
+ new Random(42),
+ spark,
+ schema,
+ 500,
+ DataGenOptions(generateNegativeZero = false))
+
+ // Write input data without Comet
+ withSQLConf(
+ CometConf.COMET_NATIVE_PARQUET_WRITE_ENABLED.key -> "false",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Denver") {
+ df.write.parquet(inputPath)
+ }
+
+ // Write with Comet native writer
+ // Use auto scan mode so native_iceberg_compat is used (which supports complex types)
+ // instead of native_comet. This overrides the COMET_PARQUET_SCAN_IMPL env var set by CI.
+ withSQLConf(
+ CometConf.COMET_NATIVE_PARQUET_WRITE_ENABLED.key -> "true",
+ CometConf.COMET_NATIVE_SCAN_IMPL.key -> "auto",
+ CometConf.COMET_SCAN_ALLOW_INCOMPATIBLE.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Halifax",
+ CometConf.getOperatorAllowIncompatConfigKey(classOf[DataWritingCommandExec]) -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true") {
+
+ val inputDf = spark.read.parquet(inputPath)
+ inputDf.write.parquet(outputPath)
+
+ // Verify round-trip
+ var sparkDf: DataFrame = null
+ var cometDf: DataFrame = null
+ withSQLConf(CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false") {
+ sparkDf = spark.read.parquet(outputPath)
+ }
+ withSQLConf(CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "true") {
+ cometDf = spark.read.parquet(outputPath)
+ }
+
+ assert(sparkDf.count() == 500, "Expected 500 rows")
+ checkAnswer(sparkDf, cometDf)
Review Comment:
ditto
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: add complex type support to native Parquet writer [datafusion-comet]
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #3214:
URL: https://github.com/apache/datafusion-comet/pull/3214#discussion_r2706444516
##########
native/core/src/execution/operators/parquet_writer.rs:
##########
@@ -535,8 +535,12 @@ impl ExecutionPlan for ParquetWriterExec {
DataFusionError::Execution(format!("Failed to close writer: {}", e))
})?;
- // Get file size
- let file_size = std::fs::metadata(&part_file)
+ // Get file size - strip file:// prefix if present for local filesystem access
+ let local_path = part_file
+ .strip_prefix("file://")
+ .or_else(|| part_file.strip_prefix("file:"))
+ .unwrap_or(&part_file);
+ let file_size = std::fs::metadata(local_path)
Review Comment:
A question unrelated to this PR: #2929 supports write to hdfs. So, is the logic here correct when local_path is an hdfs path?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: add complex type support to native Parquet writer [datafusion-comet]
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #3214:
URL: https://github.com/apache/datafusion-comet/pull/3214#discussion_r2703004660
##########
spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala:
##########
@@ -228,4 +230,299 @@ class CometParquetWriterSuite extends CometTestBase {
}
}
}
+
+ // ===== Complex Type Tests =====
+
+ private def writeComplexTypeData(
+ inputDf: DataFrame,
+ outputPath: String,
+ expectedRows: Int): Unit = {
+ withTempPath { inputDir =>
+ val inputPath = new File(inputDir, "input.parquet").getAbsolutePath
+
+ // First write the input data without Comet to get proper Arrow arrays when reading
+ withSQLConf(
+ CometConf.COMET_NATIVE_PARQUET_WRITE_ENABLED.key -> "false",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Denver") {
+ inputDf.write.parquet(inputPath)
+ }
+
+ // Now read and write with Comet native writer
+ // Use auto scan mode so native_iceberg_compat is used (which supports complex types)
+ // instead of native_comet. This overrides the COMET_PARQUET_SCAN_IMPL env var set by CI.
+ withSQLConf(
+ CometConf.COMET_NATIVE_PARQUET_WRITE_ENABLED.key -> "true",
+ CometConf.COMET_NATIVE_SCAN_IMPL.key -> "auto",
+ CometConf.COMET_SCAN_ALLOW_INCOMPATIBLE.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Halifax",
+ CometConf.getOperatorAllowIncompatConfigKey(classOf[DataWritingCommandExec]) -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true") {
+
+ val parquetDf = spark.read.parquet(inputPath)
+ parquetDf.write.parquet(outputPath)
+
+ // Verify round-trip: read with Spark and Comet, compare results
+ var sparkDf: DataFrame = null
+ var cometDf: DataFrame = null
+ withSQLConf(CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false") {
+ sparkDf = spark.read.parquet(outputPath)
+ }
+ withSQLConf(CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "true") {
+ cometDf = spark.read.parquet(outputPath)
+ }
+
+ assert(sparkDf.count() == expectedRows, s"Expected $expectedRows rows")
+ checkAnswer(sparkDf, cometDf)
Review Comment:
Will the configurations in L267/L270 take effect 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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: add complex type support to native Parquet writer [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #3214:
URL: https://github.com/apache/datafusion-comet/pull/3214#issuecomment-3765755804
With these changes, we can run the PySpark repartition benchmark fully native and it shows an almost 2x speedup compared to Spark.
<img width="781" height="1609" alt="plan" src="https://github.com/user-attachments/assets/064ee985-0cbc-4e22-bdd7-ee885b6ae39e" />
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: add complex type support to native Parquet writer [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #3214:
URL: https://github.com/apache/datafusion-comet/pull/3214#issuecomment-3765833402
> With these changes, we can run the PySpark repartition benchmark fully native, and it shows an almost 2x speedup compared to Spark (and also ~2x compared to Comet when writes are disabled and Comet does the columnar-to-row transition).
@comphead :point_up:
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: add complex type support to native Parquet writer [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #3214:
URL: https://github.com/apache/datafusion-comet/pull/3214#issuecomment-3769620049
Thanks for the thorough review @wForget. I refactored the test framework and added assertions that the plans are running as intended (Spark vs Comet).
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: add complex type support to native Parquet writer [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove merged PR #3214:
URL: https://github.com/apache/datafusion-comet/pull/3214
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org