优化group和reverse等查询聚合

xuejmnet · xuejmnet · commit b0f110829d18 · 2022-08-09T13:41:59.000+08:00
diff --git a/src/ShardingCore/Sharding/MergeEngines/Executors/ShardingMergers/AppendOrderSequenceEnumerableShardingMerger.cs b/src/ShardingCore/Sharding/MergeEngines/Executors/ShardingMergers/AppendOrderSequenceEnumerableShardingMerger.cs
@@ -14,14 +14,26 @@ public AppendOrderSequenceEnumerableShardingMerger(StreamMergeContext streamMerg
             streamMergeContext, async)
         {
         }
-
         public override IStreamMergeAsyncEnumerator<TEntity> StreamMerge(
             List<IStreamMergeAsyncEnumerator<TEntity>> parallelResults)
         {
             if (GetStreamMergeContext().HasGroupQuery())
+            {
                 return new MultiAggregateOrderStreamMergeAsyncEnumerator<TEntity>(GetStreamMergeContext(),
                     parallelResults);
+            }
+
             return new MultiOrderStreamMergeAsyncEnumerator<TEntity>(GetStreamMergeContext(), parallelResults);
         }
+
+        protected override IStreamMergeAsyncEnumerator<TEntity> StreamInMemoryMerge(List<IStreamMergeAsyncEnumerator<TEntity>> parallelResults)
+        {
+            //如果是group in memory merger需要在内存中聚合好所有的 并且最后通过内存聚合在发挥
+            if (GetStreamMergeContext().GroupQueryMemoryMerge())
+            {
+                return new MultiAggregateOrderStreamMergeAsyncEnumerator<TEntity>(GetStreamMergeContext(), parallelResults);
+            }
+            return StreamMerge(parallelResults);
+        }
     }
 }
diff --git a/src/ShardingCore/Sharding/MergeEngines/Executors/ShardingMergers/LastOrDefaultEnumerableShardingMerger.cs b/src/ShardingCore/Sharding/MergeEngines/Executors/ShardingMergers/LastOrDefaultEnumerableShardingMerger.cs
@@ -10,13 +10,5 @@ public LastOrDefaultEnumerableShardingMerger(StreamMergeContext streamMergeConte
             streamMergeContext, async)
         {
         }
-
-        protected override IStreamMergeAsyncEnumerator<TEntity> StreamInMemoryMerge(List<IStreamMergeAsyncEnumerator<TEntity>> parallelResults)
-        {
-            if (GetStreamMergeContext().IsPaginationQuery())
-                return new PaginationStreamMergeAsyncEnumerator<TEntity>(GetStreamMergeContext(), parallelResults, 0, GetStreamMergeContext().GetPaginationReWriteTake());//内存聚合分页不可以直接获取skip必须获取skip+take的数目
-
-            return base.StreamInMemoryMerge(parallelResults);
-        }
     }
 }
diff --git a/src/ShardingCore/Sharding/MergeEngines/Executors/ShardingMergers/ReverseEnumerableShardingMerger.cs b/src/ShardingCore/Sharding/MergeEngines/Executors/ShardingMergers/ReverseEnumerableShardingMerger.cs
@@ -11,23 +11,10 @@ public ReverseEnumerableShardingMerger(StreamMergeContext streamMergeContext, bo
         {
         }
 
-        // protected override IStreamMergeAsyncEnumerator<TEntity> StreamInMemoryMerge(
-        //     List<IStreamMergeAsyncEnumerator<TEntity>> parallelResults)
-        // {
-        //     // if (GetStreamMergeContext().IsPaginationQuery() && GetStreamMergeContext().HasGroupQuery())
-        //     // {
-        //     //     var multiAggregateOrderStreamMergeAsyncEnumerator =
-        //     //         new MultiAggregateOrderStreamMergeAsyncEnumerator<TEntity>(GetStreamMergeContext(),
-        //     //             parallelResults);
-        //     //     return new PaginationStreamMergeAsyncEnumerator<TEntity>(GetStreamMergeContext(),
-        //     //         new[] { multiAggregateOrderStreamMergeAsyncEnumerator }, 0,
-        //     //         GetStreamMergeContext().GetPaginationReWriteTake());
-        //     // }
-        //
-        //     if (GetStreamMergeContext().IsPaginationQuery())
-        //         return new PaginationStreamMergeAsyncEnumerator<TEntity>(GetStreamMergeContext(), parallelResults, 0,
-        //             GetStreamMergeContext().GetPaginationReWriteTake());
-        //     return base.StreamInMemoryMerge(parallelResults);
-        // }
+        public override IStreamMergeAsyncEnumerator<TEntity> StreamMerge(List<IStreamMergeAsyncEnumerator<TEntity>> parallelResults)
+        {
+            var streamMergeAsyncEnumerator = base.StreamMerge(parallelResults);
+            return new InMemoryReverseStreamMergeAsyncEnumerator<TEntity>(streamMergeAsyncEnumerator);
+        }
     }
 }
