Skip to content

Commit e2671f3

Browse files
committed
fix(table_diff): support key columns stored with non-normalized casing
Signed-off-by: Anant <75747269+Anant-gif@users.noreply.github.com>
1 parent 2c30f83 commit e2671f3

2 files changed

Lines changed: 175 additions & 3 deletions

File tree

sqlmesh/core/table_diff.py

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -282,9 +282,11 @@ def key_columns(self) -> t.Tuple[t.List[exp.Column], t.List[exp.Column], t.List[
282282
# If the columns to join on are explicitly specified, then just return them
283283
if isinstance(self._on, (list, tuple)):
284284
identifiers = [normalize_identifiers(c, dialect=dialect) for c in self._on]
285-
s_index = [exp.column(c, "s") for c in identifiers]
286-
t_index = [exp.column(c, "t") for c in identifiers]
287-
return s_index, t_index, [i.name for i in identifiers]
285+
s_names = [self._resolve_column_name(c.name, self.source_schema) for c in identifiers]
286+
t_names = [self._resolve_column_name(c.name, self.target_schema) for c in identifiers]
287+
s_index = [exp.column(c, "s") for c in s_names]
288+
t_index = [exp.column(c, "t") for c in t_names]
289+
return s_index, t_index, s_names
288290

289291
# Otherwise, we need to parse them out of the supplied "on" condition
290292
index_cols = []
@@ -295,16 +297,26 @@ def key_columns(self) -> t.Tuple[t.List[exp.Column], t.List[exp.Column], t.List[
295297
for col in self._on.find_all(exp.Column):
296298
index_cols.append(col.name)
297299
if col.table.lower() == "s":
300+
col = exp.column(self._resolve_column_name(col.name, self.source_schema), col.table)
298301
s_index.append(col)
299302
elif col.table.lower() == "t":
303+
col = exp.column(self._resolve_column_name(col.name, self.target_schema), col.table)
300304
t_index.append(col)
305+
index_cols.append(col.name)
301306

302307
index_cols = list(dict.fromkeys(index_cols))
303308
s_index = list(dict.fromkeys(s_index))
304309
t_index = list(dict.fromkeys(t_index))
305310

306311
return s_index, t_index, index_cols
307312

313+
def _resolve_column_name(self, name: str, schema: t.Dict[str, exp.DataType]) -> str:
314+
if name in schema:
315+
return name
316+
317+
lowercase_name = name.lower()
318+
return next((c for c in schema if c.lower() == lowercase_name), name)
319+
308320
@property
309321
def source_key_expression(self) -> exp.Expr:
310322
s_index, _, _ = self.key_columns

tests/core/test_table_diff.py

Lines changed: 160 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1197,6 +1197,166 @@ def test_data_diff_sample_limit():
11971197
assert len(diff.joined_sample) == 3
11981198

11991199

1200+
def test_data_diff_non_lowercase_key_columns():
1201+
engine_adapter = DuckDBConnectionConfig().create_engine_adapter()
1202+
1203+
columns_to_types = {
1204+
"KEY1": exp.DataType.build("int"),
1205+
"Key2": exp.DataType.build("varchar"),
1206+
"VALUE": exp.DataType.build("varchar"),
1207+
}
1208+
1209+
engine_adapter.create_table("src", columns_to_types)
1210+
engine_adapter.create_table("target", columns_to_types)
1211+
1212+
src_records = [
1213+
(1, "a", "value"),
1214+
(2, "b", "source"),
1215+
(3, "c", "source only"),
1216+
]
1217+
1218+
target_records = [
1219+
(1, "a", "value"),
1220+
(2, "b", "target"),
1221+
(4, "d", "target only"),
1222+
]
1223+
1224+
src_df = pd.DataFrame(data=src_records, columns=columns_to_types.keys())
1225+
target_df = pd.DataFrame(data=target_records, columns=columns_to_types.keys())
1226+
1227+
engine_adapter.insert_append("src", src_df)
1228+
engine_adapter.insert_append("target", target_df)
1229+
1230+
# casing of the supplied key should not matter
1231+
for on in (["KEY1", "Key2"], ["key1", "KEY2"]):
1232+
table_diff = TableDiff(adapter=engine_adapter, source="src", target="target", on=on)
1233+
1234+
_, _, col_names = table_diff.key_columns
1235+
assert col_names == ["KEY1", "Key2"]
1236+
1237+
diff = table_diff.row_diff()
1238+
1239+
assert diff.join_count == 2
1240+
assert diff.full_match_count == 1
1241+
assert diff.partial_match_count == 1
1242+
assert diff.s_only_count == 1
1243+
assert diff.t_only_count == 1
1244+
1245+
assert diff.s_sample["VALUE"].tolist() == ["source only"]
1246+
assert diff.t_sample["VALUE"].tolist() == ["target only"]
1247+
assert diff.joined_sample[["s_VALUE", "t_VALUE"]].values.flatten().tolist() == [
1248+
"source",
1249+
"target",
1250+
]
1251+
1252+
table_diff = TableDiff(adapter=engine_adapter, source="src", target="target", on=["KEY1"])
1253+
1254+
_, _, col_names = table_diff.key_columns
1255+
assert col_names == ["KEY1"]
1256+
1257+
diff = table_diff.row_diff()
1258+
1259+
assert diff.join_count == 2
1260+
assert diff.full_match_count == 1
1261+
assert diff.partial_match_count == 1
1262+
assert diff.s_only_count == 1
1263+
assert diff.t_only_count == 1
1264+
1265+
1266+
def test_data_diff_key_columns_with_differing_case_between_source_and_target():
1267+
engine_adapter = DuckDBConnectionConfig().create_engine_adapter()
1268+
1269+
source_columns_to_types = {
1270+
"KEY1": exp.DataType.build("int"),
1271+
"Key2": exp.DataType.build("varchar"),
1272+
"value": exp.DataType.build("varchar"),
1273+
}
1274+
target_columns_to_types = {
1275+
"key1": exp.DataType.build("int"),
1276+
"KEY2": exp.DataType.build("varchar"),
1277+
"value": exp.DataType.build("varchar"),
1278+
}
1279+
1280+
engine_adapter.create_table("src", source_columns_to_types)
1281+
engine_adapter.create_table("target", target_columns_to_types)
1282+
1283+
engine_adapter.insert_append(
1284+
"src",
1285+
pd.DataFrame(
1286+
data=[(1, "a", "value"), (2, "b", "source")],
1287+
columns=source_columns_to_types.keys(),
1288+
),
1289+
)
1290+
engine_adapter.insert_append(
1291+
"target",
1292+
pd.DataFrame(
1293+
data=[(1, "a", "value"), (2, "b", "target")],
1294+
columns=target_columns_to_types.keys(),
1295+
),
1296+
)
1297+
1298+
table_diff = TableDiff(
1299+
adapter=engine_adapter, source="src", target="target", on=["key1", "KEY2"]
1300+
)
1301+
1302+
s_index, t_index, col_names = table_diff.key_columns
1303+
assert [c.sql() for c in s_index] == ['"s.KEY1"', '"s.Key2"']
1304+
assert [c.sql() for c in t_index] == ['"t.key1"', '"t.KEY2"']
1305+
assert col_names == ["KEY1", "Key2"]
1306+
1307+
diff = table_diff.row_diff()
1308+
1309+
assert diff.join_count == 2
1310+
assert diff.full_match_count == 1
1311+
assert diff.partial_match_count == 1
1312+
assert diff.s_only_count == 0
1313+
assert diff.t_only_count == 0
1314+
1315+
# the key columns are excluded from the per column match stats
1316+
assert diff.column_stats.index.tolist() == ["value"]
1317+
1318+
1319+
def test_data_diff_non_lowercase_key_columns_in_on_condition():
1320+
engine_adapter = DuckDBConnectionConfig().create_engine_adapter()
1321+
1322+
columns_to_types = {
1323+
"KEY1": exp.DataType.build("int"),
1324+
"Key2": exp.DataType.build("varchar"),
1325+
"VALUE": exp.DataType.build("varchar"),
1326+
}
1327+
1328+
engine_adapter.create_table("src", columns_to_types)
1329+
engine_adapter.create_table("target", columns_to_types)
1330+
1331+
src_df = pd.DataFrame(
1332+
data=[(1, "a", "value"), (2, "b", "source")], columns=columns_to_types.keys()
1333+
)
1334+
target_df = pd.DataFrame(
1335+
data=[(1, "a", "value"), (2, "b", "target")], columns=columns_to_types.keys()
1336+
)
1337+
1338+
engine_adapter.insert_append("src", src_df)
1339+
engine_adapter.insert_append("target", target_df)
1340+
1341+
table_diff = TableDiff(
1342+
adapter=engine_adapter,
1343+
source="src",
1344+
target="target",
1345+
on=exp.condition('s."KEY1" = t."KEY1" AND s."Key2" = t."Key2"'),
1346+
)
1347+
1348+
_, col_names = table_diff.key_columns
1349+
assert col_names == ["KEY1", "Key2"]
1350+
1351+
diff = table_diff.row_diff()
1352+
1353+
assert diff.join_count == 2
1354+
assert diff.full_match_count == 1
1355+
assert diff.partial_match_count == 1
1356+
assert diff.s_only_count == 0
1357+
assert diff.t_only_count == 0
1358+
1359+
12001360
def test_data_diff_nulls_in_some_grain_columns():
12011361
engine_adapter = DuckDBConnectionConfig().create_engine_adapter()
12021362

0 commit comments

Comments
 (0)