diff --git a/hrpc/call.go b/hrpc/call.go
index 8e84d56e..a15e6e4c 100644
--- a/hrpc/call.go
+++ b/hrpc/call.go
@@ -259,12 +259,12 @@ type Result struct {
 	Cells   []*Cell
 	Stale   bool
 	Partial bool
-	// Exists is only set if existance_only was set in the request query.
+	// Exists is only set if existence_only was set in the request query.
 	Exists *bool
 }
 
 func (c *Result) String() string {
-	return fmt.Sprintf("cells:%v stale:%v partial:%v exists:%v ",
+	return fmt.Sprintf("cells:%v stale:%v partial:%v exists:%v",
 		c.Cells, c.Stale, c.Partial, c.Exists)
 }
 
diff --git a/hrpc/hrpc_test.go b/hrpc/hrpc_test.go
index 3e572157..c8934abf 100644
--- a/hrpc/hrpc_test.go
+++ b/hrpc/hrpc_test.go
@@ -285,6 +285,7 @@ func TestScanToProto(t *testing.T) {
 					Column:        []*pb.Column{},
 					TimeRange:     &pb.TimeRange{},
 				},
+				TrackScanMetrics: proto.Bool(false),
 			},
 		},
 		{ // explicitly set configurable attributes to default values
@@ -317,6 +318,7 @@ func TestScanToProto(t *testing.T) {
 					MaxVersions:   nil,
 					CacheBlocks:   nil,
 				},
+				TrackScanMetrics: proto.Bool(false),
 			},
 		},
 		{ // set configurable attributes to non-default values
@@ -350,6 +352,7 @@ func TestScanToProto(t *testing.T) {
 					MaxVersions: proto.Uint32(89),
 					CacheBlocks: proto.Bool(!DefaultCacheBlocks),
 				},
+				TrackScanMetrics: proto.Bool(false),
 			},
 		},
 		{ // test that pb.ScanRequest.Scan is nil when scanner id is specificed
@@ -374,6 +377,7 @@ func TestScanToProto(t *testing.T) {
 				ClientHandlesPartials:   proto.Bool(true),
 				ClientHandlesHeartbeats: proto.Bool(true),
 				Scan:                    nil,
+				TrackScanMetrics:        proto.Bool(false),
 			},
 		},
 		{ // set reversed attribute
@@ -393,6 +397,7 @@ func TestScanToProto(t *testing.T) {
 					TimeRange:     &pb.TimeRange{},
 					Reversed:      proto.Bool(true),
 				},
+				TrackScanMetrics: proto.Bool(false),
 			},
 		},
 		{ // set scan attribute
@@ -418,6 +423,7 @@ func TestScanToProto(t *testing.T) {
 						{Name: proto.String("key2"), Value: []byte("value2")},
 					},
 				},
+				TrackScanMetrics: proto.Bool(false),
 			},
 		},
 		{ // scan key range
@@ -438,6 +444,7 @@ func TestScanToProto(t *testing.T) {
 					StartRow:      startRow,
 					StopRow:       stopRow,
 				},
+				TrackScanMetrics: proto.Bool(false),
 			},
 		},
 		{ // set filters and families
@@ -459,6 +466,7 @@ func TestScanToProto(t *testing.T) {
 						TimeRange:     &pb.TimeRange{},
 						Filter:        pbFilter,
 					},
+					TrackScanMetrics: proto.Bool(false),
 				}
 			}(),
 		},
@@ -478,8 +486,30 @@ func TestScanToProto(t *testing.T) {
 					Column:        []*pb.Column{},
 					TimeRange:     &pb.TimeRange{},
 				},
+				TrackScanMetrics: proto.Bool(false),
 			},
 		},
