diff --git a/extra/mariabackup/aria_backup_client.cc b/extra/mariabackup/aria_backup_client.cc
index 700a62dc4eb..4056f72b62d 100644
--- a/extra/mariabackup/aria_backup_client.cc
+++ b/extra/mariabackup/aria_backup_client.cc
@@ -989,8 +989,8 @@ bool prepare(const char *target_dir) {
         logs.find_logs_after_last(target_dir);
         last_logno= logs.last(); // Update last_logno if extra logs were found
 
-	if (init_pagecache(maria_pagecache, 1024L*1024L, 0, 0,
-		static_cast<uint>(maria_block_size), 0, MY_WME) == 0)
+	if (multi_init_pagecache(&maria_pagecaches, 1, 1024L*1024L, 0, 0,
+		static_cast<uint>(maria_block_size), 0, MY_WME))
 		die("Got error in Aria init_pagecache() (errno: %d)", errno);
 
 	if (init_pagecache(maria_log_pagecache, 1024L*1024L,
diff --git a/mysql-test/suite/maria/maria-recovery-bitmap.opt b/mysql-test/suite/maria/maria-recovery-bitmap.opt
new file mode 100644
index 00000000000..89562ce3a46
--- /dev/null
+++ b/mysql-test/suite/maria/maria-recovery-bitmap.opt
@@ -0,0 +1 @@
+--aria-pagecache-segments=5
diff --git a/mysql-test/suite/maria/maria3.result b/mysql-test/suite/maria/maria3.result
index 3f1ac8a5d04..5ff3f667c6f 100644
--- a/mysql-test/suite/maria/maria3.result
+++ b/mysql-test/suite/maria/maria3.result
@@ -320,6 +320,7 @@ aria_pagecache_age_threshold	#
 aria_pagecache_buffer_size	#
 aria_pagecache_division_limit	#
 aria_pagecache_file_hash_size	#
+aria_pagecache_segments	#
 aria_page_checksum	#
 aria_recover_options	#
 aria_repair_threads	#
diff --git a/mysql-test/suite/sys_vars/r/sysvars_aria.result b/mysql-test/suite/sys_vars/r/sysvars_aria.result
index 18f38ea11ef..8ac0405b602 100644
--- a/mysql-test/suite/sys_vars/r/sysvars_aria.result
+++ b/mysql-test/suite/sys_vars/r/sysvars_aria.result
@@ -181,6 +181,18 @@ NUMERIC_BLOCK_SIZE	1
 ENUM_VALUE_LIST	NULL
 READ_ONLY	YES
 COMMAND_LINE_ARGUMENT	REQUIRED
+VARIABLE_NAME	ARIA_PAGECACHE_SEGMENTS
+SESSION_VALUE	NULL
+DEFAULT_VALUE	1
+VARIABLE_SCOPE	GLOBAL
+VARIABLE_TYPE	BIGINT UNSIGNED
+VARIABLE_COMMENT	The number of segments in the page_cache. Each file is put in their own segments of size pagecache_buffer_size / segments. Having many segments improves parallel performance
+NUMERIC_MIN_VALUE	1
+NUMERIC_MAX_VALUE	128
+NUMERIC_BLOCK_SIZE	1
+ENUM_VALUE_LIST	NULL
+READ_ONLY	YES
+COMMAND_LINE_ARGUMENT	REQUIRED
 VARIABLE_NAME	ARIA_PAGE_CHECKSUM
 SESSION_VALUE	NULL
 DEFAULT_VALUE	ON
diff --git a/mysql-test/suite/sys_vars/r/sysvars_server_embedded.result b/mysql-test/suite/sys_vars/r/sysvars_server_embedded.result
index cee7576a7bb..0084920554b 100644
--- a/mysql-test/suite/sys_vars/r/sysvars_server_embedded.result
+++ b/mysql-test/suite/sys_vars/r/sysvars_server_embedded.result
@@ -192,6 +192,16 @@ NUMERIC_BLOCK_SIZE	1
 ENUM_VALUE_LIST	NULL
 READ_ONLY	YES
 COMMAND_LINE_ARGUMENT	REQUIRED
+VARIABLE_NAME	ARIA_PAGECACHE_SEGMENTS
+VARIABLE_SCOPE	GLOBAL
+VARIABLE_TYPE	BIGINT UNSIGNED
+VARIABLE_COMMENT	The number of segments in the page_cache. Each file is put in their own segments of size pagecache_buffer_size / segments. Having many segments improves parallel performance
+NUMERIC_MIN_VALUE	1
+NUMERIC_MAX_VALUE	128
+NUMERIC_BLOCK_SIZE	1
+ENUM_VALUE_LIST	NULL
+READ_ONLY	YES
+COMMAND_LINE_ARGUMENT	REQUIRED
 VARIABLE_NAME	ARIA_PAGE_CHECKSUM
 VARIABLE_SCOPE	GLOBAL
 VARIABLE_TYPE	BOOLEAN
diff --git a/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result b/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result
index d91f5df6798..375a86621de 100644
--- a/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result
+++ b/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result
@@ -192,6 +192,16 @@ NUMERIC_BLOCK_SIZE	1
 ENUM_VALUE_LIST	NULL
 READ_ONLY	YES
 COMMAND_LINE_ARGUMENT	REQUIRED
+VARIABLE_NAME	ARIA_PAGECACHE_SEGMENTS
+VARIABLE_SCOPE	GLOBAL
+VARIABLE_TYPE	BIGINT UNSIGNED
+VARIABLE_COMMENT	The number of segments in the page_cache. Each file is put in their own segments of size pagecache_buffer_size / segments. Having many segments improves parallel performance
+NUMERIC_MIN_VALUE	1
+NUMERIC_MAX_VALUE	128
+NUMERIC_BLOCK_SIZE	1
+ENUM_VALUE_LIST	NULL
+READ_ONLY	YES
+COMMAND_LINE_ARGUMENT	REQUIRED
 VARIABLE_NAME	ARIA_PAGE_CHECKSUM
 VARIABLE_SCOPE	GLOBAL
 VARIABLE_TYPE	BOOLEAN
diff --git a/storage/maria/aria_chk.c b/storage/maria/aria_chk.c
index 1b6c7020902..9a4c3f20cca 100644
--- a/storage/maria/aria_chk.c
+++ b/storage/maria/aria_chk.c
@@ -1224,14 +1224,16 @@ static int maria_chk(HA_CHECK *param, char *filename)
   */
   maria_lock_database(info, F_EXTRA_LCK);
   datafile= info->dfile.file;
-  if (init_pagecache(maria_pagecache, (size_t) param->use_buffers, 0, 0,
-                     maria_block_size, 0, MY_WME) == 0)
+  if (multi_init_pagecache(&maria_pagecaches, 1, (size_t) param->use_buffers,
+                           0, 0, maria_block_size, 0, MY_WME))
   {
     _ma_check_print_error(param, "Can't initialize page cache with %lu memory",
                           (ulong) param->use_buffers);
     error= 1;
     goto end2;
   }
+  /* The pagecache is initialized. Update the table pagecaches pointers */
+  ma_change_pagecache(info);
 
   if (param->testflag & (T_REP_ANY | T_SORT_RECORDS | T_SORT_INDEX |
                          T_ZEROFILL))
@@ -1472,7 +1474,7 @@ end2:
     _ma_check_print_error(param, default_close_errmsg, my_errno, filename);
     DBUG_RETURN(1);
   }
-  end_pagecache(maria_pagecache, 1);
+  multi_end_pagecache(&maria_pagecaches);
   if (error == 0)
   {
     if (param->out_flag & O_NEW_DATA)
diff --git a/storage/maria/aria_ftdump.c b/storage/maria/aria_ftdump.c
index 571d7cc6a70..c9694818d1f 100644
--- a/storage/maria/aria_ftdump.c
+++ b/storage/maria/aria_ftdump.c
@@ -85,8 +85,9 @@ int main(int argc,char *argv[])
       usage();
   }
 
-  init_pagecache(maria_pagecache, PAGE_BUFFER_INIT, 0, 0,
-                 MARIA_KEY_BLOCK_LENGTH, 0, MY_WME);
+  multi_init_pagecache(&maria_pagecaches, 1,
+                       PAGE_BUFFER_INIT, 0, 0,
+                       MARIA_KEY_BLOCK_LENGTH, 0, MY_WME);
 
   if (!(info=maria_open(argv[0], O_RDONLY,
                         HA_OPEN_ABORT_IF_LOCKED|HA_OPEN_FROM_SQL_LAYER, 0)))
diff --git a/storage/maria/aria_pack.c b/storage/maria/aria_pack.c
index 98761cdaa83..7acba8d1bfb 100644
--- a/storage/maria/aria_pack.c
+++ b/storage/maria/aria_pack.c
@@ -613,12 +613,15 @@ static int compress(PACK_MRG_INFO *mrg,char *result_table)
   else
     fn_format(org_name,isam_file->s->open_file_name.str, "",MARIA_NAME_DEXT, 2+4+16);
 
-  if (init_pagecache(maria_pagecache, MARIA_MIN_PAGE_CACHE_SIZE, 0, 0,
-                     maria_block_size, 0, MY_WME) == 0)
+  if (multi_init_pagecache(&maria_pagecaches, 1, MARIA_MIN_PAGE_CACHE_SIZE,
+                           0, 0, maria_block_size, 0, MY_WME))
   {
     fprintf(stderr, "Can't initialize page cache\n");
     goto err;
   }
+  /* The pagecache is initialized. Update the table pagecaches pointers */
+  for (i=0 ; i < mrg->count ; i++)
+    ma_change_pagecache(mrg->file[i]);
 
   if (!test_only && result_table)
   {
@@ -867,7 +870,7 @@ static int compress(PACK_MRG_INFO *mrg,char *result_table)
   if (join_maria_file >= 0)
     my_close(join_maria_file,MYF(0));
   mrg_close(mrg);
-  end_pagecache(maria_pagecache, 1);
+  multi_end_pagecache(&maria_pagecaches);
   fprintf(stderr, "Aborted: %s is not compressed\n", org_name);
   DBUG_RETURN(-1);
 }
diff --git a/storage/maria/aria_read_log.c b/storage/maria/aria_read_log.c
index 2bc7e5f6280..8d94d0c482c 100644
--- a/storage/maria/aria_read_log.c
+++ b/storage/maria/aria_read_log.c
@@ -114,8 +114,8 @@ int main(int argc, char **argv)
     fprintf(stderr, "Can't find any log\n");
     goto err;
   }
-  if (init_pagecache(maria_pagecache, (size_t)opt_page_buffer_size, 0, 0,
-                     maria_block_size, 0, MY_WME) == 0)
+  if (multi_init_pagecache(&maria_pagecaches, 1, (size_t) opt_page_buffer_size,
+                           0, 0, maria_block_size, 0, MY_WME))
   {
     fprintf(stderr, "Got error in init_pagecache() (errno: %d)\n", errno);
     goto err;
diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index 8a92ee84f8b..647df858798 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -42,6 +42,7 @@ C_MODE_END
 #include "key.h"
 #include "log.h"
 #include "sql_parse.h"
+#include "mysql/plugin.h"
 #include "mysql/service_print_check_msg.h"
 #include "debug.h"
 
@@ -57,7 +58,8 @@ C_MODE_END
 
 #define THD_TRN (TRN*) thd_get_ha_data(thd, maria_hton)
 
-ulong pagecache_division_limit, pagecache_age_threshold, pagecache_file_hash_size;
+ulong pagecache_division_limit, pagecache_age_threshold;
+ulong pagecache_file_hash_size, pagecache_segments;
 ulonglong pagecache_buffer_size;
 const char *zerofill_error_msg=
   "Table is probably from another system and must be zerofilled or repaired ('REPAIR TABLE table_name') to be usable on this system";
@@ -258,6 +260,14 @@ static MYSQL_SYSVAR_ENUM(sync_log_dir, sync_log_dir, PLUGIN_VAR_RQCMDARG,
        "creation", NULL, NULL, TRANSLOG_SYNC_DIR_NEWFILE,
        &maria_sync_log_dir_typelib);
 
+static MYSQL_SYSVAR_ULONG(pagecache_segments, pagecache_segments,
+                          PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,
+                          "The number of segments in the page_cache. "
+                          "Each file is put in their own segments of size "
+                          "pagecache_buffer_size / segments. "
+                          "Having many segments improves parallel performance",
+                          0, 0, 1, 1, 128, 1);
+
 #ifdef USE_ARIA_FOR_TMP_TABLES
 #define USE_ARIA_FOR_TMP_TABLES_VAL 1
 #else
@@ -3926,10 +3936,11 @@ static int ha_maria_init(void *p)
   res= res ||
     ((force_start_after_recovery_failures != 0 && !aria_readonly) &&
      mark_recovery_start(log_dir)) ||
-    !init_pagecache(maria_pagecache,
-                    (size_t) pagecache_buffer_size, pagecache_division_limit,
-                    pagecache_age_threshold, maria_block_size, pagecache_file_hash_size,
-                    0) ||
+    multi_init_pagecache(&maria_pagecaches, pagecache_segments,
+                         (size_t) pagecache_buffer_size,
+                         pagecache_division_limit,
+                         pagecache_age_threshold, maria_block_size,
+                         pagecache_file_hash_size, 0) ||
     !init_pagecache(maria_log_pagecache,
                     TRANSLOG_PAGECACHE_SIZE, 0, 0,
                     TRANSLOG_PAGE_SIZE, 0, 0) ||
@@ -3946,7 +3957,6 @@ static int ha_maria_init(void *p)
     ma_checkpoint_init(checkpoint_interval);
   maria_multi_threaded= maria_in_ha_maria= TRUE;
   maria_create_trn_hook= maria_create_trn_for_mysql;
-  maria_pagecache->extra_debug= 1;
   maria_assert_if_crashed_table= debug_assert_if_crashed_table;
 
   if (res)
@@ -4052,6 +4062,7 @@ static struct st_mysql_sys_var *system_variables[]= {
   MYSQL_SYSVAR(pagecache_buffer_size),
   MYSQL_SYSVAR(pagecache_division_limit),
   MYSQL_SYSVAR(pagecache_file_hash_size),
+  MYSQL_SYSVAR(pagecache_segments),
   MYSQL_SYSVAR(recover_options),
   MYSQL_SYSVAR(repair_threads),
   MYSQL_SYSVAR(sort_buffer_size),
@@ -4177,14 +4188,31 @@ static void update_log_file_size(MYSQL_THD thd,
 
 
 static SHOW_VAR status_variables[]= {
-  {"pagecache_blocks_not_flushed", (char*) &maria_pagecache_var.global_blocks_changed, SHOW_LONG},
-  {"pagecache_blocks_unused",      (char*) &maria_pagecache_var.blocks_unused, SHOW_LONG},
-  {"pagecache_blocks_used",        (char*) &maria_pagecache_var.blocks_used, SHOW_LONG},
-  {"pagecache_read_requests",      (char*) &maria_pagecache_var.global_cache_r_requests, SHOW_LONGLONG},
-  {"pagecache_reads",              (char*) &maria_pagecache_var.global_cache_read, SHOW_LONGLONG},
-  {"pagecache_write_requests",     (char*) &maria_pagecache_var.global_cache_w_requests, SHOW_LONGLONG},
-  {"pagecache_writes",             (char*) &maria_pagecache_var.global_cache_write, SHOW_LONGLONG},
-  {"transaction_log_syncs",        (char*) &translog_syncs, SHOW_LONGLONG},
+  {"blocks_not_flushed", (char*) &pagecache_stats.global_blocks_changed, SHOW_LONG},
+  {"blocks_unused",      (char*) &pagecache_stats.blocks_unused, SHOW_LONG},
+  {"blocks_used",        (char*) &pagecache_stats.blocks_used, SHOW_LONG},
+  {"read_requests",      (char*) &pagecache_stats.global_cache_r_requests, SHOW_LONGLONG},
+  {"reads",              (char*) &pagecache_stats.global_cache_read, SHOW_LONGLONG},
+  {"write_requests",     (char*) &pagecache_stats.global_cache_w_requests, SHOW_LONGLONG},
+  {"writes",             (char*) &pagecache_stats.global_cache_write, SHOW_LONGLONG},
+  {NullS, NullS, SHOW_LONG}
+};
+
+
+static int pagecache_stats_func(THD *thd, SHOW_VAR *var, void *buff,
+                                system_status_var *, enum_var_type scope)
+{
+  multi_update_pagecache_stats();
+  var->type= SHOW_ARRAY;
+  var->value= status_variables;
+  return 0;
+}
+
+static SHOW_VAR dynamic_status_variables[]=
+{
+  /* Accessing pagecache causes multi_update_pagecache_stats() to be called */
+  {"pagecache",             (char*) &pagecache_stats_func, SHOW_FUNC},
+  {"transaction_log_syncs", (char*) &translog_syncs, SHOW_LONGLONG},
   {NullS, NullS, SHOW_LONG}
 };
 
@@ -4373,10 +4401,10 @@ maria_declare_plugin(aria)
   PLUGIN_LICENSE_GPL,
   ha_maria_init,                /* Plugin Init      */
   NULL,                         /* Plugin Deinit    */
-  0x0105,                       /* 1.5              */
-  status_variables,             /* status variables */
+  0x0106,                       /* 1.6              */
+  dynamic_status_variables,     /* status variables */
   system_variables,             /* system variables */
-  "1.5",                        /* string version   */
+  "1.6",                        /* string version   */
   MariaDB_PLUGIN_MATURITY_STABLE /* maturity         */
 }
 maria_declare_plugin_end;
diff --git a/storage/maria/ha_s3.cc b/storage/maria/ha_s3.cc
index 9ae16524b9b..b49293e5bbe 100644
--- a/storage/maria/ha_s3.cc
+++ b/storage/maria/ha_s3.cc
@@ -682,7 +682,8 @@ int ha_s3::open(const char *name, int mode, uint open_flags)
         Table is in S3. We have to modify the pagecache callbacks for the
         data file, index file and for bitmap handling.
       */
-      file->s->pagecache= &s3_pagecache;
+      file->s->pagecache= file->s->kfile.pagecache= file->dfile.pagecache=
+        &s3_pagecache;
       file->dfile.big_block_size= file->s->kfile.big_block_size=
         file->s->bitmap.file.big_block_size= file->s->base.s3_block_size;
       file->s->kfile.head_blocks= file->s->base.keystart / file->s->block_size;
diff --git a/storage/maria/ma_bitmap.c b/storage/maria/ma_bitmap.c
index bdfba862b0c..798118b116a 100644
--- a/storage/maria/ma_bitmap.c
+++ b/storage/maria/ma_bitmap.c
@@ -247,6 +247,7 @@ my_bool _ma_bitmap_init(MARIA_SHARE *share, File file,
   bitmap->share= share;
   bitmap->block_size= share->block_size;
   bitmap->file.file= file;
+  bitmap->file.pagecache= share->kfile.pagecache;
   _ma_bitmap_set_pagecache_callbacks(&bitmap->file, share);
 
   /* Size needs to be aligned on 6 */
@@ -549,7 +550,7 @@ my_bool _ma_bitmap_flush_all(MARIA_SHARE *share)
       be different.
       There should be no pinned pages as bitmap->non_flushable==0.
     */
-    if (flush_pagecache_blocks_with_filter(share->pagecache,
+    if (flush_pagecache_blocks_with_filter(bitmap->file.pagecache,
                                            &bitmap->file, FLUSH_KEEP,
                                            filter_flush_bitmap_pages,
                                            &bitmap->pages_covered) &
@@ -1079,7 +1080,7 @@ static my_bool _ma_read_bitmap_page(MARIA_HA *info,
   }
   else
   {
-    res= pagecache_read(share->pagecache,
+    res= pagecache_read(bitmap->file.pagecache,
                         &bitmap->file, page, 0,
                         bitmap->map, PAGECACHE_PLAIN_PAGE,
                         PAGECACHE_LOCK_LEFT_UNLOCKED, 0) == NULL;
@@ -2869,7 +2870,7 @@ my_bool _ma_bitmap_free_full_pages(MARIA_HA *info, const uchar *extents,
     {
       if (page == 0 && page_count == 0)
         continue;                               /* Not used extent */
-      if (pagecache_delete_pages(info->s->pagecache, &info->dfile, page,
+      if (pagecache_delete_pages(info->dfile.pagecache, &info->dfile, page,
                                  page_count, PAGECACHE_LOCK_WRITE, 1))
         DBUG_RETURN(1);
       mysql_mutex_lock(&bitmap->bitmap_lock);
@@ -3208,7 +3209,7 @@ _ma_bitmap_create_missing_into_pagecache(MARIA_SHARE *share,
       filesystem may fill gaps with zeroes physically which is a waste of
       time.
     */
-    if (pagecache_write(share->pagecache,
+    if (pagecache_write(bitmap->file.pagecache,
                         &bitmap->file, i, 0,
                         zeroes, PAGECACHE_PLAIN_PAGE,
                         PAGECACHE_LOCK_LEFT_UNLOCKED,
diff --git a/storage/maria/ma_blockrec.c b/storage/maria/ma_blockrec.c
index 84f196329ac..4da9c73ed1e 100644
--- a/storage/maria/ma_blockrec.c
+++ b/storage/maria/ma_blockrec.c
@@ -448,8 +448,10 @@ my_bool _ma_once_end_block_record(MARIA_SHARE *share)
   int res= _ma_bitmap_end(share);
   if (share->bitmap.file.file >= 0)
   {
-    if (flush_pagecache_blocks(share->pagecache, &share->bitmap.file,
-                       share->deleting ? FLUSH_IGNORE_CHANGED : FLUSH_RELEASE))
+    if (share->pagecache &&
+        flush_pagecache_blocks(share->pagecache, &share->bitmap.file,
+                               share->deleting ?
+                               FLUSH_IGNORE_CHANGED : FLUSH_RELEASE))
       res= 1;
     /*
       File must be synced as it is going out of the maria_open_list and so
diff --git a/storage/maria/ma_checkpoint.c b/storage/maria/ma_checkpoint.c
index c387f808785..e97ab0345c4 100644
--- a/storage/maria/ma_checkpoint.c
+++ b/storage/maria/ma_checkpoint.c
@@ -154,16 +154,29 @@ static int really_execute_checkpoint(void)
 {
   uint i, error= 0;
   int error_errno= 0;
+  uint dirty_pages;
   /** @brief checkpoint_start_log_horizon will be stored there */
-  char *ptr;
   const char *error_place= 0;
-  LEX_STRING record_pieces[4]; /**< only malloc-ed pieces */
+  LEX_STRING *record_pieces; /**< only malloc-ed pieces */
+  LEX_CUSTRING *log_array;
   LSN min_page_rec_lsn, min_trn_rec_lsn, min_first_undo_lsn;
   TRANSLOG_ADDRESS checkpoint_start_log_horizon;
   char checkpoint_start_log_horizon_char[LSN_STORE_SIZE];
+  uint record_pieces_count= 3+maria_pagecaches.segments;
   DBUG_ENTER("really_execute_checkpoint");
   DBUG_PRINT("enter", ("level: %d", checkpoint_in_progress));
-  bzero(&record_pieces, sizeof(record_pieces));
+
+  record_pieces= my_alloca(sizeof(*record_pieces) * record_pieces_count +
+                           sizeof(*log_array) *
+                           (TRANSLOG_INTERNAL_PARTS + 5 +
+                            maria_pagecaches.segments));
+  if (!record_pieces)
+  {
+    error_place= "allocating_memory";
+    goto err;
+  }
+  log_array= (LEX_CUSTRING*) (record_pieces + record_pieces_count);
+  bzero(record_pieces, sizeof(*record_pieces) * record_pieces_count);
 
   /*
     STEP 1: record current end-of-log position using log's lock. It is
@@ -216,9 +229,10 @@ static int really_execute_checkpoint(void)
     and thus we will have less work at Recovery.
   */
   /* Using default pagecache for now */
-  if (unlikely(pagecache_collect_changed_blocks_with_lsn(maria_pagecache,
-                                                         &record_pieces[3],
-                                                         &min_page_rec_lsn)))
+  if (unlikely(multi_pagecache_collect_changed_blocks_with_lsn(&maria_pagecaches,
+                                                               record_pieces+3,
+                                                               &min_page_rec_lsn,
+                                                               &dirty_pages)))
   {
     error_place= "collect_pages";
     goto err;
@@ -229,26 +243,49 @@ static int really_execute_checkpoint(void)
   {
     LSN lsn;
     translog_size_t total_rec_length;
+    uchar dirty_pages_buff[8];
+    LEX_CUSTRING *log_pos;
+
     /*
       the log handler is allowed to modify "str" and "length" (but not "*str")
       of its argument, so we must not pass it record_pieces directly,
       otherwise we would later not know what memory pieces to my_free().
     */
-    LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 5];
     log_array[TRANSLOG_INTERNAL_PARTS + 0].str=
       (uchar*) checkpoint_start_log_horizon_char;
     log_array[TRANSLOG_INTERNAL_PARTS + 0].length= total_rec_length=
       sizeof(checkpoint_start_log_horizon_char);
-    for (i= 0; i < (sizeof(record_pieces)/sizeof(record_pieces[0])); i++)
+
+    log_pos= log_array+ TRANSLOG_INTERNAL_PARTS + 1;
+    for (i= 0; i < 3 ; i++)
     {
-      log_array[TRANSLOG_INTERNAL_PARTS + 1 + i].str= (uchar*)record_pieces[i].str;
-      log_array[TRANSLOG_INTERNAL_PARTS + 1 + i].length= record_pieces[i].length;
+      (*log_pos).str= (uchar*)record_pieces[i].str;
+      (*log_pos).length= record_pieces[i].length;
+      log_pos++;
       total_rec_length+= (translog_size_t) record_pieces[i].length;
     }
+
+    int8store(dirty_pages_buff, (longlong) dirty_pages);
+    (*log_pos).str= dirty_pages_buff;
+    (*log_pos).length= 8;
+    log_pos++;
+    total_rec_length+= 8;
+
+    /* Copy the information about the dirty pages in the page caches */
+    for (i= 0; i < maria_pagecaches.segments; i++)
+    {
+      if (record_pieces[3+i].length)
+      {
+        (*log_pos).str= (uchar*) record_pieces[3+i].str;
+        (*log_pos).length= record_pieces[3+i].length;
+        log_pos++;
+        total_rec_length+= (translog_size_t) record_pieces[3+i].length;
+      }
+    }
     if (unlikely(translog_write_record(&lsn, LOGREC_CHECKPOINT,
                                        &dummy_transaction_object, NULL,
                                        total_rec_length,
-                                       sizeof(log_array)/sizeof(log_array[0]),
+                                       log_pos - log_array,
                                        log_array, NULL, NULL) ||
                  translog_flush(lsn)))
     {
@@ -278,8 +315,7 @@ static int really_execute_checkpoint(void)
     written the checkpoint record and control file.
   */
   /* checkpoint succeeded */
-  ptr= record_pieces[3].str;
-  pages_to_flush_before_next_checkpoint= uint4korr(ptr);
+  pages_to_flush_before_next_checkpoint= dirty_pages;
   DBUG_PRINT("checkpoint",("%u pages to flush before next checkpoint",
                           pages_to_flush_before_next_checkpoint));
 
@@ -310,8 +346,12 @@ err:
   pages_to_flush_before_next_checkpoint= 0;
 
 end:
-  for (i= 0; i < (sizeof(record_pieces)/sizeof(record_pieces[0])); i++)
-    my_free(record_pieces[i].str);
+  if (record_pieces)
+  {
+    for (i= 0; i < record_pieces_count; i++)
+      my_free(record_pieces[i].str);
+  }
+  my_afree(record_pieces);
   mysql_mutex_lock(&LOCK_checkpoint);
   checkpoint_in_progress= CHECKPOINT_NONE;
   checkpoints_total++;
@@ -570,7 +610,7 @@ pthread_handler_t ma_checkpoint_background(void *arg)
   TRANSLOG_ADDRESS log_horizon_at_last_checkpoint=
     translog_get_horizon();
   ulonglong pagecache_flushes_at_last_checkpoint=
-    maria_pagecache->global_cache_write;
+    multi_global_cache_writes(&maria_pagecaches);
   uint UNINIT_VAR(pages_bunch_size);
   struct st_filter_param filter_param;
   PAGECACHE_FILE *UNINIT_VAR(dfile); /**< data file currently being flushed */
@@ -607,6 +647,7 @@ pthread_handler_t ma_checkpoint_background(void *arg)
       }
       {
         TRANSLOG_ADDRESS horizon= translog_get_horizon();
+        ulonglong writes= multi_global_cache_writes(&maria_pagecaches);
 
         /*
           With background flushing evenly distributed over the time
@@ -624,9 +665,8 @@ pthread_handler_t ma_checkpoint_background(void *arg)
         */
         if ((ulonglong) (horizon - log_horizon_at_last_checkpoint) <=
             maria_checkpoint_min_log_activity &&
-            ((ulonglong) (maria_pagecache->global_cache_write -
-                          pagecache_flushes_at_last_checkpoint) *
-             maria_pagecache->block_size) <=
+            ((ulonglong) (writes - pagecache_flushes_at_last_checkpoint) *
+             maria_pagecaches.caches->block_size) <=
             maria_checkpoint_min_cache_activity)
         {
           /*
@@ -643,8 +683,7 @@ pthread_handler_t ma_checkpoint_background(void *arg)
           below is possibly greater than last_checkpoint_lsn.
         */
         log_horizon_at_last_checkpoint= translog_get_horizon();
-        pagecache_flushes_at_last_checkpoint=
-          maria_pagecache->global_cache_write;
+        pagecache_flushes_at_last_checkpoint= writes;
         /*
           If the checkpoint above succeeded it has set d|kfiles and
           d|kfiles_end. If is has failed, it has set
@@ -685,7 +724,7 @@ pthread_handler_t ma_checkpoint_background(void *arg)
             only the OS file descriptor.
           */
           int res=
-            flush_pagecache_blocks_with_filter(maria_pagecache,
+            flush_pagecache_blocks_with_filter(dfile->pagecache,
                                                dfile, FLUSH_KEEP_LAZY,
                                                filter_flush_file_evenly,
                                                &filter_param);
@@ -705,7 +744,7 @@ pthread_handler_t ma_checkpoint_background(void *arg)
         while (kfile != kfiles_end)
         {
           int res=
-            flush_pagecache_blocks_with_filter(maria_pagecache,
+            flush_pagecache_blocks_with_filter(kfile->pagecache,
                                                kfile, FLUSH_KEEP_LAZY,
                                                filter_flush_file_evenly,
                                                &filter_param);
@@ -983,7 +1022,7 @@ static int collect_tables(LEX_STRING *str, LSN checkpoint_start_log_horizon)
       Tables in a normal state have their two file descriptors open.
       In some rare cases like REPAIR, some descriptor may be closed or even
       -1. If that happened, the _ma_state_info_write() may fail. This is
-      prevented by enclosing all all places which close/change kfile.file with
+      prevented by enclosing all places which close/change kfile.file with
       intern_lock.
     */
     kfile= share->kfile;
@@ -1104,7 +1143,6 @@ static int collect_tables(LEX_STRING *str, LSN checkpoint_start_log_horizon)
         /** @todo all write failures should mark table corrupted */
         ma_message_no_user(0, "checkpoint bitmap page flush failed");
       }
-      DBUG_ASSERT(share->pagecache == maria_pagecache);
     }
     /*
       Clean up any unused states.
@@ -1189,12 +1227,12 @@ static int collect_tables(LEX_STRING *str, LSN checkpoint_start_log_horizon)
     {
       if (filter != NULL)
       {
-        if ((flush_pagecache_blocks_with_filter(maria_pagecache,
+        if ((flush_pagecache_blocks_with_filter(dfile.pagecache,
                                                 &dfile, FLUSH_KEEP_LAZY,
                                                 filter, &filter_param) &
              PCFLUSH_ERROR))
           ma_message_no_user(0, "checkpoint data page flush failed");
-        if ((flush_pagecache_blocks_with_filter(maria_pagecache,
+        if ((flush_pagecache_blocks_with_filter(kfile.pagecache,
                                                 &kfile, FLUSH_KEEP_LAZY,
                                                 filter, &filter_param) &
              PCFLUSH_ERROR))
diff --git a/storage/maria/ma_close.c b/storage/maria/ma_close.c
index 47935d61b88..f17f7e1b100 100644
--- a/storage/maria/ma_close.c
+++ b/storage/maria/ma_close.c
@@ -42,8 +42,10 @@ int maria_close(register MARIA_HA *info)
   DBUG_ASSERT(info->key_del_used == 0);
   /* Check that file is not part of any uncommitted transactions */
   DBUG_ASSERT(info->trn == 0 || info->trn == &dummy_transaction_object);
+  DBUG_ASSERT(info->dfile.pagecache == info->s->kfile.pagecache);
 
-  if (share->reopen == 1)
+  /* pagecache can be 0 if we come here from maria_recreate_table */
+  if (share->reopen == 1 && share->pagecache)
   {
     /*
       If we are going to close the file, flush page cache without
@@ -123,8 +125,11 @@ int maria_close(register MARIA_HA *info)
         Extra flush, just in case someone opened and closed the file
         since the start of the function (very unlikely)
       */
-      if (flush_pagecache_blocks(share->pagecache, &share->kfile,
-                        share->deleting ? FLUSH_IGNORE_CHANGED : FLUSH_RELEASE))
+      if (share->pagecache &&
+          flush_pagecache_blocks(share->pagecache, &share->kfile,
+                                 share->deleting ?
+                                 FLUSH_IGNORE_CHANGED :
+                                 FLUSH_RELEASE))
         error= my_errno;
       unmap_file(info);
       if (!internal_table &&
diff --git a/storage/maria/ma_init.c b/storage/maria/ma_init.c
index 6873c94a0ed..4f952b6527b 100644
--- a/storage/maria/ma_init.c
+++ b/storage/maria/ma_init.c
@@ -108,7 +108,7 @@ void maria_end(void)
     if (translog_status == TRANSLOG_OK || translog_status == TRANSLOG_READONLY)
       translog_destroy();
     end_pagecache(maria_log_pagecache, TRUE);
-    end_pagecache(maria_pagecache, TRUE);
+    multi_end_pagecache(&maria_pagecaches);
     ma_control_file_end();
     mysql_mutex_destroy(&THR_LOCK_maria);
     my_hash_free(&maria_stored_state);
diff --git a/storage/maria/ma_keycache.c b/storage/maria/ma_keycache.c
index 2ff8d019d1c..76e9564f705 100644
--- a/storage/maria/ma_keycache.c
+++ b/storage/maria/ma_keycache.c
@@ -46,6 +46,7 @@
     #  Error code
 */
 
+#ifdef NOT_USED
 int maria_assign_to_pagecache(MARIA_HA *info,
                               ulonglong key_map __attribute__((unused)),
                               PAGECACHE *pagecache)
@@ -112,6 +113,7 @@ int maria_assign_to_pagecache(MARIA_HA *info,
   mysql_mutex_unlock(&share->intern_lock);
   DBUG_RETURN(error);
 }
+#endif
 
 
 /*
@@ -135,6 +137,7 @@ int maria_assign_to_pagecache(MARIA_HA *info,
 */
 
 
+#ifdef NOT_USED
 void maria_change_pagecache(PAGECACHE *old_pagecache,
                             PAGECACHE *new_pagecache)
 {
@@ -162,3 +165,4 @@ void maria_change_pagecache(PAGECACHE *old_pagecache,
   mysql_mutex_unlock(&THR_LOCK_maria);
   DBUG_VOID_RETURN;
 }
+#endif
diff --git a/storage/maria/ma_open.c b/storage/maria/ma_open.c
index a4d1beb4a0f..56be2ca76db 100644
--- a/storage/maria/ma_open.c
+++ b/storage/maria/ma_open.c
@@ -106,7 +106,10 @@ static MARIA_HA *maria_clone_internal(MARIA_SHARE *share,
     goto err;
   }
   if (data_file >= 0)
+  {
     info.dfile.file= data_file;
+    info.dfile.pagecache= share->pagecache;
+  }
   else if (_ma_open_datafile(&info, share))
     goto err;
   errpos= 5;
@@ -332,9 +335,6 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags,
     share= &share_buff;
     bzero((uchar*) &share_buff,sizeof(share_buff));
     share_buff.state.key_root=key_root;
-    share_buff.pagecache= multi_pagecache_search((uchar*) name_buff,
-						 (uint) strlen(name_buff),
-                                                 maria_pagecache);
 
     if (!s3)
     {
@@ -609,7 +609,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags,
       with different block sizes.
     */
     if (share->base.block_size != maria_block_size &&
-        share_buff.pagecache->inited != 0)
+        maria_pagecaches.initialized)
     {
       DBUG_PRINT("error", ("Wrong block size %u; Expected %u",
                            (uint) share->base.block_size,
@@ -847,6 +847,10 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags,
                          share->base.pack_bytes +
                          MY_TEST(share->options & HA_OPTION_CHECKSUM));
     share->kfile.file= kfile;
+    /* Pagecaches are not initialize when using aria_chk */
+    if (maria_pagecaches.initialized)
+      share->pagecache= share->kfile.pagecache=
+        multi_get_pagecache(&maria_pagecaches);
 
     if (open_flags & HA_OPEN_COPY)
     {
@@ -2056,6 +2060,8 @@ int _ma_open_datafile(MARIA_HA *info, MARIA_SHARE *share)
   info->dfile.file= share->bitmap.file.file=
     mysql_file_open(key_file_dfile, share->data_file_name.str,
                     share->mode | O_SHARE | O_CLOEXEC, flags);
+  /* Note that share->pagecache may be 0 here if run from aria_chk */
+  info->dfile.pagecache= share->bitmap.file.pagecache= share->pagecache;
   return info->dfile.file >= 0 ? 0 : 1;
 }
 
@@ -2071,11 +2077,30 @@ int _ma_open_keyfile(MARIA_SHARE *share)
                                      share->unique_file_name.str,
                                      share->mode | O_SHARE | O_NOFOLLOW | O_CLOEXEC,
                              MYF(MY_WME | MY_NOSYMLINKS));
+  /*
+    share->kfile.pagecache is updated in the caller if needed.
+    This is needed as we don't want to change pagecache in ma_sort_index()
+    as we want to check pagecache concistency in ma_close().
+  */
   mysql_mutex_unlock(&share->intern_lock);
   return (share->kfile.file < 0);
 }
 
 
+/*
+  Update pagecaches for a table. Used by aria_check() which creates
+  the pagecache after the table has been opened.
+*/
+
+void ma_change_pagecache(MARIA_HA *info)
+{
+  MARIA_SHARE *share= info->s;
+  DBUG_ASSERT(share->pagecache == 0);
+  share->pagecache= share->bitmap.file.pagecache= share->kfile.pagecache=
+    info->dfile.pagecache= multi_get_pagecache(&maria_pagecaches);
+}
+
+
 /*
   Disable all indexes.
 
diff --git a/storage/maria/ma_pagecache.c b/storage/maria/ma_pagecache.c
index c00b3cb88fb..6fef97fa045 100644
--- a/storage/maria/ma_pagecache.c
+++ b/storage/maria/ma_pagecache.c
@@ -89,6 +89,11 @@
 #define PAGECACHE_DEBUG_LOG  "my_pagecache_debug.log"
 #define _VARARGS(X) X
 
+/* For debugging of pagecache */
+#ifdef EXTRA_DEBUG_BITMAP
+static my_bool pagecache_extra_debug= 1;
+#endif
+
 /*
   In key cache we have external raw locking here we use
   SERIALIZED_READ_FROM_CACHE to avoid problem of reading
@@ -654,7 +659,7 @@ static my_bool pagecache_fwrite(PAGECACHE *pagecache,
     debug either of the above issues.
   */
 
-  if (pagecache->extra_debug)
+  if (pagecache_extra_debug)
   {
     char buff[80];
     uint len= my_sprintf(buff,
@@ -795,6 +800,7 @@ size_t init_pagecache(PAGECACHE *pagecache, size_t use_mem,
   pagecache->disk_blocks= -1;
   if (! pagecache->inited)
   {
+    my_hash_clear(&pagecache->files_in_flush);
     if (mysql_mutex_init(key_PAGECACHE_cache_lock,
                          &pagecache->cache_lock, MY_MUTEX_INIT_FAST) ||
         my_hash_init(PSI_INSTRUMENT_ME, &pagecache->files_in_flush,
@@ -938,6 +944,7 @@ err:
     my_free(pagecache->block_root);
     pagecache->block_root= NULL;
   }
+  my_hash_free(&pagecache->files_in_flush);
   my_errno= error;
   pagecache->can_be_used= 0;
   DBUG_RETURN(0);
@@ -5233,8 +5240,6 @@ int flush_pagecache_blocks_with_filter(PAGECACHE *pagecache,
   DBUG_ENTER("flush_pagecache_blocks_with_filter");
   DBUG_PRINT("enter", ("pagecache: %p  fd: %di", pagecache, file->file));
 
-  if (pagecache->disk_blocks <= 0)
-    DBUG_RETURN(0);
   pagecache_pthread_mutex_lock(&pagecache->cache_lock);
   inc_counter_for_resize_op(pagecache);
   res= flush_pagecache_blocks_int(pagecache, file, type, filter, filter_arg);
@@ -5290,11 +5295,14 @@ int reset_pagecache_counters(const char *name __attribute__((unused)),
    are not interesting for a checkpoint record.
    The caller has the intention of doing checkpoints.
 
-   @param       pagecache   pointer to the page cache
-   @param[out]  str         pointer to where the allocated buffer, and
-                            its size, will be put
-   @param[out]  min_rec_lsn pointer to where the minimum rec_lsn of all
-                            relevant dirty pages will be put
+   @param       pagecache     pointer to the page cache
+   @param[out]  str           pointer to where the allocated buffer, and
+                              its size, will be put
+   @param[in|out] min_rec_lsn pointer to where the minimum rec_lsn of all
+                              relevant dirty pages will be put.
+                              Note the original value is used as current
+                              min value!
+   @param list_size           Number of dirty_pages
    @return Operation status
      @retval 0      OK
      @retval 1      Error
@@ -5302,16 +5310,17 @@ int reset_pagecache_counters(const char *name __attribute__((unused)),
 
 my_bool pagecache_collect_changed_blocks_with_lsn(PAGECACHE *pagecache,
                                                   LEX_STRING *str,
-                                                  LSN *min_rec_lsn)
+                                                  LSN *min_rec_lsn,
+                                                  uint *dirty_pages)
 {
   my_bool error= 0;
-  size_t stored_list_size= 0;
+  uint stored_list_size= 0;
   uint file_hash;
+  LSN minimum_rec_lsn= *min_rec_lsn;
   char *ptr;
-  LSN minimum_rec_lsn= LSN_MAX;
-  DBUG_ENTER("pagecache_collect_changed_blocks_with_LSN");
+  DBUG_ENTER("pagecache_collect_changed_blocks_with_lsn");
 
-  DBUG_ASSERT(NULL == str->str);
+  DBUG_ASSERT(!str->str);
   /*
     We lock the entire cache but will be quick, just reading/writing a few MBs
     of memory at most.
@@ -5379,20 +5388,18 @@ my_bool pagecache_collect_changed_blocks_with_lsn(PAGECACHE *pagecache,
   }
 
   compile_time_assert(sizeof(pagecache->blocks) <= 8);
-  str->length= 8 + /* number of dirty pages */
-    (2 + /* table id */
-     1 + /* data or index file */
-     5 + /* pageno */
-     LSN_STORE_SIZE /* rec_lsn */
-     ) * stored_list_size;
-  if (NULL == (str->str= my_malloc(PSI_INSTRUMENT_ME, str->length, MYF(MY_WME))))
-    goto err;
-  ptr= str->str;
-  int8store(ptr, (ulonglong)stored_list_size);
-  ptr+= 8;
-  DBUG_PRINT("info", ("found %zu dirty pages", stored_list_size));
   if (stored_list_size == 0)
     goto end;
+
+  str->length= ((2 + /* table id */
+                 1 + /* data or index file */
+                 5 + /* pageno */
+                 LSN_STORE_SIZE /* rec_lsn */
+                 ) * stored_list_size);
+  if (!(str->str= my_malloc(PSI_INSTRUMENT_ME, str->length, MYF(MY_WME))))
+    goto err;
+  ptr= str->str;
+  DBUG_PRINT("info", ("found %u dirty pages", stored_list_size));
   for (file_hash= 0; file_hash < pagecache->changed_blocks_hash_size; file_hash++)
   {
     PAGECACHE_BLOCK_LINK *block;
@@ -5426,9 +5433,11 @@ my_bool pagecache_collect_changed_blocks_with_lsn(PAGECACHE *pagecache,
 end:
   pagecache_pthread_mutex_unlock(&pagecache->cache_lock);
   *min_rec_lsn= minimum_rec_lsn;
+  *dirty_pages= stored_list_size;
   DBUG_RETURN(error);
 
 err:
+  stored_list_size= 0;
   error= 1;
   goto end;
 }
diff --git a/storage/maria/ma_pagecache.h b/storage/maria/ma_pagecache.h
index dbd86fc0def..38c183445b2 100644
--- a/storage/maria/ma_pagecache.h
+++ b/storage/maria/ma_pagecache.h
@@ -97,7 +97,6 @@ typedef struct st_S3_BLOCK
   size_t length;
 } S3_BLOCK;
 
-
 /* file descriptor for Maria */
 typedef struct st_pagecache_file
 {
@@ -120,6 +119,7 @@ typedef struct st_pagecache_file
 
   /** Cannot be NULL */
   uchar *callback_data;
+  struct st_pagecache *pagecache;
 } PAGECACHE_FILE;
 
 /* declare structures that is used by  st_pagecache */
@@ -216,7 +216,7 @@ typedef struct st_pagecache
   my_bool resize_in_flush;       /* true during flush of resize operation    */
   my_bool can_be_used;           /* usage of cache for read/write is allowed */
   my_bool in_init;		/* Set to 1 in MySQL during init/resize     */
-  my_bool extra_debug;	        /* set to 1 if one wants extra logging */
+  my_bool multi;                /* If part of segmented pagecache */
   HASH    files_in_flush;       /**< files in flush_pagecache_blocks_int() */
 } PAGECACHE;
 
@@ -338,7 +338,8 @@ extern my_bool pagecache_delete_pages(PAGECACHE *pagecache,
 extern void end_pagecache(PAGECACHE *keycache, my_bool cleanup)__attribute__((visibility("default"))) ;
 extern my_bool pagecache_collect_changed_blocks_with_lsn(PAGECACHE *pagecache,
                                                          LEX_STRING *str,
-                                                         LSN *min_lsn);
+                                                         LSN *min_lsn,
+                                                         uint *dirt_pages);
 extern int reset_pagecache_counters(const char *name, PAGECACHE *pagecache);
 extern uchar *pagecache_block_link_to_buffer(PAGECACHE_BLOCK_LINK *block);
 
@@ -347,19 +348,58 @@ extern void pagecache_add_level_by_link(PAGECACHE_BLOCK_LINK *block,
 					uint level);
 
 /* Functions to handle multiple key caches */
-extern my_bool multi_pagecache_init(void);
-extern void multi_pagecache_free(void);
-extern PAGECACHE *multi_pagecache_search(uchar *key, uint length,
-                                         PAGECACHE *def);
-extern my_bool multi_pagecache_set(const uchar *key, uint length,
-				   PAGECACHE *pagecache);
-extern void multi_pagecache_change(PAGECACHE *old_data,
-				   PAGECACHE *new_data);
+
+typedef struct st_pagecaches
+{
+  PAGECACHE *caches;
+  ulonglong requests;
+  uint segments;
+  my_bool initialized;
+} PAGECACHES;
+
+my_bool multi_init_pagecache(PAGECACHES *pagecaches, ulong segments,
+                             size_t use_mem, uint division_limit,
+                             uint age_threshold,
+                             uint block_size, uint changed_blocks_hash_size,
+                             myf my_readwrite_flags);
+extern void multi_end_pagecache(PAGECACHES *pagecaches);
+extern my_bool
+multi_pagecache_collect_changed_blocks_with_lsn(PAGECACHES *pagecaches,
+                                                LEX_STRING *str,
+                                                LSN *min_rec_lsn,
+                                                uint *dirty_pages);
+
+static inline PAGECACHE *multi_get_pagecache(PAGECACHES *pagecaches)
+{
+  return pagecaches->caches + (pagecaches->requests++ % pagecaches->segments);
+}
+
+/* Pagecache stats */
+
+struct st_pagecache_stats
+{
+  size_t blocks_used;          /* maximum number of concurrently used blocks */
+  size_t blocks_unused;          /* number of currently unused blocks        */
+  size_t blocks_changed;         /* number of currently dirty blocks         */
+
+  size_t global_blocks_changed;	/* number of currently dirty blocks          */
+  ulonglong global_cache_w_requests;/* number of write requests (write hits) */
+  ulonglong global_cache_write;     /* number of writes from cache to files  */
+  ulonglong global_cache_r_requests;/* number of read requests (read hits)   */
+  ulonglong global_cache_read;      /* number of reads from files to cache   */
+};
+
+/* Stats will be updated when multi_update_pagecache_stats() is called */
+extern struct st_pagecache_stats pagecache_stats;
+void multi_update_pagecache_stats();
+ulonglong multi_global_cache_writes(PAGECACHES *pagecaches);
+
 #ifndef DBUG_OFF
 void pagecache_file_no_dirty_page(PAGECACHE *pagecache, PAGECACHE_FILE *file);
 #else
 #define pagecache_file_no_dirty_page(A,B) {}
 #endif
 
+
 C_MODE_END
-#endif /* _keycache_h */
+#endif /* _ma_pagecache_h */
diff --git a/storage/maria/ma_pagecaches.c b/storage/maria/ma_pagecaches.c
index fd5cd2092e1..e459a02829c 100644
--- a/storage/maria/ma_pagecaches.c
+++ b/storage/maria/ma_pagecaches.c
@@ -1,4 +1,4 @@
-/* Copyright (C) 2003-2007 MySQL AB
+/* Copyright (C) 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
@@ -14,11 +14,10 @@
    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA */
 
 /*
-  Handling of multiple key caches
+  Handling of multiple key caches in Aria
 
-  The idea is to have a thread safe hash on the table name,
-  with a default key cache value that is returned if the table name is not in
-  the cache.
+  Each data and index pages for a table is put in the same cache, based on the
+  filenumber of that index file.
 */
 
 #include "maria_def.h"
@@ -31,74 +30,167 @@
   Functions to handle the pagecache objects
 *****************************************************************************/
 
-/* Variable to store all key cache objects */
-static SAFE_HASH pagecache_hash;
+/**
+   Init 'segments' number of independent pagecaches
 
-
-my_bool multi_pagecache_init(void)
-{
-  return safe_hash_init(&pagecache_hash, 16, (uchar*) maria_pagecache);
-}
-
-
-void multi_pagecache_free(void)
-{
-  safe_hash_free(&pagecache_hash);
-}
-
-/*
-  Get a key cache to be used for a specific table.
-
-  SYNOPSIS
-    multi_pagecache_search()
-    key				key to find (usually table path)
-    uint length			Length of key.
-    def				Default value if no key cache
-
-  NOTES
-    This function is coded in such a way that we will return the
-    default key cache even if one never called multi_pagecache_init.
-    This will ensure that it works with old MyISAM clients.
-
-  RETURN
-    key cache to use
+   @return 0 ok
+   @return 1 error
 */
 
-PAGECACHE *multi_pagecache_search(uchar *key, uint length,
-                                  PAGECACHE *def)
+my_bool multi_init_pagecache(PAGECACHES *pagecaches, ulong segments,
+                             size_t use_mem, uint division_limit,
+                             uint age_threshold,
+                             uint block_size, uint changed_blocks_hash_size,
+                             myf my_readwrite_flags)
 {
-  if (!pagecache_hash.hash.records)
-    return def;
-  return (PAGECACHE*) safe_hash_search(&pagecache_hash, key, length,
-                                       (void*) def);
+  PAGECACHE *pagecache;
+  DBUG_ENTER("init_pagecaches");
+
+  pagecaches->initialized= 0;
+  if (!(pagecaches->caches= ((PAGECACHE*)
+                             my_malloc(PSI_INSTRUMENT_ME,
+                                       sizeof(PAGECACHE) * segments,
+                                       MYF(MY_FAE | MY_ZEROFILL)))))
+    DBUG_RETURN(1);
+
+  pagecache= pagecaches->caches;
+  pagecaches->segments= segments;
+  for (ulong i= 0; i < segments ; i++, pagecache++)
+  {
+    if (init_pagecache(pagecache, MY_MAX(KEY_CACHE_SIZE, use_mem / segments),
+                       division_limit, age_threshold,
+                       block_size, changed_blocks_hash_size,
+                       my_readwrite_flags) == 0)
+      goto err;
+    pagecache->multi= 1;                        /* Part of segemented cache */
+  }
+  pagecaches->initialized= 1;
+  DBUG_RETURN(0);
+
+err:
+  while (pagecache-- != pagecaches->caches)
+    end_pagecache(pagecache, TRUE);
+  my_free(pagecaches->caches);
+  pagecaches->caches= 0;                        /* For easier debugging */
+  DBUG_RETURN(1);
+}
+
+
+void multi_end_pagecache(PAGECACHES *pagecaches)
+{
+  DBUG_ENTER("end_pagecaches");
+
+  if (unlikely(!pagecaches->initialized))
+    DBUG_VOID_RETURN;
+
+  for (ulong i= 0; i < pagecaches->segments ; i++)
+    end_pagecache(pagecaches->caches+i, TRUE);
+
+  my_free(pagecaches->caches);
+  pagecaches->caches= 0;
+  pagecaches->initialized= 0;
+  pagecaches->segments= 0;
+  DBUG_VOID_RETURN;
 }
 
 
 /*
-  Assosiate a key cache with a key
+  Call pagecache_collect_changed_blocks_with_len() over all
+  pagecaches
 
-
-  SYONOPSIS
-    multi_pagecache_set()
-    key				key (path to table etc..)
-    length			Length of key
-    pagecache			cache to assococite with the table
-
-  NOTES
-    This can be used both to insert a new entry and change an existing
-    entry
+  @param pagecaches        The partitioned page cache
+  @param str               Buffers for changed blocks
+  @param min_rec_lsn[out]  Min rec lsn in pagecache
+  @param dirty_pages[out]  Total dirty pages found
 */
 
-
-my_bool multi_pagecache_set(const uchar *key, uint length,
-			    PAGECACHE *pagecache)
+my_bool multi_pagecache_collect_changed_blocks_with_lsn(PAGECACHES *pagecaches,
+                                                        LEX_STRING *str,
+                                                        LSN *min_rec_lsn,
+                                                        uint *dirty_pages)
 {
-  return safe_hash_set(&pagecache_hash, key, length, (uchar*) pagecache);
+  uint pages= 0;
+  DBUG_ENTER("multi_pagecache_collect_changed_blocks_with_lsn");
+  *min_rec_lsn= LSN_MAX;
+  for (ulong i= 0; i < pagecaches->segments ; i++, str++)
+  {
+    uint tmp_dirty;
+    if (unlikely(pagecache_collect_changed_blocks_with_lsn(pagecaches->caches+i,
+                                                           str,
+                                                           min_rec_lsn,
+                                                           &tmp_dirty)))
+    {
+      /* Free the collected checkpoint information */
+      do
+      {
+        my_free(str->str);
+        str->str= 0;
+        str--;
+      } while (i-- > 0);
+      *dirty_pages= 0;
+      DBUG_RETURN(1);
+    }
+    pages+= tmp_dirty;
+  }
+  *dirty_pages= pages;
+  DBUG_RETURN(0);
 }
 
 
-void multi_pagecache_change(PAGECACHE *old_data,
-			    PAGECACHE *new_data)
+struct st_pagecache_stats pagecache_stats;
+
+/*
+  Update the global pagecache status
+  This function is called when accessing status variables
+*/
+
+void multi_update_pagecache_stats()
 {
-  safe_hash_change(&pagecache_hash, (uchar*) old_data, (uchar*) new_data);
+  struct st_pagecache_stats new;
+  bzero(&new, sizeof(new));
+  for (uint i= 0 ; i < maria_pagecaches.segments ; i++)
+  {
+    PAGECACHE *pagecache= maria_pagecaches.caches + i;
+    new.blocks_used+=             pagecache->blocks_used;
+    new.blocks_unused+=           pagecache->blocks_unused;
+    new.blocks_changed+=          pagecache->blocks_changed;
+    new.global_blocks_changed+=   pagecache->global_blocks_changed;
+    new.global_cache_w_requests+= pagecache->global_cache_w_requests;
+    new.global_cache_write+=      pagecache->global_cache_write;
+    new.global_cache_r_requests+= pagecache->global_cache_r_requests;
+    new.global_cache_read+=       pagecache->global_cache_read;
+  }
+  pagecache_stats.blocks_used=             new.blocks_used;
+  pagecache_stats.blocks_unused=           new.blocks_unused;
+  pagecache_stats.blocks_changed=          new.blocks_changed;
+  pagecache_stats.global_blocks_changed=   new.global_blocks_changed;
+  pagecache_stats.global_cache_w_requests= new.global_cache_w_requests;
+  pagecache_stats.global_cache_write=      new.global_cache_write;
+  pagecache_stats.global_cache_r_requests= new.global_cache_r_requests;
+  pagecache_stats.global_cache_read=       new.global_cache_read;
+};
+
+
+/*
+  Get the total writes to the pagecaches
+*/
+
+ulonglong multi_global_cache_writes(PAGECACHES *pagecaches)
+{
+  ulonglong count= 0;
+  for (ulong i= 0; i < pagecaches->segments ; i++)
+    count+= pagecaches->caches[i].global_cache_write;
+  return count;
+}
+
+
+/*
+  Reset pagecache statistics
+*/
+
+void multi_reset_pagecache_counters(PAGECACHES *pagecaches)
+{
+  for (ulong i= 0; i < pagecaches->segments ; i++)
+    reset_pagecache_counters(NullS, pagecaches->caches+i);
+  multi_update_pagecache_stats();
 }
diff --git a/storage/maria/ma_recovery.c b/storage/maria/ma_recovery.c
index 061c653460d..960330b114e 100644
--- a/storage/maria/ma_recovery.c
+++ b/storage/maria/ma_recovery.c
@@ -236,7 +236,7 @@ int maria_recovery_from_log(void)
   trace_file= NULL; /* no trace file for being fast */
 #endif
   tprint(trace_file, "TRACE of the last Aria recovery from mysqld\n");
-  DBUG_ASSERT(maria_pagecache->inited);
+  DBUG_ASSERT(maria_pagecaches.initialized);
   res= maria_apply_log(LSN_IMPOSSIBLE, LSN_IMPOSSIBLE, 0, MARIA_LOG_APPLY,
                        trace_file, TRUE, TRUE, &warnings_count);
   if (!res)
diff --git a/storage/maria/ma_rt_test.c b/storage/maria/ma_rt_test.c
index 1c960737528..12fd4102fc8 100644
--- a/storage/maria/ma_rt_test.c
+++ b/storage/maria/ma_rt_test.c
@@ -96,8 +96,8 @@ int main(int argc, char *argv[])
   get_options(argc, argv);
   /* Maria requires that we always have a page cache */
   if (maria_init() ||
-      (init_pagecache(maria_pagecache, maria_block_size * 16, 0, 0,
-                      maria_block_size, 0, MY_WME) == 0) ||
+      (multi_init_pagecache(&maria_pagecaches, 1, maria_block_size * 16, 0, 0,
+                            maria_block_size, 0, MY_WME)) ||
       ma_control_file_open_or_create() ||
       (init_pagecache(maria_log_pagecache,
                       TRANSLOG_PAGECACHE_SIZE, 0, 0,
diff --git a/storage/maria/ma_static.c b/storage/maria/ma_static.c
index 2c9f1dcc213..168655630f2 100644
--- a/storage/maria/ma_static.c
+++ b/storage/maria/ma_static.c
@@ -55,10 +55,9 @@ ulong maria_concurrent_insert= 2;
 my_off_t maria_max_temp_length= MAX_FILE_SIZE;
 ulong    maria_bulk_insert_tree_size=8192*1024;
 ulong    maria_data_pointer_size= 6;
+ulong    maria_pagecache_segments;
 
-PAGECACHE maria_pagecache_var;
-PAGECACHE *maria_pagecache= &maria_pagecache_var;
-
+PAGECACHES maria_pagecaches;
 PAGECACHE maria_log_pagecache_var;
 PAGECACHE *maria_log_pagecache= &maria_log_pagecache_var;
 MY_TMPDIR *maria_tmpdir;                        /* Tempdir for redo */
diff --git a/storage/maria/ma_test1.c b/storage/maria/ma_test1.c
index a14679d3a60..92528ce3b52 100644
--- a/storage/maria/ma_test1.c
+++ b/storage/maria/ma_test1.c
@@ -79,8 +79,8 @@ int main(int argc,char *argv[])
   get_options(argc,argv);
   /* Maria requires that we always have a page cache */
   if (maria_init() ||
-      (init_pagecache(maria_pagecache, maria_block_size * 16, 0, 0,
-                      maria_block_size, 0, MY_WME) == 0) ||
+      (multi_init_pagecache(&maria_pagecaches, 1, maria_block_size * 16, 0, 0,
+                      maria_block_size, 0, MY_WME)) ||
       ma_control_file_open_or_create() ||
       (init_pagecache(maria_log_pagecache,
                       TRANSLOG_PAGECACHE_SIZE, 0, 0,
diff --git a/storage/maria/ma_test2.c b/storage/maria/ma_test2.c
index 400e6193695..71766a9a2e1 100644
--- a/storage/maria/ma_test2.c
+++ b/storage/maria/ma_test2.c
@@ -44,6 +44,7 @@ static int create_flag= 0, srand_arg= 0, checkpoint= 0;
 static my_bool opt_versioning= 0;
 static uint use_blob= 0, update_count= 0;
 static ulong pagecache_size=8192*32;
+static ulong pagecache_segments= 1;
 static enum data_file_type record_type= DYNAMIC_RECORD;
 
 static uint keys=MARIA_KEYS,recant=1000;
@@ -88,8 +89,9 @@ int main(int argc, char *argv[])
 
   /* Maria requires that we always have a page cache */
   if (maria_init() ||
-      (init_pagecache(maria_pagecache, pagecache_size, 0, 0,
-		      maria_block_size, 0, MY_WME) == 0) ||
+      (multi_init_pagecache(&maria_pagecaches, pagecache_segments,
+                            pagecache_size, 0, 0,
+                            maria_block_size, 0, MY_WME)) ||
       ma_control_file_open_or_create() ||
       (init_pagecache(maria_log_pagecache,
 		      TRANSLOG_PAGECACHE_SIZE, 0, 0,
@@ -982,6 +984,7 @@ end:
     goto err;
   }
   file= 0;
+  multi_update_pagecache_stats();
   maria_panic(HA_PANIC_CLOSE);			/* Should close log */
   if (!silent)
   {
@@ -1009,12 +1012,12 @@ w_requests: %10lu\n\
 writes:     %10lu\n\
 r_requests: %10lu\n\
 reads:      %10lu\n",
-           (ulong) maria_pagecache->blocks_used,
-           (ulong) maria_pagecache->global_blocks_changed,
-           (ulong) maria_pagecache->global_cache_w_requests,
-           (ulong) maria_pagecache->global_cache_write,
-           (ulong) maria_pagecache->global_cache_r_requests,
-           (ulong) maria_pagecache->global_cache_read);
+           (ulong) pagecache_stats.blocks_used,
+           (ulong) pagecache_stats.global_blocks_changed,
+           (ulong) pagecache_stats.global_cache_w_requests,
+           (ulong) pagecache_stats.global_cache_write,
+           (ulong) pagecache_stats.global_cache_r_requests,
+           (ulong) pagecache_stats.global_cache_read);
   }
   maria_end();
   my_free(blob_buffer);
@@ -1127,6 +1130,10 @@ static void get_options(int argc, char **argv)
       pack_type=0;			/* Don't use DIFF_LENGTH */
       pack_seg=0;
       break;
+    case 'p':                           /* Segmented page cache */
+      pagecache_segments= atoi(++pos);
+      pagecache_size*= pagecache_segments;
+      break;
     case 'R':				/* Length of record pointer */
       rec_pointer_size=atoi(++pos);
       if (rec_pointer_size > 7)
diff --git a/storage/maria/ma_test3.c b/storage/maria/ma_test3.c
index 411a5f6ce0b..477d0544514 100644
--- a/storage/maria/ma_test3.c
+++ b/storage/maria/ma_test3.c
@@ -177,9 +177,9 @@ void start_test(int id)
     fprintf(stderr,"Can't open isam-file: %s\n",filename);
     exit(1);
   }
-  if (pagecacheing && rnd(2) == 0)
-    init_pagecache(maria_pagecache, 65536L, 0, 0, MARIA_KEY_BLOCK_LENGTH, 0,
-                   MY_WME);
+  multi_init_pagecache(&maria_pagecaches, 1, 65536L, 0, 0,
+                       MARIA_KEY_BLOCK_LENGTH, 0, MY_WME);
+
   printf("Process %d, pid: %ld\n",id,(long) getpid()); fflush(stdout);
 
   for (error=i=0 ; i < tests && !error; i++)
diff --git a/storage/maria/ma_test_big.sh b/storage/maria/ma_test_big.sh
old mode 100644
new mode 100755
index 6419d05e3a4..41260151cd7
--- a/storage/maria/ma_test_big.sh
+++ b/storage/maria/ma_test_big.sh
@@ -4,19 +4,23 @@
 # finding bugs in blob handling
 #
 
+mkdir -p tmp
+cd tmp
 set -e
 a=15
 while test $a -le 5000
 do
   echo $a
-  rm -f maria_log*
-  ma_test2 -s -L -K -W -P -M -T -c -b32768 -t4 -A1 -m$a > /dev/null
-  maria_read_log -a -s >& /dev/null
-  maria_chk -es test2
-  maria_read_log -a -s >& /dev/null
-  maria_chk -es test2
+  rm -f aria_log*
+  ../ma_test2 -s -L -K -W -P -M -T -c -b32768 -t4 -A1 -m$a > /dev/null
+  ../aria_read_log -a -s >& /dev/null
+  ../aria_chk -ess test2
+  ../aria_read_log -a -s >& /dev/null
+  ../aria_chk -ess test2
   rm test2.MA?
-  maria_read_log -a -s >& /dev/null
-  maria_chk -es test2
+  ../aria_read_log -a -s >& /dev/null
+  ../aria_chk -ess test2
   a=$((a+1))
 done
+cd ..
+rm -r tmp
diff --git a/storage/maria/maria_def.h b/storage/maria/maria_def.h
index 7234df00e76..247ac0b3e5f 100644
--- a/storage/maria/maria_def.h
+++ b/storage/maria/maria_def.h
@@ -208,6 +208,7 @@ extern int maria_close(MARIA_HA *file);
 extern int maria_delete(MARIA_HA *file, const uchar *buff);
 extern MARIA_HA *maria_open(const char *name, int mode,
                             uint wait_if_locked, S3_INFO *s3);
+extern void ma_change_pagecache(MARIA_HA *info);
 extern int maria_panic(enum ha_panic_function function);
 extern int maria_rfirst(MARIA_HA *file, uchar *buf, int inx);
 extern int maria_rkey(MARIA_HA *file, uchar *buf, int inx,
@@ -354,7 +355,7 @@ void _ma_update_auto_increment_key(HA_CHECK *param, MARIA_HA *info,
 typedef struct st_sort_key_blocks MA_SORT_KEY_BLOCKS;
 typedef struct st_sort_ftbuf MA_SORT_FT_BUF;
 
-extern PAGECACHE maria_pagecache_var, *maria_pagecache;
+extern PAGECACHES maria_pagecaches;
 int maria_assign_to_pagecache(MARIA_HA *info, ulonglong key_map,
 			      PAGECACHE *key_cache);
 void maria_change_pagecache(PAGECACHE *old_key_cache,
@@ -1786,7 +1787,7 @@ static inline check_result_t ma_check_index_cond(MARIA_HA *info, uint keynr,
   return ma_check_index_cond_real(info, keynr, record);
 }
 
-
+extern void ma_update_pagecache_stats();
 extern my_bool ma_yield_and_check_if_killed(MARIA_HA *info, int inx);
 extern my_bool ma_killed_standalone(MARIA_HA *);
 
diff --git a/storage/maria/test_ma_backup.c b/storage/maria/test_ma_backup.c
index 5cb2b074887..f71a3953d0a 100644
--- a/storage/maria/test_ma_backup.c
+++ b/storage/maria/test_ma_backup.c
@@ -45,8 +45,9 @@ int main(int argc __attribute__((unused)), char *argv[])
 
   /* Maria requires that we always have a page cache */
   if (maria_init() ||
-      (init_pagecache(maria_pagecache, maria_block_size * 2000, 0, 0,
-                      maria_block_size, 0, MY_WME) == 0) ||
+      multi_init_pagecache(&maria_pagecaches, 1,
+                           maria_block_size * 2000, 0, 0,
+                           maria_block_size, 0, MY_WME) ||
       ma_control_file_open_or_create() ||
       (init_pagecache(maria_log_pagecache,
                       TRANSLOG_PAGECACHE_SIZE, 0, 0,
diff --git a/storage/maria/unittest/ma_test_all-t b/storage/maria/unittest/ma_test_all-t
index 8858649fb5d..93c36fdadcc 100755
--- a/storage/maria/unittest/ma_test_all-t
+++ b/storage/maria/unittest/ma_test_all-t
@@ -289,7 +289,7 @@ sub run_check_tests
                       ["--key_multiple -a -B --key_length=480","-sm"],
                       ["--key_multiple -P -S","-sm"] );
   my @ma_test2_opt= ( ["-L -K -W -P","-sm"],
-                      ["-L -K -W -P -A","-sm"],
+                      ["-L -K -W -P -A -p4","-sm"],
                       ["-L -K -W -P -b32768", "-sm"],
                       ["-L -K -W -P -M -T -c -b32768 -t4 -m300", "-sm"],
                       ["-L -K -P -R3 -m50 -b1000000", "-sm"],
@@ -310,6 +310,8 @@ sub run_check_tests
     for ($i= 0; defined($ma_test2_opt[$i]); $i++) { $nr_tests+=2; }
     for ($i= 0; defined($ma_rt_test_opt[$i]); $i++) { $nr_tests+=2; }
     return $nr_tests;
+
+
   }
 
   for ($i= 0; defined($ma_test1_opt[$i]); $i++)
@@ -526,7 +528,7 @@ sub run_tests_on_clrs
           "$maria_exe_path/aria_chk$suffix -h$tmpdir -s -e $tmpdir/test2",
           "rm $tmpdir/test2.MA?",
           $NEW_TEST,
-          "$maria_exe_path/ma_test2$suffix -h$tmpdir -s -L -K -W -P -M -T -c -b -t2 -A1",
+          "$maria_exe_path/ma_test2$suffix -h$tmpdir -s -L -K -W -P -M -T -c -b -t2 -A1 -p4",
           "$maria_exe_path/aria_read_log$suffix -a -s -h$tmpdir ",
           "$maria_exe_path/aria_chk$suffix -h$tmpdir -s -e $tmpdir/test2",
           "rm $tmpdir/test2.MA?",
@@ -595,7 +597,7 @@ sub ok
     $len= length($tmp);
   }
   $output= `$com 2>&1`;
-  if ($verbose)
+  if ($verbose && $len <= 62)
   {
     print " " x (62 - $len);
   }