[SPARK-33567][SQL] DSv2: Use callback instead of passing Spark session and v2 relation for refreshing cache #30491
Conversation
sunchao
force-pushed
the
SPARK-33492-followup
branch
from
de468e5 to
7a9b72d
Compare
|
Test build #131696 has started for PR 30491 at commit |
...e/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
Show resolved
Hide resolved
rdblue
left a comment
rdblue
left a comment
There was a problem hiding this comment.
Looks good to me. I'll merge this in a few days, unless there are objections from others.
| writeOptions: CaseInsensitiveStringMap, | ||
| query: SparkPlan) extends V2TableWriteExec with BatchWriteHelper { | ||
| query: SparkPlan, | ||
| afterWrite: () => Unit = () => ()) extends V2TableWriteExec with BatchWriteHelper { |
There was a problem hiding this comment.
I prefer to not have the default parameter value when unnecessary.
| r.table.asWritable match { | ||
| case v1 if v1.supports(TableCapability.V1_BATCH_WRITE) => | ||
| AppendDataExecV1(v1, writeOptions.asOptions, query, r) :: Nil | ||
| AppendDataExecV1(v1, writeOptions.asOptions, query, afterWrite = refreshCache) :: Nil |
There was a problem hiding this comment.
Shall we omit afterWrite = ?
sunchao
changed the title
|
Test build #131817 has finished for PR 30491 at commit
|
| writeOptions: CaseInsensitiveStringMap, | ||
| plan: LogicalPlan, | ||
| v2Relation: DataSourceV2Relation) extends V1FallbackWriters { | ||
| afterWrite: () => Unit) extends V1FallbackWriters { |
There was a problem hiding this comment.
shall we use a consistent name "refreshCache"?
| writeOptions: CaseInsensitiveStringMap, | ||
| plan: LogicalPlan, | ||
| v2Relation: DataSourceV2Relation) extends V1FallbackWriters { | ||
| afterWrite: () => Unit) extends V1FallbackWriters { |
| session, r.table.asWritable, r, writeOptions.asOptions, planLater(query)) :: Nil | ||
| r.table.asWritable, writeOptions.asOptions, planLater(query), refreshCache(r)) :: Nil | ||
|
|
||
| case DeleteFromTable(relation, condition) => |
There was a problem hiding this comment.
For DeleteFromTable, do we need to invalidate cache too? Doesn't this command also update table data?
There was a problem hiding this comment.
Yes good catch! I think we should. I'll work on this in a separate PR.
|
Test build #131911 has finished for PR 30491 at commit
|
|
retest this please |
| catalog.invalidateTable(ident) | ||
|
|
||
| // invalidate all caches referencing the given table | ||
| // TODO(SPARK-33437): re-cache the table itself once we support caching a DSv2 table |
There was a problem hiding this comment.
@sunchao let's also fix this TODO in a separate PR.
There was a problem hiding this comment.
Sure. Will do that soon.
|
GA passed, merging to master! |
|
Test build #131951 has finished for PR 30491 at commit
|
7 participants
What changes were proposed in this pull request?
This replaces Spark session and
DataSourceV2Relationin V2 write plans by replacing them with a callbackafterWrite.Why are the changes needed?
Per discussion in #30429, it's better to not pass Spark session and
DataSourceV2Relationthrough Spark plans. Instead we can use a callback which makes the interface cleaner.Does this PR introduce any user-facing change?
No
How was this patch tested?
N/A