diff --git a/Sources/Linq2DynamoDb.DataContext.Tests/Entities/Book.cs b/Sources/Linq2DynamoDb.DataContext.Tests/Entities/Book.cs
index 37aeb9d..63a6a34 100644
--- a/Sources/Linq2DynamoDb.DataContext.Tests/Entities/Book.cs
+++ b/Sources/Linq2DynamoDb.DataContext.Tests/Entities/Book.cs
@@ -32,6 +32,9 @@ public class Book : EntityBase
 		[DynamoDBProperty(typeof(StringTimeSpanDictionaryConverter))]
 		public IDictionary<string, TimeSpan> FilmsBasedOnBook { get; set; }
 
+        [DynamoDBVersion]
+        public int? VersionNumber { get; set; }
+
 		public enum Popularity
 		{
 			Low,
diff --git a/Sources/Linq2DynamoDb.DataContext.Tests/Entities/BookPoco.cs b/Sources/Linq2DynamoDb.DataContext.Tests/Entities/BookPoco.cs
index 6db08f8..f5fa232 100644
--- a/Sources/Linq2DynamoDb.DataContext.Tests/Entities/BookPoco.cs
+++ b/Sources/Linq2DynamoDb.DataContext.Tests/Entities/BookPoco.cs
@@ -30,6 +30,9 @@ public class BookPoco {
         [DynamoDBProperty(typeof(StringTimeSpanDictionaryConverter))]
         public IDictionary<string, TimeSpan> FilmsBasedOnBook { get; set; }
 
+        [DynamoDBVersion]
+        public int? VersionNumber { get; set; }
+
         public enum Popularity {
             Low,
 
diff --git a/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/EntityCreationTests.cs b/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/EntityCreationTests.cs
index e57aaf0..67514f2 100644
--- a/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/EntityCreationTests.cs
+++ b/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/EntityCreationTests.cs
@@ -33,17 +33,18 @@ public void DataContext_EntityCreation_PersistsRecordToDynamoDb()
             Assert.IsNotNull(storedBook);
         }
 
-        [Ignore("This behavior is currently expected. SubmitChanges() uses DocumentBatchWrite, which only supports PUT operations, which by default replaces existing entities")]
         [Test]
-        [ExpectedException(typeof(InvalidOperationException), ExpectedMessage = "cannot be added, because entity with that key already exists", MatchType = MessageMatch.Contains)]
+        [ExpectedException(typeof(AggregateException))]
         public void DataContext_EntityCreation_ThrowsExceptionWhenEntityAlreadyExistsInDynamoDbButWasNeverQueriedInCurrentContext()
         {
-            var book = BooksHelper.CreateBook(popularityRating: Book.Popularity.Average);
+            var persistedBook = BooksHelper.CreateBook(popularityRating: Book.Popularity.Average);
 
-            book.PopularityRating = Book.Popularity.High;
+            var bookCopy = BooksHelper.CreateBook(name: persistedBook.Name, publishYear: persistedBook.PublishYear, persistToDynamoDb: false);
+
+            bookCopy.PopularityRating = Book.Popularity.High;
 
             var booksTable = this.Context.GetTable<Book>();
-            booksTable.InsertOnSubmit(book);
+            booksTable.InsertOnSubmit(bookCopy);
             this.Context.SubmitChanges();
         }
 
diff --git a/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/EntityModificationTests.cs b/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/EntityModificationTests.cs
index a36ffce..5887a1d 100644
--- a/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/EntityModificationTests.cs
+++ b/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/EntityModificationTests.cs
@@ -1,7 +1,9 @@
 using Linq2DynamoDb.DataContext.Tests.Entities;
 using Linq2DynamoDb.DataContext.Tests.Helpers;
 using NUnit.Framework;
+using System;
 using System.Collections.Generic;
+using System.Linq;
 
 namespace Linq2DynamoDb.DataContext.Tests.EntityManagementTests
 {
@@ -50,28 +52,7 @@ public void DataContext_EntityModification_UpdateRecordWithNewArray()
             
             CollectionAssert.AreEquivalent(storedBook.RentingHistory, storedBookAfterModification.RentingHistory);
         }
-
-        [Ignore("This behavior is currently expected. SubmitChanges() uses DocumentBatchWrite, which only supports PUT operations with default 'replace' behavior")]
-        [Test]
-        public void DataContext_EntityModification_UpdateShouldNotAffectFieldsModifiedFromOutside()
-        {
-            var book = BooksHelper.CreateBook(popularityRating: Book.Popularity.Average, persistToDynamoDb: false);
-
-            var booksTable = this.Context.GetTable<Book>();
-            booksTable.InsertOnSubmit(book);
-            this.Context.SubmitChanges();
-
-            // Update record from outside of DataTable
-            BooksHelper.CreateBook(book.Name, book.PublishYear, numPages: 15);
-
-            book.PopularityRating = Book.Popularity.High;
-            this.Context.SubmitChanges();
-
-            var storedBook = booksTable.Find(book.Name, book.PublishYear);
-            Assert.AreEqual(book.PopularityRating, storedBook.PopularityRating, "Record was not updated");
-            Assert.AreEqual(book.NumPages, 15, "Update has erased changes from outside");
-        }
-
+        
         [Test]
         public void DataContext_UpdateEntity_UpdatesRecordWhenOldRecordIsNull()
         {
diff --git a/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/Poco/PocoCreationTests.cs b/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/Poco/PocoCreationTests.cs
index de540a0..a9d6ed6 100644
--- a/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/Poco/PocoCreationTests.cs
+++ b/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/Poco/PocoCreationTests.cs
@@ -33,20 +33,6 @@ public void DataContext_EntityCreation_PersistsRecordToDynamoDb()
             Assert.IsNotNull(storedBookPoco);
         }
 
-        [Ignore("This behavior is currently expected. SubmitChanges() uses DocumentBatchWrite, which only supports PUT operations, which by default replaces existing entities")]
-        [Test]
-        [ExpectedException(typeof(InvalidOperationException), ExpectedMessage = "cannot be added, because entity with that key already exists", MatchType = MessageMatch.Contains)]
-        public void DataContext_EntityCreation_ThrowsExceptionWhenEntityAlreadyExistsInDynamoDbButWasNeverQueriedInCurrentContext()
-        {
-            var book = BookPocosHelper.CreateBookPoco(popularityRating: BookPoco.Popularity.Average);
-
-            book.PopularityRating = BookPoco.Popularity.High;
-
-            var booksTable = this.Context.GetTable<BookPoco>();
-            booksTable.InsertOnSubmit(book);
-            this.Context.SubmitChanges();
-        }
-
         [Test]
         [ExpectedException(typeof(InvalidOperationException), ExpectedMessage = "cannot be added, because entity with that key already exists", MatchType = MessageMatch.Contains)]
         public void DataContext_EntityCreation_ThrowsExceptionWhenTryingToAddSameEntityTwice()
diff --git a/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/Poco/PocoModificationTests.cs b/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/Poco/PocoModificationTests.cs
index d581bef..8a31bbd 100644
--- a/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/Poco/PocoModificationTests.cs
+++ b/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/Poco/PocoModificationTests.cs
@@ -51,27 +51,6 @@ public void DataContext_EntityModification_UpdateRecordWithNewArray()
             CollectionAssert.AreEquivalent(storedBookPoco.RentingHistory, storedBookPocoAfterModification.RentingHistory);
         }
 
-        [Ignore("This behavior is currently expected. SubmitChanges() uses DocumentBatchWrite, which only supports PUT operations with default 'replace' behavior")]
-        [Test]
-        public void DataContext_EntityModification_UpdateShouldNotAffectFieldsModifiedFromOutside()
-        {
-            var book = BookPocosHelper.CreateBookPoco(popularityRating: BookPoco.Popularity.Average, persistToDynamoDb: false);
-
-            var booksTable = this.Context.GetTable<BookPoco>();
-            booksTable.InsertOnSubmit(book);
-            this.Context.SubmitChanges();
-
-            // Update record from outside of DataTable
-            BookPocosHelper.CreateBookPoco(book.Name, book.PublishYear, numPages: 15);
-
-            book.PopularityRating = BookPoco.Popularity.High;
-            this.Context.SubmitChanges();
-
-            var storedBookPoco = booksTable.Find(book.Name, book.PublishYear);
-            Assert.AreEqual(book.PopularityRating, storedBookPoco.PopularityRating, "Record was not updated");
-            Assert.AreEqual(book.NumPages, 15, "Update has erased changes from outside");
-        }
-
         [Test]
         public void DataContext_UpdateEntity_UpdatesRecordWhenOldRecordIsNull()
         {
diff --git a/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/Versioning/EntityVersioningTests.cs b/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/Versioning/EntityVersioningTests.cs
new file mode 100644
index 0000000..2405119
--- /dev/null
+++ b/Sources/Linq2DynamoDb.DataContext.Tests/EntityManagementTests/Versioning/EntityVersioningTests.cs
@@ -0,0 +1,91 @@
+using Linq2DynamoDb.DataContext.Tests.Entities;
+using Linq2DynamoDb.DataContext.Tests.Helpers;
+using NUnit.Framework;
+using System;
+using System.Collections.Generic;
+
+namespace Linq2DynamoDb.DataContext.Tests.EntityManagementTests.Versioning 
+{
+    [TestFixture]
+    class EntityVersioningTests : DataContextTestBase
+    {
+        public override void SetUp() 
+        {
+        }
+
+        public override void TearDown() 
+        {
+        }
+
+        [Test]
+        [ExpectedException(typeof(AggregateException))]
+        public void DataContext_UpdateEntity_Does_OptimisticLocking() 
+        {
+            var contextA = TestConfiguration.GetDataContext();
+            var contextB = TestConfiguration.GetDataContext();
+
+            var originalBook = BooksHelper.CreateBook(popularityRating: Book.Popularity.Low, rentingHistory: null);
+            var booksTableA = contextA.GetTable<Book>();
+            var booksTableB = contextB.GetTable<Book>();
+
+            // Read the same entry from the database into two contexts
+            var retrievedBookA = booksTableA.Find(originalBook.Name, originalBook.PublishYear);
+            var retrievedBookB = booksTableB.Find(originalBook.Name, originalBook.PublishYear);
+
+            // Mutate a property on instance A and persist
+            retrievedBookA.PopularityRating = Book.Popularity.Average;
+            contextA.SubmitChanges();
+
+            // Mutate a property on instance B (unaware of changes to A)
+            retrievedBookB.RentingHistory = new List<string> { "history element" };
+            contextB.SubmitChanges();
+        }
+
+        [Test]
+        [ExpectedException(typeof(AggregateException))]
+        public void DataContext_AddEntity_DoesNotOverwrite_ExistingVersionedEntity() 
+        {
+            var contextA = TestConfiguration.GetDataContext();
+            var contextB = TestConfiguration.GetDataContext();
+
+            var tableA = contextA.GetTable<Book>();
+            var tableB = contextB.GetTable<Book>();
+
+            var bookA = BooksHelper.CreateBook(name: "A Tale of Two Books", publishYear: 0, persistToDynamoDb: false);
+            var bookB = BooksHelper.CreateBook(name: "A Tale of Two Books", publishYear: 0, persistToDynamoDb: false);
+
+
+            tableA.InsertOnSubmit(bookA);
+            contextA.SubmitChanges();
+
+            tableB.InsertOnSubmit(bookB);
+            contextB.SubmitChanges();
+        }
+
+        [Test]
+        [ExpectedException(typeof(AggregateException))]
+        public void DataContext_RemoveEntity_RespectsVersionConstraint() 
+        {
+            var book = BooksHelper.CreateBook(numPages: 5, persistToDynamoDb: false);
+
+            var contextA = TestConfiguration.GetDataContext();
+            var contextB = TestConfiguration.GetDataContext();
+
+            var tableA = contextA.GetTable<Book>();
+            var tableB = contextB.GetTable<Book>();
+
+            // Insert the book in Context A
+            tableA.InsertOnSubmit(book);
+            contextA.SubmitChanges();
+
+            // Find and modify the book in Context B
+            var retrievedBook = tableB.Find(book.Name, book.PublishYear);
+            retrievedBook.NumPages = 10;
+            contextB.SubmitChanges();
+
+            // Try to delete the book from Context A
+            tableA.RemoveOnSubmit(book);
+            contextA.SubmitChanges();
+        }
+    }
+}
diff --git a/Sources/Linq2DynamoDb.DataContext.Tests/Helpers/BooksHelper.cs b/Sources/Linq2DynamoDb.DataContext.Tests/Helpers/BooksHelper.cs
index a05311a..d51be37 100644
--- a/Sources/Linq2DynamoDb.DataContext.Tests/Helpers/BooksHelper.cs
+++ b/Sources/Linq2DynamoDb.DataContext.Tests/Helpers/BooksHelper.cs
@@ -28,7 +28,7 @@ public static void CleanSession()
 		{
             Logger.DebugFormat("Removing {0} records from DynamoDb", _recordsForCleanup.Count);
 
-			Parallel.ForEach(_recordsForCleanup, book => PersistenceContext.Delete(book));
+			Parallel.ForEach(_recordsForCleanup, book => PersistenceContext.Delete(book, new DynamoDBOperationConfig { SkipVersionCheck = true, ConsistentRead = true }));
 
 			_recordsForCleanup = new ConcurrentQueue<Book>();
 		}
diff --git a/Sources/Linq2DynamoDb.DataContext.Tests/Linq2DynamoDb.DataContext.Tests.csproj b/Sources/Linq2DynamoDb.DataContext.Tests/Linq2DynamoDb.DataContext.Tests.csproj
index bdd4fbf..faa3946 100644
--- a/Sources/Linq2DynamoDb.DataContext.Tests/Linq2DynamoDb.DataContext.Tests.csproj
+++ b/Sources/Linq2DynamoDb.DataContext.Tests/Linq2DynamoDb.DataContext.Tests.csproj
@@ -90,6 +90,7 @@
     <Compile Include="EntityManagementTests\Poco\PocoCreationTests.cs" />
     <Compile Include="EntityManagementTests\Poco\PocoModificationTests.cs" />
     <Compile Include="EntityManagementTests\Poco\PocoRemovalTests.cs" />
+    <Compile Include="EntityManagementTests\Versioning\EntityVersioningTests.cs" />
     <Compile Include="Helpers\BookPocoHelper.cs" />
     <Compile Include="Helpers\BooksHelper.cs" />
     <Compile Include="Helpers\MemcachedController.cs" />
@@ -140,7 +141,9 @@
       <SubType>Designer</SubType>
     </None>
   </ItemGroup>
-  <ItemGroup />
+  <ItemGroup>
+    <Service Include="{82A7F48D-3B50-4B1E-B82E-3ADA8210C358}" />
+  </ItemGroup>
   <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
   <!-- To modify your build process, add your task inside one of the targets below and uncomment it. 
        Other similar extension points exist, see Microsoft.Common.targets.
diff --git a/Sources/Linq2DynamoDb.DataContext/EntityProxy.cs b/Sources/Linq2DynamoDb.DataContext/EntityProxy.cs
index 8a8002a..1e3069d 100644
--- a/Sources/Linq2DynamoDb.DataContext/EntityProxy.cs
+++ b/Sources/Linq2DynamoDb.DataContext/EntityProxy.cs
@@ -131,6 +131,8 @@ public Document GetDocumentIfDirty()
 
         public object Entity { get { return this.GetTransparentProxy(); } }
 
+        public Document AsDocument() { return _document; }
+
         public void Commit() {}
     }
 }
