Skip to content

refactor: rename the single-destination replicator (was: duckling) #86

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jghoman merged 1 commit into from
Sep 3, 2026
Merged

refactor: rename the single-destination replicator (was: duckling) #86

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
56 changes: 28 additions & 28 deletions tests/integration/test_single_destination_e2e.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
"""End-to-end: the per-destination duckling against real catalogs.
"""End-to-end: the single-destination viaduck against real catalogs.

PG testcontainer holds both ducklake catalogs (source + destination) and
the viaduck_state cursor table on the SOURCE database (the colocated
Expand All @@ -17,11 +17,11 @@
from testcontainers.postgres import PostgresContainer

from viaduck import metrics
from viaduck.single_destination import Duckling, DucklingConfig, FatalDucklingError
from viaduck.single_destination import FatalSingleDestinationError, SingleDestinationConfig, SingleDestinationViaduck


def setup_module():
metrics.init("duckling_integration_test")
metrics.init("single_destination_integration_test")


@pytest.fixture(scope="module")
Expand All @@ -41,7 +41,7 @@ def _mkdb(pg_dsn: str, name: str) -> str:


class Env:
"""Test handle: both catalogs plus a DucklingConfig factory."""
"""Test handle: both catalogs plus a SingleDestinationConfig factory."""

def __init__(self, src, dst, src_dsn, dst_dsn, tmp_path):
self.src = src
Expand All @@ -51,7 +51,7 @@ def __init__(self, src, dst, src_dsn, dst_dsn, tmp_path):
self.tmp_path = tmp_path
self._destination_id = f"org-test-team-2-{uuid.uuid4().hex[:6]}"

def cfg(self, **kw) -> DucklingConfig:
def cfg(self, **kw) -> SingleDestinationConfig:
base = dict(
# ATTACH format on purpose: exercises the F2 conninfo translation
source_pg_uri=f"postgres:{self.src_dsn}",
Expand All @@ -67,12 +67,12 @@ def cfg(self, **kw) -> DucklingConfig:
cursor_pg_uri="",
destination_id=self._destination_id, # stable per test: a restart
# resumes the SAME cursor row
instance_id="duckling-test-0",
instance_id="single-destination-test-0",
poll_interval_s=0.1,
start_snapshot_id=0, # tests want full catch-up, not start-at-head
)
base.update(kw)
return DucklingConfig(**base)
return SingleDestinationConfig(**base)


@pytest.fixture()
Expand Down Expand Up @@ -114,7 +114,7 @@ def test_catchup_filters_and_advances(self, env):
_insert(src, [(2, "c")])
_insert(src, [(9, "d"), (2, "e")])

d = Duckling(env.cfg())
d = SingleDestinationViaduck(env.cfg())
d.boot()
try:
d.poll_once()
Expand All @@ -136,7 +136,7 @@ def test_crash_between_append_and_cursor_redelivers_once(self, env):
src, dst = env.src, env.dst
_insert(src, [(2, "a"), (2, "b")])

d = Duckling(env.cfg())
d = SingleDestinationViaduck(env.cfg())
d.boot()
d._cursor_advance = lambda *a, **k: (_ for _ in ()).throw(RuntimeError("simulated crash"))
try:
Expand All @@ -146,7 +146,7 @@ def test_crash_between_append_and_cursor_redelivers_once(self, env):
_close(d)
assert _dest_rows(dst) == [(2, "a"), (2, "b")] # committed; cursor not

d2 = Duckling(env.cfg()) # "restart": same cursor row
d2 = SingleDestinationViaduck(env.cfg()) # "restart": same cursor row
d2.boot()
try:
d2.poll_once()
Expand All @@ -160,7 +160,7 @@ def test_add_column_restart_picks_up(self, env):
src, dst = env.src, env.dst
_insert(src, [(2, "a")])

d = Duckling(env.cfg())
d = SingleDestinationViaduck(env.cfg())
d.boot()
try:
# NB: on the deployed build the ALTER resurrects the session's
Expand All @@ -173,7 +173,7 @@ def test_add_column_restart_picks_up(self, env):
_close(d)
assert _dest_rows(dst) == [(2, "a"), (2, "b")]

d2 = Duckling(env.cfg())
d2 = SingleDestinationViaduck(env.cfg())
d2.boot()
try:
assert "extra" in d2.columns
Expand All @@ -193,7 +193,7 @@ def test_delete_appearance_crashes(self, env):
_insert(src, [(2, "a")])
src.connection.execute("DELETE FROM lake.main.events WHERE team_id = 2")

d = Duckling(env.cfg())
d = SingleDestinationViaduck(env.cfg())
try:
# the baseline assertions at boot already see the delete
with pytest.raises(RuntimeError, match="contract violated|delete"):
Expand All @@ -209,7 +209,7 @@ def test_restart_resumes_from_cursor(self, env):
src, dst = env.src, env.dst
_insert(src, [(2, "a")])

d = Duckling(env.cfg())
d = SingleDestinationViaduck(env.cfg())
d.boot()
try:
d.poll_once()
Expand All @@ -218,7 +218,7 @@ def test_restart_resumes_from_cursor(self, env):
_close(d)
assert _dest_rows(dst) == [(2, "a")]

d2 = Duckling(env.cfg()) # same destination_id → same cursor row
d2 = SingleDestinationViaduck(env.cfg()) # same destination_id → same cursor row
d2.boot()
try:
assert d2._cursor == cursor_after # resumed, not restarted
Expand Down Expand Up @@ -246,7 +246,7 @@ def test_retention_clamp_then_poll_composes(self, env):
floor = snaps[-1]
pg.execute(f'DELETE FROM "{meta}".ducklake_snapshot WHERE snapshot_id < {floor}')

d = Duckling(env.cfg())
d = SingleDestinationViaduck(env.cfg())
d.boot() # writes the cursor row at start_snapshot_id=0
try:
assert d._cursor == 0
Expand Down Expand Up @@ -318,9 +318,9 @@ def test_inlined_delete_of_flushed_rows_crashes(self, env):
pg.execute(f'DELETE FROM "{meta}".ducklake_snapshot_changes')
pg.close()

d = Duckling(env.cfg(source_table="main.inlined_del2", dest_table="main.inlined_del2"))
d = SingleDestinationViaduck(env.cfg(source_table="main.inlined_del2", dest_table="main.inlined_del2"))
try:
with pytest.raises(FatalDucklingError, match="inlined deletes"):
with pytest.raises(FatalSingleDestinationError, match="inlined deletes"):
d.boot() # baseline assertions: store probe fires
finally:
_close(d)
Expand All @@ -339,7 +339,7 @@ def test_accepted_delete_below_cursor_boots_clean(self, env):
head = int(pg.execute(f'SELECT MAX(snapshot_id) FROM "{meta}".ducklake_snapshot').fetchone()[0])
pg.close()

d = Duckling(env.cfg(start_snapshot_id=head)) # the documented accept: cursor past the delete
d = SingleDestinationViaduck(env.cfg(start_snapshot_id=head)) # the documented accept: cursor past the delete
d.boot()
try:
_insert(src, [(2, "b")])
Expand All @@ -358,7 +358,7 @@ def test_dest_reordered_columns_map_by_name(self, env):
"CREATE TABLE dst.main.events (event VARCHAR, _inserted_at TIMESTAMPTZ DEFAULT now(), team_id BIGINT)"
)

