Ad Hoc Queries

By Ad Hoc queries we mean queries directed at a running system. In the typical scenario originally envisioned during system development, the initial working assumption was that a system user would know all the queries and data sources needed to obtain the processed time series.

During development, however, additional scenarios emerged, assuming that the system’s operation should not be interrupted, and that additional queries should be attached to the query execution plan. We call this kind of functionality Ad Hoc queries — attached to the system while it is running, without interrupting its operation.

Fig. 47 shows the control flow described above. A file with queries and directives is first directed to the xretractor process. Then, through shared memory, the xqry process pulls data from xretractor. Using that same process, we can send a command to the xretractor process. In this command we include the text of the additional query that xretractor should attach to the tree being processed.

What can be attached at run time

The ad hoc channel accepts SELECT only. Every data source must already exist in the plan — adding a new declaration to a running system is not supported and yields:

$ xqry -a "DECLARE a BYTE STREAM C, 1 FILE 'data3.txt'"
rcv: db Fail parse: AdHoc DECLARE not supported

The reason is executional, not syntactic: a declaration added at run time does not receive a runtime logical-index base (the processing loop skips declarations when deriving it), so the first read of such a stream by any operator would be a read outside the defined range. The restriction will go away once ad hoc DECLARE gets its own path for deriving that base.

Where an ad hoc stream begins

A plan built from the start of system operation numbers records from the logical origin computed by the compiler. A query attached ad hoc has no such history — its first record is the first slot in which the runtime saw it, not slot zero of the plan. The import is atomic: the compiled tree and its stream instances are published under a common lock, and the execution loop rebuilds the timeline without rewinding, even when the new query introduces a new rate to the system.

NOTE: This behavior is covered by the issue227_join_alignment test (the adhoc-origin case).

Example

We’ll start the example by preparing a simple query:

DECLARE a BYTE STREAM A, 1 FILE 'data1.txt'
DECLARE a BYTE STREAM B, 2 FILE 'data2.txt'
SELECT * STREAM str1 FROM A+B

We’ll save the query file under the name qplan1.rql. For the query to run correctly, we also need to prepare the files data1.txt and data2.txt. I suggest filling data1.txt with consecutive numbers from 1 to 6, each on a new line, and filling data2.txt with numbers from 10 to 15. In a directory prepared this way, we run the command:

$ xretractor qplan1.rql

If we previously performed some operations in this directory and created a str1 stream with a different schema, we’ll get an error titled “Error in data descriptor file”. It will also show information about the differences between the two descriptors. In that case, the files str1 and str1.desc should be deleted and the command run again.

The xretractor process will begin processing data. At this point, open another terminal and issue the command:

$ xqry -d
|str1|1|48|24|         |0|
|   A|1|-1| 3|data1.txt|1|
|   B|2|-1| 2|data2.txt|1|
ok.

This will display, in tabular form, what’s currently being processed in the system — how many bytes have already arrived, which files the data is being read from, and how much data has already been processed. If a more descriptive format is desired, we can issue the following command:

$ xqry -y
---
apiVersion: xqry/v1
streams:
  - name: str1
    delta: 1
    size: 214
    count: 107
  - name: A
    delta: 1
    count: 86
    location: data1.txt
  - name: B
    delta: 2
    count: 43
    location: data2.txt

The response is given in YAML form.

To add another query to the system, we need to issue the command:

$ xqry -a "SELECT * STREAM str2 FROM A#B"
snd: adhoc SELECT * STREAM str2 FROM A#B
rcv: db OK

A command in this form sends a new query to the xretractor process. Upon receiving it, the system compiles it and merges it into the query plan tree.

If we check the system’s state again, we’ll see the following picture:

$ xqry -d
|str2|2/3| 10| 10|         |0|
|   A|  1| -1| 23|data1.txt|1|
|str1|  1|312|156|         |0|
|   B|  2| -1| 12|data2.txt|1|
ok.

Or like this:

$ xqry -y
---
apiVersion: xqry/v1
streams:
  - name: str2
    delta: 2/3
    size: 7
    count: 7
  - name: A
    delta: 1
    count: 16
    location: data1.txt
  - name: str1
    delta: 1
    size: 298
    count: 149
  - name: B
    delta: 2
    count: 8
    location: data2.txt

Taking a closer look at the queries via the xqry command, we’ll see the following system response for the str1 query:

$ xqry -t str1
---
apiVersion: xqry/v1
stream:
  name: str1
  delta: 1
query: SELECT * STREAM str1 FROM A+B
fields:
  str1.A_0:
    type: BYTE
  str1.B_1:
    type: BYTE

and for the str2 query:

$ xqry -t str2
---
apiVersion: xqry/v1
stream:
  name: str2
  delta: 2/3
query: SELECT * STREAM str2 FROM A#B
fields:
  str2.a:
    type: BYTE

As you can see, the additional query str2 was correctly merged into the existing query execution plan. You can also see that far less data has accumulated compared to str1.

NOTE: The functionality described here is covered by the test: issue6_adhoc, described in the appendix Integration Tests.