SQL

Apache Druid supports two query languages: Druid SQL and native queries, which SQL queries are planned into, and which end users can also issue directly. This document describes the SQL language.

Druid SQL is a built-in SQL layer and an alternative to Druid’s native JSON-based query language, and is powered by a parser and planner based on Apache Calcite. Druid SQL translates SQL into native Druid queries on the query Broker (the first process you query), which are then passed down to data processes as native Druid queries. Other than the (slight) overhead of translating SQL on the Broker, there isn’t an additional performance penalty versus native queries.

Query syntax

Each Druid datasource appears as a table in the “druid” schema. This is also the default schema, so Druid datasources can be referenced as either druid.dataSourceName or simply dataSourceName.

Identifiers like datasource and column names can optionally be quoted using double quotes. To escape a double quote inside an identifier, use another double quote, like "My ""very own"" identifier". All identifiers are case-sensitive and no implicit case conversions are performed.

Literal strings should be quoted with single quotes, like 'foo'. Literal strings with Unicode escapes can be written like U&'fo\00F6', where character codes in hex are prefixed by a backslash. Literal numbers can be written in forms like 100 (denoting an integer), 100.0 (denoting a floating point value), or 1.0e5 (scientific notation). Literal timestamps can be written like TIMESTAMP '2000-01-01 00:00:00'. Literal intervals, used for time arithmetic, can be written like INTERVAL '1' HOUR, INTERVAL '1 02:03' DAY TO MINUTE, INTERVAL '1-2' YEAR TO MONTH, and so on.

Druid SQL supports SELECT queries with the following structure:

[ EXPLAIN PLAN FOR ]
[ WITH tableName [ ( column1, column2, ... ) ] AS ( query ) ]
SELECT [ ALL | DISTINCT ] { * | exprs }
FROM table
[ WHERE expr ]
[ GROUP BY exprs ]
[ HAVING expr ]
[ ORDER BY expr [ ASC | DESC ], expr [ ASC | DESC ], ... ]
[ LIMIT limit ]
[ UNION ALL <another query> ]

The FROM clause refers to either a Druid datasource, like druid.foo, an INFORMATION_SCHEMA table, a subquery, or a common-table-expression provided in the WITH clause. If the FROM clause references a subquery or a common-table-expression, and both levels of queries are aggregations and they cannot be combined into a single level of aggregation, the overall query will be executed as a nested GroupBy.

The WHERE clause refers to columns in the FROM table, and will be translated to native filters. The WHERE clause can also reference a subquery, like WHERE col1 IN (SELECT foo FROM ...). Queries like this are executed as semi-joins, described below.

The GROUP BY clause refers to columns in the FROM table. Using GROUP BY, DISTINCT, or any aggregation functions will trigger an aggregation query using one of Druid’s three native aggregation query types. GROUP BY can refer to an expression or a select clause ordinal position (like GROUP BY 2 to group by the second selected column).

The HAVING clause refers to columns that are present after execution of GROUP BY. It can be used to filter on either grouping expressions or aggregated values. It can only be used together with GROUP BY.

The ORDER BY clause refers to columns that are present after execution of GROUP BY. It can be used to order the results based on either grouping expressions or aggregated values. ORDER BY can refer to an expression or a select clause ordinal position (like ORDER BY 2 to order by the second selected column). For non-aggregation queries, ORDER BY can only order by the __time column. For aggregation queries, ORDER BY can order by any column.

The LIMIT clause can be used to limit the number of rows returned. It can be used with any query type. It is pushed down to data processes for queries that run with the native TopN query type, but not the native GroupBy query type. Future versions of Druid will support pushing down limits using the native GroupBy query type as well. If you notice that adding a limit doesn’t change performance very much, then it’s likely that Druid didn’t push down the limit for your query.

The “UNION ALL” operator can be used to fuse multiple queries together. Their results will be concatenated, and each query will run separately, back to back (not in parallel). Druid does not currently support “UNION” without “ALL”.

Add “EXPLAIN PLAN FOR” to the beginning of any query to see how it would be run as a native Druid query. In this case, the query will not actually be executed.

Data types and casts

Druid natively supports five basic column types: “long” (64 bit signed int), “float” (32 bit float), “double” (64 bit float) “string” (UTF-8 encoded strings and string arrays), and “complex” (catch-all for more exotic data types like hyperUnique and approxHistogram columns).

