MDEV-20306 Assert when converting encrypted Aria table to S3

Changes:
- maria_create() now uses a bit in the parameter flags to check if table
  should be encrypted instead of using maria_encrypted_tables.
- Don't encrypt tables that are to be converted to S3
- Added encrypted flag to ARIA_TABLE_CAPABILITIES
- maria_chk --description now prints if table is encrypted. Other
  operations is not allowed on encrypted tables.

include/aria_backup.h

@@ -26,6 +26,7 @@ typedef struct st_aria_table_capabilities
   enum data_file_type data_file_type;
   my_bool checksum;
   my_bool transactional;
+  my_bool encrypted;
   /* This is true if the table can be copied without any locks */
   my_bool online_backup_safe;
   /* s3 capabilities */

include/my_base.h

@@ -368,6 +368,7 @@ enum ha_base_keytype {
 #define HA_CREATE_INTERNAL_TABLE 256U
 #define HA_PRESERVE_INSERT_ORDER 512U
 #define HA_CREATE_NO_ROLLBACK    1024U
+#define HA_CREATE_ENCRYPTED      2048U
 
 /* Flags used by start_bulk_insert */
 

mysql-test/suite/s3/encryption.opt

@@ -0,0 +1,4 @@
+--plugin-load-add=$FILE_KEY_MANAGEMENT_SO
+--aria-encrypt-tables=1
+--loose-file-key-management
+--loose-file-key-management-filename=$MYSQL_TEST_DIR/std_data/keys.txt

mysql-test/suite/s3/encryption.result

@@ -0,0 +1,23 @@
+#
+# MDEV-20306
+# Assertion `!(end_of_data > info->scan.dir_end)' failed in
+# _ma_scan_block_record upon converting table from S3 to Aria
+# with encryption enabled
+#
+drop table if exists t1;
+CREATE TABLE t1 (a INT) ENGINE=Aria;
+INSERT INTO t1 VALUES (1);
+ALTER TABLE t1 ENGINE=S3;
+select * from t1;
+a
+1
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=S3 DEFAULT CHARSET=latin1 PAGE_CHECKSUM=1
+ALTER TABLE t1 ENGINE=Aria;
+select * from t1;
+a
+1
+DROP TABLE t1;

mysql-test/suite/s3/encryption.test

@@ -0,0 +1,36 @@
+--source include/have_s3.inc
+
+if (`SELECT COUNT(*)=0 FROM INFORMATION_SCHEMA.PLUGINS WHERE PLUGIN_NAME = 'file_key_management' AND PLUGIN_STATUS='ACTIVE'`)
+{
+  --skip Test requires file_key_management plugin
+}
+
+#
+# Create unique database for running the tests
+#
+--source create_database.inc
+
+--echo #
+--echo # MDEV-20306
+--echo # Assertion `!(end_of_data > info->scan.dir_end)' failed in
+--echo # _ma_scan_block_record upon converting table from S3 to Aria
+--echo # with encryption enabled
+--echo #
+
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+
+CREATE TABLE t1 (a INT) ENGINE=Aria;
+INSERT INTO t1 VALUES (1);
+ALTER TABLE t1 ENGINE=S3;
+select * from t1;
+show create table t1;
+ALTER TABLE t1 ENGINE=Aria;
+select * from t1;
+DROP TABLE t1;
+
+#
+# clean up
+#
+--source drop_database.inc

storage/maria/ha_maria.cc

@@ -3165,6 +3165,8 @@ int ha_maria::create(const char *name, TABLE *table_arg,
   (void) translog_log_debug_info(0, LOGREC_DEBUG_INFO_QUERY,
                                  (uchar*) thd->query(), thd->query_length());
 
+  if (maria_encrypt_tables && ht == maria_hton)
+    create_flags|= HA_CREATE_ENCRYPTED;
   /* TODO: Check that the following fn_format is really needed */
   error=
     maria_create(fn_format(buff, name, "", "",

storage/maria/ma_backup.c

@@ -80,6 +80,8 @@ int aria_get_capabilities(File kfile, ARIA_TABLE_CAPABILITIES *cap)
   cap->data_file_type= share.state.header.data_file_type;
   cap->s3_block_size=  share.base.s3_block_size;
   cap->compression=    share.base.compression_algorithm;
+  cap->encrypted=      MY_TEST(share.base.extra_options &
+                               MA_EXTRA_OPTIONS_ENCRYPTED);
 
   if (share.state.header.data_file_type == BLOCK_RECORD)
   {

storage/maria/ma_create.c

@@ -95,7 +95,8 @@ int maria_create(const char *name, enum data_file_type datafile_type,
   my_bool forced_packed;
   myf     sync_dir=  0;
   uchar   *log_data= NULL;
-  my_bool encrypted= maria_encrypt_tables && datafile_type == BLOCK_RECORD;
+  my_bool encrypted= ((flags & HA_CREATE_ENCRYPTED) &&
+                      datafile_type == BLOCK_RECORD);
   my_bool insert_order= MY_TEST(flags & HA_PRESERVE_INSERT_ORDER);
   uint crypt_page_header_space= 0;
   DBUG_ENTER("maria_create");

storage/maria/maria_chk.c

@@ -1097,6 +1097,15 @@ static int maria_chk(HA_CHECK *param, char *filename)
       param->testflag|= T_REP_BY_SORT;
     }
   }
+  if ((share->base.extra_options & MA_EXTRA_OPTIONS_ENCRYPTED) &&
+      !(param->testflag & T_DESCRIPT))
+  {
+    _ma_check_print_warning(param,
+                            "Table %s is encrypted. Only --description (-d) "
+                            "option is supported", filename);
+    param->warning_printed= 0;
+    goto end2;
+  }
 
   /*
     Skip the checking of the file if:
@@ -1549,6 +1558,8 @@ static void descript(HA_CHECK *param, register MARIA_HA *info, char *name)
 
   if (param->testflag & T_VERBOSE)
   {
+    if (share->base.extra_options & MA_EXTRA_OPTIONS_ENCRYPTED)
+      printf("Encrypted:           yes\n");
     printf("File-version:        %d\n",
 	   (int) share->state.header.file_version[3]);
     if (share->state.create_time)

storage/maria/s3_func.c

@@ -356,7 +356,8 @@ int aria_copy_to_s3(ms3_st *s3_client, const char *aws_bucket,
             error, path);
     goto err;
   }
-  if (cap.transactional || cap.data_file_type != BLOCK_RECORD)
+  if (cap.transactional || cap.data_file_type != BLOCK_RECORD ||
+      cap.encrypted)
   {
     fprintf(stderr,
             "Aria table %s doesn't match criteria to be copied to S3.\n"