Skip to content

[SPARK-52312][SQL] Ignore V2WriteCommand when caching DataFrame #51032

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 4 commits into from
Closed

[SPARK-52312][SQL] Ignore V2WriteCommand when caching DataFrame #51032

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
18b06c7
Ignore V2WriteCommand when caching
qiyuandong-db May 27, 2025
e9942b7
Remove repeated tests
qiyuandong-db May 28, 2025
7c6cc96
Move IgnoreCachedData check to cacheQueryInternal
qiyuandong-db May 28, 2025
05539d3
address comments
qiyuandong-db May 29, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,11 @@ trait KeepAnalyzedQuery extends Command {
/**
* Base trait for DataSourceV2 write commands
*/
trait V2WriteCommand extends UnaryCommand with KeepAnalyzedQuery with CTEInChildren {
trait V2WriteCommand
extends UnaryCommand
with KeepAnalyzedQuery
with CTEInChildren
with IgnoreCachedData {
def table: NamedRelation
def query: LogicalPlan
def isByName: Boolean
Expand Down
5 changes: 5 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,11 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
storageLevel: StorageLevel): Unit = {
if (storageLevel == StorageLevel.NONE) {
// Do nothing for StorageLevel.NONE since it will not actually cache any data.
} else if (unnormalizedPlan.isInstanceOf[IgnoreCachedData]) {
logWarning(
log"Asked to cache a plan that is inapplicable for caching: " +
log"${MDC(LOGICAL_PLAN, unnormalizedPlan)}"
)
} else if (lookupCachedDataInternal(normalizedPlan).nonEmpty) {
logWarning("Asked to cache already cached data.")
} else {
Expand Down
11 changes: 11 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/DataFrameWriterV2Suite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -875,4 +875,15 @@ class DataFrameWriterV2Suite extends QueryTest with SharedSparkSession with Befo
}
}
}

test("SPARK-52312: caching dataframe created from INSERT shouldn't re-execute the command") {
spark.sql("CREATE TABLE testcat.table_name (c1 int, c2 string) USING foo")

val insertDF = spark.sql("INSERT INTO testcat.table_name VALUES (1, 'a'), (2, 'b')")
checkAnswer(spark.table("testcat.table_name"), Seq(Row(1, "a"), Row(2, "b")))

// Caching the DataFrame created from INSERT should not re-execute the command
insertDF.cache()
checkAnswer(spark.table("testcat.table_name"), Seq(Row(1, "a"), Row(2, "b")))
}
}