MDEV-12634: Uninitialised ROW_MERGE_RESERVE_SIZE bytes written to tem…

…porary file

Fixed by removing writing key version to start of every block that
was encrypted. Instead we will use single key version from log_sys
crypt info.

After this MDEV also blocks writen to row log are encrypted and blocks
read from row log aren decrypted if encryption is configured for the
table.

innodb_status_variables[], struct srv_stats_t
	Added status variables for merge block and row log block
	encryption and decryption amounts.

Removed ROW_MERGE_RESERVE_SIZE define.

row_merge_fts_doc_tokenize
	Remove ROW_MERGE_RESERVE_SIZE

row_log_t
	Add index, crypt_tail, crypt_head to be used in case of
	encryption.

row_log_online_op, row_log_table_close_func
	Before writing a block encrypt it if encryption is enabled

row_log_table_apply_ops, row_log_apply_ops
	After reading a block decrypt it if encryption is enabled

row_log_allocate
	Allocate temporary buffers crypt_head and crypt_tail
	if needed.

row_log_free
	Free temporary buffers crypt_head and crypt_tail if they
	exist.

row_merge_encrypt_buf, row_merge_decrypt_buf
	Removed.

row_merge_buf_create, row_merge_buf_write
	Remove ROW_MERGE_RESERVE_SIZE

row_merge_build_indexes
	Allocate temporary buffer used in decryption and encryption
	if needed.

log_tmp_blocks_crypt, log_tmp_block_encrypt, log_temp_block_decrypt
	New functions used in block encryption and decryption

log_tmp_is_encrypted
	New function to check is encryption enabled.

Added test case innodb-rowlog to force creating a row log and
verify that operations are done using introduced status
variables.

mysql-test/suite/encryption/r/innodb-rowlog.result

+SET GLOBAL innodb_file_format = `Barracuda`;
+create table t1(id int not null primary key auto_increment, credit_card varchar(200), private varchar(50)) engine=innodb;
+SET SESSION debug_dbug="+d,ib_merge_wait_after_read";
+alter table t1 add index secret (credit_card), ALGORITHM=INPLACE, LOCK=NONE;
+SET GLOBAL innodb_encryption_rotate_key_age = 1;
+create table t2(id int) engine=innodb;
+SET SESSION debug_dbug="+d,ib_merge_wait_after_read";
+alter table t1 add index secret2 (private), ALGORITHM=INPLACE, LOCK=NONE;
+insert into t1(credit_card) select credit_card from t1;
+insert into t1(credit_card) select credit_card from t1;
+insert into t1(credit_card) select credit_card from t1;
+drop table t2;
+SELECT variable_value > 0 FROM information_schema.global_status WHERE variable_name = 'innodb_num_pages_encrypted';
+variable_value > 0
+1
+SELECT variable_value > 0 FROM information_schema.global_status WHERE variable_name = 'innodb_encryption_n_merge_blocks_encrypted';
+variable_value > 0
+1
+SELECT variable_value > 0 FROM information_schema.global_status WHERE variable_name = 'innodb_encryption_n_merge_blocks_decrypted';
+variable_value > 0
+1
+SELECT variable_value > 0 FROM information_schema.global_status WHERE variable_name = 'innodb_encryption_n_rowlog_blocks_encrypted';
+variable_value > 0
+1
+SELECT variable_value > 0 FROM information_schema.global_status WHERE variable_name = 'innodb_encryption_n_rowlog_blocks_decrypted';
+variable_value > 0
+1
+drop table t1;
+SET GLOBAL innodb_file_format=Antelope;
+SET GLOBAL innodb_encryption_rotate_key_age=15;

mysql-test/suite/encryption/t/innodb-rowlog.opt

+--aria-encrypt-tables
+--encrypt-tmp-disk-tables
+--innodb-encrypt-tables
+--innodb-encrypt-log
+--innodb-encryption-rotate-key-age=15
+--innodb-encryption-threads=4
+--innodb-tablespaces-encryption

mysql-test/suite/encryption/t/innodb-rowlog.test

+-- source include/have_innodb.inc
+-- source include/have_example_key_management_plugin.inc
+# needs dbug_dbug
+-- source include/have_debug.inc
+
+--disable_warnings
+--disable_query_log
+let $innodb_file_format_orig = `SELECT @@innodb_file_format`;
+let $key_age = `SELECT @@innodb_encryption_rotate_key_age`;
+--enable_query_log
+
+SET GLOBAL innodb_file_format = `Barracuda`;
+--enable_warnings
+
+let $MYSQLD_DATADIR = `SELECT @@datadir`;
+let $MYSQLD_TMPDIR = `SELECT @@tmpdir`;
+
+#
+# Create a table that will be encrypted and put some sensitive data to it (credit card numbers)
+#
+
+create table t1(id int not null primary key auto_increment, credit_card varchar(200), private varchar(50)) engine=innodb;
+
+let $rows = 15000;
+--disable_query_log
+begin;
+while ($rows)
+{
+    eval insert into t1 values(NULL, '0000-0000-0000-0000','private_data');
+    dec $rows;
+}
+commit;
+--enable_query_log
+
+--let $wait_timeout= 600
+--let $wait_condition=SELECT COUNT(*) = 0 FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0
+--source include/wait_condition.inc
+
+#
+# Now we create secondary index for credit_card column in parallel we create new rows
+# forcing alter table to wait so that row log is used.
+#
+SET SESSION debug_dbug="+d,ib_merge_wait_after_read";
+send alter table t1 add index secret (credit_card), ALGORITHM=INPLACE, LOCK=NONE;
+#
+# Force key rotation and create second index for same table
+#
+connect (con2,localhost,root,,);
+SET GLOBAL innodb_encryption_rotate_key_age = 1;
+create table t2(id int) engine=innodb;
+SET SESSION debug_dbug="+d,ib_merge_wait_after_read";
+send alter table t1 add index secret2 (private), ALGORITHM=INPLACE, LOCK=NONE;
+#
+# Create new rows to row log
+#
+connect (con1,localhost,root,,);
+insert into t1(credit_card) select credit_card from t1;
+insert into t1(credit_card) select credit_card from t1;
+insert into t1(credit_card) select credit_card from t1;
+
+connection default;
+reap;
+disconnect con1;
+
+connection con2;
+reap;
+drop table t2;
+
+connection default;
+disconnect con2;
+
+let $wait_condition= select variable_value > 0 from information_schema.global_status where variable_name = 'INNODB_NUM_PAGES_ENCRYPTED';
+--source include/wait_condition.inc
+
+#
+# Verify that both merge blocks and row log blocks are encryted and decrypted
+#
+
+SELECT variable_value > 0 FROM information_schema.global_status WHERE variable_name = 'innodb_num_pages_encrypted';
+SELECT variable_value > 0 FROM information_schema.global_status WHERE variable_name = 'innodb_encryption_n_merge_blocks_encrypted';
+SELECT variable_value > 0 FROM information_schema.global_status WHERE variable_name = 'innodb_encryption_n_merge_blocks_decrypted';
+SELECT variable_value > 0 FROM information_schema.global_status WHERE variable_name = 'innodb_encryption_n_rowlog_blocks_encrypted';
+SELECT variable_value > 0 FROM information_schema.global_status WHERE variable_name = 'innodb_encryption_n_rowlog_blocks_decrypted';
+
+drop table t1;
+
+--disable_warnings
+eval SET GLOBAL innodb_file_format=$innodb_file_format_orig;
+eval SET GLOBAL innodb_encryption_rotate_key_age=$key_age;
+--enable_warnings

storage/innobase/handler/ha_innodb.cc

