Added support for CHECK TABLE for S3 tables

Other things:
- Don't allocate an IO_CACHE for scanning tables of type BLOCK
  (It was never used in this case)
- Fixed bug in page cache that cased a hang when trying to read a
  not existing S3 block.

mysql-test/suite/s3/basic.result

@@ -2,23 +2,25 @@ drop table if exists t1;
 #
 # Test simple create of s3 table
 #
-create or replace table t1 (a int, b int, key (a)) engine=aria;
-insert into t1 select seq,seq+10 from seq_1_to_10000;
+create or replace table t1 (a int, b int, c varchar(1000), key (a), key(c)) engine=aria;
+insert into t1 select seq, seq+10, repeat(char(65+ mod(seq, 20)),mod(seq,1000)) from seq_1_to_10000;
 alter table t1 engine=s3;
 show create table t1;
 Table	Create Table
 t1	CREATE TABLE `t1` (
   `a` int(11) DEFAULT NULL,
   `b` int(11) DEFAULT NULL,
-  KEY `a` (`a`)
+  `c` varchar(1000) DEFAULT NULL,
+  KEY `a` (`a`),
+  KEY `c` (`c`)
 ) ENGINE=S3 DEFAULT CHARSET=latin1 PAGE_CHECKSUM=1
 select * from information_schema.tables where table_schema="database" and table_name="t1";;
 TABLE_CATALOG	TABLE_SCHEMA	TABLE_NAME	TABLE_TYPE	ENGINE	VERSION	ROW_FORMAT	TABLE_ROWS	AVG_ROW_LENGTH	DATA_LENGTH	MAX_DATA_LENGTH	INDEX_LENGTH	DATA_FREE	AUTO_INCREMENT	CREATE_TIME	UPDATE_TIME	CHECK_TIME	TABLE_COLLATION	CHECKSUM	CREATE_OPTIONS	TABLE_COMMENT	MAX_INDEX_LENGTH	TEMPORARY
-def	#	t1	BASE TABLE	S3	10	Page	10000	33	335872	#	122880	0	NULL	#	#	#	latin1_swedish_ci	NULL	page_checksum=1		9007199254732800	#
+def	#	t1	BASE TABLE	S3	10	Page	10000	567	5677056	#	761856	0	NULL	#	#	#	latin1_swedish_ci	NULL	page_checksum=1		2305843009213685760	#
 show table status like "t1";
 Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment	Max_index_length	Temporary
-t1	S3	10	Page	10000	33	335872	#	122880	0	NULL	#	#	#	latin1_swedish_ci	NULL	page_checksum=1		#	N
-select * from t1 limit 10;
+t1	S3	10	Page	10000	567	5677056	#	761856	0	NULL	#	#	#	latin1_swedish_ci	NULL	page_checksum=1		#	N
+select a,b from t1 limit 10;
 a	b
 1	11
 2	12
@@ -33,7 +35,7 @@ a	b
 select count(*) from t1;
 count(*)
 10000
-select * from t1 where a between 10 and 20;
+select a,b from t1 where a between 10 and 20;
 a	b
 10	20
 11	21
@@ -60,9 +62,15 @@ ERROR HY000: Table 't1' is read only
 #
 set @@use_stat_tables='never';
 truncate mysql.table_stats;
-check table t1;
+check table t1 fast;
 Table	Op	Msg_type	Msg_text
-database.t1	check	status	Table 'database.t1' is read only
+database.t1	check	status	Table is already up to date
+check table t1 quick;
+Table	Op	Msg_type	Msg_text
+database.t1	check	status	OK
+check table t1 extended;
+Table	Op	Msg_type	Msg_text
+database.t1	check	status	OK
 analyze table t1;
 Table	Op	Msg_type	Msg_text
 database.t1	analyze	status	Table 'database.t1' is read only
@@ -91,9 +99,11 @@ Table	Create Table
 t1	CREATE TABLE `t1` (
   `a` int(11) DEFAULT NULL,
   `b` int(11) DEFAULT NULL,
-  KEY `a` (`a`)
+  `c` varchar(1000) DEFAULT NULL,
+  KEY `a` (`a`),
+  KEY `c` (`c`)
 ) ENGINE=Aria DEFAULT CHARSET=latin1 PAGE_CHECKSUM=1
-select * from t1 limit 10;
+select a,b from t1 limit 10;
 a	b
 1	11
 2	12

mysql-test/suite/s3/basic.test

@@ -13,8 +13,8 @@ drop table if exists t1;
 --echo # Test simple create of s3 table
 --echo #
 
-create or replace table t1 (a int, b int, key (a)) engine=aria;
-insert into t1 select seq,seq+10 from seq_1_to_10000;
+create or replace table t1 (a int, b int, c varchar(1000), key (a), key(c)) engine=aria;
+insert into t1 select seq, seq+10, repeat(char(65+ mod(seq, 20)),mod(seq,1000)) from seq_1_to_10000;
 alter table t1 engine=s3;
 show create table t1;
 
@@ -23,9 +23,9 @@ show create table t1;
 --eval select * from information_schema.tables where table_schema="$database" and table_name="t1";
 --replace_column 8 # 12 # 13 # 14 # 19 #
 show table status like "t1";
