MDEV-20831 Table partitioned by LIST/RANGE COLUMNS(inet6) can be created, but not inserted into

This clause in CREATE TABLE:

  PARTITION BY LIST COLUMNS (inet6column)
    (PARTITION p1 VALUES IN ('::'))

was erroneously written to frm file as:

  PARTITION BY LIST COLUMNS(inet6column)
    (PARTITION p1 VALUES IN (_binary 0x3A3A))

I.e. the text value '::' was converted to HEX representation
and prefixed with _binary.

A simple fix could write `_latin1 0x3A3A` instead of `_binary 0x3A3A`,
but in case of INET6 we don't need neither character set introducers,
nor HEX encoding, because text representation of INET6 values consist
of pure ASCII characters.

So this patch changes the above clause to be printed as:

  PARTITION BY LIST COLUMNS(inet6column)
    (PARTITION p1 VALUES IN ('::'))

Details:

The old code in check_part_field() was not friendly to pluggable data types.
Replacing this function to two new virtual methods in Type_handler:

  virtual bool partition_field_check(const LEX_CSTRING &field_name,
                                     Item *item_expr) const;

  virtual bool partition_field_append_value(String *str,
                                            Item *item_expr,
                                            CHARSET_INFO *field_cs,
                                            partition_value_print_mode_t mode)
                                            const;

so data type plugins can decide whether they need to use character set
introducer and/or hex encoding when printing partition values.

plugin/type_inet/mysql-test/type_inet/type_inet6_partition.result

+#
+# MDEV-20831 Table partitioned by LIST/RANGE COLUMNS(inet6) can be created, but not inserted into
+#
+SET NAMES utf8;
+CREATE TABLE t1 (a INET6)
+PARTITION BY LIST COLUMNS(a)
+(PARTITION p00 VALUES IN (10));
+ERROR HY000: Partition column values of incorrect type
+CREATE TABLE t1 (a INET6)
+PARTITION BY LIST COLUMNS(a)
+(PARTITION p00 VALUES IN (TIME'10:20:30'));
+ERROR HY000: Partition column values of incorrect type
+CREATE TABLE t1 (a INET6)
+PARTITION BY LIST COLUMNS(a)
+(PARTITION p00 VALUES IN ('€'));
+ERROR HY000: This partition function is not allowed
+CREATE TABLE t1 (a INET6)
+PARTITION BY LIST COLUMNS(a)
+(PARTITION p00 VALUES IN ('::'),
+PARTITION pFF VALUES IN (0xFFFF000000000000000000000000FFFF));
+INSERT INTO t1 VALUES ('::');
+INSERT INTO t1 VALUES ('ffff::ffff');
+SELECT * FROM t1 PARTITION (p00);
+a
+::
+SELECT * FROM t1 PARTITION (pFF);
+a
+ffff::ffff
+DROP TABLE t1;

plugin/type_inet/mysql-test/type_inet/type_inet6_partition.test

+--source include/have_partition.inc
+
+--echo #
+--echo # MDEV-20831 Table partitioned by LIST/RANGE COLUMNS(inet6) can be created, but not inserted into
+--echo #
+
+SET NAMES utf8;
+
+--error ER_WRONG_TYPE_COLUMN_VALUE_ERROR
+CREATE TABLE t1 (a INET6)
+  PARTITION BY LIST COLUMNS(a)
+    (PARTITION p00 VALUES IN (10));
+
+--error ER_WRONG_TYPE_COLUMN_VALUE_ERROR
+CREATE TABLE t1 (a INET6)
+  PARTITION BY LIST COLUMNS(a)
+    (PARTITION p00 VALUES IN (TIME'10:20:30'));
+
+--error ER_PARTITION_FUNCTION_IS_NOT_ALLOWED
+CREATE TABLE t1 (a INET6)
+  PARTITION BY LIST COLUMNS(a)
+    (PARTITION p00 VALUES IN ('€'));
+
+CREATE TABLE t1 (a INET6)
+  PARTITION BY LIST COLUMNS(a)
+    (PARTITION p00 VALUES IN ('::'),
+     PARTITION pFF VALUES IN (0xFFFF000000000000000000000000FFFF));
+INSERT INTO t1 VALUES ('::');
+INSERT INTO t1 VALUES ('ffff::ffff');
+SELECT * FROM t1 PARTITION (p00);
+SELECT * FROM t1 PARTITION (pFF);
+DROP TABLE t1;

plugin/type_inet/sql_type_inet.cc

@@ -1449,6 +1449,40 @@ Field *Type_handler_inet6::make_conversion_table_field(MEM_ROOT *root,
 }
 
 
+bool Type_handler_inet6::partition_field_check(const LEX_CSTRING &field_name,
+                                               Item *item_expr) const
+{
+  if (item_expr->cmp_type() != STRING_RESULT)
+  {
+    my_error(ER_WRONG_TYPE_COLUMN_VALUE_ERROR, MYF(0));
+    return true;
+  }
+  return false;
+}
+
+
+bool
+Type_handler_inet6::partition_field_append_value(
+                                          String *to,
+                                          Item *item_expr,
+                                          CHARSET_INFO *field_cs,
+                                          partition_value_print_mode_t mode)
+                                          const
+{
+  StringBufferInet6 inet6str;
+  Inet6_null inet6(item_expr);
+  if (inet6.is_null())
+  {
+    my_error(ER_PARTITION_FUNCTION_IS_NOT_ALLOWED, MYF(0));
+    return true;
+  }
+  return inet6.to_string(&inet6str) ||
+         to->append('\'') ||
+         to->append(inet6str) ||
+         to->append('\'');
+}
+
+
 /***************************************************************/
 
 

plugin/type_inet/sql_type_inet.h

@@ -478,6 +478,15 @@ class Type_handler_inet6: public Type_handler
     return false;
   }
 
+  bool partition_field_check(const LEX_CSTRING &field_name,
+                             Item *item_expr) const override;
+
+  bool partition_field_append_value(String *to,
+                                    Item *item_expr,
+                                    CHARSET_INFO *field_cs,
+                                    partition_value_print_mode_t mode)
+                                    const override;
+
   Field *make_table_field(MEM_ROOT *root,
                           const LEX_CSTRING *name,
                           const Record_addr &addr,

sql/sql_partition.cc

@@ -2240,77 +2240,33 @@ static int add_partition_options(String *str, partition_element *p_elem)
 }
 
 
-/*
-  Check partition fields for result type and if they need
-  to check the character set.
-
-  SYNOPSIS
-    check_part_field()
-    sql_type              Type provided by user
-    field_name            Name of field, used for error handling
-    result_type           Out value: Result type of field
-    need_cs_check         Out value: Do we need character set check
+void
+Type_handler::partition_field_type_not_allowed(const LEX_CSTRING &field_name)
+{
+  my_error(ER_FIELD_TYPE_NOT_ALLOWED_AS_PARTITION_FIELD, MYF(0),
+           field_name.str);
+}
 
-  RETURN VALUES
-    TRUE                  Error
-    FALSE                 Ok
-*/
 
