SQL

Apache Druid supports two query languages: Druid SQL and native queries. 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

Druid SQL supports SELECT queries with the following structure:

[ EXPLAIN PLAN FOR ]
[ WITH tableName [ ( column1, column2, ... ) ] AS ( query ) ]
SELECT [ ALL | DISTINCT ] { * | exprs }
FROM { <table> | (<subquery>) | <o1> [ INNER | LEFT ] JOIN <o2> ON condition }
[ WHERE expr ]
[ GROUP BY [ exprs | GROUPING SETS ( (exprs), ... ) | ROLLUP (exprs) | CUBE (exprs) ] ]
[ HAVING expr ]
[ ORDER BY expr [ ASC | DESC ], expr [ ASC | DESC ], ... ]
[ LIMIT limit ]
[ OFFSET offset ]
[ UNION ALL <another query> ]

FROM

The FROM clause can refer to any of the following:

• Table datasources from the druid schema. This is the default schema, so Druid table datasources can be referenced as either druid.dataSourceName or simply dataSourceName.
• Lookups from the lookup schema, for example lookup.countries. Note that lookups can also be queried using the LOOKUP function.
• Subqueries.
• Joins between anything in this list, except between native datasources (table, lookup, query) and system tables. The join condition must be an equality between expressions from the left- and right-hand side of the join.
• Metadata tables from the INFORMATION_SCHEMA or sys schemas. Unlike the other options for the FROM clause, metadata tables are not considered datasources. They exist only in the SQL layer.

For more information about table, lookup, query, and join datasources, refer to the Datasources documentation.

WHERE

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 a join on the subquery, described below in the Query translation section.

GROUP BY

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 GROUP BY clause can also refer to multiple grouping sets in three ways. The most flexible is GROUP BY GROUPING SETS, for example GROUP BY GROUPING SETS ( (country, city), () ). This example is equivalent to a GROUP BY country, city followed by GROUP BY () (a grand total). With GROUPING SETS, the underlying data is only scanned one time, leading to better efficiency. Second, GROUP BY ROLLUP computes a grouping set for each level of the grouping expressions. For example GROUP BY ROLLUP (country, city) is equivalent to GROUP BY GROUPING SETS ( (country, city), (country), () ) and will produce grouped rows for each country / city pair, along with subtotals for each country, along with a grand total. Finally, GROUP BY CUBE computes a grouping set for each combination of grouping expressions. For example, GROUP BY CUBE (country, city) is equivalent to GROUP BY GROUPING SETS ( (country, city), (country), (city), () ).

Grouping columns that do not apply to a particular row will contain NULL. For example, when computing GROUP BY GROUPING SETS ( (country, city), () ), the grand total row corresponding to () will have NULL for the “country” and “city” columns. Column may also be NULL if it was NULL in the data itself. To differentiate such rows, you can use GROUPING aggregation.

When using GROUP BY GROUPING SETS, GROUP BY ROLLUP, or GROUP BY CUBE, be aware that results may not be generated in the order that you specify your grouping sets in the query. If you need results to be generated in a particular order, use the ORDER BY clause.

HAVING

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.

ORDER 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.

LIMIT

The LIMIT clause limits the number of rows returned. In some situations Druid will push down this limit to data servers, which boosts performance. Limits are always pushed down for queries that run with the native Scan or TopN query types. With the native GroupBy query type, it is pushed down when ordering on a column that you are grouping by. If you notice that adding a limit doesn’t change performance very much, then it’s possible that Druid wasn’t able to push down the limit for your query.

OFFSET

The OFFSET clause skips a certain number of rows when returning results.

If both LIMIT and OFFSET are provided, then OFFSET will be applied first, followed by LIMIT. For example, using LIMIT 100 OFFSET 10 will return 100 rows, starting from row number 10.

Together, LIMIT and OFFSET can be used to implement pagination. However, note that if the underlying datasource is modified between page fetches, then the different pages will not necessarily align with each other.

There are two important factors that can affect the performance of queries that use OFFSET:

• Skipped rows still need to be generated internally and then discarded, meaning that raising offsets to high values can cause queries to use additional resources.
• OFFSET is only supported by the Scan and GroupBy native query types. Therefore, a query with OFFSET will use one of those two types, even if it might otherwise have run as a Timeseries or TopN. Switching query engines in this way can affect performance.

UNION ALL

The “UNION ALL” operator fuses multiple queries together. Druid SQL supports the UNION ALL operator in two situations: top-level and table-level. Queries that use UNION ALL in any other way will not be able to execute.

Top-level

