MDEV-17124: mariadb 10.1.34, views and prepared statements: ERROR 1615 (HY000): Prepared statement needs to be re-preparedbb-10.2-MDEV-17124

The problem is that if table definition cache (TDC) is full of real tables
which are in tables cache, view definition can not stay there so will be
removed by its own underlying tables.
In situation above old mechanism of detection matching definition in PS
and current version always require reprepare and so prevent executing
the PS.

One work around is to increase TDC, other - improve version check for
views/triggers (which is done here). Now in suspicious cases we check:
 - timestamp (ms) of the view to be sure that version really have changed;
 - time (ms) of creation of a trigger related to time (ms) of statement
   preparation.

15 files changed, 323 insertions, 66 deletions

diff --git a/mysql-test/r/ps_ddl.result b/mysql-test/r/ps_ddl.result
index e69c6e06193..005c78a0319 100644
--- a/mysql-test/r/ps_ddl.result
+++ b/mysql-test/r/ps_ddl.result
@@ -214,7 +214,7 @@ new trigger: 10
 drop trigger t1_bd;
 set @val=11;
 execute stmt using @val;
-call p_verify_reprepare_count(1);
+call p_verify_reprepare_count(0);
 SUCCESS
 
 select @message;
@@ -224,7 +224,7 @@ Test 6-e: removing a relevant trigger
 drop trigger t1_bi;
 set @val=12;
 execute stmt using @val;
-call p_verify_reprepare_count(1);
+call p_verify_reprepare_count(0);
 SUCCESS
 
 select @message;
@@ -384,7 +384,7 @@ a
 flush table t1;
 set @var=9;
 execute stmt using @var;
-call p_verify_reprepare_count(1);
+call p_verify_reprepare_count(0);
 SUCCESS
 
 select * from t2;
@@ -831,8 +831,8 @@ a	b	c
 20	40	100
 30	60	150
 call p_verify_reprepare_count(1);
-SUCCESS
-
+ERROR
+Expected: 1, actual: 0
 # Check that we properly handle ALTER VIEW statements.
 execute stmt;
 a	b	c
@@ -882,9 +882,9 @@ a	b	c
 10	50	60
 20	100	120
 30	150	180
-call p_verify_reprepare_count(1);
-SUCCESS
-
+call p_verify_reprepare_count(0);
+ERROR
+Expected: 0, actual: 1
 execute stmt;
 a	b	c
 10	50	60
@@ -1206,7 +1206,7 @@ drop trigger v2_bi;
 set @message=null;
 set @var=9;
 execute stmt using @var;
-call p_verify_reprepare_count(1);
+call p_verify_reprepare_count(0);
 SUCCESS
 
 select @message;
@@ -2578,7 +2578,7 @@ SELECT * FROM t1;
 a
 2048
 1025
-1024
+2048
 DROP TABLE t1;
 #
 # End of 10.1 tests
diff --git a/mysql-test/r/view.result b/mysql-test/r/view.result
index 4edbabf11cc..1340a48fb83 100644
--- a/mysql-test/r/view.result
+++ b/mysql-test/r/view.result
@@ -6862,5 +6862,33 @@ r
 drop view v1;
 drop table t1;
 #
+# MDEV-17124: mariadb 10.1.34, views and prepared statements:
+# ERROR 1615 (HY000): Prepared statement needs to be re-prepared
+#
+set @tdc= @@table_definition_cache, @tc= @@table_open_cache;
+set global table_definition_cache= 400, table_open_cache= 400;
+create table tt (a int, primary key(a)) engine=MyISAM;
+create view v as select * from tt;
+insert into tt values(1),(2),(3),(4);
+prepare stmt from 'select * from tt';
+#fill table definition cache
+execute stmt;
+a
+1
+2
+3
+4
+prepare stmt from 'select * from v';
+execute stmt;
+a
+1
+2
+3
+4
+drop database db;
+drop view v;
+drop table tt;
+set global table_definition_cache= @tdc, table_open_cache= @tc;
+#
 # End of 10.2 tests
 #
diff --git a/mysql-test/t/ps_ddl.test b/mysql-test/t/ps_ddl.test
index 90226d379bf..fc1bd8d3752 100644
--- a/mysql-test/t/ps_ddl.test
+++ b/mysql-test/t/ps_ddl.test
@@ -250,7 +250,8 @@ drop trigger t1_bd;
 
 set @val=11;
 execute stmt using @val;
