#840

Committed 18 Oct 2023 09:51PM UTC coverage: 68.616% (-9.8%) from 78.391%

Build # #840

Build Type

push

circle-ci

Committed by

jiashenC

Commit Message

[BUMP]: v0.3.9+dev

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

8634 of 12583 relevant lines covered (68.62%)

0.69 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

41.67

/evadb/executor/abstract_executor.py

# coding=utf-8
# Copyright 2018-2023 EvaDB
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from abc import ABC, abstractmethod
from collections import deque
from typing import TYPE_CHECKING, Any, Generator, Iterable, List, TypeVar

if TYPE_CHECKING:
    from evadb.catalog.catalog_manager import CatalogManager
from evadb.configuration.configuration_manager import ConfigurationManager
from evadb.database import EvaDBDatabase
from evadb.models.storage.batch import Batch
from evadb.plan_nodes.abstract_plan import AbstractPlan

AbstractExecutor = TypeVar("AbstractExecutor")


class AbstractExecutor(ABC):
    """
    An abstract class for the executor engine
    Arguments:
        node (AbstractPlan): Plan node corresponding to this executor
    """

    def __init__(self, db: EvaDBDatabase, node: AbstractPlan):
        self._db = db
        self._node = node
        self._config: ConfigurationManager = db.config if db else None
        self._children = []

    # @lru_cache(maxsize=None)
    def catalog(self) -> "CatalogManager":
        """The object is intentionally generated on demand to prevent serialization issues. Having a SQLAlchemy object as a member variable can cause problems with multiprocessing. See get_catalog_instance()"""
        return self._db.catalog() if self._db else None

    def append_child(self, child: AbstractExecutor):
        """
        appends a child executor node

        Arguments:
            child {AbstractExecutor} -- child node
        """
        self._children.append(child)

    @property
    def children(self) -> List[AbstractExecutor]:
        """
        Returns the list of child executor
        Returns:
            [] -- list of children
        """
        return self._children

    @children.setter
    def children(self, children):
        self._children = children

    @property
    def node(self) -> AbstractPlan:
        return self._node

    @property
    def db(self) -> EvaDBDatabase:
        return self._db

    @property
    def config(self) -> ConfigurationManager:
        return self._config

    @abstractmethod
    def exec(self, *args, **kwargs) -> Iterable[Batch]:
        """
        This method is implemented by every executor.
        Contains logic for that executor;
        For retrieval based executor : It fetches frame batches from
        child nodes and emits it to parent node.
        """

    def __call__(self, *args, **kwargs) -> Generator[Batch, None, None]:
        yield from self.exec(*args, **kwargs)

    def bfs(self):
        """Returns a generator which visits all nodes in execution tree in
        breadth-first search (BFS) traversal order.

        Returns:
            the generator object.
        """
        queue = deque([self])
        while queue:
            node = queue.popleft()
            yield node
            for child in node.children:
                queue.append(child)

    def find_all(self, execution_type: Any):
        """Returns a generator which visits all the nodes in execution tree and yields one that matches the passed `execution_type`.

        Args:
            execution_type (Any): execution type to match with

        Returns:
            the generator object.
        """

        for node in self.bfs():
            if isinstance(node, execution_type):
                yield node

1	# coding=utf-8
2	# Copyright 2018-2023 EvaDB
3	#
4	# Licensed under the Apache License, Version 2.0 (the "License");
5	# you may not use this file except in compliance with the License.
6	# You may obtain a copy of the License at
7	#
8	# http://www.apache.org/licenses/LICENSE-2.0
9	#
10	# Unless required by applicable law or agreed to in writing, software
11	# distributed under the License is distributed on an "AS IS" BASIS,
12	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	# See the License for the specific language governing permissions and
14	# limitations under the License.
15	from abc import ABC, abstractmethod	1✔
16	from collections import deque	1✔
17	from typing import TYPE_CHECKING, Any, Generator, Iterable, List, TypeVar	1✔
18
19	if TYPE_CHECKING:
20	from evadb.catalog.catalog_manager import CatalogManager
21	from evadb.configuration.configuration_manager import ConfigurationManager	1✔
22	from evadb.database import EvaDBDatabase	1✔
23	from evadb.models.storage.batch import Batch	1✔
24	from evadb.plan_nodes.abstract_plan import AbstractPlan	×
25
26	AbstractExecutor = TypeVar("AbstractExecutor")	×
27
28
29	class AbstractExecutor(ABC):	1✔
30	"""
31	An abstract class for the executor engine
32	Arguments:
33	node (AbstractPlan): Plan node corresponding to this executor
34	"""
35
36	def __init__(self, db: EvaDBDatabase, node: AbstractPlan):	1✔
37	self._db = db	1✔
38	self._node = node	1✔
39	self._config: ConfigurationManager = db.config if db else None	×
40	self._children = []	×
41
42	# @lru_cache(maxsize=None)
43	def catalog(self) -> "CatalogManager":	1✔
44	"""The object is intentionally generated on demand to prevent serialization issues. Having a SQLAlchemy object as a member variable can cause problems with multiprocessing. See get_catalog_instance()"""
45	return self._db.catalog() if self._db else None	1✔
46
47	def append_child(self, child: AbstractExecutor):	×
48	"""
49	appends a child executor node
50
51	Arguments:
52	child {AbstractExecutor} -- child node
53	"""
54	self._children.append(child)	1✔
55
56	@property	×
57	def children(self) -> List[AbstractExecutor]:	×
58	"""
59	Returns the list of child executor
60	Returns:
61	[] -- list of children
62	"""
63	return self._children	1✔
64
65	@children.setter	×
66	def children(self, children):	×
67	self._children = children	1✔
68
69	@property	1✔
70	def node(self) -> AbstractPlan:	×
71	return self._node	1✔
72
73	@property	1✔
74	def db(self) -> EvaDBDatabase:	×
75	return self._db	1✔
76
77	@property	×
78	def config(self) -> ConfigurationManager:	×
79	return self._config	×
80
81	@abstractmethod	×
82	def exec(self, args, *kwargs) -> Iterable[Batch]:	×
83	"""
84	This method is implemented by every executor.
85	Contains logic for that executor;
86	For retrieval based executor : It fetches frame batches from
87	child nodes and emits it to parent node.
88	"""
89
90	def __call__(self, args, *kwargs) -> Generator[Batch, None, None]:	×
91	yield from self.exec(args, *kwargs)	×
92
93	def bfs(self):	×
94	"""Returns a generator which visits all nodes in execution tree in
95	breadth-first search (BFS) traversal order.
96
97	Returns:
98	the generator object.
99	"""
100	queue = deque([self])	×
101	while queue:	1✔
102	node = queue.popleft()	×
103	yield node	×
104	for child in node.children:	×
105	queue.append(child)	×
106
107	def find_all(self, execution_type: Any):	×
108	"""Returns a generator which visits all the nodes in execution tree and yields one that matches the passed `execution_type`.
109
110	Args:
111	execution_type (Any): execution type to match with
112
113	Returns:
114	the generator object.
115	"""
116
117	for node in self.bfs():	×
118	if isinstance(node, execution_type):	×
119	yield node	×

georgia-tech-db / eva / #840

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous