Remove +1 assumption for input versions in $import

src/Microsoft.Health.Fhir.SqlServer/Features/Schema/SchemaVersion.cs

@@ -83,5 +83,6 @@ public enum SchemaVersion
         V71 = 71,
         V72 = 72,
         V73 = 73,
+        V74 = 74,
     }
 }

src/Microsoft.Health.Fhir.SqlServer/Features/Schema/SchemaVersionConstants.cs

@@ -8,7 +8,7 @@ namespace Microsoft.Health.Fhir.SqlServer.Features.Schema
     public static class SchemaVersionConstants
     {
         public const int Min = (int)SchemaVersion.V69;
-        public const int Max = (int)SchemaVersion.V73;
+        public const int Max = (int)SchemaVersion.V74;
         public const int MinForUpgrade = (int)SchemaVersion.V69; // this is used for upgrade tests only
         public const int SearchParameterStatusSchemaVersion = (int)SchemaVersion.V6;
         public const int SupportForReferencesWithMissingTypeVersion = (int)SchemaVersion.V7;

src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Scripts/TransactionCheckWithInitialiScript.sql

@@ -19,6 +19,6 @@ Go
 
 INSERT INTO dbo.SchemaVersion
 VALUES
-    (73, 'started')
+    (74, 'started')
 
 Go

src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/MergeResources.sql

@@ -99,7 +99,7 @@ BEGIN TRY
                ON B.ResourceTypeId = A.ResourceTypeId AND B.ResourceId = A.ResourceId AND B.IsHistory = 0
         OPTION (MAXDOP 1, OPTIMIZE FOR (@DummyTop = 1))
 
-    IF @RaiseExceptionOnConflict = 1 AND EXISTS (SELECT * FROM @ResourceInfos WHERE PreviousVersion IS NOT NULL AND Version <> PreviousVersion + 1)
+    IF @RaiseExceptionOnConflict = 1 AND EXISTS (SELECT * FROM @ResourceInfos WHERE PreviousVersion IS NOT NULL AND Version <= PreviousVersion)
       THROW 50409, 'Resource has been recently updated or added, please compare the resource content in code for any duplicate updates', 1
 
     INSERT INTO @PreviousSurrogateIds

src/Microsoft.Health.Fhir.SqlServer/Features/Storage/SqlServerFhirDataStore.cs

@@ -164,12 +164,12 @@ internal async Task<IDictionary<DataStoreOperationIdentifier, DataStoreOperation
             (var transactionId, var minSequenceId) = await StoreClient.MergeResourcesBeginTransactionAsync(resources.Count, cancellationToken);
 
             var index = 0;
-            var mergeWrappers = new List<MergeResourceWrapper>();
+            var mergeWrappersWithVersions = new List<(MergeResourceWrapper Wrapper, bool KeepVersion, int ResourceVersion, int? ExistingVersion)>();
             var prevResourceId = string.Empty;
             var singleTransaction = enlistInTransaction;
-            foreach (var resourceExt in resources) // if list contains more that one version per resource it must be sorted by id and last updated desc.
+            foreach (var resourceExt in resources) // if list contains more that one version per resource it must be sorted by id and last updated DESC.
             {
-                var setAsHistory = prevResourceId == resourceExt.Wrapper.ResourceId;
+                var setAsHistory = prevResourceId == resourceExt.Wrapper.ResourceId; // this assumes that first resource version is the latest one
                 prevResourceId = resourceExt.Wrapper.ResourceId;
                 var weakETag = resourceExt.WeakETag;
                 int? eTag = weakETag == null
@@ -178,9 +178,9 @@ internal async Task<IDictionary<DataStoreOperationIdentifier, DataStoreOperation
 
                 var resource = resourceExt.Wrapper;
                 var identifier = resourceExt.GetIdentifier();
-                var resourceKey = resource.ToResourceKey(); // keep input version in the results to allow processing multiple versions per resource
                 existingResources.TryGetValue(resource.ToResourceKey(true), out var existingResource);
                 var hasVersionToCompare = false;
+                var existingVersion = 0;
 
                 // Check for any validation errors
                 if (existingResource != null && eTag.HasValue && !string.Equals(eTag.ToString(), existingResource.Version, StringComparison.Ordinal))
@@ -271,7 +271,7 @@ internal async Task<IDictionary<DataStoreOperationIdentifier, DataStoreOperation
                         }
                     }
 
-                    var existingVersion = int.Parse(existingResource.Version);
+                    existingVersion = int.Parse(existingResource.Version);
                     var versionPlusOne = (existingVersion + 1).ToString(CultureInfo.InvariantCulture);
                     if (!resourceExt.KeepVersion) // version is set on input
                     {
@@ -310,12 +310,34 @@ internal async Task<IDictionary<DataStoreOperationIdentifier, DataStoreOperation
                     singleTransaction = true;
                 }
 
