feat(query-engine): prototype streaming query engine

api/seqproxyapi/v1/seq_proxy_api.proto

@@ -111,6 +111,13 @@ service SeqProxyApi {
       body: "*"
     };
   }
+
+  rpc OnePhaseSearch(SearchRequest) returns (ComplexSearchResponse) {
+    option (google.api.http) = {
+      post: "/one-phase-search"
+      body: "*"
+    };
+  }
 }
 
 // Custom error code, returned by seq-db proxy.

api/storeapi/store_api.proto

@@ -26,6 +26,8 @@ service StoreApi {
   rpc Fetch(FetchRequest) returns (stream BinaryData) {}
 
   rpc Status(StatusRequest) returns (StatusResponse) {}
+
+  rpc OnePhaseSearch(OnePhaseSearchRequest) returns (stream OnePhaseSearchResponse) {}
 }
 
 message BulkRequest {
@@ -145,6 +147,7 @@ enum SearchErrorCode {
   TOO_MANY_FRACTION_TOKENS = 6;
   TOO_MANY_FIELD_VALUES = 7;
   MEMORY_LIMIT_EXCEEDED = 8;
+  DISABLED = 9;
 }
 
 message StartAsyncSearchRequest {
@@ -244,12 +247,13 @@ message IdWithHint {
   string hint = 2;
 }
 
+message FieldsFilter {
+  repeated string fields = 1;
+  // see seqproxyapi.FetchRequest.FieldsFilter.allow_list for details.
+  bool allow_list = 2;
+}
+
 message FetchRequest {
-  message FieldsFilter {
-    repeated string fields = 1;
-    // see seqproxyapi.FetchRequest.FieldsFilter.allow_list for details.
-    bool allow_list = 2;
-  }
   repeated string ids = 1;
   bool explain = 3;
   repeated IdWithHint ids_with_hints = 4;
@@ -264,3 +268,60 @@ message StatusRequest {}
 message StatusResponse {
   google.protobuf.Timestamp oldest_time = 1;
 }
+
+message OnePhaseSearchRequest {
+  string query = 1;
+  google.protobuf.Timestamp from = 2;
+  google.protobuf.Timestamp to = 3;
+  int64 size = 4;
+  int64 offset = 5;
+  bool explain = 6;
+  bool with_total = 7;
+  Order order = 8;
+  string offset_id = 9;
+  FieldsFilter fields_filter = 10;
+}
+
+message OnePhaseSearchResponse {
+  oneof ResponseType {
+    Header header = 1;
+    RecordsBatch batch = 2;
+  }
+}
+
+message Header {
+  Metadata metadata = 1;
+  repeated Typing typing = 2;
+}
+
+message Metadata {
+  uint64 total = 1;
+  SearchErrorCode code = 2;
+  repeated string errors = 3;
+  optional ExplainEntry explain = 4;
+}
+
+enum DataType {
+  BYTES = 0;
+  RAW_DOCUMENT = 1;
+  STRING = 2;
+  UINT32 = 3;
+  UINT64 = 4;
+  INT32 = 5;
+  INT64 = 6;
+  FLOAT64 = 7;
+  // TODO: array data types: StringArray, Uin64Array, Float64Array etc.
+}
+
+message Typing {
+  string title = 1;
+  DataType type = 2;
+}
+
+message RecordsBatch {
+  repeated Record records = 1;
+}
+
+message Record {
+  repeated bytes raw_data = 1;
+}

cmd/seq-db/seq-db.go

@@ -187,6 +187,7 @@ func startProxy(
 			EsVersion:                         cfg.API.ESVersion,
 			GatewayAddr:                       cfg.Address.GRPC,
 			AsyncSearchMaxDocumentsPerRequest: cfg.AsyncSearch.MaxDocumentsPerRequest,
+			EnableOnePhaseSearch:              cfg.Experimental.EnableOnePhaseSearch,
 		},
 		Search: search.Config{
 			HotStores:       hotStores,
@@ -318,6 +319,9 @@ func startStore(
 				To:    cfg.Filtering.To,
 				From:  cfg.Filtering.From,
 			},
+			OnePhaseSearch: storeapi.OnePhaseSearchConfig{
+				Enabled: cfg.Experimental.EnableOnePhaseSearch,
+			},
 		},
 		SkipMaskManagerConfig: skipmaskmanager.Config{
 			DataDir:        cfg.SkipMaskManager.DataDir,

config/config.go

@@ -300,6 +300,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"`
+		// EnableOnePhaseSearch enanles experimental one phase search method that supports new query engine.
+		EnableOnePhaseSearch bool `config:"enable_one_phase_search"`
 	} `config:"experimental"`
 }
 

consts/consts.go

@@ -111,4 +111,5 @@ var (
 	ErrTooManyFractionTokens     = errors.New("aggregation has too many fraction tokens")
 	ErrTooManyFractionsHit       = errors.New("too many fractions hit")
 	ErrMemoryLimitExceeded       = errors.New("memory limit exceeded")
+	ErrResourceDisabled          = errors.New("resource is disabled")
 )