-static int check_part_field(enum_field_types sql_type,
-                            const char *field_name,
-                            Item_result *result_type,
-                            bool *need_cs_check)
+bool
+Type_handler::partition_field_check_result_type(Item *item,
+                                                Item_result expected_type)
 {
-  if (sql_type >= MYSQL_TYPE_TINY_BLOB &&
-      sql_type <= MYSQL_TYPE_BLOB)
+  if (item->result_type() != expected_type)
   {
-    my_error(ER_BLOB_FIELD_IN_PART_FUNC_ERROR, MYF(0));
+    my_error(ER_WRONG_TYPE_COLUMN_VALUE_ERROR, MYF(0));
     return TRUE;
   }
-  switch (sql_type)
-  {
-    case MYSQL_TYPE_TINY:
-    case MYSQL_TYPE_SHORT:
-    case MYSQL_TYPE_LONG:
-    case MYSQL_TYPE_LONGLONG:
-    case MYSQL_TYPE_INT24:
-      *result_type= INT_RESULT;
-      *need_cs_check= FALSE;
-      return FALSE;
-    case MYSQL_TYPE_NEWDATE:
-    case MYSQL_TYPE_DATE:
-    case MYSQL_TYPE_TIME:
-    case MYSQL_TYPE_DATETIME:
-    case MYSQL_TYPE_TIME2:
-    case MYSQL_TYPE_DATETIME2:
-      *result_type= STRING_RESULT;
-      *need_cs_check= TRUE;
-      return FALSE;
-    case MYSQL_TYPE_VARCHAR:
-    case MYSQL_TYPE_STRING:
-    case MYSQL_TYPE_VAR_STRING:
-      *result_type= STRING_RESULT;
-      *need_cs_check= TRUE;
-      return FALSE;
-    case MYSQL_TYPE_NEWDECIMAL:
-    case MYSQL_TYPE_DECIMAL:
-    case MYSQL_TYPE_TIMESTAMP:
-    case MYSQL_TYPE_TIMESTAMP2:
-    case MYSQL_TYPE_NULL:
-    case MYSQL_TYPE_FLOAT:
-    case MYSQL_TYPE_DOUBLE:
-    case MYSQL_TYPE_BIT:
-    case MYSQL_TYPE_ENUM:
-    case MYSQL_TYPE_SET:
-    case MYSQL_TYPE_GEOMETRY:
-      goto error;
-    default:
-      goto error;
-  }
-error:
-  my_error(ER_FIELD_TYPE_NOT_ALLOWED_AS_PARTITION_FIELD, MYF(0),
-           field_name);
-  return TRUE;
+  return false;
+}
+
+
+bool
+Type_handler_blob_common::partition_field_check(const LEX_CSTRING &field_name,
+                                                Item *item_expr) const
+{
+  my_error(ER_BLOB_FIELD_IN_PART_FUNC_ERROR, MYF(0));
+  return true;
 }
 
 
@@ -2347,6 +2303,58 @@ static Create_field* get_sql_field(const char *field_name,
 }
 
 