-select * from t1 limit 10;
+select a,b from t1 limit 10;
 select count(*) from t1;
-select * from t1 where a between 10 and 20;
+select a,b from t1 where a between 10 and 20;
 --replace_column 9 #
 explain select * from t1 where a between 10 and 20;
 --error ER_OPEN_AS_READONLY
@@ -43,7 +43,11 @@ delete from t1 where a>10;
 set @@use_stat_tables='never';
 truncate mysql.table_stats;
 --replace_result $database database
-check table t1;
+check table t1 fast;
+--replace_result $database database
+check table t1 quick;
+--replace_result $database database
+check table t1 extended;
 --replace_result $database database
 analyze table t1;
 --replace_result $database database
@@ -61,7 +65,7 @@ select * from mysql.table_stats;
 
 alter table t1 engine=aria;
 show create table t1;
-select * from t1 limit 10;
+select a,b from t1 limit 10;
 select count(*) from t1;
 delete from t1 where a=1;
 drop table t1;

mysql-test/suite/s3/partition.result

@@ -64,7 +64,7 @@ count(*)
 6
 ALTER TABLE t2 CHECK PARTITION p3;
 Table	Op	Msg_type	Msg_text
-s3.t2	check	status	Table 's3.t2' is read only
+s3.t2	check	status	OK
 SELECT count(*) FROM t2;
 count(*)
 6

storage/maria/ha_maria.cc

@@ -1270,6 +1270,7 @@ int ha_maria::check(THD * thd, HA_CHECK_OPT * check_opt)
   if (!file || !param) return HA_ADMIN_INTERNAL_ERROR;
 
   unmap_file(file);
+  register_handler(file);
   maria_chk_init(param);
   param->thd= thd;
   param->op_name= "check";
@@ -1325,14 +1326,18 @@ int ha_maria::check(THD * thd, HA_CHECK_OPT * check_opt)
     {
       ulonglong old_testflag= param->testflag;
       param->testflag |= T_MEDIUM;
-      if (!(error= init_io_cache(&param->read_cache, file->dfile.file,
-                                 my_default_record_cache_size, READ_CACHE,
-                                 share->pack.header_length, 1, MYF(MY_WME))))
-      {
+
+      /* BLOCK_RECORD does not need a cache as it is using the page cache */
+      if (file->s->data_file_type != BLOCK_RECORD)
+        error= init_io_cache(&param->read_cache, file->dfile.file,
+                             my_default_record_cache_size, READ_CACHE,
+                             share->pack.header_length, 1, MYF(MY_WME));
+      if (!error)
         error= maria_chk_data_link(param, file,
                                    MY_TEST(param->testflag & T_EXTEND));
+
+      if (file->s->data_file_type != BLOCK_RECORD)
         end_io_cache(&param->read_cache);
-      }
       param->testflag= old_testflag;
     }
   }

storage/maria/ha_s3.h

@@ -44,11 +44,6 @@ class ha_s3 final :public ha_maria
     DBUG_ENTER("delete_row");
     DBUG_RETURN(HA_ERR_TABLE_READONLY);
   }
-  int check(THD *, HA_CHECK_OPT *) override
-  {
-    DBUG_ENTER("check");
-    DBUG_RETURN(HA_ERR_TABLE_READONLY);
-  }
   int analyze(THD *, HA_CHECK_OPT *) override
   {
     DBUG_ENTER("analyze");

storage/maria/ma_check.c

@@ -413,6 +413,12 @@ int maria_chk_size(HA_CHECK *param, register MARIA_HA *info)
   char buff[22],buff2[22];
   DBUG_ENTER("maria_chk_size");
 
+  if (info->s3)
+  {
+    /* We cannot check file sizes for S3 */
+    DBUG_RETURN(0);
+  }
+
   if (!(param->testflag & T_SILENT))
     puts("- check file-size");
 

storage/maria/ma_pagecache.c

@@ -2772,7 +2772,7 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
 
 #ifdef WITH_S3_STORAGE_ENGINE
 static void read_big_block(PAGECACHE *pagecache,
-                              PAGECACHE_BLOCK_LINK *block)
+                           PAGECACHE_BLOCK_LINK *block)
 {
   int page_st;
   size_t big_block_size_in_pages;
@@ -2810,6 +2810,11 @@ static void read_big_block(PAGECACHE *pagecache,
     if (block_to_read->status & PCBLOCK_ERROR)
     {
       /* We get first block with an error so all operation failed */
+      DBUG_PRINT("error", ("Got error when reading first page"));
+      block->status|= PCBLOCK_ERROR;
+      block->error= block_to_read->error;
+      remove_reader(block_to_read);
+      unreg_request(pagecache, block_to_read, 1);
       DBUG_VOID_RETURN;
     }
     if (block_to_read->status & PCBLOCK_BIG_READ)
@@ -3952,7 +3957,6 @@ void pagecache_set_write_on_delete_by_link(PAGECACHE_BLOCK_LINK *block)
 
   @retval 0 deleted or was not present at all
   @retval 1 error
-
 */
 
 static my_bool pagecache_delete_internal(PAGECACHE *pagecache,