代码修改

.CHANGELOG.md

@@ -25,6 +25,7 @@
 - [eorm: 分库分表: 结果集处理--聚合函数(不含GroupBy子句)](https://github.com/ecodeclub/eorm/pull/187)
 - [eorm: 分库分表: 范围查询支持](https://github.com/ecodeclub/eorm/pull/178)
 - [eorm: 修复单条查询时连接泄露问题](https://github.com/ecodeclub/eorm/pull/188)
+- [eorm: 分库分表: 结果集处理--聚合函数(含GroupBy子句)](https://github.com/ecodeclub/eorm/pull/193)
 
 ## v0.0.1:
 - [Init Project](https://github.com/ecodeclub/eorm/pull/1)

internal/merger/aggregatemerger/aggregator/avg.go

@@ -17,19 +17,21 @@ package aggregator
 import (
 	"reflect"
 
+	"github.com/ecodeclub/eorm/internal/merger"
+
 	"github.com/ecodeclub/eorm/internal/merger/internal/errs"
 )
 
 // AVG 用于求平均值，通过sum/count求得。
 // AVG 我们并不能预期在不同的数据库上，精度会不会损失，以及损失的话会有多少的损失。这很大程度上跟数据库类型，数据库驱动实现都有关
 type AVG struct {
-	sumColumnInfo   ColumnInfo
-	countColumnInfo ColumnInfo
+	sumColumnInfo   merger.ColumnInfo
+	countColumnInfo merger.ColumnInfo
 	avgName         string
 }
 
 // NewAVG sumInfo是sum的信息，countInfo是count的信息，avgName用于Column方法
-func NewAVG(sumInfo ColumnInfo, countInfo ColumnInfo, avgName string) *AVG {
+func NewAVG(sumInfo merger.ColumnInfo, countInfo merger.ColumnInfo, avgName string) *AVG {
 	return &AVG{
 		sumColumnInfo:   sumInfo,
 		countColumnInfo: countInfo,

internal/merger/aggregatemerger/aggregator/avg_test.go

@@ -15,6 +15,7 @@
 package aggregator
 
 import (
+	"github.com/ecodeclub/eorm/internal/merger"
 	"testing"
 
 	"github.com/ecodeclub/eorm/internal/merger/internal/errs"
@@ -82,7 +83,7 @@ func TestAvg_Aggregate(t *testing.T) {
 	}
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
-			avg := NewAVG(NewColumnInfo(tc.index[0], "SUM(grade)"), NewColumnInfo(tc.index[1], "COUNT(grade)"), "AVG(grade)")
+			avg := NewAVG(merger.NewColumnInfo(tc.index[0], "SUM(grade)"), merger.NewColumnInfo(tc.index[1], "COUNT(grade)"), "AVG(grade)")
 			val, err := avg.Aggregate(tc.input)
 			assert.Equal(t, tc.wantErr, err)
 			if err != nil {

internal/merger/aggregatemerger/aggregator/count.go

@@ -17,11 +17,13 @@ package aggregator
 import (
 	"reflect"
 
+	"github.com/ecodeclub/eorm/internal/merger"
+
 	"github.com/ecodeclub/eorm/internal/merger/internal/errs"
 )
 
 type Count struct {
-	countInfo ColumnInfo
+	countInfo merger.ColumnInfo
 }
 
 func (s *Count) Aggregate(cols [][]any) (any, error) {
@@ -49,7 +51,7 @@ func (s *Count) ColumnName() string {
 	return s.countInfo.Name
 }
 
-func NewCount(info ColumnInfo) *Count {
+func NewCount(info merger.ColumnInfo) *Count {
 	return &Count{
 		countInfo: info,
 	}

internal/merger/aggregatemerger/aggregator/count_test.go

@@ -17,6 +17,8 @@ package aggregator
 import (
 	"testing"
 
+	"github.com/ecodeclub/eorm/internal/merger"
+
 	"github.com/ecodeclub/eorm/internal/merger/internal/errs"
 
 	"github.com/stretchr/testify/assert"
@@ -78,7 +80,7 @@ func TestCount_Aggregate(t *testing.T) {
 	}
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
-			count := NewCount(NewColumnInfo(tc.countIndex, "COUNT(id)"))
+			count := NewCount(merger.NewColumnInfo(tc.countIndex, "COUNT(id)"))
 			val, err := count.Aggregate(tc.input)
 			assert.Equal(t, tc.wantErr, err)
 			if err != nil {

internal/merger/aggregatemerger/aggregator/max.go

@@ -17,11 +17,13 @@ package aggregator
 import (
 	"reflect"
 
+	"github.com/ecodeclub/eorm/internal/merger"
+
 	"github.com/ecodeclub/eorm/internal/merger/internal/errs"
 )
 
 type Max struct {
-	maxColumnInfo ColumnInfo
+	maxColumnInfo merger.ColumnInfo
 }
 
 func (m *Max) Aggregate(cols [][]any) (any, error) {
@@ -49,7 +51,7 @@ func (m *Max) ColumnName() string {
 	return m.maxColumnInfo.Name
 }
 
-func NewMax(info ColumnInfo) *Max {
+func NewMax(info merger.ColumnInfo) *Max {
 	return &Max{
 		maxColumnInfo: info,
 	}

internal/merger/aggregatemerger/aggregator/max_test.go

@@ -17,6 +17,8 @@ package aggregator
 import (
 	"testing"
 
+	"github.com/ecodeclub/eorm/internal/merger"
+
 	"github.com/ecodeclub/eorm/internal/merger/internal/errs"
 	"github.com/stretchr/testify/assert"
 )
@@ -77,7 +79,7 @@ func TestMax_Aggregate(t *testing.T) {
 	}
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
-			max := NewMax(NewColumnInfo(tc.maxIndex, "MAX(id)"))
+			max := NewMax(merger.NewColumnInfo(tc.maxIndex, "MAX(id)"))
 			val, err := max.Aggregate(tc.input)
 			assert.Equal(t, tc.wantErr, err)
 			if err != nil {

internal/merger/aggregatemerger/aggregator/min.go

@@ -17,11 +17,13 @@ package aggregator
 import (
 	"reflect"
 
+	"github.com/ecodeclub/eorm/internal/merger"
+
 	"github.com/ecodeclub/eorm/internal/merger/internal/errs"
 )
 
 type Min struct {
-	minColumnInfo ColumnInfo
+	minColumnInfo merger.ColumnInfo
 }
 
 func (m *Min) Aggregate(cols [][]any) (any, error) {
@@ -50,7 +52,7 @@ func (m *Min) ColumnName() string {
 	return m.minColumnInfo.Name
 }
 
-func NewMin(info ColumnInfo) *Min {
+func NewMin(info merger.ColumnInfo) *Min {
 	return &Min{
 		minColumnInfo: info,
 	}

internal/merger/aggregatemerger/aggregator/min_test.go

@@ -15,6 +15,7 @@
 package aggregator
 
 import (
+	"github.com/ecodeclub/eorm/internal/merger"
 	"testing"
 
 	"github.com/ecodeclub/eorm/internal/merger/internal/errs"
@@ -77,7 +78,7 @@ func TestMin_Aggregate(t *testing.T) {
 	}
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
-			min := NewMin(NewColumnInfo(tc.minIndex, "MIN(id)"))
+			min := NewMin(merger.NewColumnInfo(tc.minIndex, "MIN(id)"))
 			val, err := min.Aggregate(tc.input)
 			assert.Equal(t, tc.wantErr, err)
 			if err != nil {

internal/merger/aggregatemerger/aggregator/sum.go

@@ -17,11 +17,13 @@ package aggregator
 import (
 	"reflect"
 
+	"github.com/ecodeclub/eorm/internal/merger"
+
 	"github.com/ecodeclub/eorm/internal/merger/internal/errs"
 )
 
 type Sum struct {
-	sumColumnInfo ColumnInfo
+	sumColumnInfo merger.ColumnInfo
 }
 
 func (s *Sum) Aggregate(cols [][]any) (any, error) {
@@ -50,7 +52,7 @@ func (s *Sum) ColumnName() string {
 	return s.sumColumnInfo.Name
 }
 
-func NewSum(info ColumnInfo) *Sum {
+func NewSum(info merger.ColumnInfo) *Sum {
 	return &Sum{
 		sumColumnInfo: info,
 	}

internal/merger/aggregatemerger/aggregator/sum_test.go

@@ -15,6 +15,7 @@
 package aggregator
 
 import (
+	"github.com/ecodeclub/eorm/internal/merger"
 	"testing"
 
 	"github.com/ecodeclub/eorm/internal/merger/internal/errs"
@@ -79,7 +80,7 @@ func TestSum_Aggregate(t *testing.T) {
 	}
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
-			sum := NewSum(NewColumnInfo(tc.sumIndex, "SUM(id)"))
+			sum := NewSum(merger.NewColumnInfo(tc.sumIndex, "SUM(id)"))
 			val, err := sum.Aggregate(tc.input)
 			assert.Equal(t, tc.wantErr, err)
 			if err != nil {

internal/merger/aggregatemerger/aggregator/type.go

@@ -24,15 +24,3 @@ type Aggregator interface {
 	// ColumnName 聚合函数的别名
 	ColumnName() string
 }
-
-type ColumnInfo struct {
-	Index int
-	Name  string
-}
-
-func NewColumnInfo(index int, name string) ColumnInfo {
-	return ColumnInfo{
-		Index: index,
-		Name:  name,
-	}
-}

internal/merger/aggregatemerger/merger.go

@@ -17,7 +17,7 @@ package aggregatemerger
 import (
 	"context"
 	"database/sql"
-	"reflect"
+	"github.com/ecodeclub/eorm/internal/merger/utils"
 	"sync"
 	_ "unsafe"
 
@@ -27,9 +27,6 @@ import (
 	"go.uber.org/multierr"
 )
 
-//go:linkname convertAssign database/sql.convertAssign
-func convertAssign(dest, src any) error
-
 // Merger 该实现不支持group by操作,并且聚合函数查询应该只返回一行数据。
 type Merger struct {
 	aggregators []aggregator.Aggregator
@@ -151,32 +148,14 @@ func (r *Rows) getSqlRowsData() ([][]any, error) {
 	return rowsData, nil
 }
 func (r *Rows) getSqlRowData(row *sql.Rows) ([]any, error) {
-	colsInfo, err := row.ColumnTypes()
-	if err != nil {
-		return nil, err
-	}
-	// colsData 表示一个sql.Rows的数据
-	colsData := make([]any, 0, len(colsInfo))
+
+	var colsData []any
+	var err error
 	if row.Next() {
-		// 拿到sql.Rows字段的类型然后初始化
-		for _, colInfo := range colsInfo {
-			typ := colInfo.ScanType()
-			// sqlite3的驱动返回的是指针。循环的去除指针
-			for typ.Kind() == reflect.Pointer {
-				typ = typ.Elem()
-			}
-			newData := reflect.New(typ).Interface()
-			colsData = append(colsData, newData)
-		}
-		// 通过Scan赋值
-		err = row.Scan(colsData...)
+		colsData, err = utils.Scan(row)
 		if err != nil {
 			return nil, err
 		}
-		// 去掉reflect.New的指针
-		for i := 0; i < len(colsData); i++ {
-			colsData[i] = reflect.ValueOf(colsData[i]).Elem().Interface()
-		}
 	} else {
 		// sql.Rows迭代过程中发生报错，返回报错
 		if row.Err() != nil {
@@ -201,7 +180,7 @@ func (r *Rows) Scan(dest ...any) error {
 		return errs.ErrMergerScanNotNext
 	}
 	for i := 0; i < len(dest); i++ {
-		err := convertAssign(dest[i], r.cur[i])
+		err := utils.ConvertAssign(dest[i], r.cur[i])
 		if err != nil {
 			return err
 		}