influxdata / influxdata/influxdb
Integrate Arrow to query / read
- Dominant language
- Rust
- Stars
- 31.7k
- Forks
- 3.7k
- Avg merge
- 13h 37m
- Merged PRs (30d)
- 8
Description
This issue addresses the section titled _Research best practices when working with Arrow_ of #10088. As outlined below, Arrow was not a simple drop-in to the existing storage read path. Some exploratory attempts to integrate it to the existing code-generated cursors and arrays failed or raised significant performance concerns. Taking a step back to explore the spec, canonical C++ implementation and the existing Go code has proven fruitful in better understanding the right path. The first section details that research and also includes a high-level review of the TSM encoding and compaction process, to ensure the design was appropriate in order to move this area forward to columnar data structures. See **Design / TODOs** section below for the list of tasks to be completed.
A complimentary issue and set of TODOs will be prepared for the encoding and compaction work.
Overview
--------
Query and compaction are independent processes that leverage a number of shared APIs to manage the reading, decoding, deletion (tombstones) and merging of TSM and cache (WAL) data. The two process diverge for specific use cases, that will be identified later. Assume these processes are performed per-shard.
For the sake of simplicity, examples will refer to the `Float` data type, however, any of the other supported data types (`Integer`, `String`, etc) could be substituted.
At a high level, the common APIs used by query and compaction are:
* Block decoders, such as `DecodeFloatBlock` or the newer, columnar API, `DecodeFloatArrayBlock`
* decoders return blocks of Time / Value data
* Time / Value blocks are encapsulated as `FloatValues` or the newer, columnar `FloatArray`
* `FloatValues` are used by the InfluxQL query engine and the compaction service
* `FloatArray` is used by the `storage/read` service
* both are represented as Go slices
* `FloatValues` → `[]struct { time int64; value float64 }`
* `FloatArray` → `struct { time []int64; value []float64 }`
* provide a number of APIs, used independently by the LSM Tree merge operations for query and compaction
* `TSMReader` for accessing the mmapped TSM files
* query and compaction use different APIs provided by `TSMReader` for reading and decoding blocks
The subsequent sections discuss the differences.
### Query / Read
At a high level, the query process is encapsulated in a _cursor_. Concretely, a cursor is
* an implementation of the LSM Tree merge operation
* merges all TSM files and the cache (WAL)
* produces merged blocks of time and value data
* in ascending or descending time order
* for a single series key + field key combination
Series keys are typically read _in random order_, as driven by the query engine.
To fulfill this requirement, a cursor is constructed with the series key to be read, any associated values from the cache1 and a `KeyCursor`. The `KeyCursor` is a lower-level construct that produces merged blocks from the TSM files for the shard.
Code generation (AKA "generics") is employed to produce concrete implementations for each supported value type and whether the data should be retrieved in ascending or descending time order. The InfluxQL query engine and `storage/read` services uses alternate implementations, which have significantly different performance characteristics. The focus will remain on the newer columnar implementations, which have been primed for moving to Arrow.
A query may span multiple shards and therefore the process will create cursors for each shard in order to aggregate or filter the data.
#### Arrow Challenges
As data flows between these layers, existing `FloatArray` blocks are mutated via the `FloatArray.Include` or `FloatArray.Exclude` APIs. In some cases, new arrays are allocated via the `FloatArray.Merge` API. These allocations occur in isolation due to no shared state or context, making the process of tracking resource usage impractical; a requirement for 2.0. This process and lack of shared context is particularly problematic for Arrow, which generally considers arrays _immutable_ and requires access to a `memory.Allocator` for building new arrays. It is worth noting that immutability may facilitate caching in the future.
### Compaction
The equivalent merge operation for compaction is implemented by the `tsmKeyIterator`. The `tsmKeyIterator` is
* an implementation of the LSM Tree merge operation
* merges a specific set of TSM files
* in series-key order
* producing merged blocks of time and value data
* in ascending time order
* for all series keys across the set of TSM files
A notable difference from the query process is data is read sequentially.
#### Arrow Challenges
Similar to the `KeyCursor` used by the query process, the `tsmKeyIterator` leverages the equivalent `Include`, `Exclude` and `Merge` APIs during the merge process, however, again the lack of shared state or context presents challenges for Arrow.
**NOTE:** The compaction service has not been updated to use the columnar decoders, which is out of scope for the goal of supporting columnar access patterns through the `storage/read` service. There is also additional work required to review and implement columnar encoders and refactor `tsmKeyIterator` and the `Cache` to use the new columnar data structures.
### Summary
Given the differences between the query and compaction processes, it may not be possible or practical to unify the entire LSM Tree merge operations. It is necessary that `FloatArray` and related types be altered to utilize a shared context in order to provide an Arrow `memory.Allocator` and facilitate tracking metrics in the future.
Design / TODOs
--------------
Although `context.Context` was considered, it is not appropriate in this scenario, as the shared state is _required_. A new data type will be created as a container for the shared state and made available through a significant portion of the read process. At a minimum, it will provide access to an Arrow `memory.Allocator` and later extended to supply various metrics / counters and resource tracking.
Much of the existing code uses code-generation to produce type-specific cursors and arrays for each of the supported types, namely `Integer`, `Float`, `Unsigned`, `String` and `Boolean`. These are currently represented as simple Go slices, so the templated code to mutate and create new blocks is easily shared. Updating the fixed-width numeric types should present few challenges, as they map natively to the memory layout of the underlying architecture and Go slices. The packed types (`String` and `Boolean`) can no longer share this generic code and therefore will require the most changes. Additionally, whilst Arrow arrays don't require the validity vectors (used to track the positions of null values in an array) and nor does InfluxDB, the builders always create them. To avoid the overhead for manually building the fixed-width types, we'll simply create the arrays directly from the decoded blocks. We may also experiment with creating an additional set of builder types that do not create the validity vectors, if that permits more reuse in the templates.
### Create execute state / context
- [ ] create shared query context
- must be an interface that provides `memory.Allocator`
- future APIs to expose metrics
- implementation lives in `storage/read`
- perhaps `tsdb` can vend a `memory.Allocator` that is optimized for block sizes?
Update following to accept / use context
- [ ] `ArraySumCursor`
- [ ] `ArrayCountCursor`
- [ ] `ArrayFilterCursor`
- [ ] `CreateCursor` API
- [ ] `KeyCursor`
- [ ] `ArrayAscendingCursor`
- [ ] `ArrayDescendingCursor`
### Update `Array` types
This collection of types represent the time / value blocks. They require some significant changes to be a container for Arrow builders and arrays vs the current implementation, which are simple Go slices.
`Float`, `Unsigned`, `Integer` are fixed-width, thus their in-memory representation matches a native slice, therefore they should be relatively easy to implement.
`String` and `Boolean` are densely packed in Arrow, so these are not a drop in replacement. Some experimentation will be required to determine best / most efficient approach for implementing `Include`, `Exclude` and `Merge` APIs. Downstream filter query cursors will also require changes, when mutating these blocks.
#### Update fixed-width `Array` types:
- [ ] store shared context and create `memory.Buffer` (for data, nullability excluded)
- [ ] alias `memory.Buffer` with native Go slices types
- [ ] implement `Include`, `Exclude`
- [ ] implement `Merge`
#### Update packed `StringArray` type:
- [ ] store shared context and `memory.Buffers`
- [ ] implement `Include`, `Exclude`
- [ ] implement `Merge`
#### Update packed `BooleanArray` type:
- [ ] store shared context and `memory.Buffers`
- [ ] implement `Include`, `Exclude`
- [ ] implement `Merge`
### Update cursors to use Arrow arrays
#### `KeyCursor`
- [ ] `KeyCursor` ascending merge
- [ ] `KeyCursor` descending merge
### Typed ascending / descending cursors
These implement the TSM + Cache LSM Tree merge operation
#### Update fixed-width cursor types:
Similarly to the `Array`, the fixed-width cursors (`Integer`, `Float` and `Unsigned`) should be relatively easy to update, due to the Arrow arrays layout in memory matching Go slices.
- [ ] `ArrayAscendingCursor`
- [ ] `ArrayDescendingCursor`
#### Update packed string cursor
- [ ] `stringArrayAscendingCursor`
- [ ] `stringArrayDescendingCursor`
#### Update packed boolean cursor
- [ ] `booleanArrayAscendingCursor`
- [ ] `booleanArrayDescendingCursor`
### Update query cursors
These cursors are implemented in the `storage/read` package.
#### Update aggregate cursors
These cursors perform aggregations for the numeric cursor types (`Integer`, `Float` and `Unsigned`) and will be easy to update.
- [ ] `ArraySumCursor`
- [ ] `ArrayCountCursor`
#### Update filter cursors
This is another cursor that mutates incoming data, and will likely required differences between the fixed-width vs packed data types. By this point, I expect patterns to be established for efficiently mutating both formats.
- [ ] fixed-width, numeric `ArrayFilterCursor`
- [ ] packed `String` and `Boolean` `ArrayFilterCursor`
### Switch query cursors to public API
Move these cursors to a public-facing API, so they can be reused directly in Cloud 2.0.
- [ ] Switch query cursors to public-facing API
**NOTE:** Additional time should be scheduled increase the reusability of the OSS storage service APIs, to avoid the constant duplication of work between OSS and Cloud 2.0.
Contributor guide
Assessment
This issue has not been assessed yet.