Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Hotfix v0.2087.3 #461

Merged
merged 1 commit into from
May 24, 2024
+26 −30
Merged

Hotfix v0.2087.3 #461

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
56 changes: 26 additions & 30 deletions lib/dl_connector_mysql/dl_connector_mysql/core/async_adapters_mysql.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,40 +163,36 @@ async def _execute_by_steps(self, db_adapter_query: DBAdapterQuery) -> AsyncIter
debug_query = query if isinstance(query, str) else compile_query_for_debug(query, self._dialect)

with self.handle_execution_error(debug_query):
async with self._get_connection(
db_adapter_query.db_name
) as conn: # type: aiomysql.sa.connection.SAConnection
async with conn.begin():
await conn.execute("SET SESSION TRANSACTION READ ONLY")
result = await conn.execute(compiled_query, compiled_query_parameters)
cursor_info = ExecutionStepCursorInfo(
cursor_info=self._make_cursor_info(result.cursor),
raw_cursor_description=list(result.cursor.description),
)
yield cursor_info

row_converters = self._get_row_converters(cursor_info=cursor_info)
while True:
LOGGER.info("Fetching %s rows (conn %s)", chunk_size, conn)
rows = await result.fetchmany(chunk_size)
if not rows:
LOGGER.info("No rows remaining")
break

LOGGER.info("Rows fetched, yielding")
yield ExecutionStepDataChunk(
async with self._get_connection(db_adapter_query.db_name) as conn:
result = await conn.execute(compiled_query, compiled_query_parameters)
cursor_info = ExecutionStepCursorInfo(
cursor_info=self._make_cursor_info(result.cursor),
raw_cursor_description=list(result.cursor.description),
)
yield cursor_info

row_converters = self._get_row_converters(cursor_info=cursor_info)
while True:
LOGGER.info("Fetching %s rows (conn %s)", chunk_size, conn)
rows = await result.fetchmany(chunk_size)
if not rows:
LOGGER.info("No rows remaining")
break

LOGGER.info("Rows fetched, yielding")
yield ExecutionStepDataChunk(
tuple(
tuple(
tuple(
(col_converter(val) if col_converter is not None and val is not None else val)
for val, col_converter in zip(
[row[col_name] for col_name in cursor_info.cursor_info["names"]],
row_converters,
strict=True,
)
(col_converter(val) if col_converter is not None and val is not None else val)
for val, col_converter in zip(
[row[col_name] for col_name in cursor_info.cursor_info["names"]],
row_converters,
strict=True,
)
for row in rows
)
for row in rows
)
)

@generic_profiler_async("db-full") # type: ignore # TODO: fix
async def execute(self, query: DBAdapterQuery) -> AsyncRawExecutionResult:
Expand Down
Loading