-call p_verify_reprepare_count(1);
+# No trigger in opened table => nothing to check => no reprepare
+call p_verify_reprepare_count(0);
 select @message;
 
 --echo Test 6-e: removing a relevant trigger
@@ -259,7 +260,8 @@ drop trigger t1_bi;
 
 set @val=12;
 execute stmt using @val;
-call p_verify_reprepare_count(1);
+# No trigger in opened table => nothing to check => no reprepare
+call p_verify_reprepare_count(0);
 select @message;
 set @val=13;
 execute stmt using @val;
@@ -374,7 +376,8 @@ select * from t3;
 flush table t1;
 set @var=9;
 execute stmt using @var;
-call p_verify_reprepare_count(1);
+# flush tables now do not mean reprepare
+call p_verify_reprepare_count(0);
 select * from t2;
 select * from t3;
 drop view v1;
@@ -763,7 +766,8 @@ flush tables; # empty TDC
 create view t1 as select a, 5*a as b, 6*a as c from t2;
 lock tables t1 read, t2 read;
 execute stmt;
-call p_verify_reprepare_count(1);
+# flush tables now do not mean reprepare
+call p_verify_reprepare_count(0);
 execute stmt;
 call p_verify_reprepare_count(0);
 execute stmt;
@@ -967,7 +971,8 @@ drop trigger v2_bi;
 set @message=null;
 set @var=9;
 execute stmt using @var;
-call p_verify_reprepare_count(1);
+# No trigger in opened table => nothing to check => no reprepare
+call p_verify_reprepare_count(0);
 select @message;
 create trigger v2_bi after insert on v2 for each row set @message="v2_ai";
 set @var= 10;
diff --git a/mysql-test/t/view.test b/mysql-test/t/view.test
index 6265a514783..6914c80f635 100644
--- a/mysql-test/t/view.test
+++ b/mysql-test/t/view.test
@@ -6598,5 +6598,45 @@ drop view v1;
 drop table t1;
 
 --echo #
+--echo # MDEV-17124: mariadb 10.1.34, views and prepared statements:
+--echo # ERROR 1615 (HY000): Prepared statement needs to be re-prepared
+--echo #
+
+set @tdc= @@table_definition_cache, @tc= @@table_open_cache;
+set global table_definition_cache= 400, table_open_cache= 400;
+
+create table tt (a int, primary key(a)) engine=MyISAM;
+create view v as select * from tt;
+insert into tt values(1),(2),(3),(4);
+
+prepare stmt from 'select * from tt';
+--echo #fill table definition cache
+--disable_query_log
+--disable_result_log
+create database db;
+use db;
+--let $tables=401
+while ($tables)
+{
+  --eval create table t$tables (i int) engine=MyISAM
+  --eval select * from t$tables
+  --dec $tables
+}
+
+use test;
+
+--enable_query_log
+--enable_result_log
+execute stmt;
+prepare stmt from 'select * from v';
+execute stmt;
+
+# Cleanup
+drop database db;
+drop view v;
+drop table tt;
+set global table_definition_cache= @tdc, table_open_cache= @tc;
+
+--echo #
 --echo # End of 10.2 tests
 --echo #
