e6605954-d2af-406f-a2e4-075433fee53b

Committed 16 Sep 2023 06:13PM UTC coverage: 80.158% (+10.2%) from 69.982%

Build # e6605954-d2af-406f-a2e4-075433fee53b

Build Type

push

circle-ci

Committed by

Chitti-Ankith

Commit Message

lint fixes

Run Details

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

9522 of 11879 relevant lines covered (80.16%)

1.44 hits per line

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

84.81

/evadb/executor/executor_utils.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 glob
import os
from pathlib import Path
from typing import TYPE_CHECKING, Generator, List

if TYPE_CHECKING:
    from evadb.catalog.catalog_manager import CatalogManager

from evadb.catalog.catalog_type import VectorStoreType
from evadb.expression.abstract_expression import AbstractExpression
from evadb.expression.function_expression import FunctionExpression
from evadb.models.storage.batch import Batch
from evadb.parser.table_ref import TableInfo
from evadb.parser.types import FileFormatType
from evadb.readers.document.registry import SUPPORTED_TYPES
from evadb.utils.generic_utils import try_to_import_cv2
from evadb.utils.logging_manager import logger


class ExecutorError(Exception):
    pass


def apply_project(
    batch: Batch, project_list: List[AbstractExpression], catalog: "CatalogManager"
):
    if not batch.empty() and project_list:
        batches = [expr.evaluate(batch) for expr in project_list]
        batch = Batch.merge_column_wise(batches)

        # persist stats of function expression
        for expr in project_list:
            for func_expr in expr.find_all(FunctionExpression):
                if func_expr.function_obj and func_expr._stats:
                    function_id = func_expr.function_obj.row_id
                    catalog.upsert_function_cost_catalog_entry(
                        function_id,
                        func_expr.function_obj.name,
                        func_expr._stats.prev_cost,
                    )
    return batch


def apply_predicate(
    batch: Batch, predicate: AbstractExpression, catalog: "CatalogManager"
) -> Batch:
    if not batch.empty() and predicate is not None:
        outcomes = predicate.evaluate(batch)
        batch.drop_zero(outcomes)
        batch.reset_index()

        # persist stats of function expression
        for func_expr in predicate.find_all(FunctionExpression):
            if func_expr.function_obj and func_expr._stats:
                function_id = func_expr.function_obj.row_id
                catalog.upsert_function_cost_catalog_entry(
                    function_id, func_expr.function_obj.name, func_expr._stats.prev_cost
                )
    return batch


def handle_if_not_exists(
    catalog: "CatalogManager", table_info: TableInfo, if_not_exist=False
):
    # Table exists
    if catalog.check_table_exists(
        table_info.table_name,
        table_info.database_name,
    ):
        err_msg = "Table: {} already exists".format(table_info)
        if if_not_exist:
            logger.warn(err_msg)
            return True
        else:
            logger.error(err_msg)
            raise ExecutorError(err_msg)
    # Table does not exist
    else:
        return False


def validate_image(image_path: Path) -> bool:
    try:
        try_to_import_cv2()
        import cv2

        data = cv2.imread(str(image_path))
        return data is not None
    except Exception as e:
        logger.warning(
            f"Unexpected Exception {e} occurred while reading image file {image_path}"
        )
        return False


def iter_path_regex(path_regex: Path) -> Generator[str, None, None]:
    return glob.iglob(os.path.expanduser(path_regex), recursive=True)


def validate_video(video_path: Path) -> bool:
    try:
        try_to_import_cv2()
        import cv2

        vid = cv2.VideoCapture(str(video_path))
        if not vid.isOpened():
            return False
        return True
    except Exception as e:
        logger.warning(
            f"Unexpected Exception {e} occurred while reading video file {video_path}"
        )


def validate_document(doc_path: Path) -> bool:
    return doc_path.suffix in SUPPORTED_TYPES


def validate_pdf(doc_path: Path) -> bool:
    return doc_path.suffix == ".pdf"


def validate_media(file_path: Path, media_type: FileFormatType) -> bool:
    if media_type == FileFormatType.VIDEO:
        return validate_video(file_path)
    elif media_type == FileFormatType.IMAGE:
        return validate_image(file_path)
    elif media_type == FileFormatType.DOCUMENT:
        return validate_document(file_path)
    elif media_type == FileFormatType.PDF:
        return validate_pdf(file_path)
    else:
        raise ValueError(f"Unsupported Media type {str(media_type)}")