UNION ALL can be used at the very top outer layer of a SQL query (not in a subquery, and not in the FROM clause). In this case, the underlying queries will be run separately, back to back. Their results will be concatenated together and appear one after the other.

For example:

SELECT COUNT(*) FROM tbl WHERE my_column = 'value1'
UNION ALL
SELECT COUNT(*) FROM tbl WHERE my_column = 'value2'

With top-level UNION ALL, no further processing can be done after the UNION ALL. For example, the results of the UNION ALL cannot have GROUP BY, ORDER BY, or any other operators applied to them.

Table-level

UNION ALL can be used to query multiple tables at the same time. In this case, it must appear in a subquery in the FROM clause, and the lower-level subqueries that are inputs to the UNION ALL operator must be simple table SELECTs. Features like expressions, column aliasing, JOIN, GROUP BY, ORDER BY, and so on cannot be used. The query will run natively using a union datasource.

The same columns must be selected from each table in the same order, and those columns must either have the same types, or types that can be implicitly cast to each other (such as different numeric types). For this reason, it is generally more robust to write your queries to select specific columns. If you use SELECT *, you will need to modify your queries if a new column is added to one of the tables but not to the others.

For example:

SELECT col1, COUNT(*)
FROM (
  SELECT col1, col2, col3 FROM tbl1
  UNION ALL
  SELECT col1, col2, col3 FROM tbl2
)
GROUP BY col1

With table-level UNION ALL, the rows from the unioned tables are not guaranteed to be processed in any particular order. They may be processed in an interleaved fashion. If you need a particular result ordering, use ORDER BY on the outer query.

EXPLAIN PLAN

Add “EXPLAIN PLAN FOR” to the beginning of any query to get information about how it will be translated. In this case, the query will not actually be executed. Refer to the Query translation documentation for help interpreting EXPLAIN PLAN output.

Identifiers and literals

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.

Dynamic parameters

Druid SQL supports dynamic parameters using question mark (?) syntax, where parameters are bound to ? placeholders at execution time. To use dynamic parameters, replace any literal in the query with a ? character and provide a corresponding parameter value when you execute the query. Parameters are bound to the placeholders in the order in which they are passed. Parameters are supported in both the HTTP POST and JDBC APIs.

In certain cases, using dynamic parameters in expressions can cause type inference issues which cause your query to fail, for example:

SELECT * FROM druid.foo WHERE dim1 like CONCAT('%', ?, '%')

To solve this issue, explicitly provide the type of the dynamic parameter using the CAST keyword. Consider the fix for the preceding example:

SELECT * FROM druid.foo WHERE dim1 like CONCAT('%', CAST (? AS VARCHAR), '%')

Data types

Standard types

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.

The following table describes how Druid maps SQL types onto native types at 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).

Multi-value strings

Druid’s native type system allows strings to potentially have multiple values. These multi-value string dimensions will be reported in SQL as VARCHAR typed, and can be syntactically used like any other VARCHAR. Regular string functions that refer to multi-value string dimensions will be applied to all values for each row individually. Multi-value string dimensions can also be treated as arrays via special multi-value string functions, which can perform powerful array-aware operations.

Grouping by a multi-value expression will observe the native Druid multi-value aggregation behavior, which is similar to the UNNEST functionality available in some other SQL dialects. Refer to the documentation on multi-value string dimensions for additional details.

Because multi-value dimensions are treated by the SQL planner as VARCHAR, there are some inconsistencies between how they are handled in Druid SQL and in native queries. For example, expressions involving multi-value dimensions may be incorrectly optimized by the Druid SQL planner: multi_val_dim = 'a' AND multi_val_dim = 'b' will be optimized to false, even though it is possible for a single row to have both “a” and “b” as values for multi_val_dim. The SQL behavior of multi-value dimensions will change in a future release to more closely align with their behavior in native queries.

NULL values

The druid.generic.useDefaultValueForNull runtime property controls Druid’s NULL handling mode.

In the default value mode (true), Druid treats NULLs and empty strings interchangeably, rather than according to the SQL standard. 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. Numeric columns in this mode are not nullable; any null or missing values will be treated as zeroes.

In SQL compatible mode (false), NULLs are treated more closely to the SQL standard. The property affects both storage and querying, so for correct behavior, it should be set on all Druid service types to be available at both ingestion time and query time. There is some overhead associated with the ability to handle NULLs; see the segment internals documentation for more details.

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.

When no rows are selected, aggregate functions will return their initial value. This can occur when filtering results in no matches while aggregating values across an entire table without a grouping, or, when using filtered aggregations within a grouping. What this value is exactly varies per aggregator, but COUNT, and the various approximate count distinct sketch functions, will always return 0.

