Merge pull request #348 from DataObjects-NET/6.0-batching-problem

fix batching problem of skipping some tasks

Author: alex-kulakov

ChangeLog/6.0.12_dev.txt

@@ -1,3 +1,4 @@
 [main] Addressed DataTypeCollection.Add method issue of wrong adding storage-specifid types to the collection
 [main] Addressed rare issue of entities expire management when SessionOption.NonTransactionalReads is on
+[main] Fixed issue of skipping persist or query tasks during batching
 [sqlserver] Sql error messages for British English are correctly parsed

Orm/Xtensive.Orm.Tests/Storage/CommandProcessing/BatchingCommandProcessorParametersManagement.cs

@@ -602,13 +602,16 @@ public void InsertTest01()
       using (var session = Domain.OpenSession())
       using (var counter = new CommandCounter(session))
       using (var transaction = session.OpenTransaction()) {
+        var before = session.Query.All<ALotOfFieldsEntityValid>().Count();
+
         _ = new ALotOfFieldsEntityValid();
 
         using (counter.Attach()) {
           Assert.DoesNotThrow(() => session.SaveChanges());
         }
 
         Assert.That(counter.Count, Is.EqualTo(1));
+        Assert.That(session.Query.All<ALotOfFieldsEntityValid>().Count(), Is.EqualTo(before + 1));
       }
     }
 
@@ -618,6 +621,8 @@ public void InsertTest02()
       using (var session = Domain.OpenSession())
       using (var counter = new CommandCounter(session))
       using (var transaction = session.OpenTransaction()) {
+        var before = session.Query.All<ALotOfFieldsEntityValid>().Count();
+
         _ = new ALotOfFieldsEntityValid();
         _ = new ALotOfFieldsEntityValid();
         _ = new ALotOfFieldsEntityValid();
@@ -631,6 +636,8 @@ public void InsertTest02()
         var expectedCommandCount = Math.Ceiling(
           Math.Ceiling(4 * Domain.Model.Types[type].Fields.Count / (decimal) StorageLimit));
         Assert.That(counter.Count, Is.EqualTo(expectedCommandCount));
+
+        Assert.That(session.Query.All<ALotOfFieldsEntityValid>().Count(), Is.EqualTo(before + 4));
       }
     }
 
@@ -849,11 +856,13 @@ public void UpdateRegularTest01()
     }
 
     [Test]
-    public void PartitialExecutionAllowedTest01()
+    public void PartialExecutionAllowedTest01()
     {
       using (var session = Domain.OpenSession())
       using (var counter = new CommandCounter(session))
       using (var transaction = session.OpenTransaction()) {
+        var countBefore = session.Query.All<ALotOfFieldsEntityValid>().Count();
+
         _ = new ALotOfFieldsEntityValid();
         _ = new ALotOfFieldsEntityValid();
         _ = new ALotOfFieldsEntityValid();
@@ -867,15 +876,44 @@ public void PartitialExecutionAllowedTest01()
 
         counter.Reset();
         using (counter.Attach()) {
-          Assert.That(session.Query.All<ALotOfFieldsEntityValid>().Count(), Is.EqualTo(5));
+          Assert.That(session.Query.All<ALotOfFieldsEntityValid>().Count(), Is.EqualTo(countBefore + 3));
         }
 
         Assert.That(counter.Count, Is.EqualTo(2));
       }
     }
 
     [Test]
-    public void PartitialExecutionAllowedTest02()
+    public async Task PartialExecutionAllowedAsyncTest01()
+    {
+      using (var session = Domain.OpenSession())
+      using (var counter = new CommandCounter(session))
+      using (var transaction = session.OpenTransaction()) {
+        var countBefore = session.Query.All<ALotOfFieldsEntityValid>().Count();
+
+        _ = new ALotOfFieldsEntityValid();
+        _ = new ALotOfFieldsEntityValid();
+        _ = new ALotOfFieldsEntityValid();
+
+        //persist by query causes allowPartialExecution = true;
+        using (counter) {
+          session.Persist(PersistReason.Query);
+        }
+
+        Assert.That(counter.Count, Is.EqualTo(0));
+
+        counter.Reset();
+        using (counter.Attach()) {
+          var result = (await session.Query.All<ALotOfFieldsEntityValid>().AsAsync()).ToArray().Length;
+          Assert.That(result, Is.EqualTo(countBefore + 3));
+        }
+
+        Assert.That(counter.Count, Is.EqualTo(2));
+      }
+    }
+
+    [Test]
+    public void PartialExecutionAllowedTest02()
     {
       using (var session = Domain.OpenSession())
       using (var counter = new CommandCounter(session))
@@ -901,33 +939,76 @@ public void PartitialExecutionAllowedTest02()
         Assert.That(counter.Count, Is.EqualTo(1));
         counter.Reset();
         using (counter.Attach()) {
-          Assert.That(session.Query.All<NormalAmountOfFieldsEntity>().Count(), Is.EqualTo(26));
+          Assert.That(session.Query.All<NormalAmountOfFieldsEntity>().Count(), Is.EqualTo(batchSize + 1));
         }
         Assert.That(counter.Count, Is.EqualTo(1));
       }
     }
 
     [Test]
