Allow for schema pruning during update check for files to touch

[Kimahriman]

Description

Resolves #1201

Allows for schema pruning in the first part of an update to check for files to touch.

Code snippet I ran:

>>> import pyspark.sql.functions as F
>>> from delta.tables import DeltaTable
>>> table = DeltaTable.forPath(spark, "test")
>>> table.toDF().printSchema()
root
 |-- key: string (nullable = true)
 |-- value: long (nullable = true)

>>> table.update("key = 'c'", set={'value': F.lit(6)})

The execution plan for the find files to update:
before:

(1) Scan parquet 
Output [2]: [key#526, value#527L]
Batched: true
Location: TahoeBatchFileIndex [file:.../projects/delta/test]
PushedFilters: [IsNotNull(key), EqualTo(key,c)]
ReadSchema: struct<key:string,value:bigint>

after:

(1) Scan parquet 
Output [1]: [key#686]
Batched: true
Location: TahoeBatchFileIndex [file:.../projects/delta/test]
PushedFilters: [IsNotNull(key), EqualTo(key,c)]
ReadSchema: struct<key:string>

Only key is read, not value as well. The line swap should result in the same behavior, but doing the select before the nonDeterminstic UDF allows schema pruning to happen.

How was this patch tested?

Existing UTs plus screenshot of execution plan.

Does this PR introduce any user-facing changes?

Performance improvement for update with data predicate.

[Kimahriman]

I've randomly noticed Update taking an incredibly long time for certain tables. Today it was taking ~10 minute to find the files to update for a 1 GB table, and another 20-30 min to try to rewrite it. This was the first thing I found while trying to look more into that.

[scottsand-db]

I've randomly noticed Update taking an incredibly long time for certain tables. Today it was taking ~10 minute to find the files to update for a 1 GB table, and another 20-30 min to try to rewrite it. This was the first thing I found while trying to look more into that.

@Kimahriman - want to make a separate issue for this? with the full stack trace; so we can help debug.

[scottsand-db]

This change LGTM, but I'd like @vkorukanti to take a look as well.

Were you able to do any performance testing using this change? On the slow tables/queries you mentioned.

[Kimahriman]

No I haven't, only was able to verify the read schema change

[vkorukanti]

@Kimahriman It makes sense, the non-deterministic function is not allowing the schema pruning from happening. Is it possible to add a test? There is DeltaTestUtils.withLogicalPlansCaptured that you can try (not sure though if the schema pruning happens in the logical planing itself or in physical planning).

[Kimahriman]

Yeah it would be during the optimization stage, not sure if that can be captured, I can see

[Kimahriman]

Had to add a physical plan variant of that but got a test working

[scottsand-db]

Hi @Kimahriman - thanks for adding that test! I'm going to wait for @vkorukanti to take a look before I merge.

Since we are so busy with the next release of Delta Lake, and with the Data and AI summit next week, he might not get to this for a week or so. Thanks!