f3cb8bbb-8165-49fa-af41-b587f634b3c4

pending completion

Build # f3cb8bbb-8165-49fa-af41-b587f634b3c4

Build Type

Pull #814

circle-ci

Committed by

jiashenC

Commit Message

update results

Pull Request Pull Request #814: feat: benchmark question answering v1

Run Details

42 of 42 new or added lines in 2 files covered. (100.0%)

10095 of 10421 relevant lines covered (96.87%)

0.97 hits per line

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

94.55

/eva/executor/create_index_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 pathlib import Path

import pandas as pd

from eva.catalog.catalog_manager import CatalogManager
from eva.catalog.sql_config import IDENTIFIER_COLUMN
from eva.configuration.configuration_manager import ConfigurationManager
from eva.executor.abstract_executor import AbstractExecutor
from eva.executor.executor_utils import ExecutorError, handle_vector_store_params
from eva.models.storage.batch import Batch
from eva.plan_nodes.create_index_plan import CreateIndexPlan
from eva.storage.storage_engine import StorageEngine
from eva.third_party.vector_stores.types import FeaturePayload
from eva.third_party.vector_stores.utils import VectorStoreFactory
from eva.utils.logging_manager import logger


class CreateIndexExecutor(AbstractExecutor):
    def __init__(self, node: CreateIndexPlan):
        super().__init__(node)

    def exec(self, *args, **kwargs):
        catalog_manager = CatalogManager()
        if catalog_manager.get_index_catalog_entry_by_name(self.node.name):
            msg = f"Index {self.node.name} already exists."
            logger.error(msg)
            raise ExecutorError(msg)

        self.index_path = self._get_index_save_path()
        self.index = None
        self._create_index()

        yield Batch(
            pd.DataFrame(
                [f"Index {self.node.name} successfully added to the database."]
            )
        )

    def _get_index_save_path(self) -> Path:
        index_dir = Path(ConfigurationManager().get_value("storage", "index_dir"))
        if not index_dir.exists():
            index_dir.mkdir(parents=True, exist_ok=True)
        return str(
            index_dir
            / Path("{}_{}.index".format(self.node.vector_store_type, self.node.name))
        )

    def _create_index(self):
        try:
            # Get feature tables.
            feat_catalog_entry = self.node.table_ref.table.table_obj

            # Get feature column.
            feat_col_name = self.node.col_list[0].name
            feat_column = [
                col for col in feat_catalog_entry.columns if col.name == feat_col_name
            ][0]

            # Add features to index.
            # TODO: batch size is hardcoded for now.
            input_dim = -1
            storage_engine = StorageEngine.factory(feat_catalog_entry)
            for input_batch in storage_engine.read(feat_catalog_entry):
                if self.node.udf_func:
                    # Create index through UDF expression.
                    # UDF(input column) -> 2 dimension feature vector.
                    input_batch.modify_column_alias(feat_catalog_entry.name.lower())
                    feat_batch = self.node.udf_func.evaluate(input_batch)
                    feat_batch.drop_column_alias()
                    input_batch.drop_column_alias()
                    feat = feat_batch.column_as_numpy_array("features")
                else:
                    # Create index on the feature table directly.
                    # Pandas wraps numpy array as an object inside a numpy
                    # array. Use zero index to get the actual numpy array.
                    feat = input_batch.column_as_numpy_array(feat_col_name)

                row_id = input_batch.column_as_numpy_array(IDENTIFIER_COLUMN)

                for i in range(len(input_batch)):
                    row_feat = feat[i].reshape(1, -1)
                    if self.index is None:
                        input_dim = row_feat.shape[1]
                        self.index = VectorStoreFactory.init_vector_store(
                            self.node.vector_store_type,
                            self.node.name,
                            **handle_vector_store_params(
                                self.node.vector_store_type, self.index_path
                            ),
                        )
                        self.index.create(input_dim)

                    # Row ID for mapping back to the row.
                    self.index.add([FeaturePayload(row_id[i], row_feat)])

            # Persist index.
            self.index.persist()

            # Save to catalog.
            CatalogManager().insert_index_catalog_entry(
                self.node.name,
                self.index_path,
                self.node.vector_store_type,
                feat_column,
                self.node.udf_func.signature() if self.node.udf_func else None,
            )
        except Exception as e:
            # Delete index.
            if self.index:
                self.index.delete()

            # Throw exception back to user.
            raise ExecutorError(str(e))

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 pathlib import Path	1✔
16
17	import pandas as pd	1✔
18
19	from eva.catalog.catalog_manager import CatalogManager	1✔
20	from eva.catalog.sql_config import IDENTIFIER_COLUMN	1✔
21	from eva.configuration.configuration_manager import ConfigurationManager	1✔
22	from eva.executor.abstract_executor import AbstractExecutor	1✔
23	from eva.executor.executor_utils import ExecutorError, handle_vector_store_params	1✔
24	from eva.models.storage.batch import Batch	1✔
25	from eva.plan_nodes.create_index_plan import CreateIndexPlan	1✔
26	from eva.storage.storage_engine import StorageEngine	1✔
27	from eva.third_party.vector_stores.types import FeaturePayload	1✔
28	from eva.third_party.vector_stores.utils import VectorStoreFactory	1✔
29	from eva.utils.logging_manager import logger	1✔
30
31
32	class CreateIndexExecutor(AbstractExecutor):	1✔
33	def __init__(self, node: CreateIndexPlan):	1✔
34	super().__init__(node)	1✔
35
36	def exec(self, args, *kwargs):	1✔
37	catalog_manager = CatalogManager()	1✔
38	if catalog_manager.get_index_catalog_entry_by_name(self.node.name):	1✔
39	msg = f"Index {self.node.name} already exists."	×
40	logger.error(msg)	×
41	raise ExecutorError(msg)
42
43	self.index_path = self._get_index_save_path()	1✔
44	self.index = None	1✔
45	self._create_index()	1✔
46
47	yield Batch(	1✔
48	pd.DataFrame(
49	[f"Index {self.node.name} successfully added to the database."]
50	)
51	)
52
53	def _get_index_save_path(self) -> Path:	1✔
54	index_dir = Path(ConfigurationManager().get_value("storage", "index_dir"))	1✔
55	if not index_dir.exists():	1✔
56	index_dir.mkdir(parents=True, exist_ok=True)	×
57	return str(	1✔
58	index_dir
59	/ Path("{}_{}.index".format(self.node.vector_store_type, self.node.name))
60	)
61
62	def _create_index(self):	1✔
63	try:	1✔
64	# Get feature tables.
65	feat_catalog_entry = self.node.table_ref.table.table_obj	1✔
66
67	# Get feature column.
68	feat_col_name = self.node.col_list[0].name	1✔
69	feat_column = [	1✔
70	col for col in feat_catalog_entry.columns if col.name == feat_col_name
71	][0]
72
73	# Add features to index.
74	# TODO: batch size is hardcoded for now.
75	input_dim = -1	1✔
76	storage_engine = StorageEngine.factory(feat_catalog_entry)	1✔
77	for input_batch in storage_engine.read(feat_catalog_entry):	1✔
78	if self.node.udf_func:	1✔
79	# Create index through UDF expression.
80	# UDF(input column) -> 2 dimension feature vector.
81	input_batch.modify_column_alias(feat_catalog_entry.name.lower())	1✔
82	feat_batch = self.node.udf_func.evaluate(input_batch)	1✔
83	feat_batch.drop_column_alias()	1✔
84	input_batch.drop_column_alias()	1✔
85	feat = feat_batch.column_as_numpy_array("features")	1✔
86	else:
87	# Create index on the feature table directly.
88	# Pandas wraps numpy array as an object inside a numpy
89	# array. Use zero index to get the actual numpy array.
90	feat = input_batch.column_as_numpy_array(feat_col_name)	1✔
91
92	row_id = input_batch.column_as_numpy_array(IDENTIFIER_COLUMN)	1✔
93
94	for i in range(len(input_batch)):	1✔
95	row_feat = feat[i].reshape(1, -1)	1✔
96	if self.index is None:	1✔
97	input_dim = row_feat.shape[1]	1✔
98	self.index = VectorStoreFactory.init_vector_store(	1✔
99	self.node.vector_store_type,
100	self.node.name,
101	**handle_vector_store_params(
102	self.node.vector_store_type, self.index_path
103	),
104	)
105	self.index.create(input_dim)	1✔
106
107	# Row ID for mapping back to the row.
108	self.index.add([FeaturePayload(row_id[i], row_feat)])	1✔
109
110	# Persist index.
111	self.index.persist()	1✔
112
113	# Save to catalog.
114	CatalogManager().insert_index_catalog_entry(	1✔
115	self.node.name,
116	self.index_path,
117	self.node.vector_store_type,
118	feat_column,
119	self.node.udf_func.signature() if self.node.udf_func else None,
120	)
121	except Exception as e:
122	# Delete index.
123	if self.index:
124	self.index.delete()
125
126	# Throw exception back to user.
127	raise ExecutorError(str(e))

georgia-tech-db / eva / f3cb8bbb-8165-49fa-af41-b587f634b3c4

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