Fix/minor bugs #2348
Fix/minor bugs #2348
Conversation
| @@ -210,6 +214,8 @@ def schedule_jobs( | |||
| ids = db.execute(self.select.select_ids) | |||
| ids = [id[self.select.primary_id] for id in ids] | |||
|
|
|||
| # TODO: Check ready ids | |||
|
|
|||
There was a problem hiding this comment.
This is not required, will remove the TODO
| @override | ||
| def post_create(self, db: "Datalayer") -> None: | ||
| """Post-create hook. | ||
|
|
||
| :param db: Data layer instance. | ||
| """ | ||
| logging.info('Requesting vector index setup on CDC service') | ||
| if CFG.cluster.cdc.uri and not is_csn('cdc'): |
There was a problem hiding this comment.
csn means current service name
this check if the current process is cdc
if yes then donot create a cdc trigger
There was a problem hiding this comment.
How will this work if the services are in a single server?
There was a problem hiding this comment.
check the function, is_csn
When single server is used via testing
I check for it
superduper/backends/base/compute.py
Outdated
| def connect(self, *args, **kwargs): | ||
| pass | ||
|
|
||
| def create_handler(self, component_id, compute_kwargs={}): |
superduper/backends/base/compute.py
Outdated
| @@ -85,3 +85,9 @@ def shutdown(self) -> None: | |||
|
|
|||
| def execute_task(self, job_id, dependencies, compute_kwargs={}): | |||
| """Execute task function for distributed backends.""" | |||
|
|
|||
| def connect(self, *args, **kwargs): | |||
superduper/base/datalayer.py
Outdated
| @@ -195,7 +196,7 @@ def backfill_vector_search(self, vi, searcher): | |||
| if s.CFG.cluster.vector_search.type == 'native': | |||
| return | |||
|
|
|||
| if s.CFG.cluster.vector_search.uri and not self.server_mode: | |||
| if s.CFG.cluster.vector_search.uri and not is_csn('vector_search'): | |||
There was a problem hiding this comment.
What's this? Please add comments.
There was a problem hiding this comment.
CSN - current service name
this checks if current process is a vector search server.
superduper/base/datalayer.py
Outdated
| @@ -195,7 +196,7 @@ def backfill_vector_search(self, vi, searcher): | |||
| if s.CFG.cluster.vector_search.type == 'native': | |||
| return | |||
|
|
|||
| if s.CFG.cluster.vector_search.uri and not self.server_mode: | |||
| if s.CFG.cluster.vector_search.uri and not is_csn('vector_search'): | |||
There was a problem hiding this comment.
CSN - current service name
this checks if current process is a vector search server.
| @@ -72,21 +72,25 @@ def outputs_select(self): | |||
| """Get select statement for outputs.""" | |||
| return self.db[self.select.table].select().outputs(self.predict_id) | |||
|
|
|||
| @property | |||
| def cdc_table(self): | |||
There was a problem hiding this comment.
cdc table property which returns the component table to be listened by cdc service when added
| @override | ||
| def post_create(self, db: "Datalayer") -> None: | ||
| """Post-create hook. | ||
|
|
||
| :param db: Data layer instance. | ||
| """ | ||
| logging.info('Requesting vector index setup on CDC service') | ||
| if CFG.cluster.cdc.uri and not is_csn('cdc'): |
There was a problem hiding this comment.
check the function, is_csn
When single server is used via testing
I check for it
Description
Related Issues
Checklist
make unit_testingandmake integration-testingsuccessfully?Additional Notes or Comments