MDEV-27744 InnoDB: Failing assertion: !cursor->index->is_committed() in row0ins.cc (from row_ins_sec_index_entry_by_modify) | Assertion `0' failed in row_upd_sec_index_entry (debug) | Corruptionbb-10.3-bar-MDEV-27744

The crash happened with an indexed virtual column whose
value is evaluated using a function that has a different meaning
in sql_mode='' vs sql_mode=ORACLE:

- DECODE()
- LTRIM()
- RTRIM()
- LPAD()
- RPAD()
- REPLACE()
- SUBSTR()

For example:

CREATE TABLE t1 (
  b VARCHAR(1),
  g CHAR(1) GENERATED ALWAYS AS (SUBSTR(b,0,0)) VIRTUAL,
  KEY g(g)
);

So far we had replacement XXX_ORACLE() functions for all mentioned function,
e.g. SUBSTR_ORACLE() for SUBSTR(). So it was possible to correctly re-parse
SUBSTR_ORACLE() even in sql_mode=''.

But it was not possible to re-parse the MariaDB version of SUBSTR()
after switching to sql_mode=ORACLE. It was erroneously mis-interpreted
as SUBSTR_ORACLE().

As a result, this combination worked fine:

SET sql_mode=ORACLE;
CREATE TABLE t1 ... g CHAR(1) GENERATED ALWAYS AS (SUBSTR(b,0,0)) VIRTUAL, ...;
INSERT ...
FLUSH TABLES;
SET sql_mode='';
INSERT ...

But the other way around it crashed:

SET sql_mode='';
CREATE TABLE t1 ... g CHAR(1) GENERATED ALWAYS AS (SUBSTR(b,0,0)) VIRTUAL, ...;
INSERT ...
FLUSH TABLES;
SET sql_mode=ORACLE;
INSERT ...

At CREATE time, SUBSTR was instantiated as Item_func_substr and printed
in the FRM file as substr(). At re-open time with sql_mode=ORACLE, "substr()"
was erroneously instantiated as Item_func_substr_oracle.

Fix:

The fix proposes a symmetric solution. It provides a way to re-parse reliably
all sql_mode dependent functions to their original CREATE TABLE time meaning,
no matter what the open-time sql_mode is.

We take advantage of the same idea we previously used to resolve sql_mode
dependent data types.

Now all sql_mode dependent functions are printed by SHOW using a schema
qualifier when the current sql_mode differs from the function sql_mode:

SET sql_mode='';
CREATE TABLE t1 ... SUBSTR(a,b,c) ..;
SET sql_mode=ORACLE;
SHOW CREATE TABLE t1;   ->   mariadb_schema.substr(a,b,c)

SET sql_mode=ORACLE;
CREATE TABLE t2 ... SUBSTR(a,b,c) ..;
SET sql_mode='';
SHOW CREATE TABLE t1;   ->   oracle_schema.substr(a,b,c)

Old replacement names like substr_oracle() are still understood for
backward compatibility and used in FRM files (for downgrade compatibility),
but they are not printed by SHOW any more.

24 files changed, 996 insertions, 445 deletions

diff --git a/sql/item.h b/sql/item.h
index 5ae3d7a3435..f70e6256551 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -28,6 +28,7 @@
 #include "field.h"                              /* Derivation */
 #include "sql_type.h"
 #include "sql_time.h"
+#include "sql_schema.h"
 #include "mem_root_array.h"
 
 C_MODE_START
@@ -1479,7 +1480,8 @@ public:
                                        QT_ITEM_IDENT_SKIP_DB_NAMES |
                                        QT_ITEM_IDENT_SKIP_TABLE_NAMES |
                                        QT_NO_DATA_EXPANSION |
-                                       QT_TO_SYSTEM_CHARSET),
+                                       QT_TO_SYSTEM_CHARSET |
+                                       QT_FOR_FRM),
                      LOWEST_PRECEDENCE);
   }
   virtual void print(String *str, enum_query_type query_type);
@@ -4830,6 +4832,14 @@ public:
     return (this->*processor)(arg);
   }
   /*
+    Built-in schema, e.g. mariadb_schema, oracle_schema, maxdb_schema
+  */
+  virtual const Schema *schema() const
+  {
+    // A function does not belong to a built-in schema by default
+    return NULL;
+  }
+  /*
     This method is used for debug purposes to print the name of an
     item to the debug log. The second use of this method is as
     a helper function of print() and error messages, where it is
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index fe6b8feb4de..600fe8b0c8a 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -3307,9 +3307,24 @@ void Item_func_case_simple::print(String *str, enum_query_type query_type)
 }
 
 
+Item_func_decode_oracle *
+Item_func_decode_oracle::create(THD *thd, const LEX_CSTRING &name,
+                                List<Item> *item_list)
+{
+  if (unlikely(!item_list || item_list->elements < 3))
+  {
+    wrong_param_count_error(oracle_schema_ref.name(), name);
+    return NULL;
+  }
+  return new (thd->mem_root) Item_func_decode_oracle(thd, *item_list);
+}
+
+
 void Item_func_decode_oracle::print(String *str, enum_query_type query_type)
 {
-  str->append(func_name());
+  print_sql_mode_dependent_name(str, query_type,
+                                oracle_schema_ref,
+                                Item_func_decode_oracle::func_name());
   str->append('(');
   args[0]->print(str, query_type);
   for (uint i= 1, count= when_count() ; i <= count; i++)
diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index 671fa52635d..9143366beec 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -2238,7 +2238,10 @@ public:
   Item_func_decode_oracle(THD *thd, List<Item> &list)
    :Item_func_case_simple(thd, list)
   { }
-  const char *func_name() const { return "decode_oracle"; }
+  static Item_func_decode_oracle *create(THD *thd, const LEX_CSTRING &name,
+                                         List<Item> *list);
+  const Schema *schema() const { return &oracle_schema_ref; }
+  const char *func_name() const { return "decode"; }
   void print(String *str, enum_query_type query_type);
   bool fix_length_and_dec();
   Item *find_item();
diff --git a/sql/item_create.cc b/sql/item_create.cc
index 52e0312f89b..c75e8ce3032 100644
--- a/sql/item_create.cc
+++ b/sql/item_create.cc
@@ -598,12 +598,28 @@ protected:
 };
 
 
+class Create_func_decode : public Create_native_func
+{
+public:
+  virtual Item *create_native(THD *thd, const LEX_CSTRING *name,
+                              List<Item> *item_list)
+  {
+    return Item_func_decode::create(thd, *name, item_list);
+  }
+
+  static Create_func_decode s_singleton;
+
+protected:
+  Create_func_decode() {}
+  virtual ~Create_func_decode() {}
+};
+
+
 class Create_func_decode_oracle : public Create_native_func
 {
 public:
   virtual Item *create_native(THD *thd, const LEX_CSTRING *name,
                               List<Item> *item_list);
-
   static Create_func_decode_oracle s_singleton;
 
 protected:
@@ -2267,29 +2283,23 @@ public:
   virtual Item *create_native(THD *thd, const LEX_CSTRING *name,
                               List<Item> *item_list)
   {
-    return thd->variables.sql_mode & MODE_ORACLE ?
-           create_native_oracle(thd, name, item_list) :
-           create_native_std(thd, name, item_list);
+    return Item_func_lpad::create(thd, *name, item_list);
   }
   static Create_func_lpad s_singleton;
 
 protected:
   Create_func_lpad() {}
   virtual ~Create_func_lpad() {}
-  Item *create_native_std(THD *thd, const LEX_CSTRING *name,
-                          List<Item> *items);
-  Item *create_native_oracle(THD *thd, const LEX_CSTRING *name,
-                             List<Item> *items);
 };
 
 
-class Create_func_lpad_oracle : public Create_func_lpad
+class Create_func_lpad_oracle : public Create_native_func
 {
 public:
   Item *create_native(THD *thd, const LEX_CSTRING *name,
                       List<Item> *item_list)
   {
-    return create_native_oracle(thd, name, item_list);
+    return Item_func_lpad_oracle::create(thd, *name, item_list);
   }
   static Create_func_lpad_oracle s_singleton;
 };
@@ -2745,29 +2755,23 @@ public:
   virtual Item *create_native(THD *thd, const LEX_CSTRING *name,
                               List<Item> *item_list)
   {
-    return thd->variables.sql_mode & MODE_ORACLE ?
-           create_native_oracle(thd, name, item_list) :
-           create_native_std(thd, name, item_list);
+    return Item_func_rpad::create(thd, *name, item_list);
   }
   static Create_func_rpad s_singleton;
 
 protected:
   Create_func_rpad() {}
   virtual ~Create_func_rpad() {}
-  Item *create_native_std(THD *thd, const LEX_CSTRING *name,
-                          List<Item> *items);
-  Item *create_native_oracle(THD *thd, const LEX_CSTRING *name,
-                             List<Item> *items);
 };
 
 
-class Create_func_rpad_oracle : public Create_func_rpad
+class Create_func_rpad_oracle : public Create_native_func
 {
 public:
   Item *create_native(THD *thd, const LEX_CSTRING *name,
                       List<Item> *item_list)
   {
-    return create_native_oracle(thd, name, item_list);
+    return Item_func_rpad_oracle::create(thd, *name, item_list);
   }
   static Create_func_rpad_oracle s_singleton;
 };
@@ -3990,22 +3994,10 @@ Item*
 Create_func_concat::create_native(THD *thd, const LEX_CSTRING *name,
                                   List<Item> *item_list)
 {
-  int arg_count= 0;
-
-  if (item_list != NULL)
-    arg_count= item_list->elements;
-
-  if (unlikely(arg_count < 1))
-  {
-    my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0), name->str);
-    return NULL;
-  }
-
-  return thd->variables.sql_mode & MODE_ORACLE ?
-    new (thd->mem_root) Item_func_concat_operator_oracle(thd, *item_list) :
-    new (thd->mem_root) Item_func_concat(thd, *item_list);
+  return Item_func_concat::create(thd, *name, item_list);
 }
 
+
 Create_func_concat_operator_oracle
   Create_func_concat_operator_oracle::s_singleton;
 
@@ -4013,18 +4005,7 @@ Item*
 Create_func_concat_operator_oracle::create_native(THD *thd, const LEX_CSTRING *name,
                                                   List<Item> *item_list)
 {
-  int arg_count= 0;
-
-  if (item_list != NULL)
-    arg_count= item_list->elements;
-
-  if (unlikely(arg_count < 1))
-  {
-    my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0), name->str);
-    return NULL;
-  }
-
-  return new (thd->mem_root) Item_func_concat_operator_oracle(thd, *item_list);
+  return Item_func_concat_operator_oracle::create(thd, *name, item_list);
 }
 
 Create_func_decode_histogram Create_func_decode_histogram::s_singleton;
@@ -4035,21 +4016,19 @@ Create_func_decode_histogram::create_2_arg(THD *thd, Item *arg1, Item *arg2)
   return new (thd->mem_root) Item_func_decode_histogram(thd, arg1, arg2);
 }
 
+
+Create_func_decode Create_func_decode::s_singleton;
+
 Create_func_decode_oracle Create_func_decode_oracle::s_singleton;
 
 Item*
 Create_func_decode_oracle::create_native(THD *thd, const LEX_CSTRING *name,
                                          List<Item> *item_list)
 {
-  uint arg_count= item_list ? item_list->elements : 0;
-  if (unlikely(arg_count < 3))
-  {
-    my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0), name->str);
-    return NULL;
-  }
-  return new (thd->mem_root) Item_func_decode_oracle(thd, *item_list);
+  return Item_func_decode_oracle::create(thd, *name, item_list);
 }
 
+
 Create_func_concat_ws Create_func_concat_ws::s_singleton;
 
 Item*
@@ -5852,10 +5831,7 @@ Create_func_length Create_func_length::s_singleton;
 Item*
 Create_func_length::create_1_arg(THD *thd, Item *arg1)
 {
-  if (thd->variables.sql_mode & MODE_ORACLE)
-    return new (thd->mem_root) Item_func_char_length(thd, arg1);
-  else
-    return new (thd->mem_root) Item_func_octet_length(thd, arg1);
+  return new (thd->mem_root) Item_func_octet_length(thd, arg1);
 }
 
 Create_func_octet_length Create_func_octet_length::s_singleton;
@@ -6008,72 +5984,12 @@ Create_func_lpad Create_func_lpad::s_singleton;
 
 Create_func_lpad_oracle Create_func_lpad_oracle::s_singleton;
 
-Item*
-Create_func_lpad::create_native_std(THD *thd, const LEX_CSTRING *name,
-                                    List<Item> *item_list)
-{
-  Item *func= NULL;
-  int arg_count= item_list ? item_list->elements : 0;
-
-  switch (arg_count) {
-  case 2:
-  {
-    Item *param_1= item_list->pop();
-    Item *param_2= item_list->pop();
-    func= new (thd->mem_root) Item_func_lpad(thd, param_1, param_2);
-    break;
-  }
-  case 3:
-  {
-    Item *param_1= item_list->pop();
-    Item *param_2= item_list->pop();
-    Item *param_3= item_list->pop();
-    func= new (thd->mem_root) Item_func_lpad(thd, param_1, param_2, param_3);
-    break;
-  }
-  default:
-    my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0), name->str);
-    break;
-  }
-
-  return func;
-}
-
-
-Item*
-Create_func_lpad::create_native_oracle(THD *thd, const LEX_CSTRING *name,
-                                       List<Item> *item_list)
-{
-  int arg_count= item_list ? item_list->elements : 0;
-  switch (arg_count) {
-  case 2:
-  {
-    Item *param_1= item_list->pop();
-    Item *param_2= item_list->pop();
-    return new (thd->mem_root) Item_func_lpad_oracle(thd, param_1, param_2);
-  }
-  case 3:
-  {
-    Item *param_1= item_list->pop();
-    Item *param_2= item_list->pop();
-    Item *param_3= item_list->pop();
-    return new (thd->mem_root) Item_func_lpad_oracle(thd, param_1,
-                                                     param_2, param_3);
-  }
-  default:
-    my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0), name->str);
-    break;
-  }
-  return NULL;
-}
-
-
 Create_func_ltrim Create_func_ltrim::s_singleton;
 
 Item*
 Create_func_ltrim::create_1_arg(THD *thd, Item *arg1)
 {
-  return Lex_trim(TRIM_LEADING, arg1).make_item_func_trim(thd);
+  return Lex_trim(TRIM_LEADING, arg1).make_item_func_trim_std(thd);
 }
 
 
@@ -6544,72 +6460,12 @@ Create_func_rpad Create_func_rpad::s_singleton;
 
 Create_func_rpad_oracle Create_func_rpad_oracle::s_singleton;
 
-Item*
-Create_func_rpad::create_native_std(THD *thd, const LEX_CSTRING *name,
-                                    List<Item> *item_list)
-{
-  Item *func= NULL;
-  int arg_count= item_list ? item_list->elements : 0;
-
-  switch (arg_count) {
-  case 2:
-  {
-    Item *param_1= item_list->pop();
-    Item *param_2= item_list->pop();
-    func= new (thd->mem_root) Item_func_rpad(thd, param_1, param_2);
-    break;
-  }
-  case 3:
-  {
-    Item *param_1= item_list->pop();
-    Item *param_2= item_list->pop();
-    Item *param_3= item_list->pop();
-    func= new (thd->mem_root) Item_func_rpad(thd, param_1, param_2, param_3);
-    break;
-  }
-  default:
-    my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0), name->str);
-    break;
-  }
-
-  return func;
-}
-
-
-Item*
-Create_func_rpad::create_native_oracle(THD *thd, const LEX_CSTRING *name,
-                                       List<Item> *item_list)
-{
-  int arg_count= item_list ? item_list->elements : 0;
-  switch (arg_count) {
-  case 2:
-  {
-    Item *param_1= item_list->pop();
-    Item *param_2= item_list->pop();
-    return new (thd->mem_root) Item_func_rpad_oracle(thd, param_1, param_2);
-  }
-  case 3:
-  {
-    Item *param_1= item_list->pop();
-    Item *param_2= item_list->pop();
-    Item *param_3= item_list->pop();
-    return new (thd->mem_root) Item_func_rpad_oracle(thd, param_1,
-                                                     param_2, param_3);
-  }
-  default:
-    my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0), name->str);
-    break;
-  }
-  return NULL;
-}
-
-
 Create_func_rtrim Create_func_rtrim::s_singleton;
 
 Item*
 Create_func_rtrim::create_1_arg(THD *thd, Item *arg1)
 {
-  return Lex_trim(TRIM_TRAILING, arg1).make_item_func_trim(thd);
+  return Lex_trim(TRIM_TRAILING, arg1).make_item_func_trim_std(thd);
 }
 
 
@@ -7120,7 +6976,7 @@ Create_func_year_week::create_native(THD *thd, const LEX_CSTRING *name,
   - keep 1 line per entry, it makes grep | sort easier
 */
 
-Native_func_registry func_array[] =
+const Native_func_registry func_array[] =
 {
   { { STRING_WITH_LEN("ABS") }, BUILDER(Create_func_abs)},
   { { STRING_WITH_LEN("ACOS") }, BUILDER(Create_func_acos)},
@@ -7170,6 +7026,7 @@ Native_func_registry func_array[] =
   { { STRING_WITH_LEN("DAYOFMONTH") }, BUILDER(Create_func_dayofmonth)},
   { { STRING_WITH_LEN("DAYOFWEEK") }, BUILDER(Create_func_dayofweek)},
   { { STRING_WITH_LEN("DAYOFYEAR") }, BUILDER(Create_func_dayofyear)},
+  { { STRING_WITH_LEN("DECODE") }, BUILDER(Create_func_decode)},
   { { STRING_WITH_LEN("DEGREES") }, BUILDER(Create_func_degrees)},
   { { STRING_WITH_LEN("DECODE_HISTOGRAM") }, BUILDER(Create_func_decode_histogram)},
   { { STRING_WITH_LEN("DECODE_ORACLE") }, BUILDER(Create_func_decode_oracle)},
@@ -7474,9 +7331,25 @@ Native_func_registry func_array[] =
   { {0, 0}, NULL}
 };
 
-size_t func_array_length= sizeof(func_array) / sizeof(Native_func_registry) - 1;
 
-static HASH native_functions_hash;
+const Native_func_registry func_array_oracle_overrides[] =
+{
+  { { STRING_WITH_LEN("CONCAT") },  BUILDER(Create_func_concat_operator_oracle)},
+  { { STRING_WITH_LEN("DECODE") },  BUILDER(Create_func_decode_oracle)},
+  { { STRING_WITH_LEN("LENGTH") },  BUILDER(Create_func_char_length)},
+  { { STRING_WITH_LEN("LPAD") },    BUILDER(Create_func_lpad_oracle)},
+  { { STRING_WITH_LEN("LTRIM") },   BUILDER(Create_func_ltrim_oracle)},
+  { { STRING_WITH_LEN("RPAD") },    BUILDER(Create_func_rpad_oracle)},
+  { { STRING_WITH_LEN("RTRIM") },   BUILDER(Create_func_rtrim_oracle)},
+
+  { {0, 0}, NULL}
+};
+
+
+const size_t func_array_length= sizeof(func_array) / sizeof(Native_func_registry) - 1;
+
+Native_functions_hash native_functions_hash;
+Native_functions_hash native_functions_hash_oracle_overrides;
 
 extern "C" uchar*
 get_native_fct_hash_key(const uchar *buff, size_t *length,
@@ -7493,85 +7366,89 @@ get_native_fct_hash_key(const uchar *buff, size_t *length,
   startup only (before going multi-threaded)
 */
 
-int item_create_init()
+bool Native_functions_hash::init(size_t count)
 {
-  DBUG_ENTER("item_create_init");
+  DBUG_ENTER("Native_functions_hash::item");
 
-  if (my_hash_init(& native_functions_hash,
+  if (my_hash_init(this,
                    system_charset_info,
-                   array_elements(func_array),
+                   (ulong) count,
                    0,
                    0,
                    (my_hash_get_key) get_native_fct_hash_key,
                    NULL,                          /* Nothing to free */
                    MYF(0)))
-    DBUG_RETURN(1);
+    DBUG_RETURN(true);
 
-  DBUG_RETURN(item_create_append(func_array));
+  DBUG_RETURN(false);
 }
 
-int item_create_append(Native_func_registry array[])
+
+bool Native_functions_hash::append(const Native_func_registry array[])
 {
-  Native_func_registry *func;
+  const Native_func_registry *func;
 
-  DBUG_ENTER("item_create_append");
+  DBUG_ENTER("Native_functions_hash::append");
 
   for (func= array; func->builder != NULL; func++)
   {
-    if (my_hash_insert(& native_functions_hash, (uchar*) func))
-      DBUG_RETURN(1);
+    if (my_hash_insert(this, (uchar*) func))
+      DBUG_RETURN(true);
   }
 
 #ifndef DBUG_OFF
-  for (uint i=0 ; i < native_functions_hash.records ; i++)
+  for (uint i=0 ; i < records ; i++)
   {
-    func= (Native_func_registry*) my_hash_element(& native_functions_hash, i);
+    func= (Native_func_registry*) my_hash_element(this, i);
     DBUG_PRINT("info", ("native function: %s  length: %u",
                         func->name.str, (uint) func->name.length));
   }
 #endif
 
-  DBUG_RETURN(0);
+  DBUG_RETURN(false);
 }
 
-int item_create_remove(Native_func_registry array[])
+
+bool Native_functions_hash::remove(const Native_func_registry array[])
 {
-  Native_func_registry *func;
+  const Native_func_registry *func;
 
-  DBUG_ENTER("item_create_remove");
+  DBUG_ENTER("Native_functions_hash::remove");
 
   for (func= array; func->builder != NULL; func++)
   {
-    if (my_hash_delete(& native_functions_hash, (uchar*) func))
-      DBUG_RETURN(1);
+    if (my_hash_delete(this, (uchar*) func))
+      DBUG_RETURN(true);
   }
 
-  DBUG_RETURN(0);
+  DBUG_RETURN(false);
 }
 
+
 /*
   Empty the hash table for native functions.
   Note: this code is not thread safe, and is intended to be used at server
   shutdown only (after thread requests have been executed).
 */
 
-void item_create_cleanup()
+void Native_functions_hash::cleanup()
 {
-  DBUG_ENTER("item_create_cleanup");
-  my_hash_free(& native_functions_hash);
+  DBUG_ENTER("Native_functions_hash::cleanup");
+  my_hash_free(this);
   DBUG_VOID_RETURN;
 }
 
+
 Create_func *
-find_native_function_builder(THD *thd, const LEX_CSTRING *name)
+Native_functions_hash::find(THD *thd, const LEX_CSTRING &name) const
 {
   Native_func_registry *func;
   Create_func *builder= NULL;
 
   /* Thread safe */
-  func= (Native_func_registry*) my_hash_search(&native_functions_hash,
-                                               (uchar*) name->str,
-                                               name->length);
+  func= (Native_func_registry*) my_hash_search(this,
+                                               (uchar*) name.str,
+                                               name.length);
 
   if (func)
   {
@@ -7581,6 +7458,36 @@ find_native_function_builder(THD *thd, const LEX_CSTRING *name)
   return builder;
 }
 
+
+int item_create_init()
+{
+  return
+    native_functions_hash.init(func_array, array_elements(func_array)) ||
+    native_functions_hash_oracle_overrides.init(
+                                   func_array_oracle_overrides,
+                                   array_elements(func_array_oracle_overrides));
+}
+
+
+int item_create_append(const Native_func_registry array[])
+{
+  return native_functions_hash.append(array);
+}
+
+
+int item_create_remove(const Native_func_registry array[])
+{
+  return native_functions_hash.remove(array);
+}
+
+
+void item_create_cleanup()
+{
+  native_functions_hash.cleanup();
+  native_functions_hash_oracle_overrides.cleanup();
+}
+
+
 Create_qfunc *
 find_qualified_function_builder(THD *thd)
 {
diff --git a/sql/item_create.h b/sql/item_create.h
index 894e9777b8d..3fadaecb090 100644
--- a/sql/item_create.h
+++ b/sql/item_create.h
@@ -145,16 +145,6 @@ protected:
 
 
 /**
-  Find the native function builder associated with a given function name.
-  @param thd The current thread
-  @param name The native function name
-  @return The native function builder associated with the name, or NULL
-*/
-extern Create_func *find_native_function_builder(THD *thd,
-                                                 const LEX_CSTRING *name);
-
-
-/**
   Find the function builder for qualified functions.
   @param thd The current thread
   @return A function builder for qualified functions
@@ -215,9 +205,48 @@ struct Native_func_registry
   Create_func *builder;
 };
 
+
+class Native_functions_hash: public HASH
+{
+public:
+  Native_functions_hash()
+  {
+    bzero(this, sizeof(*this));
+  }
+  ~Native_functions_hash()
+  {
+    /*
+      No automatic free.
+      The the upper level code should call cleanup() explicitly.
+    */
+    DBUG_ASSERT(!records);
+  }
+  bool init(size_t count);
+  bool init(const Native_func_registry array[], size_t count)
+  {
+    return init(count) || append(array);
+  }
+  bool append(const Native_func_registry array[]);
+  bool remove(const Native_func_registry array[]);
+  void cleanup();
+  /**
+    Find the native function builder associated with a given function name.
+    @param thd The current thread
+    @param name The native function name
+    @return The native function builder associated with the name, or NULL
+  */
+  Create_func *find(THD *thd, const LEX_CSTRING &name) const;
+};
+
+extern Native_functions_hash native_functions_hash;
+extern Native_functions_hash native_functions_hash_oracle_overrides;
+
+extern const Native_func_registry func_array[];
+extern const size_t func_array_length;
+
 int item_create_init();
-int item_create_append(Native_func_registry array[]);
-int item_create_remove(Native_func_registry array[]);
+int item_create_append(const Native_func_registry array[]);
+int item_create_remove(const Native_func_registry array[]);
 void item_create_cleanup();
 
 Item *create_func_dyncol_create(THD *thd, List<DYNCALL_CREATE_DEF> &list);
diff --git a/sql/item_func.cc b/sql/item_func.cc
index 2c79ca531d3..4e83ddea90e 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -128,6 +128,16 @@ Item_args::Item_args(THD *thd, const Item_args *other)
 }
 
 
+void Item_func::wrong_param_count_error(const LEX_CSTRING &schema_name,
+                                        const LEX_CSTRING &func_name)
+{
+  DBUG_ASSERT(schema_name.length);
+  Database_qualified_name qname(schema_name, func_name);
+  my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0),
+           ErrConvDQName(&qname).ptr());
+}
+
+
 void Item_func::sync_with_sum_func_and_with_field(List<Item> &list)
 {
   List_iterator_fast<Item> li(list);
@@ -595,9 +605,7 @@ table_map Item_func::not_null_tables() const
 void Item_func::print(String *str, enum_query_type query_type)
 {
   str->append(func_name());
-  str->append('(');
-  print_args(str, 0, query_type);
-  str->append(')');
+  print_args_parenthesized(str, query_type);
 }
 
 
diff --git a/sql/item_func.h b/sql/item_func.h
index d7fd24d7fa2..72be810dbd8 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -56,8 +56,117 @@ protected:
   bool check_argument_types_can_return_text(uint start, uint end) const;
   bool check_argument_types_can_return_date(uint start, uint end) const;
   bool check_argument_types_can_return_time(uint start, uint end) const;
+
+  void print_schema_qualified_name(String *to,
+                                   const LEX_CSTRING &schema_name,
+                                   const char *function_name) const
+  {
+    // e.g. oracle_schema.func()
+    to->append(schema_name);
+    to->append('.');
+    to->append(function_name);
+  }
+
+  void print_name_with_optional_suffix(String *to,
+                                       const char *function_name,
+                                       const LEX_CSTRING &suffix) const
+  {
+    // e.g. func_oracle()
+    to->append(function_name);
+    if (suffix.length)
+    {
+      to->append("_");
+      to->append(suffix);
+    }
+  }
+
+  void print_sql_mode_dependent_name(String *to, enum_query_type query_type,
+                                     const Schema &schema,
+                                     const char *function_name,
+                                     const LEX_CSTRING &suffix) const
+  {
+    if (query_type & QT_ITEM_FUNC_FORCE_SCHEMA_NAME)
+    {
+      DBUG_ASSERT((query_type & QT_FOR_FRM) == 0);
+      print_schema_qualified_name(to, schema.name(), function_name);
+    }
+    else if (query_type & QT_FOR_FRM)
+    {
+      // e.g. substr_oracle()
+      DBUG_ASSERT((query_type & QT_ITEM_FUNC_FORCE_SCHEMA_NAME) == 0);
+      print_name_with_optional_suffix(to, function_name, suffix);
+    }
+    else if (&schema != Schema::find_implied(current_thd))
+      print_schema_qualified_name(to, schema.name(), function_name);
+    else
+      to->append(function_name);
+  }
+
+  void print_sql_mode_dependent_name(String *to, enum_query_type query_type,
+                                     const Schema &schema,
+                                     const char *function_name) const
+  {
+    static const LEX_CSTRING oracle= {STRING_WITH_LEN("oracle")};
+    static const LEX_CSTRING empty= {NULL, 0};
+    const LEX_CSTRING suffix= &schema == &oracle_schema_ref ? oracle : empty;
+    print_sql_mode_dependent_name(to, query_type,
+                                  schema, function_name, suffix);
+  }
+  void print_sql_mode_dependent_name(String *to, enum_query_type query_type,
+                                     const char *function_name) const
+  {
+    print_sql_mode_dependent_name(to, query_type, *schema(), function_name);
+  }
+  void print_sql_mode_dependent(String *to, enum_query_type query_type)
+  {
+    print_sql_mode_dependent_name(to, query_type, func_name());
+    print_args_parenthesized(to, query_type);
+  }
 public:
 
+  // Print an error message for a builtin-schema qualified function call
+  static void wrong_param_count_error(const LEX_CSTRING &schema_name,
+                                      const LEX_CSTRING &func_name);
+
+  // Check that the number of arguments is greater or equal to "expected"
+  static bool create_check_args_ge(const LEX_CSTRING &name,
+                                   const List<Item> *item_list,
+                                   uint expected)
+  {
+    DBUG_ASSERT(expected > 0);
+    if (!item_list || item_list->elements < expected)
+    {
+      my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0), name.str);
+      return true;
+    }
+    return false;
+  }
+
+  // Check that the number of arguments is less or equal to "expected"
+  static bool create_check_args_le(const LEX_CSTRING &name,
+                                   const List<Item> *item_list,
+                                   uint expected)
+  {
+    DBUG_ASSERT(expected > 0);
+    if (!item_list || item_list->elements > expected)
+    {
+      my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0), name.str);
+      return true;
+    }
+    return false;
+  }
+
+  // Check that the number of arguments is in the allowed range
+  static bool create_check_args_between(const LEX_CSTRING &name,
+                                        const List<Item> *item_list,
+                                        uint min_arg_count,
+                                        uint max_arg_count)
+  {
+    DBUG_ASSERT(min_arg_count < max_arg_count);
+    return create_check_args_ge(name, item_list, min_arg_count) ||
+           create_check_args_le(name, item_list, max_arg_count);
+  }
+
   table_map not_null_tables_cache;
 
   enum Functype { UNKNOWN_FUNC,EQ_FUNC,EQUAL_FUNC,NE_FUNC,LT_FUNC,LE_FUNC,
@@ -174,6 +283,12 @@ public:
   virtual void print(String *str, enum_query_type query_type);
   void print_op(String *str, enum_query_type query_type);
   void print_args(String *str, uint from, enum_query_type query_type);
+  void print_args_parenthesized(String *str, enum_query_type query_type)
+  {
+    str->append('(');
+    print_args(str, 0, query_type);
+    str->append(')');
+  }
   inline bool get_arg0_date(MYSQL_TIME *ltime, ulonglong fuzzy_date)
   {
     DBUG_ASSERT(!(fuzzy_date & TIME_TIME_ONLY));
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index e9263fb6954..bb0f95d04f0 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -574,6 +574,27 @@ String *Item_func_decode_histogram::val_str(String *str)
 
 ///////////////////////////////////////////////////////////////////////////////
 
+Item_func_concat *
+Item_func_concat::create(THD *thd, const LEX_CSTRING &name,
+                         List<Item> *item_list)
+{
+  if (create_check_args_ge(name, item_list, 1))
+    return NULL;
+  return new (thd->mem_root) Item_func_concat(thd, *item_list);
+}
+
+
+Item_func_concat_operator_oracle *
+Item_func_concat_operator_oracle::create(THD *thd,
+                                         const LEX_CSTRING &name,
+                                         List<Item> *item_list)
+{
+  if (create_check_args_ge(name, item_list, 1))
+    return NULL;
+  return new (thd->mem_root) Item_func_concat_operator_oracle(thd, *item_list);
+}
+
+
 /*
   Realloc the result buffer.
   NOTE: We should be prudent in the initial allocation unit -- the
@@ -2161,11 +2182,10 @@ void Item_func_trim::print(String *str, enum_query_type query_type)
 {
   if (arg_count == 1)
   {
-    Item_func::print(str, query_type);
+    print_sql_mode_dependent(str, query_type);
     return;
   }
-  str->append(Item_func_trim::func_name());
-  str->append(func_name_ext());
+  print_sql_mode_dependent_name(str, query_type, Item_func_trim::func_name());
   str->append('(');
   str->append(mode_name());
   str->append(' ');
@@ -2379,6 +2399,20 @@ void Item_func_decode::crypto_transform(String *res)
 }
 
 
+Item_func_decode *Item_func_decode::create(THD *thd, const LEX_CSTRING &name,
+                                           List<Item> *item_list)
+{
+  if (unlikely(!item_list || item_list->elements != 2))
+  {
+    wrong_param_count_error(mariadb_schema.name(), name);
+    return NULL;
+  }
+  Item_args args(thd, *item_list);
+  return new (thd->mem_root) Item_func_decode(thd, args.arguments()[0],
+                                                   args.arguments()[1]);
+}
+
+
 String *Item_func_database::val_str(String *str)
 {
   DBUG_ASSERT(fixed == 1);
@@ -3204,6 +3238,47 @@ static String *default_pad_str(String *pad_str, CHARSET_INFO *collation)
   return pad_str;
 }
 
+
+Item_func_lpad*
+Item_func_lpad::create(THD *thd, const LEX_CSTRING &name,
+                       List<Item> *item_list)
+{
+  if (create_check_args_between(name, item_list, 2, 3))
+    return NULL;
+  return new (thd->mem_root) Item_func_lpad(thd, *item_list);
+}
+
+
+Item_func_lpad_oracle*
+Item_func_lpad_oracle::create(THD *thd, const LEX_CSTRING &name,
+                              List<Item> *item_list)
+{
+  if (create_check_args_between(name, item_list, 2, 3))
+    return NULL;
+  return new (thd->mem_root) Item_func_lpad_oracle(thd, *item_list);
+}
+
+
+Item_func_rpad*
+Item_func_rpad::create(THD *thd, const LEX_CSTRING &name,
+                       List<Item> *item_list)
+{
+  if (create_check_args_between(name, item_list, 2, 3))
+    return NULL;
+  return new (thd->mem_root) Item_func_rpad(thd, *item_list);
+}
+
+
+Item_func_rpad_oracle*
+Item_func_rpad_oracle::create(THD *thd, const LEX_CSTRING &name,
+                              List<Item> *item_list)
+{
+  if (create_check_args_between(name, item_list, 2, 3))
+    return NULL;
+  return new (thd->mem_root) Item_func_rpad_oracle(thd, *item_list);
+}
+
+
 bool Item_func_pad::fix_length_and_dec()
 {
   if (arg_count == 3)
diff --git a/sql/item_strfunc.h b/sql/item_strfunc.h
index cd50da3dd3a..3ecf9b8c487 100644
--- a/sql/item_strfunc.h
+++ b/sql/item_strfunc.h
@@ -259,8 +259,15 @@ protected:
 public:
   Item_func_concat(THD *thd, List<Item> &list): Item_str_func(thd, list) {}
   Item_func_concat(THD *thd, Item *a, Item *b): Item_str_func(thd, a, b) {}
+  static Item_func_concat* create(THD *thd, const LEX_CSTRING &name,
+                                  List<Item> *list);
   String *val_str(String *);
   bool fix_length_and_dec();
+  const Schema *schema() const { return &mariadb_schema; }
+  void print(String *str, enum_query_type query_type)
+  {
+    print_sql_mode_dependent(str, query_type);
+  }
   const char *func_name() const { return "concat"; }
   Item *get_copy(THD *thd)
   { return get_item_copy<Item_func_concat>(thd, this); }
@@ -280,8 +287,20 @@ public:
   Item_func_concat_operator_oracle(THD *thd, Item *a, Item *b)
    :Item_func_concat(thd, a, b)
   { }
+  static Item_func_concat_operator_oracle* create(THD *thd,
+                                                  const LEX_CSTRING &name,
+                                                  List<Item> *list);
   String *val_str(String *);
-  const char *func_name() const { return "concat_operator_oracle"; }
+  const Schema *schema() const { return &oracle_schema_ref; }
+  void print(String *str, enum_query_type query_type)
+  {
+    static const LEX_CSTRING suffix= {STRING_WITH_LEN("operator_oracle")};
+    print_sql_mode_dependent_name(str, query_type,
+                                  oracle_schema_ref,
+                                  Item_func_concat::func_name(),
+                                  suffix);
+    print_args_parenthesized(str, query_type);
+  }
   Item *get_copy(THD *thd)
   {
     return get_item_copy<Item_func_concat_operator_oracle>(thd, this);
@@ -342,6 +361,11 @@ public:
   String *val_str(String *to) { return val_str_internal(to, NULL); };
   bool fix_length_and_dec();
   String *val_str_internal(String *str, String *empty_string_for_null);
+  const Schema *schema() const { return &mariadb_schema; }
+  void print(String *str, enum_query_type query_type)
+  {
+    print_sql_mode_dependent(str, query_type);
+  }
   const char *func_name() const { return "replace"; }
   Item *get_copy(THD *thd)
   { return get_item_copy<Item_func_replace>(thd, this); }
@@ -355,7 +379,7 @@ public:
   Item_func_replace_oracle(THD *thd, Item *org, Item *find, Item *replace):
     Item_func_replace(thd, org, find, replace) {}
   String *val_str(String *to) { return val_str_internal(to, &tmp_emtpystr); };
-  const char *func_name() const { return "replace_oracle"; }
+  const Schema *schema() const { return &oracle_schema_ref; }
   Item *get_copy(THD *thd)
   { return get_item_copy<Item_func_replace_oracle>(thd, this); }
 };
@@ -493,6 +517,11 @@ public:
     Item_str_func(thd, a, b, c) {}
   String *val_str(String *);
   bool fix_length_and_dec();
+  const Schema *schema() const { return &mariadb_schema; }
+  void print(String *str, enum_query_type query_type)
+  {
+    print_sql_mode_dependent(str, query_type);
+  }
   const char *func_name() const { return "substr"; }
   Item *get_copy(THD *thd)
   { return get_item_copy<Item_func_substr>(thd, this); }
@@ -516,7 +545,7 @@ public:
     maybe_null= true;
     return res;
   }
-  const char *func_name() const { return "substr_oracle"; }
+  const Schema *schema() const { return &oracle_schema_ref; }
   Item *get_copy(THD *thd)
   { return get_item_copy<Item_func_substr_oracle>(thd, this); }
 };
@@ -559,13 +588,13 @@ protected:
   {
     return trimmed_value(res, 0, res->length());
   }
-  virtual const char *func_name_ext() const { return ""; }
 public:
   Item_func_trim(THD *thd, Item *a, Item *b): Item_str_func(thd, a, b) {}
   Item_func_trim(THD *thd, Item *a): Item_str_func(thd, a) {}
   Sql_mode_dependency value_depends_on_sql_mode() const;
   String *val_str(String *);
   bool fix_length_and_dec();
+  const Schema *schema() const { return &mariadb_schema; }
   const char *func_name() const { return "trim"; }
   void print(String *str, enum_query_type query_type);
   virtual const char *mode_name() const { return "both"; }
@@ -579,12 +608,11 @@ class Item_func_trim_oracle :public Item_func_trim
 protected:
   String *make_empty_result()
   { null_value= 1; return NULL; }
-  const char *func_name_ext() const { return "_oracle"; }
 public:
   Item_func_trim_oracle(THD *thd, Item *a, Item *b):
     Item_func_trim(thd, a, b) {}
   Item_func_trim_oracle(THD *thd, Item *a): Item_func_trim(thd, a) {}
-  const char *func_name() const { return "trim_oracle"; }
+  const Schema *schema() const { return &oracle_schema_ref; }
   bool fix_length_and_dec()
   {
     bool res= Item_func_trim::fix_length_and_dec();
@@ -606,6 +634,7 @@ public:
     return Item_func::value_depends_on_sql_mode();
   }
   String *val_str(String *);
+  const Schema *schema() const { return &mariadb_schema; }
   const char *func_name() const { return "ltrim"; }
   const char *mode_name() const { return "leading"; }
   Item *get_copy(THD *thd)
@@ -618,12 +647,11 @@ class Item_func_ltrim_oracle :public Item_func_ltrim
 protected:
   String *make_empty_result()
   { null_value= 1; return NULL; }
-  const char *func_name_ext() const { return "_oracle"; }
 public:
   Item_func_ltrim_oracle(THD *thd, Item *a, Item *b):
     Item_func_ltrim(thd, a, b) {}
   Item_func_ltrim_oracle(THD *thd, Item *a): Item_func_ltrim(thd, a) {}
-  const char *func_name() const { return "ltrim_oracle"; }
+  const Schema *schema() const { return &oracle_schema_ref; }
   bool fix_length_and_dec()
   {
     bool res= Item_func_ltrim::fix_length_and_dec();
@@ -641,6 +669,7 @@ public:
   Item_func_rtrim(THD *thd, Item *a, Item *b): Item_func_trim(thd, a, b) {}
   Item_func_rtrim(THD *thd, Item *a): Item_func_trim(thd, a) {}
   String *val_str(String *);
+  const Schema *schema() const { return &mariadb_schema; }
   const char *func_name() const { return "rtrim"; }
   const char *mode_name() const { return "trailing"; }
   Item *get_copy(THD *thd)
@@ -653,12 +682,11 @@ class Item_func_rtrim_oracle :public Item_func_rtrim
 protected:
   String *make_empty_result()
   { null_value= 1; return NULL; }
-  const char *func_name_ext() const { return "_oracle"; }
 public:
   Item_func_rtrim_oracle(THD *thd, Item *a, Item *b):
     Item_func_rtrim(thd, a, b) {}
   Item_func_rtrim_oracle(THD *thd, Item *a): Item_func_rtrim(thd, a) {}
-  const char *func_name() const { return "rtrim_oracle"; }
+  const Schema *schema() const { return &oracle_schema_ref; }
   bool fix_length_and_dec()
   {
     bool res= Item_func_rtrim::fix_length_and_dec();
@@ -820,6 +848,13 @@ class Item_func_decode :public Item_func_encode
 {
 public:
   Item_func_decode(THD *thd, Item *a, Item *seed_arg): Item_func_encode(thd, a, seed_arg) {}
+  static Item_func_decode *create(THD *thd, const LEX_CSTRING &name,
+                                  List<Item> *item_list);
+  const Schema *schema() const { return &mariadb_schema; }
+  void print(String *str, enum_query_type query_type)
+  {
+    print_sql_mode_dependent(str, query_type);
+  }
   const char *func_name() const { return "decode"; }
   Item *get_copy(THD *thd)
   { return get_item_copy<Item_func_decode>(thd, this); }
@@ -1125,6 +1160,8 @@ public:
     Item_str_func(thd, arg1, arg2, arg3) {}
   Item_func_pad(THD *thd, Item *arg1, Item *arg2):
     Item_str_func(thd, arg1, arg2) {}
+  Item_func_pad(THD *thd, List<Item> &list):
+    Item_str_func(thd,list) {}
   bool fix_length_and_dec();
 };
 
@@ -1136,7 +1173,16 @@ public:
     Item_func_pad(thd, arg1, arg2, arg3) {}
   Item_func_rpad(THD *thd, Item *arg1, Item *arg2):
     Item_func_pad(thd, arg1, arg2) {}
+  Item_func_rpad(THD *thd, List<Item> &list):
+    Item_func_pad(thd,list) {}
+  static Item_func_rpad *create(THD *thd, const LEX_CSTRING &name,
+                                List<Item> *item_list);
   String *val_str(String *);
+  const Schema *schema() const { return &mariadb_schema; }
+  void print(String *str, enum_query_type query_type)
+  {
+    print_sql_mode_dependent(str, query_type);
+  }
   const char *func_name() const { return "rpad"; }
   Sql_mode_dependency value_depends_on_sql_mode() const;
   Item *get_copy(THD *thd)
@@ -1153,13 +1199,17 @@ public:
     Item_func_rpad(thd, arg1, arg2, arg3) {}
   Item_func_rpad_oracle(THD *thd, Item *arg1, Item *arg2):
     Item_func_rpad(thd, arg1, arg2) {}
+  Item_func_rpad_oracle(THD *thd, List<Item> &list):
+    Item_func_rpad(thd,list) {}
+  static Item_func_rpad_oracle *create(THD *thd, const LEX_CSTRING &name,
+                                       List<Item> *item_list);
   bool fix_length_and_dec()
   {
     bool res= Item_func_rpad::fix_length_and_dec();
     maybe_null= true;
     return res;
   }
-  const char *func_name() const { return "rpad_oracle"; }
+  const Schema *schema() const { return &oracle_schema_ref; }
   Item *get_copy(THD *thd)
   { return get_item_copy<Item_func_rpad_oracle>(thd, this); }
 };
@@ -1172,7 +1222,16 @@ public:
     Item_func_pad(thd, arg1, arg2, arg3) {}
   Item_func_lpad(THD *thd, Item *arg1, Item *arg2):
     Item_func_pad(thd, arg1, arg2) {}
+  Item_func_lpad(THD *thd, List<Item> &list):
+    Item_func_pad(thd,list) {}
+  static Item_func_lpad *create(THD *thd, const LEX_CSTRING &name,
+                                List<Item> *item_list);
   String *val_str(String *);
+  const Schema *schema() const { return &mariadb_schema; }
+  void print(String *str, enum_query_type query_type)
+  {
+    print_sql_mode_dependent(str, query_type);
+  }
   const char *func_name() const { return "lpad"; }
   Item *get_copy(THD *thd)
   { return get_item_copy<Item_func_lpad>(thd, this); }
@@ -1188,13 +1247,17 @@ public:
     Item_func_lpad(thd, arg1, arg2, arg3) {}
   Item_func_lpad_oracle(THD *thd, Item *arg1, Item *arg2):
     Item_func_lpad(thd, arg1, arg2) {}
+  Item_func_lpad_oracle(THD *thd, List<Item> &list):
+    Item_func_lpad(thd,list) {}
+  static Item_func_lpad_oracle *create(THD *thd, const LEX_CSTRING &name,
+                                       List<Item> *item_list);
   bool fix_length_and_dec()
   {
     bool res= Item_func_lpad::fix_length_and_dec();
     maybe_null= true;
     return res;
   }
-  const char *func_name() const { return "lpad_oracle"; }
+  const Schema *schema() const { return &oracle_schema_ref; }
   Item *get_copy(THD *thd)
   { return get_item_copy<Item_func_lpad_oracle>(thd, this); }
 };
diff --git a/sql/lex.h b/sql/lex.h
index 823e95bb863..1f8089c2359 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -739,7 +739,6 @@ SYMBOL sql_functions[] = {
   { "DATE_ADD",		SYM(DATE_ADD_INTERVAL)},
   { "DATE_SUB",		SYM(DATE_SUB_INTERVAL)},
   { "DATE_FORMAT",      SYM(DATE_FORMAT_SYM)},
-  { "DECODE",           SYM(DECODE_MARIADB_SYM)},
   { "DENSE_RANK",       SYM(DENSE_RANK_SYM)},
   { "EXTRACT",		SYM(EXTRACT_SYM)},
   { "FIRST_VALUE",      SYM(FIRST_VALUE_SYM)},
diff --git a/sql/mysqld.h b/sql/mysqld.h
index f7d0fce910f..aa1bf855380 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -740,12 +740,25 @@ enum enum_query_type
                         QT_ITEM_SUBSELECT_ID_ONLY,
 
   QT_SHOW_SELECT_NUMBER= (1<<10),
+
+  //QT_ITEM_IDENT_DISABLE_DB_TABLE_NAMES= (1 <<11), -- this is taken in 10.5
+
+  /// Print sql_mode-dependent functions with the schema qualifier
+  /// even if the currently implied (by sql_mode) schema is equal to
+  /// to the function schema, e.g. mariadb_schema.concat('a').
+  QT_ITEM_FUNC_FORCE_SCHEMA_NAME= (1 << 12),
+
+  /// Print for FRM file. Focus on parse-back.
+  /// e.g. VIEW expressions and virtual column expressions
+  QT_FOR_FRM= (1 << 13),
+
   /// This is used for EXPLAIN EXTENDED extra warnings / Be more detailed
   /// Be more detailed than QT_EXPLAIN.
   /// Perhaps we should eventually include QT_ITEM_IDENT_SKIP_CURRENT_DATABASE
   /// here, as it would give better readable results
   QT_EXPLAIN_EXTENDED=  QT_TO_SYSTEM_CHARSET|
-                        QT_SHOW_SELECT_NUMBER,
+                        QT_SHOW_SELECT_NUMBER|
+                        QT_ITEM_FUNC_FORCE_SCHEMA_NAME,
 
   // If an expression is constant, print the expression, not the value
   // it evaluates to. Should be used for error messages, so that they
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 371f17e4faa..a8af993dfc5 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -6649,6 +6649,50 @@ class Sql_mode_save
   sql_mode_t old_mode; // SQL mode saved at construction time.
 };
 
+
+/*
+  Save the current sql_mode. Switch off sql_mode flags which can prevent
+  normal parsing of VIEWs, expressions in generated columns.
+  Restore the old sql_mode on destructor.
+*/
+class Sql_mode_save_for_frm_handling: public Sql_mode_save
+{
+public:
+  Sql_mode_save_for_frm_handling(THD *thd)
+   :Sql_mode_save(thd)
+  {
+    /*
+      - MODE_REAL_AS_FLOAT            affect only CREATE TABLE parsing
+      + MODE_PIPES_AS_CONCAT          affect expression parsing
+      + MODE_ANSI_QUOTES              affect expression parsing
+      + MODE_IGNORE_SPACE             affect expression parsing
+      - MODE_IGNORE_BAD_TABLE_OPTIONS affect only CREATE/ALTER TABLE parsing
+      * MODE_ONLY_FULL_GROUP_BY       affect execution
+      * MODE_NO_UNSIGNED_SUBTRACTION  affect execution
+      - MODE_NO_DIR_IN_CREATE         affect table creation only
+      - MODE_POSTGRESQL               compounded from other modes
+      - MODE_ORACLE                   affects Item creation (e.g for CONCAT)
+      - MODE_MSSQL                    compounded from other modes
+      - MODE_DB2                      compounded from other modes
+      - MODE_MAXDB                    affect only CREATE TABLE parsing
+      - MODE_NO_KEY_OPTIONS           affect only SHOW
+      - MODE_NO_TABLE_OPTIONS         affect only SHOW
+      - MODE_NO_FIELD_OPTIONS         affect only SHOW
+      - MODE_MYSQL323                 affect only SHOW
+      - MODE_MYSQL40                  affect only SHOW
+      - MODE_ANSI                     compounded from other modes
+                                      (+ transaction mode)
+      ? MODE_NO_AUTO_VALUE_ON_ZERO    affect UPDATEs
+      + MODE_NO_BACKSLASH_ESCAPES     affect expression parsing
+      + MODE_EMPTY_STRING_IS_NULL     affect expression parsing
+    */
+    thd->variables.sql_mode&= ~(MODE_PIPES_AS_CONCAT | MODE_ANSI_QUOTES |
+                                MODE_IGNORE_SPACE | MODE_NO_BACKSLASH_ESCAPES |
+                                MODE_ORACLE | MODE_EMPTY_STRING_IS_NULL);
+  };
+};
+
+
 class Abort_on_warning_instant_set
 {
   THD *m_thd;
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index a9adfebff81..b2a406e7573 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -824,7 +824,7 @@ Yacc_state::~Yacc_state()
 }
 
 int Lex_input_stream::find_keyword(Lex_ident_cli_st *kwd,
-                                   uint len, bool function)
+                                   uint len, bool function) const
 {
   const char *tok= m_tok_start;
 
@@ -844,7 +844,6 @@ int Lex_input_stream::find_keyword(Lex_ident_cli_st *kwd,
       case CLOB_MARIADB_SYM:           return CLOB_ORACLE_SYM;
       case CONTINUE_MARIADB_SYM:       return CONTINUE_ORACLE_SYM;
       case DECLARE_MARIADB_SYM:        return DECLARE_ORACLE_SYM;
-      case DECODE_MARIADB_SYM:         return DECODE_ORACLE_SYM;
       case ELSEIF_MARIADB_SYM:         return ELSEIF_ORACLE_SYM;
       case ELSIF_MARIADB_SYM:          return ELSIF_ORACLE_SYM;
       case EXCEPTION_MARIADB_SYM:      return EXCEPTION_ORACLE_SYM;
@@ -914,7 +913,7 @@ bool is_lex_native_function(const LEX_CSTRING *name)
 
 bool is_native_function(THD *thd, const LEX_CSTRING *name)
 {
-  if (find_native_function_builder(thd, name))
+  if (mariadb_schema.find_native_function_builder(thd, *name))
     return true;
 
   if (is_lex_native_function(name))
@@ -1539,7 +1538,18 @@ int Lex_input_stream::lex_one_token(YYSTYPE *yylval, THD *thd)
       if (lex->parsing_options.lookup_keywords_after_qualifier)
         next_state= MY_LEX_IDENT_OR_KEYWORD;
       else
-        next_state= MY_LEX_IDENT_START;    // Next is ident (not keyword)
+      {
+        /*
+          Next is:
+          - A qualified func with a special syntax:
+            mariadb_schema.REPLACE('a','b','c')
+            mariadb_schema.SUSTRING('a',1,2)
+            mariadb_schema.TRIM('a')
+          - Or an identifier otherwise. No keyword lookup is done,
+            all keywords are treated as identifiers.
+        */
+        next_state= MY_LEX_IDENT_OR_QUALIFIED_SPECIAL_FUNC;
+      }
       if (!ident_map[(uchar) yyPeek()])    // Probably ` or "
         next_state= MY_LEX_START;
       return((int) c);