+bool
+Type_handler_general_purpose_int::partition_field_append_value(
+                                            String *str,
+                                            Item *item_expr,
+                                            CHARSET_INFO *field_cs,
+                                            partition_value_print_mode_t mode)
+                                            const
+{
+  DBUG_ASSERT(item_expr->cmp_type() == INT_RESULT);
+  StringBuffer<21> tmp;
+  longlong value= item_expr->val_int();
+  tmp.set(value, system_charset_info);
+  return str->append(tmp);
+}
+
+
+bool Type_handler::partition_field_append_value(
+                                            String *str,
+                                            Item *item_expr,
+                                            CHARSET_INFO *field_cs,
+                                            partition_value_print_mode_t mode)
+                                            const
+{
+  DBUG_ASSERT(cmp_type() != INT_RESULT);
+
+  if (field_cs && field_cs != item_expr->collation.collation)
+  {
+    if (!(item_expr= convert_charset_partition_constant(item_expr,
+                                                        field_cs)))
+    {
+      my_error(ER_PARTITION_FUNCTION_IS_NOT_ALLOWED, MYF(0));
+      return true;
+    }
+  }
+  StringBuffer<MAX_KEY_LENGTH> buf;
+  String val_conv, *res;
+  val_conv.set_charset(system_charset_info);
+  if (!(res= item_expr->val_str(&buf)))
+  {
+    my_error(ER_PARTITION_FUNCTION_IS_NOT_ALLOWED, MYF(0));
+    return true;
+  }
+  if (get_cs_converted_part_value_from_string(current_thd,
+                                              item_expr, res,
+                                              &val_conv, field_cs,
+                                              mode ==
+                                              PARTITION_VALUE_PRINT_MODE_FRM))
+    return true;
+  return str->append(val_conv);
+}
+
+
 static int add_column_list_values(String *str, partition_info *part_info,
                                   part_elem_value *list_value,
                                   HA_CREATE_INFO *create_info,
@@ -2377,8 +2385,7 @@ static int add_column_list_values(String *str, partition_info *part_info,
       else
       {
         CHARSET_INFO *field_cs;
-        bool need_cs_check= FALSE;
-        Item_result result_type= STRING_RESULT;
+        const Type_handler *th= NULL;
 
         /*
           This function is called at a very early stage, even before
@@ -2396,57 +2403,24 @@ static int add_column_list_values(String *str, partition_info *part_info,
             my_error(ER_FIELD_NOT_FOUND_PART_ERROR, MYF(0));
             return 1;
           }
-          if (check_part_field(sql_field->real_field_type(),
-                               sql_field->field_name.str,
-                               &result_type,
-                               &need_cs_check))
+          th= sql_field->type_handler();
+          if (th->partition_field_check(sql_field->field_name, item_expr))
             return 1;
-          if (need_cs_check)
-            field_cs= get_sql_field_charset(sql_field, create_info);
-          else
-            field_cs= NULL;
+          field_cs= get_sql_field_charset(sql_field, create_info);
         }
         else
         {
           Field *field= part_info->part_field_array[i];
-          result_type= field->result_type();
-          if (check_part_field(field->real_type(),
-                               field->field_name.str,
-                               &result_type,
-                               &need_cs_check))
+          th= field->type_handler();
+          if (th->partition_field_check(field->field_name, item_expr))
             return 1;
-          DBUG_ASSERT(result_type == field->result_type());
-          if (need_cs_check)
-            field_cs= field->charset();
-          else
-            field_cs= NULL;
+          field_cs= field->charset();
         }
-        if (result_type != item_expr->result_type())
-        {
-          my_error(ER_WRONG_TYPE_COLUMN_VALUE_ERROR, MYF(0));
+        if (th->partition_field_append_value(str, item_expr, field_cs,
+                                             alter_info == NULL ?
+                                             PARTITION_VALUE_PRINT_MODE_SHOW:
+                                             PARTITION_VALUE_PRINT_MODE_FRM))
           return 1;
-        }
-        if (field_cs && field_cs != item_expr->collation.collation)
-        {
-          if (!(item_expr= convert_charset_partition_constant(item_expr,
-                                                              field_cs)))
-          {
-            my_error(ER_PARTITION_FUNCTION_IS_NOT_ALLOWED, MYF(0));
-            return 1;
-          }
-        }
-        {
-          StringBuffer<MAX_KEY_LENGTH> buf;
-          String val_conv, *res;
-          val_conv.set_charset(system_charset_info);
-          res= item_expr->val_str(&buf);
-          if (get_cs_converted_part_value_from_string(current_thd,
-                                                      item_expr, res,
-                                                      &val_conv, field_cs,
-                                                   (bool)(alter_info != NULL)))
-            return 1;
-          err+= str->append(val_conv);
-        }
       }
     }
     if (i != (num_elements - 1))

sql/sql_type.h

@@ -119,6 +119,13 @@ enum scalar_comparison_op
 };
 
 
+enum partition_value_print_mode_t
+{
+  PARTITION_VALUE_PRINT_MODE_SHOW= 0,
+  PARTITION_VALUE_PRINT_MODE_FRM= 1
+};
+
+
 class Data_type_statistics
 {
 public:
@@ -3357,7 +3364,9 @@ class Type_handler
   static const
   Type_handler *aggregate_for_result_traditional(const Type_handler *h1,
                                                  const Type_handler *h2);
-
+  static void partition_field_type_not_allowed(const LEX_CSTRING &field_name);
+  static bool partition_field_check_result_type(Item *item,
+                                                Item_result expected_type);
   virtual const Name name() const= 0;
   virtual const Name version() const;
   virtual const Name &default_value() const= 0;
@@ -3495,6 +3504,17 @@ class Type_handler
   {
     return this;
   }
+  virtual bool partition_field_check(const LEX_CSTRING &field_name,
+                                     Item *item_expr) const
+  {
+    partition_field_type_not_allowed(field_name);
+    return true;
+  }
+  virtual bool partition_field_append_value(String *str,
+                                            Item *item_expr,
+                                            CHARSET_INFO *field_cs,
+                                            partition_value_print_mode_t mode)
+                                            const;
   virtual int
   stored_field_cmp_to_item(THD *thd, Field *field, Item *item) const= 0;
   virtual CHARSET_INFO *charset_for_protocol(const Item *item) const;
@@ -4865,6 +4885,16 @@ class Type_handler_general_purpose_int: public Type_handler_int_result
   {
     return type_limits_int()->char_length();
   }
+  bool partition_field_check(const LEX_CSTRING &field_name,
+                             Item *item_expr) const override
+  {
+    return partition_field_check_result_type(item_expr, INT_RESULT);
+  }
+  bool partition_field_append_value(String *str,
+                                    Item *item_expr,
+                                    CHARSET_INFO *field_cs,
+                                    partition_value_print_mode_t)
+                                    const override;
   const Vers_type_handler *vers() const override { return &vers_type_trx; }
 };
 
@@ -5659,6 +5689,11 @@ class Type_handler_time_common: public Type_handler_temporal_result
   {
     return MYSQL_TIMESTAMP_TIME;
   }
+  bool partition_field_check(const LEX_CSTRING &field_name,
+                             Item *item_expr) const override
+  {
+    return partition_field_check_result_type(item_expr, STRING_RESULT);
+  }
   Field *make_schema_field(MEM_ROOT *root,
                            TABLE *table,
                            const Record_addr &addr,
@@ -5865,6 +5900,11 @@ class Type_handler_date_common: public Type_handler_temporal_with_date
   {
     return true;
   }
+  bool partition_field_check(const LEX_CSTRING &field_name,
+                             Item *item_expr) const override
+  {
+    return partition_field_check_result_type(item_expr, STRING_RESULT);
+  }
   Field *make_schema_field(MEM_ROOT *root,
                            TABLE *table,
                            const Record_addr &addr,
@@ -5987,6 +6027,11 @@ class Type_handler_datetime_common: public Type_handler_temporal_with_date
   {
     return true;
   }
+  bool partition_field_check(const LEX_CSTRING &field_name,
+                             Item *item_expr) const override
+  {
+    return partition_field_check_result_type(item_expr, STRING_RESULT);
+  }
   Field *make_schema_field(MEM_ROOT *root,
                            TABLE *table,
                            const Record_addr &addr,
@@ -6432,6 +6477,11 @@ class Type_handler_string: public Type_handler_longstr
   {
     return varstring_type_handler(item);
   }
+  bool partition_field_check(const LEX_CSTRING &field_name,
+                             Item *item_expr) const override
+  {
+    return partition_field_check_result_type(item_expr, STRING_RESULT);
+  }
   void show_binlog_type(const Conv_source &src, const Field &dst, String *str)
     const override;
   Field *make_conversion_table_field(MEM_ROOT *root,
@@ -6520,6 +6570,11 @@ class Type_handler_varchar: public Type_handler_longstr
     return varstring_type_handler(item);
   }
   bool is_param_long_data_type() const override { return true; }
+  bool partition_field_check(const LEX_CSTRING &field_name,
+                             Item *item_expr) const override
+  {
+    return partition_field_check_result_type(item_expr, STRING_RESULT);
+  }
   void show_binlog_type(const Conv_source &src, const Field &dst, String *str)
     const override;
   Field *make_conversion_table_field(MEM_ROOT *root,
@@ -6574,6 +6629,12 @@ class Type_handler_varchar_compressed: public Type_handler_varchar
     return 0;
   }
   uint32 max_display_length_for_field(const Conv_source &src) const override;
+  bool partition_field_check(const LEX_CSTRING &field_name,
+                             Item *item_expr) const override
+  {
+    partition_field_type_not_allowed(field_name);
+    return true;
+  }
   void show_binlog_type(const Conv_source &src, const Field &dst, String *str)
     const override;
   Field *make_conversion_table_field(MEM_ROOT *root,
@@ -6648,6 +6709,8 @@ class Type_handler_blob_common: public Type_handler_longstr
     override;
   void Item_param_setup_conversion(THD *thd, Item_param *) const override;
 
+  bool partition_field_check(const LEX_CSTRING &field_name,
+                        Item *item_expr) const override;
   Field *make_schema_field(MEM_ROOT *root,
                            TABLE *table,
                            const Record_addr &addr,