<commit_before>namespace EventStore.Core.Sql
{
	using System;
	using System.Collections.Generic;
	using System.Data;
	using System.Data.Common;
	using System.Text;

	public class SqlEventStore : IStoreEvents
	{
		private const int SerializedDataIndex = 0;
		private const int VersionIndex = 1;
		private readonly IDictionary<Guid, long> versions = new Dictionary<Guid, long>();
		private readonly IDbConnection connection;
		private readonly SqlDialect dialect;
		private readonly ISerialize serializer;

		public SqlEventStore(IDbConnection connection, SqlDialect dialect, ISerialize serializer)
		{
			this.connection = connection;
			this.dialect = dialect;
			this.serializer = serializer;
		}

		public CommittedEventStream Read(Guid id)
		{
			return this.Read(id, 0, this.dialect.SelectEvents);
		}
		public CommittedEventStream ReadStartingFrom(Guid id, long version)
		{
			return this.Read(id, version, this.dialect.SelectEventsWhere);
		}
		private CommittedEventStream Read(Guid id, long version, string queryStatement)
		{
			using (var command = this.connection.CreateCommand())
			{
				command.CommandText = queryStatement;
				command.AddParameter(this.dialect.Id, id);
				command.AddParameter(this.dialect.CurrentVersion, version);
				using (var reader = this.WrapOnFailure(() => command.ExecuteReader()))
					return this.BuildStream(id, version, reader);
			}
		}
		private CommittedEventStream BuildStream(Guid id, long version, IDataReader reader)
		{
			ICollection<object> events = new LinkedList<object>();
			object snapshot = null;

			while (reader.Read())
				events.Add(this.serializer.Deserialize<object>(reader[SerializedDataIndex] as byte[]));

			if (reader.NextResult() && reader.Read())
			{
				snapshot = this.serializer.Deserialize<object>(reader[SerializedDataIndex] as byte[]);
				version = (long)reader[VersionIndex];
			}

			this.versions[id] = version + events.Count;
			return new CommittedEventStream(id, version + events.Count, events, snapshot);
		}

		public void Write(UncommittedEventStream stream)
		{
			using (var command = this.connection.CreateCommand())
			{
				long initialVersion;
				this.versions.TryGetValue(stream.Id, out initialVersion);
				this.versions[stream.Id] = initialVersion + stream.Events.Count;

				command.AddParameter(this.dialect.Id, stream.Id);
				command.AddParameter(this.dialect.InitialVersion, initialVersion);
				command.AddParameter(this.dialect.CurrentVersion, initialVersion + stream.Events.Count);
				command.AddParameter(this.dialect.Type, stream.Type == null ? null : stream.Type.FullName);<commit_msg>Added simple null validation checks.<commit_after>namespace EventStore.Core.Sql
{
	using System;
	using System.Collections.Generic;
	using System.Data;
	using System.Data.Common;
	using System.Text;

	public class SqlEventStore : IStoreEvents
	{
		private const int SerializedDataIndex = 0;
		private const int VersionIndex = 1;
		private readonly IDictionary<Guid, long> versions = new Dictionary<Guid, long>();
		private readonly IDbConnection connection;
		private readonly SqlDialect dialect;
		private readonly ISerialize serializer;

		public SqlEventStore(IDbConnection connection, SqlDialect dialect, ISerialize serializer)
		{
			this.connection = connection;
			this.dialect = dialect;
			this.serializer = serializer;
		}

		public CommittedEventStream Read(Guid id)
		{
			return this.Read(id, 0, this.dialect.SelectEvents);
		}
		public CommittedEventStream ReadStartingFrom(Guid id, long version)
		{
			return this.Read(id, version, this.dialect.SelectEventsWhere);
		}
		private CommittedEventStream Read(Guid id, long version, string queryStatement)
		{
			using (var command = this.connection.CreateCommand())
			{
				command.CommandText = queryStatement;
				command.AddParameter(this.dialect.Id, id);
				command.AddParameter(this.dialect.CurrentVersion, version);
				using (var reader = this.WrapOnFailure(() => command.ExecuteReader()))
					return this.BuildStream(id, version, reader);
			}
		}
		private CommittedEventStream BuildStream(Guid id, long version, IDataReader reader)
		{
			ICollection<object> events = new LinkedList<object>();
			object snapshot = null;

			while (reader.Read())
				events.Add(this.serializer.Deserialize<object>(reader[SerializedDataIndex] as byte[]));

			if (reader.NextResult() && reader.Read())
			{
				snapshot = this.serializer.Deserialize<object>(reader[SerializedDataIndex] as byte[]);
				version = (long)reader[VersionIndex];
			}

			this.versions[id] = version + events.Count;
			return new CommittedEventStream(id, version + events.Count, events, snapshot);
		}

		public void Write(UncommittedEventStream stream)
		{
			if (null == stream)
				return; // nothing to do

			stream.Events = stream.Events ?? new object[0]; // null protection
			if (stream.Events.Count == 0 && stream.Snapshot == null)
				return; // nothing to do

			using (var command = this.connection.CreateCommand())
			{
				long initialVersion;
				this.versions.TryGetValue(stream.Id, out initialVersion);
				this.versions[stream.Id] = initialVersion + stream.Events.Count;

				command.AddParameter(this.dialect.Id, stream.Id);
				command.AddParameter(this.dialect.InitialVersion, initialVersion);
				command.AddParameter(this.dialect.CurrentVersion, initialVersion + stream.Events.Count);
				command.AddParameter(this.dialect.Type, stream.Type == null ? null : stream.Type.FullName);
{'fraction_non_alphanumeric': 0.10661764705882353, 'fraction_numerical': 0.0014705882352941176, 'mean_word_length': 4.54638124362895, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2549967', 'n_tokens_mistral': 1808, 'n_tokens_neox': 1544, 'n_words': 367}