HyukjinKwon commented on code in PR #57234:
URL: https://github.com/apache/spark/pull/57234#discussion_r3583568107
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/toFromProtobufSqlFunctions.scala:
##########
@@ -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))
Review Comment:
The PR description says "the same path works via the DataFrame API", but the
DataFrame `from_protobuf`/`to_protobuf(..., descFilePath)` overloads call the
same local-NIO `ProtobufUtils.readDescriptorFileContent`
(`sql/api/src/main/scala/org/apache/spark/sql/protobuf/functions.scala:50,97,225`),
so they also fail on `hdfs:`/`s3:`/`abfss:`. After this PR the SQL functions
support distributed descriptor paths but the DataFrame API still does not — a
SQL-vs-DataFrame consistency gap, and the stated motivation is inaccurate.
Either extend the fix to the DataFrame path (ideally in the shared `sql/api`
reader, if it can be made Hadoop-aware without the layering issue) or correct
the description to note the DataFrame API is also affected and out of scope.
Non-blocking — scoping to SQL is a legitimate choice — but the description
should not claim parity that doesn't exist.
##########
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
Review Comment:
This comment says the malformed empty string "(where `new Path` throws) ...
is eligible for the local fallback", but the detection below sets `schemeless =
false` when `new Path` throws (`case NonFatal(_) => false`), so an
empty/malformed path is **not** eligible for the fallback — it lands in the
`case NonFatal(e)` branch and surfaces `CANNOT_PARSE_PROTOBUF_DESCRIPTOR`.
The behavior is fine (it matches the old reader, and the `to_protobuf`
empty-path test asserts `CANNOT_PARSE_PROTOBUF_DESCRIPTOR`), but the comment
misdescribes the code. Either reword the comment to drop the "eligible for the
local fallback" claim for the throwing case, or — if fallback for malformed
paths was actually intended — return `true` on throw.
##########
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)
Review Comment:
This builds the Hadoop conf from `SparkEnv.get.conf`, which picks up static
`spark.hadoop.*` but not session-level overrides applied by
`sessionState.newHadoopConf()` (`SessionState.scala:119`) — the path the
DataFrame-side `ProtobufOptions` uses (`ProtobufOptions.scala:263-264`). A
session that sets `spark.hadoop.fs.*` at runtime (e.g. per-session cloud
credentials) would have those silently ignored for descriptor reads.
Since `sql/catalyst` can't reach `SparkSession`, this may be an accepted
limitation. Could you confirm whether session-level Hadoop config needs to be
honored here? If it does, resolving the conf at the call site (where a session
is reachable) and passing it in would close the gap; if not, a one-line note on
the method would save the next reader the trace.
##########
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))
Review Comment:
For a scheme-less path, `path.getFileSystem(conf)` resolves against
`fs.defaultFS`. On a cluster whose default FS is not local, a bare path like
`/tmp/x.desc` first attempts a read on the default FS: normally that throws and
falls back to local (with a WARN on every such read), but if a file
coincidentally exists at that path on the default FS, the Hadoop read
*succeeds* and the reader returns the wrong bytes with no fallback.
This may well be intended — resolving scheme-less paths against
`fs.defaultFS` is standard Spark path behavior, so it arguably makes descriptor
paths more consistent with the rest of Spark. Worth confirming the tradeoff is
deliberate and calling it out under user-facing changes, since it's a behavior
change vs. the always-local old reader.
##########
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") {
Review Comment:
This scheme-less test succeeds through the Hadoop `LocalFileSystem` (the
test `fs.defaultFS` is `file:///`), not through the local-NIO fallback; the
`mockfs` test uses an explicit scheme, where fallback is disabled by design. So
the fallback branch (`ProtobufDescriptorFileReader.scala:79-85`) and its
suppressed-exception chaining are never actually exercised.
Consider a test that forces the scheme-less Hadoop read to fail while a
local file is readable (e.g. a mock default FS whose `open` throws on a
scheme-less path), asserting the bytes still load via the NIO fallback and, on
a double failure, that the Hadoop error is chained as suppressed.
--
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]