Timestamps (including the __time column) are treated by Druid as longs, with the value being the number of milliseconds since 1970-01-01 00:00:00 UTC, not counting leap seconds. Therefore, timestamps in Druid do not carry any timezone information, but only carry information about the exact moment in time they represent. See the Time functions section for more information about timestamp handling.

Null handling modes

By default Druid treats NULLs and empty strings interchangeably, rather than according to the SQL standard. As such, in this mode Druid SQL only has partial support for NULLs. For example, the expressions col IS NULL and col = '' are equivalent, and both will evaluate to true if col contains an empty string. Similarly, the expression COALESCE(col1, col2) will return col2 if col1 is an empty string. While the COUNT(*) aggregator counts all rows, the COUNT(expr) aggregator will count the number of rows where expr is neither null nor the empty string. String columns in Druid are NULLable. Numeric columns are NOT NULL; if you query a numeric column that is not present in all segments of your Druid datasource, then it will be treated as zero for rows from those segments.

If druid.generic.useDefaultValueForNull is set to false system-wide, at indexing time, data will be stored in a manner that allows distinguishing empty strings from NULL values for string columns, and will allow NULL values to be stored for numeric columns. Druid SQL will generally operate more properly and the SQL optimizer will work best in this mode, however this does come at a cost. See the segment documentation on SQL compatible null-handling for more details.

For mathematical operations, Druid SQL will use integer math if all operands involved in an expression are integers. Otherwise, Druid will switch to floating point math. You can force this to happen by casting one of your operands to FLOAT. At runtime, Druid may widen 32-bit floats to 64-bit for certain operators, like SUM aggregators.

Druid multi-value string dimensions will appear in the table schema as VARCHAR typed, and may be interacted with in expressions as such. Additionally, they can be treated as ARRAY ‘like’, via a handful of special multi-value operators. Expressions against multi-value string dimensions will apply the expression to all values of the row, however the caveat is that aggregations on these multi-value string columns will observe the native Druid multi-value aggregation behavior, which is equivalent to the UNNEST function available in many dialects. Refer to the documentation on multi-value string dimensions and Druid expressions documentation for additional details.

The following table describes how SQL types map onto Druid types during query runtime. Casts between two SQL types that have the same Druid runtime type will have no effect, other than exceptions noted in the table. Casts between two SQL types that have different Druid runtime types will generate a runtime cast in Druid. If a value cannot be properly cast to another value, as in CAST('foo' AS BIGINT), the runtime will substitute a default value. NULL values cast to non-nullable types will also be substituted with a default value (for example, nulls cast to numbers will be converted to zeroes).

Built-in functions

Aggregation functions

Aggregation functions can appear in the SELECT clause of any query. Any aggregator can be filtered using syntax like AGG(expr) FILTER(WHERE whereExpr). Filtered aggregators will only aggregate rows that match their filter. It’s possible for two aggregators in the same SQL query to have different filters.

Only the COUNT aggregation can accept DISTINCT.

For advice on choosing approximate aggregation functions, check out our approximate aggregations documentation.

Numeric functions

Numeric functions will return 64 bit integers or 64 bit floats, depending on their inputs.

String functions

String functions accept strings, and return a type appropriate to the function.

Time functions

Time functions can be used with Druid’s __time column, with any column storing millisecond timestamps through use of the MILLIS_TO_TIMESTAMP function, or with any column storing string timestamps through use of the TIME_PARSE function. By default, time operations use the UTC time zone. You can change the time zone by setting the connection context parameter “sqlTimeZone” to the name of another time zone, like “America/Los_Angeles”, or to an offset like “-08:00”. If you need to mix multiple time zones in the same query, or if you need to use a time zone other than the connection time zone, some functions also accept time zones as parameters. These parameters always take precedence over the connection time zone.

Literal timestamps in the connection time zone can be written using TIMESTAMP '2000-01-01 00:00:00' syntax. The simplest way to write literal timestamps in other time zones is to use TIME_PARSE, like TIME_PARSE('2000-02-01 00:00:00', NULL, 'America/Los_Angeles').

IP address functions

For the IPv4 address functions, the address argument can either be an IPv4 dotted-decimal string (e.g., ‘192.168.0.1’) or an IP address represented as an integer (e.g., 3232235521). The subnet argument should be a string formatted as an IPv4 address subnet in CIDR notation (e.g., ‘192.168.0.0/16’).

