[SPARK-37287][SQL] Pull out dynamic partition and bucket sort from FileFormatWriter

[allisonwang-db]

What changes were proposed in this pull request?

FileFormatWriter.write is used by all V1 write commands including data source and hive tables. Depending on dynamic partitions, bucketed, and sort columns in the V1 write command, FileFormatWriter can add a physical sort on top of the query plan which is not visible from plan directly.

This PR (based on #34568) intends to pull out the physical sort added by FileFormatWriter into logical planning. It adds a new logical rule V1Writes to add logical Sort operators based on the required ordering of a V1 write command. This behavior can be controlled by the new config spark.sql.optimizer.plannedWrite.enabled (default: true).

Why are the changes needed?

Improve observability of V1 write, and unify the logic of V1 and V2 write commands.

Does this PR introduce any user-facing change?

No

How was this patch tested?

New unit tests.

[c21]

Thank you @allisonwang-db for picking this up! Knowing the PR is still draft, but just leave some early questions/comments. cc @ulysses-you and @cloud-fan as well.

[c21]

nit: 3.4.0?

[c21]

nit: feeling the config name is a little bit obscure. could it be spark.sql.requireOrderingForV1Writers or something similar?

[allisonwang-db]

Indeed the name is not very descriptive. Planned write here means we want to explicitly plan file writes instead of adding various operations when executing the write. It could include things other than required ordering in the future. I am happy to brainstorm more here.

[c21]

Spark -> Spark optimizer could be clearer that sort is added during query planning.

[c21]

sorry why we need the comment here?

[allisonwang-db]

I've removed the confusing comment. It means we don't need other option values like table path when getting the writer bucket spec.

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/command/createDataSourceTables.scala

Lines 176 to 180 in 3331d4c

	val tableLocation = if (table.tableType == CatalogTableType.MANAGED) {
	Some(sessionState.catalog.defaultTablePath(table.identifier))
	} else {
	table.storage.locationUri
	}

[c21]

Is it just for debugging?

[allisonwang-db]

This is used in unit test to check when v1 writes is enabled, we should have added a logical sort and thus do not need to add a physical sort (ordering should match).

[c21]

Just brainstorming here, if we plan to add a requirement for partitioning, e.g. support shuffle before writing bucket table. Do we want to add a similar RequiresDistributionAndOrdering as v2 now or not?

[cloud-fan]

I think we can add one more method: requiredPartitioning

[c21]

Thanks @cloud-fan and @allisonwang-db for pushing on this. This is great!

I will work on supporting required partitioning for V1 in this week (https://issues.apache.org/jira/browse/SPARK-37287). The motivation is to support shuffling on bucket columns when writing Hive bucket table. cc @cloud-fan, @allisonwang-db and @ulysses-you FYI.

[gengliangwang]

So there was an optimization in #32198 which can avoid local sort if there are only a small set of partition/bucket values.
Is there optimization gone after the changes in this PR?

[cloud-fan]

That optimization is off by default. When it's turned on, we skip planned write.

[dongjoon-hyun]

Hi, @allisonwang-db and @cloud-fan

There is a correctness issue report for this configuration, SPARK-44512, for Apache Spark 3.4.0+. Could you take a look at that?

[dongjoon-hyun]

Sorry all. After checking the reported use case once more, I found that that it's a false alarm. I closed the issue as Not A Problem.