Feature/add pg source #159
Conversation
|
|
||
| session = self._service.get_service_session() | ||
| try: | ||
| logs = session.query(Logs).filter(query_filter).order_by(*Logs.__query_order__).all() |
There was a problem hiding this comment.
I'm a bit worry about query performance here.
Also do you need order since you are getting all the rows? Or maybe do in memory ordering?
There was a problem hiding this comment.
I will add partition key "timestamp" to the query to enhance its performance.
The order statement would only sort the results set, and I use it for uniform processing.
If the order statement has a significant negative impact on execution, I will do in memory ordering.
| session = self._service.get_service_session() | ||
| try: | ||
| transactions = ( | ||
| session.query(Transactions).filter(query_filter).order_by(*Transactions.__query_order__).all() |
There was a problem hiding this comment.
Same question here. Need to test performance
There was a problem hiding this comment.
Check whether enough index.
There was a problem hiding this comment.
Index has been checked.
I will add partition key "timestamp" to the query to enhance its performance.
| if self._service is None: | ||
| raise FastShutdownError("-pg or --postgres-url is required to run PGSourceJob") | ||
|
|
||
| self.build_dependency = {} |
There was a problem hiding this comment.
What does this build_dependency do?
There was a problem hiding this comment.
This is legacy code in development and will be removed in the next commit.
cli/stream.py
Outdated
| output_types = list(set(parse_output_types)) | ||
| if source_types is None and source_path.startwith("postgresql://"): | ||
| source_types = "block,transaction,log" | ||
| if source_types: |
There was a problem hiding this comment.
Should we add this extra step or we should enforce user to list all the source types?
There was a problem hiding this comment.
Instead of this step, I set a default value to “--source-types”
| @@ -375,6 +400,9 @@ def stream( | |||
| batch_web3_provider=ThreadLocalProxy(lambda: get_provider_from_uri(provider_uri, batch=False)), | |||
| job_scheduler=job_scheduler, | |||
| sync_recorder=create_recorder(sync_recorder, config), | |||
| limit_reader=create_limit_reader( | |||
There was a problem hiding this comment.
what if user put other value in source_path?
There was a problem hiding this comment.
In unexpected cases, limit_reader will request the latest block from rpc.
if source_path specify a csv_source, parameter checking would ensure that the rpc_limit_reader does not interfere with the program
No description provided.