Query Performance Optimizations

specialized indexes the storage engine, execution engine, and optimizer are all aware of two special concepts that can bring about significant query performance benefits neither of these specializations are strictly required in any table or query, but the {{ocient}} engine capitalizes on the fact that for the datasets with trillions or quadrillions of records they are often advantageous time keys much of the data in hyperscale databases is produced by machines, devices, loggers, and sensors the records these entities emit often have a temporal nature they are timestamped or otherwise sequenced relative to each other at query time, it is often the case that queries themselves have temporal predicates (e g , "select all records for this user for this year ") as well when these two conditions are met, the storage layer, execution engine, optimizer, and loading infrastructure all have special handling built in for more efficiently processing these records the time based processing of records shares similarities with zone maps in that they can be used to avoid i/o for non intersecting sets of data when predicates against certain columns are present in a query however, the specific temporal nature of the data is further utilized to make loading more efficient and also impacts segment grouping and data roll off clustering keys for many data sets there is often a commonly queried set of columns that form the basis for a variety of analytical tasks for example, in a network security database recording metadata of ip traffic, it might be the case that a substantial fraction of interesting queries are concerned with the public destination ip address, represented as a column in a record, in addition to other constraints or filters when this is true for a subset of the columns in a given table, a clustering key can be defined, which enables additional specialized handling across the storage layer, execution engine, optimizer, and loading infrastructure a clustering key shares some similarities with the "partitioning key" concept found in other databases in that it is defined in terms of one or more columns in the table and also factors into locality decisions for storage however, unlike most systems, the clustering key is not a strict partitioning, and is instead only a ”hint” that promotes locality for storage to improve i/o efficiency for queries that locate records using it because it is only a (usually followed) hint, the storage layer and engine do not suffer from issues related to partitions outsizing others or forced locating of segments on particular nodes global dictionary compression global dictionary compression, or gdc , benefits both storage utilization and query throughput for columns with low cardinality (≤ 1 million distinct values) when enabled for a given column, that column’s values are mapped to a succinct global enumeration that is stored on disk and used at query time gdc is often enabled for variable length columns representing an enumerated set of potential values (e g , city, county, and country names, or classification tags) in these scenarios it provides significant compression and also enables a number of optimizations in query and join processing in this latter category, the optimizer is able to delay or even eliminate the decompression step during execution because there exists a single global key to value mapping this often results in a significant reduction in string data processing as well as network and memory bandwidth like other forms of compression, gdc is completely transparent to end users new values are added to the global enumeration automatically, and compressed values are mapped back to their original value in query results specialized types complex types tuples, arrays, matrixes in many analytical workloads, semi structured and multi dimensional data must be analyzed ocient offers complex data types to efficiently store and analyze this type of information, in many cases dramatically reducing query response times when compared to normalized schema designs each complex type allows groupings of elements to be stored and analyzed in a denormalized fashion with the parent record, and combinations of these complex types might also be used this avoids the repetitive joins across billions of records required for nested data in a normalized schema improving query response times finally, ocient’s indexing strategies extend to complex types as described here the complex types supported in ocient include arrays, matrixes, and tuples an array is an ordered list of elements the element can have any data type including a tuple array types provide operators for efficient set operations and standard sql semantics like some and all matrixes are multi dimensional arrays that share many of the same benefits as arrays when denormalizing schemas or working with matrix data for mathematical operations tuples are structured objects in a single column created from individual data elements of the same or different underlying types arrays of tuples provide the ability to denormalize structured elements that can then be jointly queried and indexed for improved performance in some cases because the array of elements can be stored in a columnar fashion and associated with the parent row, they need not be joined to the parent table repeatedly to query them performance is improved and working set memory consumption is reduced on queries that would otherwise need to join tens of billions of rows to potentially trillions of rows the flexibility of these composable complex types creates more options for database administrators to structure data and optimize performance at scale secondary indexes a variety of secondary index types are used to improve query performance secondary indexes are built and searched on a per segment basis, allowing different segments of a given table to be indexed differently where appropriate for fixed length column types, an i/o optimized inverted secondary index enables fast equality and range queries variable length columns support hash and n gram indexes that enable fast equality and like filters, respectively complex types like tuples and matrixes are a concise way to model certain datasets in order to support efficient filtering on these types and their sub components, filters are pushed into the i/o layer and use secondary indexes when possible you can create indexes on arrays of both fixed and variable length elements to optimize contains, overlaps, and other similar operations additionally, for all()/for some() syntax allows other filters to be pushed down and applied to arrays using an index, without requiring unnest in particular, for all/for some(varchar array col) like \<filter> utilizes n gram indexes to optimize textual queries over arrays of strings per segment i/o pipelines the i/o layer is designed to minimize the amount of on disk data read and processed for each query by compiling a custom i/o pipeline for each segment of data the query accesses an i/o pipeline is a directed acyclic graph that represents an efficient i/o plan for the particular columns and filters of that query combined with indexes and data that can vary from segment to segment this plan adjusts for data demographic changes over time, and for schema or index changes also just like the query plan itself, efficient pipelines are critical for performance a simple i/o pipeline example is illustrated here this graphic example shows the i/o process stages that go into action when the ocient system compiles a pipeline these stages are as follows when a query is in the io pipeline, each segment first searches indexes to minimize the data that must be read and processed the indexes help to quickly filter the rows when possible, filter predicates are pushed down and evaluated in line, further reducing the number of rows needed to read even when an appropriate index does not exist the system reads rows that pass the filters in parallel the system returns the query output simple i/o pipeline example that contains a pipeline and its elements during execution, i/o pipeline elements work together to search indexes, read column data, and apply filters to return the exact set of matching rows from each segment much like cpu instruction pipelining, the i/o pipeline attempts to mitigate latency while fully utilizing fast storage devices requiring deep i/o queues the graph for a sophisticated query can orchestrate dozens of pipeline steps in series and in parallel this layer works to keep disks saturated by coordinating i/o across segments with read ahead and a shared priority queue a good i/o pipeline determines the set of cluster key and secondary indexes to use, and an optimal ordering based on index selectivity to filter rows at maximum speed when possible, filter predicates are pushed down and evaluated in line, enabling optimizations even when an appropriate index does not exist optimizing i/o with bloom filter and index based join push down a pair of advanced techniques are employed to dynamically reduce the amount of i/o needed when processing certain types of joins index based join push down — as a running query narrows the collection of potentially matching values, an updated list of filters can be dynamically pushed down into running i/o pipelines when a matching index is present, the i/o layer can begin using it mid query to reduce i/o otherwise, this technique has the same benefits of early filtering as a pushed down bloom filter bloom filter push down — when one side of a join is fully determined, the execution engine can build a bloom filter that gets pushed down into running i/o pipelines to enable early filtering of unneeded values at the lowest level as a probabilistic data structure, bloom filters can be tuned to provide filtering benefits for even high cardinality sets both of these techniques reduce i/o on dependent columns, memory and storage bandwidth for unnecessary rows, and unneeded join processing query execution parallelism the storage layer and i/o access patterns to promote parallelism would be of limited impact if the data processing working on the actual data was not itself highly parallel to that end, the execution engine (which is named the virtual machine or vm ) is designed to maintain parallel and disjoint flow of data during data processing a query plan is a (mostly) directed graph of plan operators in a tree like arrangement at the leaves of the tree are the raw i/o operations undertaken to bring records off of ssds and into memory for processing internally, iterative processing, computation, filtering, and joining is undertaken by successive operators to produce the final results for the query the execution engine is named the vm because it structures plan operators as assembly instructions (such as add and mov ) to be scheduled and executed the fundamental organizational unit of this execution engine follows a design pattern named resource aligned parallelism where a series of independent execution units, strongly coupled with hardware resources like cores, memory, and disks, disjointly process independent data partitions of records that flow independently and in parallel within the engine because minimal to no data sharing is required across data partitions, the execution engine not only maximizes resource utilization and throughput, but also scales very well as cpus continue to include larger numbers of cores the number of data partitions across any edge within the plan tree is usually, but not always, equal to the number of distinct vm cores each vm core is a dedicated and isolated cpu core that does not process any other database or system threads (this is achieved using a {{linux}} kernel feature named "cpu isolation") then, each plan operator is locally "compiled" into a number of operator instances that is equal to the local number of vm cores in this way, any individual operation within a plan can be spread over all allocated computational units to independently process the parallel disjoint subsets of records it is to handle of particular note is the attention paid to data skew in order to maximize aggregate throughput across all parallel operator instances it is required that they perform approximately the same amount of work if work is not evenly distributed across all parallel processors, then query execution is limited to the slowest one while the others have no work even partitioning of data is easy to achieve in some cases, and impossible in others (it depends on the operation and the data), but the ocient engine undertakes a variety of special efforts to evenly balance data flow, even in non trivial cases parallel vm core operation with balanced data flow an interesting example of this is found in sort execution if one were given a set of integers to sort, a straightforward approach to parallelize the sort operation would be to split the set into a sequence of disjoint subsets s0, s1, s2, , sn where for all n , the integers in sn are all less than the integers in sn+1 then, one could sort each subset, in parallel, to end up with a totally ordered sequence of integers across all sub sequences put another way, the system splits and groups the integers at boundary values on the number line, then sorts each interval independently the problem of data skew arises when the number of rows per interval (or subset) is not roughly equivalent, with the worst case being where one interval has all the integers and the others all have none the problem is one of boundary selection any naive choice can be subject to skew to address this in the sort case, the database engine, in conjunction with information provided by the optimizer based on metadata and statistics, adjusts the boundaries for the ranges such that each boundary has a roughly equivalent number of rows implementation efficiency when discussing parallelism, you might have noticed several terms that sound very operating system like threads, disjoint cores, compilation, and so on in fact, the database engine directly implements many general pieces of functionality that are usually in the domain of the operating system, but with database particular features that outperform standard system provided ones specialized hugepage backed memory allocators designed for high throughput and low lock contention, these allocators are backed by hugepage memory, not only deriving the benefits of fewer tlb misses, but also making them suitable for userspace i/o operations because their virtual to physical address mapping is fixed special efforts were made to make them extremely fast and efficient within the resource aligned parallelism pattern cooperative per vm core scheduler this component runs per vm core, allocating cpu time to operator instances on that core to process data in this framing, each operator instance can be considered as a parallel thread on its core however, the scheduler is not a generic thread scheduler, but instead has access and visibility into inter operator data flow, concurrent query memory utilization, and workload management limits this means that it has a much richer set of information with which to make intelligent scheduling decisions that promote maximal total system throughput numa aware data processing modern multi socket systems utilize a non uniform memory architecture (numa), whereby individual cpus (and the cores in them) have both local memory and remote memory the implication of this is that a particular memory access made against local memory by code executing on a core is not only lower latency than a remote one, but also avoids an expensive hop across a dedicated memory bus with a finite amount of bandwidth because most commodity cpu packages directly integrate a pcie bus as well, it is the case that not only is there local and remote memory, but local and remote devices as well to account for this, the vm groups locally connected ram, disks, and cores into a functional unit named a silo that for all practical purposes is treated as an isolated and independent processing unit generally speaking, records for a particular disk will be read only into local memory, and only processed on local cores, all within that silo this avoids expensive cross socket communication and greatly improves throughput related links secondary indexes docid\ pvmvtea poabqrd9q vkc understanding data types docid\ iywo24lf1xyznaktpomxz related videos at the whiteboard with ocient sql at scale {{linux}} is the registered trademark of linus torvalds in the u s and other countries