#758

Committed 04 Sep 2023 08:37PM UTC coverage: 0.0% (-78.3%) from 78.333%

Build # #758

Build Type

push

circle-ci

Committed by

hershd23

Commit Message

Increased underline length in at line 75 in text_summarization.rst
	modified:   docs/source/benchmarks/text_summarization.rst

Run Details

0 of 11303 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/create_udf_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.
import os
from pathlib import Path
from typing import Dict, List

import pandas as pd

from evadb.catalog.catalog_utils import get_metadata_properties
from evadb.catalog.models.udf_catalog import UdfCatalogEntry
from evadb.catalog.models.udf_io_catalog import UdfIOCatalogEntry
from evadb.catalog.models.udf_metadata_catalog import UdfMetadataCatalogEntry
from evadb.configuration.constants import (
    DEFAULT_TRAIN_TIME_LIMIT,
    EvaDB_INSTALLATION_DIR,
)
from evadb.database import EvaDBDatabase
from evadb.executor.abstract_executor import AbstractExecutor
from evadb.models.storage.batch import Batch
from evadb.plan_nodes.create_udf_plan import CreateUDFPlan
from evadb.third_party.huggingface.create import gen_hf_io_catalog_entries
from evadb.udfs.decorators.utils import load_io_from_udf_decorators
from evadb.utils.errors import UDFIODefinitionError
from evadb.utils.generic_utils import (
    load_udf_class_from_file,
    try_to_import_ludwig,
    try_to_import_torch,
    try_to_import_ultralytics,
)
from evadb.utils.logging_manager import logger


