d88c9094-1a92-4944-91ea-5e7e8275cb11

Committed 23 Nov 2023 10:19PM UTC coverage: 67.117% (+67.1%) from 0.0%

Build # d88c9094-1a92-4944-91ea-5e7e8275cb11

Build Type

push

circleci

Committed by

web-flow

Commit Message

Merge branch 'georgia-tech-db:staging' into cost_batching

Run Details

342 of 692 new or added lines in 47 files covered. (49.42%)

12 existing lines in 4 files now uncovered.

9189 of 13691 relevant lines covered (67.12%)

0.67 hits per line

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

77.66

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

from evadb.catalog.catalog_utils import xform_column_definitions_to_catalog_entries
from evadb.catalog.models.utils import TableCatalogEntry
from evadb.parser.create_statement import ColumnDefinition

if TYPE_CHECKING:
    from evadb.catalog.catalog_manager import CatalogManager

from evadb.catalog.catalog_type import TableType, 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 instrument_function_expression_cost(
    expr: Union[AbstractExpression, List[AbstractExpression]],
    catalog: "CatalogManager",
):
    """We are expecting an instance of a catalog. An optimization can be to avoid creating a catalog instance if there is no function expression. An easy fix is to pass the function handler and create the catalog instance only if there is a function expression. In the past, this was problematic because of Ray. We can revisit it again."""

    if expr is None:
        return

    list_expr = expr
    if not isinstance(expr, list):
        list_expr = [expr]

    # persist stats of function expression
    for expr in list_expr:
        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,
                )


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

    return batch


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

    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, catalog
) -> 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.CHROMADB:
        return {"index_path": str(Path(index_path).parent)}
    elif vector_store_type == VectorStoreType.PINECONE:
        # add the required API_KEYS
        return {
            "PINECONE_API_KEY": catalog().get_configuration_catalog_value(
                "PINECONE_API_KEY"
            ),
            "PINECONE_ENV": catalog().get_configuration_catalog_value("PINECONE_ENV"),
        }
    elif vector_store_type == VectorStoreType.WEAVIATE:
        # Weaviate Configuration
        # Weaviate API key and URL Can be obtained from cluster details on Weaviate Cloud Services (WCS) dashboard
        return {
            "WEAVIATE_API_KEY": catalog().get_configuration_catalog_value(
                "WEAVIATE_API_KEY"
            ),
            "WEAVIATE_API_URL": catalog().get_configuration_catalog_value(
                "WEAVIATE_API_URL"
            ),
        }
    elif vector_store_type == VectorStoreType.MILVUS:
        return {
            "MILVUS_URI": catalog().get_configuration_catalog_value("MILVUS_URI"),
            "MILVUS_USER": catalog().get_configuration_catalog_value("MILVUS_USER"),
            "MILVUS_PASSWORD": catalog().get_configuration_catalog_value(
                "MILVUS_PASSWORD"
            ),
            "MILVUS_DB_NAME": catalog().get_configuration_catalog_value(
                "MILVUS_DB_NAME"
            ),
            "MILVUS_TOKEN": catalog().get_configuration_catalog_value("MILVUS_TOKEN"),
        }
    else:
        raise ValueError("Unsupported vector store type: {}".format(vector_store_type))


