Add covering index based query rewriter rule

[dai-chen]

Description

This PR introduces the feature to rewrite applicable queries using covering indexes, enhancing the performance of SELECT queries by allowing faster data retrieval using these indexes.

PR Planned

• Add covering index based query rewriter rule #318
• Support query rewrite for Iceberg table (IT disabled temporarily)
• Add logging or whyNot API to explain why/whyNot index applied
• Support SQL hint for Spark conf, ex. CV rewrite, hybrid scan [TBD]
• Support partial covering index rewrite [TBD]

Changes

• Implemented query rewrite logic that detects when a covering index can be utilized for a query.
• Added new SQL test cases to ensure proper functionality and to validate that queries are only rewritten when appropriate indexes are available and enabled.
• Provided a new option to enable or disable query rewrites using covering indexes.

Testing

Added several unit and integration tests to cover:

• UT: The basic functionality of query rewrite when a suitable covering index exists.
• IT: Ensuring no rewrite occurs when covering indexes are disabled via Spark configuration.
• IT: Correct interaction of query rewrites in presence of both covering and skipping indexes.

Examples:

# Create covering index as before
spark-sql> CREATE INDEX all ON ds_tables.http_logs
         > (`@timestamp`, clientip, request, status, size);
spark-sql> REFRESH INDEX all ON ds_tables.http_logs;
Time taken: 9593.567 seconds

# Disable covering index acceleration
spark-sql> set spark.flint.optimizer.covering.enabled=false;

spark-sql> EXPLAIN SELECT clientip, request AS cnt FROM ds_tables.http_logs WHERE status = 400;
== Physical Plan ==
*(1) Project [clientip#17, request#18 AS cnt#10]
+- *(1) Filter (isnotnull(status#19) AND (status#19 = 400))
   +- FileScan json ds_tables.http_logs[clientip#17,request#18,status#19,year#21,month#22,day#23]
Batched: false, DataFilters: [isnotnull(status#19), (status#19 = 400)], Format: JSON,
Location: FlintSparkSkippingFileIndex(1 paths)[s3://.../httplogs/http_logs_partitioned_json_bz2],
PartitionFilters: [], PushedFilters: [IsNotNull(status), EqualTo(status,400)],
ReadSchema: struct

# Enable covering index acceleration
spark-sql> set spark.flint.optimizer.covering.enabled=true;

spark-sql> EXPLAIN SELECT clientip, request AS cnt FROM ds_tables.http_logs WHERE status = 400;
== Physical Plan ==
*(1) Project [clientip#17, request#18 AS cnt#39]
+- BatchScan[@timestamp#16, request#18, size#20, clientip#17, status#19]
class org.apache.spark.sql.flint.FlintScan,
PushedPredicates: [status IS NOT NULL, status = 400]
 RuntimeFilters: []

# Join query
spark-sql> EXPLAIN
         > SELECT l.clientip, l.status, s.description
         > FROM ds_tables.http_logs l
         > JOIN ds_tables.http_status s ON l.status = s.status
         > WHERE clientip = '157.12.11.0';
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [clientip#17, status#19, description#51]
   +- BroadcastHashJoin [status#19], [status#50], Inner, BuildRight, false
      :- Project [clientip#17, status#19]
      :  +- BatchScan[@timestamp#16, request#18, size#20, clientip#17, status#19]
class org.apache.spark.sql.flint.FlintScan,
PushedPredicates: [clientip IS NOT NULL, clientip = '157.12.11.0', status IS NOT NULL]
RuntimeFilters: []
      +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [plan_id=71]
         +- Filter isnotnull(status#50)
            +- FileScan parquet ds_tables.http_status[status#50,description#51] Batched: true,
DataFilters: [isnotnull(status#50)], Format: Parquet,
Location: InMemoryFileIndex(1 paths)[s3://.../httpstatus], PartitionFilters: [],
PushedFilters: [IsNotNull(status)], ReadSchema: struct

Issues Resolved

#298

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.