[Feature] OpenSearch and Apache Spark Integration

[penghuo]

More reading at #4

Phase-0: Proof of Concepts

Goals

• Verify the solution for end to end production use case
• Demo the solution

Tasks

• Use Case Analysis
  • MV definition: build desired dashboard in Superset/Grafana and figure out MV required to accelerate the queries
  • Spark transform logic: build MV by Spark DataFrame and verify if any functional gap in Spark streaming
• Setup Spark docker cluster
  • Thrfit JDBC Server and LDAP configuration [Deprecated]
  • Docker-compose file. sql#1352
• Backend Development
  • Add OpenSearch Spark Connector
    • Build OpenSearch Spark connector. sql#1302
    • Test OpenSearch Spark Connector sql#1306
    • add docker compose for testing spark connector sql#1311
    • Adds SigV4 capability opensearch-hadoop#48
  • Maximus Implementation
    • Create External Table with AUTO_REFRESH enabled sql#1363
      • Add CREATE EXTERNAL TABLE statement sql#1394
      • Collect external table statistics sql#1362
    • Store Maximus Table metadata inside OpenSearch.
      • store metadata/index/mv in opensearch.
      • sync between table and index/mv [Deprioritized]
    • Create index on external table sql#1379
      • Extend Spark grammar with CREATE INDEX
      • Store index data in OpenSearch index
      • Auto refresh by Spark streaming execution (integrate with table metadata refresh job)
    • Create materialized view on external table sql#1407
      • Extend Spark grammar with CREATE MATERIALIZED VIEW
      • Refresh MV data by Spark streaming job
      • Add tumbling window function
    • Query rewrite with secondary index and materialized view sql#1435
      • Rewrite query to use skipping index (reuse Hyperspace)
      • Rewrite streaming query (create MV) with skipping index
      • Rewrite direct query with materialized view [Deprioritized]
• Demo
  • Create dashboard in Superset/Grafana for demo use
  • Integrate and demo with OpenSearch Observability once the UI is ready
• Release
  • OpenSearch Docker Compose sql#1480

Deliverables

• Docker: https://github.com/penghuo/os-sql/tree/feature/spark-integration/docker#setup-dev-env
• Demo: https://github.com/opensearch-project/sql/discussions/1465

---

Phase-1: Spark Connector and Flint API Support

Goals

• OpenSearch Release: 2.9.0
  • Support create spark-submit datasource in OpenSearch.
  • Support build visualization in OpenSearch Dashboard using spark datasource.
• OpenSearch Spark Extension Release.
  • Support create skipping index for Hive Table.

// 1. define the context
USING FLINT
PROPERTIES (
  OPENSEARCH = "https://my-opensearch.com"
)

// 2. create skipping index
CREATE SKIPPING INDEX index_name ON TABLE [db_name].table_name
FOR COLUMNS (col1 INDEX_TYPE, ...)
[REFRESH_ON_CREATE]            <- auto refresh the table after table create
[AUTO_REFRESH TRUE]            <- auto refresh the index when new data appended.

Tasks

• Backend Development
  • Codebase
    • Create Flint project folder sql#1573
  • Spark Connector in OpenSearch SQL
    • Add spark connector sql#1780
    • Add Spark application sql#1723
  • Add Flint core API sql#1635
    • Support FlintTable batch write sql#1653
    • Add scroll search based OpenSearchReader sql#1633
    • Add Flint client and index management API sql#1634
    • add AWS sigv4 support sql#1663
  • Flint Spark Integration
    • Flint Connector
      • Spark SPI
        • Support table read sql#1631
          • Pagination Support sql#1632
          • Partition Support
          • Filter push down sql#1641
          • add flint date type support sql#1699
        • Add Streaming write support sql#1656
      • OpenSearch Sink
        • Improve schema infer in OpenSearch-Hadoop
    • Add data skipping index support #2
      • Skipping Index SQL Support
        • Flint context configuration statement
        • Add Flint create and refresh index SQL support sql#1800
        • Add Flint describe index SQL support sql#1785
        • Add Flint drop index SQL support sql#1730
      • Skipping Index API Interface
        • Create/delete/show Index
          • Add Flint Spark index management API sql#1636
        • Refresh Index
          • Add REFRESH INDEX statement
          • Add refresh index API for partition skipping index sql#1658
      • Skipping Index Provider
        • Partition index
        • Add value set skipping index sql#1708
        • Add min max skipping index sql#1726
      • Spark query optimization
        • Query rewrite for partition skipping index sql#1690
  • Fault Tolerance
    • Streaming job can resume with checkpoint
    • Generate doc ID in build index job for idempotency sql#1803
    • Provide hybrid scan setting for consistency requirement sql#1819
  • Bug Fixes, Enhancements, Refactoring
    • Terminate refreshing job when delete index #8
    • Support more filter push down in skipping strategy #7
    • Support database name in table name #10
    • Fix doc id column not ignored as expected #9
    • Add version and update doc #20

