Skip to content

introduce /druid/v3 query endpoint that gives query responseContext #3319

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
himanshug wants to merge 1 commit into from
Closed

introduce /druid/v3 query endpoint that gives query responseContext #3319

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 49 additions & 4 deletions docs/content/querying/querying.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,17 +11,62 @@ query is expressed in JSON and each of these node types expose the same
REST query interface. For normal Druid operations, queries should be issued to the broker nodes. Queries can be posted
to the queryable nodes like this -

```bash
curl -X POST '<queryable_host>:<port>/druid/v2/?pretty' -H 'Content-Type:application/json' -d @<query_json_file>
```

```bash
curl -X POST '<queryable_host>:<port>/druid/v2/?pretty' -H 'Content-Type:application/json' -d @<query_json_file>
```
In response, Druid returns a JSON array of result data objects like below. For example, here is response for a sample groupBy query.

```json
[
{
"version" : "v1",
"timestamp" : "2012-01-01T00:00:00.000Z",
"event" : {
"country" : "US",
"name" : "test name1",
"data_transfer" : 500
}
},
{
"version" : "v1",
"timestamp" : "2012-01-01T00:00:00.000Z",
"event" : {
"country" : "US",
"name" : "test name2",
"data_transfer" : 354
}
}
...
]
```

Druid's native query language is JSON over HTTP, although many members of the community have contributed different
[client libraries](../development/libraries.html) in other languages to query Druid.

Druid's native query is relatively low level, mapping closely to how computations are performed internally. Druid queries
are designed to be lightweight and complete very quickly. This means that for more complex analysis, or to build
more complex visualizations, multiple Druid queries may be required.


<div class="note caution">
Another <a href="../development/experimental.html">experimental</a> HTTP query endpoint, `/druid/v3`.
</div>
Sometimes it is desirable to be able to send some contextual information back to druid client in addition to the result data objects.
Existing response format of JSON array did not allow that. So, a new HTTP endpoint is created to handle this use case.
If client sends a druid query to `/druid/v3` like
```bash
curl -X POST '<queryable_host>:<port>/druid/v3/?pretty' -H 'Content-Type:application/json' -d @<query_json_file>
```
Then the response format is as follows.

```json
{
"result": [ ...... result data objects ..... ],
"context": { .... context object.... }
}
```
JSON array contained in "result" attribute is same as the full response object from `/druid/v2` . Note that, first "results" are streamed back to the client and then context object follows.

Available Queries
-----------------

Expand Down
156 changes: 86 additions & 70 deletions server/src/main/java/io/druid/server/QueryResource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,76 +241,7 @@ public Object accumulate(Object accumulated, Object in)
}
);

try {
final Query theQuery = query;
final QueryToolChest theToolChest = toolChest;
Response.ResponseBuilder builder = Response
.ok(
new StreamingOutput()
{
@Override
public void write(OutputStream outputStream) throws IOException, WebApplicationException
{
// json serializer will always close the yielder
CountingOutputStream os = new CountingOutputStream(outputStream);
jsonWriter.writeValue(os, yielder);

os.flush(); // Some types of OutputStream suppress flush errors in the .close() method.
os.close();

final long queryTime = System.currentTimeMillis() - start;
emitter.emit(
DruidMetrics.makeQueryTimeMetric(theToolChest, jsonMapper, theQuery, req.getRemoteAddr())
.setDimension("success", "true")
.build("query/time", queryTime)
);
emitter.emit(
DruidMetrics.makeQueryTimeMetric(theToolChest, jsonMapper, theQuery, req.getRemoteAddr())
.build("query/bytes", os.getCount())
);

requestLogger.log(
new RequestLogLine(
new DateTime(start),
req.getRemoteAddr(),
theQuery,
new QueryStats(
ImmutableMap.<String, Object>of(
"query/time", queryTime,
"query/bytes", os.getCount(),
"success", true
)
)
)
);
}
},
contentType
)
.header("X-Druid-Query-Id", queryId);

//Limit the response-context header, see https://github.com/druid-io/druid/issues/2331
//Note that Response.ResponseBuilder.header(String key,Object value).build() calls value.toString()
//and encodes the string using ASCII, so 1 char is = 1 byte
String responseCtxString = jsonMapper.writeValueAsString(responseContext);
if (responseCtxString.length() > RESPONSE_CTX_HEADER_LEN_LIMIT) {
log.warn("Response Context truncated for id [%s] . Full context is [%s].", queryId, responseCtxString);
responseCtxString = responseCtxString.substring(0, RESPONSE_CTX_HEADER_LEN_LIMIT);
}

return builder
.header("X-Druid-Response-Context", responseCtxString)
.build();
}
catch (Exception e) {
// make sure to close yielder if anything happened before starting to serialize the response.
yielder.close();
throw Throwables.propagate(e);
}
finally {
// do not close yielder here, since we do not want to close the yielder prior to
// StreamingOutput having iterated over all the results
}
return buildQueryResponse(req, query, toolChest, jsonWriter, contentType, start, yielder, responseContext);
}
catch (QueryInterruptedException e) {
try {
Expand Down Expand Up @@ -406,4 +337,89 @@ public void write(OutputStream outputStream) throws IOException, WebApplicationE
Thread.currentThread().setName(currThreadName);
}
}

