#850

Committed 08 Nov 2023 08:36PM UTC coverage: 0.0% (-77.0%) from 76.982%

Build # #850

Build Type

push

circleci

Committed by

americast

Commit Message

fix metrics logic

Run Details

0 of 1 new or added line in 1 file covered. (0.0%)

9789 existing lines in 252 files now uncovered.

0 of 12428 relevant lines covered (0.0%)

0.0 hits per line

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

0.0

/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, Union

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_database_executor import CreateDatabaseExecutor
from evadb.executor.create_executor import CreateExecutor
from evadb.executor.create_function_executor import CreateFunctionExecutor
from evadb.executor.create_index_executor import CreateIndexExecutor
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.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.set_executor import SetExecutor
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.use_executor import UseExecutor
from evadb.executor.vector_index_scan_executor import VectorIndexScanExecutor
from evadb.models.storage.batch import Batch
from evadb.parser.create_statement import CreateDatabaseStatement
from evadb.parser.set_statement import SetStatement
from evadb.parser.statement import AbstractStatement
from evadb.parser.use_statement import UseStatement
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: Union[AbstractPlan, AbstractStatement]
    ) -> 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

        # First handle cases when the plan is actually a parser statement
        if isinstance(plan, CreateDatabaseStatement):
            return CreateDatabaseExecutor(db=self._db, node=plan)
        elif isinstance(plan, UseStatement):
            return UseExecutor(db=self._db, node=plan)
        elif isinstance(plan, SetStatement):
            return SetExecutor(db=self._db, node=plan)

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

georgia-tech-db / eva / #850

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