v0.0.437 properly handle $group in wmo extraModPipeline
Some checks failed
Build Docker and Deploy / Run goext test-suite (push) Failing after 3m23s
Some checks failed
Build Docker and Deploy / Run goext test-suite (push) Failing after 3m23s
This commit is contained in:
@@ -125,3 +125,17 @@ func (c *Coll[TData]) createToken(fieldPrimary string, dirPrimary ct.SortDirecti
|
||||
Extra: ct.Extra{},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *Coll[TData]) needsDoubleSort(ctx context.Context) bool {
|
||||
for _, ppl := range c.extraModPipeline {
|
||||
for _, stage := range ppl(ctx) {
|
||||
for _, bsone := range stage {
|
||||
if bsone.Key == "$group" {
|
||||
// a group stage in extraModPipeline results in unsorted data, which means the caller must sort again after these pipeline stages...
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
@@ -36,6 +36,14 @@ func (c *Coll[TData]) Find(ctx context.Context, filter bson.M, opts ...*options.
|
||||
pipeline = langext.ArrConcat(pipeline, ppl(ctx))
|
||||
}
|
||||
|
||||
if c.needsDoubleSort(ctx) {
|
||||
for _, opt := range opts {
|
||||
if opt != nil && opt.Sort != nil {
|
||||
pipeline = append(pipeline, bson.D{{Key: "$sort", Value: opt.Sort}})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for _, opt := range opts {
|
||||
if opt != nil && opt.Projection != nil {
|
||||
pipeline = append(pipeline, bson.D{{Key: "$project", Value: opt.Projection}})
|
||||
|
@@ -35,7 +35,7 @@ func (c *Coll[TData]) List(ctx context.Context, filter ct.Filter, pageSize *int,
|
||||
sortDirSecondary = nil
|
||||
}
|
||||
|
||||
paginationPipeline, err := createPaginationPipeline(c, inTok, sortPrimary, sortDirPrimary, sortSecondary, sortDirSecondary, pageSize)
|
||||
paginationPipeline, doubleSortPipeline, err := createPaginationPipeline(c, inTok, sortPrimary, sortDirPrimary, sortSecondary, sortDirSecondary, pageSize)
|
||||
if err != nil {
|
||||
return nil, ct.CursorToken{}, exerr.
|
||||
Wrap(err, "failed to create pagination").
|
||||
@@ -56,6 +56,10 @@ func (c *Coll[TData]) List(ctx context.Context, filter ct.Filter, pageSize *int,
|
||||
pipeline = langext.ArrConcat(pipeline, ppl(ctx))
|
||||
}
|
||||
|
||||
if c.needsDoubleSort(ctx) {
|
||||
pipeline = langext.ArrConcat(pipeline, doubleSortPipeline)
|
||||
}
|
||||
|
||||
cursor, err := c.coll.Aggregate(ctx, pipeline)
|
||||
if err != nil {
|
||||
return nil, ct.CursorToken{}, exerr.Wrap(err, "mongo-aggregation failed").Any("pipeline", pipeline).Str("collection", c.Name()).Build()
|
||||
@@ -136,14 +140,14 @@ func (c *Coll[TData]) ListWithCount(ctx context.Context, filter ct.Filter, pageS
|
||||
return data, token, count, nil
|
||||
}
|
||||
|
||||
func createPaginationPipeline[TData any](coll *Coll[TData], token ct.CursorToken, fieldPrimary string, sortPrimary ct.SortDirection, fieldSecondary *string, sortSecondary *ct.SortDirection, pageSize *int) ([]bson.D, error) {
|
||||
func createPaginationPipeline[TData any](coll *Coll[TData], token ct.CursorToken, fieldPrimary string, sortPrimary ct.SortDirection, fieldSecondary *string, sortSecondary *ct.SortDirection, pageSize *int) ([]bson.D, []bson.D, error) {
|
||||
|
||||
cond := bson.A{}
|
||||
sort := bson.D{}
|
||||
|
||||
valuePrimary, err := coll.getTokenValueAsMongoType(token.ValuePrimary, fieldPrimary)
|
||||
if err != nil {
|
||||
return nil, exerr.Wrap(err, "failed to get (primary) token-value as mongo-type").Build()
|
||||
return nil, nil, exerr.Wrap(err, "failed to get (primary) token-value as mongo-type").Build()
|
||||
}
|
||||
|
||||
if sortPrimary == ct.SortASC {
|
||||
@@ -160,7 +164,7 @@ func createPaginationPipeline[TData any](coll *Coll[TData], token ct.CursorToken
|
||||
|
||||
valueSecondary, err := coll.getTokenValueAsMongoType(token.ValueSecondary, *fieldSecondary)
|
||||
if err != nil {
|
||||
return nil, exerr.Wrap(err, "failed to get (secondary) token-value as mongo-type").Build()
|
||||
return nil, nil, exerr.Wrap(err, "failed to get (secondary) token-value as mongo-type").Build()
|
||||
}
|
||||
|
||||
if *sortSecondary == ct.SortASC {
|
||||
@@ -203,15 +207,17 @@ func createPaginationPipeline[TData any](coll *Coll[TData], token ct.CursorToken
|
||||
|
||||
} else {
|
||||
|
||||
return nil, exerr.New(exerr.TypeInternal, "unknown ct mode: "+string(token.Mode)).Any("token.Mode", token.Mode).Build()
|
||||
return nil, nil, exerr.New(exerr.TypeInternal, "unknown ct mode: "+string(token.Mode)).Any("token.Mode", token.Mode).Build()
|
||||
|
||||
}
|
||||
|
||||
pipeline = append(pipeline, bson.D{{Key: "$sort", Value: sort}})
|
||||
|
||||
pipelineSort := mongo.Pipeline{bson.D{{Key: "$sort", Value: sort}}}
|
||||
|
||||
if pageSize != nil {
|
||||
pipeline = append(pipeline, bson.D{{Key: "$limit", Value: int64(*pageSize + 1)}})
|
||||
}
|
||||
|
||||
return pipeline, nil
|
||||
return pipeline, pipelineSort, nil
|
||||
}
|
||||
|
Reference in New Issue
Block a user