feature: ShardingSelector GetMulti 使用 merge

.CHANGELOG.md

@@ -23,10 +23,11 @@
 - [eorm: 分库分表: 范围查询支持](https://github.com/ecodeclub/eorm/pull/178)
 - [eorm: 分库分表: 结果集处理--聚合函数(不含GroupBy子句)](https://github.com/ecodeclub/eorm/pull/187)
 - [eorm: 修复单条查询时连接泄露问题](https://github.com/ecodeclub/eorm/pull/188)
-- [eorm: 分库分表: 结果集处理--聚合函数(含GroupBy子句)](https://github.com/ecodeclub/eorm/pull/193)
 - [eorm: 分库分表: NOT 支持](https://github.com/ecodeclub/eorm/pull/191)
+- [eorm: 分库分表: 结果集处理--聚合函数(含GroupBy子句)](https://github.com/ecodeclub/eorm/pull/193)
 - [eorm: 分库分表: Merger NullAble类型数据的支持(sortMerger)](https://github.com/ecodeclub/eorm/pull/195)
 - [eorm: 补充 NULL 语义和基本类型之间转化的测试用例](https://github.com/ecodeclub/eorm/pull/198)
+- [eorm: 分库分表: ShardingSelector GetMulti 使用 merge](https://github.com/ecodeclub/eorm/pull/199)
 
 ## v0.0.1:
 - [Init Project](https://github.com/ecodeclub/eorm/pull/1)

internal/integration/sharding_select_test.go

@@ -135,6 +135,26 @@ func (s *ShardingSelectTestSuite) TestSardingSelectorGetMulti() {
 		wantErr error
 		wantRes []*test.OrderDetail
 	}{
+		{
+			name: "found tab eq not data",
+			s: func() *eorm.ShardingSelector[test.OrderDetail] {
+				builder := eorm.NewShardingSelector[test.OrderDetail](s.shardingDB).
+					Where(eorm.C("OrderId").EQ(500))
+				return builder
+			}(),
+			wantRes: []*test.OrderDetail{},
+		},
+		{
+			name: "found tab eq",
+			s: func() *eorm.ShardingSelector[test.OrderDetail] {
+				builder := eorm.NewShardingSelector[test.OrderDetail](s.shardingDB).
+					Where(eorm.C("OrderId").EQ(123))
+				return builder
+			}(),
+			wantRes: []*test.OrderDetail{
+				{OrderId: 123, ItemId: 10, UsingCol1: "LeBron", UsingCol2: "James"},
+			},
+		},
 		{
 			name: "found tab or",
 			s: func() *eorm.ShardingSelector[test.OrderDetail] {

internal/merger/aggregatemerger/merger.go

@@ -20,9 +20,10 @@ import (
 	"sync"
 	_ "unsafe"
 
+	"github.com/ecodeclub/eorm/internal/merger"
+
 	"github.com/ecodeclub/eorm/internal/merger/utils"
 
-	"github.com/ecodeclub/eorm/internal/merger"
 	"github.com/ecodeclub/eorm/internal/merger/aggregatemerger/aggregator"
 	"github.com/ecodeclub/eorm/internal/merger/internal/errs"
 	"go.uber.org/multierr"

internal/merger/batchmerger/merger.go

@@ -19,9 +19,10 @@ import (
 	"database/sql"
 	"sync"
 
+	"github.com/ecodeclub/eorm/internal/merger"
+
 	"go.uber.org/multierr"
 
-	"github.com/ecodeclub/eorm/internal/merger"
 	"github.com/ecodeclub/eorm/internal/merger/internal/errs"
 )
 
@@ -108,6 +109,7 @@ func (r *Rows) Next() bool {
 	return canNext
 
 }
+
 func (r *Rows) nextRows() (bool, error) {
 	row := r.rowsList[r.cnt]
 	if row.Next() {

internal/merger/pagedmerger/merger.go

@@ -42,16 +42,16 @@ func NewMerger(m merger.Merger, offset int, limit int) (*Merger, error) {
 }
 
 func (m *Merger) Merge(ctx context.Context, results []*sql.Rows) (merger.Rows, error) {
-	rows, err := m.m.Merge(ctx, results)
+	rs, err := m.m.Merge(ctx, results)
 	if err != nil {
 		return nil, err
 	}
-	err = m.nextOffset(ctx, rows)
+	err = m.nextOffset(ctx, rs)
 	if err != nil {
 		return nil, err
 	}
 	return &Rows{
-		rows:  rows,
+		rows:  rs,
 		mu:    &sync.RWMutex{},
 		limit: m.limit,
 	}, nil

internal/merger/sortmerger/merger.go

@@ -21,11 +21,12 @@ import (
 	"reflect"
 	"sync"
 
+	"github.com/ecodeclub/eorm/internal/merger"
+
 	"github.com/ecodeclub/eorm/internal/merger/utils"
 
 	"go.uber.org/multierr"
 
-	"github.com/ecodeclub/eorm/internal/merger"
 	"github.com/ecodeclub/eorm/internal/merger/internal/errs"
 )
 
@@ -130,25 +131,25 @@ func (m *Merger) Merge(ctx context.Context, results []*sql.Rows) (merger.Rows, e
 }
 
 func (m *Merger) initRows(results []*sql.Rows) (*Rows, error) {
-	rows := &Rows{
+	rs := &Rows{
 		rowsList:    results,
 		sortColumns: m.sortColumns,
 		mu:          &sync.RWMutex{},
 		columns:     m.cols,
 	}
 	h := &Heap{
-		h:           make([]*node, 0, len(rows.rowsList)),
-		sortColumns: rows.sortColumns,
+		h:           make([]*node, 0, len(rs.rowsList)),
+		sortColumns: rs.sortColumns,
 	}
-	rows.hp = h
-	for i := 0; i < len(rows.rowsList); i++ {
-		err := rows.nextRows(rows.rowsList[i], i)
+	rs.hp = h
+	for i := 0; i < len(rs.rowsList); i++ {
+		err := rs.nextRows(rs.rowsList[i], i)
 		if err != nil {
-			_ = rows.Close()
+			_ = rs.Close()
 			return nil, err
 		}
 	}
-	return rows, nil
+	return rs, nil
 }
 
 func (m *Merger) checkColumns(rows *sql.Rows) error {

internal/merger/type.go

@@ -17,6 +17,8 @@ package merger
 import (
 	"context"
 	"database/sql"
+
+	"github.com/ecodeclub/eorm/internal/rows"
 )
 
 // Merger 将sql.Rows列表里的元素合并，返回一个类似sql.Rows的迭代器
@@ -25,15 +27,6 @@ type Merger interface {
 	Merge(ctx context.Context, results []*sql.Rows) (Rows, error)
 }
 
-// Rows 各方法用法及语义尽可能与sql.Rows相同
-type Rows interface {
-	Next() bool
-	Scan(dest ...any) error
-	Close() error
-	Columns() ([]string, error)
-	Err() error
-}
-
 type ColumnInfo struct {
 	Index int
 	Name  string
@@ -45,3 +38,5 @@ func NewColumnInfo(index int, name string) ColumnInfo {
 		Name:  name,
 	}
 }
+
+type Rows = rows.Rows

internal/rows/types.go

@@ -0,0 +1,28 @@
+// Copyright 2021 ecodeclub
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package rows
+
+import "database/sql"
+
+var _ Rows = &sql.Rows{}
+
+// Rows 各方法用法及语义尽可能与sql.Rows相同
+type Rows interface {
+	Next() bool
+	Scan(dest ...any) error
+	Close() error
+	Columns() ([]string, error)
+	Err() error
+}

internal/valuer/primitive.go

@@ -18,6 +18,8 @@ import (
 	"database/sql"
 	"reflect"
 
+	"github.com/ecodeclub/eorm/internal/rows"
+
 	"github.com/ecodeclub/eorm/internal/model"
 )
 
@@ -34,7 +36,7 @@ func (s primitiveValue) Field(name string) (reflect.Value, error) {
 }
 
 // SetColumns 设置列值， 支持基本类型，基于 reflect 与 unsafe Value 封装
-func (s primitiveValue) SetColumns(rows *sql.Rows) error {
+func (s primitiveValue) SetColumns(rows rows.Rows) error {
 	switch s.valType.Elem().Kind() {
 	case reflect.Struct:
 		if scanner, ok := s.val.(sql.Scanner); ok {

internal/valuer/reflect.go

@@ -15,9 +15,10 @@
 package valuer
 
 import (
-	"database/sql"
 	"reflect"
 
+	"github.com/ecodeclub/eorm/internal/rows"
+
 	"github.com/ecodeclub/eorm/internal/errs"
 	"github.com/ecodeclub/eorm/internal/model"
 )
@@ -60,7 +61,7 @@ func (r reflectValue) fieldByIndex(name string) (reflect.Value, bool) {
 	return value, true
 }
 
-func (r reflectValue) SetColumns(rows *sql.Rows) error {
+func (r reflectValue) SetColumns(rows rows.Rows) error {
 	cs, err := rows.Columns()
 	if err != nil {
 		return err

internal/valuer/unsafe.go

@@ -15,10 +15,11 @@
 package valuer
 
 import (
-	"database/sql"
 	"reflect"
 	"unsafe"
 
+	"github.com/ecodeclub/eorm/internal/rows"
+
 	"github.com/ecodeclub/eorm/internal/errs"
 	"github.com/ecodeclub/eorm/internal/model"
 )
@@ -50,7 +51,7 @@ func (u unsafeValue) Field(name string) (reflect.Value, error) {
 	return val, nil
 }
 
-func (u unsafeValue) SetColumns(rows *sql.Rows) error {
+func (u unsafeValue) SetColumns(rows rows.Rows) error {
 
 	cs, err := rows.Columns()
 	if err != nil {

internal/valuer/value.go

@@ -15,9 +15,10 @@
 package valuer
 
 import (
-	"database/sql"
 	"reflect"
 
+	"github.com/ecodeclub/eorm/internal/rows"
+
 	"github.com/ecodeclub/eorm/internal/model"
 )
 
@@ -27,7 +28,9 @@ type Value interface {
 	Field(name string) (reflect.Value, error)
 	// SetColumns 设置新值，column 是列名
 	// 要注意，val 可能存在被上层复用，从而引起篡改的问题
-	SetColumns(rows *sql.Rows) error
+	// SetColumns(rows *sql.Rows) error
+
+	SetColumns(rows rows.Rows) error
 }
 
 type Creator func(val any, meta *model.TableMeta) Value

sharding_select.go

@@ -19,6 +19,8 @@ import (
 	"database/sql"
 	"sync"
 
+	"github.com/ecodeclub/eorm/internal/merger/batchmerger"
+
 	operator "github.com/ecodeclub/eorm/internal/operator"
 
 	"github.com/gotomicro/ekit/slice"
@@ -460,16 +462,20 @@ func (s *ShardingSelector[T]) GetMulti(ctx context.Context) ([]*T, error) {
 	if err != nil {
 		return nil, err
 	}
+
+	mgr := batchmerger.NewMerger()
+	rows, err := mgr.Merge(ctx, rowsSlice)
+	if err != nil {
+		return nil, err
+	}
 	var res []*T
-	for _, rows := range rowsSlice {
-		for rows.Next() {
-			tp := new(T)
-			val := s.valCreator.NewPrimitiveValue(tp, s.meta)
-			if err = val.SetColumns(rows); err != nil {
-				return nil, err
-			}
-			res = append(res, tp)
+	for rows.Next() {
+		tp := new(T)
+		val := s.valCreator.NewPrimitiveValue(tp, s.meta)
+		if err = val.SetColumns(rows); err != nil {
+			return nil, err
 		}
+		res = append(res, tp)
 	}
 	return res, nil
 }

sharding_select_test.go

@@ -6099,6 +6099,7 @@ func TestShardingSelector_Get(t *testing.T) {
 }
 
 func TestShardingSelector_GetMulti(t *testing.T) {
+	t.Parallel()
 	r := model.NewMetaRegistry()
 	_, err := r.Register(&test.OrderDetail{},
 		model.WithTableShardingAlgorithm(&hash.Hash{
@@ -6199,6 +6200,25 @@ func TestShardingSelector_GetMulti(t *testing.T) {
 				{OrderId: 234, ItemId: 12, UsingCol1: "Kevin", UsingCol2: "Durant"},
 			},
 		},
+		{
+			name: "err merge rows diff",
+			s: func() *ShardingSelector[test.OrderDetail] {
+				b := NewShardingSelector[test.OrderDetail](shardingDB).
+					Where(C("OrderId").EQ(123).Or(C("OrderId").EQ(234)))
+				return b
+			}(),
+			mockOrder: func(mock1, mock2 sqlmock.Sqlmock) {
+				rows1 := mock1.NewRows([]string{"order_id", "ite_id", "using_col1", "using_col2"})
+				rows1.AddRow(234, 12, "Kevin", "Durant")
+				mock1.ExpectQuery("SELECT `order_id`,`item_id`,`using_col1`,`using_col2` FROM `order_detail_db_0`.`order_detail_tab_0` WHERE (`order_id`=?) OR (`order_id`=?);").
+					WithArgs(123, 234).WillReturnRows(rows1)
+				rows2 := mock2.NewRows([]string{"order_id", "item_id", "using_col1", "using_col2"})
+				rows2.AddRow(123, 10, "LeBron", "James")
+				mock2.ExpectQuery("SELECT `order_id`,`item_id`,`using_col1`,`using_col2` FROM `order_detail_db_1`.`order_detail_tab_0` WHERE (`order_id`=?) OR (`order_id`=?);").
+					WithArgs(123, 234).WillReturnRows(rows2)
+			},
+			wantErr: errors.New("merger: sql.Rows列表中的字段不同"),
+		},
 	}
 
 	for _, tc := range testCases {