66dd80ef-0c0e-472b-82bc-b677131eeb05

pending completion

Build # 66dd80ef-0c0e-472b-82bc-b677131eeb05

Build Type

Pull #924

circle-ci

Committed by

yulaicui

Commit Message

fixes

Pull Request Pull Request #924: fix tutorials 11-similarity-search-for-motif-mining.ipynb

Run Details

45 of 45 new or added lines in 8 files covered. (100.0%)

10104 of 10657 relevant lines covered (94.81%)

1.89 hits per line

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

93.75

/evadb/executor/plan_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 typing import Iterator

from evadb.database import EvaDBDatabase
from evadb.executor.abstract_executor import AbstractExecutor
from evadb.executor.apply_and_merge_executor import ApplyAndMergeExecutor
from evadb.executor.create_executor import CreateExecutor
from evadb.executor.create_index_executor import CreateIndexExecutor
from evadb.executor.create_udf_executor import CreateUDFExecutor
from evadb.executor.delete_executor import DeleteExecutor
from evadb.executor.drop_object_executor import DropObjectExecutor
from evadb.executor.exchange_executor import ExchangeExecutor
from evadb.executor.executor_utils import ExecutorError
from evadb.executor.explain_executor import ExplainExecutor
from evadb.executor.function_scan_executor import FunctionScanExecutor
from evadb.executor.groupby_executor import GroupByExecutor
from evadb.executor.hash_join_executor import HashJoinExecutor
from evadb.executor.insert_executor import InsertExecutor
from evadb.executor.join_build_executor import BuildJoinExecutor
from evadb.executor.lateral_join_executor import LateralJoinExecutor
from evadb.executor.limit_executor import LimitExecutor
from evadb.executor.load_executor import LoadDataExecutor
from evadb.executor.nested_loop_join_executor import NestedLoopJoinExecutor
from evadb.executor.orderby_executor import OrderByExecutor
from evadb.executor.pp_executor import PPExecutor
from evadb.executor.predicate_executor import PredicateExecutor
from evadb.executor.project_executor import ProjectExecutor
from evadb.executor.rename_executor import RenameExecutor
from evadb.executor.sample_executor import SampleExecutor
from evadb.executor.seq_scan_executor import SequentialScanExecutor
from evadb.executor.show_info_executor import ShowInfoExecutor
from evadb.executor.storage_executor import StorageExecutor
from evadb.executor.union_executor import UnionExecutor
from evadb.executor.vector_index_scan_executor import VectorIndexScanExecutor
from evadb.models.storage.batch import Batch
from evadb.plan_nodes.abstract_plan import AbstractPlan
from evadb.plan_nodes.types import PlanOprType
from evadb.utils.logging_manager import logger