diff --git a/sql/parse_file.cc b/sql/parse_file.cc
index 999f53bd681..92e7d82af1c 100644
--- a/sql/parse_file.cc
+++ b/sql/parse_file.cc
@@ -174,11 +174,11 @@ write_parameter(IO_CACHE *file, uchar* base, File_option *parameter)
   {
     /* string have to be allocated already */
     LEX_STRING *val_s= (LEX_STRING *)(base + parameter->offset);
-    time_t tm= my_time(0);
-
-    get_date(val_s->str, GETDATE_DATE_TIME|GETDATE_GMT|GETDATE_FIXEDLENGTH,
-	     tm);
-    val_s->length= PARSE_FILE_TIMESTAMPLENGTH;
+    ulonglong tm= my_hrtime().val;
+    // Paded to 19 characters for compatibility
+    val_s->length= snprintf(val_s->str, MICROSECOND_TIMESTAMP_BUFFER_SIZE,
+                            "%019lld", tm);
+    DBUG_ASSERT(val_s->length == MICROSECOND_TIMESTAMP_BUFFER_SIZE-1);
     if (my_b_write(file, (const uchar *)val_s->str,
                     PARSE_FILE_TIMESTAMPLENGTH))
       DBUG_RETURN(TRUE);
@@ -835,15 +835,15 @@ File_parser::parse(uchar* base, MEM_ROOT *mem_root,
 	  /* string have to be allocated already */
 	  LEX_STRING *val= (LEX_STRING *)(base + parameter->offset);
 	  /* yyyy-mm-dd HH:MM:SS = 19(PARSE_FILE_TIMESTAMPLENGTH) characters */
-	  if (ptr[PARSE_FILE_TIMESTAMPLENGTH] != '\n')
+	  if (ptr[MICROSECOND_TIMESTAMP_BUFFER_SIZE-1] != '\n')
 	  {
 	    my_error(ER_FPARSER_ERROR_IN_PARAMETER, MYF(0),
                      parameter->name.str, line);
 	    DBUG_RETURN(TRUE);
 	  }
-	  memcpy(val->str, ptr, PARSE_FILE_TIMESTAMPLENGTH);
-	  val->str[val->length= PARSE_FILE_TIMESTAMPLENGTH]= '\0';
-	  ptr+= (PARSE_FILE_TIMESTAMPLENGTH+1);
+	  memcpy(val->str, ptr, MICROSECOND_TIMESTAMP_BUFFER_SIZE-1);
+	  val->str[val->length= MICROSECOND_TIMESTAMP_BUFFER_SIZE-1]= '\0';
+	  ptr+= MICROSECOND_TIMESTAMP_BUFFER_SIZE;
 	  break;
 	}
 	case FILE_OPTIONS_STRLIST:
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 248dedf36e4..85bf6f05d02 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1839,6 +1839,10 @@ retry_share:
       goto err_lock;
     }
 
+    /* Open view */
+    if (mysql_make_view(thd, share, table_list, false))
+      goto err_lock;
+
     /*
       This table is a view. Validate its metadata version: in particular,
       that it was a view when the statement was prepared.
@@ -1846,10 +1850,6 @@ retry_share:
     if (check_and_update_table_version(thd, table_list, share))
       goto err_lock;
 
-    /* Open view */
-    if (mysql_make_view(thd, share, table_list, false))
-      goto err_lock;
-
 
     /* TODO: Don't free this */
     tdc_release_share(share);