@@ -1983,7 +1993,12 @@ int Lex_input_stream::lex_one_token(YYSTYPE *yylval, THD *thd)
         We should now be able to handle:
         [(global | local | session) .]variable_name
       */
-      return scan_ident_sysvar(thd, &yylval->ident_cli);
+      return scan_ident_common(thd, &yylval->ident_cli,
+                               GENERAL_KEYWORD_OR_FUNC_LPAREN);
+
+    case MY_LEX_IDENT_OR_QUALIFIED_SPECIAL_FUNC:
+      return scan_ident_common(thd, &yylval->ident_cli,
+                               QUALIFIED_SPECIAL_FUNC_LPAREN);
     }
   }
 }
@@ -2005,7 +2020,59 @@ bool Lex_input_stream::get_7bit_or_8bit_ident(THD *thd, uchar *last_char)
 }
 
 
-int Lex_input_stream::scan_ident_sysvar(THD *thd, Lex_ident_cli_st *str)
+/*
+  Resolve special SQL functions that have a qualified syntax in sql_yacc.yy.
+  These functions are not listed in the native function registry
+  because of a special syntax, or a reserved keyword:
+
+    mariadb_schema.SUBSTRING('a' FROM 1 FOR 2)   -- Special syntax
+    mariadb_schema.TRIM(BOTH ' ' FROM 'a')       -- Special syntax
+    mariadb_schema.REPLACE('a','b','c')          -- Verb keyword
+*/
+
+int Lex_input_stream::find_keyword_qualified_special_func(Lex_ident_cli_st *str,
+                                                          uint length) const
+{
+  /*
+    There are many other special functions, see the following grammar rules:
+      function_call_keyword
+      function_call_nonkeyword
+    Here we resolve only those that have a qualified syntax to handle
+    different behavior in different @@sql_mode settings.
+
+    Other special functions do not work in qualified context:
+      SELECT mariadb_schema.year(now()); -- Function year is not defined
+      SELECT mariadb_schema.now();       -- Function now is not defined
+
+    We don't resolve TRIM_ORACLE here, because it does not have
+    a qualified syntax yet. Search for "trim_operands" in sql_yacc.yy
+    to find more comments.
+  */
+  static LEX_CSTRING funcs[]=
+  {
+    {STRING_WITH_LEN("SUBSTRING")},
+    {STRING_WITH_LEN("SUBSTR")},
+    {STRING_WITH_LEN("TRIM")},
+    {STRING_WITH_LEN("REPLACE")}
+  };
+
+  int tokval= find_keyword(str, length, true);
+  if (!tokval)
+    return 0;
+  for (size_t i= 0; i < array_elements(funcs); i++)
+  {
+    CHARSET_INFO *cs= system_charset_info;
+    if (!cs->coll->strnncollsp(cs,
+                               (const uchar *) m_tok_start, length,
+                               (const uchar *) funcs[i].str, funcs[i].length))
+      return tokval;
+  }
+  return 0;
+}
+
+
+int Lex_input_stream::scan_ident_common(THD *thd, Lex_ident_cli_st *str,
+                                        Ident_mode mode)
 {
   uchar last_char;
   uint length;
@@ -2019,10 +2086,41 @@ int Lex_input_stream::scan_ident_sysvar(THD *thd, Lex_ident_cli_st *str)
     next_state= MY_LEX_IDENT_SEP;
   if (!(length= yyLength()))
     return ABORT_SYM;                  // Names must be nonempty.
-  if ((tokval= find_keyword(str, length, 0)))
-  {
-    yyUnget();                         // Put back 'c'
-    return tokval;                     // Was keyword
+
+  switch (mode) {
+  case GENERAL_KEYWORD_OR_FUNC_LPAREN:
+    /*
+      We can come here inside a system variable after "@@",
+      e.g. @@global.character_set_client.
+      We resolve all general purpose keywords here.
+
+      We can come here when LEX::parsing_options.lookup_keywords_after_qualifier
+      is true, i.e. within the "field_spec" Bison rule.
+      We need to resolve functions that have special rules inside sql_yacc.yy,
+      such as SUBSTR, REPLACE, TRIM, to make this work:
+        c2 varchar(4) GENERATED ALWAYS AS (mariadb_schema.substr(c1,1,4))
+    */
+    if ((tokval= find_keyword(str, length, last_char == '(')))
+    {
+      yyUnget();                         // Put back 'c'
+      return tokval;                     // Was keyword
+    }
+    break;
+  case QUALIFIED_SPECIAL_FUNC_LPAREN:
+    /*
+      We come here after '.' in various contexts:
+        SELECT @@global.character_set_client;
+        SELECT t1.a FROM t1;
+        SELECT test.f1() FROM t1;
+        SELECT mariadb_schema.trim('a');
+    */
+    if (last_char == '(' &&
+        (tokval= find_keyword_qualified_special_func(str, length)))
+    {
+      yyUnget();                         // Put back 'c'
+      return tokval;                     // Was keyword
+    }
+    break;
   }
 
   yyUnget();                       // ptr points now after last token char
@@ -2072,9 +2170,9 @@ int Lex_input_stream::scan_ident_start(THD *thd, Lex_ident_cli_st *str)
   {
     is_8bit= get_7bit_or_8bit_ident(thd, &c);
   }
+
   if (c == '.' && ident_map[(uchar) yyPeek()])
     next_state= MY_LEX_IDENT_SEP;// Next is '.'
-
   uint length= yyLength();
   yyUnget(); // ptr points now after last token char
   str->set_ident(m_tok_start, length, is_8bit);
@@ -8024,30 +8122,49 @@ bool LEX::add_grant_command(THD *thd, enum_sql_command sql_command_arg,
 }
 
 
-Item *LEX::make_item_func_substr(THD *thd, Item *a, Item *b, Item *c)
+const Schema *
+LEX::find_func_schema_by_name_or_error(const Lex_ident_sys &schema,
+                                       const Lex_ident_sys &func)
 {
-  return (thd->variables.sql_mode & MODE_ORACLE) ?
-    new (thd->mem_root) Item_func_substr_oracle(thd, a, b, c) :
-    new (thd->mem_root) Item_func_substr(thd, a, b, c);
+  Schema *res= Schema::find_by_name(schema);
+  if (res)
+    return res;
+  Database_qualified_name qname(schema, func);
+  my_error(ER_FUNCTION_NOT_DEFINED, MYF(0), ErrConvDQName(&qname).ptr());
+  return NULL;
 }
 
 
-Item *LEX::make_item_func_substr(THD *thd, Item *a, Item *b)
+Item *LEX::make_item_func_substr(THD *thd,
+                                 const Lex_ident_cli_st &schema_name_cli,
+                                 const Lex_ident_cli_st &func_name_cli,
+                                 const Lex_substring_spec_st &spec)
 {
-  return (thd->variables.sql_mode & MODE_ORACLE) ?
-    new (thd->mem_root) Item_func_substr_oracle(thd, a, b) :
-    new (thd->mem_root) Item_func_substr(thd, a, b);
+  Lex_ident_sys schema_name(thd, &schema_name_cli);
+  Lex_ident_sys func_name(thd, &func_name_cli);
+  if (schema_name.is_null() || func_name.is_null())
+    return NULL; // EOM
+  const Schema *schema= find_func_schema_by_name_or_error(schema_name,
+                                                          func_name);
+  return schema ? schema->make_item_func_substr(thd, spec) : NULL;
 }
 
 
 Item *LEX::make_item_func_replace(THD *thd,
+                                  const Lex_ident_cli_st &schema_name_cli,
+                                  const Lex_ident_cli_st &func_name_cli,
                                   Item *org,
                                   Item *find,
                                   Item *replace)
 {
-  return (thd->variables.sql_mode & MODE_ORACLE) ?
-    new (thd->mem_root) Item_func_replace_oracle(thd, org, find, replace) :
-    new (thd->mem_root) Item_func_replace(thd, org, find, replace);
+  Lex_ident_sys schema_name(thd, &schema_name_cli);
+  Lex_ident_sys func_name(thd, &func_name_cli);
+  if (schema_name.is_null() || func_name.is_null())
+    return NULL; // EOM
+  const Schema *schema= find_func_schema_by_name_or_error(schema_name,
+                                                          func_name);
+  return schema ? schema->make_item_func_replace(thd, org, find, replace) :
+                  NULL;
 }
 
 
@@ -8129,11 +8246,18 @@ Item *Lex_trim_st::make_item_func_trim_oracle(THD *thd) const
 }
 
 
-Item *Lex_trim_st::make_item_func_trim(THD *thd) const
+Item *Lex_trim_st::make_item_func_trim(THD *thd,
+                                       const Lex_ident_cli_st &schema_name_cli,
+                                       const Lex_ident_cli_st &func_name_cli)
+                                       const
 {
-  return (thd->variables.sql_mode & MODE_ORACLE) ?
-         make_item_func_trim_oracle(thd) :
-         make_item_func_trim_std(thd);
+  Lex_ident_sys schema_name(thd, &schema_name_cli);
+  Lex_ident_sys func_name(thd, &func_name_cli);
+  if (schema_name.is_null() || func_name.is_null())
+    return NULL; // EOM
+  const Schema *schema= LEX::find_func_schema_by_name_or_error(schema_name,
+                                                               func_name);
+  return schema ? schema->make_item_func_trim(thd, *this) : NULL;
 }
 
 
@@ -8165,6 +8289,19 @@ Item *LEX::make_item_func_call_generic(THD *thd, Lex_ident_cli_st *cdb,
   if (check_routine_name(&name))
     return NULL;
 
+  return make_item_func_call_generic(thd, db, name, args);
+}
+
+
+Item *LEX::make_item_func_call_generic(THD *thd,
+                                       const Lex_ident_sys &db,
+                                       const Lex_ident_sys &name,
+                                       List<Item> *args)
+{
+  const Schema *schema= Schema::find_by_name(db);
+  if (schema)
+    return schema->make_item_func_call_native(thd, name, args);
+
   Create_qfunc *builder= find_qualified_function_builder(thd);
   DBUG_ASSERT(builder);
   return builder->create_with_db(thd, &db, &name, true, args);
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 064d0de8905..7729bd82b84 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -2230,6 +2230,15 @@ public:
   void reduce_digest_token(uint token_left, uint token_right);
 
 private:
+
+  enum Ident_mode
+  {
+    GENERAL_KEYWORD_OR_FUNC_LPAREN,
+    QUALIFIED_SPECIAL_FUNC_LPAREN
+  };
+
+  int scan_ident_common(THD *thd, Lex_ident_cli_st *str, Ident_mode mode);
+
   /**
     Set the echo mode.
 
@@ -2427,7 +2436,7 @@ public:
   }
 
   /** Get the raw query buffer. */
-  const char *get_buf()
+  const char *get_buf() const
   {
     return m_buf;
   }
@@ -2439,7 +2448,7 @@ public:
   }
 
   /** Get the end of the raw query buffer. */
-  const char *get_end_of_query()
+  const char *get_end_of_query() const
   {
     return m_end_of_query;
   }
@@ -2483,7 +2492,7 @@ public:
     Get the token end position in the pre-processed buffer,
     with trailing spaces removed.
   */
-  const char *get_cpp_tok_end_rtrim()
+  const char *get_cpp_tok_end_rtrim() const
   {
     const char *p;
     for (p= m_cpp_tok_end;
@@ -2555,9 +2564,9 @@ private:
 
   bool consume_comment(int remaining_recursions_permitted);
   int lex_one_token(union YYSTYPE *yylval, THD *thd);
-  int find_keyword(Lex_ident_cli_st *str, uint len, bool function);
+  int find_keyword(Lex_ident_cli_st *str, uint len, bool function) const;
+  int find_keyword_qualified_special_func(Lex_ident_cli_st *str, uint len) const;
   LEX_CSTRING get_token(uint skip, uint length);
-  int scan_ident_sysvar(THD *thd, Lex_ident_cli_st *str);
   int scan_ident_start(THD *thd, Lex_ident_cli_st *str);
   int scan_ident_middle(THD *thd, Lex_ident_cli_st *str,
                         CHARSET_INFO **cs, my_lex_states *);
@@ -3684,12 +3693,24 @@ public:
                           const Lex_ident_cli_st *var_name,
                           const Lex_ident_cli_st *field_name);
 
-  Item *make_item_func_replace(THD *thd, Item *org, Item *find, Item *replace);
-  Item *make_item_func_substr(THD *thd, Item *a, Item *b, Item *c);
-  Item *make_item_func_substr(THD *thd, Item *a, Item *b);
+  static const Schema *
+    find_func_schema_by_name_or_error(const Lex_ident_sys &schema_name,
+                                      const Lex_ident_sys &func_name);
+  Item *make_item_func_replace(THD *thd,
+                               const Lex_ident_cli_st &schema_name,
+                               const Lex_ident_cli_st &func_name,
+                               Item *org, Item *find, Item *replace);
+  Item *make_item_func_substr(THD *thd,
+                              const Lex_ident_cli_st &schema_name,
+                              const Lex_ident_cli_st &func_name,
+                              const Lex_substring_spec_st &spec);
   Item *make_item_func_call_generic(THD *thd, Lex_ident_cli_st *db,
                                     Lex_ident_cli_st *name, List<Item> *args);
   Item *make_item_func_call_generic(THD *thd,
+                                    const Lex_ident_sys &db,
+                                    const Lex_ident_sys &name,
+                                    List<Item> *args);
+  Item *make_item_func_call_generic(THD *thd,
                                     Lex_ident_cli_st *db,
                                     Lex_ident_cli_st *pkg,
                                     Lex_ident_cli_st *name,
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index 4a0a55f8178..c7ea6600030 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -2591,11 +2591,9 @@ char *generate_partition_syntax_for_frm(THD *thd, partition_info *part_info,
                                         HA_CREATE_INFO *create_info,
                                         Alter_info *alter_info)
 {
-  sql_mode_t old_mode= thd->variables.sql_mode;
-  thd->variables.sql_mode &= ~MODE_ANSI_QUOTES;
+  Sql_mode_save_for_frm_handling sql_mode_save(thd);
   char *res= generate_partition_syntax(thd, part_info, buf_length,
                                              true, create_info, alter_info);
-  thd->variables.sql_mode= old_mode;
   return res;
 }
 
diff --git a/sql/sql_schema.cc b/sql/sql_schema.cc
index 0bf4a63c2f8..e2993f51c6d 100644
--- a/sql/sql_schema.cc
+++ b/sql/sql_schema.cc
@@ -32,6 +32,16 @@ public:
       return thd->type_handler_for_datetime();
     return src;
   }
+
+  Create_func *find_native_function_builder(THD*, const LEX_CSTRING&) const;
+
+  Item *make_item_func_substr(THD *thd,
+                              const Lex_substring_spec_st &spec) const;
+  Item *make_item_func_trim(THD *thd, const Lex_trim_st &spec) const;
+  Item *make_item_func_replace(THD *thd,
+                               Item *subj,
+                               Item *find,
+                               Item *replace) const;
 };
 
 
@@ -56,6 +66,7 @@ Schema        mariadb_schema(Lex_cstring(STRING_WITH_LEN("mariadb_schema")));
 Schema_oracle oracle_schema(Lex_cstring(STRING_WITH_LEN("oracle_schema")));
 Schema_maxdb  maxdb_schema(Lex_cstring(STRING_WITH_LEN("maxdb_schema")));
 
+const Schema &oracle_schema_ref= oracle_schema;
 
 Schema *Schema::find_by_name(const LEX_CSTRING &name)
 {
@@ -78,3 +89,86 @@ Schema *Schema::find_implied(THD *thd)
     return &maxdb_schema;
   return &mariadb_schema;
 }
+
+
+Create_func *
+Schema::find_native_function_builder(THD *thd, const LEX_CSTRING &name) const
+{
+  return native_functions_hash.find(thd, name);
+}
+
+
+Create_func *
+Schema_oracle::find_native_function_builder(THD *thd,
+                                            const LEX_CSTRING &name) const
+{
+  Create_func *create= native_functions_hash_oracle_overrides.find(thd, name);
+  if (create)
+    return create;
+  return native_functions_hash.find(thd, name);
+}
+
+
+Item *Schema::make_item_func_call_native(THD *thd,
+                                         const Lex_ident_sys &name,
+                                         List<Item> *args) const
+{
+  Create_func *builder= find_native_function_builder(thd, name);
+  if (builder)
+    return builder->create_func(thd, &name, args);
+  my_error(ER_FUNCTION_NOT_DEFINED, MYF(0), name.str);
+  return NULL;
+}
+
+
+Item *Schema::make_item_func_trim(THD *thd, const Lex_trim_st &spec) const
+{
+  return spec.make_item_func_trim_std(thd);
+}
+
+
+Item *Schema::make_item_func_substr(THD *thd,
+                                    const Lex_substring_spec_st &spec) const
+{
+  return spec.m_for ?
+    new (thd->mem_root) Item_func_substr(thd, spec.m_subject, spec.m_from,
+                                              spec.m_for) :
+    new (thd->mem_root) Item_func_substr(thd, spec.m_subject, spec.m_from);
+}
+
+
+Item *Schema_oracle::make_item_func_substr(THD *thd,
+                                    const Lex_substring_spec_st &spec) const
+{
+  return spec.m_for ?
+    new (thd->mem_root) Item_func_substr_oracle(thd, spec.m_subject,
+                                                     spec.m_from,
+                                                     spec.m_for) :
+    new (thd->mem_root) Item_func_substr_oracle(thd, spec.m_subject,
+                                                     spec.m_from);
+}
+
+
+Item *Schema_oracle::make_item_func_trim(THD *thd,
+                                         const Lex_trim_st &spec) const
+{
+  return spec.make_item_func_trim_oracle(thd);
+}
+
+
+Item *Schema::make_item_func_replace(THD *thd,
+                                     Item *subj,
+                                     Item *find,
+                                     Item *replace) const
+{
+  return new (thd->mem_root) Item_func_replace(thd, subj, find, replace);
+}
+
+
+Item *Schema_oracle::make_item_func_replace(THD *thd,
+                                            Item *subj,
+                                            Item *find,
+                                            Item *replace) const
+{
+  return new (thd->mem_root) Item_func_replace_oracle(thd, subj, find, replace);
+}
diff --git a/sql/sql_schema.h b/sql/sql_schema.h
index 7c8f284d526..36914e36990 100644
--- a/sql/sql_schema.h
+++ b/sql/sql_schema.h
@@ -19,6 +19,9 @@
 #include "mysqld.h"
 #include "lex_string.h"
 
+class Lex_ident_sys;
+class Create_func;
+
 class Schema
 {
   LEX_CSTRING m_name;
@@ -33,6 +36,34 @@ public:
   {
     return src;
   }
+
+
+  /**
+    Find the native function builder associated with a given function name.
+    @param thd The current thread
+    @param name The native function name
+    @return The native function builder associated with the name, or NULL
+  */
+  virtual Create_func *find_native_function_builder(THD *thd,
+                                                    const LEX_CSTRING &name)
+                                                    const;
+  /**
+    Find a native function builder, return an error if not found,
+    build an Item otherwise.
+  */
+  Item *make_item_func_call_native(THD *thd,
+                                   const Lex_ident_sys &name,
+                                   List<Item> *args) const;
+
+  // Builders for native SQL function with a special syntax in sql_yacc.yy
+  virtual Item *make_item_func_substr(THD *thd,
+                                      const Lex_substring_spec_st &spec) const;
+
+  virtual Item *make_item_func_trim(THD *thd, const Lex_trim_st &spec) const;
+  virtual Item *make_item_func_replace(THD *thd,
+                                       Item *subj,
+                                       Item *find,
+                                       Item *replace) const;
   /*
     For now we have *hard-coded* compatibility schemas:
       schema_mariadb, schema_oracle, schema_maxdb.
@@ -66,5 +97,6 @@ public:
 
 
 extern Schema mariadb_schema;
+extern const Schema &oracle_schema_ref;
 
 #endif // SQL_SCHEMA_H_INCLUDED
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 7b0967b4461..06ab2a212f3 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -74,9 +74,6 @@ extern size_t symbols_length;
 extern SYMBOL sql_functions[];
 extern size_t sql_functions_length;
 
-extern Native_func_registry func_array[];
-extern size_t func_array_length;
-
 enum enum_i_s_events_fields
 {
   ISE_EVENT_CATALOG= 0,
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 1b8762f90dd..f3980ad72dd 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -950,16 +950,15 @@ static int mysql_register_view(THD *thd, TABLE_LIST *view,
   view_query.length(0);
   is_query.length(0);
   {
-    sql_mode_t sql_mode= thd->variables.sql_mode & MODE_ANSI_QUOTES;
-    thd->variables.sql_mode&= ~MODE_ANSI_QUOTES;
+    Sql_mode_save_for_frm_handling sql_mode_save(thd);
 
-    lex->unit.print(&view_query, enum_query_type(QT_VIEW_INTERNAL |
+    lex->unit.print(&view_query, enum_query_type(QT_FOR_FRM |
+                                                 QT_VIEW_INTERNAL |
                                                  QT_ITEM_ORIGINAL_FUNC_NULLIF));
-    lex->unit.print(&is_query, enum_query_type(QT_TO_SYSTEM_CHARSET |
+    lex->unit.print(&is_query, enum_query_type(QT_ITEM_FUNC_FORCE_SCHEMA_NAME |
+                                               QT_TO_SYSTEM_CHARSET |
                                                QT_WITHOUT_INTRODUCERS |
                                                QT_ITEM_ORIGINAL_FUNC_NULLIF));
-
-    thd->variables.sql_mode|= sql_mode;
   }
   DBUG_PRINT("info", ("View: %.*s", view_query.length(), view_query.ptr()));
 
@@ -1420,35 +1419,7 @@ bool mysql_make_view(THD *thd, TABLE_SHARE *share, TABLE_LIST *table,
     view_select= &lex->select_lex;
     view_select->select_number= ++thd->lex->stmt_lex->current_select_number;
 
-    sql_mode_t saved_mode= thd->variables.sql_mode;
-    /* switch off modes which can prevent normal parsing of VIEW
-      - MODE_REAL_AS_FLOAT            affect only CREATE TABLE parsing
-      + MODE_PIPES_AS_CONCAT          affect expression parsing
-      + MODE_ANSI_QUOTES              affect expression parsing
-      + MODE_IGNORE_SPACE             affect expression parsing
-      - MODE_IGNORE_BAD_TABLE_OPTIONS affect only CREATE/ALTER TABLE parsing
-      * MODE_ONLY_FULL_GROUP_BY       affect execution
-      * MODE_NO_UNSIGNED_SUBTRACTION  affect execution
-      - MODE_NO_DIR_IN_CREATE         affect table creation only
-      - MODE_POSTGRESQL               compounded from other modes
-      - MODE_ORACLE                   affects Item creation (e.g for CONCAT)
-      - MODE_MSSQL                    compounded from other modes
-      - MODE_DB2                      compounded from other modes
-      - MODE_MAXDB                    affect only CREATE TABLE parsing
-      - MODE_NO_KEY_OPTIONS           affect only SHOW
-      - MODE_NO_TABLE_OPTIONS         affect only SHOW
-      - MODE_NO_FIELD_OPTIONS         affect only SHOW
-      - MODE_MYSQL323                 affect only SHOW
-      - MODE_MYSQL40                  affect only SHOW
-      - MODE_ANSI                     compounded from other modes
-                                      (+ transaction mode)
-      ? MODE_NO_AUTO_VALUE_ON_ZERO    affect UPDATEs
-      + MODE_NO_BACKSLASH_ESCAPES     affect expression parsing
-    */
-    thd->variables.sql_mode&= ~(MODE_PIPES_AS_CONCAT | MODE_ANSI_QUOTES |
-                                MODE_IGNORE_SPACE | MODE_NO_BACKSLASH_ESCAPES |
-                                MODE_ORACLE);
-
+    Sql_mode_save_for_frm_handling sql_mode_save(thd);
     /* Parse the query. */
 
     parse_status= parse_sql(thd, & parser_state, table->view_creation_ctx);
@@ -1459,8 +1430,6 @@ bool mysql_make_view(THD *thd, TABLE_SHARE *share, TABLE_LIST *table,
         (old_lex->sql_command == SQLCOM_SHOW_CREATE))
         lex->sql_command= old_lex->sql_command;
 
-    thd->variables.sql_mode= saved_mode;
-
     if (dbchanged && mysql_change_db(thd, &old_db, TRUE))
       goto err;
   }
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index e03bde31832..6478216979e 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -802,6 +802,7 @@ Virtual_column_info *add_virtual_expression(THD *thd, Item *expr)
   Lex_for_loop_st for_loop;
   Lex_for_loop_bounds_st for_loop_bounds;
   Lex_trim_st trim;
+  Lex_substring_spec_st substring_spec;
   vers_history_point_t vers_history_point;
 
   /* pointers */
@@ -1138,7 +1139,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  RELEASE_SYM                   /* SQL-2003-R */
 %token  RENAME
 %token  REPEAT_SYM                    /* MYSQL-FUNC */
-%token  REPLACE                       /* MYSQL-FUNC */
 %token  REQUIRE_SYM
 %token  RESIGNAL_SYM                  /* SQL-2003-R */
 %token  RESTRICT
@@ -1178,7 +1178,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  STDDEV_SAMP_SYM               /* SQL-2003-N */
 %token  STD_SYM
 %token  STRAIGHT_JOIN
-%token  SUBSTRING                     /* SQL-2003-N */
 %token  SUM_SYM                       /* SQL-2003-N */
 %token  SYSDATE
 %token  TABLE_REF_PRIORITY
@@ -1192,7 +1191,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  TO_SYM                        /* SQL-2003-R */
 %token  TRAILING                      /* SQL-2003-R */
 %token  TRIGGER_SYM                   /* SQL-2003-R */
-%token  TRIM                          /* SQL-2003-N */
 %token  TRUE_SYM                      /* SQL-2003-R */
 %token  ULONGLONG_NUM
 %token  UNDERSCORE_CHARSET
@@ -1244,6 +1242,14 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  <kwd>  ROWTYPE_MARIADB_SYM           // PLSQL-R
 
 /*
+  SQL functions with a special syntax
+*/
+%token  <kwd> REPLACE                        /* MYSQL-FUNC */
+%token  <kwd> SUBSTRING                      /* SQL-2003-N */
+%token  <kwd> TRIM                           /* SQL-2003-N */
+
+
+/*
   Non-reserved keywords
 */
 
@@ -1329,8 +1335,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  <kwd>  DATE_SYM                      /* SQL-2003-R, Oracle-R, PLSQL-R */
 %token  <kwd>  DAY_SYM                       /* SQL-2003-R */
 %token  <kwd>  DEALLOCATE_SYM                /* SQL-2003-R */
-%token  <kwd>  DECODE_MARIADB_SYM            /* Function, non-reserved */
-%token  <kwd>  DECODE_ORACLE_SYM             /* Function, non-reserved */
 %token  <kwd>  DEFINER_SYM
 %token  <kwd>  DELAYED_SYM
 %token  <kwd>  DELAY_KEY_WRITE_SYM
@@ -1995,7 +1999,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %type <item_list>
         expr_list opt_udf_expr_list udf_expr_list when_list when_list_opt_else
         ident_list ident_list_arg opt_expr_list
-        decode_when_list_oracle
 
 %type <sp_cursor_stmt>
         sp_cursor_stmt_lex
@@ -2194,6 +2197,7 @@ END_OF_INPUT
 %type <for_loop> sp_for_loop_index_and_bounds
 %type <for_loop_bounds> sp_for_loop_bounds
 %type <trim> trim_operands
+%type <substring_spec> substring_operands
 %type <num> opt_sp_for_loop_direction
 %type <spvar_mode> sp_opt_inout
 %type <index_hint> index_hint_type
@@ -10672,7 +10676,8 @@ function_call_keyword:
           }
         | TRIM '(' trim_operands ')'
           {
-            if (unlikely(!($$= $3.make_item_func_trim(thd))))
+            if (unlikely(!($$= Schema::find_implied(thd)->
+                                 make_item_func_trim(thd, $3))))
               MYSQL_YYABORT;
           }
         | USER_SYM '(' ')'
