Skip to content

Commit dca3cca

Browse files
raphael-intuglesujayintugle
authored andcommitted
fix: harden SQL identifier handling across adapters
1 parent adeb7cb commit dca3cca

10 files changed

Lines changed: 437 additions & 151 deletions

File tree

src/intugle/adapters/types/bigquery/bigquery.py

Lines changed: 44 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import time
2+
23
from typing import TYPE_CHECKING, Any, Optional
34

45
import numpy as np
@@ -8,7 +9,12 @@
89
from intugle.adapters.factory import AdapterFactory
910
from intugle.adapters.models import ColumnProfile, DataSetData, ProfilingOutput
1011
from intugle.adapters.types.bigquery.models import BigQueryConfig, BigQueryConnectionConfig
11-
from intugle.adapters.utils import convert_to_native
12+
from intugle.adapters.utils import (
13+
convert_to_native,
14+
quote_identifier,
15+
quote_identifier_parts,
16+
split_identifier_path,
17+
)
1218
from intugle.core import settings
1319
from intugle.core.utilities.processing import string_standardization
1420

@@ -97,16 +103,13 @@ def connect(self):
97103

98104
def _get_fqn(self, identifier: str) -> str:
99105
"""Gets the fully qualified name for a table identifier."""
100-
if "." in identifier:
101-
# Already has project or dataset prefix
102-
parts = identifier.split(".")
103-
if len(parts) == 2:
104-
# dataset.table format
105-
return f"`{self._project_id}.{identifier}`"
106-
elif len(parts) == 3:
107-
# project.dataset.table format
108-
return f"`{identifier}`"
109-
return f"`{self._project_id}.{self._dataset_id}.{identifier}`"
106+
parts = split_identifier_path(identifier, max_parts=3)
107+
if len(parts) == 1:
108+
parts = [self._project_id, self._dataset_id, parts[0]]
109+
elif len(parts) == 2:
110+
parts = [self._project_id, parts[0], parts[1]]
111+
112+
return quote_identifier_parts(parts, quote_char="`", compound=True)
110113

111114
@staticmethod
112115
def check_data(data: Any) -> BigQueryConfig:
@@ -133,20 +136,32 @@ def profile(self, data: BigQueryConfig, table_name: str) -> ProfilingOutput:
133136
"""Profile a BigQuery table."""
134137
data = self.check_data(data)
135138
fqn = self._get_fqn(data.identifier)
139+
identifier_parts = split_identifier_path(data.identifier, max_parts=3)
140+
if len(identifier_parts) == 1:
141+
project_id, dataset_id, table_identifier = self._project_id, self._dataset_id, identifier_parts[0]
142+
elif len(identifier_parts) == 2:
143+
project_id, dataset_id, table_identifier = self._project_id, identifier_parts[0], identifier_parts[1]
144+
else:
145+
project_id, dataset_id, table_identifier = identifier_parts
136146

137147
# Get total count
138148
count_query = f"SELECT COUNT(*) as count FROM {fqn}"
139149
total_count = self._execute_sql(count_query)[0]["count"]
140150