@@ -2643,7 +2643,7 @@ static bool
 check_and_update_table_version(THD *thd,
                                TABLE_LIST *tables, TABLE_SHARE *table_share)
 {
-  if (! tables->is_table_ref_id_equal(table_share))
+  if (! tables->is_table_ref_id_equal(thd, table_share))
   {
     if (thd->m_reprepare_observer &&
         thd->m_reprepare_observer->report_error(thd))
@@ -2749,7 +2749,9 @@ bool tdc_open_view(THD *thd, TABLE_LIST *table_list, uint flags)
 
   DBUG_ASSERT(share->is_view);
 
-  if (flags & CHECK_METADATA_VERSION)
+  err= mysql_make_view(thd, share, table_list, (flags & OPEN_VIEW_NO_PARSE));
+
+  if (!err && (flags & CHECK_METADATA_VERSION))
   {
     /*
       Check TABLE_SHARE-version of view only if we have been instructed to do
@@ -2764,7 +2766,6 @@ bool tdc_open_view(THD *thd, TABLE_LIST *table_list, uint flags)
       goto ret;
   }
 
-  err= mysql_make_view(thd, share, table_list, (flags & OPEN_VIEW_NO_PARSE));
 ret:
   tdc_release_share(share);
 
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 479578679f1..8907463613f 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -3607,6 +3607,7 @@ Statement::Statement(LEX *lex_arg, MEM_ROOT *mem_root_arg,
   id(id_arg),
   mark_used_columns(MARK_COLUMNS_READ),
   lex(lex_arg),
+  ms_prepare_time(0),
   db(NULL),
   db_length(0)
 {
@@ -3626,6 +3627,7 @@ void Statement::set_statement(Statement *stmt)
   mark_used_columns=   stmt->mark_used_columns;
   lex=            stmt->lex;
   query_string=   stmt->query_string;
+  ms_prepare_time=    stmt->ms_prepare_time;
 }
 
 
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 3f0fba8fc10..b9bb9c978fb 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1079,6 +1079,7 @@ public:
 
   LEX_STRING name; /* name for named prepared statements */
   LEX *lex;                                     // parse tree descriptor
+  ulonglong ms_prepare_time; // time of preparation in microseconds
   /*
     Points to the query associated with this statement. It's const, but
     we need to declare it char * because all table handlers are written
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 64e4cd30561..a4a24cc1a80 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -4177,6 +4177,9 @@ bool Prepared_statement::prepare(const char *packet, uint packet_len)
   Query_arena *old_stmt_arena;
   DBUG_ENTER("Prepared_statement::prepare");
   DBUG_ASSERT(m_sql_mode == thd->variables.sql_mode);
+
+  // The same format as for triggers to compare
+  ms_prepare_time= my_hrtime().val;
   /*
     If this is an SQLCOM_PREPARE, we also increase Com_prepare_sql.
     However, it seems handy if com_stmt_prepare is increased always,
@@ -4435,8 +4438,9 @@ Prepared_statement::execute_loop(String *expanded_query,
                                  uchar *packet_end)
 {
   Reprepare_observer reprepare_observer;
-  bool error;
+  ulonglong first_prepared= ms_prepare_time;
   int reprepare_attempt= 0;
+  bool error;
   iterations= FALSE;
 
   /*
@@ -4519,6 +4523,22 @@ reexecute:
     DBUG_ASSERT(thd->get_stmt_da()->sql_errno() == ER_NEED_REPREPARE);
     thd->clear_error();
 
+    {
+      /*
+        Check if we too fast with reprepare:
+        we can be so fast that:
+          1) make change of a trigger,
+          2) prepare,
+          3) try to exacute and reprepare
+        in 1 microsecond, so we will wait till
+        next microsecond before last reprepare
+      */
+      while (first_prepared == my_hrtime().val)
+      {
+        pthread_yield();
+      }
+    }
+
     error= reprepare();
 
     if (! error)                                /* Success */
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 9483db9eff9..138fa4bc631 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -6724,13 +6724,15 @@ static bool store_trigger(THD *thd, Trigger *trigger,
   table->field[14]->store(STRING_WITH_LEN("OLD"), cs);
   table->field[15]->store(STRING_WITH_LEN("NEW"), cs);
 
-  if (trigger->create_time)
+  if (trigger->ms_create_time)
   {
+    /* timestamp is in microseconds */
     table->field[16]->set_notnull();
     thd->variables.time_zone->gmt_sec_to_TIME(&timestamp,
-                                              (my_time_t)(trigger->create_time/100));
-    /* timestamp is with 6 digits */
-    timestamp.second_part= (trigger->create_time % 100) * 10000;
+                                              (my_time_t)
+                                              (trigger->ms_create_time/
+                                               1000000));
+    timestamp.second_part= trigger->ms_create_time % 1000000;
     ((Field_temporal_with_date*) table->field[16])->store_time_dec(&timestamp,
                                                                    2);
   }
@@ -9810,12 +9812,14 @@ static bool show_create_trigger_impl(THD *thd, Trigger *trigger)
            trigger->db_cl_name.length,
            system_charset_info);
 
-  if (trigger->create_time)
+  if (trigger->ms_create_time)
   {
     MYSQL_TIME timestamp;
     thd->variables.time_zone->gmt_sec_to_TIME(&timestamp,
-                                              (my_time_t)(trigger->create_time/100));
-    timestamp.second_part= (trigger->create_time % 100) * 10000;
+                                              (my_time_t)
+                                              (trigger->ms_create_time/
+                                               1000000));
+    timestamp.second_part= trigger->ms_create_time % 1000000;
     p->store(&timestamp, 2);
   }
   else
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index 52817ef65ae..7823bc06534 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -37,6 +37,8 @@
 #ifdef WITH_WSREP
 #include "debug_sync.h"
 #endif /* WITH_WSREP */
