branches/zip: Implement the configuration parameter and settable global

variable innodb_file_format.  Implement file format version stamping of
*.ibd files and SYS_TABLES.TYPE.

This change breaks introduces an incompatible change for for
compressed tables.  We can do this, as we have not released yet.

innodb-zip.test: Add tests for stricter KEY_BLOCK_SIZE and ROW_FORMAT
checks.

DICT_TF_COMPRESSED_MASK, DICT_TF_COMPRESSED_SHIFT: Replace with
DICT_TF_ZSSIZE_MASK, DICT_TF_ZSSIZE_SHIFT.

DICT_TF_FORMAT_MASK, DICT_TF_FORMAT_SHIFT, DICT_TF_FORMAT_51,
DICT_TF_FORMAT_ZIP: File format version, stored in table->flags,
in the .ibd file header, and in SYS_TABLES.TYPE.

dict_create_sys_tables_tuple(): Write the table flags to SYS_TABLES.TYPE
if the format is at least DICT_TF_FORMAT_ZIP.  For old formats
(DICT_TF_FORMAT_51), write DICT_TABLE_ORDINARY as the table type.

DB_TABLE_ZIP_NO_IBD: Remove the error code.  The error handling is done
in ha_innodb.cc; as a failsafe measure, dict_build_table_def_step() will
silently clear the compression and format flags instead of returning this
error.

dict_mem_table_create(): Assert that no extra bits are set in the flags.

dict_sys_tables_get_zip_size(): Rename to dict_sys_tables_get_flags().
Check all flag bits, and return ULINT_UNDEFINED if the combination is
unsupported.

dict_boot(): Document the SYS_TABLES columns N_COLS and TYPE.

dict_table_get_format(), dict_table_set_format(),
dict_table_flags_to_zip_size(): New accessors to table->flags.

dtuple_convert_big_rec(): Introduce the auxiliary variables
local_len, local_prefix_len.  Store a 768-byte prefix locally
if the file format is less than DICT_TF_FORMAT_ZIP.

dtuple_convert_back_big_rec(): Restore the columns.

srv_file_format: New variable: innodb_file_format.

fil_create_new_single_table_tablespace(): Replace the parameter zip_size
with table->flags.

fil_open_single_table_tablespace(): Replace the parameter zip_size_in_k
with table->flags.  Check the flags.

fil_space_struct, fil_space_create(), fil_op_write_log():
Replace zip_size with flags.

fil_node_open_file(): Note a TODO item for InnoDB Hot Backup.
Check that the tablespace flags match.

fil_space_get_zip_size(): Rename to fil_space_get_flags().  Add a
wrapper for fil_space_get_zip_size().

fsp_header_get_flags(): New function.

fsp_header_init_fields(): Replace zip_size with flags.

FSP_SPACE_FLAGS: New name for the tablespace flags.  This field used
to be called FSP_PAGE_ZIP_SIZE, or FSP_LOWEST_NO_WRITE.  It has always
been written as 0 in MySQL/InnoDB versions 4.1 to 5.1.

MLOG_ZIP_FILE_CREATE: Rename to MLOG_FILE_CREATE2.  Add a 32-bit
parameter for the tablespace flags.

ha_innobase::create(): Check the table attributes ROW_FORMAT and
KEY_BLOCK_SIZE.  Issue errors if they are inappropriate, or warnings
if the inherited attributes (in ALTER TABLE) will be ignored.

PAGE_ZIP_MIN_SIZE_SHIFT: New constant: the 2-logarithm of PAGE_ZIP_MIN_SIZE.

data/data0data.c

