This is an automated email from the ASF dual-hosted git repository.
dtenedor pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 308de008520e [SPARK-58109][SQL][PROTOBUF] Use Hadoop file reader for
SQL from_protobuf descriptor files
308de008520e is described below
commit 308de008520e096fe6796451c20ddb6ec656b83e
Author: Kavpreet Grewal <[email protected]>
AuthorDate: Tue Jul 14 17:16:03 2026 -0700
[SPARK-58109][SQL][PROTOBUF] Use Hadoop file reader for SQL from_protobuf
descriptor files
### What changes were proposed in this pull request?
SQL `from_protobuf`/`to_protobuf` read their descriptor-file argument with
local NIO (`Files.readAllBytes`), so a descriptor stored on a distributed
filesystem (e.g. `hdfs:`, `s3:`, `abfss:`) fails even though the same path
works via the DataFrame API. This adds `ProtobufDescriptorFileReader`, which
routes the descriptor read through the Hadoop `FileSystem` so those paths
resolve, with a fallback to the existing local read for scheme-less paths. The
read runs on the driver at plan time.
### Why are the changes needed?
Added `ProtobufDescriptorFileReadSuite` (9 tests). Existing
`ProtobufFunctionsSuite` still passes.
### Does this PR introduce _any_ user-facing change?
It allows users to use the Hadoop file reader to read Protobuf descriptor
files, with a fallback to the current Java NIO reader.
### How was this patch tested?
Added `ProtobufDescriptorFileReadSuite` (9 tests). Existing
`ProtobufFunctionsSuite` still passes.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
Closes #57234 from kavpreetgrewal/SPARK-58109-protobuf-descriptor-hadoop-fs.
Authored-by: Kavpreet Grewal <[email protected]>
Signed-off-by: Daniel Tenedorio <[email protected]>
---
.../protobuf/ProtobufDescriptorFileReadSuite.scala | 167 +++++++++++++++++++++
.../expressions/toFromProtobufSqlFunctions.scala | 9 +-
.../util/ProtobufDescriptorFileReader.scala | 98 ++++++++++++
3 files changed, 270 insertions(+), 4 deletions(-)
diff --git
a/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufDescriptorFileReadSuite.scala
b/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufDescriptorFileReadSuite.scala
new file mode 100644
index 000000000000..58b47d820fcb
--- /dev/null
+++
b/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufDescriptorFileReadSuite.scala
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.protobuf
+
+import java.io.{File, IOException}
+import java.net.URI
+import java.nio.file.Files
+
+import org.apache.hadoop.fs.{Path, RawLocalFileSystem}
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.util.ProtobufDescriptorFileReader
+import org.apache.spark.sql.protobuf.utils.{ProtobufUtils =>
ConnectorProtobufUtils}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Regression tests for reading the SQL from_protobuf/to_protobuf
descriptor-file argument via the
+ * Hadoop FileSystem (SPARK-58109). A `file:`-scheme path exercises the Hadoop
path; a scheme-less
+ * path exercises the local fallback. Distributed filesystems
(hdfs:/s3:/abfss:) can't run in a
+ * hermetic test, but all schemes flow through the same
`path.getFileSystem(conf).open` call.
+ */
+class ProtobufDescriptorFileReadSuite extends SharedSparkSession with
ProtobufTestBase {
+
+ // Register a `mockfs://` FileSystem whose open() always throws, so tests
can exercise the
+ // reader's failure handling for an explicit-scheme path. Registered via the
environment
+ // SparkConf because the reader resolves its Hadoop conf from
+ // `SparkHadoopUtil.get.newConfiguration(SparkEnv.get.conf)`, which does not
observe post-startup
+ // mutations of the runtime Hadoop configuration.
+ override protected def sparkConf: SparkConf =
+ super.sparkConf.set("spark.hadoop.fs.mockfs.impl",
classOf[ThrowingMockFileSystem].getName)
+
+ private val descPath: String = protobufDescriptorFile("functions_suite.desc")
+ private val messageName = "SimpleMessageJavaTypes"
+
+ private def runFromProtobuf(pathArg: String): Boolean = {
+ spark.range(1).selectExpr("CAST(NULL AS BINARY) AS
value").createOrReplaceTempView("t_pb")
+ spark.sql(
+ s"SELECT from_protobuf(value, '$messageName', '$pathArg', map()) IS NULL
AS r FROM t_pb")
+ .head().getBoolean(0)
+ }
+
+ test("from_protobuf reads a file: scheme descriptor path") {
+ val fileUri = new File(descPath).getCanonicalFile.toURI.toString
+ assert(fileUri.startsWith("file:"))
+ assert(runFromProtobuf(fileUri))
+ }
+
+ test("from_protobuf reads a scheme-less local descriptor path") {
+ val local = new File(descPath).getCanonicalPath
+ assert(new File(local).exists())
+ assert(runFromProtobuf(local))
+ }
+
+ test("to_protobuf reads a file: scheme descriptor path") {
+ val fileUri = new File(descPath).getCanonicalFile.toURI.toString
+ spark.range(1)
+ .selectExpr("named_struct('id', id) AS value")
+ .createOrReplaceTempView("t_pb_struct")
+ // Proves to_protobuf reads the descriptor via a file: URI and returns
non-null encoded bytes.
+ val r = spark.sql(
+ s"SELECT to_protobuf(value, '$messageName', '$fileUri', map()) IS NOT
NULL AS r " +
+ "FROM t_pb_struct").head().getBoolean(0)
+ assert(r)
+ }
+
+ test("missing descriptor file raises PROTOBUF_DESCRIPTOR_FILE_NOT_FOUND") {
+ val missing = new File(Files.createTempDirectory("pb").toFile,
"does-not-exist.desc")
+ .getCanonicalFile.toURI.toString
+ checkError(
+ exception = intercept[AnalysisException](runFromProtobuf(missing)),
+ condition = "PROTOBUF_DESCRIPTOR_FILE_NOT_FOUND",
+ parameters = Map("filePath" -> missing))
+ }
+
+ test("to_protobuf with an empty descriptor path raises
CANNOT_PARSE_PROTOBUF_DESCRIPTOR") {
+ // An empty path (new Path("") throws) must surface as
CANNOT_PARSE_PROTOBUF_DESCRIPTOR,
+ // not a raw internal error. to_protobuf, unlike from_protobuf, has no
empty-path guard.
+ spark.range(1)
+ .selectExpr("named_struct('id', id) AS value")
+ .createOrReplaceTempView("t_pb_struct_empty")
+ checkError(
+ exception = intercept[org.apache.spark.sql.AnalysisException] {
+ spark.sql(
+ s"SELECT to_protobuf(value, '$messageName', '', map()) FROM
t_pb_struct_empty"
+ ).head()
+ },
+ condition = "CANNOT_PARSE_PROTOBUF_DESCRIPTOR")
+ }
+
+ test("invalid descriptor bytes raise CANNOT_PARSE_PROTOBUF_DESCRIPTOR") {
+ // A .proto source file is not a serialized FileDescriptorSet, so
buildDescriptor rejects it.
+ // Asserted at the buildDescriptor layer (not via SQL) because the SQL
path defers the parse to
+ // execution, where it surfaces as SparkException rather than
AnalysisException.
+ val bogusBytes =
+ Files.readAllBytes(new
File("src/test/resources/protobuf/catalyst_types.proto").toPath)
+ checkError(
+ exception = intercept[AnalysisException](
+ ConnectorProtobufUtils.buildDescriptor("SomeMessage",
Some(bogusBytes))),
+ condition = "CANNOT_PARSE_PROTOBUF_DESCRIPTOR")
+ }
+
+ test("invalid descriptor bytes read via a file: path raise CANNOT_PARSE
through SQL") {
+ // Complements the buildDescriptor-layer test above: this drives the full
SQL path with a real
+ // file: descriptor path, so the bytes are actually read by the reader and
then handed to
+ // buildDescriptor -- which rejects the non-FileDescriptorSet bytes with
+ // CANNOT_PARSE_PROTOBUF_DESCRIPTOR. This proves the Hadoop-read bytes
reach buildDescriptor.
+ // A non-NULL input value is used so the descriptor is actually evaluated
(with NULL input the
+ // descriptor-building lazy val is never forced and the error would not
surface).
+ val bogus = new File("src/test/resources/protobuf/catalyst_types.proto")
+ .getCanonicalFile.toURI.toString
+ spark.range(1).selectExpr("CAST(X'00' AS BINARY) AS
value").createOrReplaceTempView("t_pb_bad")
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.sql(
+ s"SELECT from_protobuf(value, '$messageName', '$bogus', map()) FROM
t_pb_bad").collect()
+ },
+ condition = "CANNOT_PARSE_PROTOBUF_DESCRIPTOR")
+ }
+
+ test("reader routes a missing file: path to
PROTOBUF_DESCRIPTOR_FILE_NOT_FOUND") {
+ // Direct unit test of the reader's FileNotFoundException routing for an
explicit-scheme path
+ // (no fallback: an explicit scheme is not eligible for the local read).
+ val missing = new File(Files.createTempDirectory("pb").toFile, "nope.desc")
+ .getCanonicalFile.toURI.toString
+ checkError(
+ exception = intercept[AnalysisException](
+ ProtobufDescriptorFileReader.readDescriptorFileContent(missing)),
+ condition = "PROTOBUF_DESCRIPTOR_FILE_NOT_FOUND",
+ parameters = Map("filePath" -> missing))
+ }
+
+ test("reader does NOT fall back for an explicit-scheme path; the real error
is chained") {
+ // The mockfs FileSystem (registered via sparkConf) throws on open. A
mockfs:// path is
+ // explicit-scheme, so the reader must NOT retry a local NIO read of the
bogus path; it surfaces
+ // CANNOT_PARSE_PROTOBUF_DESCRIPTOR and preserves the underlying
IOException as the cause.
+ val ex = intercept[AnalysisException] {
+
ProtobufDescriptorFileReader.readDescriptorFileContent(s"mockfs://$descPath")
+ }
+ checkError(exception = ex, condition = "CANNOT_PARSE_PROTOBUF_DESCRIPTOR")
+ // The underlying FileSystem exception must be preserved as the cause (not
swallowed).
+ assert(ex.getCause != null)
+ }
+}
+
+/** A FileSystem for the `mockfs` scheme whose `open` always throws, to
exercise read failures. */
+class ThrowingMockFileSystem extends RawLocalFileSystem {
+ override def getScheme: String = "mockfs"
+ override def getUri: URI = URI.create("mockfs:///")
+ override def open(f: Path, bufferSize: Int):
org.apache.hadoop.fs.FSDataInputStream =
+ throw new IOException("mockfs open failure")
+}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/toFromProtobufSqlFunctions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/toFromProtobufSqlFunctions.scala
index b93b389ae23b..7d0fa35b3f0f 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/toFromProtobufSqlFunctions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/toFromProtobufSqlFunctions.scala
@@ -19,11 +19,10 @@ package org.apache.spark.sql.catalyst.expressions
import org.apache.spark.sql.catalyst.analysis.FunctionRegistry
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
-import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData,
ProtobufDescriptorFileReader}
import org.apache.spark.sql.errors.DataTypeErrors.toSQLType
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.types.{BinaryType, MapType, NullType, StringType,
StructType}
-import org.apache.spark.sql.util.ProtobufUtils
import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.Utils
@@ -142,7 +141,8 @@ case class FromProtobuf(
}
val descFilePathValue: Option[Array[Byte]] = descFilePath.eval() match {
case s: UTF8String if s.toString.isEmpty => None
- case s: UTF8String =>
Some(ProtobufUtils.readDescriptorFileContent(s.toString))
+ case s: UTF8String =>
+
Some(ProtobufDescriptorFileReader.readDescriptorFileContent(s.toString))
case bytes: Array[Byte] if bytes.isEmpty => None
case bytes: Array[Byte] => Some(bytes)
case null => None
@@ -292,7 +292,8 @@ case class ToProtobuf(
s.toString
}
val descFilePathValue: Option[Array[Byte]] = descFilePath.eval() match {
- case s: UTF8String =>
Some(ProtobufUtils.readDescriptorFileContent(s.toString))
+ case s: UTF8String =>
+
Some(ProtobufDescriptorFileReader.readDescriptorFileContent(s.toString))
case bytes: Array[Byte] => Some(bytes)
case null => None
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ProtobufDescriptorFileReader.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ProtobufDescriptorFileReader.scala
new file mode 100644
index 000000000000..a522806f4f46
--- /dev/null
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ProtobufDescriptorFileReader.scala
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import java.io.FileNotFoundException
+
+import scala.util.control.NonFatal
+
+import com.google.common.io.ByteStreams
+import org.apache.hadoop.fs.Path
+
+import org.apache.spark.SparkEnv
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.PATH
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.util.{ProtobufUtils => ApiProtobufUtils}
+import org.apache.spark.util.Utils
+
+/**
+ * Reads a Protobuf descriptor file for the SQL `from_protobuf` /
`to_protobuf` expressions via the
+ * Hadoop `FileSystem`, so paths on distributed file systems (e.g. `hdfs:`,
`s3:`, `abfss:`)
+ * resolve in addition to local paths. Runs on the driver at plan time.
+ */
+object ProtobufDescriptorFileReader extends Logging {
+
+ /**
+ * Reads and returns the raw bytes of the descriptor file at `filePath`.
+ *
+ * The path is read through the Hadoop `FileSystem` resolved for its scheme,
so paths on
+ * distributed file systems work. A scheme-less path that fails the Hadoop
read falls back to a
+ * local-NIO read to preserve pre-existing behavior.
+ *
+ * Must be called on the driver: it relies on the active `SparkEnv`
(`SparkEnv.get`), which is
+ * unavailable on executors.
+ *
+ * @param filePath the descriptor file path (a cloud URI, or a local path)
+ * @return the descriptor file contents as a byte array
+ * @throws org.apache.spark.sql.AnalysisException with condition
+ * `PROTOBUF_DESCRIPTOR_FILE_NOT_FOUND` if the file does not exist,
or
+ * `CANNOT_PARSE_PROTOBUF_DESCRIPTOR` if it cannot otherwise be read
+ */
+ def readDescriptorFileContent(filePath: String): Array[Byte] = {
+ // A scheme-less path (including the malformed empty string, where `new
Path` throws) may be a
+ // local file the old implementation could read, so it is eligible for the
local fallback; a
+ // path with an explicit scheme is not.
+ val schemeless = try {
+ new Path(filePath).toUri.getScheme == null
+ } catch {
+ case NonFatal(_) => false
+ }
+ try {
+ val path = new Path(filePath)
+ val fs =
path.getFileSystem(SparkHadoopUtil.get.newConfiguration(SparkEnv.get.conf))
+ Utils.tryWithResource(fs.open(path))(ByteStreams.toByteArray)
+ } catch {
+ case NonFatal(hadoopError) if schemeless =>
+ // The Hadoop route failed for a scheme-less path. Fall back to the
local-NIO read.
+ // If the local read also fails, throw its error with the Hadoop error
attached as a
+ // suppressed exception, so neither cause is lost.
+ logWarning(log"Failed to read Protobuf descriptor ${MDC(PATH,
filePath)} via the Hadoop " +
+ log"FileSystem; falling back to a local filesystem read.",
hadoopError)
+ try {
+ ApiProtobufUtils.readDescriptorFileContent(filePath)
+ } catch {
+ case NonFatal(localError) =>
+ localError.addSuppressed(hadoopError)
+ throw localError
+ }
+ case e: FileNotFoundException =>
+ // Explicit-scheme path that does not exist: surface the not-found
error class.
+ logWarning(log"Protobuf descriptor file ${MDC(PATH, filePath)} was not
found.", e)
+ throw QueryCompilationErrors.cannotFindDescriptorFileError(filePath, e)
+ case NonFatal(e) =>
+ // Explicit-scheme path that failed for another reason (e.g.
permission denied, IO error).
+ // Do not fall back -- the local read could not serve this path either
+ logWarning(
+ log"Failed to read Protobuf descriptor ${MDC(PATH, filePath)} via
the Hadoop FileSystem.",
+ e)
+ throw QueryCompilationErrors.descriptorParseError(e)
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
