Parsing and semantic analysis

The parse phase is the one that checks your SQL syntax. If you make a SQL syntax error, Drill will display a long error message that attempts to explain the error. (Error reporting is a work in progress in Drill.)

The next step looks up names, and this can seem complex to those accustomed to databases with schemas. At prepare time, Drill knows the names of databases (called storage plug-ins in Drill) and tables. However, Drill does not know the names of columns. So, it postpones some decisions about columns to runtime. In particular, Drill does not know the data types of columns until it runs the query. This means that type-related errors that you might expect to be caught at prepare time are, in Drill, found only at runtime.

Furthermore, Drill will accept any column name, even the names of columns that don’t actually exist in your table. Again, Drill does not know the set of available columns at prepare time. Because tables are just (collections of) files, it might even be that newer versions of the file have the requested column, but older files do not. Drill’s schema-on-read system can handle such scenarios, whereas a system based on metadata cannot.

Logical and physical plans

Once the names are known, Drill uses Calcite to work out a logical plan for the query. For example, Drill might identify that you want to join tables A and B. The next step, the physical plan, works out the best way to implement the join. (Like most query engines, Drill implements multiple join operators and chooses the lowest-cost alternative for each query.) 

The logical and physical plans are key to understanding how Drill operates and how to optimize queries. The Drill Web Console provides a query profile for each query. The query profile includes a visualization of the query plan in a form that is a hybrid of the logical and physical plans. (The diagram shows decisions made in the physical plan but omits many of the gory details, making the diagram closer to the logical plan.)

A query plan is a set of operators grouped into major fragments. An operator is a distinct operation such as “scan a file,” “filter a result set,” or “perform a join.” Often, several operators can be combined into a pipeline. For example, the scan operation is typically combined with a filter operation.

Because Drill is distributed, data must sometimes be exchanged among Drillbits (called a “shuffle” in MapReduce and Spark). For example, to perform a GROUP BY operation, like rows are collected from all scan operators and sent to the Drillbit that will perform the aggregation for each group.

The set of operations that can be done without an exchange is grouped into a thread of execution called a major fragment. The Drill web console query profile view uses colors to show the set of operators that belong to the same major fragment. Every major fragment starts with either a scan or an exchange receiver. Each major fragment ends with either an exchange sender or (for the root fragment) a send back to the Drill client.

Operators make up a tree. The root is the screen operator (the one that talks to the Drill client). The leaves are scan operators that read data into Drill. Likewise, major fragments form a tree, but at a higher level of abstraction. The Drill web console displays the tree with the screen (root) operator at the top, and the scans (leaves) at the bottom.

Data flows from the leaves to the intermediate nodes and ultimately to the root (screen). From the perspective of an internal node, Drill uses the term upstream to indicate the source of data, which is toward the leaves. Similarly, the term downstream indicates the destination of the data, which is toward the root. (Confusingly, in the Drill web console, upstream is toward the bottom of the image; downstream is toward the top).

Distribution

The final step of query preparation is to distribute the physical plan across nodes. Because each major fragment is a single thread of execution, Drill performs distribution at the level of the major fragment. If, for instance, a major fragment scans a file, Drill will slice the fragment into pieces: one for each input file (if scanning a directory) and for each HDFS block in each file. The result is a set of slices (called minor fragments)—the smallest possible units of work in Drill.

Drill distributes the set of minor fragments to Drillbits for execution. The slicing process is a bit complex but well worth understanding because it helps you to understand the query profile and how best to tune query execution.

When parallelizing fragments, Drill considers where best to place each. For a scan, each fragment corresponds to one HDFS block for one file, so the best place to execute that fragment is on a node with the data. Because HDFS uses three-way replication, Drill has three “best” choices for the node to hold the fragment. Preferring to place a fragment on the same node as the data is called data affinity. Data affinity is only possible when Drill runs on the same node as the data. When compute and storage are separated (as when using Amazon S3), data affinity is ignored and Drill simply divides up work randomly.

