Repository: opendatacube/datacube-core (language: python)

Issue:
Get rid of common_timestamp function in 1.9.x
The `common_timestamp` sql function is no longer required as it can be more efficiently replaced with the postgresql double colon (`::`) type-cast operator.  This is not supported by SQLAlchemy 1.4, and so cannot be done in the 1.8 branch, but should be relatively easy to factor out in 1.9

(See https://github.com/opendatacube/datacube-explorer/pull/595)



Discussion / hints:

Fix patch:
diff --git a/datacube/drivers/postgis/_api.py b/datacube/drivers/postgis/_api.py
index 43be61f4e..59d07cc19 100644
--- a/datacube/drivers/postgis/_api.py
+++ b/datacube/drivers/postgis/_api.py
@@ -24,7 +24,6 @@
 from sqlalchemy import select, text, and_, or_, func
 from sqlalchemy.dialects.postgresql import INTERVAL
 from sqlalchemy.exc import IntegrityError
-from sqlalchemy.engine import Row
 
 from typing import Iterable, Sequence, Optional, Set, Any
 from typing import cast as type_cast
@@ -50,22 +49,40 @@
 
 # Make a function because it's broken
 def _dataset_select_fields() -> tuple:
+    return tuple(f.alchemy_expression for f in _dataset_fields())
+
+
+def _dataset_fields() -> tuple:
+    native_flds = get_native_fields()
     return (
-        Dataset,
-        # All active URIs, from newest to oldest
-        func.array(
-            select(
-                SelectedDatasetLocation.uri
-            ).where(
-                and_(
-                    SelectedDatasetLocation.dataset_ref == Dataset.id,
-                    SelectedDatasetLocation.archived == None
-                )
-            ).order_by(
-                SelectedDatasetLocation.added.desc(),
-                SelectedDatasetLocation.id.desc()
-            ).label('uris')
-        ).label('uris')
+        native_flds["id"],
+        native_flds["indexed_time"],
+        native_flds["indexed_by"],
+        native_flds["product_id"],
+        native_flds["metadata_type_id"],
+        native_flds["metadata_doc"],
+        NativeField(
+            'archived',
+            'Archived date',
+            Dataset.archived
+        ),
+        NativeField("uris",
+                    "all uris",
+                    func.array(
+                        select(
+                            SelectedDatasetLocation.uri
+                        ).where(
+                            and_(
+                                SelectedDatasetLocation.dataset_ref == Dataset.id,
+                                SelectedDatasetLocation.archived == None
+                            )
+                        ).order_by(
+                            SelectedDatasetLocation.added.desc(),
+                            SelectedDatasetLocation.id.desc()
+                        ).label('uris')
+                    ),
+                    alchemy_table=Dataset.__table__  # type: ignore[attr-defined]
+        )
     )
 
 
@@ -230,6 +247,29 @@ def extract_dataset_fields(ds_metadata, fields):
     return result
 
 
+# Min/Max aggregating time fields for temporal_extent methods
+time_min = DateDocField('acquisition_time_min',
+                        'Min of time when dataset was acquired',
+                        Dataset.metadata_doc,
+                        False,  # is it indexed
+                        offset=[
+                            ['properties', 'dtr:start_datetime'],
+                            ['properties', 'datetime']
+                        ],
+                        selection='least')
+
+
+time_max = DateDocField('acquisition_time_max',
+                        'Max of time when dataset was acquired',
+                        Dataset.metadata_doc,
+                        False,  # is it indexed
+                        offset=[
+                            ['properties', 'dtr:end_datetime'],
+                            ['properties', 'datetime']
+                        ],
+                        selection='greatest')
+
+
 class PostgisDbAPI:
     def __init__(self, parentdb, connection):
         self._db = parentdb
@@ -476,19 +516,6 @@ def all_dataset_ids(self, archived: bool | None = False):
             )
         return self._connection.execute(query).fetchall()
 
