<commit_before>using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Linq.Expressions;
using DataBoss.Data;

namespace DataBoss.DataPackage
{
	public interface ITransformedDataRecord : IDataRecord
	{
		IDataRecord Source { get; }
	}

	public class DataReaderTransform : IDataReader, ITransformedDataRecord
	{
		public static class FieldInfo<T>
		{
			public static readonly Type FieldType = typeof(T).TryGetNullableTargetType(out var targetType) ? targetType : typeof(T);
			public static readonly bool IsNullable = typeof(T).IsNullable();
			public static readonly int ColumnSize = DataBossDbType.From(typeof(T)).ColumnSize ?? -1;
			public static readonly Func<T, bool> IsNull = CreateIsNullMethod();

			static Func<T, bool> CreateIsNullMethod() {
				var p0 = Expression.Parameter(typeof(T));
				Expression CreateNullCheck(Expression input) {
					if (p0.Type.IsNullable())
						return Expression.Not(Expression.MakeMemberAccess(input, typeof(T).GetMember("HasValue").Single()));
					if (p0.Type.IsValueType)
						return Expression.Constant(false, typeof(bool));

					return Expression.ReferenceEqual(input, Expression.Constant(null, p0.Type));
				}
				return Expression.Lambda<Func<T, bool>>(CreateNullCheck(p0), p0).Compile();
			}
		}

		delegate IFieldAccessor AccessorBuilder(int ordinal, bool allowDBNull, int columnSize);

		static readonly ConcurrentDictionary<Type, AccessorBuilder> fieldAccessorCtorCache = new();

		static IFieldAccessor CreateFieldAccessor(int ordinal, bool isNullable, int columnSize, Type type) {
			var ctor = fieldAccessorCtorCache.GetOrAdd(type, fieldType => {
				var ps = Array.ConvertAll(
					typeof(AccessorBuilder).GetMethod("Invoke").GetParameters(),
					x => Expression.Parameter(x.ParameterType, x.Name));
				var c = typeof(SourceFieldAccessor<>).MakeGenericType(fieldType).GetConstructor(Array.ConvertAll(ps, x => x.Type));
				return Expression.Lambda<AccessorBuilder>(Expression.New(c, ps), ps).Compile();
			});

			return ctor(ordinal, isNullable, columnSize);
		}

		interface IFieldAccessor
		{
			bool AllowDBNull { get; }
			int ColumnSize { get; }
			Type FieldType { get; }
			string Name { get; set; }

			bool IsDBNull(DataReaderTransform record);

			object GetValue(DataReaderTransform record);
			T GetFieldValue<T>(DataReaderTransform record);
		}

		abstract class SourceFieldAccessor : IFieldAccessor
		{
			public SourceFieldAccessor(int ordinal, bool allowDBNull, int columnSize) {
				this.Ordinal = ordinal;
				this.AllowDBNull = allowDBNull;
				this.ColumnSize = columnSize;
			}

			public bool AllowDBNull { get; }
			public int ColumnSize { get; }
			public int Ordinal { get; }
			public abstract Type FieldType { get; }
			public string Name { get; set; }

			public bool IsDBNull(DataReaderTransform record) => record.Source.IsDBNull(Ordinal);

			public abstract object GetValue(DataReaderTransform record);
			public T GetFieldValue<T>(DataReaderTransform record) => record.Source.GetFieldValue<T>(Ordinal);
		}

		class SourceFieldAccessor<TField> : SourceFieldAccessor
		{
			public SourceFieldAccessor(int ordinal, bool allowDBNull, int columnSize) : base(ordinal, allowDBNull, columnSize) 
			{ }

			public override Type FieldType => FieldInfo<TField>.FieldType;
			public override object GetValue(DataReaderTransform record) => GetFieldValue<TField>(record);
		}

		abstract class UserFieldAccessor<TField> : IFieldAccessor
		{
			TField currentValue;
			bool isDirty;

			public UserFieldAccessor() {
				this.isDirty = true;
			}

			public virtual bool AllowDBNull => FieldInfo<TField>.IsNullable;
			public int ColumnSize => FieldInfo<TField>.ColumnSize;
			public Type FieldType => FieldInfo<TField>.FieldType;
			public string Name { get; set; }

			public T GetFieldValue<T>(DataReaderTransform record) =>
				(T)(object)GetCurrentValue(record);

			public object GetValue(DataReaderTransform record) {
				var value = GetCurrentValue(record);
				return FieldInfo<TField>.IsNull(value) ? DBNull.Value : value;
			}

			public bool IsDBNull(DataReaderTransform record) => AllowDBNull && FieldInfo<TField>.IsNull(GetCurrentValue(record));

			TField GetCurrentValue(DataReaderTransform record) {
				if (isDirty) {
					currentValue = GetRecordValue(record);
					isDirty = false;
				}
				return currentValue;
			}

			protected abstract TField GetRecordValue(DataReaderTransform record);

			public void Dirty() { isDirty = true; }
		}

		class FieldTransform<T> : UserFieldAccessor<T>
		{
			readonly Func<ITransformedDataRecord, T> transform;

