Querier (bullet-core 1.2.3 API)

java.lang.Object
- com.yahoo.bullet.querying.Querier

All Implemented Interfaces:

Closable, Monoidal
```
public class Querier
extends Object
implements Monoidal
```
This manages a RunningQuery that is currently being executed. It can consume(BulletRecord) records for the query, and combine(byte[]) serialized data from another instance of the running query. It can also merge itself with another instance of running query using Monoidal.merge(Monoidal). Use finish() to retrieve the final results and terminate the query.
Ideally to implement Bullet, you would parallelize into two stages:
1. The Filter Stage - this is where you partition and distribute your data across workers. Each worker should have a copy of the query so no matter where the data ends up, the query as a whole across all the workers, will see it.
2. The Join Stage - this is where you group the results for all the parallelized outputs in the filter stage for a particular query (using the query ID as an identifier to combine the intermediate outputs).
Filter Stage
1. For each Query message from the PubSub, check to see if it is has a KILL or COMPLETE signal. If yes, remove any existing Querier objects for that query (identified by the ID) If no, create an instance of Querier for the query in Querier.Mode.PARTITION mode if and only if you are going to be persisting the querier for the duration of the query. If you are throwing away the querier, such as after processing your partitioned data in mini-batches and recreating it every new mini-batch, then you need not change the mode. If any exceptions or errors initializing, throw away the querier since the errors are handled in the Join stage below.
2. For every BulletRecord, call consume(BulletRecord) on all the Querier objects unless isDone()
3. If isDone(), call getData() and also remove the Querier.
4. If isClosed(), use getData() to emit the intermediate data to the Join stage for the query ID. Then, call reset().
5. Optional: if you are processing record by record (instead of micro-batches) and honoring isClosed(), you should check if isExceedingRateLimit() is true after calling getData(). If yes, you should cancel the query and emit a RateLimitError to the Join stage to kill the query. You can use getRateLimitError() to get the RateLimitError to pass to the Join stage.
6. Optional: If your data volume is very, very small (Heuristic: less than 1 per your 0.1 * bullet.query.window.min.emit.every.ms). across your partitions), you should run the isDone() and isClosed() and do the emits either on a timer or at fixed intervals so that your queries are checked for results and maintain their windowing guarantees.
You can also use hasNewData() to check if there is any new data to emit if you need to know a successful consumption or combining happened. If you do not want to call getData(), you can serialize Querier using non-native serialization frameworks and use Monoidal.merge(Monoidal) in the Join stage to merge them into an empty Querier for the query. This will be equivalent to calling combine(byte[]) on getData().
Pseudo Code

