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