Druid: Enable customized result format for `scan-query`

[stevenchen3]

As described in the document (see here), scan-query currently supports two kinds of resultFormat, compatedList and list, and they both return data in a JSON array while keeping real actual raw data nested (e.g., events). For example:

[{
    "segmentId" : "wikipedia_editstream_2012-12-29T00:00:00.000Z_2013-01-10T08:00:00.000Z_2013-01-10T08:13:47.830Z_v9",
    "columns" : [
      "timestamp",
      "robot",
      "namespace",
      "anonymous",
      "unpatrolled",
      "page",
      "language",
      "newpage",
      "user",
      "count",
      "added",
      "delta",
      "variation",
      "deleted"
    ],
    "events" : [ {
        "timestamp" : "2013-01-01T00:00:00.000Z",
        "robot" : "1",
        "namespace" : "article",
        "anonymous" : "0",
        "unpatrolled" : "0",
        "page" : "11._korpus_(NOVJ)",
        "language" : "sl",
        "newpage" : "0",
        "user" : "EmausBot",
        "count" : 1.0,
        "added" : 39.0,
        "delta" : 39.0,
        "variation" : 39.0,
        "deleted" : 0.0
    }]
}]

Before receiving the last chunk of data, it's very difficult to extract events on a per-chunk basis. If the amount of data returned is too huge, this will cause memory pressure on the client side. If I understand correctly, however, this does not well support HTTP streaming while enabling the capability of doing some processing on per event basis before piping to next sink.

Use http4s streaming as an example, if the returned format can be customized like below (output from http://mockbin.org/stream/10):

{"type":"stream","chunk":1}
{"type":"stream","chunk":2}
{"type":"stream","chunk":3}

I can easily carry out some processing per JSON object without too much memory consumption. Such as (below snippets can be found at here):

def f(obj: Json): String = obj.spaces2 // use `circe` to format Json string

implicit val f = io.circe.jawn.CirceSupportParser.facade
val client = PooledHttp1Client()

// query druid using `scan-query`
val streams = res <- client.streaming(POST(uri, payload))(_.body.chunks.parseJsonStream)
streams.map(x => f(x))
  .through(lines)
  .through(utf8Encode)
  .to(stdout) // pipe to stdout as an example
  .onFinalize(client.shutdown).run

It would be nice if there's a plan to support this feature.

[KenjiTakahashi]

FYI: It was discussed in the scan-query PR: https://github.com/druid-io/druid/pull/3307. TL;DR is that it needs to be done on the upper level of serialization, above specific query, and is not trivial to introduce. Not saying it can't be done, though, and I'd like to see that happen too.

For now, depending on your case, you could be able to use a streaming Jackson deserializer to avoid memory pressure.

[gianm]

If you have a parser that can parse JSON arrays in a streaming fashion (like Jackson) then as @KenjiTakahashi pointed out, you could parse this output one chunk at a time.

If you have one that can't parse JSON arrays like that, but _can_ parse a stream of objects concatenated together with potential garbage between them, then you could try skipping the first character (first checking that it's in fact a [) and then parsing the rest of the stream as a series of objects. They will have , between them but maybe you can skip that.

I think we would also accept a patch that adds an option to the broker to return results as concatenated objects rather than as JSON arrays. It sounds like a simple and potentially useful patch to me (to help in situations like this).

[gianm]

That option would apply to all queries, btw, not just scan.

[stevenchen3]

@KenjiTakahashi thanks for the pointer of the PR discussions and suggesting Jackson. I found alternative streaming Json decoder (circe-fs2) that works for me at this moment.

Looks like scan-query enforces client code to handle streaming data to avoid memory pressure. Just curious why to disable the pagination feature in scan-query? Would it be good to let client code choose which way to go? :)

[KenjiTakahashi]

While we're discussing it anyway :-), there are two things I see here to sort out

1. How to handle this in the API (without breaking the current one)? I don't think that adding option to Broker specifically is good, there are cases where people are querying historicals directly. I originally thought of it as an option that client supplies along with the query. But now that I think of it again, it will probably be easier to introduce a server-side switch (but for common, not just Broker).
2. I remember looking at the code and the obvious approach of removing outer array and adding \n with setRaw (or similar method, I'm not sure about the name now) worked for JSON, but was completely breaking SMILE (it didn't even want to compile, if I remember correctly).

@stevenchen3 scan-query implementation is completely separate from select, the original implementer probably just didn't need pagination and it was less work for him to not add it.
Also, I think changing the return to what you (and me too, earlier :-)) described would be better than select style pagination.

[stevenchen3]

@KenjiTakahashi thanks for the explanation

[stale[bot]]

This issue has been marked as stale due to 280 days of inactivity. It will be closed in 2 weeks if no further activity occurs. If this issue is still relevant, please simply write any comment. Even if closed, you can still revive the issue at any time or discuss it on the [email protected] list. Thank you for your contributions.

[stale[bot]]

This issue has been closed due to lack of activity. If you think that is incorrect, or the issue requires additional review, you can revive the issue at any time.