diff --git a/addons/calendar/models/calendar_event.py b/addons/calendar/models/calendar_event.py
index c8472e865dc..f5561963e81 100644
--- a/addons/calendar/models/calendar_event.py
+++ b/addons/calendar/models/calendar_event.py
@@ -481,35 +481,37 @@ class Meeting(models.Model):
             return super(Meeting, self.with_context(prefetch_fields=False))._compute_field_value(field)
         return super()._compute_field_value(field)
 
-    def _read(self, fields):
+    def _fetch_query(self, query, fields):
         if self.env.is_system():
-            super()._read(fields)
-            return
+            return super()._fetch_query(query, fields)
 
-        fields = set(fields)
-        private_fields = fields - self._get_public_fields()
+        public_fnames = self._get_public_fields()
+        private_fields = [field for field in fields if field.name not in public_fnames]
         if not private_fields:
-            super()._read(fields)
-            return
+            return super()._fetch_query(query, fields)
 
-        private_fields.add('partner_ids')
-        super()._read(fields | {'privacy', 'user_id', 'partner_ids'})
+        fields_to_fetch = list(fields) + [self._fields[name] for name in ('privacy', 'user_id', 'partner_ids')]
+        events = super()._fetch_query(query, fields_to_fetch)
+
+        # determine private events to which the user does not participate
         current_partner_id = self.env.user.partner_id
-        others_private_events = self.filtered(
+        others_private_events = events.filtered(
             lambda e: e.privacy == 'private' \
                   and e.user_id != self.env.user \
                   and current_partner_id not in e.partner_ids
         )
         if not others_private_events:
-            return
+            return events
 
-        for field_name in private_fields:
-            field = self._fields[field_name]
+        private_fields.append(self._fields['partner_ids'])
+        for field in private_fields:
             replacement = field.convert_to_cache(
-                _('Busy') if field_name == 'name' else False,
+                _('Busy') if field.name == 'name' else False,
                 others_private_events)
             self.env.cache.update(others_private_events, field, repeat(replacement))
 
+        return events
+
     def write(self, values):
         detached_events = self.env['calendar.event']
         recurrence_update_setting = values.pop('recurrence_update', None)
diff --git a/addons/calendar/tests/test_access_rights.py b/addons/calendar/tests/test_access_rights.py
index b36de6c87eb..8eac1328757 100644
--- a/addons/calendar/tests/test_access_rights.py
+++ b/addons/calendar/tests/test_access_rights.py
@@ -93,6 +93,8 @@ class TestAccessRights(TransactionCase):
             privacy='public',
             location='In Hell',
         )
+        # invalidate cache before reading, otherwise read() might leak private data
+        self.env.invalidate_all()
         [private_location, public_location] = self.read_event(self.raoul, private + public, 'location')
         self.assertEqual(private_location, False, "Private value should be obfuscated")
         self.assertEqual(public_location, 'In Hell', "Public value should not be obfuscated")
diff --git a/addons/calendar/tests/test_event_recurrence.py b/addons/calendar/tests/test_event_recurrence.py
index 7ffd752d68e..799a1382edc 100644
--- a/addons/calendar/tests/test_event_recurrence.py
+++ b/addons/calendar/tests/test_event_recurrence.py
@@ -418,7 +418,7 @@ class TestUpdateRecurrentEvents(TestRecurrentEvents):
             'start': event.start + relativedelta(days=4),
             'stop': event.stop + relativedelta(days=5),
         })
-        self.assertFalse(self.recurrence.calendar_event_ids, "Inactive event should not create recurrent events")
+        self.assertFalse(self.recurrence.exists(), "Inactive event should not create recurrent events")
 
     def test_shift_all_with_outlier(self):
         outlier = self.events[1]
diff --git a/addons/crm/models/crm_lead.py b/addons/crm/models/crm_lead.py
index 7bdaa7e4958..58b1ec38fa7 100644
--- a/addons/crm/models/crm_lead.py
+++ b/addons/crm/models/crm_lead.py
@@ -774,13 +774,13 @@ class Lead(models.Model):
         return result
 
     @api.model
-    def search(self, domain, offset=0, limit=None, order=None):
+    def search_fetch(self, domain, field_names, offset=0, limit=None, order=None):
         """ Override to support ordering on my_activity_date_deadline.
 
-        Ordering through web client calls search_read with an order parameter set.
-        Search_read then calls search. In this override we therefore override search
-        to intercept a search without count with an order on my_activity_date_deadline.
-        In that case we do the search in two steps.
+        Ordering through web client calls search_read() with an order parameter
+        set. Method search_read() then calls search_fetch(). Here we override
+        search_fetch() to intercept a search with an order on field
+        my_activity_date_deadline. In that case we do the search in two steps.
 
         First step: fill with deadline-based results
 
@@ -805,8 +805,7 @@ class Lead(models.Model):
         side effects. Search_count is not affected by this override.
         """
         if not order or 'my_activity_date_deadline' not in order:
-            return super().search(domain, offset, limit, order)
-
+            return super().search_fetch(domain, field_names, offset, limit, order)
         order_items = [order_item.strip().lower() for order_item in (order or self._order).split(',')]
 
         # Perform a read_group on my activities to get a mapping lead_id / deadline
@@ -826,7 +825,7 @@ class Lead(models.Model):
 
         # Search leads linked to those activities and order them. See docstring
         # of this method for more details.
-        search_res = super().search(my_lead_domain, order=my_lead_order)
+        search_res = super().search_fetch(my_lead_domain, field_names, order=my_lead_order)
         my_lead_ids_ordered = sorted(search_res.ids, key=lambda lead_id: my_lead_mapping[lead_id], reverse=not activity_asc)
         # keep only requested window (offset + limit, or offset+)
         my_lead_ids_keep = my_lead_ids_ordered[offset:(offset + limit)] if limit else my_lead_ids_ordered[offset:]
@@ -848,9 +847,9 @@ class Lead(models.Model):
             lead_offset = 0
         lead_order = ', '.join(item for item in order_items if 'my_activity_date_deadline' not in item)
 