d = Duckling(env.cfg())
d = SingleDestinationViaduck(env.cfg())
d.boot()
try:
d.poll_once()
Expand All @@ -377,7 +377,7 @@ def test_dest_managed_column_default_filled(self, env):
"CREATE TABLE dst.main.events (team_id BIGINT, event VARCHAR, _inserted_at TIMESTAMPTZ DEFAULT now())"
)

d = Duckling(env.cfg())
d = SingleDestinationViaduck(env.cfg())
d.boot()
try:
d.poll_once()
Expand All @@ -398,9 +398,9 @@ def test_inlined_delete_pages_via_witness(self, env):
src.connection.execute("INSERT INTO lake.main.inlined_del VALUES (2, 'a'), (2, 'b')")
src.connection.execute("DELETE FROM lake.main.inlined_del WHERE event = 'a'")

d = Duckling(env.cfg(source_table="main.inlined_del", dest_table="main.inlined_del"))
d = SingleDestinationViaduck(env.cfg(source_table="main.inlined_del", dest_table="main.inlined_del"))
try:
with pytest.raises(FatalDucklingError, match="delete/drop activity"):
with pytest.raises(FatalSingleDestinationError, match="delete/drop activity"):
d.boot()
finally:
d.feed.close()
Expand All @@ -410,7 +410,7 @@ def test_inlined_delete_pages_via_witness(self, env):

class TestVariantSourceColumn:
"""The production shape: events_nrt carries millpond's VARIANT dual-write
companions. The duckling must boot, read, and deliver as if they don't
companions. The single-destination viaduck must boot, read, and deliver as if they don't
exist (load_table's EXCLUDABLE_SOURCE_TYPES exclusion)."""

def test_boot_and_deliver_with_variant_column(self, env):
Expand All @@ -420,7 +420,7 @@ def test_boot_and_deliver_with_variant_column(self, env):
)
src.connection.execute("INSERT INTO lake.main.events_v VALUES (2, 'a', {\"k\": 1}), (3, 'b', NULL)")

d = Duckling(env.cfg(source_table="main.events_v", dest_table="main.events_v"))
d = SingleDestinationViaduck(env.cfg(source_table="main.events_v", dest_table="main.events_v"))
d.boot()
try:
assert "properties_variant" not in d.columns
Expand All @@ -439,7 +439,7 @@ def test_variant_added_mid_stream(self, env):
src, dst = env.src, env.dst
_insert(src, [(2, "a")])

d = Duckling(env.cfg())
d = SingleDestinationViaduck(env.cfg())
d.boot()
try:
src.connection.execute("ALTER TABLE lake.main.events ADD COLUMN properties_variant VARIANT")
Expand All @@ -451,7 +451,7 @@ def test_variant_added_mid_stream(self, env):
_close(d)
assert _dest_rows(dst) == [(2, "a"), (2, "b")]

d2 = Duckling(env.cfg())
d2 = SingleDestinationViaduck(env.cfg())
d2.boot() # restart: load_table excludes the VARIANT again
try:
assert "properties_variant" not in d2.columns
Expand All @@ -474,7 +474,7 @@ def test_inline_rows_served_with_page(self, env):
src.connection.execute("INSERT INTO lake.main.inlined VALUES (2, 'i1'), (3, 'i2')")
src.connection.execute("RESET ducklake_default_data_inlining_row_limit")

d = Duckling(env.cfg(source_table="main.inlined", dest_table="main.inlined"))
d = SingleDestinationViaduck(env.cfg(source_table="main.inlined", dest_table="main.inlined"))
d.boot()
try:
d.poll_once()
Expand Down
Loading
Loading