[HPOS CLI] Add support for backfilling specific properties or metadata (#45171)

* `wc hpos diff` command enhancements

* Add support for partial backfills to LegacyDataHandler

* Add unit tests

* Add support for partial backfills to `wp wc hpos backfill` CLI tool

* Add changelog

* Augment update_order_from_object() with args parameter

* Implement update_order_from_object() for HPOS

* Simplify partial backfill logic in LegacyDataHandler

* Move prop name validation to legacy data handler

* Apply changes to order before passing to backfill functions

* Simplify update_order_from_object() in HPOS datastore

* Simplify order update from object across datastores

* Undo change

* Minor fix for CPT version (which only takes order differences into account)

* Improve readability around ‘woocommerce_orders_table_datastore_should_save_after_meta_change’ filter

* Commit docblock suggestion

* PHPCS fixes

* Enhance unit test

plugins/woocommerce/changelog/fix-41910

@@ -0,0 +1,4 @@
+Significance: minor
+Type: enhancement
+
+Add support for partial backfilling from or to the HPOS datastore using the CLI.

plugins/woocommerce/includes/data-stores/abstract-wc-order-data-store-cpt.php

@@ -317,7 +317,7 @@ abstract class Abstract_WC_Order_Data_Store_CPT extends WC_Data_Store_WP impleme
 				/**
 				 * Fires immediately after an order is trashed.
 				 *
-				 * @since
+				 * @since 2.7.0
 				 *
 				 * @param int      $order_id ID of the order that has been trashed.
 				 */

plugins/woocommerce/src/Database/Migrations/CustomOrderTable/CLIRunner.php

@@ -1063,6 +1063,10 @@ ORDER BY $meta_table.order_id ASC, $meta_table.meta_key ASC;
 				$hpos_value = is_a( $hpos_value, \WC_DateTime::class ) ? $hpos_value->format( DATE_ATOM ) : $hpos_value;
 				$cpt_value  = is_a( $cpt_value, \WC_DateTime::class ) ? $cpt_value->format( DATE_ATOM ) : $cpt_value;
 
+				// Format for NULL.
+				$hpos_value = is_null( $hpos_value ) ? '' : $hpos_value;
+				$cpt_value  = is_null( $cpt_value ) ? '' : $cpt_value;
+
 				return array(
 					'property' => $key,
 					'hpos'     => $hpos_value,
@@ -1109,6 +1113,12 @@ ORDER BY $meta_table.order_id ASC, $meta_table.meta_key ASC;
 	 *   - posts
 	 * ---
 	 *
+	 * [--meta_keys=<meta_keys>]
+	 * : Comma separated list of meta keys to backfill.
+	 *
+	 * [--props=<props>]
+	 * : Comma separated list of order properties to backfill.
+	 *
 	 * @since 8.6.0
 	 *
 	 * @param array $args       Positional arguments passed to the command.
@@ -1136,8 +1146,14 @@ ORDER BY $meta_table.order_id ASC, $meta_table.meta_key ASC;
 			WP_CLI::error( __( 'Please use different source (--from) and destination (--to) datastores.', 'woocommerce' ) );
 		}
 
+		$fields = array_intersect_key( $assoc_args, array_flip( array( 'meta_keys', 'props' ) ) );
+		foreach ( $fields as &$field_names ) {
+			$field_names = is_string( $field_names ) ? array_map( 'trim', explode( ',', $field_names ) ) : $field_names;
+			$field_names = array_unique( array_filter( array_filter( $field_names, 'is_string' ) ) );
+		}
+
 		try {
-			$legacy_handler->backfill_order_to_datastore( $order_id, $from, $to );
+			$legacy_handler->backfill_order_to_datastore( $order_id, $from, $to, $fields );
 		} catch ( \Exception $e ) {
 			WP_CLI::error(
 				sprintf(

plugins/woocommerce/src/Internal/DataStores/Orders/LegacyDataHandler.php

@@ -6,7 +6,9 @@
 namespace Automattic\WooCommerce\Internal\DataStores\Orders;
 
 use Automattic\WooCommerce\Database\Migrations\CustomOrderTable\PostsToOrdersMigrationController;
+use Automattic\WooCommerce\Internal\Utilities\DatabaseUtil;
 use Automattic\WooCommerce\Utilities\ArrayUtil;
+use WC_Abstract_Order;
 
 defined( 'ABSPATH' ) || exit;
 
@@ -57,7 +59,7 @@ class LegacyDataHandler {
 	 * @param array $order_ids If provided, total is computed only among IDs in this array, which can be either individual IDs or ranges like "100-200".
 	 * @return int Number of orders.
 	 */
-	public function count_orders_for_cleanup( $order_ids = array() ) : int {
+	public function count_orders_for_cleanup( $order_ids = array() ): int {
 		global $wpdb;
 		return (int) $wpdb->get_var( $this->build_sql_query_for_cleanup( $order_ids, 'count' ) ); // phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared -- prepared in build_sql_query_for_cleanup().
 	}
@@ -154,11 +156,11 @@ class LegacyDataHandler {
 		$order = wc_get_order( $order_id );
 		if ( ! $order ) {
 			// translators: %d is an order ID.
-			throw new \Exception( sprintf( __( '%d is not a valid order ID.', 'woocommerce' ), $order_id ) );
+			throw new \Exception( esc_html( sprintf( __( '%d is not a valid order ID.', 'woocommerce' ), $order_id ) ) );
 		}
 
 		if ( ! $skip_checks && ! $this->is_order_newer_than_post( $order ) ) {
-			throw new \Exception( sprintf( __( 'Data in posts table appears to be more recent than in HPOS tables.', 'woocommerce' ) ) );
+			throw new \Exception( esc_html( sprintf( __( 'Data in posts table appears to be more recent than in HPOS tables.', 'woocommerce' ) ) ) );
 		}
 
 		// Delete all metadata.
@@ -192,7 +194,7 @@ class LegacyDataHandler {
 	 */
 	private function is_order_newer_than_post( \WC_Abstract_Order $order ): bool {
 		if ( ! is_a( $order->get_data_store()->get_current_class_name(), OrdersTableDataStore::class, true ) ) {
-			throw new \Exception( __( 'Order is not an HPOS order.', 'woocommerce' ) );
+			throw new \Exception( esc_html__( 'Order is not an HPOS order.', 'woocommerce' ) );
 		}
 
 		$post = get_post( $order->get_id() );
@@ -251,7 +253,7 @@ class LegacyDataHandler {
 				$val2 = get_post_meta( $order_id, $key, true );
 			}
 
-			if ( $val1 != $val2 ) { // phpcs:ignore WordPress.PHP.StrictComparisons.LooseComparison
+			if ( $val1 != $val2 ) { // phpcs:ignore WordPress.PHP.StrictComparisons.LooseComparison,Universal.Operators.StrictComparisons.LooseNotEqual
 				$diff[ $key ] = array( $val1, $val2 );
 			}
 		}
@@ -283,7 +285,7 @@ class LegacyDataHandler {
 
 		if ( ! $order_type ) {
 			// translators: %d is an order ID.
-			throw new \Exception( sprintf( __( '%d is not an order or has an invalid order type.', 'woocommerce' ), $order_id ) );
+			throw new \Exception( esc_html( sprintf( __( '%d is not an order or has an invalid order type.', 'woocommerce' ), $order_id ) ) );
 		}
 
 		$classname = $order_type['class_name'];
@@ -321,27 +323,85 @@ class LegacyDataHandler {
 	 * @param int    $order_id               Order ID.
 	 * @param string $source_data_store      Datastore to use as source. Should be either 'hpos' or 'posts'.
 	 * @param string $destination_data_store Datastore to use as destination. Should be either 'hpos' or 'posts'.
+	 * @param array  $fields                 List of metakeys or order properties to limit the backfill to.
 	 * @return void
 	 * @throws \Exception When an error occurs.
 	 */
-	public function backfill_order_to_datastore( int $order_id, string $source_data_store, string $destination_data_store ) {
+	public function backfill_order_to_datastore( int $order_id, string $source_data_store, string $destination_data_store, array $fields = array() ) {
 		$valid_data_stores = array( 'posts', 'hpos' );
 
 		if ( ! in_array( $source_data_store, $valid_data_stores, true ) || ! in_array( $destination_data_store, $valid_data_stores, true ) || $destination_data_store === $source_data_store ) {
-			throw new \Exception( sprintf( 'Invalid datastore arguments: %1$s -> %2$s.', $source_data_store, $destination_data_store ) );
+			throw new \Exception( esc_html( sprintf( 'Invalid datastore arguments: %1$s -> %2$s.', $source_data_store, $destination_data_store ) ) );
 		}
 
-		$order = $this->get_order_from_datastore( $order_id, $source_data_store );
+		$fields    = array_filter( $fields );
+		$src_order = $this->get_order_from_datastore( $order_id, $source_data_store );
 
-		switch ( $destination_data_store ) {
+		// Backfill entire orders.
-			case 'posts':
+		if ( ! $fields ) {
-				$order->get_data_store()->backfill_post_record( $order );
+			if ( 'posts' === $destination_data_store ) {
-				break;
+				$src_order->get_data_store()->backfill_post_record( $src_order );
-			case 'hpos':
+			} elseif ( 'hpos' === $destination_data_store ) {
-				$this->posts_to_cot_migrator->migrate_orders( array( $order_id ) );
+				$this->posts_to_cot_migrator->migrate_orders( array( $src_order->get_id() ) );
-				break;
+			}
-			default:
+
-				break;
+			return;
+		}
+
+		$this->validate_backfill_fields( $fields, $src_order );
+
+		$dest_order = $this->get_order_from_datastore( $src_order->get_id(), $destination_data_store );
+
+		if ( 'posts' === $destination_data_store ) {
+			$datastore = $this->data_store->get_cpt_data_store_instance();
+		} elseif ( 'hpos' === $destination_data_store ) {
+			$datastore = $this->data_store;
+		}
+
+		if ( ! $datastore || ! method_exists( $datastore, 'update_order_from_object' ) ) {
+			throw new \Exception( esc_html__( 'The backup datastore does not support updating orders.', 'woocommerce' ) );
+		}
+
+		// Backfill meta.
+		if ( ! empty( $fields['meta_keys'] ) ) {
+			foreach ( $fields['meta_keys'] as $meta_key ) {
+				$dest_order->delete_meta_data( $meta_key );
+
+				foreach ( $src_order->get_meta( $meta_key, false, 'edit' ) as $meta ) {
+					$dest_order->add_meta_data( $meta_key, $meta->value );
+				}
+			}
+		}
+
+		// Backfill props.
+		if ( ! empty( $fields['props'] ) ) {
+			$new_values = array_combine(
+				$fields['props'],
+				array_map(
+					fn( $prop_name ) => $src_order->{"get_{$prop_name}"}(),
+					$fields['props']
+				)
+			);
+
+			$dest_order->set_props( $new_values );
+
+			if ( 'hpos' === $destination_data_store ) {
+				$dest_order->apply_changes();
+				$limit_cb = function ( $rows, $order ) use ( $dest_order, $fields ) {
+					if ( $dest_order->get_id() === $order->get_id() ) {
+						$rows = $this->limit_hpos_update_to_props( $rows, $fields['props'] );
+					}
+
+					return $rows;
+				};
+				add_filter( 'woocommerce_orders_table_datastore_db_rows_for_order', $limit_cb, 10, 2 );
+			}
+		}
+
+		$datastore->update_order_from_object( $dest_order );
+
+		if ( 'hpos' === $destination_data_store && isset( $limit_cb ) ) {
+			remove_filter( 'woocommerce_orders_table_datastore_db_rows_for_order', $limit_cb );
 		}
 	}
 
@@ -372,13 +432,121 @@ class LegacyDataHandler {
 	 * @return string[] Property names.
 	 */
 	private function get_order_base_props(): array {
-		return array_column(
+		$base_props = array();
-			call_user_func_array(
+
-				'array_merge',
+		foreach ( $this->data_store->get_all_order_column_mappings() as $mapping ) {
-				array_values( $this->data_store->get_all_order_column_mappings() )
+			$base_props = array_merge( $base_props, array_column( $mapping, 'name' ) );
-			),
+		}
-			'name'
+
-		);
+		return $base_props;
 	}
 
+	/**
+	 * Filters a set of HPOS row updates to those matching a specific set of order properties.
+	 * Called via the `woocommerce_orders_table_datastore_db_rows_for_order` filter in `backfill_order_to_datastore`.
+	 *
+	 * @param array    $rows  Details for the db update.
+	 * @param string[] $props Order property names.
+	 * @return array
+	 * @see OrdersTableDataStore::get_db_rows_for_order()
+	 */
+	private function limit_hpos_update_to_props( array $rows, array $props ) {
+		// Determine HPOS columns corresponding to the props in the $props array.
+		$allowed_columns = array();
+		foreach ( $this->data_store->get_all_order_column_mappings() as &$mapping ) {
+			foreach ( $mapping as $column_name => &$column_data ) {
+				if ( ! isset( $column_data['name'] ) || ! in_array( $column_data['name'], $props, true ) ) {
+					continue;
+				}
+
+				$allowed_columns[ $column_data['name'] ] = $column_name;
+			}
+		}
+
+		foreach ( $rows as $i => &$db_update ) {
+			// Prevent accidental update of another prop by limiting columns to explicitly requested props.
+			if ( ! array_intersect_key( $db_update['data'], array_flip( $allowed_columns ) ) ) {
+				unset( $rows[ $i ] );
+				continue;
+			}
+
+			$allowed_column_names_with_ids = array_merge(
+				$allowed_columns,
+				array( 'id', 'order_id', 'address_type' )
+			);
+
+			$db_update['data']   = array_intersect_key( $db_update['data'], array_flip( $allowed_column_names_with_ids ) );
+			$db_update['format'] = array_intersect_key( $db_update['format'], array_flip( $allowed_column_names_with_ids ) );
+		}
+
+		return $rows;
+	}
+
+	/**
+	 * Validates meta_keys and property names for a partial order backfill.
+	 *
+	 * @param array              $fields An array possibly having entries with index 'meta_keys' and/or 'props',
+	 *                                   corresponding to an array of order meta keys and/or order properties.
+	 * @param \WC_Abstract_Order $order  The order being validated.
+	 * @throws \Exception When a validation error occurs.
+	 * @return void
+	 */
+	private function validate_backfill_fields( array $fields, \WC_Abstract_Order $order ) {
+		if ( ! $fields ) {
+			return;
+		}
+
+		if ( ! empty( $fields['meta_keys'] ) ) {
+			$internal_meta_keys = array_unique(
+				array_merge(
+					$this->data_store->get_internal_meta_keys(),
+					$this->data_store->get_cpt_data_store_instance()->get_internal_meta_keys()
+				)
+			);
+
+			$possibly_internal_keys = array_intersect( $internal_meta_keys, $fields['meta_keys'] );
+			if ( ! empty( $possibly_internal_keys ) ) {
+				throw new \Exception(
+					esc_html(
+						sprintf(
+							// translators: %s is a comma separated list of metakey names.
+							_n(
+								'%s is an internal meta key. Use --props to set it.',
+								'%s are internal meta keys. Use --props to set them.',
+								count( $possibly_internal_keys ),
+								'woocommerce'
+							),
+							implode( ', ', $possibly_internal_keys )
+						)
+					)
+				);
+			}
+		}
+
+		if ( ! empty( $fields['props'] ) ) {
+			$invalid_props = array_filter(
+				$fields['props'],
+				function ( $prop_name ) use ( $order ) {
+					return ! method_exists( $order, "get_{$prop_name}" );
+				}
+			);
+
+			if ( ! empty( $invalid_props ) ) {
+				throw new \Exception(
+					esc_html(
+						sprintf(
+							// translators: %s is a list of order property names.
+							_n(
+								'%s is not a valid order property.',
+								'%s are not valid order properties.',
+								count( $invalid_props ),
+								'woocommerce'
+							),
+							implode( ', ', $invalid_props )
+						)
+					)
+				);
+			}
+		}
+	}
 }

plugins/woocommerce/src/Internal/DataStores/Orders/OrdersTableDataStore.php

@@ -531,7 +531,7 @@ class OrdersTableDataStore extends \Abstract_WC_Order_Data_Store_CPT implements
 	 *
 	 * @return string Alias.
 	 */
-	private function get_order_table_alias() : string {
+	private function get_order_table_alias(): string {
 		return 'o';
 	}
 
@@ -540,7 +540,7 @@ class OrdersTableDataStore extends \Abstract_WC_Order_Data_Store_CPT implements
 	 *
 	 * @return string Alias.
 	 */
-	private function get_op_table_alias() : string {
+	private function get_op_table_alias(): string {
 		return 'p';
 	}
 
@@ -551,7 +551,7 @@ class OrdersTableDataStore extends \Abstract_WC_Order_Data_Store_CPT implements
 	 *
 	 * @return string Alias.
 	 */
-	private function get_address_table_alias( string $type ) : string {
+	private function get_address_table_alias( string $type ): string {
 		return 'billing' === $type ? 'b' : 's';
 	}
 
@@ -629,6 +629,41 @@ class OrdersTableDataStore extends \Abstract_WC_Order_Data_Store_CPT implements
 		do_action( 'woocommerce_hpos_post_record_backfilled', $order );
 	}
 
+	/**
+	 * Updates an order (in this datastore) from another order object.
+	 *
+	 * @param \WC_Abstract_Order $order Source order.
+	 * @return bool Whether the order was updated.
+	 */
+	public function update_order_from_object( $order ) {
+		$hpos_order = new \WC_Order();
+		$hpos_order->set_id( $order->get_id() );
+		$this->read( $hpos_order );
+		$hpos_order->set_props( $order->get_data() );
+
+		// Meta keys.
+		foreach ( $hpos_order->get_meta_data() as &$meta ) {
+			$hpos_order->delete_meta_data( $meta->key );
+		}
+
+		foreach ( $order->get_meta_data() as &$meta ) {
+			$hpos_order->add_meta_data( $meta->key, $meta->value );
+		}
+
+		add_filter( 'woocommerce_orders_table_datastore_should_save_after_meta_change', '__return_false' );
+		$hpos_order->save_meta_data();
+		remove_filter( 'woocommerce_orders_table_datastore_should_save_after_meta_change', '__return_false' );
+
+		$db_rows = $this->get_db_rows_for_order( $hpos_order, 'update', true );
+		foreach ( $db_rows as $db_update ) {
+			ksort( $db_update['data'] );
+			ksort( $db_update['format'] );
+			$this->database_util->insert_on_duplicate_key_update( $db_update['table'], $db_update['data'], array_values( $db_update['format'] ) );
+		}
+
+		return true;
+	}
+
 	/**
 	 * Get information about whether permissions are granted yet.
 	 *
@@ -1105,7 +1140,7 @@ WHERE
 	 * @param int $order_id The order id to check.
 	 * @return bool True if an order exists with the given name.
 	 */
-	public function order_exists( $order_id ) : bool {
+	public function order_exists( $order_id ): bool {
 		global $wpdb;
 
 		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared
@@ -1144,7 +1179,7 @@ WHERE
 		$data      = $this->get_order_data_for_ids( $order_ids );
 
 		if ( count( $data ) !== count( $order_ids ) ) {
-			throw new \Exception( __( 'Invalid order IDs in call to read_multiple()', 'woocommerce' ) );
+			throw new \Exception( esc_html__( 'Invalid order IDs in call to read_multiple()', 'woocommerce' ) );
 		}
 
 		$data_synchronizer = wc_get_container()->get( DataSynchronizer::class );
@@ -1188,7 +1223,7 @@ WHERE
 	 *
 	 * @return bool Whether the order should be synced.
 	 */
-	private function should_sync_order( \WC_Abstract_Order $order ) : bool {
+	private function should_sync_order( \WC_Abstract_Order $order ): bool {
 		$draft_order    = in_array( $order->get_status(), array( 'draft', 'auto-draft' ), true );
 		$already_synced = in_array( $order->get_id(), self::$reading_order_ids, true );
 		return ! $draft_order && ! $already_synced;
@@ -1225,7 +1260,7 @@ WHERE
 	 *
 	 * @return array Filtered meta data.
 	 */
-	public function filter_raw_meta_data( &$object, $raw_meta_data ) {
+	public function filter_raw_meta_data( &$object, $raw_meta_data ) { // phpcs:ignore Universal.NamingConventions.NoReservedKeywordParameterNames.objectFound
 		$filtered_meta_data = parent::filter_raw_meta_data( $object, $raw_meta_data );
 		$allowed_keys       = array(
 			'_billing_address_index',
@@ -1233,7 +1268,7 @@ WHERE
 		);
 		$allowed_meta       = array_filter(
 			$raw_meta_data,
-			function( $meta ) use ( $allowed_keys ) {
+			function ( $meta ) use ( $allowed_keys ) {
 				return in_array( $meta->meta_key, $allowed_keys, true );
 			}
 		);
@@ -1802,7 +1837,7 @@ FROM $order_meta_table
 			);
 
 			if ( ! $post_id ) {
-				throw new \Exception( __( 'Could not create order in posts table.', 'woocommerce' ) );
+				throw new \Exception( esc_html__( 'Could not create order in posts table.', 'woocommerce' ) );
 			}
 
 			$order->set_id( $post_id );
@@ -1826,7 +1861,7 @@ FROM $order_meta_table
 
 			if ( false === $result ) {
 				// translators: %s is a table name.
-				throw new \Exception( sprintf( __( 'Could not persist order to database table "%s".', 'woocommerce' ), $update['table'] ) );
+				throw new \Exception( esc_html( sprintf( __( 'Could not persist order to database table "%s".', 'woocommerce' ), $update['table'] ) ) );
 			}
 		}
 
@@ -1999,7 +2034,24 @@ FROM $order_meta_table
 		 */
 		$ext_rows = apply_filters( 'woocommerce_orders_table_datastore_extra_db_rows_for_order', array(), $order, $context );
 
-		return array_merge( $result, $ext_rows );
+		/**
+		 * Filters the rows that are going to be inserted or updated during an order save.
+		 *
+		 * @since 8.8.0
+		 * @internal Use 'woocommerce_orders_table_datastore_extra_db_rows_for_order' for adding rows to the database save.
+		 *
+		 * @param array     $rows    Array of rows to be inserted/updated. See 'woocommerce_orders_table_datastore_extra_db_rows_for_order' for exact format.
+		 * @param \WC_Order $order   The order object.
+		 * @param string    $context The context of the operation: 'create' or 'update'.
+		 */
+		$result = apply_filters(
+			'woocommerce_orders_table_datastore_db_rows_for_order',
+			array_merge( $result, $ext_rows ),
+			$order,
+			$context
+		);
+
+		return $result;
 	}
 
 	/**
@@ -2112,7 +2164,7 @@ FROM $order_meta_table
 				/**
 				 * Fires immediately after an order is deleted.
 				 *
-				 * @since
+				 * @since 2.7.0
 				 *
 				 * @param int $order_id ID of the order that has been deleted.
 				 */
@@ -2137,7 +2189,7 @@ FROM $order_meta_table
 				/**
 				 * Fires immediately after an order is trashed.
 				 *
-				 * @since
+				 * @since 2.7.0
 				 *
 				 * @param int $order_id ID of the order that has been trashed.
 				 */
@@ -2198,7 +2250,7 @@ FROM $order_meta_table
 	 *
 	 * @return void
 	 */
-	private function upshift_or_delete_child_orders( $order ) : void {
+	private function upshift_or_delete_child_orders( $order ): void {
 		global $wpdb;
 
 		$order_table     = self::get_orders_table_name();
@@ -2703,7 +2755,6 @@ FROM $order_meta_table
 		if ( $save ) {
 			$order->save_meta_data();
 		}
-
 	}
 
 	/**
@@ -2871,7 +2922,7 @@ CREATE TABLE $meta_table (
 	 * @param  WC_Data $object WC_Data object.
 	 * @return array
 	 */
-	public function read_meta( &$object ) {
+	public function read_meta( &$object ) { // phpcs:ignore Universal.NamingConventions.NoReservedKeywordParameterNames.objectFound
 		$raw_meta_data = $this->data_store_meta->read_meta( $object );
 		return $this->filter_raw_meta_data( $object, $raw_meta_data );
 	}
@@ -2884,7 +2935,7 @@ CREATE TABLE $meta_table (
 	 *
 	 * @return bool
 	 */
-	public function delete_meta( &$object, $meta ) {
+	public function delete_meta( &$object, $meta ) { // phpcs:ignore Universal.NamingConventions.NoReservedKeywordParameterNames.objectFound
 		global $wpdb;
 
 		if ( $this->should_backfill_post_record() && isset( $meta->id ) ) {
@@ -2932,7 +2983,7 @@ CREATE TABLE $meta_table (
 	 *
 	 * @return int|bool  meta ID or false on failure
 	 */
-	public function add_meta( &$object, $meta ) {
+	public function add_meta( &$object, $meta ) { // phpcs:ignore Universal.NamingConventions.NoReservedKeywordParameterNames.objectFound
 		$add_meta        = $this->data_store_meta->add_meta( $object, $meta );
 		$meta->id        = $add_meta;
 		$changes_applied = $this->after_meta_change( $object, $meta );
@@ -2954,7 +3005,7 @@ CREATE TABLE $meta_table (
 	 *
 	 * @return bool The number of rows updated, or false on error.
 	 */
-	public function update_meta( &$object, $meta ) {
+	public function update_meta( &$object, $meta ) { // phpcs:ignore Universal.NamingConventions.NoReservedKeywordParameterNames.objectFound
 		$update_meta     = $this->data_store_meta->update_meta( $object, $meta );
 		$changes_applied = $this->after_meta_change( $object, $meta );
 
@@ -3007,6 +3058,17 @@ CREATE TABLE $meta_table (
 		$current_time      = $this->legacy_proxy->call_function( 'current_time', 'mysql', 1 );
 		$current_date_time = new \WC_DateTime( $current_time, new \DateTimeZone( 'GMT' ) );
 
-		return $order->get_date_modified() < $current_date_time && empty( $order->get_changes() ) && ( ! is_object( $meta ) || ! in_array( $meta->key, $this->ephemeral_meta_keys, true ) );
+		$should_save =
+			$order->get_date_modified() < $current_date_time && empty( $order->get_changes() )
+			&& ( ! is_object( $meta ) || ! in_array( $meta->key, $this->ephemeral_meta_keys, true ) );
+
+		/**
+		 * Allows code to skip a full order save() when metadata is changed.
+		 *
+		 * @since 8.8.0
+		 *
+		 * @param bool $should_save Whether to trigger a full save after metadata is changed.
+		 */
+		return apply_filters( 'woocommerce_orders_table_datastore_should_save_after_meta_change', $should_save );
 	}
 }

plugins/woocommerce/tests/php/src/Internal/DataStores/Orders/LegacyDataHandlerTests.php

@@ -189,4 +189,70 @@ class LegacyDataHandlerTests extends WC_Unit_Test_Case {
 		$this->assertEquals( $order_hpos->get_meta( 'meta_key' ), $order_cpt->get_meta( 'meta_key' ) );
 	}
 
+	/**
+	 * Checks that partial backfills from/to either datastore work correctly.
+	 *
+	 * @since 8.8.0
+	 *
+	 * @return void
+	 */
+	public function test_datastore_partial_backfill() {
+		// Test order.
+		$this->enable_cot_sync();
+		$order = new \WC_Order();
+		$order->set_status( 'on-hold' );
+		$order->add_meta_data( 'my_meta', 'hpos+posts' );
+		$order->save();
+		$this->disable_cot_sync();
+
+		$order_hpos = $this->sut->get_order_from_datastore( $order->get_id(), 'hpos' );
+		$order_hpos->set_status( 'completed' );
+		$order_hpos->set_billing_first_name( 'Mr. HPOS' );
+		$order_hpos->set_billing_address_1( 'HPOS Street' );
+		$order_hpos->update_meta_data( 'my_meta', 'hpos' );
+		$order_hpos->update_meta_data( 'other_meta', 'hpos' );
+		$order_hpos->save();
+
+		// Fetch the posts version and make sure it's different.
+		$order_cpt = $this->sut->get_order_from_datastore( $order->get_id(), 'posts' );
+		$this->assertNotEquals( $order_hpos->get_billing_first_name(), $order_cpt->get_billing_first_name() );
+		$this->assertNotEquals( $order_hpos->get_status(), $order_cpt->get_status() );
+		$this->assertNotEquals( $order_hpos->get_meta( 'my_meta' ), $order_cpt->get_meta( 'my_meta' ) );
+
+		// Backfill "my_meta" to posts and confirm it has been backfilled.
+		$this->sut->backfill_order_to_datastore( $order->get_id(), 'hpos', 'posts', array( 'meta_keys' => array( 'my_meta' ) ) );
+		$order_cpt = $this->sut->get_order_from_datastore( $order->get_id(), 'posts' );
+		$this->assertNotEquals( $order_hpos->get_billing_first_name(), $order_cpt->get_billing_first_name() );
+		$this->assertNotEquals( $order_hpos->get_status(), $order_cpt->get_status() );
+		$this->assertEquals( $order_hpos->get_meta( 'my_meta' ), $order_cpt->get_meta( 'my_meta' ) );
+
+		// Backfill status and confirm it has been backfilled.
+		$this->sut->backfill_order_to_datastore( $order->get_id(), 'hpos', 'posts', array( 'props' => array( 'status' ) ) );
+		$order_cpt = $this->sut->get_order_from_datastore( $order->get_id(), 'posts' );
+		$this->assertNotEquals( $order_hpos->get_billing_first_name(), $order_cpt->get_billing_first_name() );
+		$this->assertEquals( $order_hpos->get_status(), $order_cpt->get_status() );
+
+		// Update the CPT version now.
+		$order_cpt->set_billing_first_name( 'Mr. Post' );
+		$order_cpt->set_billing_address_1( 'CPT Street' );
+		$order_cpt->save();
+
+		// Re-load the HPOS version and confirm billing first name is different.
+		$order_hpos = $this->sut->get_order_from_datastore( $order->get_id(), 'hpos' );
+		$this->assertNotEquals( $order_hpos->get_billing_first_name(), $order_cpt->get_billing_first_name() );
+
+		// Backfill name and confirm.
+		$this->sut->backfill_order_to_datastore( $order->get_id(), 'posts', 'hpos', array( 'props' => array( 'billing_first_name' ) ) );
+		$order_hpos = $this->sut->get_order_from_datastore( $order->get_id(), 'hpos' );
+		$this->assertEquals( $order_hpos->get_billing_first_name(), $order_cpt->get_billing_first_name() );
+		$this->assertEquals( $order_hpos->get_status(), $order_cpt->get_status() );
+
+		// Re-enable sync, trigger an order sync and confirm that meta/props that we didn't partially backfill already are correctly synced.
+		$this->enable_cot_sync();
+		wc_get_container()->get( DataSynchronizer::class )->process_batch( array( $order_cpt->get_id() ) );
+		$order_cpt = $this->sut->get_order_from_datastore( $order->get_id(), 'posts' );
+		$this->assertEquals( $order_cpt->get_billing_address_1(), $order_hpos->get_billing_address_1() );
+		$this->assertEquals( $order_cpt->get_meta( 'other_meta', true, 'edit' ), $order_hpos->get_meta( 'other_meta', true, 'edit' ) );
+	}
+
 }