Case 1: Query
```
 (String id, String queryBody, Metadata metadata) = Query
 if (metadata.hasSignal(Signal.KILL) || metadata.hasSignal(Signal.COMPLETE))
     remove Querier for id
 else
     create new Querier(id, queryBody, config) and initialize it;
 
```
Case 2: BulletRecord record
```
 for Querier q : queriers:
     if (q.isDone())
         emit(q.getData())
         remove q
     else
         if (q.isClosed())
             emit(q.getData())
             q.reset()
             q.consume(record)
         if (q.isExceedingRateLimit())
             emit(q.getRateLimitError())
             remove q
 
```
Case 3: Periodically (for very small data volumes)
```
 for Querier q : queriers:
     if (q.isDone())
         emit(q.getData())
         remove q
     if (q.isClosed())
         emit(q.getData())
         q.reset()
     if (q.isExceedingRateLimit())
         emit(q.getRateLimitError())
         remove q
 
```
Join Stage
1. For each Query message from the PubSub, if it is a KILL signal similar to the Filter stage, kill the query and return. Otherwise create an instance of Querier for the query in Querier.Mode.ALL mode. If any exceptions or errors initializing it, make BulletError objects from them and return them as a Clip back through the PubSub.
2. For each KILL message from the Filter stage, call finish(), and add to the Meta a RateLimitError. Emit this through the regular PubSub Publisher for results. Also emit a KILL signal PubSubMessage to a Publisher for queries so that it is fed back to the Filter stage.
3. For each (id, byte[]) data from the Filter stage, call combine(byte[]) for the querier for that id.
4. If isDone(), call getResult() to emit the final result and remove the querier. Emit a COMPLETE signal to the PubSub Publisher for queries to feed back the complete status to the Filter stage.
5. If isClosed(), use getResult() ()} to emit the intermediate result and call reset()
6. Optional: as with the Filter stage, if the querier isExceedingRateLimit(), you can use getRateLimitError() and then RateLimitError.makeMeta() to create and emit a Clip.add(Meta). Make sure to remove the querier and send a KILL signal to a PubSub Publisher for queries to feed back the kill status to the Filter stage.
7. Optional: Similar to the filter stage you should run isDone() and isClosed() periodically if your data volume is too low.
8. Optional Delayed start and End (recommended if you are processing event by event): Since data from the Filter stage partitions may not arrive at the same time and since the querier may be isClosed() at the same time the data from the Filter stage partitions arrive, you should not immediately emit getResult() if isClosed() and then reset(). There are two ways to handle this. You could delay the start of the query by a bit in the Join stage so that windows from the Filter stage always arrive a bit earlier. Or you could buffer the results in the Join stage for a bit for each window as results trickle in. The issue with the latter approach is that you will slowly add the buffer time to the duration of your windows in your Join stage and eventually you will get two windows in one. The former approach does not have this problem. However, that approach could lead to results that are sent immediately (for record based windows) being dropped while the delay is happening. To solve these issues, you should buffer the final results for all queries for whom shouldBuffer() is true. This should be true for time-based windows and false for all record-based windows or queries with no window. So you can use the negation of shouldBuffer() to find out if the latter queries can be delayed. This delay will ensure that results from the filter phase are collected in their entirety before emitting from the Join phase. To aid you in doing this, you can buffer it and use restart() to mark the delayed start of the query.
Pseudo Code

Case 1: Query
```
 (String id, String queryBody, Metadata metadata) = Query
 if (metadata.hasSignal(Signal.KILL))
     remove Querier for id
     return
 try {
     create new Querier(id, queryBody, config)
     initialize it (see note above regarding delaying start) and if errors present:
         emit(Clip.of(Meta.of(errors.get()));
 catch (Exception e) {
     Clip clip = Clip.of(Meta.of(asList(BulletError.makeError(e, queryBody)))
     emit(clip)
 
```
Case 2: KILL messages from Filter
```
 (String id, RateLimitError error) = KILL message

 querier = Querier for id
 Clip clip = querier.finish()
 clip.add(Meta.of(error))
 emit(clip)
 queryPubSubPublisher.emit((id, KILL signal))
 remove querier
 
```
Case 3: Data message from Filter
```
 (String id, byte[] data) = Data

 querier = Querier for id
 querier.combine(data)
 if (querier.isDone())
     Clip clip = querier.finish()
     emit(clip)
     queryPubSubPublisher.emit(
 else if (querier.isClosed())
     Clip clip = querier.getResult()
     // See note above regarding buffering if querier.shouldBuffer()
     emit(clip)
     querier.reset()

 if (querier.isExceedingRateLimit())
     Clip clip = merge q.finish() with q.getRateLimitError()
     emit(clip)
     queryPubSubPublisher.emit((id, KILL signal))
     remove q
 
```
Case 4: Periodically (for very small data volumes)
```
 for Querier q : queriers:
     if (q.isDone())
         emit(q.finish())
         remove q
     if (q.isClosed())
         emit(q.getResult())
         q.reset()
     if (q.isExceedingRateLimit())
         Clip clip = merge q.finish() with q.getRateLimitError()
         emit(clip)
         queryPubSubPublisher.emit((id, KILL signal))
         remove q
 
```

Nested Class Summary

Nested Classes
Modifier and Type Class and Description

static class Querier.Mode
This is used to determine if this operates in partitioned mode or not.

Nested Classes
Modifier and Type	Class and Description
`static class`	`Querier.Mode` This is used to determine if this operates in partitioned mode or not.