Suppose that your file has 1,000 blocks, but your cluster has only 100 CPUs. Drill could create 1,000 different scan fragments, but doing so would simply create far more threads than CPUs, causing unnecessary CPU contention. Instead, Drill picks an ideal slice target per node. (By default, this number is 70% of the number of cores on the node.) Drill will create no more than this number of minor fragments. It then assigns files or file blocks to minor fragments, combining blocks if needed. In this example, with 100 cores, Drill will create 70 units of work, each of which will read approximately 14 blocks. Of course, if your query reads just a few blocks, Drill will create fewer minor fragments than the maximum.

Parallelizing of internal operators is similar, but without data affinity considerations. Instead, Drill might group data based on hash keys (for joins and grouping) or other considerations.

The slice target mentioned earlier applies to the slices of each major fragment. At present, Drill slices each major fragment independently. If your query is complex and has, say, 10 major fragments, Drill will parallelize each to 70% of the number of cores on your system. The result will be seven threads per core. Further, each query is parallelized independently of other queries (which might be planned on different Drillbits). As a result, in extreme cases Drill might create far more threads than the system has cores, leading to thrashing. If you find yourself pushing Drill to these extremes, consider enabling admission control (discussed in Chapter 9) to limit the number of queries.

Drill uses ZooKeeper to determine the set of Drillbits in the cluster. Thus, if you add a new Drillbit to the cluster, Drill will use it for new queries as soon as the Drillbit registers itself with ZooKeeper.

It is worth noting that the root fragment (the one with the screen operator) is special: it always has just one minor fragment, and that minor fragment always runs on the Foreman node for that session.

For those familiar with Apache Spark, Drill and Spark use similar structures but different names. Table 3-1 provides a quick reference.

Data representation

We have seen that Drill distributes work to Drillbits in the form of minor fragments, each of which contains a set of operators. Leaf fragments contain a set of one or more scan operators. The next question is: how does Drill represent the rows within the query?

Drill is a SQL engine, meaning that data is represented as rows and columns. JDBC and ODBC present query results one row at a time, with methods to access each column.

Drill, however, is a big data engine; processing data at the row level would not be very efficient. Thus, even though Drill processes data row by row, it does so by grouping rows into record batches that range in size from a few thousand up to 65,536 rows. Record batches are the unit of exchange at shuffles, and the units of data returned to the Drill client. The JDBC and ODBC drivers receive record batches and then iterate over them to return individual rows.

Suppose that your query reads a block of data that contains a million rows. Drill will read this data as a set of, say, 20 record batches with 50,000 records per batch. (The actual batch size varies depending on data format and size.) The minor fragment passes the record batches from one operator to the next, applying varying transforms. When a batch reaches the top of the fragment, Drill sends the batch over the wire to the next minor fragment up the tree.

Query engines typically use one of two formats for data: row oriented or columnar. Drill is a columnar engine. This means that Drill stores data as columns, termed value vectors. Think of a value vector as an array of values. For example, if you have rows of data that contain an integer column N, all values for N are stored together in an array. Rows are simply a common index into a group of value vectors. The Drill record batch is thus simply a collection of value vectors.

Some of Drill’s older materials talked about vectorized processing, the idea being that Drill might eventually use the Single Instruction, Multiple Data (SIMD) instructions of modern processors to speed up execution. At present, this is mostly still an aspiration, for two reasons. First, Drill uses SQL data types, which often include null values, which SIMD instructions do not support. Second, because Drill is SQL-based, it tends to work row by row rather than column by column, which makes it difficult to use a SIMD instruction to process a column. (The Drill project welcomes contributions to solve challenges such as this.)

You will encounter the columnar representation in two ways. First, the query profile displays the number of record batches processed by each operator and fragment. Second, if you create Drill extensions (as described in later chapters), you will work directly with Drill’s value vectors and record batches.

Drill’s value vectors were the starting point for the Apache Arrow project, which provided a general-purpose columnar format for a variety of tools. Although Drill does not yet use Arrow, such use is possible in the future. (Drill has implemented substantial additional functionality not yet available in Arrow, so additional work is needed before the two can reconverge.)

Long-lived Drillbits