@@ -561,11 +561,21 @@ dtuple_convert_big_rec(
 	dict_field_t*	ifield;
 	ulint		size;
 	ulint		n_fields;
+	ulint		local_len;
+	ulint		local_prefix_len;
 
 	if (UNIV_UNLIKELY(!dict_index_is_clust(index))) {
 		return(NULL);
 	}
 
+	if (dict_table_get_format(index->table) < DICT_TF_FORMAT_ZIP) {
+		/* up to MySQL 5.1: store a 768-byte prefix locally */
+		local_len = BTR_EXTERN_FIELD_REF_SIZE + DICT_MAX_INDEX_COL_LEN;
+	} else {
+		/* new-format table: do not store any BLOB prefix locally */
+		local_len = BTR_EXTERN_FIELD_REF_SIZE;
+	}
+
 	ut_a(dtuple_check_typed_no_assert(entry));
 
 	size = rec_get_converted_size(index, entry, *n_ext);
@@ -601,6 +611,8 @@ dtuple_convert_big_rec(
 		ulint			i;
 		ulint			longest		= 0;
 		ulint			longest_i	= ULINT_MAX;
+		byte*			data;
+		big_rec_field_t*	b;
 
 		for (i = dict_index_get_n_unique_in_tree(index);
 		     i < dtuple_get_n_fields(entry); i++) {
@@ -615,13 +627,13 @@ dtuple_convert_big_rec(
 			if (ifield->fixed_len
 			    || dfield_is_null(dfield)
 			    || dfield_is_ext(dfield)
+			    || dfield_get_len(dfield) <= local_len
 			    || dfield_get_len(dfield)
 			    <= BTR_EXTERN_FIELD_REF_SIZE * 2) {
 				goto skip_field;
 			}
 
-			savings = dfield_get_len(dfield)
-				- BTR_EXTERN_FIELD_REF_SIZE;
+			savings = dfield_get_len(dfield) - local_len;
 
 			/* Check that there would be savings */
 			if (longest >= savings) {
@@ -651,25 +663,32 @@ skip_field:
 
 		dfield = dtuple_get_nth_field(entry, longest_i);
 		ifield = dict_index_get_nth_field(index, longest_i);
-		vector->fields[n_fields].field_no = longest_i;
+		local_prefix_len = local_len - BTR_EXTERN_FIELD_REF_SIZE;
 
-		vector->fields[n_fields].len = dfield_get_len(dfield);
+		b = &vector->fields[n_fields];
+		b->field_no = longest_i;
+		b->len = dfield_get_len(dfield) - local_prefix_len;
+		b->data = (char*) dfield_get_data(dfield) + local_prefix_len;
 
-		vector->fields[n_fields].data = dfield_get_data(dfield);
+		/* Allocate the locally stored part of the column. */
+		data = mem_heap_alloc(heap, local_len);
 
-		/* Set the extern field reference in dfield to zero */
-		dfield_set_data(dfield,
-				mem_heap_zalloc(heap,
-						BTR_EXTERN_FIELD_REF_SIZE),
-				BTR_EXTERN_FIELD_REF_SIZE);
-		dfield_set_ext(dfield);
+		/* Copy the local prefix. */
+		memcpy(data, dfield_get_data(dfield), local_prefix_len);
+		/* Clear the extern field reference (BLOB pointer). */
+		memset(data + local_prefix_len, 0, BTR_EXTERN_FIELD_REF_SIZE);
 #if 0
 		/* The following would fail the Valgrind checks in
 		page_cur_insert_rec_low() and page_cur_insert_rec_zip().
 		The BLOB pointers in the record will be initialized after
 		the record and the BLOBs have been written. */
-		UNIV_MEM_ALLOC(dfield->data, BTR_EXTERN_FIELD_REF_SIZE);
+		UNIV_MEM_ALLOC(data + local_prefix_len,
+			       BTR_EXTERN_FIELD_REF_SIZE);
 #endif
+
+		dfield_set_data(dfield, data, local_len);
+		dfield_set_ext(dfield);
+
 		n_fields++;
 		(*n_ext)++;
 		ut_ad(n_fields < dtuple_get_n_fields(entry));
@@ -692,16 +711,26 @@ dtuple_convert_back_big_rec(
 	big_rec_t*	vector)	/* in, own: big rec vector; it is
 				freed in this function */
 {
+	big_rec_field_t*		b	= vector->fields;
+	const big_rec_field_t* const	end	= b + vector->n_fields;
+
+	for (; b < end; b++) {
 		dfield_t*	dfield;
-	ulint		i;
+		ulint		local_len;
 
-	for (i = 0; i < vector->n_fields; i++) {
+		dfield = dtuple_get_nth_field(entry, b->field_no);
+		local_len = dfield_get_len(dfield);
 
-		dfield = dtuple_get_nth_field(entry,
-					      vector->fields[i].field_no);
 		ut_ad(dfield_is_ext(dfield));
+		ut_ad(local_len >= BTR_EXTERN_FIELD_REF_SIZE);
+
+		local_len -= BTR_EXTERN_FIELD_REF_SIZE;
+
+		ut_ad(local_len <= DICT_MAX_INDEX_COL_LEN);
+
 		dfield_set_data(dfield,
-				vector->fields[i].data, vector->fields[i].len);
+				(char*) b->data - local_len,
+				b->len + local_len);
 	}
 
 	mem_heap_free(vector->heap);

dict/dict0boot.c

@@ -252,7 +252,10 @@ dict_boot(void)
 
 	dict_mem_table_add_col(table, heap, "NAME", DATA_BINARY, 0, 0);
 	dict_mem_table_add_col(table, heap, "ID", DATA_BINARY, 0, 0);
+	/* ROW_FORMAT = (N_COLS >> 31) ? COMPACT : REDUNDANT */
 	dict_mem_table_add_col(table, heap, "N_COLS", DATA_INT, 0, 4);
+	/* TYPE is either DICT_TABLE_ORDINARY, or (TYPE & DICT_TF_COMPACT)
+	and (TYPE & DICT_TF_FORMAT_MASK) are nonzero and TYPE = table->flags */
 	dict_mem_table_add_col(table, heap, "TYPE", DATA_INT, 0, 4);
 	dict_mem_table_add_col(table, heap, "MIX_ID", DATA_BINARY, 0, 0);
 	dict_mem_table_add_col(table, heap, "MIX_LEN", DATA_INT, 0, 4);

dict/dict0crea.c

@@ -77,11 +77,13 @@ dict_create_sys_tables_tuple(
 	dfield = dtuple_get_nth_field(entry, 3);
 
 	ptr = mem_heap_alloc(heap, 4);
-	if (table->flags & DICT_TF_COMPRESSED_MASK) {
+	if (table->flags & ~DICT_TF_COMPACT) {
 		ut_a(table->flags & DICT_TF_COMPACT);
-		mach_write_to_4(ptr, DICT_TABLE_COMPRESSED_BASE
-				+ ((table->flags & DICT_TF_COMPRESSED_MASK)
-				>> DICT_TF_COMPRESSED_SHIFT));
+		ut_a(dict_table_get_format(table) >= DICT_TF_FORMAT_ZIP);
+		ut_a((table->flags & DICT_TF_ZSSIZE_MASK)
+		     <= (DICT_TF_ZSSIZE_MAX << DICT_TF_ZSSIZE_SHIFT));
+		ut_a(!(table->flags & (~0 << DICT_TF_BITS)));
+		mach_write_to_4(ptr, table->flags);
 	} else {
 		mach_write_to_4(ptr, DICT_TABLE_ORDINARY);
 	}
@@ -255,9 +257,13 @@ dict_build_table_def_step(
 			is_path = FALSE;
 		}
 
+		ut_ad(dict_table_get_format(table) <= DICT_TF_FORMAT_MAX);
+		ut_ad(!dict_table_zip_size(table)
+		      || dict_table_get_format(table) >= DICT_TF_FORMAT_ZIP);
+
 		error = fil_create_new_single_table_tablespace(
 			&space, path_or_name, is_path,
-			dict_table_zip_size(table),
+			table->flags,
 			FIL_IBD_FILE_INITIAL_SIZE);
 		table->space = (unsigned int) space;
 
@@ -272,10 +278,8 @@ dict_build_table_def_step(
 
 		mtr_commit(&mtr);
 	} else {
-		/* Create in the system tablespace: disallow compression */
-		if (table->flags & DICT_TF_COMPRESSED_MASK) {
-			return(DB_TABLE_ZIP_NO_IBD);
-		}
+		/* Create in the system tablespace: disallow new features */
+		table->flags &= DICT_TF_COMPACT;
 	}
 
 	row = dict_create_sys_tables_tuple(table, node->heap);

dict/dict0load.c

@@ -223,11 +223,11 @@ loop:
 }
 
 /************************************************************************
-Determine the compressed page size of a table described in SYS_TABLES. */
+Determine the flags of a table described in SYS_TABLES. */
 static
 ulint
-dict_sys_tables_get_zip_size(
-/*=========================*/
+dict_sys_tables_get_flags(
+/*======================*/
 				/* out: compressed page size in kilobytes;
 				or 0 if the tablespace is uncompressed,
 				ULINT_UNDEFINED on error */
@@ -236,29 +236,53 @@ dict_sys_tables_get_zip_size(
 	const byte*	field;
 	ulint		len;
 	ulint		n_cols;
-	ulint		table_type;
+	ulint		flags;
 
 	field = rec_get_nth_field_old(rec, 5, &len);
 	ut_a(len == 4);
 
-	table_type = mach_read_from_4(field);
+	flags = mach_read_from_4(field);
+
+	if (UNIV_LIKELY(flags == DICT_TABLE_ORDINARY)) {
+		return(0);
+	}
 
 	field = rec_get_nth_field_old(rec, 4, &len);
 	n_cols = mach_read_from_4(field);
 
-	if (UNIV_EXPECT(n_cols & 0x80000000UL, 0x80000000UL)
-			&& UNIV_LIKELY(table_type > DICT_TABLE_COMPRESSED_BASE)
-			&& UNIV_LIKELY(table_type
-					<= DICT_TABLE_COMPRESSED_BASE + 16)) {
+	if (UNIV_UNLIKELY(!(n_cols & 0x80000000UL))) {
+		/* New file formats require ROW_FORMAT=COMPACT. */
+		return(ULINT_UNDEFINED);
+	}
+
+	switch (flags & (DICT_TF_FORMAT_MASK | DICT_TF_COMPACT)) {
+	default:
+	case DICT_TF_FORMAT_51 << DICT_TF_FORMAT_SHIFT:
+	case DICT_TF_FORMAT_51 << DICT_TF_FORMAT_SHIFT | DICT_TF_COMPACT:
+		/* flags should be DICT_TABLE_ORDINARY,
+		or DICT_TF_FORMAT_MASK should be nonzero. */
+		return(ULINT_UNDEFINED);
 
-		return(table_type - DICT_TABLE_COMPRESSED_BASE);
+	case DICT_TF_FORMAT_ZIP << DICT_TF_FORMAT_SHIFT | DICT_TF_COMPACT:
+#if DICT_TF_FORMAT_MAX > DICT_TF_FORMAT_ZIP
+# error "missing case labels for DICT_TF_FORMAT_ZIP .. DICT_TF_FORMAT_MAX"
+#endif
+		/* We support this format. */
+		break;
 	}
 
-	if (UNIV_LIKELY(table_type == DICT_TABLE_ORDINARY)) {
-		return(0);
+	if (UNIV_UNLIKELY((flags & DICT_TF_ZSSIZE_MASK)
+			  > (DICT_TF_ZSSIZE_MAX << DICT_TF_ZSSIZE_SHIFT))) {
+		/* Unsupported compressed page size. */
+		return(ULINT_UNDEFINED);
 	}
 
+	if (UNIV_UNLIKELY(flags & (~0 << DICT_TF_BITS))) {
+		/* Some unused bits are set. */
 		return(ULINT_UNDEFINED);
+	}
+
+	return(flags);
 }
 
 /************************************************************************
@@ -321,14 +345,28 @@ loop:
 		const byte*	field;
 		ulint		len;
 		ulint		space_id;
-		ulint		zip_size_in_k;
+		ulint		flags;
 		char*		name;
 
 		field = rec_get_nth_field_old(rec, 0, &len);
 		name = mem_strdupl((char*) field, len);
 
-		zip_size_in_k = dict_sys_tables_get_zip_size(rec);
-		ut_a(zip_size_in_k != ULINT_UNDEFINED);
+		flags = dict_sys_tables_get_flags(rec);
+		if (UNIV_UNLIKELY(flags == ULINT_UNDEFINED)) {
+
+			field = rec_get_nth_field_old(rec, 5, &len);
+			flags = mach_read_from_4(field);
+
+			ut_print_timestamp(stderr);
+			fputs("  InnoDB: Error: table ", stderr);
+			ut_print_filename(stderr, name);
+			fprintf(stderr, "\n"
+				"InnoDB: in InnoDB data dictionary"
+				" has unknown type %lx.\n",
+				(ulong) flags);
+
+			goto loop;
+		}
 
 		field = rec_get_nth_field_old(rec, 9, &len);
 		ut_a(len == 4);
@@ -352,8 +390,7 @@ loop:
 			object and check that the .ibd file exists. */
 
 			fil_open_single_table_tablespace(FALSE, space_id,
-							 zip_size_in_k * 1024,
-							 name);
+							 flags, name);
 		}
 
 		mem_free(name);
@@ -784,7 +821,6 @@ dict_load_table(
 	ulint		n_cols;
 	ulint		flags;
 	ulint		err;
-	ulint		zip_size_in_k;
 	mtr_t		mtr;
 
 	ut_ad(mutex_own(&(dict_sys->mutex)));
@@ -833,8 +869,22 @@ err_exit:
 
 	/* Check if the tablespace exists and has the right name */
 	if (space != 0) {
-		zip_size_in_k = dict_sys_tables_get_zip_size(rec);
-		ut_a(zip_size_in_k != ULINT_UNDEFINED);
+		flags = dict_sys_tables_get_flags(rec);
+
+		if (UNIV_UNLIKELY(flags == ULINT_UNDEFINED)) {
+unknown_table_type:
+			field = rec_get_nth_field_old(rec, 5, &len);
+			flags = mach_read_from_4(field);
+
+			ut_print_timestamp(stderr);
+			fputs("  InnoDB: Error: table ", stderr);
+			ut_print_filename(stderr, name);
+			fprintf(stderr, "\n"
+				"InnoDB: in InnoDB data dictionary"
+				" has unknown type %lx.\n",
+				(ulong) flags);
+			goto err_exit;
+		}
 
 		if (fil_space_for_table_exists_in_mem(space, name, FALSE,
 						      FALSE, FALSE)) {
@@ -853,7 +903,7 @@ err_exit:
 				name, (ulong)space);
 			/* Try to open the tablespace */
 			if (!fil_open_single_table_tablespace(
-				    TRUE, space, zip_size_in_k << 10, name)) {
+				    TRUE, space, flags, name)) {
 				/* We failed to find a sensible tablespace
 				file */
 
@@ -861,7 +911,7 @@ err_exit:
 			}
 		}
 	} else {
-		zip_size_in_k = 0;
+		flags = 0;
 	}
 
 	ut_a(name_of_col_is(sys_tables, sys_index, 4, "N_COLS"));
@@ -869,11 +919,12 @@ err_exit:
 	field = rec_get_nth_field_old(rec, 4, &len);
 	n_cols = mach_read_from_4(field);
 
-	flags = zip_size_in_k << DICT_TF_COMPRESSED_SHIFT;
-
 	/* The high-order bit of N_COLS is the "compact format" flag. */
 	if (n_cols & 0x80000000UL) {
 		flags |= DICT_TF_COMPACT;
+	} else if (UNIV_UNLIKELY(flags)) {
+		/* Only ROW_FORMAT=COMPACT tables can have flags set. */
+		goto unknown_table_type;
 	}
 
 	table = dict_mem_table_create(name, space, n_cols & ~0x80000000UL,
@@ -886,17 +937,6 @@ err_exit:
 	field = rec_get_nth_field_old(rec, 3, &len);
 	table->id = mach_read_from_8(field);
 
-	zip_size_in_k = dict_sys_tables_get_zip_size(rec);
-
-	if (UNIV_UNLIKELY(zip_size_in_k == ULINT_UNDEFINED)) {
-		field = rec_get_nth_field_old(rec, 5, &len);
-		ut_print_timestamp(stderr);
-		fprintf(stderr,
-			"  InnoDB: table %s: unknown table type %lu\n",
-			name, (ulong) mach_read_from_4(field));
-		goto err_exit;
-	}
-
 	btr_pcur_close(&pcur);
 	mtr_commit(&mtr);
 

dict/dict0mem.c

@@ -42,6 +42,7 @@ dict_mem_table_create(
 	mem_heap_t*	heap;
 
 	ut_ad(name);
+	ut_a(!(flags & (~0 << DICT_TF_BITS)));
 
 	heap = mem_heap_create(DICT_HEAP_SIZE);
 

handler/ha_innodb.cc

@@ -5030,11 +5030,6 @@ create_table_def(
 
 	innodb_check_for_record_too_big_error(flags & DICT_TF_COMPACT, error);
 
-	if (error == DB_TABLE_ZIP_NO_IBD) {
-		my_error(ER_ILLEGAL_HA_CREATE_OPTION, MYF(0),
-			 innobase_hton_name, "KEY_BLOCK_SIZE");
-	}
-
 	error = convert_error_code_to_mysql(error, NULL);
 
 	DBUG_RETURN(error);
@@ -5236,6 +5231,7 @@ ha_innobase::create(
 	THD*		thd = ha_thd();
 	ib_longlong	auto_inc_value;
 	ulint		flags;
+	const ulint	file_format = srv_file_format;
 
 	DBUG_ENTER("ha_innobase::create");
 
@@ -5291,14 +5287,68 @@ ha_innobase::create(
 
 	flags = 0;
 
+	switch (create_info->key_block_size) {
+	case 0:
 		if (form->s->row_type != ROW_TYPE_REDUNDANT) {
 			flags |= DICT_TF_COMPACT;
+		}
 
-		switch (create_info->key_block_size) {
-		case 1: case 2: case 4: case 8: case 16:
-			flags |= create_info->key_block_size
-				<< DICT_TF_COMPRESSED_SHIFT;
+		goto key_block_size_ok;
+	case 1:
+		flags |= 1 << DICT_TF_ZSSIZE_SHIFT | DICT_TF_COMPACT;
+		break;
+	case 2:
+		flags |= 2 << DICT_TF_ZSSIZE_SHIFT | DICT_TF_COMPACT;
+		break;
+	case 4:
+		flags |= 3 << DICT_TF_ZSSIZE_SHIFT | DICT_TF_COMPACT;
+		break;
+	case 8:
+		flags |= 4 << DICT_TF_ZSSIZE_SHIFT | DICT_TF_COMPACT;
+		break;
+	case 16:
+		flags |= 5 << DICT_TF_ZSSIZE_SHIFT | DICT_TF_COMPACT;
+		break;
+#if DICT_TF_ZSSIZE_MAX != 5
+# error "DICT_TF_ZSSIZE_MAX != 5"
+#endif
+key_block_size_wrong:
+	default:
+		if (create_info->used_fields & HA_CREATE_USED_KEY_BLOCK_SIZE) {
+			my_error(ER_ILLEGAL_HA_CREATE_OPTION, MYF(0),
+				 innobase_hton_name, "KEY_BLOCK_SIZE");
+			error = -1;
+			goto cleanup;
+		} else {
+			sql_print_warning("InnoDB: ignoring"
+					  " KEY_BLOCK_SIZE=%lu\n",
+					  create_info->key_block_size);
+			flags &= ~DICT_TF_ZSSIZE_MASK;
+			goto key_block_size_ok;
+		}
+	}
+
+	if (!srv_file_per_table || file_format < DICT_TF_FORMAT_ZIP) {
+
+		goto key_block_size_wrong;
+	}
+
+key_block_size_ok:
+
+	if (UNIV_EXPECT(flags & DICT_TF_COMPACT, DICT_TF_COMPACT)) {
+		/* New formats are only available if ROW_FORMAT=COMPACT. */
+		flags |= file_format << DICT_TF_FORMAT_SHIFT;
 	}
+
+	if ((create_info->used_fields & HA_CREATE_USED_ROW_FORMAT)
+	    && form->s->row_type != ((flags & DICT_TF_COMPACT)
+				     ? ROW_TYPE_COMPACT
+				     : ROW_TYPE_REDUNDANT)) {
+
+		my_error(ER_ILLEGAL_HA_CREATE_OPTION, MYF(0),
+			 innobase_hton_name, "ROW_FORMAT");
+		error = -1;
+		goto cleanup;
 	}
 
 	error = create_table_def(trx, form, norm_name,
@@ -8323,6 +8373,11 @@ static MYSQL_SYSVAR_BOOL(file_per_table, srv_file_per_table,
   "Stores each InnoDB table to an .ibd file in the database dir.",
   NULL, NULL, FALSE);
 
+static MYSQL_SYSVAR_UINT(file_format, srv_file_format,
+  PLUGIN_VAR_RQCMDARG,
+  "File format to use for new tables in .ibd files.",
+  NULL, NULL, DICT_TF_FORMAT_51, DICT_TF_FORMAT_51, DICT_TF_FORMAT_MAX, 0);
+
 static MYSQL_SYSVAR_ULONG(flush_log_at_trx_commit, srv_flush_log_at_trx_commit,
   PLUGIN_VAR_OPCMDARG,
   "Set to 0 (write and flush once per second),"
@@ -8500,6 +8555,7 @@ static struct st_mysql_sys_var* innobase_system_variables[]= {
   MYSQL_SYSVAR(fast_shutdown),
   MYSQL_SYSVAR(file_io_threads),
   MYSQL_SYSVAR(file_per_table),
+  MYSQL_SYSVAR(file_format),
   MYSQL_SYSVAR(flush_log_at_trx_commit),
   MYSQL_SYSVAR(flush_method),
   MYSQL_SYSVAR(force_recovery),

ibuf/ibuf0ibuf.c

@@ -1718,6 +1718,7 @@ ibuf_add_free_page(
 {
 	mtr_t	mtr;
 	page_t*	header_page;
+	ulint	flags;
 	ulint	zip_size;
 	ulint	page_no;
 	page_t*	page;
@@ -1730,7 +1731,8 @@ ibuf_add_free_page(
 
 	/* Acquire the fsp latch before the ibuf header, obeying the latching
 	order */
-	mtr_x_lock(fil_space_get_latch(space, &zip_size), &mtr);
+	mtr_x_lock(fil_space_get_latch(space, &flags), &mtr);
+	zip_size = dict_table_flags_to_zip_size(flags);
 
 	header_page = ibuf_header_page_get(space, &mtr);
 
@@ -1807,6 +1809,7 @@ ibuf_remove_free_page(
 	mtr_t	mtr;
 	mtr_t	mtr2;
 	page_t*	header_page;
+	ulint	flags;
 	ulint	zip_size;
 	ulint	page_no;
 	page_t*	page;
@@ -1819,7 +1822,8 @@ ibuf_remove_free_page(
 
 	/* Acquire the fsp latch before the ibuf header, obeying the latching
 	order */
-	mtr_x_lock(fil_space_get_latch(space, &zip_size), &mtr);
+	mtr_x_lock(fil_space_get_latch(space, &flags), &mtr);
+	zip_size = dict_table_flags_to_zip_size(flags);
 
 	header_page = ibuf_header_page_get(space, &mtr);
 

include/db0err.h

@@ -74,8 +74,6 @@ enum db_err {
 
 	DB_PRIMARY_KEY_IS_NULL,		/* a column in the PRIMARY KEY
 					was found to be NULL */
-	DB_TABLE_ZIP_NO_IBD,		/* trying to create a compressed
-					table in the system tablespace */
 
 	/* The following are partial failure codes */
 	DB_FAIL = 1000,

include/dict0dict.h

@@ -631,6 +631,32 @@ dict_table_is_comp(
 					compact page format */
 	const dict_table_t*	table);	/* in: table */
 /************************************************************************
+Determine the file format of a table. */
+UNIV_INLINE
+ulint
+dict_table_get_format(
+/*==================*/
+					/* out: file format version */
+	const dict_table_t*	table);	/* in: table */
+/************************************************************************
+Set the file format of a table. */
+UNIV_INLINE
+void
+dict_table_set_format(
+/*==================*/
+	dict_table_t*	table,	/* in/out: table */
+	ulint		format);/* in: file format version */
+/************************************************************************
+Extract the compressed page size from table flags. */
+UNIV_INLINE
+ulint
+dict_table_flags_to_zip_size(
+/*=========================*/
+			/* out: compressed page size,
+			or 0 if not compressed */
+	ulint	flags)	/* in: flags */
+	__attribute__((const));
+/************************************************************************
 Check whether the table uses the compressed compact page format. */
 UNIV_INLINE
 ulint

include/dict0dict.ic

@@ -349,6 +349,57 @@ dict_table_is_comp(
 	return(UNIV_LIKELY(table->flags & DICT_TF_COMPACT));
 }
 
+/************************************************************************
+Determine the file format of a table. */
+UNIV_INLINE
+ulint
+dict_table_get_format(
+/*==================*/
+					/* out: file format version */
+	const dict_table_t*	table)	/* in: table */
+{
+	ut_ad(table);
+
+	return((table->flags & DICT_TF_FORMAT_MASK) >> DICT_TF_FORMAT_SHIFT);
+}
+
+/************************************************************************
+Determine the file format of a table. */
+UNIV_INLINE
+void
+dict_table_set_format(
+/*==================*/
+	dict_table_t*	table,	/* in/out: table */
+	ulint		format)	/* in: file format version */
+{
+	ut_ad(table);
+
+	table->flags = (table->flags & ~DICT_TF_FORMAT_MASK)
+		| (format << DICT_TF_FORMAT_SHIFT);
+}
+
+/************************************************************************
+Extract the compressed page size from table flags. */
+UNIV_INLINE
+ulint
+dict_table_flags_to_zip_size(
+/*=========================*/
+			/* out: compressed page size,
+			or 0 if not compressed */
+	ulint	flags)	/* in: flags */
+{
+	ulint	zip_size = flags & DICT_TF_ZSSIZE_MASK;
+
+	if (UNIV_UNLIKELY(zip_size)) {
+		zip_size = ((PAGE_ZIP_MIN_SIZE >> 1)
+			 << (zip_size >> DICT_TF_ZSSIZE_SHIFT));
+
+		ut_ad(zip_size <= UNIV_PAGE_SIZE);
+	}
+
+	return(zip_size);
+}
+
 /************************************************************************
 Check whether the table uses the compressed compact page format. */
 UNIV_INLINE
@@ -361,8 +412,7 @@ dict_table_zip_size(
 {
 	ut_ad(table);
 
-	return(UNIV_UNLIKELY((table->flags & DICT_TF_COMPRESSED_MASK)
-			     << (10 - DICT_TF_COMPRESSED_SHIFT)));
+	return(dict_table_flags_to_zip_size(table->flags));
 }
 
 /************************************************************************

include/dict0mem.h

@@ -40,12 +40,32 @@ combination of types */
 #define	DICT_TABLE_CLUSTER		3 /* this means that the table is
 					  really a cluster definition */
 #endif
-#define DICT_TABLE_COMPRESSED_BASE	0x8000	/* compressed tablespace */
 
-/* Table flags */
-#define DICT_TF_COMPACT			1	/* compact page format */
-#define DICT_TF_COMPRESSED_MASK		62	/* compressed page size, KiB */
-#define DICT_TF_COMPRESSED_SHIFT	1
+/* Table flags.  All unused bits must be 0. */
+#define DICT_TF_COMPACT			1	/* Compact page format.
+						This must be set for
+						new file formats
+						(later than
+						DICT_TF_FORMAT_51). */
+
+/* compressed page size (0=uncompressed, up to 15 compressed sizes) */
+#define DICT_TF_ZSSIZE_SHIFT		1
+#define DICT_TF_ZSSIZE_MASK		(15 << DICT_TF_ZSSIZE_SHIFT)
+#define DICT_TF_ZSSIZE_MAX (UNIV_PAGE_SIZE_SHIFT - PAGE_ZIP_MIN_SIZE_SHIFT + 1)
+
+
+#define DICT_TF_FORMAT_SHIFT		5	/* file format */
+#define DICT_TF_FORMAT_MASK		(127 << DICT_TF_FORMAT_SHIFT)
+#define DICT_TF_FORMAT_51		0	/* InnoDB/MySQL up to 5.1 */
+#define DICT_TF_FORMAT_ZIP		1	/* InnoDB plugin for 5.1:
+						compressed tables,
+						new BLOB treatment */
+#define DICT_TF_FORMAT_MAX		DICT_TF_FORMAT_ZIP
+
+#define DICT_TF_BITS			6	/* number of flag bits */
+#if (1 << (DICT_TF_BITS - DICT_TF_FORMAT_SHIFT)) <= DICT_TF_FORMAT_MAX
+# error "DICT_TF_BITS is insufficient for DICT_TF_FORMAT_MAX"
+#endif
 
 /**************************************************************************
 Creates a table memory object. */
@@ -314,6 +334,7 @@ struct dict_table_struct{
 	unsigned	space:32;
 				/* space where the clustered index of the
 				table is placed */
+	unsigned	flags:DICT_TF_BITS;/* DICT_TF_COMPACT, ... */
 	unsigned	ibd_file_missing:1;
 				/* TRUE if this is in a single-table
 				tablespace and the .ibd file is missing; then
@@ -326,7 +347,6 @@ struct dict_table_struct{
 				TABLESPACE */
 	unsigned	cached:1;/* TRUE if the table object has been added
 				to the dictionary cache */
-	unsigned	flags:8;/* DICT_TF_COMPACT, ... */
 	unsigned	n_def:10;/* number of columns defined so far */
 	unsigned	n_cols:10;/* number of columns */
 	dict_col_t*	cols;	/* array of column descriptions */

include/fil0fil.h

@@ -222,6 +222,15 @@ fil_space_get_size(
 			/* out: space size, 0 if space not found */
 	ulint	id);	/* in: space id */
 /***********************************************************************
+Returns the flags of the space. The tablespace must be cached
+in the memory cache. */
+UNIV_INTERN
+ulint
+fil_space_get_flags(
+/*================*/
+			/* out: flags, ULINT_UNDEFINED if space not found */
+	ulint	id);	/* in: space id */
+/***********************************************************************
 Returns the compressed page size of the space, or 0 if the space
 is not compressed. The tablespace must be cached in the memory cache. */
 UNIV_INTERN
@@ -413,8 +422,8 @@ fil_create_new_single_table_tablespace(
 					table */
 	ibool		is_temp,	/* in: TRUE if a table created with
 					CREATE TEMPORARY TABLE */
-	ulint		zip_size,	/* in: compressed page size,
-					or 0 if uncompressed tablespace */
+	ulint		flags,		/* in: compressed page size and
+					file format version, or 0 */
 	ulint		size);		/* in: the initial size of the
 					tablespace file in pages,
 					must be >= FIL_IBD_FILE_INITIAL_SIZE */
@@ -440,8 +449,7 @@ fil_open_single_table_tablespace(
 					faster (the OS caches them) than
 					accessing the first page of the file */
 	ulint		id,		/* in: space id */
-	ulint		zip_size,	/* in: compressed page size,
-					or 0 if uncompressed tablespace */
+	ulint		flags,		/* in: tablespace flags */
 	const char*	name);		/* in: table name in the
 					databasename/tablename format */
 /************************************************************************

include/fsp0fsp.h

@@ -82,6 +82,14 @@ fsp_header_get_space_id(
 				/* out: space id, ULINT UNDEFINED if error */
 	const page_t*	page);	/* in: first page of a tablespace */
 /**************************************************************************
+Reads the space flags from the first page of a tablespace. */
+UNIV_INTERN
+ulint
+fsp_header_get_flags(
+/*=================*/
+				/* out: flags */
+	const page_t*	page);	/* in: first page of a tablespace */
+/**************************************************************************
 Reads the compressed page size from the first page of a tablespace. */
 UNIV_INTERN
 ulint
@@ -100,8 +108,8 @@ fsp_header_init_fields(
 /*===================*/
 	page_t*	page,		/* in/out: first page in the space */
 	ulint	space_id,	/* in: space id */
-	ulint	zip_size);	/* in: compressed page size in bytes;
-				0 for uncompressed pages */
+	ulint	flags);		/* in: compressed page size and
+				file format version, or 0 */
 /**************************************************************************
 Initializes the space header of a new created space and creates also the
 insert buffer tree root if space == 0. */

include/mtr0mtr.h

@@ -130,8 +130,8 @@ flag value must give the length also! */
 						/* copy compact record list end
 						to a new created index page */
 #define MLOG_COMP_PAGE_REORGANIZE ((byte)46)	/* reorganize an index page */
-#define MLOG_ZIP_FILE_CREATE	((byte)47)	/* log record about creating a
-						compressed .ibd file */
+#define MLOG_FILE_CREATE2	((byte)47)	/* log record about creating
+						an .ibd file, with format */
 #define MLOG_ZIP_WRITE_NODE_PTR	((byte)48)	/* write the node pointer of
 						a record on a compressed
 						non-leaf B-tree page */

include/page0types.h

@@ -46,7 +46,8 @@ struct page_zip_des_struct
 					PAGE_ZIP_MIN_SIZE << (ssize - 1). */
 };
 
-#define PAGE_ZIP_MIN_SIZE	1024	/* smallest page_zip_des_struct.size */
+#define PAGE_ZIP_MIN_SIZE_SHIFT	10	/* log2 of smallest compressed size */
+#define PAGE_ZIP_MIN_SIZE	(1 << PAGE_ZIP_MIN_SIZE_SHIFT)
 
 /** Number of page compressions, indexed by page_zip_des_t::ssize */
 extern ulint	page_zip_compress_count[8];

include/page0zip.ic

@@ -107,7 +107,7 @@ page_zip_get_size(
 		return(0);
 	}
 
-	size = 512 << page_zip->ssize;
+	size = (PAGE_ZIP_MIN_SIZE >> 1) << page_zip->ssize;
 
 	ut_ad(size >= PAGE_ZIP_MIN_SIZE);
 	ut_ad(size <= UNIV_PAGE_SIZE);

include/srv0srv.h

@@ -60,6 +60,8 @@ extern char*	srv_arch_dir;
 /* store to its own file each table created by an user; data
 dictionary tables are in the system tablespace 0 */
 extern my_bool	srv_file_per_table;
+/* The file format to use on new *.ibd files. */
+extern uint	srv_file_format;
 /* Place locks to records only i.e. do not use next-key locking except
 on duplicate key checking and foreign key checking */
 extern ibool	srv_locks_unsafe_for_binlog;

log/log0recv.c

@@ -962,7 +962,7 @@ recv_parse_or_apply_log_rec_body(
 	case MLOG_FILE_CREATE:
 	case MLOG_FILE_RENAME:
 	case MLOG_FILE_DELETE:
-	case MLOG_ZIP_FILE_CREATE:
+	case MLOG_FILE_CREATE2:
 		ptr = fil_op_log_parse_or_replay(ptr, end_ptr, type, 0);
 		break;
 	case MLOG_ZIP_WRITE_NODE_PTR:
@@ -1990,7 +1990,7 @@ loop:
 #endif/* UNIV_LOG_DEBUG */
 
 		} else if (type == MLOG_FILE_CREATE
-			   || type == MLOG_ZIP_FILE_CREATE
+			   || type == MLOG_FILE_CREATE2
 			   || type == MLOG_FILE_RENAME
 			   || type == MLOG_FILE_DELETE) {
 			ut_a(space);

mysql-test/innodb-zip.result

-set global innodb_file_per_table=1;
+set global innodb_file_per_table=off;
+set global innodb_file_format=0;
+create table t1(a int primary key) engine=innodb row_format=dynamic;
+ERROR HY000: Table storage engine 'InnoDB' does not support the create option 'ROW_FORMAT'
+create table t1(a int primary key) engine=innodb row_format=redundant;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) NOT NULL,
+  PRIMARY KEY (`a`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 ROW_FORMAT=REDUNDANT
+drop table t1;
+create table t1(a int primary key) engine=innodb row_format=compact;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) NOT NULL,
+  PRIMARY KEY (`a`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 ROW_FORMAT=COMPACT
+drop table t1;
+create table t1(a int primary key) engine=innodb key_block_size=9;
+ERROR HY000: Table storage engine 'InnoDB' does not support the create option 'KEY_BLOCK_SIZE'
+create table t1(a int primary key) engine=innodb
+key_block_size=1 row_format=redundant;
+ERROR HY000: Table storage engine 'InnoDB' does not support the create option 'KEY_BLOCK_SIZE'
+set global innodb_file_per_table=on;
+create table t1(a int primary key) engine=innodb
+key_block_size=1 row_format=redundant;
+ERROR HY000: Table storage engine 'InnoDB' does not support the create option 'KEY_BLOCK_SIZE'
+set global innodb_file_format=1;
+create table t1(a int primary key) engine=innodb
+key_block_size=1 row_format=redundant;
+ERROR HY000: Table storage engine 'InnoDB' does not support the create option 'ROW_FORMAT'
+create table t1(a int primary key) engine=innodb
+key_block_size=1 row_format=compact;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) NOT NULL,
+  PRIMARY KEY (`a`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 ROW_FORMAT=COMPACT KEY_BLOCK_SIZE=1
+drop table t1;
+create table t1(a int primary key) engine=innodb
+key_block_size=1;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) NOT NULL,
+  PRIMARY KEY (`a`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 KEY_BLOCK_SIZE=1
+drop table t1;
+create table t1(a int primary key) engine=innodb key_block_size=9;
+ERROR HY000: Table storage engine 'InnoDB' does not support the create option 'KEY_BLOCK_SIZE'
 create table t1(a int not null, b text, index(b(10))) engine=innodb
 key_block_size=1;
 insert into t1 values (1,1);
@@ -14,3 +66,4 @@ a	left(b,40)	is_equal
 1	1	1
 drop table t1;
 set global innodb_file_per_table=0;
+set global innodb_file_format=0;

mysql-test/innodb-zip.test

 -- source include/have_innodb.inc
 
-let $default=`select @@innodb_file_per_table`;
-set global innodb_file_per_table=1;
+let $per_table=`select @@innodb_file_per_table`;
+let $format=`select @@innodb_file_format`;
+set global innodb_file_per_table=off;
+set global innodb_file_format=0;
+--error 1478
+create table t1(a int primary key) engine=innodb row_format=dynamic;
+create table t1(a int primary key) engine=innodb row_format=redundant;
+show create table t1;
+drop table t1;
+create table t1(a int primary key) engine=innodb row_format=compact;
+show create table t1;
+drop table t1;
+--error 1478
+create table t1(a int primary key) engine=innodb key_block_size=9;
+
+--error 1478
+create table t1(a int primary key) engine=innodb
+key_block_size=1 row_format=redundant;
+set global innodb_file_per_table=on;
+--error 1478
+create table t1(a int primary key) engine=innodb
+key_block_size=1 row_format=redundant;
+set global innodb_file_format=1;
+--error 1478
+create table t1(a int primary key) engine=innodb
+key_block_size=1 row_format=redundant;
+
+create table t1(a int primary key) engine=innodb
+key_block_size=1 row_format=compact;
+show create table t1;
+drop table t1;
+
+create table t1(a int primary key) engine=innodb
+key_block_size=1;
+show create table t1;
+drop table t1;
+
+--error 1478
+create table t1(a int primary key) engine=innodb key_block_size=9;
+
 create table t1(a int not null, b text, index(b(10))) engine=innodb
 key_block_size=1;
 
@@ -31,4 +69,5 @@ disconnect a;
 disconnect b;
 
 drop table t1;
-eval set global innodb_file_per_table=$default;
+eval set global innodb_file_per_table=$per_table;
+eval set global innodb_file_format=$format;

row/row0mysql.c

@@ -2559,8 +2559,7 @@ row_import_tablespace_for_mysql(
 	ibuf_delete_for_discarded_space(table->space);
 
 	success = fil_open_single_table_tablespace(TRUE, table->space,
-						   dict_table_zip_size(table),
-						   table->name);
+						   table->flags, table->name);
 	if (success) {
 		table->ibd_file_missing = FALSE;
 		table->tablespace_discarded = FALSE;

srv/srv0srv.c

@@ -88,9 +88,15 @@ UNIV_INTERN char*	srv_arch_dir	= NULL;
 /* store to its own file each table created by an user; data
 dictionary tables are in the system tablespace 0 */
 UNIV_INTERN my_bool	srv_file_per_table;
+/* The file format to use on new *.ibd files. */
+UNIV_INTERN uint	srv_file_format;
+#if DICT_TF_FORMAT_51
+# error "DICT_TF_FORMAT_51 must be 0!"
+#endif
 /* Place locks to records only i.e. do not use next-key locking except
 on duplicate key checking and foreign key checking */
 UNIV_INTERN ibool	srv_locks_unsafe_for_binlog = FALSE;
+
 UNIV_INTERN ulint	srv_n_data_files = 0;
 UNIV_INTERN char**	srv_data_file_names = NULL;
 /* size in database pages */

srv/srv0start.c

@@ -665,7 +665,7 @@ open_or_create_log_file(
 	if (k == 0 && i == 0) {
 		arch_space_id = 2 * k + 1 + SRV_LOG_SPACE_FIRST_ID;
 
-		fil_space_create("arch_log_space", arch_space_id, FIL_LOG);
+		fil_space_create("arch_log_space", arch_space_id, 0, FIL_LOG);
 	} else {
 		arch_space_id = ULINT_UNDEFINED;
 	}

trx/trx0rseg.c

@@ -243,11 +243,13 @@ trx_rseg_create(
 	ulint*	id,		/* out: rseg id */
 	mtr_t*	mtr)		/* in: mtr */
 {
+	ulint		flags;
 	ulint		zip_size;
 	ulint		page_no;
 	trx_rseg_t*	rseg;
 
-	mtr_x_lock(fil_space_get_latch(space, &zip_size), mtr);
+	mtr_x_lock(fil_space_get_latch(space, &flags), mtr);
+	zip_size = dict_table_flags_to_zip_size(flags);
 	mutex_enter(&kernel_mutex);
 
 	page_no = trx_rseg_header_create(space, zip_size, max_size, id, mtr);