@@ -10691,6 +10696,26 @@ function_call_keyword:
           }
         ;
 
+substring_operands:
+          expr ',' expr ',' expr
+          {
+            $$= Lex_substring_spec_st::init($1, $3, $5);
+          }
+        | expr ',' expr
+          {
+            $$= Lex_substring_spec_st::init($1, $3);
+          }
+        | expr FROM expr FOR_SYM expr
+          {
+            $$= Lex_substring_spec_st::init($1, $3, $5);
+          }
+        | expr FROM expr
+          {
+            $$= Lex_substring_spec_st::init($1, $3);
+          }
+        ;
+
+
 /*
   Function calls using non reserved keywords, with special syntaxic forms.
   Dedicated grammar rules are needed because of the syntax,
@@ -10755,18 +10780,6 @@ function_call_nonkeyword:
             if (unlikely($$ == NULL))
               MYSQL_YYABORT;
           }
-        | DECODE_MARIADB_SYM '(' expr ',' expr ')'
-          {
-            $$= new (thd->mem_root) Item_func_decode(thd, $3, $5);
-            if (unlikely($$ == NULL))
-              MYSQL_YYABORT;
-          }
-        | DECODE_ORACLE_SYM '(' expr ',' decode_when_list_oracle ')'
-          {
-            $5->push_front($3, thd->mem_root);
-            if (unlikely(!($$= new (thd->mem_root) Item_func_decode_oracle(thd, *$5))))
-              MYSQL_YYABORT;
-          }
         | EXTRACT_SYM '(' interval FROM expr ')'
           {
             $$=new (thd->mem_root) Item_extract(thd, $3, $5);
@@ -10805,24 +10818,10 @@ function_call_nonkeyword:
             if (unlikely($$ == NULL))
               MYSQL_YYABORT;
           }
-        | SUBSTRING '(' expr ',' expr ',' expr ')'
-          {
-            if (unlikely(!($$= Lex->make_item_func_substr(thd, $3, $5, $7))))
-              MYSQL_YYABORT;
-          }
-        | SUBSTRING '(' expr ',' expr ')'
-          {
-            if (unlikely(!($$= Lex->make_item_func_substr(thd, $3, $5))))
-              MYSQL_YYABORT;
-          }
-        | SUBSTRING '(' expr FROM expr FOR_SYM expr ')'
+        | SUBSTRING '(' substring_operands ')'
           {
-            if (unlikely(!($$= Lex->make_item_func_substr(thd, $3, $5, $7))))
-              MYSQL_YYABORT;
-          }
-        | SUBSTRING '(' expr FROM expr ')'
-          {
-            if (unlikely(!($$= Lex->make_item_func_substr(thd, $3, $5))))
+            if (unlikely(!($$= Schema::find_implied(thd)->
+                                 make_item_func_substr(thd, $3))))
               MYSQL_YYABORT;
           }
         | SYSDATE opt_time_precision
@@ -11038,7 +11037,8 @@ function_call_conflict:
           }
         | REPLACE '(' expr ',' expr ',' expr ')'
           {
-            if (unlikely(!($$= Lex->make_item_func_replace(thd, $3, $5, $7))))
+            if (unlikely(!($$= Schema::find_implied(thd)->
+                                 make_item_func_replace(thd, $3, $5, $7))))
               MYSQL_YYABORT;
           }
         | REVERSE_SYM '(' expr ')'
@@ -11226,7 +11226,8 @@ function_call_generic:
 
               This will be revised with WL#2128 (SQL PATH)
             */
