From b520678c757ddf509efa1f710f2ae68a0315708d Mon Sep 17 00:00:00 2001 From: Tej Kashi Date: Wed, 30 Jul 2025 13:18:09 -0400 Subject: [PATCH 1/2] [ACE-103] Conservative datatype handling * Handle transactions better while updating merkle trees --- cli/scripts/ace.py | 129 ++++++++++++++++++-------------------- cli/scripts/ace_config.py | 3 + cli/scripts/ace_core.py | 8 ++- cli/scripts/ace_mtree.py | 9 +-- 4 files changed, 73 insertions(+), 76 deletions(-) diff --git a/cli/scripts/ace.py b/cli/scripts/ace.py index ecf50d43..0fed3a05 100755 --- a/cli/scripts/ace.py +++ b/cli/scripts/ace.py @@ -727,63 +727,47 @@ def check_diff_file_format(diff_file_path: str, task) -> dict: return diff_json -def convert_pg_type_to_json(item: str, type: str): +def convert_pg_type_to_json(item, type): """ Converts a value from a postgres column to a json-compatible type. """ # TODO: Need to revisit this. - try: - # List of types that should be treated as strings - string_types = [ - "char", - "text", - "time", - "bytea", - "uuid", - "date", - "timestamp", - "interval", - "inet", - "macaddr", - "xml", - "money", - "point", - "line", - "polygon", - ] - - # Types that can be directly represented in JSON - json_compatible_types = [ - "json", - "jsonb", - "boolean", - "integer", - "bigint", - "smallint", - "numeric", - "real", - "double precision", - ] - - type_lower = type.lower() - - if not item or item == "" or item.lower() == "null" or item.lower() == "none": - return None - elif "[]" in type_lower: - return ast.literal_eval(item) - elif any(s in type_lower for s in json_compatible_types): - # For JSON-compatible types, parse them using AST - return ast.literal_eval(item) - elif any(s in type_lower for s in string_types): - return item - else: - # Default to treating as string if type is unknown - return item + type_lower = type.lower() - except Exception as e: - raise AceException( - f"Could not convert value {item} to {type} while writing to json: {e}" - ) + # Types that should be parsed into native JSON types (not strings) + json_compatible_types = [ + "json", + "jsonb", + "boolean", + "integer", + "bigint", + "smallint", + "numeric", + "real", + "double precision", + ] + + is_parsable = ( + any(s in type_lower for s in json_compatible_types) or "[]" in type_lower + ) + + if not is_parsable: + # For string-like types (text, varchar, etc.), we return the value + # directly. This correctly preserves string literals like 'None' or + # 'null'. A database NULL would arrive here as item=None from the driver. + return item + + # For parsable types (numeric, boolean, json, array), we can interpret + # 'null' and 'none'. + if item is None or str(item).lower() in ("", "null", "none"): + return None + + try: + # For JSON-compatible types, parse them using AST + return ast.literal_eval(str(item)) + except (ValueError, SyntaxError): + # If conversion fails, treat as a string + return str(item) def convert_json_to_pg_type(rows, cols_list, col_types) -> list[tuple]: @@ -841,31 +825,40 @@ def convert_json_to_pg_type(rows, cols_list, col_types) -> list[tuple]: modified_row = tuple() for col_name in cols_list: col_type = col_types[col_name] - elem = str(row[col_name]) + elem = row[col_name] + type_lower = col_type.lower() try: - type_lower = col_type.lower() - - if ( - not elem - or elem == "" - or elem.lower() == "null" - or elem.lower() == "none" - ): - modified_row += (None,) - elif "[]" in type_lower: - modified_row += (ast.literal_eval(elem),) - elif any(s in type_lower for s in string_types): + # If the column type is a string type, we don't need to do anything + # special. A value of None will be converted to NULL by psycopg. + if any(s in type_lower for s in string_types): if type_lower == "bytea": - modified_row += (bytes.fromhex(elem),) + # We stored bytea as hex, so we need to convert it back + if elem is not None: + modified_row += (bytes.fromhex(elem),) + else: + modified_row += (None,) else: modified_row += (elem,) + continue + + # For non-string types, if the value is None, or looks like null, + # it should be treated as such. + if elem is None or str(elem).lower() in ("null", "none", ""): + modified_row += (None,) + continue + + elem_str = str(elem) + + if "[]" in type_lower: + modified_row += (ast.literal_eval(elem_str),) elif any(s in type_lower for s in json_compatible_types): - item = ast.literal_eval(elem) - if type_lower == "jsonb" or type_lower == "json": + item = ast.literal_eval(elem_str) + if type_lower in ("jsonb", "json"): item = json.dumps(item) modified_row += (item,) else: + # Fallback for any other types modified_row += (elem,) except (ValueError, SyntaxError): diff --git a/cli/scripts/ace_config.py b/cli/scripts/ace_config.py index 1580b1c4..39379e0e 100644 --- a/cli/scripts/ace_config.py +++ b/cli/scripts/ace_config.py @@ -12,6 +12,9 @@ STATEMENT_TIMEOUT = 0 # in milliseconds CONNECTION_TIMEOUT = 10 # in seconds +# Whether to use repeatable read isolation for Merkle tree updates +USE_REPEATABLE_READ = False + # Default values for ACE table-diff MAX_DIFF_ROWS = 1_000_000 MIN_DIFF_BLOCK_SIZE = 1000 diff --git a/cli/scripts/ace_core.py b/cli/scripts/ace_core.py index 91567668..b0c35b45 100644 --- a/cli/scripts/ace_core.py +++ b/cli/scripts/ace_core.py @@ -574,12 +574,16 @@ def compare_checksums(worker_id, shared_objects, worker_state, pkey1, pkey2): for row_key in t1_only: worker_diffs[node_pair_key][host1].append( - dict(zip(cols, (str(x) for x in row_key))) + dict( + zip(cols, (str(x) if x is not None else None for x in row_key)) + ) ) for row_key in t2_only: worker_diffs[node_pair_key][host2].append( - dict(zip(cols, (str(x) for x in row_key))) + dict( + zip(cols, (str(x) if x is not None else None for x in row_key)) + ) ) total_diffs += max(len(t1_only), len(t2_only)) diff --git a/cli/scripts/ace_mtree.py b/cli/scripts/ace_mtree.py index a621b66b..32eb7980 100644 --- a/cli/scripts/ace_mtree.py +++ b/cli/scripts/ace_mtree.py @@ -1025,7 +1025,6 @@ def split_blocks(conn, schema, table, key, blocks, block_size): i += 1 - conn.commit() pbar.close() return list(modified_positions) @@ -1399,7 +1398,6 @@ def merge_blocks(conn, schema, table, key, blocks, block_size): if i >= len(blocks): break - conn.commit() pbar.close() return list(modified_positions) @@ -1408,7 +1406,7 @@ def update_mtree(mtree_task: MerkleTreeTask, skip_all_checks=False) -> None: """ Update a Merkle tree by recomputing hashes for dirty leaf nodes and new blocks. Also processes any pending block rebalancing operations. - Uses repeatable read isolation to ensure consistency during the update. + Uses repeatable read isolation if config.USE_REPEATABLE_READ is True. Args: cluster_name (str): Name of the cluster @@ -1455,7 +1453,8 @@ def update_mtree(mtree_task: MerkleTreeTask, skip_all_checks=False) -> None: for node in mtree_task.fields.cluster_nodes: _, conn = mtree_task.connection_pool.connect(node) - conn.set_isolation_level(IsolationLevel.REPEATABLE_READ) + if config.USE_REPEATABLE_READ: + conn.set_isolation_level(IsolationLevel.REPEATABLE_READ) print(f"\nUpdating Merkle tree on node: {node['name']}") @@ -1476,7 +1475,6 @@ def update_mtree(mtree_task: MerkleTreeTask, skip_all_checks=False) -> None: if not blocks_to_update: print(f"No updates needed for {node['name']}") - conn.commit() continue # First identify blocks that might need splitting based on insert count @@ -1557,7 +1555,6 @@ def update_mtree(mtree_task: MerkleTreeTask, skip_all_checks=False) -> None: if not blocks_to_update: print(f"No updates needed for {node['name']}") - conn.commit() continue print(f"Found {len(blocks_to_update)} blocks to update") From fa5c44296d77954438c82ff937af80d2cdd40baa Mon Sep 17 00:00:00 2001 From: Tej Kashi Date: Thu, 31 Jul 2025 16:03:52 -0400 Subject: [PATCH 2/2] Add additional tests for datatypes --- cli/scripts/ace-tests/test_data_types.py | 632 +++++++++++++++++++---- 1 file changed, 521 insertions(+), 111 deletions(-) diff --git a/cli/scripts/ace-tests/test_data_types.py b/cli/scripts/ace-tests/test_data_types.py index 950bf2dd..ff5d7aec 100644 --- a/cli/scripts/ace-tests/test_data_types.py +++ b/cli/scripts/ace-tests/test_data_types.py @@ -1,3 +1,6 @@ +from datetime import datetime, timedelta, date, time +from decimal import Decimal +from ipaddress import IPv4Address import pytest import psycopg import json @@ -31,50 +34,25 @@ def setup_datatypes(self, nodes): bytea_col BYTEA, point_col POINT, text_col TEXT, - text_array_col TEXT[] + text_array_col TEXT[], + bool_col BOOLEAN, + bigint_col BIGINT, + smallint_col SMALLINT, + numeric_col NUMERIC(10, 4), + real_col REAL, + time_col TIME, + date_col DATE, + timestamp_col TIMESTAMP, + interval_col INTERVAL, + inet_col INET, + macaddr_col MACADDR, + money_col MONEY ) """ ) # Insert sample data - cur.execute( - """ - INSERT INTO datatypes_test VALUES - ( - 'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11', - 42, - 3.14159, - ARRAY[1, 2, 3, 4, 5], - '{"key": "value", "nested": {"foo": "bar"}}', - decode('DEADBEEF', 'hex'), - point(1.5, 2.5), - 'sample text', - ARRAY['apple', 'banana', 'cherry'] - ), - ( - 'b0eebc99-9c0b-4ef8-bb6d-6bb9bd380a12', - 100, - 2.71828, - ARRAY[10, 20, 30], - '{"numbers": [1, 2, 3], "active": true}', - decode('BADDCAFE', 'hex'), - point(3.7, 4.2), - 'another sample', - ARRAY['dog', 'cat', 'bird'] - ), - ( - 'c0eebc99-9c0b-4ef8-bb6d-6bb9bd380a13', - -17, - 0.577216, - ARRAY[]::INTEGER[], - '{"empty": true}', - NULL, - point(0, 0), - 'third sample', - ARRAY[]::TEXT[] - ) - """ - ) + self._insert_initial_data(cur) repset_add_datatypes_sql = """ SELECT spock.repset_add_table('test_repset', 'datatypes_test') @@ -106,6 +84,145 @@ def setup_datatypes(self, nodes): except Exception as e: pytest.fail(f"Failed to setup/cleanup datatypes test: {str(e)}") + def _insert_initial_data(self, cur): + """Helper method to insert the initial dataset.""" + cur.execute(self._get_initial_data_sql()) + + def _get_initial_data_sql(self): + """Returns the SQL for inserting the initial dataset.""" + return """ + INSERT INTO datatypes_test VALUES + ( + 'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11', + 42, + 3.14159, + ARRAY[1, 2, 3, 4, 5], + '{"key": "value", "nested": {"foo": "bar"}}', + decode('DEADBEEF', 'hex'), + point(1.5, 2.5), + 'sample text', + ARRAY['apple', 'banana', 'cherry'], + true, + 9223372036854775807, + 32767, + 12345.6789, + 123.456, + '12:34:56', + '2024-01-01', + '2024-01-01 12:34:56', + '30 days', + '192.168.1.1', + '08:00:2b:01:02:03', + 12345.67 + ), + ( + 'b0eebc99-9c0b-4ef8-bb6d-6bb9bd380a12', + 100, + 2.71828, + ARRAY[10, 20, 30], + '{"numbers": [1, 2, 3], "active": true}', + decode('BADDCAFE', 'hex'), + point(3.7, 4.2), + 'another sample', + ARRAY['dog', 'cat', 'bird'], + false, + -9223372036854775808, + -32768, + -12345.6789, + -123.456, + '23:59:59', + '2023-12-31', + '2023-12-31 23:59:59', + '-5 days', + '10.0.0.1', + '00:1A:2B:3C:4D:5E', + -12345.67 + ), + ( + 'c0eebc99-9c0b-4ef8-bb6d-6bb9bd380a13', + -17, + 0.577216, + ARRAY[]::INTEGER[], + '{"empty": true}', + NULL, + point(0, 0), + 'third sample', + ARRAY[]::TEXT[], + true, + 0, + 0, + 0, + 0, + '00:00:00', + '1970-01-01', + '1970-01-01 00:00:00', + '0 seconds', + '0.0.0.0', + '00:00:00:00:00:00', + 0 + ), + ( + 'd0eebc99-9c0b-4ef8-bb6d-6bb9bd380a14', + NULL, + NULL, + ARRAY[NULL, 1, NULL], + NULL, + NULL, + NULL, + 'null', + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL + ), + ( + 'e0eebc99-9c0b-4ef8-bb6d-6bb9bd380a15', + 1, + 'NaN', + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL, + NULL + ) + """ + + def reset_data(self, nodes): + """Reset data in the test table before each test function.""" + try: + for node in nodes: + conn = psycopg.connect(host=node, dbname="demo", user="admin") + cur = conn.cursor() + cur.execute("TRUNCATE TABLE datatypes_test") + self._insert_initial_data(cur) + conn.commit() + cur.close() + conn.close() + except Exception as e: + pytest.fail(f"Failed to reset data between tests: {str(e)}") + # Override the table_name parameter for all parameterized tests @pytest.mark.parametrize("table_name", ["public.datatypes_test"]) def test_simple_table_diff(self, cli, capsys, table_name): @@ -113,16 +230,28 @@ def test_simple_table_diff(self, cli, capsys, table_name): @pytest.mark.parametrize("table_name", ["public.datatypes_test"]) @pytest.mark.parametrize( - "column_name,test_value", + "column_name,test_value,expected_diffs", [ - ("int_col", "9999"), - ("float_col", "123.456"), - ("array_col", "ARRAY[99, 98, 97]"), - ("json_col", '\'{"test": "modified"}\''), - ("bytea_col", "decode('FEEDFACE', 'hex')"), - ("point_col", "point(99.9, 99.9)"), - ("text_col", "'modified-text'"), - ("text_array_col", "ARRAY['modified', 'text', 'array']"), + ("int_col", "9999", 5), + ("float_col", "123.456", 5), + ("array_col", "ARRAY[99, 98, 97]", 5), + ("json_col", '\'{"test": "modified"}\'', 5), + ("bytea_col", "decode('FEEDFACE', 'hex')", 5), + ("point_col", "point(99.9, 99.9)", 5), + ("text_col", "'modified-text'", 5), + ("text_array_col", "ARRAY['modified', 'text', 'array']", 5), + ("bool_col", "false", 5), + ("bigint_col", "1234567890123456789", 5), + ("smallint_col", "-32768", 5), + ("numeric_col", "98765.4321", 5), + ("real_col", "987.654", 5), + ("time_col", "'11:22:33'", 5), + ("date_col", "'2025-05-25'", 5), + ("timestamp_col", "'2025-05-25 11:22:33'", 5), + ("interval_col", "'90 days'", 5), + ("inet_col", "'192.168.100.200'", 5), + ("macaddr_col", "'01:23:45:67:89:ab'", 5), + ("money_col", "9876.54", 5), ], ) @pytest.mark.parametrize("key_column", ["id"]) @@ -134,6 +263,7 @@ def test_table_diff_with_differences( table_name, column_name, test_value, + expected_diffs, key_column, diff_file_path, ): @@ -154,11 +284,6 @@ def test_table_diff_with_differences( """ ) - modified_rows = cur.fetchall() - modified_indices = { - str(row[0]) for row in modified_rows - } # Convert UUID to string - conn.commit() cur.close() conn.close() @@ -185,55 +310,79 @@ def test_table_diff_with_differences( # Verify number of differences assert ( - len(diff_data["diffs"]["n1/n2"]["n2"]) == 3 - ), "Expected 3 differences," + len(diff_data["diffs"]["n1/n2"]["n2"]) == expected_diffs + ), f"Expected {expected_diffs} differences," f" found {len(diff_data['diffs']['n1/n2']['n2'])}" - # Verify modified rows are in diff - diff_indices = { - str(diff["id"]) for diff in diff_data["diffs"]["n1/n2"]["n2"] - } - assert ( - modified_indices == diff_indices - ), "Modified rows don't match diff file records" - # Verify the differences are correctly reported for diff in diff_data["diffs"]["n1/n2"]["n2"]: + diff_val = diff[column_name] + expected_val_str = test_value.strip("'") + if column_name == "json_col": assert ( - diff[column_name].get("test") == "modified" + diff_val.get("test") == "modified" ), f"Modified row {diff['id']} doesn't have expected JSON value" elif column_name == "array_col": - assert diff[column_name] == [ + assert diff_val == [ 99, 98, 97, ], f"Modified row {diff['id']} doesn't have expected array value" elif column_name == "text_array_col": - assert diff[column_name] == [ + assert diff_val == [ "modified", "text", "array", ], ( - f"Modified row {diff['id']} doesn't have expected " - "text array value" + f"Modified row {diff['id']} doesn't have expected text" + " array value" ) elif column_name == "point_col": assert ( - diff[column_name] == "(99.9,99.9)" + diff_val == "(99.9,99.9)" ), f"Modified row {diff['id']} doesn't have expected point value" elif column_name == "bytea_col": - print("bytea col: ", diff[column_name]) assert ( - diff[column_name] == "feedface" + diff_val == "feedface" ), f"Modified row {diff['id']} doesn't have expected bytea value" + elif column_name == "macaddr_col": + assert ( + diff_val == "01:23:45:67:89:ab" + ), f"Modified row {diff['id']} doesn't have expected macaddr value" + elif column_name == "money_col": + cleaned_diff = diff_val.replace("$", "").replace(",", "") + assert float(cleaned_diff) == float( + expected_val_str + ), f"Modified row {diff['id']} doesn't have expected money value" + elif column_name == "bool_col": + assert str(diff_val).lower() == expected_val_str.lower(), ( + f"Modified row {diff['id']} " + "doesn't have expected boolean value" + ) + elif column_name == "interval_col": + # Interval representation can vary, so we check equality + # directly in Postgres + conn = psycopg.connect(host="n1", dbname="demo", user="admin") + cur = conn.cursor() + cur.execute( + "SELECT %s::interval = %s::interval", + (str(diff_val), expected_val_str), + ) + is_equal = cur.fetchone()[0] + cur.close() + conn.close() + assert is_equal, ( + f"Modified row {diff['id']} " + f"doesn't have expected interval value. " + f"Got {diff_val}, expected equivalence to {expected_val_str}" + ) else: - assert str(diff[column_name]) in ( - test_value.strip("'"), - "9999", - "123.456", - "modified-text", - ), f"Modified row {diff['id']} doesn't have expected value" + assert str(diff_val) == expected_val_str, ( + f"Modified row {diff['id']} " + f"doesn't have expected value, got {diff_val} " + f"expected {expected_val_str}" + ) except Exception as e: pytest.fail(f"Failed to test differences for {column_name}: {str(e)}") @@ -245,16 +394,28 @@ def test_simple_table_repair(self, cli, capsys, table_name, diff_file_path): @pytest.mark.parametrize("table_name", ["public.datatypes_test"]) @pytest.mark.parametrize("key_column", ["id"]) @pytest.mark.parametrize( - "column_name,test_value", + "column_name,test_value,expected_rerun_diffs", [ - ("int_col", "1234"), - ("float_col", "98.765"), - ("array_col", "ARRAY[11, 22, 33]"), - ("json_col", '\'{"rerun": "modified"}\''), - ("bytea_col", "decode('ABCDEF12', 'hex')"), - ("point_col", "point(88.8, 88.8)"), - ("text_col", "'rerun-modified'"), - ("text_array_col", "ARRAY['rerun', 'modified', 'array']"), + ("int_col", "1234", 5), + ("float_col", "98.765", 5), + ("array_col", "ARRAY[11, 22, 33]", 5), + ("json_col", '\'{"rerun": "modified"}\'', 5), + ("bytea_col", "decode('ABCDEF12', 'hex')", 5), + ("point_col", "point(88.8, 88.8)", 5), + ("text_col", "'rerun-modified'", 5), + ("text_array_col", "ARRAY['rerun', 'modified', 'array']", 5), + ("bool_col", "true", 5), + ("bigint_col", "-1234567890123456789", 5), + ("smallint_col", "32767", 5), + ("numeric_col", "-98765.4321", 5), + ("real_col", "-987.654", 5), + ("time_col", "'01:02:03'", 5), + ("date_col", "'2022-02-02'", 5), + ("timestamp_col", "'2022-02-02 01:02:03'", 5), + ("interval_col", "'60 days'", 5), + ("inet_col", "'127.0.0.1'", 5), + ("macaddr_col", "'fe:dc:ba:98:76:54'", 5), + ("money_col", "-9876.54", 5), ], ) def test_table_rerun_temptable( @@ -266,6 +427,7 @@ def test_table_rerun_temptable( key_column, column_name, test_value, + expected_rerun_diffs, diff_file_path, ): """Test table rerun temptable with various data types""" @@ -323,42 +485,115 @@ def test_table_rerun_temptable( diff_data = json.load(f) assert ( - len(diff_data["diffs"]["n1/n2"]["n2"]) == 3 - ), f"Expected 3 differences, found {len(diff_data['diffs']['n1/n2']['n2'])}" + len(diff_data["diffs"]["n1/n2"]["n2"]) == expected_rerun_diffs + ), f"Expected {expected_rerun_diffs} differences, " + f"found {len(diff_data['diffs']['n1/n2']['n2'])}" # Verify the differences are correctly reported for diff in diff_data["diffs"]["n1/n2"]["n2"]: + diff_val = diff[column_name] + expected_val_str = test_value.strip("'") + if column_name == "json_col": assert ( - diff[column_name].get("rerun") == "modified" + diff_val.get("rerun") == "modified" ), f"Modified row {diff['id']} doesn't have expected JSON value" elif column_name == "array_col": - assert diff[column_name] == [ + assert diff_val == [ 11, 22, 33, ], f"Modified row {diff['id']} doesn't have expected array value" elif column_name == "text_array_col": - assert diff[column_name] == [ + assert diff_val == [ "rerun", "modified", "array", ], f"Modified row {diff['id']} doesn't have expected text array value" elif column_name == "point_col": assert ( - diff[column_name] == "(88.8,88.8)" + diff_val == "(88.8,88.8)" ), f"Modified row {diff['id']} doesn't have expected point value" elif column_name == "bytea_col": assert ( - diff[column_name] == "abcdef12" + diff_val == "abcdef12" ), f"Modified row {diff['id']} doesn't have expected bytea value" + elif column_name == "macaddr_col": + assert ( + diff_val == "fe:dc:ba:98:76:54" + ), f"Modified row {diff['id']} doesn't have expected macaddr value" + elif column_name == "money_col": + cleaned_diff = diff_val.replace("$", "").replace(",", "") + assert float(cleaned_diff) == float( + expected_val_str + ), f"Modified row {diff['id']} doesn't have expected money value" + elif column_name == "bool_col": + assert ( + str(diff_val).lower() == expected_val_str.lower() + ), f"Modified row {diff['id']} doesn't have expected boolean value" + elif column_name == "interval_col": + # Interval representation can vary, so we check equality in the DB + conn = psycopg.connect(host="n1", dbname="demo", user="admin") + cur = conn.cursor() + cur.execute( + "SELECT %s::interval = %s::interval", + (str(diff_val), expected_val_str), + ) + is_equal = cur.fetchone()[0] + cur.close() + conn.close() + assert is_equal, ( + f"Modified row {diff['id']} " + f"doesn't have expected interval value. " + f"Got {diff_val}, expected equivalence to {expected_val_str}" + ) else: - assert str(diff[column_name]) in ( - test_value.strip("'"), - "1234", - "98.765", - "rerun-modified", - ), f"Modified row {diff['id']} doesn't have expected value" + assert str(diff_val) == expected_val_str, ( + f"Modified row {diff['id']} " + f"doesn't have expected value, got {diff_val} " + f"expected {expected_val_str}" + ) + + def _verify_repaired_value(self, column_name, repaired_value, expected_value): + """Helper function to verify repaired values based on data type""" + if column_name == "bytea_col": + assert ( + repaired_value == expected_value + ), "Repaired bytea value doesn't match expected value" + elif column_name == "point_col": + if isinstance(repaired_value, str): + repaired_tuple = tuple( + map(float, repaired_value.strip("()").split(",")) + ) + else: + repaired_tuple = repaired_value + + expected_tuple = tuple(map(float, expected_value.strip("()").split(","))) + assert ( + repaired_tuple == expected_tuple + ), "Repaired point value doesn't match expected value" + elif column_name == "numeric_col": + assert repaired_value == Decimal( + expected_value + ), f"Repaired value for {column_name} doesn't match" + elif column_name == "inet_col": + assert repaired_value == IPv4Address( + expected_value + ), f"Repaired value for {column_name} doesn't match" + elif column_name in ["time_col", "date_col", "timestamp_col", "interval_col"]: + assert ( + repaired_value == expected_value + ), f"Repaired value for {column_name} doesn't match" + elif column_name == "money_col": + cleaned_repaired = str(repaired_value).replace("$", "").replace(",", "") + cleaned_expected = str(expected_value).replace("$", "").replace(",", "") + assert float(cleaned_repaired) == float( + cleaned_expected + ), f"Repaired money value doesn't match for {column_name}" + else: + assert ( + repaired_value == expected_value + ), f"Repaired value doesn't match expected value for {column_name}" @pytest.mark.parametrize("table_name", ["public.datatypes_test"]) @pytest.mark.parametrize( @@ -376,6 +611,22 @@ def test_table_rerun_temptable( "ARRAY['modified', 'text', 'array']", ["modified", "text", "array"], ), + ("bool_col", "false", False), + ("bigint_col", "1234567890123456789", 1234567890123456789), + ("smallint_col", "-32768", -32768), + ("numeric_col", "98765.4321", Decimal("98765.4321")), + ("real_col", "987.654", 987.654), + ("time_col", "'11:22:33'", time(11, 22, 33)), + ("date_col", "'2025-05-25'", date(2025, 5, 25)), + ( + "timestamp_col", + "'2025-05-25 11:22:33'", + datetime(2025, 5, 25, 11, 22, 33), + ), + ("interval_col", "'90 days'", timedelta(days=90)), + ("inet_col", "'192.168.100.200'", "192.168.100.200"), + ("macaddr_col", "'01:23:45:67:89:ab'", "01:23:45:67:89:ab"), + ("money_col", "9876.54", "$9,876.54"), ], ) def test_table_repair_datatypes( @@ -441,18 +692,177 @@ def test_table_repair_datatypes( conn.close() # Compare with expected value - if column_name == "bytea_col": - assert ( - repaired_value == expected_value - ), "Repaired bytea value doesn't match expected value" - elif column_name == "point_col": - assert ( - str(repaired_value) == expected_value - ), "Repaired point value doesn't match expected value" - else: - assert ( - repaired_value == expected_value - ), f"Repaired value doesn't match expected value for {column_name}" + self._verify_repaired_value(column_name, repaired_value, expected_value) except Exception as e: pytest.fail(f"Test failed: {str(e)}") + + @pytest.mark.parametrize("id_to_update", ["d0eebc99-9c0b-4ef8-bb6d-6bb9bd380a14"]) + def test_null_and_string_literal_handling( + self, cli, capsys, diff_file_path, id_to_update + ): + """ + Verify that NULL values and string literals like 'null' are + handled correctly. + """ + try: + # Our prior repair unfortunately reset a lot of fields, so we reset + # the data first here + self.reset_data(nodes=["n1", "n2"]) + + # On n2, update text_col from 'null' to 'not null' and + # int_col from NULL to a number + conn = psycopg.connect(host="n2", dbname="demo", user="admin") + cur = conn.cursor() + cur.execute("SELECT spock.repair_mode(true)") + # This specific id has "null" as a literal in the text_col + cur.execute( + """ + UPDATE datatypes_test + SET text_col = 'not null anymore', int_col = 123 + WHERE id = %s + """, + (id_to_update,), + ) + conn.commit() + cur.close() + conn.close() + + # Run table-diff + cli.table_diff(cluster_name="eqn-t9da", table_name="public.datatypes_test") + captured = capsys.readouterr() + clean_output = re.sub( + r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])", "", captured.out + ) + match = re.search(r"diffs written out to (.+\.json)", clean_output.lower()) + assert match, "Diff file path not found in output" + diff_file_path.path = match.group(1) + + # Verify diff file + with open(diff_file_path.path, "r") as f: + diff_data = json.load(f) + + diffs_n1 = diff_data["diffs"]["n1/n2"]["n1"] + diffs_n2 = diff_data["diffs"]["n1/n2"]["n2"] + + assert len(diffs_n1) == 1, "Expected 1 difference on n1" + assert len(diffs_n2) == 1, "Expected 1 difference on n2" + + # Check n1 (original values) + assert diffs_n1[0]["id"] == id_to_update + assert diffs_n1[0]["text_col"] == "null" + assert diffs_n1[0]["int_col"] is None + + # Check n2 (modified values) + assert diffs_n2[0]["id"] == id_to_update + assert diffs_n2[0]["text_col"] == "not null anymore" + assert diffs_n2[0]["int_col"] == 123 + + # Run table-repair + cli.table_repair( + cluster_name="eqn-t9da", + table_name="public.datatypes_test", + diff_file=diff_file_path.path, + source_of_truth="n2", + ) + + # Verify repair on n1 + conn = psycopg.connect(host="n1", dbname="demo", user="admin") + cur = conn.cursor() + cur.execute( + """ + SELECT text_col, int_col FROM datatypes_test + WHERE id = %s + """, + (id_to_update,), + ) + repaired_text, repaired_int = cur.fetchone() + cur.close() + conn.close() + + assert repaired_text == "not null anymore" + assert repaired_int == 123 + + except Exception as e: + pytest.fail(f"Test for null handling failed: {str(e)}") + + @pytest.mark.parametrize("id_to_update", ["e0eebc99-9c0b-4ef8-bb6d-6bb9bd380a15"]) + def test_ast_literal_eval_fallback(self, cli, capsys, diff_file_path, id_to_update): + """ + Verify that the fallback to string representation works when + ast.literal_eval fails. + """ + try: + # Resetting again here + self.reset_data(nodes=["n1", "n2"]) + + # On n2, update float_col from NaN to a valid number + conn = psycopg.connect(host="n2", dbname="demo", user="admin") + cur = conn.cursor() + cur.execute("SELECT spock.repair_mode(true)") + cur.execute( + """ + UPDATE datatypes_test + SET float_col = 1.23 + WHERE id = %s + """, + (id_to_update,), + ) + conn.commit() + cur.close() + conn.close() + + # Run table-diff + cli.table_diff(cluster_name="eqn-t9da", table_name="public.datatypes_test") + captured = capsys.readouterr() + clean_output = re.sub( + r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])", "", captured.out + ) + match = re.search(r"diffs written out to (.+\.json)", clean_output.lower()) + assert match, "Diff file path not found in output" + diff_file_path.path = match.group(1) + + # Verify diff file + with open(diff_file_path.path, "r") as f: + diff_data = json.load(f) + + diffs_n1 = diff_data["diffs"]["n1/n2"]["n1"] + diffs_n2 = diff_data["diffs"]["n1/n2"]["n2"] + + assert len(diffs_n1) == 1, "Expected 1 difference on n1" + assert len(diffs_n2) == 1, "Expected 1 difference on n2" + + # Check n1 (original 'NaN' value) + assert diffs_n1[0]["id"] == id_to_update + assert diffs_n1[0]["float_col"] == "nan" + + # Check n2 (modified value) + assert diffs_n2[0]["id"] == id_to_update + assert diffs_n2[0]["float_col"] == 1.23 + + # Run table-repair + cli.table_repair( + cluster_name="eqn-t9da", + table_name="public.datatypes_test", + diff_file=diff_file_path.path, + source_of_truth="n2", + ) + + # Verify repair on n1 + conn = psycopg.connect(host="n1", dbname="demo", user="admin") + cur = conn.cursor() + cur.execute( + """ + SELECT float_col FROM datatypes_test + WHERE id = %s + """, + (id_to_update,), + ) + repaired_float = cur.fetchone()[0] + cur.close() + conn.close() + + assert repaired_float == 1.23 + + except Exception as e: + pytest.fail(f"Test for ast.literal_eval fallback failed: {str(e)}")