feat(stream search): serve regular methods through stream search

api/seqproxyapi/v1/seq_proxy_api.proto

@@ -497,8 +497,8 @@ enum DataType {
   INT32 = 6;
   INT64 = 7;
   FLOAT64 = 8;
-  // TODO: later we will need array data types, such as:
-  // StringArray, Uin64Array, Float64Array etc.
+  FLOAT64_ARRAY = 9;
+  STRING_ARRAY = 10;
 }
 
 message ResponseData {

api/storeapi/store_api.proto

@@ -321,8 +321,8 @@ enum DataType {
   INT32 = 6;
   INT64 = 7;
   FLOAT64 = 8;
-  // TODO: later we will need array data types, such as:
-  // StringArray, Uin64Array, Float64Array etc.
+  FLOAT64_ARRAY = 9;
+  STRING_ARRAY = 10;
 }
 
 message ResponseData {

cmd/seq-db/seq-db.go

@@ -188,6 +188,7 @@ func startProxy(
 			EsVersion:                         cfg.API.ESVersion,
 			GatewayAddr:                       cfg.Address.GRPC,
 			AsyncSearchMaxDocumentsPerRequest: cfg.AsyncSearch.MaxDocumentsPerRequest,
+			TryStreamSearch:                   cfg.Experimental.TryStreamSearch,
 		},
 		Search: search.Config{
 			HotStores:       hotStores,

config/config.go

@@ -341,6 +341,8 @@ type Config struct {
 		// Specify how many tokens can be checked using regular expressions.
 		// If zero then there is no limit.
 		MaxRegexTokensCheck int `config:"max_regex_tokens_check" default:"0"`
+		// If true, suitable ComplexSearch queries will be served through stream search implementation.
+		TryStreamSearch bool `config:"try_stream_search"`
 	} `config:"experimental"`
 }
 

pkg/seqproxyapi/v1/marshaler.go

@@ -262,71 +262,75 @@ func (i *AsyncSearchesListItem) MarshalJSON() ([]byte, error) {
 	return pbMarshaller.Marshal(i)
 }
 
-// streamSearchColumns is the number of columns carried by a record produced by
-// StreamSearch. Both document and aggregation records use 3 cells, so the
-// layout is distinguished by the third cell's content rather than its count:
-//   - documents: [id, time(uint64 nanoseconds), data] — data is a JSON document;
-//   - aggregation buckets: [key, value(float64), ts(uint64 nanoseconds)] — ts is
-//     a little-endian uint64, never valid JSON.
-const streamSearchColumns = 3
+// streamSearchDocColumns is the number of columns carried by a document record
+// produced by StreamSearch: [id, time, data].
+const streamSearchDocColumns = 3
+
+// streamSearchAggColumns is the number of columns carried by an aggregation
+// record: [key, value, ts, quantiles]. quantiles holds the computed quantile
+// values (a little-endian float64 array); it is empty for non-quantile
+// aggregations. The record kind is told from the column count, so documents
+// (3 cells) and aggregations (4 cells) never collide.
+const streamSearchAggColumns = 4
 
 // MarshalJSON formats a StreamSearch record into a human-readable JSON array.
+// The record kind is distinguished by the column count:
 //
-// Document and aggregation records share the same 3-cell width, so the record
-// kind is inferred from the third cell: when it holds a valid JSON document the
-// record is treated as a document, otherwise as an aggregation bucket.
-//
-//   - documents: [id, time(nanoseconds, little-endian uint64), data] where data
-//     is inlined as a json.RawMessage and time is rendered as an RFC3339Nano
-//     string;
-//   - aggregation buckets: [key, value(little-endian float64),
-//     ts(nanoseconds, little-endian uint64)] where value is rendered as a number
-//     (NaN/Inf become quoted strings) and ts is rendered as an RFC3339Nano
-//     string (empty when the bucket has no timestamp).
+//   - documents (3 cells): [id, time(nanoseconds, little-endian uint64), data]
+//     where data is inlined as a json.RawMessage and time is rendered as an
+//     RFC3339Nano string;
+//   - aggregation buckets (4 cells): [key, value(little-endian float64),
+//     ts(nanoseconds, little-endian uint64), quantiles(little-endian float64
+//     array)] where value is rendered as a number (NaN/Inf become quoted
+//     strings), ts is rendered as an RFC3339Nano string (empty when the bucket
+//     has no timestamp) and quantiles is rendered as an array of numbers
+//     (empty when no quantiles were requested).
 //
 // For any other layout the raw cells are emitted as-is (base64 for bytes).
 func (r *Record) MarshalJSON() ([]byte, error) {
 	cells := r.GetRawData()
 	switch len(cells) {
-	case streamSearchColumns:
-		// Both layouts use 3 cells. A document's third cell is a JSON document;
-		// an aggregation bucket's third cell is an 8-byte uint64 timestamp,
-		// which is never valid JSON.
-		if json.Valid(cells[2]) {
-			// cells[1] is a little-endian uint64 storing the document MID (nanoseconds).
-			var ts time.Time
-			if len(cells[1]) == 8 {
-				ts = seq.MID(encoding.Uint64FromBytes(cells[1])).Time()
-			}
-			return json.Marshal([]any{
-				string(cells[0]), // id
-				ts.UTC().Format(time.RFC3339Nano),
-				json.RawMessage(cells[2]), // data, inlined as JSON
-			})
-		}
-		// cells[1] is a little-endian float64 storing the aggregation value.
-		var value float64
+	case streamSearchDocColumns:
+		// cells[1] is a little-endian uint64 storing the document MID (nanoseconds).
+		var ts time.Time
 		if len(cells[1]) == 8 {
-			value = encoding.Float64FromBytes(cells[1])
-		}
-		val := json.RawMessage(strconv.FormatFloat(value, 'f', -1, 64))
-		if math.IsNaN(value) || math.IsInf(value, 0) {
-			val = json.RawMessage(strconv.Quote(string(val)))
-		}
-		// cells[2] is a little-endian uint64 storing the bucket MID (nanoseconds).
-		// A zero MID (no timestamp) is rendered as an empty string.
-		formattedTime := ""
-		if len(cells[2]) == 8 {
-			if ns := encoding.Uint64FromBytes(cells[2]); ns != 0 {
-				formattedTime = time.Unix(0, int64(ns)).UTC().Format(time.RFC3339Nano)
-			}
+			ts = seq.MID(encoding.Uint64FromBytes(cells[1])).Time()
 		}
 		return json.Marshal([]any{
-			string(cells[0]), // key
-			val,              // value
-			formattedTime,    // ts
+			string(cells[0]), // id
+			ts.UTC().Format(time.RFC3339Nano),
+			json.RawMessage(cells[2]), // data, inlined as JSON
 		})
+	case streamSearchAggColumns:
+		return marshalAggRecord(cells)
 	default:
 		return json.Marshal(cells)
 	}
 }
+
+// marshalAggRecord renders an aggregation record: [key, value, ts, quantiles].
+func marshalAggRecord(cells [][]byte) ([]byte, error) {
+	// cells[1] is a little-endian float64 storing the aggregation value.
+	var value float64
+	if len(cells[1]) == 8 {
+		value = encoding.Float64FromBytes(cells[1])
+	}
+	val := json.RawMessage(strconv.FormatFloat(value, 'f', -1, 64))
+	if math.IsNaN(value) || math.IsInf(value, 0) {
+		val = json.RawMessage(strconv.Quote(string(val)))
+	}
+	// cells[2] is a little-endian uint64 storing the bucket MID (nanoseconds).
+	// A zero MID (no timestamp) is rendered as an empty string.
+	formattedTime := ""
+	if len(cells[2]) == 8 {
+		if ns := encoding.Uint64FromBytes(cells[2]); ns != 0 {
+			formattedTime = time.Unix(0, int64(ns)).UTC().Format(time.RFC3339Nano)
+		}
+	}
+	return json.Marshal([]any{
+		string(cells[0]),                         // key
+		val,                                      // value
+		formattedTime,                            // ts
+		encoding.Float64ArrayFromBytes(cells[3]), // quantiles
+	})
+}

pkg/seqproxyapi/v1/marshaler_test.go

@@ -172,23 +172,39 @@ func TestRecordMarshalJSON(t *testing.T) {
 			[]byte("service-a"),
 			encoding.Float64ToBytes(42.5),
 			encoding.Uint64ToBytes(uint64(tsNs)),
+			encoding.Float64ArrayToBytes(nil), // no quantiles
 		}}
 
 		raw, err := json.Marshal(rec)
 		r.NoError(err)
-		r.Equal(`["service-a",42.5,"2025-07-08T10:19:08.742Z"]`, string(raw))
+		r.Equal(`["service-a",42.5,"2025-07-08T10:19:08.742Z",[]]`, string(raw))
+	})
+
+	t.Run("aggregation bucket with quantiles", func(t *testing.T) {
+		tsNs := time.Date(2025, 7, 8, 10, 19, 8, 742000000, time.UTC).UnixNano()
+		rec := &Record{RawData: [][]byte{
+			[]byte("service-a"),
+			encoding.Float64ToBytes(42.5),
+			encoding.Uint64ToBytes(uint64(tsNs)),
+			encoding.Float64ArrayToBytes([]float64{5.5, 9.1}),
+		}}
+
+		raw, err := json.Marshal(rec)
+		r.NoError(err)
+		r.Equal(`["service-a",42.5,"2025-07-08T10:19:08.742Z",[5.5,9.1]]`, string(raw))
 	})
 
 	t.Run("aggregation bucket with NaN", func(t *testing.T) {
 		rec := &Record{RawData: [][]byte{
 			[]byte("service-a"),
 			encoding.Float64ToBytes(math.NaN()),
 			make([]byte, 8),
+			encoding.Float64ArrayToBytes(nil),
 		}}
 
 		raw, err := json.Marshal(rec)
 		r.NoError(err)
-		r.Equal(`["service-a","NaN",""]`, string(raw))
+		r.Equal(`["service-a","NaN","",[]]`, string(raw))
 	})
 
 	t.Run("unknown layout falls back to raw bytes", func(t *testing.T) {