(spark) branch master updated: [SPARK-36680][SQL][FOLLOWUP] Files with options should be put into resolveDataSource function

[we...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new 1a428c160664 [SPARK-36680][SQL][FOLLOWUP] Files with options should be put into resolveDataSource function
1a428c160664 is described below

commit 1a428c1606645057ef94ac8a6cadbb947b9208a6
Author: lizongze <li...@xiaomi.com>
AuthorDate: Thu Jul 18 15:58:38 2024 +0800

    [SPARK-36680][SQL][FOLLOWUP] Files with options should be put into resolveDataSource function
    
    ### What changes were proposed in this pull request?
    
    When reading csv, json and other files, pass the options parameter to the rules.resolveDataSource method to make the options parameter effective.
    
    This is a bug fix for [#46707](https://github.com/apache/spark/pull/46707) szehon-ho
    
    ### Why are the changes needed?
    
    For the following SQL, the options parameter passed in does not take effect. This is because the rules.resolveDataSource method does not pass the options parameter during the datasource construction process
    
    ```
     SELECT * FROM csv.`/test/data.csv` WITH (`header` = true, 'delimiter' = '|')
     ```
    
    ### Does this PR introduce _any_ user-facing change?
    
    No
    
    ### How was this patch tested?
    
    Unit test in SQLQuerySuite
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    No
    
    Closes #47370 from logze/hint-options.
    
    Authored-by: lizongze <li...@xiaomi.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../spark/sql/execution/datasources/rules.scala    |  8 +++++++-
 .../scala/org/apache/spark/sql/SQLQuerySuite.scala | 22 +++++++++++++++++++++-
 2 files changed, 28 insertions(+), 2 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
index e4c3cd20dedb..5265c24b140d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
@@ -19,6 +19,8 @@ package org.apache.spark.sql.execution.datasources
 
 import java.util.Locale
 
+import scala.jdk.CollectionConverters._
+
 import org.apache.spark.sql.{AnalysisException, SaveMode, SparkSession}
 import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.catalog._
@@ -50,7 +52,11 @@ class ResolveSQLOnFile(sparkSession: SparkSession) extends Rule[LogicalPlan] {
 
   private def resolveDataSource(unresolved: UnresolvedRelation): DataSource = {
     val ident = unresolved.multipartIdentifier
-    val dataSource = DataSource(sparkSession, paths = Seq(ident.last), className = ident.head)
+    val dataSource = DataSource(
+      sparkSession,
+      paths = Seq(ident.last),
+      className = ident.head,
+      options = unresolved.options.asScala.toMap)
     // `dataSource.providingClass` may throw ClassNotFoundException, the caller side will try-catch
     // it and return the original plan, so that the analyzer can report table not found later.
     val isFileFormat = classOf[FileFormat].isAssignableFrom(dataSource.providingClass)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
index 56c364e20846..302b05e9b5ce 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
@@ -43,7 +43,7 @@ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
 import org.apache.spark.sql.execution.aggregate._
 import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
 import org.apache.spark.sql.execution.command.DataWritingCommandExec
-import org.apache.spark.sql.execution.datasources.{InsertIntoHadoopFsRelationCommand, LogicalRelation}
+import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, InsertIntoHadoopFsRelationCommand, LogicalRelation}
 import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
 import org.apache.spark.sql.execution.datasources.v2.orc.OrcScan
 import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
@@ -4856,6 +4856,26 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
         |"""
     )
   }
+
+  test("SPARK-36680: Files hint options should be put into resolveDataSource function") {
+    val df1 = spark.range(100).toDF()
+    withTempPath { f =>
+      df1.write.json(f.getCanonicalPath)
+      val df2 = sql(
+        s"""
+           |SELECT id
+           |FROM json.`${f.getCanonicalPath}`
+           |WITH (`key1` = 1, `key2` = 2)
+        """.stripMargin
+      )
+      checkAnswer(df2, df1)
+      val relations = df2.queryExecution.analyzed.collect {
+        case LogicalRelation(fs: HadoopFsRelation, _, _, _) => fs
+      }
+      assert(relations.size == 1)
+      assert(relations.head.options == Map("key1" -> "1", "key2" -> "2"))
+    }
+  }
 }
 
 case class Foo(bar: Option[String])


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org