diff --git a/Sources/Linq2DynamoDb.DataContext/EntityWrapper.cs b/Sources/Linq2DynamoDb.DataContext/EntityWrapper.cs
index 5590958..c94ff66 100644
--- a/Sources/Linq2DynamoDb.DataContext/EntityWrapper.cs
+++ b/Sources/Linq2DynamoDb.DataContext/EntityWrapper.cs
@@ -2,6 +2,8 @@
 using System.Linq;
 using Amazon.DynamoDBv2.DocumentModel;
 using Linq2DynamoDb.DataContext.Utils;
+using System.Reflection;
+using Amazon.DynamoDBv2.DataModel;
 
 namespace Linq2DynamoDb.DataContext
 {
@@ -14,6 +16,31 @@ internal class EntityWrapper : IEntityWrapper
         private readonly IEntityKeyGetter _keyGetter;
         private Document _doc, _newDoc;
 
+        private PropertyInfo _entityVersionNumberProperty;
+        private bool _hasResolvedEntityVersionNumberProperty;
+
+        /// <summary>
+        /// Gets PropertyInfo for the entity's property that has [DynamoDBVersion] 
+        /// attribute or returns null if there is none.
+        /// </summary>
+        private PropertyInfo EntityVersionNumberProperty {
+            get {
+                if (!_hasResolvedEntityVersionNumberProperty) {
+                    _entityVersionNumberProperty = Entity
+                        .GetType()
+                        .GetProperties(BindingFlags.Public | BindingFlags.Instance)
+                        .Where(property =>
+                            property
+                                .GetCustomAttributes(typeof(DynamoDBVersionAttribute), true)
+                                .SingleOrDefault() != null
+                        ).SingleOrDefault();
+
+                    _hasResolvedEntityVersionNumberProperty = true;
+                }
+                return _entityVersionNumberProperty;
+            }
+        }
+
         internal EntityWrapper(object entity, Func<object, Document> conversionFunctor, IEntityKeyGetter keyGetter)
         {
             this.Entity = entity;
@@ -38,6 +65,8 @@ internal EntityWrapper(Document doc, Type entityType, Func<object, Document> con
         /// </summary>
         internal bool IsCommited { get; private set; }
 
+        public Document AsDocument() {  return this._conversionFunctor(this.Entity); }
+
         /// <summary>
         /// Returns a new document, if the entity was modified since the last call to this method.
         /// Otherwise returns null.
@@ -45,7 +74,7 @@ internal EntityWrapper(Document doc, Type entityType, Func<object, Document> con
         /// <returns></returns>
         public Document GetDocumentIfDirty()
         {
-            this._newDoc = this._conversionFunctor(this.Entity);
+            this._newDoc = AsDocument();
 
             if (this._doc == null)
             {
@@ -101,9 +130,28 @@ public void Commit()
         {
             this._doc = this._newDoc;
             this._newDoc = null;
+            this.UpdateEntityVersionNumber();
             this.IsCommited = true;
         }
 
+        /// <summary>
+        /// Sets the value of the Entity's propety that has the DynamoDBVersionAttribute to 
+        /// the value in _doc. When adding/updating the document the version number will
+        /// change in the _doc on the way to DynamoDB 
+        /// </summary>
+        private void UpdateEntityVersionNumber() 
+        {
+            if (EntityVersionNumberProperty == default(PropertyInfo))
+            {
+                return;
+            }
+
+            EntityVersionNumberProperty.SetValue(
+                Entity,
+                this._doc[EntityVersionNumberProperty.Name].ToObject(EntityVersionNumberProperty.PropertyType)
+            );
+        }
+
         #region Redirecting Equals() and GetHashCode() to the underlying entity
 
         public override bool Equals(object obj)
diff --git a/Sources/Linq2DynamoDb.DataContext/IEntityWrapper.cs b/Sources/Linq2DynamoDb.DataContext/IEntityWrapper.cs
index 47215eb..02ea054 100644
--- a/Sources/Linq2DynamoDb.DataContext/IEntityWrapper.cs
+++ b/Sources/Linq2DynamoDb.DataContext/IEntityWrapper.cs
@@ -7,5 +7,6 @@ public interface IEntityWrapper
         object Entity { get; }
         Document GetDocumentIfDirty();
         void Commit();
+        Document AsDocument();
     }
 }
diff --git a/Sources/Linq2DynamoDb.DataContext/TableDefinitionWrapper.cs b/Sources/Linq2DynamoDb.DataContext/TableDefinitionWrapper.cs
index d0b441e..b407a54 100644
--- a/Sources/Linq2DynamoDb.DataContext/TableDefinitionWrapper.cs
+++ b/Sources/Linq2DynamoDb.DataContext/TableDefinitionWrapper.cs
@@ -99,7 +99,14 @@ public void RemoveEntity(object removedEntity)
                 }
             }
 
-            this._removedEntities.Add(removedKey);
+            var entityWrapperToRemove = removedEntity is IEntityWrapper
+                    ? (IEntityWrapper)removedEntity
+                    : new EntityWrapper(removedEntity, this.ToDocumentConversionFunctor, this.EntityKeyGetter);
+
+            this._removedEntities.Add(
+                removedKey, 
+                entityWrapperToRemove
+            );
 
             // removing the entity from the list of loaded entities as well
             this._loadedEntities.Remove(removedKey);
@@ -126,19 +133,19 @@ public override void ClearModifications()
         internal Action<Exception> ThingsToDoUponSubmit;
 
         /// <summary>
-        /// Entities, loaded from DynamoDb
+        /// Entities that were loaded from DynamoDb or 
         /// </summary>
         private readonly Dictionary<EntityKey, IEntityWrapper> _loadedEntities = new Dictionary<EntityKey, IEntityWrapper>();
 
         /// <summary>
-        /// Entities, that were added locally
+        /// Entities to add
         /// </summary>
         private readonly List<EntityWrapper> _addedEntities = new List<EntityWrapper>();
 
         /// <summary>
         /// Keys of removed entities
         /// </summary>
-        private readonly HashSet<EntityKey> _removedEntities = new HashSet<EntityKey>();
+        private readonly Dictionary<EntityKey, IEntityWrapper> _removedEntities = new Dictionary<EntityKey, IEntityWrapper>();
 
         /// <summary>
         /// Implements converting objects into documents (caches everything it needs for that)
@@ -156,7 +163,7 @@ protected internal Task SubmitChangesAsync()
         {
             var entitiesToAdd = new Dictionary<EntityKey, Document>();
             var entitiesToUpdate = new Dictionary<EntityKey, Document>();
-            var entitiesToRemove = new List<EntityKey>();
+            var entitiesToRemove = new Dictionary<EntityKey, Document>();
 
             // all newly added (even already saved and not modified any more) entities
             var addedEntitiesDictionary = new Dictionary<EntityKey, IEntityWrapper>();
@@ -175,7 +182,7 @@ protected internal Task SubmitChangesAsync()
                     var modifiedKey = this.EntityKeyGetter.GetKey(modifiedDoc);
 
                     // no need to modify the entity, if it was removed
-                    if (this._removedEntities.Contains(modifiedKey))
+                    if (this._removedEntities.ContainsKey(modifiedKey))
                     {
                         continue;
                     }
@@ -211,16 +218,16 @@ protected internal Task SubmitChangesAsync()
                 }
 
                 // removing removed entities
-                foreach (var removedKey in this._removedEntities)
+                foreach (var removedKvp in this._removedEntities)
                 {
                     // if the entity was removed and then added anew - then it shouldn't be removed from the table
-                    if (addedEntitiesDictionary.ContainsKey(removedKey))
+                    if (addedEntitiesDictionary.ContainsKey(removedKvp.Key))
                     {
                         continue;
                     }
 
-                    this.Log("Removing entity with key {0}", removedKey);
-                    entitiesToRemove.Add(removedKey);
+                    this.Log("Removing entity with key {0}", removedKvp);
+                    entitiesToRemove.Add(removedKvp.Key, removedKvp.Value.AsDocument());
                 }
             }
             catch (Exception ex)
@@ -259,13 +266,15 @@ protected internal Task SubmitChangesAsync()
                         // clearing the list of removed entities, as they should not be removed twice
                         this._removedEntities.Clear();
 
-                        // clearing IsDirty-flag on all newly added entities (even those, that are not dirty)
+                        // clearing IsDirty-flag on all newly added entities (even those, that are not dirty) 
+                        // and update the version attribute (if applicable)
                         foreach (var addedEntity in addedEntitiesDictionary.Values)
                         {
                             addedEntity.Commit();
                         }
 
                         // clearing IsDirty-flag on updated entities (even those, that are not dirty)
+                        // and update the version attribute (if applicable)
                         foreach (var updatedEntityWrapper in this._loadedEntities.Values)
                         {
                             updatedEntityWrapper.Commit();
@@ -360,30 +369,21 @@ protected override void InitReader(ISupervisableEnumerable reader)
         /// <summary>
         /// Fills in and executes an update batch
         /// </summary>
-        private Task ExecuteUpdateBatchAsync(IDictionary<EntityKey, Document> addedEntities, IDictionary<EntityKey, Document> modifiedEntities, ICollection<EntityKey> removedEntities)
+        private Task ExecuteUpdateBatchAsync(IDictionary<EntityKey, Document> addedEntities, IDictionary<EntityKey, Document> modifiedEntities, IDictionary<EntityKey, Document> removedEntities)
         {
-            var batch = this.TableDefinition.CreateBatchWrite();
+            var dynamoDbUpdateTasks = AddEntriesAsync(addedEntities)
+                .Concat(AddEntriesAsync(modifiedEntities))
+                .Concat(DeleteEntitiesAsync(removedEntities))
+                .ToArray();
 
-            foreach (var key in addedEntities)
-            {
-                batch.AddDocumentToPut(key.Value);
-            }
-            foreach (var key in modifiedEntities)
+            if (string.IsNullOrEmpty(VersionPropertyName)) 
             {
-                batch.AddDocumentToPut(key.Value);
-            }
-            foreach (var key in removedEntities)
-            {
-                batch.AddKeyToDelete(this.EntityKeyGetter.GetKeyDictionary(key));
+                // Only update the cache in a separate thread if there is no Version Property
+                this.Cache.UpdateCacheAndIndexes(addedEntities, modifiedEntities, removedEntities.Keys);
             }
 
-            // Updating cache in current thread and DynamoDb in a separate thread
-            var dynamoDbUpdateTask = batch.ExecuteAsync();
 
-            // this should never throw any exceptions
-            this.Cache.UpdateCacheAndIndexes(addedEntities, modifiedEntities, removedEntities);
-
-            return dynamoDbUpdateTask.ContinueWith
+            return Task.WhenAll(dynamoDbUpdateTasks).ContinueWith
             (
                 updateTask =>
                 {
@@ -391,14 +391,98 @@ private Task ExecuteUpdateBatchAsync(IDictionary<EntityKey, Document> addedEntit
                     {
                         // If the update failed - then removing all traces of modified entities from cache,
                         // as the update might partially succeed (in that case some of entities in cache might become stale)
-                        this.Cache.RemoveEntities(addedEntities.Keys.Union(modifiedEntities.Keys).Union(removedEntities));
+                        this.Cache.RemoveEntities(addedEntities.Keys.Union(modifiedEntities.Keys).Union(removedEntities.Keys));
 
                         throw updateTask.Exception;
                     }
+
+                    if (!string.IsNullOrEmpty(VersionPropertyName))
+                    {
+                        // Update the cache if there is a Version Property. Version properties change 
+                        // during insertion/updating so we want to have the latest values for versions 
+                        // of added and modified entities. 
+                        this.Cache.UpdateCacheAndIndexes(addedEntities, modifiedEntities, removedEntities.Keys);
+                    }
                 }
             );
         }
 
+        /// <summary>
+        /// Performs create/update operations with DynamoDB, respecting the [DynamoDBVersion] attribute if present
+        /// </summary>
+        /// <remarks>
+        /// If the entity is new (Version property has not been set) set the version property 0. If 
+        /// updating an existing entity increment the Version property by 1.
+        /// </remarks>
+        /// <param name="entitiesToAdd">Dictionary of entity keys and documents to add/update</param>
+        /// <returns>An IEnumerable containing one task for each entity to add or update.</returns>
+        private IEnumerable<Task> AddEntriesAsync(IDictionary<EntityKey, Document> entitiesToAdd) 
+        {
+            return entitiesToAdd
+                .Values
+                .Select(entry => {
+                    var putConfiguration = new PutItemOperationConfig();
+                    if (!string.IsNullOrEmpty(VersionPropertyName)) 
+                    {
+                        if (entry[VersionPropertyName] != null && 
+                            entry[VersionPropertyName].ToPrimitive(typeof(object)).Value == null)
+                        {
+                            entry[VersionPropertyName] = 0;
+
+                            putConfiguration.ConditionalExpression = new Expression() 
+                            {
+                                ExpressionStatement = $"attribute_not_exists(#key)",
+                                ExpressionAttributeNames = new Dictionary<string, string>
+                                {
+                                    { "#key", HashKeyPropetyName }
+                                }
+                            };
+                        }
+                        else
+                        {
+                            putConfiguration.Expected = new Document(new Dictionary<string, DynamoDBEntry>()
+                            {
+                                { VersionPropertyName, entry[VersionPropertyName] }
+                            });
+
+                            // TODO: We assume that the Version is an int here but it could be 
+                            // any numeric (signed or unsigned) so we may overflow or lose precision when 
+                            // converting or incrementing. This isn't strcitly a problem for optimstic locking 
+                            // as long as the verison number is different from the fetched version number.
+                            entry[VersionPropertyName] = (entry[VersionPropertyName].AsInt() + 1);
+                        }
+                    }
+
+                    return TableDefinition.PutItemAsync(entry, putConfiguration);
+                });    
+        }
+
+
+        /// <summary>
+        /// Performs Delete for entitites on DynamoDB, respecting the [DynamoDBVersion] attribute if present
+        /// </summary>
+        /// <param name="removedEntities">Dictionary of entity keys and documents to remove</param>
+        /// <returns>An IEnumerable containing one task for each entity to add or update.</returns>
+        private IEnumerable<Task> DeleteEntitiesAsync(IDictionary<EntityKey, Document> removedEntities) 
+        {
+            return removedEntities
+                .Select(entry => {
+                    var deleteConfiguration = new DeleteItemOperationConfig();
+                    if (!string.IsNullOrEmpty(VersionPropertyName)) 
+                    {
+                        deleteConfiguration.Expected = new Document(new Dictionary<string, DynamoDBEntry>() 
+                        {
+                            { VersionPropertyName, entry.Value[VersionPropertyName] }
+                        });
+                    }
+
+                    return TableDefinition.DeleteItemAsync(
+                        this.EntityKeyGetter.GetKeyDictionary(entry.Key),
+                        deleteConfiguration
+                    );
+                });
+        }
+
 
         private TranslationResult GetTranslationResultForFind(params object[] keyValues)
         {
diff --git a/Sources/Linq2DynamoDb.DataContext/TableDefinitionWrapperBase.cs b/Sources/Linq2DynamoDb.DataContext/TableDefinitionWrapperBase.cs
index eb35cec..9324614 100644
--- a/Sources/Linq2DynamoDb.DataContext/TableDefinitionWrapperBase.cs
+++ b/Sources/Linq2DynamoDb.DataContext/TableDefinitionWrapperBase.cs
@@ -8,6 +8,8 @@
 using Linq2DynamoDb.DataContext.Caching;
 using Linq2DynamoDb.DataContext.ExpressionUtils;
 using Linq2DynamoDb.DataContext.Utils;
+using Amazon.DynamoDBv2.DataModel;
+using System.Reflection;
 
 namespace Linq2DynamoDb.DataContext
 {
@@ -46,6 +48,57 @@ internal TableDefinitionWrapperBase(Table tableDefinition, Type tableEntityType,
         /// </summary>
         internal event Action<string> OnLog;
 
+        /// <summary>
+        ///  The name of the underlying entity's property that has the  [DynamoDBVersion] attribute
+        /// </summary>
+        internal string VersionPropertyName 
+        {
+            get 
+            {
+                if (!_hasResolvedVersionPropertyName) 
+                {
+                    var versionProperty = TableEntityType
+                        .GetProperties(BindingFlags.Public | BindingFlags.Instance)
+                        .Where(property =>
+                            property
+                                .GetCustomAttributes(typeof(DynamoDBVersionAttribute), true)
+                                .SingleOrDefault() != null
+                        ).SingleOrDefault();
+
+                    _versionPropertyName = versionProperty != default(PropertyInfo)
+                        ? versionProperty.Name
+                        : null;
+
+                    _hasResolvedVersionPropertyName = true;
+                }
+                return _versionPropertyName;
+            }
+        }
+
+        /// <summary>
+        /// THe name of the underlying entity's property that has the [DynamoDBHashKey] attribute
+        /// </summary>
+        internal string HashKeyPropetyName 
+        {
+            get 
+            {
+                if(!_hasResolvedHashKeyPropertyName) 
+                {
+                    var hashKeyProperty = TableEntityType
+                        .GetProperties(BindingFlags.Public | BindingFlags.Instance)
+                        .Where(property => 
+                            property
+                                .GetCustomAttributes(typeof(DynamoDBHashKeyAttribute), true)
+                                .SingleOrDefault() != null
+                        ).Single();
+
+                    _hashKeyPropertyName = hashKeyProperty.Name;
+                }
+
+                return _hashKeyPropertyName;
+            }
+        }
+
         #endregion
 
         #region Public Methods
@@ -173,6 +226,26 @@ public virtual void ClearModifications()
         /// </summary>
         protected readonly Primitive HashKeyValue;
 
+        /// <summary>
+        /// Have we resolved the property with the [DynamoDBVersion] attribute?
+        /// </summary>
+        private bool _hasResolvedVersionPropertyName;
+
+        /// <summary>
+        /// Cache the name of the property with the [DynamoDBVersion] attribute
+        /// </summary>
+        private string _versionPropertyName;
+
+        /// <summary>
+        /// Have we resolved the property with the [DynamoDBHashKey] attribute?
+        /// </summary>
+        private bool _hasResolvedHashKeyPropertyName;
+
+        /// <summary>
+        /// Cache the name of the propety with the [DynamoDBHashKey] attribute
+        /// </summary>
+        private string _hashKeyPropertyName;
+
         private IEntityKeyGetter _entityKeyGetter;
         /// <summary>
         /// Implements key extraction (caches everything it needs for that)