<commit_before>namespace EventStore.Persistence.SqlPersistence
{
	using System;
	using System.Collections.Generic;
	using System.Data;
	using System.Linq;
	using System.Threading;
	using System.Transactions;
	using Logging;
	using Persistence;
	using Serialization;

	public class SqlPersistenceEngine : IPersistStreams
	{
		private static readonly ILog Logger = LogFactory.BuildLogger(typeof(SqlPersistenceEngine));
		private static readonly DateTime EpochTime = new DateTime(1970, 1, 1);
		private readonly IConnectionFactory connectionFactory;
		private readonly ISqlDialect dialect;
		private readonly ISerialize serializer;
		private readonly TransactionScopeOption scopeOption;
		private readonly int pageSize;
		private int initialized;
		private bool disposed;

		public SqlPersistenceEngine(
			IConnectionFactory connectionFactory,
			ISqlDialect dialect,
			ISerialize serializer,
			TransactionScopeOption scopeOption,
			int pageSize)
		{
			if (connectionFactory == null)
				throw new ArgumentNullException("connectionFactory");

			if (dialect == null)
				throw new ArgumentNullException("dialect");

			if (serializer == null)
				throw new ArgumentNullException("serializer");

			if (pageSize < 0)
				throw new ArgumentException("pageSize");

			this.connectionFactory = connectionFactory;
			this.dialect = dialect;
			this.serializer = serializer;
			this.scopeOption = scopeOption;
			this.pageSize = pageSize;

			Logger.Debug(Messages.UsingScope, this.scopeOption.ToString());
		}

		public void Dispose()
		{
			this.Dispose(true);
			GC.SuppressFinalize(this);
		}
		protected virtual void Dispose(bool disposing)
		{
			if (!disposing || this.disposed)
				return;

			Logger.Debug(Messages.ShuttingDownPersistence);
			this.disposed = true;
		}

		public virtual void Initialize()
		{
			if (Interlocked.Increment(ref this.initialized) > 1)
				return;

			Logger.Debug(Messages.InitializingStorage);
			this.ExecuteCommand(Guid.Empty, statement =>
				statement.ExecuteWithoutExceptions(this.dialect.InitializeStorage));
		}

		public virtual IEnumerable<Commit> GetFrom(Guid streamId, int minRevision, int maxRevision)
		{
			Logger.Debug(Messages.GettingAllCommitsBetween, streamId, minRevision, maxRevision);
			return this.ExecuteQuery(streamId, query =>
			{
				var statement = this.dialect.GetCommitsFromStartingRevision;
				query.AddParameter(this.dialect.StreamId, streamId);
				query.AddParameter(this.dialect.StreamRevision, minRevision);
				query.AddParameter(this.dialect.MaxStreamRevision, maxRevision);
				query.AddParameter(this.dialect.CommitSequence, 0);
				return query.ExecutePagedQuery(statement,
						(q, r) => q.SetParameter(this.dialect.CommitSequence, r.CommitSequence()))
					.Select(x => x.GetCommit(this.serializer));
			});
		}

		public virtual IEnumerable<Commit> GetFrom(DateTime start)
		{
			start = start < EpochTime ? EpochTime : start;

			Logger.Debug(Messages.GettingAllCommitsFrom, start);
			return this.ExecuteQuery(Guid.Empty, query =>
			{
				var statement = this.dialect.GetCommitsFromInstant;
				query.AddParameter(this.dialect.CommitStamp, start);
				return query.ExecutePagedQuery(statement, (q, r) => { })
					.Select(x => x.GetCommit(this.serializer));
			});
		}

		public virtual IEnumerable<Commit> GetFromTo(DateTime start, DateTime end)
		{
			start = start < EpochTime ? EpochTime : start;
			end = end < EpochTime ? EpochTime : end;<commit_msg>Rounds down the start datetime in GetFrom and GetFromTo to the nearest second in SqlPersistenceEngine to accomodate differences in DateTime precisions in variours SQL engines.<commit_after>namespace EventStore.Persistence.SqlPersistence
{
	using System;
	using System.Collections.Generic;
	using System.Data;
	using System.Linq;
	using System.Threading;
	using System.Transactions;
	using Logging;
	using Persistence;
	using Serialization;

	public class SqlPersistenceEngine : IPersistStreams
	{
		private static readonly ILog Logger = LogFactory.BuildLogger(typeof(SqlPersistenceEngine));
		private static readonly DateTime EpochTime = new DateTime(1970, 1, 1);
		private readonly IConnectionFactory connectionFactory;
		private readonly ISqlDialect dialect;
		private readonly ISerialize serializer;
		private readonly TransactionScopeOption scopeOption;
		private readonly int pageSize;
		private int initialized;
		private bool disposed;

		public SqlPersistenceEngine(
			IConnectionFactory connectionFactory,
			ISqlDialect dialect,
			ISerialize serializer,
			TransactionScopeOption scopeOption,
			int pageSize)
		{
			if (connectionFactory == null)
				throw new ArgumentNullException("connectionFactory");

			if (dialect == null)
				throw new ArgumentNullException("dialect");

			if (serializer == null)
				throw new ArgumentNullException("serializer");

			if (pageSize < 0)
				throw new ArgumentException("pageSize");

			this.connectionFactory = connectionFactory;
			this.dialect = dialect;
			this.serializer = serializer;
			this.scopeOption = scopeOption;
			this.pageSize = pageSize;

			Logger.Debug(Messages.UsingScope, this.scopeOption.ToString());
		}

		public void Dispose()
		{
			this.Dispose(true);
			GC.SuppressFinalize(this);
		}
		protected virtual void Dispose(bool disposing)
		{
			if (!disposing || this.disposed)
				return;

			Logger.Debug(Messages.ShuttingDownPersistence);
			this.disposed = true;
		}

		public virtual void Initialize()
		{
			if (Interlocked.Increment(ref this.initialized) > 1)
				return;

			Logger.Debug(Messages.InitializingStorage);
			this.ExecuteCommand(Guid.Empty, statement =>
				statement.ExecuteWithoutExceptions(this.dialect.InitializeStorage));
		}

		public virtual IEnumerable<Commit> GetFrom(Guid streamId, int minRevision, int maxRevision)
		{
			Logger.Debug(Messages.GettingAllCommitsBetween, streamId, minRevision, maxRevision);
			return this.ExecuteQuery(streamId, query =>
			{
				var statement = this.dialect.GetCommitsFromStartingRevision;
				query.AddParameter(this.dialect.StreamId, streamId);
				query.AddParameter(this.dialect.StreamRevision, minRevision);
				query.AddParameter(this.dialect.MaxStreamRevision, maxRevision);
				query.AddParameter(this.dialect.CommitSequence, 0);
				return query.ExecutePagedQuery(statement,
						(q, r) => q.SetParameter(this.dialect.CommitSequence, r.CommitSequence()))
					.Select(x => x.GetCommit(this.serializer));
			});
		}

		public virtual IEnumerable<Commit> GetFrom(DateTime start)
		{
            start = start.AddTicks(-(start.Ticks % TimeSpan.TicksPerSecond)); // Rounds down to the nearest second.
			start = start < EpochTime ? EpochTime : start;

			Logger.Debug(Messages.GettingAllCommitsFrom, start);
			return this.ExecuteQuery(Guid.Empty, query =>
			{
				var statement = this.dialect.GetCommitsFromInstant;
				query.AddParameter(this.dialect.CommitStamp, start);
				return query.ExecutePagedQuery(statement, (q, r) => { })
					.Select(x => x.GetCommit(this.serializer));
			});
		}

		public virtual IEnumerable<Commit> GetFromTo(DateTime start, DateTime end)
		{
            start = start.AddTicks(-(start.Ticks % TimeSpan.TicksPerSecond)); // Rounds down to the nearest second.
			start = start < EpochTime ? EpochTime : start;
			end = end < EpochTime ? EpochTime : end;
{'fraction_non_alphanumeric': 0.09519230769230769, 'fraction_numerical': 0.0024725274725274724, 'mean_word_length': 4.532674772036474, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 25, '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': '5218669', 'n_tokens_mistral': 2474, 'n_tokens_neox': 2084, 'n_words': 483}