-    # Not currently implemented.
-    # def insert_dataset_source(self, classifier, dataset_id, source_dataset_id):
-        # r = self._connection.execute(
-        #     insert(DatasetSource).on_conflict_do_nothing(
-        #         index_elements=['classifier', 'dataset_ref']
-        #     ).values(
-        #         classifier=classifier,
-        #         dataset_ref=dataset_id,
-        #         source_dataset_ref=source_dataset_id
-        #     )
-        # )
-        # return r.rowcount > 0
-
     def archive_dataset(self, dataset_id):
         r = self._connection.execute(
             update(Dataset).where(
@@ -548,10 +575,10 @@ def get_datasets(self, dataset_ids):
         ).fetchall()
 
     def get_derived_datasets(self, dataset_id):
-        raise NotImplementedError
+        raise NotImplementedError()
 
     def get_dataset_sources(self, dataset_id):
-        raise NotImplementedError
+        raise NotImplementedError()
 
     def search_datasets_by_metadata(self, metadata, archived):
         """
@@ -621,14 +648,13 @@ def search_datasets_query(self,
         assert source_exprs is None
         assert not with_source_ids
 
-        if select_fields:
-            select_columns = tuple(
-                f.alchemy_expression.label(f.name)
-                for f in select_fields
-            )
-        else:
-            select_columns = _dataset_select_fields()
+        if not select_fields:
+            select_fields = _dataset_fields()
 
+        select_columns = tuple(
+            f.alchemy_expression.label(f.name)
+            for f in select_fields
+        )
         if geom:
             SpatialIndex, spatialquery = self.geospatial_query(geom)
         else:
@@ -663,12 +689,21 @@ def search_datasets(self, expressions,
         :type with_source_ids: bool
         :type select_fields: tuple[datacube.drivers.postgis._fields.PgField]
         :type expressions: tuple[datacube.drivers.postgis._fields.PgExpression]
+
+        :return: An iterable of tuples of decoded values
         """
+        if select_fields is None:
+            select_fields = _dataset_fields()
         select_query = self.search_datasets_query(expressions, source_exprs,
                                                   select_fields, with_source_ids,
                                                   limit, geom=geom, archived=archived)
         _LOG.debug("search_datasets SQL: %s", str(select_query))
-        return self._connection.execute(select_query)
+
+        def decode_row(raw: Iterable[Any]) -> dict[str, Any]:
+            return {f.name: f.normalise_value(r) for r, f in zip(raw, select_fields)}
+
+        for row in self._connection.execute(select_query):
+            yield decode_row(row)
 
     def bulk_simple_dataset_search(self, products=None, batch_size=0):
         """
@@ -690,7 +725,7 @@ def bulk_simple_dataset_search(self, products=None, batch_size=0):
         query = select(
             *_dataset_bulk_select_fields()
         ).select_from(Dataset).where(
-            Dataset.archived == None
+            Dataset.archived.is_(None)
         )
         if products:
             query = query.where(Dataset.product_ref.in_(products))
@@ -733,10 +768,12 @@ def insert_lineage_bulk(self, values):
         )
         return res.rowcount, requested - res.rowcount
 
-    def get_duplicates(self, match_fields: Sequence[PgField], expressions: Sequence[PgExpression]) -> Iterable[Row]:
+    def get_duplicates(self,
+                       match_fields: Sequence[PgField],
+                       expressions: Sequence[PgExpression]) -> Iterable[dict[str, Any]]:
         # TODO
         if "time" in [f.name for f in match_fields]:
-            return self.get_duplicates_with_time(match_fields, expressions)
+            yield from self.get_duplicates_with_time(match_fields, expressions)
 
         group_expressions = tuple(f.alchemy_expression for f in match_fields)
         join_tables = PostgisDbAPI._join_tables(expressions, match_fields)
@@ -749,27 +786,31 @@ def get_duplicates(self, match_fields: Sequence[PgField], expressions: Sequence[
             query = query.join(*joins)
 
         query = query.where(
-            and_(Dataset.archived == None, *(PostgisDbAPI._alchemify_expressions(expressions)))
+            and_(Dataset.archived.is_(None), *(PostgisDbAPI._alchemify_expressions(expressions)))
         ).group_by(
             *group_expressions
         ).having(
             func.count(Dataset.id) > 1
         )
-        return self._connection.execute(query)
+        for row in self._connection.execute(query):
+            drow = {"ids": row.ids}
+            for f in match_fields:
+                drow[f.name] = getattr(row, f.name)
+            yield drow
 
     def get_duplicates_with_time(
             self, match_fields: Sequence[PgField], expressions: Sequence[PgExpression]
-    ) -> Iterable[Row]:
+    ) -> Iterable[dict[str, Any]]:
         fields = []
-        for f in match_fields:
-            if f.name == "time":
-                time_field = type_cast(DateRangeDocField, f).expression_with_leniency
+        for fld in match_fields:
+            if fld.name == "time":
+                time_field = type_cast(DateRangeDocField, fld)
             else:
-                fields.append(f.alchemy_expression)
+                fields.append(fld.alchemy_expression)
 
         join_tables = PostgisDbAPI._join_tables(expressions, match_fields)
 
-        cols = [Dataset.id, time_field.label('time'), *fields]
+        cols = [Dataset.id, time_field.expression_with_leniency.label('time'), *fields]
         query = select(
             *cols
         ).select_from(Dataset)
@@ -777,7 +818,7 @@ def get_duplicates_with_time(
             query = query.join(*joins)
 
         query = query.where(
-            and_(Dataset.archived == None, *(PostgisDbAPI._alchemify_expressions(expressions)))
+            and_(Dataset.archived.is_(None), *(PostgisDbAPI._alchemify_expressions(expressions)))
         )
 
         t1 = query.alias("t1")
@@ -785,7 +826,7 @@ def get_duplicates_with_time(
 
         time_overlap = select(
             t1.c.id,
-            text("t1.time * t2.time as time_intersect"),
+            t1.c.time.intersection(t2.c.time).label('time_intersect'),
             *fields
         ).select_from(
             t1.join(
@@ -797,7 +838,7 @@ def get_duplicates_with_time(
         query = select(
             func.array_agg(func.distinct(time_overlap.c.id)).label("ids"),
             *fields,  # type: ignore[arg-type]
-            text("(lower(time_intersect) at time zone 'UTC', upper(time_intersect) at time zone 'UTC') as time")
+            text("time_intersect as time")
         ).select_from(
             time_overlap  # type: ignore[arg-type]
         ).group_by(
@@ -805,7 +846,16 @@ def get_duplicates_with_time(
         ).having(
             func.count(time_overlap.c.id) > 1
         )
-        return self._connection.execute(query)
+
+        for row in self._connection.execute(query):
+            # TODO: Use decode_rows above - would require creating a field class for the ids array.
+            drow: dict[str, Any] = {
+                "ids": row.ids,
+            }
+            for f in fields:
+                drow[f.key] = getattr(row, f.key)  # type: ignore[union-attr]
+            drow["time"] = time_field.normalise_value((row.time.lower, row.time.upper))
+            yield drow
 
     def count_datasets(self, expressions, archived: bool | None = False, geom: Geometry | None = None):
         """
@@ -1474,33 +1524,20 @@ def remove_lineage_relations(self,
     def temporal_extent_by_prod(self, product_id: int) -> tuple[datetime.datetime, datetime.datetime]:
         query = self.temporal_extent_full().where(Dataset.product_ref == product_id)
         res = self._connection.execute(query)
-        return res.first()
+        for tmin, tmax in res:
+            return (self.time_min.normalise_value(tmin), self.time_max.normalise_value(tmax))
+        raise RuntimeError("Product has no datasets and therefore no temporal extent")
 
     def temporal_extent_by_ids(self, ids: Iterable[DSID]) -> tuple[datetime.datetime, datetime.datetime]:
         query = self.temporal_extent_full().where(Dataset.id.in_(ids))
         res = self._connection.execute(query)
-        return res.first()
+        for tmin, tmax in res:
+            return (self.time_min.normalise_value(tmin), self.time_max.normalise_value(tmax))
+        raise ValueError("no dataset ids provided")
 
     def temporal_extent_full(self) -> Select:
         # Hardcode eo3 standard time locations - do not use this approach in a legacy index driver.
-        time_min = DateDocField('aquisition_time_min',
-                                'Min of time when dataset was acquired',
-                                Dataset.metadata_doc,
-                                False,  # is it indexed
-                                offset=[
-                                    ['properties', 'dtr:start_datetime'],
-                                    ['properties', 'datetime']
-                                ],
-                                selection='least')
-        time_max = DateDocField('aquisition_time_max',
-                                'Max of time when dataset was acquired',
-                                Dataset.metadata_doc,
-                                False,  # is it indexed
-                                offset=[
-                                    ['properties', 'dtr:end_datetime'],
-                                    ['properties', 'datetime']
-                                ],
-                                selection='greatest')
+
         return select(
-            func.min(time_min.alchemy_expression), func.max(time_max.alchemy_expression)
+            func.min(self.time_min.alchemy_expression), func.max(self.time_max.alchemy_expression)
         )
diff --git a/datacube/drivers/postgis/_core.py b/datacube/drivers/postgis/_core.py
index 0c95e7578..a105c471e 100644
--- a/datacube/drivers/postgis/_core.py
+++ b/datacube/drivers/postgis/_core.py
@@ -128,7 +128,6 @@ def ensure_db(engine, with_permissions=True):
             c.execute(text(f"""
             grant usage on schema {SCHEMA_NAME} to odc_user;
             grant select on all tables in schema {SCHEMA_NAME} to odc_user;
-            grant execute on function {SCHEMA_NAME}.common_timestamp(text) to odc_user;
 
             grant insert on {SCHEMA_NAME}.dataset,
                             {SCHEMA_NAME}.location,
diff --git a/datacube/drivers/postgis/_fields.py b/datacube/drivers/postgis/_fields.py
index 35c1eb59f..bfb1154fc 100755
--- a/datacube/drivers/postgis/_fields.py
+++ b/datacube/drivers/postgis/_fields.py
@@ -8,14 +8,16 @@
 import math
 
 from collections import namedtuple
+from datetime import timezone
 from datetime import datetime, date
 from decimal import Decimal
-from typing import Any, Callable, Tuple, Union
+from typing import Any, Callable, Type, Tuple, Union
 
 from psycopg2.extras import NumericRange, DateTimeTZRange
+from sqlalchemy.types import TIMESTAMP
 from sqlalchemy import cast, func, and_
 from sqlalchemy.dialects import postgresql as postgres
-from sqlalchemy.dialects.postgresql import NUMRANGE, TSTZRANGE
+from sqlalchemy.dialects.postgresql import NUMRANGE, TSTZRANGE, Range as PgRange
 from sqlalchemy.dialects.postgresql import INTERVAL
 from sqlalchemy.sql import ColumnElement, FromClause, ColumnExpressionArgument
 from sqlalchemy.orm import aliased
@@ -27,7 +29,7 @@
 
 from datacube.drivers.postgis._schema import Dataset, search_field_index_map
 from datacube.utils import cached_property
-from datacube.utils.dates import tz_aware
+from datacube.utils.dates import tz_as_utc
 
 DatasetJoinArgs = tuple[FromClause] | tuple[FromClause, ColumnExpressionArgument]
 
@@ -41,17 +43,30 @@ class PgField(Field):
     def __init__(self,
                  name: str, description: str,
                  alchemy_column: ColumnElement,
-                 indexed: bool):
+                 indexed: bool,
+                 alchemy_table: FromClause | None = None):
         super().__init__(name, description)
 
         # The underlying SQLAlchemy column. (eg. DATASET.c.metadata)
         self.alchemy_column = alchemy_column
         self.indexed = indexed
+        self.alchemy_table = alchemy_table
 
     @property
     def select_alchemy_table(self) -> FromClause:
+        if self.alchemy_table is not None:
+            return self.alchemy_table
         return self.alchemy_column.table
 
+    def normalise_value(self, value):
+        """
+        Wrap the given value with any necessary type casts/conversions for this field.
+
+        Overridden by other classes as needed.
+        """
+        # Default do nothing (eg. string datatypes)
+        return value
+
     @cached_property
     def search_index_table(self) -> FromClause:
         if self.indexed:
@@ -122,9 +137,10 @@ def __init__(self, name: str, description: str,
                  alchemy_column: ColumnElement,
                  alchemy_expression: ColumnExpressionArgument | None = None,
                  join_clause: ColumnExpressionArgument | None = None,
+                 alchemy_table: FromClause | None = None,
                  # Should this be selected by default when selecting all fields?
                  affects_row_selection: bool = False):
-        super(NativeField, self).__init__(name, description, alchemy_column, False)
+        super(NativeField, self).__init__(name, description, alchemy_column, indexed=False, alchemy_table=alchemy_table)
         self._expression = alchemy_expression
         self.affects_row_selection = affects_row_selection
         self.join_clause = join_clause
@@ -171,7 +187,8 @@ def parse_value(self, value):
 
     def _alchemy_offset_value(self,
                               doc_offsets: Tuple[Tuple[str]],
-                              agg_function: Callable[[Any], ColumnElement]) -> ColumnElement:
+                              agg_function: Callable[[Any], ColumnElement],
+                              type_: Type | None = None) -> ColumnElement:
         """
         Get an sqlalchemy value for the given offsets of this field's sqlalchemy column.
         If there are multiple they will be combined using the given aggregate function.
@@ -192,6 +209,8 @@ def _alchemy_offset_value(self,
             doc_offsets = [doc_offsets]
 
         alchemy_values = [self.value_to_alchemy(self.alchemy_column[offset].astext) for offset in doc_offsets]
+        if type_ is not None:
+            alchemy_values = [cast(v, type_) for v in alchemy_values]
         # If there's multiple fields, we aggregate them (eg. "min()"). Otherwise use the one.
         return agg_function(*alchemy_values) if len(alchemy_values) > 1 else alchemy_values[0]
 
@@ -312,14 +331,25 @@ def value_to_alchemy(self, value: DateFieldLike) -> DateFieldLike:
         Wrap a value as needed for this field type.
         """
         if isinstance(value, datetime):
-            return tz_aware(value)
-        # SQLAlchemy expression or string are parsed in pg as dates.
+            return self.normalise_value(value)
+        elif isinstance(value, str):
+            return tz_as_utc(datetime.fromisoformat(value))
         elif isinstance(value, (ColumnElement, str)):
-            return func.odc.common_timestamp(value)
+            # SQLAlchemy expression or string are parsed in pg as dates.
+            # NB: Do not cast here - casting here breaks expected behaviour in other timezones
+            return value
         else:
             raise ValueError("Value not readable as date: %r" % value)
 
+    def normalise_value(self, value):
+        if isinstance(value, str):
+            value = datetime.fromisoformat(value)
+        return tz_as_utc(value)
+
     def search_value_to_alchemy(self, value):
+        if isinstance(value, datetime):
+            value = tz_as_utc(value)
+        value = cast(value, TIMESTAMP(timezone=True))
         return func.tstzrange(
             value, value,
             # Inclusive on both sides.
@@ -340,9 +370,13 @@ def day(self):
             '{}_day'.format(self.name),
             'Day of {}'.format(self.description),
             self.alchemy_column,
-            alchemy_expression=cast(func.date_trunc('day', self.alchemy_expression), postgres.TIMESTAMP)
+            alchemy_expression=cast(func.date_trunc('day', self.alchemy_expression), TIMESTAMP(timezone=True))
         )
 
+    @property
+    def alchemy_expression(self):
+        return self._alchemy_offset_value(self.offset, self.aggregation.pg_calc).label(self.name)
+
 
 class RangeDocField(PgDocField):
     """
@@ -431,6 +465,11 @@ class DateRangeDocField(RangeDocField):
 
     def value_to_alchemy(self, value):
         low, high = value
+        # Is OK to cast, because we are wrapping it in timezone-aware datatype.
+        if isinstance(low, (ColumnElement, str)):
+            low = cast(low, TIMESTAMP(timezone=True))
+        if isinstance(high, (ColumnElement, str)):
+            high = cast(high, TIMESTAMP(timezone=True))
         return func.tstzrange(
             low, high,
             # Inclusive on both sides.
@@ -438,6 +477,27 @@ def value_to_alchemy(self, value):
             type_=TSTZRANGE,
         )
 
+    def search_value_to_alchemy(self, value):
+        low, high = value
+        if isinstance(low, datetime):
+            low = tz_as_utc(low)
+        if isinstance(high, datetime):
+            high = tz_as_utc(high)
+        return func.tstzrange(
+            low, high,
+            # Inclusive on both sides.
+            '[]',
+            type_=TSTZRANGE,
+        )
+
+    def normalise_value(self, value):
+        if isinstance(value, datetime):
+            return tz_as_utc(value)
+        elif isinstance(value, PgRange):
+            return PgRange(lower=tz_as_utc(value.lower), upper=tz_as_utc(value.upper), bounds=value.bounds)
+        else:
+            return tuple(tz_as_utc(v) for v in value)
+
     def between(self, low, high):
         """
         :rtype: Expression
@@ -448,8 +508,8 @@ def between(self, low, high):
         if isinstance(low, datetime) and isinstance(high, datetime):
             return RangeBetweenExpression(
                 self,
-                tz_aware(low),
-                tz_aware(high),
+                tz_as_utc(low).astimezone(timezone.utc),
+                tz_as_utc(high).astimezone(timezone.utc),
                 _range_class=DateTimeTZRange
             )
         else:
@@ -458,9 +518,10 @@ def between(self, low, high):
 
     @property
     def expression_with_leniency(self):
+        low = cast(self.lower.alchemy_expression, TIMESTAMP(timezone=True)) - cast('500 milliseconds', INTERVAL)
+        high = cast(self.greater.alchemy_expression, TIMESTAMP(timezone=True)) + cast('500 milliseconds', INTERVAL)
         return func.tstzrange(
-            self.lower.alchemy_expression - cast('500 milliseconds', INTERVAL),
-            self.greater.alchemy_expression + cast('500 milliseconds', INTERVAL),
+            low, high,
             # Inclusive on both sides.
             '[]',
             type_=TSTZRANGE,
diff --git a/datacube/drivers/postgis/sql.py b/datacube/drivers/postgis/sql.py
index b21940378..5766c6a38 100644
--- a/datacube/drivers/postgis/sql.py
+++ b/datacube/drivers/postgis/sql.py
@@ -6,7 +6,7 @@
 Custom types for postgres & sqlalchemy
 """
 
-from sqlalchemy import TIMESTAMP, text
+from sqlalchemy import text
 from sqlalchemy.types import Double
 from sqlalchemy.dialects.postgresql.ranges import AbstractRange, Range
 from sqlalchemy.ext.compiler import compiles
@@ -52,10 +52,7 @@ def visit_create_view(element, compiler, **kw):
 """
 
 TYPES_INIT_SQL = """
-create or replace function {schema}.common_timestamp(text)
-returns timestamp with time zone as $$
-select ($1)::timestamp at time zone 'utc';
-$$ language sql immutable returns null on null input;
+drop function if exists {schema}.common_timestamp(text);
 
 create type {schema}.float8range as range (
     subtype = float8,
@@ -74,21 +71,6 @@ def visit_float8range(element, compiler, **kw):
     return "FLOAT8RANGE"
 
 
-# Register the function with SQLAlchemhy.
-# pylint: disable=too-many-ancestors
-class CommonTimestamp(GenericFunction):
-    type = TIMESTAMP(timezone=True)
-    package = 'odc'
-    identifier = 'common_timestamp'
-    inherit_cache = False
-
-    name = 'common_timestamp'
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.packagenames = ['%s' % SCHEMA_NAME]
-
-
 # pylint: disable=too-many-ancestors
 class Float8Range(GenericFunction):
     type = FLOAT8RANGE  # type: ignore[assignment]
diff --git a/datacube/index/postgis/_datasets.py b/datacube/index/postgis/_datasets.py
index 5f76639f0..d6105bc5d 100755
--- a/datacube/index/postgis/_datasets.py
+++ b/datacube/index/postgis/_datasets.py
@@ -266,22 +266,22 @@ def search_product_duplicates(self, product: Product, *args):
 
         Returns each set of those field values and the datasets that have them.
         """
+        dataset_fields = product.metadata_type.dataset_fields
 
         def load_field(f: Union[str, fields.Field]) -> fields.Field:
             if isinstance(f, str):
-                return product.metadata_type.dataset_fields[f]
+                return dataset_fields[f]
             assert isinstance(f, fields.Field), "Not a field: %r" % (f,)
             return f
 
         group_fields = [cast(PgField, load_field(f)) for f in args]
-        expressions = [cast(PgExpression, product.metadata_type.dataset_fields.get('product') == product.name)]
+        expressions = [cast(PgExpression, dataset_fields.get('product') == product.name)]
 
         with self._db_connection() as connection:
             for record in connection.get_duplicates(group_fields, expressions):
-                as_dict = record._asdict()
-                if 'ids' in as_dict.keys():
-                    ids = as_dict.pop('ids')
-                    yield namedtuple('search_result', as_dict.keys())(**as_dict), set(ids)
+                if 'ids' in record:
+                    ids = record.pop('ids')
+                    yield namedtuple('search_result', record.keys())(**record), set(ids)
 
     def can_update(self, dataset, updates_allowed=None):
         """
@@ -597,19 +597,21 @@ def _make(self, dataset_res, full_info=False, product=None,
         :param bool full_info: Include all available fields
         """
         kwargs = {}
-        if dataset_res.uris:
-            uris = [uri for uri in dataset_res.uris if uri]
+        if not isinstance(dataset_res, dict):
+            dataset_res = dataset_res._asdict()
+        if "uris" in dataset_res:
+            uris = [uri for uri in dataset_res["uris"] if uri]
             if len(uris) == 1:
                 kwargs["uri"] = uris[0]
             else:
                 kwargs["uris"] = uris
 
         return Dataset(
-            product=product or self.products.get(dataset_res.product_ref),
-            metadata_doc=dataset_res.metadata,
-            indexed_by=dataset_res.added_by if full_info else None,
-            indexed_time=dataset_res.added if full_info else None,
-            archived_time=dataset_res.archived,
+            product=product or self.products.get(dataset_res["product_id"]),
+            metadata_doc=dataset_res["metadata_doc"],
+            indexed_by=dataset_res["indexed_by"] if full_info else None,
+            indexed_time=dataset_res["indexed_time"] if full_info else None,
+            archived_time=dataset_res["archived"],
             source_tree=source_tree,
             derived_tree=derived_tree,
             **kwargs
@@ -716,11 +718,9 @@ def search_returning(self,
                                                      limit=limit,
                                                      archived=archived):
             for columns in results:
-                coldict = columns._asdict()
-
                 def extract_field(f):
                     # Custom fields are not type-aware and returned as stringified json.
-                    return json.loads(coldict.get(f)) if f in custom_fields else coldict.get(f)
+                    return json.loads(columns.get(f)) if f in custom_fields else columns.get(f)
                 kwargs = {f: extract_field(f) for f in field_name_d}
                 yield result_type(**kwargs)
 
diff --git a/docs/about/whats_new.rst b/docs/about/whats_new.rst
index 0b02ca43e..449149799 100644
--- a/docs/about/whats_new.rst
+++ b/docs/about/whats_new.rst
@@ -17,6 +17,8 @@ v1.9.next
 - Record lineage when adding datasets with postgis index (:pull:`1632`)
 - Update schema logic (:pull:`1634`)
 - Drop valid-area check and anti-meridian fix 3857 extents (:pull:1635)
+- Remove problematic "common_timestamp" postgresql function from postgis driver. Some internal API changes
+  required to accommodate and preserve all previous index-driver level behaviour. (:pull:`1623`)
 
 v1.9.0-rc9 (3rd July 2024)
 ==========================


Test patch:
diff --git a/integration_tests/index/test_search_eo3.py b/integration_tests/index/test_search_eo3.py
index 9b371cf24..1a56a430f 100644
--- a/integration_tests/index/test_search_eo3.py
+++ b/integration_tests/index/test_search_eo3.py
@@ -931,18 +931,37 @@ def test_find_duplicates_with_time(index, nrt_dataset, final_dataset, ls8_eo3_da
     assert len(all_datasets) == 3
 
     dupe_fields = namedtuple('search_result', ['region_code', 'time'])
-    expected_result = [
+
+    expected_result_old = [
         (
             dupe_fields('090086', '("2023-04-30 23:50:33.884549","2023-04-30 23:50:34.884549")'),
             {nrt_dataset.id, final_dataset.id}
         )
     ]
+    expected_result_new = [
+        (
+            dupe_fields(
+                '090086',
+                (
+                    datetime.datetime(
+                        2023, 4, 30,
+                        23, 50, 33, 884549,
+                        tzinfo=datetime.timezone.utc),
+                    datetime.datetime(
+                        2023, 4, 30,
+                        23, 50, 34, 884549,
+                        tzinfo=datetime.timezone.utc)
+                ),
+            ),
+            {nrt_dataset.id, final_dataset.id}
+        )
+    ]
     res = sorted(index.datasets.search_product_duplicates(
         nrt_dataset.product,
         'region_code', 'time',
     ))
 
-    assert res == expected_result
+    assert res == expected_result_old or res == expected_result_new
 
 
 def test_csv_search_via_cli_eo3(clirunner: Any,