* RecordCache.cs - added. Each instance of record cache belongs to specific table and manages pool of records ( indexes into data containers) to be used by data rows.

* DataContainer.cs - added. Provides implementation for data containers that holds data in arrays of primitives or objects. Each data container belongs to specific DataColumn.
* DbDataAdapter.cs - changes to complete data storage redesign. Bug fix in FillTable ( to fetch exact number of records required ).

svn path=/trunk/mcs/; revision=27655

4 files changed, 948 insertions, 25 deletions

diff --git a/mcs/class/System.Data/System.Data.Common/ChangeLog b/mcs/class/System.Data/System.Data.Common/ChangeLog
index 56156d54796..5018969f92d 100755
--- a/mcs/class/System.Data/System.Data.Common/ChangeLog
+++ b/mcs/class/System.Data/System.Data.Common/ChangeLog
@@ -1,3 +1,11 @@
+2004-05-19  Boris Kirzner <borisk@mainsoft.com>
+	* RecordCache.cs - added. Each instance of record cache belongs to specific table
+	and manages pool of records ( indexes into data containers) to be used by data rows.
+	* DataContainer.cs - added. Provides implementation for data containers that holds data in arrays 
+	of primitives or objects. Each data container belongs to specific DataColumn.
+	* DbDataAdapter.cs - changes to complete data storage redesign. Bug fix in FillTable 
+	( to fetch exact number of records required ).
+	
 2004-05-13 Umadevi S (sumadevi@novell.com)
 	* DbDataPermissionAttribute.cs - Added KeyRestrictions property with a TODO tag
 
