feat(StreamSearch): add stream search on stores

api/storeapi/store_api.proto

@@ -26,6 +26,8 @@ service StoreApi {
   rpc Fetch(FetchRequest) returns (stream BinaryData) {}
 
   rpc Status(StatusRequest) returns (StatusResponse) {}
+
+  rpc StreamSearch(stream StreamSearchRequest) returns (stream StreamSearchResponse) {}
 }
 
 message BulkRequest {
@@ -265,3 +267,83 @@ message StatusRequest {}
 message StatusResponse {
   google.protobuf.Timestamp oldest_time = 1;
 }
+
+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;
+}
+
+message Error {
+  SearchErrorCode code = 1;
+  string message = 2;
+}

parser/seqql.go

@@ -29,12 +29,13 @@ func (q *SeqQLQuery) SeqQLString() string {
 // callers reject them via SeqQLQuery.ValidatePipes.
 var streamOnlyPipes = map[string]struct{}{
 	"stats":  {},
+	"filter": {},
 	"sort":   {},
 	"limit":  {},
 	"offset": {},
 }
 
-// ValidateStreamPipes returns an error if the query contains any stream-only pipe (stats,
+// ValidateStreamPipes returns an error if the query contains any stream-only pipe (stats, filter,
 // 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) ValidateStreamPipes() error {

parser/seqql_pipes.go

@@ -12,27 +12,29 @@ type Pipe interface {
 }
 
 // pipeOrder defines the allowed order of pipes in a SeqQL query:
-// stats | fields | sort | limit | offset. Any pipe may be omitted, but the
+// stats | filter | fields | sort | limit | offset. Any pipe may be omitted, but the
 // present ones must appear in this exact sequence. The value is the position of
 // the pipe in that sequence.
 var pipeOrder = map[string]int{
 	"stats":  0,
-	"fields": 1,
-	"sort":   2,
-	"limit":  3,
-	"offset": 4,
+	"filter": 1,
+	"fields": 2,
+	"sort":   3,
+	"limit":  4,
+	"offset": 5,
 }
 
 var pipeNameFromOrder = map[int]string{
 	0: "stats",
-	1: "fields",
-	2: "sort",
-	3: "limit",
-	4: "offset",
+	1: "filter",
+	2: "fields",
+	3: "sort",
+	4: "limit",
+	5: "offset",
 }
 
 // parsePipes parses the pipe stage of a SeqQL query. The pipes must appear in
-// the fixed order stats | fields | sort | limit | offset; pipes may be omitted,
+// the fixed order stats | filter | fields | sort | limit | offset; pipes may be omitted,
 // but none may appear out of order.
 func parsePipes(lex *lexer) ([]Pipe, error) {
 	seen := make(map[string]struct{})
@@ -55,6 +57,13 @@ func parsePipes(lex *lexer) ([]Pipe, error) {
 				return nil, fmt.Errorf("parsing 'fields' pipe: %s", err)
 			}
 			pipes = append(pipes, p)
+		case lex.IsKeyword("filter"):
+			name = "filter"
+			p, err := parsePipeFilter(lex)
+			if err != nil {
+				return nil, fmt.Errorf("parsing 'filter' pipe: %s", err)
+			}
+			pipes = append(pipes, p)
 		case lex.IsKeyword("stats"):
 			name = "stats"
 			p, err := parsePipeStats(lex)
@@ -101,6 +110,62 @@ func parsePipes(lex *lexer) ([]Pipe, error) {
 	return pipes, nil
 }
 
+type FilterCondition struct {
+	Field string
+	Value string
+}
+
+type PipeFilter struct {
+	Condition FilterCondition
+}
+
+func (f *PipeFilter) Name() string {
+	return "filter"
+}
+
+func (f *PipeFilter) DumpSeqQL(o *strings.Builder) {
+	o.WriteString("filter ")
+	o.WriteString(quoteTokenIfNeeded(f.Condition.Field))
+	o.WriteString(":")
+	o.WriteString(quoteTokenIfNeeded(f.Condition.Value))
+}
+
+func parsePipeFilter(lex *lexer) (*PipeFilter, error) {
+	if !lex.IsKeyword("filter") {
+		return nil, fmt.Errorf("missing 'filter' keyword")
+	}
+	lex.Next()
+
+	field, err := parseCompositeTokenReplaceWildcards(lex)
+	if err != nil {
+		return nil, fmt.Errorf("parsing field name: %s", err)
+	}
+	if field == "" {
+		return nil, fmt.Errorf("empty field name")
+	}
+
+	if !lex.IsKeyword(":") {
+		return nil, fmt.Errorf("missing ':' after %q", field)
+	}
+	lex.Next()
+
+	if lex.IsKeyword("") {
+		return nil, fmt.Errorf("missing filter value for field %q", field)
+	}
+
+	value, err := parseCompositeTokenReplaceWildcards(lex)
+	if err != nil {
+		return nil, fmt.Errorf("parsing filter value: %s", err)
+	}
+
+	return &PipeFilter{
+		Condition: FilterCondition{
+			Field: field,
+			Value: value,
+		},
+	}, nil
+}
+
 type PipeFields struct {
 	Fields []string
 	Except bool
@@ -457,7 +522,7 @@ var reservedKeywords = uniqueTokens([]string{
 	"|",
 
 	// Pipe specific keywords.
-	"fields", "except", "limit", "offset", "sort", "stats", "by", "interval", "unique_count", "asc", "desc",
+	"fields", "except", "filter", "limit", "offset", "sort", "stats", "by", "interval", "unique_count", "asc", "desc",
 })
 
 func needQuoteToken(s string) bool {

parser/seqql_pipes_test.go

@@ -39,6 +39,31 @@ func TestParsePipeFieldsExcept(t *testing.T) {
 	test(`* | fields except k8s_namespace`, `* | fields except k8s_namespace`)
 }
 
+func TestParsePipeFilter(t *testing.T) {
+	test := func(q, expected string) {
+		t.Helper()
+		query, err := ParseSeqQL(q, nil)
+		require.NoError(t, err)
+		require.Equal(t, expected, query.SeqQLString())
+	}
+
+	test(`service:my_service | filter field:"some value"`, `service:my_service | filter field:"some value"`)
+	test(`service:my_service | filter field:value`, `service:my_service | filter field:value`)
+}
+
+func TestParsePipeFilterErrors(t *testing.T) {
+	test := func(q string) {
+		t.Helper()
+		_, err := ParseSeqQL(q, nil)
+		require.Error(t, err)
+	}
+
+	test(`service:my_service | filter`)
+	test(`service:my_service | filter field`)
+	test(`service:my_service | filter :value`)
+	test(`service:my_service | filter a:1 | filter b:2`)
+}
+
 func TestParsePipeStats(t *testing.T) {
 	test := func(q, expected string) {
 		t.Helper()
@@ -171,6 +196,7 @@ func TestParseSeqQLRejectsStreamPipes(t *testing.T) {
 	test(`service:my_service | sort asc`)
 	test(`service:my_service | limit 10`)
 	test(`service:my_service | offset 10`)
+	test(`service:my_service | filter field:value`)
 }
 
 func TestValidatePipesAllowsFields(t *testing.T) {
@@ -189,13 +215,14 @@ func TestParsePipeOrder(t *testing.T) {
 	}
 
 	test(
-		"service:my_service | stats count by (service) | fields message | sort asc | limit 10 | offset 5",
-		"service:my_service | stats count by (service) | fields message | sort asc | limit 10 | offset 5",
+		"service:my_service | stats count by (service) | filter field:value | fields message | sort asc | limit 10 | offset 5",
+		"service:my_service | stats count by (service) | filter field:value | fields message | sort asc | limit 10 | offset 5",
 	)
 	test("service:my_service | stats count by (service) | limit 10", "service:my_service | stats count by (service) | limit 10")
 	test("service:my_service | fields message | offset 5", "service:my_service | fields message | offset 5")
 	test("service:my_service | sort asc | limit 10", "service:my_service | sort asc | limit 10")
 	test("service:my_service | fields message | sort asc", "service:my_service | fields message | sort asc")
+	test("service:my_service | filter field:value | limit 10", "service:my_service | filter field:value | limit 10")
 }
 
 func TestParsePipeOrderErrors(t *testing.T) {
@@ -211,4 +238,5 @@ func TestParsePipeOrderErrors(t *testing.T) {
 	test(`service:my_service | offset 5 | limit 10`)
 	test(`service:my_service | offset 5 | sort asc | limit 10`)
 	test(`service:my_service | offset 5 | limit 10 | sort asc | fields message | stats count by (service)`)
+	test(`service:my_service | fields message | filter field:value`)
 }

pkg/seqproxyapi/v1/marshaler.go

@@ -2,16 +2,17 @@ package seqproxyapi
 
 import (
 	"bytes"
-	"encoding/binary"
 	"encoding/json"
 	"math"
 	"strconv"
 	"time"
 
-	"github.com/ozontech/seq-db/seq"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/ozontech/seq-db/query/encoding"
+	"github.com/ozontech/seq-db/seq"
 )
 
 var (
@@ -295,7 +296,7 @@ func (r *Record) MarshalJSON() ([]byte, error) {
 			// cells[1] is a little-endian uint64 storing the document MID (nanoseconds).
 			var ts time.Time
 			if len(cells[1]) == 8 {
-				ts = seq.MID(binary.LittleEndian.Uint64(cells[1])).Time()
+				ts = seq.MID(encoding.Uint64FromBytes(cells[1])).Time()
 			}
 			return json.Marshal([]any{
 				string(cells[0]), // id
@@ -306,7 +307,7 @@ func (r *Record) MarshalJSON() ([]byte, error) {
 		// cells[1] is a little-endian float64 storing the aggregation value.
 		var value float64
 		if len(cells[1]) == 8 {
-			value = math.Float64frombits(binary.LittleEndian.Uint64(cells[1]))
+			value = encoding.Float64FromBytes(cells[1])
 		}
 		val := json.RawMessage(strconv.FormatFloat(value, 'f', -1, 64))
 		if math.IsNaN(value) || math.IsInf(value, 0) {
@@ -316,7 +317,7 @@ func (r *Record) MarshalJSON() ([]byte, error) {
 		// A zero MID (no timestamp) is rendered as an empty string.
 		formattedTime := ""
 		if len(cells[2]) == 8 {
-			if ns := binary.LittleEndian.Uint64(cells[2]); ns != 0 {
+			if ns := encoding.Uint64FromBytes(cells[2]); ns != 0 {
 				formattedTime = time.Unix(0, int64(ns)).UTC().Format(time.RFC3339Nano)
 			}
 		}