-    public void PartialExecutionDeniedTest01()
+    public async Task PartialExecutionAllowedAsyncTest02()
     {
       using (var session = Domain.OpenSession())
       using (var counter = new CommandCounter(session))
       using (var transaction = session.OpenTransaction()) {
-        _ = new ALotOfFieldsEntityValid();
-        _ = new ALotOfFieldsEntityValid();
-        _ = new ALotOfFieldsEntityValid();
+        var batchSize = session.Configuration.BatchSize;
+        var currentBatchCapacity = batchSize;
 
-        //manual persist causes allowPartialExecution = false;
+        Console.WriteLine(batchSize);
+        // one complete batch;
+        while (currentBatchCapacity > 0) {
+          _ = new NormalAmountOfFieldsEntity();
+          currentBatchCapacity--;
+        }
+
+        // extra task to have extra batch
+        _ = new NormalAmountOfFieldsEntity();
+
+        //persist by query causes allowPartialExecution = true;
         using (counter.Attach()) {
-          session.SaveChanges();
+          session.Persist(PersistReason.Query);
         }
 
+        Assert.That(counter.Count, Is.EqualTo(1));
+        counter.Reset();
+        using (counter.Attach()) {
+          var result = (await session.Query.All<NormalAmountOfFieldsEntity>().AsAsync()).ToArray().Length;
+          Assert.That(result, Is.EqualTo(batchSize + 1));
+        }
+        Assert.That(counter.Count, Is.EqualTo(1));
+      }
+    }
+
+    [Test]
+    public void PartialExecutionAllowedTest03()
+    {
+      using (var session = Domain.OpenSession())
+      using (var counter = new CommandCounter(session))
+      using (var transaction = session.OpenTransaction()) {
+        var batchSize = session.Configuration.BatchSize;
+        var currentBatchCapacity = batchSize;
+
+        Console.WriteLine(batchSize);
+        // one complete batch;
+        while (currentBatchCapacity > 0) {
+          _ = new NormalAmountOfFieldsEntity();
+          currentBatchCapacity--;
+        }
+
+        // extra task to have extra batch
+        _ = new NormalAmountOfFieldsEntity();
+
+        //persist by query causes allowPartialExecution = true;
+        using (counter.Attach()) {
+          var result = session.Query.All<NormalAmountOfFieldsEntity>().ToArray();
+          Assert.That(result.Length, Is.EqualTo(batchSize + 1));
+        }
         Assert.That(counter.Count, Is.EqualTo(2));
       }
     }
 
     [Test]
-    public void PartialExecutionDeniedTest02()
+    public async Task PartialExecutionAllowedAsyncTest03()
     {
       using (var session = Domain.OpenSession())
       using (var counter = new CommandCounter(session))
@@ -947,14 +1028,45 @@ public void PartialExecutionDeniedTest02()
 
         //persist by query causes allowPartialExecution = true;
         using (counter.Attach()) {
-          session.Query.All<NormalAmountOfFieldsEntity>().Run();
+          var result = (await session.Query.All<NormalAmountOfFieldsEntity>().AsAsync()).ToArray();
+          Assert.That(result.Length, Is.EqualTo(batchSize + 1));
         }
         Assert.That(counter.Count, Is.EqualTo(2));
       }
     }
 
     [Test]
