upsert is using read replica for select

[Arcus16]

Describe the bug
When using em.upsert() with read replicas and preferReadReplicas: true. The select query that follows the insert query uses the read connections. Probably due to replication lag the hydration fails. I tried setting preferReadReplicas: false and it works as expected.
Stack trace

[Nest] 71051  - 10/23/2023, 3:47:33 PM     LOG [DatabaseModule] [query] insert into `users` (`email`, `reference_uuid`) values ('jolane@example.com', 'fed6019f-1e39-4341-9282-b3647e318e1c') on duplicate key update `reference_uuid` = values(`reference_uuid`) [took 11 ms] (via write connection 'localhost')
[Nest] 71051  - 10/23/2023, 3:47:33 PM     LOG [DatabaseModule] [query] select `u0`.`id`, `u0`.`first_name`, `u0`.`last_name`, `u0`.`updated_at`, `u0`.`created_at` from `users` as `u0` where `u0`.`email` = 'jolane@example.com' limit 1 [took 5 ms, 0 result] (via read connection 'read-1')
  console.error
    JIT runtime error: Cannot read properties of null (reading 'id')
    
      function(entity, data, factory, newEntity, convertCustomTypes, schema) {
    >   if (typeof data.id !== 'undefined') entity.id = data.id;
                       ^
        if (typeof data.firstName !== 'undefined') {
          if (convertCustomTypes) {
            const value = convertToJSValue_firstName(data.firstName);
            entity.firstName = value;
          } else {
            entity.firstName = data.firstName;
          }
        }
        if (typeof data.lastName !== 'undefined') {
          if (convertCustomTypes) {
            const value = convertToJSValue_lastName(data.lastName);
            entity.lastName = value;
          } else {
            entity.lastName = data.lastName;
          }
        }
        if (typeof data.email !== 'undefined') entity.email = data.email;
        const createCollectionItem_accounts = (value, entity) => {
          if (isPrimaryKey(value, false)) return factory.createReference('Account', value, { convertCustomTypes, schema, merge: true });
          if (value && value.__entity) return value;
          return factory.create('Account', value, { newEntity, convertCustomTypes, schema, merge: true });
        }
        if (data.accounts && !Array.isArray(data.accounts) && typeof data.accounts === 'object') {
          data.accounts = [data.accounts];
        }
        if (Array.isArray(data.accounts)) {
          const items = data.accounts.map(value => createCollectionItem_accounts(value, entity));
          const coll = Collection.create(entity, 'accounts', items, newEntity);
          if (newEntity) {
            coll.setDirty();
          } else {
            coll.takeSnapshot(true);
          }
        } else if (!entity.accounts && data.accounts instanceof Collection) {
          entity.accounts = data.accounts;
        } else if (!entity.accounts) {
          const coll = Collection.create(entity, 'accounts', undefined, !!data.accounts || newEntity);
          coll.setDirty(false);
        }
        if (typeof data.updatedAt !== 'undefined') {
          if (convertCustomTypes) {
            const value = convertToJSValue_updatedAt(data.updatedAt);
            entity.updatedAt = value;
          } else {
            entity.updatedAt = data.updatedAt;
          }
        }
        if (typeof data.createdAt !== 'undefined') {
          if (convertCustomTypes) {
            const value = convertToJSValue_createdAt(data.createdAt);
            entity.createdAt = value;
          } else {
            entity.createdAt = data.createdAt;
          }
        }
        if (typeof data.referenceUuid !== 'undefined') {
          if (convertCustomTypes) {
            const value = convertToJSValue_referenceUuid(data.referenceUuid);
            data.referenceUuid = convertToDatabaseValue_referenceUuid(value);
            entity.referenceUuid = value;
          } else {
            entity.referenceUuid = data.referenceUuid;
          }
        }
      }

      67 |         if (!user) {
    > 68 |             user = await this.em.upsert(User, { ...omit(['id'], params), referenceUuid: uuidv4() });
         |                    ^
      70 |         }
      71 |

      at Function.callCompiledFunction (node_modules/@mikro-orm/core/utils/Utils.js:840:25)
      at ObjectHydrator.hydrate (node_modules/@mikro-orm/core/hydration/ObjectHydrator.js:27:23)
      at SqlEntityManager.upsert (node_modules/@mikro-orm/core/EntityManager.js:519:30)

Expected behavior
em.upsert will always use a write connection regardless of preferReadReplicas flag.

Versions

Dependency	Version
node	20.8.0
typescript	5.2.2
mikro-orm	5.8.10
@mikro-orm/mysql	5.8.10

[Arcus16]

I tried to debug it, and it works as expected when I add. connectionType: 'write' here, but I am not sure if this is the right fix.

[B4nan]

Yes, that place looks good, also the same will need to happen for upsertMany:

mikro-orm/packages/core/src/EntityManager.ts

Lines 915 to 919 in 22e140e

	const data2 = await this.driver.find(meta.className, where, {
	fields: returning.concat(...add).concat(...uniqueFields as string[]),
	ctx: em.transactionContext,
	convertCustomTypes: true,
	});

[Arcus16]

Ok, I will prepare PR with the fix, but right now, I am unsure how to write a test for this (I never saw Mikro-orm tests), so I will probably need help with that.

[B4nan]

You could just add them in here, and assert the connection type the same way as the other tests (via logs):

https://github.com/mikro-orm/mikro-orm/blob/master/tests/features/read-replicas.test.ts

[Arcus16]

I opened PR with fix