-        other_lead_res = super().search(
+        other_lead_res = super().search_fetch(
             expression.AND([[('id', 'not in', my_lead_ids_skip)], domain]),
-            lead_offset, lead_limit, lead_order,
+            field_names, lead_offset, lead_limit, lead_order,
         )
         return self.browse(my_lead_ids_keep) + other_lead_res
 
diff --git a/addons/hr/models/hr_employee.py b/addons/hr/models/hr_employee.py
index cf126018dd0..781919ce783 100644
--- a/addons/hr/models/hr_employee.py
+++ b/addons/hr/models/hr_employee.py
@@ -184,17 +184,45 @@ class HrEmployeePrivate(models.Model):
             return super(HrEmployeePrivate, self).name_get()
         return self.env['hr.employee.public'].browse(self.ids).name_get()
 
-    def _read(self, fields):
+    def search_fetch(self, domain, field_names, offset=0, limit=None, order=None):
         if self.check_access_rights('read', raise_exception=False):
-            return super(HrEmployeePrivate, self)._read(fields)
+            return super().search_fetch(domain, field_names, offset, limit, order)
 
         # HACK: retrieve publicly available values from hr.employee.public and
         # copy them to the cache of self; non-public data will be missing from
         # cache, and interpreted as an access error
-        self.flush_recordset(fields)
+        self._check_private_fields(field_names)
+        self.flush_model(field_names)
+        public = self.env['hr.employee.public'].search_fetch(domain, field_names, offset, limit, order)
+        employees = self.browse(public._ids)
+        employees._copy_cache_from(public, field_names)
+        return employees
+
+    def fetch(self, field_names):
+        if self.check_access_rights('read', raise_exception=False):
+            return super().fetch(field_names)
+
+        # HACK: retrieve publicly available values from hr.employee.public and
+        # copy them to the cache of self; non-public data will be missing from
+        # cache, and interpreted as an access error
+        self._check_private_fields(field_names)
+        self.flush_recordset(field_names)
         public = self.env['hr.employee.public'].browse(self._ids)
-        public.read(fields)
-        for fname in fields:
+        public.fetch(field_names)
+        self._copy_cache_from(public, field_names)
+
+    def _check_private_fields(self, field_names):
+        """ Check whether ``field_names`` contain private fields. """
+        public_fields = self.env['hr.employee.public']._fields
+        private_fields = [fname for fname in field_names if fname not in public_fields]
+        if private_fields:
+            raise AccessError(_('The fields "%s" you try to read is not available on the public employee profile.') % (','.join(private_fields)))
+
+    def _copy_cache_from(self, public, field_names):
+        # HACK: retrieve publicly available values from hr.employee.public and
+        # copy them to the cache of self; non-public data will be missing from
+        # cache, and interpreted as an access error
+        for fname in field_names:
             values = self.env.cache.get_values(public, public._fields[fname])
             if self._fields[fname].translate:
                 values = [(value.copy() if value else None) for value in values]
@@ -221,14 +249,6 @@ class HrEmployeePrivate(models.Model):
                     user_id=responsible_user_id)
         employees_scheduled.write({'work_permit_scheduled_activity': True})
 
-    def read(self, fields=None, load='_classic_read'):
-        if self.check_access_rights('read', raise_exception=False):
-            return super(HrEmployeePrivate, self).read(fields, load=load)
-        private_fields = set(fields or self._fields).difference(self.env['hr.employee.public']._fields)
-        if private_fields:
-            raise AccessError(_('The fields "%s" you try to read is not available on the public employee profile.') % (','.join(private_fields)))
-        return self.env['hr.employee.public'].browse(self.ids).read(fields, load=load)
-
     @api.model
     def get_view(self, view_id=None, view_type='form', **options):
         if self.check_access_rights('read', raise_exception=False):
diff --git a/addons/hr_expense/models/hr_expense.py b/addons/hr_expense/models/hr_expense.py
index bdb4654c6a5..f104e70bac0 100644
--- a/addons/hr_expense/models/hr_expense.py
+++ b/addons/hr_expense/models/hr_expense.py
@@ -1025,14 +1025,15 @@ class HrExpenseSheet(models.Model):
     # ORM Overrides
     # ----------------------------------------
 
-    def read(self, fields=None, load='_classic_read'):
+    def _read_format(self, fnames, load='_classic_read'):
         # setting the context in the field on the view is not enough
-        return super(HrExpenseSheet, self.with_context(show_payment_journal_id=True)).read(fields=fields, load=load)
+        self = self.with_context(show_payment_journal_id=True)
+        return super()._read_format(fnames, load)
 
     def onchange(self, values, field_name, field_onchange):
         # setting the context in the field on the view is not enough
-        self_ctx = self.with_context(show_payment_journal_id=True)
-        return super(HrExpenseSheet, self_ctx).onchange(values, field_name, field_onchange)
+        self = self.with_context(show_payment_journal_id=True)
+        return super().onchange(values, field_name, field_onchange)
 
     @api.model_create_multi
     def create(self, vals_list):
diff --git a/addons/hr_holidays/report/hr_leave_report_calendar.py b/addons/hr_holidays/report/hr_leave_report_calendar.py
index 0f26167a9dc..41271119638 100644
--- a/addons/hr_holidays/report/hr_leave_report_calendar.py
+++ b/addons/hr_holidays/report/hr_leave_report_calendar.py
@@ -73,13 +73,14 @@ class LeaveReportCalendar(models.Model):
         );
         """)
 
-    def _read(self, fields):
-        res = super()._read(fields)
+    def _fetch_query(self, query, fields):
+        records = super()._fetch_query(query, fields)
         if self.env.context.get('hide_employee_name') and 'employee_id' in self.env.context.get('group_by', []):
-            name_field = self._fields['name']
-            for record in self.with_user(SUPERUSER_ID):
-                self.env.cache.set(record, name_field, record.name.split(':')[-1].strip())
-        return res
+            self.env.cache.update(records, self._fields['name'], [
+                record.name.split(':')[-1].strip()
+                for record in records.with_user(SUPERUSER_ID)
+            ])
+        return records
 
     @api.model
     def get_unusual_days(self, date_from, date_to=None):
diff --git a/addons/hr_timesheet/tests/test_performance.py b/addons/hr_timesheet/tests/test_performance.py
index 13b3f5275fc..30b27968806 100644
--- a/addons/hr_timesheet/tests/test_performance.py
+++ b/addons/hr_timesheet/tests/test_performance.py
@@ -14,7 +14,7 @@ class TestPerformanceTimesheet(TestCommonTimesheet):
         self.env.invalidate_all()
         projects.clear_caches()
         tasks.clear_caches()
-        with self.assertQueryCount(4):
+        with self.assertQueryCount(5):
             self.env['account.analytic.line']._timesheet_preprocess([
                 {'task_id': task.id} for task in tasks for _i in range(10)
             ])
diff --git a/addons/hr_work_entry_holidays/tests/test_performance.py b/addons/hr_work_entry_holidays/tests/test_performance.py
index 461a907e964..746a1a8f283 100644
--- a/addons/hr_work_entry_holidays/tests/test_performance.py
+++ b/addons/hr_work_entry_holidays/tests/test_performance.py
@@ -32,7 +32,7 @@ class TestWorkEntryHolidaysPerformance(TestWorkEntryHolidaysBase):
         self.richard_emp.generate_work_entries(date(2018, 1, 1), date(2018, 1, 2))
         leave = self.create_leave(datetime(2018, 1, 1, 7, 0), datetime(2018, 1, 1, 18, 0))
 
-        with self.assertQueryCount(__system__=91, admin=95):
+        with self.assertQueryCount(__system__=92, admin=96):
             leave.action_validate()
         leave.action_refuse()
 
@@ -108,11 +108,10 @@ class TestWorkEntryHolidaysPerformancesBigData(TestWorkEntryHolidaysBase):
         cls.leaves.action_approve()
         cls.leaves.action_validate()
 
-
     def test_work_entries_generation_perf(self):
         # Test Case 7: Try to generate work entries for
         # a hundred employees over a month
-        with self.assertQueryCount(__system__=406, admin=406):
+        with self.assertQueryCount(__system__=413):
             work_entries = self.contracts._generate_work_entries(date(2020, 7, 1), date(2020, 8, 31))
 
         # Original work entries to generate when we don't adapt date_generated_from and
diff --git a/addons/project/models/project.py b/addons/project/models/project.py
index f830d226a21..85ade69f5b7 100644
--- a/addons/project/models/project.py
+++ b/addons/project/models/project.py
@@ -1683,10 +1683,10 @@ class Task(models.Model):
             can normally see.
             (In other words, this compute is only used in project sharing views to see all assignees for each task)
         """