-            builder= find_native_function_builder(thd, &$1);
+            builder= Schema::find_implied(thd)->
+                       find_native_function_builder(thd, $1);
             if (builder)
             {
               item= builder->create_func(thd, &$1, $4);
@@ -11268,6 +11269,34 @@ function_call_generic:
             if (unlikely(!($$= Lex->make_item_func_call_generic(thd, &$1, &$3, &$5, $7))))
               MYSQL_YYABORT;
           }
+        | ident_cli '.' REPLACE '(' expr ',' expr ',' expr ')'
+          {
+            if (unlikely(!($$= Lex->make_item_func_replace(thd, $1, $3,
+                                                           $5, $7, $9))))
+              MYSQL_YYABORT;
+          }
+        | ident_cli '.' SUBSTRING '(' substring_operands ')'
+          {
+            if (unlikely(!($$= Lex->make_item_func_substr(thd, $1, $3, $5))))
+              MYSQL_YYABORT;
+          }
+        | ident_cli '.' TRIM '(' trim_operands ')'
+          {
+            if (unlikely(!($$= $5.make_item_func_trim(thd, $1, $3))))
+              MYSQL_YYABORT;
+          }
+          /*
+            We don't add a qualified syntax for TRIM_ORACLE here,
+            as this syntax is not absolutely required:
+              SELECT mariadb_schema.TRIM_ORACLE(..);
+            What absolutely required is only:
+              SELECT mariadb_schema.TRIM(..);
+            Adding a qualified syntax for TRIM_ORACLE would be tricky because
+            it is a non-reserved keyword. To avoid new shift/reduce conflicts
+            it would require grammar changes, like introducing a new rule
+            ident_step2_cli (which would include everything that ident_cli
+            includes but TRIM_ORACLE).
+          */
         ;
 
 fulltext_options:
@@ -11935,25 +11964,6 @@ when_list_opt_else:
           }
         ;
 
-decode_when_list_oracle:
-          expr ',' expr
-          {
-            $$= new (thd->mem_root) List<Item>;
-            if (unlikely($$ == NULL) ||
-                unlikely($$->push_back($1, thd->mem_root)) ||
-                unlikely($$->push_back($3, thd->mem_root)))
-              MYSQL_YYABORT;
-
-          }
-        | decode_when_list_oracle ',' expr
-          {
-            $$= $1;
-            if (unlikely($$->push_back($3, thd->mem_root)))
-              MYSQL_YYABORT;
-          }
-        ;
-
-
 /* Equivalent to <table reference> in the SQL:2003 standard. */
 /* Warning - may return NULL in case of incomplete SELECT */
 table_ref:
@@ -16153,8 +16163,6 @@ keyword_sp_var_and_label:
         | DATAFILE_SYM
         | DATE_FORMAT_SYM
         | DAY_SYM
-        | DECODE_MARIADB_SYM
-        | DECODE_ORACLE_SYM
         | DEFINER_SYM
         | DELAY_KEY_WRITE_SYM
         | DES_KEY_FILE
diff --git a/sql/sql_yacc_ora.yy b/sql/sql_yacc_ora.yy
index 89f7412ea89..e12c45f8d47 100644
--- a/sql/sql_yacc_ora.yy
+++ b/sql/sql_yacc_ora.yy
@@ -195,6 +195,7 @@ void ORAerror(THD *thd, const char *s)
   Lex_for_loop_st for_loop;
   Lex_for_loop_bounds_st for_loop_bounds;
   Lex_trim_st trim;
