peter-toth commented on code in PR #58614:
URL: https://github.com/apache/spark/pull/58614#discussion_r3995617103
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -1255,6 +1255,66 @@ abstract class AvroSuite
assertExceptionMsg[FileNotFoundException](e, "File not_exists.avsc does
not exist")
}
+ // spark.sql.avro.schemaUrlAllowedSchemes is a static SQL config, so it
cannot be set with
+ // withSQLConf; these drive AvroOptions directly under a SQLConf provided
via withExistingConf.
+ // A scheme-less local path resolves to the default file system ("file").
Review Comment:
**Finding 9.** This says a scheme-less local path resolves to the default
file system, but `testFile` is
`Thread.currentThread().getContextClassLoader.getResource(fileName).toString`
(`sql/core/src/test/scala/org/apache/spark/sql/QueryTest.scala:508`), so
`avroSchemaUrl` here is `file:/.../test_sub.avsc`. Its `getScheme` is `"file"`,
and `Option(uri.getScheme)` is never `None` in any test on this PR.
So the `getOrElse(FileSystem.getDefaultUri(conf).getScheme)` at
`sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:92` has no
coverage. I replaced it with a constant and all seven new tests still passed:
the four here and the three in `AvroSchemaUrlAllowlistSuite`.
The same claim is at `AvroSuite.scala:4281` and in the description
("including a scheme-less local path that resolves to `file`"). It came from my
round-2 comment at
[r3987277368](https://github.com/apache/spark/pull/58614#discussion_r3987277368),
where I said every test URL is scheme-less. That was wrong. The scheme-side
fold is pinned correctly now, but not for the reason I gave.
This case covers the branch. I ran it both ways: green on this head, red
with the fallback replaced.
```scala
test("SPARK-59329: avroSchemaUrl allowlist resolves a scheme-less path " +
"against the default file system") {
// testFile returns a "file:" URL, so strip the scheme to get a
genuinely scheme-less path.
val avroSchemaUrl = new URI(testFile("test_sub.avsc")).getPath
assert(new URI(avroSchemaUrl).getScheme == null)
val hadoopConf = spark.sessionState.newHadoopConf()
// The default file system is "file", so allowing "file" permits the
scheme-less path ...
val allowed = new SQLConf()
allowed.setConf(SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES, Seq("file"))
SQLConf.withExistingConf(allowed) {
assert(new AvroOptions(Map("avroSchemaUrl" -> avroSchemaUrl),
hadoopConf).schema.isDefined)
}
// ... and an allowlist without it rejects the same path.
val disallowed = new SQLConf()
disallowed.setConf(SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES, Seq("s3a"))
SQLConf.withExistingConf(disallowed) {
val e = intercept[AnalysisException] {
new AvroOptions(Map("avroSchemaUrl" -> avroSchemaUrl), hadoopConf)
}
assert(e.getMessage.contains("The scheme 'file'"))
}
}
```
##########
sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:
##########
@@ -76,7 +76,29 @@ private[sql] class AvroOptions(
parameters.get(AVRO_SCHEMA).map(AvroUtils.parseAvroSchema).orElse({
val avroUrlSchema = parameters.get(AVRO_SCHEMA_URL).map(url => {
log.debug("loading avro schema from url: " + url)
- val fs = FileSystem.get(new URI(url), conf)
+ val uri = new URI(url)
+ // Optional operator-configured allowlist of URI schemes for
avroSchemaUrl. Empty by
+ // default, which permits any scheme and leaves the file-system
resolution below unchanged.
+ // When set, the scheme is resolved and checked before the file system
for the URL is
+ // instantiated, so a disallowed scheme is rejected with a clear error
rather than a
+ // lower-level failure while opening it. A scheme-less URL takes the
default file system's
+ // scheme, so it can be permitted by allowing that scheme.
+ val allowedSchemes =
SQLConf.get.getConf(SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES)
+ .map(_.toLowerCase(Locale.ROOT))
+ if (allowedSchemes.nonEmpty) {
+ // FileSystem.getDefaultUri always carries a scheme (it throws
otherwise), so a
+ // scheme-less URL resolves to the default file system's scheme.
+ val scheme = Option(uri.getScheme)
+ .getOrElse(FileSystem.getDefaultUri(conf).getScheme)
+ .toLowerCase(Locale.ROOT)
+ if (!allowedSchemes.contains(scheme)) {
+ throw QueryCompilationErrors.avroOptionsException(
+ AVRO_SCHEMA_URL,
+ s"The scheme '$scheme' of avroSchemaUrl '$url' is not in the
allowlist " +
+ s"configured by
${SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES.key}.")
Review Comment:
**Finding 10.** The message names the key but not what it parsed to, so the
mistake an operator is most likely to make reads as a contradiction.
`spark.sql.avro.schemaUrlAllowedSchemes=file://` gives `Seq("file://")`, which
matches nothing, and every read then fails with `The scheme 'file' of
avroSchemaUrl '...' is not in the allowlist configured by
spark.sql.avro.schemaUrlAllowedSchemes` while `file` is exactly what they wrote.
```suggestion
s"${allowedSchemes.mkString("[", ", ", "]")} configured by "
+
s"${SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES.key}.")
```
That reads `... is not in the allowlist [file://] configured by ...`. Both
existing assertions still match; I ran the four `AvroSuite` cases and
`AvroSchemaUrlAllowlistSuite` with it.
The other route is #58613's: reject a malformed entry up front with
`checkValue`. Either works.
--
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]