-        if self.ids:
+        if self._origin:
             # fetch 'user_ids' in superuser mode (and override value in cache
             # browse is useful to avoid miscache because of the newIds contained in self
-            self.browse(self.ids)._read(['user_ids'])
+            self._origin.fetch(['user_ids'])
         for task in self.with_context(prefetch_fields=False):
             task.portal_user_names = ', '.join(task.user_ids.mapped('name'))
 
diff --git a/addons/purchase/models/purchase.py b/addons/purchase/models/purchase.py
index 7a020d05478..a19c0513228 100644
--- a/addons/purchase/models/purchase.py
+++ b/addons/purchase/models/purchase.py
@@ -655,7 +655,7 @@ class PurchaseOrder(models.Model):
             # invoices related to the purchase order, we read them in sudo to fill the
             # cache.
             self.invalidate_model(['invoice_ids'])
-            self.sudo()._read(['invoice_ids'])
+            self.sudo().fetch(['invoice_ids'])
             invoices = self.invoice_ids
 
         result = self.env['ir.actions.act_window']._for_xml_id('account.action_move_in_invoice_type')
diff --git a/odoo/addons/base/models/ir_attachment.py b/odoo/addons/base/models/ir_attachment.py
index edd7ccae63b..a64145e3046 100644
--- a/odoo/addons/base/models/ir_attachment.py
+++ b/odoo/addons/base/models/ir_attachment.py
@@ -603,10 +603,6 @@ class IrAttachment(models.Model):
 
         return self.browse(result)._as_query(order)
 
-    def _read(self, fields):
-        self.check('read')
-        return super(IrAttachment, self)._read(fields)
-
     def write(self, vals):
         self.check('write', values=vals)
         # remove computed field depending of datas
diff --git a/odoo/addons/base/models/res_users.py b/odoo/addons/base/models/res_users.py
index b04267093c6..fcb5eceb6ba 100644
--- a/odoo/addons/base/models/res_users.py
+++ b/odoo/addons/base/models/res_users.py
@@ -458,19 +458,14 @@ class Users(models.Model):
     def onchange_parent_id(self):
         return self.partner_id.onchange_parent_id()
 
-    def _read(self, fields):
-        super(Users, self)._read(fields)
-        if set(USER_PRIVATE_FIELDS).intersection(fields):
+    def _fetch_query(self, query, fields):
+        records = super()._fetch_query(query, fields)
+        if not set(USER_PRIVATE_FIELDS).isdisjoint(field.name for field in fields):
             if self.check_access_rights('write', raise_exception=False):
-                return
-            for record in self:
-                for f in USER_PRIVATE_FIELDS:
-                    try:
-                        record._cache[f]
-                        record._cache[f] = '********'
-                    except Exception:
-                        # skip SpecialValue (e.g. for missing record or access right)
-                        pass
+                return records
+            for fname in USER_PRIVATE_FIELDS:
+                self.env.cache.update(records, self._fields[fname], repeat('********'))
+        return records
 
     @api.constrains('company_id', 'company_ids', 'active')
     def _check_company(self):
diff --git a/odoo/addons/test_access_rights/models.py b/odoo/addons/test_access_rights/models.py
index 770f1142e6f..2139edb606d 100644
--- a/odoo/addons/test_access_rights/models.py
+++ b/odoo/addons/test_access_rights/models.py
@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 # Part of Odoo. See LICENSE file for full copyright and licensing details.
 
-from odoo import fields, models
+from odoo import api, fields, models
 
 class SomeObj(models.Model):
     _name = 'test_access_right.some_obj'
@@ -43,10 +43,11 @@ class ObjCateg(models.Model):
 
     name = fields.Char(required=True)
 
-    def search(self, domain, *args, **kwargs):
+    @api.model
+    def search_fetch(self, domain, field_names, offset=0, limit=None, order=None):
         if self.env.context.get('only_media'):
             domain += [('name', '=', 'Media')]
-        return super(ObjCateg, self).search(domain, **kwargs)
+        return super().search_fetch(domain, field_names, offset, limit, order)
 
 
 class FakeTicket(models.Model):
diff --git a/odoo/addons/test_new_api/models/test_new_api.py b/odoo/addons/test_new_api/models/test_new_api.py
index 20713b64c9a..c9be9f79cb4 100644
--- a/odoo/addons/test_new_api/models/test_new_api.py
+++ b/odoo/addons/test_new_api/models/test_new_api.py
@@ -69,13 +69,13 @@ class Category(models.Model):
             # assign name of last category, and reassign display_name (to normalize it)
             cat.name = names[-1].strip()
 
-    def _read(self, fields):
+    def _fetch_query(self, query, fields):
         # DLE P45: `test_31_prefetch`,
         # with self.assertRaises(AccessError):
         #     cat1.name
         if self.search_count([('id', 'in', self._ids), ('name', '=', 'NOACCESS')]):
             raise AccessError('Sorry')
-        return super(Category, self)._read(fields)
+        return super()._fetch_query(query, fields)
 
 
 class Discussion(models.Model):
diff --git a/odoo/addons/test_new_api/tests/test_new_fields.py b/odoo/addons/test_new_api/tests/test_new_fields.py
index 5b4f0131393..8b560e4f06b 100644
--- a/odoo/addons/test_new_api/tests/test_new_fields.py
+++ b/odoo/addons/test_new_api/tests/test_new_fields.py
@@ -2609,16 +2609,21 @@ class TestFields(TransactionCaseWithUserDemo):
 
         # only one query as admin: reading pivot table
         with self.assertQueryCount(1):
+            # trick: if value is in cache, read() does not make any query
+            record.invalidate_recordset(['tags'])
             record.read(['tags'])
 
         user = self.env['res.users'].create({'name': "user", 'login': "user"})
         record_user = record.with_user(user)
 
         # prep the following query count by caching access check related data
+        record_user.invalidate_recordset(['tags'])
         record_user.read(['tags'])
 
         # only one query as user: reading pivot table
         with self.assertQueryCount(1):
+            # trick: if value is in cache, read() does not make any query
+            record_user.invalidate_recordset(['tags'])
             record_user.read(['tags'])
 
         # create a passing ir.rule
@@ -2628,11 +2633,14 @@ class TestFields(TransactionCaseWithUserDemo):
         })
 
         # prep the following query count by caching access check related data
+        record_user.invalidate_recordset(['tags'])
         record_user.read(['tags'])
 
         # still only 1 query: reading pivot table
         # access rules are checked in python in this case
         with self.assertQueryCount(1):
+            # trick: if value is in cache, read() does not make any query
+            record_user.invalidate_recordset(['tags'])
             record_user.read(['tags'])
 
         # create a blocking ir.rule