Deliverables

• Demo (without OS Spark connector): https://github.com/opensearch-project/sql/discussions/1877

---

Phase-2: Covering Index and Materialized View Support

Goals

• OpenSearch release: preview in 2.11
• User story: [flint] As a user, I can Accelerate my Specific "time-based or Periodic" SparkSQL queries for a dataset using OnDemand Indices in OpenSearch sql#1376
• User guide: guidance for performance, cost estimation and OpenSearch/Spark cluster sizing

Non-Goals

• Covering index and MV are not accessible in Spark and thus no query rewrite support

Tasks

• Backend Development
  • [FEATURE] Add covering index support #23
    • API support
      • Add covering index in Flint Spark API #22
      • Exactly-once guarantee for covering index and MV incremental refresh #143 [TBD]
      • Add where clause support for covering index #85
      • Support included columns [Low priority]
    • SQL support
      • Add create, drop and refresh covering index SQL support #32
      • Add describe and show covering index SQL support #36
  • [FEATURE] Add materialized view support #25
    • API support
      • Add materialized view in Flint Spark API #71
      • Handle large backlog when cold start [Postponed]
    • SQL support
      • Add create and drop materialized view SQL support #73
      • Add show and desc materialized view SQL support #81
      • Add missing refresh materialized view statement #97
    • Windowing Function
      • Add tumble windowing function #37
  • [FEATURE] Add index refresh job management commands #57
    • Persisting runtime env info into Flint metadata #58
    • Add show index job SQL statement #82
    • Add recover index job statement #119
  • [FEATURE] Flint index state management #95
    • Add Flint index metadata log and transaction support #110
  • Bug Fixes, Enhancements, Refactoring
    • Fix catalog-to-spark data type mapping #17
    • Support Spark data types VarcharType and CharType #21
    • Refactoring Flint metadata and build index API #24
      • Refactor Flint index metadata #70
      • Build API refactor is done in MV API PR
    • [FEATURE] Support more options in create index statement #26
      • Support refresh interval and checkpoint location option #39
      • Add index settings option in create statement #44
      • Add more index options for streaming job #76
      • Separate auto start option from auto refresh [Low priority]
    • Support IF NOT EXISTS in create statement #42
    • Bug Fix, close scroll context #47
    • Fix catalog name missing in Flint index name #48
    • add basic auth #49
    • Fix index name with uppercase letter issue #60
    • [FEATURE] Improve validation for SQL statement #65
      • Add validation for index options #66
    • [BUG] Streaming job application exits unexpectedly #53
    • Add checkpoint mandatory configuration #92
    • Fix column name with special character when create materialized view #98
    • Add more info in Flint metadata log #125
    • Fix Flint index name and MV name bug #127
    • Revert EMR info in Flint metadata #142
    • Improve Flint index monitor error handling #158
    • Request index not exist handling #169
    • Add backoff retry capability in rest client #170
• Frontend Development
  • Table, index and MV management UI which allow user manage the table, index and MV.
  • SQL explain UI which allow user to know (1) the benefit of create index or MV and (2) the query execution plan and the cost of each operation.
• Performance Test
  • Setup Thrfit JDBC Server and LDAP configuration sql#1400

[dai-chen]

Phase-3: Flint Index Enhancement and Optimization

Goals

• OpenSearch release: TBD
• User story:
  • [EPIC] Zero-ETL - OpenSearch Table #185
  • [EPIC] Zero-ETL - AWS ALB Logs Integration #186
  • [EPIC] Zero-ETL - AWS WAF logs Integration #198
• User guide:
  • Guidance for performance, cost estimation and OpenSearch/Spark cluster sizing
  • [EPIC] Zero-ETL - Spark streaming job computation cost reduction #196

Non-Goals

• N/A

Tasks

• Spark Application
  • Bug Fixes
    • [BUG] Flint job hangs if no streaming job launched #227
    • [BUG] Stop Spark context and JVM explicitly in FlintJob #270
• PPL Extension
  • Bug Fixes
    • [BUG]ppl parse issue for backtick wrapped fields #130
    • [BUG] PPL has issues working with backticks in spark catalog namspaces #260