def create_table_catalog_entry_for_native_table(
    table_info: TableInfo, column_list: List[ColumnDefinition]
):
    column_catalog_entries = xform_column_definitions_to_catalog_entries(column_list)

    # Assemble table.
    table_catalog_entry = TableCatalogEntry(
        name=table_info.table_name,
        file_url=None,
        table_type=TableType.NATIVE_DATA,
        columns=column_catalog_entries,
        database_name=table_info.database_name,
    )
    return table_catalog_entry

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	1✔
16	import os	1✔
17	from pathlib import Path	1✔
18	from typing import TYPE_CHECKING, Generator, List, Union	1✔
19
20	from evadb.catalog.catalog_utils import xform_column_definitions_to_catalog_entries	1✔
21	from evadb.catalog.models.utils import TableCatalogEntry	1✔
22	from evadb.parser.create_statement import ColumnDefinition	1✔
23
24	if TYPE_CHECKING:
25	from evadb.catalog.catalog_manager import CatalogManager
26
27	from evadb.catalog.catalog_type import TableType, VectorStoreType	1✔
28	from evadb.expression.abstract_expression import AbstractExpression	1✔
29	from evadb.expression.function_expression import FunctionExpression	1✔
30	from evadb.models.storage.batch import Batch	1✔
31	from evadb.parser.table_ref import TableInfo	1✔
32	from evadb.parser.types import FileFormatType	1✔
33	from evadb.readers.document.registry import SUPPORTED_TYPES	1✔
34	from evadb.utils.generic_utils import try_to_import_cv2	1✔
35	from evadb.utils.logging_manager import logger	1✔
36
37
38	class ExecutorError(Exception):	1✔
39	pass	1✔
40
41
42	def instrument_function_expression_cost(	1✔
43	expr: Union[AbstractExpression, List[AbstractExpression]],
44	catalog: "CatalogManager",
45	):
46	"""We are expecting an instance of a catalog. An optimization can be to avoid creating a catalog instance if there is no function expression. An easy fix is to pass the function handler and create the catalog instance only if there is a function expression. In the past, this was problematic because of Ray. We can revisit it again."""
47
48	if expr is None:	1✔
49	return	1✔
50
51	list_expr = expr	1✔
52	if not isinstance(expr, list):	1✔
53	list_expr = [expr]	1✔
54
55	# persist stats of function expression
56	for expr in list_expr:	1✔
57	for func_expr in expr.find_all(FunctionExpression):	1✔
58	if func_expr.function_obj and func_expr._stats:	1✔
59	function_id = func_expr.function_obj.row_id	1✔
60	catalog.upsert_function_cost_catalog_entry(	1✔
61	function_id,
62	func_expr.function_obj.name,
63	func_expr._stats.prev_cost,
64	)
65
66
67	def apply_project(batch: Batch, project_list: List[AbstractExpression]):	1✔
68	if not batch.empty() and project_list:	1✔
69	batches = [expr.evaluate(batch) for expr in project_list]	1✔
70	batch = Batch.merge_column_wise(batches)	1✔
71
72	return batch	1✔
73
74
75	def apply_predicate(batch: Batch, predicate: AbstractExpression) -> Batch:	1✔
76	if not batch.empty() and predicate is not None:	1✔
77	outcomes = predicate.evaluate(batch)	1✔
78	batch.drop_zero(outcomes)	1✔
79	batch.reset_index()	1✔
80
81	return batch	1✔
82
83
84	def handle_if_not_exists(	1✔
85	catalog: "CatalogManager", table_info: TableInfo, if_not_exist=False
86	):
87	# Table exists
88	if catalog.check_table_exists(	1✔
89	table_info.table_name,
90	table_info.database_name,
91	):
92	err_msg = "Table: {} already exists".format(table_info)	×
93	if if_not_exist:	×
94	logger.warn(err_msg)	×
95	return True	×
96	else:
97	logger.error(err_msg)	×
98	raise ExecutorError(err_msg)
99	# Table does not exist
100	else:
101	return False	1✔
102
103
104	def validate_image(image_path: Path) -> bool:	1✔
105	try:	1✔
106	try_to_import_cv2()	1✔
107	import cv2	1✔
108
109	data = cv2.imread(str(image_path))	1✔
110	return data is not None	1✔
111	except Exception as e:
112	logger.warning(
113	f"Unexpected Exception {e} occurred while reading image file {image_path}"
114	)
115	return False
116
117
118	def iter_path_regex(path_regex: Path) -> Generator[str, None, None]:	1✔
119	return glob.iglob(os.path.expanduser(path_regex), recursive=True)	1✔
120
121
122	def validate_video(video_path: Path) -> bool:	1✔
123	try:	1✔
124	try_to_import_cv2()	1✔
125	import cv2	1✔
126
127	vid = cv2.VideoCapture(str(video_path))	1✔
128	if not vid.isOpened():	1✔
129	return False	×
130	return True	1✔
131	except Exception as e:
132	logger.warning(
133	f"Unexpected Exception {e} occurred while reading video file {video_path}"
134	)
135
136
137	def validate_document(doc_path: Path) -> bool:	1✔
138	return doc_path.suffix in SUPPORTED_TYPES	×
139
140
141	def validate_pdf(doc_path: Path) -> bool:	1✔
142	return doc_path.suffix == ".pdf"	1✔
143
144
145	def validate_media(file_path: Path, media_type: FileFormatType) -> bool:	1✔
146	if media_type == FileFormatType.VIDEO:	1✔
147	return validate_video(file_path)	1✔
148	elif media_type == FileFormatType.IMAGE:	1✔
149	return validate_image(file_path)	1✔
150	elif media_type == FileFormatType.DOCUMENT:	1✔
151	return validate_document(file_path)	×
152	elif media_type == FileFormatType.PDF:	1✔
153	return validate_pdf(file_path)	1✔
154	else:
155	raise ValueError(f"Unsupported Media type {str(media_type)}")
156
157
158	def handle_vector_store_params(	1✔
159	vector_store_type: VectorStoreType, index_path: str, catalog
160	) -> dict:
161	"""Handle vector store parameters based on the vector store type and index path.
162
163	Args:
164	vector_store_type (VectorStoreType): The type of vector store.
165	index_path (str): The path to store the index.
166
167	Returns:
168	dict: Dictionary containing the appropriate vector store parameters.
169
170
171	Raises:
172	ValueError: If the vector store type in the node is not supported.
173	"""
174	if vector_store_type == VectorStoreType.FAISS:	1✔
175	return {"index_path": index_path}	1✔
176	elif vector_store_type == VectorStoreType.QDRANT:	×
177	return {"index_db": str(Path(index_path).parent)}	×
178	elif vector_store_type == VectorStoreType.CHROMADB:	×
179	return {"index_path": str(Path(index_path).parent)}	×
180	elif vector_store_type == VectorStoreType.PINECONE:	×
181	# add the required API_KEYS
182	return {	×
183	"PINECONE_API_KEY": catalog().get_configuration_catalog_value(
184	"PINECONE_API_KEY"
185	),
186	"PINECONE_ENV": catalog().get_configuration_catalog_value("PINECONE_ENV"),
187	}
NEW 188	elif vector_store_type == VectorStoreType.WEAVIATE:	×
189	# Weaviate Configuration
190	# Weaviate API key and URL Can be obtained from cluster details on Weaviate Cloud Services (WCS) dashboard
NEW 191	return {	×
192	"WEAVIATE_API_KEY": catalog().get_configuration_catalog_value(
193	"WEAVIATE_API_KEY"
194	),
195	"WEAVIATE_API_URL": catalog().get_configuration_catalog_value(
196	"WEAVIATE_API_URL"
197	),
198	}
199	elif vector_store_type == VectorStoreType.MILVUS:	×
200	return {	×
201	"MILVUS_URI": catalog().get_configuration_catalog_value("MILVUS_URI"),
202	"MILVUS_USER": catalog().get_configuration_catalog_value("MILVUS_USER"),
203	"MILVUS_PASSWORD": catalog().get_configuration_catalog_value(
204	"MILVUS_PASSWORD"
205	),
206	"MILVUS_DB_NAME": catalog().get_configuration_catalog_value(
207	"MILVUS_DB_NAME"
208	),
209	"MILVUS_TOKEN": catalog().get_configuration_catalog_value("MILVUS_TOKEN"),
210	}
211	else:
212	raise ValueError("Unsupported vector store type: {}".format(vector_store_type))
213
214
215	def create_table_catalog_entry_for_native_table(	1✔
216	table_info: TableInfo, column_list: List[ColumnDefinition]
217	):
218	column_catalog_entries = xform_column_definitions_to_catalog_entries(column_list)	×
219
220	# Assemble table.
221	table_catalog_entry = TableCatalogEntry(	×
222	name=table_info.table_name,
223	file_url=None,
224	table_type=TableType.NATIVE_DATA,
225	columns=column_catalog_entries,
226	database_name=table_info.database_name,
227	)
228	return table_catalog_entry	×

georgia-tech-db / eva / d88c9094-1a92-4944-91ea-5e7e8275cb11

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