@@ -2683,28 +2691,27 @@ class TestFields(TransactionCaseWithUserDemo):
         self.env.invalidate_all()
 
         with self.assertQueries(["""
-            SELECT
-                "test_new_api_prefetch"."id" AS "id",
-                "test_new_api_prefetch"."name"->>'en_US' AS "name",
-                "test_new_api_prefetch"."description"->>'en_US' AS "description",
-                "test_new_api_prefetch"."html_description"->>'en_US' AS "html_description",
-                "test_new_api_prefetch"."create_uid" AS "create_uid",
-                "test_new_api_prefetch"."create_date" AS "create_date",
-                "test_new_api_prefetch"."write_uid" AS "write_uid",
-                "test_new_api_prefetch"."write_date" AS "write_date"
+            SELECT "test_new_api_prefetch"."id",
+                   "test_new_api_prefetch"."name"->>'en_US',
+                   "test_new_api_prefetch"."description"->>'en_US',
+                   "test_new_api_prefetch"."html_description"->>'en_US',
+                   "test_new_api_prefetch"."create_uid",
+                   "test_new_api_prefetch"."create_date",
+                   "test_new_api_prefetch"."write_uid",
+                   "test_new_api_prefetch"."write_date"
             FROM "test_new_api_prefetch"
-            WHERE "test_new_api_prefetch".id IN %s
+            WHERE ("test_new_api_prefetch"."id" IN %s)
         """]):
             records.mapped('name')  # fetch all fields with prefetch=True
 
         with self.assertQueries(["""
             SELECT
-                "test_new_api_prefetch"."id" AS "id",
-                "test_new_api_prefetch"."harry" AS "harry",
-                "test_new_api_prefetch"."hermione" AS "hermione",
-                "test_new_api_prefetch"."ron" AS "ron"
+                "test_new_api_prefetch"."id",
+                "test_new_api_prefetch"."harry",
+                "test_new_api_prefetch"."hermione",
+                "test_new_api_prefetch"."ron"
             FROM "test_new_api_prefetch"
-            WHERE "test_new_api_prefetch".id IN %s
+            WHERE ("test_new_api_prefetch"."id" IN %s)
         """]):
             records.mapped('harry')  # fetch all fields with prefetch='Harry Potter'
             records.mapped('hermione')  # fetched already
@@ -2712,11 +2719,11 @@ class TestFields(TransactionCaseWithUserDemo):
 
         with self.assertQueries(["""
             SELECT
-                "test_new_api_prefetch"."id" AS "id",
-                "test_new_api_prefetch"."hansel" AS "hansel",
-                "test_new_api_prefetch"."gretel" AS "gretel"
+                "test_new_api_prefetch"."id",
+                "test_new_api_prefetch"."hansel",
+                "test_new_api_prefetch"."gretel"
             FROM "test_new_api_prefetch"
-            WHERE "test_new_api_prefetch".id IN %s
+            WHERE ("test_new_api_prefetch"."id" IN %s)
         """]):
             records.mapped('hansel')  # fetch all fields with prefetch='Hansel and Gretel'
             records.mapped('gretel')  # fetched already
@@ -3760,10 +3767,10 @@ def select(model, *fnames):
     """ Return the expected query string to SELECT the given columns. """
     table = model._table
     terms = ", ".join(
-        f'"{table}"."{fname}" AS "{fname}"'
+        f'"{table}"."{fname}"'
         for fname in ['id'] + list(fnames)
     )
-    return f'SELECT {terms} FROM "{table}" WHERE "{table}".id IN %s'
+    return f'SELECT {terms} FROM "{table}" WHERE ("{table}"."id" IN %s)'
 
 
 def insert(model, *fnames, rowcount=1):
