Add covering index in Flint Spark API

[dai-chen]

Description

1. Add covering index implementation to FlintSpark API
2. Extract FlintSparkIndexBuilder and FlintSparkSuite with common code from skipping and covering index

TODO

1. Generate unique ID column for Flint sink Idempotency
2. Support filtering condition for partial indexing
3. Update user manual after SQL support ready
4. Refactoring Flint metadata and build index API #24

Example

Manual test example as below and will move to user manual doc once SQL support is ready:

$ spark-shell \
  --jars /xxx/flint-spark-integration-assembly-0.1.0-SNAPSHOT.jar \
  --conf spark.sql.extensions=org.opensearch.flint.spark.FlintSparkExtensions  \
  --conf spark.datasource.flint.host=xxx.us-east-1.es.amazonaws.com  \
  --conf spark.datasource.flint.port=-1  \
  --conf spark.datasource.flint.scheme=https \
  --conf spark.datasource.flint.auth=sigv4  \
  --conf spark.datasource.flint.region=us-east-1

import org.opensearch.flint.spark._
import org.opensearch.flint.spark.covering._
import org.opensearch.flint.spark.FlintSpark._

val flint = new FlintSpark(spark)
val flintIndexName = FlintSparkCoveringIndex.getFlintIndexName("orderkey", "stream.lineitem_tiny")

(flint.coveringIndex()
  .name("orderkey")
  .onTable("stream.lineitem_tiny")
  .addIndexColumns("l_orderkey", "l_quantity", "l_returnflag", "l_shipdate")
  .create())

flint.refreshIndex(flintIndexName, RefreshMode.INCREMENTAL)

OpenSearch index looks like below:

GET flint_stream_lineitem_tiny_orderkey_index/_mapping
{
  "flint_stream_lineitem_tiny_orderkey_index": {
    "mappings": {
      "_meta": {
        "name": "flint_stream_lineitem_tiny_orderkey_index",
        "source": "stream.lineitem_tiny",
        "kind": "covering",
        "indexedColumns": [
          {
            "columnType": "bigint",
            "columnName": "l_orderkey"
          },
          {
            "columnType": "float",
            "columnName": "l_quantity"
          },
          {
            "columnType": "string",
            "columnName": "l_returnflag"
          },
          {
            "columnType": "date",
            "columnName": "l_shipdate"
          }
        ]
      },
      "properties": {
        "l_orderkey": {
          "type": "long"
        },
        "l_quantity": {
          "type": "float"
        },
        "l_returnflag": {
          "type": "keyword"
        },
        "l_shipdate": {
          "type": "date",
          "format": "strict_date"
        }
      }
    }
  }
}

GET flint_stream_lineitem_tiny_orderkey_index/_search
{
     ......
    "hits": [
      {
        "_index": "flint_stream_lineitem_tiny_orderkey_index",
        "_id": "TfBEQ4oB9arJAA1F1Nkv",
        "_score": 1,
        "_source": {
          "l_orderkey": 27863683,
          "l_quantity": 22,
          "l_returnflag": "A",
          "l_shipdate": "1993-07-31"
        }
      },
      {
        "_index": "flint_stream_lineitem_tiny_orderkey_index",
        "_id": "TvBEQ4oB9arJAA1F1Nkv",
        "_score": 1,
        "_source": {
          "l_orderkey": 58918182,
          "l_quantity": 7,
          "l_returnflag": "R",
          "l_shipdate": "1993-07-31"
        }
      },
      ......

Issues Resolved

#23

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.