+#include <my_time.h>
+#include <mysql_time.h>
 
 LEX_STRING *make_lex_string(LEX_STRING *lex_str, const char* str, uint length,
                             MEM_ROOT *mem_root)
@@ -211,7 +213,7 @@ static File_option triggers_file_parameters[]=
   },
   {
     { C_STRING_WITH_LEN("created") },
-    my_offsetof(class Table_triggers_list, create_times),
+    my_offsetof(class Table_triggers_list, ms_create_times),
     FILE_OPTIONS_ULLLIST
   },
   { { 0, 0 }, 0, FILE_OPTIONS_STRING }
@@ -888,6 +890,10 @@ bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables,
   if (!(trigger= new (&table->mem_root) Trigger(this, 0)))
     goto err_without_cleanup;
 
+  /* Time with in microseconds */
+  trigger->ms_create_time= make_ms_time(thd->query_start(),
+                                        thd->query_start_sec_part());
+
   /* Create trigger_name.TRN file to ensure trigger name is unique */
   if (sql_create_definition_file(NULL, &trigname_file, &trigname_file_type,
                                  (uchar*)&trigname, trigname_file_parameters))
@@ -896,8 +902,6 @@ bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables,
   /* Populate the trigger object */
 
   trigger->sql_mode= thd->variables.sql_mode;
-  /* Time with 2 decimals, like in MySQL 5.7 */
-  trigger->create_time= ((ulonglong) thd->query_start())*100 + thd->query_start_sec_part()/10000;
   build_trig_stmt_query(thd, tables, stmt_query, &trigger_definition,
                         &trigger->definer, trg_definer_holder);
 
@@ -965,7 +969,7 @@ void Table_triggers_list::empty_lists()
   client_cs_names.empty();
   connection_cl_names.empty();
   db_cl_names.empty();
-  create_times.empty();
+  ms_create_times.empty();
 }
 
 
@@ -1001,7 +1005,7 @@ bool Trigger::add_to_file_list(void* param_arg)
       base->client_cs_names.push_back(&client_cs_name, mem_root) ||
       base->connection_cl_names.push_back(&connection_cl_name, mem_root) ||
       base->db_cl_names.push_back(&db_cl_name, mem_root) ||
-      base->create_times.push_back(&create_time, mem_root))
+      base->ms_create_times.push_back(&ms_create_time, mem_root))
     return 1;
   return 0;
 }
@@ -1380,7 +1384,8 @@ bool Table_triggers_list::check_n_load(THD *thd, const char *db,
       List_iterator_fast<LEX_STRING> it_client_cs_name(trigger_list->client_cs_names);
       List_iterator_fast<LEX_STRING> it_connection_cl_name(trigger_list->connection_cl_names);
       List_iterator_fast<LEX_STRING> it_db_cl_name(trigger_list->db_cl_names);
-      List_iterator_fast<ulonglong> it_create_times(trigger_list->create_times);
+      List_iterator_fast<ulonglong>
+        it_create_times(trigger_list->ms_create_times);
       LEX *old_lex= thd->lex;
       LEX lex;
       sp_rcontext *save_spcont= thd->spcont;
@@ -1466,7 +1471,13 @@ bool Table_triggers_list::check_n_load(THD *thd, const char *db,
 
         trigger->sql_mode= sql_mode;
         trigger->definition= *trg_create_str;
-        trigger->create_time= trg_create_time ? *trg_create_time : 0;
+        trigger->ms_create_time= trg_create_time ? *trg_create_time : 0;
+        /*
+          Fix time if in 100th of second (comparison with max uint * 100
+          (max possible timestamp in the old format))
+        */
+        if (trigger->ms_create_time < 429496729400ULL)
+          trigger->ms_create_time*= 10000;
         trigger->name= sp ? sp->m_name : empty_lex_str;
         trigger->on_table_name.str= (char*) lex.raw_trg_on_table_name_begin;
         trigger->on_table_name.length= (lex.raw_trg_on_table_name_end -
diff --git a/sql/sql_trigger.h b/sql/sql_trigger.h
index e2e6e1b7acc..70f246d4e0d 100644
--- a/sql/sql_trigger.h
+++ b/sql/sql_trigger.h
@@ -112,7 +112,7 @@ public:
   GRANT_INFO subject_table_grants;
   sql_mode_t sql_mode;
   /* Store create time. Can't be mysql_time_t as this holds also sub seconds */
-  ulonglong create_time;
+  ulonglong ms_create_time; // Create time timestamp in microseconds
   trg_event_type event;
   trg_action_time_type action_time;
   uint action_order;
@@ -195,7 +195,7 @@ public:
   */
   List<ulonglong> definition_modes_list;
   /** Create times for triggers */
-  List<ulonglong> create_times;
+  List<ulonglong> ms_create_times;
 
   List<LEX_STRING>  definers_list;
 
@@ -331,4 +331,14 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create);
 extern const char * const TRG_EXT;
 extern const char * const TRN_EXT;
 
+
+/**
+  Make time compatible with MySQL 5.7 trigger time.
+*/
+
+inline ulonglong make_ms_time(my_time_t time, ulong time_sec_part)
+{
+  return ((ulonglong) time)*1000000 + time_sec_part;
+}
+
 #endif /* SQL_TRIGGER_INCLUDED */
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 020830483c5..3b51f6d2451 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -764,7 +764,7 @@ static File_option view_parameters[]=
   my_offsetof(TABLE_LIST, with_check),
   FILE_OPTIONS_ULONGLONG},
  {{ C_STRING_WITH_LEN("timestamp")},
-  my_offsetof(TABLE_LIST, timestamp),
+  my_offsetof(TABLE_LIST, ms_timestamp),
   FILE_OPTIONS_TIMESTAMP},
  {{ C_STRING_WITH_LEN("create-version")},
   my_offsetof(TABLE_LIST, file_version),
@@ -788,6 +788,16 @@ static File_option view_parameters[]=
   FILE_OPTIONS_STRING}
 };
 