@@ -4282,14 +4289,20 @@ class TestModifiedPerformance(common.TransactionCase):
 
         self.modified_line_a_child.price
         with self.assertQueries(["""
-        SELECT "test_new_api_modified_line"."id" AS "id", "test_new_api_modified_line"."modified_id" AS "modified_id",
-               "test_new_api_modified_line"."quantity" AS "quantity", "test_new_api_modified_line"."price" AS "price",
-               "test_new_api_modified_line"."parent_id" AS "parent_id", "test_new_api_modified_line"."create_uid" AS "create_uid",
-               "test_new_api_modified_line"."create_date" AS "create_date", "test_new_api_modified_line"."write_uid" AS "write_uid",
-               "test_new_api_modified_line"."write_date" AS "write_date"
-         FROM "test_new_api_modified_line"
-        WHERE "test_new_api_modified_line".id IN %s
-        """] * 2, flush=False):
+            SELECT "test_new_api_modified_line"."id",
+                   "test_new_api_modified_line"."modified_id",
+                   "test_new_api_modified_line"."quantity",
+                   "test_new_api_modified_line"."parent_id",
+                   "test_new_api_modified_line"."create_uid",
+                   "test_new_api_modified_line"."create_date"
+            FROM "test_new_api_modified_line"
+            WHERE ("test_new_api_modified_line"."id" IN %s)
+        """, """
+            SELECT "test_new_api_modified_line"."id",
+                   "test_new_api_modified_line"."parent_id"
+            FROM "test_new_api_modified_line"
+            WHERE ("test_new_api_modified_line"."id" IN %s)
+        """], flush=False):
             # Two requests:
             # - one for fetch modified_line_a_child_child data (invalidate just before)
             # - one because modified_line_a_child.parent_id (invalidate just before because we invalidate inverse in `_invalidate_cache`,
diff --git a/odoo/addons/test_new_api/tests/test_properties.py b/odoo/addons/test_new_api/tests/test_properties.py
index 60c40adea5c..0bd1bc53507 100644
--- a/odoo/addons/test_new_api/tests/test_properties.py
+++ b/odoo/addons/test_new_api/tests/test_properties.py
@@ -174,53 +174,53 @@ class PropertiesCase(TransactionCase):
 
         expected_queries = [
             # read the properties field value
-            ''' SELECT "test_new_api_message"."id" AS "id",
-                       "test_new_api_message"."attributes" AS "attributes"
+            ''' SELECT "test_new_api_message"."id",
+                       "test_new_api_message"."attributes"
                 FROM "test_new_api_message"
-                WHERE "test_new_api_message".id IN %s
+                WHERE ("test_new_api_message"."id" IN %s)
             ''',
-            ''' SELECT "test_new_api_message"."id" AS "id",
-                       "test_new_api_message"."discussion" AS "discussion",
-                       "test_new_api_message"."body" AS "body",
-                       "test_new_api_message"."author" AS "author",
-                       "test_new_api_message"."name" AS "name",
-                       "test_new_api_message"."important" AS "important",
-                       "test_new_api_message"."label"->>'en_US' AS "label",
-                       "test_new_api_message"."priority" AS "priority",
-                       "test_new_api_message"."create_uid" AS "create_uid",
-                       "test_new_api_message"."create_date" AS "create_date",
-                       "test_new_api_message"."write_uid" AS "write_uid",
-                       "test_new_api_message"."write_date" AS "write_date"
+            ''' SELECT "test_new_api_message"."id",
+                       "test_new_api_message"."discussion",
+                       "test_new_api_message"."body",
+                       "test_new_api_message"."author",
+                       "test_new_api_message"."name",
+                       "test_new_api_message"."important",
+                       "test_new_api_message"."label"->>'en_US',
+                       "test_new_api_message"."priority",
+                       "test_new_api_message"."create_uid",
+                       "test_new_api_message"."create_date",
+                       "test_new_api_message"."write_uid",
+                       "test_new_api_message"."write_date"
                 FROM "test_new_api_message"
-                WHERE "test_new_api_message".id IN %s
+                WHERE ("test_new_api_message"."id" IN %s)
             ''',
             # read the definition on the definition record
-            ''' SELECT "test_new_api_discussion"."id" AS "id",
-                       "test_new_api_discussion"."name" AS "name",
-                       "test_new_api_discussion"."moderator" AS "moderator",
-                       "test_new_api_discussion"."message_concat" AS "message_concat",
-                       "test_new_api_discussion"."history" AS "history",
-                       "test_new_api_discussion"."attributes_definition" AS "attributes_definition",
-                       "test_new_api_discussion"."create_uid" AS "create_uid",
-                       "test_new_api_discussion"."create_date" AS "create_date",
-                       "test_new_api_discussion"."write_uid" AS "write_uid",
-                       "test_new_api_discussion"."write_date" AS "write_date"
+            ''' SELECT "test_new_api_discussion"."id",
+                       "test_new_api_discussion"."name",
+                       "test_new_api_discussion"."moderator",
+                       "test_new_api_discussion"."message_concat",
+                       "test_new_api_discussion"."history",
+                       "test_new_api_discussion"."attributes_definition",
+                       "test_new_api_discussion"."create_uid",
+                       "test_new_api_discussion"."create_date",
+                       "test_new_api_discussion"."write_uid",
+                       "test_new_api_discussion"."write_date"
                 FROM "test_new_api_discussion"
-                WHERE "test_new_api_discussion".id IN %s
+                WHERE ("test_new_api_discussion"."id" IN %s)
             ''',
             # check the many2one existence
             ''' SELECT "test_new_api_partner"."id"
                 FROM "test_new_api_partner"
                 WHERE "test_new_api_partner".id IN %s
             ''',
-            ''' SELECT "test_new_api_partner"."id" AS "id",
-                       "test_new_api_partner"."name" AS "name",
-                       "test_new_api_partner"."create_uid" AS "create_uid",
-                       "test_new_api_partner"."create_date" AS "create_date",
-                       "test_new_api_partner"."write_uid" AS "write_uid",
-                       "test_new_api_partner"."write_date" AS "write_date"
+            ''' SELECT "test_new_api_partner"."id",
+                       "test_new_api_partner"."name",
+                       "test_new_api_partner"."create_uid",
+                       "test_new_api_partner"."create_date",
+                       "test_new_api_partner"."write_uid",
+                       "test_new_api_partner"."write_date"
                 FROM "test_new_api_partner"
-                WHERE "test_new_api_partner".id IN %s
+                WHERE ("test_new_api_partner"."id" IN %s)
             ''',
         ]
 
diff --git a/odoo/addons/test_performance/tests/test_performance.py b/odoo/addons/test_performance/tests/test_performance.py
index ed4f1b91e58..047166efb08 100644
--- a/odoo/addons/test_performance/tests/test_performance.py
+++ b/odoo/addons/test_performance/tests/test_performance.py
@@ -79,7 +79,7 @@ class TestPerformance(SavepointCaseWithUserDemo):
         records.write({'line_ids': [Command.create({})]})
         self.env.invalidate_all()
 
-        with self.assertQueryCount(2):
+        with self.assertQueryCount(1):
             records.line_ids
 
     @users('__system__', 'demo')
@@ -115,6 +115,51 @@ class TestPerformance(SavepointCaseWithUserDemo):
             for record in records:
                 self.assertEqual(record.with_context(key=3).value_ctx, 3)
 
+    def test_fetch(self):
+        """ Fetch only when necessary. """
+        records = self.env['test_performance.base'].search([])
+        self.assertEqual(len(records), 5)
+
+        with self.assertQueryCount(1):
+            # not in cache yet
+            records.fetch(['name', 'partner_id'])
+
+        with self.assertQueryCount(0):
+            # already in cache
+            records.mapped('name')
+            records.mapped('partner_id')
+
+        with self.assertQueryCount(1):
+            # this one shouldn't have been fetched
+            records.mapped('value')
+
+        with self.assertQueryCount(0):
+            # 'name' and 'value' are already in cache
+            records.fetch(['name', 'value'])
+
+    @warmup
+    def test_search_fetch(self):
+        """ Search and fetch all at once. """
+        records = self.env['test_performance.base'].search([])
+        self.assertEqual(len(records), 5)
+
+        with self.assertQueryCount(2):
+            self.env.invalidate_all()
+            for record in records.search([]):
+                record.partner_id
+
+        # search() can do everything in a single query!
+        with self.assertQueryCount(1):
+            self.env.invalidate_all()
+            for record in records.search_fetch([], ['partner_id']):
+                record.partner_id
+
+        # the case where you don't fetch the right field
+        with self.assertQueryCount(2):
+            self.env.invalidate_all()
+            for record in records.search_fetch([], ['value_pc']):
+                record.partner_id
+
     @warmup
     def test_search_read(self):
         """ Search and fetch all at once. """
@@ -122,18 +167,18 @@ class TestPerformance(SavepointCaseWithUserDemo):
         records = Model.search([])
         self.assertEqual(len(records), 5)
 
-        # one query for search, one query for read, one query for display_name
+        # one query for search and read, one query for display_name
         expected = records.read(['partner_id', 'value_pc'])
-        with self.assertQueryCount(3):
+        with self.assertQueryCount(2):
             self.env.invalidate_all()
             self.assertEqual(
                 Model.search_read([], ['partner_id', 'value_pc']),
                 expected,
             )
 
-        # one query for search, one query for read
+        # one query for search and read
         expected = records.read(['partner_id', 'value_pc'], load=False)
-        with self.assertQueryCount(2):
+        with self.assertQueryCount(1):
             self.env.invalidate_all()
             self.assertEqual(
                 Model.search_read([], ['partner_id', 'value_pc'], load=False),
@@ -260,7 +305,7 @@ class TestPerformance(SavepointCaseWithUserDemo):
         lines = rec1.line_ids
 
         # set N lines in rec2: O(1) queries
-        with self.assertQueryCount(8):
+        with self.assertQueryCount(7):
             self.env.invalidate_all()
             rec2.write({'line_ids': [Command.set(lines[0].ids)]})
         self.assertEqual(rec1.line_ids, lines[1:])
@@ -498,18 +543,18 @@ class TestPerformance(SavepointCaseWithUserDemo):
 
         # fetching 'name' prefetches all fields on all records
         queries = [
-            ''' SELECT "test_performance_base"."id" AS "id",
-                       "test_performance_base"."name" AS "name",
-                       "test_performance_base"."value" AS "value",
-                       "test_performance_base"."value_pc" AS "value_pc",
-                       "test_performance_base"."partner_id" AS "partner_id",
-                       "test_performance_base"."total" AS "total",
-                       "test_performance_base"."create_uid" AS "create_uid",
-                       "test_performance_base"."create_date" AS "create_date",
-                       "test_performance_base"."write_uid" AS "write_uid",
-                       "test_performance_base"."write_date" AS "write_date"
+            ''' SELECT "test_performance_base"."id",
+                       "test_performance_base"."name",
+                       "test_performance_base"."value",
+                       "test_performance_base"."value_pc",
+                       "test_performance_base"."partner_id",
+                       "test_performance_base"."total",
+                       "test_performance_base"."create_uid",
+                       "test_performance_base"."create_date",
+                       "test_performance_base"."write_uid",
+                       "test_performance_base"."write_date"
                 FROM "test_performance_base"
-                WHERE "test_performance_base".id IN %s
+                WHERE ("test_performance_base"."id" IN %s)
             ''',
         ]
         with self.assertQueries(queries, flush=False):
@@ -648,7 +693,7 @@ class TestMapped(TransactionCase):
         self.env.invalidate_all()
 
         # expected same performance as recs.line_ids.mapped('value')
-        with self.assertQueryCount(3):
+        with self.assertQueryCount(2):
             for rec in recs:
                 rec.line_ids.mapped('value')
 
diff --git a/odoo/fields.py b/odoo/fields.py
index 7fc05f92f27..f9e76c739c2 100644
--- a/odoo/fields.py
+++ b/odoo/fields.py
@@ -4293,20 +4293,18 @@ class One2many(_RelationalMulti):
         comodel = records.env[self.comodel_name].with_context(**context)
         inverse = self.inverse_name
         inverse_field = comodel._fields[inverse]
-        domain = self.get_domain_list(records) + [(inverse, 'in', records.ids)]
-        lines = comodel.search(domain)
 
-        if len(records) == 1:
-            # optimization: all lines have the same value for 'inverse_field',
-            # so we don't need to fetch it from database
-            records.env.cache.insert_missing(records, self, [lines._ids])
-            records.env.cache.insert_missing(lines, inverse_field, itertools.repeat(records.id))
-            return
+        # optimization: fetch the inverse and active fields with search()
+        domain = self.get_domain_list(records) + [(inverse, 'in', records.ids)]
+        field_names = [inverse]
+        if comodel._active_name:
+            field_names.append(comodel._active_name)
+        lines = comodel.search_fetch(domain, field_names)
 
         # group lines by inverse field (without prefetching other fields)
         get_id = (lambda rec: rec.id) if inverse_field.type == 'many2one' else int
         group = defaultdict(list)
-        for line in lines.with_context(prefetch_fields=False):
+        for line in lines:
             # line[inverse] may be a record or an integer
             group[get_id(line[inverse])].append(line.id)
 
diff --git a/odoo/models.py b/odoo/models.py
index 14c0d6bd1a8..a91e57d7d45 100644
--- a/odoo/models.py
+++ b/odoo/models.py
@@ -168,6 +168,8 @@ class MetaModel(api.Meta):
 
         if '__init__' in attrs and len(inspect.signature(attrs['__init__']).parameters) != 4:
             _logger.warning("The method %s.__init__ doesn't match the new signature in module %s", name, attrs.get('__module__'))
+        if callable(attrs.get('_read')):
+            warnings.warn(f"{self.__module__}.{self.__name__}: method BaseModel._read() has been replaced by BaseModel._fetch_query()")
 
         if not attrs.get('_register', True):
             return
@@ -1468,6 +1470,9 @@ class BaseModel(metaclass=MetaModel):
         :param domain: :ref:`A search domain <reference/orm/domains>`. Use an empty
                      list to match all records.
         :param limit: maximum number of record to count (upperbound) (default: all)
+
+        This is a high-level method, which should not be overridden. Its actual
+        implementation is done by method :meth:`_search`.
         """
         query = self._search(domain, limit=limit)
         return len(query)
