6237307473

Committed 19 Sep 2023 04:02PM UTC coverage: 57.015% (-0.4%) from 57.44%

Build # 6237307473

Build Type

push

github

Committed by

web-flow

Commit Message

Feature/rc4 (#1570)

* Syntax tidying
* Dependency updates
* Event handling singletons (ThrowImmediately and co)

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: James A Sutherland <>
Co-authored-by: James Friel <jfriel001@dundee.ac.uk>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Run Details

10734 of 20259 branches covered (0.0%)

Branch coverage included in aggregate %.

5922 of 5922 new or added lines in 565 files covered. (100.0%)

30687 of 52390 relevant lines covered (58.57%)

7361.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.97

/Rdmp.Core/QueryBuilding/CohortQueryBuilderResult.cs

﻿// Copyright (c) The University of Dundee 2018-2019
// This file is part of the Research Data Management Platform (RDMP).
// RDMP is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
// RDMP is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
// You should have received a copy of the GNU General Public License along with RDMP. If not, see <https://www.gnu.org/licenses/>.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using FAnsi;
using FAnsi.Discovery;
using FAnsi.Discovery.QuerySyntax;
using Rdmp.Core.CohortCreation.Execution;
using Rdmp.Core.Curation.Data;
using Rdmp.Core.Curation.Data.Aggregation;
using Rdmp.Core.Curation.Data.Cohort;
using Rdmp.Core.MapsDirectlyToDatabaseTable;
using Rdmp.Core.Providers;
using Rdmp.Core.QueryBuilding.Parameters;
using Rdmp.Core.QueryCaching.Aggregation;
using Rdmp.Core.ReusableLibraryCode.DataAccess;
using Rdmp.Core.ReusableLibraryCode.Settings;

namespace Rdmp.Core.QueryBuilding;

/// <summary>
/// Builds a subset of a <see cref="CohortIdentificationConfiguration"/> e.g. a single <see cref="CohortAggregateContainer"/> (UNION / INTERSECT / EXCEPT) or a
/// cohort set.  This includes identifying all <see cref="Dependencies"/> and resolving the dependencies servers / the <see cref="CacheServer"/> to determine
/// whether a valid query can be assembled from the sub-components and deciding where it can be run (e.g. should the query run on the cache server or the data server
/// or are they both the same server so query sections can be mixed depending on cache hit/miss for each bit).
/// </summary>
public class CohortQueryBuilderResult
{
    public ExternalDatabaseServer CacheServer { get; }
    public CachedAggregateConfigurationResultsManager CacheManager { get; }

    public bool IsForContainer { get; private set; }
    public ICoreChildProvider ChildProvider { get; }
    public CohortQueryBuilderHelper Helper { get; }
    public QueryBuilderCustomArgs Customise { get; }
    public CancellationToken CancellationToken { get; }

    private readonly StringBuilder _log = new();

    /// <summary>
    /// Log of all activities undertaken while building
    /// </summary>
    public string Log => _log.ToString();

    /// <summary>
    /// The allowable caching state based on the <see cref="Dependencies"/>, whether there is a
    /// <see cref="CacheServer"/> and if they are on the same or separate servers from one another
    /// </summary>
    public CacheUsage CacheUsageDecision { get; private set; }

    private List<CohortQueryBuilderDependency> _dependencies = new();
    private bool _alreadyBuilt;

    public IReadOnlyCollection<CohortQueryBuilderDependency> Dependencies => _dependencies;

    /// <summary>
    /// Only Populated after Building.  If all <see cref="Dependencies"/> are on the same server as one another
    /// then this will contain all tables that must be queried otherwise it will be null
    /// </summary>
    public DataAccessPointCollection DependenciesSingleServer { get; private set; }

    /// <summary>
    /// The final SQL that should be executed on the <see cref="TargetServer"/>
    /// </summary>
    public string Sql { get; private set; }

    /// <summary>
    /// The location at which the <see cref="Sql"/> should be run (may be a data server or a cache server or they may be one and the same!)
    /// </summary>
    public DiscoveredServer TargetServer { get; set; }

    public IOrderable StopContainerWhenYouReach { get; set; }
    public int CountOfSubQueries => Dependencies.Count;
    public int CountOfCachedSubQueries { get; private set; }

    public IReadOnlyCollection<IPluginCohortCompiler> PluginCohortCompilers { get; } =
        Array.Empty<PluginCohortCompiler>().ToList().AsReadOnly();

    /// <summary>
    /// Creates a new result for a single <see cref="AggregateConfiguration"/> or <see cref="CohortAggregateContainer"/>
    /// </summary>
    /// <param name="cacheServer"></param>
    /// <param name="childProvider"></param>
    /// <param name="helper"></param>
    /// <param name="customise"></param>
    /// <param name="cancellationToken"></param>
    public CohortQueryBuilderResult(ExternalDatabaseServer cacheServer, ICoreChildProvider childProvider,
        CohortQueryBuilderHelper helper, QueryBuilderCustomArgs customise, CancellationToken cancellationToken)
    {
        CacheServer = cacheServer;
        ChildProvider = childProvider;
        Helper = helper;
        Customise = customise;
        CancellationToken = cancellationToken;

        if (cacheServer != null)
        {
            CacheManager = new CachedAggregateConfigurationResultsManager(CacheServer);

            try
            {
                PluginCohortCompilers = PluginCohortCompilerFactory.CreateAll();
            }
            catch (Exception ex)
            {
                throw new Exception("Failed to build list of IPluginCohortCompilers", ex);
            }
        }
    }


    public void BuildFor(CohortAggregateContainer container, ParameterManager parameterManager)
    {
        ThrowIfAlreadyBuilt();
        IsForContainer = true;

        _log.AppendLine($"Starting Build for {container}");
        //gather dependencies
        foreach (var cohortSet in ChildProvider.GetAllChildrenRecursively(container).OfType<AggregateConfiguration>()
                     .Where(IsEnabled).OrderBy(ac => ac.Order))
            AddDependency(cohortSet);

        if (!Dependencies.Any())
            throw new QueryBuildingException(
                $"There are no AggregateConfigurations under the SET container '{container}'");

        LogDependencies();

        BuildDependenciesSql(parameterManager.ParametersFoundSoFarInQueryGeneration[ParameterLevel.Global].ToArray());

        MakeCacheDecision();

        Sql = BuildSql(container, parameterManager);
    }

    public void BuildFor(AggregateConfiguration configuration, ParameterManager parameterManager)
    {
        ThrowIfAlreadyBuilt();
        IsForContainer = false;

        _log.AppendLine($"Starting Build for {configuration}");
        var d = AddDependency(configuration);

        LogDependencies();

        BuildDependenciesSql(parameterManager.ParametersFoundSoFarInQueryGeneration[ParameterLevel.Global].ToArray());

        MakeCacheDecision();


        Sql = BuildSql(d, parameterManager);
    }

    private string BuildSql(CohortAggregateContainer container, ParameterManager parameterManager)
    {
        Dictionary<CohortQueryBuilderDependency, string> sqlDictionary;

        //if we are fully cached on everything
        if (Dependencies.All(d => d.SqlFullyCached != null))
        {
            SetTargetServer(GetCacheServer(), "all dependencies are fully cached"); //run on the cache server

            //all are cached
            CountOfCachedSubQueries = CountOfSubQueries;

            sqlDictionary =
                Dependencies.ToDictionary(k => k,
                    v => v.SqlFullyCached.Use(parameterManager)); //run the fully cached sql
        }
        else
        {
            var uncached =
                $"CacheUsageDecision is {CacheUsageDecision} and the following were not cached:{string.Join(Environment.NewLine, Dependencies.Where(d => d.SqlFullyCached == null))}";

            switch (CacheUsageDecision)
            {
                case CacheUsage.MustUse:
                    throw new QueryBuildingException(
                        $"Could not build final SQL because some queries are not fully cached and {uncached}");

                case CacheUsage.Opportunistic:

                    //The cache and dataset are on the same server so run it
                    SetTargetServer(DependenciesSingleServer.GetDistinctServer(),
                        $"not all dependencies are cached while {uncached}");

                    CountOfCachedSubQueries = Dependencies.Count(d => d.SqlFullyCached != null);

                    sqlDictionary =
                        Dependencies.ToDictionary(k => k,
                            v => v.SqlFullyCached?.Use(parameterManager) ??
                                 v.SqlPartiallyCached?.Use(parameterManager) ??
                                 v.SqlCacheless.Use(parameterManager)); //run the fully cached sql
                    break;

                case CacheUsage.AllOrNothing:

                    //It's not fully cached so we have to run it entirely uncached
                    SetTargetServer(DependenciesSingleServer.GetDistinctServer(),
                        $"not all dependencies are cached while {uncached}");

                    //cannot use any of the caches because it's all or nothing
                    CountOfCachedSubQueries = 0;
                    sqlDictionary =
                        Dependencies.ToDictionary(k => k,
                            v => v.SqlCacheless.Use(parameterManager)); //run the fully uncached sql
                    break;
                default:
                    throw new ArgumentOutOfRangeException();
            }
        }

        return WriteContainers(container, TargetServer.GetQuerySyntaxHelper(), sqlDictionary, 0);
    }

    private void SetTargetServer(DiscoveredServer target, string reason)
    {
        if (TargetServer != null)
            throw new InvalidOperationException("You are only supposed to pick a target server once");

        TargetServer = target;
        _log.AppendLine($"Picked TargetServer as {target} because {reason}");
    }

    private string WriteContainers(CohortAggregateContainer container, IQuerySyntaxHelper syntaxHelper,
        Dictionary<CohortQueryBuilderDependency, string> sqlDictionary, int tabs)
    {
        var sql = "";

        //Things we need to output
        var toWriteOut = container.GetOrderedContents().Where(IsEnabled).ToArray();

        if (toWriteOut.Any())
            sql += Environment.NewLine + TabIn("(", tabs) + Environment.NewLine;
        else
            throw new QueryBuildingException($"Container '{container}' is empty, Disable it if you don't want it run'");

        var firstEntityWritten = false;
        foreach (var toWrite in toWriteOut)
        {
            if (firstEntityWritten)
                sql += Environment.NewLine +
                       TabIn(
                           GetSetOperationSql(container.Operation, syntaxHelper.DatabaseType) + Environment.NewLine +
                           Environment.NewLine, tabs);

            if (toWrite is AggregateConfiguration)
                sql += TabIn(sqlDictionary.Single(kvp => Equals(kvp.Key.CohortSet, toWrite)).Value, tabs);

            if (toWrite is CohortAggregateContainer sub)
                sql += WriteContainers(sub, syntaxHelper, sqlDictionary, tabs + 1);

            //we have now written the first thing at this level of recursion - all others will need to be separated by the OPERATION e.g. UNION
            firstEntityWritten = true;

            if (StopContainerWhenYouReach != null && StopContainerWhenYouReach.Equals(toWrite))
                if (tabs != 0)
                    throw new NotSupportedException(
                        "Stopping prematurely only works when the aggregate to stop at is in the top level container");
                else
                    break;
        }

        //if we outputted anything
        if (toWriteOut.Any())
            sql += Environment.NewLine + TabIn(")", tabs) + Environment.NewLine;

        return sql;
    }

    private bool IsEnabled(IOrderable arg) => IsEnabled(arg, ChildProvider);

    /// <summary>
    /// Objects are enabled if they do not support disabling (<see cref="IDisableable"/>) or are <see cref="IDisableable.IsDisabled"/> = false
    /// </summary>
    /// <returns></returns>
    public static bool IsEnabled(IOrderable arg, ICoreChildProvider childProvider)
    {
        var parentDisabled = childProvider.GetDescendancyListIfAnyFor(arg)?.Parents.Any(p => p is IDisableable
        {
            IsDisabled: true
        });

        //if a parent is disabled
        if (parentDisabled.HasValue && parentDisabled.Value)
            return false;

        // skip empty containers unless strict validation is enabled
        if (arg is CohortAggregateContainer container &&
            !UserSettings.StrictValidationForCohortBuilderContainers)
            if (!container.GetOrderedContents().Any())
                return false;

        //or you yourself are disabled
        return arg is not IDisableable { IsDisabled: true };
    }

    /// <summary>
    /// Returns the SQL keyword for the <paramref name="currentContainerOperation"/>
    /// </summary>
    /// <param name="currentContainerOperation"></param>
    /// <param name="dbType"></param>
    /// <returns></returns>
    protected virtual string GetSetOperationSql(SetOperation currentContainerOperation, DatabaseType dbType)
    {
        return currentContainerOperation switch
        {
            SetOperation.UNION => "UNION",
            SetOperation.INTERSECT => "INTERSECT",
            SetOperation.EXCEPT => dbType == DatabaseType.Oracle ? "MINUS" : "EXCEPT",
            _ => throw new ArgumentOutOfRangeException(nameof(currentContainerOperation), currentContainerOperation,
                null)
        };
    }

    private string BuildSql(CohortQueryBuilderDependency dependency, ParameterManager parameterManager)
    {
        //if we are fully cached on everything
        if (dependency.SqlFullyCached != null)
        {
            SetTargetServer(GetCacheServer(), " dependency is cached"); //run on the cache server
            CountOfCachedSubQueries++; //it is cached
            return dependency.SqlFullyCached.Use(parameterManager); //run the fully cached sql
        }

        switch (CacheUsageDecision)
        {
            case CacheUsage.MustUse:
                throw new QueryBuildingException(
                    $"Could not build final SQL because {dependency} is not fully cached and CacheUsageDecision is {CacheUsageDecision}");

            case CacheUsage.Opportunistic:

                //The cache and dataset are on the same server so run it
                SetTargetServer(DependenciesSingleServer.GetDistinctServer(), "data and cache are on the same server");
                return dependency.SqlPartiallyCached?.Use(parameterManager) ??
                       dependency.SqlCacheless.Use(parameterManager);
            case CacheUsage.AllOrNothing:

                //It's not fully cached so we have to run it entirely uncached
                SetTargetServer(DependenciesSingleServer.GetDistinctServer(),
                    "cache and data are on separate servers / access credentials and not all datasets are in the cache");
                return dependency.SqlCacheless.Use(parameterManager);
            default:
                throw new ArgumentOutOfRangeException();
        }
    }

    private DiscoveredServer GetCacheServer() => CacheServer.Discover(DataAccessContext.InternalDataProcessing).Server;

    private void BuildDependenciesSql(ISqlParameter[] globals)
    {
        foreach (var d in Dependencies)
            d.Build(this, globals, CancellationToken);
    }


    private CohortQueryBuilderDependency AddDependency(AggregateConfiguration cohortSet)
    {
        if (cohortSet.Catalogue.IsApiCall())
        {
            if (CacheManager == null) throw new Exception($"Caching must be enabled to execute API call '{cohortSet}'");

            if (!PluginCohortCompilers.Any(c => c.ShouldRun(cohortSet)))
                throw new Exception(
                    $"No PluginCohortCompilers claimed to support '{cohortSet}' in their ShouldRun method");
        }

        var join = ChildProvider.AllJoinUses.Where(j => j.AggregateConfiguration_ID == cohortSet.ID).ToArray();

        if (join.Length > 1)
            throw new NotSupportedException(
                $"There are {join.Length} joins configured to AggregateConfiguration {cohortSet}");

        var d = new CohortQueryBuilderDependency(cohortSet, join.SingleOrDefault(), ChildProvider,
            PluginCohortCompilers);
        _dependencies.Add(d);

        return d;
    }

    private void MakeCacheDecision()
    {
        if (CacheServer == null)
        {
            SetCacheUsage(CacheUsage.AllOrNothing, "there is no cache server");
        }
        else
        {
            _log.AppendLine($"Cache Server:{CacheServer.Server} (DatabaseType:{CacheServer.DatabaseType})");
            SetCacheUsage(CacheUsage.Opportunistic, "there is a cache server (so starting with Opportunistic)");
        }

        DependenciesSingleServer = new DataAccessPointCollection(true);

        foreach (var dependency in Dependencies)
        {
            _log.AppendLine($"Evaluating '{dependency.CohortSet}'");
            foreach (var dependantTable in dependency.CohortSet.Catalogue.GetTableInfoList(false))
                HandleDependency(dependency, false, dependantTable);

            if (dependency.JoinedTo != null)
            {
                _log.AppendLine($"Evaluating '{dependency.JoinedTo}'");
                foreach (var dependantTable in dependency.JoinedTo.Catalogue.GetTableInfoList(false))
                    HandleDependency(dependency, true, dependantTable);
            }
        }
    }

    private void HandleDependency(CohortQueryBuilderDependency dependency, bool isPatientIndexTable,
        ITableInfo dependantTable)
    {
        _log.AppendLine(
            $"Found dependant table '{dependantTable}' (Server:{dependantTable.Server} DatabaseType:{dependantTable.DatabaseType})");

        //if dependencies are on different servers / access credentials
        if (DependenciesSingleServer != null)
            if (!DependenciesSingleServer.TryAdd(dependantTable))
            {
                //we can no longer establish a consistent connection to all the dependencies
                _log.AppendLine($"Found problematic dependant table '{dependantTable}'");

                //if there's no cache server that's a problem!
                if (CacheServer == null)
                    throw new QueryBuildingException(
                        $"Table {dependantTable} is on a different server (or uses different access credentials) from previously seen dependencies and no QueryCache is configured");

                //there is a cache server, perhaps we can dodge 'dependantTable' by going to cache instead
                var canUseCacheForDependantTable =
                    (isPatientIndexTable ? dependency.SqlJoinableCached : dependency.SqlFullyCached)
                    != null;

                //can we go to the cache server instead?
                if (canUseCacheForDependantTable && DependenciesSingleServer.TryAdd(CacheServer))
                {
                    _log.AppendLine($"Avoided problematic dependant table '{dependantTable}' by using the cache");
                }
                else
                {
                    DependenciesSingleServer = null;

                    //there IS a cache so we now Must use it
                    if (CacheUsageDecision != CacheUsage.MustUse)
                        SetCacheUsage(CacheUsage.MustUse,
                            $"Table {dependantTable} is on a different server (or uses different access credentials) from previously seen dependencies.  Therefore the QueryCache MUST be used for all dependencies");
                }
            }

        if (DependenciesSingleServer != null &&
            CacheServer != null &&
            CacheUsageDecision == CacheUsage.Opportunistic)
        {
            //We can only do opportunistic joins if the Cache and Data server are on the same server
            var canCombine = DependenciesSingleServer.AddWouldBePossible(CacheServer);

            if (!canCombine)
                SetCacheUsage(CacheUsage.AllOrNothing,
                    "All datasets are on one server/access credentials while Cache is on a separate one");
        }
    }

    private void LogDependencies()
    {
        foreach (var d in Dependencies)
        {
            _log.AppendLine($"Dependency '{d}' is {d.DescribeCachedState()}");
            _log.AppendLine(
                $"Dependency '{d}' IsExtractionIdentifier column is {d.ExtractionIdentifierColumn?.GetRuntimeName() ?? "NULL"}");
        }
    }


    private void SetCacheUsage(CacheUsage value, string thereIsNoCacheServer)
    {
        CacheUsageDecision = value;
        _log.AppendLine($"Setting {nameof(CacheUsageDecision)} to {value} because {thereIsNoCacheServer}");
    }

    private void ThrowIfAlreadyBuilt()
    {
        if (_alreadyBuilt)
            throw new InvalidOperationException("Dependencies have already been built");

        _alreadyBuilt = true;
    }

    public static string TabIn(string str, int numberOfTabs)
    {
        if (string.IsNullOrWhiteSpace(str))
            return str;

        var tabs = new string('\t', numberOfTabs);
        return tabs + str.Replace(Environment.NewLine, Environment.NewLine + tabs);
    }
}

1	// Copyright (c) The University of Dundee 2018-2019
2	// This file is part of the Research Data Management Platform (RDMP).
3	// RDMP is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
4	// RDMP is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
5	// You should have received a copy of the GNU General Public License along with RDMP. If not, see <https://www.gnu.org/licenses/>.
6
7	using System;
8	using System.Collections.Generic;
9	using System.Linq;
10	using System.Text;
11	using System.Threading;
12	using FAnsi;
13	using FAnsi.Discovery;
14	using FAnsi.Discovery.QuerySyntax;
15	using Rdmp.Core.CohortCreation.Execution;
16	using Rdmp.Core.Curation.Data;
17	using Rdmp.Core.Curation.Data.Aggregation;
18	using Rdmp.Core.Curation.Data.Cohort;
19	using Rdmp.Core.MapsDirectlyToDatabaseTable;
20	using Rdmp.Core.Providers;
21	using Rdmp.Core.QueryBuilding.Parameters;
22	using Rdmp.Core.QueryCaching.Aggregation;
23	using Rdmp.Core.ReusableLibraryCode.DataAccess;
24	using Rdmp.Core.ReusableLibraryCode.Settings;
25
26	namespace Rdmp.Core.QueryBuilding;
27
28	/// <summary>
29	/// Builds a subset of a <see cref="CohortIdentificationConfiguration"/> e.g. a single <see cref="CohortAggregateContainer"/> (UNION / INTERSECT / EXCEPT) or a
30	/// cohort set. This includes identifying all <see cref="Dependencies"/> and resolving the dependencies servers / the <see cref="CacheServer"/> to determine
31	/// whether a valid query can be assembled from the sub-components and deciding where it can be run (e.g. should the query run on the cache server or the data server
32	/// or are they both the same server so query sections can be mixed depending on cache hit/miss for each bit).
33	/// </summary>
34	public class CohortQueryBuilderResult
35	{
36	public ExternalDatabaseServer CacheServer { get; }	1,450✔
37	public CachedAggregateConfigurationResultsManager CacheManager { get; }	854✔
38
39	public bool IsForContainer { get; private set; }	336✔
40	public ICoreChildProvider ChildProvider { get; }	1,504✔
41	public CohortQueryBuilderHelper Helper { get; }	×
42	public QueryBuilderCustomArgs Customise { get; }	454✔
43	public CancellationToken CancellationToken { get; }	430✔
44
45	private readonly StringBuilder _log = new();	336✔
46
47	/// <summary>
48	/// Log of all activities undertaken while building
49	/// </summary>
50	public string Log => _log.ToString();	264✔
51
52	/// <summary>
53	/// The allowable caching state based on the <see cref="Dependencies"/>, whether there is a
54	/// <see cref="CacheServer"/> and if they are on the same or separate servers from one another
55	/// </summary>
56	public CacheUsage CacheUsageDecision { get; private set; }	870✔
57
58	private List<CohortQueryBuilderDependency> _dependencies = new();	336✔
59	private bool _alreadyBuilt;
60
61	public IReadOnlyCollection<CohortQueryBuilderDependency> Dependencies => _dependencies;	1,810✔
62
63	/// <summary>
64	/// Only Populated after Building. If all <see cref="Dependencies"/> are on the same server as one another
65	/// then this will contain all tables that must be queried otherwise it will be null
66	/// </summary>
67	public DataAccessPointCollection DependenciesSingleServer { get; private set; }	2,118✔
68
69	/// <summary>
70	/// The final SQL that should be executed on the <see cref="TargetServer"/>
71	/// </summary>
72	public string Sql { get; private set; }	632✔
73
74	/// <summary>
75	/// The location at which the <see cref="Sql"/> should be run (may be a data server or a cache server or they may be one and the same!)
76	/// </summary>
77	public DiscoveredServer TargetServer { get; set; }	1,036✔
78
79	public IOrderable StopContainerWhenYouReach { get; set; }	590✔
80	public int CountOfSubQueries => Dependencies.Count;	316✔
81	public int CountOfCachedSubQueries { get; private set; }	420✔
82
83	public IReadOnlyCollection<IPluginCohortCompiler> PluginCohortCompilers { get; } =	444✔
84	Array.Empty<PluginCohortCompiler>().ToList().AsReadOnly();	336✔
85
86	/// <summary>
87	/// Creates a new result for a single <see cref="AggregateConfiguration"/> or <see cref="CohortAggregateContainer"/>
88	/// </summary>
89	/// <param name="cacheServer"></param>
90	/// <param name="childProvider"></param>
91	/// <param name="helper"></param>
92	/// <param name="customise"></param>
93	/// <param name="cancellationToken"></param>
94	public CohortQueryBuilderResult(ExternalDatabaseServer cacheServer, ICoreChildProvider childProvider,	336✔
95	CohortQueryBuilderHelper helper, QueryBuilderCustomArgs customise, CancellationToken cancellationToken)	336✔
96	{
97	CacheServer = cacheServer;	336✔
98	ChildProvider = childProvider;	336✔
99	Helper = helper;	336✔
100	Customise = customise;	336✔
101	CancellationToken = cancellationToken;	336✔
102
103	if (cacheServer != null)	336✔
104	{
105	CacheManager = new CachedAggregateConfigurationResultsManager(CacheServer);	154✔
106
107	try
108	{
109	PluginCohortCompilers = PluginCohortCompilerFactory.CreateAll();	154✔
110	}	154✔
111	catch (Exception ex)	×
112	{
113	throw new Exception("Failed to build list of IPluginCohortCompilers", ex);	×
114	}
115	}
116	}	336✔
117
118
119	public void BuildFor(CohortAggregateContainer container, ParameterManager parameterManager)
120	{
121	ThrowIfAlreadyBuilt();	146✔
122	IsForContainer = true;	146✔
123
124	_log.AppendLine($"Starting Build for {container}");	146✔
125	//gather dependencies
126	foreach (var cohortSet in ChildProvider.GetAllChildrenRecursively(container).OfType<AggregateConfiguration>()	772✔
127	.Where(IsEnabled).OrderBy(ac => ac.Order))	386✔
128	AddDependency(cohortSet);	240✔
129
130	if (!Dependencies.Any())	146✔
131	throw new QueryBuildingException(	2✔
132	$"There are no AggregateConfigurations under the SET container '{container}'");	2✔
133
134	LogDependencies();	144✔
135
136	BuildDependenciesSql(parameterManager.ParametersFoundSoFarInQueryGeneration[ParameterLevel.Global].ToArray());	144✔
137
138	MakeCacheDecision();	136✔
139
140	Sql = BuildSql(container, parameterManager);	136✔
141	}	134✔
142
143	public void BuildFor(AggregateConfiguration configuration, ParameterManager parameterManager)
144	{
145	ThrowIfAlreadyBuilt();	190✔
146	IsForContainer = false;	190✔
147
148	_log.AppendLine($"Starting Build for {configuration}");	190✔
149	var d = AddDependency(configuration);	190✔
150
151	LogDependencies();	190✔
152
153	BuildDependenciesSql(parameterManager.ParametersFoundSoFarInQueryGeneration[ParameterLevel.Global].ToArray());	190✔
154
155	MakeCacheDecision();	182✔
156
157
158	Sql = BuildSql(d, parameterManager);	182✔
159	}	182✔
160
161	private string BuildSql(CohortAggregateContainer container, ParameterManager parameterManager)
162	{
163	Dictionary<CohortQueryBuilderDependency, string> sqlDictionary;
164
165	//if we are fully cached on everything
166	if (Dependencies.All(d => d.SqlFullyCached != null))	282✔
167	{
168	SetTargetServer(GetCacheServer(), "all dependencies are fully cached"); //run on the cache server	52✔
169
170	//all are cached
171	CountOfCachedSubQueries = CountOfSubQueries;	52✔
172
173	sqlDictionary =	52✔
174	Dependencies.ToDictionary(k => k,	62✔
175	v => v.SqlFullyCached.Use(parameterManager)); //run the fully cached sql	114✔
176	}
177	else
178	{
179	var uncached =	84✔
180	$"CacheUsageDecision is {CacheUsageDecision} and the following were not cached:{string.Join(Environment.NewLine, Dependencies.Where(d => d.SqlFullyCached == null))}";	254✔
181
182	switch (CacheUsageDecision)	84!
183	{
184	case CacheUsage.MustUse:
185	throw new QueryBuildingException(	×
186	$"Could not build final SQL because some queries are not fully cached and {uncached}");	×
187
188	case CacheUsage.Opportunistic:
189
190	//The cache and dataset are on the same server so run it
191	SetTargetServer(DependenciesSingleServer.GetDistinctServer(),	6✔
192	$"not all dependencies are cached while {uncached}");	6✔
193
194	CountOfCachedSubQueries = Dependencies.Count(d => d.SqlFullyCached != null);	12✔
195
196	sqlDictionary =	6✔
197	Dependencies.ToDictionary(k => k,	6✔
198	v => v.SqlFullyCached?.Use(parameterManager) ??	12!
199	v.SqlPartiallyCached?.Use(parameterManager) ??	12✔
200	v.SqlCacheless.Use(parameterManager)); //run the fully cached sql	12✔
201	break;	6✔
202
203	case CacheUsage.AllOrNothing:
204
205	//It's not fully cached so we have to run it entirely uncached
206	SetTargetServer(DependenciesSingleServer.GetDistinctServer(),	78✔
207	$"not all dependencies are cached while {uncached}");	78✔
208
209	//cannot use any of the caches because it's all or nothing
210	CountOfCachedSubQueries = 0;	78✔
211	sqlDictionary =	78✔
212	Dependencies.ToDictionary(k => k,	164✔
213	v => v.SqlCacheless.Use(parameterManager)); //run the fully uncached sql	242✔
214	break;	78✔
215	default:
216	throw new ArgumentOutOfRangeException();	×
217	}
218	}
219
220	return WriteContainers(container, TargetServer.GetQuerySyntaxHelper(), sqlDictionary, 0);	136✔
221	}
222
223	private void SetTargetServer(DiscoveredServer target, string reason)
224	{
225	if (TargetServer != null)	318!
226	throw new InvalidOperationException("You are only supposed to pick a target server once");	×
227
228	TargetServer = target;	318✔
229	_log.AppendLine($"Picked TargetServer as {target} because {reason}");	318✔
230	}	318✔
231
232	private string WriteContainers(CohortAggregateContainer container, IQuerySyntaxHelper syntaxHelper,
233	Dictionary<CohortQueryBuilderDependency, string> sqlDictionary, int tabs)
234	{
235	var sql = "";	156✔
236
237	//Things we need to output
238	var toWriteOut = container.GetOrderedContents().Where(IsEnabled).ToArray();	156✔
239
240	if (toWriteOut.Any())	156!
241	sql += Environment.NewLine + TabIn("(", tabs) + Environment.NewLine;	156✔
242	else
243	throw new QueryBuildingException($"Container '{container}' is empty, Disable it if you don't want it run'");	×
244
245	var firstEntityWritten = false;	156✔
246	foreach (var toWrite in toWriteOut)	806✔
247	{
248	if (firstEntityWritten)	250✔
249	sql += Environment.NewLine +	94✔
250	TabIn(	94✔
251	GetSetOperationSql(container.Operation, syntaxHelper.DatabaseType) + Environment.NewLine +	94✔
252	Environment.NewLine, tabs);	94✔
253
254	if (toWrite is AggregateConfiguration)	250✔
255	sql += TabIn(sqlDictionary.Single(kvp => Equals(kvp.Key.CohortSet, toWrite)).Value, tabs);	730✔
256
257	if (toWrite is CohortAggregateContainer sub)	248✔
258	sql += WriteContainers(sub, syntaxHelper, sqlDictionary, tabs + 1);	20✔
259
260	//we have now written the first thing at this level of recursion - all others will need to be separated by the OPERATION e.g. UNION
261	firstEntityWritten = true;	248✔
262
263	if (StopContainerWhenYouReach != null && StopContainerWhenYouReach.Equals(toWrite))	248✔
264	if (tabs != 0)	4!
265	throw new NotSupportedException(	×
266	"Stopping prematurely only works when the aggregate to stop at is in the top level container");	×
267	else
268	break;
269	}
270
271	//if we outputted anything
272	if (toWriteOut.Any())	154✔
273	sql += Environment.NewLine + TabIn(")", tabs) + Environment.NewLine;	154✔
274
275	return sql;	154✔
276	}
277
278	private bool IsEnabled(IOrderable arg) => IsEnabled(arg, ChildProvider);	498✔
279
280	/// <summary>
281	/// Objects are enabled if they do not support disabling (<see cref="IDisableable"/>) or are <see cref="IDisableable.IsDisabled"/> = false
282	/// </summary>
283	/// <returns></returns>
284	public static bool IsEnabled(IOrderable arg, ICoreChildProvider childProvider)
285	{
286	var parentDisabled = childProvider.GetDescendancyListIfAnyFor(arg)?.Parents.Any(p => p is IDisableable	2,070✔
287	{	2,070✔
288	IsDisabled: true	2,070✔
289	});	2,070✔
290
291	//if a parent is disabled
292	if (parentDisabled.HasValue && parentDisabled.Value)	498!
293	return false;	×
294
295	// skip empty containers unless strict validation is enabled
296	if (arg is CohortAggregateContainer container &&	498!
297	!UserSettings.StrictValidationForCohortBuilderContainers)	498✔
298	if (!container.GetOrderedContents().Any())	×
299	return false;	×
300
301	//or you yourself are disabled
302	return arg is not IDisableable { IsDisabled: true };	498!
303	}
304
305	/// <summary>
306	/// Returns the SQL keyword for the <paramref name="currentContainerOperation"/>
307	/// </summary>
308	/// <param name="currentContainerOperation"></param>
309	/// <param name="dbType"></param>
310	/// <returns></returns>
311	protected virtual string GetSetOperationSql(SetOperation currentContainerOperation, DatabaseType dbType)
312	{
313	return currentContainerOperation switch	94!
314	{	94✔
315	SetOperation.UNION => "UNION",	32✔
316	SetOperation.INTERSECT => "INTERSECT",	4✔
317	SetOperation.EXCEPT => dbType == DatabaseType.Oracle ? "MINUS" : "EXCEPT",	58!
318	_ => throw new ArgumentOutOfRangeException(nameof(currentContainerOperation), currentContainerOperation,	×
319	null)	×
320	};	94✔
321	}
322
323	private string BuildSql(CohortQueryBuilderDependency dependency, ParameterManager parameterManager)
324	{
325	//if we are fully cached on everything
326	if (dependency.SqlFullyCached != null)	182✔
327	{
328	SetTargetServer(GetCacheServer(), " dependency is cached"); //run on the cache server	10✔
329	CountOfCachedSubQueries++; //it is cached	10✔
330	return dependency.SqlFullyCached.Use(parameterManager); //run the fully cached sql	10✔
331	}
332
333	switch (CacheUsageDecision)	172!
334	{
335	case CacheUsage.MustUse:
336	throw new QueryBuildingException(	×
337	$"Could not build final SQL because {dependency} is not fully cached and CacheUsageDecision is {CacheUsageDecision}");	×
338
339	case CacheUsage.Opportunistic:
340
341	//The cache and dataset are on the same server so run it
342	SetTargetServer(DependenciesSingleServer.GetDistinctServer(), "data and cache are on the same server");	52✔
343	return dependency.SqlPartiallyCached?.Use(parameterManager) ??	52✔
344	dependency.SqlCacheless.Use(parameterManager);	52✔
345	case CacheUsage.AllOrNothing:
346
347	//It's not fully cached so we have to run it entirely uncached
348	SetTargetServer(DependenciesSingleServer.GetDistinctServer(),	120✔
349	"cache and data are on separate servers / access credentials and not all datasets are in the cache");	120✔
350	return dependency.SqlCacheless.Use(parameterManager);	120✔
351	default:
352	throw new ArgumentOutOfRangeException();	×
353	}
354	}
355
356	private DiscoveredServer GetCacheServer() => CacheServer.Discover(DataAccessContext.InternalDataProcessing).Server;	62✔
357
358	private void BuildDependenciesSql(ISqlParameter[] globals)
359	{
360	foreach (var d in Dependencies)	1,512✔
361	d.Build(this, globals, CancellationToken);	430✔
362	}	318✔
363
364
365	private CohortQueryBuilderDependency AddDependency(AggregateConfiguration cohortSet)
366	{
367	if (cohortSet.Catalogue.IsApiCall())	430✔
368	{
369	if (CacheManager == null) throw new Exception($"Caching must be enabled to execute API call '{cohortSet}'");	14!
370
371	if (!PluginCohortCompilers.Any(c => c.ShouldRun(cohortSet)))	28!
372	throw new Exception(	×
373	$"No PluginCohortCompilers claimed to support '{cohortSet}' in their ShouldRun method");	×
374	}
375
376	var join = ChildProvider.AllJoinUses.Where(j => j.AggregateConfiguration_ID == cohortSet.ID).ToArray();	1,746✔
377
378	if (join.Length > 1)	430!
379	throw new NotSupportedException(	×
380	$"There are {join.Length} joins configured to AggregateConfiguration {cohortSet}");	×
381
382	var d = new CohortQueryBuilderDependency(cohortSet, join.SingleOrDefault(), ChildProvider,	430✔
383	PluginCohortCompilers);	430✔
384	_dependencies.Add(d);	430✔
385
386	return d;	430✔
387	}
388
389	private void MakeCacheDecision()
390	{
391	if (CacheServer == null)	318✔
392	{
393	SetCacheUsage(CacheUsage.AllOrNothing, "there is no cache server");	174✔
394	}
395	else
396	{
397	_log.AppendLine($"Cache Server:{CacheServer.Server} (DatabaseType:{CacheServer.DatabaseType})");	144✔
398	SetCacheUsage(CacheUsage.Opportunistic, "there is a cache server (so starting with Opportunistic)");	144✔
399	}
400
401	DependenciesSingleServer = new DataAccessPointCollection(true);	318✔
402
403	foreach (var dependency in Dependencies)	1,464✔
404	{
405	_log.AppendLine($"Evaluating '{dependency.CohortSet}'");	414✔
406	foreach (var dependantTable in dependency.CohortSet.Catalogue.GetTableInfoList(false))	1,632✔
407	HandleDependency(dependency, false, dependantTable);	402✔
408
409	if (dependency.JoinedTo != null)	414✔
410	{
411	_log.AppendLine($"Evaluating '{dependency.JoinedTo}'");	62✔
412	foreach (var dependantTable in dependency.JoinedTo.Catalogue.GetTableInfoList(false))	240✔
413	HandleDependency(dependency, true, dependantTable);	58✔
414	}
415	}
416	}	318✔
417
418	private void HandleDependency(CohortQueryBuilderDependency dependency, bool isPatientIndexTable,
419	ITableInfo dependantTable)
420	{
421	_log.AppendLine(	460✔
422	$"Found dependant table '{dependantTable}' (Server:{dependantTable.Server} DatabaseType:{dependantTable.DatabaseType})");	460✔
423
424	//if dependencies are on different servers / access credentials
425	if (DependenciesSingleServer != null)	460✔
426	if (!DependenciesSingleServer.TryAdd(dependantTable))	460✔
427	{
428	//we can no longer establish a consistent connection to all the dependencies
429	_log.AppendLine($"Found problematic dependant table '{dependantTable}'");	4✔
430
431	//if there's no cache server that's a problem!
432	if (CacheServer == null)	4!
433	throw new QueryBuildingException(	×
434	$"Table {dependantTable} is on a different server (or uses different access credentials) from previously seen dependencies and no QueryCache is configured");	×
435
436	//there is a cache server, perhaps we can dodge 'dependantTable' by going to cache instead
437	var canUseCacheForDependantTable =	4!
438	(isPatientIndexTable ? dependency.SqlJoinableCached : dependency.SqlFullyCached)	4✔
439	!= null;	4✔
440
441	//can we go to the cache server instead?
442	if (canUseCacheForDependantTable && DependenciesSingleServer.TryAdd(CacheServer))	4!
443	{
444	_log.AppendLine($"Avoided problematic dependant table '{dependantTable}' by using the cache");	4✔
445	}
446	else
447	{
448	DependenciesSingleServer = null;	×
449
450	//there IS a cache so we now Must use it
451	if (CacheUsageDecision != CacheUsage.MustUse)	×
452	SetCacheUsage(CacheUsage.MustUse,	×
453	$"Table {dependantTable} is on a different server (or uses different access credentials) from previously seen dependencies. Therefore the QueryCache MUST be used for all dependencies");	×
454	}
455	}
456
457	if (DependenciesSingleServer != null &&	460✔
458	CacheServer != null &&	460✔
459	CacheUsageDecision == CacheUsage.Opportunistic)	460✔
460	{
461	//We can only do opportunistic joins if the Cache and Data server are on the same server
462	var canCombine = DependenciesSingleServer.AddWouldBePossible(CacheServer);	160✔
463
464	if (!canCombine)	160✔
465	SetCacheUsage(CacheUsage.AllOrNothing,	36✔
466	"All datasets are on one server/access credentials while Cache is on a separate one");	36✔
467	}
468	}	460✔
469
470	private void LogDependencies()
471	{
472	foreach (var d in Dependencies)	1,528✔
473	{
474	_log.AppendLine($"Dependency '{d}' is {d.DescribeCachedState()}");	430✔
475	_log.AppendLine(	430✔
476	$"Dependency '{d}' IsExtractionIdentifier column is {d.ExtractionIdentifierColumn?.GetRuntimeName() ?? "NULL"}");	430✔
477	}
478	}	334✔
479
480
481	private void SetCacheUsage(CacheUsage value, string thereIsNoCacheServer)
482	{
483	CacheUsageDecision = value;	354✔
484	_log.AppendLine($"Setting {nameof(CacheUsageDecision)} to {value} because {thereIsNoCacheServer}");	354✔
485	}	354✔
486
487	private void ThrowIfAlreadyBuilt()
488	{
489	if (_alreadyBuilt)	336!
490	throw new InvalidOperationException("Dependencies have already been built");	×
491
492	_alreadyBuilt = true;	336✔
493	}	336✔
494
495	public static string TabIn(string str, int numberOfTabs)
496	{
497	if (string.IsNullOrWhiteSpace(str))	632!
498	return str;	×
499
500	var tabs = new string('\t', numberOfTabs);	632✔
501	return tabs + str.Replace(Environment.NewLine, Environment.NewLine + tabs);	632✔
502	}
503	}

HicServices / RDMP / 6237307473

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous