Merge commit '94d525d88f713b36419f28bfda4d82ee68637d83' into v9/dev

# Conflicts:
#	build/NuSpecs/UmbracoCms.Web.nuspec
#	src/Umbraco.Core/Composing/Current.cs
#	src/Umbraco.Core/Persistence/NPocoDatabaseExtensions-Bulk.cs
#	src/Umbraco.Core/Runtime/CoreRuntime.cs
#	src/Umbraco.Infrastructure/Migrations/Upgrade/UmbracoPlan.cs
#	src/Umbraco.Infrastructure/Persistence/NPocoDatabaseExtensions.cs
#	src/Umbraco.Infrastructure/Persistence/UmbracoDatabase.cs
#	src/Umbraco.Persistence.SqlCe/SqlCeSyntaxProvider.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/BTree.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/ContentCacheDataModel.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/ContentCacheDataSerializationResult.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/ContentCacheDataSerializerEntityType.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/ContentData.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/ContentNestedData.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/CultureVariation.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/IContentCacheDataSerializer.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/IContentCacheDataSerializerFactory.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/IDictionaryOfPropertyDataSerializer.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/JsonContentNestedDataSerializer.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/JsonContentNestedDataSerializerFactory.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/LazyCompressedString.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/MsgPackContentNestedDataSerializer.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/MsgPackContentNestedDataSerializerFactory.cs
#	src/Umbraco.PublishedCache.NuCache/DataSource/PropertyData.cs
#	src/Umbraco.PublishedCache.NuCache/NuCacheSerializerComponent.cs
#	src/Umbraco.PublishedCache.NuCache/NuCacheSerializerComposer.cs
#	src/Umbraco.Tests.Integration/Umbraco.Infrastructure/Services/ContentTypeServiceVariantsTests.cs
#	src/Umbraco.Tests/App.config
#	src/Umbraco.Tests/PublishedContent/NuCacheChildrenTests.cs
#	src/Umbraco.Tests/PublishedContent/NuCacheTests.cs
#	src/Umbraco.Tests/Scoping/ScopedNuCacheTests.cs
#	src/Umbraco.Web.UI.NetCore/umbraco/config/lang/da.xml
#	src/Umbraco.Web.UI/web.Template.Debug.config
#	src/Umbraco.Web.UI/web.Template.config
#	src/Umbraco.Web/Composing/ModuleInjector.cs
#	src/Umbraco.Web/Editors/NuCacheStatusController.cs
#	src/Umbraco.Web/PublishedCache/NuCache/DataSource/ContentNestedData.cs
#	src/Umbraco.Web/PublishedCache/NuCache/DataSource/DatabaseDataSource.cs
#	src/Umbraco.Web/PublishedCache/NuCache/NuCacheComposer.cs
#	src/Umbraco.Web/PublishedCache/NuCache/PublishedSnapshotService.cs
#	src/Umbraco.Web/Runtime/WebRuntime.cs

src/Umbraco.PublishedCache.NuCache/ContentNodeKit.cs

@@ -16,6 +16,7 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache
 
         public bool IsNull => ContentTypeId < 0;
 
+        public static ContentNodeKit Empty { get; } = new ContentNodeKit();
         public static ContentNodeKit Null { get; } = new ContentNodeKit { ContentTypeId = -1 };
 
         public void Build(

src/Umbraco.PublishedCache.NuCache/DataSource/BTree.ContentDataSerializer.cs

@@ -3,10 +3,22 @@ using CSharpTest.Net.Serialization;
 
 namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
 {
-    class ContentDataSerializer : ISerializer<ContentData>
+    /// <summary>
+    /// Serializes/Deserializes data to BTree data source for <see cref="ContentData"/>
+    /// </summary>
+    internal class ContentDataSerializer : ISerializer<ContentData>
     {
-        private static readonly DictionaryOfPropertyDataSerializer PropertiesSerializer = new DictionaryOfPropertyDataSerializer();
-        private static readonly DictionaryOfCultureVariationSerializer CultureVariationsSerializer = new DictionaryOfCultureVariationSerializer();
+        public ContentDataSerializer(IDictionaryOfPropertyDataSerializer dictionaryOfPropertyDataSerializer = null)
+        {
+            _dictionaryOfPropertyDataSerializer = dictionaryOfPropertyDataSerializer;
+            if(_dictionaryOfPropertyDataSerializer == null)
+            {
+                _dictionaryOfPropertyDataSerializer = DefaultPropertiesSerializer;
+            }
+        }
+        private static readonly DictionaryOfPropertyDataSerializer DefaultPropertiesSerializer = new DictionaryOfPropertyDataSerializer();
+        private static readonly DictionaryOfCultureVariationSerializer DefaultCultureVariationsSerializer = new DictionaryOfCultureVariationSerializer();
+        private readonly IDictionaryOfPropertyDataSerializer _dictionaryOfPropertyDataSerializer;
 
         public ContentData ReadFrom(Stream stream)
         {
@@ -19,8 +31,8 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
                 VersionDate = PrimitiveSerializer.DateTime.ReadFrom(stream),
                 WriterId = PrimitiveSerializer.Int32.ReadFrom(stream),
                 TemplateId = PrimitiveSerializer.Int32.ReadFrom(stream),
-                Properties = PropertiesSerializer.ReadFrom(stream),
-                CultureInfos = CultureVariationsSerializer.ReadFrom(stream)
+                Properties = _dictionaryOfPropertyDataSerializer.ReadFrom(stream), // TODO: We don't want to allocate empty arrays
+                CultureInfos = DefaultCultureVariationsSerializer.ReadFrom(stream) // TODO: We don't want to allocate empty arrays
             };
         }
 
@@ -36,8 +48,8 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
             {
                 PrimitiveSerializer.Int32.WriteTo(value.TemplateId.Value, stream);
             }
-            PropertiesSerializer.WriteTo(value.Properties, stream);
-            CultureVariationsSerializer.WriteTo(value.CultureInfos, stream);
+            _dictionaryOfPropertyDataSerializer.WriteTo(value.Properties, stream);
+            DefaultCultureVariationsSerializer.WriteTo(value.CultureInfos, stream);
         }
     }
 }

src/Umbraco.PublishedCache.NuCache/DataSource/BTree.ContentNodeKitSerializer.cs

@@ -5,7 +5,17 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
 {
     internal class ContentNodeKitSerializer : ISerializer<ContentNodeKit>
     {
-        static readonly ContentDataSerializer DataSerializer = new ContentDataSerializer();
+        public ContentNodeKitSerializer(ContentDataSerializer contentDataSerializer = null)
+        {
+            _contentDataSerializer = contentDataSerializer;
+            if(_contentDataSerializer == null)
+            {
+                _contentDataSerializer = DefaultDataSerializer;
+            }
+        }
+        static readonly ContentDataSerializer DefaultDataSerializer = new ContentDataSerializer();
+        private readonly ContentDataSerializer _contentDataSerializer;
+
         //static readonly ListOfIntSerializer ChildContentIdsSerializer = new ListOfIntSerializer();
 
         public ContentNodeKit ReadFrom(Stream stream)
@@ -26,10 +36,10 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
             };
             var hasDraft = PrimitiveSerializer.Boolean.ReadFrom(stream);
             if (hasDraft)
-                kit.DraftData = DataSerializer.ReadFrom(stream);
+                kit.DraftData = _contentDataSerializer.ReadFrom(stream);
             var hasPublished = PrimitiveSerializer.Boolean.ReadFrom(stream);
             if (hasPublished)
-                kit.PublishedData = DataSerializer.ReadFrom(stream);
+                kit.PublishedData = _contentDataSerializer.ReadFrom(stream);
             return kit;
         }
 
@@ -47,11 +57,11 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
 
             PrimitiveSerializer.Boolean.WriteTo(value.DraftData != null, stream);
             if (value.DraftData != null)
-                DataSerializer.WriteTo(value.DraftData, stream);
+                _contentDataSerializer.WriteTo(value.DraftData, stream);
 
             PrimitiveSerializer.Boolean.WriteTo(value.PublishedData != null, stream);
             if (value.PublishedData != null)
-                DataSerializer.WriteTo(value.PublishedData, stream);
+                _contentDataSerializer.WriteTo(value.PublishedData, stream);
         }
     }
 }

src/Umbraco.PublishedCache.NuCache/DataSource/BTree.DictionaryOfCultureVariationSerializer.cs

@@ -6,6 +6,9 @@ using Umbraco.Extensions;
 
 namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
 {
+    /// <summary>
+    /// Serializes/Deserializes culture variant data as a dictionary for BTree
+    /// </summary>
     internal class DictionaryOfCultureVariationSerializer : SerializerBase, ISerializer<IReadOnlyDictionary<string, CultureVariation>>
     {
         public IReadOnlyDictionary<string, CultureVariation> ReadFrom(Stream stream)
@@ -18,8 +21,13 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
             var dict = new Dictionary<string, CultureVariation>(StringComparer.InvariantCultureIgnoreCase);
             for (var i = 0; i < pcount; i++)
             {
-                var languageId = PrimitiveSerializer.String.ReadFrom(stream);
-                var cultureVariation = new CultureVariation { Name = ReadStringObject(stream), UrlSegment = ReadStringObject(stream), Date = ReadDateTime(stream) };
+                var languageId = string.Intern(PrimitiveSerializer.String.ReadFrom(stream));
+                var cultureVariation = new CultureVariation
+                {
+                    Name = ReadStringObject(stream),
+                    UrlSegment = ReadStringObject(stream),
+                    Date = ReadDateTime(stream)
+                };
                 dict[languageId] = cultureVariation;
             }
             return dict;

src/Umbraco.PublishedCache.NuCache/DataSource/BTree.DictionaryOfPropertyDataSerializer.cs

@@ -6,44 +6,47 @@ using Umbraco.Extensions;
 
 namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
 {
-    internal class DictionaryOfPropertyDataSerializer : SerializerBase, ISerializer<IDictionary<string, PropertyData[]>>
+    /// <summary>
+    /// Serializes/Deserializes property data as a dictionary for BTree
+    /// </summary>
+    internal class DictionaryOfPropertyDataSerializer : SerializerBase, ISerializer<IDictionary<string, PropertyData[]>>, IDictionaryOfPropertyDataSerializer
     {
         public IDictionary<string, PropertyData[]> ReadFrom(Stream stream)
         {
-            var dict = new Dictionary<string, PropertyData[]>(StringComparer.InvariantCultureIgnoreCase);
 
             // read properties count
             var pcount = PrimitiveSerializer.Int32.ReadFrom(stream);
 
+            var dict = new Dictionary<string, PropertyData[]>(pcount,StringComparer.InvariantCultureIgnoreCase);
             // read each property
             for (var i = 0; i < pcount; i++)
             {
                 // read property alias
-                var key = PrimitiveSerializer.String.ReadFrom(stream);
+                var key = string.Intern(PrimitiveSerializer.String.ReadFrom(stream));
 
                 // read values count
                 var vcount = PrimitiveSerializer.Int32.ReadFrom(stream);
 
                 // create pdata and add to the dictionary
-                var pdatas = new List<PropertyData>();
+                var pdatas = new PropertyData[vcount];
 
                 // for each value, read and add to pdata
                 for (var j = 0; j < vcount; j++)
                 {
                     var pdata = new PropertyData();
-                    pdatas.Add(pdata);
+                    pdatas[j] =pdata;
 
                     // everything that can be null is read/written as object
                     //  even though - culture and segment should never be null here, as 'null' represents
                     //  the 'current' value, and string.Empty should be used to represent the invariant or
                     //  neutral values - PropertyData throws when getting nulls, so falling back to
                     //  string.Empty here - what else?
-                    pdata.Culture = ReadStringObject(stream) ?? string.Empty;
-                    pdata.Segment = ReadStringObject(stream) ?? string.Empty;
+                    pdata.Culture = ReadStringObject(stream, true) ?? string.Empty;
+                    pdata.Segment = ReadStringObject(stream, true) ?? string.Empty;
                     pdata.Value = ReadObject(stream);
                 }
 
-                dict[key] = pdatas.ToArray();
+                dict[key] = pdatas;
             }
             return dict;
         }

src/Umbraco.PublishedCache.NuCache/DataSource/BTree.cs

@@ -7,10 +7,10 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
 {
     internal class BTree
     {
-        public static BPlusTree<int, ContentNodeKit> GetTree(string filepath, bool exists, NuCacheSettings settings)
+        public static BPlusTree<int, ContentNodeKit> GetTree(string filepath, bool exists, NuCacheSettings settings, ContentDataSerializer contentDataSerializer = null)
         {
             var keySerializer = new PrimitiveSerializer();
-            var valueSerializer = new ContentNodeKitSerializer();
+            var valueSerializer = new ContentNodeKitSerializer(contentDataSerializer);
             var options = new BPlusTree<int, ContentNodeKit>.OptionsV2(keySerializer, valueSerializer)
             {
                 CreateFile = exists ? CreatePolicy.IfNeeded : CreatePolicy.Always,
@@ -37,6 +37,7 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
             //btree.
 
             return tree;
+
         }
 
         private static int GetBlockSize(NuCacheSettings settings)

src/Umbraco.PublishedCache.NuCache/DataSource/ContentCacheDataModel.cs

@@ -0,0 +1,45 @@
+using System.Collections.Generic;
+using Newtonsoft.Json;
+using Umbraco.Cms.Infrastructure.Serialization;
+using System.Runtime.Serialization;
+
+namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
+{
+    /// <summary>
+    /// The content model stored in the content cache database table serialized as JSON
+    /// </summary>
+    [DataContract] // NOTE: Use DataContract annotations here to control how MessagePack serializes/deserializes the data to use INT keys
+    public class ContentCacheDataModel
+    {
+        // TODO: We don't want to allocate empty arrays
+        //dont serialize empty properties
+        [DataMember(Order = 0)]
+        [JsonProperty("pd")]
+        [JsonConverter(typeof(AutoInterningStringKeyCaseInsensitiveDictionaryConverter<PropertyData[]>))]
+        public Dictionary<string, PropertyData[]> PropertyData { get; set; }
+
+        [DataMember(Order = 1)]
+        [JsonProperty("cd")]
+        [JsonConverter(typeof(AutoInterningStringKeyCaseInsensitiveDictionaryConverter<CultureVariation>))]
+        public Dictionary<string, CultureVariation> CultureData { get; set; }
+
+        [DataMember(Order = 2)]
+        [JsonProperty("us")]
+        public string UrlSegment { get; set; }
+
+        //Legacy properties used to deserialize existing nucache db entries
+        [IgnoreDataMember]
+        [JsonProperty("properties")]
+        [JsonConverter(typeof(CaseInsensitiveDictionaryConverter<PropertyData[]>))]
+        private Dictionary<string, PropertyData[]> LegacyPropertyData { set => PropertyData = value; }
+
+        [IgnoreDataMember]
+        [JsonProperty("cultureData")]
+        [JsonConverter(typeof(CaseInsensitiveDictionaryConverter<CultureVariation>))]
+        private Dictionary<string, CultureVariation> LegacyCultureData { set => CultureData = value; }
+
+        [IgnoreDataMember]
+        [JsonProperty("urlSegment")]
+        private string LegacyUrlSegment { set => UrlSegment = value; }
+    }
+}

src/Umbraco.PublishedCache.NuCache/DataSource/ContentCacheDataSerializationResult.cs

@@ -0,0 +1,43 @@
+using System;
+using System.Collections.Generic;
+
+namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
+{
+    /// <summary>
+    /// The serialization result from <see cref="IContentCacheDataSerializer"/> for which the serialized value
+    /// will be either a string or a byte[]
+    /// </summary>
+    public struct ContentCacheDataSerializationResult : IEquatable<ContentCacheDataSerializationResult>
+    {
+        public ContentCacheDataSerializationResult(string stringData, byte[] byteData)
+        {
+            StringData = stringData;
+            ByteData = byteData;
+        }
+
+        public string StringData { get; }
+        public byte[] ByteData { get; }
+
+        public override bool Equals(object obj)
+            => obj is ContentCacheDataSerializationResult result && Equals(result);
+
+        public bool Equals(ContentCacheDataSerializationResult other)
+            => StringData == other.StringData &&
+                   EqualityComparer<byte[]>.Default.Equals(ByteData, other.ByteData);
+
+        public override int GetHashCode()
+        {
+            var hashCode = 1910544615;
+            hashCode = hashCode * -1521134295 + EqualityComparer<string>.Default.GetHashCode(StringData);
+            hashCode = hashCode * -1521134295 + EqualityComparer<byte[]>.Default.GetHashCode(ByteData);
+            return hashCode;
+        }
+
+        public static bool operator ==(ContentCacheDataSerializationResult left, ContentCacheDataSerializationResult right)
+            => left.Equals(right);
+
+        public static bool operator !=(ContentCacheDataSerializationResult left, ContentCacheDataSerializationResult right)
+            => !(left == right);
+    }
+
+}

src/Umbraco.PublishedCache.NuCache/DataSource/ContentCacheDataSerializerEntityType.cs

@@ -0,0 +1,13 @@
+using System;
+
+namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
+{
+    [Flags]
+    public enum ContentCacheDataSerializerEntityType
+    {
+        Document = 1,
+        Media = 2,
+        Member = 4
+    }
+
+}

src/Umbraco.PublishedCache.NuCache/DataSource/ContentData.cs

@@ -3,7 +3,9 @@ using System.Collections.Generic;
 
 namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
 {
-    // represents everything that is specific to edited or published version
+    /// <summary>
+    /// Represents everything that is specific to an edited or published content version
+    /// </summary>
     public class ContentData
     {
         public string Name { get; set; }

src/Umbraco.PublishedCache.NuCache/DataSource/ContentNestedData.cs

@@ -1,36 +0,0 @@
-using System.Collections.Generic;
-using Newtonsoft.Json;
-using Umbraco.Cms.Infrastructure.Serialization;
-
-namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
-{
-    /// <summary>
-    /// The content item 1:M data that is serialized to JSON
-    /// </summary>
-    internal class ContentNestedData
-    {
-        // dont serialize empty properties
-        [JsonProperty("pd")]
-        [JsonConverter(typeof(CaseInsensitiveDictionaryConverter<PropertyData[]>))]
-        public Dictionary<string, PropertyData[]> PropertyData { get; set; }
-
-        [JsonProperty("cd")]
-        [JsonConverter(typeof(CaseInsensitiveDictionaryConverter<CultureVariation>))]
-        public Dictionary<string, CultureVariation> CultureData { get; set; }
-
-        [JsonProperty("us")]
-        public string UrlSegment { get; set; }
-
-        // Legacy properties used to deserialize existing nucache db entries
-        [JsonProperty("properties")]
-        [JsonConverter(typeof(CaseInsensitiveDictionaryConverter<PropertyData[]>))]
-        private Dictionary<string, PropertyData[]> LegacyPropertyData { set { PropertyData = value; } }
-
-        [JsonProperty("cultureData")]
-        [JsonConverter(typeof(CaseInsensitiveDictionaryConverter<CultureVariation>))]
-        private Dictionary<string, CultureVariation> LegacyCultureData { set { CultureData = value; } }
-
-        [JsonProperty("urlSegment")]
-        private string LegacyUrlSegment { set { UrlSegment = value; } }
-    }
-}

src/Umbraco.PublishedCache.NuCache/DataSource/ContentSourceDto.cs

@@ -1,12 +1,13 @@
-using System;
+using System;
+using Umbraco.Cms.Core.Models;
 
 namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
 {
     // read-only dto
-    internal class ContentSourceDto
+    internal class ContentSourceDto : IReadOnlyContentBase
     {
         public int Id { get; set; }
-        public Guid Uid { get; set; }
+        public Guid Key { get; set; }
         public int ContentTypeId { get; set; }
 
         public int Level { get; set; }
@@ -27,6 +28,7 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
         public int EditWriterId { get; set; }
         public int EditTemplateId { get; set; }
         public string EditData { get; set; }
+        public byte[] EditDataRaw { get; set; }
 
         // published data
         public int PublishedVersionId { get; set; }
@@ -35,5 +37,11 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
         public int PubWriterId { get; set; }
         public int PubTemplateId { get; set; }
         public string PubData { get; set; }
+        public byte[] PubDataRaw { get; set; }
+
+        // Explicit implementation
+        DateTime IReadOnlyContentBase.UpdateDate => EditVersionDate;
+        string IReadOnlyContentBase.Name => EditName;
+        int IReadOnlyContentBase.WriterId => EditWriterId;
     }
 }

src/Umbraco.PublishedCache.NuCache/DataSource/CultureVariation.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Runtime.Serialization;
 using Newtonsoft.Json;
 
 namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
@@ -6,30 +7,39 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
     /// <summary>
     /// Represents the culture variation information on a content item
     /// </summary>
+    [DataContract] // NOTE: Use DataContract annotations here to control how MessagePack serializes/deserializes the data to use INT keys
     public class CultureVariation
     {
+        [DataMember(Order = 0)]
         [JsonProperty("nm")]
         public string Name { get; set; }
 
+        [DataMember(Order = 1)]
         [JsonProperty("us")]
         public string UrlSegment { get; set; }
 
+        [DataMember(Order = 2)]
         [JsonProperty("dt")]
         public DateTime Date { get; set; }
 
+        [DataMember(Order = 3)]
         [JsonProperty("isd")]
         public bool IsDraft { get; set; }
 
         //Legacy properties used to deserialize existing nucache db entries
+        [IgnoreDataMember]
         [JsonProperty("name")]
         private string LegacyName { set { Name = value; } }
 
+        [IgnoreDataMember]
         [JsonProperty("urlSegment")]
         private string LegacyUrlSegment { set { UrlSegment = value; } }
 
+        [IgnoreDataMember]
         [JsonProperty("date")]
         private DateTime LegacyDate { set { Date = value; } }
 
+        [IgnoreDataMember]
         [JsonProperty("isDraft")]
         private bool LegacyIsDraft { set { IsDraft = value; } }
     }

src/Umbraco.PublishedCache.NuCache/DataSource/IContentCacheDataSerializer.cs

@@ -0,0 +1,25 @@
+using Umbraco.Cms.Core.Models;
+
+namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
+{
+
+    /// <summary>
+    /// Serializes/Deserializes <see cref="ContentCacheDataModel"/> document to the SQL Database as a string
+    /// </summary>
+    /// <remarks>
+    /// Resolved from the <see cref="IContentCacheDataSerializerFactory"/>. This cannot be resolved from DI.
+    /// </remarks>
+    public interface IContentCacheDataSerializer
+    {
+        /// <summary>
+        /// Deserialize the data into a <see cref="ContentCacheDataModel"/>
+        /// </summary>
+        ContentCacheDataModel Deserialize(IReadOnlyContentBase content, string stringData, byte[] byteData);
+
+        /// <summary>
+        /// Serializes the <see cref="ContentCacheDataModel"/> 
+        /// </summary>
+        ContentCacheDataSerializationResult Serialize(IReadOnlyContentBase content, ContentCacheDataModel model);
+    }
+
+}

src/Umbraco.PublishedCache.NuCache/DataSource/IContentCacheDataSerializerFactory.cs

@@ -0,0 +1,16 @@
+namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
+{
+    public interface IContentCacheDataSerializerFactory
+    {
+        /// <summary>
+        /// Gets or creates a new instance of <see cref="IContentCacheDataSerializer"/>
+        /// </summary>
+        /// <returns></returns>
+        /// <remarks>
+        /// This method may return the same instance, however this depends on the state of the application and if any underlying data has changed.
+        /// This method may also be used to initialize anything before a serialization/deserialization session occurs.
+        /// </remarks>
+        IContentCacheDataSerializer Create(ContentCacheDataSerializerEntityType types);
+    }
+
+}

src/Umbraco.PublishedCache.NuCache/DataSource/IDictionaryOfPropertyDataSerializer.cs

@@ -0,0 +1,11 @@
+using System.Collections.Generic;
+using System.IO;
+
+namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
+{
+    internal interface IDictionaryOfPropertyDataSerializer
+    {
+        IDictionary<string, PropertyData[]> ReadFrom(Stream stream);
+        void WriteTo(IDictionary<string, PropertyData[]> value, Stream stream);
+    }
+}

src/Umbraco.PublishedCache.NuCache/DataSource/JsonContentNestedDataSerializer.cs

@@ -0,0 +1,91 @@
+using Newtonsoft.Json;
+using System;
+using System.Buffers;
+using System.Collections.Generic;
+using System.IO;
+using Umbraco.Cms.Core.Models;
+using Umbraco.Cms.Infrastructure.Serialization;
+
+namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
+{
+
+    public class JsonContentNestedDataSerializer : IContentCacheDataSerializer
+    {
+        // by default JsonConvert will deserialize our numeric values as Int64
+        // which is bad, because they were Int32 in the database - take care
+        private readonly JsonSerializerSettings _jsonSerializerSettings = new JsonSerializerSettings
+        {
+            Converters = new List<JsonConverter> { new ForceInt32Converter() },
+
+            // Explicitly specify date handling so that it's consistent and follows the same date handling as MessagePack
+            DateParseHandling = DateParseHandling.DateTime,
+            DateFormatHandling = DateFormatHandling.IsoDateFormat,
+            DateTimeZoneHandling = DateTimeZoneHandling.Utc,
+            DateFormatString = "o"
+        };
+        private readonly JsonNameTable _propertyNameTable = new DefaultJsonNameTable();
+        public ContentCacheDataModel Deserialize(IReadOnlyContentBase content, string stringData, byte[] byteData)
+        {
+            if (stringData == null && byteData != null)
+                throw new NotSupportedException($"{typeof(JsonContentNestedDataSerializer)} does not support byte[] serialization");
+
+            JsonSerializer serializer = JsonSerializer.Create(_jsonSerializerSettings);
+            using (JsonTextReader reader = new JsonTextReader(new StringReader(stringData)))
+            {
+                // reader will get buffer from array pool
+                reader.ArrayPool = JsonArrayPool.Instance;
+                reader.PropertyNameTable = _propertyNameTable;
+                return serializer.Deserialize<ContentCacheDataModel>(reader);
+            }
+        }
+
+        public ContentCacheDataSerializationResult Serialize(IReadOnlyContentBase content, ContentCacheDataModel model)
+        {
+            // note that numeric values (which are Int32) are serialized without their
+            // type (eg "value":1234) and JsonConvert by default deserializes them as Int64
+
+            var json = JsonConvert.SerializeObject(model);
+            return new ContentCacheDataSerializationResult(json, null);
+        }
+    }
+    public class JsonArrayPool : IArrayPool<char>
+    {
+        public static readonly JsonArrayPool Instance = new JsonArrayPool();
+
+        public char[] Rent(int minimumLength)
+        {
+            // get char array from System.Buffers shared pool
+            return ArrayPool<char>.Shared.Rent(minimumLength);
+        }
+
+        public void Return(char[] array)
+        {
+            // return char array to System.Buffers shared pool
+            ArrayPool<char>.Shared.Return(array);
+        }
+    }
+    public class AutomaticJsonNameTable : DefaultJsonNameTable
+    {
+        int nAutoAdded = 0;
+        int maxToAutoAdd;
+
+        public AutomaticJsonNameTable(int maxToAdd)
+        {
+            this.maxToAutoAdd = maxToAdd;
+        }
+
+        public override string Get(char[] key, int start, int length)
+        {
+            var s = base.Get(key, start, length);
+
+            if (s == null && nAutoAdded < maxToAutoAdd)
+            {
+                s = new string(key, start, length);
+                Add(s);
+                nAutoAdded++;
+            }
+
+            return s;
+        }
+    }
+}

src/Umbraco.PublishedCache.NuCache/DataSource/JsonContentNestedDataSerializerFactory.cs

@@ -0,0 +1,10 @@
+using System;
+
+namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
+{
+    internal class JsonContentNestedDataSerializerFactory : IContentCacheDataSerializerFactory
+    {
+        private readonly Lazy<JsonContentNestedDataSerializer> _serializer = new Lazy<JsonContentNestedDataSerializer>();
+        public IContentCacheDataSerializer Create(ContentCacheDataSerializerEntityType types) => _serializer.Value;
+    }
+}

src/Umbraco.PublishedCache.NuCache/DataSource/LazyCompressedString.cs

@@ -0,0 +1,109 @@
+using K4os.Compression.LZ4;
+using System;
+using System.Diagnostics;
+using System.Text;
+using Umbraco.Cms.Core.Exceptions;
+
+namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
+{
+    /// <summary>
+    /// Lazily decompresses a LZ4 Pickler compressed UTF8 string
+    /// </summary>
+    [DebuggerDisplay("{Display}")]
+    internal struct LazyCompressedString
+    {
+        private byte[] _bytes;
+        private string _str;
+        private readonly object _locker;
+
+        /// <summary>
+        /// Constructor
+        /// </summary>
+        /// <param name="bytes">LZ4 Pickle compressed UTF8 String</param>
+        public LazyCompressedString(byte[] bytes)
+        {
+            _locker = new object();
+            _bytes = bytes;
+            _str = null;
+        }
+
+        public byte[] GetBytes()
+        {
+            if (_bytes == null)
+            {
+                throw new InvalidOperationException("The bytes have already been expanded");
+            }
+
+            return _bytes;
+        }
+
+        /// <summary>
+        /// Returns the decompressed string from the bytes. This methods can only be called once.
+        /// </summary>
+        /// <returns></returns>
+        /// <exception cref="InvalidOperationException">Throws if this is called more than once</exception>
+        public string DecompressString()
+        {
+            if (_str != null)
+            {
+                return _str;
+            }
+
+            lock (_locker)
+            {
+                if (_str != null)
+                {
+                    // double check
+                    return _str;
+                }
+
+                if (_bytes == null)
+                {
+                    throw new InvalidOperationException("Bytes have already been cleared");
+                }
+
+                _str = Encoding.UTF8.GetString(LZ4Pickler.Unpickle(_bytes));
+                _bytes = null;
+            }
+            return _str;
+        }
+
+        /// <summary>
+        /// Used to display debugging output since ToString() can only be called once
+        /// </summary>
+        private string Display
+        {
+            get
+            {
+                if (_str != null)
+                {
+                    return $"Decompressed: {_str}";
+                }
+
+                lock (_locker)
+                {
+                    if (_str != null)
+                    {
+                        // double check
+                        return $"Decompressed: {_str}";
+                    }
+
+                    if (_bytes == null)
+                    {
+                        // This shouldn't happen
+                        throw new PanicException("Bytes have already been cleared");
+                    }
+                    else
+                    {
+                        return $"Compressed Bytes: {_bytes.Length}";
+                    }
+                }
+            }
+        }
+
+        public override string ToString() => DecompressString();
+
+        public static implicit operator string(LazyCompressedString l) => l.ToString();
+    }
+
+}

src/Umbraco.PublishedCache.NuCache/DataSource/MsgPackContentNestedDataSerializer.cs

@@ -0,0 +1,126 @@
+using K4os.Compression.LZ4;
+using MessagePack;
+using MessagePack.Resolvers;
+using System;
+using System.Linq;
+using System.Text;
+using Umbraco.Cms.Core.Models;
+using Umbraco.Cms.Core.PropertyEditors;
+
+namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
+{
+
+    /// <summary>
+    /// Serializes/Deserializes <see cref="ContentCacheDataModel"/> document to the SQL Database as bytes using MessagePack
+    /// </summary>
+    public class MsgPackContentNestedDataSerializer : IContentCacheDataSerializer
+    {
+        private readonly MessagePackSerializerOptions _options;
+        private readonly IPropertyCacheCompression _propertyOptions;
+
+        public MsgPackContentNestedDataSerializer(IPropertyCacheCompression propertyOptions)
+        {
+            _propertyOptions = propertyOptions ?? throw new ArgumentNullException(nameof(propertyOptions));
+
+            var defaultOptions = ContractlessStandardResolver.Options;
+            var resolver = CompositeResolver.Create(
+
+                // TODO: We want to be able to intern the strings for aliases when deserializing like we do for Newtonsoft but I'm unsure exactly how
+                // to do that but it would seem to be with a custom message pack resolver but I haven't quite figured out based on the docs how
+                // to do that since that is part of the int key -> string mapping operation, might have to see the source code to figure that one out.
+                // There are docs here on how to build one of these: https://github.com/neuecc/MessagePack-CSharp/blob/master/README.md#low-level-api-imessagepackformattert
+                // and there are a couple examples if you search on google for them but this will need to be a separate project.
+                // NOTE: resolver custom types first
+                // new ContentNestedDataResolver(),
+
+                // finally use standard resolver
+                defaultOptions.Resolver
+            );
+
+            _options = defaultOptions
+                .WithResolver(resolver)
+                .WithCompression(MessagePackCompression.Lz4BlockArray);            
+        }
+
+        public string ToJson(byte[] bin)
+        {
+            var json = MessagePackSerializer.ConvertToJson(bin, _options);
+            return json;
+        }
+
+        public ContentCacheDataModel Deserialize(IReadOnlyContentBase content, string stringData, byte[] byteData)
+        {
+            if (byteData != null)
+            {
+                var cacheModel = MessagePackSerializer.Deserialize<ContentCacheDataModel>(byteData, _options);
+                Expand(content, cacheModel);
+                return cacheModel;
+            }
+            else if (stringData != null)
+            {
+                // NOTE: We don't really support strings but it's possible if manually used (i.e. tests)
+                var bin = Convert.FromBase64String(stringData);
+                var cacheModel = MessagePackSerializer.Deserialize<ContentCacheDataModel>(bin, _options);
+                Expand(content, cacheModel);
+                return cacheModel;
+            }
+            else
+            {
+                return null;
+            }
+        }
+
+        public ContentCacheDataSerializationResult Serialize(IReadOnlyContentBase content, ContentCacheDataModel model)
+        {
+            Compress(content, model);
+            var bytes = MessagePackSerializer.Serialize(model, _options);
+            return new ContentCacheDataSerializationResult(null, bytes);
+        }
+
+        /// <summary>
+        /// Used during serialization to compress properties
+        /// </summary>
+        /// <param name="model"></param>
+        /// <remarks>
+        /// This will essentially 'double compress' property data. The MsgPack data as a whole will already be compressed
+        /// but this will go a step further and double compress property data so that it is stored in the nucache file
+        /// as compressed bytes and therefore will exist in memory as compressed bytes. That is, until the bytes are
+        /// read/decompressed as a string to be displayed on the front-end. This allows for potentially a significant
+        /// memory savings but could also affect performance of first rendering pages while decompression occurs.
+        /// </remarks>
+        private void Compress(IReadOnlyContentBase content, ContentCacheDataModel model)
+        {
+            foreach(var propertyAliasToData in model.PropertyData)
+            {
+                if (_propertyOptions.IsCompressed(content, propertyAliasToData.Key))
+                {
+                    foreach(var property in propertyAliasToData.Value.Where(x => x.Value != null && x.Value is string))
+                    {
+                        property.Value = LZ4Pickler.Pickle(Encoding.UTF8.GetBytes((string)property.Value), LZ4Level.L00_FAST);
+                    }
+                }
+            }
+        }
+
+        /// <summary>
+        /// Used during deserialization to map the property data as lazy or expand the value
+        /// </summary>
+        /// <param name="nestedData"></param>
+        private void Expand(IReadOnlyContentBase content, ContentCacheDataModel nestedData)
+        {
+            foreach (var propertyAliasToData in nestedData.PropertyData)
+            {
+                if (_propertyOptions.IsCompressed(content, propertyAliasToData.Key))
+                {
+                    foreach (var property in propertyAliasToData.Value.Where(x => x.Value != null))
+                    {
+                        if (property.Value is byte[] byteArrayValue)
+                        {
+                            property.Value = new LazyCompressedString(byteArrayValue);
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

src/Umbraco.PublishedCache.NuCache/DataSource/MsgPackContentNestedDataSerializerFactory.cs

@@ -0,0 +1,69 @@
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using Umbraco.Cms.Core.Models;
+using Umbraco.Cms.Core.PropertyEditors;
+using Umbraco.Cms.Core.Services;
+
+namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
+{
+    internal class MsgPackContentNestedDataSerializerFactory : IContentCacheDataSerializerFactory
+    {
+        private readonly IContentTypeService _contentTypeService;
+        private readonly IMediaTypeService _mediaTypeService;
+        private readonly IMemberTypeService _memberTypeService;
+        private readonly PropertyEditorCollection _propertyEditors;
+        private readonly IPropertyCacheCompressionOptions _compressionOptions;
+        private readonly ConcurrentDictionary<(int, string), bool> _isCompressedCache = new ConcurrentDictionary<(int, string), bool>();
+
+        public MsgPackContentNestedDataSerializerFactory(
+            IContentTypeService contentTypeService,
+            IMediaTypeService mediaTypeService,
+            IMemberTypeService memberTypeService,
+            PropertyEditorCollection propertyEditors,
+            IPropertyCacheCompressionOptions compressionOptions)
+        {
+            _contentTypeService = contentTypeService;
+            _mediaTypeService = mediaTypeService;
+            _memberTypeService = memberTypeService;
+            _propertyEditors = propertyEditors;
+            _compressionOptions = compressionOptions;
+        }
+
+        public IContentCacheDataSerializer Create(ContentCacheDataSerializerEntityType types)
+        {
+            // Depending on which entity types are being requested, we need to look up those content types
+            // to initialize the compression options.
+            // We need to initialize these options now so that any data lookups required are completed and are not done while the content cache
+            // is performing DB queries which will result in errors since we'll be trying to query with open readers.
+            // NOTE: The calls to GetAll() below should be cached if the data has not been changed.
+
+            var contentTypes = new Dictionary<int, IContentTypeComposition>();
+            if ((types & ContentCacheDataSerializerEntityType.Document) == ContentCacheDataSerializerEntityType.Document)
+            {
+                foreach(var ct in _contentTypeService.GetAll())
+                {
+                    contentTypes[ct.Id] = ct;
+                }
+            }
+            if ((types & ContentCacheDataSerializerEntityType.Media) == ContentCacheDataSerializerEntityType.Media)
+            {
+                foreach (var ct in _mediaTypeService.GetAll())
+                {
+                    contentTypes[ct.Id] = ct;
+                }
+            }
+            if ((types & ContentCacheDataSerializerEntityType.Member) == ContentCacheDataSerializerEntityType.Member)
+            {
+                foreach (var ct in _memberTypeService.GetAll())
+                {
+                    contentTypes[ct.Id] = ct;
+                }
+            }
+
+            var compression = new PropertyCacheCompression(_compressionOptions, contentTypes, _propertyEditors, _isCompressedCache);
+            var serializer = new MsgPackContentNestedDataSerializer(compression);
+
+            return serializer;
+        }
+    }
+}

src/Umbraco.PublishedCache.NuCache/DataSource/PropertyData.cs

@@ -1,14 +1,20 @@
 using System;
 using System.ComponentModel;
+using System.Runtime.Serialization;
 using Newtonsoft.Json;
+using Umbraco.Cms.Infrastructure.Serialization;
 
 namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
 {
+
+    [DataContract] // NOTE: Use DataContract annotations here to control how MessagePack serializes/deserializes the data to use INT keys
     public class PropertyData
     {
         private string _culture;
         private string _segment;
 
+        [DataMember(Order = 0)]
+        [JsonConverter(typeof(AutoInterningStringConverter))]
         [DefaultValue("")]
         [JsonProperty(DefaultValueHandling = DefaultValueHandling.IgnoreAndPopulate, PropertyName = "c")]
         public string Culture
@@ -17,6 +23,8 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
             set => _culture = value ?? throw new ArgumentNullException(nameof(value)); // TODO: or fallback to string.Empty? CANNOT be null
         }
 
+        [DataMember(Order = 1)]
+        [JsonConverter(typeof(AutoInterningStringConverter))]
         [DefaultValue("")]
         [JsonProperty(DefaultValueHandling = DefaultValueHandling.IgnoreAndPopulate, PropertyName = "s")]
         public string Segment
@@ -25,23 +33,26 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
             set => _segment = value ?? throw new ArgumentNullException(nameof(value)); // TODO: or fallback to string.Empty? CANNOT be null
         }
 
+        [DataMember(Order = 2)]
         [JsonProperty("v")]
         public object Value { get; set; }
 
-
         // Legacy properties used to deserialize existing nucache db entries
+        [IgnoreDataMember]
         [JsonProperty("culture")]
         private string LegacyCulture
         {
             set => Culture = value;
         }
 
+        [IgnoreDataMember]
         [JsonProperty("seg")]
         private string LegacySegment
         {
             set => Segment = value;
         }
 
+        [IgnoreDataMember]
         [JsonProperty("val")]
         private object LegacyValue
         {

src/Umbraco.PublishedCache.NuCache/DataSource/SerializerBase.cs

@@ -6,100 +6,218 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.DataSource
 {
     internal abstract class SerializerBase
     {
+        private const char PrefixNull = 'N';
+        private const char PrefixString = 'S';
+        private const char PrefixInt32 = 'I';
+        private const char PrefixUInt16 = 'H';
+        private const char PrefixUInt32 = 'J';
+        private const char PrefixLong = 'L';
+        private const char PrefixFloat = 'F';
+        private const char PrefixDouble = 'B';
+        private const char PrefixDateTime = 'D';
+        private const char PrefixByte = 'O';
+        private const char PrefixByteArray = 'A';
+        private const char PrefixCompressedStringByteArray = 'C';
+        private const char PrefixSignedByte = 'E';
+        private const char PrefixBool = 'M';
+        private const char PrefixGuid = 'G';
+        private const char PrefixTimeSpan = 'T';
+        private const char PrefixInt16 = 'Q';
+        private const char PrefixChar = 'R';
+
         protected string ReadString(Stream stream) => PrimitiveSerializer.String.ReadFrom(stream);
         protected int ReadInt(Stream stream) => PrimitiveSerializer.Int32.ReadFrom(stream);
         protected long ReadLong(Stream stream) => PrimitiveSerializer.Int64.ReadFrom(stream);
         protected float ReadFloat(Stream stream) => PrimitiveSerializer.Float.ReadFrom(stream);
         protected double ReadDouble(Stream stream) => PrimitiveSerializer.Double.ReadFrom(stream);
         protected DateTime ReadDateTime(Stream stream) => PrimitiveSerializer.DateTime.ReadFrom(stream);
+        protected byte[] ReadByteArray(Stream stream) => PrimitiveSerializer.Bytes.ReadFrom(stream);
 
-        private T? ReadObject<T>(Stream stream, char t, Func<Stream, T> read)
+        private T? ReadStruct<T>(Stream stream, char t, Func<Stream, T> read)
             where T : struct
         {
             var type = PrimitiveSerializer.Char.ReadFrom(stream);
-            if (type == 'N') return null;
+            if (type == PrefixNull) return null;
             if (type != t)
                 throw new NotSupportedException($"Cannot deserialize type '{type}', expected '{t}'.");
             return read(stream);
         }
 
-        protected string ReadStringObject(Stream stream) // required 'cos string is not a struct
+        protected string ReadStringObject(Stream stream, bool intern = false) // required 'cos string is not a struct
         {
             var type = PrimitiveSerializer.Char.ReadFrom(stream);
-            if (type == 'N') return null;
-            if (type != 'S')
-                throw new NotSupportedException($"Cannot deserialize type '{type}', expected 'S'.");
-            return PrimitiveSerializer.String.ReadFrom(stream);
+            if (type == PrefixNull) return null;
+            if (type != PrefixString)
+                throw new NotSupportedException($"Cannot deserialize type '{type}', expected '{PrefixString}'.");
+            return intern
+                ? string.Intern(PrimitiveSerializer.String.ReadFrom(stream))
+                : PrimitiveSerializer.String.ReadFrom(stream);
         }
 
-        protected int? ReadIntObject(Stream stream) => ReadObject(stream, 'I', ReadInt);
-        protected long? ReadLongObject(Stream stream) => ReadObject(stream, 'L', ReadLong);
-        protected float? ReadFloatObject(Stream stream) => ReadObject(stream, 'F', ReadFloat);
-        protected double? ReadDoubleObject(Stream stream) => ReadObject(stream, 'B', ReadDouble);
-        protected DateTime? ReadDateTimeObject(Stream stream) => ReadObject(stream, 'D', ReadDateTime);
+        protected int? ReadIntObject(Stream stream) => ReadStruct(stream, PrefixInt32, ReadInt);
+        protected long? ReadLongObject(Stream stream) => ReadStruct(stream, PrefixLong, ReadLong);
+        protected float? ReadFloatObject(Stream stream) => ReadStruct(stream, PrefixFloat, ReadFloat);
+        protected double? ReadDoubleObject(Stream stream) => ReadStruct(stream, PrefixDouble, ReadDouble);
+        protected DateTime? ReadDateTimeObject(Stream stream) => ReadStruct(stream, PrefixDateTime, ReadDateTime);
 
         protected object ReadObject(Stream stream)
             => ReadObject(PrimitiveSerializer.Char.ReadFrom(stream), stream);
 
+        /// <summary>
+        /// Reads in a value based on its char type
+        /// </summary>
+        /// <param name="type"></param>
+        /// <param name="stream"></param>
+        /// <returns></returns>
+        /// <remarks>
+        /// This will incur boxing because the result is an object but in most cases the value will be a struct.
+        /// When the type is known use the specific methods like <see cref="ReadInt(Stream)"/> instead
+        /// </remarks>
         protected object ReadObject(char type, Stream stream)
         {
             switch (type)
             {
-                case 'N':
+                case PrefixNull:
                     return null;
-                case 'S':
+                case PrefixString:
                     return PrimitiveSerializer.String.ReadFrom(stream);
-                case 'I':
+                case PrefixInt32:
                     return PrimitiveSerializer.Int32.ReadFrom(stream);
-                case 'L':
+                case PrefixUInt16:
+                    return PrimitiveSerializer.UInt16.ReadFrom(stream);
+                case PrefixUInt32:
+                    return PrimitiveSerializer.UInt32.ReadFrom(stream);
+                case PrefixByte:
+                    return PrimitiveSerializer.Byte.ReadFrom(stream);
+                case PrefixLong:
                     return PrimitiveSerializer.Int64.ReadFrom(stream);
-                case 'F':
+                case PrefixFloat:
                     return PrimitiveSerializer.Float.ReadFrom(stream);
-                case 'B':
+                case PrefixDouble:
                     return PrimitiveSerializer.Double.ReadFrom(stream);
-                case 'D':
+                case PrefixDateTime:
                     return PrimitiveSerializer.DateTime.ReadFrom(stream);
+                case PrefixByteArray:
+                    return PrimitiveSerializer.Bytes.ReadFrom(stream);
+                case PrefixSignedByte:
+                    return PrimitiveSerializer.SByte.ReadFrom(stream);
+                case PrefixBool:
+                    return PrimitiveSerializer.Boolean.ReadFrom(stream);
+                case PrefixGuid:
+                    return PrimitiveSerializer.Guid.ReadFrom(stream);
+                case PrefixTimeSpan:
+                    return PrimitiveSerializer.TimeSpan.ReadFrom(stream);
+                case PrefixInt16:
+                    return PrimitiveSerializer.Int16.ReadFrom(stream);
+                case PrefixChar:
+                    return PrimitiveSerializer.Char.ReadFrom(stream);
+                case PrefixCompressedStringByteArray:
+                    return new LazyCompressedString(PrimitiveSerializer.Bytes.ReadFrom(stream));
                 default:
                     throw new NotSupportedException($"Cannot deserialize unknown type '{type}'.");
             }
         }
 
+        /// <summary>
+        /// Writes a value to the stream ensuring it's char type is prefixed to the value for reading later
+        /// </summary>
+        /// <param name="value"></param>
+        /// <param name="stream"></param>
+        /// <remarks>
+        /// This method will incur boxing if the value is a struct. When the type is known use the <see cref="PrimitiveSerializer"/>
+        /// to write the value directly.
+        /// </remarks>
         protected void WriteObject(object value, Stream stream)
         {
             if (value == null)
             {
-                PrimitiveSerializer.Char.WriteTo('N', stream);
+                PrimitiveSerializer.Char.WriteTo(PrefixNull, stream);
             }
             else if (value is string stringValue)
             {
-                PrimitiveSerializer.Char.WriteTo('S', stream);
+                PrimitiveSerializer.Char.WriteTo(PrefixString, stream);
                 PrimitiveSerializer.String.WriteTo(stringValue, stream);
             }
             else if (value is int intValue)
             {
-                PrimitiveSerializer.Char.WriteTo('I', stream);
+                PrimitiveSerializer.Char.WriteTo(PrefixInt32, stream);
                 PrimitiveSerializer.Int32.WriteTo(intValue, stream);
             }
+            else if (value is byte byteValue)
+            {
+                PrimitiveSerializer.Char.WriteTo(PrefixByte, stream);
+                PrimitiveSerializer.Byte.WriteTo(byteValue, stream);
+            }
+            else if (value is ushort ushortValue)
+            {
+                PrimitiveSerializer.Char.WriteTo(PrefixUInt16, stream);
+                PrimitiveSerializer.UInt16.WriteTo(ushortValue, stream);
+            }
             else if (value is long longValue)
             {
-                PrimitiveSerializer.Char.WriteTo('L', stream);
+                PrimitiveSerializer.Char.WriteTo(PrefixLong, stream);
                 PrimitiveSerializer.Int64.WriteTo(longValue, stream);
             }
             else if (value is float floatValue)
             {
-                PrimitiveSerializer.Char.WriteTo('F', stream);
+                PrimitiveSerializer.Char.WriteTo(PrefixFloat, stream);
                 PrimitiveSerializer.Float.WriteTo(floatValue, stream);
             }
             else if (value is double doubleValue)
             {
-                PrimitiveSerializer.Char.WriteTo('B', stream);
+                PrimitiveSerializer.Char.WriteTo(PrefixDouble, stream);
                 PrimitiveSerializer.Double.WriteTo(doubleValue, stream);
             }
             else if (value is DateTime dateValue)
             {
-                PrimitiveSerializer.Char.WriteTo('D', stream);
+                PrimitiveSerializer.Char.WriteTo(PrefixDateTime, stream);
                 PrimitiveSerializer.DateTime.WriteTo(dateValue, stream);
             }
+            else if (value is uint uInt32Value)
+            {
+                PrimitiveSerializer.Char.WriteTo(PrefixUInt32, stream);
+                PrimitiveSerializer.UInt32.WriteTo(uInt32Value, stream);
+            }
+            else if (value is byte[] byteArrayValue)
+            {
+                PrimitiveSerializer.Char.WriteTo(PrefixByteArray, stream);
+                PrimitiveSerializer.Bytes.WriteTo(byteArrayValue, stream);
+            }
+            else if (value is LazyCompressedString lazyCompressedString)
+            {
+                PrimitiveSerializer.Char.WriteTo(PrefixCompressedStringByteArray, stream);
+                PrimitiveSerializer.Bytes.WriteTo(lazyCompressedString.GetBytes(), stream);
+            }
+            else if (value is sbyte signedByteValue)
+            {
+                PrimitiveSerializer.Char.WriteTo(PrefixSignedByte, stream);
+                PrimitiveSerializer.SByte.WriteTo(signedByteValue, stream);
+            }
+            else if (value is bool boolValue)
+            {
+                PrimitiveSerializer.Char.WriteTo(PrefixBool, stream);
+                PrimitiveSerializer.Boolean.WriteTo(boolValue, stream);
+            }
+            else if (value is Guid guidValue)
+            {
+                PrimitiveSerializer.Char.WriteTo(PrefixGuid, stream);
+                PrimitiveSerializer.Guid.WriteTo(guidValue, stream);
+            }
+            else if (value is TimeSpan timespanValue)
+            {
+                PrimitiveSerializer.Char.WriteTo(PrefixTimeSpan, stream);
+                PrimitiveSerializer.TimeSpan.WriteTo(timespanValue, stream);
+            }
+            else if (value is short int16Value)
+            {
+                PrimitiveSerializer.Char.WriteTo(PrefixInt16, stream);
+                PrimitiveSerializer.Int16.WriteTo(int16Value, stream);
+            }
+            else if (value is char charValue)
+            {
+                PrimitiveSerializer.Char.WriteTo(PrefixChar, stream);
+                PrimitiveSerializer.Char.WriteTo(charValue, stream);
+            }
             else
                 throw new NotSupportedException("Value type " + value.GetType().FullName + " cannot be serialized.");
         }

src/Umbraco.PublishedCache.NuCache/DependencyInjection/UmbracoBuilderExtensions.cs

@@ -1,12 +1,17 @@
+using System;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
+using Microsoft.Extensions.Options;
+using Umbraco.Cms.Core.Configuration.Models;
 using Umbraco.Cms.Core.DependencyInjection;
 using Umbraco.Cms.Core.Models;
 using Umbraco.Cms.Core.Notifications;
+using Umbraco.Cms.Core.PropertyEditors;
 using Umbraco.Cms.Core.PublishedCache;
 using Umbraco.Cms.Core.Scoping;
 using Umbraco.Cms.Core.Services;
 using Umbraco.Cms.Infrastructure.PublishedCache;
+using Umbraco.Cms.Infrastructure.PublishedCache.DataSource;
 using Umbraco.Cms.Infrastructure.PublishedCache.Persistence;
 
 namespace Umbraco.Extensions
@@ -49,6 +54,23 @@ namespace Umbraco.Extensions
 
             builder.AddNuCacheNotifications();
 
+            builder.AddNotificationHandler<UmbracoApplicationStartingNotification, NuCacheStartupHandler>();
+            builder.Services.AddSingleton<IContentCacheDataSerializerFactory>(s =>
+            {
+                IOptions<NuCacheSettings> options = s.GetRequiredService<IOptions<NuCacheSettings>>();
+                switch (options.Value.NuCacheSerializerType)
+                {
+                    case NuCacheSerializerType.JSON:
+                        return new JsonContentNestedDataSerializerFactory();
+                    case NuCacheSerializerType.MessagePack:
+                        return ActivatorUtilities.CreateInstance<MsgPackContentNestedDataSerializerFactory>(s);
+                    default:
+                        throw new IndexOutOfRangeException();
+                }
+            });
+            builder.Services.AddSingleton<IPropertyCacheCompressionOptions, NoopPropertyCacheCompressionOptions>();
+            builder.Services.AddSingleton(s => new ContentDataSerializer(new DictionaryOfPropertyDataSerializer()));
+
             // add the NuCache health check (hidden from type finder)
             // TODO: no NuCache health check yet
             // composition.HealthChecks().Add<NuCacheIntegrityHealthCheck>();
@@ -61,6 +83,7 @@ namespace Umbraco.Extensions
             builder
                 .AddNotificationHandler<LanguageSavedNotification, PublishedSnapshotServiceEventHandler>()
                 .AddNotificationHandler<MemberDeletingNotification, PublishedSnapshotServiceEventHandler>()
+#pragma warning disable CS0618 // Type or member is obsolete
                 .AddNotificationHandler<ContentRefreshNotification, PublishedSnapshotServiceEventHandler>()
                 .AddNotificationHandler<MediaRefreshNotification, PublishedSnapshotServiceEventHandler>()
                 .AddNotificationHandler<MemberRefreshNotification, PublishedSnapshotServiceEventHandler>()
@@ -68,6 +91,7 @@ namespace Umbraco.Extensions
                 .AddNotificationHandler<MediaTypeRefreshedNotification, PublishedSnapshotServiceEventHandler>()
                 .AddNotificationHandler<MemberTypeRefreshedNotification, PublishedSnapshotServiceEventHandler>()
                 .AddNotificationHandler<ScopedEntityRemoveNotification, PublishedSnapshotServiceEventHandler>()
+#pragma warning restore CS0618 // Type or member is obsolete
                 ;
 
             return builder;

src/Umbraco.PublishedCache.NuCache/NuCacheStartupHandler.cs

@@ -0,0 +1,67 @@
+using System.Configuration;
+using Microsoft.Extensions.Logging;
+using Umbraco.Cms.Core.Events;
+using Umbraco.Cms.Core.Logging;
+using Umbraco.Cms.Core.Notifications;
+using Umbraco.Cms.Core.PublishedCache;
+using Umbraco.Cms.Core.Services;
+
+namespace Umbraco.Cms.Infrastructure.PublishedCache
+{
+    /// <summary>
+    /// Rebuilds the database cache if required when the serializer changes
+    /// </summary>
+    public class NuCacheStartupHandler : INotificationHandler<UmbracoApplicationStartingNotification>
+    {
+        // TODO: Eventually we should kill this since at some stage we shouldn't even support JSON since we know
+        // this is faster.
+
+        internal const string Nucache_Serializer_Key = "Umbraco.Web.PublishedCache.NuCache.Serializer";
+        private const string JSON_SERIALIZER_VALUE = "JSON";
+        private readonly IPublishedSnapshotService _service;
+        private readonly IKeyValueService _keyValueService;
+        private readonly IProfilingLogger _profilingLogger;
+        private readonly ILogger<NuCacheStartupHandler> _logger;
+
+        public NuCacheStartupHandler(
+            IPublishedSnapshotService service,
+            IKeyValueService keyValueService,
+            IProfilingLogger profilingLogger,
+            ILogger<NuCacheStartupHandler> logger)
+        {
+            _service = service;
+            _keyValueService = keyValueService;
+            _profilingLogger = profilingLogger;
+            _logger = logger;
+        }
+
+        public void Handle(UmbracoApplicationStartingNotification notification)
+            => RebuildDatabaseCacheIfSerializerChanged();
+
+        private void RebuildDatabaseCacheIfSerializerChanged()
+        {
+            var serializer = ConfigurationManager.AppSettings[Nucache_Serializer_Key];
+            var currentSerializer = _keyValueService.GetValue(Nucache_Serializer_Key);
+
+            if (currentSerializer == null)
+            {
+                currentSerializer = JSON_SERIALIZER_VALUE;
+            }
+            if (serializer == null)
+            {
+                serializer = JSON_SERIALIZER_VALUE;
+            }
+
+            if (serializer != currentSerializer)
+            {
+                _logger.LogWarning("Database NuCache was serialized using {CurrentSerializer}. Currently configured NuCache serializer {Serializer}. Rebuilding Nucache", currentSerializer, serializer);
+
+                using (_profilingLogger.TraceDuration<NuCacheStartupHandler>($"Rebuilding NuCache database with {currentSerializer} serializer"))
+                {
+                    _service.Rebuild();
+                    _keyValueService.SetValue(Nucache_Serializer_Key, serializer);
+                }
+            }
+        }
+    }
+}

src/Umbraco.PublishedCache.NuCache/Persistence/INuCacheContentRepository.cs

@@ -21,19 +21,22 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.Persistence
         void RefreshContent(IContent content);
 
         /// <summary>
-        /// Refreshes the nucache database row for the <see cref="IContentBase"/> (used for media/members)
+        /// Refreshes the nucache database row for the <see cref="IMedia"/>
         /// </summary>
-        void RefreshEntity(IContentBase content);
+        void RefreshMedia(IMedia content);
+
+        /// <summary>
+        /// Refreshes the nucache database row for the <see cref="IMember"/>
+        /// </summary>
+        void RefreshMember(IMember content);
 
         /// <summary>
         /// Rebuilds the caches for content, media and/or members based on the content type ids specified
         /// </summary>
-        /// <param name="groupSize">The operation batch size to process the items</param>
         /// <param name="contentTypeIds">If not null will process content for the matching content types, if empty will process all content</param>
         /// <param name="mediaTypeIds">If not null will process content for the matching media types, if empty will process all media</param>
         /// <param name="memberTypeIds">If not null will process content for the matching members types, if empty will process all members</param>
         void Rebuild(
-            int groupSize = 5000,
             IReadOnlyCollection<int> contentTypeIds = null,
             IReadOnlyCollection<int> mediaTypeIds = null,
             IReadOnlyCollection<int> memberTypeIds = null);

src/Umbraco.PublishedCache.NuCache/Persistence/INuCacheContentService.cs

@@ -71,19 +71,22 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.Persistence
         void RefreshContent(IContent content);
 
         /// <summary>
-        /// Refreshes the nucache database row for the <see cref="IContentBase"/> (used for media/members)
+        /// Refreshes the nucache database row for the <see cref="IMedia"/>
         /// </summary>
-        void RefreshEntity(IContentBase content);
+        void RefreshMedia(IMedia media);
+
+        /// <summary>
+        /// Refreshes the nucache database row for the <see cref="IMember"/>
+        /// </summary>
+        void RefreshMember(IMember member);
 
         /// <summary>
         /// Rebuilds the database caches for content, media and/or members based on the content type ids specified
         /// </summary>
-        /// <param name="groupSize">The operation batch size to process the items</param>
         /// <param name="contentTypeIds">If not null will process content for the matching content types, if empty will process all content</param>
         /// <param name="mediaTypeIds">If not null will process content for the matching media types, if empty will process all media</param>
         /// <param name="memberTypeIds">If not null will process content for the matching members types, if empty will process all members</param>
         void Rebuild(
-            int groupSize = 5000,
             IReadOnlyCollection<int> contentTypeIds = null,
             IReadOnlyCollection<int> mediaTypeIds = null,
             IReadOnlyCollection<int> memberTypeIds = null);

src/Umbraco.PublishedCache.NuCache/Persistence/NuCacheContentRepository.cs

@@ -3,9 +3,11 @@ using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
 using Newtonsoft.Json;
 using NPoco;
 using Umbraco.Cms.Core.Cache;
+using Umbraco.Cms.Core.Configuration.Models;
 using Umbraco.Cms.Core.Models;
 using Umbraco.Cms.Core.Persistence.Querying;
 using Umbraco.Cms.Core.Persistence.Repositories;
@@ -16,7 +18,6 @@ using Umbraco.Cms.Infrastructure.Persistence;
 using Umbraco.Cms.Infrastructure.Persistence.Dtos;
 using Umbraco.Cms.Infrastructure.Persistence.Repositories.Implement;
 using Umbraco.Cms.Infrastructure.PublishedCache.DataSource;
-using Umbraco.Cms.Infrastructure.Serialization;
 using Umbraco.Extensions;
 using static Umbraco.Cms.Core.Persistence.SqlExtensionsStatics;
 using Constants = Umbraco.Cms.Core.Constants;
@@ -25,13 +26,14 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.Persistence
 {
     public class NuCacheContentRepository : RepositoryBase, INuCacheContentRepository
     {
-        private const int PageSize = 500;
         private readonly ILogger<NuCacheContentRepository> _logger;
         private readonly IMemberRepository _memberRepository;
         private readonly IDocumentRepository _documentRepository;
         private readonly IMediaRepository _mediaRepository;
         private readonly IShortStringHelper _shortStringHelper;
         private readonly UrlSegmentProviderCollection _urlSegmentProviders;
+        private readonly IContentCacheDataSerializerFactory _contentCacheDataSerializerFactory;
+        private readonly IOptions<NuCacheSettings> _nucacheSettings;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="NuCacheContentRepository"/> class.
@@ -44,7 +46,9 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.Persistence
             IDocumentRepository documentRepository,
             IMediaRepository mediaRepository,
             IShortStringHelper shortStringHelper,
-            UrlSegmentProviderCollection urlSegmentProviders)
+            UrlSegmentProviderCollection urlSegmentProviders,
+            IContentCacheDataSerializerFactory contentCacheDataSerializerFactory,
+            IOptions<NuCacheSettings> nucacheSettings)
             : base(scopeAccessor, appCaches)
         {
             _logger = logger;
@@ -53,6 +57,8 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.Persistence
             _mediaRepository = mediaRepository;
             _shortStringHelper = shortStringHelper;
             _urlSegmentProviders = urlSegmentProviders;
+            _contentCacheDataSerializerFactory = contentCacheDataSerializerFactory;
+            _nucacheSettings = nucacheSettings;
         }
 
         public void DeleteContentItem(IContentBase item)
@@ -60,8 +66,10 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.Persistence
 
         public void RefreshContent(IContent content)
         {
+            IContentCacheDataSerializer serializer = _contentCacheDataSerializerFactory.Create(ContentCacheDataSerializerEntityType.Document);
+
             // always refresh the edited data
-            OnRepositoryRefreshed(content, false);
+            OnRepositoryRefreshed(serializer, content, false);
 
             if (content.PublishedState == PublishedState.Unpublishing)
             {
@@ -71,24 +79,36 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.Persistence
             else if (content.PublishedState == PublishedState.Publishing)
             {
                 // if publishing, refresh the published data
-                OnRepositoryRefreshed(content, true);
+                OnRepositoryRefreshed(serializer, content, true);
             }
         }
 
-        public void RefreshEntity(IContentBase content)
-            => OnRepositoryRefreshed(content, false);
+        public void RefreshMedia(IMedia media)
+        {
+            IContentCacheDataSerializer serializer = _contentCacheDataSerializerFactory.Create(ContentCacheDataSerializerEntityType.Media);
 
-        private void OnRepositoryRefreshed(IContentBase content, bool published)
+            OnRepositoryRefreshed(serializer, media, false);
+        }
+
+        public void RefreshMember(IMember member)
+        {
+            IContentCacheDataSerializer serializer = _contentCacheDataSerializerFactory.Create(ContentCacheDataSerializerEntityType.Member);
+
+            OnRepositoryRefreshed(serializer, member, false);
+        }
+
+        private void OnRepositoryRefreshed(IContentCacheDataSerializer serializer, IContentBase content, bool published)
         {
             // use a custom SQL to update row version on each update
             // db.InsertOrUpdate(dto);
-            ContentNuDto dto = GetDto(content, published);
+            ContentNuDto dto = GetDto(content, published, serializer);
 
             Database.InsertOrUpdate(
                 dto,
-                "SET data=@data, rv=rv+1 WHERE nodeId=@id AND published=@published",
+                "SET data=@data, dataRaw=@dataRaw, rv=rv+1 WHERE nodeId=@id AND published=@published",
                 new
                 {
+                    dataRaw = dto.RawData ?? Array.Empty<byte>(),
                     data = dto.Data,
                     id = dto.NodeId,
                     published = dto.Published
@@ -96,18 +116,22 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.Persistence
         }
 
         public void Rebuild(
-            int groupSize = 5000,
             IReadOnlyCollection<int> contentTypeIds = null,
             IReadOnlyCollection<int> mediaTypeIds = null,
             IReadOnlyCollection<int> memberTypeIds = null)
         {
-            RebuildContentDbCache(groupSize, contentTypeIds);
-            RebuildContentDbCache(groupSize, mediaTypeIds);
-            RebuildContentDbCache(groupSize, memberTypeIds);
+            IContentCacheDataSerializer serializer = _contentCacheDataSerializerFactory.Create(
+                ContentCacheDataSerializerEntityType.Document
+                | ContentCacheDataSerializerEntityType.Media
+                | ContentCacheDataSerializerEntityType.Member);
+
+            RebuildContentDbCache(serializer, _nucacheSettings.Value.SqlPageSize, contentTypeIds);
+            RebuildMediaDbCache(serializer, _nucacheSettings.Value.SqlPageSize, mediaTypeIds);
+            RebuildMemberDbCache(serializer, _nucacheSettings.Value.SqlPageSize, memberTypeIds);
         }
 
         // assumes content tree lock
-        private void RebuildContentDbCache(int groupSize, IReadOnlyCollection<int> contentTypeIds)
+        private void RebuildContentDbCache(IContentCacheDataSerializer serializer, int groupSize, IReadOnlyCollection<int> contentTypeIds)
         {
             Guid contentObjectType = Constants.ObjectTypes.Document;
 
@@ -156,12 +180,12 @@ WHERE cmsContentNu.nodeId IN (
                 foreach (IContent c in descendants)
                 {
                     // always the edited version
-                    items.Add(GetDto(c, false));
+                    items.Add(GetDto(c, false, serializer));
 
                     // and also the published version if it makes any sense
                     if (c.Published)
                     {
-                        items.Add(GetDto(c, true));
+                        items.Add(GetDto(c, true, serializer));
                     }
 
                     count++;
@@ -173,7 +197,7 @@ WHERE cmsContentNu.nodeId IN (
         }
 
         // assumes media tree lock
-        private void RebuildMediaDbCache(int groupSize, IReadOnlyCollection<int> contentTypeIds)
+        private void RebuildMediaDbCache(IContentCacheDataSerializer serializer, int groupSize, IReadOnlyCollection<int> contentTypeIds)
         {
             var mediaObjectType = Constants.ObjectTypes.Media;
 
@@ -217,14 +241,14 @@ WHERE cmsContentNu.nodeId IN (
             {
                 // the tree is locked, counting and comparing to total is safe
                 var descendants = _mediaRepository.GetPage(query, pageIndex++, groupSize, out total, null, Ordering.By("Path"));
-                var items = descendants.Select(m => GetDto(m, false)).ToList();
+                var items = descendants.Select(m => GetDto(m, false, serializer)).ToList();
                 Database.BulkInsertRecords(items);
                 processed += items.Count;
             } while (processed < total);
         }
 
         // assumes member tree lock
-        private void RebuildMemberDbCache(int groupSize, IReadOnlyCollection<int> contentTypeIds)
+        private void RebuildMemberDbCache(IContentCacheDataSerializer serializer, int groupSize, IReadOnlyCollection<int> contentTypeIds)
         {
             Guid memberObjectType = Constants.ObjectTypes.Member;
 
@@ -267,7 +291,7 @@ WHERE cmsContentNu.nodeId IN (
             do
             {
                 IEnumerable<IMember> descendants = _memberRepository.GetPage(query, pageIndex++, groupSize, out total, null, Ordering.By("Path"));
-                ContentNuDto[] items = descendants.Select(m => GetDto(m, false)).ToArray();
+                ContentNuDto[] items = descendants.Select(m => GetDto(m, false, serializer)).ToArray();
                 Database.BulkInsertRecords(items);
                 processed += items.Length;
             } while (processed < total);
@@ -327,7 +351,7 @@ AND cmsContentNu.nodeId IS NULL
             return count == 0;
         }
 
-        private ContentNuDto GetDto(IContentBase content, bool published)
+        private ContentNuDto GetDto(IContentBase content, bool published, IContentCacheDataSerializer serializer)
         {
             // should inject these in ctor
             // BUT for the time being we decide not to support ConvertDbToXml/String
@@ -382,32 +406,31 @@ AND cmsContentNu.nodeId IS NULL
             }
 
             // the dictionary that will be serialized
-            var nestedData = new ContentNestedData
+            var contentCacheData = new ContentCacheDataModel
             {
                 PropertyData = propertyData,
                 CultureData = cultureData,
                 UrlSegment = content.GetUrlSegment(_shortStringHelper, _urlSegmentProviders)
             };
 
+            var serialized = serializer.Serialize(ReadOnlyContentBaseAdapter.Create(content), contentCacheData);
+
             var dto = new ContentNuDto
             {
                 NodeId = content.Id,
                 Published = published,
-
-                // note that numeric values (which are Int32) are serialized without their
-                // type (eg "value":1234) and JsonConvert by default deserializes them as Int64
-                Data = JsonConvert.SerializeObject(nestedData)
+                Data = serialized.StringData,
+                RawData = serialized.ByteData
             };
 
             return dto;
         }
 
         // we want arrays, we want them all loaded, not an enumerable
-        private Sql<ISqlContext> ContentSourcesSelect(Func<Sql<ISqlContext>, Sql<ISqlContext>> joins = null)
+        private Sql<ISqlContext> SqlContentSourcesSelect(Func<ISqlContext, Sql<ISqlContext>> joins = null)
         {
-            var sql = Sql()
-
-                .Select<NodeDto>(x => Alias(x.NodeId, "Id"), x => Alias(x.UniqueId, "Uid"),
+            var sqlTemplate = SqlContext.Templates.Get(Constants.SqlTemplates.NuCacheDatabaseDataSource.ContentSourcesSelect, tsql =>
+                tsql.Select<NodeDto>(x => Alias(x.NodeId, "Id"), x => Alias(x.UniqueId, "Key"),
                     x => Alias(x.Level, "Level"), x => Alias(x.Path, "Path"), x => Alias(x.SortOrder, "SortOrder"), x => Alias(x.ParentId, "ParentId"),
                     x => Alias(x.CreateDate, "CreateDate"), x => Alias(x.UserId, "CreatorId"))
                 .AndSelect<ContentDto>(x => Alias(x.ContentTypeId, "ContentTypeId"))
@@ -422,12 +445,17 @@ AND cmsContentNu.nodeId IS NULL
                 .AndSelect<ContentNuDto>("nuEdit", x => Alias(x.Data, "EditData"))
                 .AndSelect<ContentNuDto>("nuPub", x => Alias(x.Data, "PubData"))
 
-                .From<NodeDto>();
+                .AndSelect<ContentNuDto>("nuEdit", x => Alias(x.RawData, "EditDataRaw"))
+                .AndSelect<ContentNuDto>("nuPub", x => Alias(x.RawData, "PubDataRaw"))
+
+                .From<NodeDto>());
+
+            var sql = sqlTemplate.Sql();
+
+            // TODO: I'm unsure how we can format the below into SQL templates also because right.Current and right.Published end up being parameters
 
             if (joins != null)
-            {
-                sql = joins(sql);
-            }
+                sql = sql.Append(joins(sql.SqlContext));
 
             sql = sql
                 .InnerJoin<ContentDto>().On<NodeDto, ContentDto>((left, right) => left.NodeId == right.NodeId)
@@ -437,94 +465,118 @@ AND cmsContentNu.nodeId IS NULL
                 .InnerJoin<DocumentVersionDto>().On<ContentVersionDto, DocumentVersionDto>((left, right) => left.Id == right.Id)
 
                 .LeftJoin<ContentVersionDto>(j =>
-                    j.InnerJoin<DocumentVersionDto>("pdver").On<ContentVersionDto, DocumentVersionDto>((left, right) => left.Id == right.Id && right.Published, "pcver", "pdver"), "pcver")
+                    j.InnerJoin<DocumentVersionDto>("pdver").On<ContentVersionDto, DocumentVersionDto>((left, right) => left.Id == right.Id && right.Published == true, "pcver", "pdver"), "pcver")
                 .On<NodeDto, ContentVersionDto>((left, right) => left.NodeId == right.NodeId, aliasRight: "pcver")
 
-                .LeftJoin<ContentNuDto>("nuEdit").On<NodeDto, ContentNuDto>((left, right) => left.NodeId == right.NodeId && !right.Published, aliasRight: "nuEdit")
-                .LeftJoin<ContentNuDto>("nuPub").On<NodeDto, ContentNuDto>((left, right) => left.NodeId == right.NodeId && right.Published, aliasRight: "nuPub");
+                .LeftJoin<ContentNuDto>("nuEdit").On<NodeDto, ContentNuDto>((left, right) => left.NodeId == right.NodeId && right.Published == false, aliasRight: "nuEdit")
+                .LeftJoin<ContentNuDto>("nuPub").On<NodeDto, ContentNuDto>((left, right) => left.NodeId == right.NodeId && right.Published == true, aliasRight: "nuPub");
 
             return sql;
         }
 
-        public ContentNodeKit GetContentSource(int id)
+        private Sql<ISqlContext> SqlContentSourcesSelectUmbracoNodeJoin(ISqlContext sqlContext)
         {
-            var sql = ContentSourcesSelect()
-                .Where<NodeDto>(x => x.NodeObjectType == Constants.ObjectTypes.Document && x.NodeId == id && !x.Trashed)
-                .OrderBy<NodeDto>(x => x.Level, x => x.ParentId, x => x.SortOrder);
+            var syntax = sqlContext.SqlSyntax;
 
-            var dto = Database.Fetch<ContentSourceDto>(sql).FirstOrDefault();
-            return dto == null ? new ContentNodeKit() : CreateContentNodeKit(dto);
+            var sqlTemplate = sqlContext.Templates.Get(Constants.SqlTemplates.NuCacheDatabaseDataSource.SourcesSelectUmbracoNodeJoin, builder =>
+                builder.InnerJoin<NodeDto>("x")
+                    .On<NodeDto, NodeDto>((left, right) => left.NodeId == right.NodeId || SqlText<bool>(left.Path, right.Path, (lp, rp) => $"({lp} LIKE {syntax.GetConcat(rp, "',%'")})"), aliasRight: "x"));
+
+            var sql = sqlTemplate.Sql();
+            return sql;
         }
 
-        public IEnumerable<ContentNodeKit> GetAllContentSources()
+        private Sql<ISqlContext> SqlWhereNodeId(ISqlContext sqlContext, int id)
         {
-            var sql = ContentSourcesSelect()
-                .Where<NodeDto>(x => x.NodeObjectType == Constants.ObjectTypes.Document && !x.Trashed)
-                .OrderBy<NodeDto>(x => x.Level, x => x.ParentId, x => x.SortOrder);
+            var syntax = sqlContext.SqlSyntax;
 
-            // We need to page here. We don't want to iterate over every single row in one connection cuz this can cause an SQL Timeout.
-            // We also want to read with a db reader and not load everything into memory, QueryPaged lets us do that.
+            var sqlTemplate = sqlContext.Templates.Get(Constants.SqlTemplates.NuCacheDatabaseDataSource.WhereNodeId, builder =>
+                builder.Where<NodeDto>(x => x.NodeId == SqlTemplate.Arg<int>("id")));
 
-            foreach (var row in Database.QueryPaged<ContentSourceDto>(PageSize, sql))
-            {
-                yield return CreateContentNodeKit(row);
-            }
+            var sql = sqlTemplate.Sql(id);
+            return sql;
         }
 
-        public IEnumerable<ContentNodeKit> GetBranchContentSources(int id)
+        private Sql<ISqlContext> SqlWhereNodeIdX(ISqlContext sqlContext, int id)
         {
-            var syntax = SqlSyntax;
-            var sql = ContentSourcesSelect(
-                s => s.InnerJoin<NodeDto>("x").On<NodeDto, NodeDto>((left, right) => left.NodeId == right.NodeId || SqlText<bool>(left.Path, right.Path, (lp, rp) => $"({lp} LIKE {syntax.GetConcat(rp, "',%'")})"), aliasRight: "x"))
-                    .Where<NodeDto>(x => x.NodeObjectType == Constants.ObjectTypes.Document && !x.Trashed)
-                    .Where<NodeDto>(x => x.NodeId == id, "x")
-                    .OrderBy<NodeDto>(x => x.Level, x => x.ParentId, x => x.SortOrder);
+            var syntax = sqlContext.SqlSyntax;
 
-            // We need to page here. We don't want to iterate over every single row in one connection cuz this can cause an SQL Timeout.
-            // We also want to read with a db reader and not load everything into memory, QueryPaged lets us do that.
+            var sqlTemplate = sqlContext.Templates.Get(Constants.SqlTemplates.NuCacheDatabaseDataSource.WhereNodeIdX, s =>
+                s.Where<NodeDto>(x => x.NodeId == SqlTemplate.Arg<int>("id"), "x"));
 
-            foreach (var row in Database.QueryPaged<ContentSourceDto>(PageSize, sql))
-            {
-                yield return CreateContentNodeKit(row);
-            }
+            var sql = sqlTemplate.Sql(id);
+            return sql;
         }
 
-        public IEnumerable<ContentNodeKit> GetTypeContentSources(IEnumerable<int> ids)
+        private Sql<ISqlContext> SqlOrderByLevelIdSortOrder(ISqlContext sqlContext)
         {
-            if (!ids.Any())
-                yield break;
+            var syntax = sqlContext.SqlSyntax;
 
-            var sql = ContentSourcesSelect()
-                .Where<NodeDto>(x => x.NodeObjectType == Constants.ObjectTypes.Document && !x.Trashed)
-                .WhereIn<ContentDto>(x => x.ContentTypeId, ids)
-                .OrderBy<NodeDto>(x => x.Level, x => x.ParentId, x => x.SortOrder);
+            var sqlTemplate = sqlContext.Templates.Get(Constants.SqlTemplates.NuCacheDatabaseDataSource.OrderByLevelIdSortOrder, s =>
+                s.OrderBy<NodeDto>(x => x.Level, x => x.ParentId, x => x.SortOrder));
 
-            // We need to page here. We don't want to iterate over every single row in one connection cuz this can cause an SQL Timeout.
-            // We also want to read with a db reader and not load everything into memory, QueryPaged lets us do that.
-
-            foreach (var row in Database.QueryPaged<ContentSourceDto>(PageSize, sql))
-            {
-                yield return CreateContentNodeKit(row);
-            }
+            var sql = sqlTemplate.Sql();
+            return sql;
         }
 
-        private Sql<ISqlContext> MediaSourcesSelect(Func<Sql<ISqlContext>, Sql<ISqlContext>> joins = null)
+        private Sql<ISqlContext> SqlObjectTypeNotTrashed(ISqlContext sqlContext, Guid nodeObjectType)
         {
-            var sql = Sql()
+            var syntax = sqlContext.SqlSyntax;
 
-                .Select<NodeDto>(x => Alias(x.NodeId, "Id"), x => Alias(x.UniqueId, "Uid"),
+            var sqlTemplate = sqlContext.Templates.Get(Constants.SqlTemplates.NuCacheDatabaseDataSource.ObjectTypeNotTrashedFilter, s =>
+                s.Where<NodeDto>(x => x.NodeObjectType == SqlTemplate.Arg<Guid?>("nodeObjectType") && x.Trashed == SqlTemplate.Arg<bool>("trashed")));
+
+            var sql = sqlTemplate.Sql(nodeObjectType, false);
+            return sql;
+        }
+
+        /// <summary>
+        /// Returns a slightly more optimized query to use for the document counting when paging over the content sources
+        /// </summary>
+        /// <param name="scope"></param>
+        /// <returns></returns>
+        private Sql<ISqlContext> SqlContentSourcesCount(Func<ISqlContext, Sql<ISqlContext>> joins = null)
+        {
+            var sqlTemplate = SqlContext.Templates.Get(Constants.SqlTemplates.NuCacheDatabaseDataSource.ContentSourcesCount, tsql =>
+                tsql.Select<NodeDto>(x => Alias(x.NodeId, "Id"))
+                    .From<NodeDto>()
+                    .InnerJoin<ContentDto>().On<NodeDto, ContentDto>((left, right) => left.NodeId == right.NodeId)
+                    .InnerJoin<DocumentDto>().On<NodeDto, DocumentDto>((left, right) => left.NodeId == right.NodeId));
+
+            var sql = sqlTemplate.Sql();
+
+            if (joins != null)
+                sql = sql.Append(joins(sql.SqlContext));
+
+            // TODO: We can't use a template with this one because of the 'right.Current' and 'right.Published' ends up being a parameter so not sure how we can do that
+            sql = sql
+                .InnerJoin<ContentVersionDto>().On<NodeDto, ContentVersionDto>((left, right) => left.NodeId == right.NodeId && right.Current)
+                .InnerJoin<DocumentVersionDto>().On<ContentVersionDto, DocumentVersionDto>((left, right) => left.Id == right.Id)
+                .LeftJoin<ContentVersionDto>(j =>
+                        j.InnerJoin<DocumentVersionDto>("pdver").On<ContentVersionDto, DocumentVersionDto>((left, right) => left.Id == right.Id && right.Published, "pcver", "pdver"), "pcver")
+                    .On<NodeDto, ContentVersionDto>((left, right) => left.NodeId == right.NodeId, aliasRight: "pcver");
+
+            return sql;
+        }
+
+        private Sql<ISqlContext> SqlMediaSourcesSelect(Func<ISqlContext, Sql<ISqlContext>> joins = null)
+        {
+            var sqlTemplate = SqlContext.Templates.Get(Constants.SqlTemplates.NuCacheDatabaseDataSource.MediaSourcesSelect, tsql =>
+                tsql.Select<NodeDto>(x => Alias(x.NodeId, "Id"), x => Alias(x.UniqueId, "Key"),
                     x => Alias(x.Level, "Level"), x => Alias(x.Path, "Path"), x => Alias(x.SortOrder, "SortOrder"), x => Alias(x.ParentId, "ParentId"),
                     x => Alias(x.CreateDate, "CreateDate"), x => Alias(x.UserId, "CreatorId"))
                 .AndSelect<ContentDto>(x => Alias(x.ContentTypeId, "ContentTypeId"))
                 .AndSelect<ContentVersionDto>(x => Alias(x.Id, "VersionId"), x => Alias(x.Text, "EditName"), x => Alias(x.VersionDate, "EditVersionDate"), x => Alias(x.UserId, "EditWriterId"))
                 .AndSelect<ContentNuDto>("nuEdit", x => Alias(x.Data, "EditData"))
-                .From<NodeDto>();
+                .AndSelect<ContentNuDto>("nuEdit", x => Alias(x.RawData, "EditDataRaw"))
+                .From<NodeDto>());
+
+            var sql = sqlTemplate.Sql();
 
             if (joins != null)
-            {
-                sql = joins(sql);
-            }
+                sql = sql.Append(joins(sql.SqlContext));
 
+            // TODO: We can't use a template with this one because of the 'right.Published' ends up being a parameter so not sure how we can do that
             sql = sql
                 .InnerJoin<ContentDto>().On<NodeDto, ContentDto>((left, right) => left.NodeId == right.NodeId)
                 .InnerJoin<ContentVersionDto>().On<NodeDto, ContentVersionDto>((left, right) => left.NodeId == right.NodeId && right.Current)
@@ -533,78 +585,226 @@ AND cmsContentNu.nodeId IS NULL
             return sql;
         }
 
-        public ContentNodeKit GetMediaSource(int id)
+        private Sql<ISqlContext> SqlMediaSourcesCount(Func<ISqlContext, Sql<ISqlContext>> joins = null)
         {
-            var sql = MediaSourcesSelect()
-                .Where<NodeDto>(x => x.NodeObjectType == Constants.ObjectTypes.Media && x.NodeId == id && !x.Trashed)
-                .OrderBy<NodeDto>(x => x.Level, x => x.ParentId, x => x.SortOrder);
+            var sqlTemplate = SqlContext.Templates.Get(Constants.SqlTemplates.NuCacheDatabaseDataSource.MediaSourcesCount, tsql =>
+               tsql.Select<NodeDto>(x => Alias(x.NodeId, "Id")).From<NodeDto>());
 
-            var dto = Database.Fetch<ContentSourceDto>(sql).FirstOrDefault();
-            return dto == null ? new ContentNodeKit() : CreateMediaNodeKit(dto);
+            var sql = sqlTemplate.Sql();
+
+            if (joins != null)
+                sql = sql.Append(joins(sql.SqlContext));
+
+            // TODO: We can't use a template with this one because of the 'right.Current' ends up being a parameter so not sure how we can do that
+            sql = sql
+                .InnerJoin<ContentDto>().On<NodeDto, ContentDto>((left, right) => left.NodeId == right.NodeId)
+                .InnerJoin<ContentVersionDto>().On<NodeDto, ContentVersionDto>((left, right) => left.NodeId == right.NodeId && right.Current);
+
+            return sql;
         }
 
-        public IEnumerable<ContentNodeKit> GetAllMediaSources()
+        public ContentNodeKit GetContentSource(int id)
         {
-            var sql = MediaSourcesSelect()
-                .Where<NodeDto>(x => x.NodeObjectType == Constants.ObjectTypes.Media && !x.Trashed)
-                .OrderBy<NodeDto>(x => x.Level, x => x.ParentId, x => x.SortOrder);
+            var sql = SqlContentSourcesSelect()
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Document))
+                .Append(SqlWhereNodeId(SqlContext, id))
+                .Append(SqlOrderByLevelIdSortOrder(SqlContext));
+
+            var dto = Database.Fetch<ContentSourceDto>(sql).FirstOrDefault();
+
+            if (dto == null) return ContentNodeKit.Empty;
+
+            var serializer = _contentCacheDataSerializerFactory.Create(ContentCacheDataSerializerEntityType.Document);
+            return CreateContentNodeKit(dto, serializer);
+        }
+
+        public IEnumerable<ContentNodeKit> GetAllContentSources()
+        {
+            var sql = SqlContentSourcesSelect()
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Document))
+                .Append(SqlOrderByLevelIdSortOrder(SqlContext));
+
+            // Use a more efficient COUNT query
+            var sqlCountQuery = SqlContentSourcesCount()
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Document));
+
+            var sqlCount = SqlContext.Sql("SELECT COUNT(*) FROM (").Append(sqlCountQuery).Append(") npoco_tbl");
+
+            var serializer = _contentCacheDataSerializerFactory.Create(ContentCacheDataSerializerEntityType.Document);
 
             // We need to page here. We don't want to iterate over every single row in one connection cuz this can cause an SQL Timeout.
             // We also want to read with a db reader and not load everything into memory, QueryPaged lets us do that.
 
-            foreach (var row in Database.QueryPaged<ContentSourceDto>(PageSize, sql))
+            foreach (var row in Database.QueryPaged<ContentSourceDto>(_nucacheSettings.Value.SqlPageSize, sql, sqlCount))
             {
-                yield return CreateMediaNodeKit(row);
+                yield return CreateContentNodeKit(row, serializer);
+            }
+        }
+
+        public IEnumerable<ContentNodeKit> GetBranchContentSources(int id)
+        {
+            var sql = SqlContentSourcesSelect(SqlContentSourcesSelectUmbracoNodeJoin)
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Document))
+                .Append(SqlWhereNodeIdX(SqlContext, id))
+                .Append(SqlOrderByLevelIdSortOrder(SqlContext));
+
+            // Use a more efficient COUNT query
+            var sqlCountQuery = SqlContentSourcesCount(SqlContentSourcesSelectUmbracoNodeJoin)
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Document))
+                .Append(SqlWhereNodeIdX(SqlContext, id));
+            var sqlCount = SqlContext.Sql("SELECT COUNT(*) FROM (").Append(sqlCountQuery).Append(") npoco_tbl");
+
+            var serializer = _contentCacheDataSerializerFactory.Create(ContentCacheDataSerializerEntityType.Document);
+
+            // We need to page here. We don't want to iterate over every single row in one connection cuz this can cause an SQL Timeout.
+            // We also want to read with a db reader and not load everything into memory, QueryPaged lets us do that.
+
+            foreach (var row in Database.QueryPaged<ContentSourceDto>(_nucacheSettings.Value.SqlPageSize, sql, sqlCount))
+            {
+                yield return CreateContentNodeKit(row, serializer);
+            }
+        }
+
+        public IEnumerable<ContentNodeKit> GetTypeContentSources(IEnumerable<int> ids)
+        {
+            if (!ids.Any())
+                yield break;
+
+            var sql = SqlContentSourcesSelect()
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Document))
+                .WhereIn<ContentDto>(x => x.ContentTypeId, ids)
+                .Append(SqlOrderByLevelIdSortOrder(SqlContext));
+
+            // Use a more efficient COUNT query
+            var sqlCountQuery = SqlContentSourcesCount()
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Document))
+                .WhereIn<ContentDto>(x => x.ContentTypeId, ids);
+            var sqlCount = SqlContext.Sql("SELECT COUNT(*) FROM (").Append(sqlCountQuery).Append(") npoco_tbl");
+
+            var serializer = _contentCacheDataSerializerFactory.Create(ContentCacheDataSerializerEntityType.Document);
+
+            // We need to page here. We don't want to iterate over every single row in one connection cuz this can cause an SQL Timeout.
+            // We also want to read with a db reader and not load everything into memory, QueryPaged lets us do that.
+
+            foreach (var row in Database.QueryPaged<ContentSourceDto>(_nucacheSettings.Value.SqlPageSize, sql, sqlCount))
+            {
+                yield return CreateContentNodeKit(row, serializer);
+            }
+        }
+
+        public ContentNodeKit GetMediaSource(IScope scope, int id)
+        {
+            var sql = SqlMediaSourcesSelect()
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Media))
+                .Append(SqlWhereNodeId(SqlContext, id))
+                .Append(SqlOrderByLevelIdSortOrder(scope.SqlContext));
+
+            var dto = scope.Database.Fetch<ContentSourceDto>(sql).FirstOrDefault();
+
+            if (dto == null)
+                return ContentNodeKit.Empty;
+
+            var serializer = _contentCacheDataSerializerFactory.Create(ContentCacheDataSerializerEntityType.Media);
+            return CreateMediaNodeKit(dto, serializer);
+        }
+
+        public ContentNodeKit GetMediaSource(int id)
+        {
+            var sql = SqlMediaSourcesSelect()
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Media))
+                .Append(SqlWhereNodeId(SqlContext, id))
+                .Append(SqlOrderByLevelIdSortOrder(SqlContext));
+
+            var dto = Database.Fetch<ContentSourceDto>(sql).FirstOrDefault();
+
+            if (dto == null)
+                return ContentNodeKit.Empty;
+
+            var serializer = _contentCacheDataSerializerFactory.Create(ContentCacheDataSerializerEntityType.Media);
+            return CreateMediaNodeKit(dto, serializer);
+        }
+
+        public IEnumerable<ContentNodeKit> GetAllMediaSources()
+        {
+            var sql = SqlMediaSourcesSelect()
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Media))
+                .Append(SqlOrderByLevelIdSortOrder(SqlContext));
+
+            // Use a more efficient COUNT query
+            var sqlCountQuery = SqlMediaSourcesCount()
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Media));
+            var sqlCount = SqlContext.Sql("SELECT COUNT(*) FROM (").Append(sqlCountQuery).Append(") npoco_tbl");
+
+            var serializer = _contentCacheDataSerializerFactory.Create(ContentCacheDataSerializerEntityType.Media);
+
+            // We need to page here. We don't want to iterate over every single row in one connection cuz this can cause an SQL Timeout.
+            // We also want to read with a db reader and not load everything into memory, QueryPaged lets us do that.
+
+            foreach (var row in Database.QueryPaged<ContentSourceDto>(_nucacheSettings.Value.SqlPageSize, sql, sqlCount))
+            {
+                yield return CreateMediaNodeKit(row, serializer);
             }
         }
 
         public IEnumerable<ContentNodeKit> GetBranchMediaSources(int id)
         {
-            var syntax = SqlSyntax;
-            var sql = MediaSourcesSelect(
-                s => s.InnerJoin<NodeDto>("x").On<NodeDto, NodeDto>((left, right) => left.NodeId == right.NodeId || SqlText<bool>(left.Path, right.Path, (lp, rp) => $"({lp} LIKE {syntax.GetConcat(rp, "',%'")})"), aliasRight: "x"))
-                .Where<NodeDto>(x => x.NodeObjectType == Constants.ObjectTypes.Media && !x.Trashed)
-                .Where<NodeDto>(x => x.NodeId == id, "x")
-                .OrderBy<NodeDto>(x => x.Level, x => x.ParentId, x => x.SortOrder);
+            var sql = SqlMediaSourcesSelect(SqlContentSourcesSelectUmbracoNodeJoin)
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Media))
+                .Append(SqlWhereNodeIdX(SqlContext, id))
+                .Append(SqlOrderByLevelIdSortOrder(SqlContext));
+
+            // Use a more efficient COUNT query
+            var sqlCountQuery = SqlMediaSourcesCount(SqlContentSourcesSelectUmbracoNodeJoin)
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Media))
+                .Append(SqlWhereNodeIdX(SqlContext, id));
+            var sqlCount = SqlContext.Sql("SELECT COUNT(*) FROM (").Append(sqlCountQuery).Append(") npoco_tbl");
+
+            var serializer = _contentCacheDataSerializerFactory.Create(ContentCacheDataSerializerEntityType.Media);
 
             // We need to page here. We don't want to iterate over every single row in one connection cuz this can cause an SQL Timeout.
             // We also want to read with a db reader and not load everything into memory, QueryPaged lets us do that.
 
-            foreach (var row in Database.QueryPaged<ContentSourceDto>(PageSize, sql))
+            foreach (var row in Database.QueryPaged<ContentSourceDto>(_nucacheSettings.Value.SqlPageSize, sql, sqlCount))
             {
-                yield return CreateMediaNodeKit(row);
+                yield return CreateMediaNodeKit(row, serializer);
             }
         }
 
         public IEnumerable<ContentNodeKit> GetTypeMediaSources(IEnumerable<int> ids)
         {
             if (!ids.Any())
-            {
                 yield break;
-            }
 
-            var sql = MediaSourcesSelect()
-                    .Where<NodeDto>(x => x.NodeObjectType == Constants.ObjectTypes.Media && !x.Trashed)
+            var sql = SqlMediaSourcesSelect()
+                    .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Media))
                     .WhereIn<ContentDto>(x => x.ContentTypeId, ids)
-                    .OrderBy<NodeDto>(x => x.Level, x => x.ParentId, x => x.SortOrder);
+                    .Append(SqlOrderByLevelIdSortOrder(SqlContext));
+
+            // Use a more efficient COUNT query
+            var sqlCountQuery = SqlMediaSourcesCount()
+                .Append(SqlObjectTypeNotTrashed(SqlContext, Constants.ObjectTypes.Media))
+                .WhereIn<ContentDto>(x => x.ContentTypeId, ids);
+            var sqlCount = SqlContext.Sql("SELECT COUNT(*) FROM (").Append(sqlCountQuery).Append(") npoco_tbl");
+
+            var serializer = _contentCacheDataSerializerFactory.Create(ContentCacheDataSerializerEntityType.Media);
 
             // We need to page here. We don't want to iterate over every single row in one connection cuz this can cause an SQL Timeout.
             // We also want to read with a db reader and not load everything into memory, QueryPaged lets us do that.
 
-            foreach (var row in Database.QueryPaged<ContentSourceDto>(PageSize, sql))
+            foreach (var row in Database.QueryPaged<ContentSourceDto>(_nucacheSettings.Value.SqlPageSize, sql, sqlCount))
             {
-                yield return CreateMediaNodeKit(row);
+                yield return CreateMediaNodeKit(row, serializer);
             }
         }
 
-        private ContentNodeKit CreateContentNodeKit(ContentSourceDto dto)
+        private ContentNodeKit CreateContentNodeKit(ContentSourceDto dto, IContentCacheDataSerializer serializer)
         {
             ContentData d = null;
             ContentData p = null;
 
             if (dto.Edited)
             {
-                if (dto.EditData == null)
+                if (dto.EditData == null && dto.EditDataRaw == null)
                 {
                     if (Debugger.IsAttached)
                     {
@@ -615,7 +815,7 @@ AND cmsContentNu.nodeId IS NULL
                 }
                 else
                 {
-                    var nested = DeserializeNestedData(dto.EditData);
+                    var deserializedContent = serializer.Deserialize(dto, dto.EditData, dto.EditDataRaw);
 
                     d = new ContentData
                     {
@@ -625,16 +825,16 @@ AND cmsContentNu.nodeId IS NULL
                         VersionId = dto.VersionId,
                         VersionDate = dto.EditVersionDate,
                         WriterId = dto.EditWriterId,
-                        Properties = nested.PropertyData,
-                        CultureInfos = nested.CultureData,
-                        UrlSegment = nested.UrlSegment
+                        Properties = deserializedContent.PropertyData, // TODO: We don't want to allocate empty arrays
+                        CultureInfos = deserializedContent.CultureData,
+                        UrlSegment = deserializedContent.UrlSegment
                     };
                 }
             }
 
             if (dto.Published)
             {
-                if (dto.PubData == null)
+                if (dto.PubData == null && dto.PubDataRaw == null)
                 {
                     if (Debugger.IsAttached)
                     {
@@ -645,24 +845,24 @@ AND cmsContentNu.nodeId IS NULL
                 }
                 else
                 {
-                    var nested = DeserializeNestedData(dto.PubData);
+                    var deserializedContent = serializer.Deserialize(dto, dto.PubData, dto.PubDataRaw);
 
                     p = new ContentData
                     {
                         Name = dto.PubName,
-                        UrlSegment = nested.UrlSegment,
+                        UrlSegment = deserializedContent.UrlSegment,
                         Published = true,
                         TemplateId = dto.PubTemplateId,
                         VersionId = dto.VersionId,
                         VersionDate = dto.PubVersionDate,
                         WriterId = dto.PubWriterId,
-                        Properties = nested.PropertyData,
-                        CultureInfos = nested.CultureData
+                        Properties = deserializedContent.PropertyData, // TODO: We don't want to allocate empty arrays
+                        CultureInfos = deserializedContent.CultureData
                     };
                 }
             }
 
-            var n = new ContentNode(dto.Id, dto.Uid,
+            var n = new ContentNode(dto.Id, dto.Key,
                 dto.Level, dto.Path, dto.SortOrder, dto.ParentId, dto.CreateDate, dto.CreatorId);
 
             var s = new ContentNodeKit
@@ -676,12 +876,12 @@ AND cmsContentNu.nodeId IS NULL
             return s;
         }
 
-        private static ContentNodeKit CreateMediaNodeKit(ContentSourceDto dto)
+        private ContentNodeKit CreateMediaNodeKit(ContentSourceDto dto, IContentCacheDataSerializer serializer)
         {
-            if (dto.EditData == null)
+            if (dto.EditData == null && dto.EditDataRaw == null)
                 throw new InvalidOperationException("No data for media " + dto.Id);
 
-            var nested = DeserializeNestedData(dto.EditData);
+            var deserializedMedia = serializer.Deserialize(dto, dto.EditData, dto.EditDataRaw);
 
             var p = new ContentData
             {
@@ -691,11 +891,11 @@ AND cmsContentNu.nodeId IS NULL
                 VersionId = dto.VersionId,
                 VersionDate = dto.EditVersionDate,
                 WriterId = dto.CreatorId, // what-else?
-                Properties = nested.PropertyData,
-                CultureInfos = nested.CultureData
+                Properties = deserializedMedia.PropertyData, // TODO: We don't want to allocate empty arrays
+                CultureInfos = deserializedMedia.CultureData
             };
 
-            var n = new ContentNode(dto.Id, dto.Uid,
+            var n = new ContentNode(dto.Id, dto.Key,
                 dto.Level, dto.Path, dto.SortOrder, dto.ParentId, dto.CreateDate, dto.CreatorId);
 
             var s = new ContentNodeKit
@@ -707,19 +907,5 @@ AND cmsContentNu.nodeId IS NULL
 
             return s;
         }
-
-        private static readonly JsonSerializerSettings NestedContentDataJsonSerializerSettings = new JsonSerializerSettings
-        {
-            Converters = new List<JsonConverter> { new ForceInt32Converter() }
-        };
-
-        private static ContentNestedData DeserializeNestedData(string data)
-        {
-            // by default JsonConvert will deserialize our numeric values as Int64
-            // which is bad, because they were Int32 in the database - take care
-
-            return JsonConvert.DeserializeObject<ContentNestedData>(data, NestedContentDataJsonSerializerSettings
-            );
-        }
     }
 }

src/Umbraco.PublishedCache.NuCache/Persistence/NuCacheContentService.cs

@@ -12,7 +12,11 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.Persistence
     {
         private readonly INuCacheContentRepository _repository;
 
-        public NuCacheContentService(INuCacheContentRepository repository, IScopeProvider provider, ILoggerFactory loggerFactory, IEventMessagesFactory eventMessagesFactory)
+        public NuCacheContentService(
+            INuCacheContentRepository repository,
+            IScopeProvider provider,
+            ILoggerFactory loggerFactory,
+            IEventMessagesFactory eventMessagesFactory)
             : base(provider, loggerFactory, eventMessagesFactory)
         {
             _repository = repository;
@@ -67,12 +71,15 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.Persistence
             => _repository.RefreshContent(content);
 
         /// <inheritdoc/>
-        public void RefreshEntity(IContentBase content)
-            => _repository.RefreshEntity(content);
+        public void RefreshMedia(IMedia media)
+            => _repository.RefreshMedia(media);
+
+        /// <inheritdoc/>
+        public void RefreshMember(IMember member)
+            => _repository.RefreshMember(member);
 
         /// <inheritdoc/>
         public void Rebuild(
-            int groupSize = 5000,
             IReadOnlyCollection<int> contentTypeIds = null,
             IReadOnlyCollection<int> mediaTypeIds = null,
             IReadOnlyCollection<int> memberTypeIds = null)
@@ -84,7 +91,7 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache.Persistence
                 scope.ReadLock(Constants.Locks.MediaTree);
                 scope.ReadLock(Constants.Locks.MemberTree);
 
-                _repository.Rebuild(groupSize, contentTypeIds, mediaTypeIds, memberTypeIds);
+                _repository.Rebuild(contentTypeIds, mediaTypeIds, memberTypeIds);
                 scope.Complete();
             }
         }

src/Umbraco.PublishedCache.NuCache/PublishedContent.cs

@@ -274,7 +274,18 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache
                             throw new PanicException($"failed to get content with id={id}");
                     }
 
-                    id = UnwrapIPublishedContent(content)._contentNode.NextSiblingContentId;
+                    var next = UnwrapIPublishedContent(content)._contentNode.NextSiblingContentId;
+
+#if DEBUG
+                    // I've seen this happen but I think that may have been due to corrupt DB data due to my own
+                    // bugs, but I'm leaving this here just in case we encounter it again while we're debugging.
+                    if (next == id)
+                    {
+                        throw new PanicException($"The current content id {id} is the same as it's next sibling id {next}");
+                    }
+#endif
+
+                    id = next;
                 }
             }
         }