Field Summary

Fields
Modifier and Type Field and Description

static String TRY_AGAIN_LATER

Fields
Modifier and Type	Field and Description
`static String`	`TRY_AGAIN_LATER`

Constructor Summary

Constructors
Constructor and Description
`Querier(Querier.Mode mode, RunningQuery query, BulletConfig config)` Constructor that takes a `Querier.Mode`, `RunningQuery` instance and a configuration to use.
`Querier(RunningQuery query, BulletConfig config)` Constructor that takes a `RunningQuery` instance and a configuration to use.

Method Summary

All Methods Instance Methods Concrete Methods
Modifier and Type	Method and Description
`void`	`combine(byte[] data)` Presents the query with a serialized data representation of a prior result for the query.
`void`	`consume(com.yahoo.bullet.record.BulletRecord record)` Consume a `BulletRecord` for this query.
`Clip`	`finish()` Terminate the query and return the final result.
`byte[]`	`getData()` Get the result emitted so far after the last window.
`Meta`	`getMetadata()` Returns the `Meta` of the result so far.
`Query`	`getQuery()` Gets the `Query` instance for this Query.
`RateLimitError`	`getRateLimitError()` Returns a `RateLimitError` if the rate limit had exceeded the rate from a prior call to `isExceedingRateLimit()`.
`List<com.yahoo.bullet.record.BulletRecord>`	`getRecords()` Returns the `List` of `BulletRecord` result so far.
`Clip`	`getResult()` Gets the resulting `Clip` of the results so far.
`boolean`	`hasNewData()` Returns whether there is any new data to emit at all since the last `reset()`.
`boolean`	`isClosed()` Depending on the `Querier.Mode.ALL` mode this is operating in, returns true if and only if the query window is closed and you should emit the result at this time.
`boolean`	`isDone()` Returns true if the query has expired and will never accept any more data.
`boolean`	`isExceedingRateLimit()` Returns whether this is exceeding the rate limit.
`void`	`reset()` Resets this object.
`void`	`restart()` Forces a restart of a valid query to mark the correct start of this object if it was previously created but delayed in starting it (by using the negation of `shouldBuffer()`.
`boolean`	`shouldBuffer()` Returns if this query should buffer before emitting the final results.
`String`	`toString()`

Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Methods inherited from interface com.yahoo.bullet.common.Monoidal
merge

- Field Detail
  - TRY_AGAIN_LATER
```
public static final String TRY_AGAIN_LATER
```
    See Also:
    
    Constant Field Values
- Constructor Detail
  - Querier
```
public Querier(RunningQuery query,
               BulletConfig config)
```
    Constructor that takes a RunningQuery instance and a configuration to use. This also starts executing the query.
    
    Parameters:
    
    query - The running query.
    
    config - The validated BulletConfig configuration to use.
  - Querier
```
public Querier(Querier.Mode mode,
               RunningQuery query,
               BulletConfig config)
```
    Constructor that takes a Querier.Mode, RunningQuery instance and a configuration to use. This also starts executing the query.
    
    Parameters:
    
    mode - The mode for this querier.
    
    query - The running query.
    
    config - The validated BulletConfig configuration to use.
- Method Detail
  - restart
```
public void restart()
```
    Forces a restart of a valid query to mark the correct start of this object if it was previously created but delayed in starting it (by using the negation of shouldBuffer(). You might be using this if you were delaying the start of the query in the Join phase. This does not revalidate the query or reset any data this might have already consumed.
  - consume
```
public void consume(com.yahoo.bullet.record.BulletRecord record)
```
    Consume a BulletRecord for this query. The record may or not be actually incorporated into the query results. This depends on whether the query can accept more data, if it is expired or not or if the record matches any query filtering criteria.
    
    Specified by:
    
    consume in interface Monoidal
    
    Parameters:
    
    record - The BulletRecord to consume.
  - combine
```
public void combine(byte[] data)
```
    Presents the query with a serialized data representation of a prior result for the query. These will be included into the query results even if the query is isClosed() or isDone().
    
    Specified by:
    
    combine in interface Monoidal
    
    Parameters:
    
    data - The serialized data that represents a partial query result.
  - getData
```
public byte[] getData()
```
    Get the result emitted so far after the last window. Post aggregations are NOT applied.
    
    Specified by:
    
    getData in interface Monoidal
    
    Returns:
    
    The byte[] representation of the serialized result.
  - getRecords
```
public List<com.yahoo.bullet.record.BulletRecord> getRecords()
```
    Returns the List of BulletRecord result so far. See getResult() for the full result with metadata.
    
    Specified by:
    
    getRecords in interface Monoidal
    
    Returns:
    
    The records that are part of the result.
  - getMetadata
```
public Meta getMetadata()
```
    Returns the Meta of the result so far. See getResult() for the full result with the data.
    
    Specified by:
    
    getMetadata in interface Monoidal
    
    Returns:
    
    The metadata part of the result.
  - getResult
```
public Clip getResult()
```
    Gets the resulting Clip of the results so far.
    
    Specified by:
    
    getResult in interface Monoidal
    
    Returns:
    
    A non-null Clip representing the aggregated result.
  - isClosed
```
public boolean isClosed()
```
    Depending on the Querier.Mode.ALL mode this is operating in, returns true if and only if the query window is closed and you should emit the result at this time.
    
    Specified by:
    
    isClosed in interface Closable
    
    Returns:
    
    boolean denoting if query has closed.
  - reset
```
public void reset()
```
    Resets this object. You should call this if you have called getResult() or getData() after verifying whether this is isClosed().
    
    Specified by:
    
    reset in interface Monoidal
  - getQuery
```
public Query getQuery()
```
    Gets the Query instance for this Query.
    
    Returns:
    
    The Query instance for this object.
  - isDone
```
public boolean isDone()
```
    Returns true if the query has expired and will never accept any more data.
    
    Returns:
    
    A boolean denoting whether the query has expired.
  - hasNewData
```
public boolean hasNewData()
```
    Returns whether there is any new data to emit at all since the last reset(). Use this method if you are driving how data is consumed by this instance (for instance, microbatches) and need to emit data outside the windowing standards.
    
    Returns:
    
    A boolean denoting whether we have any new data that can be emitted.
  - isExceedingRateLimit
```
public boolean isExceedingRateLimit()
```
    Returns whether this is exceeding the rate limit. It is up to the user to perform any action such as killing the query if this is
    
    Returns:
    
    A boolean denoting whether we have exceeded the rate limit.
  - getRateLimitError
```
public RateLimitError getRateLimitError()
```
    Returns a RateLimitError if the rate limit had exceeded the rate from a prior call to isExceedingRateLimit().
    
    Returns:
    
    A rate limit error or null if the rate limit was not exceeded.
  - shouldBuffer
```
public boolean shouldBuffer()
```
    Returns if this query should buffer before emitting the final results. You can use this to wait for the final results in your Join or Combine stage after a query is isDone().
    
    Returns:
    
    A boolean that is true if the query results should be buffered in the Join phase.
  - finish
```
public Clip finish()
```
    Terminate the query and return the final result.
    
    Returns:
    
    The final non-null Clip representing the final result.
  - toString
```
public String toString()
```
    Overrides:
    
    toString in class Object

Class Querier

Filter Stage

Pseudo Code

Case 1: Query

Case 2: BulletRecord record

Case 3: Periodically (for very small data volumes)

Join Stage

Pseudo Code

Case 1: Query

Case 2: KILL messages from Filter

Case 3: Data message from Filter

Case 4: Periodically (for very small data volumes)

Nested Class Summary

Field Summary

Constructor Summary

Method Summary

Methods inherited from class java.lang.Object

Methods inherited from interface com.yahoo.bullet.common.Monoidal

Field Detail

TRY_AGAIN_LATER

Constructor Detail

Querier

Querier

Method Detail

restart

consume

combine

getData

getRecords

getMetadata

getResult

isClosed

reset

getQuery

isDone

hasNewData

isExceedingRateLimit

getRateLimitError

shouldBuffer

finish

toString