Support directly reading and writing Druid data from Spark

### Motivation

To paraphrase the [design doc](https://docs.google.com/document/d/112VsrCKhtqtUTph5yXMzsaoxtz9wX1U2poi1vxuDswY/edit#), Druid is an excellent OLAP tool that aims to serve real-time analytics. Apache Spark, on the other hand, is a de-facto standard in the industry for big data processing, which is mostly batch-oriented by nature. While Druid has a few ways to query and ingest data, most of them are not suitable to be used from within ETL workflows, and specifically - via Spark (a few use-cases are detailed in [this slack thread](https://the-asf.slack.com/archives/CJ8D1JTB8/p1581452302483600)).

For ingesting data, the primary concern is the inefficiency of needing to write data produced in a Spark application to an intermediate location, using additional resources to read the data back into memory and prepare it for ingestion, and then finally ingesting the data into Druid via a Druid indexing task. If a Spark application can write files directly to deep storage in the format needed by Druid, these wasteful intermediate steps can be avoided.

On the reading side, some users want to use Spark to perform more complicated or arbitrary operations on data stored in Druid than Druid supports, or to join Druid data against other data stores, or simply to create very large reports. While this could be done via the existing Druid query interfaces, most of these use cases are batch-oriented, and so are both computationally expensive to produce and unlikely to be repeated with any frequency. To avoid degrading the performance of a Druid cluster for interactive workloads or triggering thrash as segments page in and out of memory, Spark applications could instead read the necessary data directly from the backing segment files.

### Proposed changes

There are 3 primary ways to read and write data in Spark: the original RDD APIs, the DataSource V1 API, and the DataSource V2 API. A brief overview of some of the pros and cons of each is below:

RDD APIs
* Pros
* Most flexible
* Cons
* Unable to take full advantage of most Spark improvements over the years such as the Catalyst optimizer, predicate pushdown and input pruning on the read side
* No built-in support for transactional writes and clean up
* Difficult for users to tune properly
* Doesn't use standard Spark data APIs (`spark.read.format("druid")` and `df.write.format("druid")`)

DataSource V1
* Pros
* Stable API (supported across all 2.x releases)
* Cons
* In the process of being supplanted by DataSource API V2
* Still coupled to legacy Spark internals (`SQLContext`, etc.)
* Doesn't handle writing very well
* Difficult to push down predicates

DataSource V2
* Pros
* Current focus of Spark development efforts
* Supports predicate pushdown, better column pruning, transactional writes, columnar reads, streaming reads and writes, and more
* Cons
* Evolving API: We can support Spark 2.4, which was released in November 2018, but not earlier versions. We will need to make relatively minor code changes to support Spark 3.x, which has a circulating release candidate but no clear release data and will need time to see widespread adoption.
* Little partitioning information is available to the writer

[This presentation](https://www.slideshare.net/databricks/apache-spark-data-source-v2-with-wenchen-fan-and-gengliang-wang) goes into more detail if desired.

Because DataSource V2 is more powerful and full-featured and is the current focus of Spark development efforts, we should target this API for our development efforts. Moreover, since Spark 2.4 has been released for a year and a half and Spark traditionally sees slow uptake across major versions, Spark 2.4 will likely continue to be the dominant deployed version for some time, lessening the argument for designing against the upcoming 3.x modifications to the API.

#### Direct Reader

The direct Spark reader will have two modes of operation:

1. In the first and most common case, users will provide a data source name, one or more intervals (via Spark filters), and optionally various connection and configuration parameters. We will issue segment metadata queries to determine the schema of data source, query the metadata store to obtain the load specs for the active segments covering the desired intervals for the given data source, and then read each segment into a partition.
1. Ideally, we could construct the schema purely from queries to the `INFORMATION_SCHEMA.COLUMNS` and not have to issue potentially expensive segment metadata queries over large intervals. However, the `INFORMATION_SCHEMA` tables don't contain information on the specific complex serde used to encode a metric or on whether or not a particular column has entries with multiple values (see #9707).
2. Similarly, the `sys` tables exposed through the SQL query interface do not contain the load specs for the given segments, so segment locations will need to be queried through the coordinator API or the metadata server. Since we need to interact with the metadata server for the writer anyway and querying the metadata server is much more performant than querying the coordinator API for segment loadspecs, we will read segment locations on deep storage via SQL queries to the metadata server.
2. In the second case, we will allow users to provide a list of segments directly and bypass querying the metadata store. This will support cases where users have alternative ways of determining segment locations.

#### Direct Writer

Perhaps the biggest disadvantage of the two DataSource APIs vis-à-vis the older RDD APIs is the lack of control over the partitioning of the input data. This gives users much greater control over how data is partitioned, but also allows them to easily make suboptimal choices without meaning too. Druid requires each segment to have extensive information about broader partitioning choices in order to take advantage of most of its features (e.g. a segment must know how many other segments there are in an interval to support atomic updates, and must know _which_ segment of n it is to support contiguity checks and minor compaction, etc.). Without this information, Druid loses some flexibility and consistency guarantees.

Frustratingly, Spark shares very little information with its writers (only a partition's id, as well as task and epoch ids). This information is insufficient on its own to construct most Druid shard specs. To work around this, we can allow users to optionally provide additional partition information as options to the writers while falling back to the knowledge-less case if they don't. This will allow users who require certain Druid features to achieve them without unduly burdening users who don't.

At a high level, the Druid writer will work by constructing one or more incremental indices per partition (one per segment interval with rows in the partition unless a user provided more information) and then merging these together into segment files, publishing them to deep storage, and reporting the resulting data segments to the driver. If all partitions successfully write their segments, Spark will update the metadata store directly and the Druid coordinator will begin the loading process the next time it runs. If one or more partition write fails, Spark will delete all temporary files associated with the write attempt on its executors and delete the already written segments from deep storage. In this way, the writer should avoid polluting deep storage with segments from failed writes, and users won't be left with partially updated data sources.

#### Registries to shadow extensions

Because reading and writing data will happen on Spark clusters instead of Druid clusters, we can't take full advantage of the extensibility and dependency injection of Druid (e.g. we can't support custom metadata stores or complex metrics we don't know about the same way Druid does). To support custom extensions, we can use a registry pattern instead. By default, we can register the core implementations of various extension points and expose public APIs that allow users to register their own as well. Where we need use certain features in code, such as when we interact with deep storage or construct shard specs, we can pattern match against the registered functions for the given task. As with passing partitioning information to the writer, this pattern allows users who need more complex functionality to take advantage of it while "just working" for users who don't.

### Rationale

The basic motivation for direct readers and writers in Spark is discussed above, as are the pros and cons of the various Spark APIs for reading data into and writing data out of Spark applications. Fuller discussions of the benefits and drawbacks of various technical implementations in Spark as well as existing third-party alternatives can be found in the [design doc](https://docs.google.com/document/d/112VsrCKhtqtUTph5yXMzsaoxtz9wX1U2poi1vxuDswY/edit#).

### Operational impact

There will be no operational impacts on existing Druid clusters. This proposal would add readers and writers that can be called from within a Spark job.

### Test plan

Because the reader and writer are symmetric, we can add integration tests that write segments to ephemeral local storage and read them back and verify correctness. Beyond these integration tests and unit tests, the biggest challenge will be testing all the various deep storage and metadata server possibilities. Here we will have to rely to some degree on the existing test support for the underlying extensions.

### Future work

The primary goal of this proposal is to add readers and writers with all core features supported. The Spark DataSource V2 API allows us to envision support for many future enhancements, even if they're not part of the initial scope. Some highlights include:

- Columnar reads: Because one way Druid data is stored on disk is columnar, we can take advantage of Spark's support for columnar reads to greatly increase the read speed of many workloads.
- Structured streaming support for reading: The DataSource API V2 optionally supports reading updating data into structured streams. We can use this to support continuous reading from Druid data sources into Spark streaming jobs. The writer can also support outputting data from continually updating tables, but it probably makes more sense for users to simply emit records to Kafka or Kinesis and allow Druid indexing tasks to handle the ingestion since other users of Druid won't be able to query data being produced in Spark until it's been written to deep storage and loaded on historicals.
- Statistics: If we do use segment metadata queries to extract schema information for the reader, we can optionally pull segment statistics as well and use those to plan more efficient reads.

### Annex

[Spark read/write to Druid - design doc](https://docs.google.com/document/d/112VsrCKhtqtUTph5yXMzsaoxtz9wX1U2poi1vxuDswY/edit#)

Start by reviewing the proposed Spark DataSource V2 reader and writer design, then inspect linked pull request #10920. Trace the planned segment metadata, deep-storage, and metadata-store interactions and the proposed unit and integration test coverage. Done means direct reads and writes work with the described cleanup and correctness guarantees.

Written by the indexing model from the issue text.