• Flint Extension
  • Features
    • [RFC] Flint serverless skipping index #118
      • Automatic skipping algorithm selection
      • Serverless skipping index building
      • Serverless skipping index storage
    • [FEATURE] Push mode streaming support #29
    • [FEATURE] Incremental refresh index on Hive source table #91
    • [RFC] Automatic Workload-Driven Query Acceleration by OpenSearch #128
    • [FEATURE] Support COPY operation #129
    • [FEATURE] Enrich skipping index data structure #193
    • [RFC] Proposal: introduce new Flint Table concept #217
    • [RFC] Accelerate Spark SQL queries by covering index #298
  • Enhancements
    • [FEATURE] Improve validation for SQL statement #65
      • Improve pre-validation for Flint index refresh options  #297
      • Allow non-existent checkpoint location path in index validation #313
    • [FEATURE] Covering index and materialized view refresh idempotency #88
    • [FEATURE] Support partial indexing for skipping and covering index #89
      • Add partial indexing support for skipping index #124
      • Use index filter in metadata to avoid semi-join
    • [FEATURE] Redefine drop index semantic as logical deletion #104
      • Add vacuum index API and SQL support #189
      • Change delete index API to logical delete #191
      • Rule out logical deleted skipping index in query rewrite #289
    • [FEATURE] Large backlog processing when materialized view cold start #90
    • [FEATURE] Skipping index and materialized view refresh synchronization #93
    • [FEATURE] Support incremental refresh in refresh index statement #100 [Low Priority]
    • [FEATURE] Add stateCode if job executed failed #149
    • [FEATURE] Flint metadata log improvement #121
    • [FEATURE] Checkpoint folder data cleanup limitation #102
    • [FEATURE] Spark streaming job rate limit #187
    • [FEATURE] Index refresh job management phase 2 #190
    • [FEATURE] Improve special character handling for generating Flint index name #192
    • [FEATURE] Support on-demand triggered incremental refresh #195
    • [FEATURE] Create skipping index on IP type field #203
    • [FEATURE] Support expression in indexed column #204
    • [FEATURE] Add bloom filter skipping index type #206
      • Implement BloomFilter skipping index building logic #242
      • Implement BloomFilter query rewrite (without pushdown optimization) #248
      • Implement BloomFilter query pushdown optimization #271
      • Implement adaptive BloomFilter algorithm #251
      • Add BloomFilter skipping index SQL support #283
    • [FEATURE] Avoid expensive S3 listing by using file list in skipping index #218
    • [FEATURE] Automate skipping index column and algorithm selection #221
      • Implement analyze skipping index statement #284
    • [FEATURE] Introduce bytes-based batch size config to optimize Flint write throughput  #304
  • Bug Fixes
    • [BUG] Find correct catalog name in query rewriter #54
    • [BUG] Missing errors when connecting to a Parquet table with incorrect types #101
    • [BUG] Flint not supporting Complex schema #112
    • [BUG] Field with null value is omitted when write back to result index #139
    • [BUG] Create skipping index failed on nested field of struct type #194
    • [BUG] Value set cannot handle high cardinality column #197
      • Restrict the maximum size of value set by default limit #208
      • Configure value set max size in SQL statement #210
      • Add new CollectSetLimit function to replace CollectSet [TBD]
    • [BUG] Skipping index query rewriter cannot rewrite query on SQL view #199
    • [BUG] If user provide invalid s3 location as checkpoint, streaming query still execute, but using local disk as checkpoint. #200
    • [BUG] Value_Set doc too large, Flint Request size exceeded 104857600 bytes  #202
    • [BUG] Flint data source cannot read nested field value #233
    • [BUG] Index state stuck in refreshing when Flint index deleted directly #237
    • [BUG] Gracefully terminate index refresh job when Flint index deleted accidentally #244
    • [BUG] Flint index stuck in transient state #249
    • [BUG] Show index statement failed if user index name has Flint prefix #295
  • Refactoring
    • Refactor low level OpenSearch client for general use #74
    • [FEATURE] Error message is not meaning full, "Fail to run query, cause: Failed to refresh Flint index" #201
    • Remove unnecessary locking when commit and rollback transaction #226
    • [FEATURE] Standardize Flint log output format #235
• Testing
  • Documentation
    • Add Doctest framework #28
  • Performance
    • Derived dataset auto refresh benchmark #38
    • Performance test on large dataset sql#1399
    • Estimate the cost reduction by Maximus solution sql#1401
    • Figure out recommendation on Spark cluster config and sizing sql#1402

[dai-chen]

Follow-up tracking issue: #365