Even though MapReduce and thus Hive are known for extreme scalability, they are not particularly fast. Part of this is due to the way that MapReduce itself works: YARN must start a new process for each mapper or reducer.

Spark is faster because each application reuses the same set of Spark executors for each of its jobs. However, each application uses a separate set of executors, which incurs startup costs.

Drill takes the Spark idea one step further: it uses a single executor (the Drillbit) per node and for all minor fragments on that node. This architecture is similar to that used for Impala (the impalad process) and for Hive LLAP (Live Long and Process).

Because Drillbits are long-running, Drill can execute minor fragments immediately upon receipt without the overhead of starting a new process.

This speed-up comes at a cost, however: a loss of workload isolation. In some cases, excessive memory use by one query can cause another query to run out of memory. Excessive compution by one query can slow down all queries. This “noisy neighbor” problem occurs in many concurrent systems, and it typically becomes a problem in Drill only under heavy load. Because even separate processes will run into problems under heavy load, Drill’s design is not an unreasonable trade-off. Drill’s admission control mechanism can help reduce the noisy-neighbor problem if it does occur.

Code generation

Spark requires separate executors for each application because Spark users express their applications as code (in Scala, Java, or Python). Similarly, MapReduce must distribute application code to each mapper or reducer. Drill avoids the code distribution step for two reasons:

• Drill uses SQL as its input language, avoiding the need to execute application-specific code.

• Drill generates Java code on each Drillbit for each query, allowing a single Drillbit to handle any number of queries.

Earlier, we discussed that Drill translates a SQL statement into a physical plan sent to the Drillbits. To gain maximum performance, each operator translates its part of the physical plan to Java code, which the operator then executes against each of its incoming record batches.

Local code generation not only avoids the cost and complexity of code distribution but is an essential aspect of Drill’s schema-on-read approach. Only when an operator sees the first batch of data will it know the type of each column. For example, a query might contain the clause A + B. Until we know the types of the columns, we don’t know whether we need to perform integer, floating-point, or decimal arithmetic. Further, different scan operators might see different versions of the file schema, and so each fragment might, in some cases, see a slightly different schema and generate different code. Thus, Drill cannot generate code up front, nor can the same code be used by all fragments.

Drill processes this sample expression in a Project operator. As soon as that operator sees the first batch of data it can determine, say, that A is an INT and B is a DOUBLE (floating-point number). This allows the operator to generate Java code to convert A from INT to DOUBLE, and then add the two DOUBLEs.

This is a trivial example. Code generation is used heavily in filter, aggregation, join, and other operations. This not only allows Drill to use the same generic Drillbit for all queries, but also execute (Drill-generated) code specific to each query.

Network exchanges

The final key performance tool in Drill is its “in-memory” processing model. MapReduce writes shuffle data to disk after the map phase, to be read in the reduce phase. Drill avoids intermediate disk files by transferring data directly over the network from the Drillbit running an upstream (sending) fragment to the Drillbit running the downstream (receiving) fragment.

The Drill value vector format is designed to optimize exchanges: the same data buffers used for processing data within operators is used in Drill’s Netty-based network layer. Drill uses a sophisticated handshake mechanism to implement backpressure: a signal to the producing fragments to stop sending if the downstream fragment can’t keep up. You will see this delay reflected in the times reported in the query profile.

If a query runs slower than you expect, it might be due to a related issue: that some operators (such as the sort operator), cannot finish until they receive all of their inputs. If data is skewed (some fragments process much more than others), the sort must wait for the slowest sender, even if all other fragments are completed. The query profile timing numbers and timing charts help you to identify such cases.

Although Drill uses network exchanges to avoid shuffle files, there are times when Drill will still write to disk. Buffering operators, such as sort, hash aggregate, and hash join operators, must store all (or for a join, part) of their input in memory. If the query includes more data than will fit in memory, Drill operators will spill data to a temporary file on the local disk (or, if configured to do so, to the distributed filesystem). Operators that support spilling gather spill statistics that are available in the query profile. Although spilling ensures that queries complete, spilling is slower than in-memory processing. If a query begins to slow unexpectedly as data sizes increase, check the query profile to see whether operators have begun to spill.