@@ -1477,7 +1482,7 @@ class BaseModel(metaclass=MetaModel):
     def search(self, domain, offset=0, limit=None, order=None):
         """ search(domain[, offset=0][, limit=None][, order=None])
 
-        Searches for records based on the ``domain``
+        Search for the records that satisfy the given ``domain``
         :ref:`search domain <reference/orm/domains>`.
 
         :param domain: :ref:`A search domain <reference/orm/domains>`. Use an empty
@@ -1487,9 +1492,57 @@ class BaseModel(metaclass=MetaModel):
         :param str order: sort string
         :returns: at most ``limit`` records matching the search criteria
         :raise AccessError: if user is not allowed to access requested information
+
+        This is a high-level method, which should not be overridden. Its actual
+        implementation is done by method :meth:`_search`.
         """
+        return self.search_fetch(domain, [], offset=offset, limit=limit, order=order)
+
+    @api.model
+    @api.returns('self')
+    def search_fetch(self, domain, field_names, offset=0, limit=None, order=None):
+        """ search_fetch(domain, field_names[, offset=0][, limit=None][, order=None])
+
+        Search for the records that satisfy the given ``domain``
+        :ref:`search domain <reference/orm/domains>`, and fetch the given fields
+        to the cache.  This method is like a combination of methods :meth:`search`
+        and :meth:`fetch`, but it performs both tasks with a minimal number of
+        SQL queries.
+
+        :param domain: :ref:`A search domain <reference/orm/domains>`. Use an empty
+                     list to match all records.
+        :param field_names: a collection of field names to fetch
+        :param int offset: number of results to ignore (default: none)
+        :param int limit: maximum number of records to return (default: all)
+        :param str order: sort string
+        :returns: at most ``limit`` records matching the search criteria
+        :raise AccessError: if user is not allowed to access requested information
+        """
+        # first determine a query that satisfies the domain and access rules
         query = self._search(domain, offset=offset, limit=limit, order=order or self._order)
-        return self.browse(query)
+
+        if query.is_empty():
+            # optimization: don't execute the query at all
+            return self.browse()
+
+        # determine fields to fetch
+        fields_to_fetch = OrderedSet()
+        if field_names:
+            field_names = self.check_field_access_rights('read', field_names)
+        for field_name in field_names:
+            field = self._fields.get(field_name)
+            if not field:
+                raise ValueError(f"Invalid field {field_name!r} on model {self._name!r}")
+            if field.store:
+                fields_to_fetch.add(field)
+            elif field.compute:
+                # optimization: fetch direct field dependencies
+                for dotname in self.pool.field_depends[field]:
+                    dep = self._fields[dotname.split('.', 1)[0]]
+                    if dep.prefetch is True and (not dep.groups or self.user_has_groups(dep.groups)):
+                        fields_to_fetch.add(dep)
+
+        return self._fetch_query(query, fields_to_fetch)
 
     #
     # display_name, name_get, name_create, name_search
