From e98da53a745cecc40d06733f0286b6d4982610ac Mon Sep 17 00:00:00 2001 From: liupoyi-1031 <242609140+liupoyi-1031@users.noreply.github.com> Date: Sat, 19 Sep 2026 15:24:52 +0000 Subject: [PATCH] fix: preserve Spark row index read errors Fall back before validating row index column types so Spark can surface invalid schemas as FAILED_READ_FILE. --- dev/diffs/4.0.4.diff | 10 ---------- dev/diffs/4.1.3.diff | 10 ---------- .../scala/org/apache/comet/rules/CometScanRule.scala | 6 +++++- 3 files changed, 5 insertions(+), 21 deletions(-) diff --git a/dev/diffs/4.0.4.diff b/dev/diffs/4.0.4.diff index 12eaabd1bed..1d9978f4186 100644 --- a/dev/diffs/4.0.4.diff +++ b/dev/diffs/4.0.4.diff @@ -3056,16 +3056,6 @@ index 08fd8a9ecb5..06967aec8e1 100644 case _ => } assert(numPartitions > 0) -@@ -303,6 +312,9 @@ class ParquetRowIndexSuite extends QueryTest with SharedSparkSession { - val conf = RowIndexTestConf(useDataSourceV2 = useDataSourceV2) - - test(s"invalid row index column type - ${conf.desc}") { -+ // https://github.com/apache/datafusion-comet/issues/3886 -+ // Comet throws RuntimeException instead of SparkException -+ assume(false) - withSQLConf(conf.sqlConfs: _*) { - withTempPath{ path => - val df = spark.range(0, 10, 1, 1).toDF("id") diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaPruningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaPruningSuite.scala index 5c0b7def039..151184bc98c 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaPruningSuite.scala diff --git a/dev/diffs/4.1.3.diff b/dev/diffs/4.1.3.diff index bec73bfb988..af819e4a1b9 100644 --- a/dev/diffs/4.1.3.diff +++ b/dev/diffs/4.1.3.diff @@ -3222,16 +3222,6 @@ index 08fd8a9ecb5..06967aec8e1 100644 case _ => } assert(numPartitions > 0) -@@ -303,6 +312,9 @@ class ParquetRowIndexSuite extends QueryTest with SharedSparkSession { - val conf = RowIndexTestConf(useDataSourceV2 = useDataSourceV2) - - test(s"invalid row index column type - ${conf.desc}") { -+ // https://github.com/apache/datafusion-comet/issues/3886 -+ // Comet throws RuntimeException instead of SparkException -+ assume(false) - withSQLConf(conf.sqlConfs: _*) { - withTempPath{ path => - val df = spark.range(0, 10, 1, 1).toDF("id") diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaPruningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaPruningSuite.scala index 5c0b7def039..151184bc98c 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaPruningSuite.scala diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala index d6a5223bee8..5f37761a051 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -346,7 +346,11 @@ case class CometScanRule(session: SparkSession) "input_file_block_start, or input_file_block_length") return None } - if (ShimFileFormat.findRowIndexColumnIndexInSchema(scanExec.requiredSchema) >= 0) { + // Check the name directly instead of calling findRowIndexColumnIndexInSchema, which validates + // the temporary column's type and can throw a raw RuntimeException. Falling back lets Spark's + // Parquet reader wrap that validation failure as FAILED_READ_FILE. + if (scanExec.requiredSchema.fieldNames.contains( + ShimFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME)) { withFallbackReason(scanExec, "Native Parquet scan does not support row index generation") return None }