Comparison operators

Multi-value string functions

All ‘array’ references in the multi-value string function documentation can refer to multi-value string columns or ARRAY literals.

Sketch operators

These functions operate on expressions or columns that return sketch objects.

HLL sketch operators

The following functions operate on DataSketches HLL sketches. The DataSketches extension must be loaded to use the following functions.

Theta sketch operators

The following functions operate on theta sketches. The DataSketches extension must be loaded to use the following functions.

Quantiles sketch operators

The following functions operate on quantiles sketches. The DataSketches extension must be loaded to use the following functions.

Other functions

Unsupported features

Druid does not support all SQL features, including:

• OVER clauses, and analytic functions such as LAG and LEAD.
• JOIN clauses, other than semi-joins as described above.
• OFFSET clauses.
• DDL and DML.

Additionally, some Druid features are not supported by the SQL language. Some unsupported Druid features include:

• Multi-value dimensions.
• Set operations on DataSketches aggregators.
• Spatial filters.
• Query cancellation.

Query execution

Queries without aggregations will use Druid’s Scan native query type.

Aggregation queries (using GROUP BY, DISTINCT, or any aggregation functions) will use one of Druid’s three native aggregation query types. Two (Timeseries and TopN) are specialized for specific types of aggregations, whereas the other (GroupBy) is general-purpose.

• Timeseries is used for queries that GROUP BY FLOOR(__time TO <unit>) or TIME_FLOOR(__time, period), have no other grouping expressions, no HAVING or LIMIT clauses, no nesting, and either no ORDER BY, or an ORDER BY that orders by same expression as present in GROUP BY. It also uses Timeseries for “grand total” queries that have aggregation functions but no GROUP BY. This query type takes advantage of the fact that Druid segments are sorted by time.

• TopN is used by default for queries that group by a single expression, do have ORDER BY and LIMIT clauses, do not have HAVING clauses, and are not nested. However, the TopN query type will deliver approximate ranking and results in some cases; if you want to avoid this, set “useApproximateTopN” to “false”. TopN results are always computed in memory. See the TopN documentation for more details.

• GroupBy is used for all other aggregations, including any nested aggregation queries. Druid’s GroupBy is a traditional aggregation engine: it delivers exact results and rankings and supports a wide variety of features. GroupBy aggregates in memory if it can, but it may spill to disk if it doesn’t have enough memory to complete your query. Results are streamed back from data processes through the Broker if you ORDER BY the same expressions in your GROUP BY clause, or if you don’t have an ORDER BY at all. If your query has an ORDER BY referencing expressions that don’t appear in the GROUP BY clause (like aggregation functions) then the Broker will materialize a list of results in memory, up to a max of your LIMIT, if any. See the GroupBy documentation for details about tuning performance and memory use.

If your query does nested aggregations (an aggregation subquery in your FROM clause) then Druid will execute it as a nested GroupBy. In nested GroupBys, the innermost aggregation is distributed, but all outer aggregations beyond that take place locally on the query Broker.

Semi-join queries containing WHERE clauses like col IN (SELECT expr FROM ...) are executed with a special process. The Broker will first translate the subquery into a GroupBy to find distinct values of expr. Then, the broker will rewrite the subquery to a literal filter, like col IN (val1, val2, ...) and run the outer query. The configuration parameter druid.sql.planner.maxSemiJoinRowsInMemory controls the maximum number of values that will be materialized for this kind of plan.

For all native query types, filters on the __time column will be translated into top-level query “intervals” whenever possible, which allows Druid to use its global time index to quickly prune the set of data that must be scanned. In addition, Druid will use indexes local to each data process to further speed up WHERE evaluation. This can typically be done for filters that involve boolean combinations of references to and functions of single columns, like WHERE col1 = 'a' AND col2 = 'b', but not WHERE col1 = col2.

Approximate algorithms

Druid SQL will use approximate algorithms in some situations:

• The COUNT(DISTINCT col) aggregation functions by default uses a variant of HyperLogLog, a fast approximate distinct counting algorithm. Druid SQL will switch to exact distinct counts if you set “useApproximateCountDistinct” to “false”, either through query context or through Broker configuration.
• GROUP BY queries over a single column with ORDER BY and LIMIT may be executed using the TopN engine, which uses an approximate algorithm. Druid SQL will switch to an exact grouping algorithm if you set “useApproximateTopN” to “false”, either through query context or through Broker configuration.
• The APPROX_COUNT_DISTINCT and APPROX_QUANTILE aggregation functions always use approximate algorithms, regardless of configuration.