+  Lex_substring_spec_st substring_spec;
   vers_history_point_t vers_history_point;
 
   /* pointers */
@@ -532,7 +533,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  RELEASE_SYM                   /* SQL-2003-R */
 %token  RENAME
 %token  REPEAT_SYM                    /* MYSQL-FUNC */
-%token  REPLACE                       /* MYSQL-FUNC */
 %token  REQUIRE_SYM
 %token  RESIGNAL_SYM                  /* SQL-2003-R */
 %token  RESTRICT
@@ -572,7 +572,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  STDDEV_SAMP_SYM               /* SQL-2003-N */
 %token  STD_SYM
 %token  STRAIGHT_JOIN
-%token  SUBSTRING                     /* SQL-2003-N */
 %token  SUM_SYM                       /* SQL-2003-N */
 %token  SYSDATE
 %token  TABLE_REF_PRIORITY
@@ -586,7 +585,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  TO_SYM                        /* SQL-2003-R */
 %token  TRAILING                      /* SQL-2003-R */
 %token  TRIGGER_SYM                   /* SQL-2003-R */
-%token  TRIM                          /* SQL-2003-N */
 %token  TRUE_SYM                      /* SQL-2003-R */
 %token  ULONGLONG_NUM
 %token  UNDERSCORE_CHARSET
