Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
6 changes: 6 additions & 0 deletions sqlmesh/core/snapshot/definition.py
Original file line number Diff line number Diff line change
Expand Up @@ -481,6 +481,12 @@ class SnapshotTableInfo(PydanticModel, SnapshotInfoMixin, frozen=True):
def __lt__(self, other: SnapshotTableInfo) -> bool:
return self.name < other.name

def __eq__(self, other: t.Any) -> bool:
return isinstance(other, SnapshotTableInfo) and self.fingerprint == other.fingerprint

def __hash__(self) -> int:
return hash((self.__class__, self.name, self.fingerprint))

def table_name(self, is_deployable: bool = True) -> str:
"""Full table name pointing to the materialized location of the snapshot.

Expand Down
42 changes: 42 additions & 0 deletions tests/core/test_snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -1035,6 +1035,48 @@ def test_fingerprint_virtual_properties(model: Model, parent_model: Model):
assert updated_fingerprint.data_hash == fingerprint.data_hash


def test_tableinfo_equality():
snapshot_a = SnapshotTableInfo(
name="test_schema.a",
fingerprint=SnapshotFingerprint(data_hash="1", metadata_hash="1"),
version="test_version",
physical_schema="test_physical_schema",
parents=[],
dev_table_suffix="dev",
)

snapshot_b = SnapshotTableInfo(
name="test_schema.b",
fingerprint=SnapshotFingerprint(data_hash="1", metadata_hash="1"),
version="test_version",
physical_schema="test_physical_schema",
parents=[],
dev_table_suffix="dev",
)

snapshot_c = SnapshotTableInfo(
name="test_schema.c",
fingerprint=SnapshotFingerprint(data_hash="1", metadata_hash="1"),
version="test_version",
physical_schema="test_physical_schema",
parents=[snapshot_a.snapshot_id, snapshot_b.snapshot_id],
dev_table_suffix="dev",
)

# parents in different order than snapshot_c
snapshot_c2 = SnapshotTableInfo(
name="test_schema.c",
fingerprint=SnapshotFingerprint(data_hash="1", metadata_hash="1"),
version="test_version",
physical_schema="test_physical_schema",
parents=[snapshot_b.snapshot_id, snapshot_a.snapshot_id],
dev_table_suffix="dev",
)

assert snapshot_c is not snapshot_c2
assert snapshot_c == snapshot_c2


def test_stamp(model: Model):
original_fingerprint = fingerprint_from_node(model, nodes={})

Expand Down