+
+static File_option view_timestamp_parameters[]=
+{
+
+ {{ C_STRING_WITH_LEN("timestamp")}, 0, FILE_OPTIONS_TIMESTAMP},
+ {{NullS, 0}, 0, FILE_OPTIONS_STRING}
+};
+
+
+
 static LEX_STRING view_file_type[]= {{(char*) STRING_WITH_LEN("VIEW") }};
 
 
@@ -805,8 +815,8 @@ int mariadb_fix_view(THD *thd, TABLE_LIST *view, bool wrong_checksum,
                      &path, path_buff, sizeof(path_buff),
                      &file, view);
   /* init timestamp */
-  if (!view->timestamp.str)
-    view->timestamp.str= view->timestamp_buffer;
+  if (!view->ms_timestamp.str)
+    view->ms_timestamp.str= view->timestamp_buffer;
 
   if (swap_alg && view->algorithm != VIEW_ALGORITHM_UNDEFINED)
   {
@@ -1006,8 +1016,8 @@ loop_out:
                      &path, path_buff, sizeof(path_buff),
                      &file, view);
   /* init timestamp */
-  if (!view->timestamp.str)
-    view->timestamp.str= view->timestamp_buffer;
+  if (!view->ms_timestamp.str)
+    view->ms_timestamp.str= view->timestamp_buffer;
 
   /* check old .frm */
   {
@@ -1131,7 +1141,31 @@ err:
   DBUG_RETURN(error);
 }
 
+/**
+  Check is TABLE_LEST and SHARE match
+  @param[in]  view                TABLE_LIST of the view
+  @param[in]  share               Share object of view
+
+  @return false on error or misspatch
+*/
 
+bool mariadb_view_version_get(TABLE_SHARE *share)
+{
+  DBUG_ASSERT(share->is_view);
+
+  if (!(share->tabledef_version.str=
+        (uchar*) alloc_root(&share->mem_root,
+                            MICROSECOND_TIMESTAMP_BUFFER_SIZE)))
+    return TRUE;
+
+  DBUG_ASSERT(share->view_def != NULL);
+  if (share->view_def->parse((uchar *) &share->tabledef_version, NULL,
+                             view_timestamp_parameters, 1,
+                             &file_parser_dummy_hook))
+    return TRUE;
+  DBUG_ASSERT(share->tabledef_version.length == MICROSECOND_TIMESTAMP_BUFFER_SIZE-1);
+  return FALSE;
+}
 
 /**
   read VIEW .frm and create structures
@@ -1193,6 +1227,10 @@ bool mysql_make_view(THD *thd, TABLE_SHARE *share, TABLE_LIST *table,
     mysql_derived_reinit(thd, NULL, table);
 
     DEBUG_SYNC(thd, "after_cached_view_opened");
+    if (!share->tabledef_version.length)
+    {
+      mariadb_view_version_get(share);
+    }
     DBUG_RETURN(0);
   }
 
@@ -1229,8 +1267,8 @@ bool mysql_make_view(THD *thd, TABLE_SHARE *share, TABLE_LIST *table,
   arena= thd->activate_stmt_arena_if_needed(&backup);
 
   /* init timestamp */
