Fall back instead of asserting when native scan partition values cannot be serialized

[peterxcli]

What is the problem the feature request solves?

partition2Proto assumes that accepting a partition schema guarantees every concrete partition value can be serialized:

val valueProto = exprToProto(Literal(value, attr.dataType), Seq.empty)
assert(
  valueProto.isDefined,
  s"Unsupported partition value: $value, type: ${attr.dataType}")
valueProto.get

That assumption spans separate support checks. CometScanRule validates the partition schema through CometScanTypeChecker, while partition2Proto later relies on CometLiteral / exprToProto supporting each concrete value. If those capabilities drift, Comet has already selected the native scan and raises AssertionError instead of falling back to Spark.

PR #5161 exposed this with YearMonthIntervalType: the scan checker admitted the partition type before the literal path supported it. That specific interval gap is fixed by the PR, but the generic assertion remains. There is no currently known unsupported partition literal that passes the gate at the PR head; this issue tracks hardening the invariant before another type exposes it.

The helper is shared by the native Parquet and CSV scan paths.

Describe the potential solution

• Use one shared capability check for partition-schema admission and partition-literal serialization so the two cannot drift.
• Validate partition-value serialization before committing to the native scan, or otherwise propagate serialization failure to a point where Spark fallback is still possible.
• Record a useful fallback reason instead of surfacing AssertionError as an internal failure.
• Add a regression test that deliberately exercises an unsupported partition literal and verifies clean fallback.

Additional context

• partition2Proto:

  datafusion-comet/spark/src/main/scala/org/apache/comet/serde/operator/package.scala

  Lines 48 to 65 in 2963b08

  	def partition2Proto(
  	partition: FilePartition,
  	partitionSchema: StructType): OperatorOuterClass.SparkFilePartition = {
  	val partitionBuilder = OperatorOuterClass.SparkFilePartition.newBuilder()
  	partition.files.foreach(file => {
  	// Process the partition values
  	val partitionValues = file.partitionValues
  	assert(partitionValues.numFields == partitionSchema.length)
  	val partitionVals =
  	partitionValues.toSeq(partitionSchema).zipWithIndex.map { case (value, i) =>
  	val attr = partitionSchema(i)
  	val valueProto = exprToProto(Literal(value, attr.dataType), Seq.empty)
  	// In `CometScanRule`, we have already checked that all partition values are
  	// supported. So, we can safely use `get` here.
  	assert(
  	valueProto.isDefined,
  	s"Unsupported partition value: $value, type: ${attr.dataType}")
  	valueProto.get

• Parquet caller:

  datafusion-comet/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala

  Lines 215 to 228 in 2963b08

  	// Get file partitions from CometScanExec (handles bucketing, etc.)
  	val filePartitions = scan.getFilePartitions()
  	// Serialize each partition's files
  	import org.apache.comet.serde.operator.partition2Proto
  	val perPartitionBytes = filePartitions.map { filePartition =>
  	val partitionProto = partition2Proto(filePartition, relation.partitionSchema)
  	val partitionNativeScan = org.apache.comet.serde.OperatorOuterClass.NativeScan
  	.newBuilder()
  	.setFilePartition(partitionProto)
  	.build()
  	partitionNativeScan.toByteArray
  	}.toArray

• CSV caller:

  datafusion-comet/spark/src/main/scala/org/apache/spark/sql/comet/CometCsvNativeScanExec.scala

  Lines 95 to 105 in 2963b08

  	val filePartitions = op.inputPartitions.map(_.asInstanceOf[FilePartition])
  	val csvOptionsProto = csvOptions2Proto(options)
  	val dataSchemaProto = schema2Proto(csvScan.dataSchema.fields)
  	val readSchemaFieldNames = csvScan.readDataSchema.fieldNames
  	val projectionVector = csvScan.dataSchema.fields.zipWithIndex
  	.filter { case (field, _) =>
  	readSchemaFieldNames.contains(field.name)
  	}
  	.map(_._2.asInstanceOf[Integer])
  	val partitionSchemaProto = schema2Proto(csvScan.readPartitionSchema.fields)
  	val partitionsProto = filePartitions.map(partition2Proto(_, csvScan.readPartitionSchema))

• PR: feat: support ANSI interval types in native Parquet scans #5161
• Review discussion and original reproduction: feat: support ANSI interval types in native Parquet scans #5161 (comment)

[peterxcli]

take