Client APIs

JSON over HTTP

You can make Druid SQL queries using JSON over HTTP by posting to the endpoint /druid/v2/sql/. The request should be a JSON object with a “query” field, like {"query" : "SELECT COUNT(*) FROM data_source WHERE foo = 'bar'"}.

You can use curl to send SQL queries from the command-line:

$ cat query.json
{"query":"SELECT COUNT(*) AS TheCount FROM data_source"}
$ curl -XPOST -H'Content-Type: application/json' http://BROKER:8082/druid/v2/sql/ -d @query.json
[{"TheCount":24433}]

There are a variety of connection context parameters you can provide by adding a “context” map, like:

{
  "query" : "SELECT COUNT(*) FROM data_source WHERE foo = 'bar' AND __time > TIMESTAMP '2000-01-01 00:00:00'",
  "context" : {
    "sqlTimeZone" : "America/Los_Angeles"
  }
}

Metadata is available over the HTTP API by querying system tables.

Responses

Druid SQL supports a variety of result formats. You can specify these by adding a “resultFormat” parameter, like:

{
  "query" : "SELECT COUNT(*) FROM data_source WHERE foo = 'bar' AND __time > TIMESTAMP '2000-01-01 00:00:00'",
  "resultFormat" : "object"
}

The supported result formats are:

You can additionally request a header by setting “header” to true in your request, like:

{
  "query" : "SELECT COUNT(*) FROM data_source WHERE foo = 'bar' AND __time > TIMESTAMP '2000-01-01 00:00:00'",
  "resultFormat" : "arrayLines",
  "header" : true
}

In this case, the first result returned will be a header. For the csv, array, and arrayLines formats, the header will be a list of column names. For the object and objectLines formats, the header will be an object where the keys are column names, and the values are null.

Errors that occur before the response body is sent will be reported in JSON, with an HTTP 500 status code, in the same format as native Druid query errors. If an error occurs while the response body is being sent, at that point it is too late to change the HTTP status code or report a JSON error, so the response will simply end midstream and an error will be logged by the Druid server that was handling your request.

As a caller, it is important that you properly handle response truncation. This is easy for the “object” and “array” formats, since truncated responses will be invalid JSON. For the line-oriented formats, you should check the trailer they all include: one blank line at the end of the result set. If you detect a truncated response, either through a JSON parsing error or through a missing trailing newline, you should assume the response was not fully delivered due to an error.

JDBC

You can make Druid SQL queries using the Avatica JDBC driver. Once you’ve downloaded the Avatica client jar, add it to your classpath and use the connect string jdbc:avatica:remote:url=http://BROKER:8082/druid/v2/sql/avatica/.

Example code:

// Connect to /druid/v2/sql/avatica/ on your Broker.
String url = "jdbc:avatica:remote:url=http://localhost:8082/druid/v2/sql/avatica/";
// Set any connection context parameters you need here (see "Connection context" below).
// Or leave empty for default behavior.
Properties connectionProperties = new Properties();
try (Connection connection = DriverManager.getConnection(url, connectionProperties)) {
  try (
      final Statement statement = connection.createStatement();
      final ResultSet resultSet = statement.executeQuery(query)
  ) {
    while (resultSet.next()) {
      // Do something
    }
  }
}

Table metadata is available over JDBC using connection.getMetaData() or by querying the “INFORMATION_SCHEMA” tables. Parameterized queries (using ? or other placeholders) don’t work properly, so avoid those.

Connection stickiness

Druid’s JDBC server does not share connection state between Brokers. This means that if you’re using JDBC and have multiple Druid Brokers, you should either connect to a specific Broker, or use a load balancer with sticky sessions enabled. The Druid Router process provides connection stickiness when balancing JDBC requests, and can be used to achieve the necessary stickiness even with a normal non-sticky load balancer. Please see the Router documentation for more details.

Note that the non-JDBC JSON over HTTP API is stateless and does not require stickiness.

Connection context