src/Umbraco.PublishedCache.NuCache/PublishedSnapshotService.cs

@@ -46,6 +46,8 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache
         private readonly IPublishedModelFactory _publishedModelFactory;
         private readonly IDefaultCultureAccessor _defaultCultureAccessor;
         private readonly IHostingEnvironment _hostingEnvironment;
+        private readonly IContentCacheDataSerializerFactory _contentCacheDataSerializerFactory;
+        private readonly ContentDataSerializer _contentDataSerializer;
         private readonly NuCacheSettings _config;
 
         private bool _isReady;
@@ -90,7 +92,9 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache
             IOptions<GlobalSettings> globalSettings,
             IPublishedModelFactory publishedModelFactory,
             IHostingEnvironment hostingEnvironment,
-            IOptions<NuCacheSettings> config)
+            IOptions<NuCacheSettings> config,
+            IContentCacheDataSerializerFactory contentCacheDataSerializerFactory,
+            ContentDataSerializer contentDataSerializer)
         {
             _options = options;
             _syncBootStateAccessor = syncBootStateAccessor;
@@ -107,6 +111,8 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache
             _defaultCultureAccessor = defaultCultureAccessor;
             _globalSettings = globalSettings.Value;
             _hostingEnvironment = hostingEnvironment;
+            _contentCacheDataSerializerFactory = contentCacheDataSerializerFactory;
+            _contentDataSerializer = contentDataSerializer;
             _config = config.Value;
             _publishedModelFactory = publishedModelFactory;
         }
