[EPIC] Improve Comet Native writer

[comphead]

What is the problem the feature request solves?

This epic to track Comet native writer issues

• The Comet writer adds unnecessary transition CometColumnarToRow #2958
• Comet native writer crashes for non Arrow input data #2944
• Comet writer doesnt support INSERT OVERWRITE #2957
• Support HDFS writes with Comet writer via OpenDAL #2890
• Comet writer doesn't respect OVERWRITE mode #2970
• Comet writer executes twice #2971
• Comet writer doesn't create _SUCCESS file #2985
• Comet writer should support 2PC and staging output #3015
• Comet writer to support nested types #3023
• Comet writer to respect object_store_settings sent from Scala #3032
• Comet writer should support dynamicPartitionOverwrite mode #3041
• Run Spark tests with Comet native writer #3209
• [Native Writer] INSERT INTO ... SELECT fails due to stale catalog cache after write #3521
• empty file should be skipped while write to file
• INSERT INTO TABLE - complex type but different names
• Insert overwrite table command should output correct schema: basic
• parquet timestamp conversion
• SPARK-29174 Support LOCAL in INSERT OVERWRITE DIRECTORY to data source
• SPARK-33901: ctas should should not change table's schema
• SPARK-37160: CREATE TABLE AS SELECT with CHAR_AS_VARCHAR
• SPARK-38336 INSERT INTO statements with tables with default columns: positive tests
• SPARK-38811 INSERT INTO on columns added with ALTER TABLE ADD COLUMNS: Positive tests
• SPARK-43071: INSERT INTO from queries whose final operators are not projections
• write path implements onTaskCommit API correctly
• Write Spark version into Parquet metadata
• Native writer doesn't report written bytes #3608
• respect parquet block size

Additional tests which had to be ignored for spark 4 :

• ctas with union
• SPARK-48817: test multi inserts

Describe the potential solution

No response

Additional context

No response

[coderfender]

Creating new issues and tagging in this EPIC to address failing spark tests

[coderfender]

I think we can classify test failures fall into 3 main categories :

Empty file/dir creation

1. empty file should be skipped while write to file
2. SPARK-33901: ctas should not change table's schema
3. SPARK-37160: CREATE TABLE AS SELECT with CHAR_AS_VARCHAR

Insert overwrite bugs

1. SPARK-29174 Support LOCAL in INSERT OVERWRITE DIRECTORY
2. SPARK-38336 INSERT INTO with default columns
3. SPARK-38811 INSERT INTO on ALTER TABLE ADD COLUMNS
4. SPARK-43071: INSERT INTO from non-projection queries
5. Insert overwrite table command should output correct schema: basic

Other

1. parquet timestamp conversion (timestamp96)
2. INSERT INTO TABLE - complex type but different names
3. Write Spark version into Parquet metadata
4. write path implements onTaskCommit API correctly

[coderfender]

linking newly created issues with the EPIC for lineage tracking

Empty file/dir creation

empty file should be skipped while write to file
#3417

SPARK-33901: ctas should not change table's schema
#3418

SPARK-37160: CREATE TABLE AS SELECT with CHAR_AS_VARCHAR
#3419

Insert overwrite bugs

SPARK-29174 Support LOCAL in INSERT OVERWRITE DIRECTORY
#3420

SPARK-38336 INSERT INTO with default columns
#3421

SPARK-38811 INSERT INTO on ALTER TABLE ADD COLUMNS
#3422

SPARK-43071: INSERT INTO from non-projection queries
#3423

Insert overwrite table command should output correct schema: basic
#3424

Other

parquet timestamp conversion (timestamp96)
#3425

INSERT INTO TABLE - complex type but different names
#3426

Write Spark version into Parquet metadata
#3427

write path implements onTaskCommit API correctly
#3428

##Spark 4x tests

ctas with union
#3429

SPARK-48817: test multi insert
#3430

[comphead]

Here test suites that Spark run for the writer

  Core DataFrame Writer Test Suites

  DataFrameReaderWriterSuite - sql/core/src/test/scala/org/apache/spark/sql/test/DataFrameReaderWriterSuite.scala:145
  DataFrameWriterV2Suite - sql/core/src/test/scala/org/apache/spark/sql/DataFrameWriterV2Suite.scala:44
  DataFrameSuite - sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala:60
  DataFrameCallbackSuite - sql/core/src/test/scala/org/apache/spark/sql/util/DataFrameCallbackSuite.scala:37

  Java Test Suites

  JavaDataFrameReaderWriterSuite - sql/core/src/test/java/test/org/apache/spark/sql/JavaDataFrameReaderWriterSuite.java:32
  JavaDataFrameWriterV2Suite - sql/core/src/test/java/test/org/apache/spark/sql/JavaDataFrameWriterV2Suite.java:35

  Format-Specific Writer Test Suites

  FileFormatWriterSuite - sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileFormatWriterSuite.scala:24
  FileWriterFactorySuite - sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/FileWriterFactorySuite.scala:29
  JDBCWriteSuite - sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCWriteSuite.scala:39

  Partitioning and Bucketing Writer Test Suites

  PartitionedWriteSuite - sql/core/src/test/scala/org/apache/spark/sql/sources/PartitionedWriteSuite.scala:50
  BucketedWriteSuite - sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedWriteSuite.scala:44
  BucketedWriteWithoutHiveSupportSuite - sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedWriteSuite.scala:35

  V1/V2 Write Command Test Suites

  V1WriteCommandSuite - sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/V1WriteCommandSuite.scala:128
  V1WriteFallbackSuite - sql/core/src/test/scala/org/apache/spark/sql/connector/V1WriteFallbackSuite.scala:45
  SaveIntoDataSourceCommandSuite - sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SaveIntoDataSourceCommandSuite.scala:26

  Insert and Save/Load Test Suites

  InsertSuite - sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala:59
  SaveLoadSuite - sql/core/src/test/scala/org/apache/spark/sql/sources/SaveLoadSuite.scala:32
  CreateTableAsSelectSuite - sql/core/src/test/scala/org/apache/spark/sql/sources/CreateTableAsSelectSuite.scala:30

  Streaming Writer Test Suites

  DataStreamReaderWriterSuite - sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamReaderWriterSuite.scala:113
  FileStreamSinkV1Suite - sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSinkSuite.scala:707
  FileStreamSinkV2Suite - sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSinkSuite.scala:758
  ForeachWriterSuite - sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/sources/ForeachWriterSuite.scala:33
  PythonForeachWriterSuite - sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonForeachWriterSuite.scala:35

  Other Writer Test Suites

  ArrowWriterSuite - sql/core/src/test/scala/org/apache/spark/sql/execution/arrow/ArrowWriterSuite.scala:2

Trying to create a separate pipeline with writer enabled to run those tests

[Shekharrajak]

We can consider this feature to be available in native writer #3595 - please have a look.

[comphead]

We can consider this feature to be available in native writer #3595 - please have a look.

For Iceberg I'm not sure tbh, as iceberg-rs should have its own writer afaik, @mbutrovich please confirm

[Shekharrajak]

Please have a look : apache/iceberg-rust#2269 - this tracker can help in completing the iceberg native write support.