[SPARK-33521][SQL] Universal type conversion in resolving V2 partition specs

[MaxGekk]

What changes were proposed in this pull request?

In the PR, I propose to changes the resolver of partition specs used in V2 ALTER TABLE .. ADD/DROP PARTITION (at the moment), and re-use CAST in conversion partition values to desired types according to the partition schema.

Why are the changes needed?

Currently, the resolver of V2 partition specs supports just a few types:

spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolvePartitionSpec.scala

Line 72 in 23e9920

// TODO: Support other datatypes, such as DateType

, and fails on other types like date/timestamp.

Does this PR introduce any user-facing change?

Yes

How was this patch tested?

By running AlterTablePartitionV2SQLSuite

[SparkQA]

Test build #131572 has started for PR 30474 at commit d0108e0.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/36173/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/36173/

[HyukjinKwon]

Is this matched to the V1 partitioning type coercion? I remember it has a bit different rules, see PartitioningUtils.inferPartitionColumnValue.

[MaxGekk]

We don't have unified tests for V1 and V2 ALTER TABLE .. ADD/DROP PARTITION at the moment. I plan to do that soon. As soon as we have such tests we will see the differences and fix them.

For now, I just try to make implementation simpler - cast partition values according to the partition schema.

[MaxGekk]

I think V1 does the same, see

spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala

Lines 149 to 165 in dfa6fb4

	/**
	* Given the partition schema, returns a row with that schema holding the partition values.
	*/
	def toRow(partitionSchema: StructType, defaultTimeZondId: String): InternalRow = {
	val caseInsensitiveProperties = CaseInsensitiveMap(storage.properties)
	val timeZoneId = caseInsensitiveProperties.getOrElse(
	DateTimeUtils.TIMEZONE_OPTION, defaultTimeZondId)
	InternalRow.fromSeq(partitionSchema.map { field =>
	val partValue = if (spec(field.name) == ExternalCatalogUtils.DEFAULT_PARTITION_NAME) {
	null
	} else {
	spec(field.name)
	}
	Cast(Literal(partValue), field.dataType, Option(timeZoneId)).eval()
	})
	}
	}

Let me extract the code to PartitioningUtils, and re-use it in V2.

[HyukjinKwon]

Gotya

[MaxGekk]

I reused the code from DSv1 #30482 and fixed an issue. @HyukjinKwon Please, review it.

[SparkQA]

Test build #131579 has finished for PR 30474 at commit d0108e0.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

thanks, merging to master!