Add refresh index API for partition skipping index

[dai-chen]

Description

Please see more details in changes in doc: https://github.com/dai-chen/sql-1/blob/add-partion-index-building/flint/docs/index.md

Flint Metadata Spec

Update metadata with more info (index kind, column name and type). In this way, FlintSparkSkippingIndex can be deserialized out of _meta JSON.

Flint Spark API

Add refreshIndex API with FULL (batch) and INCREMENTAL (streaming) mode.

Partition Index Implementation

Followed the first approach below for now. Need to evaluate the performance later.

• Approach 1: treat partition columns as normal and use FIRST_VALUE function to aggregate
• Approach 2: treat partition columns as special and add partition columns to GROUP BY

scala> spark.sql("select first_value(year, true) as year, first_value(month, true) as month, first_value(day, true), first_value(hour, true), input_file_name() from alb_logs group by input_file_name()").explain
== Physical Plan ==
*(2) HashAggregate(keys=[_nondeterministic#146], functions=[first_value(year#33, true), first_value(month#34, true), first_value(day#35, true), first_value(hour#36, true)])
+- Exchange hashpartitioning(_nondeterministic#146, 200), ENSURE_REQUIREMENTS, [id=#112]
   +- *(1) HashAggregate(keys=[_nondeterministic#146], functions=[partial_first_value(year#33, true), partial_first_value(month#34, true), partial_first_value(day#35, true), partial_first_value(hour#36, true)])
      +- *(1) Project [year#33, month#34, day#35, hour#36, input_file_name() AS _nondeterministic#146]
         +- FileScan csv default.alb_logs[year#33,month#34,day#35,hour#36] Batched: false, DataFilters: [], Format: CSV, Location: CatalogFileIndex[s3a://maximus.dev.daichen.us-west-2/alb-log-daily], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<>

scala> spark.sql("select year, month, day, hour, input_file_name() from alb_logs group by year, month, day, hour, input_file_name()").explain
== Physical Plan ==
*(2) HashAggregate(keys=[year#33, month#34, day#35, hour#36, _nondeterministic#169], functions=[])
+- Exchange hashpartitioning(year#33, month#34, day#35, hour#36, _nondeterministic#169, 200), ENSURE_REQUIREMENTS, [id=#134]
   +- *(1) HashAggregate(keys=[year#33, month#34, day#35, hour#36, _nondeterministic#169], functions=[])
      +- *(1) Project [year#33, month#34, day#35, hour#36, input_file_name() AS _nondeterministic#169]
         +- FileScan csv default.alb_logs[year#33,month#34,day#35,hour#36] Batched: false, DataFilters: [], Format: CSV, Location: CatalogFileIndex[s3a://maximus.dev.daichen.us-west-2/alb-log-daily], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<>

Sample index data:

curl "localhost:9200/flint_test_skipping_index/_search?pretty"
{
  "took" : 121,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 18,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "flint_test_skipping_index",
        "_id" : "LWt3VIgBQCGf4ClzQDd8",
        "_score" : 1.0,
        "_source" : {
          "file_path" : "file:///.../spark-warehouse/test/year=2023/month=4/part-00000-c2f88187-da00-40aa-981a-471df2e15ca1.c000.csv",
          "year" : 2023,
          "month" : 4
        }
      },
      {
        "_index" : "flint_test_skipping_index",
        "_id" : "L2t6VIgBQCGf4ClzVzcU",
        "_score" : 1.0,
        "_source" : {
          "file_path" : "file:///.../spark-warehouse/test/year=2023/month=5/part-00000-10f5c696-8fd8-4b28-8d41-62ac9c39f905.c000.csv",
          "year" : 2023,
          "month" : 5
        }
      }
      ...

Issues Resolved

opensearch-project/opensearch-spark#2

Check List

• New functionality includes testing.
  • All tests pass, including unit test, integration test and doctest
• New functionality has been documented.
  • New functionality has javadoc added
  • New functionality has user manual doc added
• Commits are signed per the DCO using --signoff

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.

[codecov]

Codecov Report

Merging #1658 (9c6a5cd) into feature/flint (56e6520) will not change coverage.
The diff coverage is n/a.

@@               Coverage Diff                @@
##             feature/flint    opensearch-project/sql#1658   +/-   ##
================================================
  Coverage            97.19%   97.19%           
  Complexity            4107     4107           
================================================
  Files                  371      371           
  Lines                10464    10464           
  Branches               706      706           
================================================
  Hits                 10170    10170           
  Misses                 287      287           
  Partials                 7        7           

Flag	Coverage Δ
sql-engine	97.19% <ø> (ø)

Flags with carried forward coverage won't be shown. Click here to find out more.

[penghuo]

FYI, another way to parse json. https://github.com/penghuo/os-sql/blob/flint-datatype-pushdown/flint/flint-spark-integration/src/main/scala/org/apache/spark/sql/flint/mapping/FlintMapping.scala

[penghuo]

why not use Spark StructType for columnType if it is spark sql type?

[dai-chen]

sure, I may remove this columnType field from _meta and this class after we refactored the Spark SQL type => Flint type mapping

[penghuo]

does the flintType align need to align with AggregateFunction output type?

[dai-chen]

yes, it needs to align. As comment above, I may need to use AggregateFunction output type if columnType is unnecessary. I've noted this down. Thanks!