peter-toth commented on code in PR #58615:
URL: https://github.com/apache/spark/pull/58615#discussion_r3987125426
##########
sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala:
##########
@@ -200,6 +200,24 @@ private[spark] object HiveUtils extends Logging {
.booleanConf
.createWithDefault(false)
+ val INITIALIZE_METASTORE_FORMAT_CLASSES =
+ buildConf("spark.sql.hive.initializeMetastoreFormatClasses")
+ .doc("When true, an InputFormat/OutputFormat class name stored in the
Hive metastore is " +
+ "resolved with its static initializer run at resolution time. When
false, the class is " +
+ "resolved without running its static initializer, which then runs
later when the format " +
+ "is instantiated for a scan/write. Only the class name is needed when
converting " +
+ "metastore metadata, so setting this to false avoids running a format
class's static " +
+ "initializer during a metadata operation. Note that the conversion
also runs when " +
+ "planning a scan or write and when inferring the schema of a Hive
serde table, so with " +
+ "false a format class whose static initializer fails no longer fails
fast on the driver " +
+ "at resolution time; the failure instead surfaces later on an executor
when the format " +
+ "is instantiated (as NoClassDefFoundError: Could not initialize class
...). Keep this " +
+ "true (the default) to preserve the fail-fast behavior.")
Review Comment:
**Finding 4.** This sentence is mine from round 1 and it is wrong. For a
Hive serde scan the InputFormat is instantiated on the driver, not on an
executor.
`HadoopRDD.getPartitions` calls `getInputFormat(jobConf)`, which is
`ReflectionUtils.newInstance`
(`core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala:217`).
`NewHadoopRDD.getPartitions` calls
`inputFormatClass.getConstructor().newInstance()`
(`core/src/main/scala/org/apache/spark/rdd/NewHadoopRDD.scala:149`). Both run
on the driver, when the job is submitted. So with `false` a broken static
initializer still fails on the driver for a scan. What moves is the timing,
from analysis or DDL to split computation.
The write side does reach an executor. `HiveFileFormat.prepareWrite` only
puts the class name into the conf
(`sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala:78`),
and `jobConf.value.getOutputFormat` at `:103` instantiates it inside the
`OutputWriterFactory`, on the task side.
Two more things worth saying while the doc is being rewritten:
- `false` still loads the class, so a missing class still throws
`ClassNotFoundException` right here. Only the static initializer is deferred.
As written, a reader can take `false` to mean the resolution cannot fail at all.
- With `convertMetastoreParquet` / `convertMetastoreOrc` the format is never
instantiated, so the initializer never runs anywhere. That is the case where
`false` actually buys something, and the doc does not mention it.
```suggestion
.doc("When true, an InputFormat/OutputFormat class name stored in the
Hive metastore is " +
"resolved with its static initializer run at resolution time. When
false, the class is " +
"still loaded, so a missing class still fails here, but its static
initializer is not " +
"run. Only the class name is needed when converting metastore
metadata, so setting " +
"this to false avoids running a format class's static initializer
during a metadata " +
"operation. Note that the conversion also runs when planning a scan
or write and when " +
"inferring the schema of a Hive serde table, so with false a format
class whose static " +
"initializer fails no longer fails fast at resolution time. It fails
when the format " +
"is instantiated instead (as NoClassDefFoundError: Could not
initialize class ...), " +
"which for a Hive serde scan is on the driver when the input splits
are computed, and " +
"for a Hive serde write is on an executor. A table read through the
built-in " +
"Parquet/ORC reader never instantiates the format, so there the
initializer never " +
"runs. Keep this true (the default) to preserve the fail-fast
behavior.")
```
##########
sql/hive/src/test/scala/org/apache/spark/sql/hive/client/HiveClientImplSuite.scala:
##########
@@ -20,9 +20,41 @@ package org.apache.spark.sql.hive.client
import org.apache.hadoop.hive.metastore.api.FieldSchema
import org.apache.spark.{SparkFunSuite, SparkUnsupportedOperationException}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, CatalogTableType}
+import org.apache.spark.sql.hive.{HiveUtils, StaticInitFlags,
StaticInitInputFormat}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
class HiveClientImplSuite extends SparkFunSuite {
+ test("SPARK-59330: toHiveTable skips the format class static initializer
when " +
+ "spark.sql.hive.initializeMetastoreFormatClasses is false") {
+ val table = CatalogTable(
+ identifier = TableIdentifier("t", Some("default")),
+ tableType = CatalogTableType.MANAGED,
+ storage = CatalogStorageFormat.empty.copy(
+ inputFormat = Some(classOf[StaticInitInputFormat].getName)),
+ schema = new StructType().add("a", "int"))
+
+ def toHiveTableWith(initialize: Boolean): Unit = {
+ val conf = new SQLConf()
+ conf.setConf(HiveUtils.INITIALIZE_METASTORE_FORMAT_CLASSES, initialize)
+ SQLConf.withExistingConf(conf) {
+ HiveClientImpl.toHiveTable(table)
+ }
+ }
+
+ // The false half must run first: class initialization is one-way per JVM.
Resolving the
+ // format class name without initializing it must not run the static
initializer.
+ toHiveTableWith(initialize = false)
+ assert(!StaticInitFlags.inputFormatInitialized)
+
+ // With initialization enabled (the default), resolving the class name
runs the initializer.
+ toHiveTableWith(initialize = true)
+ assert(StaticInitFlags.inputFormatInitialized)
Review Comment:
**Finding 5.** `toOutputFormat` took the same `initialize =
initializeFormatClasses` change at
`sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala:1180`,
and nothing runs it. The two call sites are independent, so an edit to one of
them is unguarded.
Same shape as the input format, sketch below is untested. Add
`outputFormatInitialized` to `StaticInitFlags`, add a `StaticInitOutputFormat
implements org.apache.hadoop.hive.ql.io.HiveOutputFormat<Void, Void>` whose
static block flips it and whose `getHiveRecordWriter` / `getRecordWriter` /
`checkOutputSpecs` throw `UnsupportedOperationException`, then put it on the
same storage format:
```scala
storage = CatalogStorageFormat.empty.copy(
inputFormat = Some(classOf[StaticInitInputFormat].getName),
outputFormat = Some(classOf[StaticInitOutputFormat].getName)),
```
and assert both flags in each half.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
