[Feature]Sharding support 1 (#18533)

1.  Finish collecting tombstones from remote CN.
2. Handle sharding reader

Todo:
- [ ] Handle order by
- [ ] Improve the performance for sharding.

Approved by: @daviszhen, @badboynt1, @zhangxu19830126, @m-schen, @reusee, @ouyuanning, @qingxinhome, @aunjgr, @heni02, @aressu1985, @iamlinjunhong, @XuPeng-SH

pkg/cnservice/server.go

@@ -25,6 +25,8 @@ import (
 
 	"github.com/fagongzi/goetty/v2"
 	"github.com/google/uuid"
+	"go.uber.org/zap"
+
 	"github.com/matrixorigin/matrixone/pkg/bootstrap"
 	"github.com/matrixorigin/matrixone/pkg/catalog"
 	"github.com/matrixorigin/matrixone/pkg/clusterservice"
@@ -66,7 +68,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/disttae"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/blockio"
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
-	"go.uber.org/zap"
 )
 
 func NewService(
@@ -764,10 +765,13 @@ func (s *service) initShardService() {
 			shardservice.ReadApproxObjectsNum:         disttae.HandleShardingReadApproxObjectsNum,
 			shardservice.ReadRanges:                   disttae.HandleShardingReadRanges,
 			shardservice.ReadGetColumMetadataScanInfo: disttae.HandleShardingReadGetColumMetadataScanInfo,
-			shardservice.ReadReader:                   disttae.HandleShardingReadReader,
+			shardservice.ReadBuildReader:              disttae.HandleShardingReadBuildReader,
 			shardservice.ReadPrimaryKeysMayBeModified: disttae.HandleShardingReadPrimaryKeysMayBeModified,
 			shardservice.ReadMergeObjects:             disttae.HandleShardingReadMergeObjects,
 			shardservice.ReadVisibleObjectStats:       disttae.HandleShardingReadVisibleObjectStats,
+			shardservice.ReadClose:                    disttae.HandleShardingReadClose,
+			shardservice.ReadNext:                     disttae.HandleShardingReadNext,
+			shardservice.ReadCollectTombstones:        disttae.HandleShardingReadCollectTombstones,
 		},
 		s.storeEngine,
 	)

pkg/common/morpc/buffer_pool.go

@@ -16,6 +16,7 @@ package morpc
 
 import (
 	"github.com/fagongzi/goetty/v2/buf"
+
 	"github.com/matrixorigin/matrixone/pkg/common/reuse"
 )
 
@@ -91,5 +92,8 @@ func (b *Buffer) Mark() {
 }
 
 func (b *Buffer) GetMarkedData() []byte {
+	if b.markIdx == b.buf.GetWriteIndex() {
+		return nil
+	}
 	return b.buf.RawSlice(b.markIdx, b.buf.GetWriteIndex())
 }