Druid SQL supports setting connection parameters on the client. The parameters in the table below affect SQL planning. All other context parameters you provide will be attached to Druid queries and can affect how they run. See Query context for details on the possible options.

Note that to specify an unique identifier for SQL query, use sqlQueryId instead of queryId. Setting queryId for a SQL request has no effect, all native queries underlying SQL will use auto-generated queryId.

Connection context can be specified as JDBC connection properties or as a “context” object in the JSON API.

Metadata tables

Druid Brokers infer table and column metadata for each datasource from segments loaded in the cluster, and use this to plan SQL queries. This metadata is cached on Broker startup and also updated periodically in the background through SegmentMetadata queries. Background metadata refreshing is triggered by segments entering and exiting the cluster, and can also be throttled through configuration.

Druid exposes system information through special system tables. There are two such schemas available: Information Schema and Sys Schema. Information schema provides details about table and column types. The “sys” schema provides information about Druid internals like segments/tasks/servers.

INFORMATION SCHEMA

You can access table and column metadata through JDBC using connection.getMetaData(), or through the INFORMATION_SCHEMA tables described below. For example, to retrieve metadata for the Druid datasource “foo”, use the query:

SELECT * FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = 'druid' AND TABLE_NAME = 'foo'

SCHEMATA table

TABLES table

COLUMNS table

SYSTEM SCHEMA

The “sys” schema provides visibility into Druid segments, servers and tasks.

SEGMENTS table

Segments table provides details on all Druid segments, whether they are published yet or not.

For example to retrieve all segments for datasource “wikipedia”, use the query:

SELECT * FROM sys.segments WHERE datasource = 'wikipedia'

Another example to retrieve segments total_size, avg_size, avg_num_rows and num_segments per datasource:

SELECT
    datasource,
    SUM("size") AS total_size,
    CASE WHEN SUM("size") = 0 THEN 0 ELSE SUM("size") / (COUNT(*) FILTER(WHERE "size" > 0)) END AS avg_size,
    CASE WHEN SUM(num_rows) = 0 THEN 0 ELSE SUM("num_rows") / (COUNT(*) FILTER(WHERE num_rows > 0)) END AS avg_num_rows,
    COUNT(*) AS num_segments
FROM sys.segments
GROUP BY 1
ORDER BY 2 DESC

Caveat: Note that a segment can be served by more than one stream ingestion tasks or Historical processes, in that case it would have multiple replicas. These replicas are weakly consistent with each other when served by multiple ingestion tasks, until a segment is eventually served by a Historical, at that point the segment is immutable. Broker prefers to query a segment from Historical over an ingestion task. But if a segment has multiple realtime replicas, for e.g.. Kafka index tasks, and one task is slower than other, then the sys.segments query results can vary for the duration of the tasks because only one of the ingestion tasks is queried by the Broker and it is not guaranteed that the same task gets picked every time. The num_rows column of segments table can have inconsistent values during this period. There is an open issue about this inconsistency with stream ingestion tasks.

SERVERS table

Servers table lists all discovered servers in the cluster.

To retrieve information about all servers, use the query:

SELECT * FROM sys.servers;

SERVER_SEGMENTS table

SERVER_SEGMENTS is used to join servers with segments table

JOIN between “servers” and “segments” can be used to query the number of segments for a specific datasource, grouped by server, example query:

SELECT count(segments.segment_id) as num_segments from sys.segments as segments
INNER JOIN sys.server_segments as server_segments
ON segments.segment_id  = server_segments.segment_id
INNER JOIN sys.servers as servers
ON servers.server = server_segments.server
WHERE segments.datasource = 'wikipedia'
GROUP BY servers.server;

TASKS table

The tasks table provides information about active and recently-completed indexing tasks. For more information check out the documentation for ingestion tasks.

For example, to retrieve tasks information filtered by status, use the query

SELECT * FROM sys.tasks WHERE status='FAILED';

SUPERVISORS table

The supervisors table provides information about supervisors.

For example, to retrieve supervisor tasks information filtered by health status, use the query

SELECT * FROM sys.supervisors WHERE healthy=0;

Note that sys tables may not support all the Druid SQL Functions.

Server configuration

The Druid SQL server is configured through the following properties on the Broker.

SQL Metrics

Broker will emit the following metrics for SQL.

Authorization Permissions

Please see Defining SQL permissions for information on what permissions are needed for making SQL queries in a secured cluster.