diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/RelationCatalog.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/RelationCatalog.java index e088e5d3a82c..dd5ad4c61eee 100644 --- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/RelationCatalog.java +++ b/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. + *
+ * 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. diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala index 22bed3fbe769..cf599ef03ca9 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala +++ b/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 } diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRelationCatalog.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRelationCatalog.scala index d6f526b30ce0..1ea38c57ad6d 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRelationCatalog.scala +++ b/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 = { diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala index 803dd35513f4..1e7dbd564744 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala +++ b/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) {