feat: stream search API

api/seqproxyapi/v1/seq_proxy_api.proto

@@ -120,6 +120,13 @@ service SeqProxyApi {
       body: "*"
     };
   }
+
+  rpc StreamSearch(stream StreamSearchRequest) returns (stream StreamSearchResponse) {
+    option (google.api.http) = {
+      post: "/stream-search"
+      body: "*"
+    };
+  }
 }
 
 // Custom error code, returned by seq-db proxy.
@@ -436,3 +443,78 @@ message ExportAsyncSearchRequest {
 message ExportResponse {
   Document doc = 1; // Response document.
 }
+
+message StreamSearchRequest {
+  oneof RequestType {
+    StreamSearchQuery query = 1;
+    StreamControl control = 2;
+  }
+}
+
+message StreamSearchQuery {
+  string query = 1;                   // Search query.
+  google.protobuf.Timestamp from = 2; // Lower bound for search (inclusive).
+  google.protobuf.Timestamp to = 3;   // Upper bound for search (inclusive).
+  bool explain = 4;                   // Should request be explained (tracing will be provided with the result).
+  string offset_id = 5;               // ID offset for pagination.
+  bool with_total = 6;                // Should total number of documents be returned in response.
+}
+
+message StreamControl {
+  ControlAction action = 1;
+}
+
+enum ControlAction {
+  CONTROL_ACTION_UNSPECIFIED = 0;
+  FINALIZE = 1;   // Indicates correct stream termination, will get Summary after
+  CANCEL = 2;     // Some client error, termination stream immediately, no need for Summary
+}
+
+message StreamSearchResponse {
+  oneof ResponseType {
+    ResponseHeader header = 1;
+    ResponseData data = 2;
+    ResponseSummary summary = 3;
+  }
+}
+
+message ResponseHeader {
+  repeated Typing typing = 1;
+}
+
+message Typing {
+  string title = 1;
+  DataType type = 2;
+}
+
+enum DataType {
+  BYTES = 0;
+  SEQ_ID = 1;
+  RAW_DOCUMENT = 2;
+  STRING = 3;
+  UINT32 = 4;
+  UINT64 = 5;
+  INT32 = 6;
+  INT64 = 7;
+  FLOAT64 = 8;
+  // TODO: later we will need array data types, such as:
+  // StringArray, Uin64Array, Float64Array etc.
+}
+
+message ResponseData {
+  RecordsBatch batch = 1;
+}
+
+message RecordsBatch {
+  repeated Record records = 1;
+}
+
+message Record {
+  repeated bytes raw_data = 1;
+}
+
+message ResponseSummary {
+  uint64 total = 1;
+  Error error = 2;
+  optional ExplainEntry explain = 3;
+}

asyncsearcher/async_searcher.go

@@ -228,6 +228,9 @@ func (as *AsyncSearcher) StartSearch(r AsyncSearchRequest, fracProvider fraction
 	if err != nil {
 		return err
 	}
+	if err := ast.ValidatePipes(); err != nil {
+		return err
+	}
 	r.Params.AST = ast.Root
 
 	now := timeNow()
@@ -346,6 +349,9 @@ func (as *AsyncSearcher) doSearch(id string, fracProvider fractionAcquirer) {
 		if err != nil {
 			panic(fmt.Errorf("BUG: search query must be valid: %s", err))
 		}
+		if err := ast.ValidatePipes(); err != nil {
+			panic(fmt.Errorf("BUG: search query must be valid: %s", err))
+		}
 		info.Request.Params.AST = ast.Root
 	}
 

parser/seqql.go

@@ -25,6 +25,27 @@ func (q *SeqQLQuery) SeqQLString() string {
 	return b.String()
 }
 
+// streamOnlyPipes are the pipes valid only for the StreamSearch method. Non-stream
+// callers reject them via SeqQLQuery.ValidatePipes.
+var streamOnlyPipes = map[string]struct{}{
+	"stats":  {},
+	"sort":   {},
+	"limit":  {},
+	"offset": {},
+}
+
+// ValidatePipes returns an error if the query contains any stream-only pipe (stats,
+// sort, limit, offset). It is used by methods that do not support stream-only pipes
+// to reject them after parsing. ParseSeqQL itself does not perform this check.
+func (q *SeqQLQuery) ValidatePipes() error {
+	for _, p := range q.Pipes {
+		if _, ok := streamOnlyPipes[p.Name()]; ok {
+			return fmt.Errorf("pipe '%s' is not allowed", p.Name())
+		}
+	}
+	return nil
+}
+
 func parse(q string, mapping seq.Mapping) (SeqQLQuery, error) {
 	lex := newLexer(q)
 

parser/seqql_filter_test.go

@@ -403,7 +403,10 @@ func TestParseSeqQLError(t *testing.T) {
 	// Test pipes.
 	test(`message:--||`, `unknown pipe: |`)
 	test(`source_type:access* | fields message | fields except login:admin`, `parsing 'fields' pipe: unexpected symbol ":"`)
-	test(`source_type:access* | fields message | fields login`, `multiple field filters is not allowed`)
+	test(`source_type:access* | stats count by (service) | stats count by (login)`, `multiple 'stats' pipes are not allowed`)
+	test(`source_type:access* | sort asc | sort desc`, `multiple 'sort' pipes are not allowed`)
+	test(`source_type:access* | limit 10 | limit 20`, `multiple 'limit' pipes are not allowed`)
+	test(`source_type:access* | offset 10 | offset 20`, `multiple 'offset' pipes are not allowed`)
 	test(`* | fields event, `, `parsing 'fields' pipe: trailing comma not allowed`)
 }