From 360df9ed8fba8614733e7a8e642b8c21c5e04cb6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Wed, 9 Sep 2026 12:03:57 +0300 Subject: [PATCH] cleanup: remove unreachable view execution branch --- src/openstatspec/sql/workflow.py | 106 ++++++++++++------------------- 1 file changed, 41 insertions(+), 65 deletions(-) diff --git a/src/openstatspec/sql/workflow.py b/src/openstatspec/sql/workflow.py index 7fbb154..4d6c954 100644 --- a/src/openstatspec/sql/workflow.py +++ b/src/openstatspec/sql/workflow.py @@ -2053,72 +2053,48 @@ def execute_transformation( preflight(profile, variables) row_count = 0 phase = "staging" - if output_mode == "materialized": - relation = Table( - staging_name, MetaData(), - Column("__row_ordinal", BigInteger, primary_key=True, nullable=False), - *(Column( - item["physical_name"], - binary64_type() if item["storage_kind"] == "numeric" else Text(), - nullable=item["is_nullable"], - ) for item in schema), - ) - relation.create(connection) - with _sqlite_read_authorizer(connection, parent_reads): - query_result = connection.exec_driver_sql( - full_query, values, - execution_options={"stream_results": True}, - ) - while True: - chunk = query_result.fetchmany(1000) - if not chunk: - break - rows = [] - for raw in chunk: - row_count += 1 - output = {"__row_ordinal": row_count} - for item, value in zip(schema, tuple(raw)): - if value is not None: - value = float(value) if item["storage_kind"] == "numeric" else str(value) - output[item["physical_name"]] = value - rows.append(output) - connection.execute(insert(relation), rows) - relation_query = ( - f"SELECT {', '.join(quote(item['physical_name']) for item in schema)} " - f"FROM {quote(staging_name)}" - ) - _validate_order_key( - connection, relation_query, order_descriptors, {}, {("main", staging_name)} - ) - connection.exec_driver_sql( - f"ALTER TABLE {quote(staging_name)} RENAME TO {quote(relation_name)}" - ) - _create_derived_triggers( - connection, derived_id, relation_name - ) - else: - projected = ", ".join( - f"q.{quote(item['name'])} AS {quote(item['physical_name'])}" - for item in schema - ) - window_order = ", ".join( - ( - f"q.{quote(item['expression'])}" - + (f" COLLATE {item['collation']}" if item["collation"] else "") - + f" {item['direction'].upper()} NULLS {item['nulls'].upper()}" - ) - for item in order_descriptors - ) - view_sql = ( - f"CREATE VIEW {quote(relation_name)} AS " - f"SELECT ROW_NUMBER() OVER (ORDER BY {window_order}) " - f"AS {quote('__row_ordinal')}, {projected} " - f"FROM ({full_query}) q" + relation = Table( + staging_name, MetaData(), + Column("__row_ordinal", BigInteger, primary_key=True, nullable=False), + *(Column( + item["physical_name"], + binary64_type() if item["storage_kind"] == "numeric" else Text(), + nullable=item["is_nullable"], + ) for item in schema), + ) + relation.create(connection) + with _sqlite_read_authorizer(connection, parent_reads): + query_result = connection.exec_driver_sql( + full_query, values, + execution_options={"stream_results": True}, ) - connection.exec_driver_sql(view_sql) - row_count = int(connection.execute( - text(f"SELECT COUNT(*) FROM {quote(relation_name)}") - ).scalar_one()) + while True: + chunk = query_result.fetchmany(1000) + if not chunk: + break + rows = [] + for raw in chunk: + row_count += 1 + output = {"__row_ordinal": row_count} + for item, value in zip(schema, tuple(raw)): + if value is not None: + value = float(value) if item["storage_kind"] == "numeric" else str(value) + output[item["physical_name"]] = value + rows.append(output) + connection.execute(insert(relation), rows) + relation_query = ( + f"SELECT {', '.join(quote(item['physical_name']) for item in schema)} " + f"FROM {quote(staging_name)}" + ) + _validate_order_key( + connection, relation_query, order_descriptors, {}, {("main", staging_name)} + ) + connection.exec_driver_sql( + f"ALTER TABLE {quote(staging_name)} RENAME TO {quote(relation_name)}" + ) + _create_derived_triggers( + connection, derived_id, relation_name + ) phase = "publication_validation" content_hash = _relation_snapshot_hash( connection, relation_schema=None, relation_name=relation_name,