feat(client): Add DoRaw and DoRawWriteTo for raw RESP protocol access (#3713)

* Add DoRaw and DoRawWriteTo for raw RESP protocol access

* fix augment comments

* Pipeline retry fix

* Fix Missing NoRetry check in cluster and ring retry loops

command.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"context"
 	"fmt"
+	"io"
 	"maps"
 	"net"
 	"regexp"
@@ -215,6 +216,11 @@ type Cmder interface {
 	SetErr(error)
 	Err() error
 
+	// NoRetry returns true if the command should not be retried on failure.
+	// Commands that write directly to an io.Writer should return true since
+	// partial writes cannot be undone on retry.
+	NoRetry() bool
+
 	// GetCmdType returns the command type for fast value extraction
 	GetCmdType() CmdType
 }
@@ -236,6 +242,18 @@ func cmdsFirstErr(cmds []Cmder) error {
 	return nil
 }
 
+// cmdsContainNoRetry returns true if any command in the slice has NoRetry() == true.
+// If a pipeline contains a non-retryable command (e.g., RawWriteToCmd), the entire
+// pipeline must not be retried to prevent data corruption from partial writes.
+func cmdsContainNoRetry(cmds []Cmder) bool {
+	for _, cmd := range cmds {
+		if cmd.NoRetry() {
+			return true
+		}
+	}
+	return false
+}
+
 func writeCmds(wr *proto.Writer, cmds []Cmder) error {
 	for _, cmd := range cmds {
 		if err := writeCmd(wr, cmd); err != nil {
@@ -398,6 +416,14 @@ func (cmd *baseCmd) readRawReply(rd *proto.Reader) (err error) {
 	return err
 }
 
+// NoRetry returns true if the command should not be retried on failure.
+// By default, commands can be retried. Commands that write directly to an
+// io.Writer (like RawWriteToCmd) should override this to return true since
+// partial writes cannot be undone on retry.
+func (cmd *baseCmd) NoRetry() bool {
+	return false
+}
+
 func (cmd *baseCmd) GetCmdType() CmdType {
 	return cmd.cmdType
 }
@@ -720,6 +746,122 @@ func (cmd *Cmd) Clone() Cmder {
 
 //------------------------------------------------------------------------------
 
+// RawCmd returns raw RESP protocol bytes without parsing.
+type RawCmd struct {
+	baseCmd
+	val []byte
+}
+
+var _ Cmder = (*RawCmd)(nil)
+
+func NewRawCmd(ctx context.Context, args ...interface{}) *RawCmd {
+	return &RawCmd{
+		baseCmd: baseCmd{
+			ctx:     ctx,
+			args:    args,
+			cmdType: CmdTypeGeneric,
+		},
+	}
+}
+
+func (cmd *RawCmd) SetVal(val []byte) {
+	cmd.val = val
+}
+
+func (cmd *RawCmd) Val() []byte {
+	return cmd.val
+}
+
+func (cmd *RawCmd) Result() ([]byte, error) {
+	return cmd.val, cmd.err
+}
+
+func (cmd *RawCmd) Bytes() ([]byte, error) {
+	return cmd.val, cmd.err
+}
+
+func (cmd *RawCmd) String() string {
+	return cmdString(cmd, cmd.val)
+}
+
+func (cmd *RawCmd) readReply(rd *proto.Reader) (err error) {
+	cmd.val, err = rd.ReadRawReply()
+	return err
+}
+
+func (cmd *RawCmd) Clone() Cmder {
+	var val []byte
+	if cmd.val != nil {
+		val = make([]byte, len(cmd.val))
+		copy(val, cmd.val)
+	}
+	return &RawCmd{
+		baseCmd: cmd.cloneBaseCmd(),
+		val:     val,
+	}
+}
+
+//------------------------------------------------------------------------------
+
+// RawWriteToCmd streams raw RESP protocol bytes directly to an io.Writer without intermediate allocations.
+type RawWriteToCmd struct {
+	baseCmd
+	w       io.Writer
+	written int64
+}
+
+var _ Cmder = (*RawWriteToCmd)(nil)
+
+func NewRawWriteToCmd(ctx context.Context, w io.Writer, args ...interface{}) *RawWriteToCmd {
+	return &RawWriteToCmd{
+		baseCmd: baseCmd{
+			ctx:     ctx,
+			args:    args,
+			cmdType: CmdTypeGeneric,
+		},
+		w: w,
+	}
+}
+
+func (cmd *RawWriteToCmd) SetVal(written int64) {
+	cmd.written = written
+}
+
+func (cmd *RawWriteToCmd) Val() int64 {
+	return cmd.written
+}
+
+func (cmd *RawWriteToCmd) Result() (int64, error) {
+	return cmd.written, cmd.err
+}
+
+func (cmd *RawWriteToCmd) String() string {
+	return cmdString(cmd, cmd.written)
+}
+
+func (cmd *RawWriteToCmd) readReply(rd *proto.Reader) (err error) {
+	cmd.written, err = rd.ReadRawReplyWriteTo(cmd.w)
+	return err
+}
+
+// NoRetry returns true because RawWriteToCmd writes directly to an io.Writer.
+// If a retry occurs, partial data from failed attempts would be appended to
+// the writer, causing data corruption. The caller must handle retries manually
+// if needed, using a fresh writer for each attempt.
+func (cmd *RawWriteToCmd) NoRetry() bool {
+	return true
+}
+
+func (cmd *RawWriteToCmd) Clone() Cmder {
+	return &RawWriteToCmd{
+		baseCmd: cmd.cloneBaseCmd(),
+		w:       cmd.w,
+		written: cmd.written,
+	}
+}
+
+//------------------------------------------------------------------------------
+
 type SliceCmd struct {
 	baseCmd
 

command_test.go

@@ -1,7 +1,9 @@
 package redis_test
 
 import (
+	"bytes"
 	"errors"
+	"strconv"
 	"time"
 
 	"github.com/redis/go-redis/v9"
@@ -10,6 +12,35 @@ import (
 	. "github.com/bsm/gomega"
 )
 
+// parseRESPType returns the RESP type character and validates basic structure
+func parseRESPType(data []byte) (byte, bool) {
+	if len(data) < 3 {
+		return 0, false
+	}
+	// Check for CRLF at the end
+	if !bytes.HasSuffix(data, []byte("\r\n")) {
+		return 0, false
+	}
+	return data[0], true
+}
+
+// countRESPElements counts the number of elements in a RESP array/map response
+func countRESPElements(data []byte) (int, bool) {
+	if len(data) < 4 {
+		return 0, false
+	}
+	// Find first CRLF
+	idx := bytes.Index(data, []byte("\r\n"))
+	if idx < 2 {
+		return 0, false
+	}
+	count, err := strconv.Atoi(string(data[1:idx]))
+	if err != nil {
+		return 0, false
+	}
+	return count, true
+}
+
 var _ = Describe("Cmd", func() {
 	var client *redis.Client
 
@@ -94,3 +125,419 @@ var _ = Describe("Cmd", func() {
 		Expect(err).To(Equal(e))
 	})
 })
+
+var _ = Describe("RawCmd", func() {
+	var client *redis.Client
+
+	BeforeEach(func() {
+		client = redis.NewClient(redisOptions())
+		Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())
+	})
+
+	AfterEach(func() {
+		Expect(client.Close()).NotTo(HaveOccurred())
+	})
+
+	It("returns raw RESP bytes for simple string (PING)", func() {
+		cmd := client.DoRaw(ctx, "PING")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+		rawBytes, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(string(rawBytes)).To(Equal("+PONG\r\n"))
+	})
+
+	It("returns raw RESP bytes for bulk string (GET)", func() {
+		err := client.Set(ctx, "rawtest", "hello", 0).Err()
+		Expect(err).NotTo(HaveOccurred())
+
+		cmd := client.DoRaw(ctx, "GET", "rawtest")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+		rawBytes, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(string(rawBytes)).To(Equal("$5\r\nhello\r\n"))
+	})
+
+	It("returns raw RESP bytes for integer (INCR)", func() {
+		err := client.Set(ctx, "counter", "10", 0).Err()
+		Expect(err).NotTo(HaveOccurred())
+
+		cmd := client.DoRaw(ctx, "INCR", "counter")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+		rawBytes, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(string(rawBytes)).To(Equal(":11\r\n"))
+	})
+
+	It("returns raw RESP bytes for array (KEYS) with validation", func() {
+		err := client.Set(ctx, "key1", "val1", 0).Err()
+		Expect(err).NotTo(HaveOccurred())
+		err = client.Set(ctx, "key2", "val2", 0).Err()
+		Expect(err).NotTo(HaveOccurred())
+
+		cmd := client.DoRaw(ctx, "KEYS", "key*")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+		rawBytes, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+
+		// Validate RESP structure
+		respType, valid := parseRESPType(rawBytes)
+		Expect(valid).To(BeTrue())
+		Expect(respType).To(Equal(byte('*')), "Should be array type")
+
+		// Verify element count
+		count, ok := countRESPElements(rawBytes)
+		Expect(ok).To(BeTrue())
+		Expect(count).To(Equal(2))
+
+		// Verify keys are present in the raw bytes
+		rawStr := string(rawBytes)
+		Expect(rawStr).To(ContainSubstring("key1"))
+		Expect(rawStr).To(ContainSubstring("key2"))
+	})
+
+	It("returns raw RESP bytes for HGETALL with complete validation", func() {
+		err := client.HSet(ctx, "myhash", "field1", "value1", "field2", "value2").Err()
+		Expect(err).NotTo(HaveOccurred())
+
+		cmd := client.DoRaw(ctx, "HGETALL", "myhash")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+		rawBytes, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+
+		// Validate RESP structure
+		respType, valid := parseRESPType(rawBytes)
+		Expect(valid).To(BeTrue(), "Response should be valid RESP")
+
+		// HGETALL returns array (RESP2) or map (RESP3)
+		Expect(respType).To(BeElementOf(byte('*'), byte('%')), "Should be array or map type")
+
+		// Count elements: array has 4 elements (2 key-value pairs), map has 2 pairs
+		count, ok := countRESPElements(rawBytes)
+		Expect(ok).To(BeTrue())
+		if respType == '*' {
+			Expect(count).To(Equal(4), "Array should have 4 elements (2 key-value pairs)")
+		} else {
+			Expect(count).To(Equal(2), "Map should have 2 pairs")
+		}
+
+		rawStr := string(rawBytes)
+		Expect(rawStr).To(ContainSubstring("field1"))
+		Expect(rawStr).To(ContainSubstring("value1"))
+		Expect(rawStr).To(ContainSubstring("field2"))
+		Expect(rawStr).To(ContainSubstring("value2"))
+
+		// Verify response ends with CRLF
+		Expect(rawBytes[len(rawBytes)-2:]).To(Equal([]byte("\r\n")))
+	})
+
+	It("returns raw RESP bytes for nil value", func() {
+		cmd := client.DoRaw(ctx, "GET", "nonexistent")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+		rawBytes, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		rawStr := string(rawBytes)
+		Expect(rawStr == "$-1\r\n" || rawStr == "_\r\n").To(BeTrue())
+	})
+
+	It("returns raw RESP bytes for empty array", func() {
+		cmd := client.DoRaw(ctx, "KEYS", "nonexistent*")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+		rawBytes, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(string(rawBytes)).To(Equal("*0\r\n"))
+	})
+
+	It("has Val and Bytes methods", func() {
+		cmd := client.DoRaw(ctx, "PING")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+
+		val := cmd.Val()
+		Expect(string(val)).To(Equal("+PONG\r\n"))
+
+		bytes, err := cmd.Bytes()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(string(bytes)).To(Equal("+PONG\r\n"))
+	})
+
+	It("implements Stringer", func() {
+		cmd := client.DoRaw(ctx, "PING")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+		Expect(cmd.String()).To(ContainSubstring("PING"))
+	})
+
+	It("allows to set custom error", func() {
+		e := errors.New("custom error")
+		cmd := redis.RawCmd{}
+		cmd.SetErr(e)
+		_, err := cmd.Result()
+		Expect(err).To(Equal(e))
+	})
+
+	It("can be cloned", func() {
+		cmd := client.DoRaw(ctx, "PING")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+
+		cloned := cmd.Clone().(*redis.RawCmd)
+		Expect(cloned.Val()).To(Equal(cmd.Val()))
+		originalVal := cmd.Val()
+		clonedVal := cloned.Val()
+		Expect(&originalVal[0]).NotTo(BeIdenticalTo(&clonedVal[0]))
+	})
+
+	It("returns raw RESP bytes for error response", func() {
+		cmd := client.DoRaw(ctx, "INVALIDCMD")
+		rawBytes, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(string(rawBytes)).To(HavePrefix("-"))
+		Expect(string(rawBytes)).To(ContainSubstring("ERR"))
+	})
+
+	It("returns raw RESP bytes for empty hash (nil map)", func() {
+		// HGETALL on non-existent hash returns empty array/map
+		cmd := client.DoRaw(ctx, "HGETALL", "nonexistent_hash")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+		rawBytes, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+
+		respType, valid := parseRESPType(rawBytes)
+		Expect(valid).To(BeTrue())
+		// Should be empty array or empty map
+		Expect(respType).To(BeElementOf(byte('*'), byte('%')))
+
+		count, ok := countRESPElements(rawBytes)
+		Expect(ok).To(BeTrue())
+		Expect(count).To(Equal(0), "Empty hash should return 0 elements")
+	})
+
+	It("returns raw RESP bytes for nested array structures (SMEMBERS)", func() {
+		// Create a set with multiple members
+		err := client.SAdd(ctx, "myset", "member1", "member2", "member3").Err()
+		Expect(err).NotTo(HaveOccurred())
+
+		cmd := client.DoRaw(ctx, "SMEMBERS", "myset")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+		rawBytes, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+
+		// Validate basic RESP structure
+		respType, valid := parseRESPType(rawBytes)
+		Expect(valid).To(BeTrue())
+		// SMEMBERS returns array (RESP2) or set (RESP3)
+		Expect(respType).To(BeElementOf(byte('*'), byte('~')))
+
+		// Verify count
+		count, ok := countRESPElements(rawBytes)
+		Expect(ok).To(BeTrue())
+		Expect(count).To(Equal(3))
+
+		// Verify members are present
+		rawStr := string(rawBytes)
+		Expect(rawStr).To(ContainSubstring("member1"))
+		Expect(rawStr).To(ContainSubstring("member2"))
+		Expect(rawStr).To(ContainSubstring("member3"))
+	})
+
+	It("returns raw RESP bytes for LRANGE with exact byte validation", func() {
+		// Create a list with multiple elements
+		err := client.RPush(ctx, "mylist", "elem1", "elem2", "elem3").Err()
+		Expect(err).NotTo(HaveOccurred())
+
+		cmd := client.DoRaw(ctx, "LRANGE", "mylist", "0", "-1")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+		rawBytes, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+
+		// LRANGE returns elements in order, so we can validate exact bytes
+		expectedResp := "*3\r\n$5\r\nelem1\r\n$5\r\nelem2\r\n$5\r\nelem3\r\n"
+		Expect(string(rawBytes)).To(Equal(expectedResp))
+	})
+
+	It("returns raw RESP bytes for LPUSH with exact integer response", func() {
+		// LPUSH returns the length of the list after push
+		cmd := client.DoRaw(ctx, "LPUSH", "newlist", "item")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+		rawBytes, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(string(rawBytes)).To(Equal(":1\r\n"))
+	})
+
+	It("returns raw RESP bytes for ECHO with exact bulk string", func() {
+		cmd := client.DoRaw(ctx, "ECHO", "testmsg")
+		Expect(cmd.Err()).NotTo(HaveOccurred())
+		rawBytes, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(string(rawBytes)).To(Equal("$7\r\ntestmsg\r\n"))
+	})
+})
+
+var _ = Describe("RawWriteToCmd", func() {
+	var client *redis.Client
+
+	BeforeEach(func() {
+		client = redis.NewClient(redisOptions())
+		Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())
+	})
+
+	AfterEach(func() {
+		Expect(client.Close()).NotTo(HaveOccurred())
+	})
+
+	It("streams raw RESP bytes for PING", func() {
+		var buf bytes.Buffer
+		cmd := client.DoRawWriteTo(ctx, &buf, "PING")
+		written, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(written).To(Equal(int64(7)))
+		Expect(buf.String()).To(Equal("+PONG\r\n"))
+	})
+
+	It("streams raw RESP bytes for bulk string (GET)", func() {
+		client.Set(ctx, "rawkey", "rawvalue", 0)
+		var buf bytes.Buffer
+		cmd := client.DoRawWriteTo(ctx, &buf, "GET", "rawkey")
+		written, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(written).To(BeNumerically(">", 0))
+		Expect(buf.String()).To(Equal("$8\r\nrawvalue\r\n"))
+	})
+
+	It("streams raw RESP bytes for integer (INCR)", func() {
+		client.Set(ctx, "writeto_counter", "10", 0)
+		var buf bytes.Buffer
+		cmd := client.DoRawWriteTo(ctx, &buf, "INCR", "writeto_counter")
+		written, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(written).To(Equal(int64(5)))
+		Expect(buf.String()).To(Equal(":11\r\n"))
+	})
+
+	It("streams raw RESP bytes for nil value (GET non-existent)", func() {
+		var buf bytes.Buffer
+		cmd := client.DoRawWriteTo(ctx, &buf, "GET", "nonexistent_key")
+		written, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(written).To(Equal(int64(buf.Len())))
+		rawStr := buf.String()
+		// RESP2 returns $-1\r\n, RESP3 returns _\r\n
+		Expect(rawStr == "$-1\r\n" || rawStr == "_\r\n").To(BeTrue())
+	})
+
+	It("streams raw RESP bytes for array (LRANGE) with exact validation", func() {
+		client.RPush(ctx, "writeto_list", "a", "b", "c")
+		var buf bytes.Buffer
+		cmd := client.DoRawWriteTo(ctx, &buf, "LRANGE", "writeto_list", "0", "-1")
+		written, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(written).To(Equal(int64(buf.Len())))
+		// LRANGE returns elements in order
+		expectedResp := "*3\r\n$1\r\na\r\n$1\r\nb\r\n$1\r\nc\r\n"
+		Expect(buf.String()).To(Equal(expectedResp))
+	})
+
+	It("streams raw RESP bytes for empty array (LRANGE empty list)", func() {
+		var buf bytes.Buffer
+		cmd := client.DoRawWriteTo(ctx, &buf, "LRANGE", "nonexistent_list", "0", "-1")
+		written, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(written).To(Equal(int64(buf.Len())))
+		Expect(buf.String()).To(Equal("*0\r\n"))
+	})
+
+	It("streams raw RESP bytes for HGETALL", func() {
+		client.HSet(ctx, "rawhash", "f1", "v1", "f2", "v2")
+		var buf bytes.Buffer
+		cmd := client.DoRawWriteTo(ctx, &buf, "HGETALL", "rawhash")
+		written, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(written).To(BeNumerically(">", 0))
+		raw := buf.String()
+		Expect(raw).To(ContainSubstring("f1"))
+		Expect(raw).To(ContainSubstring("v1"))
+	})
+
+	It("returns correct byte count", func() {
+		var buf bytes.Buffer
+		cmd := client.DoRawWriteTo(ctx, &buf, "PING")
+		written, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(written).To(Equal(int64(buf.Len())))
+	})
+
+	It("implements Stringer", func() {
+		var buf bytes.Buffer
+		cmd := client.DoRawWriteTo(ctx, &buf, "PING")
+		Expect(cmd.String()).To(ContainSubstring("PING"))
+	})
+
+	It("can be cloned", func() {
+		var buf bytes.Buffer
+		cmd := client.DoRawWriteTo(ctx, &buf, "PING")
+		clone := cmd.Clone().(*redis.RawWriteToCmd)
+		Expect(clone.Val()).To(Equal(cmd.Val()))
+	})
+
+	It("streams HGETALL with complete validation", func() {
+		client.HSet(ctx, "rawhash2", "key1", "val1", "key2", "val2")
+		var buf bytes.Buffer
+		cmd := client.DoRawWriteTo(ctx, &buf, "HGETALL", "rawhash2")
+		written, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+
+		// Verify written bytes match buffer length
+		Expect(written).To(Equal(int64(buf.Len())))
+
+		// Validate RESP structure
+		rawBytes := buf.Bytes()
+		respType, valid := parseRESPType(rawBytes)
+		Expect(valid).To(BeTrue())
+		Expect(respType).To(BeElementOf(byte('*'), byte('%')))
+
+		// Verify all data present
+		rawStr := buf.String()
+		Expect(rawStr).To(ContainSubstring("key1"))
+		Expect(rawStr).To(ContainSubstring("val1"))
+		Expect(rawStr).To(ContainSubstring("key2"))
+		Expect(rawStr).To(ContainSubstring("val2"))
+	})
+
+	It("streams empty hash correctly", func() {
+		var buf bytes.Buffer
+		cmd := client.DoRawWriteTo(ctx, &buf, "HGETALL", "nonexistent_hash")
+		written, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(written).To(Equal(int64(buf.Len())))
+
+		rawBytes := buf.Bytes()
+		respType, valid := parseRESPType(rawBytes)
+		Expect(valid).To(BeTrue())
+		Expect(respType).To(BeElementOf(byte('*'), byte('%')))
+
+		count, ok := countRESPElements(rawBytes)
+		Expect(ok).To(BeTrue())
+		Expect(count).To(Equal(0))
+	})
+
+	It("streams error response correctly", func() {
+		var buf bytes.Buffer
+		cmd := client.DoRawWriteTo(ctx, &buf, "INVALIDCMD")
+		_, err := cmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+
+		rawStr := buf.String()
+		Expect(rawStr).To(HavePrefix("-"))
+		Expect(rawStr).To(ContainSubstring("ERR"))
+	})
+})
+
+var _ = Describe("NoRetry behavior", func() {
+	It("RawCmd does not have NoRetry set", func() {
+		cmd := redis.NewRawCmd(ctx, "PING")
+		Expect(cmd.NoRetry()).To(BeFalse())
+	})
+
+	It("RawWriteToCmd has NoRetry set to prevent data corruption", func() {
+		var buf bytes.Buffer
+		cmd := redis.NewRawWriteToCmd(ctx, &buf, "PING")
+		Expect(cmd.NoRetry()).To(BeTrue())
+	})
+})

commands.go

@@ -449,6 +449,20 @@ func (c cmdable) Do(ctx context.Context, args ...interface{}) *Cmd {
 	return cmd
 }
 
+// DoRaw executes a command and returns the raw RESP protocol bytes without parsing.
+func (c cmdable) DoRaw(ctx context.Context, args ...interface{}) *RawCmd {
+	cmd := NewRawCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+// DoRawWriteTo executes a command and streams raw RESP bytes directly to w without intermediate allocations.
+func (c cmdable) DoRawWriteTo(ctx context.Context, w io.Writer, args ...interface{}) *RawWriteToCmd {
+	cmd := NewRawWriteToCmd(ctx, w, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
 // Quit closes the connection.
 //
 // Deprecated: Just close the connection instead as of Redis 7.2.0.

internal/proto/reader.go

@@ -646,3 +646,193 @@ func IsNilReply(line []byte) bool {
 		(line[0] == RespString || line[0] == RespArray) &&
 		line[1] == '-' && line[2] == '1'
 }
+
+// ReadRawReply reads the next RESP reply and returns it as raw bytes without parsing.
+func (r *Reader) ReadRawReply() ([]byte, error) {
+	return r.readRawReplyBuf(nil)
+}
+
+func (r *Reader) readRawReplyBuf(buf []byte) ([]byte, error) {
+	line, err := r.readLine()
+	if err != nil {
+		return buf, err
+	}
+
+	buf = append(buf, line...)
+	buf = append(buf, '\r', '\n')
+
+	switch line[0] {
+	case RespStatus, RespError, RespInt, RespNil, RespFloat, RespBool, RespBigInt:
+		return buf, nil
+
+	case RespString, RespVerbatim, RespBlobError:
+		n, err := replyLen(line)
+		if err != nil {
+			if err == Nil {
+				return buf, nil
+			}
+			return buf, err
+		}
+		curLen := len(buf)
+		buf = append(buf, make([]byte, n+2)...)
+		_, err = io.ReadFull(r.rd, buf[curLen:])
+		return buf, err
+
+	case RespArray, RespSet, RespPush:
+		n, err := replyLen(line)
+		if err != nil {
+			if err == Nil {
+				return buf, nil
+			}
+			return buf, err
+		}
+		for i := 0; i < n; i++ {
+			buf, err = r.readRawReplyBuf(buf)
+			if err != nil {
+				return buf, err
+			}
+		}
+		return buf, nil
+
+	case RespMap:
+		n, err := replyLen(line)
+		if err != nil {
+			if err == Nil {
+				return buf, nil
+			}
+			return buf, err
+		}
+		for i := 0; i < n*2; i++ {
+			buf, err = r.readRawReplyBuf(buf)
+			if err != nil {
+				return buf, err
+			}
+		}
+		return buf, nil
+
+	case RespAttr:
+		// Per RESP3 spec, an attribute is always followed by the actual command reply.
+		// We need to read the attribute's key-value pairs AND the following reply.
+		n, err := replyLen(line)
+		if err != nil {
+			if err == Nil {
+				return buf, nil
+			}
+			return buf, err
+		}
+		// Read the attribute key-value pairs
+		for i := 0; i < n*2; i++ {
+			buf, err = r.readRawReplyBuf(buf)
+			if err != nil {
+				return buf, err
+			}
+		}
+		// Read the command reply that follows the attribute
+		return r.readRawReplyBuf(buf)
+	}
+
+	return buf, fmt.Errorf("redis: can't read raw reply: %.100q", line)
+}
+
+var crlf = []byte{'\r', '\n'}
+
+// ReadRawReplyWriteTo streams the next RESP reply directly to w without intermediate allocations.
+// Returns the number of bytes written and any error encountered.
+func (r *Reader) ReadRawReplyWriteTo(w io.Writer) (int64, error) {
+	return r.readRawReplyWriteTo(w)
+}
+
+func (r *Reader) readRawReplyWriteTo(w io.Writer) (int64, error) {
+	line, err := r.readLine()
+	if err != nil {
+		return 0, err
+	}
+
+	var written int64
+	n, err := w.Write(line)
+	written += int64(n)
+	if err != nil {
+		return written, err
+	}
+	n, err = w.Write(crlf)
+	written += int64(n)
+	if err != nil {
+		return written, err
+	}
+
+	switch line[0] {
+	case RespStatus, RespError, RespInt, RespNil, RespFloat, RespBool, RespBigInt:
+		return written, nil
+
+	case RespString, RespVerbatim, RespBlobError:
+		dataLen, err := replyLen(line)
+		if err != nil {
+			if err == Nil {
+				return written, nil
+			}
+			return written, err
+		}
+		copied, err := io.CopyN(w, r.rd, int64(dataLen)+2)
+		written += copied
+		return written, err
+
+	case RespArray, RespSet, RespPush:
+		count, err := replyLen(line)
+		if err != nil {
+			if err == Nil {
+				return written, nil
+			}
+			return written, err
+		}
+		for i := 0; i < count; i++ {
+			n, err := r.readRawReplyWriteTo(w)
+			written += n
+			if err != nil {
+				return written, err
+			}
+		}
+		return written, nil
+
+	case RespMap:
+		count, err := replyLen(line)
+		if err != nil {
+			if err == Nil {
+				return written, nil
+			}
+			return written, err
+		}
+		for i := 0; i < count*2; i++ {
+			n, err := r.readRawReplyWriteTo(w)
+			written += n
+			if err != nil {
+				return written, err
+			}
+		}
+		return written, nil
+
+	case RespAttr:
+		// Per RESP3 spec, an attribute is always followed by the actual command reply.
+		// We need to read the attribute's key-value pairs AND the following reply.
+		count, err := replyLen(line)
+		if err != nil {
+			if err == Nil {
+				return written, nil
+			}
+			return written, err
+		}
+		// Read the attribute key-value pairs
+		for i := 0; i < count*2; i++ {
+			n, err := r.readRawReplyWriteTo(w)
+			written += n
+			if err != nil {
+				return written, err
+			}
+		}
+		// Read the command reply that follows the attribute
+		n, err := r.readRawReplyWriteTo(w)
+		written += n
+		return written, err
+	}
+
+	return written, fmt.Errorf("redis: can't read raw reply: %.100q", line)
+}

osscluster.go

@@ -1300,7 +1300,7 @@ func (c *ClusterClient) process(ctx context.Context, cmd Cmder) error {
 			continue
 		}
 
-		if shouldRetry(lastErr, cmd.readTimeout() == nil) {
+		if shouldRetry(lastErr, cmd.readTimeout() == nil) && !cmd.NoRetry() {
 			// First retry the same node.
 			if attempt == 0 {
 				continue
@@ -1716,7 +1716,7 @@ func (c *ClusterClient) processPipelineNodeConn(
 		if isBadConn(err, false, node.Client.getAddr()) {
 			node.MarkAsFailing()
 		}
-		if shouldRetry(err, true) {
+		if shouldRetry(err, true) && !cmdsContainNoRetry(cmds) {
 			_ = c.mapCmdsByNode(ctx, failedCmds, cmds)
 		}
 		setCmdsErr(cmds, err)
@@ -1752,7 +1752,7 @@ func (c *ClusterClient) pipelineReadCmds(
 		}
 
 		if !isRedisError(err) {
-			if shouldRetry(err, true) {
+			if shouldRetry(err, true) && !cmdsContainNoRetry(cmds) {
 				_ = c.mapCmdsByNode(ctx, failedCmds, cmds)
 			}
 			setCmdsErr(cmds[i+1:], err)
@@ -1760,7 +1760,7 @@ func (c *ClusterClient) pipelineReadCmds(
 		}
 	}
 
-	if err := cmds[0].Err(); err != nil && shouldRetry(err, true) {
+	if err := cmds[0].Err(); err != nil && shouldRetry(err, true) && !cmdsContainNoRetry(cmds) {
 		_ = c.mapCmdsByNode(ctx, failedCmds, cmds)
 		return err
 	}
@@ -1963,7 +1963,7 @@ func (c *ClusterClient) processTxPipelineNodeConn(
 	if err := cn.WithWriter(c.context(ctx), c.opt.WriteTimeout, func(wr *proto.Writer) error {
 		return writeCmds(wr, cmds)
 	}); err != nil {
-		if shouldRetry(err, true) {
+		if shouldRetry(err, true) && !cmdsContainNoRetry(cmds) {
 			_ = c.mapCmdsByNode(ctx, failedCmds, cmds)
 		}
 		setCmdsErr(cmds, err)

redis.go

@@ -716,7 +716,9 @@ func (c *baseClient) process(ctx context.Context, cmd Cmder) error {
 		if cn != nil {
 			lastConn = cn
 		}
-		if err == nil || !retry {
+		// Don't retry if command explicitly disables retries (e.g., RawWriteToCmd
+		// which writes directly to an io.Writer and cannot undo partial writes)
+		if err == nil || !retry || cmd.NoRetry() {
 			// Record total operation duration
 			if opDurationCallback != nil {
 				operationDuration := time.Since(operationStart)
@@ -1041,7 +1043,10 @@ func (c *baseClient) generalProcessPipeline(
 			canRetry, err = p(ctx, cn, cmds)
 			return err
 		})
-		if lastErr == nil || !canRetry || !shouldRetry(lastErr, true) {
+		// Don't retry if any command in the pipeline explicitly disables retries
+		// (e.g., RawWriteToCmd which writes directly to an io.Writer and cannot
+		// undo partial writes on retry)
+		if lastErr == nil || !canRetry || !shouldRetry(lastErr, true) || cmdsContainNoRetry(cmds) {
 			// The error should be set here only when failing to obtain the conn.
 			if !isRedisError(lastErr) {
 				setCmdsErr(cmds, lastErr)

ring.go

@@ -802,7 +802,7 @@ func (c *Ring) process(ctx context.Context, cmd Cmder) error {
 		}
 
 		lastErr = shard.Client.Process(ctx, cmd)
-		if lastErr == nil || !shouldRetry(lastErr, cmd.readTimeout() == nil) {
+		if lastErr == nil || !shouldRetry(lastErr, cmd.readTimeout() == nil) || cmd.NoRetry() {
 			return lastErr
 		}
 	}