protected Response buildQueryResponse(
final HttpServletRequest req,
final Query query,
final QueryToolChest toolChest,
final ObjectWriter jsonWriter,
final String contentType,
final long startTime,
final Yielder yielder,
final Map<String, Object> responseContext
)
throws IOException
{
try {
Response.ResponseBuilder builder = Response
.ok(
new StreamingOutput()
{
@Override
public void write(OutputStream outputStream) throws IOException, WebApplicationException
{
// json serializer will always close the yielder
CountingOutputStream os = new CountingOutputStream(outputStream);

try {
Copy link
Copy Markdown
Contributor

@drcrallen drcrallen Aug 5, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same comment as below about

try(final CountingOutputStream os = new CountingOutputStream(outputStream))
{
  .....
  os.flush();
}

Copy link
Copy Markdown
Contributor Author

@himanshug himanshug Aug 5, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same comment as above

jsonWriter.writeValue(os, yielder);
os.flush(); // Some types of OutputStream suppress flush errors in the .close() method.
} finally {
os.close();
}

final long queryTime = System.currentTimeMillis() - startTime;
emitter.emit(
DruidMetrics.makeQueryTimeMetric(toolChest, jsonMapper, query, req.getRemoteAddr())
.setDimension("success", "true")
.build("query/time", queryTime)
);
emitter.emit(
DruidMetrics.makeQueryTimeMetric(toolChest, jsonMapper, query, req.getRemoteAddr())
.build("query/bytes", os.getCount())
);

requestLogger.log(
new RequestLogLine(
new DateTime(startTime),
req.getRemoteAddr(),
query,
new QueryStats(
ImmutableMap.<String, Object>of(
"query/time", queryTime,
"query/bytes", os.getCount(),
"success", true
)
)
)
);
}
},
contentType
)
.header("X-Druid-Query-Id", query.getId());

//Limit the response-context header, see https://github.com/druid-io/druid/issues/2331
//Note that Response.ResponseBuilder.header(String key,Object value).build() calls value.toString()
//and encodes the string using ASCII, so 1 char is = 1 byte
String responseCtxString = jsonMapper.writeValueAsString(responseContext);
if (responseCtxString.length() > RESPONSE_CTX_HEADER_LEN_LIMIT) {
log.warn("Response Context truncated for id [%s] . Full context is [%s].", query.getId(), responseCtxString);
responseCtxString = responseCtxString.substring(0, RESPONSE_CTX_HEADER_LEN_LIMIT);
}

return builder
.header("X-Druid-Response-Context", responseCtxString)
.build();
}
catch (Exception e) {
// make sure to close yielder if anything happened before starting to serialize the response.
yielder.close();
throw Throwables.propagate(e);
}
finally {
// do not close yielder here, since we do not want to close the yielder prior to
// StreamingOutput having iterated over all the results
}
}
}
Loading