@@ -638,6 +636,14 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  <kwd>  ROWTYPE_MARIADB_SYM           // PLSQL-R
 
 /*
+  SQL functions with a special syntax
+*/
+%token  <kwd> REPLACE                        /* MYSQL-FUNC */
+%token  <kwd> SUBSTRING                      /* SQL-2003-N */
+%token  <kwd> TRIM                           /* SQL-2003-N */
+
+
+/*
   Non-reserved keywords
 */
 
@@ -723,8 +729,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  <kwd>  DATE_SYM                      /* SQL-2003-R, Oracle-R, PLSQL-R */
 %token  <kwd>  DAY_SYM                       /* SQL-2003-R */
 %token  <kwd>  DEALLOCATE_SYM                /* SQL-2003-R */
-%token  <kwd>  DECODE_MARIADB_SYM            /* Function, non-reserved */
-%token  <kwd>  DECODE_ORACLE_SYM             /* Function, non-reserved */
 %token  <kwd>  DEFINER_SYM
 %token  <kwd>  DELAYED_SYM
 %token  <kwd>  DELAY_KEY_WRITE_SYM
@@ -1396,7 +1400,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %type <item_list>
         expr_list opt_udf_expr_list udf_expr_list when_list when_list_opt_else
         ident_list ident_list_arg opt_expr_list