@@ -2935,8 +2988,8 @@ class BaseModel(metaclass=MetaModel):
     def read(self, fields=None, load='_classic_read'):
         """ read([fields])
 
-        Reads the requested fields for the records in ``self``, low-level/RPC
-        method.
+        Read the requested fields for the records in ``self``, and return their
+        values as a list of dicts.
 
         :param list fields: field names to return (default is all fields)
         :param str load: loading mode, currently the only option is to set to
@@ -2946,25 +2999,13 @@ class BaseModel(metaclass=MetaModel):
         :rtype: list
         :raise AccessError: if user is not allowed to access requested information
         :raise ValueError: if a requested field does not exist
+
+        This is a high-level method that is not supposed to be overridden. In
+        order to modify how fields are read from database, see methods
+        :meth:`_fetch_query` and :meth:`_read_format`.
         """
         fields = self.check_field_access_rights('read', fields)
-
-        # fetch stored fields from the database to the cache
-        stored_fields = OrderedSet()
-        for name in fields:
-            field = self._fields.get(name)
-            if not field:
-                raise ValueError("Invalid field %r on model %r" % (name, self._name))
-            if field.store:
-                stored_fields.add(name)
-            elif field.compute:
-                # optimization: prefetch direct field dependencies
-                for dotname in self.pool.field_depends[field]:
-                    f = self._fields[dotname.split('.')[0]]
-                    if f.prefetch is True and (not f.groups or self.user_has_groups(f.groups)):
-                        stored_fields.add(f.name)
-        self._read(stored_fields)
-
+        self.fetch(fields)
         return self._read_format(fnames=fields, load=load)
 
     def update_field_translations(self, field_name, translations):
@@ -3100,7 +3141,8 @@ class BaseModel(metaclass=MetaModel):
         """Returns a list of dictionaries mapping field names to their values,
         with one dictionary per record that exists.
 
-        The output format is similar to the one expected from the `read` method.
+        The output format is the one expected from the `read` method, which uses
+        this method as its implementation for formatting values.
 
         The current method is different from `read` because it retrieves its
         values from the cache without doing a query when it is avoidable.
@@ -3140,103 +3182,127 @@ class BaseModel(metaclass=MetaModel):
                 fnames.append(field.name)
         else:
             fnames = [field.name]
-        self._read(fnames)
+        self.fetch(fnames)
 
-    def _read(self, field_names):
-        """ Read the given fields of the records in ``self`` from the database,
-            and store them in cache. Skip fields that are not stored.
+    def fetch(self, field_names):
+        """ Make sure the given fields are in memory for the records in ``self``,
+        by fetching what is necessary from the database.  Non-stored fields are
+        mostly ignored, except for their stored dependencies. This method should
+        be called to optimize code.
 
-            :param field_names: list of field names to read
+        This method is implemented thanks to methods :meth:`_search` and
+        :meth:`_fetch_query`, and should not be overridden.
         """
-        if not self:
+        if not self or not field_names:
             return
-        self.check_access_rights('read')
 
-        # determine columns fields and those with their own read() method
-        column_fields = []
-        other_fields = []
-        translated_field_names = []
-        for name in field_names:
-            if name == 'id':
-                continue
-            field = self._fields.get(name)
+        # determine fields to fetch
+        fields_to_fetch = OrderedSet()
+        cache = self.env.cache
+        field_names = self.check_field_access_rights('read', field_names)
+        for field_name in field_names:
+            field = self._fields.get(field_name)
             if not field:
-                _logger.warning("%s._read() with unknown field %r", self._name, name)
+                raise ValueError(f"Invalid field {field_name!r} on model {self._name!r}")
+            if not any(cache.get_missing_ids(self, field)):
                 continue
-            if field.base_field.store and field.base_field.column_type:
-                column_fields.append(field)
-            elif field.store and not field.column_type:
-                # non-column fields: for the sake of simplicity, we ignore inherited fields
-                other_fields.append(field)
-            if field.store and field.translate:
-                translated_field_names.append(field.name)
+            if field.store:
+                fields_to_fetch.add(field)
+            elif field.compute:
+                # optimization: fetch direct field dependencies
+                for dotname in self.pool.field_depends[field]:
+                    dep = self._fields[dotname.split('.', 1)[0]]
+                    if dep.prefetch is True and (not dep.groups or self.user_has_groups(dep.groups)):
+                        fields_to_fetch.add(dep)
 
-            if field.type == 'properties':
-                # force calling fields.read for properties field because
-                # we want to read all relational properties in batch
-                # (and check their existence in batch as well)
-                other_fields.append(field)
-
-        if column_fields:
-            cr, context = self.env.cr, self.env.context
-
-            # If a read() follows a write(), we must flush the updates that have
-            # an impact on checking security rules, as they are injected into
-            # the query.  However, we don't need to flush the fields to fetch,
-            # as explained below when putting values in cache.
-
-            # Since only one language translation is fetched from database,
-            # we must flush these translated fields before read
-            # E.g. in database, the {'en_US': 'English'},
-            # write record.with_context(lang='en_US').name = 'English2'
-            # then record.with_context(lang='fr_FR').name => cache miss => _read
-            # 'English2'should is flushed before query as it is the fallback of empty 'fr_FR'
-            if translated_field_names:
-                self.flush_recordset(translated_field_names)
-            self._flush_search([], order='id')
-
-            # make a query object for selecting ids, and apply security rules to it
-            query = Query(cr, self._table, self._table_query)
-            self._apply_ir_rules(query, 'read')
-
-            # the query may involve several tables: we need fully-qualified names
-            def qualify(field):
-                qname = self._inherits_join_calc(self._table, field.name, query)
-                if field.type == 'binary' and (
-                        context.get('bin_size') or context.get('bin_size_' + field.name)):
-                    # PG 9.2 introduces conflicting pg_size_pretty(numeric) -> need ::cast
-                    qname = f'pg_size_pretty(length({qname})::bigint)'
-                return f'{qname} AS "{field.name}"'
-
-            # selected fields are: 'id' followed by column_fields
-            qual_names = [qualify(field) for field in [self._fields['id']] + column_fields]
-
-            # determine the actual query to execute (last parameter is added below)
-            query.add_where(f'"{self._table}".id IN %s')
-            query_str, params = query.select(*qual_names)
-
-            result = []
-            for sub_ids in cr.split_for_in_conditions(self.ids):
-                cr.execute(query_str, params + [sub_ids])
-                result += cr.fetchall()
-        else:
+        if not fields_to_fetch:
+            # there is nothing to fetch, but we expect an error anyway in case
+            # self is not accessible
+            self.check_access_rights('read')
             try:
                 self.check_access_rule('read')
             except MissingError:
-                # Method _read() should never raise a MissingError, but method
+                # Method fetch() should never raise a MissingError, but method
+                # check_access_rule() can, because it must read fields on self.
+                # So we restrict 'self' to existing records (to avoid an extra
+                # exists() at the end of the method).
+                self.exists().check_access_rule('read')
+            return
+
+        # first determine a query that satisfies the domain and access rules
+        if any(field.column_type for field in fields_to_fetch):
+            query = self.with_context(active_test=False)._search([('id', 'in', self.ids)])
+        else:
+            self.check_access_rights('read')
+            try:
+                self.check_access_rule('read')
+            except MissingError:
+                # Method fetch() should never raise a MissingError, but method
                 # check_access_rule() can, because it must read fields on self.
                 # So we restrict 'self' to existing records (to avoid an extra
                 # exists() at the end of the method).
                 self = self.exists()
                 self.check_access_rule('read')
+            query = self._as_query(ordered=False)
 
