[SPARK-58392][SQL] Pass all options while loading relations

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/RelationCatalog.java

@@ -22,6 +22,7 @@
 import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
 import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
 import org.apache.spark.sql.catalyst.analysis.NoSuchViewException;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
 
 /**
  * Catalog API for connectors that expose both tables and views in a single shared identifier
@@ -134,6 +135,26 @@ public interface RelationCatalog extends TableCatalog, ViewCatalog {
    */
   Relation loadRelation(Identifier ident) throws NoSuchTableException;
 
+  /**
+   * Load the relation for an identifier that may resolve to either a table or a view, forwarding
+   * all user-specified options.
+   * <p>
+   * Behaves like {@link #loadRelation(Identifier)} but also receives the options passed to the
+   * read. The default implementation ignores {@code options} and delegates to
+   * {@link #loadRelation(Identifier)}; catalogs that want to receive the user options while
+   * reading a relation must override this method.
+   *
+   * @param ident the identifier
+   * @param options all options passed to the read
+   * @return a {@link Table} for tables, or a {@link View} for views
+   * @throws NoSuchTableException if neither a table nor a view exists at {@code ident}
+   * @since 4.3.0
+   */
+  default Relation loadRelation(Identifier ident, CaseInsensitiveStringMap options)
+      throws NoSuchTableException {
+    return loadRelation(ident);
+  }
+
   /**
    * List the tables and views in a namespace, returned as {@link TableSummary} entries with
    * the kind preserved on each summary.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala

@@ -256,7 +256,7 @@ class RelationResolution(
             val relation: Option[Relation] = catalog match {
               case mc: RelationCatalog if finalTimeTravelSpec.isEmpty && writePrivileges == null =>
                 try {
-                  Some(mc.loadRelation(ident))
+                  Some(mc.loadRelation(ident, finalOptions))
                 } catch {
                   case _: NoSuchTableException => None
                 }

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRelationCatalog.scala

@@ -47,6 +47,14 @@ class InMemoryRelationCatalog extends RelationCatalog with SupportsNamespaces {
     Option(store.get(key)).getOrElse(throw new NoSuchTableException(ident))
   }
 
+  private var _lastLoadRelationOptions: Option[CaseInsensitiveStringMap] = None
+  def lastLoadRelationOptions: Option[CaseInsensitiveStringMap] = _lastLoadRelationOptions
+
+  override def loadRelation(ident: Identifier, options: CaseInsensitiveStringMap): Relation = {
+    _lastLoadRelationOptions = Some(options)
+    loadRelation(ident)
+  }
+
   // ----- TableCatalog -----------------------------------------------------------------
 
   override def createTable(ident: Identifier, info: TableInfo): Table = {

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala

@@ -20,7 +20,7 @@ package org.apache.spark.sql.connector
 import org.apache.spark.sql.{AnalysisException, Row}
 import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
 import org.apache.spark.sql.catalyst.plans.logical._
-import org.apache.spark.sql.connector.catalog.InMemoryBaseTable
+import org.apache.spark.sql.connector.catalog.{InMemoryBaseTable, InMemoryRelationCatalog}
 import org.apache.spark.sql.execution.CommandResultExec
 import org.apache.spark.sql.execution.datasources.v2._
 import org.apache.spark.sql.functions.lit
@@ -65,6 +65,23 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
     }
   }
 
+  test("Propagate options to RelationCatalog.loadRelation on read") {
+    registerCatalog("testrelcat", classOf[InMemoryRelationCatalog])
+    val t1 = "testrelcat.ns1.ns2.table"
+    withTable(t1) {
+      sql(s"CREATE TABLE $t1 (id bigint, data string) USING parquet")
+
+      val relCatalog = catalog("testrelcat").asInstanceOf[InMemoryRelationCatalog]
+      assert(relCatalog.lastLoadRelationOptions.isEmpty)
+
+      spark.read.option("customOption", "customValue").table(t1)
+        .queryExecution.analyzed
+      val recorded = relCatalog.lastLoadRelationOptions
+      assert(recorded.isDefined)
+      assert(recorded.get.get("customOption") == "customValue")
+    }
+  }
+
   test("SPARK-50286: Propagate options for DataFrameReader") {
     val t1 = s"${catalogAndNamespace}table"
     withTable(t1) {