MDEV-21195 INSERT chooses wrong partition for RANGE partitioning by DECIMAL column

Use FLOOR rounding for DECIMAL_RESULT item_expr in partition function.

mysql-test/main/partition_range.result

@@ -1009,3 +1009,47 @@ select * from t1 partition (p1);
 d
 2000-01-01 00:00:01.000000
 DROP TABLE t1, t2;
+#
+# MDEV-21195 INSERT chooses wrong partition for RANGE partitioning by DECIMAL column
+#
+create or replace table t (
+d decimal(2,1)) partition by range (d)
+(partition p1 values less than (10));
+insert into t values (9.9);
+create or replace table t (
+d decimal(2,1)) partition by range (d)
+(partition p1 values less than (10),
+partition p2 values less than (20));
+insert into t values (9.9);
+select * from t partition (p1);
+d
+9.9
+select * from t partition (p2);
+d
+create or replace table t (
+d decimal(2,1)) partition by range (d)
+(partition p1 values less than (-3));
+insert into t values (-3.3);
+create or replace table t (
+d decimal(2,1)) partition by range (d+1)
+(partition p1 values less than (10),
+partition p2 values less than (20));
+insert into t values (8.9);
+select * from t partition (p1);
+d
+8.9
+select * from t partition (p2);
+d
+set time_zone='+00:00';
+create or replace table t (
+d timestamp(1)) partition by range (unix_timestamp(d))
+(partition p1 values less than (1577836800),
+partition p2 values less than (2000000000));
+insert into t values (from_unixtime(1577836799.9));
+select * from t partition (p1);
+d
+2019-12-31 23:59:59.9
+select * from t partition (p2);
+d
+set time_zone=default;
+drop table t;

mysql-test/main/partition_range.test

@@ -999,3 +999,49 @@ select * from t1 partition (p0);
 select * from t1 partition (p1);
 
 DROP TABLE t1, t2;
+
+--echo #
+--echo # MDEV-21195 INSERT chooses wrong partition for RANGE partitioning by DECIMAL column
+--echo #
+create or replace table t (
+  d decimal(2,1)) partition by range (d)
+  (partition p1 values less than (10));
+
+insert into t values (9.9);
+
+create or replace table t (
+  d decimal(2,1)) partition by range (d)
+  (partition p1 values less than (10),
+   partition p2 values less than (20));
+
+insert into t values (9.9);
+select * from t partition (p1);
+select * from t partition (p2);
+
+create or replace table t (
+  d decimal(2,1)) partition by range (d)
+  (partition p1 values less than (-3));
+
+insert into t values (-3.3);
+
+create or replace table t (
+  d decimal(2,1)) partition by range (d+1)
+  (partition p1 values less than (10),
+   partition p2 values less than (20));
+
+insert into t values (8.9);
+select * from t partition (p1);
+select * from t partition (p2);
+
+set time_zone='+00:00';
+create or replace table t (
+  d timestamp(1)) partition by range (unix_timestamp(d))
+  (partition p1 values less than (1577836800),
+   partition p2 values less than (2000000000));
+
+insert into t values (from_unixtime(1577836799.9));
+select * from t partition (p1);
+select * from t partition (p2);
+
+set time_zone=default;
+drop table t;

sql/my_decimal.cc

@@ -345,12 +345,12 @@ void my_decimal_trim(ulonglong *precision, uint *scale)
 */
 
 int my_decimal2int(uint mask, const decimal_t *d, bool unsigned_flag,
-		   longlong *l)
+		   longlong *l, decimal_round_mode round_type)
 {
   int res;
   my_decimal rounded;
   /* decimal_round can return only E_DEC_TRUNCATED */
-  decimal_round(d, &rounded, 0, HALF_UP);
+  decimal_round(d, &rounded, 0, round_type);
   res= (unsigned_flag ?
         decimal2ulonglong(&rounded, (ulonglong *) l) :
         decimal2longlong(&rounded, l));

sql/my_decimal.h

@@ -348,7 +348,7 @@ my_decimal *seconds2my_decimal(bool sign, ulonglong sec, ulong microsec,
                         (TIME)->second_part, (DECIMAL))
 
 int my_decimal2int(uint mask, const decimal_t *d, bool unsigned_flag,
-		   longlong *l);
+		   longlong *l, decimal_round_mode round_type= HALF_UP);
 
 inline
 int my_decimal2double(uint, const decimal_t *d, double *result)

sql/sql_partition.cc

@@ -2840,14 +2840,34 @@ bool partition_key_modified(TABLE *table, const MY_BITMAP *fields)
 
 static inline int part_val_int(Item *item_expr, longlong *result)
 {
-  *result= item_expr->val_int();
+  switch (item_expr->cmp_type())
+  {
+  case DECIMAL_RESULT:
+  {
+    my_decimal buf;
+    my_decimal *val= item_expr->val_decimal(&buf);
+    if (val && my_decimal2int(E_DEC_FATAL_ERROR, val, item_expr->unsigned_flag,
+                              result, FLOOR) != E_DEC_OK)
+      return true;
+    break;
+  }
+  case INT_RESULT:
+    *result= item_expr->val_int();
+    break;
+  case STRING_RESULT:
+  case REAL_RESULT:
+  case ROW_RESULT:
+  case TIME_RESULT:
+    DBUG_ASSERT(0);
+    break;
+  }
   if (item_expr->null_value)
   {
     if (unlikely(current_thd->is_error()))
-      return TRUE;
+      return true;
     *result= LONGLONG_MIN;
   }
-  return FALSE;
+  return false;
 }