141151
# Get column information from INFORMATION_SCHEMA
152+
information_schema = quote_identifier_parts(
153+
[project_id, dataset_id, "INFORMATION_SCHEMA", "COLUMNS"],
154+
quote_char="`",
155+
compound=True,
156+
)
142157
schema_query = f"""
143158
SELECT column_name, data_type
144-
FROM `{self._project_id}.{self._dataset_id}.INFORMATION_SCHEMA.COLUMNS`
159+
FROM {information_schema}
145160
WHERE table_name = @table_name
146161
ORDER BY ordinal_position
147162
"""
148163
job_config = bigquery.QueryJobConfig(
149-
query_parameters=[bigquery.ScalarQueryParameter("table_name", "STRING", data.identifier)]
164+
query_parameters=[bigquery.ScalarQueryParameter("table_name", "STRING", table_identifier)]
150165
)
151166
query_job = self.client.query(schema_query, job_config=job_config)
152167
rows = [dict(row) for row in query_job.result()]
@@ -172,13 +187,14 @@ def column_profile(
172187
"""Profile a specific column in a BigQuery table."""
173188
data = self.check_data(data)
174189
fqn = self._get_fqn(data.identifier)
190+
safe_column_name = quote_identifier(column_name, quote_char="`")
175191
start_ts = time.time()
176192

177193
# Null and distinct counts
178194
query = f"""
179195
SELECT
180-
COUNTIF(`{column_name}` IS NULL) as null_count,
181-
COUNT(DISTINCT `{column_name}`) as distinct_count
196+
COUNTIF({safe_column_name} IS NULL) as null_count,
197+
COUNT(DISTINCT {safe_column_name}) as distinct_count
182198
FROM {fqn}
183199
"""
184200
result = self._execute_sql(query)[0]
@@ -188,9 +204,9 @@ def column_profile(
188204

189205
# Sampling for distinct values
190206
sample_query = f"""
191-
SELECT DISTINCT CAST(`{column_name}` AS STRING) as value
207+
SELECT DISTINCT CAST({safe_column_name} AS STRING) as value
192208
FROM {fqn}
193-
WHERE `{column_name}` IS NOT NULL
209+
WHERE {safe_column_name} IS NOT NULL
194210
LIMIT {dtype_sample_limit}
195211
"""
196212
distinct_values_result = self._execute_sql(sample_query)
@@ -209,9 +225,9 @@ def column_profile(
209225
remaining_sample_size = dtype_sample_limit - len(distinct_values)
210226
if remaining_sample_size > 0:
211227
additional_samples_query = f"""
212-
SELECT CAST(`{column_name}` AS STRING) as value
228+
SELECT CAST({safe_column_name} AS STRING) as value
213229
FROM {fqn}
214-
WHERE `{column_name}` IS NOT NULL
230+
WHERE {safe_column_name} IS NOT NULL
215231
ORDER BY RAND()
216232
LIMIT {remaining_sample_size}
217233
"""
@@ -295,18 +311,20 @@ def intersect_count(
295311
data2 = self.check_data(table2.data)
296312
fqn1 = self._get_fqn(data1.identifier)
297313
fqn2 = self._get_fqn(data2.identifier)
314+
col1 = quote_identifier(column1_name, quote_char="`")
315+
col2 = quote_identifier(column2_name, quote_char="`")
298316

299317
query = f"""
300318
SELECT COUNT(*) as count
301319
FROM (
302-
SELECT DISTINCT `{column1_name}` as key
320+
SELECT DISTINCT {col1} as key
303321
FROM {fqn1}
304-
WHERE `{column1_name}` IS NOT NULL
322+
WHERE {col1} IS NOT NULL
305323
) t1
306324
INNER JOIN (
307-
SELECT DISTINCT `{column2_name}` as key
325+
SELECT DISTINCT {col2} as key
308326
FROM {fqn2}
309-
WHERE `{column2_name}` IS NOT NULL
327+
WHERE {col2} IS NOT NULL
310328
) t2
311329
ON t1.key = t2.key
312330
"""
@@ -319,8 +337,7 @@ def get_composite_key_uniqueness(
319337
data = self.check_data(dataset_data)
320338
fqn = self._get_fqn(data.identifier)
321339

322-
# Build column list with backticks
323-
safe_columns = [f"`{col}`" for col in columns]
340+
safe_columns = [quote_identifier(col, quote_char="`") for col in columns]
324341
columns_str = ", ".join(safe_columns)
325342

326343
# Build null filter
@@ -352,9 +369,8 @@ def intersect_composite_keys_count(
352369
fqn1 = self._get_fqn(data1.identifier)
353370
fqn2 = self._get_fqn(data2.identifier)
354371

355-
# Build column lists with backticks
356-
safe_columns1 = [f"`{col}`" for col in columns1]
357-
safe_columns2 = [f"`{col}`" for col in columns2]
372+
safe_columns1 = [quote_identifier(col, quote_char="`") for col in columns1]
373+
safe_columns2 = [quote_identifier(col, quote_char="`") for col in columns2]
358374

359375
# Subquery for distinct keys from table 1
360376
distinct_cols1 = ", ".join(safe_columns1)

0 commit comments

Comments
 (0)