-  if (!table->timestamp.str)
-    table->timestamp.str= table->timestamp_buffer;
+  if (!table->ms_timestamp.str)
+    table->ms_timestamp.str= table->timestamp_buffer;
   /* prepare default values for old format */
   table->view_suid= TRUE;
   table->definer.user.str= table->definer.host.str= 0;
@@ -1246,6 +1284,19 @@ bool mysql_make_view(THD *thd, TABLE_SHARE *share, TABLE_LIST *table,
                                       required_view_parameters,
                                       &file_parser_dummy_hook)))
     goto end;
+  if (!share->tabledef_version.length)
+  {
+    share->tabledef_version.str= (const uchar *)
+                                 memdup_root(&share->mem_root,
+                                             (const void *)
+                                             table->ms_timestamp.str,
+                                             (share->tabledef_version.length=
+                                              table->ms_timestamp.length));
+  }
+  if (!table->tabledef_version.length)
+  {
+    table->set_view_def_version(&table->ms_timestamp);
+  }
 
   /*
     check old format view .frm
@@ -2151,7 +2202,7 @@ mysql_rename_view(THD *thd,
       object for it.
     */
     view_def.reset();
-    view_def.timestamp.str= view_def.timestamp_buffer;
+    view_def.ms_timestamp.str= view_def.timestamp_buffer;
     view_def.view_suid= TRUE;
 
     /* get view definition and source */
diff --git a/sql/table.cc b/sql/table.cc
index ca6ce02e4f2..24412966b36 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -8480,6 +8480,53 @@ bool TABLE_LIST::is_with_table()
 }
 
 
+bool TABLE_LIST::is_table_ref_id_equal(THD* thd, TABLE_SHARE *s)
+{
+  enum enum_table_ref_type tp= s->get_table_ref_type();
+  if (m_table_ref_type == tp)
+  {
+    bool res= m_table_ref_version == s->get_table_ref_version();
+
+    /*
+      If definition is different check content version
+    */
+    if (tabledef_version.length &&
+        tabledef_version.length == s->tabledef_version.length &&
+        memcmp(tabledef_version.str, s->tabledef_version.str,
+               tabledef_version.length) == 0)
+    {
+      if (table && table->triggers)
+      {
+
+        ulonglong ms_stmt_prepare= thd->ms_prepare_time;
+        if (ms_stmt_prepare)
+          for(uint i= 0; i < TRG_EVENT_MAX; i++)
+            for (uint j= 0; j < TRG_ACTION_MAX; j++)
+            {
+              Trigger *tr=
+                table->triggers->get_trigger((trg_event_type)i,
+                                             (trg_action_time_type)j);
+              if (tr)
+                if (ms_stmt_prepare <= tr->ms_create_time)
+                {
+                  set_tabledef_version(s);
+                  return FALSE;
+                }
+            }
+      }
+      set_table_id(s);
+      return TRUE;
+    }
+    else
+      tabledef_version.length= 0;
+    return res;
+  }
+  else
+    set_tabledef_version(s);
+  return FALSE;
+}
+
+
 uint TABLE_SHARE::actual_n_key_parts(THD *thd)
 {
   return use_ext_keys &&
diff --git a/sql/table.h b/sql/table.h
index 14d6b787b4e..82508388f84 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -32,8 +32,18 @@
 #include "filesort_utils.h"
 #include "parse_file.h"
 
-/* buffer for timestamp (19+1) */
-#define VIEW_TIME_STAMP_BUFFER_SIZE (PARSE_FILE_TIMESTAMPLENGTH + 1)
+/*
+  Buffer for unix timestamp in microseconds:
+  9,223,372,036,854,775,807 (signed int64 maximal value)
+  1 234 567 890 123 456 789
+
+  Note: we can use unsigned for calculation, but practically they
+  are the same by probability to overflow them (signed int64 in
+  microseconds is enough for almost 3e5 years) and signed allow to
+  avoid increasing the buffer (the old buffer for human readable
+  date was 19+1).
+*/
+#define MICROSECOND_TIMESTAMP_BUFFER_SIZE (19 + 1)
 
 /* Structs that defines the TABLE */
 
@@ -2146,6 +2156,12 @@ struct TABLE_LIST
     to view with SQL SECURITY DEFINER)
   */
   Security_context *security_ctx;