@@ -1087,6 +1087,18 @@ static SHOW_VAR innodb_status_variables[]= {
   {"encryption_key_rotation_list_length",
   (char*)&export_vars.innodb_key_rotation_list_length,
    SHOW_LONGLONG},
+  {"encryption_n_merge_blocks_encrypted",
+  (char*)&export_vars.innodb_n_merge_blocks_encrypted,
+   SHOW_LONGLONG},
+  {"encryption_n_merge_blocks_decrypted",
+  (char*)&export_vars.innodb_n_merge_blocks_decrypted,
+   SHOW_LONGLONG},
+  {"encryption_n_rowlog_blocks_encrypted",
+  (char*)&export_vars.innodb_n_rowlog_blocks_encrypted,
+   SHOW_LONGLONG},
+  {"encryption_n_rowlog_blocks_decrypted",
+  (char*)&export_vars.innodb_n_rowlog_blocks_decrypted,
+   SHOW_LONGLONG},
 
   /* scrubing */
   {"scrub_background_page_reorganizations",

storage/innobase/include/log0crypt.h

 /*****************************************************************************
 
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (C) 2014, 2016, MariaDB Corporation. All Rights Reserved.
+Copyright (C) 2014, 2017, MariaDB Corporation. All Rights Reserved.
 
 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
@@ -29,6 +29,7 @@ Modified           Jan Lindström jan.lindstrom@mariadb.com
 #include "univ.i"
 #include "ut0byte.h"
 #include "my_crypt.h"
+#include "os0file.h"
 
 typedef int Crypt_result;
 
@@ -125,4 +126,45 @@ log_crypt_print_checkpoint_keys(
 /*============================*/
 	const byte* log_block);
 
+/** Encrypt temporary log block.
+@param[in]	src_block	block to encrypt or decrypt
+@param[in]	size		size of the block
+@param[out]	dst_block	destination block
+@param[in]	offs		offset to block
+@param[in]	space_id	tablespace id
+@return true if successfull, false in case of failure
+*/
+UNIV_INTERN
+bool
+log_tmp_block_encrypt(
+	const byte*		src_block,
+	ulint			size,
+	byte*			dst_block,
+	os_offset_t		offs,
+	ulint			space_id)
+	MY_ATTRIBUTE((warn_unused_result));
+
+/** Decrypt temporary log block.
+@param[in]	src_block	block to encrypt or decrypt
+@param[in]	size		size of the block
+@param[out]	dst_block	destination block
+@param[in]	offs		offset to block
+@param[in]	space_id	tablespace id
+@return true if successfull, false in case of failure
+*/
+UNIV_INTERN
+bool
+log_tmp_block_decrypt(
+	const byte*		src_block,
+	ulint			size,
+	byte*			dst_block,
+	os_offset_t		offs,
+	ulint			space_id)
+	MY_ATTRIBUTE((warn_unused_result));
+
+/** Find out is temporary log files encrypted.
+@return true if temporary log file should be encrypted, false if not */
+UNIV_INTERN
+bool
+log_tmp_is_encrypted() MY_ATTRIBUTE((warn_unused_result));
 #endif  // log0crypt.h

storage/innobase/include/row0ftsort.h

 /*****************************************************************************
 
 Copyright (c) 2010, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2016, MariaDB Corporation.
+Copyright (c) 2015, 2017, MariaDB Corporation.
 
 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
@@ -72,7 +72,6 @@ struct fts_psort_common_t {
 						store Doc ID during sort, if
 						Doc ID will not be big enough
 						to use 8 bytes value */
-	fil_space_crypt_t*	crypt_data;	/*!< crypt data or NULL */
 };
 
 struct fts_psort_t {

storage/innobase/include/row0merge.h

 /*****************************************************************************
 
 Copyright (c) 2005, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2016, MariaDB Corporation.
+Copyright (c) 2015, 2017, MariaDB Corporation.
 
 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
@@ -41,9 +41,6 @@ Created 13/06/2005 Jan Lindstrom
 #include "lock0types.h"
 #include "srv0srv.h"
 
-/* Reserve free space from every block for key_version */
-#define ROW_MERGE_RESERVE_SIZE 4
-
 /* Cluster index read task is mandatory */
 #define COST_READ_CLUSTERED_INDEX            1.0
 
@@ -352,17 +349,16 @@ row_merge_buf_sort(
 Write a merge block to the file system.
 @return TRUE if request was successful, FALSE if fail */
 UNIV_INTERN
-ibool
+bool
 row_merge_write(
 /*============*/
 	int		fd,	/*!< in: file descriptor */
 	ulint		offset,	/*!< in: offset where to write,
 				in number of row_merge_block_t elements */
 	const void*	buf,	/*!< in: data */
-	fil_space_crypt_t*	crypt_data,	/*!< in: table crypt data */
 	void*		crypt_buf,		/*!< in: crypt buf or NULL */
-	ulint		space);			/*!< in: space id */
-
+	ulint		space)			/*!< in: space id */
+	MY_ATTRIBUTE((warn_unused_result));
 /********************************************************************//**
 Empty a sort buffer.
 @return sort buffer */
@@ -400,10 +396,10 @@ row_merge_sort(
 	const bool		update_progress, /*!< in: update progress status variable or not */
 	const float		pct_progress, /*!< in: total progress percent until now */
 	const float		pct_cost, /*!< in: current progress percent */
-	fil_space_crypt_t*	crypt_data,/*!< in: table crypt data */
 	row_merge_block_t*	crypt_block, /*!< in: crypt buf or NULL */
 	ulint			space)	   /*!< in: space id */
-	__attribute__((nonnull(1,2,3,4,5)));
+	MY_ATTRIBUTE((warn_unused_result));
+
 /*********************************************************************//**
 Allocate a sort buffer.
 @return own: sort buffer */
@@ -433,7 +429,7 @@ row_merge_file_destroy(
 Read a merge block from the file system.
 @return TRUE if request was successful, FALSE if fail */
 UNIV_INTERN
-ibool
+bool
 row_merge_read(
 /*===========*/
 	int			fd,	/*!< in: file descriptor */
@@ -441,9 +437,9 @@ row_merge_read(
 					in number of row_merge_block_t
 					elements */
 	row_merge_block_t*	buf,	/*!< out: data */
-	fil_space_crypt_t*	crypt_data,/*!< in: table crypt data */
 	row_merge_block_t*	crypt_buf, /*!< in: crypt buf or NULL */
-	ulint			space);	   /*!< in: space id */
+	ulint			space)	   /*!< in: space id */
+	MY_ATTRIBUTE((warn_unused_result));
 
 /********************************************************************//**
 Read a merge record.
@@ -462,8 +458,8 @@ row_merge_read_rec(
 					or NULL on end of list
 					(non-NULL on I/O error) */
 	ulint*			offsets,/*!< out: offsets of mrec */
-	fil_space_crypt_t*	crypt_data,/*!< in: table crypt data */
 	row_merge_block_t*	crypt_block, /*!< in: crypt buf or NULL */
 	ulint			space)	   /*!< in: space id */
-	__attribute__((nonnull(1,2,3,4,6,7,8), warn_unused_result));
+	MY_ATTRIBUTE((warn_unused_result));
+
 #endif /* row0merge.h */

storage/innobase/include/srv0srv.h

@@ -137,6 +137,14 @@ struct srv_stats_t {
 	ulint_ctr_64_t          pages_encrypted;
    	/* Number of pages decrypted */
 	ulint_ctr_64_t          pages_decrypted;
+	/* Number of merge blocks encrypted */
+	ulint_ctr_64_t          n_merge_blocks_encrypted;
+	/* Number of merge blocks decrypted */
+	ulint_ctr_64_t          n_merge_blocks_decrypted;
+	/* Number of row log blocks encrypted */
+	ulint_ctr_64_t          n_rowlog_blocks_encrypted;
+	/* Number of row log blocks decrypted */
+	ulint_ctr_64_t          n_rowlog_blocks_decrypted;
 
 	/** Number of data read in total (in bytes) */
 	ulint_ctr_1_t		data_read;
@@ -1033,6 +1041,15 @@ struct export_var_t{
 	ib_int64_t innodb_pages_decrypted;      /*!< Number of pages
 						decrypted */
 
+	/*!< Number of merge blocks encrypted */
+	ib_int64_t innodb_n_merge_blocks_encrypted;
+	/*!< Number of merge blocks decrypted */
+	ib_int64_t innodb_n_merge_blocks_decrypted;
+	/*!< Number of row log blocks encrypted */
+	ib_int64_t innodb_n_rowlog_blocks_encrypted;
+	/*!< Number of row log blocks decrypted */
+	ib_int64_t innodb_n_rowlog_blocks_decrypted;
+
 	ulint innodb_sec_rec_cluster_reads;	/*!< srv_sec_rec_cluster_reads */
 	ulint innodb_sec_rec_cluster_reads_avoided;/*!< srv_sec_rec_cluster_reads_avoided */
 

storage/innobase/log/log0crypt.cc

 /*****************************************************************************
 
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (C) 2014, 2016, MariaDB Corporation. All Rights Reserved.
+Copyright (C) 2014, 2017, MariaDB Corporation. All Rights Reserved.
 
 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
@@ -237,6 +237,129 @@ log_blocks_crypt(
 	return rc;
 }
 
+/** Encrypt/decrypt temporary log blocks.
+
+@param[in]	src_block	block to encrypt or decrypt
+@param[in]	size		size of the block
+@param[out]	dst_block	destination block
+@param[in]	what		ENCRYPTION_FLAG_ENCRYPT or
+				ENCRYPTION_FLAG_DECRYPT
+@param[in]	offs		offset to block
+@param[in]	space_id	tablespace id
+@return true if successfull, false in case of failure
+*/
+static
+bool
+log_tmp_blocks_crypt(
+	const byte*		src_block,
+	ulint			size,
+	byte*			dst_block,
+	int			what,
+	os_offset_t		offs,
+	ulint			space_id)
+{
+	Crypt_result rc = MY_AES_OK;
+	uint dst_len;
+	byte aes_ctr_counter[MY_AES_BLOCK_SIZE];
+	byte is_encrypt= what == ENCRYPTION_FLAG_ENCRYPT;
+	const crypt_info_t* info = static_cast<const crypt_info_t*>(&crypt_info[0]);
+
+	// AES_CTR_COUNTER = space_id + offs
+
+	bzero(aes_ctr_counter, MY_AES_BLOCK_SIZE);
+	mach_write_to_8(aes_ctr_counter, space_id);
+	mach_write_to_8(aes_ctr_counter + 8, offs);
+
+	rc = encryption_crypt(src_block, size,
+				dst_block, &dst_len,
+				(unsigned char*)(info->crypt_key), 16,
+				aes_ctr_counter, MY_AES_BLOCK_SIZE,
+				what | ENCRYPTION_FLAG_NOPAD,
+				LOG_DEFAULT_ENCRYPTION_KEY,
+				info->key_version);
+
+	if (rc != MY_AES_OK) {
+		ib_logf(IB_LOG_LEVEL_ERROR,
+			"%s failed for temporary log file with rc = %d",
+			is_encrypt ? "Encryption" : "Decryption",
+			rc);
+		return false;
+	}
+
+	return true;
+}
+
+/** Get crypt info
+@return pointer to log crypt info or NULL
+*/
+inline
+const crypt_info_t*
+get_crypt_info()
+{
+	mutex_enter(&log_sys->mutex);
+	const crypt_info_t* info = get_crypt_info(log_sys->next_checkpoint_no);
+	mutex_exit(&log_sys->mutex);
+
+	return info;
+}
+
+/** Find out is temporary log files encrypted.
+@return true if temporary log file should be encrypted, false if not */
+UNIV_INTERN
+bool
+log_tmp_is_encrypted()
+{
+	const crypt_info_t* info = get_crypt_info();
+
+	if (info == NULL || info->key_version == UNENCRYPTED_KEY_VER) {
+		return false;
+	}
+
+	return true;
+}
+
+/** Encrypt temporary log block.
+@param[in]	src_block	block to encrypt or decrypt
+@param[in]	size		size of the block
+@param[out]	dst_block	destination block
+@param[in]	offs		offset to block
+@param[in]	space_id	tablespace id
+@return true if successfull, false in case of failure
+*/
+UNIV_INTERN
+bool
+log_tmp_block_encrypt(
+	const byte*		src_block,
+	ulint			size,
+	byte*			dst_block,
+	os_offset_t		offs,
+	ulint			space_id)
+{
+	return (log_tmp_blocks_crypt(src_block, size, dst_block,
+				ENCRYPTION_FLAG_ENCRYPT, offs, space_id));
+}
+
+/** Decrypt temporary log block.
+@param[in]	src_block	block to encrypt or decrypt
+@param[in]	size		size of the block
+@param[out]	dst_block	destination block
+@param[in]	offs		offset to block
+@param[in]	space_id	tablespace id
+@return true if successfull, false in case of failure
+*/
+UNIV_INTERN
+bool
+log_tmp_block_decrypt(
+	const byte*		src_block,
+	ulint			size,
+	byte*			dst_block,
+	os_offset_t		offs,
+	ulint			space_id)
+{
+	return (log_tmp_blocks_crypt(src_block, size, dst_block,
+				     ENCRYPTION_FLAG_DECRYPT, offs, space_id));
+}
+
 /*********************************************************************//**
 Generate crypt key from crypt msg.
 @return true if successfull, false if not. */

storage/innobase/row/row0ftsort.cc

@@ -39,7 +39,7 @@ Created 10/13/2010 Jimmy Yang
 		b[N] = row_merge_read_rec(				\
 			block[N], buf[N], b[N], index,			\
 			fd[N], &foffs[N], &mrec[N], offsets[N],		\
-			crypt_data, crypt_block[N], space);		\
+			crypt_block[N], space);				\
 		if (UNIV_UNLIKELY(!b[N])) {				\
 			if (mrec[N]) {					\
 				goto exit;				\
@@ -191,7 +191,6 @@ row_fts_psort_info_init(
 	fts_psort_t*		merge_info = NULL;
 	ulint			block_size;
 	ibool			ret = TRUE;
-	fil_space_crypt_t*	crypt_data = NULL;
 	bool			encrypted = false;
 
 	block_size = 3 * srv_sort_buf_size;
@@ -222,21 +221,8 @@ row_fts_psort_info_init(
 	common_info->merge_event = os_event_create();
 	common_info->opt_doc_id_size = opt_doc_id_size;
 
-	/* Theoretically the tablespace can be dropped straight away.
-	In practice, the DDL completion will wait for this thread to
-	finish. */
-	if (fil_space_t* space = fil_space_acquire(new_table->space)) {
-		crypt_data = space->crypt_data;
-		fil_space_release(space);
-	}
-
-	if (crypt_data && crypt_data->should_encrypt()) {
-		common_info->crypt_data = crypt_data;
+	if (log_tmp_is_encrypted()) {
 		encrypted = true;
-	} else {
-		/* Not needed */
-		common_info->crypt_data = NULL;
-		crypt_data = NULL;
 	}
 
 	ut_ad(trx->mysql_thd != NULL);
@@ -574,11 +560,9 @@ row_merge_fts_doc_tokenize(
 		cur_len += len;
 		dfield_dup(field, buf->heap);
 
-		/* Reserve one byte for the end marker of row_merge_block_t
-		and we have reserved ROW_MERGE_RESERVE_SIZE (= 4) for
-		encryption key_version in the beginning of the buffer. */
+		/* Reserve one byte for the end marker of row_merge_block_t */
 		if (buf->total_size + data_size[idx] + cur_len
-			>= (srv_sort_buf_size - 1 - ROW_MERGE_RESERVE_SIZE)) {
+			>= (srv_sort_buf_size - 1)) {
 
 			buf_full = TRUE;
 			break;
@@ -672,7 +656,6 @@ fts_parallel_tokenization(
 	fts_tokenize_ctx_t	t_ctx;
 	ulint			retried = 0;
 	dberr_t			error = DB_SUCCESS;
-	fil_space_crypt_t*	crypt_data = NULL;
 
 	ut_ad(psort_info->psort_common->trx->mysql_thd != NULL);
 
@@ -693,7 +676,6 @@ fts_parallel_tokenization(
 
 	block = psort_info->merge_block;
 	crypt_block = psort_info->crypt_block;
-	crypt_data = psort_info->psort_common->crypt_data;
 	zip_size = dict_table_zip_size(table);
 
 	row_merge_fts_get_next_doc_item(psort_info, &doc_item);
@@ -788,7 +770,6 @@ fts_parallel_tokenization(
 		if (!row_merge_write(merge_file[t_ctx.buf_used]->fd,
 				     merge_file[t_ctx.buf_used]->offset++,
 				     block[t_ctx.buf_used],
-				     crypt_data,
 				     crypt_block[t_ctx.buf_used],
 				     table->space)) {
 			error = DB_TEMP_FILE_WRITE_FAILURE;
@@ -884,7 +865,6 @@ fts_parallel_tokenization(
 				if (!row_merge_write(merge_file[i]->fd,
 						merge_file[i]->offset++,
 						block[i],
-						crypt_data,
 						crypt_block[i],
 						table->space)) {
 					error = DB_TEMP_FILE_WRITE_FAILURE;
@@ -924,7 +904,7 @@ fts_parallel_tokenization(
 				       psort_info->psort_common->dup,
 				       merge_file[i], block[i], &tmpfd[i],
 				       false, 0.0/* pct_progress */, 0.0/* pct_cost */,
-				       crypt_data, crypt_block[i], table->space);
+				       crypt_block[i], table->space);
 
 		if (error != DB_SUCCESS) {
 			close(tmpfd[i]);
@@ -1436,7 +1416,6 @@ row_fts_merge_insert(
 	ulint			start;
 	fts_psort_insert_t	ins_ctx;
 	ulint			count_diag = 0;
-	fil_space_crypt_t*	crypt_data = NULL;
 	ulint			space;
 
 	ut_ad(index);
@@ -1450,7 +1429,6 @@ row_fts_merge_insert(
 	ins_ctx.trx->op_info = "inserting index entries";
 
 	ins_ctx.opt_doc_id_size = psort_info[0].psort_common->opt_doc_id_size;
-	crypt_data = psort_info[0].psort_common->crypt_data;
 
 	heap = mem_heap_create(500 + sizeof(mrec_buf_t));
 
@@ -1544,7 +1522,6 @@ row_fts_merge_insert(
 			    && (!row_merge_read(
 					fd[i], foffs[i],
 					(row_merge_block_t*) block[i],
-					crypt_data,
 					(row_merge_block_t*) crypt_block[i],
 					space))) {
 				error = DB_CORRUPTION;

storage/innobase/row/row0log.cc

@@ -182,6 +182,7 @@ struct row_log_t {
 	dict_table_t*	table;	/*!< table that is being rebuilt,
 				or NULL when this is a secondary
 				index that is being created online */
+	dict_index_t*	index;	/*!< index to be build */
 	bool		same_pk;/*!< whether the definition of the PRIMARY KEY
 				has remained the same */
 	const dtuple_t*	add_cols;
@@ -197,8 +198,14 @@ struct row_log_t {
 	row_log_buf_t	tail;	/*!< writer context;
 				protected by mutex and index->lock S-latch,
 				or by index->lock X-latch only */
+	byte*		crypt_tail; /*!< writer context;
+				temporary buffer used in encryption,
+				decryption or NULL*/
 	row_log_buf_t	head;	/*!< reader context; protected by MDL only;
 				modifiable by row_log_apply_ops() */
+	byte*		crypt_head; /*!< reader context;
+				temporary buffer used in encryption,
+				decryption or NULL */
 	const char*	path;	/*!< where to create temporary file during
 				log operation */
 };
@@ -349,6 +356,7 @@ row_log_online_op(
 			= (os_offset_t) log->tail.blocks
 			* srv_sort_buf_size;
 		ibool			ret;
+		byte * buf = log->tail.block;
 
 		if (byte_offset + srv_sort_buf_size >= srv_online_max_size) {
 			goto write_failed;
@@ -368,11 +376,29 @@ row_log_online_op(
 			goto err_exit;
 		}
 
+		/* If encryption is enabled encrypt buffer before writing it
+		to file system. */
+		if (log_tmp_is_encrypted()) {
+			if (!log_tmp_block_encrypt(log->tail.block,
+						   srv_sort_buf_size,
+						   log->crypt_tail,
+						   byte_offset,
+						   index->table->space)) {
+				log->error = DB_DECRYPTION_FAILED;
+				goto err_exit;
+			}
+
+			srv_stats.n_rowlog_blocks_encrypted.inc();
+			buf = log->crypt_tail;
+		}
+
 		ret = os_file_write_int_fd(
 			"(modification log)",
 			log->fd,
-			log->tail.block, byte_offset, srv_sort_buf_size);
+			buf, byte_offset, srv_sort_buf_size);
+
 		log->tail.blocks++;
+
 		if (!ret) {
 write_failed:
 			/* We set the flag directly instead of invoking
@@ -380,7 +406,9 @@ row_log_online_op(
 			because the index is not "public" yet. */
 			index->type |= DICT_CORRUPT;
 		}
+
 		UNIV_MEM_INVALID(log->tail.block, srv_sort_buf_size);
+
 		memcpy(log->tail.block, log->tail.buf + avail_size,
 		       mrec_size - avail_size);
 		log->tail.bytes = mrec_size - avail_size;
@@ -465,6 +493,7 @@ row_log_table_close_func(
 			= (os_offset_t) log->tail.blocks
 			* srv_sort_buf_size;
 		ibool			ret;
+		byte * buf = log->tail.block;
 
 		if (byte_offset + srv_sort_buf_size >= srv_online_max_size) {
 			goto write_failed;
@@ -484,11 +513,29 @@ row_log_table_close_func(
 			goto err_exit;
 		}
 
+		/* If encryption is enabled encrypt buffer before writing it
+		to file system. */
+		if (log_tmp_is_encrypted()) {
+			if (!log_tmp_block_encrypt(log->tail.block,
+						   srv_sort_buf_size,
+						   log->crypt_tail,
+						   byte_offset,
+						   log->index->table->space)) {
+				log->error = DB_DECRYPTION_FAILED;
+				goto err_exit;
+			}
+
+			srv_stats.n_rowlog_blocks_encrypted.inc();
+			buf = log->crypt_tail;
+		}
+
 		ret = os_file_write_int_fd(
 			"(modification log)",
 			log->fd,
-			log->tail.block, byte_offset, srv_sort_buf_size);
+			buf, byte_offset, srv_sort_buf_size);
+
 		log->tail.blocks++;
+
 		if (!ret) {
 write_failed:
 			log->error = DB_ONLINE_LOG_TOO_BIG;
@@ -2622,10 +2669,29 @@ row_log_table_apply_ops(
 			goto func_exit;
 		}
 
+		byte * buf = index->online_log->head.block;
+
 		success = os_file_read_no_error_handling_int_fd(
 			index->online_log->fd,
-			index->online_log->head.block, ofs,
+			buf, ofs,
 			srv_sort_buf_size);
+
+		/* If encryption is enabled decrypt buffer after reading it
+		from file system. */
+		if (log_tmp_is_encrypted()) {
+			if (!log_tmp_block_decrypt(buf,
+						   srv_sort_buf_size,
+						   index->online_log->crypt_head,
+						   ofs,
+						   index->table->space)) {
+				error = DB_DECRYPTION_FAILED;
+				goto func_exit;
+			}
+
+			srv_stats.n_rowlog_blocks_decrypted.inc();
+			memcpy(buf, index->online_log->crypt_head, srv_sort_buf_size);
+		}
+
 		if (!success) {
 			fprintf(stderr, "InnoDB: unable to read temporary file"
 				" for table %s\n", index->table_name);
@@ -2932,9 +2998,22 @@ row_log_allocate(
 	log->head.blocks = log->head.bytes = 0;
 	log->head.total = 0;
 	log->path = path;
+	log->crypt_tail = log->crypt_head = NULL;
+	log->index = index;
 	dict_index_set_online_status(index, ONLINE_INDEX_CREATION);
 	index->online_log = log;
 
+	if (log_tmp_is_encrypted()) {
+		ulint size = srv_sort_buf_size;
+		log->crypt_head = static_cast<byte *>(os_mem_alloc_large(&size));
+		log->crypt_tail = static_cast<byte *>(os_mem_alloc_large(&size));
+
+		if (!log->crypt_head || !log->crypt_tail) {
+			row_log_free(log);
+			DBUG_RETURN(false);
+		}
+	}
+
 	/* While we might be holding an exclusive data dictionary lock
 	here, in row_log_abort_sec() we will not always be holding it. Use
 	atomic operations in both cases. */
@@ -2957,6 +3036,15 @@ row_log_free(
 	row_log_block_free(log->tail);
 	row_log_block_free(log->head);
 	row_merge_file_destroy_low(log->fd);
+
+	if (log->crypt_head) {
+		os_mem_free_large(log->crypt_head, srv_sort_buf_size);
+	}
+
+	if (log->crypt_tail) {
+		os_mem_free_large(log->crypt_tail, srv_sort_buf_size);
+	}
+
 	mutex_free(&log->mutex);
 	ut_free(log);
 	log = 0;
@@ -3451,11 +3539,29 @@ row_log_apply_ops(
 			goto func_exit;
 		}
 
+		byte* buf = index->online_log->head.block;
+
 		success = os_file_read_no_error_handling_int_fd(
 			index->online_log->fd,
-			index->online_log->head.block, ofs,
+			buf, ofs,
 			srv_sort_buf_size);
 
+		/* If encryption is enabled decrypt buffer after reading it
+		from file system. */
+		if (log_tmp_is_encrypted()) {
+			if (!log_tmp_block_decrypt(buf,
+						   srv_sort_buf_size,
+						   index->online_log->crypt_head,
+						   ofs,
+						   index->table->space)) {
+				error = DB_DECRYPTION_FAILED;
+				goto func_exit;
+			}
+
+			srv_stats.n_rowlog_blocks_decrypted.inc();
+			memcpy(buf, index->online_log->crypt_head, srv_sort_buf_size);
+		}
+
 		if (!success) {
 			fprintf(stderr, "InnoDB: unable to read temporary file"
 				" for index %s\n", index->name + 1);

storage/innobase/srv/srv0srv.cc

@@ -1619,6 +1619,10 @@ srv_export_innodb_status(void)
 	export_vars.innodb_pages_page_compression_error = srv_stats.pages_page_compression_error;
 	export_vars.innodb_pages_decrypted = srv_stats.pages_decrypted;
 	export_vars.innodb_pages_encrypted = srv_stats.pages_encrypted;
+	export_vars.innodb_n_merge_blocks_encrypted = srv_stats.n_merge_blocks_encrypted;
+	export_vars.innodb_n_merge_blocks_decrypted = srv_stats.n_merge_blocks_decrypted;
+	export_vars.innodb_n_rowlog_blocks_encrypted = srv_stats.n_rowlog_blocks_encrypted;
+	export_vars.innodb_n_rowlog_blocks_decrypted = srv_stats.n_rowlog_blocks_decrypted;
 
 	export_vars.innodb_defragment_compression_failures =
 		btr_defragment_compression_failures;

storage/xtradb/handler/ha_innodb.cc

@@ -1309,6 +1309,18 @@ static SHOW_VAR innodb_status_variables[]= {
   {"encryption_key_rotation_list_length",
   (char*)&export_vars.innodb_key_rotation_list_length,
    SHOW_LONGLONG},
+  {"encryption_n_merge_blocks_encrypted",
+  (char*)&export_vars.innodb_n_merge_blocks_encrypted,
+   SHOW_LONGLONG},
+  {"encryption_n_merge_blocks_decrypted",
+  (char*)&export_vars.innodb_n_merge_blocks_decrypted,
+   SHOW_LONGLONG},
+  {"encryption_n_rowlog_blocks_encrypted",
+  (char*)&export_vars.innodb_n_rowlog_blocks_encrypted,
+   SHOW_LONGLONG},
+  {"encryption_n_rowlog_blocks_decrypted",
+  (char*)&export_vars.innodb_n_rowlog_blocks_decrypted,
+   SHOW_LONGLONG},
 
   /* Scrubing feature */
   {"scrub_background_page_reorganizations",

storage/xtradb/include/log0crypt.h

 /*****************************************************************************
 
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (C) 2014, 2016, MariaDB Corporation. All Rights Reserved.
+Copyright (C) 2014, 2017, MariaDB Corporation. All Rights Reserved.
 
 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
@@ -29,6 +29,7 @@ Modified           Jan Lindström jan.lindstrom@mariadb.com
 #include "univ.i"
 #include "ut0byte.h"
 #include "my_crypt.h"
+#include "os0file.h"
 
 typedef int Crypt_result;
 
@@ -125,4 +126,45 @@ log_crypt_print_checkpoint_keys(
 /*============================*/
 	const byte* log_block);
 
+/** Encrypt temporary log block.
+@param[in]	src_block	block to encrypt or decrypt
+@param[in]	size		size of the block
+@param[out]	dst_block	destination block
+@param[in]	offs		offset to block
+@param[in]	space_id	tablespace id
+@return true if successfull, false in case of failure
+*/
+UNIV_INTERN
+bool
+log_tmp_block_encrypt(
+	const byte*		src_block,
+	ulint			size,
+	byte*			dst_block,
+	os_offset_t		offs,
+	ulint			space_id)
+	MY_ATTRIBUTE((warn_unused_result));
+
+/** Decrypt temporary log block.
+@param[in]	src_block	block to encrypt or decrypt
+@param[in]	size		size of the block
+@param[out]	dst_block	destination block
+@param[in]	offs		offset to block
+@param[in]	space_id	tablespace id
+@return true if successfull, false in case of failure
+*/
+UNIV_INTERN
+bool
+log_tmp_block_decrypt(
+	const byte*		src_block,
+	ulint			size,
+	byte*			dst_block,
+	os_offset_t		offs,
+	ulint			space_id)
+	MY_ATTRIBUTE((warn_unused_result));
+
+/** Find out is temporary log files encrypted.
+@return true if temporary log file should be encrypted, false if not */
+UNIV_INTERN
+bool
+log_tmp_is_encrypted() MY_ATTRIBUTE((warn_unused_result));
 #endif  // log0crypt.h

storage/xtradb/include/row0ftsort.h

 /*****************************************************************************
 
 Copyright (c) 2010, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2016, MariaDB Corporation.
+Copyright (c) 2016, 2017, MariaDB Corporation.
 
 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
@@ -72,7 +72,6 @@ struct fts_psort_common_t {
 						store Doc ID during sort, if
 						Doc ID will not be big enough
 						to use 8 bytes value */
-	fil_space_crypt_t*	crypt_data;	/*!< crypt data or NULL */
 };
 
 struct fts_psort_t {

storage/xtradb/include/row0merge.h

 /*****************************************************************************
 
 Copyright (c) 2005, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2016, MariaDB Corporation.
+Copyright (c) 2015, 2017, MariaDB Corporation.
 
 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
@@ -41,9 +41,6 @@ Created 13/06/2005 Jan Lindstrom
 #include "lock0types.h"
 #include "srv0srv.h"
 
-/* Reserve free space from every block for key_version */
-#define ROW_MERGE_RESERVE_SIZE 4
-
 /* Cluster index read task is mandatory */
 #define COST_READ_CLUSTERED_INDEX            1.0
 
@@ -352,17 +349,16 @@ row_merge_buf_sort(
 Write a merge block to the file system.
 @return TRUE if request was successful, FALSE if fail */
 UNIV_INTERN
-ibool
+bool
 row_merge_write(
 /*============*/
 	int		fd,	/*!< in: file descriptor */
 	ulint		offset,	/*!< in: offset where to write,
 				in number of row_merge_block_t elements */
 	const void*	buf,	/*!< in: data */
-	fil_space_crypt_t*	crypt_data,	/*!< in: table crypt data */
 	void*		crypt_buf,		/*!< in: crypt buf or NULL */
-	ulint		space);			/*!< in: space id */
-
+	ulint		space)			/*!< in: space id */
+	MY_ATTRIBUTE((warn_unused_result));
 /********************************************************************//**
 Empty a sort buffer.
 @return sort buffer */
@@ -400,10 +396,9 @@ row_merge_sort(
 	const bool		update_progress, /*!< in: update progress status variable or not */
 	const float		pct_progress, /*!< in: total progress percent until now */
 	const float		pct_cost, /*!< in: current progress percent */
-	fil_space_crypt_t*	crypt_data,/*!< in: table crypt data */
 	row_merge_block_t*	crypt_block, /*!< in: crypt buf or NULL */
-	ulint			space)	   /*!< in: space id */
-	__attribute__((nonnull(1,2,3,4,5)));
+	ulint			space)	  /*!< in: space id */
+	MY_ATTRIBUTE((warn_unused_result));
 /*********************************************************************//**
 Allocate a sort buffer.
 @return own: sort buffer */
@@ -433,7 +428,7 @@ row_merge_file_destroy(
 Read a merge block from the file system.
 @return TRUE if request was successful, FALSE if fail */
 UNIV_INTERN
-ibool
+bool
 row_merge_read(
 /*===========*/
 	int			fd,	/*!< in: file descriptor */
@@ -441,10 +436,9 @@ row_merge_read(
 					in number of row_merge_block_t
 					elements */
 	row_merge_block_t*	buf,	/*!< out: data */
-	fil_space_crypt_t*	crypt_data,/*!< in: table crypt data */
 	row_merge_block_t*	crypt_buf, /*!< in: crypt buf or NULL */
-	ulint			space);	   /*!< in: space id */
-
+	ulint			space)	   /*!< in: space id */
+	MY_ATTRIBUTE((warn_unused_result));
 /********************************************************************//**
 Read a merge record.
 @return pointer to next record, or NULL on I/O error or end of list */
@@ -462,8 +456,8 @@ row_merge_read_rec(
 					or NULL on end of list
 					(non-NULL on I/O error) */
 	ulint*			offsets,/*!< out: offsets of mrec */
-	fil_space_crypt_t*	crypt_data,/*!< in: table crypt data */
 	row_merge_block_t*	crypt_block, /*!< in: crypt buf or NULL */
-	ulint			space)	   /*!< in: space id */
-	__attribute__((nonnull(1,2,3,4,6,7,8), warn_unused_result));
+	ulint			space)	/*!< in: space id */
+	MY_ATTRIBUTE((warn_unused_result));
+
 #endif /* row0merge.h */

storage/xtradb/include/srv0srv.h

@@ -137,6 +137,14 @@ struct srv_stats_t {
 	ulint_ctr_64_t          pages_encrypted;
    	/* Number of pages decrypted */
 	ulint_ctr_64_t          pages_decrypted;
+	/* Number of merge blocks encrypted */
+	ulint_ctr_64_t          n_merge_blocks_encrypted;
+	/* Number of merge blocks decrypted */
+	ulint_ctr_64_t          n_merge_blocks_decrypted;
+	/* Number of row log blocks encrypted */
+	ulint_ctr_64_t          n_rowlog_blocks_encrypted;
+	/* Number of row log blocks decrypted */
+	ulint_ctr_64_t          n_rowlog_blocks_decrypted;
 
 	/** Number of data read in total (in bytes) */
 	ulint_ctr_1_t		data_read;
@@ -1276,6 +1284,15 @@ struct export_var_t{
 	ib_int64_t innodb_pages_decrypted;      /*!< Number of pages
 						decrypted */
 
+	/*!< Number of merge blocks encrypted */
+	ib_int64_t innodb_n_merge_blocks_encrypted;
+	/*!< Number of merge blocks decrypted */
+	ib_int64_t innodb_n_merge_blocks_decrypted;
+	/*!< Number of row log blocks encrypted */
+	ib_int64_t innodb_n_rowlog_blocks_encrypted;
+	/*!< Number of row log blocks decrypted */
+	ib_int64_t innodb_n_rowlog_blocks_decrypted;
+
 	ulint innodb_sec_rec_cluster_reads;	/*!< srv_sec_rec_cluster_reads */
 	ulint innodb_sec_rec_cluster_reads_avoided;/*!< srv_sec_rec_cluster_reads_avoided */
 

storage/xtradb/log/log0crypt.cc

 /*****************************************************************************
 
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (C) 2014, 2016, MariaDB Corporation. All Rights Reserved.
+Copyright (C) 2014, 2017, MariaDB Corporation. All Rights Reserved.
 
 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
@@ -236,6 +236,129 @@ log_blocks_crypt(
 	return rc;
 }
 
+/** Encrypt/decrypt temporary log blocks.
+
+@param[in]	src_block	block to encrypt or decrypt
+@param[in]	size		size of the block
+@param[out]	dst_block	destination block
+@param[in]	what		ENCRYPTION_FLAG_ENCRYPT or
+				ENCRYPTION_FLAG_DECRYPT
+@param[in]	offs		offset to block
+@param[in]	space_id	tablespace id
+@return true if successfull, false in case of failure
+*/
+static
+bool
+log_tmp_blocks_crypt(
+	const byte*		src_block,
+	ulint			size,
+	byte*			dst_block,
+	int			what,
+	os_offset_t		offs,
+	ulint			space_id)
+{
+	Crypt_result rc = MY_AES_OK;
+	uint dst_len;
+	byte aes_ctr_counter[MY_AES_BLOCK_SIZE];
+	byte is_encrypt= what == ENCRYPTION_FLAG_ENCRYPT;
+	const crypt_info_t* info = static_cast<const crypt_info_t*>(&crypt_info[0]);
+
+	// AES_CTR_COUNTER = space_id + offs
+
+	bzero(aes_ctr_counter, MY_AES_BLOCK_SIZE);
+	mach_write_to_8(aes_ctr_counter, space_id);
+	mach_write_to_8(aes_ctr_counter + 8, offs);
+
+	rc = encryption_crypt(src_block, size,
+				dst_block, &dst_len,
+				(unsigned char*)(info->crypt_key), 16,
+				aes_ctr_counter, MY_AES_BLOCK_SIZE,
+				what | ENCRYPTION_FLAG_NOPAD,
+				LOG_DEFAULT_ENCRYPTION_KEY,
+				info->key_version);
+
+	if (rc != MY_AES_OK) {
+		ib_logf(IB_LOG_LEVEL_ERROR,
+			"%s failed for temporary log file with rc = %d",
+			is_encrypt ? "Encryption" : "Decryption",
+			rc);
+		return false;
+	}
+
+	return true;
+}
+
+/** Get crypt info
+@return pointer to log crypt info or NULL
+*/
+inline
+const crypt_info_t*
+get_crypt_info()
+{
+	mutex_enter(&log_sys->mutex);
+	const crypt_info_t* info = get_crypt_info(log_sys->next_checkpoint_no);
+	mutex_exit(&log_sys->mutex);
+
+	return info;
+}
+
+/** Find out is temporary log files encrypted.
+@return true if temporary log file should be encrypted, false if not */
+UNIV_INTERN
+bool
+log_tmp_is_encrypted()
+{
+	const crypt_info_t* info = get_crypt_info();
+
+	if (info == NULL || info->key_version == UNENCRYPTED_KEY_VER) {
+		return false;
+	}
+
+	return true;
+}
+
+/** Encrypt temporary log block.
+@param[in]	src_block	block to encrypt or decrypt
+@param[in]	size		size of the block
+@param[out]	dst_block	destination block
+@param[in]	offs		offset to block
+@param[in]	space_id	tablespace id
+@return true if successfull, false in case of failure
+*/
+UNIV_INTERN
+bool
+log_tmp_block_encrypt(
+	const byte*		src_block,
+	ulint			size,
+	byte*			dst_block,
+	os_offset_t		offs,
+	ulint			space_id)
+{
+	return (log_tmp_blocks_crypt(src_block, size, dst_block,
+				ENCRYPTION_FLAG_ENCRYPT, offs, space_id));
+}
+
+/** Decrypt temporary log block.
+@param[in]	src_block	block to encrypt or decrypt
+@param[in]	size		size of the block
+@param[out]	dst_block	destination block
+@param[in]	offs		offset to block
+@param[in]	space_id	tablespace id
+@return true if successfull, false in case of failure
+*/
+UNIV_INTERN
+bool
+log_tmp_block_decrypt(
+	const byte*		src_block,
+	ulint			size,
+	byte*			dst_block,
+	os_offset_t		offs,
+	ulint			space_id)
+{
+	return (log_tmp_blocks_crypt(src_block, size, dst_block,
+				ENCRYPTION_FLAG_DECRYPT, offs, space_id));
+}
+
 /*********************************************************************//**
 Generate crypt key from crypt msg.
 @return true if successfull, false if not. */

storage/xtradb/row/row0ftsort.cc

@@ -40,7 +40,7 @@ Created 10/13/2010 Jimmy Yang
 		b[N] = row_merge_read_rec(				\
 			block[N], buf[N], b[N], index,			\
 			fd[N], &foffs[N], &mrec[N], offsets[N],		\
-			crypt_data, crypt_block[N], space);		\
+			crypt_block[N], space);				\
 		if (UNIV_UNLIKELY(!b[N])) {				\
 			if (mrec[N]) {					\
 				goto exit;				\
@@ -194,7 +194,6 @@ row_fts_psort_info_init(
 	fts_psort_t*		merge_info = NULL;
 	ulint			block_size;
 	ibool			ret = TRUE;
-	fil_space_crypt_t*	crypt_data = NULL;
 	bool			encrypted = false;
 
 	block_size = 3 * srv_sort_buf_size;
@@ -225,21 +224,8 @@ row_fts_psort_info_init(
 	common_info->merge_event = os_event_create();
 	common_info->opt_doc_id_size = opt_doc_id_size;
 
-	/* Theoretically the tablespace can be dropped straight away.
-	In practice, the DDL completion will wait for this thread to
-	finish. */
-	if (fil_space_t* space = fil_space_acquire(new_table->space)) {
-		crypt_data = space->crypt_data;
-		fil_space_release(space);
-	}
-
-	if (crypt_data && crypt_data->should_encrypt()) {
-		common_info->crypt_data = crypt_data;
+	if (log_tmp_is_encrypted()) {
 		encrypted = true;
-	} else {
-		/* Not needed */
-		common_info->crypt_data = NULL;
-		crypt_data = NULL;
 	}
 
 	ut_ad(trx->mysql_thd != NULL);
@@ -577,11 +563,9 @@ row_merge_fts_doc_tokenize(
 		cur_len += len;
 		dfield_dup(field, buf->heap);
 
-		/* Reserve one byte for the end marker of row_merge_block_t
-		and we have reserved ROW_MERGE_RESERVE_SIZE (= 4) for
-		encryption key_version in the beginning of the buffer. */
+		/* Reserve one byte for the end marker of row_merge_block_t */
 		if (buf->total_size + data_size[idx] + cur_len
-			>= (srv_sort_buf_size - 1 - ROW_MERGE_RESERVE_SIZE)) {
+			>= (srv_sort_buf_size - 1)) {
 
 			buf_full = TRUE;
 			break;
@@ -675,7 +659,6 @@ fts_parallel_tokenization(
 	fts_tokenize_ctx_t	t_ctx;
 	ulint			retried = 0;
 	dberr_t			error = DB_SUCCESS;
-	fil_space_crypt_t*	crypt_data = NULL;
 
 	ut_ad(psort_info->psort_common->trx->mysql_thd != NULL);
 
@@ -696,7 +679,6 @@ fts_parallel_tokenization(
 
 	block = psort_info->merge_block;
 	crypt_block = psort_info->crypt_block;
-	crypt_data = psort_info->psort_common->crypt_data;
 	zip_size = dict_table_zip_size(table);
 
 	row_merge_fts_get_next_doc_item(psort_info, &doc_item);
@@ -791,7 +773,6 @@ fts_parallel_tokenization(
 		if (!row_merge_write(merge_file[t_ctx.buf_used]->fd,
 				     merge_file[t_ctx.buf_used]->offset++,
 				     block[t_ctx.buf_used],
-				     crypt_data,
 				     crypt_block[t_ctx.buf_used],
 				     table->space)) {
 			error = DB_TEMP_FILE_WRITE_FAILURE;
@@ -887,7 +868,6 @@ fts_parallel_tokenization(
 				if (!row_merge_write(merge_file[i]->fd,
 						merge_file[i]->offset++,
 						block[i],
-						crypt_data,
 						crypt_block[i],
 						table->space)) {
 					error = DB_TEMP_FILE_WRITE_FAILURE;
@@ -927,7 +907,7 @@ fts_parallel_tokenization(
 				       psort_info->psort_common->dup,
 				       merge_file[i], block[i], &tmpfd[i],
 				       false, 0.0/* pct_progress */, 0.0/* pct_cost */,
-				       crypt_data, crypt_block[i], table->space);
+				       crypt_block[i], table->space);
 
 		if (error != DB_SUCCESS) {
 			close(tmpfd[i]);
@@ -1439,7 +1419,6 @@ row_fts_merge_insert(
 	ulint			start;
 	fts_psort_insert_t	ins_ctx;
 	ulint			count_diag = 0;
-	fil_space_crypt_t*	crypt_data = NULL;
 	ulint			space;
 
 	ut_ad(index);
@@ -1453,7 +1432,6 @@ row_fts_merge_insert(
 	ins_ctx.trx->op_info = "inserting index entries";
 
 	ins_ctx.opt_doc_id_size = psort_info[0].psort_common->opt_doc_id_size;
-	crypt_data = psort_info[0].psort_common->crypt_data;
 
 	heap = mem_heap_create(500 + sizeof(mrec_buf_t));
 
@@ -1547,7 +1525,6 @@ row_fts_merge_insert(
 			    && (!row_merge_read(
 					fd[i], foffs[i],
 					(row_merge_block_t*) block[i],
-					crypt_data,
 					(row_merge_block_t*) crypt_block[i],
 					space))) {
 				error = DB_CORRUPTION;

storage/xtradb/row/row0log.cc

@@ -182,6 +182,7 @@ struct row_log_t {
 	dict_table_t*	table;	/*!< table that is being rebuilt,
 				or NULL when this is a secondary
 				index that is being created online */
+	dict_index_t*	index;	/*!< index to be build */
 	bool		same_pk;/*!< whether the definition of the PRIMARY KEY
 				has remained the same */
 	const dtuple_t*	add_cols;
@@ -197,8 +198,14 @@ struct row_log_t {
 	row_log_buf_t	tail;	/*!< writer context;
 				protected by mutex and index->lock S-latch,
 				or by index->lock X-latch only */
+	byte*		crypt_tail; /*!< writer context;
+				temporary buffer used in encryption,
+				decryption or NULL*/
 	row_log_buf_t	head;	/*!< reader context; protected by MDL only;
 				modifiable by row_log_apply_ops() */
+	byte*		crypt_head; /*!< reader context;
+				temporary buffer used in encryption,
+				decryption or NULL */
 	const char*	path;	/*!< where to create temporary file during
 				log operation */
 };
@@ -349,6 +356,7 @@ row_log_online_op(
 			= (os_offset_t) log->tail.blocks
 			* srv_sort_buf_size;
 		ibool			ret;
+		byte * buf = log->tail.block;
 
 		if (byte_offset + srv_sort_buf_size >= srv_online_max_size) {
 			goto write_failed;
@@ -368,11 +376,29 @@ row_log_online_op(
 			goto err_exit;
 		}
 
+		/* If encryption is enabled encrypt buffer before writing it
+		to file system. */
+		if (log_tmp_is_encrypted()) {
+			if (!log_tmp_block_encrypt(log->tail.block,
+						   srv_sort_buf_size,
+						   log->crypt_tail,
+						   byte_offset,
+						   index->table->space)) {
+				log->error = DB_DECRYPTION_FAILED;
+				goto err_exit;
+			}
+
+			srv_stats.n_rowlog_blocks_encrypted.inc();
+			buf = log->crypt_tail;
+		}
+
 		ret = os_file_write_int_fd(
 			"(modification log)",
 			log->fd,
-			log->tail.block, byte_offset, srv_sort_buf_size);
+			buf, byte_offset, srv_sort_buf_size);
+
 		log->tail.blocks++;
+
 		if (!ret) {
 write_failed:
 			/* We set the flag directly instead of invoking
@@ -380,7 +406,9 @@ row_log_online_op(
 			because the index is not "public" yet. */
 			index->type |= DICT_CORRUPT;
 		}
+
 		UNIV_MEM_INVALID(log->tail.block, srv_sort_buf_size);
+
 		memcpy(log->tail.block, log->tail.buf + avail_size,
 		       mrec_size - avail_size);
 		log->tail.bytes = mrec_size - avail_size;
@@ -465,6 +493,7 @@ row_log_table_close_func(
 			= (os_offset_t) log->tail.blocks
 			* srv_sort_buf_size;
 		ibool			ret;
+		byte * buf = log->tail.block;
 
 		if (byte_offset + srv_sort_buf_size >= srv_online_max_size) {
 			goto write_failed;
@@ -484,11 +513,29 @@ row_log_table_close_func(
 			goto err_exit;
 		}
 
+		/* If encryption is enabled encrypt buffer before writing it
+		to file system. */
+		if (log_tmp_is_encrypted()) {
+			if (!log_tmp_block_encrypt(log->tail.block,
+						   srv_sort_buf_size,
+						   log->crypt_tail,
+						   byte_offset,
+						   log->index->table->space)) {
+				log->error = DB_DECRYPTION_FAILED;
+				goto err_exit;
+			}
+
+			srv_stats.n_rowlog_blocks_encrypted.inc();
+			buf = log->crypt_tail;
+		}
+
 		ret = os_file_write_int_fd(
 			"(modification log)",
 			log->fd,
-			log->tail.block, byte_offset, srv_sort_buf_size);
+			buf, byte_offset, srv_sort_buf_size);
+
 		log->tail.blocks++;
+
 		if (!ret) {
 write_failed:
 			log->error = DB_ONLINE_LOG_TOO_BIG;
@@ -2619,10 +2666,29 @@ row_log_table_apply_ops(
 			goto func_exit;
 		}
 
+		byte * buf = index->online_log->head.block;
+
 		success = os_file_read_no_error_handling_int_fd(
 			index->online_log->fd,
-			index->online_log->head.block, ofs,
+			buf, ofs,
 			srv_sort_buf_size);
+
+		/* If encryption is enabled decrypt buffer after reading it
+		from file system. */
+		if (log_tmp_is_encrypted()) {
+			if (!log_tmp_block_decrypt(buf,
+						   srv_sort_buf_size,
+						   index->online_log->crypt_head,
+						   ofs,
+						   index->table->space)) {
+				error = DB_DECRYPTION_FAILED;
+				goto func_exit;
+			}
+
+			srv_stats.n_rowlog_blocks_decrypted.inc();
+			memcpy(buf, index->online_log->crypt_head, srv_sort_buf_size);
+		}
+
 		if (!success) {
 			fprintf(stderr, "InnoDB: unable to read temporary file"
 				" for table %s\n", index->table_name);
@@ -2929,9 +2995,22 @@ row_log_allocate(
 	log->head.blocks = log->head.bytes = 0;
 	log->head.total = 0;
 	log->path = path;
+	log->crypt_tail = log->crypt_head = NULL;
+	log->index = index;
 	dict_index_set_online_status(index, ONLINE_INDEX_CREATION);
 	index->online_log = log;
 
+	if (log_tmp_is_encrypted()) {
+		ulint size = srv_sort_buf_size;
+		log->crypt_head = static_cast<byte *>(os_mem_alloc_large(&size));
+		log->crypt_tail = static_cast<byte *>(os_mem_alloc_large(&size));
+
+		if (!log->crypt_head || !log->crypt_tail) {
+			row_log_free(log);
+			DBUG_RETURN(false);
+		}
+	}
+
 	/* While we might be holding an exclusive data dictionary lock
 	here, in row_log_abort_sec() we will not always be holding it. Use
 	atomic operations in both cases. */
@@ -2954,6 +3033,15 @@ row_log_free(
 	row_log_block_free(log->tail);
 	row_log_block_free(log->head);
 	row_merge_file_destroy_low(log->fd);
+
+	if (log->crypt_head) {
+		os_mem_free_large(log->crypt_head, srv_sort_buf_size);
+	}
+
+	if (log->crypt_tail) {
+		os_mem_free_large(log->crypt_tail, srv_sort_buf_size);
+	}
+
 	mutex_free(&log->mutex);
 	ut_free(log);
 	log = 0;
@@ -3445,11 +3533,29 @@ row_log_apply_ops(
 			goto func_exit;
 		}
 
+		byte* buf = index->online_log->head.block;
+
 		success = os_file_read_no_error_handling_int_fd(
 			index->online_log->fd,
-			index->online_log->head.block, ofs,
+			buf, ofs,
 			srv_sort_buf_size);
 
+		/* If encryption is enabled decrypt buffer after reading it
+		from file system. */
+		if (log_tmp_is_encrypted()) {
+			if (!log_tmp_block_decrypt(buf,
+						   srv_sort_buf_size,
+						   index->online_log->crypt_head,
+						   ofs,
+						   index->table->space)) {
+				error = DB_DECRYPTION_FAILED;
+				goto func_exit;
+			}
+
+			srv_stats.n_rowlog_blocks_decrypted.inc();
+			memcpy(buf, index->online_log->crypt_head, srv_sort_buf_size);
+		}
+
 		if (!success) {
 			fprintf(stderr, "InnoDB: unable to read temporary file"
 				" for index %s\n", index->name + 1);

storage/xtradb/srv/srv0srv.cc

@@ -2060,6 +2060,10 @@ srv_export_innodb_status(void)
 	export_vars.innodb_pages_page_compression_error = srv_stats.pages_page_compression_error;
 	export_vars.innodb_pages_decrypted = srv_stats.pages_decrypted;
 	export_vars.innodb_pages_encrypted = srv_stats.pages_encrypted;
+	export_vars.innodb_n_merge_blocks_encrypted = srv_stats.n_merge_blocks_encrypted;
+	export_vars.innodb_n_merge_blocks_decrypted = srv_stats.n_merge_blocks_decrypted;
+	export_vars.innodb_n_rowlog_blocks_encrypted = srv_stats.n_rowlog_blocks_encrypted;
+	export_vars.innodb_n_rowlog_blocks_decrypted = srv_stats.n_rowlog_blocks_decrypted;
 
 	export_vars.innodb_defragment_compression_failures =
 		btr_defragment_compression_failures;