diff --git a/mcs/class/System.Data/System.Data.Common/DataContainer.cs b/mcs/class/System.Data/System.Data.Common/DataContainer.cs
new file mode 100644
index 00000000000..8e48fec4745
--- /dev/null
+++ b/mcs/class/System.Data/System.Data.Common/DataContainer.cs
@@ -0,0 +1,847 @@
+using System;
+using System.Collections;
+
+namespace System.Data.Common
+{
+	internal abstract class AbstractDataContainer
+	{
+		#region Fields
+
+		BitArray _nullValues;
+		System.Type _type;
+		DataColumn _column;
+
+		#endregion //Fields
+
+		#region Properties
+
+		internal abstract object this[int index] {
+			get;
+			set;
+		}
+
+		internal virtual int Capacity {
+			get { 
+				return (_nullValues != null) ? _nullValues.Count : 0; 
+			}
+			set { 
+				if (_nullValues == null) {
+					_nullValues = new BitArray(value);
+				}
+				else {
+					_nullValues.Length = value;
+				}
+			}
+		}
+
+		internal Type Type {
+			get {
+				return _type;
+			}
+		}
+
+		protected DataColumn Column {
+			get {
+				return _column;
+			}
+		}
+
+		#endregion //Properties
+
+		#region Methods
+
+		internal static AbstractDataContainer CreateInstance(Type type, DataColumn column)
+		{
+			AbstractDataContainer container;
+			switch (Type.GetTypeCode(type)) {
+				case TypeCode.Int16 :
+					container = new Int16DataContainer();
+					break;
+				case TypeCode.Int32 : 
+					container = new Int32DataContainer();
+					break;
+				case TypeCode.Int64 :
+					container = new Int64DataContainer();
+					break;
+				case TypeCode.String :
+					container = new StringDataContainer();
+					break;
+				case TypeCode.Boolean:
+					container = new BitDataContainer();
+					break;
+				case TypeCode.Byte :
+					container = new ByteDataContainer();
+					break;
+				//case TypeCode.Char :
+				case TypeCode.DateTime :
+					container = new DateTimeDataContainer();
+					break;
+				//case TypeCode.Decimal :
+				case TypeCode.Double :
+					container = new DoubleDataContainer();
+					break;
+				//case TypeCode.SByte :
+				case TypeCode.Single :
+					container = new SingleDataContainer();
+					break;
+				//case TypeCode.UInt16 :
+				//case TypeCode.UInt32 :
+				//case TypeCode.UInt64 :
+				default :
+					container = new ObjectDataContainer();
+					break;
+			}
+			container._type = type;
+			container._column = column;
+			return container;
+		}
+
+		internal bool IsNull(int index)
+		{
+			return (_nullValues != null) ? _nullValues[index] : true;
+		}
+
+		protected void SetNull(int index,bool isNull)
+		{
+			_nullValues[index] = isNull;
+		}
+
+		internal virtual void CopyValue(AbstractDataContainer fromContainer, int fromIndex, int toIndex)
+		{
+			_nullValues[toIndex] = fromContainer._nullValues[fromIndex];
+		}
+
+		internal virtual void CopyValue(int fromIndex, int toIndex)
+		{
+			_nullValues[toIndex] = _nullValues[fromIndex];
+		}
+
+		internal abstract void SetItemFromDataRecord(int index, IDataRecord record, int field);
+
+		internal abstract int CompareValues(int index1, int index2);
+
+		#endregion //Methods
+
+		sealed class Int16DataContainer : AbstractDataContainer
+		{
+			#region Fields
+		
+			short[] _values;
+
+			#endregion //Fields
+
+			#region Properties
+
+			internal override object this[int index] {
+				get {
+					return _values[index];
+				}
+				set {
+					if (value == null || value ==  DBNull.Value) {
+						SetValue(index,0);
+					}
+					else if( value is int ) {
+						SetValue(index,(short)value);
+					}
+					else {
+						SetValue(index,Convert.ToInt16(value));
+					}
+				}
+			}
+
+			internal override int Capacity {
+				set {
+					base.Capacity = value;
+					if (_values == null) {
+						_values = new short[value];
+					}
+					else {
+						short[] tmp = new short[value];
+						Array.Copy(_values,0,tmp,0,_values.Length);
+						_values = tmp;
+					}
+				}
+			}
+
+			#endregion //Properties
+
+			#region Methods
+			
+			private void SetValue(int index, short value)
+			{
+				_values[index] = value;
+				SetNull(index,value == 0);
+			}
+
+			internal override void SetItemFromDataRecord(int index, IDataRecord record, int field)
+			{
+				// if exception thrown, it should be caught 
+				// in the  caller method
+				SetValue(index,record.GetInt16(field));
+			}
+
+			internal override void CopyValue(int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromIndex, toIndex);
+				_values[toIndex] = _values[fromIndex];
+			}
+
+			internal override void CopyValue(AbstractDataContainer fromContainer, int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromContainer, fromIndex, toIndex);
+				_values[toIndex] = ((Int16DataContainer)fromContainer)._values[fromIndex];
+			}
+
+			internal override int CompareValues(int index1, int index2)
+			{
+				return (_values[index1] - _values[index2]);
+			}
+
+			#endregion //Methods
+		}
+
+		sealed class Int32DataContainer : AbstractDataContainer
+		{
+			#region Fields
+		
+			int[] _values;
+
+			#endregion //Fields
+
+			#region Properties
+
+			internal override object this[int index] {
+				get {
+					return _values[index];
+				}
+				set {
+					if (value == null || value ==  DBNull.Value) {
+						SetValue(index,0);
+					}
+					else if( value is int ) {
+						SetValue(index,(int)value);
+					}
+					else {
+						SetValue(index,Convert.ToInt32(value));
+					}
+				}
+			}
+
+			internal override int Capacity {
+				set {
+					base.Capacity = value;
+					if (_values == null) {
+						_values = new int[value];
+					}
+					else {
+						int[] tmp = new int[value];
+						Array.Copy(_values,0,tmp,0,_values.Length);
+						_values = tmp;
+					}
+				}
+			}
+
+			#endregion //Properties
+
+			#region Methods
+			
+			private void SetValue(int index, int value)
+			{
+				_values[index] = value;
+				SetNull(index,value == 0);
+			}
+
+			internal override void SetItemFromDataRecord(int index, IDataRecord record, int field)
+			{
+				// if exception thrown, it should be caught 
+				// in the  caller method
+				SetValue(index,record.GetInt32(field));
+			}
+
+			internal override void CopyValue(int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromIndex, toIndex);
+				_values[toIndex] = _values[fromIndex];
+			}
+
+			internal override void CopyValue(AbstractDataContainer fromContainer, int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromContainer, fromIndex, toIndex);
+				_values[toIndex] = ((Int32DataContainer)fromContainer)._values[fromIndex];
+			}
+
+			internal override int CompareValues(int index1, int index2)
+			{
+				int i1 = _values[index1];
+				int i2 = _values[index2];
+
+				if (i1 == i2) {
+					return 0;
+				}
+				return (i1 > i2) ? 1 : -1;
+			}
+
+			#endregion //Methods
+		}
+
+		sealed class Int64DataContainer : AbstractDataContainer
+		{
+			#region Fields
+		
+			long[] _values;
+
+			#endregion //Fields
+
+			#region Properties
+
+			internal override object this[int index] {
+				get {
+					return _values[index];
+				}
+				set {
+					if (value == null || value ==  DBNull.Value) {
+						SetValue(index,0);
+					}
+					else if( value is long ) {
+						SetValue(index,(long)value);
+					}
+					else {
+						SetValue(index,Convert.ToInt64(value));
+					}
+				}
+			}
+
+			internal override int Capacity {
+				set {
+					base.Capacity = value;
+					if (_values == null) {
+						_values = new long[value];
+					}
+					else {
+						long[] tmp = new long[value];
+						Array.Copy(_values,0,tmp,0,_values.Length);
+						_values = tmp;
+					}
+				}
+			}
+
+			#endregion //Properties
+
+			#region Methods
+			
+			private void SetValue(int index, long value)
+			{
+				_values[index] = value;
+				SetNull(index,value == 0);
+			}
+
+			internal override void SetItemFromDataRecord(int index, IDataRecord record, int field)
+			{
+				// if exception thrown, it should be caught 
+				// in the  caller method
+				SetValue(index,record.GetInt64(field));
+			}
+
+			internal override void CopyValue(int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromIndex, toIndex);
+				_values[toIndex] = _values[fromIndex];
+			}
+
+			internal override void CopyValue(AbstractDataContainer fromContainer, int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromContainer, fromIndex, toIndex);
+				_values[toIndex] = ((Int64DataContainer)fromContainer)._values[fromIndex];
+			}
+
+			internal override int CompareValues(int index1, int index2)
+			{
+				long l1 = _values[index1];
+				long l2 = _values[index2];
+
+				if (l1 == l2) {
+					return 0;
+				}
+				return (l1 > l2) ? 1 : -1;
+			}
+
+			#endregion //Methods
+		}
+
+		sealed class SingleDataContainer : AbstractDataContainer
+		{
+			#region Fields
+		
+			float[] _values;
+
+			#endregion //Fields
+
+			#region Properties
+
+			internal override object this[int index] {
+				get {
+					return _values[index];
+				}
+				set {
+					if (value == null || value ==  DBNull.Value) {
+						SetValue(index,0);
+					}
+					else if( value is float ) {
+						SetValue(index,(float)value);
+					}
+					else {
+						SetValue(index,Convert.ToSingle(value));
+					}
+				}
+			}
+
+			internal override int Capacity {
+				set {
+					base.Capacity = value;
+					if (_values == null) {
+						_values = new float[value];
+					}
+					else {
+						float[] tmp = new float[value];
+						Array.Copy(_values,0,tmp,0,_values.Length);
+						_values = tmp;
+					}
+				}
+			}
+
+			#endregion //Properties
+
+			#region Methods
+			
+			private void SetValue(int index, float value)
+			{
+				_values[index] = value;
+				SetNull(index,value == 0);
+			}
+
+			internal override void SetItemFromDataRecord(int index, IDataRecord record, int field)
+			{
+				// if exception thrown, it should be caught 
+				// in the  caller method
+				SetValue(index,record.GetFloat(field));
+			}
+
+			internal override void CopyValue(int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromIndex, toIndex);
+				_values[toIndex] = _values[fromIndex];
+			}
+
+			internal override void CopyValue(AbstractDataContainer fromContainer, int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromContainer, fromIndex, toIndex);
+				_values[toIndex] = ((SingleDataContainer)fromContainer)._values[fromIndex];
+			}
+
+			internal override int CompareValues(int index1, int index2)
+			{
+				return (int)(_values[index1] - _values[index2]);
+			}
+
+			#endregion //Methods
+		}
+
+		sealed class DoubleDataContainer : AbstractDataContainer
+		{
+			#region Fields
+		
+			double[] _values;
+
+			#endregion //Fields
+
+			#region Properties
+
+			internal override object this[int index] {
+				get {
+					return _values[index];
+				}
+				set {
+					if (value == null || value ==  DBNull.Value) {
+						SetValue(index,0);
+					}
+					else if( value is double ) {
+						SetValue(index,(double)value);
+					}
+					else {
+						SetValue(index,Convert.ToDouble(value));
+					}
+				}
+			}
+
+			internal override int Capacity {
+				set {
+					base.Capacity = value;
+					if (_values == null) {
+						_values = new double[value];
+					}
+					else {
+						double[] tmp = new double[value];
+						Array.Copy(_values,0,tmp,0,_values.Length);
+						_values = tmp;
+					}
+				}
+			}
+
+			#endregion //Properties
+
+			#region Methods
+			
+			private void SetValue(int index, double value)
+			{
+				_values[index] = value;
+				SetNull(index,value == 0);
+			}
+
+			internal override void SetItemFromDataRecord(int index, IDataRecord record, int field)
+			{
+				// if exception thrown, it should be caught 
+				// in the  caller method
+				SetValue(index,record.GetDouble(field));
+			}
+
+			internal override void CopyValue(int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromIndex, toIndex);
+				_values[toIndex] = _values[fromIndex];
+			}
+
+			internal override void CopyValue(AbstractDataContainer fromContainer, int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromContainer, fromIndex, toIndex);
+				_values[toIndex] = ((DoubleDataContainer)fromContainer)._values[fromIndex];
+			}
+
+			internal override int CompareValues(int index1, int index2)
+			{
+				return (int)(_values[index1] - _values[index2]);
+			}
+
+			#endregion //Methods
+		}
+
+		sealed class ByteDataContainer : AbstractDataContainer
+		{
+			#region Fields
+		
+			byte[] _values;
+
+			#endregion //Fields
+
+			#region Properties
+
+			internal override object this[int index] {
+				get {
+					return _values[index];
+				}
+				set {
+					if (value == null || value ==  DBNull.Value) {
+						SetValue(index,0);
+					}
+					else if( value is byte ) {
+						SetValue(index,(byte)value);
+					}
+					else {
+						SetValue(index,Convert.ToByte(value));
+					}
+				}
+			}
+
+			internal override int Capacity {
+				set {
+					base.Capacity = value;
+					if (_values == null) {
+						_values = new byte[value];
+					}
+					else {
+						byte[] tmp = new byte[value];
+						Array.Copy(_values,0,tmp,0,_values.Length);
+						_values = tmp;
+					}
+				}
+			}
+
+			#endregion //Properties
+
+			#region Methods
+			
+			private void SetValue(int index, byte value)
+			{
+				_values[index] = value;
+				SetNull(index,value == 0);
+			}
+
+			internal override void SetItemFromDataRecord(int index, IDataRecord record, int field)
+			{
+				// if exception thrown, it should be caught 
+				// in the  caller method
+				SetValue(index,record.GetByte(field));
+			}
+
+			internal override void CopyValue(int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromIndex, toIndex);
+				_values[toIndex] = _values[fromIndex];
+			}
+
+			internal override void CopyValue(AbstractDataContainer fromContainer, int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromContainer, fromIndex, toIndex);
+				_values[toIndex] = ((ByteDataContainer)fromContainer)._values[fromIndex];
+			}
+
+			internal override int CompareValues(int index1, int index2)
+			{
+				return (_values[index1] - _values[index2]);
+			}
+
+			#endregion //Methods
+		}
+
+		sealed class BitDataContainer : AbstractDataContainer
+		{
+			#region Fields
+		
+			// we don't need _values - using _nullValues instead
+
+			#endregion //Fields
+
+			#region Properties
+
+			internal override object this[int index] {
+				get {
+					return !IsNull(index);
+				}
+				set {
+					if (value == null || value ==  DBNull.Value) {
+						SetValue(index,false);
+					}
+					else if( value is bool ) {
+						SetValue(index,(bool)value);
+					}
+					else {
+						SetValue(index,Convert.ToBoolean(value));
+					}
+				}
+			}
+
+			internal override int Capacity {
+				set {
+					base.Capacity = value;
+				}
+			}
+
+			#endregion //Properties
+
+			#region Methods
+			
+			private void SetValue(int index, bool value)
+			{
+				SetNull(index,!value);
+			}
+
+			internal override void SetItemFromDataRecord(int index, IDataRecord record, int field)
+			{
+				// if exception thrown, it should be caught 
+				// in the  caller method
+				SetValue(index,record.GetBoolean(field));
+			}
+
+			internal override void CopyValue(int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromIndex, toIndex);
+			}
+
+			internal override int CompareValues(int index1, int index2)
+			{
+				return ((int)this[index1] - (int)this[index2]);
+			}
+
+			#endregion //Methods
+		}
+
+		class ObjectDataContainer : AbstractDataContainer
+		{
+			#region Fields
+		
+			object[] _values;
+
+			#endregion //Fields
+
+			#region Properties
+
+			internal override object this[int index] {
+				get {
+					return _values[index];
+				}
+				set {
+					SetValue(index,value);
+				}
+			}
+
+			internal override int Capacity {
+				set {
+					base.Capacity = value;
+					if (_values == null) {
+						_values = new object[value];
+					}
+					else {
+						object[] tmp = new object[value];
+						Array.Copy(_values,0,tmp,0,_values.Length);
+						_values = tmp;
+					}
+				}
+			}
+
+			#endregion //Properties
+
+			#region Methods
+			
+			protected virtual void SetValue(int index, object value)
+			{
+				if(value == null) {
+					value = DBNull.Value;
+				}
+				_values[index] = value;
+				SetNull(index,value == DBNull.Value);
+			}
+
+			internal override void SetItemFromDataRecord(int index, IDataRecord record, int field)
+			{
+				// if exception thrown, it should be caught 
+				// in the  caller method
+				SetValue(index,record.GetValue(field));
+			}
+
+			internal override void CopyValue(int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromIndex, toIndex);
+				_values[toIndex] = _values[fromIndex];
+			}
+
+			internal override void CopyValue(AbstractDataContainer fromContainer, int fromIndex, int toIndex)
+			{
+				base.CopyValue(fromContainer, fromIndex, toIndex);
+				_values[toIndex] = ((ObjectDataContainer)fromContainer)._values[fromIndex];
+			}
+
+			internal override int CompareValues(int index1, int index2)
+			{
+				object obj1 = _values[index1];
+				object obj2 = _values[index2];
+				if(obj1 == obj2) {
+					return 0;
+				}
+				else if (obj1 is IComparable) {
+					try {
+						return ((IComparable)obj1).CompareTo(obj2);
+					}
+					catch {
+						//just suppress
+					}
+
+					if (obj2 is IComparable) {
+						obj2 = Convert.ChangeType(obj2, Type.GetTypeCode(obj1.GetType()));
+						return ((IComparable)obj1).CompareTo(obj2);
+					}
+				}
+
+				return String.Compare(obj1.ToString(), obj2.ToString());
+			}
+
+			#endregion //Methods
+	 
+		}
+
+		sealed class StringDataContainer : ObjectDataContainer
+		{
+			#region Methods
+
+			private void SetValue(int index, string value)
+			{
+				if (value != null && Column.MaxLength >= 0 && Column.MaxLength < value.Length ) {
+					throw new ArgumentException("Cannot set column '" + Column.ColumnName + "' to '" + value + "'. The value violates the MaxLength limit of this column.");
+				}
+				base.SetValue(index,value);
+			}
+			
+			protected override void SetValue(int index, object value)
+			{
+				if ( value != null && value != DBNull.Value && !(value is string)) {
+					SetValue(index, Convert.ToString(value));
+					return;
+				}
+
+				base.SetValue(index, value);
+			}
+
+			internal override void SetItemFromDataRecord(int index, IDataRecord record, int field)
+			{
+				// if exception thrown, it should be caught 
+				// in the  caller method
+				SetValue(index,record.GetString(field));
+			}
+
+			internal override int CompareValues(int index1, int index2)
+			{
+				bool isNull1 = IsNull(index1);
+				bool isNull2 = IsNull(index2);
+
+				if (isNull1) {
+					return isNull2 ? 0 : -1;
+				}
+				else {
+					if (isNull2) {
+						return 1;
+					}
+				}
+				return String.Compare((string)this[index1], (string)this[index2], !Column.Table.CaseSensitive);
+			}
+
+			#endregion //Methods 
+		}
+
+		sealed class DateTimeDataContainer : ObjectDataContainer
+		{
+			#region Methods
+			
+			protected override void SetValue(int index, object value)
+			{
+				if ( value != null && value != DBNull.Value && !(value is DateTime)) {
+					value = Convert.ToDateTime(value);
+				}
+
+				base.SetValue(index,value);
+			}
+
+			internal override void SetItemFromDataRecord(int index, IDataRecord record, int field)
+			{
+				// if exception thrown, it should be caught 
+				// in the  caller method
+				base.SetValue(index,record.GetDateTime(field));
+			}
+
+			internal override int CompareValues(int index1, int index2)
+			{
+				bool isNull1 = IsNull(index1);
+				bool isNull2 = IsNull(index2);
+
+				if (isNull1) {
+					return isNull2 ? 0 : -1;
+				}
+				else {
+					if (isNull2) {
+						return 1;
+					}
+				}
+				return DateTime.Compare((DateTime)this[index1], (DateTime)this[index2]);
+			}
+
+			#endregion //Methods 
+		}
+	}
+}
diff --git a/mcs/class/System.Data/System.Data.Common/DbDataAdapter.cs b/mcs/class/System.Data/System.Data.Common/DbDataAdapter.cs
index da1cd2b0ff0..41c787fd57a 100644
--- a/mcs/class/System.Data/System.Data.Common/DbDataAdapter.cs
+++ b/mcs/class/System.Data/System.Data.Common/DbDataAdapter.cs
@@ -250,11 +250,6 @@ namespace System.Data.Common {
 			if (maxRecords < 0)
 				throw new ArgumentException ("The maxRecords parameter was less than 0.");
 
-			//if (dataReader.FieldCount == 0) {
-			//	dataReader.Close ();
-			//	return 0;
-			//}
-
 			DataTable dataTable;
 			int resultIndex = 0;
 			int count = 0;
@@ -274,7 +269,6 @@ namespace System.Data.Common {
 						}
 
 						if (!FillTable (dataTable, dataReader, startRecord, maxRecords, ref count)) {
-							//break;
 							continue;
 						}
 
@@ -284,11 +278,12 @@ namespace System.Data.Common {
 						maxRecords = 0;
 					}
 				} while (dataReader.NextResult ());