+  uchar tabledef_version_buf[MY_UUID_SIZE >
+                               MICROSECOND_TIMESTAMP_BUFFER_SIZE-1 ?
+                             MY_UUID_SIZE + 1 :
+                             MICROSECOND_TIMESTAMP_BUFFER_SIZE];
+  LEX_CUSTRING tabledef_version;
+
   /*
     This view security context (non-zero only for views with
     SQL SECURITY DEFINER)
@@ -2159,7 +2175,7 @@ struct TABLE_LIST
   LEX_STRING	source;			/* source of CREATE VIEW */
   LEX_STRING	view_db;		/* saved view database */
   LEX_STRING	view_name;		/* saved view name */
-  LEX_STRING	timestamp;		/* GMT time stamp of last operation */
+  LEX_STRING	ms_timestamp;		/* GMT time stamp of last operation */
   st_lex_user   definer;                /* definer of view */
   ulonglong	file_version;		/* version of file's field set */
   ulonglong	mariadb_version;	/* version of server on creation */
@@ -2244,7 +2260,7 @@ struct TABLE_LIST
   /* FRMTYPE_ERROR if any type is acceptable */
   enum frm_type_enum required_type;
   handlerton	*db_type;		/* table_type for handler */
-  char		timestamp_buffer[VIEW_TIME_STAMP_BUFFER_SIZE];
+  char		timestamp_buffer[MICROSECOND_TIMESTAMP_BUFFER_SIZE];
   /*
     This TABLE_LIST object is just placeholder for prelocking, it will be
     used for implicit LOCK TABLES only and won't be used in real statement.
@@ -2442,12 +2458,7 @@ struct TABLE_LIST
 
     @sa check_and_update_table_version()
   */
-  inline bool is_table_ref_id_equal(TABLE_SHARE *s) const
-  {
-    return (m_table_ref_type == s->get_table_ref_type() &&
-            m_table_ref_version == s->get_table_ref_version());
-  }
-
+  bool is_table_ref_id_equal(THD *thd, TABLE_SHARE *s);
   /**
     Record the value of metadata version of the corresponding
     table definition cache element in this parse tree node.
@@ -2464,6 +2475,26 @@ struct TABLE_LIST
     m_table_ref_version= table_ref_version_arg;
   }
 
+  void set_table_id(TABLE_SHARE *s)
+  {
+    set_table_ref_id(s);
+    set_tabledef_version(s);
+  }
+
+  void set_tabledef_version(TABLE_SHARE *s)
+  {
+    if (!tabledef_version.length && s->tabledef_version.length)
+    {
+      DBUG_ASSERT(s->tabledef_version.length <
+                  sizeof(tabledef_version_buf));
+      tabledef_version.str= tabledef_version_buf;
+      memcpy(tabledef_version_buf, s->tabledef_version.str,
+             (tabledef_version.length= s->tabledef_version.length));
+      // safety
+      tabledef_version_buf[tabledef_version.length]= 0;
+    }
+  }
+
   /* Set of functions returning/setting state of a derived table/view. */
   inline bool is_non_derived()
   {
@@ -2590,6 +2621,12 @@ struct TABLE_LIST
     }
   }
 
+  inline void set_view_def_version(LEX_STRING *version)
+  {
+    m_table_ref_type= TABLE_REF_VIEW;
+    tabledef_version.str= (const uchar *) version->str;
+    tabledef_version.length= version->length;
+  }
 private:
   bool prep_check_option(THD *thd, uint8 check_opt_type);
   bool prep_where(THD *thd, Item **conds, bool no_where_clause);