Changes to create storage directory for storage engines.

4 files changed, 2498 insertions, 0 deletions

diff --git a/storage/innobase/rem/Makefile.am b/storage/innobase/rem/Makefile.am
new file mode 100644
index 00000000000..e2b2fdaf669
--- /dev/null
+++ b/storage/innobase/rem/Makefile.am
@@ -0,0 +1,24 @@
+# Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB
+# & Innobase Oy
+# 
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+# 
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+# 
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+
+include ../include/Makefile.i
+
+noinst_LIBRARIES =	librem.a
+
+librem_a_SOURCES =	rem0rec.c rem0cmp.c
+
+EXTRA_PROGRAMS =	
diff --git a/storage/innobase/rem/makefilewin b/storage/innobase/rem/makefilewin
new file mode 100644
index 00000000000..51ca4a92012
--- /dev/null
+++ b/storage/innobase/rem/makefilewin
@@ -0,0 +1,12 @@
+include ..\include\makefile.i
+
+rem.lib: rem0rec.obj rem0cmp.obj
+	lib -out:..\libs\rem.lib rem0rec.obj rem0cmp.obj
+
+rem0rec.obj: rem0rec.c
+	$(CCOM) $(CFL) -c rem0rec.c
+
+rem0cmp.obj: rem0cmp.c
+	$(CCOM) $(CFL) -c rem0cmp.c
+
+
diff --git a/storage/innobase/rem/rem0cmp.c b/storage/innobase/rem/rem0cmp.c
new file mode 100644
index 00000000000..74348b865a8
--- /dev/null
+++ b/storage/innobase/rem/rem0cmp.c
@@ -0,0 +1,1043 @@
+/***********************************************************************
+Comparison services for records
+
+(c) 1994-1996 Innobase Oy
+
+Created 7/1/1994 Heikki Tuuri
+************************************************************************/
+
+#include "rem0cmp.h"
+
+#ifdef UNIV_NONINL
+#include "rem0cmp.ic"
+#endif
+
+#include "srv0srv.h"
+
+/*		ALPHABETICAL ORDER
+		==================
+		
+The records are put into alphabetical order in the following
+way: let F be the first field where two records disagree.
+If there is a character in some position n where the the
+records disagree, the order is determined by comparison of
+the characters at position n, possibly after
+collating transformation. If there is no such character,
+but the corresponding fields have different lengths, then
+if the data type of the fields is paddable,
+shorter field is padded with a padding character. If the
+data type is not paddable, longer field is considered greater.
+Finally, the SQL null is bigger than any other value.
+
+At the present, the comparison functions return 0 in the case,
+where two records disagree only in the way that one 
+has more fields than the other. */
+
+#ifdef UNIV_DEBUG
+/*****************************************************************
+Used in debug checking of cmp_dtuple_... .
+This function is used to compare a data tuple to a physical record. If
+dtuple has n fields then rec must have either m >= n fields, or it must
+differ from dtuple in some of the m fields rec has. */
+static
+int
+cmp_debug_dtuple_rec_with_match(
+/*============================*/	
+				/* out: 1, 0, -1, if dtuple is greater, equal, 
+				less than rec, respectively, when only the 
+				common first fields are compared */
+	dtuple_t*	dtuple,	/* in: data tuple */
+	rec_t*		rec,	/* in: physical record which differs from
+				dtuple in some of the common fields, or which
+				has an equal number or more fields than
+				dtuple */
+	const ulint*	offsets,/* in: array returned by rec_get_offsets() */
+	ulint*	 	matched_fields);/* in/out: number of already
+				completely  matched fields; when function
+				returns, contains the value for current
+				comparison */
+#endif /* UNIV_DEBUG */
+/*****************************************************************
+This function is used to compare two data fields for which the data type
+is such that we must use MySQL code to compare them. The prototype here
+must be a copy of the the one in ha_innobase.cc! */
+extern
+int
+innobase_mysql_cmp(
+/*===============*/
+					/* out: 1, 0, -1, if a is greater,
+					equal, less than b, respectively */
+	int		mysql_type,	/* in: MySQL type */
+	uint		charset_number,	/* in: number of the charset */
+	unsigned char*	a,		/* in: data field */
+	unsigned int	a_length,	/* in: data field length,
+					not UNIV_SQL_NULL */
+	unsigned char*	b,		/* in: data field */
+	unsigned int	b_length);	/* in: data field length,
+					not UNIV_SQL_NULL */
+
+/*************************************************************************
+Transforms the character code so that it is ordered appropriately for the
+language. This is only used for the latin1 char set. MySQL does the
+comparisons for other char sets. */
+UNIV_INLINE
+ulint
+cmp_collate(
+/*========*/
+			/* out: collation order position */
+	ulint	code)	/* in: code of a character stored in database record */
+{	
+	return((ulint) srv_latin1_ordering[code]);
+}
+
+/*****************************************************************
+Returns TRUE if two types are equal for comparison purposes. */
+
+ibool
+cmp_types_are_equal(
+/*================*/
+				/* out: TRUE if the types are considered
+				equal in comparisons */
+	dtype_t*	type1,	/* in: type 1 */
+	dtype_t*	type2)	/* in: type 2 */
+{
+	if (dtype_is_non_binary_string_type(type1->mtype, type1->prtype)
+	    && dtype_is_non_binary_string_type(type2->mtype, type2->prtype)) {
+
+		/* Both are non-binary string types: they can be compared if
+		and only if the charset-collation is the same */
+
+		if (dtype_get_charset_coll(type1->prtype)
+				== dtype_get_charset_coll(type2->prtype)) {
+			return(TRUE);
+		}
+
+		return(FALSE);
+        }
+
+	if (dtype_is_binary_string_type(type1->mtype, type1->prtype)
+	    && dtype_is_binary_string_type(type2->mtype, type2->prtype)) {
+
+		/* Both are binary string types: they can be compared */
+
+		return(TRUE);
+	}
+	
+        if (type1->mtype != type2->mtype) {
+
+		return(FALSE);
+	}
+
+        if (type1->mtype == DATA_INT
+	    && (type1->prtype & DATA_UNSIGNED)
+		!= (type2->prtype & DATA_UNSIGNED)) {
+
+		/* The storage format of an unsigned integer is different
+		from a signed integer: in a signed integer we OR
+		0x8000... to the value of positive integers. */
+	
+		return(FALSE);
+	}
+
+        if (type1->mtype == DATA_INT && type1->len != type2->len) {
+	
+		return(FALSE);
+	}
+
+	return(TRUE);
+}
+
+/*****************************************************************
+Innobase uses this function to compare two data fields for which the data type
+is such that we must compare whole fields or call MySQL to do the comparison */
+static
+int
+cmp_whole_field(
+/*============*/	
+					/* out: 1, 0, -1, if a is greater,
+					equal, less than b, respectively */
+	dtype_t*	type,		/* in: data type */
+	unsigned char*	a,		/* in: data field */
+	unsigned int	a_length,	/* in: data field length,
+					not UNIV_SQL_NULL */
+	unsigned char*	b,		/* in: data field */
+	unsigned int	b_length)	/* in: data field length,
+					not UNIV_SQL_NULL */
+{
+	float		f_1;
+	float		f_2;
+	double		d_1;
+	double		d_2;
+	int		swap_flag	= 1;
+	ulint		data_type;
+
+	data_type = type->mtype;
+
+	switch (data_type) {
+
+    	case DATA_DECIMAL:
+		/* Remove preceding spaces */
+		for (; a_length && *a == ' '; a++, a_length--);
+		for (; b_length && *b == ' '; b++, b_length--);
+
+      		if (*a == '-') {
+			if (*b != '-') {
+	  			return(-1);
+	  		}
+
+			a++; b++;
+			a_length--;
+			b_length--;
+
+			swap_flag = -1;
+			
+      		} else if (*b == '-') {
+
+			return(1);
+		}
+		
+      		while (a_length > 0 && (*a == '+' || *a == '0')) {
+			a++; a_length--;
+      		}
+
+      		while (b_length > 0 && (*b == '+' || *b == '0')) {
+			b++; b_length--;
+      		}
+      		
+      		if (a_length != b_length) {
+			if (a_length < b_length) {
+				return(-swap_flag);
+			}
+
+			return(swap_flag);
+		}
+
+		while (a_length > 0 && *a == *b) {
+
+			a++; b++; a_length--;
+		}
+			
+		if (a_length == 0) {
+
+			return(0);
+		}
+
+		if (*a > *b) {
+			return(swap_flag);
+		}
+
+		return(-swap_flag);
+    	case DATA_DOUBLE:
+		d_1 = mach_double_read(a);
+		d_2 = mach_double_read(b);
+		
+      		if (d_1 > d_2) {
+      			return(1);
+      		} else if (d_2 > d_1) {
+      			return(-1);
+      		}
+
+      		return(0);
+	
+	case DATA_FLOAT:
+		f_1 = mach_float_read(a);
+		f_2 = mach_float_read(b);
+
+      		if (f_1 > f_2) {
+      			return(1);
+      		} else if (f_2 > f_1) {
+      			return(-1);
+      		}
+
+      		return(0);
+	case DATA_VARMYSQL:
+	case DATA_MYSQL:
+	case DATA_BLOB:
+		if (data_type == DATA_BLOB
+		    && 0 != (type->prtype & DATA_BINARY_TYPE)) {
+
+			ut_print_timestamp(stderr);
+		        fprintf(stderr,
+"  InnoDB: Error: comparing a binary BLOB with a character set sensitive\n"
+"InnoDB: comparison!\n");
+		}		   
+
+		return(innobase_mysql_cmp(
+				(int)(type->prtype & DATA_MYSQL_TYPE_MASK),
+				(uint)dtype_get_charset_coll(type->prtype),
+				a, a_length, b, b_length));
+	default:
+	        fprintf(stderr,
+			"InnoDB: unknown type number %lu\n",
+			(ulong) data_type);
+	        ut_error;
+	}
+
+	return(0);
+}
+
+/*****************************************************************
+This function is used to compare two data fields for which we know the
+data type. */
+
+int
+cmp_data_data_slow(
+/*===============*/	
+				/* out: 1, 0, -1, if data1 is greater, equal, 
+				less than data2, respectively */
+	dtype_t*	cur_type,/* in: data type of the fields */
+	byte*		data1,	/* in: data field (== a pointer to a memory
+				buffer) */
+	ulint		len1,	/* in: data field length or UNIV_SQL_NULL */
+	byte*		data2,	/* in: data field (== a pointer to a memory
+				buffer) */
+	ulint		len2)	/* in: data field length or UNIV_SQL_NULL */
+{
+	ulint	data1_byte;
+	ulint	data2_byte;
+	ulint	cur_bytes;
+
+	ut_ad(dtype_validate(cur_type));
+
+	if (len1 == UNIV_SQL_NULL || len2 == UNIV_SQL_NULL) {
+
+		if (len1 == len2) {
+
+			return(0);
+		}
+
+		if (len1 == UNIV_SQL_NULL) {
+			/* We define the SQL null to be the smallest possible
+			value of a field in the alphabetical order */
+
+			return(-1);
+		}
+
+		return(1);
+	}
+	
+	if (cur_type->mtype >= DATA_FLOAT
+	    || (cur_type->mtype == DATA_BLOB
+	        && 0 == (cur_type->prtype & DATA_BINARY_TYPE)
+		&& dtype_get_charset_coll(cur_type->prtype) !=
+				data_mysql_latin1_swedish_charset_coll)) {
+
+		return(cmp_whole_field(cur_type,
+					data1, (unsigned) len1,
+					data2, (unsigned) len2));
+	}
+	
+	/* Compare then the fields */
+
+	cur_bytes = 0;
+
+	for (;;) {
+		if (len1 <= cur_bytes) {
+			if (len2 <= cur_bytes) {
+
+				return(0);
+			}
+			
+			data1_byte = dtype_get_pad_char(cur_type);
+
+			if (data1_byte == ULINT_UNDEFINED) {
+
+				return(-1);
+			}
+		} else {
+			data1_byte = *data1;
+		}
+
+		if (len2 <= cur_bytes) {
+			data2_byte = dtype_get_pad_char(cur_type);
+
+			if (data2_byte == ULINT_UNDEFINED) {
+
+				return(1);
+			}
+		} else {
+			data2_byte = *data2;
+		}
+			
+		if (data1_byte == data2_byte) {
+			/* If the bytes are equal, they will remain such even
+			after the collation transformation below */
+
+			goto next_byte;
+		}
+
+		if (cur_type->mtype <= DATA_CHAR
+		    || (cur_type->mtype == DATA_BLOB
+		        && 0 == (cur_type->prtype & DATA_BINARY_TYPE))) {
+
+			data1_byte = cmp_collate(data1_byte);
+			data2_byte = cmp_collate(data2_byte);
+		}
+			
+		if (data1_byte > data2_byte) {
+
+				return(1);
+		} else if (data1_byte < data2_byte) {
+
+				return(-1);
+		}
+	next_byte:
+		/* Next byte */
+		cur_bytes++;
+		data1++;
+		data2++;
+	}
+
+	return(0);		/* Not reached */
+}
+
+/*****************************************************************
+This function is used to compare a data tuple to a physical record.
+Only dtuple->n_fields_cmp first fields are taken into account for
+the the data tuple! If we denote by n = n_fields_cmp, then rec must
+have either m >= n fields, or it must differ from dtuple in some of
+the m fields rec has. If rec has an externally stored field we do not
+compare it but return with value 0 if such a comparison should be
+made. */
+
+int
+cmp_dtuple_rec_with_match(
+/*======================*/	
+				/* out: 1, 0, -1, if dtuple is greater, equal, 
+				less than rec, respectively, when only the 
+				common first fields are compared, or
+				until the first externally stored field in
+				rec */
+	dtuple_t*	dtuple,	/* in: data tuple */
+	rec_t*		rec,	/* in: physical record which differs from
+				dtuple in some of the common fields, or which
+				has an equal number or more fields than
+				dtuple */
+	const ulint*	offsets,/* in: array returned by rec_get_offsets() */
+	ulint*	 	matched_fields, /* in/out: number of already completely 
+				matched fields; when function returns,
+				contains the value for current comparison */
+	ulint*	  	matched_bytes) /* in/out: number of already matched 
+				bytes within the first field not completely
+				matched; when function returns, contains the
+				value for current comparison */
+{
+	dtype_t*	cur_type;	/* pointer to type of the current
+					field in dtuple */
+	dfield_t*	dtuple_field;	/* current field in logical record */
+	ulint		dtuple_f_len;	/* the length of the current field 
+					in the logical record */
+	byte*		dtuple_b_ptr;	/* pointer to the current byte in 
+					logical field data */
+	ulint		dtuple_byte;	/* value of current byte to be compared 
+					in dtuple*/
+	ulint		rec_f_len;	/* length of current field in rec */
+	byte*		rec_b_ptr;	/* pointer to the current byte in 
+					rec field */
+	ulint		rec_byte;	/* value of current byte to be 
+					compared in rec */
+	ulint		cur_field;	/* current field number */
+	ulint		cur_bytes; 	/* number of already matched bytes 
+					in current field */
+	int		ret = 3333;	/* return value */
+
+	ut_ad(dtuple && rec && matched_fields && matched_bytes);
+	ut_ad(dtuple_check_typed(dtuple));
+	ut_ad(rec_offs_validate(rec, NULL, offsets));
+
+	cur_field = *matched_fields;
+	cur_bytes = *matched_bytes;
+
+	ut_ad(cur_field <= dtuple_get_n_fields_cmp(dtuple));
+	ut_ad(cur_field <= rec_offs_n_fields(offsets));
+
+	/* Match fields in a loop; stop if we run out of fields in dtuple
+	or find an externally stored field */
+
+	while (cur_field < dtuple_get_n_fields_cmp(dtuple)) {
+
+		dtuple_field = dtuple_get_nth_field(dtuple, cur_field);
+		cur_type = dfield_get_type(dtuple_field);
+
+		dtuple_f_len = dfield_get_len(dtuple_field);	
+
+		rec_b_ptr = rec_get_nth_field(rec, offsets,
+						cur_field, &rec_f_len);
+
+		/* If we have matched yet 0 bytes, it may be that one or
+		both the fields are SQL null, or the record or dtuple may be
+		the predefined minimum record, or the field is externally
+		stored */
+
+		if (cur_bytes == 0) {
+			if (cur_field == 0) {
+
+				if (rec_get_info_bits(rec,
+						rec_offs_comp(offsets))
+				    & REC_INFO_MIN_REC_FLAG) {
+
+					if (dtuple_get_info_bits(dtuple)
+				    	    & REC_INFO_MIN_REC_FLAG) {
+
+				    		ret = 0;
+				    	} else {
+				    		ret = 1;
+				    	}
+
+					goto order_resolved;
+				}
+
+				if (dtuple_get_info_bits(dtuple)
+				    & REC_INFO_MIN_REC_FLAG) {
+ 				    	ret = -1;
+
+					goto order_resolved;
+				}
+			}
+
+			if (rec_offs_nth_extern(offsets, cur_field)) {
+				/* We do not compare to an externally
+				stored field */
+
+				ret = 0;
+
+				goto order_resolved;
+			}
+
+		    	if (dtuple_f_len == UNIV_SQL_NULL
+		            || rec_f_len == UNIV_SQL_NULL) {
+
+				if (dtuple_f_len == rec_f_len) {
+
+					goto next_field;
+				}
+
+				if (rec_f_len == UNIV_SQL_NULL) {
+					/* We define the SQL null to be the 
+					smallest possible value of a field 
+					in the alphabetical order */
+
+					ret = 1;
+				} else {
+					ret = -1;
+				}
+
+				goto order_resolved;
+			}
+		}
+
+		if (cur_type->mtype >= DATA_FLOAT
+	    	    || (cur_type->mtype == DATA_BLOB
+	        	&& 0 == (cur_type->prtype & DATA_BINARY_TYPE)
+			&& dtype_get_charset_coll(cur_type->prtype) !=
+				data_mysql_latin1_swedish_charset_coll)) {
+
+			ret = cmp_whole_field(
+				cur_type,
+				dfield_get_data(dtuple_field),
+				(unsigned) dtuple_f_len,
+				rec_b_ptr, (unsigned) rec_f_len);
+
+			if (ret != 0) {
+				cur_bytes = 0;
+
+				goto order_resolved;
+			} else {
+				goto next_field;
+			}
+		}
+		
+		/* Set the pointers at the current byte */
+
+		rec_b_ptr = rec_b_ptr + cur_bytes;
+		dtuple_b_ptr = (byte*)dfield_get_data(dtuple_field) 
+								+ cur_bytes;
+		/* Compare then the fields */		
+		
+		for (;;) {
+			if (rec_f_len <= cur_bytes) {
+				if (dtuple_f_len <= cur_bytes) {
+
+					goto next_field;
+				}
+			
+				rec_byte = dtype_get_pad_char(cur_type);
+
+				if (rec_byte == ULINT_UNDEFINED) {
+					ret = 1;
+
+					goto order_resolved;
+				}
+			} else {
+				rec_byte = *rec_b_ptr;
+			}
+
+			if (dtuple_f_len <= cur_bytes) {
+				dtuple_byte = dtype_get_pad_char(cur_type);
+
+				if (dtuple_byte == ULINT_UNDEFINED) {
+					ret = -1;
+
+					goto order_resolved;
+				}
+			} else {
+				dtuple_byte = *dtuple_b_ptr;
+			}
+			
+			if (dtuple_byte == rec_byte) {
+				/* If the bytes are equal, they will
+				remain such even after the collation
+				transformation below */
+
+				goto next_byte;
+			}
+
+			if (cur_type->mtype <= DATA_CHAR
+			    || (cur_type->mtype == DATA_BLOB
+			        && 0 ==
+				    (cur_type->prtype & DATA_BINARY_TYPE))) {
+
+				rec_byte = cmp_collate(rec_byte);
+				dtuple_byte = cmp_collate(dtuple_byte);
+			}
+			
+			if (dtuple_byte > rec_byte) {
+				ret = 1;
+				goto order_resolved;
+
+			} else if (dtuple_byte < rec_byte) {
+				ret = -1;
+				goto order_resolved;
+			}
+		next_byte:
+			/* Next byte */
+			cur_bytes++;
+			rec_b_ptr++;
+			dtuple_b_ptr++;
+		}
+
+	next_field:
+		cur_field++;
+		cur_bytes = 0;
+	}
+
+	ut_ad(cur_bytes == 0);
+
+	ret = 0;	/* If we ran out of fields, dtuple was equal to rec
+			up to the common fields */
+order_resolved:
+	ut_ad((ret >= - 1) && (ret <= 1));
+	ut_ad(ret == cmp_debug_dtuple_rec_with_match(dtuple, rec, offsets,
+							matched_fields));
+	ut_ad(*matched_fields == cur_field); /* In the debug version, the
+						above cmp_debug_... sets
+						*matched_fields to a value */
+	*matched_fields = cur_field;
+	*matched_bytes = cur_bytes;
+
+	return(ret);
+}
+
+/******************************************************************
+Compares a data tuple to a physical record. */
+
+int
+cmp_dtuple_rec(
+/*===========*/
+				/* out: 1, 0, -1, if dtuple is greater, equal, 
+				less than rec, respectively; see the comments
+				for cmp_dtuple_rec_with_match */
+	dtuple_t* 	dtuple,	/* in: data tuple */
+	rec_t*		rec,	/* in: physical record */
+	const ulint*	offsets)/* in: array returned by rec_get_offsets() */
+{
+	ulint	matched_fields	= 0;
+	ulint	matched_bytes	= 0;
+
+	ut_ad(rec_offs_validate(rec, NULL, offsets));
+	return(cmp_dtuple_rec_with_match(dtuple, rec, offsets,
+					&matched_fields, &matched_bytes));
+}
+
+/******************************************************************
+Checks if a dtuple is a prefix of a record. The last field in dtuple
+is allowed to be a prefix of the corresponding field in the record. */
+
+ibool
+cmp_dtuple_is_prefix_of_rec(
+/*========================*/
+				/* out: TRUE if prefix */
+	dtuple_t*	dtuple,	/* in: data tuple */
+	rec_t*		rec,	/* in: physical record */
+	const ulint*	offsets)/* in: array returned by rec_get_offsets() */
+{
+	ulint	n_fields;
+	ulint	matched_fields	= 0;
+	ulint	matched_bytes	= 0;
+
+	ut_ad(rec_offs_validate(rec, NULL, offsets));
+	n_fields = dtuple_get_n_fields(dtuple);
+	
+	if (n_fields > rec_offs_n_fields(offsets)) {
+
+		return(FALSE);
+	}
+	
+	cmp_dtuple_rec_with_match(dtuple, rec, offsets,
+					&matched_fields, &matched_bytes);
+	if (matched_fields == n_fields) {
+
+		return(TRUE);
+	}
+
+	if (matched_fields == n_fields - 1
+			&& matched_bytes == dfield_get_len(
+				dtuple_get_nth_field(dtuple, n_fields - 1))) {
+		return(TRUE);
+	}
+
+	return(FALSE);	
+}
+
+/*****************************************************************
+This function is used to compare two physical records. Only the common
+first fields are compared, and if an externally stored field is
+encountered, then 0 is returned. */
+
+int
+cmp_rec_rec_with_match(
+/*===================*/	
+				/* out: 1, 0 , -1 if rec1 is greater, equal,
+				less, respectively, than rec2; only the common
+				first fields are compared */
+	rec_t*		rec1,	/* in: physical record */
+	rec_t*		rec2,	/* in: physical record */
+	const ulint*	offsets1,/* in: rec_get_offsets(rec1, index) */
+	const ulint*	offsets2,/* in: rec_get_offsets(rec2, index) */
+	dict_index_t*	index,	/* in: data dictionary index */
+	ulint*	 	matched_fields, /* in/out: number of already completely 
+				matched fields; when the function returns,
+				contains the value the for current
+				comparison */
+	ulint*	  	matched_bytes) /* in/out: number of already matched 
+				bytes within the first field not completely
+				matched; when the function returns, contains
+				the value for the current comparison */
+{
+	dtype_t* cur_type;	/* pointer to type struct of the
+				current field in index */
+	ulint	rec1_n_fields;	/* the number of fields in rec */
+	ulint	rec1_f_len;	/* length of current field in rec */
+	byte*	rec1_b_ptr;	/* pointer to the current byte in rec field */
+	ulint	rec1_byte;	/* value of current byte to be compared in
+				rec */
+	ulint	rec2_n_fields;	/* the number of fields in rec */
+	ulint	rec2_f_len;	/* length of current field in rec */
+	byte*	rec2_b_ptr;	/* pointer to the current byte in rec field */
+	ulint	rec2_byte;	/* value of current byte to be compared in
+				rec */
+	ulint	cur_field;	/* current field number */
+	ulint	cur_bytes; 	/* number of already matched bytes in current
+				field */
+	int	ret = 3333;	/* return value */
+	ibool	comp;
+
+	ut_ad(rec1 && rec2 && index);
+	ut_ad(rec_offs_validate(rec1, index, offsets1));
+	ut_ad(rec_offs_validate(rec2, index, offsets2));
+	ut_ad(rec_offs_comp(offsets1) == rec_offs_comp(offsets2));
+
+	comp = rec_offs_comp(offsets1);
+	rec1_n_fields = rec_offs_n_fields(offsets1);
+	rec2_n_fields = rec_offs_n_fields(offsets2);
+
+	cur_field = *matched_fields;
+	cur_bytes = *matched_bytes;
+
+	/* Match fields in a loop */
+
+	while ((cur_field < rec1_n_fields) && (cur_field < rec2_n_fields)) {
+
+		if (index->type & DICT_UNIVERSAL) {
+			cur_type = dtype_binary;
+		} else {
+			cur_type = dict_col_get_type(
+			      dict_field_get_col(
+				dict_index_get_nth_field(index, cur_field)));
+		}
+
+		rec1_b_ptr = rec_get_nth_field(rec1, offsets1,
+						cur_field, &rec1_f_len);
+		rec2_b_ptr = rec_get_nth_field(rec2, offsets2,
+						cur_field, &rec2_f_len);
+
+		if (cur_bytes == 0) {
+			if (cur_field == 0) {
+				/* Test if rec is the predefined minimum
+				record */
+				if (rec_get_info_bits(rec1, comp)
+						& REC_INFO_MIN_REC_FLAG) {
+	
+					if (rec_get_info_bits(rec2, comp)
+					    & REC_INFO_MIN_REC_FLAG) {
+						ret = 0;
+					} else {
+						ret = -1;
+					}
+	
+					goto order_resolved;
+	
+				} else if (rec_get_info_bits(rec2, comp)
+					   & REC_INFO_MIN_REC_FLAG) {
+	
+					ret = 1;
+	
+					goto order_resolved;
+				}
+			}
+
+			if (rec_offs_nth_extern(offsets1, cur_field)
+			    || rec_offs_nth_extern(offsets2, cur_field)) {
+				/* We do not compare to an externally
+				stored field */
+
+				ret = 0;
+
+				goto order_resolved;
+			}
+			
+		    	if (rec1_f_len == UNIV_SQL_NULL
+		            || rec2_f_len == UNIV_SQL_NULL) {
+
+				if (rec1_f_len == rec2_f_len) {
+
+					goto next_field;
+
+				} else if (rec2_f_len == UNIV_SQL_NULL) {
+
+					/* We define the SQL null to be the 
+					smallest possible value of a field 
+					in the alphabetical order */
+
+					ret = 1;
+				} else {
+					ret = -1;
+				}
+
+				goto order_resolved;
+			}
+		}
+
+		if (cur_type->mtype >= DATA_FLOAT
+	    	    || (cur_type->mtype == DATA_BLOB
+	        	&& 0 == (cur_type->prtype & DATA_BINARY_TYPE)
+			&& dtype_get_charset_coll(cur_type->prtype) !=
+				data_mysql_latin1_swedish_charset_coll)) {
+
+			ret = cmp_whole_field(cur_type,
+					rec1_b_ptr, (unsigned) rec1_f_len,
+					rec2_b_ptr, (unsigned) rec2_f_len);
+			if (ret != 0) {
+				cur_bytes = 0;
+
+				goto order_resolved;
+			} else {
+				goto next_field;
+			}
+		}
+
+		/* Set the pointers at the current byte */
+		rec1_b_ptr = rec1_b_ptr + cur_bytes;
+		rec2_b_ptr = rec2_b_ptr + cur_bytes;
+
+		/* Compare then the fields */
+		for (;;) {
+			if (rec2_f_len <= cur_bytes) {
+
+				if (rec1_f_len <= cur_bytes) {
+
+					goto next_field;
+				}
+
+				rec2_byte = dtype_get_pad_char(cur_type);
+
+				if (rec2_byte == ULINT_UNDEFINED) {
+					ret = 1;
+
+					goto order_resolved;
+				}
+			} else {
+				rec2_byte = *rec2_b_ptr;
+			}
+
+			if (rec1_f_len <= cur_bytes) {
+				rec1_byte = dtype_get_pad_char(cur_type);
+
+				if (rec1_byte == ULINT_UNDEFINED) {
+					ret = -1;
+
+					goto order_resolved;
+				}
+			} else {
+				rec1_byte = *rec1_b_ptr;
+			}
+			
+			if (rec1_byte == rec2_byte) {
+				/* If the bytes are equal, they will remain
+				such even after the collation transformation 
+				below */
+
+				goto next_byte;
+			}
+
+			if (cur_type->mtype <= DATA_CHAR
+			    || (cur_type->mtype == DATA_BLOB
+			        && 0 ==
+				    (cur_type->prtype & DATA_BINARY_TYPE))) {
+
+				rec1_byte = cmp_collate(rec1_byte);
+				rec2_byte = cmp_collate(rec2_byte);
+			}
+
+			if (rec1_byte < rec2_byte) {
+				ret = -1;
+				goto order_resolved;
+			} else if (rec1_byte > rec2_byte) {
+				ret = 1;
+				goto order_resolved;
+			}
+		next_byte:
+			/* Next byte */
+
+			cur_bytes++;
+			rec1_b_ptr++;
+			rec2_b_ptr++;
+		}
+
+	next_field:
+		cur_field++;
+		cur_bytes = 0;
+	}
+
+	ut_ad(cur_bytes == 0);
+
+	ret = 0;	/* If we ran out of fields, rec1 was equal to rec2 up
+			to the common fields */
+order_resolved:
+
+	ut_ad((ret >= - 1) && (ret <= 1));
+
+	*matched_fields = cur_field;	
+	*matched_bytes = cur_bytes;
+
+	return(ret);
+}
+
+#ifdef UNIV_DEBUG
+/*****************************************************************
+Used in debug checking of cmp_dtuple_... .
+This function is used to compare a data tuple to a physical record. If
+dtuple has n fields then rec must have either m >= n fields, or it must
+differ from dtuple in some of the m fields rec has. If encounters an
+externally stored field, returns 0. */
+static
+int
+cmp_debug_dtuple_rec_with_match(
+/*============================*/	
+				/* out: 1, 0, -1, if dtuple is greater, equal, 
+				less than rec, respectively, when only the 
+				common first fields are compared */
+	dtuple_t*	dtuple,	/* in: data tuple */
+	rec_t*		rec,	/* in: physical record which differs from
+				dtuple in some of the common fields, or which
+				has an equal number or more fields than
+				dtuple */
+	const ulint*	offsets,/* in: array returned by rec_get_offsets() */
+	ulint*	 	matched_fields) /* in/out: number of already
+				completely matched fields; when function
+				returns, contains the value for current
+				comparison */
+{
+	dtype_t*	cur_type;	/* pointer to type of the current
+					field in dtuple */
+	dfield_t*	dtuple_field;	/* current field in logical record */
+	ulint		dtuple_f_len;	/* the length of the current field 
+					in the logical record */
+	byte*		dtuple_f_data;	/* pointer to the current logical
+					field data */
+	ulint		rec_f_len;	/* length of current field in rec */
+	byte*		rec_f_data;	/* pointer to the current rec field */
+	int		ret = 3333;	/* return value */
+	ulint		cur_field;	/* current field number */
+	
+	ut_ad(dtuple && rec && matched_fields);
+	ut_ad(dtuple_check_typed(dtuple));
+	ut_ad(rec_offs_validate(rec, NULL, offsets));
+	
+	ut_ad(*matched_fields <= dtuple_get_n_fields_cmp(dtuple));
+	ut_ad(*matched_fields <= rec_offs_n_fields(offsets));
+
+	cur_field = *matched_fields;
+
+	if (cur_field == 0) {
+		if (rec_get_info_bits(rec, rec_offs_comp(offsets))
+					& REC_INFO_MIN_REC_FLAG) {
+
+			if (dtuple_get_info_bits(dtuple)
+				    	    & REC_INFO_MIN_REC_FLAG) {
+				ret = 0;
+			} else {
+				ret = 1;
+			}
+
+			goto order_resolved;
+		}
+
+		if (dtuple_get_info_bits(dtuple) & REC_INFO_MIN_REC_FLAG) {
+ 			ret = -1;
+
+			goto order_resolved;
+		}
+	}
+
+	/* Match fields in a loop; stop if we run out of fields in dtuple */
+
+	while (cur_field < dtuple_get_n_fields_cmp(dtuple)) {
+
+		dtuple_field = dtuple_get_nth_field(dtuple, cur_field);
+
+		cur_type = dfield_get_type(dtuple_field);
+
+		dtuple_f_data = dfield_get_data(dtuple_field);
+		dtuple_f_len = dfield_get_len(dtuple_field);
+		
+		rec_f_data = rec_get_nth_field(rec, offsets,
+						cur_field, &rec_f_len);
+
+		if (rec_offs_nth_extern(offsets, cur_field)) {
+			/* We do not compare to an externally stored field */
+
+			ret = 0;
+
+			goto order_resolved;
+		}
+
+		ret = cmp_data_data(cur_type, dtuple_f_data, dtuple_f_len,
+							rec_f_data, rec_f_len);
+		if (ret != 0) {
+			goto order_resolved;
+		} 
+
+		cur_field++;
+	}
+
+	ret = 0;	/* If we ran out of fields, dtuple was equal to rec
+			up to the common fields */
+order_resolved:
+	ut_ad((ret >= - 1) && (ret <= 1));
+
+	*matched_fields = cur_field;
+
+	return(ret);
+}
+#endif /* UNIV_DEBUG */
diff --git a/storage/innobase/rem/rem0rec.c b/storage/innobase/rem/rem0rec.c
new file mode 100644
index 00000000000..542c746209b
--- /dev/null
+++ b/storage/innobase/rem/rem0rec.c
@@ -0,0 +1,1419 @@
+/************************************************************************
+Record manager
+
+(c) 1994-2001 Innobase Oy
+
+Created 5/30/1994 Heikki Tuuri
+*************************************************************************/
+
+#include "rem0rec.h"
+
+#ifdef UNIV_NONINL
+#include "rem0rec.ic"
+#endif
+
+#include "mtr0mtr.h"
+#include "mtr0log.h"
+
+/*			PHYSICAL RECORD (OLD STYLE)
+			===========================
+
+The physical record, which is the data type of all the records
+found in index pages of the database, has the following format
+(lower addresses and more significant bits inside a byte are below
+represented on a higher text line):
+
+| offset of the end of the last field of data, the most significant
+  bit is set to 1 if and only if the field is SQL-null,
+  if the offset is 2-byte, then the second most significant
+  bit is set to 1 if the field is stored on another page:
+  mostly this will occur in the case of big BLOB fields |
+... 
+| offset of the end of the first field of data + the SQL-null bit |
+| 4 bits used to delete mark a record, and mark a predefined
+  minimum record in alphabetical order |
+| 4 bits giving the number of records owned by this record
+  (this term is explained in page0page.h) |
+| 13 bits giving the order number of this record in the
+  heap of the index page |
+| 10 bits giving the number of fields in this record |
+| 1 bit which is set to 1 if the offsets above are given in
+  one byte format, 0 if in two byte format |
+| two bytes giving an absolute pointer to the next record in the page |
+ORIGIN of the record
+| first field of data | 
+... 
+| last field of data |
+
+The origin of the record is the start address of the first field 
+of data. The offsets are given relative to the origin. 
+The offsets of the data fields are stored in an inverted
+order because then the offset of the first fields are near the 
+origin, giving maybe a better processor cache hit rate in searches.
+
+The offsets of the data fields are given as one-byte 
+(if there are less than 127 bytes of data in the record) 
+or two-byte unsigned integers. The most significant bit
+is not part of the offset, instead it indicates the SQL-null
+if the bit is set to 1. */
+
+/*			PHYSICAL RECORD (NEW STYLE)
+			===========================
+
+The physical record, which is the data type of all the records
+found in index pages of the database, has the following format
+(lower addresses and more significant bits inside a byte are below
+represented on a higher text line):
+
+| length of the last non-null variable-length field of data:
+  if the maximum length is 255, one byte; otherwise,
+  0xxxxxxx (one byte, length=0..127), or 1exxxxxxxxxxxxxx (two bytes,
+  length=128..16383, extern storage flag) |
+...
+| length of first variable-length field of data |
+| SQL-null flags (1 bit per nullable field), padded to full bytes |
+| 4 bits used to delete mark a record, and mark a predefined
+  minimum record in alphabetical order |
+| 4 bits giving the number of records owned by this record
+  (this term is explained in page0page.h) |
+| 13 bits giving the order number of this record in the
+  heap of the index page |
+| 3 bits record type: 000=conventional, 001=node pointer (inside B-tree),
+  010=infimum, 011=supremum, 1xx=reserved |
+| two bytes giving a relative pointer to the next record in the page |
+ORIGIN of the record
+| first field of data |
+...
+| last field of data |
+
+The origin of the record is the start address of the first field
+of data. The offsets are given relative to the origin.
+The offsets of the data fields are stored in an inverted
+order because then the offset of the first fields are near the
+origin, giving maybe a better processor cache hit rate in searches.
+
+The offsets of the data fields are given as one-byte
+(if there are less than 127 bytes of data in the record)
+or two-byte unsigned integers. The most significant bit
+is not part of the offset, instead it indicates the SQL-null
+if the bit is set to 1. */
+
+/* CANONICAL COORDINATES. A record can be seen as a single
+string of 'characters' in the following way: catenate the bytes
+in each field, in the order of fields. An SQL-null field
+is taken to be an empty sequence of bytes. Then after
+the position of each field insert in the string 
+the 'character' <FIELD-END>, except that after an SQL-null field
+insert <NULL-FIELD-END>. Now the ordinal position of each
+byte in this canonical string is its canonical coordinate.
+So, for the record ("AA", SQL-NULL, "BB", ""), the canonical
+string is "AA<FIELD_END><NULL-FIELD-END>BB<FIELD-END><FIELD-END>".
+We identify prefixes (= initial segments) of a record
+with prefixes of the canonical string. The canonical
+length of the prefix is the length of the corresponding
+prefix of the canonical string. The canonical length of
+a record is the length of its canonical string.
+
+For example, the maximal common prefix of records
+("AA", SQL-NULL, "BB", "C") and ("AA", SQL-NULL, "B", "C")
+is "AA<FIELD-END><NULL-FIELD-END>B", and its canonical
+length is 5.
+
+A complete-field prefix of a record is a prefix which ends at the
+end of some field (containing also <FIELD-END>).
+A record is a complete-field prefix of another record, if
+the corresponding canonical strings have the same property. */
+
+ulint	rec_dummy;	/* this is used to fool compiler in
+			rec_validate */
+
+/*******************************************************************
+Validates the consistency of an old-style physical record. */
+static
+ibool
+rec_validate_old(
+/*=============*/
+			/* out: TRUE if ok */
+	rec_t*	rec);	/* in: physical record */
+
+/**********************************************************
+The following function determines the offsets to each field in the
+record.  The offsets are written to a previously allocated array of
+ulint, where rec_offs_n_fields(offsets) has been initialized to the
+number of fields in the record.  The rest of the array will be
+initialized by this function.  rec_offs_base(offsets)[0] will be set
+to the extra size (if REC_OFFS_COMPACT is set, the record is in the
+new format), and rec_offs_base(offsets)[1..n_fields] will be set to
+offsets past the end of fields 0..n_fields, or to the beginning of
+fields 1..n_fields+1.  When the high-order bit of the offset at [i+1]
+is set (REC_OFFS_SQL_NULL), the field i is NULL.  When the second
+high-order bit of the offset at [i+1] is set (REC_OFFS_EXTERNAL), the
+field i is being stored externally. */
+static
+void
+rec_init_offsets(
+/*=============*/
+				/* out: the offsets */
+	rec_t*		rec,	/* in: physical record */
+	dict_index_t*	index,	/* in: record descriptor */
+	ulint*		offsets)/* in/out: array of offsets;
+				in: n=rec_offs_n_fields(offsets) */
+{
+	ulint	n_fields = rec_offs_n_fields(offsets);
+	ulint	i	= 0;
+	ulint	offs;
+
+	rec_offs_make_valid(rec, index, offsets);
+
+	if (index->table->comp) {
+		const byte*	nulls;
+		const byte*	lens;
+		dict_field_t*	field;
+		dtype_t*	type;
+		ulint		null_mask;
+		ulint		status = rec_get_status(rec);
+		ulint		n_node_ptr_field = ULINT_UNDEFINED;
+
+		switch (status) {
+		case REC_STATUS_INFIMUM:
+		case REC_STATUS_SUPREMUM:
+			/* the field is 8 bytes long */
+			rec_offs_base(offsets)[0] =
+				REC_N_NEW_EXTRA_BYTES | REC_OFFS_COMPACT;
+			rec_offs_base(offsets)[1] = 8;
+			return;
+		case REC_STATUS_NODE_PTR:
+			n_node_ptr_field =
+				dict_index_get_n_unique_in_tree(index);
+			break;
+		case REC_STATUS_ORDINARY:
+			break;
+		}
+
+		nulls = rec - (REC_N_NEW_EXTRA_BYTES + 1);
+		lens = nulls - (index->n_nullable + 7) / 8;
+		offs = 0;
+		null_mask = 1;
+
+		/* read the lengths of fields 0..n */
+		for (; i < n_fields; i++) {
+			ibool	is_null = FALSE, is_external = FALSE;
+			ulint	len;
+			if (i == n_node_ptr_field) {
+				len = 4;
+				goto resolved;
+			}
+
+			field = dict_index_get_nth_field(index, i);
+			type = dict_col_get_type(dict_field_get_col(field));
+			if (!(dtype_get_prtype(type) & DATA_NOT_NULL)) {
+				/* nullable field => read the null flag */
+				is_null = (*nulls & null_mask) != 0;
+				null_mask <<= 1;
+				if (null_mask == 0x100) {
+					nulls--;
+					null_mask = 1;
+				}
+			}
+
+			if (is_null) {
+				/* No length is stored for NULL fields. */
+				len = 0;
+			} else if (!field->fixed_len) {
+				/* Variable-length field: read the length */
+				len = *lens--;
+				if (dtype_get_len(type) > 255
+				    || dtype_get_mtype(type) == DATA_BLOB) {
+					if (len & 0x80) {
+						/* 1exxxxxxx xxxxxxxx */
+						is_external = !!(len & 0x40);
+						len &= 0x3f;
+						len <<= 8;
+						len |= *lens--;
+					}
+				}
+			} else {
+				len = field->fixed_len;
+			}
+		resolved:
+			offs += len;
+			len = offs;
+			if (is_external) {
+				len |= REC_OFFS_EXTERNAL;
+			}
+			if (is_null) {
+				len |= REC_OFFS_SQL_NULL;
+			}
+			rec_offs_base(offsets)[i + 1] = len;
+		}
+
+		*rec_offs_base(offsets) =
+			(rec - (lens + 1)) | REC_OFFS_COMPACT;
+	} else {
+		/* Old-style record: determine extra size and end offsets */
+		offs = REC_N_OLD_EXTRA_BYTES;
+		if (rec_get_1byte_offs_flag(rec)) {
+			offs += n_fields;
+			*rec_offs_base(offsets) = offs;
+			/* Determine offsets to fields */
+			for (; i < n_fields; i++) {
+				offs = rec_1_get_field_end_info(rec, i);
+				if (offs & REC_1BYTE_SQL_NULL_MASK) {
+					offs &= ~REC_1BYTE_SQL_NULL_MASK;
+					offs |= REC_OFFS_SQL_NULL;
+				}
+				rec_offs_base(offsets)[1 + i] = offs;
+			}
+		} else {
+			offs += 2 * n_fields;
+			*rec_offs_base(offsets) = offs;
+			/* Determine offsets to fields */
+			for (; i < n_fields; i++) {
+				offs = rec_2_get_field_end_info(rec, i);
+				if (offs & REC_2BYTE_SQL_NULL_MASK) {
+					offs &= ~REC_2BYTE_SQL_NULL_MASK;
+					offs |= REC_OFFS_SQL_NULL;
+				}
+				if (offs & REC_2BYTE_EXTERN_MASK) {
+					offs &= ~REC_2BYTE_EXTERN_MASK;
+					offs |= REC_OFFS_EXTERNAL;
+				}
+				rec_offs_base(offsets)[1 + i] = offs;
+			}
+		}
+	}
+}
+
+/**********************************************************
+The following function determines the offsets to each field
+in the record.  It can reuse a previously returned array. */
+
+ulint*
+rec_get_offsets_func(
+/*=================*/
+				/* out: the new offsets */
+	rec_t*		rec,	/* in: physical record */
+	dict_index_t*	index,	/* in: record descriptor */
+	ulint*		offsets,/* in: array consisting of offsets[0]
+				allocated elements, or an array from
+				rec_get_offsets(), or NULL */
+	ulint		n_fields,/* in: maximum number of initialized fields
+				(ULINT_UNDEFINED if all fields) */
+	mem_heap_t**	heap,	/* in/out: memory heap */
+	const char*	file,	/* in: file name where called */
+	ulint		line)	/* in: line number where called */
+{
+	ulint	n;
+	ulint	size;
+
+	ut_ad(rec);
+	ut_ad(index);
+	ut_ad(heap);
+
+	if (index->table->comp) {
+		switch (rec_get_status(rec)) {
+		case REC_STATUS_ORDINARY:
+			n = dict_index_get_n_fields(index);
+			break;
+		case REC_STATUS_NODE_PTR:
+			n = dict_index_get_n_unique_in_tree(index) + 1;
+			break;
+		case REC_STATUS_INFIMUM:
+		case REC_STATUS_SUPREMUM:
+			/* infimum or supremum record */
+			n = 1;
+			break;
+		default:
+			ut_error;
+			return(NULL);
+		}
+	} else {
+		n = rec_get_n_fields_old(rec);
+	}
+
+	if (n_fields < n) {
+		n = n_fields;
+	}
+
+	size = n + (1 + REC_OFFS_HEADER_SIZE);
+
+	if (!offsets || rec_offs_get_n_alloc(offsets) < size) {
+		if (!*heap) {
+			*heap = mem_heap_create_func(size * sizeof(ulint),
+				NULL, MEM_HEAP_DYNAMIC, file, line);
+		}
+		offsets = mem_heap_alloc(*heap, size * sizeof(ulint));
+		rec_offs_set_n_alloc(offsets, size);
+	}
+
+	rec_offs_set_n_fields(offsets, n);
+	rec_init_offsets(rec, index, offsets);
+	return(offsets);
+}
+
+/****************************************************************
+The following function is used to get a pointer to the nth
+data field in an old-style record. */
+
+byte*
+rec_get_nth_field_old(
+/*==================*/
+ 			/* out: pointer to the field */
+ 	rec_t*	rec, 	/* in: record */
+ 	ulint	n,	/* in: index of the field */
+	ulint*	len)	/* out: length of the field; UNIV_SQL_NULL if SQL
+			null */
+{
+	ulint	os;
+	ulint	next_os;
+
+	ut_ad(rec && len);
+	ut_ad(n < rec_get_n_fields_old(rec));
+
+	if (n > REC_MAX_N_FIELDS) {
+		fprintf(stderr, "Error: trying to access field %lu in rec\n",
+								(ulong) n);
+		ut_error;
+	}
+
+	if (rec == NULL) {
+		fputs("Error: rec is NULL pointer\n", stderr);
+		ut_error;
+	}
+	
+	if (rec_get_1byte_offs_flag(rec)) {
+		os = rec_1_get_field_start_offs(rec, n);
+
+		next_os = rec_1_get_field_end_info(rec, n);
+
+		if (next_os & REC_1BYTE_SQL_NULL_MASK) {
+			*len = UNIV_SQL_NULL;
+
+			return(rec + os);
+		}
+
+		next_os = next_os & ~REC_1BYTE_SQL_NULL_MASK;
+	} else {
+		os = rec_2_get_field_start_offs(rec, n);
+	
+		next_os = rec_2_get_field_end_info(rec, n);
+
+		if (next_os & REC_2BYTE_SQL_NULL_MASK) {
+			*len = UNIV_SQL_NULL;
+
+			return(rec + os);
+		}
+
+		next_os = next_os & ~(REC_2BYTE_SQL_NULL_MASK
+						| REC_2BYTE_EXTERN_MASK);
+	}
+	
+	*len = next_os - os;
+
+	ut_ad(*len < UNIV_PAGE_SIZE);
+
+	return(rec + os);
+}
+
+/**************************************************************
+The following function returns the size of a data tuple when converted to
+a new-style physical record. */
+
+ulint
+rec_get_converted_size_new(
+/*=======================*/
+				/* out: size */
+	dict_index_t*	index,	/* in: record descriptor */
+	dtuple_t*	dtuple)	/* in: data tuple */
+{
+	ulint		size		= REC_N_NEW_EXTRA_BYTES
+					+ (index->n_nullable + 7) / 8;
+	dict_field_t*	field;
+	dtype_t*	type;
+	ulint		i;
+	ulint		n_fields;
+	ut_ad(index && dtuple);
+	ut_ad(index->table->comp);
+
+	switch (dtuple_get_info_bits(dtuple) & REC_NEW_STATUS_MASK) {
+	case REC_STATUS_ORDINARY:
+		n_fields = dict_index_get_n_fields(index);
+		ut_ad(n_fields == dtuple_get_n_fields(dtuple));
+		break;
+	case REC_STATUS_NODE_PTR:
+		n_fields = dict_index_get_n_unique_in_tree(index);
+		ut_ad(n_fields + 1 == dtuple_get_n_fields(dtuple));
+		ut_ad(dtuple_get_nth_field(dtuple, n_fields)->len == 4);
+		size += 4; /* child page number */
+		break;
+	case REC_STATUS_INFIMUM:
+	case REC_STATUS_SUPREMUM:
+		/* infimum or supremum record, 8 bytes */
+		return(size + 8); /* no extra data needed */
+	default:
+		ut_a(0);
+		return(ULINT_UNDEFINED);
+	}
+
+	/* read the lengths of fields 0..n */
+	for (i = 0; i < n_fields; i++) {
+		ulint	len	= dtuple_get_nth_field(dtuple, i)->len;
+		field = dict_index_get_nth_field(index, i);
+		type = dict_col_get_type(dict_field_get_col(field));
+		ut_ad(len != UNIV_SQL_NULL ||
+			!(dtype_get_prtype(type) & DATA_NOT_NULL));
+
+		if (len == UNIV_SQL_NULL) {
+			/* No length is stored for NULL fields. */
+			continue;
+		}
+
+		ut_ad(len <= dtype_get_len(type)
+			|| dtype_get_mtype(type) == DATA_BLOB);
+		ut_ad(!field->fixed_len || len == field->fixed_len);
+
+		if (field->fixed_len) {
+		} else if (len < 128 || (dtype_get_len(type) < 256
+				&& dtype_get_mtype(type) != DATA_BLOB)) {
+			size++;
+		} else {
+			size += 2;
+		}
+		size += len;
+	}
+
+	return(size);
+}
+
+/***************************************************************
+Sets the value of the ith field SQL null bit of an old-style record. */
+
+void
+rec_set_nth_field_null_bit(
+/*=======================*/
+	rec_t*	rec,	/* in: record */
+	ulint	i,	/* in: ith field */
+	ibool	val)	/* in: value to set */
+{
+	ulint	info;
+
+	if (rec_get_1byte_offs_flag(rec)) {
+
+		info = rec_1_get_field_end_info(rec, i);
+
+		if (val) {
+			info = info | REC_1BYTE_SQL_NULL_MASK;
+		} else {
+			info = info & ~REC_1BYTE_SQL_NULL_MASK;
+		}
+
+		rec_1_set_field_end_info(rec, i, info);
+
+		return;
+	}
+
+	info = rec_2_get_field_end_info(rec, i);
+
+	if (val) {
+		info = info | REC_2BYTE_SQL_NULL_MASK;
+	} else {
+		info = info & ~REC_2BYTE_SQL_NULL_MASK;
+	}
+
+	rec_2_set_field_end_info(rec, i, info);
+}
+
+/***************************************************************
+Sets the value of the ith field extern storage bit of an old-style record. */
+
+void
+rec_set_nth_field_extern_bit_old(
+/*=============================*/
+	rec_t*	rec,	/* in: old-style record */
+	ulint	i,	/* in: ith field */
+	ibool	val,	/* in: value to set */
+	mtr_t*	mtr)	/* in: mtr holding an X-latch to the page where
+			rec is, or NULL; in the NULL case we do not
+			write to log about the change */
+{
+	ulint	info;
+
+	ut_a(!rec_get_1byte_offs_flag(rec));
+	ut_a(i < rec_get_n_fields_old(rec));
+	
+	info = rec_2_get_field_end_info(rec, i);
+
+	if (val) {
+		info = info | REC_2BYTE_EXTERN_MASK;
+	} else {
+		info = info & ~REC_2BYTE_EXTERN_MASK;
+	}
+
+	if (mtr) {
+		mlog_write_ulint(rec - REC_N_OLD_EXTRA_BYTES - 2 * (i + 1),
+						info, MLOG_2BYTES, mtr);
+	} else {
+		rec_2_set_field_end_info(rec, i, info);
+	}
+}
+
+/***************************************************************
+Sets the value of the ith field extern storage bit of a new-style record. */
+
+void
+rec_set_nth_field_extern_bit_new(
+/*=============================*/
+	rec_t*		rec,	/* in: record */
+	dict_index_t*	index,	/* in: record descriptor */
+	ulint		ith,	/* in: ith field */
+	ibool		val,	/* in: value to set */
+	mtr_t*		mtr)	/* in: mtr holding an X-latch to the page
+				where rec is, or NULL; in the NULL case
+				we do not write to log about the change */
+{
+	byte*		nulls	= rec - (REC_N_NEW_EXTRA_BYTES + 1);
+	byte*		lens	= nulls - (index->n_nullable + 7) / 8;
+	dict_field_t*	field;
+	dtype_t*	type;
+	ulint		i;
+	ulint		n_fields;
+	ulint		null_mask	= 1;
+	ut_ad(rec && index);
+	ut_ad(index->table->comp);
+	ut_ad(rec_get_status(rec) == REC_STATUS_ORDINARY);
+
+	n_fields = dict_index_get_n_fields(index);
+
+	ut_ad(ith < n_fields);
+
+	/* read the lengths of fields 0..n */
+	for (i = 0; i < n_fields; i++) {
+		ibool	is_null;
+		ulint	len;
+		field = dict_index_get_nth_field(index, i);
+		type = dict_col_get_type(dict_field_get_col(field));
+		is_null = !(dtype_get_prtype(type) & DATA_NOT_NULL);
+		if (is_null) {
+			/* nullable field => read the null flag */
+			is_null = !!(*nulls & null_mask);
+			null_mask <<= 1;
+			if (null_mask == 0x100)
+				nulls--, null_mask = 1;
+		}
+		if (is_null || field->fixed_len) {
+			/* No length (or extern bit) is stored for
+			fields that are NULL or fixed-length. */
+			ut_ad(i != ith);
+			continue;
+		}
+		len = *lens--;
+		if (dtype_get_len(type) > 255
+				|| dtype_get_mtype(type) == DATA_BLOB) {
+			if (len & 0x80) { /* 1exxxxxx: 2-byte length */
+				if (i == ith) {
+					if (!val == !(len & 0x20)) {
+						return; /* no change */
+					}
+					/* toggle the extern bit */
+					len ^= 0x40;
+					if (mtr) {
+						mlog_write_ulint(lens + 1, len,
+							MLOG_1BYTE, mtr);
+					} else {
+						lens[1] = (byte) len;
+					}
+					return;
+				}
+				lens--;
+			} else {
+				/* short fields cannot be external */
+				ut_ad(i != ith);
+			}
+		} else {
+			/* short fields cannot be external */
+			ut_ad(i != ith);
+		}
+	}
+}
+
+/***************************************************************
+Sets TRUE the extern storage bits of fields mentioned in an array. */
+
+void
+rec_set_field_extern_bits(
+/*======================*/
+	rec_t*		rec,	/* in: record */
+	dict_index_t*	index,	/* in: record descriptor */
+	const ulint*	vec,	/* in: array of field numbers */
+	ulint		n_fields,/* in: number of fields numbers */
+	mtr_t*		mtr)	/* in: mtr holding an X-latch to the
+				page where rec is, or NULL;
+				in the NULL case we do not write
+				to log about the change */
+{
+	ulint	i;
+	
+	for (i = 0; i < n_fields; i++) {
+		rec_set_nth_field_extern_bit(rec, index, vec[i], TRUE, mtr);
+	}
+}
+
+/*************************************************************** 
+Sets an old-style record field to SQL null.
+The physical size of the field is not changed. */
+
+void
+rec_set_nth_field_sql_null(
+/*=======================*/
+	rec_t*	rec, 	/* in: record */
+	ulint	n)	/* in: index of the field */
+{
+	ulint	offset;
+
+	offset = rec_get_field_start_offs(rec, n);
+
+	data_write_sql_null(rec + offset, rec_get_nth_field_size(rec, n));
+
+	rec_set_nth_field_null_bit(rec, n, TRUE);
+}
+
+/*************************************************************
+Builds an old-style physical record out of a data tuple and
+stores it beginning from the start of the given buffer. */
+static
+rec_t*
+rec_convert_dtuple_to_rec_old(
+/*==========================*/			
+			/* out: pointer to the origin of
+			physical record */
+	byte*	buf,	/* in: start address of the physical record */
+	dtuple_t* dtuple)/* in: data tuple */
+{
+	dfield_t* 	field;
+	ulint		n_fields;
+	ulint		data_size;
+	rec_t* 		rec;
+	ulint		end_offset;
+	ulint		ored_offset;
+	byte*		data;
+	ulint		len;
+	ulint		i;
+	
+	ut_ad(buf && dtuple);
+	ut_ad(dtuple_validate(dtuple));
+	ut_ad(dtuple_check_typed(dtuple));
+
+	n_fields = dtuple_get_n_fields(dtuple);
+	data_size = dtuple_get_data_size(dtuple);
+
+	ut_ad(n_fields > 0);
+
+	/* Calculate the offset of the origin in the physical record */	
+
+	rec = buf + rec_get_converted_extra_size(data_size, n_fields);
+	
+	/* Store the number of fields */
+	rec_set_n_fields_old(rec, n_fields);
+
+	/* Set the info bits of the record */
+	rec_set_info_bits(rec, FALSE,
+			dtuple_get_info_bits(dtuple) & REC_INFO_BITS_MASK);
+
+	/* Store the data and the offsets */
+
+	end_offset = 0;
+
+	if (data_size <= REC_1BYTE_OFFS_LIMIT) {
+
+	    rec_set_1byte_offs_flag(rec, TRUE);
+
+	    for (i = 0; i < n_fields; i++) {
+
+		field = dtuple_get_nth_field(dtuple, i);
+
+		data = dfield_get_data(field);
+		len = dfield_get_len(field);
+		
+		if (len == UNIV_SQL_NULL) {
+			len = dtype_get_sql_null_size(dfield_get_type(field));
+			data_write_sql_null(rec + end_offset, len);
+		
+			end_offset += len;
+			ored_offset = end_offset | REC_1BYTE_SQL_NULL_MASK;
+		} else {
+			/* If the data is not SQL null, store it */
+			ut_memcpy(rec + end_offset, data, len);
+
+			end_offset += len;
+			ored_offset = end_offset;
+		}
+
+		rec_1_set_field_end_info(rec, i, ored_offset);
+	    }
+	} else {
+	    rec_set_1byte_offs_flag(rec, FALSE);
+
+	    for (i = 0; i < n_fields; i++) {
+
+		field = dtuple_get_nth_field(dtuple, i);
+
+		data = dfield_get_data(field);
+		len = dfield_get_len(field);
+		
+		if (len == UNIV_SQL_NULL) {
+			len = dtype_get_sql_null_size(dfield_get_type(field));
+			data_write_sql_null(rec + end_offset, len);
+		
+			end_offset += len;
+			ored_offset = end_offset | REC_2BYTE_SQL_NULL_MASK;
+		} else {
+			/* If the data is not SQL null, store it */
+			ut_memcpy(rec + end_offset, data, len);
+
+			end_offset += len;
+			ored_offset = end_offset;
+		}
+
+		rec_2_set_field_end_info(rec, i, ored_offset);
+	    }
+	}
+
+	return(rec);
+}
+
+/*************************************************************
+Builds a new-style physical record out of a data tuple and
+stores it beginning from the start of the given buffer. */
+static
+rec_t*
+rec_convert_dtuple_to_rec_new(
+/*==========================*/
+				/* out: pointer to the origin
+				of physical record */
+	byte*		buf,	/* in: start address of the physical record */
+	dict_index_t*	index,	/* in: record descriptor */
+	dtuple_t*	dtuple)	/* in: data tuple */
+{
+	dfield_t*	field;
+	dtype_t*	type;
+	rec_t* 		rec		= buf + REC_N_NEW_EXTRA_BYTES;
+	byte*		end;
+	byte*		nulls;
+	byte*		lens;
+	ulint		len;
+	ulint		i;
+	ulint		fixed_len;
+	ulint		null_mask	= 1;
+	const ulint	n_fields	= dtuple_get_n_fields(dtuple);
+	const ulint	status		= dtuple_get_info_bits(dtuple)
+					& REC_NEW_STATUS_MASK;
+	ut_ad(index->table->comp);
+
+	ut_ad(n_fields > 0);
+	switch (status) {
+	case REC_STATUS_ORDINARY:
+		ut_ad(n_fields <= dict_index_get_n_fields(index));
+		break;
+	case REC_STATUS_NODE_PTR:
+		ut_ad(n_fields == dict_index_get_n_unique_in_tree(index) + 1);
+		break;
+	case REC_STATUS_INFIMUM:
+	case REC_STATUS_SUPREMUM:
+		ut_ad(n_fields == 1);
+		goto init;
+	default:
+		ut_a(0);
+		return(0);
+	}
+
+	/* Calculate the offset of the origin in the physical record.
+	We must loop over all fields to do this. */
+	rec += (index->n_nullable + 7) / 8;
+
+	for (i = 0; i < n_fields; i++) {
+		field = dtuple_get_nth_field(dtuple, i);
+		type = dfield_get_type(field);
+		len = dfield_get_len(field);
+		if (status == REC_STATUS_NODE_PTR && i == n_fields - 1) {
+			fixed_len = 4;
+			ut_ad(dtype_get_prtype(type) & DATA_NOT_NULL);
+			ut_ad(len == 4);
+			continue;
+		}
+		fixed_len = dict_index_get_nth_field(index, i)->fixed_len;
+
+		if (!(dtype_get_prtype(type) & DATA_NOT_NULL)) {
+			if (len == UNIV_SQL_NULL)
+				continue;
+		}
+		/* only nullable fields can be null */
+		ut_ad(len != UNIV_SQL_NULL);
+		if (fixed_len) {
+			ut_ad(len == fixed_len);
+		} else {
+			ut_ad(len <= dtype_get_len(type)
+				|| dtype_get_mtype(type) == DATA_BLOB);
+			rec++;
+			if (len >= 128 && (dtype_get_len(type) >= 256
+				|| dtype_get_mtype(type) == DATA_BLOB)) {
+				rec++;
+			}
+		}
+	}
+
+init:
+	end = rec;
+	nulls = rec - (REC_N_NEW_EXTRA_BYTES + 1);
+	lens = nulls - (index->n_nullable + 7) / 8;
+	/* clear the SQL-null flags */
+	memset (lens + 1, 0, nulls - lens);
+
+	/* Set the info bits of the record */
+	rec_set_status(rec, status);
+
+	rec_set_info_bits(rec, TRUE,
+			dtuple_get_info_bits(dtuple) & REC_INFO_BITS_MASK);
+
+	/* Store the data and the offsets */
+
+	for (i = 0; i < n_fields; i++) {
+		field = dtuple_get_nth_field(dtuple, i);
+		type = dfield_get_type(field);
+		len = dfield_get_len(field);
+
+		if (status == REC_STATUS_NODE_PTR && i == n_fields - 1) {
+			fixed_len = 4;
+			ut_ad(dtype_get_prtype(type) & DATA_NOT_NULL);
+			ut_ad(len == 4);
+			goto copy;
+		}
+		fixed_len = dict_index_get_nth_field(index, i)->fixed_len;
+
+		if (!(dtype_get_prtype(type) & DATA_NOT_NULL)) {
+			/* nullable field */
+			ut_ad(index->n_nullable > 0);
+			ut_ad(*nulls < null_mask);
+			/* set the null flag if necessary */
+			if (len == UNIV_SQL_NULL) {
+				*nulls |= null_mask;
+			}
+			null_mask <<= 1;
+			if (null_mask == 0x100)
+				nulls--, null_mask = 1;
+			if (len == UNIV_SQL_NULL)
+				continue;
+		}
+		/* only nullable fields can be null */
+		ut_ad(len != UNIV_SQL_NULL);
+		if (fixed_len) {
+			ut_ad(len == fixed_len);
+		} else {
+			ut_ad(len <= dtype_get_len(type)
+				|| dtype_get_mtype(type) == DATA_BLOB);
+			if (len < 128 || (dtype_get_len(type) < 256
+				&& dtype_get_mtype(type) != DATA_BLOB)) {
+				*lens-- = (byte) len;
+			}
+			else {
+				/* the extern bits will be set later */
+				ut_ad(len < 16384);
+				*lens-- = (byte) (len >> 8) | 0x80;
+				*lens-- = (byte) len;
+			}
+		}
+	copy:
+		memcpy(end, dfield_get_data(field), len);
+		end += len;
+	}
+
+	return(rec);
+}
+
+/*************************************************************
+Builds a physical record out of a data tuple and
+stores it beginning from the start of the given buffer. */
+
+rec_t*
+rec_convert_dtuple_to_rec(
+/*======================*/
+					/* out: pointer to the origin
+					of physical record */
+	byte*		buf,		/* in: start address of the
+					physical record */
+	dict_index_t*	index,		/* in: record descriptor */
+	dtuple_t*	dtuple)		/* in: data tuple */
+{
+	rec_t*	rec;
+
+	ut_ad(buf && index && dtuple);
+	ut_ad(dtuple_validate(dtuple));
+	ut_ad(dtuple_check_typed(dtuple));
+
+	if (index->table->comp) {
+		rec = rec_convert_dtuple_to_rec_new(buf, index, dtuple);
+	} else {
+		rec = rec_convert_dtuple_to_rec_old(buf, dtuple);
+	}
+
+#ifdef UNIV_DEBUG
+	{
+		mem_heap_t*	heap	= NULL;
+		ulint		offsets_[REC_OFFS_NORMAL_SIZE];
+		const ulint*	offsets;
+		*offsets_ = (sizeof offsets_) / sizeof *offsets_;
+
+		offsets = rec_get_offsets(rec, index,
+					offsets_, ULINT_UNDEFINED, &heap);
+		ut_ad(rec_validate(rec, offsets));
+		if (heap) {
+			mem_heap_free(heap);
+		}
+	}
+#endif /* UNIV_DEBUG */
+	return(rec);
+}
+
+/******************************************************************
+Copies the first n fields of a physical record to a data tuple. The fields
+are copied to the memory heap. */
+
+void
+rec_copy_prefix_to_dtuple(
+/*======================*/
+	dtuple_t*	tuple,		/* in: data tuple */
+	rec_t*		rec,		/* in: physical record */
+	dict_index_t*	index,		/* in: record descriptor */
+	ulint		n_fields,	/* in: number of fields to copy */
+	mem_heap_t*	heap)		/* in: memory heap */
+{
+	dfield_t*	field;
+	byte*		data;
+	ulint		len;
+	byte*		buf = NULL;
+	ulint		i;
+	ulint		offsets_[REC_OFFS_NORMAL_SIZE];
+	ulint*		offsets	= offsets_;
+	*offsets_ = (sizeof offsets_) / sizeof *offsets_;
+
+	offsets = rec_get_offsets(rec, index, offsets, n_fields, &heap);
+
+	ut_ad(rec_validate(rec, offsets));
+	ut_ad(dtuple_check_typed(tuple));
+
+	dtuple_set_info_bits(tuple,
+			rec_get_info_bits(rec, index->table->comp));
+
+	for (i = 0; i < n_fields; i++) {
+
+		field = dtuple_get_nth_field(tuple, i);
+		data = rec_get_nth_field(rec, offsets, i, &len);
+
+		if (len != UNIV_SQL_NULL) {
+			buf = mem_heap_alloc(heap, len);
+
+			ut_memcpy(buf, data, len);
+		}
+
+		dfield_set_data(field, buf, len);
+	}
+}
+
+/******************************************************************
+Copies the first n fields of an old-style physical record
+to a new physical record in a buffer. */
+static
+rec_t*
+rec_copy_prefix_to_buf_old(
+/*=======================*/
+				/* out, own: copied record */
+	rec_t*	rec,		/* in: physical record */
+	ulint	n_fields,	/* in: number of fields to copy */
+	ulint	area_end,	/* in: end of the prefix data */
+	byte**	buf,		/* in/out: memory buffer for the copied prefix,
+				or NULL */
+	ulint*	buf_size)	/* in/out: buffer size */
+{
+	rec_t*	copy_rec;
+	ulint	area_start;
+	ulint	prefix_len;
+
+	if (rec_get_1byte_offs_flag(rec)) {
+		area_start = REC_N_OLD_EXTRA_BYTES + n_fields;
+	} else {
+		area_start = REC_N_OLD_EXTRA_BYTES + 2 * n_fields;
+	}
+
+	prefix_len = area_start + area_end;
+
+	if ((*buf == NULL) || (*buf_size < prefix_len)) {
+		if (*buf != NULL) {
+			mem_free(*buf);
+		}
+
+		*buf = mem_alloc(prefix_len);
+		*buf_size = prefix_len;
+	}
+
+	ut_memcpy(*buf, rec - area_start, prefix_len);
+
+	copy_rec = *buf + area_start;
+
+	rec_set_n_fields_old(copy_rec, n_fields);
+
+	return(copy_rec);
+}
+
+/******************************************************************
+Copies the first n fields of a physical record to a new physical record in
+a buffer. */
+
+rec_t*
+rec_copy_prefix_to_buf(
+/*===================*/
+					/* out, own: copied record */
+	rec_t*		rec,		/* in: physical record */
+	dict_index_t*	index,		/* in: record descriptor */
+	ulint		n_fields,	/* in: number of fields to copy */
+	byte**		buf,		/* in/out: memory buffer
+					for the copied prefix, or NULL */
+	ulint*		buf_size)	/* in/out: buffer size */
+{
+	byte*		nulls	= rec - (REC_N_NEW_EXTRA_BYTES + 1);
+	byte*		lens	= nulls - (index->n_nullable + 7) / 8;
+	dict_field_t*	field;
+	dtype_t*	type;
+	ulint		i;
+	ulint		prefix_len	= 0;
+	ibool		is_null;
+	ulint		null_mask	= 1;
+	ulint		status;
+
+	if (!index->table->comp) {
+		ut_ad(rec_validate_old(rec));
+		return(rec_copy_prefix_to_buf_old(rec, n_fields,
+			rec_get_field_start_offs(rec, n_fields),
+			buf, buf_size));
+	}
+
+	status = rec_get_status(rec);
+
+	switch (status) {
+	case REC_STATUS_ORDINARY:
+		ut_ad(n_fields <= dict_index_get_n_fields(index));
+		break;
+	case REC_STATUS_NODE_PTR:
+		/* it doesn't make sense to copy the child page number field */
+		ut_ad(n_fields <= dict_index_get_n_unique_in_tree(index));
+		break;
+	case REC_STATUS_INFIMUM:
+	case REC_STATUS_SUPREMUM:
+		/* infimum or supremum record: no sense to copy anything */
+	default:
+		ut_a(0);
+		return(NULL);
+	}
+
+	/* read the lengths of fields 0..n */
+	for (i = 0; i < n_fields; i++) {
+		field = dict_index_get_nth_field(index, i);
+		type = dict_col_get_type(dict_field_get_col(field));
+		is_null = !(dtype_get_prtype(type) & DATA_NOT_NULL);
+		if (is_null) {
+			/* nullable field => read the null flag */
+			is_null = !!(*nulls & null_mask);
+			null_mask <<= 1;
+			if (null_mask == 0x100)
+				nulls--, null_mask = 1;
+		}
+
+		if (is_null) {
+		} else if (field->fixed_len) {
+			prefix_len += field->fixed_len;
+		} else {
+			ulint	len = *lens--;
+			if (dtype_get_len(type) > 255
+				|| dtype_get_mtype(type) == DATA_BLOB) {
+				if (len & 0x80) {
+					/* 1exxxxxx */
+					len &= 0x3f;
+					len <<= 8;
+					len |= *lens--;
+				}
+			}
+			prefix_len += len;
+		}
+	}
+
+	prefix_len += rec - (lens + 1);
+
+	if ((*buf == NULL) || (*buf_size < prefix_len)) {
+		if (*buf != NULL) {
+			mem_free(*buf);
+		}
+
+		*buf = mem_alloc(prefix_len);
+		*buf_size = prefix_len;
+	}
+
+	memcpy(*buf, lens + 1, prefix_len);
+
+	return(*buf + (rec - (lens + 1)));
+}
+
+/*******************************************************************
+Validates the consistency of an old-style physical record. */
+static
+ibool
+rec_validate_old(
+/*=============*/
+			/* out: TRUE if ok */
+	rec_t*	rec)	/* in: physical record */
+{
+	byte*	data;
+	ulint	len;
+	ulint	n_fields;
+	ulint	len_sum		= 0;
+	ulint	sum		= 0;
+	ulint	i;
+
+	ut_a(rec);
+	n_fields = rec_get_n_fields_old(rec);
+
+	if ((n_fields == 0) || (n_fields > REC_MAX_N_FIELDS)) {
+		fprintf(stderr, "InnoDB: Error: record has %lu fields\n",
+							(ulong) n_fields);
+		return(FALSE);
+	}
+	
+	for (i = 0; i < n_fields; i++) {
+		data = rec_get_nth_field_old(rec, i, &len);
+		
+		if (!((len < UNIV_PAGE_SIZE) || (len == UNIV_SQL_NULL))) {
+			fprintf(stderr,
+			"InnoDB: Error: record field %lu len %lu\n", (ulong) i,
+							(ulong) len);
+			return(FALSE);
+		}	
+
+		if (len != UNIV_SQL_NULL) {
+			len_sum += len;
+			sum += *(data + len -1); /* dereference the
+						end of the field to
+						cause a memory trap
+						if possible */
+		} else {
+			len_sum += rec_get_nth_field_size(rec, i);
+		}
+	}
+
+	if (len_sum != rec_get_data_size_old(rec)) {
+		fprintf(stderr,
+		"InnoDB: Error: record len should be %lu, len %lu\n",
+				(ulong) len_sum,
+			        rec_get_data_size_old(rec));
+		return(FALSE);
+	}
+
+	rec_dummy = sum; /* This is here only to fool the compiler */
+
+	return(TRUE);
+}
+
+/*******************************************************************
+Validates the consistency of a physical record. */
+
+ibool
+rec_validate(
+/*=========*/
+				/* out: TRUE if ok */
+	rec_t*		rec,	/* in: physical record */
+	const ulint*	offsets)/* in: array returned by rec_get_offsets() */
+{
+	const byte*	data;
+	ulint		len;
+	ulint		n_fields;
+	ulint		len_sum		= 0;
+	ulint		sum		= 0;
+	ulint		i;
+
+	ut_a(rec);
+	n_fields = rec_offs_n_fields(offsets);
+
+	if ((n_fields == 0) || (n_fields > REC_MAX_N_FIELDS)) {
+		fprintf(stderr, "InnoDB: Error: record has %lu fields\n",
+							(ulong) n_fields);
+		return(FALSE);
+	}
+
+	ut_a(rec_offs_comp(offsets) || n_fields <= rec_get_n_fields_old(rec));
+
+	for (i = 0; i < n_fields; i++) {
+		data = rec_get_nth_field(rec, offsets, i, &len);
+
+		if (!((len < UNIV_PAGE_SIZE) || (len == UNIV_SQL_NULL))) {
+			fprintf(stderr,
+			"InnoDB: Error: record field %lu len %lu\n", (ulong) i,
+							(ulong) len);
+			return(FALSE);
+		}
+
+		if (len != UNIV_SQL_NULL) {
+			len_sum += len;
+			sum += *(data + len -1); /* dereference the
+						end of the field to
+						cause a memory trap
+						if possible */
+		} else if (!rec_offs_comp(offsets)) {
+			len_sum += rec_get_nth_field_size(rec, i);
+		}
+	}
+
+	if (len_sum != (ulint)(rec_get_end(rec, offsets) - rec)) {
+		fprintf(stderr,
+		"InnoDB: Error: record len should be %lu, len %lu\n",
+				(ulong) len_sum,
+				(ulong) (rec_get_end(rec, offsets) - rec));
+		return(FALSE);
+	}	
+
+	rec_dummy = sum; /* This is here only to fool the compiler */
+
+	if (!rec_offs_comp(offsets)) {
+		ut_a(rec_validate_old(rec));
+	}
+
+	return(TRUE);
+}
+
+/*******************************************************************
+Prints an old-style physical record. */
+
+void
+rec_print_old(
+/*==========*/
+	FILE*		file,	/* in: file where to print */
+	rec_t*		rec)	/* in: physical record */
+{
+	const byte*	data;
+	ulint		len;
+	ulint		n;
+	ulint		i;
+
+	ut_ad(rec);
+
+	n = rec_get_n_fields_old(rec);
+
+	fprintf(file, "PHYSICAL RECORD: n_fields %lu;"
+		" %u-byte offsets; info bits %lu\n",
+		(ulong) n,
+		rec_get_1byte_offs_flag(rec) ? 1 : 2,
+		(ulong) rec_get_info_bits(rec, FALSE));
+
+	for (i = 0; i < n; i++) {
+
+		data = rec_get_nth_field_old(rec, i, &len);
+
+		fprintf(file, " %lu:", (ulong) i);
+
+		if (len != UNIV_SQL_NULL) {
+			if (len <= 30) {
+
+				ut_print_buf(file, data, len);
+			} else {
+				ut_print_buf(file, data, 30);
+
+				fputs("...(truncated)", file);
+			}
+		} else {
+			fprintf(file, " SQL NULL, size %lu ",
+				rec_get_nth_field_size(rec, i));
+		}
+		putc(';', file);
+	}
+
+	putc('\n', file);
+
+	rec_validate_old(rec);
+}
+
+/*******************************************************************
+Prints a physical record. */
+
+void
+rec_print_new(
+/*==========*/
+	FILE*		file,	/* in: file where to print */
+	rec_t*		rec,	/* in: physical record */
+	const ulint*	offsets)/* in: array returned by rec_get_offsets() */
+{
+	const byte*	data;
+	ulint		len;
+	ulint		i;
+
+	ut_ad(rec_offs_validate(rec, NULL, offsets));
+
+	if (!rec_offs_comp(offsets)) {
+		rec_print_old(file, rec);
+		return;
+	}
+
+	ut_ad(rec);
+
+	fprintf(file, "PHYSICAL RECORD: n_fields %lu;"
+		" compact format; info bits %lu\n",
+		(ulong) rec_offs_n_fields(offsets),
+		(ulong) rec_get_info_bits(rec, TRUE));
+	
+	for (i = 0; i < rec_offs_n_fields(offsets); i++) {
+
+		data = rec_get_nth_field(rec, offsets, i, &len);
+
+		fprintf(file, " %lu:", (ulong) i);
+	
+		if (len != UNIV_SQL_NULL) {
+			if (len <= 30) {
+
+				ut_print_buf(file, data, len);
+			} else {
+				ut_print_buf(file, data, 30);
+
+				fputs("...(truncated)", file);
+			}
+		} else {
+			fputs(" SQL NULL", file);
+		}
+		putc(';', file);
+	}
+
+	putc('\n', file);
+
+	rec_validate(rec, offsets);
+}
+
+/*******************************************************************
+Prints a physical record. */
+
+void
+rec_print(
+/*======*/
+	FILE*		file,	/* in: file where to print */
+	rec_t*		rec,	/* in: physical record */
+	dict_index_t*	index)	/* in: record descriptor */
+{
+	ut_ad(index);
+
+	if (!index->table->comp) {
+		rec_print_old(file, rec);
+		return;
+	} else {
+		mem_heap_t*	heap	= NULL;
+		ulint		offsets_[REC_OFFS_NORMAL_SIZE];
+		*offsets_ = (sizeof offsets_) / sizeof *offsets_;
+
+		rec_print_new(file, rec, rec_get_offsets(rec, index, offsets_,
+						ULINT_UNDEFINED, &heap));
+		if (heap) {
+			mem_heap_free(heap);
+		}
+	}
+}