-			} finally {
+			} 
+			finally {
 				dataReader.Close ();
 			}
 
-            return count;
+                        return count;
 		}
 		
 		protected virtual int Fill (DataSet dataSet, int startRecord, int maxRecords, string srcTable, IDbCommand command, CommandBehavior behavior) 
@@ -312,31 +307,32 @@ namespace System.Data.Common {
 			int counterStart = counter;
 			
 			int[] mapping = BuildSchema (dataReader, dataTable, SchemaType.Mapped);
-			
-			object[] readerArray = new object[dataReader.FieldCount];
-			object[] tableArray = new object[mapping.Length];
-
-			for (int i = 0; i < startRecord; i++)
-				dataReader.Read ();
 
-			while (dataReader.Read () && (maxRecords == 0 || (counterStart - counter) < maxRecords)) {
-
-				// we get the values from the datareader
-				dataReader.GetValues (readerArray);
-				// copy from datareader columns to table columns according to given mapping
-				for (int i = 0; i < mapping.Length; i++)
-					tableArray[i] = readerArray[mapping[i]];
+				for (int i = 0; i < startRecord; i++) {
+					dataReader.Read ();
+				}
 
+			while (dataReader.Read () && (maxRecords == 0 || (counter - counterStart) < maxRecords)) {
 				try {
 					dataTable.BeginLoadData ();
-					dataTable.LoadDataRow (tableArray, AcceptChangesDuringFill);
+					dataTable.LoadDataRow (dataReader, mapping, AcceptChangesDuringFill);
 					dataTable.EndLoadData ();
 					counter++;
-				} catch (Exception e) {
+				} 
+				catch (Exception e) {
+					object[] readerArray = new object[dataReader.FieldCount];
+					object[] tableArray = new object[mapping.Length];
+					// we get the values from the datareader
+					dataReader.GetValues (readerArray);
+					// copy from datareader columns to table columns according to given mapping
+					for (int i = 0; i < mapping.Length; i++) {
+						tableArray[i] = readerArray[mapping[i]];
+					}
 					FillErrorEventArgs args = CreateFillErrorEvent (dataTable, tableArray, e);
 					OnFillError (args);
-					if(!args.Continue)
+					if(!args.Continue) {
 						return false;
+					}
 				}
 			}
 			return true;
@@ -473,7 +469,7 @@ namespace System.Data.Common {
 			foreach (DataRow schemaRow in reader.GetSchemaTable ().Rows) {
 				// generate a unique column name in the source table.
 				string sourceColumnName;
-				if (schemaRow ["ColumnName"].Equals (DBNull.Value))
+				if (schemaRow.IsNull("ColumnName"))
 					sourceColumnName = DefaultSourceColumnName;
 				else 
 					sourceColumnName = (string) schemaRow ["ColumnName"];
diff --git a/mcs/class/System.Data/System.Data.Common/RecordCache.cs b/mcs/class/System.Data/System.Data.Common/RecordCache.cs
new file mode 100644
index 00000000000..0cb454887c4
--- /dev/null
+++ b/mcs/class/System.Data/System.Data.Common/RecordCache.cs
@@ -0,0 +1,72 @@
+using System;
+using System.Collections;
+
+namespace System.Data.Common
+{
+	internal class RecordCache
+	{
+		#region Fields
+
+		const int MIN_CACHE_SIZE = 128;
+
+		Stack _records = new Stack(16);
+		int _nextFreeIndex = 0;
+		int _currentCapacity = 0;
+		DataTable _table;
+
+		#endregion // Fields
+
+		#region Constructors
+
+		internal RecordCache(DataTable table)
+		{
+			_table = table;
+		}
+
+		#endregion //Constructors
+
+		#region Properties
+
+		internal int CurrentCapacity 
+		{
+			get {
+				return _currentCapacity;
+			}
+		}
+
+		#endregion // Properties
+
+		#region Methods
+
+		internal int NewRecord()
+		{
+			if (_records.Count > 0) {
+				return (int)_records.Pop();
+			}
+			else {
+				DataColumnCollection cols = _table.Columns;
+				if (_nextFreeIndex >= _currentCapacity) {
+					_currentCapacity *= 2;
+					if ( _currentCapacity < MIN_CACHE_SIZE ) {
+						_currentCapacity = MIN_CACHE_SIZE;
+					}
+					foreach(DataColumn col in cols) {
+						col.DataContainer.Capacity = _currentCapacity;
+					}
+				}
+				return _nextFreeIndex++;
+			}
+		}
+
+		internal void DisposeRecord(int index)
+		{
+			if ( index < 0 ) {
+				throw new ArgumentException();
+			}
+			_records.Push(index);
+		}
+
+		#endregion // Methods
+
+	}
+}