You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "andreaschat-db (via GitHub)" <gi...@apache.org> on 2026/05/26 15:06:13 UTC
[PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
andreaschat-db opened a new pull request, #56121:
URL: https://github.com/apache/spark/pull/56121
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
Currently, the DSv2 Transaction API skips dataframe caching. This can cause significant performance regression to existing workloads. Dataframes cached prior to the transaction should be allowed to be reused within the transaction.
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Existing and new tests.
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
Opus 4.7
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "andreaschat-db (via GitHub)" <gi...@apache.org>.
andreaschat-db commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3357852296
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -684,6 +666,18 @@ abstract class InMemoryBaseTable(
options: CaseInsensitiveStringMap)
extends BatchScanBaseClass(_data, readSchema, tableSchema) with SupportsRuntimeFiltering {
+ // Snapshot of the table version when this scan was built.
+ val builtAtTableVersion: Int = InMemoryBaseTable.this.tableVersion
+
+ // The current table version, read fresh on each access.
+ def currentTableVersion: Int = InMemoryBaseTable.this.tableVersion
Review Comment:
Yeah that was a workaround to a test infra limitation. So essentially all table references point to the same mutable `InMemoryTable`. I changed the test data store to return a copy of the InMemoryTable at load. That is more reminiscent to a snapshot semantic. So when cache references a table this is isolated from the main copy of the table.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3358267785
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -489,16 +492,37 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
result
}
+ // Decides whether the cached entry can be substituted into a plan being executed inside
+ // the given transaction. Collects only the scans whose table belongs to the transaction's
+ // catalog and asks the connector whether reusing the cached snapshot is compatible with its
+ // isolation contract.
+ private def validateCachedEntryForTransaction(cd: CachedData, txn: Transaction): Boolean = {
+ val txnCatalogName = txn.catalog().name()
+ val txnTables = cd.cachedRepresentation.cacheBuilder.logicalPlan.collectWithSubqueries {
Review Comment:
Why do we have to collect `txnTables` before traversing the physical plan?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "andreaschat-db (via GitHub)" <gi...@apache.org>.
andreaschat-db commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3362262266
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2532,20 +2532,25 @@ class Analyzer(
* outer plan to get evaluated.
*/
private def resolveSubQueries(plan: LogicalPlan, outer: LogicalPlan): LogicalPlan = {
+ // A subquery containing a V2TableReference also needs re-analysis: the placeholder is
Review Comment:
Narrowed the fix. Now we should only replace the reference in already resolved plans.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -489,16 +492,37 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
result
}
+ // Decides whether the cached entry can be substituted into a plan being executed inside
+ // the given transaction. Collects only the scans whose table belongs to the transaction's
+ // catalog and asks the connector whether reusing the cached snapshot is compatible with its
+ // isolation contract.
+ private def validateCachedEntryForTransaction(cd: CachedData, txn: Transaction): Boolean = {
+ val txnCatalogName = txn.catalog().name()
+ val txnTables = cd.cachedRepresentation.cacheBuilder.logicalPlan.collectWithSubqueries {
+ case r: DataSourceV2Relation if r.catalog.exists(_.name() == txnCatalogName) => r.table
+ }.toSet
+ val scans = collectWithSubqueries(cd.cachedRepresentation.cacheBuilder.cachedPlan) {
+ case b: BatchScanExec if txnTables.contains(b.table) => b.scan
Review Comment:
I think this is by default implemented as a reference equality. If a connector decides to override this is responsible for provide a sane implementation. Is there any chance we replace the table instance between logical and physical plans? Any copy would break this.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -320,6 +320,7 @@ object ExtractV2ScanInfo {
}
object DataSourceV2Relation {
+
Review Comment:
Removed.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "andreaschat-db (via GitHub)" <gi...@apache.org>.
andreaschat-db commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3372610858
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2549,6 +2548,15 @@ class Analyzer(
case a: FunctionTableSubqueryArgumentExpression if !a.plan.resolved =>
resolveSubQuery(a, outer)(
(plan, outerAttrs) => a.copy(plan = plan, outerAttrs = outerAttrs))
+ // The subquery's plan is already resolved. Replace any V2TableReferences without
+ // re-running any analyzer rules.
+ case se: SubqueryExpression
+ if se.plan.resolved &&
+ se.plan.collectFirstWithSubqueries { case _: V2TableReference => () }.isDefined =>
+ val newPlan = se.plan.transformWithSubqueries {
+ case r: V2TableReference => relationResolution.resolveReference(r)
Review Comment:
@gengliangwang answered this below. In short, yes. Relation cache is maintained per query.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on PR #56121:
URL: https://github.com/apache/spark/pull/56121#issuecomment-4651442404
@huaxingao both https://issues.apache.org/jira/browse/SPARK-56695 and https://issues.apache.org/jira/browse/SPARK-56995 are resolved now.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3364232447
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala:
##########
@@ -291,15 +291,13 @@ class QueryExecution(
assertAnalyzed()
assertSupported()
- // During a transaction, skip cache substitution. This is to avoid replacing relations
- // loaded by the transactional catalog with potentially stale relations cached before
- // the transaction was active.
- if (transactionOpt.isDefined) {
- normalized
- } else {
- // Clone the plan to avoid sharing the plan instance between different stages like
- // analyzing, optimizing and planning.
- sparkSession.sharedState.cacheManager.useCachedData(normalized.clone())
+ // Clone the plan to avoid sharing the plan instance between different stages like
+ // analyzing, optimizing and planning.
+ val planToRewrite = normalized.clone()
+ val cacheManager = sparkSession.sharedState.cacheManager
+ transactionOpt match {
Review Comment:
This seems correct.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3358241452
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -320,6 +320,7 @@ object ExtractV2ScanInfo {
}
object DataSourceV2Relation {
+
Review Comment:
Is this needed?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3358181796
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2532,20 +2532,25 @@ class Analyzer(
* outer plan to get evaluated.
*/
private def resolveSubQueries(plan: LogicalPlan, outer: LogicalPlan): LogicalPlan = {
+ // A subquery containing a V2TableReference also needs re-analysis: the placeholder is
Review Comment:
We only want to replace V2TableReference, we don't want to re-analyze everything.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3358261867
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2532,20 +2532,25 @@ class Analyzer(
* outer plan to get evaluated.
*/
private def resolveSubQueries(plan: LogicalPlan, outer: LogicalPlan): LogicalPlan = {
+ // A subquery containing a V2TableReference also needs re-analysis: the placeholder is
Review Comment:
Hopefully, we can reuse `RelationResolution$resolveReference` as is.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #56121:
URL: https://github.com/apache/spark/pull/56121#issuecomment-4558716669
@aokolnychyi @andreaschat-db
Is this PR aiming for 4.2?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3364223519
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2549,6 +2548,15 @@ class Analyzer(
case a: FunctionTableSubqueryArgumentExpression if !a.plan.resolved =>
resolveSubQuery(a, outer)(
(plan, outerAttrs) => a.copy(plan = plan, outerAttrs = outerAttrs))
+ // The subquery's plan is already resolved. Replace any V2TableReferences without
Review Comment:
This seems much safer than before.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3358176995
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2532,20 +2532,25 @@ class Analyzer(
* outer plan to get evaluated.
*/
private def resolveSubQueries(plan: LogicalPlan, outer: LogicalPlan): LogicalPlan = {
+ // A subquery containing a V2TableReference also needs re-analysis: the placeholder is
Review Comment:
I understand the motivation but I feel it is too broad and too dangerous. Let me think.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on PR #56121:
URL: https://github.com/apache/spark/pull/56121#issuecomment-4650937640
Thanks, merging to master/4.x/4.2
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #56121:
URL: https://github.com/apache/spark/pull/56121#issuecomment-4566071459
@aokolnychyi @andreaschat-db
Could you share when you expect this PR to be merged and backported to branch-4.2? Also, should we treat this as a blocker for cutting Spark 4.2.0 RC1?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3364760596
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/transactions/Transaction.java:
##########
@@ -66,6 +67,20 @@ public interface Transaction extends Closeable {
*/
void abort();
+ /**
+ * Attempts to register materialized scans against this transaction's read set.
+ * <p>
+ * An example use case is cache reuse. Spark passes the scans of a candidate cached subtree
+ * for the transaction's catalog and the connector decides whether to accept them.
+ * <p>
+ * The connector must either accept (returning {@code true} after adding any relevant scans
+ * to the read set) or refuse (returning {@code false} without modifying the read set).
+ *
+ * @param scans the scans to register.
Review Comment:
Two small doc points on this `@Evolving` API:
1. The accept/refuse decision is all-or-nothing for the whole `scans` array, but the doc doesn't say so — "must either accept … or refuse" can read as if it were per-scan. Worth one sentence that the return value covers the entire batch: the connector accepts reuse of all the passed scans or none, there's no partial acceptance — a `false` makes Spark bypass the whole cached subtree.
2. `@param scans the scans to register.` is circular (it restates the method name). For a public API, describe what they are:
```suggestion
* @param scans the materialized scans Spark offers for registration against this transaction's read set.
```
Both are contract-level, so they keep the body generic per @aokolnychyi's earlier point about not over-fitting to caching. The caller-specific scope — that Spark passes only this one cached subtree's scans, already filtered to the transaction's catalog — is better documented at the call site (`CacheManager.validateCachedEntryForTransaction`) than baked into this contract.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3364230850
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -489,16 +492,37 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
result
}
+ // Decides whether the cached entry can be substituted into a plan being executed inside
+ // the given transaction. Collects only the scans whose table belongs to the transaction's
+ // catalog and asks the connector whether reusing the cached snapshot is compatible with its
+ // isolation contract.
+ private def validateCachedEntryForTransaction(cd: CachedData, txn: Transaction): Boolean = {
+ val txnCatalogName = txn.catalog().name()
+ val txnTables = cd.cachedRepresentation.cacheBuilder.logicalPlan.collectWithSubqueries {
Review Comment:
@cloud-fan @gengliangwang, any ideas how to handle this better?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang closed pull request #56121: [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API
URL: https://github.com/apache/spark/pull/56121
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "andreaschat-db (via GitHub)" <gi...@apache.org>.
andreaschat-db commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3372893799
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -489,16 +492,37 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
result
}
+ // Decides whether the cached entry can be substituted into a plan being executed inside
+ // the given transaction. Collects only the scans whose table belongs to the transaction's
+ // catalog and asks the connector whether reusing the cached snapshot is compatible with its
+ // isolation contract.
+ private def validateCachedEntryForTransaction(cd: CachedData, txn: Transaction): Boolean = {
+ val txnCatalogName = txn.catalog().name()
+ val txnTables = cd.cachedRepresentation.cacheBuilder.logicalPlan.collectWithSubqueries {
+ case r: DataSourceV2Relation if r.catalog.exists(_.name() == txnCatalogName) => r.table
+ }.toSet
+ val scans = collectWithSubqueries(cd.cachedRepresentation.cacheBuilder.cachedPlan) {
+ case b: BatchScanExec if txnTables.contains(b.table) => b.scan
+ }
+ scans.isEmpty || txn.registerScans(scans.toArray)
Review Comment:
Modified the condition to be more defensive.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3364221842
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2549,6 +2548,15 @@ class Analyzer(
case a: FunctionTableSubqueryArgumentExpression if !a.plan.resolved =>
resolveSubQuery(a, outer)(
(plan, outerAttrs) => a.copy(plan = plan, outerAttrs = outerAttrs))
+ // The subquery's plan is already resolved. Replace any V2TableReferences without
+ // re-running any analyzer rules.
+ case se: SubqueryExpression
+ if se.plan.resolved &&
+ se.plan.collectFirstWithSubqueries { case _: V2TableReference => () }.isDefined =>
+ val newPlan = se.plan.transformWithSubqueries {
+ case r: V2TableReference => relationResolution.resolveReference(r)
Review Comment:
Does this use the same relation cache as the main resolution rule?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3358283006
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -489,16 +492,37 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
result
}
+ // Decides whether the cached entry can be substituted into a plan being executed inside
+ // the given transaction. Collects only the scans whose table belongs to the transaction's
+ // catalog and asks the connector whether reusing the cached snapshot is compatible with its
+ // isolation contract.
+ private def validateCachedEntryForTransaction(cd: CachedData, txn: Transaction): Boolean = {
+ val txnCatalogName = txn.catalog().name()
+ val txnTables = cd.cachedRepresentation.cacheBuilder.logicalPlan.collectWithSubqueries {
+ case r: DataSourceV2Relation if r.catalog.exists(_.name() == txnCatalogName) => r.table
+ }.toSet
+ val scans = collectWithSubqueries(cd.cachedRepresentation.cacheBuilder.cachedPlan) {
+ case b: BatchScanExec if txnTables.contains(b.table) => b.scan
Review Comment:
I wonder if this is reliable and whether we should use reference equality rather than equals/hashCode.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3364612158
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -489,16 +492,37 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
result
}
+ // Decides whether the cached entry can be substituted into a plan being executed inside
+ // the given transaction. Collects only the scans whose table belongs to the transaction's
+ // catalog and asks the connector whether reusing the cached snapshot is compatible with its
+ // isolation contract.
+ private def validateCachedEntryForTransaction(cd: CachedData, txn: Transaction): Boolean = {
+ val txnCatalogName = txn.catalog().name()
+ val txnTables = cd.cachedRepresentation.cacheBuilder.logicalPlan.collectWithSubqueries {
+ case r: DataSourceV2Relation if r.catalog.exists(_.name() == txnCatalogName) => r.table
+ }.toSet
+ val scans = collectWithSubqueries(cd.cachedRepresentation.cacheBuilder.cachedPlan) {
+ case b: BatchScanExec if txnTables.contains(b.table) => b.scan
+ }
+ scans.isEmpty || txn.registerScans(scans.toArray)
Review Comment:
`scans.isEmpty` is the accept condition for two different situations, and only one is safe by construction. `scans` is empty when (a) the entry has no relations from the txn catalog — fine, reuse it; but also when (b) the entry *does* contain txn-catalog relations (`txnTables.nonEmpty`) whose physical scans aren't `BatchScanExec` — e.g. a nested cached relation served as `InMemoryTableScanExec`, a v1-fallback scan, or a streaming scan. In case (b) the entry has already passed the version-aware `sameResult` check, but `registerScans` is silently skipped, so a connector whose isolation contract rejects reuse for reasons beyond version equality never gets to veto.
Is case (b) reachable for the target connectors? If so, consider distinguishing the two — accept on `txnTables.isEmpty`, but treat `txnTables.nonEmpty && scans.isEmpty` as "can't prove compatibility" (refuse, or at least assert it doesn't happen) rather than folding both into one `scans.isEmpty`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -489,16 +492,37 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
result
}
+ // Decides whether the cached entry can be substituted into a plan being executed inside
+ // the given transaction. Collects only the scans whose table belongs to the transaction's
+ // catalog and asks the connector whether reusing the cached snapshot is compatible with its
+ // isolation contract.
+ private def validateCachedEntryForTransaction(cd: CachedData, txn: Transaction): Boolean = {
Review Comment:
`validateCachedEntryForTransaction` (and the `canUse` predicate it becomes — documented as "predicate filtering which cached entries are eligible") read as pure, but they mutate the connector's read set via `txn.registerScans`, and that runs inside `lookupCachedDataInternal`'s `find`. It's correct today — registration aligns 1:1 with the entry that gets substituted — but the side effect is load-bearing for the transaction's read tracking and invisible at the call site. A short doc note on `canUse` (or a name that signals the effect, e.g. `registerAndCheck`) would keep a future change to `lookupCachedDataInternal` (memoizing the predicate, reordering, calling it for diagnostics) from silently breaking read tracking.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoDataFrameSuite.scala:
##########
@@ -1233,4 +1233,318 @@ class MergeIntoDataFrameSuite extends RowLevelOperationSuiteBase {
assert(e.message.contains("incompatible changes to table `cat`.`ns1`.`source_table`"))
}
}
+
+ // Cache-substitution tests for the txn path: connector approves stale-free cached scans via
+ // Transaction.registerScans.
+ test("cached merge source is reused when the table is unchanged since caching") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |{ "pk": 3, "salary": 300, "dep": "hr" }
+ |""".stripMargin)
+
+ val tableVersionBeforeCache = table.version()
+
+ val sourceDF = spark.table(tableNameAsString).where("salary < 250").as("source")
+ sourceDF.cache()
+ sourceDF.count()
+
+ assert(table.version() == tableVersionBeforeCache, "sanity: caching does not bump version")
+
+ val (txn, txnTables) = executeTransaction {
+ sourceDF
+ .mergeInto(tableNameAsString, $"source.pk" === targetTableCol("pk"))
+ .whenMatched()
+ .update(Map("salary" -> targetTableCol("salary").plus(1)))
+ .merge()
+ }
+
+ assert(txn.currentState == Committed)
+ assert(txn.isClosed)
+ assert(txnTables.size == 1)
+ assert(table.version() == "2")
+ assert(txn.registeredScans.nonEmpty, "registerScans should have accepted the cached scan")
+
+ val targetTxnTable = txnTables(tableNameAsString)
+ assert(targetTxnTable.scanEvents.size == 4)
+ val sourceFilterScans = targetTxnTable.scanEvents.flatten.count {
+ case sources.LessThan("salary", 250) => true
+ case _ => false
+ }
+ assert(sourceFilterScans == 2,
+ s"expected two salary < 250 scan events, got " +
+ s"${targetTxnTable.scanEvents.map(_.toSeq).mkString("[", ", ", "]")}")
+ val targetScans = targetTxnTable.scanEvents.count(_.isEmpty)
+ assert(targetScans == 2,
+ s"expected two target-side scans with no pushed filters, got " +
+ s"${targetTxnTable.scanEvents.map(_.toSeq).mkString("[", ", ", "]")}")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Seq(
+ Row(1, 101, "hr"),
+ Row(2, 201, "software"),
+ Row(3, 300, "hr")))
+ }
+
+ test("cached merge source is dropped when the table version moves on between caching and MERGE") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |{ "pk": 3, "salary": 300, "dep": "hr" }
+ |""".stripMargin)
+
+ val sourceDF = spark.table(tableNameAsString).where("salary < 250").as("source")
+ sourceDF.cache()
+ sourceDF.count()
+ val versionAtCache = table.version()
+
+ // Bump the version directly to simulate an out-of-band committer. A Spark-side write would
+ // also bump the version but would trigger CacheManager.refreshCache and defeat the test.
+ table.increaseVersion()
+ assert(table.version() != versionAtCache, "sanity: bump should change the version")
+
+ val (txn, txnTables) = executeTransaction {
+ sourceDF
+ .mergeInto(tableNameAsString, $"source.pk" === targetTableCol("pk"))
+ .whenMatched()
+ .update(Map("salary" -> targetTableCol("salary").plus(1)))
+ .merge()
+ }
+
+ assert(txn.currentState == Committed)
+ assert(txn.isClosed)
+ assert(txnTables.size == 1)
+ assert(table.version() == "3")
+ assert(txn.registeredScans.isEmpty, "registerScans should refuse the stale cached scan")
Review Comment:
This message misattributes the mechanism: here the stale entry is rejected by the version-aware `sameResult`/`Table.equals` check *before* `registerScans` is consulted — exactly as your own comments state (`txns.scala`: "Staleness is handled upstream"; `InMemoryRowLevelOperationTableCatalog`: "compare unequal so cache substitution fails before `Transaction.registerScans` is consulted"). So `registerScans` is never called in this test; the "connector rejecting registerScans" test is the one that exercises the refuse path. Suggest wording that points at the version mismatch, e.g.:
```suggestion
assert(txn.registeredScans.isEmpty, "stale entry fails version-aware sameResult, so registerScans is never consulted")
```
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -116,6 +116,21 @@ abstract class InMemoryBaseTable(
}
}
+ // Version-aware equality: two tables refer to the same metastore entity at the same state.
+ // Fallback to reference equality when `id()` is null (no metastore identity).
Review Comment:
"Fallback" is a noun; as a verb it's two words.
```suggestion
// Fall back to reference equality when `id()` is null (no metastore identity).
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2548,6 +2548,15 @@ class Analyzer(
case a: FunctionTableSubqueryArgumentExpression if !a.plan.resolved =>
resolveSubQuery(a, outer)(
(plan, outerAttrs) => a.copy(plan = plan, outerAttrs = outerAttrs))
+ // The subquery's plan is already resolved. Replace any V2TableReferences without
+ // re-running any analyzer rules.
+ case se: SubqueryExpression
+ if se.plan.resolved &&
+ se.plan.collectFirstWithSubqueries { case _: V2TableReference => () }.isDefined =>
+ val newPlan = se.plan.transformWithSubqueries {
+ case r: V2TableReference => relationResolution.resolveReference(r)
Review Comment:
Re: the question above on whether this uses the same relation cache as the main resolution rule — yes. This case and the top-level `case r: V2TableReference` (Analyzer.scala:1131) both call `relationResolution.resolveReference`, which routes through `getOrLoadRelation`/`relationCache` (RelationResolution.scala:460-480). `relationCache` is `AnalysisContext.get.relationCache`, which is per-analysis, so within one query the relation is loaded once — the first occurrence (top-level or subquery) goes through the txn catalog and tracks the read, and later occurrences reuse a consistent snapshot. Across queries the cache is fresh. So the sharing looks correct and is actually what keeps a self-referencing table consistent within the query.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3364612171
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoDataFrameSuite.scala:
##########
@@ -1233,4 +1233,318 @@ class MergeIntoDataFrameSuite extends RowLevelOperationSuiteBase {
assert(e.message.contains("incompatible changes to table `cat`.`ns1`.`source_table`"))
}
}
+
+ // Cache-substitution tests for the txn path: connector approves stale-free cached scans via
+ // Transaction.registerScans.
+ test("cached merge source is reused when the table is unchanged since caching") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |{ "pk": 3, "salary": 300, "dep": "hr" }
+ |""".stripMargin)
+
+ val tableVersionBeforeCache = table.version()
+
+ val sourceDF = spark.table(tableNameAsString).where("salary < 250").as("source")
+ sourceDF.cache()
+ sourceDF.count()
+
+ assert(table.version() == tableVersionBeforeCache, "sanity: caching does not bump version")
+
+ val (txn, txnTables) = executeTransaction {
+ sourceDF
+ .mergeInto(tableNameAsString, $"source.pk" === targetTableCol("pk"))
+ .whenMatched()
+ .update(Map("salary" -> targetTableCol("salary").plus(1)))
+ .merge()
+ }
+
+ assert(txn.currentState == Committed)
+ assert(txn.isClosed)
+ assert(txnTables.size == 1)
+ assert(table.version() == "2")
+ assert(txn.registeredScans.nonEmpty, "registerScans should have accepted the cached scan")
+
+ val targetTxnTable = txnTables(tableNameAsString)
+ assert(targetTxnTable.scanEvents.size == 4)
+ val sourceFilterScans = targetTxnTable.scanEvents.flatten.count {
+ case sources.LessThan("salary", 250) => true
+ case _ => false
+ }
+ assert(sourceFilterScans == 2,
+ s"expected two salary < 250 scan events, got " +
+ s"${targetTxnTable.scanEvents.map(_.toSeq).mkString("[", ", ", "]")}")
+ val targetScans = targetTxnTable.scanEvents.count(_.isEmpty)
+ assert(targetScans == 2,
+ s"expected two target-side scans with no pushed filters, got " +
+ s"${targetTxnTable.scanEvents.map(_.toSeq).mkString("[", ", ", "]")}")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Seq(
+ Row(1, 101, "hr"),
+ Row(2, 201, "software"),
+ Row(3, 300, "hr")))
+ }
+
+ test("cached merge source is dropped when the table version moves on between caching and MERGE") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |{ "pk": 3, "salary": 300, "dep": "hr" }
+ |""".stripMargin)
+
+ val sourceDF = spark.table(tableNameAsString).where("salary < 250").as("source")
+ sourceDF.cache()
+ sourceDF.count()
+ val versionAtCache = table.version()
+
+ // Bump the version directly to simulate an out-of-band committer. A Spark-side write would
+ // also bump the version but would trigger CacheManager.refreshCache and defeat the test.
+ table.increaseVersion()
+ assert(table.version() != versionAtCache, "sanity: bump should change the version")
+
+ val (txn, txnTables) = executeTransaction {
+ sourceDF
+ .mergeInto(tableNameAsString, $"source.pk" === targetTableCol("pk"))
+ .whenMatched()
+ .update(Map("salary" -> targetTableCol("salary").plus(1)))
+ .merge()
+ }
+
+ assert(txn.currentState == Committed)
+ assert(txn.isClosed)
+ assert(txnTables.size == 1)
+ assert(table.version() == "3")
+ assert(txn.registeredScans.isEmpty, "registerScans should refuse the stale cached scan")
Review Comment:
This message misattributes the mechanism: here the stale entry is rejected by the version-aware `sameResult`/`Table.equals` check *before* `registerScans` is consulted — exactly as your own comments state (`txns.scala`: "Staleness is handled upstream"; `InMemoryRowLevelOperationTableCatalog`: "compare unequal so cache substitution fails before `Transaction.registerScans` is consulted"). So `registerScans` is never called in this test; the "connector rejecting registerScans" test is the one that exercises the refuse path. Suggest wording that points at the version mismatch, e.g.:
```suggestion
assert(txn.registeredScans.isEmpty, "stale entry fails version-aware sameResult, so registerScans is never consulted")
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "andreaschat-db (via GitHub)" <gi...@apache.org>.
andreaschat-db commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3372713677
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -489,16 +492,37 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
result
}
+ // Decides whether the cached entry can be substituted into a plan being executed inside
+ // the given transaction. Collects only the scans whose table belongs to the transaction's
+ // catalog and asks the connector whether reusing the cached snapshot is compatible with its
+ // isolation contract.
+ private def validateCachedEntryForTransaction(cd: CachedData, txn: Transaction): Boolean = {
Review Comment:
Added a note in `validateCachedEntryForTransaction` explaining it has a side effect.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "andreaschat-db (via GitHub)" <gi...@apache.org>.
andreaschat-db commented on PR #56121:
URL: https://github.com/apache/spark/pull/56121#issuecomment-4561847927
> @aokolnychyi @andreaschat-db
>
> Is this PR aiming for 4.2?
Yes. It is an important fix for the DSv2 Transaction API.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "andreaschat-db (via GitHub)" <gi...@apache.org>.
andreaschat-db commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3355493723
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/transactions/Transaction.java:
##########
@@ -66,6 +68,29 @@ public interface Transaction extends Closeable {
*/
void abort();
+ /**
+ * Attempts to register a list of materialized scans against this transaction's read set.
+ * <p>
+ * Spark calls this when considering reuse of a cached subtree during the transaction. The
Review Comment:
Generalised description.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/transactions/Transaction.java:
##########
@@ -66,6 +68,29 @@ public interface Transaction extends Closeable {
*/
void abort();
+ /**
+ * Attempts to register a list of materialized scans against this transaction's read set.
+ * <p>
+ * Spark calls this when considering reuse of a cached subtree during the transaction. The
+ * list contains every materialized scan in the candidate cached subtree. That may include
+ * scans that belong to catalogs other than this transaction's catalog. The connector can
+ * decide which scans to register or ignore.
+ * <p>
+ * The connector decides whether reusing the cached snapshots is compatible with the
+ * transaction's isolation contract. It must either accept the cache entry (returning
+ * {@code true} after adding any of its own scans to the read set) or refuse it (returning
+ * {@code false} without modifying the read set).
+ * <p>
+ * This method may be called multiple times during a single query with overlapping scan
+ * lists. Registering a scan that is already in the read set must be a no-op.
+ *
+ * @param scans Every materialized scan in the candidate cached subtree.
+ * @return true if the connector accepts reuse of the cache entry; false otherwise.
+ */
+ default boolean registerScans(List<Scan> scans) {
Review Comment:
Makes sense. Removed default.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -122,6 +123,21 @@ case class DataSourceV2Relation(
copy(output = output.map(_.newInstance()))
}
+ // Replace the `table` field with an identity placeholder. This allows two relations
+ // that point at the same logical table compare equal even when one is wrapped (e.g.
+ // by a transaction-aware catalog). Note, when `Table.id()` is null we key only on catalog
+ // and identifier, thus, drop-and-recreate is not distinguished.
+ override def doCanonicalize(): LogicalPlan = {
Review Comment:
Arrrg this is what I was looking for. Makes sense. Dropped this.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -479,8 +481,13 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
lookupCachedDataInternal(normalized)
}
- private def lookupCachedDataInternal(plan: LogicalPlan): Option[CachedData] = {
- val result = cachedData.find(cd => plan.sameResult(cd.plan))
+ private def lookupCachedDataInternal(
Review Comment:
Passing a filter function was my original version :D. I was thinking that I might be over-engineering it. But why not.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -489,16 +496,34 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
result
}
+ // Decides whether the cached entry can be substituted into a plan being executed inside
+ // the given transaction. It identifies the scans and attempts to register them in the connector.
+ // The connector returns true if reusing the cached snapshot is consistent with its isolation
+ // contract. Note, the scans might belong to different catalogs. The connector can decide which
+ // one to register and which ones to ignore.
+ private def validateCachedEntryForTransaction(cd: CachedData, txn: Transaction): Boolean = {
Review Comment:
Fixed. I explained in a relevant comment in `Transaction.java` what my issue was. I can now get the catalog from the logical plan and match this against the scan.table found in the materialized plan by relying on `table.equals`.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/transactions/Transaction.java:
##########
@@ -66,6 +68,29 @@ public interface Transaction extends Closeable {
*/
void abort();
+ /**
+ * Attempts to register a list of materialized scans against this transaction's read set.
+ * <p>
+ * Spark calls this when considering reuse of a cached subtree during the transaction. The
+ * list contains every materialized scan in the candidate cached subtree. That may include
+ * scans that belong to catalogs other than this transaction's catalog. The connector can
+ * decide which scans to register or ignore.
+ * <p>
+ * The connector decides whether reusing the cached snapshots is compatible with the
Review Comment:
Trimmed the description.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/transactions/Transaction.java:
##########
@@ -66,6 +68,29 @@ public interface Transaction extends Closeable {
*/
void abort();
+ /**
+ * Attempts to register a list of materialized scans against this transaction's read set.
+ * <p>
+ * Spark calls this when considering reuse of a cached subtree during the transaction. The
+ * list contains every materialized scan in the candidate cached subtree. That may include
+ * scans that belong to catalogs other than this transaction's catalog. The connector can
Review Comment:
We can have scans from multiple catalogs in a single command. We only support predicate tracking at the transaction catalog. My trouble here was identifying In Spark which scan belongs to which catalog. So I was delegating that to the connector. I was missing that the connector overrides `Table.equals`. Now this can work. I am passing the filtered scan list.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3358272753
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -489,16 +492,37 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
result
}
+ // Decides whether the cached entry can be substituted into a plan being executed inside
+ // the given transaction. Collects only the scans whose table belongs to the transaction's
+ // catalog and asks the connector whether reusing the cached snapshot is compatible with its
+ // isolation contract.
+ private def validateCachedEntryForTransaction(cd: CachedData, txn: Transaction): Boolean = {
+ val txnCatalogName = txn.catalog().name()
+ val txnTables = cd.cachedRepresentation.cacheBuilder.logicalPlan.collectWithSubqueries {
Review Comment:
Ah, we lost the catalog in `BatchScanExec`... That's a bummer.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "andreaschat-db (via GitHub)" <gi...@apache.org>.
andreaschat-db commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3372635544
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/transactions/Transaction.java:
##########
@@ -66,6 +67,20 @@ public interface Transaction extends Closeable {
*/
void abort();
+ /**
+ * Attempts to register materialized scans against this transaction's read set.
+ * <p>
+ * An example use case is cache reuse. Spark passes the scans of a candidate cached subtree
+ * for the transaction's catalog and the connector decides whether to accept them.
+ * <p>
+ * The connector must either accept (returning {@code true} after adding any relevant scans
+ * to the read set) or refuse (returning {@code false} without modifying the read set).
+ *
+ * @param scans the scans to register.
Review Comment:
Clarified both 1 and 2.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #56121:
URL: https://github.com/apache/spark/pull/56121#discussion_r3350507931
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/transactions/Transaction.java:
##########
@@ -66,6 +68,29 @@ public interface Transaction extends Closeable {
*/
void abort();
+ /**
+ * Attempts to register a list of materialized scans against this transaction's read set.
+ * <p>
+ * Spark calls this when considering reuse of a cached subtree during the transaction. The
+ * list contains every materialized scan in the candidate cached subtree. That may include
+ * scans that belong to catalogs other than this transaction's catalog. The connector can
+ * decide which scans to register or ignore.
+ * <p>
+ * The connector decides whether reusing the cached snapshots is compatible with the
+ * transaction's isolation contract. It must either accept the cache entry (returning
+ * {@code true} after adding any of its own scans to the read set) or refuse it (returning
+ * {@code false} without modifying the read set).
+ * <p>
+ * This method may be called multiple times during a single query with overlapping scan
+ * lists. Registering a scan that is already in the read set must be a no-op.
+ *
+ * @param scans Every materialized scan in the candidate cached subtree.
+ * @return true if the connector accepts reuse of the cache entry; false otherwise.
+ */
+ default boolean registerScans(List<Scan> scans) {
Review Comment:
Do we want to provide a default implementation here? The API hasn't been released and it may be better to explicitly check what behavior the connector prefers.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/transactions/Transaction.java:
##########
@@ -66,6 +68,29 @@ public interface Transaction extends Closeable {
*/
void abort();
+ /**
+ * Attempts to register a list of materialized scans against this transaction's read set.
+ * <p>
+ * Spark calls this when considering reuse of a cached subtree during the transaction. The
+ * list contains every materialized scan in the candidate cached subtree. That may include
+ * scans that belong to catalogs other than this transaction's catalog. The connector can
+ * decide which scans to register or ignore.
+ * <p>
+ * The connector decides whether reusing the cached snapshots is compatible with the
+ * transaction's isolation contract. It must either accept the cache entry (returning
+ * {@code true} after adding any of its own scans to the read set) or refuse it (returning
+ * {@code false} without modifying the read set).
+ * <p>
+ * This method may be called multiple times during a single query with overlapping scan
+ * lists. Registering a scan that is already in the read set must be a no-op.
+ *
+ * @param scans Every materialized scan in the candidate cached subtree.
+ * @return true if the connector accepts reuse of the cache entry; false otherwise.
+ */
+ default boolean registerScans(List<Scan> scans) {
Review Comment:
Shall we pass scans as an array (Scan[]) or as a list (List<Scan>)? Would an array be more consistent with the rest of the APIs in Data Source V2? It is your call, @andreaschat-db. I am OK either way.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/transactions/Transaction.java:
##########
@@ -66,6 +68,29 @@ public interface Transaction extends Closeable {
*/
void abort();
+ /**
+ * Attempts to register a list of materialized scans against this transaction's read set.
+ * <p>
+ * Spark calls this when considering reuse of a cached subtree during the transaction. The
Review Comment:
Caching is one use case. You may mention it as an example but without too much details. I don't want to limit the use of this method as there may be valid use cases in the future.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -489,16 +496,34 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
result
}
+ // Decides whether the cached entry can be substituted into a plan being executed inside
+ // the given transaction. It identifies the scans and attempts to register them in the connector.
+ // The connector returns true if reusing the cached snapshot is consistent with its isolation
+ // contract. Note, the scans might belong to different catalogs. The connector can decide which
+ // one to register and which ones to ignore.
+ private def validateCachedEntryForTransaction(cd: CachedData, txn: Transaction): Boolean = {
Review Comment:
I think we need to limit this to only the catalog for which we started the transaction.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/transactions/Transaction.java:
##########
@@ -66,6 +68,29 @@ public interface Transaction extends Closeable {
*/
void abort();
+ /**
+ * Attempts to register a list of materialized scans against this transaction's read set.
+ * <p>
+ * Spark calls this when considering reuse of a cached subtree during the transaction. The
+ * list contains every materialized scan in the candidate cached subtree. That may include
+ * scans that belong to catalogs other than this transaction's catalog. The connector can
Review Comment:
Hm, this doesn't seem to make sense. I thought we would only pass scans for the current catalog. We don't support transactions across catalogs, don't we?
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/transactions/Transaction.java:
##########
@@ -66,6 +68,29 @@ public interface Transaction extends Closeable {
*/
void abort();
+ /**
+ * Attempts to register a list of materialized scans against this transaction's read set.
+ * <p>
+ * Spark calls this when considering reuse of a cached subtree during the transaction. The
+ * list contains every materialized scan in the candidate cached subtree. That may include
+ * scans that belong to catalogs other than this transaction's catalog. The connector can
+ * decide which scans to register or ignore.
+ * <p>
+ * The connector decides whether reusing the cached snapshots is compatible with the
Review Comment:
Methods like `Table$equals()` and `Table$hashCode()` must be version-aware by definition, meaning we will only have a cache match if the state is compatible.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -684,6 +666,18 @@ abstract class InMemoryBaseTable(
options: CaseInsensitiveStringMap)
extends BatchScanBaseClass(_data, readSchema, tableSchema) with SupportsRuntimeFiltering {
+ // Snapshot of the table version when this scan was built.
+ val builtAtTableVersion: Int = InMemoryBaseTable.this.tableVersion
+
+ // The current table version, read fresh on each access.
+ def currentTableVersion: Int = InMemoryBaseTable.this.tableVersion
Review Comment:
DSv2 versions must be pinned at table loading.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -122,6 +123,21 @@ case class DataSourceV2Relation(
copy(output = output.map(_.newInstance()))
}
+ // Replace the `table` field with an identity placeholder. This allows two relations
+ // that point at the same logical table compare equal even when one is wrapped (e.g.
+ // by a transaction-aware catalog). Note, when `Table.id()` is null we key only on catalog
+ // and identifier, thus, drop-and-recreate is not distinguished.
+ override def doCanonicalize(): LogicalPlan = {
Review Comment:
This breaks version-aware and context-aware comparison in caching as looking at the identifiers is not sufficient to tell if the table state is the same. We must rely on correct equality and hash code of `Table` implementations. In other words, instead of normalizing `DataSourceV2Relation` on the Spark side, connectors must correctly implement `equals` and `hashCode` for `Table`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -479,8 +481,13 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
lookupCachedDataInternal(normalized)
}
- private def lookupCachedDataInternal(plan: LogicalPlan): Option[CachedData] = {
- val result = cachedData.find(cd => plan.sameResult(cd.plan))
+ private def lookupCachedDataInternal(
Review Comment:
What about making these core methods a bit more generic?
```
private def lookupCachedDataInternal(
plan: LogicalPlan,
canUse: CachedData => Boolean = _ => true): Option[CachedData] = {
// impl
}
private[sql] def useCachedData(
plan: LogicalPlan,
canUse: CachedData => Boolean = _ => true): LogicalPlan = {
// impl
}
```
We can then offer one more with a transaction that would internally create canUse. This method will be used from QueryExecution.
```
private[sql] def useCachedData(plan: LogicalPlan, txn: Transaction): LogicalPlan = {
useCachedData(plan, transactionEntryFilter(txn))
}
```
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/transactions/Transaction.java:
##########
@@ -66,6 +68,29 @@ public interface Transaction extends Closeable {
*/
void abort();
+ /**
+ * Attempts to register a list of materialized scans against this transaction's read set.
+ * <p>
+ * Spark calls this when considering reuse of a cached subtree during the transaction. The
+ * list contains every materialized scan in the candidate cached subtree. That may include
+ * scans that belong to catalogs other than this transaction's catalog. The connector can
+ * decide which scans to register or ignore.
+ * <p>
+ * The connector decides whether reusing the cached snapshots is compatible with the
Review Comment:
We can do some sanity checks for versions but it should not be in the public docs.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56995][SQL][DML] Allow dataframe caching in the DSv2 Transaction API [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on PR #56121:
URL: https://github.com/apache/spark/pull/56121#issuecomment-4615297648
@huaxingao, hopefully, @andreaschat-db will update the PR tomorrow and we can get it in.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org