def handle_vector_store_params(
    vector_store_type: VectorStoreType, index_path: str
) -> dict:
    """Handle vector store parameters based on the vector store type and index path.

    Args:
        vector_store_type (VectorStoreType): The type of vector store.
        index_path (str): The path to store the index.

    Returns:
        dict: Dictionary containing the appropriate vector store parameters.


    Raises:
        ValueError: If the vector store type in the node is not supported.
    """
    if vector_store_type == VectorStoreType.FAISS:
        return {"index_path": index_path}
    elif vector_store_type == VectorStoreType.QDRANT:
        return {"index_db": str(Path(index_path).parent)}
    elif vector_store_type == VectorStoreType.PINECONE:
        return {}
    else:
        raise ValueError("Unsupported vector store type: {}".format(vector_store_type))

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 glob	2✔
16	import os	2✔
17	from pathlib import Path	2✔
18	from typing import TYPE_CHECKING, Generator, List	2✔
19
20	if TYPE_CHECKING:
21	from evadb.catalog.catalog_manager import CatalogManager
22
23	from evadb.catalog.catalog_type import VectorStoreType	2✔
24	from evadb.expression.abstract_expression import AbstractExpression	2✔
25	from evadb.expression.function_expression import FunctionExpression	2✔
26	from evadb.models.storage.batch import Batch	2✔
27	from evadb.parser.table_ref import TableInfo	2✔
28	from evadb.parser.types import FileFormatType	2✔
29	from evadb.readers.document.registry import SUPPORTED_TYPES	2✔
30	from evadb.utils.generic_utils import try_to_import_cv2	2✔
31	from evadb.utils.logging_manager import logger	2✔
32
33
34	class ExecutorError(Exception):	2✔
35	pass	2✔
36
37
38	def apply_project(	2✔
39	batch: Batch, project_list: List[AbstractExpression], catalog: "CatalogManager"
40	):
41	if not batch.empty() and project_list:	2✔
42	batches = [expr.evaluate(batch) for expr in project_list]	2✔
43	batch = Batch.merge_column_wise(batches)	2✔
44
45	# persist stats of function expression
46	for expr in project_list:	2✔
47	for func_expr in expr.find_all(FunctionExpression):	2✔
48	if func_expr.function_obj and func_expr._stats:	1✔
49	function_id = func_expr.function_obj.row_id	1✔
50	catalog.upsert_function_cost_catalog_entry(	1✔
51	function_id,
52	func_expr.function_obj.name,
53	func_expr._stats.prev_cost,
54	)
55	return batch	2✔
56
57
58	def apply_predicate(	2✔
59	batch: Batch, predicate: AbstractExpression, catalog: "CatalogManager"
60	) -> Batch:
61	if not batch.empty() and predicate is not None:	2✔
62	outcomes = predicate.evaluate(batch)	2✔
63	batch.drop_zero(outcomes)	2✔
64	batch.reset_index()	2✔
65
66	# persist stats of function expression
67	for func_expr in predicate.find_all(FunctionExpression):	2✔
68	if func_expr.function_obj and func_expr._stats:	1✔
69	function_id = func_expr.function_obj.row_id	1✔
70	catalog.upsert_function_cost_catalog_entry(	1✔
71	function_id, func_expr.function_obj.name, func_expr._stats.prev_cost
72	)
73	return batch	2✔
74
75
76	def handle_if_not_exists(	2✔
77	catalog: "CatalogManager", table_info: TableInfo, if_not_exist=False
78	):
79	# Table exists
80	if catalog.check_table_exists(	2✔
81	table_info.table_name,
82	table_info.database_name,
83	):
84	err_msg = "Table: {} already exists".format(table_info)	×
85	if if_not_exist:	×
86	logger.warn(err_msg)	×
87	return True	×
88	else:
89	logger.error(err_msg)	×
90	raise ExecutorError(err_msg)
91	# Table does not exist
92	else:
93	return False	2✔
94
95
96	def validate_image(image_path: Path) -> bool:	2✔
97	try:	1✔
98	try_to_import_cv2()	1✔
99	import cv2	1✔
100
101	data = cv2.imread(str(image_path))	1✔
102	return data is not None	1✔
103	except Exception as e:
104	logger.warning(
105	f"Unexpected Exception {e} occurred while reading image file {image_path}"
106	)
107	return False
108
109
110	def iter_path_regex(path_regex: Path) -> Generator[str, None, None]:	2✔
111	return glob.iglob(os.path.expanduser(path_regex), recursive=True)	2✔
112
113
114	def validate_video(video_path: Path) -> bool:	2✔
115	try:	2✔
116	try_to_import_cv2()	2✔
117	import cv2	2✔
118
119	vid = cv2.VideoCapture(str(video_path))	2✔
120	if not vid.isOpened():	2✔
121	return False	×
122	return True	2✔
123	except Exception as e:
124	logger.warning(
125	f"Unexpected Exception {e} occurred while reading video file {video_path}"
126	)
127
128
129	def validate_document(doc_path: Path) -> bool:	2✔
130	return doc_path.suffix in SUPPORTED_TYPES	×
131
132
133	def validate_pdf(doc_path: Path) -> bool:	2✔
134	return doc_path.suffix == ".pdf"	1✔
135
136
137	def validate_media(file_path: Path, media_type: FileFormatType) -> bool:	2✔
138	if media_type == FileFormatType.VIDEO:	2✔
139	return validate_video(file_path)	2✔
140	elif media_type == FileFormatType.IMAGE:	1✔
141	return validate_image(file_path)	1✔
142	elif media_type == FileFormatType.DOCUMENT:	1✔
143	return validate_document(file_path)	×
144	elif media_type == FileFormatType.PDF:	1✔
145	return validate_pdf(file_path)	1✔
146	else:
147	raise ValueError(f"Unsupported Media type {str(media_type)}")
148
149
150	def handle_vector_store_params(	2✔
151	vector_store_type: VectorStoreType, index_path: str
152	) -> dict:
153	"""Handle vector store parameters based on the vector store type and index path.
154
155	Args:
156	vector_store_type (VectorStoreType): The type of vector store.
157	index_path (str): The path to store the index.
158
159	Returns:
160	dict: Dictionary containing the appropriate vector store parameters.
161
162
163	Raises:
164	ValueError: If the vector store type in the node is not supported.
165	"""
166	if vector_store_type == VectorStoreType.FAISS:	1✔
167	return {"index_path": index_path}	1✔
168	elif vector_store_type == VectorStoreType.QDRANT:	×
169	return {"index_db": str(Path(index_path).parent)}	×
170	elif vector_store_type == VectorStoreType.PINECONE:	×
171	return {}	×
172	else:
173	raise ValueError("Unsupported vector store type: {}".format(vector_store_type))

georgia-tech-db / eva / e6605954-d2af-406f-a2e4-075433fee53b

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