You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2026/09/14 23:46:00 UTC
(spark) branch branch-4.3 updated: [SPARK-59329][SQL] Add an allowlist of URI schemes for the Avro schemaUrl option
This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new c025759c4c95 [SPARK-59329][SQL] Add an allowlist of URI schemes for the Avro schemaUrl option
c025759c4c95 is described below
commit c025759c4c95ef4f14494e32ea8b1affd326bec9
Author: Hyukjin Kwon <gu...@apache.org>
AuthorDate: Tue Sep 15 08:41:22 2026 +0900
[SPARK-59329][SQL] Add an allowlist of URI schemes for the Avro schemaUrl option
Add an opt-in `spark.sql.avro.schemaUrlAllowedSchemes` (internal, default empty = no
restriction). When non-empty, an `avroSchemaUrl` whose resolved URI scheme is not in the
allowlist is rejected before the schema is opened. A scheme-less path is resolved against the
default file system before its scheme is checked, so it can be permitted by allowing that file
system's scheme.
The config is a **static** SQL config (`buildStaticConf`), declared in `StaticSQLConf` alongside
the other SQL static configs: it is fixed when the `SparkSession` is created and cannot be changed
at runtime. It restricts the scheme an `avroSchemaUrl` may name; it does not restrict which file
system serves that scheme (that is decided by `fs.<scheme>.impl`, which a session can still set).
Gives operators an optional control over which URI schemes are accepted for the `avroSchemaUrl`
option. Empty by default, so there is no behavior change unless configured.
Scope: this targets the Avro `avroSchemaUrl` option specifically. Other options that open a
user-supplied URL and have the same shape (for example XML's `rowValidationXSDPath`) are not in
scope for this PR, and there is no shared helper yet.
No by default. When `spark.sql.avro.schemaUrlAllowedSchemes` is set, an `avroSchemaUrl` whose
scheme is not listed is rejected. Because the config is static, it must be set when the
`SparkSession` is created (for example with `--conf`) and cannot be set at runtime.
New tests in `connector/avro`:
- `AvroSchemaUrlAllowlistSuite` sets the static allowlist on the session via `sparkConf` and
drives a real `spark.read ... load()`: an allowed scheme is permitted, a disallowed scheme is
rejected, and a session cannot relax the allowlist at runtime (`CANNOT_MODIFY_STATIC_CONFIG`).
- `AvroSuite` cases drive `AvroOptions` directly under a provided `SQLConf`: an allowed `file:`
scheme is permitted, a disallowed scheme is rejected before the file system is opened, a
genuinely scheme-less path resolves against the default file system (permitted when `file` is
allowed, rejected otherwise), the empty default imposes no restriction, the scheme/allowlist
comparison is case-insensitive on both sides, and the rejection message echoes the parsed
allowlist.
Generated-by: Claude Opus 4.8
This pull request and its description were written by Isaac.
Closes #58614 from HyukjinKwon/SPARK-59329-avro-schema-url-schemes.
Authored-by: Hyukjin Kwon <gu...@apache.org>
Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
(cherry picked from commit 5839ba0b095d10eb86438ace057691024cb9d074)
Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
---
.../org/apache/spark/sql/avro/AvroSuite.scala | 147 +++++++++++++++++++++
.../apache/spark/sql/internal/StaticSQLConf.scala | 16 +++
.../org/apache/spark/sql/avro/AvroOptions.scala | 29 +++-
3 files changed, 189 insertions(+), 3 deletions(-)
diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
index e7214519577c..e645fcad3911 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
+++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
@@ -49,6 +49,7 @@ import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.LegacyBehaviorPolicy
import org.apache.spark.sql.internal.LegacyBehaviorPolicy._
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.StaticSQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
import org.apache.spark.sql.v2.avro.AvroScan
@@ -1254,6 +1255,108 @@ 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.
+ // testFile returns a "file:" URL, so its scheme is an explicit "file"; the scheme-less path that
+ // resolves against the default file system is covered by its own test below.
+ test("SPARK-59329: avroSchemaUrl scheme allowlist permits an allowed scheme") {
+ val avroSchemaUrl = testFile("test_sub.avsc")
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ val conf = new SQLConf()
+ conf.setConf(StaticSQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES, Seq("file"))
+ SQLConf.withExistingConf(conf) {
+ val options = new AvroOptions(Map("avroSchemaUrl" -> avroSchemaUrl), hadoopConf)
+ assert(options.schema.isDefined)
+ }
+ }
+
+ test("SPARK-59329: avroSchemaUrl allowlist rejects a disallowed scheme " +
+ "before opening the file system") {
+ // An explicit non-"file" scheme is rejected by the allowlist check, which runs before the
+ // file system for the URL is instantiated -- so this surfaces the clean allowlist error
+ // rather than a lower-level failure from trying to load the s3a file system. The URL uses an
+ // upper-case "S3A" scheme so the lower-case "s3a" in the message pins the scheme-side case
+ // folding: dropping the fold on the scheme leaves no lower-case "s3a" in the message.
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ val conf = new SQLConf()
+ conf.setConf(StaticSQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES, Seq("file"))
+ SQLConf.withExistingConf(conf) {
+ val e = intercept[AnalysisException] {
+ new AvroOptions(Map("avroSchemaUrl" -> "S3A://bucket/user.avsc"), hadoopConf)
+ }
+ assert(e.getCondition == "STDS_INVALID_OPTION_VALUE.WITH_MESSAGE")
+ assert(e.getMessage.contains("avroSchemaUrl"))
+ assert(e.getMessage.contains("not in the allowlist"))
+ assert(e.getMessage.contains("The scheme 's3a'"))
+ }
+ }
+
+ test("SPARK-59329: avroSchemaUrl scheme allowlist is disabled by default") {
+ val avroSchemaUrl = testFile("test_sub.avsc")
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ // The empty default skips the scheme check entirely, preserving the previous behavior.
+ val conf = new SQLConf()
+ SQLConf.withExistingConf(conf) {
+ assert(conf.getConf(StaticSQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES).isEmpty)
+ val options = new AvroOptions(Map("avroSchemaUrl" -> avroSchemaUrl), hadoopConf)
+ assert(options.schema.isDefined)
+ }
+ }
+
+ test("SPARK-59329: avroSchemaUrl scheme allowlist is case-insensitive") {
+ val avroSchemaUrl = testFile("test_sub.avsc")
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ // Allowlist entries and the URL scheme are compared case-insensitively: an upper-case "FILE"
+ // entry still permits the "file" scheme. Dropping the allowlist's case folding fails this.
+ val conf = new SQLConf()
+ conf.setConf(StaticSQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES, Seq("FILE"))
+ SQLConf.withExistingConf(conf) {
+ val options = new AvroOptions(Map("avroSchemaUrl" -> avroSchemaUrl), hadoopConf)
+ assert(options.schema.isDefined)
+ }
+ }
+
+ 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.
+ // This exercises the FileSystem.getDefaultUri fallback that the other cases do not reach.
+ 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(StaticSQLConf.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, reporting the resolved "file" scheme.
+ val disallowed = new SQLConf()
+ disallowed.setConf(StaticSQLConf.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'"))
+ }
+ }
+
+ test("SPARK-59329: the allowlist rejection echoes the parsed allowlist") {
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ // "file://" is the shape an operator is most likely to write by mistake: it parses to one
+ // entry ("file://", not "file") that matches nothing, so every read then fails with a scheme
+ // that looks like it should be allowed. Echoing what the config parsed to is what makes the
+ // message readable, so pin it: dropping the parsed allowlist from the message fails here.
+ val conf = new SQLConf()
+ conf.setConf(StaticSQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES, Seq("file://"))
+ SQLConf.withExistingConf(conf) {
+ val e = intercept[AnalysisException] {
+ new AvroOptions(Map("avroSchemaUrl" -> testFile("test_sub.avsc")), hadoopConf)
+ }
+ assert(e.getMessage.contains("The scheme 'file'"))
+ assert(e.getMessage.contains("not in the allowlist [file://]"))
+ }
+ }
+
test("support user provided avro schema with defaults for missing fields") {
val avroSchema =
"""
@@ -4128,3 +4231,47 @@ class AvroV2Suite extends AvroSuite with ExplainSuiteHelper {
}
}
}
+
+// The allowlist is a static SQL config, so it cannot be set with `withSQLConf`; it is fixed on the
+// session here via `sparkConf`. These go through a real `spark.read ... load()` so they pin the
+// production path the option guards: that the value set on the session reaches the `SQLConf.get`
+// the check reads, that the check fires in a read, and that a session cannot relax it.
+class AvroSchemaUrlAllowlistSuite extends QueryTest with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf =
+ super.sparkConf.set(StaticSQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES.key, "file")
+
+ private val testAvro = testFile("test.avro")
+
+ test("SPARK-59329: an allowed avroSchemaUrl scheme is permitted through a read") {
+ // testFile returns a "file:" URL, whose scheme "file" is allowed.
+ val result = spark.read.option("avroSchemaUrl", testFile("test_sub.avsc"))
+ .format("avro").load(testAvro).collect()
+ val expected = spark.read.format("avro").load(testAvro).select("string").collect()
+ assert(result.sameElements(expected))
+ }
+
+ test("SPARK-59329: a disallowed avroSchemaUrl scheme is rejected through a read") {
+ val e = intercept[AnalysisException] {
+ spark.read.option("avroSchemaUrl", "s3a://bucket/user.avsc")
+ .format("avro").load(testAvro).collect()
+ }
+ assert(e.getCondition == "STDS_INVALID_OPTION_VALUE.WITH_MESSAGE")
+ assert(e.getMessage.contains("not in the allowlist"))
+ }
+
+ test("SPARK-59329: a session cannot relax the avroSchemaUrl scheme allowlist") {
+ // buildStaticConf makes the allowlist an operator-level boundary: neither the DataFrame conf
+ // API nor SQL SET can widen it at runtime.
+ val key = StaticSQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES.key
+ Seq[() => Unit](
+ () => spark.conf.set(key, "s3a"),
+ () => spark.sql(s"SET $key=s3a").collect()
+ ).foreach { f =>
+ checkError(
+ exception = intercept[AnalysisException](f()),
+ condition = "CANNOT_MODIFY_STATIC_CONFIG",
+ parameters = Map("key" -> s""""$key""""))
+ }
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala
index 2d0076f614fb..8a3b03fc7be9 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala
@@ -378,6 +378,22 @@ object StaticSQLConf {
.booleanConf
.createWithDefault(false)
+ val AVRO_SCHEMA_URL_ALLOWED_SCHEMES =
+ buildStaticConf("spark.sql.avro.schemaUrlAllowedSchemes")
+ .internal()
+ .doc("A comma-separated allowlist of URI schemes permitted for the 'avroSchemaUrl' Avro " +
+ "option. Empty by default, which permits any scheme and preserves the previous behavior; " +
+ "when non-empty, an avroSchemaUrl whose scheme is not listed is rejected before it is " +
+ "opened. As a static configuration it can only be set when starting the driver, and not " +
+ "from a session. It restricts the scheme an avroSchemaUrl may name; it does not restrict " +
+ "which file system serves that scheme. That is decided by fs.<scheme>.impl, which a " +
+ "session can still set.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .stringConf
+ .toSequence
+ .createWithDefault(Nil)
+
val KAFKA_DISALLOWED_OPTIONS =
buildStaticConf("spark.sql.kafka.disallowedOptions")
.internal()
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala
index 9903456a031f..1c5d29e02eb8 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala
@@ -18,7 +18,7 @@
package org.apache.spark.sql.avro
import java.net.URI
-import java.util.HashMap
+import java.util.{HashMap, Locale}
import org.apache.avro.Schema
import org.apache.hadoop.conf.Configuration
@@ -29,7 +29,7 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.{DataSourceOptions, FileSourceOptions}
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, FailFastMode, ParseMode}
import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
/**
* Options for Avro Reader and Writer stored in case insensitive manner.
@@ -76,7 +76,30 @@ 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(StaticSQLConf.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"${allowedSchemes.mkString("[", ", ", "]")} configured by " +
+ s"${StaticSQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES.key}.")
+ }
+ }
+ val fs = FileSystem.get(uri, conf)
val in = fs.open(new Path(url))
try {
new Schema.Parser().setValidateDefaults(false).parse(in)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org