class PlanExecutor:
    """
    This is an interface between plan tree and execution tree.
    We traverse the plan tree and build execution tree from it

    Arguments:
        plan (AbstractPlan): Physical plan tree which needs to be executed
        evadb (EvaDBDatabase): database to execute the query on
    """

    def __init__(self, evadb: EvaDBDatabase, plan: AbstractPlan):
        self._db = evadb
        self._plan = plan

    def _build_execution_tree(self, plan: AbstractPlan) -> AbstractExecutor:
        """build the execution tree from plan tree

        Arguments:
            plan {AbstractPlan} -- Input Plan tree

        Returns:
            AbstractExecutor -- Compiled Execution tree
        """
        root = None
        if plan is None:
            return root

        # Get plan node type
        plan_opr_type = plan.opr_type

        if plan_opr_type == PlanOprType.SEQUENTIAL_SCAN:
            executor_node = SequentialScanExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.UNION:
            executor_node = UnionExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.STORAGE_PLAN:
            executor_node = StorageExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.PP_FILTER:
            executor_node = PPExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.CREATE:
            executor_node = CreateExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.RENAME:
            executor_node = RenameExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.DROP_OBJECT:
            executor_node = DropObjectExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.INSERT:
            executor_node = InsertExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.CREATE_UDF:
            executor_node = CreateUDFExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.LOAD_DATA:
            executor_node = LoadDataExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.GROUP_BY:
            executor_node = GroupByExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.ORDER_BY:
            executor_node = OrderByExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.LIMIT:
            executor_node = LimitExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.SAMPLE:
            executor_node = SampleExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.NESTED_LOOP_JOIN:
            executor_node = NestedLoopJoinExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.LATERAL_JOIN:
            logger.warn(
                "LateralJoin Executor should not be part of the execution plan."
                "Please raise an issue with the current query. Thanks!"
            )
            executor_node = LateralJoinExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.HASH_JOIN:
            executor_node = HashJoinExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.HASH_BUILD:
            executor_node = BuildJoinExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.FUNCTION_SCAN:
            executor_node = FunctionScanExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.EXCHANGE:
            executor_node = ExchangeExecutor(db=self._db, node=plan)
            inner_executor = self._build_execution_tree(plan.inner_plan)
            executor_node.build_inner_executor(inner_executor)
        elif plan_opr_type == PlanOprType.PROJECT:
            executor_node = ProjectExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.PREDICATE_FILTER:
            executor_node = PredicateExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.SHOW_INFO:
            executor_node = ShowInfoExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.EXPLAIN:
            executor_node = ExplainExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.CREATE_INDEX:
            executor_node = CreateIndexExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.APPLY_AND_MERGE:
            executor_node = ApplyAndMergeExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.VECTOR_INDEX_SCAN:
            executor_node = VectorIndexScanExecutor(db=self._db, node=plan)
        elif plan_opr_type == PlanOprType.DELETE:
            executor_node = DeleteExecutor(db=self._db, node=plan)

        # EXPLAIN does not need to build execution tree for its children
        if plan_opr_type != PlanOprType.EXPLAIN:
            # Build Executor Tree for children
            for children in plan.children:
                executor_node.append_child(self._build_execution_tree(children))

        return executor_node

    def execute_plan(
        self,
        do_not_raise_exceptions: bool = False,
        do_not_print_exceptions: bool = False,
    ) -> Iterator[Batch]:
        """execute the plan tree"""
        try:
            execution_tree = self._build_execution_tree(self._plan)
            output = execution_tree.exec()
            if output is not None:
                yield from output
        except Exception as e:
            if do_not_raise_exceptions is False:
                if do_not_print_exceptions is False:
                    logger.exception(str(e))
                raise ExecutorError(e)

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 typing import Iterator	2✔
16
17	from evadb.database import EvaDBDatabase	2✔
18	from evadb.executor.abstract_executor import AbstractExecutor	2✔
19	from evadb.executor.apply_and_merge_executor import ApplyAndMergeExecutor	2✔
20	from evadb.executor.create_executor import CreateExecutor	2✔
21	from evadb.executor.create_index_executor import CreateIndexExecutor	2✔
22	from evadb.executor.create_udf_executor import CreateUDFExecutor	2✔
23	from evadb.executor.delete_executor import DeleteExecutor	2✔
24	from evadb.executor.drop_object_executor import DropObjectExecutor	2✔
25	from evadb.executor.exchange_executor import ExchangeExecutor	2✔
26	from evadb.executor.executor_utils import ExecutorError	2✔
27	from evadb.executor.explain_executor import ExplainExecutor	2✔
28	from evadb.executor.function_scan_executor import FunctionScanExecutor	2✔
29	from evadb.executor.groupby_executor import GroupByExecutor	2✔
30	from evadb.executor.hash_join_executor import HashJoinExecutor	2✔
31	from evadb.executor.insert_executor import InsertExecutor	2✔
32	from evadb.executor.join_build_executor import BuildJoinExecutor	2✔
33	from evadb.executor.lateral_join_executor import LateralJoinExecutor	2✔
34	from evadb.executor.limit_executor import LimitExecutor	2✔
35	from evadb.executor.load_executor import LoadDataExecutor	2✔
36	from evadb.executor.nested_loop_join_executor import NestedLoopJoinExecutor	2✔
37	from evadb.executor.orderby_executor import OrderByExecutor	2✔
38	from evadb.executor.pp_executor import PPExecutor	2✔
39	from evadb.executor.predicate_executor import PredicateExecutor	2✔
40	from evadb.executor.project_executor import ProjectExecutor	2✔
41	from evadb.executor.rename_executor import RenameExecutor	2✔
42	from evadb.executor.sample_executor import SampleExecutor	2✔
43	from evadb.executor.seq_scan_executor import SequentialScanExecutor	2✔
44	from evadb.executor.show_info_executor import ShowInfoExecutor	2✔
45	from evadb.executor.storage_executor import StorageExecutor	2✔
46	from evadb.executor.union_executor import UnionExecutor	2✔
47	from evadb.executor.vector_index_scan_executor import VectorIndexScanExecutor	2✔
48	from evadb.models.storage.batch import Batch	2✔
49	from evadb.plan_nodes.abstract_plan import AbstractPlan	2✔
50	from evadb.plan_nodes.types import PlanOprType	2✔
51	from evadb.utils.logging_manager import logger	2✔
52
53
54	class PlanExecutor:	2✔
55	"""
56	This is an interface between plan tree and execution tree.
57	We traverse the plan tree and build execution tree from it
58
59	Arguments:
60	plan (AbstractPlan): Physical plan tree which needs to be executed
61	evadb (EvaDBDatabase): database to execute the query on
62	"""
63
64	def __init__(self, evadb: EvaDBDatabase, plan: AbstractPlan):	2✔
65	self._db = evadb	2✔
66	self._plan = plan	2✔
67
68	def _build_execution_tree(self, plan: AbstractPlan) -> AbstractExecutor:	2✔
69	"""build the execution tree from plan tree
70
71	Arguments:
72	plan {AbstractPlan} -- Input Plan tree
73
74	Returns:
75	AbstractExecutor -- Compiled Execution tree
76	"""
77	root = None	2✔
78	if plan is None:	2✔
79	return root	×
80
81	# Get plan node type
82	plan_opr_type = plan.opr_type	2✔
83
84	if plan_opr_type == PlanOprType.SEQUENTIAL_SCAN:	2✔
85	executor_node = SequentialScanExecutor(db=self._db, node=plan)	2✔
86	elif plan_opr_type == PlanOprType.UNION:	2✔
87	executor_node = UnionExecutor(db=self._db, node=plan)	2✔
88	elif plan_opr_type == PlanOprType.STORAGE_PLAN:	2✔
89	executor_node = StorageExecutor(db=self._db, node=plan)	2✔
90	elif plan_opr_type == PlanOprType.PP_FILTER:	2✔
91	executor_node = PPExecutor(db=self._db, node=plan)	2✔
92	elif plan_opr_type == PlanOprType.CREATE:	2✔
93	executor_node = CreateExecutor(db=self._db, node=plan)	2✔
94	elif plan_opr_type == PlanOprType.RENAME:	2✔
95	executor_node = RenameExecutor(db=self._db, node=plan)	2✔
96	elif plan_opr_type == PlanOprType.DROP_OBJECT:	2✔
97	executor_node = DropObjectExecutor(db=self._db, node=plan)	2✔
98	elif plan_opr_type == PlanOprType.INSERT:	2✔
99	executor_node = InsertExecutor(db=self._db, node=plan)	2✔
100	elif plan_opr_type == PlanOprType.CREATE_UDF:	2✔
101	executor_node = CreateUDFExecutor(db=self._db, node=plan)	2✔
102	elif plan_opr_type == PlanOprType.LOAD_DATA:	2✔
103	executor_node = LoadDataExecutor(db=self._db, node=plan)	2✔
104	elif plan_opr_type == PlanOprType.GROUP_BY:	2✔
105	executor_node = GroupByExecutor(db=self._db, node=plan)	2✔
106	elif plan_opr_type == PlanOprType.ORDER_BY:	2✔
107	executor_node = OrderByExecutor(db=self._db, node=plan)	2✔
108	elif plan_opr_type == PlanOprType.LIMIT:	2✔
109	executor_node = LimitExecutor(db=self._db, node=plan)	2✔
110	elif plan_opr_type == PlanOprType.SAMPLE:	2✔
111	executor_node = SampleExecutor(db=self._db, node=plan)	×
112	elif plan_opr_type == PlanOprType.NESTED_LOOP_JOIN:	2✔
113	executor_node = NestedLoopJoinExecutor(db=self._db, node=plan)	2✔
114	elif plan_opr_type == PlanOprType.LATERAL_JOIN:	2✔
115	logger.warn(
116	"LateralJoin Executor should not be part of the execution plan."
117	"Please raise an issue with the current query. Thanks!"
118	)
119	executor_node = LateralJoinExecutor(db=self._db, node=plan)	×
120	elif plan_opr_type == PlanOprType.HASH_JOIN:	2✔
121	executor_node = HashJoinExecutor(db=self._db, node=plan)	2✔
122	elif plan_opr_type == PlanOprType.HASH_BUILD:	2✔
123	executor_node = BuildJoinExecutor(db=self._db, node=plan)	2✔
124	elif plan_opr_type == PlanOprType.FUNCTION_SCAN:	2✔
125	executor_node = FunctionScanExecutor(db=self._db, node=plan)	×
126	elif plan_opr_type == PlanOprType.EXCHANGE:	2✔
127	executor_node = ExchangeExecutor(db=self._db, node=plan)	×
128	inner_executor = self._build_execution_tree(plan.inner_plan)	×
129	executor_node.build_inner_executor(inner_executor)	×
130	elif plan_opr_type == PlanOprType.PROJECT:	2✔
131	executor_node = ProjectExecutor(db=self._db, node=plan)	2✔
132	elif plan_opr_type == PlanOprType.PREDICATE_FILTER:	2✔
133	executor_node = PredicateExecutor(db=self._db, node=plan)	2✔
134	elif plan_opr_type == PlanOprType.SHOW_INFO:	2✔
135	executor_node = ShowInfoExecutor(db=self._db, node=plan)	2✔
136	elif plan_opr_type == PlanOprType.EXPLAIN:	2✔
137	executor_node = ExplainExecutor(db=self._db, node=plan)	2✔
138	elif plan_opr_type == PlanOprType.CREATE_INDEX:	2✔
139	executor_node = CreateIndexExecutor(db=self._db, node=plan)	2✔
140	elif plan_opr_type == PlanOprType.APPLY_AND_MERGE:	2✔
141	executor_node = ApplyAndMergeExecutor(db=self._db, node=plan)	2✔
142	elif plan_opr_type == PlanOprType.VECTOR_INDEX_SCAN:	2✔
143	executor_node = VectorIndexScanExecutor(db=self._db, node=plan)	2✔
144	elif plan_opr_type == PlanOprType.DELETE:	2✔
145	executor_node = DeleteExecutor(db=self._db, node=plan)	2✔
146
147	# EXPLAIN does not need to build execution tree for its children
148	if plan_opr_type != PlanOprType.EXPLAIN:	2✔
149	# Build Executor Tree for children
150	for children in plan.children:	2✔
151	executor_node.append_child(self._build_execution_tree(children))	2✔
152
153	return executor_node	2✔
154
155	def execute_plan(
156	self,
157	do_not_raise_exceptions: bool = False,
158	do_not_print_exceptions: bool = False,
159	) -> Iterator[Batch]:
160	"""execute the plan tree"""
161	try:	2✔
162	execution_tree = self._build_execution_tree(self._plan)	2✔
163	output = execution_tree.exec()	2✔
164	if output is not None:	2✔
165	yield from output	2✔
166	except Exception as e:
167	if do_not_raise_exceptions is False:
168	if do_not_print_exceptions is False:
169	logger.exception(str(e))
170	raise ExecutorError(e)

georgia-tech-db / eva / 66dd80ef-0c0e-472b-82bc-b677131eeb05

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