From 96753c908214b64ab98d40e61ae65ae064fd9a67 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?S=C3=A9bastien=20Pertus?= <spertus@microsoft.com>
Date: Wed, 18 Jan 2023 12:09:27 +0100
Subject: [PATCH] handling Pkeys in batch part files

---
 .../Serialization/LocalJsonSerializer.cs      | 27 ++++++++++++-------
 Projects/Dotmim.Sync.Core/Set/SyncTable.cs    |  7 ++++-
 2 files changed, 24 insertions(+), 10 deletions(-)

diff --git a/Projects/Dotmim.Sync.Core/Serialization/LocalJsonSerializer.cs b/Projects/Dotmim.Sync.Core/Serialization/LocalJsonSerializer.cs
index bea5ae78e..759c9a1d6 100644
--- a/Projects/Dotmim.Sync.Core/Serialization/LocalJsonSerializer.cs
+++ b/Projects/Dotmim.Sync.Core/Serialization/LocalJsonSerializer.cs
@@ -85,7 +85,7 @@ public void CloseFile()
         /// <summary>
         /// Open the file and write header
         /// </summary>
-        public void OpenFile(string path, SyncTable shemaTable, SyncRowState state, bool append = false)
+        public void OpenFile(string path, SyncTable schemaTable, SyncRowState state, bool append = false)
         {
             if (this.writer != null)
             {
@@ -114,21 +114,26 @@ public void OpenFile(string path, SyncTable shemaTable, SyncRowState state, bool
             this.writer.WriteStartArray();
             this.writer.WriteStartObject();
             this.writer.WritePropertyName("n");
-            this.writer.WriteValue(shemaTable.TableName);
+            this.writer.WriteValue(schemaTable.TableName);
             this.writer.WritePropertyName("s");
-            this.writer.WriteValue(shemaTable.SchemaName);
+            this.writer.WriteValue(schemaTable.SchemaName);
             this.writer.WritePropertyName("st");
             this.writer.WriteValue((int)state);
 
             this.writer.WritePropertyName("c");
             this.writer.WriteStartArray();
-            foreach (var c in shemaTable.Columns)
+            foreach (var c in schemaTable.Columns)
             {
                 this.writer.WriteStartObject();
                 this.writer.WritePropertyName("n");
                 this.writer.WriteValue(c.ColumnName);
                 this.writer.WritePropertyName("t");
                 this.writer.WriteValue(c.DataType);
+                if (schemaTable.IsPrimaryKey(c.ColumnName))
+                {
+                    this.writer.WritePropertyName("p");
+                    this.writer.WriteValue(1);
+                }
                 this.writer.WriteEndObject();
             }
 
@@ -137,11 +142,11 @@ public void OpenFile(string path, SyncTable shemaTable, SyncRowState state, bool
             this.writer.WriteStartArray();
             this.writer.WriteWhitespace(Environment.NewLine);
         }
-
+        
         /// <summary>
-        /// Append a syncrow to the writer
+        /// Append a sync row to the writer
         /// </summary>
-        public async Task WriteRowToFileAsync(SyncRow row, SyncTable shemaTable)
+        public async Task WriteRowToFileAsync(SyncRow row, SyncTable schemaTable)
         {
             writer.WriteStartArray();
 
@@ -149,7 +154,7 @@ public async Task WriteRowToFileAsync(SyncRow row, SyncTable shemaTable)
 
             if (this.writingRowAsync != null)
             {
-                var str = await this.writingRowAsync(shemaTable, innerRow);
+                var str = await this.writingRowAsync(schemaTable, innerRow);
                 writer.WriteValue(str);
             }
             else
@@ -426,7 +431,7 @@ private SyncTable GetSchemaTableFromReader(JsonTextReader reader, JsonSerializer
             // get columns from array
             var includedColumns = serializer.Deserialize<List<JObject>>(reader);
 
-            // if we dont have columns specified, we are assuming it's the same columns
+            // if we don't have columns specified, we are assuming it's the same columns
             if (includedColumns == null || includedColumns.Count == 0)
                 return null;
 
@@ -437,9 +442,13 @@ private SyncTable GetSchemaTableFromReader(JsonTextReader reader, JsonSerializer
                 // column name & type from file
                 var includedColumnName = includedColumns[i]["n"].Value<string>();
                 var includedColumnType = SyncColumn.GetTypeFromAssemblyQualifiedName(includedColumns[i]["t"].Value<string>());
+                var isPrimaryKey = includedColumns[i].ContainsKey("p");
 
                 // Adding the column 
                 schemaTable.Columns.Add(new SyncColumn(includedColumnName, includedColumnType));
+
+                if (isPrimaryKey)
+                    schemaTable.PrimaryKeys.Add(includedColumnName);
             }
 
             return schemaTable;
diff --git a/Projects/Dotmim.Sync.Core/Set/SyncTable.cs b/Projects/Dotmim.Sync.Core/Set/SyncTable.cs
index 80e74b2bb..fe01e9924 100644
--- a/Projects/Dotmim.Sync.Core/Set/SyncTable.cs
+++ b/Projects/Dotmim.Sync.Core/Set/SyncTable.cs
@@ -224,13 +224,18 @@ public IEnumerable<SyncColumn> GetPrimaryKeysColumns()
         {
             foreach (var column in this.Columns.OrderBy(c => c.Ordinal))
             {
-                var isPrimaryKey = this.PrimaryKeys.Any(pkey => column.ColumnName.Equals(pkey, SyncGlobalization.DataSourceStringComparison));
+                var isPrimaryKey = IsPrimaryKey(column);
 
                 if (isPrimaryKey)
                     yield return column;
             }
         }
 
+        public bool IsPrimaryKey(SyncColumn column)
+        {
+            return this.PrimaryKeys.Any(primaryKey => column.ColumnName.Equals(primaryKey, SyncGlobalization.DataSourceStringComparison));
+        }
+
         /// <summary>
         /// Get all filters for a selected sync table
         /// </summary>