HyukjinKwon commented on code in PR #57234: URL: https://github.com/apache/spark/pull/57234#discussion_r3583568120
########## 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. ########## 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. ########## 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. -- 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]
