add cohort temp table

Author: JFriel

Rdmp.Core/DataExport/DataExtraction/Pipeline/Sources/ExecuteDatasetExtractionSource.cs

@@ -7,10 +7,12 @@
 using System;
 using System.Collections.Generic;
 using System.Data;
+using System.Data.Common;
 using System.Diagnostics;
 using System.Linq;
 using System.Threading.Tasks;
 using FAnsi;
+using FAnsi.Discovery;
 using FAnsi.Discovery.QuerySyntax;
 using Rdmp.Core.Curation.Data;
 using Rdmp.Core.DataExport.Data;
@@ -99,13 +101,18 @@ public class ExecuteDatasetExtractionSource : IPluginDataFlowSource<DataTable>,
 
     private DbDataCommandDataFlowSource _hostedSource;
 
+    private IExternalCohortTable _externalCohortTable;
+    private string _whereSQL;
+    private DbConnection _con;
+    private string _uuid;
     protected virtual void Initialize(ExtractDatasetCommand request)
     {
         Request = request;
 
         if (request == ExtractDatasetCommand.EmptyCommand)
             return;
-
+        _externalCohortTable = request.ExtractableCohort.ExternalCohortTable;
+        _whereSQL = request.ExtractableCohort.WhereSQL();
         _timeSpentValidating = new Stopwatch();
         _timeSpentCalculatingDISTINCT = new Stopwatch();
         _timeSpentBuckettingDates = new Stopwatch();
@@ -156,6 +163,32 @@ private void Initialize(ExtractGlobalsCommand request)
 
     private RowPeeker _peeker = new();
 
+    private static Random random = new Random();
+
+    private static string RandomString(int length)
+    {
+        const string chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
+        return new string(Enumerable.Repeat(chars, length)
+            .Select(s => s[random.Next(s.Length)]).ToArray());
+    }
+
+    private void CreateCohortTempTable(DbConnection con)
+    {
+        var db = _externalCohortTable.Discover();
+        _uuid = $"#{RandomString(24)}";
+        var sql = $"""
+            SELECT *
+            INTO {_uuid}
+            FROM(
+            SELECT * FROM {_externalCohortTable.TableName}
+            WHERE {_whereSQL}
+            ) as cohortTempTable
+        
+            """;
+        using var cmd = db.Server.GetCommand(sql, con);
+        cmd.ExecuteNonQuery();
+    }
+
     public virtual DataTable GetChunk(IDataLoadEventListener listener, GracefulCancellationToken cancellationToken)
     {
         // we are in the Global Commands case, let's return an empty DataTable (not null)
@@ -185,15 +218,18 @@ public virtual DataTable GetChunk(IDataLoadEventListener listener, GracefulCance
 
         if (_hostedSource == null)
         {
+            _con = DatabaseCommandHelper.GetConnection(Request.GetDistinctLiveDatabaseServer().Builder);
+            _con.Open();
+            CreateCohortTempTable(_con);
             StartAudit(Request.QueryBuilder.SQL);
 
             if (Request.DatasetBundle.DataSet.DisableExtraction)
                 throw new Exception(
                     $"Cannot extract {Request.DatasetBundle.DataSet} because DisableExtraction is set to true");
-            var x = GetCommandSQL(listener);
+
             _hostedSource = new DbDataCommandDataFlowSource(GetCommandSQL(listener),
                 $"ExecuteDatasetExtraction {Request.DatasetBundle.DataSet}",
-                Request.GetDistinctLiveDatabaseServer().Builder,
+               _con,
                 ExecutionTimeout)
             {
                 // If we are running in batches then always allow empty extractions
@@ -451,7 +487,10 @@ private string GetCommandSQL(IDataLoadEventListener listener)
 
         listener.OnNotify(this, new NotifyEventArgs(ProgressEventType.Information,
             $"/*Decided on extraction SQL:*/{Environment.NewLine}{sql}"));
-
+        if(_uuid is not null)
+        {
+            sql = sql.Replace(_externalCohortTable.TableName, _uuid);
+        }
         return sql;
     }
 

Rdmp.Core/DataLoad/Engine/Pipeline/Sources/DbDataCommandDataFlowSource.cs

@@ -49,6 +49,17 @@ public DbDataCommandDataFlowSource(string sql, string taskBeingPerformed, DbConn
         BatchSize = 10000;
     }
 
+    public DbDataCommandDataFlowSource(string sql, string taskBeingPerformed, DbConnection con,
+        int timeout)
+    {
+        Sql = sql;
+        _taskBeingPerformed = taskBeingPerformed;
+        _con = con;
+        _timeout = timeout;
+
+        BatchSize = 10000;
+    }
+
     private int _numberOfColumns;
 
     private bool _firstChunk = true;
@@ -59,8 +70,11 @@ public DataTable GetChunk(IDataLoadEventListener job, GracefulCancellationToken
     {
         if (_reader == null)
         {
-            _con = DatabaseCommandHelper.GetConnection(_builder);
-            _con.Open();
+            _con = _con ==null?DatabaseCommandHelper.GetConnection(_builder):_con;
+            if(_con != null && _con.State == ConnectionState.Closed)
+{
+                _con.Open();
+            }
 
             job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Information,
                 $"Running SQL:{Environment.NewLine}{Sql}"));
@@ -209,11 +223,13 @@ private void CloseReader(IDataLoadEventListener listener)
     public DataTable TryGetPreview()
     {
         var chunk = new DataTable();
-        using var con = DatabaseCommandHelper.GetConnection(_builder);
-        con.Open();
-        using var da = DatabaseCommandHelper.GetDataAdapter(DatabaseCommandHelper.GetCommand(Sql, con));
+        _con = _con == null ? DatabaseCommandHelper.GetConnection(_builder) : _con;
+        if (_con != null && _con.State == ConnectionState.Closed)
+        {
+            _con.Open();
+        }
+        using var da = DatabaseCommandHelper.GetDataAdapter(DatabaseCommandHelper.GetCommand(Sql, _con));
         var read = da.Fill(0, 100, chunk);
-
         return read == 0 ? null : chunk;
     }
 }

Rdmp.Core/QueryBuilding/ExtractionQueryBuilder.cs

@@ -6,8 +6,11 @@
 
 using System;
 using System.Collections.Generic;
+using System.Data.Common;
 using System.Linq;
+using FAnsi.Discovery;
 using FAnsi.Discovery.QuerySyntax;
+using NPOI.SS.Formula.Functions;
 using Rdmp.Core.CohortCommitting;
 using Rdmp.Core.DataExport;
 using Rdmp.Core.DataExport.Data;
@@ -122,7 +125,11 @@ public QueryBuilder GetSQLCommandForFullExtractionSet(ExtractDatasetCommand requ
         var externalCohortTable =
             _repository.GetObjectByID<ExternalCohortTable>(request.ExtractableCohort.ExternalCohortTable_ID);
         //todo here is where we want to create a temp table
-        CreateCohortTempTable(externalCohortTable);
+
+        //var db = externalCohortTable.Discover();
+        //var con = db.Server.GetConnection();
+        //con.Open();
+        //var tempTable = CreateCohortTempTable(externalCohortTable, request.ExtractableCohort.WhereSQL(),con.db);
         if (request.ExtractableCohort != null)
         {
             //the JOIN with the cohort table:
@@ -143,7 +150,31 @@ public QueryBuilder GetSQLCommandForFullExtractionSet(ExtractDatasetCommand requ
         return queryBuilder;
     }
 
-    private void CreateCohortTempTable(ExternalCohortTable externalCohortTable) { }
+    //private static Random random = new Random();
+
+    //private static string RandomString(int length)
+    //{
+    //    const string chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
+    //    return new string(Enumerable.Repeat(chars, length)
+    //        .Select(s => s[random.Next(s.Length)]).ToArray());
+    //}
+
+    //private string CreateCohortTempTable(ExternalCohortTable externalCohortTable, string whereSQL,DbConnection con, DiscoveredDatabase db)
+    //{
+    //    var uuid = $"#{RandomString(24)}";
+    //    var sql = $"""
+    //        SELECT *
+    //        INTO {uuid}
+    //        FROM(
+    //        SELECT * FROM {externalCohortTable.TableName}
+    //        WHERE {whereSQL}
+    //        ) as cohortTempTable
+        
+    //        """;
+    //            using var cmd = db.Server.GetCommand(sql, con);
+    //    cmd.ExecuteNonQuery();
+    //    return uuid;
+    //}
 
     private void HandleBatching(ExtractDatasetCommand request, QueryBuilder queryBuilder,
         IQuerySyntaxHelper syntaxHelper)