			public FieldTransform(Func<ITransformedDataRecord, T> transform) {
				this.transform = transform;
			}

			protected override T GetRecordValue(DataReaderTransform record) => transform(record);
		}

		class FieldTransform<TField, T> : UserFieldAccessor<T>
		{
			readonly IFieldAccessor source;
			readonly Func<TField, T> transform;
			readonly bool allowDBNull;

			public FieldTransform(IFieldAccessor source, Func<TField, T> transform, bool allowDBNull) {
				this.source = source;
				this.transform = transform;
				this.allowDBNull = allowDBNull;
			}

			public override bool AllowDBNull => allowDBNull;

			protected override T GetRecordValue(DataReaderTransform record) => transform(source.GetFieldValue<TField>(record));
		}

		interface IRecordTransform 
		{
			void Reset();
		}

		class RecordTransform<TRecord, T> : UserFieldAccessor<T>, IRecordTransform
		{
			Func<IDataReader, TRecord> readRecord;
			readonly Func<TRecord, T> selector;

			public RecordTransform(Func<TRecord, T> selector) {
				this.selector = selector;
			}

			protected override T GetRecordValue(DataReaderTransform record) =>
				selector(ReadRecord(record));

			TRecord ReadRecord(DataReaderTransform reader) {
				if(readRecord == null) {
					var recordFields = new HashSet<string>(reader.fields.Where(x => x is IRecordTransform).Select(x => x.Name));
					var fields = FieldMap.Create(reader, x => !recordFields.Contains(x.ColumnName));
					readRecord = ConverterFactory.Default.BuildConverter<IDataReader, TRecord>(fields).Compiled;
				}
				return readRecord(reader);
			}

			void IRecordTransform.Reset() {
				readRecord = null;
			}
		}

		readonly IDataReader inner;
		readonly List<IFieldAccessor> fields = new();
		Action onRead;