@@ -164,8 +170,8 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache
             _localMediaDbExists = File.Exists(localMediaDbPath);
 
             // if both local databases exist then GetTree will open them, else new databases will be created
-            _localContentDb = BTree.GetTree(localContentDbPath, _localContentDbExists, _config);
-            _localMediaDb = BTree.GetTree(localMediaDbPath, _localMediaDbExists, _config);
+            _localContentDb = BTree.GetTree(localContentDbPath, _localContentDbExists, _config, _contentDataSerializer);
+            _localMediaDb = BTree.GetTree(localMediaDbPath, _localMediaDbExists, _config, _contentDataSerializer);
 
             _logger.LogInformation("Registered with MainDom, localContentDbExists? {LocalContentDbExists}, localMediaDbExists? {LocalMediaDbExists}", _localContentDbExists, _localMediaDbExists);
         }
@@ -345,10 +351,9 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache
             // contentStore is wlocked (1 thread)
             // content (and types) are read-locked
 
-            var contentTypes = _serviceContext.ContentTypeService.GetAll()
-                .Select(x => _publishedContentTypeFactory.CreateContentType(x));
+            var contentTypes = _serviceContext.ContentTypeService.GetAll().ToList();
 
-            _contentStore.SetAllContentTypesLocked(contentTypes);
+            _contentStore.SetAllContentTypesLocked(contentTypes.Select(x => _publishedContentTypeFactory.CreateContentType(x)));
 
             using (_profilingLogger.TraceDuration<PublishedSnapshotService>("Loading content from database"))
             {
@@ -1117,11 +1122,10 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache
 
         /// <inheritdoc />
         public void Rebuild(
-            int groupSize = 5000,
             IReadOnlyCollection<int> contentTypeIds = null,
             IReadOnlyCollection<int> mediaTypeIds = null,
             IReadOnlyCollection<int> memberTypeIds = null)
-            => _publishedContentService.Rebuild(groupSize, contentTypeIds, mediaTypeIds, memberTypeIds);
+            => _publishedContentService.Rebuild(contentTypeIds, mediaTypeIds, memberTypeIds);
 
         public async Task CollectAsync()
         {

src/Umbraco.PublishedCache.NuCache/PublishedSnapshotServiceEventHandler.cs

@@ -83,9 +83,9 @@ namespace Umbraco.Cms.Infrastructure.PublishedCache
 
         public void Handle(MemberDeletingNotification notification) => _publishedContentService.DeleteContentItems(notification.DeletedEntities);
 
-        public void Handle(MemberRefreshNotification notification) => _publishedContentService.RefreshEntity(notification.Entity);
+        public void Handle(MemberRefreshNotification notification) => _publishedContentService.RefreshMember(notification.Entity);
 
-        public void Handle(MediaRefreshNotification notification) => _publishedContentService.RefreshEntity(notification.Entity);
+        public void Handle(MediaRefreshNotification notification) => _publishedContentService.RefreshMedia(notification.Entity);
 
         public void Handle(ContentRefreshNotification notification) => _publishedContentService.RefreshContent(notification.Entity);
 

src/Umbraco.PublishedCache.NuCache/Umbraco.PublishedCache.NuCache.csproj

@@ -1,51 +1,53 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
-    <PropertyGroup>
-        <TargetFramework>netstandard2.0</TargetFramework>
-        <RootNamespace>Umbraco.Cms.Infrastructure.PublishedCache</RootNamespace>
-        <LangVersion>8</LangVersion>
-        <PackageId>Umbraco.Cms.PublishedCache.NuCache</PackageId>
-        <Title>Umbraco CMS Published Cache</Title>
-        <Description>Contains the Published Cache assembly needed to run Umbraco Cms. This package only contains the assembly, and can be used for package development. Use the template in the Umbraco.Templates package to setup Umbraco</Description>
-    </PropertyGroup>
+  <PropertyGroup>
+    <TargetFramework>netstandard2.0</TargetFramework>
+    <RootNamespace>Umbraco.Cms.Infrastructure.PublishedCache</RootNamespace>
+    <LangVersion>8</LangVersion>
+    <PackageId>Umbraco.Cms.PublishedCache.NuCache</PackageId>
+    <Title>Umbraco CMS Published Cache</Title>
+    <Description>Contains the Published Cache assembly needed to run Umbraco Cms. This package only contains the assembly, and can be used for package development. Use the template in the Umbraco.Templates package to setup Umbraco</Description>
+  </PropertyGroup>
 
-    <PropertyGroup Condition=" '$(Configuration)' == 'Release' ">
-        <DocumentationFile>bin\Release\Umbraco.PublishedCache.NuCache.xml</DocumentationFile>
-    </PropertyGroup>
+  <PropertyGroup Condition=" '$(Configuration)' == 'Release' ">
+    <DocumentationFile>bin\Release\Umbraco.PublishedCache.NuCache.xml</DocumentationFile>
+  </PropertyGroup>
 
-    <ItemGroup>
-        <PackageReference Include="CSharpTest.Net.Collections-NetStd2" Version="14.906.1403.1084" />
-        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0">
-          <PrivateAssets>all</PrivateAssets>
-          <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
-        </PackageReference>
-        <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
-        <PackageReference Include="Umbraco.Code" Version="1.1.1">
-          <PrivateAssets>all</PrivateAssets>
-        </PackageReference>
-    </ItemGroup>
+  <ItemGroup>
+    <PackageReference Include="CSharpTest.Net.Collections-NetStd2" Version="14.906.1403.1084" />
+    <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+    <PackageReference Include="MessagePack" Version="2.2.85" />
+    <PackageReference Include="K4os.Compression.LZ4" Version="1.2.6" />
+    <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
+    <PackageReference Include="Umbraco.Code" Version="1.1.1">
+      <PrivateAssets>all</PrivateAssets>
+    </PackageReference>
+  </ItemGroup>
 
-    <ItemGroup>
-        <ProjectReference Include="..\Umbraco.Core\Umbraco.Core.csproj" />
-        <ProjectReference Include="..\Umbraco.Infrastructure\Umbraco.Infrastructure.csproj" />
-    </ItemGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\Umbraco.Core\Umbraco.Core.csproj" />
+    <ProjectReference Include="..\Umbraco.Infrastructure\Umbraco.Infrastructure.csproj" />
+  </ItemGroup>
 
-    <ItemGroup>
-      <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
-        <_Parameter1>Umbraco.Tests</_Parameter1>
-      </AssemblyAttribute>
-      <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
-        <_Parameter1>Umbraco.Tests.UnitTests</_Parameter1>
-      </AssemblyAttribute>
-        <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
-            <_Parameter1>Umbraco.Tests.Integration</_Parameter1>
-        </AssemblyAttribute>
-        <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
-            <_Parameter1>Umbraco.Tests.Benchmarks</_Parameter1>
-        </AssemblyAttribute>
-        <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
-            <_Parameter1>DynamicProxyGenAssembly2</_Parameter1>
-        </AssemblyAttribute>
-    </ItemGroup>
+  <ItemGroup>
+    <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
+      <_Parameter1>Umbraco.Tests</_Parameter1>
+    </AssemblyAttribute>
+    <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
+      <_Parameter1>Umbraco.Tests.UnitTests</_Parameter1>
+    </AssemblyAttribute>
+    <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
+      <_Parameter1>Umbraco.Tests.Integration</_Parameter1>
+    </AssemblyAttribute>
+    <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
+      <_Parameter1>Umbraco.Tests.Benchmarks</_Parameter1>
+    </AssemblyAttribute>
+    <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
+      <_Parameter1>DynamicProxyGenAssembly2</_Parameter1>
+    </AssemblyAttribute>
+  </ItemGroup>
 
 </Project>