Make reply capture opt-in (--capture-replies) so it doesn't inflate query latency

benchmark_runner/redisearch_test.go

@@ -145,13 +145,46 @@
 	if parsed.Totals.TotalOps != 100 {
 		t.Errorf("TotalOps = %d, want 100 (pipeline=3 must flush and count the trailing window of a 100-row input)", parsed.Totals.TotalOps)
 	}
-	// Byte accounting (#111/#112/#114): sent bytes land in TxBytes, and reply
-	// bytes are actually captured (RxBytes>0 — the 100 HSET integer replies).
+	// Byte accounting (#111/#112/#114): sent bytes land in TxBytes.
 	if parsed.Totals.TxBytes == 0 {
 		t.Errorf("TxBytes = 0, want > 0 (sent bytes must be counted)")
 	}
-	if parsed.Totals.RxBytes == 0 {
-		t.Errorf("RxBytes = 0, want > 0 (reply bytes must be captured)")
+	// Reply capture is off by default (#117), so RxBytes is 0 here; see
+	// TestFTSBCaptureRepliesControlsRxBytes for the --capture-replies path.
+	if parsed.Totals.RxBytes != 0 {
+		t.Errorf("RxBytes = %d, want 0 without --capture-replies", parsed.Totals.RxBytes)
+	}
+}
+
+// #117: --capture-replies controls whether reply bytes are decoded and counted.
+// Default off (RxBytes==0, no client-side unmarshal on the latency hot path);
+// on -> RxBytes>0 (the HSET integer replies are decoded).
+func TestFTSBCaptureRepliesControlsRxBytes(t *testing.T) {
+	startRedisContainer(t)
+
+	rx := func(args ...string) uint64 {
+		data := runFTSBReadJSON(t, "../testdata/results.capture.json",
+			append([]string{"--input", "../testdata/minimal.csv"}, args...)...)
+		var parsed struct {
+			Totals struct {
+				TotalOps int    `json:"TotalOps"`
+				RxBytes  uint64 `json:"RxBytes"`
+			} `json:"Totals"`
+		}
+		if err := json.Unmarshal(data, &parsed); err != nil {
+			t.Fatalf("parse: %v", err)
+		}
+		if parsed.Totals.TotalOps != 100 {
+			t.Fatalf("TotalOps = %d, want 100", parsed.Totals.TotalOps)
+		}
+		return parsed.Totals.RxBytes
+	}
+
+	if got := rx(); got != 0 {
+		t.Errorf("default RxBytes = %d, want 0 (capture off)", got)
+	}
+	if got := rx("--capture-replies"); got == 0 {
+		t.Errorf("RxBytes with --capture-replies = 0, want > 0")
 	}
 }
 

cmd/ftsb_redisearch/cmd_processor.go

@@ -286,13 +286,24 @@ type pendingCmd struct {
 }
 
 func sendFlatCmd(p *processor, client radix.Client, cmdType, cmdQueryId, cmd string, docfields []string, txBytesCount uint64, pending []pendingCmd) ([]pendingCmd, bool) {
-	reply := new(interface{})
+	// By default use a nil receiver: radix reads and DISCARDS the reply (no
+	// allocation, no reflection) so the measured latency isn't inflated by
+	// client-side unmarshalling -- which is significant for large FT.SEARCH /
+	// FT.AGGREGATE replies (issue #117). --capture-replies opts into decoding the
+	// reply so getRxLen can populate RxBytes. Command errors are surfaced by radix
+	// regardless of the receiver.
+	var reply *interface{}
+	var rcv interface{} // nil interface -> discard
+	if captureReplies {
+		reply = new(interface{})
+		rcv = reply
+	}
 	key := ""
 	if len(docfields) > 0 {
 		key = docfields[0]
 	}
 	pending = append(pending, pendingCmd{
-		action:     radix.Cmd(reply, cmd, docfields...),
+		action:     radix.Cmd(rcv, cmd, docfields...),
 		reply:      reply,
 		cmdType:    cmdType,
 		cmdQueryId: cmdQueryId,

cmd/ftsb_redisearch/main.go

@@ -21,6 +21,7 @@ var (
 	pipeline       int
 	clusterMode    bool
 	continueOnErr  bool
+	captureReplies bool
 	timeout        time.Duration
 	versionFlag    bool
 	logFile        string
@@ -34,6 +35,7 @@ func init() {
 	flag.StringVar(&password, "a", "", "Password for Redis Auth.")
 	flag.IntVar(&debug, "debug", 0, "Debug printing (choices: 0, 1, 2). (default 0)")
 	flag.BoolVar(&continueOnErr, "continue-on-error", true, "If set to true, it will continue the benchmark and print the error message to stderr.")
+	flag.BoolVar(&captureReplies, "capture-replies", false, "If true, decode each command's reply so RxBytes is populated. Off by default: capturing fully unmarshals every reply on the client hot path (allocation + reflection inside the measured latency window), which inflates FT.SEARCH/FT.AGGREGATE latency. Command errors are detected regardless of this setting.")
 	flag.BoolVar(&clusterMode, "cluster-mode", false, "If set to true, it will run the client in cluster mode.")
 	flag.IntVar(&pipeline, "pipeline", 1, "Pipeline <numreq> requests. Default 1 (no pipeline).")
 	flag.IntVar(&timeoutSeconds, "timeout", 60, "Redis connection timeout in seconds.")
@@ -64,6 +66,7 @@ func (b *benchmark) GetConfigurationParametersMap() map[string]interface{} {
 	configs["host"] = host
 	configs["clusterMode"] = clusterMode
 	configs["continueOnError"] = continueOnErr
+	configs["captureReplies"] = captureReplies
 	configs["debug"] = debug
 	configs["pipeline"] = pipeline
 	configs["logFile"] = logFile