diff --git a/src/ShardingCore/Sharding/MergeEngines/Executors/ShardingMergers/SequenceEnumerableShardingMerger.cs b/src/ShardingCore/Sharding/MergeEngines/Executors/ShardingMergers/SequenceEnumerableShardingMerger.cs
@@ -25,6 +25,11 @@ public override IStreamMergeAsyncEnumerator<TEntity> StreamMerge(
 
         protected override IStreamMergeAsyncEnumerator<TEntity> StreamInMemoryMerge(List<IStreamMergeAsyncEnumerator<TEntity>> parallelResults)
         {
+            //如果是group in memory merger需要在内存中聚合好所有的 并且最后通过内存聚合在发挥
+            if (GetStreamMergeContext().GroupQueryMemoryMerge())
+            {
+                return new MultiAggregateOrderStreamMergeAsyncEnumerator<TEntity>(GetStreamMergeContext(), parallelResults);
+            }
             return StreamMerge(parallelResults);
         }
     }

Original file line number	Diff line number	Diff line change
`@@ -14,14 +14,26 @@ public AppendOrderSequenceEnumerableShardingMerger(StreamMergeContext streamMerg`
`14`	`14`	`streamMergeContext, async)`
`15`	`15`	`{`
`16`	`16`	`}`
`17`		`-`
`18`	`17`	`public override IStreamMergeAsyncEnumerator<TEntity> StreamMerge(`
`19`	`18`	`List<IStreamMergeAsyncEnumerator<TEntity>> parallelResults)`
`20`	`19`	`{`
`21`	`20`	`if (GetStreamMergeContext().HasGroupQuery())`
	`21`	`+ {`
`22`	`22`	`return new MultiAggregateOrderStreamMergeAsyncEnumerator<TEntity>(GetStreamMergeContext(),`
`23`	`23`	`parallelResults);`
	`24`	`+ }`
	`25`	`+`
`24`	`26`	`return new MultiOrderStreamMergeAsyncEnumerator<TEntity>(GetStreamMergeContext(), parallelResults);`
`25`	`27`	`}`
	`28`	`+`
	`29`	`+ protected override IStreamMergeAsyncEnumerator<TEntity> StreamInMemoryMerge(List<IStreamMergeAsyncEnumerator<TEntity>> parallelResults)`
	`30`	`+ {`
	`31`	`+ //如果是group in memory merger需要在内存中聚合好所有的并且最后通过内存聚合在发挥`
	`32`	`+ if (GetStreamMergeContext().GroupQueryMemoryMerge())`
	`33`	`+ {`
	`34`	`+ return new MultiAggregateOrderStreamMergeAsyncEnumerator<TEntity>(GetStreamMergeContext(), parallelResults);`
	`35`	`+ }`
	`36`	`+ return StreamMerge(parallelResults);`
	`37`	`+ }`
`26`	`38`	`}`
`27`	`39`	`}`
Original file line number	Diff line number	Diff line change
`@@ -10,13 +10,5 @@ public LastOrDefaultEnumerableShardingMerger(StreamMergeContext streamMergeConte`
`10`	`10`	`streamMergeContext, async)`
`11`	`11`	`{`
`12`	`12`	`}`
`13`		`-`
`14`		`- protected override IStreamMergeAsyncEnumerator<TEntity> StreamInMemoryMerge(List<IStreamMergeAsyncEnumerator<TEntity>> parallelResults)`
`15`		`- {`
`16`		`- if (GetStreamMergeContext().IsPaginationQuery())`
`17`		`- return new PaginationStreamMergeAsyncEnumerator<TEntity>(GetStreamMergeContext(), parallelResults, 0, GetStreamMergeContext().GetPaginationReWriteTake());//内存聚合分页不可以直接获取skip必须获取skip+take的数目`
`18`		`-`
`19`		`- return base.StreamInMemoryMerge(parallelResults);`
`20`		`- }`
`21`	`13`	`}`
`22`	`14`	`}`
Original file line number	Diff line number	Diff line change
`@@ -25,6 +25,11 @@ public override IStreamMergeAsyncEnumerator<TEntity> StreamMerge(`
`25`	`25`
`26`	`26`	`protected override IStreamMergeAsyncEnumerator<TEntity> StreamInMemoryMerge(List<IStreamMergeAsyncEnumerator<TEntity>> parallelResults)`
`27`	`27`	`{`
	`28`	`+ //如果是group in memory merger需要在内存中聚合好所有的并且最后通过内存聚合在发挥`
	`29`	`+ if (GetStreamMergeContext().GroupQueryMemoryMerge())`
	`30`	`+ {`
	`31`	`+ return new MultiAggregateOrderStreamMergeAsyncEnumerator<TEntity>(GetStreamMergeContext(), parallelResults);`
	`32`	`+ }`
`28`	`33`	`return StreamMerge(parallelResults);`
`29`	`34`	`}`
`30`	`35`	`}`