From 97fce90bd1da07806cda904b04d6f3a0c9d6c4c6 Mon Sep 17 00:00:00 2001
From: "jianwei.zhao" <jianwei.zhao@alibaba-inc.com>
Date: Sun, 19 May 2019 17:14:38 +0800
Subject: [PATCH]   [Feature] InnoDB Data File Purge Asynchronously

    Summary:
    --------
    InnoDB big data file unlink will cause serious stable problem on POSIX file system,
    So InnoDB launch a background thread to truncate data file little by little.

    The data file will firstly renamed into a temporary file when single tablespace is deleted,
    then File Purge thread will truncate file asynchronously and slowly.

    Add new DDL log (PURGE_FILE_LOG) to promise the DDL atomic.

    Several variables control the behaviors:
      INNODB_DATA_FILE_PURGE: Whether enable the async purge strategy
      INNODB_DATA_FILE_PURGE_IMMEDIATE: Unlink data file rather than truncate
      INNODB_DATA_FILE_PURGE_ALL_AT_SHUTDOWN: Cleanup all when normal shutdown
      INNODB_DATA_FILE_PURGE_DIR: Temporary file directory
      INNODB_DATA_FILE_PURGE_INTERVAL: Purge time interval (by milliseconds)
      INNODB_DATA_FILE_PURGE_MAX_SIZE: Purge max size every time (by MB)
      INNODB_PRINT_DATA_FILE_PURGE_PROCESS: Print the process of file purge worker
---
 .../r/feature_file_purge_issue56496.result    |  56 +++
 ...feature_file_purge_issue56496_basic.result |  89 +++++
 ...ture_file_purge_issue56496_shutdown.result |  44 +++
 .../rds/t/feature_file_purge_issue56496.test  |  95 +++++
 .../feature_file_purge_issue56496_basic.test  | 134 +++++++
 ...eature_file_purge_issue56496_shutdown.test |  97 +++++
 share/errmsg-utf8.txt                         |  10 +
 storage/innobase/CMakeLists.txt               |   4 +-
 storage/innobase/dict/dict0dict.cc            |   2 +-
 storage/innobase/fil/fil0fil.cc               |  42 ++-
 storage/innobase/fil/fil0purge.cc             | 350 ++++++++++++++++++
 storage/innobase/handler/ha_innodb.cc         |  56 ++-
 storage/innobase/include/fil0fil.h            |   4 +-
 storage/innobase/include/fil0purge.h          | 186 ++++++++++
 storage/innobase/include/log0ddl.h            |  44 ++-
 storage/innobase/include/srv0file.h           |  79 ++++
 storage/innobase/include/srv0srv.h            |   3 +
 storage/innobase/include/sync0types.h         |   1 +
 storage/innobase/log/log0ddl.cc               |  94 ++++-
 storage/innobase/row/row0mysql.cc             |   2 +-
 storage/innobase/srv/srv0file.cc              | 158 ++++++++
 storage/innobase/srv/srv0srv.cc               |   9 +
 storage/innobase/srv/srv0start.cc             |  32 +-
 storage/innobase/sync/sync0debug.cc           |   5 +
 24 files changed, 1565 insertions(+), 31 deletions(-)
 create mode 100644 mysql-test/suite/rds/r/feature_file_purge_issue56496.result
 create mode 100644 mysql-test/suite/rds/r/feature_file_purge_issue56496_basic.result
 create mode 100644 mysql-test/suite/rds/r/feature_file_purge_issue56496_shutdown.result
 create mode 100644 mysql-test/suite/rds/t/feature_file_purge_issue56496.test
 create mode 100644 mysql-test/suite/rds/t/feature_file_purge_issue56496_basic.test
 create mode 100644 mysql-test/suite/rds/t/feature_file_purge_issue56496_shutdown.test
 create mode 100644 storage/innobase/fil/fil0purge.cc
 create mode 100644 storage/innobase/include/fil0purge.h
 create mode 100644 storage/innobase/include/srv0file.h
 create mode 100644 storage/innobase/srv/srv0file.cc