Only the COUNT, ARRAY_AGG, and STRING_AGG aggregations can accept the DISTINCT keyword.

The order of aggregation operations across segments is not deterministic. This means that non-commutative aggregation functions can produce inconsistent results across the same query.

Functions that operate on an input type of “float” or “double” may also see these differences in aggregation results across multiple query runs because of this. If precisely the same value is desired across multiple query runs, consider using the ROUND function to smooth out the inconsistencies between queries.

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

Scalar functions

Numeric functions

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 will widen 32-bit floats to 64-bit for most expressions.

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').

Reduction functions

Reduction functions operate on zero or more expressions and return a single expression. If no expressions are passed as arguments, then the result is NULL. The expressions must all be convertible to a common data type, which will be the type of the result:

• If all argument are NULL, the result is NULL. Otherwise, NULL arguments are ignored.
• If the arguments comprise a mix of numbers and strings, the arguments are interpreted as strings.
• If all arguments are integer numbers, the arguments are interpreted as longs.
• If all arguments are numbers and at least one argument is a double, the arguments are interpreted as doubles.

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

Sketch functions

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

HLL sketch functions

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

Theta sketch functions

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

Quantiles sketch functions

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

Other scalar functions

Multi-value string functions

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

Query translation

Druid SQL translates SQL queries to native queries before running them, and understanding how this translation works is key to getting good performance.

Best practices

Consider this (non-exhaustive) list of things to look out for when looking into the performance implications of how your SQL queries are translated to native queries.

1. If you wrote a filter on the primary time column __time, make sure it is being correctly translated to an "intervals" filter, as described in the Time filters section below. If not, you may need to change the way you write the filter.

2. Try to avoid subqueries underneath joins: they affect both performance and scalability. This includes implicit subqueries generated by conditions on mismatched types, and implicit subqueries generated by conditions that use expressions to refer to the right-hand side.

3. Currently, Druid does not support pushing down predicates (condition and filter) past a Join (i.e. into Join’s children). Druid only supports pushing predicates into the join if they originated from above the join. Hence, the location of predicates and filters in your Druid SQL is very important. Also, as a result of this, comma joins should be avoided.

4. Read through the Query execution page to understand how various types of native queries will be executed.

5. Be careful when interpreting EXPLAIN PLAN output, and use request logging if in doubt. Request logs will show the exact native query that was run. See the next section for more details.

6. If you encounter a query that could be planned better, feel free to raise an issue on GitHub. A reproducible test case is always appreciated.

Interpreting EXPLAIN PLAN output

The EXPLAIN PLAN functionality can help you understand how a given SQL query will be translated to native. For simple queries that do not involve subqueries or joins, the output of EXPLAIN PLAN is easy to interpret. The native query that will run is embedded as JSON inside a “DruidQueryRel” line:

> EXPLAIN PLAN FOR SELECT COUNT(*) FROM wikipedia
DruidQueryRel(query=[{"queryType":"timeseries","dataSource":"wikipedia","intervals":"-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z","granularity":"all","aggregations":[{"type":"count","name":"a0"}]}], signature=[{a0:LONG}])

For more complex queries that do involve subqueries or joins, EXPLAIN PLAN is somewhat more difficult to interpret. For example, consider this query:

> EXPLAIN PLAN FOR
> SELECT
>     channel,
>     COUNT(*)
> FROM wikipedia
> WHERE channel IN (SELECT page FROM wikipedia GROUP BY page ORDER BY COUNT(*) DESC LIMIT 10)
> GROUP BY channel
DruidJoinQueryRel(condition=[=($1, $3)], joinType=[inner], query=[{"queryType":"groupBy","dataSource":{"type":"table","name":"__join__"},"intervals":{"type":"intervals","intervals":["-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"]},"granularity":"all","dimensions":["channel"],"aggregations":[{"type":"count","name":"a0"}]}], signature=[{d0:STRING, a0:LONG}])
  DruidQueryRel(query=[{"queryType":"scan","dataSource":{"type":"table","name":"wikipedia"},"intervals":{"type":"intervals","intervals":["-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"]},"resultFormat":"compactedList","columns":["__time","channel","page"],"granularity":"all"}], signature=[{__time:LONG, channel:STRING, page:STRING}])
  DruidQueryRel(query=[{"queryType":"topN","dataSource":{"type":"table","name":"wikipedia"},"dimension":"page","metric":{"type":"numeric","metric":"a0"},"threshold":10,"intervals":{"type":"intervals","intervals":["-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"]},"granularity":"all","aggregations":[{"type":"count","name":"a0"}]}], signature=[{d0:STRING}])