class CreateUDFExecutor(AbstractExecutor):
    def __init__(self, db: EvaDBDatabase, node: CreateUDFPlan):
        super().__init__(db, node)
        self.udf_dir = Path(EvaDB_INSTALLATION_DIR) / "udfs"

    def handle_huggingface_udf(self):
        """Handle HuggingFace UDFs

        HuggingFace UDFs are special UDFs that are not loaded from a file.
        So we do not need to call the setup method on them like we do for other UDFs.
        """
        # We need at least one deep learning framework for HuggingFace
        # Torch or Tensorflow
        try_to_import_torch()
        impl_path = f"{self.udf_dir}/abstract/hf_abstract_udf.py"
        io_list = gen_hf_io_catalog_entries(self.node.name, self.node.metadata)
        return (
            self.node.name,
            impl_path,
            self.node.udf_type,
            io_list,
            self.node.metadata,
        )

    def handle_ludwig_udf(self):
        """Handle ludwig UDFs

        Use Ludwig's auto_train engine to train/tune models.
        """
        try_to_import_ludwig()
        from ludwig.automl import auto_train

        assert (
            len(self.children) == 1
        ), "Create ludwig UDF expects 1 child, finds {}.".format(len(self.children))

        aggregated_batch_list = []
        child = self.children[0]
        for batch in child.exec():
            aggregated_batch_list.append(batch)
        aggregated_batch = Batch.concat(aggregated_batch_list, copy=False)
        aggregated_batch.drop_column_alias()

        arg_map = {arg.key: arg.value for arg in self.node.metadata}
        auto_train_results = auto_train(
            dataset=aggregated_batch.frames,
            target=arg_map["predict"],
            tune_for_memory=arg_map.get("tune_for_memory", False),
            time_limit_s=arg_map.get("time_limit", DEFAULT_TRAIN_TIME_LIMIT),
            output_directory=self.db.config.get_value("storage", "tmp_dir"),
        )
        model_path = os.path.join(
            self.db.config.get_value("storage", "model_dir"), self.node.name
        )
        auto_train_results.best_model.save(model_path)
        self.node.metadata.append(UdfMetadataCatalogEntry("model_path", model_path))

        impl_path = Path(f"{self.udf_dir}/ludwig.py").absolute().as_posix()
        io_list = self._resolve_udf_io(None)
        return (
            self.node.name,
            impl_path,
            self.node.udf_type,
            io_list,
            self.node.metadata,
        )

    def handle_ultralytics_udf(self):
        """Handle Ultralytics UDFs"""
        try_to_import_ultralytics()

        impl_path = (
            Path(f"{self.udf_dir}/yolo_object_detector.py").absolute().as_posix()
        )
        udf = self._try_initializing_udf(
            impl_path, udf_args=get_metadata_properties(self.node)
        )
        io_list = self._resolve_udf_io(udf)
        return (
            self.node.name,
            impl_path,
            self.node.udf_type,
            io_list,
            self.node.metadata,
        )

    def handle_generic_udf(self):
        """Handle generic UDFs

        Generic UDFs are loaded from a file. We check for inputs passed by the user during CREATE or try to load io from decorators.
        """
        impl_path = self.node.impl_path.absolute().as_posix()
        udf = self._try_initializing_udf(impl_path)
        io_list = self._resolve_udf_io(udf)

        return (
            self.node.name,
            impl_path,
            self.node.udf_type,
            io_list,
            self.node.metadata,
        )

    def exec(self, *args, **kwargs):
        """Create udf executor

        Calls the catalog to insert a udf catalog entry.
        """
        # check catalog if it already has this udf entry
        if self.catalog().get_udf_catalog_entry_by_name(self.node.name):
            if self.node.if_not_exists:
                msg = f"UDF {self.node.name} already exists, nothing added."
                yield Batch(pd.DataFrame([msg]))
                return
            else:
                msg = f"UDF {self.node.name} already exists."
                logger.error(msg)
                raise RuntimeError(msg)

        # if it's a type of HuggingFaceModel, override the impl_path
        if self.node.udf_type == "HuggingFace":
            name, impl_path, udf_type, io_list, metadata = self.handle_huggingface_udf()
        elif self.node.udf_type == "ultralytics":
            name, impl_path, udf_type, io_list, metadata = self.handle_ultralytics_udf()
        elif self.node.udf_type == "Ludwig":
            name, impl_path, udf_type, io_list, metadata = self.handle_ludwig_udf()
        else:
            name, impl_path, udf_type, io_list, metadata = self.handle_generic_udf()

        self.catalog().insert_udf_catalog_entry(
            name, impl_path, udf_type, io_list, metadata
        )
        yield Batch(
            pd.DataFrame([f"UDF {self.node.name} successfully added to the database."])
        )

    def _try_initializing_udf(
        self, impl_path: str, udf_args: Dict = {}
    ) -> UdfCatalogEntry:
        """Attempts to initialize UDF given the implementation file path and arguments.

        Args:
            impl_path (str): The file path of the UDF implementation file.
            udf_args (Dict, optional): Dictionary of arguments to pass to the UDF. Defaults to {}.

        Returns:
            UdfCatalogEntry: A UdfCatalogEntry object that represents the initialized UDF.

        Raises:
            RuntimeError: If an error occurs while initializing the UDF.
        """

        # load the udf class from the file
        try:
            # loading the udf class from the file
            udf = load_udf_class_from_file(impl_path, self.node.name)
            # initializing the udf class calls the setup method internally
            udf(**udf_args)
        except Exception as e:
            err_msg = f"Error creating UDF: {str(e)}"
            # logger.error(err_msg)
            raise RuntimeError(err_msg)

        return udf

    def _resolve_udf_io(self, udf: UdfCatalogEntry) -> List[UdfIOCatalogEntry]:
        """Private method that resolves the input/output definitions for a given UDF.
        It first searches for the input/outputs in the CREATE statement. If not found, it resolves them using decorators. If not found there as well, it raises an error.

        Args:
            udf (UdfCatalogEntry): The UDF for which to resolve input and output definitions.

        Returns:
            A List of UdfIOCatalogEntry objects that represent the resolved input and
            output definitions for the UDF.

        Raises:
            RuntimeError: If an error occurs while resolving the UDF input/output
            definitions.
        """
        io_list = []
        try:
            if self.node.inputs:
                io_list.extend(self.node.inputs)
            else:
                # try to load the inputs from decorators, the inputs from CREATE statement take precedence
                io_list.extend(load_io_from_udf_decorators(udf, is_input=True))

            if self.node.outputs:
                io_list.extend(self.node.outputs)
            else:
                # try to load the outputs from decorators, the outputs from CREATE statement take precedence
                io_list.extend(load_io_from_udf_decorators(udf, is_input=False))

        except UDFIODefinitionError as e:
            err_msg = f"Error creating UDF, input/output definition incorrect: {str(e)}"
            logger.error(err_msg)
            raise RuntimeError(err_msg)

        return io_list

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	import os	×
16	from pathlib import Path	×
17	from typing import Dict, List	×
18
19	import pandas as pd	×
20
21	from evadb.catalog.catalog_utils import get_metadata_properties	×
22	from evadb.catalog.models.udf_catalog import UdfCatalogEntry	×
23	from evadb.catalog.models.udf_io_catalog import UdfIOCatalogEntry	×
24	from evadb.catalog.models.udf_metadata_catalog import UdfMetadataCatalogEntry	×
25	from evadb.configuration.constants import (	×
26	DEFAULT_TRAIN_TIME_LIMIT,
27	EvaDB_INSTALLATION_DIR,
28	)
29	from evadb.database import EvaDBDatabase	×
30	from evadb.executor.abstract_executor import AbstractExecutor	×
31	from evadb.models.storage.batch import Batch	×
32	from evadb.plan_nodes.create_udf_plan import CreateUDFPlan	×
33	from evadb.third_party.huggingface.create import gen_hf_io_catalog_entries	×
34	from evadb.udfs.decorators.utils import load_io_from_udf_decorators	×
35	from evadb.utils.errors import UDFIODefinitionError	×
36	from evadb.utils.generic_utils import (	×
37	load_udf_class_from_file,
38	try_to_import_ludwig,
39	try_to_import_torch,
40	try_to_import_ultralytics,
41	)
42	from evadb.utils.logging_manager import logger	×
43
44
45	class CreateUDFExecutor(AbstractExecutor):	×
46	def __init__(self, db: EvaDBDatabase, node: CreateUDFPlan):	×
47	super().__init__(db, node)	×
48	self.udf_dir = Path(EvaDB_INSTALLATION_DIR) / "udfs"	×
49
50	def handle_huggingface_udf(self):	×
51	"""Handle HuggingFace UDFs
52
53	HuggingFace UDFs are special UDFs that are not loaded from a file.
54	So we do not need to call the setup method on them like we do for other UDFs.
55	"""
56	# We need at least one deep learning framework for HuggingFace
57	# Torch or Tensorflow
58	try_to_import_torch()	×
59	impl_path = f"{self.udf_dir}/abstract/hf_abstract_udf.py"	×
60	io_list = gen_hf_io_catalog_entries(self.node.name, self.node.metadata)	×
61	return (	×
62	self.node.name,
63	impl_path,
64	self.node.udf_type,
65	io_list,
66	self.node.metadata,
67	)
68
69	def handle_ludwig_udf(self):	×
70	"""Handle ludwig UDFs
71
72	Use Ludwig's auto_train engine to train/tune models.
73	"""
74	try_to_import_ludwig()	×
75	from ludwig.automl import auto_train	×
76
77	assert (	×
78	len(self.children) == 1
79	), "Create ludwig UDF expects 1 child, finds {}.".format(len(self.children))
80
81	aggregated_batch_list = []	×
82	child = self.children[0]	×
83	for batch in child.exec():	×
84	aggregated_batch_list.append(batch)	×
85	aggregated_batch = Batch.concat(aggregated_batch_list, copy=False)	×
86	aggregated_batch.drop_column_alias()	×
87
88	arg_map = {arg.key: arg.value for arg in self.node.metadata}	×
89	auto_train_results = auto_train(	×
90	dataset=aggregated_batch.frames,
91	target=arg_map["predict"],
92	tune_for_memory=arg_map.get("tune_for_memory", False),
93	time_limit_s=arg_map.get("time_limit", DEFAULT_TRAIN_TIME_LIMIT),
94	output_directory=self.db.config.get_value("storage", "tmp_dir"),
95	)
96	model_path = os.path.join(	×
97	self.db.config.get_value("storage", "model_dir"), self.node.name
98	)
99	auto_train_results.best_model.save(model_path)	×
100	self.node.metadata.append(UdfMetadataCatalogEntry("model_path", model_path))	×
101
102	impl_path = Path(f"{self.udf_dir}/ludwig.py").absolute().as_posix()	×
103	io_list = self._resolve_udf_io(None)	×
104	return (	×
105	self.node.name,
106	impl_path,
107	self.node.udf_type,
108	io_list,
109	self.node.metadata,
110	)
111
112	def handle_ultralytics_udf(self):	×
113	"""Handle Ultralytics UDFs"""
114	try_to_import_ultralytics()	×
115
116	impl_path = (	×
117	Path(f"{self.udf_dir}/yolo_object_detector.py").absolute().as_posix()
118	)
119	udf = self._try_initializing_udf(	×
120	impl_path, udf_args=get_metadata_properties(self.node)
121	)
122	io_list = self._resolve_udf_io(udf)	×
123	return (	×
124	self.node.name,
125	impl_path,
126	self.node.udf_type,
127	io_list,
128	self.node.metadata,
129	)
130
131	def handle_generic_udf(self):	×
132	"""Handle generic UDFs
133
134	Generic UDFs are loaded from a file. We check for inputs passed by the user during CREATE or try to load io from decorators.
135	"""
136	impl_path = self.node.impl_path.absolute().as_posix()	×
137	udf = self._try_initializing_udf(impl_path)	×
138	io_list = self._resolve_udf_io(udf)	×
139
140	return (	×
141	self.node.name,
142	impl_path,
143	self.node.udf_type,
144	io_list,
145	self.node.metadata,
146	)
147
148	def exec(self, args, *kwargs):	×
149	"""Create udf executor
150
151	Calls the catalog to insert a udf catalog entry.
152	"""
153	# check catalog if it already has this udf entry
154	if self.catalog().get_udf_catalog_entry_by_name(self.node.name):	×
155	if self.node.if_not_exists:	×
156	msg = f"UDF {self.node.name} already exists, nothing added."	×
157	yield Batch(pd.DataFrame([msg]))	×
158	return	×
159	else:
160	msg = f"UDF {self.node.name} already exists."	×
161	logger.error(msg)	×
162	raise RuntimeError(msg)
163
164	# if it's a type of HuggingFaceModel, override the impl_path
165	if self.node.udf_type == "HuggingFace":	×
166	name, impl_path, udf_type, io_list, metadata = self.handle_huggingface_udf()	×
167	elif self.node.udf_type == "ultralytics":	×
168	name, impl_path, udf_type, io_list, metadata = self.handle_ultralytics_udf()	×
169	elif self.node.udf_type == "Ludwig":	×
170	name, impl_path, udf_type, io_list, metadata = self.handle_ludwig_udf()	×
171	else:
172	name, impl_path, udf_type, io_list, metadata = self.handle_generic_udf()	×
173
174	self.catalog().insert_udf_catalog_entry(	×
175	name, impl_path, udf_type, io_list, metadata
176	)
177	yield Batch(	×
178	pd.DataFrame([f"UDF {self.node.name} successfully added to the database."])
179	)
180
181	def _try_initializing_udf(	×
182	self, impl_path: str, udf_args: Dict = {}
183	) -> UdfCatalogEntry:
184	"""Attempts to initialize UDF given the implementation file path and arguments.
185
186	Args:
187	impl_path (str): The file path of the UDF implementation file.
188	udf_args (Dict, optional): Dictionary of arguments to pass to the UDF. Defaults to {}.
189
190	Returns:
191	UdfCatalogEntry: A UdfCatalogEntry object that represents the initialized UDF.
192
193	Raises:
194	RuntimeError: If an error occurs while initializing the UDF.
195	"""
196
197	# load the udf class from the file
198	try:	×
199	# loading the udf class from the file
200	udf = load_udf_class_from_file(impl_path, self.node.name)	×
201	# initializing the udf class calls the setup method internally
202	udf(**udf_args)	×
203	except Exception as e:
204	err_msg = f"Error creating UDF: {str(e)}"
205	# logger.error(err_msg)
206	raise RuntimeError(err_msg)
207
208	return udf	×
209
210	def _resolve_udf_io(self, udf: UdfCatalogEntry) -> List[UdfIOCatalogEntry]:	×
211	"""Private method that resolves the input/output definitions for a given UDF.
212	It first searches for the input/outputs in the CREATE statement. If not found, it resolves them using decorators. If not found there as well, it raises an error.
213
214	Args:
215	udf (UdfCatalogEntry): The UDF for which to resolve input and output definitions.
216
217	Returns:
218	A List of UdfIOCatalogEntry objects that represent the resolved input and
219	output definitions for the UDF.
220
221	Raises:
222	RuntimeError: If an error occurs while resolving the UDF input/output
223	definitions.
224	"""
225	io_list = []	×
226	try:	×
227	if self.node.inputs:	×
228	io_list.extend(self.node.inputs)	×
229	else:
230	# try to load the inputs from decorators, the inputs from CREATE statement take precedence
231	io_list.extend(load_io_from_udf_decorators(udf, is_input=True))	×
232
233	if self.node.outputs:	×
234	io_list.extend(self.node.outputs)	×
235	else:
236	# try to load the outputs from decorators, the outputs from CREATE statement take precedence
237	io_list.extend(load_io_from_udf_decorators(udf, is_input=False))	×
238
239	except UDFIODefinitionError as e:
240	err_msg = f"Error creating UDF, input/output definition incorrect: {str(e)}"
241	logger.error(err_msg)
242	raise RuntimeError(err_msg)
243
244	return io_list	×

georgia-tech-db / eva / #758

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