Skip to content
This repository was archived by the owner on May 30, 2022. It is now read-only.

Performance improvements #1

Open
laurentS wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Performance improvements #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
dd37227
Cache _serialize_table_record_field_name to speedup loading
laurentS Apr 4, 2021
ffc8976
Cache datetime formatting function
laurentS Apr 4, 2021
7041e5c
Use pickle instead of deepcopy to speedup row copying
laurentS Apr 4, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion target_postgres/postgres.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from copy import deepcopy
import csv
from functools import lru_cache
import io
import json
import logging
Expand All @@ -19,6 +20,14 @@

RESERVED_NULL_DEFAULT = 'NULL'

@lru_cache(maxsize=128)

@ericboucher ericboucher Feb 9, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

let's make maxsize a constant at the top maybe?

def _format_datetime(value):
"""
Format a datetime value. This is only called from the
PostgresTarget.serialize_table_record_datetime_value
but this non-method version allows caching
"""
return arrow.get(value).format('YYYY-MM-DD HH:mm:ss.SSSSZZ')

def _update_schema_0_to_1(table_metadata, table_schema):
"""
Expand Down Expand Up @@ -546,7 +555,7 @@ def serialize_table_record_null_value(self, remote_schema, streamed_schema, fiel
return value

def serialize_table_record_datetime_value(self, remote_schema, streamed_schema, field, value):
return arrow.get(value).format('YYYY-MM-DD HH:mm:ss.SSSSZZ')
return _format_datetime(value)

def persist_csv_rows(self,
cur,
Expand Down
31 changes: 21 additions & 10 deletions target_postgres/sql_base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@
#

from copy import deepcopy
from functools import lru_cache, partial
import pickle
import time

import singer
Expand Down Expand Up @@ -628,16 +630,23 @@ def log_message(msg):

return self._get_table_schema(connection, table_name)

def _serialize_table_record_field_name(self, remote_schema, path, value_json_schema):
def _serialize_table_record_field_name(self, remote_schema, path, value_json_schema_tuple):
"""
Returns the appropriate remote field (column) name for `path`.

:param remote_schema: TABLE_SCHEMA(remote)
:param path: (string, ...)
:value_json_schema: dict, JSON Schema
:value_json_schema: tuple, JSON Schema
:return: string
"""

# rebuild the dict that needs to be passed further down the call stack
if len(value_json_schema_tuple) == 1:
value_json_schema = { 'type': value_json_schema_tuple[0] }
else:
value_json_schema = {'type': value_json_schema_tuple[0],
'format': value_json_schema_tuple[1]}

simple_json_schema = json_schema.simple_type(value_json_schema)

mapping = self._get_mapping(remote_schema,
Expand Down Expand Up @@ -725,9 +734,15 @@ def _serialize_table_records(
default_row = dict([(field, NULL_DEFAULT) for field in remote_fields])

paths = streamed_schema['schema']['properties'].keys()

## create a partial function with only hashable args so we can use lru_cache on it
_cached_field_name = partial(self._serialize_table_record_field_name, remote_schema)
cached_field_name = lru_cache(maxsize=None)(_cached_field_name)

for record in records:

row = deepcopy(default_row)
## pickling/unpickling is much faster than deepcopy
row = pickle.loads(pickle.dumps(default_row))

for path in paths:
json_schema_string_type, value = record.get(path, (None, None))
Expand All @@ -747,18 +762,14 @@ def _serialize_table_records(
and value is not None:
value = self.serialize_table_record_datetime_value(remote_schema, streamed_schema, path,
value)
value_json_schema = {'type': json_schema.STRING,
'format': json_schema.DATE_TIME_FORMAT}
value_json_schema_tuple = (json_schema.STRING, json_schema.DATE_TIME_FORMAT)
field_name = cached_field_name(path, value_json_schema_tuple)
else:
value_json_schema = {'type': json_schema_string_type}
field_name = cached_field_name(path, (json_schema_string_type,))

## Serialize NULL default value
value = self.serialize_table_record_null_value(remote_schema, streamed_schema, path, value)

field_name = self._serialize_table_record_field_name(remote_schema,
path,
value_json_schema)

## `field_name` is unset
if row[field_name] == NULL_DEFAULT:
row[field_name] = value
Expand Down