		public DataReaderTransform(IDataReader inner) {
			this.inner = inner;
			var schema = inner.GetDataReaderSchemaTable();
			var sourceFields = new IFieldAccessor[schema.Count];
			for(var i = 0; i != schema.Count; ++i) {
				var field = schema[i];
				var accessor = CreateFieldAccessor(
					field.Ordinal,
					field.AllowDBNull, 
					field.ColumnSize ?? -1, 
					field.ColumnType);
				accessor.Name = field.ColumnName;
				sourceFields[field.Ordinal] = accessor;
			}<commit_msg>RecordTransforms can now access any prior fields + source.<commit_after>using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Linq.Expressions;
using DataBoss.Data;

namespace DataBoss.DataPackage
{
	public interface ITransformedDataRecord : IDataRecord
	{
		IDataRecord Source { get; }
	}

	public class DataReaderTransform : IDataReader, ITransformedDataRecord
	{
		public static class FieldInfo<T>
		{
			public static readonly Type FieldType = typeof(T).TryGetNullableTargetType(out var targetType) ? targetType : typeof(T);
			public static readonly bool IsNullable = typeof(T).IsNullable();
			public static readonly int ColumnSize = DataBossDbType.From(typeof(T)).ColumnSize ?? -1;
			public static readonly Func<T, bool> IsNull = CreateIsNullMethod();

			static Func<T, bool> CreateIsNullMethod() {
				var p0 = Expression.Parameter(typeof(T));
				Expression CreateNullCheck(Expression input) {
					if (p0.Type.IsNullable())
						return Expression.Not(Expression.MakeMemberAccess(input, typeof(T).GetMember("HasValue").Single()));
					if (p0.Type.IsValueType)
						return Expression.Constant(false, typeof(bool));

					return Expression.ReferenceEqual(input, Expression.Constant(null, p0.Type));
				}
				return Expression.Lambda<Func<T, bool>>(CreateNullCheck(p0), p0).Compile();
			}
		}

		delegate IFieldAccessor AccessorBuilder(int ordinal, bool allowDBNull, int columnSize);

		static readonly ConcurrentDictionary<Type, AccessorBuilder> fieldAccessorCtorCache = new();

		static IFieldAccessor CreateFieldAccessor(int ordinal, bool isNullable, int columnSize, Type type) {
			var ctor = fieldAccessorCtorCache.GetOrAdd(type, fieldType => {
				var ps = Array.ConvertAll(
					typeof(AccessorBuilder).GetMethod("Invoke").GetParameters(),
					x => Expression.Parameter(x.ParameterType, x.Name));
				var c = typeof(SourceFieldAccessor<>).MakeGenericType(fieldType).GetConstructor(Array.ConvertAll(ps, x => x.Type));
				return Expression.Lambda<AccessorBuilder>(Expression.New(c, ps), ps).Compile();
			});

			return ctor(ordinal, isNullable, columnSize);
		}

		interface IFieldAccessor
		{
			bool AllowDBNull { get; }
			int ColumnSize { get; }
			Type FieldType { get; }
			string Name { get; set; }

			bool IsDBNull(DataReaderTransform record);

			object GetValue(DataReaderTransform record);
			T GetFieldValue<T>(DataReaderTransform record);
		}

		abstract class SourceFieldAccessor : IFieldAccessor
		{
			public SourceFieldAccessor(int ordinal, bool allowDBNull, int columnSize) {
				this.Ordinal = ordinal;
				this.AllowDBNull = allowDBNull;
				this.ColumnSize = columnSize;
			}

			public bool AllowDBNull { get; }
			public int ColumnSize { get; }
			public int Ordinal { get; }
			public abstract Type FieldType { get; }
			public string Name { get; set; }

			public bool IsDBNull(DataReaderTransform record) => record.Source.IsDBNull(Ordinal);

			public abstract object GetValue(DataReaderTransform record);
			public T GetFieldValue<T>(DataReaderTransform record) => record.Source.GetFieldValue<T>(Ordinal);
		}

		class SourceFieldAccessor<TField> : SourceFieldAccessor
		{
			public SourceFieldAccessor(int ordinal, bool allowDBNull, int columnSize) : base(ordinal, allowDBNull, columnSize) 
			{ }

			public override Type FieldType => FieldInfo<TField>.FieldType;
			public override object GetValue(DataReaderTransform record) => GetFieldValue<TField>(record);
		}

		abstract class UserFieldAccessor<TField> : IFieldAccessor
		{
			TField currentValue;
			bool isDirty;

			public UserFieldAccessor() {
				this.isDirty = true;
			}

			public virtual bool AllowDBNull => FieldInfo<TField>.IsNullable;
			public int ColumnSize => FieldInfo<TField>.ColumnSize;
			public Type FieldType => FieldInfo<TField>.FieldType;
			public string Name { get; set; }

			public T GetFieldValue<T>(DataReaderTransform record) =>
				(T)(object)GetCurrentValue(record);

			public object GetValue(DataReaderTransform record) {
				var value = GetCurrentValue(record);
				return FieldInfo<TField>.IsNull(value) ? DBNull.Value : value;
			}

			public bool IsDBNull(DataReaderTransform record) => AllowDBNull && FieldInfo<TField>.IsNull(GetCurrentValue(record));

			TField GetCurrentValue(DataReaderTransform record) {
				if (isDirty) {
					currentValue = GetRecordValue(record);
					isDirty = false;
				}
				return currentValue;
			}

			protected abstract TField GetRecordValue(DataReaderTransform record);

			public void Dirty() { isDirty = true; }
		}

		class FieldTransform<T> : UserFieldAccessor<T>
		{
			readonly Func<ITransformedDataRecord, T> transform;

			public FieldTransform(Func<ITransformedDataRecord, T> transform) {
				this.transform = transform;
			}

			protected override T GetRecordValue(DataReaderTransform record) => transform(record);
		}

		class FieldTransform<TField, T> : UserFieldAccessor<T>
		{
			readonly IFieldAccessor source;
			readonly Func<TField, T> transform;
			readonly bool allowDBNull;

			public FieldTransform(IFieldAccessor source, Func<TField, T> transform, bool allowDBNull) {
				this.source = source;
				this.transform = transform;
				this.allowDBNull = allowDBNull;
			}

			public override bool AllowDBNull => allowDBNull;

			protected override T GetRecordValue(DataReaderTransform record) => transform(source.GetFieldValue<TField>(record));
		}

		interface IRecordTransform 
		{
			void Reset();
		}

		class RecordTransform<TRecord, T> : UserFieldAccessor<T>, IRecordTransform
		{
			Func<IDataReader, TRecord> readRecord;
			readonly Func<TRecord, T> selector;

			public RecordTransform(Func<TRecord, T> selector) {
				this.selector = selector;
			}

			protected override T GetRecordValue(DataReaderTransform record) =>
				selector(ReadRecord(record));

			TRecord ReadRecord(DataReaderTransform reader) {
				if(readRecord == null) {
					var self = reader.fields.IndexOf(this);
					var fields = FieldMap.Create(reader, x => x.Ordinal < self || reader.fields[x.Ordinal] is SourceFieldAccessor);
					readRecord = ConverterFactory.Default.BuildConverter<IDataReader, TRecord>(fields).Compiled;
				}
				return readRecord(reader);
			}

			void IRecordTransform.Reset() {
				readRecord = null;
			}
		}

		readonly IDataReader inner;
		readonly List<IFieldAccessor> fields = new();
		Action onRead;

		public DataReaderTransform(IDataReader inner) {
			this.inner = inner;
			var schema = inner.GetDataReaderSchemaTable();
			var sourceFields = new IFieldAccessor[schema.Count];
			for(var i = 0; i != schema.Count; ++i) {
				var field = schema[i];
				var accessor = CreateFieldAccessor(
					field.Ordinal,
					field.AllowDBNull, 
					field.ColumnSize ?? -1, 
					field.ColumnType);
				accessor.Name = field.ColumnName;
				sourceFields[field.Ordinal] = accessor;
			}
{'fraction_non_alphanumeric': 0.10131837715784106, 'fraction_numerical': 0.0013110933061402869, 'mean_word_length': 4.465764331210191, 'pattern_counts': {'":': 0, '<': 79, '<?xml version=': 0, '>': 114, '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': '3742935', 'n_tokens_mistral': 4723, 'n_tokens_neox': 3939, 'n_words': 942}