-        decode_when_list_oracle
 
 %type <sp_cursor_stmt>
         sp_cursor_stmt_lex
@@ -1615,6 +1618,7 @@ END_OF_INPUT
 %type <for_loop> sp_for_loop_index_and_bounds
 %type <for_loop_bounds> sp_for_loop_bounds
 %type <trim> trim_operands
+%type <substring_spec> substring_operands
 %type <num> opt_sp_for_loop_direction
 %type <spvar_mode> sp_opt_inout
 %type <index_hint> index_hint_type
@@ -10629,7 +10633,8 @@ function_call_keyword:
           }
         | TRIM '(' trim_operands ')'
           {
-            if (unlikely(!($$= $3.make_item_func_trim(thd))))
+            if (unlikely(!($$= Schema::find_implied(thd)->
+                                 make_item_func_trim(thd, $3))))
               MYSQL_YYABORT;
           }
         | USER_SYM '(' ')'
@@ -10648,6 +10653,26 @@ function_call_keyword:
           }
         ;
 
+substring_operands:
+          expr ',' expr ',' expr
+          {
+            $$= Lex_substring_spec_st::init($1, $3, $5);
+          }
+        | expr ',' expr
+          {
+            $$= Lex_substring_spec_st::init($1, $3);
+          }
+        | expr FROM expr FOR_SYM expr
+          {
+            $$= Lex_substring_spec_st::init($1, $3, $5);
+          }
+        | expr FROM expr
+          {
+            $$= Lex_substring_spec_st::init($1, $3);
+          }
+        ;
+
+
 /*
   Function calls using non reserved keywords, with special syntaxic forms.
   Dedicated grammar rules are needed because of the syntax,
@@ -10712,18 +10737,6 @@ function_call_nonkeyword:
             if (unlikely($$ == NULL))
               MYSQL_YYABORT;
           }
-        | DECODE_MARIADB_SYM '(' expr ',' expr ')'
-          {
-            $$= new (thd->mem_root) Item_func_decode(thd, $3, $5);
-            if (unlikely($$ == NULL))
-              MYSQL_YYABORT;
-          }
-        | DECODE_ORACLE_SYM '(' expr ',' decode_when_list_oracle ')'
-          {
-            $5->push_front($3, thd->mem_root);
-            if (unlikely(!($$= new (thd->mem_root) Item_func_decode_oracle(thd, *$5))))
-              MYSQL_YYABORT;
-          }
         | EXTRACT_SYM '(' interval FROM expr ')'
           {
             $$=new (thd->mem_root) Item_extract(thd, $3, $5);
@@ -10762,24 +10775,10 @@ function_call_nonkeyword:
             if (unlikely($$ == NULL))
               MYSQL_YYABORT;
           }
-        | SUBSTRING '(' expr ',' expr ',' expr ')'
+        | SUBSTRING '(' substring_operands ')'
           {
-            if (unlikely(!($$= Lex->make_item_func_substr(thd, $3, $5, $7))))
-              MYSQL_YYABORT;
-          }
-        | SUBSTRING '(' expr ',' expr ')'
-          {
-            if (unlikely(!($$= Lex->make_item_func_substr(thd, $3, $5))))
-              MYSQL_YYABORT;
-          }
-        | SUBSTRING '(' expr FROM expr FOR_SYM expr ')'
-          {
-            if (unlikely(!($$= Lex->make_item_func_substr(thd, $3, $5, $7))))
-              MYSQL_YYABORT;
-          }
-        | SUBSTRING '(' expr FROM expr ')'
-          {
-            if (unlikely(!($$= Lex->make_item_func_substr(thd, $3, $5))))
+            if (unlikely(!($$= Schema::find_implied(thd)->
+                                 make_item_func_substr(thd, $3))))
               MYSQL_YYABORT;
           }
         | SYSDATE opt_time_precision
@@ -10995,7 +10994,8 @@ function_call_conflict:
           }
         | REPLACE '(' expr ',' expr ',' expr ')'
           {
-            if (unlikely(!($$= Lex->make_item_func_replace(thd, $3, $5, $7))))
+            if (unlikely(!($$= Schema::find_implied(thd)->
+                                 make_item_func_replace(thd, $3, $5, $7))))
               MYSQL_YYABORT;
           }
         | REVERSE_SYM '(' expr ')'
@@ -11183,7 +11183,8 @@ function_call_generic:
 
               This will be revised with WL#2128 (SQL PATH)
             */
-            builder= find_native_function_builder(thd, &$1);
+            builder= Schema::find_implied(thd)->
+                       find_native_function_builder(thd, $1);
             if (builder)
             {
               item= builder->create_func(thd, &$1, $4);
@@ -11225,6 +11226,22 @@ function_call_generic:
             if (unlikely(!($$= Lex->make_item_func_call_generic(thd, &$1, &$3, &$5, $7))))
               MYSQL_YYABORT;
           }
+        | ident_cli '.' REPLACE '(' expr ',' expr ',' expr ')'
+          {
+            if (unlikely(!($$= Lex->make_item_func_replace(thd, $1, $3,
+                                                           $5, $7, $9))))
+              MYSQL_YYABORT;
+          }
+        | ident_cli '.' SUBSTRING '(' substring_operands ')'
+          {
+            if (unlikely(!($$= Lex->make_item_func_substr(thd, $1, $3, $5))))
+              MYSQL_YYABORT;
+          }
+        | ident_cli '.' TRIM '(' trim_operands ')'
+          {
+            if (unlikely(!($$= $5.make_item_func_trim(thd, $1, $3))))
+              MYSQL_YYABORT;
+          }
         ;
 
 fulltext_options:
@@ -11892,25 +11909,6 @@ when_list_opt_else:
           }
         ;
 
-decode_when_list_oracle:
-          expr ',' expr
-          {
-            $$= new (thd->mem_root) List<Item>;
-            if (unlikely($$ == NULL) ||
-                unlikely($$->push_back($1, thd->mem_root)) ||
-                unlikely($$->push_back($3, thd->mem_root)))
-              MYSQL_YYABORT;
-
-          }
-        | decode_when_list_oracle ',' expr
-          {
-            $$= $1;
-            if (unlikely($$->push_back($3, thd->mem_root)))
-              MYSQL_YYABORT;
-          }
-        ;
-
-
 /* Equivalent to <table reference> in the SQL:2003 standard. */
 /* Warning - may return NULL in case of incomplete SELECT */
 table_ref:
@@ -16169,8 +16167,6 @@ keyword_sp_var_and_label:
         | DATAFILE_SYM
         | DATE_FORMAT_SYM
         | DAY_SYM
-        | DECODE_MARIADB_SYM
-        | DECODE_ORACLE_SYM
         | DEFINER_SYM
         | DELAY_KEY_WRITE_SYM
         | DES_KEY_FILE
diff --git a/sql/structs.h b/sql/structs.h
index c6f9fef2277..690e2306f40 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -32,6 +32,7 @@ struct TABLE;
 class Type_handler;
 class Field;
 class Index_statistics;
+struct Lex_ident_cli_st;
 
 class THD;
 
@@ -763,7 +764,9 @@ public:
   }
   Item *make_item_func_trim_std(THD *thd) const;
   Item *make_item_func_trim_oracle(THD *thd) const;
-  Item *make_item_func_trim(THD *thd) const;
+  Item *make_item_func_trim(THD *thd,
+                            const Lex_ident_cli_st &schema_name,
+                            const Lex_ident_cli_st &func_name) const;
 };
 
 
@@ -774,6 +777,25 @@ public:
 };
 
 
+class Lex_substring_spec_st
+{
+public:
+  Item *m_subject;
+  Item *m_from;
+  Item *m_for;
+  static Lex_substring_spec_st init(Item *subject,
+                                    Item *from,
+                                    Item *xfor= NULL)
+  {
+    Lex_substring_spec_st res;
+    res.m_subject= subject;
+    res.m_from= from;
+    res.m_for= xfor;
+    return res;
+  }
+};
+
+
 class Load_data_param
 {
 protected:
diff --git a/sql/table.cc b/sql/table.cc
index a1ee2f3b8b3..4bc2244581a 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -1040,7 +1040,7 @@ bool parse_vcol_defs(THD *thd, MEM_ROOT *mem_root, TABLE *table,
   Field **vfield_ptr= table->vfield;
   Field **dfield_ptr= table->default_field;
   Virtual_column_info **check_constraint_ptr= table->check_constraints;
-  sql_mode_t saved_mode= thd->variables.sql_mode;
+  Sql_mode_save_for_frm_handling sql_mode_save(thd);
   Query_arena backup_arena;
   Virtual_column_info *vcol= 0;
   StringBuffer<MAX_FIELD_WIDTH> expr_str;
@@ -1067,7 +1067,6 @@ bool parse_vcol_defs(THD *thd, MEM_ROOT *mem_root, TABLE *table,
   thd->stmt_arena= table->expr_arena;
   thd->update_charset(&my_charset_utf8mb4_general_ci, table->s->table_charset);
   expr_str.append(&parse_vcol_keyword);
-  thd->variables.sql_mode &= ~(MODE_NO_BACKSLASH_ESCAPES | MODE_EMPTY_STRING_IS_NULL);
 
   while (pos < end)
   {
@@ -1190,7 +1189,6 @@ end:
   thd->stmt_arena= backup_stmt_arena_ptr;
   if (save_character_set_client)
     thd->update_charset(save_character_set_client, save_collation);
-  thd->variables.sql_mode= saved_mode;
   DBUG_RETURN(res);
 }
 
diff --git a/sql/unireg.cc b/sql/unireg.cc
index efbb5e773b4..3b37944afa5 100644
--- a/sql/unireg.cc
+++ b/sql/unireg.cc
@@ -189,10 +189,7 @@ LEX_CUSTRING build_frm_image(THD *thd, const LEX_CSTRING *table,
     create_info->null_bits++;
   data_offset= (create_info->null_bits + 7) / 8;
 
-  sql_mode_t save_sql_mode= thd->variables.sql_mode;
-  thd->variables.sql_mode &= ~MODE_ANSI_QUOTES;
   error= pack_vcols(&vcols, create_fields, create_info->check_constraint_list);
-  thd->variables.sql_mode= save_sql_mode;
 
   if (unlikely(error))
     DBUG_RETURN(frm);
@@ -652,6 +649,7 @@ static bool pack_expression(String *buf, Virtual_column_info *vcol,
 static bool pack_vcols(String *buf, List<Create_field> &create_fields,
                              List<Virtual_column_info> *check_constraint_list)
 {
+  Sql_mode_save_for_frm_handling sql_mode_save(current_thd);
   List_iterator<Create_field> it(create_fields);
   Create_field *field;