Here, there is a join with two inputs. The way to read this is to consider each line of the EXPLAIN PLAN output as something that might become a query, or might just become a simple datasource. The query field they all have is called a “partial query” and represents what query would be run on the datasource represented by that line, if that line ran by itself. In some cases — like the “scan” query in the second line of this example — the query does not actually run, and it ends up being translated to a simple table datasource. See the Join translation section for more details about how this works.

We can see this for ourselves using Druid’s request logging feature. After enabling logging and running this query, we can see that it actually runs as the following native query.

{
  "queryType": "groupBy",
  "dataSource": {
    "type": "join",
    "left": "wikipedia",
    "right": {
      "type": "query",
      "query": {
        "queryType": "topN",
        "dataSource": "wikipedia",
        "dimension": {"type": "default", "dimension": "page", "outputName": "d0"},
        "metric": {"type": "numeric", "metric": "a0"},
        "threshold": 10,
        "intervals": "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z",
        "granularity": "all",
        "aggregations": [
          { "type": "count", "name": "a0"}
        ]
      }
    },
    "rightPrefix": "j0.",
    "condition": "(\"page\" == \"j0.d0\")",
    "joinType": "INNER"
  },
  "intervals": "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z",
  "granularity": "all",
  "dimensions": [
    {"type": "default", "dimension": "channel", "outputName": "d0"}
  ],
  "aggregations": [
    { "type": "count", "name": "a0"}
  ]
}

Query types

Druid SQL uses four different native query types.

• Scan is used for queries that do not aggregate (no GROUP BY, no DISTINCT).

• 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.

Time filters

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. Consider this (non-exhaustive) list of time filters that will be recognized and translated to “intervals”:

• __time >= TIMESTAMP '2000-01-01 00:00:00' (comparison to absolute time)
• __time >= CURRENT_TIMESTAMP - INTERVAL '8' HOUR (comparison to relative time)
• FLOOR(__time TO DAY) = TIMESTAMP '2000-01-01 00:00:00' (specific day)

Refer to the Interpreting EXPLAIN PLAN output section for details on confirming that time filters are being translated as you expect.

Joins

SQL join operators are translated to native join datasources as follows:

1. Joins that the native layer can handle directly are translated literally, to a join datasource whose left, right, and condition are faithful translations of the original SQL. This includes any SQL join where the right-hand side is a lookup or subquery, and where the condition is an equality where one side is an expression based on the left-hand table, the other side is a simple column reference to the right-hand table, and both sides of the equality are the same data type.

2. If a join cannot be handled directly by a native join datasource as written, Druid SQL will insert subqueries to make it runnable. For example, foo INNER JOIN bar ON foo.abc = LOWER(bar.def) cannot be directly translated, because there is an expression on the right-hand side instead of a simple column access. A subquery will be inserted that effectively transforms this clause to foo INNER JOIN (SELECT LOWER(def) AS def FROM bar) t ON foo.abc = t.def.

3. Druid SQL does not currently reorder joins to optimize queries.

Refer to the Interpreting EXPLAIN PLAN output section for details on confirming that joins are being translated as you expect.

Refer to the Query execution page for information about how joins are executed.

Subqueries

Subqueries in SQL are generally translated to native query datasources. Refer to the Query execution page for information about how subqueries are executed.

Note: Subqueries in the WHERE clause, like WHERE col1 IN (SELECT foo FROM ...) are translated to inner joins.

Approximations

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.

• Aggregation functions that are labeled as using sketches or approximations, such as APPROX_COUNT_DISTINCT, are always approximate, regardless of configuration.

A known issue with approximate functions based on data sketches

The APPROX_QUANTILE_DS and DS_QUANTILES_SKETCH functions can fail with an IllegalStateException if one of the sketches for the query hits maxStreamLength: the maximum number of items to store in each sketch. See GitHub issue 11544 for more details. To workaround the issue, increase value of the maximum string length with the approxQuantileDsMaxStreamLength parameter in the query context. Since it is set to 1,000,000,000 by default, you don’t need to override it in most cases. See accuracy information in the DataSketches documentation for how many bytes are required per stream length. This query context parameter is a temporary solution to avoid the known issue. It may be removed in a future release after the bug is fixed.

Unsupported features

Druid does not support all SQL features. In particular, the following features are not supported.

