[SPARK-33492][SQL] DSv2: Append/Overwrite/ReplaceTable should invalidate cache

[sunchao]

What changes were proposed in this pull request?

This adds changes in the following places:

• logic to also refresh caches referencing the target table in v2 AppendDataExec, OverwriteByExpressionExec, OverwritePartitionsDynamicExec, as well as their v1 fallbacks AppendDataExecV1 and OverwriteByExpressionExecV1.
• logic to invalidate caches referencing the target table in v2 ReplaceTableAsSelectExec and its atomic version AtomicReplaceTableAsSelectExec. These are only supported in v2 at the moment though.

In addition to the above, in order to test the v1 write fallback behavior, I extended InMemoryTableWithV1Fallback to also support batch reads.

Why are the changes needed?

Currently in DataSource v2 we don't refresh or invalidate caches referencing the target table when the table content is changed by operations such as append, overwrite, or replace table. This is different from DataSource v1, and could potentially cause data correctness issue if the staled caches are queried later.

Does this PR introduce any user-facing change?

Yes. Now When a data source v2 is cached (either directly or indirectly), all the relevant caches will be refreshed or invalidated if the table is replaced.

How was this patch tested?

Added unit tests for the new code path.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35965/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35965/

[SparkQA]

Test build #131361 has finished for PR 30429 at commit ef1a45b.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[dongjoon-hyun]

+1, LGTM.
Could you review this, @rdblue and @cloud-fan ?

[dongjoon-hyun]

Retest this please.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/36037/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/36037/

[sunchao]

The failing test in github action is unrelated - I should have rebased to the latest master.

[SparkQA]

Test build #131431 has finished for PR 30429 at commit ef1a45b.

• This patch fails SparkR unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[dongjoon-hyun]

With an integrated test result,

• SparkR test passed in GitHub Action although it failed at Jenkins.
• Scala 2.13 failure is irrelevant to this one. It's already fixed in the master by @sunchao
• K8s IT passed.

I'll merge this. Thanks, @sunchao .

[rdblue]

Sorry I'm late to help review, but this overall looks good to me.

The one minor issue I have is passing the Spark session and logical plan through the plans as new parameters. I would rather use a callback to refresh instead, so that DataSourceV2Strategy would pass what to do after the commit:

    case AppendData(r: DataSourceV2Relation, query, writeOptions, _) =>
      val refresh = () => session.sharedState.cacheManager.recacheByPlan(session, r)
      r.table.asWritable match {
        case v1 if v1.supports(TableCapability.V1_BATCH_WRITE) =>
          AppendDataExecV1(v1, writeOptions.asOptions, query, afterWrite = refresh) :: Nil
        case v2 =>
          AppendDataExec(v2, r, writeOptions.asOptions, planLater(query), afterWrite = refresh) :: Nil
      }

[cloud-fan]

+1, it makes the query plan more general.

[sunchao]

Thanks @rdblue and @cloud-fan ! yes this seems like a nice improvement. I'll address your comment in a follow-up PR.

[dongjoon-hyun]

Thank you, @rdblue , @cloud-fan , @sunchao .