-                mergeWrappers.Add(new MergeResourceWrapper(resource, resourceExt.KeepHistory, hasVersionToCompare));
+                mergeWrappersWithVersions.Add((new MergeResourceWrapper(resource, resourceExt.KeepHistory, hasVersionToCompare), resourceExt.KeepVersion, int.Parse(resource.Version), existingVersion));
                 index++;
                 results.Add(identifier, new DataStoreOperationOutcome(new UpsertOutcome(resource, resource.Version == InitialVersion ? SaveOutcomeType.Created : SaveOutcomeType.Updated)));
             }
 
-            if (mergeWrappers.Count > 0) // Do not call DB with empty input
+            // Resources with input versions (keepVersion=true) might not have hasVersionToCompare set. Fix it here.
+            // Resources with keepVersion=true must be in separate call, and not mixed with keepVersion=false ones.
+            // Sort them in groups by resource id and order by version.
+            // In each group find the smallest version higher then existing
+            prevResourceId = string.Empty;
+            var notSetInResoureGroup = false;
+            foreach (var mergeWrapper in mergeWrappersWithVersions.Where(x => x.KeepVersion && x.ExistingVersion != 0).OrderBy(x => x.Wrapper.ResourceWrapper.ResourceId).ThenBy(x => x.ResourceVersion))
+            {
+                if (prevResourceId != mergeWrapper.Wrapper.ResourceWrapper.ResourceId) // this should reset flag on each resource id group including first.
+                {
+                    notSetInResoureGroup = true;
+                }
+
+                prevResourceId = mergeWrapper.Wrapper.ResourceWrapper.ResourceId;
+
+                if (notSetInResoureGroup && mergeWrapper.ResourceVersion > mergeWrapper.ExistingVersion)
+                {
+                    mergeWrapper.Wrapper.HasVersionToCompare = true;
+                    notSetInResoureGroup = false;
+                }
+            }
+
+            if (mergeWrappersWithVersions.Count > 0) // Do not call DB with empty input
             {
                 await using (new Timer(async _ => await _sqlStoreClient.MergeResourcesPutTransactionHeartbeatAsync(transactionId, MergeResourcesTransactionHeartbeatPeriod, cancellationToken), null, TimeSpan.FromSeconds(RandomNumberGenerator.GetInt32(100) / 100.0 * MergeResourcesTransactionHeartbeatPeriod.TotalSeconds), MergeResourcesTransactionHeartbeatPeriod))
                 {
@@ -325,7 +347,7 @@ internal async Task<IDictionary<DataStoreOperationIdentifier, DataStoreOperation
                     {
                         try
                         {
-                            await MergeResourcesWrapperAsync(transactionId, singleTransaction, mergeWrappers, enlistInTransaction, timeoutRetries, cancellationToken);
+                            await MergeResourcesWrapperAsync(transactionId, singleTransaction, mergeWrappersWithVersions.Select(_ => _.Wrapper).ToList(), enlistInTransaction, timeoutRetries, cancellationToken);
                             break;
                         }
                         catch (Exception e)

src/Microsoft.Health.Fhir.SqlServer/Features/Storage/TvpRowGeneration/Merge/MergeResourceWrapper.cs

@@ -29,6 +29,6 @@ internal MergeResourceWrapper(ResourceWrapper resourceWrapper, bool keepHistory,
         /// <summary>
         /// Flag indicating whether version in resource wrapper == (existing version in the database + 1)
         /// </summary>
-        public bool HasVersionToCompare { get; private set; }
+        public bool HasVersionToCompare { get; internal set; }
     }
 }

src/Microsoft.Health.Fhir.SqlServer/Microsoft.Health.Fhir.SqlServer.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
   <!-- Properties used by sql task to generate full script -->
   <PropertyGroup>
-    <LatestSchemaVersion>73</LatestSchemaVersion>
+    <LatestSchemaVersion>74</LatestSchemaVersion>
     <GeneratedFullScriptPath>Features\Schema\Migrations\$(LatestSchemaVersion).sql</GeneratedFullScriptPath>
   </PropertyGroup>
 

test/Microsoft.Health.Fhir.Shared.Tests.E2E/Rest/Import/ImportTests.cs

@@ -92,6 +92,41 @@ public async Task GivenIncrementalLoad_MultipleInputVersionsOutOfOrderSomeNotExp
             Assert.Equal(GetLastUpdated("2002"), result.Resource.Meta.LastUpdated);
         }
 
+        [Fact]
+        public async Task GivenIncrementalLoad_MultipleNonSequentialInputVersions_ResourceExisting()
+        {
+            var id = Guid.NewGuid().ToString("N");
+
+            // set existing
+            var ndJson = PrepareResource(id, "10000", "2000");
+            var location = (await ImportTestHelper.UploadFileAsync(ndJson, _fixture.StorageAccount)).location;
+            var request = CreateImportRequest(location, ImportMode.IncrementalLoad);
+            await ImportCheckAsync(request, null);
+
+            // set input. something before and something after existing
+            ndJson = PrepareResource(id, "9000", "1999");
+            var ndJson2 = PrepareResource(id, "10100", "2001");
+            var ndJson3 = PrepareResource(id, "10300", "2003");
+
+            // note order of records
+            location = (await ImportTestHelper.UploadFileAsync(ndJson2 + ndJson + ndJson3, _fixture.StorageAccount)).location;
+            request = CreateImportRequest(location, ImportMode.IncrementalLoad);
+            await ImportCheckAsync(request, null);
+
+            // check current
+            var result = await _client.ReadAsync<Patient>(ResourceType.Patient, id);
+            Assert.Equal("10300", result.Resource.Meta.VersionId);
+            Assert.Equal(GetLastUpdated("2003"), result.Resource.Meta.LastUpdated);
+
+            // check history
+            result = await _client.VReadAsync<Patient>(ResourceType.Patient, id, "9000");
+            Assert.Equal(GetLastUpdated("1999"), result.Resource.Meta.LastUpdated);
+            result = await _client.VReadAsync<Patient>(ResourceType.Patient, id, "10100");
+            Assert.Equal(GetLastUpdated("2001"), result.Resource.Meta.LastUpdated);
+            result = await _client.VReadAsync<Patient>(ResourceType.Patient, id, "10000");
+            Assert.Equal(GetLastUpdated("2000"), result.Resource.Meta.LastUpdated);
+        }
+
         [Fact]
         public async Task GivenIncrementalLoad_MultipleInputVersions_ResourceExisting_VersionConflict()
         {

tools/BlobRewriter/App.config

@@ -21,5 +21,7 @@
     <add key="SplitBySize" value="true" />
     <!-- Filter on blob names. If empoty string - no filtering. -->
     <add key="NameFilter" value="" />
+    <!-- If true tries to replace {"resourceType": by {"meta":{"versionId":"seconds from last updated","lastUpdated":"YYYY-MM-DDThh:mm:ss"},"resourceType": -->
+    <add key="AddMeta" value="false" />
   </appSettings>
 </configuration>

tools/BlobRewriter/Program.cs

@@ -33,6 +33,7 @@ public static class Program
         private static readonly bool WritesEnabled = bool.Parse(ConfigurationManager.AppSettings["WritesEnabled"]);
         private static readonly bool SplitBySize = bool.Parse(ConfigurationManager.AppSettings["SplitBySize"]);
         private static readonly string NameFilter = ConfigurationManager.AppSettings["NameFilter"];
+        private static readonly bool AddMeta = bool.Parse(ConfigurationManager.AppSettings["AddMeta"]);
 
         public static void Main()
         {
@@ -60,7 +61,7 @@ public static void Main()
                 var blob = blobInt.Item2.First();
 
                 var lines = SplitBySize
-                          ? LinesPerBlob == 0 ? CopyBlob(sourceContainer, blob.Name, targetContainer, ref targetBlobs) : SplitBlobBySize(sourceContainer, blob.Name, targetContainer, ref targetBlobs)
+                          ? LinesPerBlob == 0 ? CopyBlob(sourceContainer, blob.Name, targetContainer, ref targetBlobs, blobIndex) : SplitBlobBySize(sourceContainer, blob.Name, targetContainer, ref targetBlobs)
                           : SplitBlobByResourceId(sourceContainer, blob.Name, targetContainer, ref targetBlobs);
                 Interlocked.Add(ref totalLines, lines);
                 Interlocked.Increment(ref sourceBlobs);
@@ -148,7 +149,7 @@ private static long SplitBlobBySize(BlobContainerClient sourceContainer, string
             return lines;
         }
 
-        private static long CopyBlob(BlobContainerClient sourceContainer, string blobName, BlobContainerClient targetContainer, ref long targetBlobs)
+        private static long CopyBlob(BlobContainerClient sourceContainer, string blobName, BlobContainerClient targetContainer, ref long targetBlobs, int blobIndex)
         {
             var lines = 0L;
             using var stream = targetContainer.GetBlockBlobClient(blobName).OpenWrite(true);
@@ -158,7 +159,17 @@ private static long CopyBlob(BlobContainerClient sourceContainer, string blobNam
                 lines++;
                 if (WritesEnabled)
                 {
-                    writer.WriteLine(line);
+                    if (AddMeta)
+                    {
+                        var date = DateTime.UtcNow.AddMinutes(-SourceBlobs).AddMinutes(blobIndex).AddMilliseconds(lines);
+                        var seconds = ((int)(date - DateTime.Parse("1970-01-01")).TotalSeconds).ToString();
+                        var lineWithMeta = line.Replace("{\"resourceType\":", "{\"meta\":{\"versionId\":\"" + seconds + "\",\"lastUpdated\":\"" + date.ToString("yyyy-MM-ddTHH:mm:ss.fff") + "\"},\"resourceType\":", StringComparison.OrdinalIgnoreCase);
+                        writer.WriteLine(lineWithMeta);
+                    }
+                    else
+                    {
+                        writer.WriteLine(line);
+                    }
                 }
             }