-    public async Task PartialExecutionDeniedAsyncTest02()
+    public void PartialExecutionAllowedTest04()
+    {
+      using (var session = Domain.OpenSession())
+      using (var counter = new CommandCounter(session))
+      using (var transaction = session.OpenTransaction()) {
+        var countBefore = session.Query.All<ALotOfFieldsEntityValid>().Count();
+
+        var batchSize = session.Configuration.BatchSize;
+        var currentBatchCapacity = batchSize;
+
+        Console.WriteLine(batchSize);
+        // one complete batch;
+        while (currentBatchCapacity > 0) {
+          _ = new ALotOfFieldsEntityValid();
+          currentBatchCapacity--;
+        }
+
+        // extra task to have extra batch
+        _ = new ALotOfFieldsEntityValid();
+
+        //persist by query causes allowPartialExecution = true;
+        using (counter.Attach()) {
+          var result = session.Query.All<ALotOfFieldsEntityValid>().ToArray();
+          Assert.That(result.Length, Is.EqualTo(countBefore + batchSize + 1));
+        }
+        Assert.That(counter.Count, Is.EqualTo(13));
+      }
+    }
+
+    [Test]
+    public async Task PartialExecutionAllowedAsyncTest04()
     {
       using (var session = Domain.OpenSession())
       using (var counter = new CommandCounter(session))
@@ -974,8 +1086,51 @@ public async Task PartialExecutionDeniedAsyncTest02()
 
         //persist by query causes allowPartialExecution = true;
         using (counter.Attach()) {
-          (await session.Query.All<NormalAmountOfFieldsEntity>().AsAsync()).Run();
+          var result = (await session.Query.All<NormalAmountOfFieldsEntity>().AsAsync()).ToArray();
+          Assert.That(result.Length, Is.EqualTo(batchSize + 1));
+        }
+        Assert.That(counter.Count, Is.EqualTo(2));
+      }
+    }
+
+    [Test]
+    public void PartialExecutionAllowedTest05()
+    {
+      Require.ProviderIs(StorageProvider.SqlServer);
+
+      var sessionOf25ItemsBatch = new SessionConfiguration(WellKnown.Sessions.Default, SessionOptions.Default | SessionOptions.AutoActivation) { BatchSize = 25 };
+      using (var session = Domain.OpenSession(sessionOf25ItemsBatch))
+      using (var tx = session.OpenTransaction()) {
+        Assert.AreEqual(0, session.Query.All<OneHundredFieldsEntity>().Count());
+
+        for (var i = 0; i < session.Configuration.BatchSize; i++) {
+          var item = new OneHundredFieldsEntity();
+
+          for (var j = 1; j < 98; j++) {
+            item["Value" + j] = i;
+          }
+        }
+
+        var count = session.Query.All<OneHundredFieldsEntity>().ToArray().Length;
+        Assert.AreEqual(session.Configuration.BatchSize, count);
+      }
+    }
+
+    [Test]
+    public void PartialExecutionDeniedTest01()
+    {
+      using (var session = Domain.OpenSession())
+      using (var counter = new CommandCounter(session))
+      using (var transaction = session.OpenTransaction()) {
+        _ = new ALotOfFieldsEntityValid();
+        _ = new ALotOfFieldsEntityValid();
+        _ = new ALotOfFieldsEntityValid();
+
+        //manual persist causes allowPartialExecution = false;
+        using (counter.Attach()) {
+          session.SaveChanges();
         }
+
         Assert.That(counter.Count, Is.EqualTo(2));
       }
     }

Orm/Xtensive.Orm.Tests/Storage/CommandProcessing/ParametersManagementModel.cs

@@ -107,6 +107,15 @@ public class ALotOfFieldsEntityVersionized : Entity
     // fields are added dynamically
   }
 
+  [HierarchyRoot]
+  public class OneHundredFieldsEntity : Entity
+  {
+    [Field, Key]
+    public int Id { get; private set; }
+
+    // fields are added dynamically
+  }
+
 
   [HierarchyRoot(InheritanceSchema.ClassTable)]
   public class SeveralPersistActionsEntityValidA : Entity
@@ -174,6 +183,10 @@ public void OnDefinitionsBuilt(BuildingContext context, DomainModelDef model)
         _ = t.DefineField(fieldName, typeof(int));
       }
 
+      t = model.Types[typeof(OneHundredFieldsEntity)];
+      foreach (var fieldName in GetFieldNames(98)) {
+        _ = t.DefineField(fieldName, typeof(int));
+      }
 
       // each entitity field count is valid
       // overall entity field count is valid