+		// set TrackScanMetrics
+		{
+			s: func() *Scan {
+				s, _ := NewScanStr(ctx, "", TrackScanMetrics())
+				return s
+			}(),
+			expProto: func() *pb.ScanRequest {
+				return &pb.ScanRequest{
+					Region:                  rs,
+					NumberOfRows:            proto.Uint32(DefaultNumberOfRows),
+					CloseScanner:            proto.Bool(false),
+					ClientHandlesPartials:   proto.Bool(true),
+					ClientHandlesHeartbeats: proto.Bool(true),
+					Scan: &pb.Scan{
+						MaxResultSize: proto.Uint64(DefaultMaxResultSize),
+						TimeRange:     &pb.TimeRange{},
+					},
+					TrackScanMetrics: proto.Bool(true),
+				}
+			}(),
+		},
 	}
 
 	for i, tcase := range tests {
diff --git a/hrpc/scan.go b/hrpc/scan.go
index 86b5a482..b3bd98a5 100644
--- a/hrpc/scan.go
+++ b/hrpc/scan.go
@@ -49,9 +49,14 @@ type Scanner interface {
 
 	// Close should be called if it is desired to stop scanning before getting all of results.
 	// If you call Next() after calling Close() you might still get buffered results.
-	// Othwerwise, in case all results have been delivered or in case of an error, the Scanner
+	// Otherwise, in case all results have been delivered or in case of an error, the Scanner
 	// will be closed automatically. It's okay to close an already closed scanner.
 	Close() error
+	// GetScanMetrics returns the scan metrics for the scanner.
+	// The scan metrics are non-nil only if the Scan has TrackScanMetrics() enabled.
+	// GetScanMetrics should only be called after the scanner has been closed with an io.EOF
+	// (ie there are no more rows left to be returned by calls to Next()).
+	GetScanMetrics() map[string]int64
 }
 
 // Scan represents a scanner on an HBase table.
@@ -64,10 +69,11 @@ type Scan struct {
 
 	scannerID uint64
 
-	maxResultSize uint64
-	numberOfRows  uint32
-	reversed      bool
-	attribute     []*pb.NameBytesPair
+	maxResultSize    uint64
+	numberOfRows     uint32
+	reversed         bool
+	attribute        []*pb.NameBytesPair
+	trackScanMetrics bool
 
 	closeScanner        bool
 	allowPartialResults bool
@@ -178,6 +184,11 @@ func (s *Scan) NumberOfRows() uint32 {
 	return s.numberOfRows
 }
 
+// TrackScanMetrics returns true if the client is requesting to track scan metrics.
+func (s *Scan) TrackScanMetrics() bool {
+	return s.trackScanMetrics
+}
+
 // ToProto converts this Scan into a protobuf message
 func (s *Scan) ToProto() proto.Message {
 	scan := &pb.ScanRequest{
@@ -189,6 +200,7 @@ func (s *Scan) ToProto() proto.Message {
 		// tell server that we "handle" heartbeats by ignoring them
 		// since we don't really time out our scans (unless context was cancelled)
 		ClientHandlesHeartbeats: proto.Bool(true),
+		TrackScanMetrics:        &s.trackScanMetrics,
 	}
 	if s.scannerID != math.MaxUint64 {
 		scan.ScannerId = &s.scannerID
@@ -334,6 +346,19 @@ func AllowPartialResults() func(Call) error {
 	}
 }
 
+// TrackScanMetrics is an option for scan requests.
+// Enables tracking scan metrics from HBase, which will be returned in the scan response.
+func TrackScanMetrics() func(Call) error {
+	return func(g Call) error {
+		scan, ok := g.(*Scan)
+		if !ok {
+			return errors.New("'TrackScanMetrics' option can only be used with Scan queries")
+		}
+		scan.trackScanMetrics = true
+		return nil
+	}
+}
+
 // Reversed is a Scan-only option which allows you to scan in reverse key order
 // To use it the startKey would be greater than the end key
 func Reversed() func(Call) error {
diff --git a/integration_test.go b/integration_test.go
index 7918d1dc..8e5b3413 100644
--- a/integration_test.go
+++ b/integration_test.go
@@ -16,6 +16,7 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"math"
 	"os"
 	"os/exec"
 	"reflect"
@@ -25,8 +26,6 @@ import (
 	"testing"
 	"time"
 
-	"math"
-
 	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/tsuna/gohbase"
@@ -1022,6 +1021,110 @@ func TestScanTimeRangeVersions(t *testing.T) {
 	}
 }
 
+func TestScanWithScanMetrics(t *testing.T) {
+	var (
+		key          = "TestScanWithScanMetrics"
+		now          = time.Now()
+		r1           = fmt.Sprintf("%s_%d", key, 1)
+		r2           = fmt.Sprintf("%s_%d", key, 2)
+		r3           = fmt.Sprintf("%s_%d", key, 3)
+		val          = []byte("1")
+		family       = "cf"
+		ctx          = context.Background()
+		rowsScanned  = "ROWS_SCANNED"
+		rowsFiltered = "ROWS_FILTERED"
+	)
+
+	c := gohbase.NewClient(*host)
+	defer c.Close()
+
+	for _, r := range []string{r1, r2, r3} {
+		err := insertKeyValue(c, r, family, val, hrpc.Timestamp(now))
+		if err != nil {
+			t.Fatalf("Put failed: %s", err)
+		}
+	}
+
+	tcases := []struct {
+		description          string
+		filters              func(call hrpc.Call) error
+		expectedRowsScanned  int64
+		expectedRowsFiltered int64
+		noScanMetrics        bool
+	}{
+		{
+			description:          "scan metrics not enabled",
+			expectedRowsScanned:  0,
+			expectedRowsFiltered: 0,
+			noScanMetrics:        true,
+		},
+		{
+			description:          "2 rows scanned",
+			expectedRowsScanned:  2,
+			expectedRowsFiltered: 0,
+		},
+		{
+			description:          "1 row scanned 1 row filtered",
+			filters:              hrpc.Filters(filter.NewPrefixFilter([]byte(r1))),
+			expectedRowsScanned:  1,
+			expectedRowsFiltered: 1,
+		},
+	}
+
+	for _, tc := range tcases {
+		t.Run(tc.description, func(t *testing.T) {
+			var (
+				scan *hrpc.Scan
+				err  error
+			)
+			if tc.noScanMetrics {
+				scan, err = hrpc.NewScanRangeStr(ctx, table, r1, r3)
+			} else if tc.filters == nil {
+				scan, err = hrpc.NewScanRangeStr(ctx, table, r1, r3, hrpc.TrackScanMetrics())
+			} else {
+				scan, err = hrpc.NewScanRangeStr(ctx, table, r1, r3, hrpc.TrackScanMetrics(),
+					tc.filters)
+			}
+			if err != nil {
+				t.Fatalf("Scan req failed: %s", err)
+			}
+
+			var results []*hrpc.Result
+			scanner := c.Scan(scan)
+			for {
+				var r *hrpc.Result
+				r, err = scanner.Next()
+				if err == io.EOF {
+					break
+				}
+				if err != nil {
+					t.Fatal(err)
+				}
+				results = append(results, r)
+			}
+
+			actualMetrics := scanner.GetScanMetrics()
+
+			if tc.noScanMetrics && actualMetrics != nil {
+				t.Fatalf("Expected nil scan metrics, got %v", actualMetrics)
+			}
+
+			scanned := actualMetrics[rowsScanned]
+			if tc.expectedRowsScanned != scanned {
+				t.Errorf("Did not get expected rows scanned - expected: %d, actual %d",
+					tc.expectedRowsScanned, scanned)
+			}
+
+			filtered := actualMetrics[rowsFiltered]
+			if tc.expectedRowsFiltered != filtered {
+				t.Errorf("Did not get expected rows filtered - expected: %d, actual %d",
+					tc.expectedRowsFiltered, filtered)
+			}
+		})
+	}
+
+}
+
 func TestPutTTL(t *testing.T) {
 	key := "TestPutTTL"
 	c := gohbase.NewClient(*host)
@@ -1034,13 +1137,13 @@ func TestPutTTL(t *testing.T) {
 		t.Fatalf("Put failed: %s", err)
 	}
 
-	//Wait ttl duration and try to get the value
+	// Wait ttl duration and try to get the value
 	time.Sleep(ttl)
 
 	get, err := hrpc.NewGetStr(context.Background(), table, key,
 		hrpc.Families(map[string][]string{"cf": nil}))
 
-	//Make sure we dont get a result back
+	// Make sure we don't get a result back
 	res, err := c.Get(get)
 	if err != nil {
 		t.Fatalf("Get failed: %s", err)
diff --git a/pb/Cell.pb.go b/pb/Cell.pb.go
index 681c97d3..4b24cffe 100644
--- a/pb/Cell.pb.go
+++ b/pb/Cell.pb.go
@@ -19,8 +19,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: Cell.proto
 
 package pb
diff --git a/pb/Client.pb.go b/pb/Client.pb.go
index 4e215be9..04e65654 100644
--- a/pb/Client.pb.go
+++ b/pb/Client.pb.go
@@ -19,8 +19,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: Client.proto
 
 package pb
@@ -1538,6 +1538,10 @@ type ScanResponse struct {
 	// timing out. Seeing a heartbeat message communicates to the Client that the
 	// server would have continued to scan had the time limit not been reached.
 	HeartbeatMessage *bool `protobuf:"varint,9,opt,name=heartbeat_message,json=heartbeatMessage" json:"heartbeat_message,omitempty"`
+	// This field is filled in if the client has requested that scan metrics be tracked.
+	// The metrics tracked here are sent back to the client to be tracked together with
+	// the existing client side metrics.
+	ScanMetrics *ScanMetrics `protobuf:"bytes,10,opt,name=scan_metrics,json=scanMetrics" json:"scan_metrics,omitempty"`
 }
 
 func (x *ScanResponse) Reset() {
@@ -1635,6 +1639,60 @@ func (x *ScanResponse) GetHeartbeatMessage() bool {
 	return false
 }
 
+func (x *ScanResponse) GetScanMetrics() *ScanMetrics {
+	if x != nil {
+		return x.ScanMetrics
+	}
+	return nil
+}
+
+type ScanMetrics struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Metrics []*NameInt64Pair `protobuf:"bytes,1,rep,name=metrics" json:"metrics,omitempty"`
+}
+
+func (x *ScanMetrics) Reset() {
+	*x = ScanMetrics{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_Client_proto_msgTypes[14]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ScanMetrics) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ScanMetrics) ProtoMessage() {}
+
+func (x *ScanMetrics) ProtoReflect() protoreflect.Message {
+	mi := &file_Client_proto_msgTypes[14]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ScanMetrics.ProtoReflect.Descriptor instead.
+func (*ScanMetrics) Descriptor() ([]byte, []int) {
+	return file_Client_proto_rawDescGZIP(), []int{14}
+}
+
+func (x *ScanMetrics) GetMetrics() []*NameInt64Pair {
+	if x != nil {
+		return x.Metrics
+	}
+	return nil
+}
+
 // *
 // Atomically bulk load multiple HFiles (say from different column families)
 // into an open region.
@@ -1651,7 +1709,7 @@ type BulkLoadHFileRequest struct {
 func (x *BulkLoadHFileRequest) Reset() {
 	*x = BulkLoadHFileRequest{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[14]
+		mi := &file_Client_proto_msgTypes[15]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1664,7 +1722,7 @@ func (x *BulkLoadHFileRequest) String() string {
 func (*BulkLoadHFileRequest) ProtoMessage() {}
 
 func (x *BulkLoadHFileRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[14]
+	mi := &file_Client_proto_msgTypes[15]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1677,7 +1735,7 @@ func (x *BulkLoadHFileRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use BulkLoadHFileRequest.ProtoReflect.Descriptor instead.
 func (*BulkLoadHFileRequest) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{14}
+	return file_Client_proto_rawDescGZIP(), []int{15}
 }
 
 func (x *BulkLoadHFileRequest) GetRegion() *RegionSpecifier {
@@ -1712,7 +1770,7 @@ type BulkLoadHFileResponse struct {
 func (x *BulkLoadHFileResponse) Reset() {
 	*x = BulkLoadHFileResponse{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[15]
+		mi := &file_Client_proto_msgTypes[16]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1725,7 +1783,7 @@ func (x *BulkLoadHFileResponse) String() string {
 func (*BulkLoadHFileResponse) ProtoMessage() {}
 
 func (x *BulkLoadHFileResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[15]
+	mi := &file_Client_proto_msgTypes[16]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1738,7 +1796,7 @@ func (x *BulkLoadHFileResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use BulkLoadHFileResponse.ProtoReflect.Descriptor instead.
 func (*BulkLoadHFileResponse) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{15}
+	return file_Client_proto_rawDescGZIP(), []int{16}
 }
 
 func (x *BulkLoadHFileResponse) GetLoaded() bool {
@@ -1762,7 +1820,7 @@ type CoprocessorServiceCall struct {
 func (x *CoprocessorServiceCall) Reset() {
 	*x = CoprocessorServiceCall{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[16]
+		mi := &file_Client_proto_msgTypes[17]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1775,7 +1833,7 @@ func (x *CoprocessorServiceCall) String() string {
 func (*CoprocessorServiceCall) ProtoMessage() {}
 
 func (x *CoprocessorServiceCall) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[16]
+	mi := &file_Client_proto_msgTypes[17]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1788,7 +1846,7 @@ func (x *CoprocessorServiceCall) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use CoprocessorServiceCall.ProtoReflect.Descriptor instead.
 func (*CoprocessorServiceCall) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{16}
+	return file_Client_proto_rawDescGZIP(), []int{17}
 }
 
 func (x *CoprocessorServiceCall) GetRow() []byte {
@@ -1830,7 +1888,7 @@ type CoprocessorServiceResult struct {
 func (x *CoprocessorServiceResult) Reset() {
 	*x = CoprocessorServiceResult{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[17]
+		mi := &file_Client_proto_msgTypes[18]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1843,7 +1901,7 @@ func (x *CoprocessorServiceResult) String() string {
 func (*CoprocessorServiceResult) ProtoMessage() {}
 
 func (x *CoprocessorServiceResult) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[17]
+	mi := &file_Client_proto_msgTypes[18]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1856,7 +1914,7 @@ func (x *CoprocessorServiceResult) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use CoprocessorServiceResult.ProtoReflect.Descriptor instead.
 func (*CoprocessorServiceResult) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{17}
+	return file_Client_proto_rawDescGZIP(), []int{18}
 }
 
 func (x *CoprocessorServiceResult) GetValue() *NameBytesPair {
@@ -1878,7 +1936,7 @@ type CoprocessorServiceRequest struct {
 func (x *CoprocessorServiceRequest) Reset() {
 	*x = CoprocessorServiceRequest{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[18]
+		mi := &file_Client_proto_msgTypes[19]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1891,7 +1949,7 @@ func (x *CoprocessorServiceRequest) String() string {
 func (*CoprocessorServiceRequest) ProtoMessage() {}
 
 func (x *CoprocessorServiceRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[18]
+	mi := &file_Client_proto_msgTypes[19]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1904,7 +1962,7 @@ func (x *CoprocessorServiceRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use CoprocessorServiceRequest.ProtoReflect.Descriptor instead.
 func (*CoprocessorServiceRequest) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{18}
+	return file_Client_proto_rawDescGZIP(), []int{19}
 }
 
 func (x *CoprocessorServiceRequest) GetRegion() *RegionSpecifier {
@@ -1933,7 +1991,7 @@ type CoprocessorServiceResponse struct {
 func (x *CoprocessorServiceResponse) Reset() {
 	*x = CoprocessorServiceResponse{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[19]
+		mi := &file_Client_proto_msgTypes[20]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1946,7 +2004,7 @@ func (x *CoprocessorServiceResponse) String() string {
 func (*CoprocessorServiceResponse) ProtoMessage() {}
 
 func (x *CoprocessorServiceResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[19]
+	mi := &file_Client_proto_msgTypes[20]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1959,7 +2017,7 @@ func (x *CoprocessorServiceResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use CoprocessorServiceResponse.ProtoReflect.Descriptor instead.
 func (*CoprocessorServiceResponse) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{19}
+	return file_Client_proto_rawDescGZIP(), []int{20}
 }
 
 func (x *CoprocessorServiceResponse) GetRegion() *RegionSpecifier {
@@ -1993,7 +2051,7 @@ type Action struct {
 func (x *Action) Reset() {
 	*x = Action{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[20]
+		mi := &file_Client_proto_msgTypes[21]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -2006,7 +2064,7 @@ func (x *Action) String() string {
 func (*Action) ProtoMessage() {}
 
 func (x *Action) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[20]
+	mi := &file_Client_proto_msgTypes[21]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -2019,7 +2077,7 @@ func (x *Action) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Action.ProtoReflect.Descriptor instead.
 func (*Action) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{20}
+	return file_Client_proto_rawDescGZIP(), []int{21}
 }
 
 func (x *Action) GetIndex() uint32 {
@@ -2066,7 +2124,7 @@ type RegionAction struct {
 func (x *RegionAction) Reset() {
 	*x = RegionAction{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[21]
+		mi := &file_Client_proto_msgTypes[22]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -2079,7 +2137,7 @@ func (x *RegionAction) String() string {
 func (*RegionAction) ProtoMessage() {}
 
 func (x *RegionAction) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[21]
+	mi := &file_Client_proto_msgTypes[22]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -2092,7 +2150,7 @@ func (x *RegionAction) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use RegionAction.ProtoReflect.Descriptor instead.
 func (*RegionAction) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{21}
+	return file_Client_proto_rawDescGZIP(), []int{22}
 }
 
 func (x *RegionAction) GetRegion() *RegionSpecifier {
@@ -2141,7 +2199,7 @@ const (
 func (x *RegionLoadStats) Reset() {
 	*x = RegionLoadStats{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[22]
+		mi := &file_Client_proto_msgTypes[23]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -2154,7 +2212,7 @@ func (x *RegionLoadStats) String() string {
 func (*RegionLoadStats) ProtoMessage() {}
 
 func (x *RegionLoadStats) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[22]
+	mi := &file_Client_proto_msgTypes[23]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -2167,7 +2225,7 @@ func (x *RegionLoadStats) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use RegionLoadStats.ProtoReflect.Descriptor instead.
 func (*RegionLoadStats) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{22}
+	return file_Client_proto_rawDescGZIP(), []int{23}
 }
 
 func (x *RegionLoadStats) GetMemstoreLoad() int32 {
@@ -2203,7 +2261,7 @@ type MultiRegionLoadStats struct {
 func (x *MultiRegionLoadStats) Reset() {
 	*x = MultiRegionLoadStats{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[23]
+		mi := &file_Client_proto_msgTypes[24]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -2216,7 +2274,7 @@ func (x *MultiRegionLoadStats) String() string {
 func (*MultiRegionLoadStats) ProtoMessage() {}
 
 func (x *MultiRegionLoadStats) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[23]
+	mi := &file_Client_proto_msgTypes[24]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -2229,7 +2287,7 @@ func (x *MultiRegionLoadStats) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use MultiRegionLoadStats.ProtoReflect.Descriptor instead.
 func (*MultiRegionLoadStats) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{23}
+	return file_Client_proto_rawDescGZIP(), []int{24}
 }
 
 func (x *MultiRegionLoadStats) GetRegion() []*RegionSpecifier {
@@ -2264,14 +2322,14 @@ type ResultOrException struct {
 	ServiceResult *CoprocessorServiceResult `protobuf:"bytes,4,opt,name=service_result,json=serviceResult" json:"service_result,omitempty"`
 	// current load on the region
 	//
-	// Deprecated: Do not use.
+	// Deprecated: Marked as deprecated in Client.proto.
 	LoadStats *RegionLoadStats `protobuf:"bytes,5,opt,name=loadStats" json:"loadStats,omitempty"`
 }
 
 func (x *ResultOrException) Reset() {
 	*x = ResultOrException{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[24]
+		mi := &file_Client_proto_msgTypes[25]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -2284,7 +2342,7 @@ func (x *ResultOrException) String() string {
 func (*ResultOrException) ProtoMessage() {}
 
 func (x *ResultOrException) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[24]
+	mi := &file_Client_proto_msgTypes[25]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -2297,7 +2355,7 @@ func (x *ResultOrException) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use ResultOrException.ProtoReflect.Descriptor instead.
 func (*ResultOrException) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{24}
+	return file_Client_proto_rawDescGZIP(), []int{25}
 }
 
 func (x *ResultOrException) GetIndex() uint32 {
@@ -2328,7 +2386,7 @@ func (x *ResultOrException) GetServiceResult() *CoprocessorServiceResult {
 	return nil
 }
 
-// Deprecated: Do not use.
+// Deprecated: Marked as deprecated in Client.proto.
 func (x *ResultOrException) GetLoadStats() *RegionLoadStats {
 	if x != nil {
 		return x.LoadStats
@@ -2351,7 +2409,7 @@ type RegionActionResult struct {
 func (x *RegionActionResult) Reset() {
 	*x = RegionActionResult{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[25]
+		mi := &file_Client_proto_msgTypes[26]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -2364,7 +2422,7 @@ func (x *RegionActionResult) String() string {
 func (*RegionActionResult) ProtoMessage() {}
 
 func (x *RegionActionResult) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[25]
+	mi := &file_Client_proto_msgTypes[26]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -2377,7 +2435,7 @@ func (x *RegionActionResult) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use RegionActionResult.ProtoReflect.Descriptor instead.
 func (*RegionActionResult) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{25}
+	return file_Client_proto_rawDescGZIP(), []int{26}
 }
 
 func (x *RegionActionResult) GetResultOrException() []*ResultOrException {
@@ -2414,7 +2472,7 @@ type MultiRequest struct {
 func (x *MultiRequest) Reset() {
 	*x = MultiRequest{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[26]
+		mi := &file_Client_proto_msgTypes[27]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -2427,7 +2485,7 @@ func (x *MultiRequest) String() string {
 func (*MultiRequest) ProtoMessage() {}
 
 func (x *MultiRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[26]
+	mi := &file_Client_proto_msgTypes[27]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -2440,7 +2498,7 @@ func (x *MultiRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use MultiRequest.ProtoReflect.Descriptor instead.
 func (*MultiRequest) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{26}
+	return file_Client_proto_rawDescGZIP(), []int{27}
 }
 
 func (x *MultiRequest) GetRegionAction() []*RegionAction {
@@ -2478,7 +2536,7 @@ type MultiResponse struct {
 func (x *MultiResponse) Reset() {
 	*x = MultiResponse{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[27]
+		mi := &file_Client_proto_msgTypes[28]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -2491,7 +2549,7 @@ func (x *MultiResponse) String() string {
 func (*MultiResponse) ProtoMessage() {}
 
 func (x *MultiResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[27]
+	mi := &file_Client_proto_msgTypes[28]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -2504,7 +2562,7 @@ func (x *MultiResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use MultiResponse.ProtoReflect.Descriptor instead.
 func (*MultiResponse) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{27}
+	return file_Client_proto_rawDescGZIP(), []int{28}
 }
 
 func (x *MultiResponse) GetRegionActionResult() []*RegionActionResult {
@@ -2540,7 +2598,7 @@ type MutationProto_ColumnValue struct {
 func (x *MutationProto_ColumnValue) Reset() {
 	*x = MutationProto_ColumnValue{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[28]
+		mi := &file_Client_proto_msgTypes[29]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -2553,7 +2611,7 @@ func (x *MutationProto_ColumnValue) String() string {
 func (*MutationProto_ColumnValue) ProtoMessage() {}
 
 func (x *MutationProto_ColumnValue) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[28]
+	mi := &file_Client_proto_msgTypes[29]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -2598,7 +2656,7 @@ type MutationProto_ColumnValue_QualifierValue struct {
 func (x *MutationProto_ColumnValue_QualifierValue) Reset() {
 	*x = MutationProto_ColumnValue_QualifierValue{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[29]
+		mi := &file_Client_proto_msgTypes[30]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -2611,7 +2669,7 @@ func (x *MutationProto_ColumnValue_QualifierValue) String() string {
 func (*MutationProto_ColumnValue_QualifierValue) ProtoMessage() {}
 
 func (x *MutationProto_ColumnValue_QualifierValue) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[29]
+	mi := &file_Client_proto_msgTypes[30]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -2674,7 +2732,7 @@ type BulkLoadHFileRequest_FamilyPath struct {
 func (x *BulkLoadHFileRequest_FamilyPath) Reset() {
 	*x = BulkLoadHFileRequest_FamilyPath{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_Client_proto_msgTypes[30]
+		mi := &file_Client_proto_msgTypes[31]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -2687,7 +2745,7 @@ func (x *BulkLoadHFileRequest_FamilyPath) String() string {
 func (*BulkLoadHFileRequest_FamilyPath) ProtoMessage() {}
 
 func (x *BulkLoadHFileRequest_FamilyPath) ProtoReflect() protoreflect.Message {
-	mi := &file_Client_proto_msgTypes[30]
+	mi := &file_Client_proto_msgTypes[31]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -2700,7 +2758,7 @@ func (x *BulkLoadHFileRequest_FamilyPath) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use BulkLoadHFileRequest_FamilyPath.ProtoReflect.Descriptor instead.
 func (*BulkLoadHFileRequest_FamilyPath) Descriptor() ([]byte, []int) {
-	return file_Client_proto_rawDescGZIP(), []int{14, 0}
+	return file_Client_proto_rawDescGZIP(), []int{15, 0}
 }
 
 func (x *BulkLoadHFileRequest_FamilyPath) GetFamily() []byte {
@@ -2956,7 +3014,7 @@ var file_Client_proto_rawDesc = []byte{
 	0x73, 0x18, 0x09, 0x20, 0x01, 0x28, 0x08, 0x52, 0x10, 0x74, 0x72, 0x61, 0x63, 0x6b, 0x53, 0x63,
 	0x61, 0x6e, 0x4d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x73, 0x12, 0x1b, 0x0a, 0x05, 0x72, 0x65, 0x6e,
 	0x65, 0x77, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x08, 0x3a, 0x05, 0x66, 0x61, 0x6c, 0x73, 0x65, 0x52,
-	0x05, 0x72, 0x65, 0x6e, 0x65, 0x77, 0x22, 0xe1, 0x02, 0x0a, 0x0c, 0x53, 0x63, 0x61, 0x6e, 0x52,
+	0x05, 0x72, 0x65, 0x6e, 0x65, 0x77, 0x22, 0x95, 0x03, 0x0a, 0x0c, 0x53, 0x63, 0x61, 0x6e, 0x52,
 	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x28, 0x0a, 0x10, 0x63, 0x65, 0x6c, 0x6c, 0x73,
 	0x5f, 0x70, 0x65, 0x72, 0x5f, 0x72, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x18, 0x01, 0x20, 0x03, 0x28,
 	0x0d, 0x52, 0x0e, 0x63, 0x65, 0x6c, 0x6c, 0x73, 0x50, 0x65, 0x72, 0x52, 0x65, 0x73, 0x75, 0x6c,
@@ -2978,7 +3036,14 @@ var file_Client_proto_rawDesc = []byte{
 	0x73, 0x75, 0x6c, 0x74, 0x73, 0x49, 0x6e, 0x52, 0x65, 0x67, 0x69, 0x6f, 0x6e, 0x12, 0x2b, 0x0a,
 	0x11, 0x68, 0x65, 0x61, 0x72, 0x74, 0x62, 0x65, 0x61, 0x74, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61,
 	0x67, 0x65, 0x18, 0x09, 0x20, 0x01, 0x28, 0x08, 0x52, 0x10, 0x68, 0x65, 0x61, 0x72, 0x74, 0x62,
-	0x65, 0x61, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0xe9, 0x01, 0x0a, 0x14, 0x42,
+	0x65, 0x61, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x32, 0x0a, 0x0c, 0x73, 0x63,
+	0x61, 0x6e, 0x5f, 0x6d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0b,
+	0x32, 0x0f, 0x2e, 0x70, 0x62, 0x2e, 0x53, 0x63, 0x61, 0x6e, 0x4d, 0x65, 0x74, 0x72, 0x69, 0x63,
+	0x73, 0x52, 0x0b, 0x73, 0x63, 0x61, 0x6e, 0x4d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x73, 0x22, 0x3a,
+	0x0a, 0x0b, 0x53, 0x63, 0x61, 0x6e, 0x4d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x73, 0x12, 0x2b, 0x0a,
+	0x07, 0x6d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x11,
+	0x2e, 0x70, 0x62, 0x2e, 0x4e, 0x61, 0x6d, 0x65, 0x49, 0x6e, 0x74, 0x36, 0x34, 0x50, 0x61, 0x69,
+	0x72, 0x52, 0x07, 0x6d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x73, 0x22, 0xe9, 0x01, 0x0a, 0x14, 0x42,
 	0x75, 0x6c, 0x6b, 0x4c, 0x6f, 0x61, 0x64, 0x48, 0x46, 0x69, 0x6c, 0x65, 0x52, 0x65, 0x71, 0x75,
 	0x65, 0x73, 0x74, 0x12, 0x2b, 0x0a, 0x06, 0x72, 0x65, 0x67, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20,
 	0x02, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x70, 0x62, 0x2e, 0x52, 0x65, 0x67, 0x69, 0x6f, 0x6e, 0x53,
@@ -3154,7 +3219,7 @@ func file_Client_proto_rawDescGZIP() []byte {
 }
 
 var file_Client_proto_enumTypes = make([]protoimpl.EnumInfo, 4)
-var file_Client_proto_msgTypes = make([]protoimpl.MessageInfo, 31)
+var file_Client_proto_msgTypes = make([]protoimpl.MessageInfo, 32)
 var file_Client_proto_goTypes = []interface{}{
 	(Consistency)(0),                                 // 0: pb.Consistency
 	(MutationProto_Durability)(0),                    // 1: pb.MutationProto.Durability
@@ -3174,108 +3239,112 @@ var file_Client_proto_goTypes = []interface{}{
 	(*Scan)(nil),                                     // 15: pb.Scan
 	(*ScanRequest)(nil),                              // 16: pb.ScanRequest
 	(*ScanResponse)(nil),                             // 17: pb.ScanResponse
-	(*BulkLoadHFileRequest)(nil),                     // 18: pb.BulkLoadHFileRequest
-	(*BulkLoadHFileResponse)(nil),                    // 19: pb.BulkLoadHFileResponse
-	(*CoprocessorServiceCall)(nil),                   // 20: pb.CoprocessorServiceCall
-	(*CoprocessorServiceResult)(nil),                 // 21: pb.CoprocessorServiceResult
-	(*CoprocessorServiceRequest)(nil),                // 22: pb.CoprocessorServiceRequest
-	(*CoprocessorServiceResponse)(nil),               // 23: pb.CoprocessorServiceResponse
-	(*Action)(nil),                                   // 24: pb.Action
-	(*RegionAction)(nil),                             // 25: pb.RegionAction
-	(*RegionLoadStats)(nil),                          // 26: pb.RegionLoadStats
-	(*MultiRegionLoadStats)(nil),                     // 27: pb.MultiRegionLoadStats
-	(*ResultOrException)(nil),                        // 28: pb.ResultOrException
-	(*RegionActionResult)(nil),                       // 29: pb.RegionActionResult
-	(*MultiRequest)(nil),                             // 30: pb.MultiRequest
-	(*MultiResponse)(nil),                            // 31: pb.MultiResponse
-	(*MutationProto_ColumnValue)(nil),                // 32: pb.MutationProto.ColumnValue
-	(*MutationProto_ColumnValue_QualifierValue)(nil), // 33: pb.MutationProto.ColumnValue.QualifierValue
-	(*BulkLoadHFileRequest_FamilyPath)(nil),          // 34: pb.BulkLoadHFileRequest.FamilyPath
-	(*NameBytesPair)(nil),                            // 35: pb.NameBytesPair
-	(*Filter)(nil),                                   // 36: pb.Filter
-	(*TimeRange)(nil),                                // 37: pb.TimeRange
-	(*ColumnFamilyTimeRange)(nil),                    // 38: pb.ColumnFamilyTimeRange
-	(*Cell)(nil),                                     // 39: pb.Cell
-	(*RegionSpecifier)(nil),                          // 40: pb.RegionSpecifier
-	(CompareType)(0),                                 // 41: pb.CompareType
-	(*Comparator)(nil),                               // 42: pb.Comparator
+	(*ScanMetrics)(nil),                              // 18: pb.ScanMetrics
+	(*BulkLoadHFileRequest)(nil),                     // 19: pb.BulkLoadHFileRequest
+	(*BulkLoadHFileResponse)(nil),                    // 20: pb.BulkLoadHFileResponse
+	(*CoprocessorServiceCall)(nil),                   // 21: pb.CoprocessorServiceCall
+	(*CoprocessorServiceResult)(nil),                 // 22: pb.CoprocessorServiceResult
+	(*CoprocessorServiceRequest)(nil),                // 23: pb.CoprocessorServiceRequest
+	(*CoprocessorServiceResponse)(nil),               // 24: pb.CoprocessorServiceResponse
+	(*Action)(nil),                                   // 25: pb.Action
+	(*RegionAction)(nil),                             // 26: pb.RegionAction
+	(*RegionLoadStats)(nil),                          // 27: pb.RegionLoadStats
+	(*MultiRegionLoadStats)(nil),                     // 28: pb.MultiRegionLoadStats
+	(*ResultOrException)(nil),                        // 29: pb.ResultOrException
+	(*RegionActionResult)(nil),                       // 30: pb.RegionActionResult
+	(*MultiRequest)(nil),                             // 31: pb.MultiRequest
+	(*MultiResponse)(nil),                            // 32: pb.MultiResponse
+	(*MutationProto_ColumnValue)(nil),                // 33: pb.MutationProto.ColumnValue
+	(*MutationProto_ColumnValue_QualifierValue)(nil), // 34: pb.MutationProto.ColumnValue.QualifierValue
+	(*BulkLoadHFileRequest_FamilyPath)(nil),          // 35: pb.BulkLoadHFileRequest.FamilyPath
+	(*NameBytesPair)(nil),                            // 36: pb.NameBytesPair
+	(*Filter)(nil),                                   // 37: pb.Filter
+	(*TimeRange)(nil),                                // 38: pb.TimeRange
+	(*ColumnFamilyTimeRange)(nil),                    // 39: pb.ColumnFamilyTimeRange
+	(*Cell)(nil),                                     // 40: pb.Cell
+	(*RegionSpecifier)(nil),                          // 41: pb.RegionSpecifier
+	(CompareType)(0),                                 // 42: pb.CompareType
+	(*Comparator)(nil),                               // 43: pb.Comparator
+	(*NameInt64Pair)(nil),                            // 44: pb.NameInt64Pair
 }
 var file_Client_proto_depIdxs = []int32{
 	6,  // 0: pb.Get.column:type_name -> pb.Column
-	35, // 1: pb.Get.attribute:type_name -> pb.NameBytesPair
-	36, // 2: pb.Get.filter:type_name -> pb.Filter
-	37, // 3: pb.Get.time_range:type_name -> pb.TimeRange
+	36, // 1: pb.Get.attribute:type_name -> pb.NameBytesPair
+	37, // 2: pb.Get.filter:type_name -> pb.Filter
+	38, // 3: pb.Get.time_range:type_name -> pb.TimeRange
 	0,  // 4: pb.Get.consistency:type_name -> pb.Consistency
-	38, // 5: pb.Get.cf_time_range:type_name -> pb.ColumnFamilyTimeRange
-	39, // 6: pb.Result.cell:type_name -> pb.Cell
-	40, // 7: pb.GetRequest.region:type_name -> pb.RegionSpecifier
+	39, // 5: pb.Get.cf_time_range:type_name -> pb.ColumnFamilyTimeRange
+	40, // 6: pb.Result.cell:type_name -> pb.Cell
+	41, // 7: pb.GetRequest.region:type_name -> pb.RegionSpecifier
 	7,  // 8: pb.GetRequest.get:type_name -> pb.Get
 	8,  // 9: pb.GetResponse.result:type_name -> pb.Result
-	41, // 10: pb.Condition.compare_type:type_name -> pb.CompareType
-	42, // 11: pb.Condition.comparator:type_name -> pb.Comparator
+	42, // 10: pb.Condition.compare_type:type_name -> pb.CompareType
+	43, // 11: pb.Condition.comparator:type_name -> pb.Comparator
 	2,  // 12: pb.MutationProto.mutate_type:type_name -> pb.MutationProto.MutationType
-	32, // 13: pb.MutationProto.column_value:type_name -> pb.MutationProto.ColumnValue
-	35, // 14: pb.MutationProto.attribute:type_name -> pb.NameBytesPair
+	33, // 13: pb.MutationProto.column_value:type_name -> pb.MutationProto.ColumnValue
+	36, // 14: pb.MutationProto.attribute:type_name -> pb.NameBytesPair
 	1,  // 15: pb.MutationProto.durability:type_name -> pb.MutationProto.Durability
-	37, // 16: pb.MutationProto.time_range:type_name -> pb.TimeRange
-	40, // 17: pb.MutateRequest.region:type_name -> pb.RegionSpecifier
+	38, // 16: pb.MutationProto.time_range:type_name -> pb.TimeRange
+	41, // 17: pb.MutateRequest.region:type_name -> pb.RegionSpecifier
 	12, // 18: pb.MutateRequest.mutation:type_name -> pb.MutationProto
 	11, // 19: pb.MutateRequest.condition:type_name -> pb.Condition
 	8,  // 20: pb.MutateResponse.result:type_name -> pb.Result
 	6,  // 21: pb.Scan.column:type_name -> pb.Column
-	35, // 22: pb.Scan.attribute:type_name -> pb.NameBytesPair
-	36, // 23: pb.Scan.filter:type_name -> pb.Filter
-	37, // 24: pb.Scan.time_range:type_name -> pb.TimeRange
+	36, // 22: pb.Scan.attribute:type_name -> pb.NameBytesPair
+	37, // 23: pb.Scan.filter:type_name -> pb.Filter
+	38, // 24: pb.Scan.time_range:type_name -> pb.TimeRange
 	0,  // 25: pb.Scan.consistency:type_name -> pb.Consistency
-	38, // 26: pb.Scan.cf_time_range:type_name -> pb.ColumnFamilyTimeRange
-	40, // 27: pb.ScanRequest.region:type_name -> pb.RegionSpecifier
+	39, // 26: pb.Scan.cf_time_range:type_name -> pb.ColumnFamilyTimeRange
+	41, // 27: pb.ScanRequest.region:type_name -> pb.RegionSpecifier
 	15, // 28: pb.ScanRequest.scan:type_name -> pb.Scan
 	8,  // 29: pb.ScanResponse.results:type_name -> pb.Result
-	40, // 30: pb.BulkLoadHFileRequest.region:type_name -> pb.RegionSpecifier
-	34, // 31: pb.BulkLoadHFileRequest.family_path:type_name -> pb.BulkLoadHFileRequest.FamilyPath
-	35, // 32: pb.CoprocessorServiceResult.value:type_name -> pb.NameBytesPair
-	40, // 33: pb.CoprocessorServiceRequest.region:type_name -> pb.RegionSpecifier
-	20, // 34: pb.CoprocessorServiceRequest.call:type_name -> pb.CoprocessorServiceCall
-	40, // 35: pb.CoprocessorServiceResponse.region:type_name -> pb.RegionSpecifier
-	35, // 36: pb.CoprocessorServiceResponse.value:type_name -> pb.NameBytesPair
-	12, // 37: pb.Action.mutation:type_name -> pb.MutationProto
-	7,  // 38: pb.Action.get:type_name -> pb.Get
-	20, // 39: pb.Action.service_call:type_name -> pb.CoprocessorServiceCall
-	40, // 40: pb.RegionAction.region:type_name -> pb.RegionSpecifier
-	24, // 41: pb.RegionAction.action:type_name -> pb.Action
-	40, // 42: pb.MultiRegionLoadStats.region:type_name -> pb.RegionSpecifier
-	26, // 43: pb.MultiRegionLoadStats.stat:type_name -> pb.RegionLoadStats
-	8,  // 44: pb.ResultOrException.result:type_name -> pb.Result
-	35, // 45: pb.ResultOrException.exception:type_name -> pb.NameBytesPair
-	21, // 46: pb.ResultOrException.service_result:type_name -> pb.CoprocessorServiceResult
-	26, // 47: pb.ResultOrException.loadStats:type_name -> pb.RegionLoadStats
-	28, // 48: pb.RegionActionResult.resultOrException:type_name -> pb.ResultOrException
-	35, // 49: pb.RegionActionResult.exception:type_name -> pb.NameBytesPair
-	25, // 50: pb.MultiRequest.regionAction:type_name -> pb.RegionAction
-	11, // 51: pb.MultiRequest.condition:type_name -> pb.Condition
-	29, // 52: pb.MultiResponse.regionActionResult:type_name -> pb.RegionActionResult
-	27, // 53: pb.MultiResponse.regionStatistics:type_name -> pb.MultiRegionLoadStats
-	33, // 54: pb.MutationProto.ColumnValue.qualifier_value:type_name -> pb.MutationProto.ColumnValue.QualifierValue
-	3,  // 55: pb.MutationProto.ColumnValue.QualifierValue.delete_type:type_name -> pb.MutationProto.DeleteType
-	9,  // 56: pb.ClientService.Get:input_type -> pb.GetRequest
-	13, // 57: pb.ClientService.Mutate:input_type -> pb.MutateRequest
-	16, // 58: pb.ClientService.Scan:input_type -> pb.ScanRequest
-	18, // 59: pb.ClientService.BulkLoadHFile:input_type -> pb.BulkLoadHFileRequest
-	22, // 60: pb.ClientService.ExecService:input_type -> pb.CoprocessorServiceRequest
-	22, // 61: pb.ClientService.ExecRegionServerService:input_type -> pb.CoprocessorServiceRequest
-	30, // 62: pb.ClientService.Multi:input_type -> pb.MultiRequest
-	10, // 63: pb.ClientService.Get:output_type -> pb.GetResponse
-	14, // 64: pb.ClientService.Mutate:output_type -> pb.MutateResponse
-	17, // 65: pb.ClientService.Scan:output_type -> pb.ScanResponse
-	19, // 66: pb.ClientService.BulkLoadHFile:output_type -> pb.BulkLoadHFileResponse
-	23, // 67: pb.ClientService.ExecService:output_type -> pb.CoprocessorServiceResponse
-	23, // 68: pb.ClientService.ExecRegionServerService:output_type -> pb.CoprocessorServiceResponse
-	31, // 69: pb.ClientService.Multi:output_type -> pb.MultiResponse
-	63, // [63:70] is the sub-list for method output_type
-	56, // [56:63] is the sub-list for method input_type
-	56, // [56:56] is the sub-list for extension type_name
-	56, // [56:56] is the sub-list for extension extendee
-	0,  // [0:56] is the sub-list for field type_name
+	18, // 30: pb.ScanResponse.scan_metrics:type_name -> pb.ScanMetrics
+	44, // 31: pb.ScanMetrics.metrics:type_name -> pb.NameInt64Pair
+	41, // 32: pb.BulkLoadHFileRequest.region:type_name -> pb.RegionSpecifier
+	35, // 33: pb.BulkLoadHFileRequest.family_path:type_name -> pb.BulkLoadHFileRequest.FamilyPath
+	36, // 34: pb.CoprocessorServiceResult.value:type_name -> pb.NameBytesPair
+	41, // 35: pb.CoprocessorServiceRequest.region:type_name -> pb.RegionSpecifier
+	21, // 36: pb.CoprocessorServiceRequest.call:type_name -> pb.CoprocessorServiceCall
+	41, // 37: pb.CoprocessorServiceResponse.region:type_name -> pb.RegionSpecifier
+	36, // 38: pb.CoprocessorServiceResponse.value:type_name -> pb.NameBytesPair
+	12, // 39: pb.Action.mutation:type_name -> pb.MutationProto
+	7,  // 40: pb.Action.get:type_name -> pb.Get
+	21, // 41: pb.Action.service_call:type_name -> pb.CoprocessorServiceCall
+	41, // 42: pb.RegionAction.region:type_name -> pb.RegionSpecifier
+	25, // 43: pb.RegionAction.action:type_name -> pb.Action
+	41, // 44: pb.MultiRegionLoadStats.region:type_name -> pb.RegionSpecifier
+	27, // 45: pb.MultiRegionLoadStats.stat:type_name -> pb.RegionLoadStats
+	8,  // 46: pb.ResultOrException.result:type_name -> pb.Result
+	36, // 47: pb.ResultOrException.exception:type_name -> pb.NameBytesPair
+	22, // 48: pb.ResultOrException.service_result:type_name -> pb.CoprocessorServiceResult
+	27, // 49: pb.ResultOrException.loadStats:type_name -> pb.RegionLoadStats
+	29, // 50: pb.RegionActionResult.resultOrException:type_name -> pb.ResultOrException
+	36, // 51: pb.RegionActionResult.exception:type_name -> pb.NameBytesPair
+	26, // 52: pb.MultiRequest.regionAction:type_name -> pb.RegionAction
+	11, // 53: pb.MultiRequest.condition:type_name -> pb.Condition
+	30, // 54: pb.MultiResponse.regionActionResult:type_name -> pb.RegionActionResult
+	28, // 55: pb.MultiResponse.regionStatistics:type_name -> pb.MultiRegionLoadStats
+	34, // 56: pb.MutationProto.ColumnValue.qualifier_value:type_name -> pb.MutationProto.ColumnValue.QualifierValue
+	3,  // 57: pb.MutationProto.ColumnValue.QualifierValue.delete_type:type_name -> pb.MutationProto.DeleteType
+	9,  // 58: pb.ClientService.Get:input_type -> pb.GetRequest
+	13, // 59: pb.ClientService.Mutate:input_type -> pb.MutateRequest
+	16, // 60: pb.ClientService.Scan:input_type -> pb.ScanRequest
+	19, // 61: pb.ClientService.BulkLoadHFile:input_type -> pb.BulkLoadHFileRequest
+	23, // 62: pb.ClientService.ExecService:input_type -> pb.CoprocessorServiceRequest
+	23, // 63: pb.ClientService.ExecRegionServerService:input_type -> pb.CoprocessorServiceRequest
+	31, // 64: pb.ClientService.Multi:input_type -> pb.MultiRequest
+	10, // 65: pb.ClientService.Get:output_type -> pb.GetResponse
+	14, // 66: pb.ClientService.Mutate:output_type -> pb.MutateResponse
+	17, // 67: pb.ClientService.Scan:output_type -> pb.ScanResponse
+	20, // 68: pb.ClientService.BulkLoadHFile:output_type -> pb.BulkLoadHFileResponse
+	24, // 69: pb.ClientService.ExecService:output_type -> pb.CoprocessorServiceResponse
+	24, // 70: pb.ClientService.ExecRegionServerService:output_type -> pb.CoprocessorServiceResponse
+	32, // 71: pb.ClientService.Multi:output_type -> pb.MultiResponse
+	65, // [65:72] is the sub-list for method output_type
+	58, // [58:65] is the sub-list for method input_type
+	58, // [58:58] is the sub-list for extension type_name
+	58, // [58:58] is the sub-list for extension extendee
+	0,  // [0:58] is the sub-list for field type_name
 }
 
 func init() { file_Client_proto_init() }
@@ -3457,7 +3526,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[14].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*BulkLoadHFileRequest); i {
+			switch v := v.(*ScanMetrics); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3469,7 +3538,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[15].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*BulkLoadHFileResponse); i {
+			switch v := v.(*BulkLoadHFileRequest); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3481,7 +3550,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[16].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*CoprocessorServiceCall); i {
+			switch v := v.(*BulkLoadHFileResponse); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3493,7 +3562,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[17].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*CoprocessorServiceResult); i {
+			switch v := v.(*CoprocessorServiceCall); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3505,7 +3574,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[18].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*CoprocessorServiceRequest); i {
+			switch v := v.(*CoprocessorServiceResult); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3517,7 +3586,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[19].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*CoprocessorServiceResponse); i {
+			switch v := v.(*CoprocessorServiceRequest); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3529,7 +3598,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[20].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Action); i {
+			switch v := v.(*CoprocessorServiceResponse); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3541,7 +3610,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[21].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*RegionAction); i {
+			switch v := v.(*Action); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3553,7 +3622,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[22].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*RegionLoadStats); i {
+			switch v := v.(*RegionAction); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3565,7 +3634,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[23].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*MultiRegionLoadStats); i {
+			switch v := v.(*RegionLoadStats); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3577,7 +3646,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[24].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*ResultOrException); i {
+			switch v := v.(*MultiRegionLoadStats); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3589,7 +3658,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[25].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*RegionActionResult); i {
+			switch v := v.(*ResultOrException); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3601,7 +3670,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[26].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*MultiRequest); i {
+			switch v := v.(*RegionActionResult); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3613,7 +3682,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[27].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*MultiResponse); i {
+			switch v := v.(*MultiRequest); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3625,7 +3694,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[28].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*MutationProto_ColumnValue); i {
+			switch v := v.(*MultiResponse); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3637,7 +3706,7 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[29].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*MutationProto_ColumnValue_QualifierValue); i {
+			switch v := v.(*MutationProto_ColumnValue); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -3649,6 +3718,18 @@ func file_Client_proto_init() {
 			}
 		}
 		file_Client_proto_msgTypes[30].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*MutationProto_ColumnValue_QualifierValue); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_Client_proto_msgTypes[31].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*BulkLoadHFileRequest_FamilyPath); i {
 			case 0:
 				return &v.state
@@ -3667,7 +3748,7 @@ func file_Client_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_Client_proto_rawDesc,
 			NumEnums:      4,
-			NumMessages:   31,
+			NumMessages:   32,
 			NumExtensions: 0,
 			NumServices:   1,
 		},
diff --git a/pb/Client.proto b/pb/Client.proto
index e814111d..58821f0f 100644
--- a/pb/Client.proto
+++ b/pb/Client.proto
@@ -108,7 +108,7 @@ message Result {
   //  in the query.
   optional bool exists = 3;
 
-  // Whether or not the results are coming from possibly stale data 
+  // Whether or not the results are coming from possibly stale data
   optional bool stale = 4 [default = false];
 
   // Whether or not the entire result could be returned. Results will be split when
@@ -323,17 +323,21 @@ message ScanResponse {
   // reasons such as the size in bytes or quantity of results accumulated. This field
   // will true when more results exist in the current region.
   optional bool more_results_in_region = 8;
-  
+
   // This field is filled in if the server is sending back a heartbeat message.
   // Heartbeat messages are sent back to the client to prevent the scanner from
   // timing out. Seeing a heartbeat message communicates to the Client that the
   // server would have continued to scan had the time limit not been reached.
   optional bool heartbeat_message = 9;
-  
+
   // This field is filled in if the client has requested that scan metrics be tracked.
-  // The metrics tracked here are sent back to the client to be tracked together with 
+  // The metrics tracked here are sent back to the client to be tracked together with
   // the existing client side metrics.
-  //optional ScanMetrics scan_metrics = 10; // not in gohbase
+  optional ScanMetrics scan_metrics = 10;
+}
+
+message ScanMetrics {
+    repeated NameInt64Pair metrics = 1;
 }
 
 /**
@@ -476,7 +480,7 @@ service ClientService {
 
   rpc ExecService(CoprocessorServiceRequest)
     returns(CoprocessorServiceResponse);
-    
+
   rpc ExecRegionServerService(CoprocessorServiceRequest)
     returns(CoprocessorServiceResponse);
 
diff --git a/pb/ClusterId.pb.go b/pb/ClusterId.pb.go
index 91ad881a..fca6af83 100644
--- a/pb/ClusterId.pb.go
+++ b/pb/ClusterId.pb.go
@@ -19,8 +19,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: ClusterId.proto
 
 package pb
diff --git a/pb/ClusterStatus.pb.go b/pb/ClusterStatus.pb.go
index 3b76d085..5db6f2f6 100644
--- a/pb/ClusterStatus.pb.go
+++ b/pb/ClusterStatus.pb.go
@@ -19,8 +19,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: ClusterStatus.proto
 
 package pb
diff --git a/pb/Comparator.pb.go b/pb/Comparator.pb.go
index e2f50fa7..e0200184 100644
--- a/pb/Comparator.pb.go
+++ b/pb/Comparator.pb.go
@@ -19,8 +19,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: Comparator.proto
 
 package pb
diff --git a/pb/ErrorHandling.pb.go b/pb/ErrorHandling.pb.go
index 821bfd6c..6e64f9c4 100644
--- a/pb/ErrorHandling.pb.go
+++ b/pb/ErrorHandling.pb.go
@@ -19,8 +19,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: ErrorHandling.proto
 
 package pb
diff --git a/pb/FS.pb.go b/pb/FS.pb.go
index d81c294e..b65aa586 100644
--- a/pb/FS.pb.go
+++ b/pb/FS.pb.go
@@ -19,8 +19,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: FS.proto
 
 package pb
diff --git a/pb/Filter.pb.go b/pb/Filter.pb.go
index 77edf5eb..352b2b4e 100644
--- a/pb/Filter.pb.go
+++ b/pb/Filter.pb.go
@@ -19,8 +19,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: Filter.proto
 
 package pb
diff --git a/pb/HBase.pb.go b/pb/HBase.pb.go
index c886eae1..c491f655 100644
--- a/pb/HBase.pb.go
+++ b/pb/HBase.pb.go
@@ -19,8 +19,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: HBase.proto
 
 package pb
diff --git a/pb/Master.pb.go b/pb/Master.pb.go
index b881d776..c47cb1b7 100644
--- a/pb/Master.pb.go
+++ b/pb/Master.pb.go
@@ -20,8 +20,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: Master.proto
 
 package pb
diff --git a/pb/Procedure.pb.go b/pb/Procedure.pb.go
index 8bf80e39..91d8e539 100644
--- a/pb/Procedure.pb.go
+++ b/pb/Procedure.pb.go
@@ -17,8 +17,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: Procedure.proto
 
 package pb
diff --git a/pb/Quota.pb.go b/pb/Quota.pb.go
index 2574e8cc..8ffbc432 100644
--- a/pb/Quota.pb.go
+++ b/pb/Quota.pb.go
@@ -17,8 +17,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: Quota.proto
 
 package pb
diff --git a/pb/RPC.pb.go b/pb/RPC.pb.go
index 51e16e8a..b85bf836 100644
--- a/pb/RPC.pb.go
+++ b/pb/RPC.pb.go
@@ -17,8 +17,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: RPC.proto
 
 package pb
diff --git a/pb/Tracing.pb.go b/pb/Tracing.pb.go
index cd1adea4..d701429d 100644
--- a/pb/Tracing.pb.go
+++ b/pb/Tracing.pb.go
@@ -17,8 +17,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: Tracing.proto
 
 package pb
@@ -42,9 +42,9 @@ type RPCTInfo struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	// Deprecated: Do not use.
+	// Deprecated: Marked as deprecated in Tracing.proto.
 	TraceId *int64 `protobuf:"varint,1,opt,name=trace_id,json=traceId" json:"trace_id,omitempty"`
-	// Deprecated: Do not use.
+	// Deprecated: Marked as deprecated in Tracing.proto.
 	ParentId *int64            `protobuf:"varint,2,opt,name=parent_id,json=parentId" json:"parent_id,omitempty"`
 	Headers  map[string]string `protobuf:"bytes,3,rep,name=headers" json:"headers,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
 }
@@ -81,7 +81,7 @@ func (*RPCTInfo) Descriptor() ([]byte, []int) {
 	return file_Tracing_proto_rawDescGZIP(), []int{0}
 }
 
-// Deprecated: Do not use.
+// Deprecated: Marked as deprecated in Tracing.proto.
 func (x *RPCTInfo) GetTraceId() int64 {
 	if x != nil && x.TraceId != nil {
 		return *x.TraceId
@@ -89,7 +89,7 @@ func (x *RPCTInfo) GetTraceId() int64 {
 	return 0
 }
 
-// Deprecated: Do not use.
+// Deprecated: Marked as deprecated in Tracing.proto.
 func (x *RPCTInfo) GetParentId() int64 {
 	if x != nil && x.ParentId != nil {
 		return *x.ParentId
diff --git a/pb/ZooKeeper.pb.go b/pb/ZooKeeper.pb.go
index f4b90b62..440c6fa8 100644
--- a/pb/ZooKeeper.pb.go
+++ b/pb/ZooKeeper.pb.go
@@ -20,8 +20,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.27.1
-// 	protoc        v3.21.5
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
 // source: ZooKeeper.proto
 
 package pb
diff --git a/scanner.go b/scanner.go
index 9860ff78..83f0b3fc 100644
--- a/scanner.go
+++ b/scanner.go
@@ -18,7 +18,12 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
-const noScannerID = math.MaxUint64
+const (
+	noScannerID = math.MaxUint64
+
+	rowsScanned  = "ROWS_SCANNED"
+	rowsFiltered = "ROWS_FILTERED"
+)
 
 // rowPadding used to pad the row key when constructing a row before
 var rowPadding = []byte{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff}
@@ -30,15 +35,22 @@ type scanner struct {
 	// curRegionScannerID is the id of scanner on current region
 	curRegionScannerID uint64
 	// startRow is the start row in the current region
-	startRow []byte
-	results  []*pb.Result
-	closed   bool
+	startRow    []byte
+	results     []*pb.Result
+	closed      bool
+	scanMetrics map[string]int64
 }
 
 func (s *scanner) fetch() ([]*pb.Result, error) {
 	// keep looping until we have error, some non-empty result or until close
 	for {
 		resp, region, err := s.request()
+		if s.rpc.TrackScanMetrics() && resp.ScanMetrics != nil {
+			metrics := resp.ScanMetrics.GetMetrics()
+			for _, m := range metrics {
+				s.scanMetrics[m.GetName()] += m.GetValue()
+			}
+		}
 		if err != nil {
 			s.Close()
 			return nil, err
@@ -60,12 +72,16 @@ func (s *scanner) fetch() ([]*pb.Result, error) {
 
 func (s *scanner) peek() (*pb.Result, error) {
 	if len(s.results) == 0 {
+		var (
+			err error
+			rs  []*pb.Result
+		)
 		if s.closed {
 			// done scanning
 			return nil, io.EOF
 		}
 
-		rs, err := s.fetch()
+		rs, err = s.fetch()
 		if err != nil {
 			return nil, err
 		}
@@ -111,11 +127,16 @@ func (s *scanner) coalesce(result, partial *pb.Result) (*pb.Result, bool) {
 }
 
 func newScanner(c RPCClient, rpc *hrpc.Scan) *scanner {
+	var sm map[string]int64
+	if rpc.TrackScanMetrics() {
+		sm = make(map[string]int64)
+	}
 	return &scanner{
 		RPCClient:          c,
 		rpc:                rpc,
 		startRow:           rpc.StartRow(),
 		curRegionScannerID: noScannerID,
+		scanMetrics:        sm,
 	}
 }
 
@@ -123,7 +144,8 @@ func toLocalResult(r *pb.Result) *hrpc.Result {
 	if r == nil {
 		return nil
 	}
-	return hrpc.ToLocalResult(r)
+	res := hrpc.ToLocalResult(r)
+	return res
 }
 
 func (s *scanner) Next() (*hrpc.Result, error) {
@@ -141,7 +163,7 @@ func (s *scanner) Next() (*hrpc.Result, error) {
 
 	if s.rpc.AllowPartialResults() {
 		// if client handles partials, just return it
-		result, err := s.peek()
+		result, err = s.peek()
 		if err != nil {
 			return nil, err
 		}
@@ -259,6 +281,14 @@ func (s *scanner) Close() error {
 	return nil
 }
 
+// GetScanMetrics returns the scan metrics for the scanner.
+// The scan metrics are non-nil only if the Scan has TrackScanMetrics() enabled.
+// GetScanMetrics should only be called after the scanner has been closed with an io.EOF
+// (there are no more rows left to be returned by calls to Next()).
+func (s *scanner) GetScanMetrics() map[string]int64 {
+	return s.scanMetrics
+}
+
 // isDone check if this scanner is done fetching new results
 func (s *scanner) isDone(resp *pb.ScanResponse, region hrpc.RegionInfo) bool {
 	if resp.MoreResults != nil && !*resp.MoreResults {
diff --git a/scanner_test.go b/scanner_test.go
index b4e03e62..e5f8f984 100644
--- a/scanner_test.go
+++ b/scanner_test.go
@@ -9,7 +9,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
+	"reflect"
+	"sync"
+	"testing"
 
+	"github.com/tsuna/gohbase/filter"
 	"github.com/tsuna/gohbase/hrpc"
 	"github.com/tsuna/gohbase/pb"
 	"github.com/tsuna/gohbase/region"
@@ -17,11 +22,6 @@ import (
 	"github.com/tsuna/gohbase/test/mock"
 	"go.uber.org/mock/gomock"
 	"google.golang.org/protobuf/proto"
-
-	"io"
-	"reflect"
-	"sync"
-	"testing"
 )
 
 func cp(i uint64) *uint64 {
@@ -259,6 +259,215 @@ func TestAllowPartialResults(t *testing.T) {
 	testPartialResults(t, scan, expected)
 }
 
+func TestScanMetrics(t *testing.T) {
+	scanned, filtered := rowsScanned, rowsFiltered
+	i0, i1, i2, i4 := int64(0), int64(1), int64(2), int64(4)
+	tcases := []struct {
+		description          string
+		trackScanMetrics     func(call hrpc.Call) error
+		filter               func(call hrpc.Call) error
+		results              []*pb.Result
+		scanMetrics          *pb.ScanMetrics
+		expectedResults      []*hrpc.Result
+		expectedRowsScanned  int64
+		expectedRowsFiltered int64
+	}{
+		{
+			description: "ScanMetrics not enabled",
+			results: []*pb.Result{
+				{Cell: cells[:3]},
+			},
+			scanMetrics:     nil,
+			expectedResults: []*hrpc.Result{hrpc.ToLocalResult(&pb.Result{Cell: cells[:3]})},
+		},
+		{
+			description:          "Empty results",
+			trackScanMetrics:     hrpc.TrackScanMetrics(),
+			results:              nil,
+			scanMetrics:          nil,
+			expectedResults:      nil,
+			expectedRowsScanned:  0,
+			expectedRowsFiltered: 0,
+		},
+		{
+			description:      "ScanMetrics: 1 row scanned",
+			trackScanMetrics: hrpc.TrackScanMetrics(),
+			results: []*pb.Result{
+				{Cell: cells[:3]},
+			},
+			scanMetrics: &pb.ScanMetrics{
+				Metrics: []*pb.NameInt64Pair{
+					{
+						Name:  &scanned,
+						Value: &i1,
+					},
+					{
+						Name:  &filtered,
+						Value: &i0,
+					},
+				},
+			},
+			expectedResults:      []*hrpc.Result{toLocalResult(&pb.Result{Cell: cells[:3]})},
+			expectedRowsScanned:  1,
+			expectedRowsFiltered: 0,
+		},
+		{
+			description:      "ScanMetrics: 2 rows scanned",
+			trackScanMetrics: hrpc.TrackScanMetrics(),
+			results: []*pb.Result{
+				{Cell: cells[:5]},
+			},
+			scanMetrics: &pb.ScanMetrics{
+				Metrics: []*pb.NameInt64Pair{
+					{
+						Name:  &scanned,
+						Value: &i2,
+					},
+					{
+						Name:  &filtered,
+						Value: &i0,
+					},
+				},
+			},
+			expectedResults:      []*hrpc.Result{toLocalResult(&pb.Result{Cell: cells[:5]})},
+			expectedRowsScanned:  2,
+			expectedRowsFiltered: 0,
+		},
+		{
+			description:      "ScanMetrics: 4 rows scanned, 2 row filtered",
+			trackScanMetrics: hrpc.TrackScanMetrics(),
+			filter:           hrpc.Filters(filter.NewPrefixFilter([]byte("b"))),
+			results: []*pb.Result{
+				{Cell: cells},
+			},
+			scanMetrics: &pb.ScanMetrics{
+				Metrics: []*pb.NameInt64Pair{
+					{
+						Name:  &scanned,
+						Value: &i4,
+					},
+					{
+						Name:  &filtered,
+						Value: &i2,
+					},
+				},
+			},
+			expectedResults:      []*hrpc.Result{toLocalResult(&pb.Result{Cell: cells})},
+			expectedRowsScanned:  4,
+			expectedRowsFiltered: 2,
+		},
+		{
+			description:      "ScanMetrics: 2 rows scanned, 1 row filtered",
+			trackScanMetrics: hrpc.TrackScanMetrics(),
+			filter:           hrpc.Filters(filter.NewPrefixFilter([]byte("a"))),
+			results: []*pb.Result{
+				{Cell: cells[:5]},
+			},
+			scanMetrics: &pb.ScanMetrics{
+				Metrics: []*pb.NameInt64Pair{
+					{
+						Name:  &scanned,
+						Value: &i2,
+					},
+					{
+						Name:  &filtered,
+						Value: &i1,
+					},
+				},
+			},
+			expectedResults:      []*hrpc.Result{toLocalResult(&pb.Result{Cell: cells[:5]})},
+			expectedRowsScanned:  2,
+			expectedRowsFiltered: 1,
+		},
+		{
+			description:      "ScanMetrics: 0 rows scanned, 1 row filtered",
+			trackScanMetrics: hrpc.TrackScanMetrics(),
+			filter:           hrpc.Filters(filter.NewPrefixFilter([]byte("a"))),
+			results: []*pb.Result{
+				{Cell: cells[:3]},
+			},
+			scanMetrics: &pb.ScanMetrics{
+				Metrics: []*pb.NameInt64Pair{
+					{
+						Name:  &scanned,
+						Value: &i0,
+					},
+					{
+						Name:  &filtered,
+						Value: &i1,
+					},
+				},
+			},
+			expectedResults:      []*hrpc.Result{toLocalResult(&pb.Result{Cell: cells[:3]})},
+			expectedRowsScanned:  0,
+			expectedRowsFiltered: 1,
+		},
+	}
+
+	ctrl := test.NewController(t)
+	defer ctrl.Finish()
+	c := mock.NewMockRPCClient(ctrl)
+
+	for _, tcase := range tcases {
+		t.Run(tcase.description, func(t *testing.T) {
+			ctx := context.Background()
+			var scan *hrpc.Scan
+			var err error
+			if tcase.trackScanMetrics != nil && tcase.filter != nil {
+				scan, err = hrpc.NewScan(ctx, table, tcase.trackScanMetrics, tcase.filter)
+			} else if tcase.trackScanMetrics != nil {
+				scan, err = hrpc.NewScan(ctx, table, tcase.trackScanMetrics)
+			} else {
+				scan, err = hrpc.NewScan(ctx, table)
+			}
+
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			sc := newScanner(c, scan)
+
+			c.EXPECT().SendRPC(&scanMatcher{scan: scan}).Return(&pb.ScanResponse{
+				Results:     tcase.results,
+				ScanMetrics: tcase.scanMetrics,
+			}, nil).Times(1)
+
+			var res []*hrpc.Result
+			for {
+				var r *hrpc.Result
+				r, err = sc.Next()
+				if err == io.EOF {
+					break
+				}
+				if err != nil {
+					t.Fatal(err)
+				}
+				res = append(res, r)
+			}
+
+			actualMetrics := sc.GetScanMetrics()
+
+			if tcase.trackScanMetrics == nil && actualMetrics != nil {
+				t.Fatalf("Got non-nil scan metrics when not enabled: %v", actualMetrics)
+			}
+
+			if tcase.expectedRowsScanned != actualMetrics[rowsScanned] {
+				t.Errorf("Did not get expected rows scanned - expected: %d, actual %d",
+					tcase.expectedRowsScanned, actualMetrics[rowsScanned])
+			}
+
+			if tcase.expectedRowsFiltered != actualMetrics[rowsFiltered] {
+				t.Errorf("Did not get expected rows filtered - expected: %d, actual %d",
+					tcase.expectedRowsFiltered, actualMetrics[rowsFiltered])
+			}
+
+			if !reflect.DeepEqual(tcase.expectedResults, res) {
+				t.Fatalf("expected: %+v\ngot: %+v", tcase.expectedResults, res)
+			}
+		})
+	}
+}
+
 func TestErrorScanFromID(t *testing.T) {
 	scan, err := hrpc.NewScan(context.Background(), table)
 	if err != nil {