peter-toth commented on code in PR #57585:
URL: https://github.com/apache/spark/pull/57585#discussion_r3726718760
##########
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
Review Comment:
**Finding 5.** This records only the last bag, while the sibling fixture
#57582 touched records every call: `InMemoryTableCatalog.scala:58-63` keeps
`_loadTableCalls: ArrayBuffer[(TableContext, CaseInsensitiveStringMap)]` and
exposes `loadTableCalls` / `resetLoadTableCalls()` / `lastLoadTableOptions` on
top of it. That accumulator is what lets `DataSourceV2OptionSuite` assert the
things a last-call recorder cannot express -- "one load per distinct option
bag" (`:568`), "identical bags load once" (`:610`), "analysis *and* refresh
both forwarded" (`:542`) -- and the `RelationCatalog` equivalents of all three
are missing here; finding 1 needs the third. Suggest mirroring the sibling
(plus the `scala.collection.mutable` import):
```scala
private val _loadRelationCalls =
mutable.ArrayBuffer.empty[CaseInsensitiveStringMap]
def loadRelationCalls: Seq[CaseInsensitiveStringMap] =
_loadRelationCalls.toSeq
def resetLoadRelationCalls(): Unit = _loadRelationCalls.clear()
def lastLoadRelationOptions: Option[CaseInsensitiveStringMap] =
_loadRelationCalls.lastOption
```
and putting it in the "Test-only accessors" section at `:227` next to
`getStoredInfo` / `getStoredView`, rather than between `loadRelation` and the
`----- TableCatalog -----` banner.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala:
##########
@@ -81,6 +81,23 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
}
}
+ test("Propagate options to RelationCatalog.loadRelation on read") {
Review Comment:
**Finding 8.** Nit: no JIRA prefix, and the test sits between the
`SPARK-36680` and `SPARK-50286` cases rather than with the option-forwarding
group at `:462`-`:540` / the `SPARK-58389:` block at `:542`+ that it is the
sequel to. Something like `test("SPARK-58392: options are forwarded to
loadRelation - DataFrame API")`, placed next to `"options are forwarded to
loadTable - DataFrame API"`, would read as the pair it is.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/RelationCatalog.java:
##########
@@ -134,6 +135,26 @@ public interface RelationCatalog extends TableCatalog,
ViewCatalog {
*/
Relation loadRelation(Identifier ident) throws NoSuchTableException;
+ /**
Review Comment:
**Finding 3.** The class javadoc's "Single-RPC perf entry points" list
(`RelationCatalog.java:109`) still describes `loadRelation(Identifier)` as "the
resolver's per-identifier read path", which this PR makes false -- the resolver
now calls the overload, and the 1-arg one is left serving the DDL lookup in
`Analyzer.lookupTableOrView`
(`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:1275`)
plus the derived `loadTable` / `loadView` / `tableExists` / `viewExists`
defaults. The new overload is not in that list at all. Rough shape:
```
* <li>{@link #loadRelation(Identifier)} -- returns a {@link Table} for a
table or a
* {@link View} for a view; callers discriminate via {@code
instanceof}. Saves the
* {@code loadTable} -> {@code loadView} fallback on a cold cache.
Used for lookups that
* carry no read options (DDL and misc commands), and as the base that
the
* {@code loadTable} / {@code loadView} defaults derive from.</li>
* <li>{@link #loadRelation(Identifier, CaseInsensitiveStringMap)} -- the
resolver's
* per-identifier read path: same contract, plus the user's read
options.</li>
```
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/RelationCatalog.java:
##########
@@ -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
Review Comment:
**Finding 4.** "all options passed to the read" is accurate about the bag's
contents but not about when it arrives: the resolver routes here only when
there is no time travel and no write privileges
(`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:267`).
A `versionAsOf` read or an `INSERT INTO ... WITH (...)` target never reaches
this method -- both go to `TableCatalog.loadTable(Identifier, TableContext,
CaseInsensitiveStringMap)`. #57582 was explicit about the analogous contract
("An override replaces that dispatch and must honor `context` itself...");
without the same note here, an implementor who overrides only this method
reasonably believes they have covered reads. Suggest adding:
```
* Spark calls this for a plain read only -- no time travel and no write
privileges, both of
* which apply to tables only and route through
* {@link TableCatalog#loadTable(Identifier, TableContext,
CaseInsensitiveStringMap)} instead.
```
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala:
##########
@@ -81,6 +81,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")
Review Comment:
**Finding 6.** `loadRelation` is the one load method that can return a
`View`, and this new overload is what a combined table+view catalog will use
for both kinds -- but only the table branch is covered. A view case is cheap:
`CREATE VIEW testrelcat.ns1.ns2.v AS SELECT 1 AS x` seeds one through
`InMemoryRelationCatalog.createView` (the same way
`DataSourceV2MetadataViewSuite.seedV2View` does against its own
`RelationCatalog`), then `spark.read.option("customOption",
"customValue").table("testrelcat.ns1.ns2.v")` and the same recorded-bag
assertion. Worth having: a connector may well want to pick a view definition
variant from the read options, and `RelationResolution.createRelation` takes a
visibly different branch for `View`
(`createDataSourceV1Scan(V1Table.toCatalogTable(...))`) than for `Table`.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/RelationCatalog.java:
##########
@@ -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)
Review Comment:
**Finding 1.** The options only reach a `RelationCatalog` from the
resolver's plain-read branch. Every other load-with-options goes through
`TableCatalog.loadTable(Identifier, TableContext, CaseInsensitiveStringMap)`,
which `RelationCatalog` does not override -- so for an empty context its
default dispatches to `loadTable(ident)` (`TableCatalog.java:238`), which
`RelationCatalog` derives from `loadRelation(ident)`
(`RelationCatalog.java:196`), and the options are gone. All three of these call
it with a non-empty options bag and an empty context:
- `V2TableRefreshUtil.refresh` -> `CatalogV2Util.getTable(catalog, ident,
options = r.options)`
(`sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala:99`)
- `CacheManager.tryRefreshPlan` -> the same call
(`sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:422`)
- the `SupportsCatalogOptions` `spark.read.format(...).option(...).load()`
path
(`sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala:144`)
The refresh one is the one that bites: it reloads and then does
`r.copy(table = currentTable)`, so a `Table` that analysis shaped from the
options is silently replaced at execution time by the one loaded without them.
Measured with a `RelationCatalog` fixture that returns a real v2 `Table` and
records both overloads ("bare" = 1-arg). For a single
`spark.read.option("split-size", "5").table(t).collect()`:
relationLoads = [options:5, bare] // analysis forwards, the refresh
does not
Adding this default turns it into `[options:5, options:5]`, with
`DataSourceV2OptionSuite` (29 tests, including the whole `SPARK-58389:` block)
still green:
```java
/**
* {@inheritDoc}
* <p>
* The default implementation derives from {@link #loadRelation(Identifier,
* CaseInsensitiveStringMap)} for a plain read, so the user options reach
the relation-level
* entry point. Time-travel and write-privilege loads keep {@link
TableCatalog}'s dispatch --
* both apply to tables only.
*/
@Override
default Table loadTable(
Identifier ident,
TableContext context,
CaseInsensitiveStringMap options) throws NoSuchTableException {
if (context.timeTravel().isPresent() ||
!context.writePrivileges().isEmpty()) {
return TableCatalog.super.loadTable(ident, context, options);
}
if (loadRelation(ident, options) instanceof Table t) {
return t;
}
throw new NoSuchTableException(ident);
}
```
Worth noting why the new test cannot see this: `InMemoryRelationCatalog`
stores tables as `DelegatingTable`, which `RelationResolution.createRelation`
routes down the v1 scan path, so the read never becomes a
`DataSourceV2Relation` and never reaches the refresh at all. A fixture
returning a plain v2 `Table` (plus the call accumulator from finding 5) is what
a regression test needs -- the `RelationCatalog` analogue of `"SPARK-58389:
execution refresh forwards options on a plain table read"`.
--
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]
