You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/03/18 16:41:07 UTC
(spark) branch branch-4.1 updated: Revert "[SPARK-56045][SQL] Add flag for ignoring Parquet UNKNOWN type annotation and revert to old behavior"
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new fd2d262c57ae Revert "[SPARK-56045][SQL] Add flag for ignoring Parquet UNKNOWN type annotation and revert to old behavior"
fd2d262c57ae is described below
commit fd2d262c57ae3b51c14932b6cf975c14cca86112
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Wed Mar 18 09:40:52 2026 -0700
Revert "[SPARK-56045][SQL] Add flag for ignoring Parquet UNKNOWN type annotation and revert to old behavior"
This reverts commit 6589d497ba94af4e67db217b6fef6255a4ec6997.
---
.../org/apache/spark/sql/internal/SQLConf.scala | 14 --------------
.../datasources/parquet/ParquetFileFormat.scala | 12 ++----------
.../datasources/parquet/ParquetSchemaConverter.scala | 19 ++++---------------
.../datasources/v2/parquet/ParquetScan.scala | 3 ---
.../test/resources/test-data/void_in_parquet.parquet | Bin 2022 -> 0 bytes
.../datasources/parquet/ParquetIOSuite.scala | 18 ------------------
6 files changed, 6 insertions(+), 60 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 5855ea48a2f5..100149a39211 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -1617,17 +1617,6 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
- val PARQUET_READER_RESPECT_UNKNOWN_TYPE_ANNOTATION =
- buildConf("spark.sql.parquet.reader.respectUnknownTypeAnnotation.enabled")
- .internal()
- .doc("When enabled, respects the UNKNOWN type annotation in Parquet files during schema " +
- "inference and infers NullType. When disabled, ignores the UNKNOWN annotation " +
- "and uses the physical type instead.")
- .version("4.1.2")
- .withBindingPolicy(ConfigBindingPolicy.SESSION)
- .booleanConf
- .createWithDefault(false)
-
val PARQUET_FIELD_ID_READ_ENABLED =
buildConf("spark.sql.parquet.fieldId.read.enabled")
.doc("Field ID is a native field of the Parquet schema spec. When enabled, Parquet readers " +
@@ -7733,9 +7722,6 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def parquetIgnoreVariantAnnotation: Boolean = getConf(SQLConf.PARQUET_IGNORE_VARIANT_ANNOTATION)
- def parquetReaderRespectUnknownTypeAnnotation: Boolean =
- getConf(SQLConf.PARQUET_READER_RESPECT_UNKNOWN_TYPE_ANNOTATION)
-
def ignoreMissingParquetFieldId: Boolean = getConf(SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID)
def legacyParquetNanosAsLong: Boolean = getConf(SQLConf.LEGACY_PARQUET_NANOS_AS_LONG)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
index 1fdb6be4b737..08e545cb8c20 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
@@ -148,9 +148,6 @@ class ParquetFileFormat
hadoopConf.setBoolean(
SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.key,
sqlConf.legacyParquetNanosAsLong)
- hadoopConf.setBoolean(
- SQLConf.PARQUET_READER_RESPECT_UNKNOWN_TYPE_ANNOTATION.key,
- sqlConf.parquetReaderRespectUnknownTypeAnnotation)
}
/**
@@ -449,9 +446,7 @@ object ParquetFileFormat extends Logging {
sqlConf.isParquetBinaryAsString,
sqlConf.isParquetINT96AsTimestamp,
inferTimestampNTZ = sqlConf.parquetInferTimestampNTZEnabled,
- nanosAsLong = sqlConf.legacyParquetNanosAsLong,
- respectUnknownTypeAnnotation =
- sqlConf.parquetReaderRespectUnknownTypeAnnotation)
+ nanosAsLong = sqlConf.legacyParquetNanosAsLong)
val seen = mutable.HashSet[String]()
val finalSchemas: Seq[StructType] = footers.flatMap { footer =>
@@ -550,8 +545,6 @@ object ParquetFileFormat extends Logging {
val assumeInt96IsTimestamp = sqlConf.isParquetINT96AsTimestamp
val inferTimestampNTZ = sqlConf.parquetInferTimestampNTZEnabled
val nanosAsLong = sqlConf.legacyParquetNanosAsLong
- val respectUnknownTypeAnnotation =
- sqlConf.parquetReaderRespectUnknownTypeAnnotation
val reader = (files: Seq[FileStatus], conf: Configuration, ignoreCorruptFiles: Boolean) => {
// Converter used to convert Parquet `MessageType` to Spark SQL `StructType`
@@ -559,8 +552,7 @@ object ParquetFileFormat extends Logging {
assumeBinaryIsString = assumeBinaryIsString,
assumeInt96IsTimestamp = assumeInt96IsTimestamp,
inferTimestampNTZ = inferTimestampNTZ,
- nanosAsLong = nanosAsLong,
- respectUnknownTypeAnnotation = respectUnknownTypeAnnotation)
+ nanosAsLong = nanosAsLong)
readParquetFootersInParallel(conf, files, ignoreCorruptFiles)
.map(ParquetFileFormat.readSchemaFromFooter(_, converter))
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala
index d5b81a616d56..9e6f4447ca79 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala
@@ -60,9 +60,7 @@ class ParquetToSparkSchemaConverter(
nanosAsLong: Boolean = SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.defaultValue.get,
useFieldId: Boolean = SQLConf.PARQUET_FIELD_ID_READ_ENABLED.defaultValue.get,
val ignoreVariantAnnotation: Boolean =
- SQLConf.PARQUET_IGNORE_VARIANT_ANNOTATION.defaultValue.get,
- val respectUnknownTypeAnnotation: Boolean =
- SQLConf.PARQUET_READER_RESPECT_UNKNOWN_TYPE_ANNOTATION.defaultValue.get) {
+ SQLConf.PARQUET_IGNORE_VARIANT_ANNOTATION.defaultValue.get) {
def this(conf: SQLConf) = this(
assumeBinaryIsString = conf.isParquetBinaryAsString,
@@ -71,9 +69,7 @@ class ParquetToSparkSchemaConverter(
inferTimestampNTZ = conf.parquetInferTimestampNTZEnabled,
nanosAsLong = conf.legacyParquetNanosAsLong,
useFieldId = conf.parquetFieldIdReadEnabled,
- ignoreVariantAnnotation = conf.parquetIgnoreVariantAnnotation,
- respectUnknownTypeAnnotation =
- conf.parquetReaderRespectUnknownTypeAnnotation)
+ ignoreVariantAnnotation = conf.parquetIgnoreVariantAnnotation)
def this(conf: Configuration) = this(
assumeBinaryIsString = conf.get(SQLConf.PARQUET_BINARY_AS_STRING.key).toBoolean,
@@ -84,10 +80,7 @@ class ParquetToSparkSchemaConverter(
useFieldId = conf.getBoolean(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key,
SQLConf.PARQUET_FIELD_ID_READ_ENABLED.defaultValue.get),
ignoreVariantAnnotation = conf.getBoolean(SQLConf.PARQUET_IGNORE_VARIANT_ANNOTATION.key,
- SQLConf.PARQUET_IGNORE_VARIANT_ANNOTATION.defaultValue.get),
- respectUnknownTypeAnnotation = conf.getBoolean(
- SQLConf.PARQUET_READER_RESPECT_UNKNOWN_TYPE_ANNOTATION.key,
- SQLConf.PARQUET_READER_RESPECT_UNKNOWN_TYPE_ANNOTATION.defaultValue.get))
+ SQLConf.PARQUET_IGNORE_VARIANT_ANNOTATION.defaultValue.get))
/**
* Converts Parquet [[MessageType]] `parquetSchema` to a Spark SQL [[StructType]].
@@ -232,11 +225,7 @@ class ParquetToSparkSchemaConverter(
primitiveColumn: PrimitiveColumnIO,
sparkReadType: Option[DataType] = None): ParquetColumn = {
val parquetType = primitiveColumn.getType.asPrimitiveType()
- val typeAnnotation = primitiveColumn.getType.getLogicalTypeAnnotation match {
- case unknown: UnknownLogicalTypeAnnotation =>
- if (respectUnknownTypeAnnotation) unknown else null
- case other => other
- }
+ val typeAnnotation = primitiveColumn.getType.getLogicalTypeAnnotation
val typeName = primitiveColumn.getPrimitive
def typeString =
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
index d0c7859964e0..5a427aad5f89 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
@@ -161,9 +161,6 @@ case class ParquetScan(
hadoopConf.setBoolean(
SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.key,
conf.legacyParquetNanosAsLong)
- hadoopConf.setBoolean(
- SQLConf.PARQUET_READER_RESPECT_UNKNOWN_TYPE_ANNOTATION.key,
- conf.parquetReaderRespectUnknownTypeAnnotation)
val broadcastedConf =
SerializableConfiguration.broadcast(sparkSession.sparkContext, hadoopConf)
diff --git a/sql/core/src/test/resources/test-data/void_in_parquet.parquet b/sql/core/src/test/resources/test-data/void_in_parquet.parquet
deleted file mode 100644
index d0c8db6dfc8c..000000000000
Binary files a/sql/core/src/test/resources/test-data/void_in_parquet.parquet and /dev/null differ
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala
index c0aa3dd22d44..3072657a0954 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala
@@ -765,24 +765,6 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSparkSession
}
}
- gridTest("Read external file with UNKNOWN type annotation")(
- Seq(true, false)
- ) { respectUnknown =>
- withSQLConf(
- SQLConf.PARQUET_READER_RESPECT_UNKNOWN_TYPE_ANNOTATION.key ->
- respectUnknown.toString
- ) {
- withAllParquetReaders {
- // Parquet file column void_col has physical type INT32 and logical type UNKNOWN and was
- // not written by Spark, so this goes through Parquet --> Spark type conversion.
- val df = readResourceParquetFile("test-data/void_in_parquet.parquet")
- val inferredType = df.schema("void_col").dataType
- val expected = if (respectUnknown) NullType else IntegerType
- assert(inferredType == expected)
- }
- }
- }
-
test("vectorized reader: missing all struct fields") {
for {
offheapEnabled <- Seq(true, false)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org