Optimize common case: SELECT COUNT(*) FROM Table Fix #1192

[felipepessoto]

Description

Running the query "SELECT COUNT(*) FROM Table" takes a lot of time for big tables, Spark scan all the parquet files just to return the number of rows, that information is available from Delta Logs.

Resolves #1192

How was this patch tested?

Created unit tests to validate the optimization works, including cases not covered by this optimization.

Does this PR introduce any user-facing changes?

Only performance improvement

[sezruby]

Ref) The feature may need to be revisited while delivering deletion vector #1367

[zsxwing]

This is a great improvement. A high level question: currently if there is any file missing numRecords, we will skip the optimization. I'm wondering if we can split the files to two types:

• files containing numRecords. We still apply the optimization.
• files not containing numRecords. We read the files to get the result.

At last, we can sum the results from the above two steps.

[zsxwing]

Ref) The feature may need to be revisited while delivering deletion vector #1367

Good point. Deletion Vector will introduce the following concept to solve this problem ( #1372 ):

In the presence of Deletion Vectors the statistics may be somewhat outdated, i.e. not reflecting deleted rows yet. The flag stats.tightBounds indicates whether we have tight bounds (i.e. the min/maxValue exists[^1] in the valid state of the file) or wide bounds (i.e. the minValue is <= all valid values in the file, and the maxValue >= all valid values in the file). These upper/lower bounds are sufficient information for data skipping.

[felipepessoto]

This is a great improvement. A high level question: currently if there is any file missing numRecords, we will skip the optimization. I'm wondering if we can split the files to two types:

• files containing numRecords. We still apply the optimization.
• files not containing numRecords. We read the files to get the result.

At last, we can sum the results from the above two steps.

I'm not sure if it is possible considering how Delta and Spark interact. In that case would we read the parquet data files during the plan rewrite?
It can make things too complicated, and the trade-off may not be worth it, especially if later if decide to implement GROUP BY Partition and other aggregations like MIN/MAX.

In my opinion would be better to recommend user to recompute stats if they are missing.

[Kimahriman]

Are there any plans to implement a v2 reader that has some of this type of capability more directly vs adding a bunch of custom analyzer/optimizer rules?

[felipepessoto]

@Kimahriman, I don't have any information about v2 reader plans. Could you please provide more details how it would help?

[Kimahriman]

Was more a question for the maintainers, seems like what SupportsPushDownAggregates was created for essentially

[zsxwing]

Are there any plans to implement a v2 reader that has some of this type of capability more directly vs adding a bunch of custom analyzer/optimizer rules?

No plan right now. It would be a huge effort as we need to rewrite a lot of code. TBH, I haven't looked at whether v2 APIs are sufficient for Delta today. We may need to add more changes to Spark in order to do that (such as supporting generated columns and check constraints).

[zsxwing]

I'm not sure if it is possible considering how Delta and Spark interact. In that case would we read the parquet data files during the plan rewrite?

I was thinking that we create two logical plans: one logical plan that uses a file index that returns files that don't have stats, and the other one is the new one you create in this PR. And we can just union them. But totally agree that this is complicated and we can still from the simplest one first: optimize only if all files contain numRecords.

By the way, just curious. Do you think if this optimization would give us a better TPCDS benchmark result?

[felipepessoto]

I don't believe it will help with TPCDS. I can't promise it because I'm still investigating it, but I plan to use Delta stats to do something similar to ANALYZE TABLE, that would probably help with TPCDS, but I'm facing some issues with some queries: https://issues.apache.org/jira/browse/SPARK-39971?page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel&focusedCommentId=17582211.

I'll open another issues/PR once I have more concrete information

[felipepessoto]

Hi. Anybody had a chance to review it?

[zsxwing]

Sorry for the delay. Left a few comments.

[felipepessoto]

Sorry for the delay. Left a few comments.

Thanks @zsxwing. I believe to have addressed all the comments

[felipepessoto]

Hi @zsxwing, just checking if you had a chance to validate the changes?
Thanks.

[zsxwing]

Sorry for the delay. It looks much better. Left a few minor comments and questions.

[felipepessoto]

@zsxwing I've addressed the new comments. Thanks!

[zsxwing]

Thanks for your patience! Left one minor comment to clean up the code. Otherwise, LGTM.

[felipepessoto]

@zsxwing, done, thanks.

[zsxwing]

LGTM! Thanks. Will merge this soon.