This commit is contained in:
xuejmnet 2021-09-03 07:37:39 +08:00
parent 880b45e486
commit 7fa315cd0d
4 changed files with 37 additions and 5 deletions

View File

@ -4,6 +4,7 @@ using System.Linq.Expressions;
using System.Reflection; using System.Reflection;
using System.Text; using System.Text;
using Microsoft.EntityFrameworkCore.Infrastructure; using Microsoft.EntityFrameworkCore.Infrastructure;
using Microsoft.EntityFrameworkCore.Internal;
namespace ShardingCore.Sharding.PaginationConfigurations namespace ShardingCore.Sharding.PaginationConfigurations
{ {

View File

@ -5,8 +5,10 @@ using System.Linq.Expressions;
using System.Text; using System.Text;
using System.Threading; using System.Threading;
using ShardingCore.Core.ShardingPage.Abstractions; using ShardingCore.Core.ShardingPage.Abstractions;
using ShardingCore.Core.VirtualTables;
using ShardingCore.Extensions; using ShardingCore.Extensions;
using ShardingCore.Sharding.Abstractions; using ShardingCore.Sharding.Abstractions;
using ShardingCore.Sharding.PaginationConfigurations;
using ShardingCore.Sharding.StreamMergeEngines.EnumeratorStreamMergeEngines; using ShardingCore.Sharding.StreamMergeEngines.EnumeratorStreamMergeEngines;
using ShardingCore.Sharding.StreamMergeEngines.EnumeratorStreamMergeEngines.Abstractions; using ShardingCore.Sharding.StreamMergeEngines.EnumeratorStreamMergeEngines.Abstractions;
@ -23,11 +25,13 @@ namespace ShardingCore.Sharding.ShardingQueryExecutors
{ {
private readonly StreamMergeContext<TEntity> _streamMergeContext; private readonly StreamMergeContext<TEntity> _streamMergeContext;
private readonly IShardingPageManager _shardingPageManager; private readonly IShardingPageManager _shardingPageManager;
private readonly IVirtualTableManager _virtualTableManager;
public EnumeratorShardingQueryExecutor(StreamMergeContext<TEntity> streamMergeContext) public EnumeratorShardingQueryExecutor(StreamMergeContext<TEntity> streamMergeContext)
{ {
_streamMergeContext = streamMergeContext; _streamMergeContext = streamMergeContext;
_shardingPageManager = ShardingContainer.GetService<IShardingPageManager>(); _shardingPageManager = ShardingContainer.GetService<IShardingPageManager>();
_virtualTableManager = ShardingContainer.GetService<IVirtualTableManager>();
} }
public IEnumeratorStreamMergeEngine<TEntity> ExecuteAsync(CancellationToken cancellationToken = new CancellationToken()) public IEnumeratorStreamMergeEngine<TEntity> ExecuteAsync(CancellationToken cancellationToken = new CancellationToken())
{ {
@ -36,12 +40,33 @@ namespace ShardingCore.Sharding.ShardingQueryExecutors
{ {
return new SingleQueryEnumeratorAsyncStreamMergeEngine<TEntity>(_streamMergeContext); return new SingleQueryEnumeratorAsyncStreamMergeEngine<TEntity>(_streamMergeContext);
} }
//未开启系统分表 //未开启系统分表或者本次查询涉及多张分表
if (_shardingPageManager.Current == null) if (!_streamMergeContext.IsPaginationQuery()||!_streamMergeContext.IsSingleShardingTableQuery()||_shardingPageManager.Current == null)
{ {
return new DefaultShardingEnumeratorAsyncStreamMergeEngine<TEntity>(_streamMergeContext); return new DefaultShardingEnumeratorAsyncStreamMergeEngine<TEntity>(_streamMergeContext);
} }
var shardingEntityType = _streamMergeContext.RouteResults.First().ReplaceTables.Single(o=>o.IsShardingTable()).EntityType;
var virtualTable = _virtualTableManager.GetVirtualTable(_streamMergeContext.GetShardingDbContext().ShardingDbContextType,shardingEntityType);
if (!virtualTable.EnablePagination)
return new DefaultShardingEnumeratorAsyncStreamMergeEngine<TEntity>(_streamMergeContext);
if (_streamMergeContext.Orders.IsEmpty())
{
var append = virtualTable.PaginationMetadata.PaginationConfigs.FirstOrDefault(o=>o.AppendIfOrderNone);
if (append != null)
{
123
return new SequenceEnumeratorAsyncStreamMergeEngine<TEntity>(_streamMergeContext);
}
}
var propertyOrder = _streamMergeContext.Orders.First();
//PaginationMatchEnum.Owner
111
var paginationConfig = virtualTable.PaginationMetadata.PaginationConfigs.FirstOrDefault(o=>o.PropertyName==propertyOrder.PropertyExpression);
if (paginationConfig==null)
return new DefaultShardingEnumeratorAsyncStreamMergeEngine<TEntity>(_streamMergeContext);
//调用顺序排序
paginationConfig
} }

View File

@ -94,6 +94,12 @@ namespace ShardingCore.Sharding
return Skip.HasValue || Take.HasValue; return Skip.HasValue || Take.HasValue;
} }
public bool IsPaginationQuery()
{
return Skip.GetValueOrDefault() > 0 || Take.GetValueOrDefault() > 0;
}
public bool HasGroupQuery() public bool HasGroupQuery()
{ {
return this.GroupByContext.GroupExpression != null; return this.GroupByContext.GroupExpression != null;

View File

@ -60,7 +60,7 @@ namespace ShardingCore.Sharding.StreamMergeEngines.EnumeratorStreamMergeEngines
public override IStreamMergeAsyncEnumerator<TEntity> GetStreamMergeAsyncEnumerator(IStreamMergeAsyncEnumerator<TEntity>[] streamsAsyncEnumerators) public override IStreamMergeAsyncEnumerator<TEntity> GetStreamMergeAsyncEnumerator(IStreamMergeAsyncEnumerator<TEntity>[] streamsAsyncEnumerators)
{ {
if (StreamMergeContext.HasSkipTake()) if (StreamMergeContext.IsPaginationQuery())
return new PaginationStreamMergeAsyncEnumerator<TEntity>(StreamMergeContext, streamsAsyncEnumerators); return new PaginationStreamMergeAsyncEnumerator<TEntity>(StreamMergeContext, streamsAsyncEnumerators);
if (StreamMergeContext.HasGroupQuery()) if (StreamMergeContext.HasGroupQuery())
return new MultiAggregateOrderStreamMergeAsyncEnumerator<TEntity>(StreamMergeContext, streamsAsyncEnumerators); return new MultiAggregateOrderStreamMergeAsyncEnumerator<TEntity>(StreamMergeContext, streamsAsyncEnumerators);