• JOIN between native datasources (table, lookup, subquery) and system tables.
• JOIN conditions that are not an equality between expressions from the left- and right-hand sides.
• JOIN conditions containing a constant value inside the condition.
• JOIN conditions on a column which contains a multi-value dimension.
• OVER clauses, and analytic functions such as LAG and LEAD.
• ORDER BY for a non-aggregating query, except for ORDER BY __time or ORDER BY __time DESC, which are supported. This restriction only applies to non-aggregating queries; you can ORDER BY any column in an aggregating query.
• DDL and DML.
• Using Druid-specific functions like TIME_PARSE and APPROX_QUANTILE_DS on system tables.

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

• Inline datasources.
• Spatial filters.
• Query cancellation.
• Multi-value dimensions are only partially implemented in Druid SQL. There are known inconsistencies between their behavior in SQL queries and in native queries due to how they are currently treated by the SQL planner.

Client APIs

HTTP POST

You can make Druid SQL queries using HTTP via POST 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'"}.

Request

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"
  }
}

Parameterized SQL queries are also supported:

{
  "query" : "SELECT COUNT(*) FROM data_source WHERE foo = ? AND __time > ?",
  "parameters": [
    { "type": "VARCHAR", "value": "bar"},
    { "type": "TIMESTAMP", "value": "2000-01-01 00:00:00" }
  ]
}

Metadata is available over HTTP POST by querying metadata tables.

Responses

Druid SQL’s HTTP POST API 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. We recommend using Avatica JDBC driver version 1.17.0 or later. Note that as of the time of this writing, Avatica 1.17.0, the latest version, does not support passing connection string parameters from the URL to Druid, so you must pass them using a Properties object. 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()) {
      // process result set
    }
  }
}

It is also possible to use a protocol buffers JDBC connection with Druid, this offer reduced bloat and potential performance improvements for larger result sets. To use it apply the following connection url instead, everything else remains the same

String url = "jdbc:avatica:remote:url=http://localhost:8082/druid/v2/sql/avatica-protobuf/;serialization=protobuf";

The protobuf endpoint is also known to work with the official Golang Avatica driver

Table metadata is available over JDBC using connection.getMetaData() or by querying the “INFORMATION_SCHEMA” tables.

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.

Dynamic Parameters

You can also use parameterized queries in JDBC code, as in this example;

PreparedStatement statement = connection.prepareStatement("SELECT COUNT(*) AS cnt FROM druid.foo WHERE dim1 = ? OR dim1 = ?");
statement.setString(1, "abc");
statement.setString(2, "def");
final ResultSet resultSet = statement.executeQuery();

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.

String url = "jdbc:avatica:remote:url=http://localhost:8082/druid/v2/sql/avatica/";
// Set any query context parameters you need here.
Properties connectionProperties = new Properties();
connectionProperties.setProperty("sqlTimeZone", "America/Los_Angeles");
connectionProperties.setProperty("useCache", "false");
try (Connection connection = DriverManager.getConnection(url, connectionProperties)) {
  // create and execute statements, process result sets, etc
}

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'

Note: INFORMATION_SCHEMA tables do not currently support Druid-specific functions like TIME_PARSE and APPROX_QUANTILE_DS. Only standard SQL functions can be used.

SCHEMATA table

INFORMATION_SCHEMA.SCHEMATA provides a list of all known schemas, which include druid for standard Druid Table datasources, lookup for Lookups, sys for the virtual System metadata tables, and INFORMATION_SCHEMA for these virtual tables. Tables are allowed to have the same name across different schemas, so the schema may be included in an SQL statement to distinguish them, e.g. lookup.table vs druid.table.

TABLES table

INFORMATION_SCHEMA.TABLES provides a list of all known tables and schemas.

COLUMNS table

INFORMATION_SCHEMA.COLUMNS provides a list of all known columns across all tables and schema.

SYSTEM SCHEMA

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

Note: “sys” tables do not currently support Druid-specific functions like TIME_PARSE and APPROX_QUANTILE_DS. Only standard SQL functions can be used.

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

If you want to retrieve segment that was compacted (ANY compaction):

SELECT * FROM sys.segments WHERE last_compaction_state is not null

or if you want to retrieve segment that was compacted only by a particular compaction spec (such as that of the auto compaction):

SELECT * FROM sys.segments WHERE last_compaction_state == 'SELECT * FROM sys.segments where last_compaction_state = 'CompactionState{partitionsSpec=DynamicPartitionsSpec{maxRowsPerSegment=5000000, maxTotalRows=9223372036854775807}, indexSpec={bitmap={type=roaring, compressRunOnSerialization=true}, dimensionCompression=lz4, metricCompression=lz4, longEncoding=longs, segmentLoader=null}}'

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;

Server configuration

Druid SQL planning occurs on the Broker and is configured by Broker runtime properties.

Security

Please see Defining SQL permissions in the basic security documentation for information on permissions needed for making SQL queries.