QueryBuilder: fix outer join filters and refactor process list

Author: r0hansaxena

src/aiida/cmdline/commands/cmd_process.py

@@ -86,7 +86,7 @@ def verdi_process():
 @options.FAILED()
 @options.PAST_DAYS()
 @options.LIMIT()
-@options.ROOT()
+@options.ONLY_ROOTS()
 @options.RAW()
 @click.pass_context
 @decorators.with_dbenv()
@@ -101,7 +101,7 @@ def process_list(
     failed,
     past_days,
     limit,
-    root,
+    only_roots,
     project,
     raw,
     order_by,
@@ -127,7 +127,9 @@ def process_list(
         relationships['with_node'] = group
 
     builder = CalculationQueryBuilder()
-    filters = builder.get_filters(all_entries, process_state, process_label, paused, exit_status, failed, root=root)
+    filters = builder.get_filters(
+        all_entries, process_state, process_label, paused, exit_status, failed, only_roots=only_roots
+    )
     query_set = builder.get_query_set(
         relationships=relationships, filters=filters, order_by={order_by: order_dir}, past_days=past_days, limit=limit
     )

src/aiida/cmdline/params/options/__init__.py

@@ -82,6 +82,7 @@
     'NODES',
     'NON_INTERACTIVE',
     'OLDER_THAN',
+    'ONLY_ROOTS',
     'ONLY_TOP_LEVEL_CALCS',
     'ONLY_TOP_LEVEL_WORKFLOWS',
     'ORDER_BY',
@@ -103,7 +104,6 @@
     'RAW',
     'RELABEL_GROUPS',
     'REPOSITORY_PATH',
-    'ROOT',
     'SCHEDULER',
     'SILENT',
     'SORT',

src/aiida/cmdline/params/options/main.py

@@ -93,6 +93,7 @@
     'NODES',
     'NON_INTERACTIVE',
     'OLDER_THAN',
+    'ONLY_ROOTS',
     'ONLY_TOP_LEVEL_CALCS',
     'ONLY_TOP_LEVEL_WORKFLOWS',
     'ORDER_BY',
@@ -114,7 +115,6 @@
     'RAW',
     'RELABEL_GROUPS',
     'REPOSITORY_PATH',
-    'ROOT',
     'SCHEDULER',
     'SILENT',
     'SORT',
@@ -643,9 +643,9 @@ def set_log_level(ctx: click.Context, _param: click.Parameter, value: t.Any) ->
     default=False,
     help='Include all entries, disregarding all other filter options and flags.',
 )
-ROOT = OverridableOption(
-    '--root',
-    'root',
+ONLY_ROOTS = OverridableOption(
+    '--only-roots',
+    'only_roots',
     is_flag=True,
     default=False,
     help='Only include root processes (those without a caller).',

src/aiida/orm/querybuilder.py

@@ -426,11 +426,11 @@ def append(
             # So for those cases we need to construct the correct filters,
             # corresponding to the provided classes and value of `subclassing`.
             if ormclass == EntityTypes.NODE:
-                self._add_node_type_filter(tag, classifiers, subclassing)
-                self._add_process_type_filter(tag, classifiers, subclassing)
+                self._add_node_type_filter(tag, classifiers, subclassing, outerjoin=outerjoin)
+                self._add_process_type_filter(tag, classifiers, subclassing, outerjoin=outerjoin)
 
             elif ormclass == EntityTypes.GROUP:
-                self._add_group_type_filter(tag, classifiers, subclassing)
+                self._add_group_type_filter(tag, classifiers, subclassing, outerjoin=outerjoin)
 
             # The order has to be first _add_node_type_filter and then add_filter.
             # If the user adds a query on the type column, it overwrites what I did
@@ -733,12 +733,15 @@ def _process_filters(filters: FilterType) -> Dict[str, Any]:
 
         return processed_filters
 
-    def _add_node_type_filter(self, tagspec: str, classifiers: List[Classifier], subclassing: bool):
+    def _add_node_type_filter(
+        self, tagspec: str, classifiers: List[Classifier], subclassing: bool, outerjoin: bool = False
+    ):
         """Add a filter based on node type.
 
         :param tagspec: The tag, which has to exist already as a key in self._filters
         :param classifiers: a dictionary with classifiers
         :param subclassing: if True, allow for subclasses of the ormclass
+        :param outerjoin: if True, wrap the filter with OR id IS NULL
         """
         if len(classifiers) > 1:
             # If a list was passed to QueryBuilder.append, this propagates to a list in the classifiers
@@ -748,14 +751,21 @@ def _add_node_type_filter(self, tagspec: str, classifiers: List[Classifier], sub
         else:
             entity_type_filter = _get_node_type_filter(classifiers[0], subclassing)
 
-        self.add_filter(tagspec, {'node_type': entity_type_filter})
+        filters: dict = {'node_type': entity_type_filter}
+        if outerjoin:
+            filters = {'or': [filters, {'id': None}]}
 
-    def _add_process_type_filter(self, tagspec: str, classifiers: List[Classifier], subclassing: bool) -> None:
+        self.add_filter(tagspec, filters)
+
+    def _add_process_type_filter(
+        self, tagspec: str, classifiers: List[Classifier], subclassing: bool, outerjoin: bool = False
+    ) -> None:
         """Add a filter based on process type.
 
         :param tagspec: The tag, which has to exist already as a key in self._filters
         :param classifiers: a dictionary with classifiers
         :param subclassing: if True, allow for subclasses of the process type
+        :param outerjoin: if True, wrap the filter with OR id IS NULL
 
         Note: This function handles the case when process_type_string is None.
         """
@@ -767,28 +777,41 @@ def _add_process_type_filter(self, tagspec: str, classifiers: List[Classifier],
                     process_type_filter['or'].append(_get_process_type_filter(classifier, subclassing))
 
             if len(process_type_filter['or']) > 0:
-                self.add_filter(tagspec, {'process_type': process_type_filter})
+                filters: dict = {'process_type': process_type_filter}
+                if outerjoin:
+                    filters = {'or': [filters, {'id': None}]}
+                self.add_filter(tagspec, filters)
 
         elif classifiers[0].process_type_string is not None:
             process_type_filter = _get_process_type_filter(classifiers[0], subclassing)
-            self.add_filter(tagspec, {'process_type': process_type_filter})
+            filters = {'process_type': process_type_filter}
+            if outerjoin:
+                filters = {'or': [filters, {'id': None}]}
+            self.add_filter(tagspec, filters)
 
-    def _add_group_type_filter(self, tagspec: str, classifiers: List[Classifier], subclassing: bool) -> None:
+    def _add_group_type_filter(
+        self, tagspec: str, classifiers: List[Classifier], subclassing: bool, outerjoin: bool = False
+    ) -> None:
         """Add a filter based on group type.
 
         :param tagspec: The tag, which has to exist already as a key in self._filters
         :param classifiers: a dictionary with classifiers
         :param subclassing: if True, allow for subclasses of the ormclass
+        :param outerjoin: if True, wrap the filter with OR id IS NULL
         """
         if len(classifiers) > 1:
             # If a list was passed to QueryBuilder.append, this propagates to a list in the classifiers
-            type_string_filter: dict = {'or': []}
+            entity_type_filter: dict = {'or': []}
             for classifier in classifiers:
-                type_string_filter['or'].append(_get_group_type_filter(classifier, subclassing))
+                entity_type_filter['or'].append(_get_group_type_filter(classifier, subclassing))
         else:
-            type_string_filter = _get_group_type_filter(classifiers[0], subclassing)
+            entity_type_filter = _get_group_type_filter(classifiers[0], subclassing)
+
+        filters: dict = {'type_string': entity_type_filter}
+        if outerjoin:
+            filters = {'or': [filters, {'id': None}]}
 
-        self.add_filter(tagspec, {'type_string': type_string_filter})
+        self.add_filter(tagspec, filters)
 
     def add_projection(self, tag_spec: Union[str, EntityClsType], projection_spec: ProjectType) -> None:
         r"""Adds a projection

src/aiida/storage/psql_dos/orm/querybuilder/joiner.py

@@ -378,7 +378,7 @@ def _join_node_inputs(self, joined_entity, entity_to_join, isouterjoin: bool, **
         aliased_edge = aliased(self._entities.Link)
 
         def new_query(q):
-            return q.join(aliased_edge, aliased_edge.output_id == joined_entity.id).join(
+            return q.join(aliased_edge, aliased_edge.output_id == joined_entity.id, isouter=isouterjoin).join(
                 entity_to_join, aliased_edge.input_id == entity_to_join.id, isouter=isouterjoin
             )
 

src/aiida/storage/psql_dos/orm/querybuilder/main.py

@@ -328,6 +328,7 @@ def _build(self, data: QueryDictType) -> BuiltQuery:
         for tag, filter_specs in data['filters'].items():
             if not filter_specs:
                 continue
+
             alias = tag_to_alias.get(tag)
             if not alias:
                 raise ValueError(f'Unknown tag {tag!r} in filters, known: {list(tag_to_alias)}')

src/aiida/tools/query/calculation.py

@@ -13,11 +13,9 @@
 import typing as t
 from collections.abc import Iterable
 
-from aiida.common.lang import classproperty
-
 from aiida import orm
 from aiida.common import timezone
-from aiida.common.links import LinkType
+from aiida.common.lang import classproperty
 
 from .mapping import CalculationProjectionMapper, ProjectionMapper
 
@@ -86,7 +84,7 @@ def get_filters(
         exit_status: str | None = None,
         failed: bool = False,
         node_types: list[Node] | None = None,
-        root: bool = False,
+        only_roots: bool = False,
     ) -> dict[str, t.Any]:
         """Return a set of QueryBuilder filters based on typical command line options.
 
@@ -97,7 +95,7 @@ def get_filters(
         :param paused: Boolean, if True, filter for processes that are paused.
         :param exit_status: Filter for this exit status.
         :param failed: Boolean to filter only failed processes.
-        :param root: Boolean, if True, filter for root processes (those without a caller).
+        :param only_roots: Boolean, if True, filter for root processes (those without a caller).
         :return: Dictionary of filters suitable for a QueryBuilder.append() call.
         """
         from aiida.engine import ProcessState
@@ -134,8 +132,8 @@ def get_filters(
             filters[process_state_attribute] = {'==': ProcessState.FINISHED.value}
             filters[exit_status_attribute] = {'==': exit_status}
 
-        if root:
-            filters['root'] = True
+        if only_roots:
+            filters['only_roots'] = True
 
         return filters
 
@@ -171,19 +169,18 @@ def get_query_set(
         if filters is None:
             filters = {}
 
-        root = filters.pop('root', False)
+        only_roots = filters.pop('only_roots', False)
 
         if past_days is not None:
             filters['ctime'] = {'>': timezone.now() - datetime.timedelta(days=past_days)}
 
         builder = orm.QueryBuilder()
         builder.append(cls=orm.ProcessNode, filters=filters, project=unique_projections, tag='process')
 
-        if root:
+        if only_roots:
             builder.append(
                 orm.ProcessNode,
                 with_outgoing='process',
-                edge_filters={'type': {'in': (LinkType.CALL_WORK.value, LinkType.CALL_CALC.value)}},
                 tag='caller',
                 outerjoin=True,
             )