diff --git a/mysql-test/suite/rds/r/feature_file_purge_issue56496.result b/mysql-test/suite/rds/r/feature_file_purge_issue56496.result
new file mode 100644
index 00000000000..771dedc61a7
--- /dev/null
+++ b/mysql-test/suite/rds/r/feature_file_purge_issue56496.result
@@ -0,0 +1,56 @@
+#
+# File purge
+#
+SET @old_log_error_verbosity = @@global.log_error_verbosity;
+SET @@global.log_error_verbosity=3;
+SET @old_innodb_data_file_purge = @@global.innodb_data_file_purge;
+SET @@global.innodb_data_file_purge = ON;
+SET @old_innodb_print_data_file_purge_process = @@global.innodb_print_data_file_purge_process;
+SET @@global.innodb_print_data_file_purge_process = ON;
+SHOW GLOBAL VARIABLES LIKE '%data_file_purge%';
+Variable_name	Value
+innodb_data_file_purge	ON
+innodb_data_file_purge_all_at_shutdown	OFF
+innodb_data_file_purge_dir	
+innodb_data_file_purge_immediate	OFF
+innodb_data_file_purge_interval	100
+innodb_data_file_purge_max_size	512
+innodb_print_data_file_purge_process	ON
+FLUSH LOGS;
+#
+# normal table
+#
+create database db_1;
+use db_1;
+create table t(id int, col1 int)engine = innodb;
+drop table t;
+drop database db_1;
+#
+# drop database
+#
+create database db_1;
+use db_1;
+create table t1(id int, col1 int)engine = innodb;
+create table t2(id int, col1 int)engine = innodb;
+drop database db_1;
+#
+# tablespace
+#
+create tablespace ts_1 add datafile 'ts_1.ibd' FILE_BLOCK_SIZE=1k;
+drop tablespace ts_1;
+#
+# data directory
+#
+create database db_1;
+use db_1;
+CREATE TABLE t_3 (a int, b text) KEY_BLOCK_SIZE=1 DATA DIRECTORY='MYSQL_TMP_DIR';
+drop table t_3;
+drop database db_1;
+File purge record:
+SET @@global.log_error_verbosity=@old_log_error_verbosity;
+SET @@global.innodb_data_file_purge=@old_innodb_data_file_purge;
+SET @@global.innodb_print_data_file_purge_process = @old_innodb_print_data_file_purge_process;
+
+###
+### done
+###
diff --git a/mysql-test/suite/rds/r/feature_file_purge_issue56496_basic.result b/mysql-test/suite/rds/r/feature_file_purge_issue56496_basic.result
new file mode 100644
index 00000000000..7cdfdf9ec84
--- /dev/null
+++ b/mysql-test/suite/rds/r/feature_file_purge_issue56496_basic.result
@@ -0,0 +1,89 @@
+SET global innodb_print_data_file_purge_process = ON;
+SET global innodb_data_file_purge = ON;
+SHOW GLOBAL VARIABLES LIKE '%data_file_purge%';
+Variable_name	Value
+innodb_data_file_purge	ON
+innodb_data_file_purge_all_at_shutdown	OFF
+innodb_data_file_purge_dir	
+innodb_data_file_purge_immediate	OFF
+innodb_data_file_purge_interval	100
+innodb_data_file_purge_max_size	512
+innodb_print_data_file_purge_process	ON
+FLUSH LOGS;
+SET SESSION debug= '+d, skip_dd_table_access_check';
+SELECT count(*) AS `Expected as 0` FROM mysql.innodb_ddl_log;
+Expected as 0
+0
+# Test normal case.
+CREATE TABLE  t1(a INT, b INT, c INT, key(a), key(b));
+TRUNCATE TABLE t1;
+DROP TABLE t1;
+show create table t1;
+ERROR 42S02: Table 'test.t1' doesn't exist
+# Test drop table crash/recovery rollback(before purge file)
+set global innodb_ddl_log_crash_reset_debug = 1;
+set session debug = '+d, ddl_log_crash_before_purge_file_log_counter_1';
+CREATE TABLE t1 (a INT, b INT, c INT, key(a), key(b));
+DROP TABLE t1;;
+show create table t1;
+ERROR 42S02: Table 'test.t1' doesn't exist
+SET SESSION debug= '+d,skip_dd_table_access_check';
+SELECT count(*) AS `Expected as 0` FROM mysql.innodb_ddl_log;
+Expected as 0
+0
+# Test drop table crash/recovery rollback(after purge file)
+set global innodb_ddl_log_crash_reset_debug = 1;
+set session debug = '+d, ddl_log_crash_after_purge_file_log_counter_1';
+CREATE TABLE t1 (a INT, b INT, c INT, key(a), key(b));
+DROP TABLE t1;;
+show create table t1;
+ERROR 42S02: Table 'test.t1' doesn't exist
+SET SESSION debug= '+d,skip_dd_table_access_check';
+SELECT count(*) AS `Expected as 0` FROM mysql.innodb_ddl_log;
+Expected as 0
+0
+# Test truncate table crash/recovery rollback(before purge file)
+set global innodb_ddl_log_crash_reset_debug = 1;
+set session debug = '+d, ddl_log_crash_before_purge_file_log_counter_1';
+CREATE TABLE t1 (a INT, b INT, c INT, key(a), key(b));
+TRUNCATE TABLE t1;;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL,
+  `c` int(11) DEFAULT NULL,
+  KEY `a` (`a`),
+  KEY `b` (`b`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci
+drop table t1;
+SET SESSION debug= '+d,skip_dd_table_access_check';
+SELECT count(*) AS `Expected as 0` FROM mysql.innodb_ddl_log;
+Expected as 0
+0
+# Test drop table crash/recovery rollback(after purge file)
+set global innodb_ddl_log_crash_reset_debug = 1;
+set session debug = '+d, ddl_log_crash_after_purge_file_log_counter_1';
+CREATE TABLE t1 (a INT, b INT, c INT, key(a), key(b));
+TRUNCATE TABLE t1;;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL,
+  `c` int(11) DEFAULT NULL,
+  KEY `a` (`a`),
+  KEY `b` (`b`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci
+drop table t1;
+SET SESSION debug= '+d,skip_dd_table_access_check';
+SELECT count(*) AS `Expected as 0` FROM mysql.innodb_ddl_log;
+Expected as 0
+0
+SET GLOBAL innodb_ddl_log_crash_reset_debug = 0;
+SET GLOBAL innodb_data_file_purge= 0;
+SET GLOBAL innodb_print_data_file_purge_process = 0;
+
+###
+### done
+###
diff --git a/mysql-test/suite/rds/r/feature_file_purge_issue56496_shutdown.result b/mysql-test/suite/rds/r/feature_file_purge_issue56496_shutdown.result
new file mode 100644
index 00000000000..20261f0ad55
--- /dev/null
+++ b/mysql-test/suite/rds/r/feature_file_purge_issue56496_shutdown.result
@@ -0,0 +1,44 @@
+call mtr.add_suppression("Cannot open temp data file for read-write");
+SET global innodb_print_data_file_purge_process = ON;
+SET global innodb_data_file_purge = ON;
+SET global innodb_data_file_purge_all_at_shutdown = ON;
+create database db_1;
+use db_1;
+SHOW GLOBAL VARIABLES LIKE '%data_file_purge%';
+Variable_name	Value
+innodb_data_file_purge	ON
+innodb_data_file_purge_all_at_shutdown	ON
+innodb_data_file_purge_dir	
+innodb_data_file_purge_immediate	OFF
+innodb_data_file_purge_interval	100
+innodb_data_file_purge_max_size	512
+innodb_print_data_file_purge_process	ON
+FLUSH LOGS;
+SET SESSION debug= '+d, skip_dd_table_access_check';
+SELECT count(*) AS `Expected as 0` FROM mysql.innodb_ddl_log;
+Expected as 0
+0
+# Test normal case.
+CREATE TABLE  t1(a INT, b INT, c INT, key(a), key(b)) engine = innodb;
+TRUNCATE TABLE t1;
+DROP TABLE t1;
+show create table t1;
+ERROR 42S02: Table 'db_1.t1' doesn't exist
+# Test rename error
+SET DEBUG='+d,Data_file_purge_rename_fail';
+CREATE TABLE  t1(a INT, b INT, c INT, key(a), key(b)) engine = innodb;
+DROP TABLE t1;
+show create table t1;
+ERROR 42S02: Table 'db_1.t1' doesn't exist
+SET DEBUG='-d,Data_file_purge_rename_fail';
+CREATE TABLE  t1(a INT, b INT, c INT, key(a), key(b)) engine = innodb;
+DROP TABLE t1;
+File purge cleanup completely
+SET GLOBAL innodb_data_file_purge= 0;
+SET GLOBAL innodb_data_file_purge_all_at_shutdown= 0;
+SET GLOBAL innodb_print_data_file_purge_process = 0;
+drop database db_1;
+
+###
+### done
+###
diff --git a/mysql-test/suite/rds/t/feature_file_purge_issue56496.test b/mysql-test/suite/rds/t/feature_file_purge_issue56496.test
new file mode 100644
index 00000000000..98313a8e26f
--- /dev/null
+++ b/mysql-test/suite/rds/t/feature_file_purge_issue56496.test
@@ -0,0 +1,95 @@
+--echo #
+--echo # File purge
+--echo #
+
+SET @old_log_error_verbosity = @@global.log_error_verbosity;
+SET @@global.log_error_verbosity=3;
+
+SET @old_innodb_data_file_purge = @@global.innodb_data_file_purge;
+SET @@global.innodb_data_file_purge = ON;
+
+SET @old_innodb_print_data_file_purge_process = @@global.innodb_print_data_file_purge_process;
+SET @@global.innodb_print_data_file_purge_process = ON;
+
+SHOW GLOBAL VARIABLES LIKE '%data_file_purge%';
+
+let $log_error_= `SELECT @@GLOBAL.log_error`;
+if($log_error_ == "stderr")
+{
+  let $log_error_ = $MYSQLTEST_VARDIR/log/mysqld.1.err;
+}
+FLUSH LOGS;
+
+
+--echo #
+--echo # normal table
+--echo #
+
+create database db_1;
+use db_1;
+create table t(id int, col1 int)engine = innodb;
+drop table t;
+drop database db_1;
+
+
+
+--echo #
+--echo # drop database
+--echo #
+create database db_1;
+use db_1;
+create table t1(id int, col1 int)engine = innodb;
+create table t2(id int, col1 int)engine = innodb;
+drop database db_1;
+
+--echo #
+--echo # tablespace
+--echo #
+create tablespace ts_1 add datafile 'ts_1.ibd' FILE_BLOCK_SIZE=1k;
+drop tablespace ts_1;
+
+--echo #
+--echo # data directory
+--echo #
+create database db_1;
+use db_1;
+
+--replace_result $MYSQL_TMP_DIR MYSQL_TMP_DIR
+eval CREATE TABLE t_3 (a int, b text) KEY_BLOCK_SIZE=1 DATA DIRECTORY='$MYSQL_TMP_DIR';
+drop table t_3;
+drop database db_1;
+
+--rmdir $MYSQL_TMP_DIR/db_1
+
+let GREP_FILE=$log_error_;
+
+--sleep 10
+
+perl;
+   use strict;
+   use File::stat;
+   my $file= $ENV{'GREP_FILE'} or die("grep file not set");
+   my $pattern="File purge";
+   my $cnt = 0;
+
+   open(FILE, "$file") or die("Unable to open $file: $!");
+   while (<FILE>) {
+     my $line = $_;
+     if ($line =~ /$pattern/) {
+       $cnt++;
+     }
+   }
+   if ($cnt > 10) {
+     print "File purge record:\n";
+   }
+   close(FILE);
+EOF
+
+SET @@global.log_error_verbosity=@old_log_error_verbosity;
+SET @@global.innodb_data_file_purge=@old_innodb_data_file_purge;
+SET @@global.innodb_print_data_file_purge_process = @old_innodb_print_data_file_purge_process;
+
+--echo
+--echo ###
+--echo ### done
+--echo ###
diff --git a/mysql-test/suite/rds/t/feature_file_purge_issue56496_basic.test b/mysql-test/suite/rds/t/feature_file_purge_issue56496_basic.test
new file mode 100644
index 00000000000..b30835ae334
--- /dev/null
+++ b/mysql-test/suite/rds/t/feature_file_purge_issue56496_basic.test
@@ -0,0 +1,134 @@
+--source include/have_debug.inc
+--source include/not_valgrind.inc
+--source include/not_crashrep.inc
+--source include/have_log_bin.inc
+
+LET $old_innodb_data_file_purge = `SELECT @@innodb_data_file_purge`;
+LET $old_innodb_print_data_file_purge_process = `SELECT @@innodb_print_data_file_purge_process`;
+LET $old_innodb_ddl_log_crash_reset_debug = `SELECT @@innodb_ddl_log_crash_reset_debug`;
+
+SET global innodb_print_data_file_purge_process = ON;
+SET global innodb_data_file_purge = ON;
+
+SHOW GLOBAL VARIABLES LIKE '%data_file_purge%';
+
+let $log_error_= `SELECT @@GLOBAL.log_error`;
+if($log_error_ == "stderr")
+{
+  let $log_error_ = $MYSQLTEST_VARDIR/log/mysqld.1.err;
+}
+FLUSH LOGS;
+
+
+SET SESSION debug= '+d, skip_dd_table_access_check';
+SELECT count(*) AS `Expected as 0` FROM mysql.innodb_ddl_log;
+
+LET $innodb_ddl_log_crash_reset_debug_orig = `SELECT @@innodb_ddl_log_crash_reset_debug`;
+
+--echo # Test normal case.
+CREATE TABLE  t1(a INT, b INT, c INT, key(a), key(b));
+TRUNCATE TABLE t1;
+DROP TABLE t1;
+
+--error ER_NO_SUCH_TABLE
+show create table t1;
+
+
+--echo # Test drop table crash/recovery rollback(before purge file)
+set global innodb_ddl_log_crash_reset_debug = 1;
+eval set session debug = '+d, ddl_log_crash_before_purge_file_log_counter_1';
+
+--exec echo "wait" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+
+--eval CREATE TABLE t1 (a INT, b INT, c INT, key(a), key(b))
+
+--error 0,CR_SERVER_LOST,ER_INTERNAL_ERROR
+--eval DROP TABLE t1;
+
+--exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+--enable_reconnect
+let $WAIT_COUNT=6000;
+--source include/wait_time_until_connected_again.inc
+
+-- error ER_NO_SUCH_TABLE
+show create table t1;
+
+SET SESSION debug= '+d,skip_dd_table_access_check';
+SELECT count(*) AS `Expected as 0` FROM mysql.innodb_ddl_log;
+
+
+--echo # Test drop table crash/recovery rollback(after purge file)
+set global innodb_ddl_log_crash_reset_debug = 1;
+eval set session debug = '+d, ddl_log_crash_after_purge_file_log_counter_1';
+
+--exec echo "wait" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+
+--eval CREATE TABLE t1 (a INT, b INT, c INT, key(a), key(b))
+
+--error 0,CR_SERVER_LOST,ER_INTERNAL_ERROR
+--eval DROP TABLE t1;
+
+--exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+--enable_reconnect
+let $WAIT_COUNT=6000;
+--source include/wait_time_until_connected_again.inc
+
+-- error ER_NO_SUCH_TABLE
+show create table t1;
+
+SET SESSION debug= '+d,skip_dd_table_access_check';
+SELECT count(*) AS `Expected as 0` FROM mysql.innodb_ddl_log;
+
+
+--echo # Test truncate table crash/recovery rollback(before purge file)
+set global innodb_ddl_log_crash_reset_debug = 1;
+eval set session debug = '+d, ddl_log_crash_before_purge_file_log_counter_1';
+
+--exec echo "wait" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+
+--eval CREATE TABLE t1 (a INT, b INT, c INT, key(a), key(b))
+
+--error 0,CR_SERVER_LOST,ER_INTERNAL_ERROR
+--eval TRUNCATE TABLE t1;
+
+--exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+--enable_reconnect
+let $WAIT_COUNT=6000;
+--source include/wait_time_until_connected_again.inc
+
+show create table t1;
+drop table t1;
+
+SET SESSION debug= '+d,skip_dd_table_access_check';
+SELECT count(*) AS `Expected as 0` FROM mysql.innodb_ddl_log;
+
+
+--echo # Test drop table crash/recovery rollback(after purge file)
+set global innodb_ddl_log_crash_reset_debug = 1;
+eval set session debug = '+d, ddl_log_crash_after_purge_file_log_counter_1';
+
+--exec echo "wait" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+
+--eval CREATE TABLE t1 (a INT, b INT, c INT, key(a), key(b))
+
+--eval TRUNCATE TABLE t1;
+
+--exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+--enable_reconnect
+let $WAIT_COUNT=6000;
+--source include/wait_time_until_connected_again.inc
+
+show create table t1;
+drop table t1;
+
+SET SESSION debug= '+d,skip_dd_table_access_check';
+SELECT count(*) AS `Expected as 0` FROM mysql.innodb_ddl_log;
+
+eval SET GLOBAL innodb_ddl_log_crash_reset_debug = $old_innodb_ddl_log_crash_reset_debug;
+eval SET GLOBAL innodb_data_file_purge= $old_innodb_data_file_purge;
+eval SET GLOBAL innodb_print_data_file_purge_process = $old_innodb_print_data_file_purge_process;
+
+--echo
+--echo ###
+--echo ### done
+--echo ###
diff --git a/mysql-test/suite/rds/t/feature_file_purge_issue56496_shutdown.test b/mysql-test/suite/rds/t/feature_file_purge_issue56496_shutdown.test
new file mode 100644
index 00000000000..cd1aaeb41e7
--- /dev/null
+++ b/mysql-test/suite/rds/t/feature_file_purge_issue56496_shutdown.test
@@ -0,0 +1,97 @@
+--source include/have_debug.inc
+--source include/not_valgrind.inc
+--source include/not_crashrep.inc
+--source include/have_log_bin.inc
+
+call mtr.add_suppression("Cannot open temp data file for read-write");
+
+LET MYSQLD_DATADIR =`SELECT @@datadir`;
+
+LET $old_innodb_data_file_purge = `SELECT @@innodb_data_file_purge`;
+LET $old_innodb_data_file_purge_all_at_shutdown = `SELECT @@innodb_data_file_purge_all_at_shutdown`;
+LET $old_innodb_print_data_file_purge_process = `SELECT @@innodb_print_data_file_purge_process`;
+
+SET global innodb_print_data_file_purge_process = ON;
+SET global innodb_data_file_purge = ON;
+SET global innodb_data_file_purge_all_at_shutdown = ON;
+
+create database db_1;
+use db_1;
+
+SHOW GLOBAL VARIABLES LIKE '%data_file_purge%';
+
+let $log_error_= `SELECT @@GLOBAL.log_error`;
+if($log_error_ == "stderr")
+{
+  let $log_error_ = $MYSQLTEST_VARDIR/log/mysqld.1.err;
+}
+FLUSH LOGS;
+
+
+SET SESSION debug= '+d, skip_dd_table_access_check';
+SELECT count(*) AS `Expected as 0` FROM mysql.innodb_ddl_log;
+
+--echo # Test normal case.
+CREATE TABLE  t1(a INT, b INT, c INT, key(a), key(b)) engine = innodb;
+TRUNCATE TABLE t1;
+DROP TABLE t1;
+
+--error ER_NO_SUCH_TABLE
+show create table t1;
+
+
+--echo # Test rename error
+SET DEBUG='+d,Data_file_purge_rename_fail';
+CREATE TABLE  t1(a INT, b INT, c INT, key(a), key(b)) engine = innodb;
+DROP TABLE t1;
+
+--error ER_NO_SUCH_TABLE
+show create table t1;
+
+--remove_file $MYSQLD_DATADIR/db_1/t1.ibd
+
+SET DEBUG='-d,Data_file_purge_rename_fail';
+CREATE TABLE  t1(a INT, b INT, c INT, key(a), key(b)) engine = innodb;
+DROP TABLE t1;
+
+--exec echo "wait" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+--exec $MYSQLADMIN -u root -S $MASTER_MYSOCK -P $MASTER_MYPORT shutdown
+--source include/wait_until_disconnected.inc
+--enable_reconnect
+--exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+--source include/wait_until_connected_again.inc
+
+let GREP_FILE=$log_error_;
+
+--sleep 10
+
+perl;
+   use strict;
+   use File::stat;
+   my $file= $ENV{'GREP_FILE'} or die("grep file not set");
+   my $pattern="Data File Purge : complete cleanup";
+   my $cnt = 0;
+
+   open(FILE, "$file") or die("Unable to open $file: $!");
+   while (<FILE>) {
+     my $line = $_;
+     if ($line =~ /$pattern/) {
+       $cnt++;
+     }
+   }
+   if ($cnt > 0) {
+    print "File purge cleanup completely\n";
+   }
+   close(FILE);
+EOF
+
+eval SET GLOBAL innodb_data_file_purge= $old_innodb_data_file_purge;
+eval SET GLOBAL innodb_data_file_purge_all_at_shutdown= $old_innodb_data_file_purge_all_at_shutdown;
+eval SET GLOBAL innodb_print_data_file_purge_process = $old_innodb_print_data_file_purge_process;
+
+drop database db_1;
+
+--echo
+--echo ###
+--echo ### done
+--echo ###
diff --git a/share/errmsg-utf8.txt b/share/errmsg-utf8.txt
index d8501e49099..d6d6989c8f0 100644
--- a/share/errmsg-utf8.txt
+++ b/share/errmsg-utf8.txt
@@ -19229,6 +19229,16 @@ ER_JSON_VALUE_OUT_OF_RANGE_FOR_FUNC_INDEX 22003
 ER_LDAP_EMPTY_USERDN_PASSWORD
   eng "Empty user dn or password is not allowed, not attempting LDAP bind."
 
+ER_IB_MSG_FILE_PURGE
+  eng "%s"
+
+ER_IB_MSG_FP_CLEANUP
+  eng "Data File Purge : cleanup tmp data files, still have %lu"
+
+ER_IB_MSG_FP_COMPLETE
+  eng "Data File Purge : complete cleanup"
+
+
 #
 # End of 8.0 error messages intended to be logged to the server error log.
 #
diff --git a/storage/innobase/CMakeLists.txt b/storage/innobase/CMakeLists.txt
index ed4083f43a4..45d30c1c8e7 100644
--- a/storage/innobase/CMakeLists.txt
+++ b/storage/innobase/CMakeLists.txt
@@ -207,7 +207,9 @@ SET(INNOBASE_SOURCES
 	ut/ut0rnd.cc
 	ut/ut0ut.cc
 	ut/ut0vec.cc
-	ut/ut0wqueue.cc)
+	ut/ut0wqueue.cc
+	srv/srv0file.cc
+	fil/fil0purge.cc)
 
 IF(WITH_INNODB)
   # Legacy option
diff --git a/storage/innobase/dict/dict0dict.cc b/storage/innobase/dict/dict0dict.cc
index 7d9e9aa7b89..74ddae13d58 100644
--- a/storage/innobase/dict/dict0dict.cc
+++ b/storage/innobase/dict/dict0dict.cc
@@ -1537,7 +1537,7 @@ dberr_t dict_table_rename_in_cache(
       return (DB_OUT_OF_MEMORY);
     }
 
-    err = fil_delete_tablespace(table->space, BUF_REMOVE_ALL_NO_WRITE);
+    err = fil_delete_tablespace(table->space, BUF_REMOVE_ALL_NO_WRITE, NULL);
 
     ut_a(err == DB_SUCCESS || err == DB_TABLESPACE_NOT_FOUND ||
          err == DB_IO_ERROR);
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index 5c760ec2139..c3d972f2b78 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -862,8 +862,10 @@ class Fil_shard {
   @param[in]	space_id	Tablespace ID
   @param[in]	buf_remove	Specify the action to take on the pages
                                   for this table in the buffer pool.
+  @param[in]  new_filepath  Rename the original datafile to new_filepath.
   @return DB_SUCCESS, DB_TABLESPCE_NOT_FOUND or DB_IO_ERROR */
-  dberr_t space_delete(space_id_t space_id, buf_remove_t buf_remove)
+  dberr_t space_delete(space_id_t space_id, buf_remove_t buf_remove,
+                       const char *new_filepath)
       MY_ATTRIBUTE((warn_unused_result));
 
   /** Truncate the tablespace to needed size.
@@ -3988,8 +3990,10 @@ datafile, fil_space_t & fil_node_t entries from the file_system_t cache.
 @param[in]	space_id	Tablespace ID
 @param[in]	buf_remove	Specify the action to take on the pages
                                 for this table in the buffer pool.
+@param[in]  new_filepath  Rename the original datafile to new_filepath.
 @return DB_SUCCESS, DB_TABLESPCE_NOT_FOUND or DB_IO_ERROR */
-dberr_t Fil_shard::space_delete(space_id_t space_id, buf_remove_t buf_remove) {
+dberr_t Fil_shard::space_delete(space_id_t space_id, buf_remove_t buf_remove,
+                                const char *new_filepath) {
   char *path = nullptr;
   fil_space_t *space = nullptr;
 
@@ -4098,14 +4102,26 @@ dberr_t Fil_shard::space_delete(space_id_t space_id, buf_remove_t buf_remove) {
     space_free_low(space);
     ut_a(space == nullptr);
 
-    if (!os_file_delete(innodb_data_file_key, path) &&
-        !os_file_delete_if_exists(innodb_data_file_key, path, nullptr)) {
-      /* Note: This is because we have removed the
-      tablespace instance from the cache. */
+    if (new_filepath) {
+      os_file_type_t type;
+      bool exists;
+      os_file_status(path, &exists, &type);
 
-      err = DB_IO_ERROR;
-    }
+      DBUG_EXECUTE_IF("Data_file_purge_rename_fail", exists = false;);
+
+      if (!exists ||
+          (exists && !os_file_rename(innodb_data_file_key, path, new_filepath)))
+        /* Not found or rename error should report error. */
+        err = DB_IO_ERROR;
+    } else {
+      if (!os_file_delete(innodb_data_file_key, path) &&
+          !os_file_delete_if_exists(innodb_data_file_key, path, nullptr)) {
+        /* Note: This is because we have removed the
+        tablespace instance from the cache. */
 
+        err = DB_IO_ERROR;
+      }
+    }
   } else {
     mutex_release();
 
@@ -4125,11 +4141,13 @@ datafile, fil_space_t & fil_node_t entries from the file_system_t cache.
 @param[in]	space_id	Tablespace ID
 @param[in]	buf_remove	Specify the action to take on the pages
                                 for this table in the buffer pool.
+@param[in]  new_filepath  Rename the original datafile to new_filepath.
 @return DB_SUCCESS, DB_TABLESPCE_NOT_FOUND or DB_IO_ERROR */
-dberr_t fil_delete_tablespace(space_id_t space_id, buf_remove_t buf_remove) {
+dberr_t fil_delete_tablespace(space_id_t space_id, buf_remove_t buf_remove,
+                              const char *new_filepath) {
   auto shard = fil_system->shard_by_id(space_id);
 
-  return (shard->space_delete(space_id, buf_remove));
+  return (shard->space_delete(space_id, buf_remove, new_filepath));
 }
 
 /** Prepare for truncating a single-table tablespace.
@@ -4226,7 +4244,7 @@ bool fil_replace_tablespace(space_id_t old_space_id, space_id_t new_space_id,
   bool is_encrypted = FSP_FLAGS_GET_ENCRYPTION(space->flags);
 
   /* Delete the old file and space object. */
-  dberr_t err = fil_delete_tablespace(old_space_id, BUF_REMOVE_ALL_NO_WRITE);
+  dberr_t err = fil_delete_tablespace(old_space_id, BUF_REMOVE_ALL_NO_WRITE, nullptr);
   if (err != DB_SUCCESS) {
     return (false);
   }
@@ -4357,7 +4375,7 @@ memory cache. Discarding is like deleting a tablespace, but
 dberr_t fil_discard_tablespace(space_id_t space_id) {
   dberr_t err;
 
-  err = fil_delete_tablespace(space_id, BUF_REMOVE_ALL_NO_WRITE);
+  err = fil_delete_tablespace(space_id, BUF_REMOVE_ALL_NO_WRITE, nullptr);
 
   switch (err) {
     case DB_SUCCESS:
diff --git a/storage/innobase/fil/fil0purge.cc b/storage/innobase/fil/fil0purge.cc
new file mode 100644
index 00000000000..4d332531b54
--- /dev/null
+++ b/storage/innobase/fil/fil0purge.cc
@@ -0,0 +1,350 @@
+/*****************************************************************************
+
+Copyright (c) 1995, 2019, Alibaba and/or its affiliates. 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, version 2.0,
+as published by the Free Software Foundation.
+
+This program is also distributed with certain software (including
+but not limited to OpenSSL) that is licensed under separate terms,
+as designated in a particular file or component or in included license
+documentation.  The authors of MySQL hereby grant you an additional
+permission to link the program and your derivative works with the
+separately licensed software that they have included with MySQL.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License, version 2.0, for more details.
+
+You should have received a copy of the GNU General Public License
+along with this program; if not, write to the Free Software
+Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+*****************************************************************************/
+
+/** @file fil/fil0purge.cc
+ Implementation of data file purge operation.
+
+ Created  1/11/2019 Jianwei.zhao
+ *******************************************************/
+
+#include "fil0purge.h"
+#include "fil0fil.h"
+#include "os0file.h"
+#include "row0mysql.h"
+#include "srv0file.h"
+#include "ut0mutex.h"
+
+/* Global file purge system */
+File_purge *file_purge_sys = nullptr;
+
+/** Constructor */
+File_purge::File_purge(ulint thread_id, time_t start_time)
+    : m_thread_id(thread_id),
+      m_start_time(start_time),
+      m_id(0),
+      m_dir(nullptr) {
+  mutex_create(LATCH_ID_FILE_PURGE_LIST, &m_mutex);
+  UT_LIST_INIT(m_list, &file_purge_node_t::LIST);
+}
+
+File_purge::~File_purge() {
+  /**
+    Clear the file nodes, those will be purged again when DDL log recovery after
+    reboot
+ */
+  for (file_purge_node_t *node = UT_LIST_GET_FIRST(m_list); node != nullptr;
+       node = UT_LIST_GET_FIRST(m_list)) {
+    if (node->m_file_path) ut_free(node->m_file_path);
+    UT_LIST_REMOVE(m_list, node);
+    ut_free(node);
+  }
+
+  mutex_free(&m_mutex);
+}
+
+/* Get next unique id number */
+ulint File_purge::next_id() {
+  ulint number;
+  number = m_id++;
+  return number;
+}
+/**
+  Add file into purge list
+
+  @param[in]    id        Log DDL record id
+  @param[int]   path      The temporary file name
+
+  @retval       false     Success
+  @retval       true      Failure
+*/
+bool File_purge::add_file(ulint id, const char *path) {
+  bool success = false;
+
+  pfs_os_file_t handle = os_file_create_simple_no_error_handling(
+      innodb_data_file_key, path, OS_FILE_OPEN, OS_FILE_READ_WRITE, FALSE,
+      &success);
+
+  if (!success) {
+    ib::error(ER_IB_MSG_392) << "Cannot open temp data file for read-write: '"
+                             << path << "'"
+                             << " when add file into purge list";
+
+    /** Temp file maybe has been purged */
+    ut_free(const_cast<char *>(path));
+    log_ddl->remove_by_id(id);
+    return true;
+  }
+
+  os_file_close(handle);
+
+  file_purge_node_t *node = static_cast<file_purge_node_t *>(
+      ut_malloc_nokey(sizeof(file_purge_node_t)));
+
+  node->m_log_ddl_id = id;
+  node->m_file_path = const_cast<char *>(path);
+
+  mutex_enter(&m_mutex);
+  UT_LIST_ADD_LAST(m_list, node);
+  mutex_exit(&m_mutex);
+
+  /* Wake up background thread */
+  srv_wakeup_file_purge_thread();
+
+  if (srv_print_data_file_purge_process)
+    ib::info(ER_IB_MSG_FILE_PURGE) << "File purge add file : " << id << ";"
+                                   << path;
+  return false;
+}
+
+/**
+  Purge the first file node by purge max size.
+
+  @param[in]  size        Purge max size (bytes)
+  @param[in]  force       Whether unlink file directly
+
+  @retval     -1          Error
+  @retval     0           Success & no file purge
+  @retval     >0          How many purged operation
+*/
+int File_purge::purge_file(ulint size, bool force) {
+  bool success = false;
+  uint truncated = 0;
+  file_purge_node_t *node = nullptr;
+  pfs_os_file_t handle;
+
+  mutex_enter(&m_mutex);
+  node = UT_LIST_GET_FIRST(m_list);
+  mutex_exit(&m_mutex);
+
+  if (node) {
+    handle = os_file_create_simple_no_error_handling(
+        innodb_data_file_key, node->m_file_path, OS_FILE_OPEN,
+        OS_FILE_READ_WRITE, FALSE, &success);
+    if (!success) {
+      ulint err = os_file_get_last_error(true);
+      ib::error(ER_IB_MSG_392) << "Cannot open temp data file for read-write: '"
+                               << node->m_file_path << "'"
+                               << " when purge file from list";
+      /* Maybe delete by DBA, so remove file directly */
+      if (err == OS_FILE_NOT_FOUND)
+        remove_file(node);
+
+      return -1;
+    }
+
+    if (srv_print_data_file_purge_process)
+      ib::info(ER_IB_MSG_FILE_PURGE) << "File purge purge file : "
+                                     << node->m_file_path;
+
+    os_offset_t file_size = os_file_get_size(handle);
+    if (!force && file_size > size) {
+      /* Truncate predefined size once */
+      os_file_truncate(node->m_file_path, handle, size);
+      os_file_close(handle);
+      truncated++;
+    } else {
+      /* Direct delete the file when file_size < predefined size */
+      os_file_close(handle);
+      os_file_delete(innodb_data_file_key, node->m_file_path);
+      remove_file(node);
+      truncated++;
+    }
+  } else {
+    return 0;
+  }
+
+  return truncated;
+}
+
+/**
+  The data file list length
+
+  @retval       >=0
+*/
+ulint File_purge::length() {
+  ulint len;
+  mutex_enter(&m_mutex);
+  len = UT_LIST_GET_LEN(m_list);
+  mutex_exit(&m_mutex);
+
+  return len;
+}
+/**
+  Purge all the data file cached in list.
+
+  @param[in]    size      Purge max size (bytes)
+  @param[in]    force     Purge little by little
+                          or unlink directly.
+*/
+void File_purge::purge_all(ulint size, bool force) {
+  while (length() > 0) {
+    purge_file(size, force);
+  }
+}
+/**
+  Remove the file node from list
+
+  @param[in]    node      File purge node pointer
+*/
+void File_purge::remove_file(file_purge_node_t *node) {
+  ulint log_id = node->m_log_ddl_id;
+
+  mutex_enter(&m_mutex);
+  UT_LIST_REMOVE(m_list, node);
+  mutex_exit(&m_mutex);
+
+#ifdef UNIV_DEBUG
+  bool exist;
+  os_file_type_t type;
+  os_file_status(node->m_file_path, &exist, &type);
+  ut_ad(exist == false);
+#endif
+
+  if (srv_print_data_file_purge_process)
+    ib::info(ER_IB_MSG_FILE_PURGE) << "File purge remove file : "
+                                   << node->m_file_path;
+  ut_free(node->m_file_path);
+  ut_free(node);
+
+  log_ddl->remove_by_id(log_id);
+}
+/**
+  Generate a unique temporary file name.
+
+  @param[in]    filepath      Original file name
+
+  @retval       file name     Generated file name
+*/
+char *File_purge::generate_file(const char *filepath) {
+  std::string new_path;
+
+  new_path.assign(get_dir());
+
+  std::string temp_filename;
+  temp_filename.assign(PREFIX);
+  temp_filename.append(std::to_string((ulong)m_start_time));
+  temp_filename.append("_");
+  temp_filename.append(std::to_string(next_id()));
+
+  char *new_file = Fil_path::make(new_path, temp_filename, NO_EXT, false);
+
+  if (srv_print_data_file_purge_process)
+    ib::info(ER_IB_MSG_FILE_PURGE) << "File purge generate file : " << filepath
+                                   << ";" << new_file;
+  return new_file;
+}
+
+/**
+  Drop a single-table tablespace and rename the data file as temporary file.
+  This deletes the fil_space_t if found and rename the file on disk.
+
+  @param[in]      space_id      Tablespace id
+  @param[in]      filepath      File path of tablespace to delete
+
+  @retval         error code */
+dberr_t row_purge_single_table_tablespace(space_id_t space_id,
+                                          const char *filepath) {
+  dberr_t err = DB_SUCCESS;
+  uint64_t log_id;
+
+  char *new_filepath = file_purge_sys->generate_file(filepath);
+
+  log_ddl->write_purge_file_log(&log_id, file_purge_sys->get_thread_id(),
+                                new_filepath);
+
+  if (srv_print_data_file_purge_process)
+    ib::info(ER_IB_MSG_FILE_PURGE) << "File purge write log : " << log_id << ";"
+                                   << new_filepath;
+
+  if (!fil_space_exists_in_mem(space_id, nullptr, true, false, NULL, 0)) {
+    if (fil_purge_file(filepath, new_filepath)) {
+      ib::info(ER_IB_MSG_989) << "Purge data file " << filepath;
+    }
+  } else {
+    err = fil_delete_tablespace(space_id, BUF_REMOVE_FLUSH_NO_WRITE,
+                                new_filepath);
+  }
+
+  file_purge_sys->add_file(log_id, new_filepath);
+
+  return err;
+}
+
+/**
+  Rename the ibd data file and delete the releted files
+
+  @param[in]      old_filepath  The original data file
+  @param[in]      new_filepath  The new data file
+
+  @retval         TRUE         Success
+  @retval         FALSE        Failure
+*/
+bool fil_purge_file(const char *old_filepath, const char *new_filepath) {
+  bool success = true;
+  bool exist;
+  os_file_type_t type;
+
+  /* If old file has been renamed or dropped, just skip it */
+  os_file_status(old_filepath, &exist, &type);
+
+  if (exist)
+    success = os_file_rename(innodb_data_file_key, old_filepath, new_filepath);
+
+  char *cfg_filepath = Fil_path::make_cfg(old_filepath);
+
+  if (cfg_filepath != nullptr) {
+    os_file_delete_if_exists(innodb_data_file_key, cfg_filepath, nullptr);
+
+    ut_free(cfg_filepath);
+  }
+
+  char *cfp_filepath = Fil_path::make_cfp(old_filepath);
+
+  if (cfp_filepath != nullptr) {
+    os_file_delete_if_exists(innodb_data_file_key, cfp_filepath, nullptr);
+
+    ut_free(cfp_filepath);
+  }
+
+  return (success);
+}
+
+/**
+  Drop or purge single table tablespace
+
+  @param[in]    space_id      tablespace id
+  @param[in]    filepath      tablespace data file path
+
+
+  @retval     DB_SUCCESS or error
+*/
+dberr_t row_drop_or_purge_single_table_tablespace(space_id_t space_id,
+                                                  const char *filepath) {
+  if (srv_data_file_purge)
+    return row_purge_single_table_tablespace(space_id, filepath);
+  else
+    return row_drop_tablespace(space_id, filepath);
+}
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 710e6b454b1..263c352743f 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -187,6 +187,8 @@ this program; if not, write to the Free Software Foundation, Inc.,
 #include <unistd.h>
 #endif /* HAVE_UNISTD_H */
 
+#include "srv0file.h"
+
 #ifndef UNIV_HOTBACKUP
 /** Stop printing warnings, if the count exceeds this threshold. */
 static const size_t MOVED_FILES_PRINT_THRESHOLD = 32;
@@ -664,7 +666,8 @@ static PSI_mutex_info all_innodb_mutexes[] = {
     PSI_MUTEX_KEY(zip_pad_mutex, 0, 0, PSI_DOCUMENT_ME),
     PSI_MUTEX_KEY(master_key_id_mutex, 0, 0, PSI_DOCUMENT_ME),
     PSI_MUTEX_KEY(sync_array_mutex, 0, 0, PSI_DOCUMENT_ME),
-    PSI_MUTEX_KEY(row_drop_list_mutex, 0, 0, PSI_DOCUMENT_ME)};
+    PSI_MUTEX_KEY(row_drop_list_mutex, 0, 0, PSI_DOCUMENT_ME),
+    PSI_MUTEX_KEY(file_purge_list_mutex, 0, 0, PSI_DOCUMENT_ME)};
 #endif /* UNIV_PFS_MUTEX */
 
 #ifdef UNIV_PFS_RWLOCK
@@ -729,7 +732,8 @@ static PSI_thread_info all_innodb_threads[] = {
     PSI_KEY(fts_optimize_thread, 0, 0, PSI_DOCUMENT_ME),
     PSI_KEY(fts_parallel_merge_thread, 0, 0, PSI_DOCUMENT_ME),
     PSI_KEY(fts_parallel_tokenization_thread, 0, 0, PSI_DOCUMENT_ME),
-    PSI_KEY(srv_ts_alter_encrypt_thread, 0, 0, PSI_DOCUMENT_ME)};
+    PSI_KEY(srv_ts_alter_encrypt_thread, 0, 0, PSI_DOCUMENT_ME),
+    PSI_KEY(srv_file_purge_thread, 0, 0, PSI_DOCUMENT_ME)};
 #endif /* UNIV_PFS_THREAD */
 
 #ifdef UNIV_PFS_IO
@@ -4520,6 +4524,7 @@ static int innodb_init_params() {
                       + 1   /* fts_optimize_thread */
                       + 1   /* recv_writer_thread */
                       + 1   /* trx_rollback_or_clean_all_recovered */
+                      + 1   /* srv_file_purge_thread */
                       + 128 /* added as margin, for use of
                             InnoDB Memcached etc. */
                       + max_connections + srv_n_read_io_threads +
@@ -21671,6 +21676,46 @@ static MYSQL_SYSVAR_STR(directories, innobase_directories,
                         "'innodb-data-home-dir;innodb-undo-directory;datadir'",
                         NULL, NULL, NULL);
 
+/* Data file purge system variables  */
+static MYSQL_SYSVAR_BOOL(data_file_purge, srv_data_file_purge,
+                         PLUGIN_VAR_OPCMDARG,
+                         "Data file purge little by little (off by default)",
+                         NULL, NULL, FALSE);
+
+static MYSQL_SYSVAR_BOOL(data_file_purge_immediate,
+                         srv_data_file_purge_immediate, PLUGIN_VAR_OPCMDARG,
+                         "Data file unlink by purge thread directly "
+                         "even though async purge",
+                         NULL, NULL, FALSE);
+
+static MYSQL_SYSVAR_BOOL(data_file_purge_all_at_shutdown,
+                         srv_data_file_purge_all_at_shutdown,
+                         PLUGIN_VAR_OPCMDARG,
+                         "Data file unlink directly even though async purge when shutdown",
+                         NULL, NULL, FALSE);
+
+static MYSQL_SYSVAR_ULONG(data_file_purge_interval,
+                          srv_data_file_purge_interval, PLUGIN_VAR_OPCMDARG,
+                          "Time interval each of data file purge operation (milliseconds)",
+                          NULL, NULL, 100, 0, 10000, 0);
+
+static MYSQL_SYSVAR_ULONG(data_file_purge_max_size, srv_data_file_purge_max_size,
+                          PLUGIN_VAR_OPCMDARG,
+                          "Max size each of data file purge operation (MB)",
+                          NULL, NULL, 512, 16, 1024 * 1024 * 1024, 0);
+
+static MYSQL_SYSVAR_STR(data_file_purge_dir, srv_data_file_purge_dir,
+                        PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY |
+                        PLUGIN_VAR_NOPERSIST,
+                        "The directory that data file will be removed into",
+                        NULL, NULL, NULL);
+
+static MYSQL_SYSVAR_BOOL(print_data_file_purge_process,
+                         srv_print_data_file_purge_process, PLUGIN_VAR_OPCMDARG,
+                        "Print all data file purge process to MySQL error log (off by default)",
+                        NULL, NULL, FALSE);
+/* End of data file purge system variables  */
+
 static SYS_VAR *innobase_system_variables[] = {
     MYSQL_SYSVAR(api_trx_level),
     MYSQL_SYSVAR(api_bk_commit_interval),
@@ -21876,6 +21921,13 @@ static SYS_VAR *innobase_system_variables[] = {
     MYSQL_SYSVAR(ddl_log_crash_reset_debug),
 #endif /* UNIV_DEBUG */
     MYSQL_SYSVAR(parallel_read_threads),
+    MYSQL_SYSVAR(data_file_purge),
+    MYSQL_SYSVAR(data_file_purge_immediate),
+    MYSQL_SYSVAR(data_file_purge_all_at_shutdown),
+    MYSQL_SYSVAR(data_file_purge_interval),
+    MYSQL_SYSVAR(data_file_purge_max_size),
+    MYSQL_SYSVAR(data_file_purge_dir),
+    MYSQL_SYSVAR(print_data_file_purge_process),
     NULL};
 
 mysql_declare_plugin(innobase){
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index 53b54c98f71..0cd5e59bfd8 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -1260,8 +1260,10 @@ datafile, fil_space_t & fil_node_t entries from the file_system_t cache.
 @param[in]	space_id	Tablespace ID
 @param[in]	buf_remove	Specify the action to take on the pages
 for this table in the buffer pool.
+@param[in]  new_filepath  Rename the original datafile to new_filepath.
 @return DB_SUCCESS, DB_TABLESPCE_NOT_FOUND or DB_IO_ERROR */
-dberr_t fil_delete_tablespace(space_id_t space_id, buf_remove_t buf_remove)
+dberr_t fil_delete_tablespace(space_id_t space_id, buf_remove_t buf_remove,
+                              const char *new_filepath)
     MY_ATTRIBUTE((warn_unused_result));
 
 /** Open a single-table tablespace and optionally check the space id is
diff --git a/storage/innobase/include/fil0purge.h b/storage/innobase/include/fil0purge.h
new file mode 100644
index 00000000000..0e1c0cbe060
--- /dev/null
+++ b/storage/innobase/include/fil0purge.h
@@ -0,0 +1,186 @@
+/*****************************************************************************
+
+Copyright (c) 1995, 2019, Alibaba and/or its affiliates. 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, version 2.0, as published by the
+Free Software Foundation.
+
+This program is also distributed with certain software (including but not
+limited to OpenSSL) that is licensed under separate terms, as designated in a
+particular file or component or in included license documentation. The authors
+of MySQL hereby grant you an additional permission to link the program and
+your derivative works with the separately licensed software that they have
+included with MySQL.
+
+This program is distributed in the hope that it will be useful, but WITHOUT
+ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+FOR A PARTICULAR PURPOSE. See the GNU General Public License, version 2.0,
+for more details.
+
+You should have received a copy of the GNU General Public License along with
+this program; if not, write to the Free Software Foundation, Inc.,
+51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+*****************************************************************************/
+
+/** @file include/fil0purge.h
+ The file purge interface
+
+ Created 1/11/2019 Jianwei.zhao
+ *******************************************************/
+#ifndef fil0purge_h
+#define fil0purge_h
+
+#include <atomic>
+
+#include "univ.i"
+#include "ut0mutex.h"
+
+/** File node structure for background purge thread. */
+struct file_purge_node_t {
+  using List_node = UT_LIST_NODE_T(file_purge_node_t);
+
+  /** File name */
+  char *m_file_path;
+
+  /** Unique id within log DDL table */
+  ulint m_log_ddl_id;
+
+  /** Link to file_purge_system->list */
+  List_node LIST;
+};
+
+/* File purge system */
+class File_purge {
+  using File_list = UT_LIST_BASE_NODE_T(file_purge_node_t);
+
+  /* */
+  static constexpr const char *PREFIX = "#FP_";
+
+ public:
+
+  /** Constructor */
+  explicit File_purge(ulint thread_id, time_t start_time);
+
+  virtual ~File_purge();
+
+  ulint get_thread_id() { return m_thread_id; }
+  /**
+    Add file into purge list
+
+    @param[in]    id        Log DDL record id
+    @param[int]   path      The temporary file name
+
+    @retval       false     Success
+    @retval       true      Failure
+  */
+  bool add_file(ulint id, const char *path);
+
+  /**
+    Purge the first file node by purge max size.
+
+    @param[in]    size      Purge max size (bytes)
+    @param[in]    force     Whether unlink file directly
+
+    @retval       -1        Error
+    @retval       0         Success & no file purge
+    @retval       >0        How many purged operation
+  */
+  int purge_file(ulint size, bool force);
+  /**
+    The data file list length
+
+    @retval       >=0
+  */
+  ulint length();
+  /**
+    Purge all the data file cached in list.
+
+    @param[in]    size      Purge max size (bytes)
+    @param[in]    force     Purge little by little
+                            or unlink directly.
+  */
+  void purge_all(ulint size, bool force);
+
+  /**
+    Remove the file node from list
+
+    @param[in]    node      File purge node pointer
+  */
+  void remove_file(file_purge_node_t *node);
+
+  /**
+    Generate a unique temporary file name.
+
+    @param[in]    filepath      Original file name
+
+    @retval       file name     Generated file name
+  */
+  char *generate_file(const char *filepath);
+
+  /* Set work directory */
+  void set_dir(const char *dir) { m_dir = dir; }
+
+  /* Get work directory */
+  const char *get_dir() { return m_dir; }
+
+  /* Get next unique id number */
+  ulint next_id();
+
+  /** Disable */
+  File_purge(File_purge &) = delete;
+  File_purge(const File_purge &) = delete;
+  File_purge &operator=(const File_purge &) = delete;
+
+ private:
+  /** Protect file node list */
+  ib_mutex_t m_mutex;
+  /** File node list */
+  File_list m_list;
+  /** File purge thread id */
+  ulint m_thread_id;
+  /** Server start time */
+  time_t m_start_time;
+  /* Unique id within instance */
+  std::atomic<ulint> m_id;
+  /** The directory will hold renamed file */
+  const char *m_dir;
+};
+
+extern File_purge *file_purge_sys;
+
+/**
+  Drop a single-table tablespace and rename the data file as temporary file.
+  This deletes the fil_space_t if found and rename the file on disk.
+
+  @param[in]      space_id      Tablespace id
+  @param[in]      filepath      File path of tablespace to delete
+
+  @retval         error code */
+extern dberr_t row_purge_single_table_tablespace(space_id_t space_id,
+                                                 const char *filepath);
+
+/**
+  Rename the ibd data file and delete the releted files
+
+  @param[in]      old_filepath  The original data file
+  @param[in]      new_filepath  The new data file
+
+  @retval         False         Success
+  @retval         True          Failure
+*/
+extern bool fil_purge_file(const char *old_filepath, const char *new_filepath);
+
+/**
+  Drop or purge single table tablespace
+
+  @param[in]    space_id      tablespace id
+  @param[in]    filepath      tablespace data file path
+
+
+  @retval     DB_SUCCESS or error
+*/
+extern dberr_t row_drop_or_purge_single_table_tablespace(space_id_t space_id,
+                                                         const char *filepath);
+#endif
diff --git a/storage/innobase/include/log0ddl.h b/storage/innobase/include/log0ddl.h
index 6a0a295b03e..e272ca53a84 100644
--- a/storage/innobase/include/log0ddl.h
+++ b/storage/innobase/include/log0ddl.h
@@ -66,8 +66,12 @@ enum class Log_Type : uint32_t {
   /** Alter Encrypt a tablespace */
   ALTER_ENCRYPT_TABLESPACE_LOG,
 
+  /** Purge file async */
+  PURGE_FILE_LOG = 64,
+
   /** Biggest log type */
-  BIGGEST_LOG = ALTER_ENCRYPT_TABLESPACE_LOG
+  BIGGEST_LOG = PURGE_FILE_LOG
+
 };
 
 /** DDL log record */
@@ -144,6 +148,14 @@ class DDL_Record {
   @return true if record is deletable. */
   bool get_deletable() const { return (m_deletable); }
 
+  /** If this record can be removed.
+  @return true if record is removed. */
+  bool get_removable() const { return (m_removable); }
+
+  /** Set removability of this record.
+  @param[in] removable removability. */
+  void set_removable(bool removable) { m_removable = removable; }
+
   /** Get the old file path/name present in the DDL log record.
   @return old file path/name. */
   const char *get_old_file_path() const { return (m_old_file_path); }
@@ -209,6 +221,10 @@ class DDL_Record {
 
   /** If this record can be deleted */
   bool m_deletable;
+
+  /** If this record can be removed. like PURGE_FILE_LOG will be deleted
+  by background thread, so front thread set its removable = false */
+  bool m_removable;
 };
 
 /** Forward declaration */
@@ -452,6 +468,18 @@ class Log_DDL {
   @return DB_SUCCESS or error */
   dberr_t write_remove_cache_log(trx_t *trx, dict_table_t *table);
 
+  /** Write a PURGE file log record
+  @param[out] id          record id
+  @param[in]  thread id   Purge thread
+  @param[in]  file path   New file path
+  @return DB_SUCCESS or error */
+  dberr_t write_purge_file_log(uint64_t *id, ulint thread_id,
+                               const char *file_path);
+
+  /** Remove the record by id
+  @param[in]  id          record id
+  @return DB_SUCCESS or error */
+  dberr_t remove_by_id(uint64_t id);
   /** Replay DDL log record
   @param[in,out]	record	DDL log record
   return DB_SUCCESS or error */
@@ -592,6 +620,20 @@ class Log_DDL {
   @param[in]	table_name	table name */
   void replay_remove_cache_log(table_id_t table_id, const char *table_name);
 
+  /** Insert a PURGE file log record
+  @param[in]  id            log id
+  @param[in]	thread_id	    thread id
+  @param[in]	new_file_path	new file path
+  @return DB_SUCCESS or error */
+  dberr_t insert_purge_file_log(uint64_t id, ulint thread_id,
+                                const char *new_file_path);
+
+  /** Replay PURGE file log
+  @param[in]  id            log id
+  @param[in]	thread_id	    thread id
+  @param[in]	new_file_path	new file path */
+  void replay_purge_file_log(uint64_t id, ulint thread_id,
+                             const char *file_path);
   /** Delete log record by id
   @param[in]	trx		transaction instance
   @param[in]	id		log id
diff --git a/storage/innobase/include/srv0file.h b/storage/innobase/include/srv0file.h
new file mode 100644
index 00000000000..a2be56f9694
--- /dev/null
+++ b/storage/innobase/include/srv0file.h
@@ -0,0 +1,79 @@
+/*****************************************************************************
+
+Copyright (c) 2000, 2019, Alibaba and/or its affiliates. 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, version 2.0, as published by the
+Free Software Foundation.
+
+This program is also distributed with certain software (including but not
+limited to OpenSSL) that is licensed under separate terms, as designated in a
+particular file or component or in included license documentation. The authors
+of MySQL hereby grant you an additional permission to link the program and
+your derivative works with the separately licensed software that they have
+included with MySQL.
+
+This program is distributed in the hope that it will be useful, but WITHOUT
+ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+FOR A PARTICULAR PURPOSE. See the GNU General Public License, version 2.0,
+for more details.
+
+You should have received a copy of the GNU General Public License along with
+this program; if not, write to the Free Software Foundation, Inc.,
+51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+*****************************************************************************/
+
+/** @file include/srv0file.h
+ Interface of data file purge operation.
+
+ Created 1/11/2019 Jianwei Zhao
+ *******************************************************/
+#ifndef srv0file_h
+#define srv0file_h
+
+
+#ifdef UNIV_PFS_THREAD
+/* File purge thread PFS key */
+extern mysql_pfs_key_t srv_file_purge_thread_key;
+#endif
+
+#ifdef UNIV_PFS_MUTEX
+/* File purge list mutex PFS key */
+extern mysql_pfs_key_t file_purge_list_mutex_key;
+#endif
+
+/** Whether enable the data file purge asynchronously little by little */
+extern bool srv_data_file_purge;
+
+/** Whether unlink the file immediately by purge thread */
+extern bool srv_data_file_purge_immediate;
+
+/** Whether purge all when normal shutdown */
+extern bool srv_data_file_purge_all_at_shutdown;
+
+/** Time interval (milliseconds) every data file purge operation */
+extern ulong srv_data_file_purge_interval;
+
+/** Max size (MB) every data file purge operation */
+extern ulong srv_data_file_purge_max_size;
+
+/** The directory that purged data file will be removed into */
+extern char *srv_data_file_purge_dir;
+
+/** Whether to print data file purge process */
+extern bool srv_print_data_file_purge_process;
+
+/** Data file purge system initialize when InnoDB server boots */
+extern void srv_file_purge_init();
+
+/** Data file purge system destroy when InnoDB server shutdown */
+extern void srv_file_purge_destroy();
+
+/** Data file purge thread runtime */
+extern void srv_file_purge_thread(void);
+
+/** Wakeup the background thread when shutdown */
+extern void srv_wakeup_file_purge_thread(void);
+
+#endif
diff --git a/storage/innobase/include/srv0srv.h b/storage/innobase/include/srv0srv.h
index e2509817b0d..97b423dea3b 100644
--- a/storage/innobase/include/srv0srv.h
+++ b/storage/innobase/include/srv0srv.h
@@ -162,6 +162,9 @@ struct Srv_threads {
 
   /** true if tablespace alter encrypt thread is created */
   bool m_ts_alter_encrypt_thread_active;
+
+  /** true if file purge thread is created */
+  bool m_file_purge_thread_active;
 };
 
 struct Srv_cpu_usage {
diff --git a/storage/innobase/include/sync0types.h b/storage/innobase/include/sync0types.h
index 79857dd3fa1..0d796e753f2 100644
--- a/storage/innobase/include/sync0types.h
+++ b/storage/innobase/include/sync0types.h
@@ -451,6 +451,7 @@ enum latch_id_t {
   LATCH_ID_CLONE_SYS,
   LATCH_ID_CLONE_TASK,
   LATCH_ID_CLONE_SNAPSHOT,
+  LATCH_ID_FILE_PURGE_LIST,
   LATCH_ID_TEST_MUTEX,
   LATCH_ID_MAX = LATCH_ID_TEST_MUTEX
 };
diff --git a/storage/innobase/log/log0ddl.cc b/storage/innobase/log/log0ddl.cc
index 5bd6c06590a..768206cd5e4 100644
--- a/storage/innobase/log/log0ddl.cc
+++ b/storage/innobase/log/log0ddl.cc
@@ -46,6 +46,7 @@ this program; if not, write to the Free Software Foundation, Inc.,
 #include "dict0dd.h"
 #include "dict0mem.h"
 #include "dict0stats.h"
+#include "fil0purge.h"
 #include "ha_innodb.h"
 #include "log0ddl.h"
 #include "mysql/plugin.h"
@@ -76,6 +77,7 @@ bool innodb_ddl_log_crash_reset_debug;
 2. DELETE SPACE
 3. RENAME SPACE
 4. DROP
+5. PURGE FILE
 Other RENAME_TABLE and REMOVE CACHE doesn't touch the data files at all,
 so would be skipped */
 
@@ -121,6 +123,12 @@ static uint32_t crash_before_alter_encrypt_space_log_counter = 1;
 /** Crash injection counter used after writing ALTER ENCRYPT TABLESPACE log */
 static uint32_t crash_after_alter_encrypt_space_log_counter = 1;
 
+/** Crash injection counter used before writing PURGE_FILE log */
+static uint32_t crash_before_purge_file_log_counter = 1;
+
+/** Crash injection counter used after writing PURGE_FILE log */
+static uint32_t crash_after_purge_file_log_counter = 1;
+
 void ddl_log_crash_reset(THD *thd, SYS_VAR *var, void *var_ptr,
                          const void *save) {
   const bool reset = *static_cast<const bool *>(save);
@@ -140,6 +148,8 @@ void ddl_log_crash_reset(THD *thd, SYS_VAR *var, void *var_ptr,
     crash_before_drop_log_counter = 1;
     crash_after_drop_log_counter = 1;
     crash_after_replay_counter = 1;
+    crash_before_purge_file_log_counter = 1;
+    crash_after_purge_file_log_counter = 1;
   }
 }
 
@@ -155,7 +165,8 @@ DDL_Record::DDL_Record()
       m_old_file_path(nullptr),
       m_new_file_path(nullptr),
       m_heap(nullptr),
-      m_deletable(true) {}
+      m_deletable(true),
+      m_removable(true) {}
 
 DDL_Record::~DDL_Record() {
   if (m_heap != nullptr) {
@@ -791,7 +802,7 @@ dberr_t DDL_Log_Table::remove(const DDL_Records &records) {
   dberr_t ret = DB_SUCCESS;
 
   for (auto record : records) {
-    if (record->get_deletable()) {
+    if (record->get_deletable() && record->get_removable()) {
       dberr_t err = remove(record->get_id());
 
       ut_ad(err == DB_SUCCESS || err == DB_TOO_MANY_CONCURRENT_TRXS);
@@ -1120,6 +1131,72 @@ dberr_t Log_DDL::insert_rename_space_log(uint64_t id, ulint thread_id,
   return (error);
 }
 
+dberr_t Log_DDL::write_purge_file_log(uint64_t *id, ulint thread_id,
+                                      const char *file_path) {
+  *id = next_id();
+
+  dberr_t err;
+  err = insert_purge_file_log(*id, thread_id, file_path);
+  ut_ad(err == DB_SUCCESS);
+
+  return (err);
+}
+
+void Log_DDL::replay_purge_file_log(uint64_t id, ulint thread_id,
+                                    const char *file_path) {
+  char *path = mem_strdup(file_path);
+  file_purge_sys->add_file(id, path);
+}
+
+dberr_t Log_DDL::insert_purge_file_log(uint64_t id, ulint thread_id,
+                                       const char *file_path) {
+  dberr_t error;
+  trx_t *trx = trx_allocate_for_background();
+  trx_start_internal(trx);
+  trx->ddl_operation = true;
+
+  DBUG_INJECT_CRASH("ddl_log_crash_before_purge_file_log_counter",
+                    crash_before_purge_file_log_counter++);
+  DDL_Record record;
+  record.set_id(id);
+  record.set_thread_id(thread_id);
+  record.set_type(Log_Type::PURGE_FILE_LOG);
+  record.set_new_file_path(file_path);
+  {
+    DDL_Log_Table ddl_log(trx);
+    error = ddl_log.insert(record);
+    ut_ad(error == DB_SUCCESS);
+  }
+
+  trx_commit_for_mysql(trx);
+  trx_free_for_background(trx);
+
+  DBUG_INJECT_CRASH("ddl_log_crash_after_purge_file_log_counter",
+                    crash_after_purge_file_log_counter++);
+  if (srv_print_ddl_logs) {
+    ib::info(ER_IB_MSG_649) << "DDL log insert : " << record;
+  }
+  return (error);
+}
+
+dberr_t Log_DDL::remove_by_id(uint64_t id) {
+  dberr_t error;
+  trx_t *trx = trx_allocate_for_background();
+  trx_start_internal(trx);
+  trx->ddl_operation = true;
+
+  {
+    DDL_Log_Table ddl_log(trx);
+    error = ddl_log.remove(id);
+    ut_ad(error == DB_SUCCESS);
+  }
+
+  trx_commit_for_mysql(trx);
+  trx_free_for_background(trx);
+
+  return (error);
+}
+
 dberr_t Log_DDL::write_alter_encrypt_space_log(space_id_t space_id) {
   /* Missing current_thd, it happens during crash recovery */
   if (!current_thd) {
@@ -1401,6 +1478,7 @@ dberr_t Log_DDL::replay_all() {
 
   DDL_Log_Table ddl_log;
   DDL_Records records;
+  DDL_Records purge_records;
 
   dberr_t err = ddl_log.search_all(records);
   ut_ad(err == DB_SUCCESS);
@@ -1414,6 +1492,11 @@ dberr_t Log_DDL::replay_all() {
       ts_encrypt_ddl_records.push_back(record);
       record->set_deletable(false);
     }
+    /* PURGE_FILE_LOG will be deleted by background thread, so front thread
+    set its removable = false */
+    if (record->get_type() == Log_Type::PURGE_FILE_LOG) {
+      record->set_removable(false);
+    }
   }
 
   err = delete_by_ids(records);
@@ -1531,6 +1614,11 @@ dberr_t Log_DDL::replay(DDL_Record &record) {
       replay_alter_encrypt_space_log(record.get_space_id());
       break;
 
+    case Log_Type::PURGE_FILE_LOG:
+      replay_purge_file_log(record.get_id(), record.get_thread_id(),
+                            record.get_new_file_path());
+      break;
+
     default:
       ut_error;
   }
@@ -1615,7 +1703,7 @@ void Log_DDL::replay_delete_space_log(space_id_t space_id,
   DBUG_EXECUTE_IF("ddl_log_replay_delete_space_crash_before_drop",
                   DBUG_SUICIDE(););
 
-  row_drop_tablespace(space_id, file_path);
+  row_drop_or_purge_single_table_tablespace(space_id, file_path);
 
   /* If this is an undo space_id, allow the undo number for it
   to be reused. */
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index 7897b0450f6..cc8ffb44b49 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -3756,7 +3756,7 @@ dberr_t row_drop_tablespace(space_id_t space_id, const char *filepath) {
     }
 
   } else {
-    err = fil_delete_tablespace(space_id, BUF_REMOVE_FLUSH_NO_WRITE);
+    err = fil_delete_tablespace(space_id, BUF_REMOVE_FLUSH_NO_WRITE, NULL);
 
     if (err != DB_SUCCESS && err != DB_TABLESPACE_NOT_FOUND) {
       ib::error(ER_IB_MSG_991)
diff --git a/storage/innobase/srv/srv0file.cc b/storage/innobase/srv/srv0file.cc
new file mode 100644
index 00000000000..04ac88d2916
--- /dev/null
+++ b/storage/innobase/srv/srv0file.cc
@@ -0,0 +1,158 @@
+/*****************************************************************************
+
+Copyright (c) 2000, 2019, Alibaba and/or its affiliates. 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, version 2.0, as published by the
+Free Software Foundation.
+
+This program is also distributed with certain software (including but not
+limited to OpenSSL) that is licensed under separate terms, as designated in a
+particular file or component or in included license documentation. The authors
+of MySQL hereby grant you an additional permission to link the program and
+your derivative works with the separately licensed software that they have
+included with MySQL.
+
+This program is distributed in the hope that it will be useful, but WITHOUT
+ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+FOR A PARTICULAR PURPOSE. See the GNU General Public License, version 2.0,
+for more details.
+
+You should have received a copy of the GNU General Public License along with
+this program; if not, write to the Free Software Foundation, Inc.,
+51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+*****************************************************************************/
+
+/** @file srv/srv0file.cc
+ Service of data file operation.
+
+ Created 1/11/2019 Jianwei Zhao
+ *******************************************************/
+#include "fil0purge.h"
+#include "my_inttypes.h"
+#include "sql/mysqld.h"
+#include "sql/mysqld_thd_manager.h"
+#include "srv0start.h"
+
+#include "srv0file.h"
+
+#ifdef UNIV_PFS_THREAD
+/* File purge thread PFS key */
+mysql_pfs_key_t srv_file_purge_thread_key;
+#endif
+
+#ifdef UNIV_PFS_MUTEX
+/* File purge list mutex PFS key */
+mysql_pfs_key_t file_purge_list_mutex_key;
+#endif
+
+/** Whether enable the data file purge background little by little */
+bool srv_data_file_purge = false;
+
+/** Whether unlink the file immediately by purge thread */
+bool srv_data_file_purge_immediate = false;
+
+/** Whether purge all when normal shutdown */
+bool srv_data_file_purge_all_at_shutdown = false;
+
+/** Time interval (milliseconds) every data file purge operation */
+ulong srv_data_file_purge_interval = 100;
+
+/** Max size (MB) every data file purge operation */
+ulong srv_data_file_purge_max_size = 512;
+
+/** The directory that purged data file will be removed into */
+char *srv_data_file_purge_dir = nullptr;
+
+/** Whether to print data file purge process */
+bool srv_print_data_file_purge_process = false;
+
+/** Indicate whether file purge system initted */
+static bool file_purge_system_inited = false;
+
+/** Purge thread event condition */
+os_event_t file_purge_event;
+
+/** Data file purge system initialize when InnoDB server boots */
+void srv_file_purge_init() {
+  file_purge_sys = UT_NEW_NOKEY(
+      File_purge(Global_THD_manager::reserved_thread_id, server_start_time));
+
+  file_purge_event = os_event_create(0);
+
+  /** If not setting special directory, inherit MySQL datadir directly. */
+  if (srv_data_file_purge_dir) {
+    os_file_type_t type;
+    bool exists;
+    bool success = os_file_status(srv_data_file_purge_dir, &exists, &type);
+
+    if (success && exists) {
+      /* If directory has existed, set dir directly */
+      file_purge_sys->set_dir(srv_data_file_purge_dir);
+    } else if ((os_file_create_subdirs_if_needed(srv_data_file_purge_dir) ==
+                DB_SUCCESS) &&
+               (os_file_create_directory(srv_data_file_purge_dir, false))) {
+      /* If not exist, try to create dir */
+      file_purge_sys->set_dir(srv_data_file_purge_dir);
+    } else {
+      /* Defaultly use innodb data dir */
+      file_purge_sys->set_dir(MySQL_datadir_path);
+    }
+  } else { /* srv_data_file_purge_dir is null */
+    file_purge_sys->set_dir(MySQL_datadir_path);
+  }
+
+  file_purge_system_inited = true;
+}
+
+/** Data file purge system destroy when InnoDB server shutdown */
+void srv_file_purge_destroy() {
+  UT_DELETE(file_purge_sys);
+  os_event_destroy(file_purge_event);
+  file_purge_system_inited = false;
+}
+
+/* Data file purge thread runtime */
+void srv_file_purge_thread(void) {
+  int64_t sig_count;
+  ut_a(file_purge_sys);
+
+loop:
+  int truncated =
+      file_purge_sys->purge_file(srv_data_file_purge_max_size * 1024 * 1024,
+                                 srv_data_file_purge_immediate);
+
+  if (truncated <= 0) {
+    sig_count = os_event_reset(file_purge_event);
+    os_event_wait_time_low(file_purge_event, 5000000, sig_count);
+  } else if (truncated > 0) {
+    os_thread_sleep(srv_data_file_purge_interval * 1000);
+  }
+
+  if (srv_shutdown_state >= SRV_SHUTDOWN_CLEANUP) goto exit_func;
+
+  goto loop;
+
+exit_func:
+  /**
+    Purge all renamed tmp data file requirement when shutdown:
+      - innodb_fast_shutdown = 0 or 1;
+      - innodb_data_file_purge_all_at_shutdown is true;
+
+    It will unlink files regardless of file size.
+  */
+  if (srv_fast_shutdown < 2 && srv_data_file_purge_all_at_shutdown) {
+    ib::info(ER_IB_MSG_FP_CLEANUP, file_purge_sys->length());
+
+    file_purge_sys->purge_all(srv_data_file_purge_max_size * 1024 * 1024, true);
+
+    ib::info(ER_IB_MSG_FP_COMPLETE);
+  }
+  srv_threads.m_file_purge_thread_active = false;
+  return;
+}
+
+/** Wakeup the background thread when shutdown */
+void srv_wakeup_file_purge_thread() { os_event_set(file_purge_event); }
+
diff --git a/storage/innobase/srv/srv0srv.cc b/storage/innobase/srv/srv0srv.cc
index 3771239650d..5c5682348c1 100644
--- a/storage/innobase/srv/srv0srv.cc
+++ b/storage/innobase/srv/srv0srv.cc
@@ -89,6 +89,8 @@ this program; if not, write to the Free Software Foundation, Inc.,
 #endif /* !UNIV_HOTBACKUP */
 #include "ut0mem.h"
 
+#include "srv0file.h"
+
 #ifdef UNIV_HOTBACKUP
 #include "page0size.h"
 #else
@@ -1155,6 +1157,8 @@ void srv_free(void) {
   ut_free(srv_sys);
 
   srv_sys = 0;
+
+  srv_file_purge_destroy();
 }
 
 /** Initializes the synchronization primitives, memory system, and the thread
@@ -1168,6 +1172,7 @@ static void srv_general_init() {
   que_init();
   row_mysql_init();
   undo_spaces_init();
+  srv_file_purge_init();
 }
 
 /** Boots the InnoDB server. */
@@ -1781,6 +1786,8 @@ const char *srv_any_background_threads_are_active() {
     thread_active = "buf_dump_thread";
   } else if (srv_threads.m_buf_resize_thread_active) {
     thread_active = "buf_resize_thread";
+  } else if (srv_threads.m_file_purge_thread_active) {
+    thread_active = "file_purge_thread";
   }
 
   os_event_set(srv_error_event);
@@ -1789,6 +1796,8 @@ const char *srv_any_background_threads_are_active() {
   os_event_set(lock_sys->timeout_event);
   os_event_set(srv_buf_resize_event);
 
+  srv_wakeup_file_purge_thread();
+
   return (thread_active);
 }
 
diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
index 6d50bd57097..998db95fa23 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -118,6 +118,8 @@ this program; if not, write to the Free Software Foundation, Inc.,
 #include "ut0crc32.h"
 #include "ut0new.h"
 
+#include "srv0file.h"
+
 /** fil_space_t::flags for hard-coded tablespaces */
 extern uint32_t predefined_flags;
 
@@ -146,15 +148,17 @@ static bool srv_start_has_been_called = false;
 determine which threads need to be stopped if we need to abort during
 the initialisation step. */
 enum srv_start_state_t {
-  SRV_START_STATE_NONE = 0,     /*!< No thread started */
-  SRV_START_STATE_LOCK_SYS = 1, /*!< Started lock-timeout
-                                thread. */
-  SRV_START_STATE_IO = 2,       /*!< Started IO threads */
-  SRV_START_STATE_MONITOR = 4,  /*!< Started montior thread */
-  SRV_START_STATE_MASTER = 8,   /*!< Started master threadd. */
-  SRV_START_STATE_PURGE = 16,   /*!< Started purge thread(s) */
-  SRV_START_STATE_STAT = 32     /*!< Started bufdump + dict stat
-                                and FTS optimize thread. */
+  SRV_START_STATE_NONE = 0,       /*!< No thread started */
+  SRV_START_STATE_LOCK_SYS = 1,   /*!< Started lock-timeout
+                                  thread. */
+  SRV_START_STATE_IO = 2,         /*!< Started IO threads */
+  SRV_START_STATE_MONITOR = 4,    /*!< Started montior thread */
+  SRV_START_STATE_MASTER = 8,     /*!< Started master threadd. */
+  SRV_START_STATE_PURGE = 16,     /*!< Started purge thread(s) */
+  SRV_START_STATE_STAT = 32,      /*!< Started bufdump + dict stat
+                                  and FTS optimize thread. */
+  SRV_START_STATE_FILE_PURGE = 64,/*!< Started file purge thread. */
+
 };
 
 /** Track server thrd starting phases */
@@ -1686,6 +1690,11 @@ void srv_shutdown_all_bg_threads() {
       }
     }
 
+    /* Wakeup file purge background thread */
+    if (srv_start_state_is_set(SRV_START_STATE_FILE_PURGE)) {
+      srv_wakeup_file_purge_thread();
+    }
+
     if (srv_start_state_is_set(SRV_START_STATE_IO)) {
       /* e. Exit the i/o threads */
       if (!srv_read_only_mode) {
@@ -2725,6 +2734,11 @@ dberr_t srv_start(bool create_new_db, const std::string &scan_directories) {
     os_thread_create(srv_monitor_thread_key, srv_monitor_thread);
 
     srv_start_state_set(SRV_START_STATE_MONITOR);
+
+    /* Create file purge thread */
+    srv_threads.m_file_purge_thread_active = true;
+    os_thread_create(srv_file_purge_thread_key, srv_file_purge_thread);
+    srv_start_state_set(SRV_START_STATE_FILE_PURGE);
   }
 
   srv_sys_tablespaces_open = true;
diff --git a/storage/innobase/sync/sync0debug.cc b/storage/innobase/sync/sync0debug.cc
index 32051d497be..f7e2e7e967b 100644
--- a/storage/innobase/sync/sync0debug.cc
+++ b/storage/innobase/sync/sync0debug.cc
@@ -56,6 +56,8 @@ this program; if not, write to the Free Software Foundation, Inc.,
 
 #include "ut0new.h"
 
+#include "srv0file.h"
+
 #ifdef UNIV_DEBUG
 
 bool srv_sync_debug;
@@ -1480,6 +1482,9 @@ static void sync_latch_meta_init() UNIV_NOTHROW {
   LATCH_ADD_MUTEX(CLONE_SNAPSHOT, SYNC_NO_ORDER_CHECK,
                   clone_snapshot_mutex_key);
 
+  LATCH_ADD_MUTEX(FILE_PURGE_LIST, SYNC_NO_ORDER_CHECK,
+                  file_purge_list_mutex_key);
+
   latch_id_t id = LATCH_ID_NONE;
 
   /* The array should be ordered on latch ID.We need to