public Etag AddAttachment(string key, Etag etag, Stream data, RavenJObject headers)
		{
			Api.JetSetCurrentIndex(session, Files, "by_name");
			Api.MakeKey(session, Files, key, Encoding.Unicode, MakeKeyGrbit.NewKey);
			var isUpdate = Api.TrySeek(session, Files, SeekGrbit.SeekEQ);
			if (isUpdate)
			{
				var existingEtag = Etag.Parse(Api.RetrieveColumn(session, Files, tableColumnsCache.FilesColumns["etag"]));
				if (existingEtag != etag && etag != null)
				{
					throw new ConcurrencyException("PUT attempted on attachment '" + key +
						"' using a non current etag")
					{
						ActualETag = existingEtag,
						ExpectedETag = etag
					};
				}
			}
			else
			{
				if (data == null)
					throw new InvalidOperationException("When adding new attachment, the attachment data must be specified");

				if (Api.TryMoveFirst(session, Details))
					Api.EscrowUpdate(session, Details, tableColumnsCache.DetailsColumns["attachment_count"], 1);
			}

			Etag newETag = uuidGenerator.CreateSequentialUuid(UuidType.Attachments);
			using (var update = new Update(session, Files, isUpdate ? JET_prep.Replace : JET_prep.Insert))
			{
				Api.SetColumn(session, Files, tableColumnsCache.FilesColumns["name"], key, Encoding.Unicode);
				if (data != null)
				{
					long written;
					using (var columnStream = new ColumnStream(session, Files, tableColumnsCache.FilesColumns["data"]))
					{
						if (isUpdate)
							columnStream.SetLength(0);
						using (var stream = new BufferedStream(columnStream))
						{
							data.CopyTo(stream);
							written = stream.Position;
							stream.Flush();
						}
					}
					if (written == 0) // empty attachment
					{
						Api.SetColumn(session, Files, tableColumnsCache.FilesColumns["data"], new byte[0]);
					}
				}

				Api.SetColumn(session, Files, tableColumnsCache.FilesColumns["etag"], newETag.TransformToValueForEsentSorting());
				Api.SetColumn(session, Files, tableColumnsCache.FilesColumns["metadata"], headers.ToString(Formatting.None), Encoding.Unicode);

				update.Save();
			}
			logger.Debug("Adding attachment {0}", key);

			return newETag;
		}

		public void Set(string name, string key, RavenJObject data, UuidType uuidType)
		{
			Api.JetSetCurrentIndex(session, Lists, "by_name_and_key");
			Api.MakeKey(session, Lists, name, Encoding.Unicode, MakeKeyGrbit.NewKey);
			Api.MakeKey(session, Lists, key, Encoding.Unicode, MakeKeyGrbit.None);

			var exists = Api.TrySeek(session, Lists, SeekGrbit.SeekEQ);


			using (var update = new Update(session, Lists, exists ? JET_prep.Replace : JET_prep.Insert))
			{
				Api.SetColumn(session, Lists, tableColumnsCache.ListsColumns["name"], name, Encoding.Unicode);
				Api.SetColumn(session, Lists, tableColumnsCache.ListsColumns["key"], key, Encoding.Unicode);
				Api.SetColumn(session, Lists, tableColumnsCache.ListsColumns["etag"], uuidGenerator.CreateSequentialUuid(uuidType).TransformToValueForEsentSorting());
				Api.SetColumn(session, Lists, tableColumnsCache.ListsColumns["created_at"], SystemTime.UtcNow);

				using (var columnStream = new ColumnStream(session, Lists, tableColumnsCache.ListsColumns["data"]))
				{
					if (exists)
						columnStream.SetLength(0);
					using (Stream stream = new BufferedStream(columnStream))
					{
						data.WriteTo(stream);
						stream.Flush();
					}
				}
				update.Save();
			}
		}

 /// <summary>
 /// Write a serialized form of an object to a column.
 /// </summary>
 /// <param name="sesid">The session to use.</param>
 /// <param name="tableid">The table to write to. An update should be prepared.</param>
 /// <param name="columnid">The column to write to.</param>
 /// <param name="value">The object to write. The object must be serializable.</param>
 public static void SerializeObjectToColumn(JET_SESID sesid, JET_TABLEID tableid, JET_COLUMNID columnid, object value)
 {
     if (null == value)
     {
         Api.SetColumn(sesid, tableid, columnid, null);
     }
     else
     {
         using (var stream = new ColumnStream(sesid, tableid, columnid))
         {
             var serializer = new BinaryFormatter
             {
                 Context = new StreamingContext(StreamingContextStates.Persistence)
             };
             serializer.Serialize(stream, value);
         }
     }
 }

 public void VerifySeekFromEndThrowsExceptionOnOverflow()
 {
     using (var t = new Transaction(this.sesid))
     using (var u = new Update(this.sesid, this.tableid, JET_prep.Insert))
     using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
     {
         stream.Write(new byte[10], 0, 10);
         try
         {
             stream.Seek(Int64.MaxValue, SeekOrigin.End);
             Assert.Fail("Expected OverflowException");
         }
         catch (OverflowException)
         {
             // expected
         }
     }
 }

		public void AddSignature(string name, int level, Action<Stream> action)
		{
			using (var update = new Update(session, Signatures, JET_prep.Insert))
			{
				Api.SetColumn(session, Signatures, tableColumnsCache.SignaturesColumns["name"], name, Encoding.Unicode);
				Api.SetColumn(session, Signatures, tableColumnsCache.SignaturesColumns["level"], level);
				Api.SetColumn(session, Signatures, tableColumnsCache.SignaturesColumns["created_at"], DateTime.UtcNow);

				using (var stream = new ColumnStream(session, Signatures, tableColumnsCache.SignaturesColumns["data"]))
				using (var buffer = new BufferedStream(stream))
				{
					action(buffer);
					buffer.Flush();
					stream.Flush();
				}

				update.Save();
			}
		}

		public int InsertPage(byte[] buffer, int size)
		{
			var key = new HashKey(buffer, size);

			Api.JetSetCurrentIndex(session, Pages, "by_keys");

			Api.MakeKey(session, Pages, key.Weak, MakeKeyGrbit.NewKey);
			Api.MakeKey(session, Pages, key.Strong, MakeKeyGrbit.None);

			if (Api.TrySeek(session, Pages, SeekGrbit.SeekEQ))
			{
				Api.EscrowUpdate(session, Pages, tableColumnsCache.PagesColumns["usage_count"], 1);
				return Api.RetrieveColumnAsInt32(session, Pages, tableColumnsCache.PagesColumns["id"]).Value;
			}

            var bookMarkBuffer = new byte[bookmarkMost];
			var actualSize = 0;
			using (var update = new Update(session, Pages, JET_prep.Insert))
			{
				Api.SetColumn(session, Pages, tableColumnsCache.PagesColumns["page_strong_hash"], key.Strong);
				Api.SetColumn(session, Pages, tableColumnsCache.PagesColumns["page_weak_hash"], key.Weak);

				using (var columnStream = new ColumnStream(session, Pages, tableColumnsCache.PagesColumns["data"]))
				{
					using (Stream stream = new BufferedStream(columnStream))
					using (var finalStream = fileCodecs.Aggregate(stream, (current, codec) => codec.EncodePage(current)))
					{
						finalStream.Write(buffer, 0, size);
						finalStream.Flush();
					}
				}

				try
				{
					update.Save(bookMarkBuffer, bookMarkBuffer.Length, out actualSize);
				}
				catch (EsentKeyDuplicateException)
				{
					// it means that page is being inserted by another thread
					throw new ConcurrencyException("The same file page is being created");
				}
			}

			Api.JetGotoBookmark(session, Pages, bookMarkBuffer, actualSize);

			return Api.RetrieveColumnAsInt32(session, Pages, tableColumnsCache.PagesColumns["id"]).Value;
		}

 public void WriteThrowsExceptionWhenBufferOffsetIsNegative()
 {
     using (var t = new Transaction(this.sesid))
     using (var u = new Update(this.sesid, this.tableid, JET_prep.Insert))
     using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
     {
         var buffer = new byte[10];
         stream.Write(buffer, -1, 1);
     }
 }

        public void WriteAtNonZeroOffset()
        {
            var bookmark = new byte[SystemParameters.BookmarkMost];
            int bookmarkSize;

            var data = Any.BytesOfLength(1024);
            int offset = data.Length / 2;

            using (var transaction = new Transaction(this.sesid))
            using (var update = new Update(this.sesid, this.tableid, JET_prep.Insert))
            using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
            {
                stream.Write(data, offset, data.Length - offset);
                update.Save(bookmark, bookmark.Length, out bookmarkSize);
                transaction.Commit(CommitTransactionGrbit.LazyFlush);
            }

            Api.JetGotoBookmark(this.sesid, this.tableid, bookmark, bookmarkSize);
            using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
            {
                var retrieved = new byte[data.Length - offset];
                stream.Read(retrieved, 0, retrieved.Length);
                for (int i = 0; i < retrieved.Length; ++i)
                {
                    Assert.AreEqual(retrieved[i], data[i + offset]);
                }
            }
        }

 public void SettingPositionThrowsExceptionWhenPositionIsTooLong()
 {
     using (var t = new Transaction(this.sesid))
     using (var u = new Update(this.sesid, this.tableid, JET_prep.Insert))
     using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
     {
         stream.Position = 0x800000000;
     }
 }

        public void SetColumnStreamPosition()
        {
            var bookmark = new byte[SystemParameters.BookmarkMost];
            int bookmarkSize;

            using (var transaction = new Transaction(this.sesid))
            using (var update = new Update(this.sesid, this.tableid, JET_prep.Insert))
            using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
            {
                stream.Write(Any.BytesOfLength(1024), 0, 1024);
                update.Save(bookmark, bookmark.Length, out bookmarkSize);
                transaction.Commit(CommitTransactionGrbit.LazyFlush);
            }

            Api.JetGotoBookmark(this.sesid, this.tableid, bookmark, bookmarkSize);
            using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
            {
                stream.Position = 10;
                Assert.AreEqual(10, stream.Position);
            }
        }

 public void ColumnStreamSetLengthThrowsExceptionWhenLengthIsTooLong()
 {
     using (var t = new Transaction(this.sesid))
     using (var u = new Update(this.sesid, this.tableid, JET_prep.Insert))
     using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
     {
         stream.SetLength(0x800000000);
     }
 }

        public void ColumnStreamCanSerializeObject()
        {
            var expected = new Dictionary<string, long> { { "foo", 1 }, { "bar", 2 }, { "baz", 3 } };

            using (var t = new Transaction(this.sesid))
            using (var u = new Update(this.sesid, this.tableid, JET_prep.Insert))
            using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
            {
                var serializer = new BinaryFormatter();
                serializer.Serialize(stream, expected);
                u.Save();
                t.Commit(CommitTransactionGrbit.LazyFlush);
            }

            Api.JetMove(this.sesid, this.tableid, JET_Move.First, MoveGrbit.None);
            using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
            {
                var deseriaizer = new BinaryFormatter();
                var actual = (Dictionary<string, long>) deseriaizer.Deserialize(stream);
                CollectionAssert.AreEqual(expected, actual);
            }
        }

        public void ColumnStreamCanSerializeBasicType()
        {
            var expected = Any.Int64;

            using (var t = new Transaction(this.sesid))
            using (var u = new Update(this.sesid, this.tableid, JET_prep.Insert))
            using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
            {
                var serializer = new BinaryFormatter
                {
                    Context = new StreamingContext(StreamingContextStates.Persistence)
                };
                serializer.Serialize(stream, expected);
                u.Save();
                t.Commit(CommitTransactionGrbit.LazyFlush);
            }

            Api.JetMove(this.sesid, this.tableid, JET_Move.First, MoveGrbit.None);
            using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
            {
                var deseriaizer = new BinaryFormatter();
                var actual = (long)deseriaizer.Deserialize(stream);
                Assert.AreEqual(expected, actual);
            }
        }

 public void ReadThrowsExceptionWhenNumberOfBytesIsNegative()
 {
     using (var t = new Transaction(this.sesid))
     using (var u = new Update(this.sesid, this.tableid, JET_prep.Insert))
     using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
     {
         var buffer = new byte[10];
         stream.Read(buffer, 0, -1);
     }
 }

 public void ReadThrowsExceptionWhenBufferOffsetIsTooBig()
 {
     using (var t = new Transaction(this.sesid))
     using (var u = new Update(this.sesid, this.tableid, JET_prep.Insert))
     using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
     {
         var buffer = new byte[10];
         stream.Read(buffer, buffer.Length, 1);
     }
 }

        public void SetAndRetrieveMultiValueColumnStream()
        {
            string[] data = { Any.String, Any.String, Any.String, Any.String, Any.String, Any.String };

            Api.JetBeginTransaction(this.sesid);
            Api.JetPrepareUpdate(this.sesid, this.tableid, JET_prep.Insert);
            for (int i = 0; i < data.Length; ++i)
            {
                var column = new ColumnStream(this.sesid, this.tableid, this.columnidLongText);
                column.Itag = i + 1;
                using (var writer = new StreamWriter(column))
                {
                    writer.WriteLine(data[i]);
                }
            }

            this.UpdateAndGotoBookmark();
            Api.JetCommitTransaction(this.sesid, CommitTransactionGrbit.LazyFlush);

            for (int i = 0; i < data.Length; ++i)
            {
                var column = new ColumnStream(this.sesid, this.tableid, this.columnidLongText);
                column.Itag = i + 1;
                using (var reader = new StreamReader(column))
                {
                    string actual = reader.ReadLine();
                    Assert.AreEqual(data[i], actual);
                }
            }
        }

        public void SetColumnStreamLength()
        {
            var bookmark = new byte[SystemParameters.BookmarkMost];
            int bookmarkSize;

            const long Length = 1345;

            using (var transaction = new Transaction(this.sesid))
            using (var update = new Update(this.sesid, this.tableid, JET_prep.Insert))
            using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
            {
                stream.SetLength(Length);
                Assert.AreEqual(Length, stream.Length);

                update.Save(bookmark, bookmark.Length, out bookmarkSize);
                transaction.Commit(CommitTransactionGrbit.LazyFlush);
            }

            Api.JetGotoBookmark(this.sesid, this.tableid, bookmark, bookmarkSize);
            using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
            {
                Assert.AreEqual(Length, stream.Length);
            }
        }

 public void ColumnStreamSupportsWrite()
 {
     using (var t = new Transaction(this.sesid))
     using (var u = new Update(this.sesid, this.tableid, JET_prep.Insert))
     using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
     {
         Assert.IsTrue(stream.CanWrite);
     }
 }

        public void SetColumnStreamToZeroLength()
        {
            var bookmark = new byte[SystemParameters.BookmarkMost];
            int bookmarkSize;

            using (var transaction = new Transaction(this.sesid))
            using (var update = new Update(this.sesid, this.tableid, JET_prep.Insert))
            using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
            {
                byte[] data = Any.Bytes;
                stream.Write(data, 0, data.Length);

                stream.SetLength(0);
                Assert.AreEqual(0, stream.Length);

                update.Save(bookmark, bookmark.Length, out bookmarkSize);
                transaction.Commit(CommitTransactionGrbit.LazyFlush);
            }

            Api.JetGotoBookmark(this.sesid, this.tableid, bookmark, bookmarkSize);
            CollectionAssert.AreEqual(new byte[0], Api.RetrieveColumn(this.sesid, this.tableid, this.columnidLongText));
        }

 public void ColumnStreamThrowsExceptionWhenSeekOffsetIsTooLarge()
 {
     using (var t = new Transaction(this.sesid))
     using (var u = new Update(this.sesid, this.tableid, JET_prep.Insert))
     using (var stream = new ColumnStream(this.sesid, this.tableid, this.columnidLongText))
     {
         stream.Seek(0x800000000, SeekOrigin.Begin);
     }
 }