e0a0b839-f01e-4338-9a63-3fc1982b296e

Build # e0a0b839-f01e-4338-9a63-3fc1982b296e

Build Type

circle-ci

Committed by Yulai Cui

Commit Message

updates

Pull Request Pull Request #804: feat: create udf interface for pythonic api

Run Details

21 of 21 new or added lines in 3 files covered. (100.0%)

10091 of 10338 relevant lines covered (97.61%)

0.98 hits per line

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

95.83

/eva/executor/create_executor.py

# coding=utf-8
# Copyright 2018-2023 EVA
#
# 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 eva.catalog.catalog_manager import CatalogManager
from eva.executor.abstract_executor import AbstractExecutor
from eva.executor.executor_utils import ExecutorError, handle_if_not_exists
from eva.plan_nodes.create_plan import CreatePlan
from eva.plan_nodes.types import PlanOprType
from eva.storage.storage_engine import StorageEngine
from eva.utils.logging_manager import logger


class CreateExecutor(AbstractExecutor):
    def __init__(self, node: CreatePlan):
        super().__init__(node)
        self.catalog = CatalogManager()

    def exec(self, *args, **kwargs):
        if not handle_if_not_exists(self.node.table_info, self.node.if_not_exists):
            logger.debug(f"Creating table {self.node.table_info}")

            catalog_entry = self.catalog.create_and_insert_table_catalog_entry(
                self.node.table_info, self.node.column_list
            )
            storage_engine = StorageEngine.factory(catalog_entry)
            storage_engine.create(table=catalog_entry)

            if self.children != []:
                child = self.children[0]
                # only support seq scan based materialization
                if child.node.opr_type not in {PlanOprType.SEQUENTIAL_SCAN, PlanOprType.PROJECT}:
                    err_msg = "Invalid query {}, expected {} or {}".format(
                        child.node.opr_type,
                        PlanOprType.SEQUENTIAL_SCAN,
                        PlanOprType.PROJECT,
                    )
                    raise ExecutorError(err_msg)

                # Populate the table
                for batch in child.exec():
                    batch.drop_column_alias()
                    storage_engine.write(catalog_entry, batch)

1	# coding=utf-8
2	# Copyright 2018-2023 EVA
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 eva.catalog.catalog_manager import CatalogManager	1✔
16	from eva.executor.abstract_executor import AbstractExecutor	1✔
17	from eva.executor.executor_utils import ExecutorError, handle_if_not_exists	1✔
18	from eva.plan_nodes.create_plan import CreatePlan	1✔
19	from eva.plan_nodes.types import PlanOprType	1✔
20	from eva.storage.storage_engine import StorageEngine	1✔
21	from eva.utils.logging_manager import logger	1✔
22
23
24	class CreateExecutor(AbstractExecutor):	1✔
25	def __init__(self, node: CreatePlan):	1✔
26	super().__init__(node)	1✔
27	self.catalog = CatalogManager()	1✔
28
29	def exec(self, args, *kwargs):	1✔
30	if not handle_if_not_exists(self.node.table_info, self.node.if_not_exists):	1✔
31	logger.debug(f"Creating table {self.node.table_info}")	1✔
32
33	catalog_entry = self.catalog.create_and_insert_table_catalog_entry(	1✔
34	self.node.table_info, self.node.column_list
35	)
36	storage_engine = StorageEngine.factory(catalog_entry)	1✔
37	storage_engine.create(table=catalog_entry)	1✔
38
39	if self.children != []:	1✔
40	child = self.children[0]	1✔
41	# only support seq scan based materialization
42	if child.node.opr_type not in {PlanOprType.SEQUENTIAL_SCAN, PlanOprType.PROJECT}:	1✔
43	err_msg = "Invalid query {}, expected {} or {}".format(	×
44	child.node.opr_type,
45	PlanOprType.SEQUENTIAL_SCAN,
46	PlanOprType.PROJECT,
47	)
48	raise ExecutorError(err_msg)
49
50	# Populate the table
51	for batch in child.exec():	1✔
52	batch.drop_column_alias()	1✔
53	storage_engine.write(catalog_entry, batch)	1✔

georgia-tech-db / eva / e0a0b839-f01e-4338-9a63-3fc1982b296e

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