-            result = [(id_,) for id_ in self.ids]
+        # fetch the fields
+        fetched = self._fetch_query(query, fields_to_fetch)
 
-        fetched = self.browse()
-        if result:
-            # result = [(id1, a1, b1), (id2, a2, b2), ...]
+        # possibly raise exception for the records that could not be read
+        if fetched != self:
+            forbidden = (self - fetched).exists()
+            if forbidden:
+                raise self.env['ir.rule']._make_access_error('read', forbidden)
+
+    def _fetch_query(self, query, fields):
+        """ Fetch the given fields (iterable of :class:`Field` instances) from
+        the given query, put them in cache, and return the fetched records.
+
+        This method may be overridden to change what fields to actually fetch,
+        or to change the values that are put in cache.
+        """
+        # determine columns fields and those with their own read() method
+        column_fields = OrderedSet()
+        other_fields = OrderedSet()
+        for field in fields:
+            if field.name == 'id':
+                continue
+            assert field.store
+            (column_fields if field.column_type else other_fields).add(field)
+            if field.type == 'properties':
+                # force calling fields.read for properties field in order to
+                # read all relational properties in batch
+                other_fields.add(field)
+
+        # necessary to retrieve the en_US value of fields without a translation
+        translated_field_names = [field.name for field in column_fields if field.translate]
+        if translated_field_names:
+            self.flush_model(translated_field_names)
+
+        context = self.env.context
+
+        if column_fields:
+            # the query may involve several tables: we need fully-qualified names
+            select_terms = [f'"{self._table}"."id"']
+            for field in column_fields:
+                qname = self._inherits_join_calc(self._table, field.name, query)
+                if field.type == 'binary' and (
+                        context.get('bin_size') or context.get('bin_size_' + field.name)):
+                    # PG 9.2 introduces conflicting pg_size_pretty(numeric) -> need ::cast
+                    qname = f'pg_size_pretty(length({qname})::bigint)'
+                select_terms.append(qname)
+
+            # select the given columns from the rows in the query
+            query_str, params = query.select(*select_terms)
+            self.env.cr.execute(query_str, params)
+            rows = self.env.cr.fetchall()
+
+            if not rows:
+                return self.browse()
+
+            # rows = [(id1, a1, b1), (id2, a2, b2), ...]
             # column_values = [(id1, id2, ...), (a1, a2, ...), (b1, b2, ...)]
-            column_values = zip(*result)
+            column_values = zip(*rows)
             ids = next(column_values)
             fetched = self.browse(ids)
 
@@ -3248,25 +3314,15 @@ class BaseModel(metaclass=MetaModel):
                 # store values in cache, but without overwriting
                 self.env.cache.insert_missing(fetched, field, values)
 
-            # process non-column fields
+        else:
+            fetched = self.browse(query)
+
+        # process non-column fields
+        if fetched:
             for field in other_fields:
                 field.read(fetched)
 
-        # possibly raise exception for the records that could not be read
-        missing = self - fetched
-        if missing:
-            extras = fetched - self
-            if extras:
-                raise AccessError(_(
-                    "Database fetch misses ids (%(missing)s) and has extra ids (%(extra)s),"
-                    " may be caused by a type incoherence in a previous request",
-                    missing=missing._ids,
-                    extra=extras._ids,
-                ))
-            # mark non-existing records in missing
-            forbidden = missing.exists()
-            if forbidden:
-                raise self.env['ir.rule']._make_access_error('read', forbidden)
+        return fetched
 
     def get_metadata(self):
         """Return some metadata about the given records.
@@ -4632,6 +4688,12 @@ class BaseModel(metaclass=MetaModel):
         :param access_rights_uid: optional user ID to use when checking access rights
                                   (not for ir.rules, this is only for ir.model.access)
         :return: a :class:`Query` object that represents the matching records
+
+        This method may be overridden to modify the domain being searched, or to
+        do some post-filtering of the resulting query object. Be careful with
+        the latter option, though, as it might hurt performance. Indeed, by
+        default the returned query object is not actually executed, and it can
+        be injected as a value in a domain in order to generate sub-queries.
         """
         model = self.with_user(access_rights_uid) if access_rights_uid else self
         model.check_access_rights('read')
@@ -4937,7 +4999,7 @@ class BaseModel(metaclass=MetaModel):
 
     @api.model
     def search_read(self, domain=None, fields=None, offset=0, limit=None, order=None, **read_kwargs):
-        """Perform a :meth:`search` followed by a :meth:`read`.
+        """ Perform a :meth:`search_fetch` followed by a :meth:`_read_format`.
 
         :param domain: Search domain, see ``args`` parameter in :meth:`search`.
             Defaults to an empty domain that will match all records.
@@ -4955,30 +5017,19 @@ class BaseModel(metaclass=MetaModel):
         :return: List of dictionaries containing the asked fields.
         :rtype: list(dict).
         """
-        records = self.search(domain or [], offset=offset, limit=limit, order=order)
-        if not records:
-            return []
+        fields = self.check_field_access_rights('read', fields)
+        records = self.search_fetch(domain or [], fields, offset=offset, limit=limit, order=order)
 
-        if fields and fields == ['id']:
-            # shortcut read if we only want the ids
-            return [{'id': record.id} for record in records]
-
-        # read() ignores active_test, but it would forward it to any downstream search call
-        # (e.g. for x2m or function fields), and this is not the desired behavior, the flag
-        # was presumably only meant for the main search().
-        # TODO: Move this to read() directly?
+        # Method _read_format() ignores 'active_test', but it would forward it
+        # to any downstream search call(e.g. for x2m or computed fields), and
+        # this is not the desired behavior. The flag was presumably only meant
+        # for the main search().
         if 'active_test' in self._context:
             context = dict(self._context)
             del context['active_test']
             records = records.with_context(context)
 
-        result = records.read(fields, **read_kwargs)
-        if len(result) <= 1:
-            return result
-
-        # reorder read
-        index = {vals['id']: vals for vals in result}
-        return [index[record.id] for record in records if record.id in index]
+        return records._read_format(fnames=fields, **read_kwargs)
 
     def toggle_active(self):
         "Inverses the value of :attr:`active` on the records in ``self``."
@@ -6321,7 +6372,7 @@ class BaseModel(metaclass=MetaModel):
                 fnames = [subname
                           for subname in subnames
                           if lines._fields[subname].base_field.store]
-                lines._read(fnames)
+                lines.fetch(fnames)
                 # copy the cache of lines to their corresponding new records;
                 # this avoids computing computed stored fields on new_lines
                 new_lines = lines.browse(map(NewId, line_ids))
diff --git a/odoo/tools/query.py b/odoo/tools/query.py
index 0e7e13c6d1d..00f1db65037 100644
--- a/odoo/tools/query.py
+++ b/odoo/tools/query.py
@@ -193,6 +193,10 @@ class Query(object):
         )
         return query_str, params
 
+    def is_empty(self):
+        """ Return whether the query is known to return nothing. """
+        return self._ids == ()
+
     def get_sql(self):
         """ Returns (query_from, query_where, query_params). """
         tables = [_from_table(table, alias) for alias, table in self._tables.items()]