MDEV-23446 UPDATE does not insert history row if the row is not changed

Add history row outside of compare_record() check. For TRX_ID
versioning we have to fail can_compare_record to force InnoDB update
which adds history row; and there in ha_innobase::update_row() is
additional "row changed" check where we force history row anyway.

mysql-test/suite/versioning/r/update.result

@@ -370,3 +370,32 @@ insert into t1 (a) values (1), (2);
 update ignore t1 set a= 3;
 delete history from t1;
 drop table t1;
+#
+# MDEV-23446 UPDATE does not insert history row if the row is not changed
+#
+create table t1 (
+a int,
+row_start SYS_DATATYPE as row start invisible,
+row_end SYS_DATATYPE as row end invisible,
+period for system_time (row_start, row_end)) with system versioning;
+insert into t1 values (1);
+update t1 set a= 1;
+select *, check_row(row_start, row_end) from t1 for system_time all order by row_end;
+a	check_row(row_start, row_end)
+1	HISTORICAL ROW
+1	CURRENT ROW
+# multi-update
+create or replace table t2 like t1;
+create or replace table t3 like t1;
+insert into t2 values (1);
+insert into t3 values (1);
+update t2, t3 set t2.a= 1, t3.a= 1 where t2.a = t3.a;
+select *, check_row(row_start, row_end) from t2 for system_time all order by row_end;
+a	check_row(row_start, row_end)
+1	HISTORICAL ROW
+1	CURRENT ROW
+select *, check_row(row_start, row_end) from t2 for system_time all order by row_end;
+a	check_row(row_start, row_end)
+1	HISTORICAL ROW
+1	CURRENT ROW
+drop tables t1, t2, t3;

mysql-test/suite/versioning/t/update.test

@@ -301,4 +301,29 @@ delete history from t1;
 # cleanup
 drop table t1;
 
+--echo #
+--echo # MDEV-23446 UPDATE does not insert history row if the row is not changed
+--echo #
+replace_result $sys_datatype_expl SYS_DATATYPE;
+eval create table t1 (
+  a int,
+  row_start $sys_datatype_expl as row start invisible,
+  row_end $sys_datatype_expl as row end invisible,
+  period for system_time (row_start, row_end)) with system versioning;
+insert into t1 values (1);
+update t1 set a= 1;
+select *, check_row(row_start, row_end) from t1 for system_time all order by row_end;
+
+--echo # multi-update
+create or replace table t2 like t1;
+create or replace table t3 like t1;
+insert into t2 values (1);
+insert into t3 values (1);
+update t2, t3 set t2.a= 1, t3.a= 1 where t2.a = t3.a;
+select *, check_row(row_start, row_end) from t2 for system_time all order by row_end;
+select *, check_row(row_start, row_end) from t2 for system_time all order by row_end;
+
+# cleanup
+drop tables t1, t2, t3;
+
 source suite/versioning/common_finish.inc;

sql/sql_update.cc

@@ -52,8 +52,9 @@
    compare_record(TABLE*).
  */
 bool records_are_comparable(const TABLE *table) {
-  return ((table->file->ha_table_flags() & HA_PARTIAL_COLUMN_READ) == 0) ||
-    bitmap_is_subset(table->write_set, table->read_set);
+  return !table->versioned(VERS_TRX_ID) &&
+          (((table->file->ha_table_flags() & HA_PARTIAL_COLUMN_READ) == 0) ||
+           bitmap_is_subset(table->write_set, table->read_set));
 }
 
 
@@ -960,25 +961,27 @@ int mysql_update(THD *thd,
         }
         else if (likely(!error))
         {
-          if (has_vers_fields && table->versioned())
-          {
-            if (table->versioned(VERS_TIMESTAMP))
-            {
-              store_record(table, record[2]);
-              table->mark_columns_per_binlog_row_image();
-              error= vers_insert_history_row(table);
-              restore_record(table, record[2]);
-            }
-            if (likely(!error))
-              updated_sys_ver++;
-          }
-          if (likely(!error))
-            updated++;
+          if (has_vers_fields && table->versioned(VERS_TRX_ID))
+            updated_sys_ver++;
+          updated++;
         }
 
         if (unlikely(error) &&
             (!ignore || table->file->is_fatal_error(error, HA_CHECK_ALL)))
         {
+          goto error;
+        }
+      }
+
+      if (likely(!error) && has_vers_fields && table->versioned(VERS_TIMESTAMP))
+      {
+        store_record(table, record[2]);
+        table->mark_columns_per_binlog_row_image();
+        error= vers_insert_history_row(table);
+        restore_record(table, record[2]);
+        if (unlikely(error))
+        {
+error:
           /*
             If (ignore && error is ignorable) we don't have to
             do anything; otherwise...
@@ -989,10 +992,11 @@ int mysql_update(THD *thd,
             flags|= ME_FATALERROR; /* Other handler errors are fatal */
 
           prepare_record_for_error_message(error, table);
-	  table->file->print_error(error,MYF(flags));
-	  error= 1;
-	  break;
-	}
+          table->file->print_error(error,MYF(flags));
+          error= 1;
+          break;
+        }
+        updated_sys_ver++;
       }
 
       if (table->triggers &&
@@ -2410,6 +2414,7 @@ int multi_update::send_data(List<Item> &not_used_values)
           if (!ignore ||
               table->file->is_fatal_error(error, HA_CHECK_ALL))
           {
+error:
             /*
               If (ignore && error == is ignorable) we don't have to
               do anything; otherwise...
@@ -2431,19 +2436,8 @@ int multi_update::send_data(List<Item> &not_used_values)
             error= 0;
             updated--;
           }
-          else if (has_vers_fields && table->versioned())
+          else if (has_vers_fields && table->versioned(VERS_TRX_ID))
           {
-            if (table->versioned(VERS_TIMESTAMP))
-            {
-              store_record(table, record[2]);
-              if (vers_insert_history_row(table))
-              {
-                restore_record(table, record[2]);
-                error= 1;
-                break;
-              }
-              restore_record(table, record[2]);
-            }
             updated_sys_ver++;
           }
           /* non-transactional or transactional table got modified   */
@@ -2457,6 +2451,17 @@ int multi_update::send_data(List<Item> &not_used_values)
           }
         }
       }
+      if (has_vers_fields && table->versioned(VERS_TIMESTAMP))
+      {
+        store_record(table, record[2]);
+        if (vers_insert_history_row(table))
+        {
+          restore_record(table, record[2]);
+          goto error;
+        }
+        restore_record(table, record[2]);
+        updated_sys_ver++;
+      }
       if (table->triggers &&
           unlikely(table->triggers->process_triggers(thd, TRG_EVENT_UPDATE,
                                                      TRG_ACTION_AFTER, TRUE)))

storage/innobase/handler/ha_innodb.cc

@@ -8799,6 +8799,20 @@ ha_innobase::update_row(
 		MySQL that the row is not really updated and it
 		should not increase the count of updated rows.
 		This is fix for http://bugs.mysql.com/29157 */
+		if (m_prebuilt->versioned_write
+		    && thd_sql_command(m_user_thd) != SQLCOM_ALTER_TABLE
+		    /* Multiple UPDATE of same rows in single transaction create
+		       historical rows only once. */
+		    && trx->id != table->vers_start_id()) {
+			error = row_insert_for_mysql((byte*) old_row,
+						     m_prebuilt,
+						     ROW_INS_HISTORICAL);
+			if (error != DB_SUCCESS) {
+				goto func_exit;
+			}
+			innobase_srv_conc_exit_innodb(m_prebuilt);
+			innobase_active_small();
+		}
 		DBUG_RETURN(HA_ERR_RECORD_IS_THE_SAME);
 	} else {
 		const bool vers_set_fields = m_prebuilt->versioned_write