Function test_join

python/pathway/tests/test_persistence.py:524–540

(tmp_path, mode)

)
@only_with_license_key("mode", [api.PersistenceMode.OPERATOR_PERSISTING])
def test_join(tmp_path, mode):
    class InputSchema(pw.Schema):
        a: int = pw.column_definition(primary_key=True)
        b: int

    def logic(t_1: pw.Table, t_2: pw.Table) -> pw.Table:
        return t_1.join(t_2, t_1.a == t_2.a).select(
            pw.this.a, b=pw.left.b, c=pw.right.b
        )

    run, _, input_path_2 = get_two_tables_runner(tmp_path, mode, logic, InputSchema)

    run(["a,b", "1,2", "2,4"], ["a,b", "1,3"], {"1,2,3,1"})
    run(["a,b", "3,3"], ["a,b", "2,6", "1,4"], {"2,4,6,1", "1,2,4,1"})
    os.remove(input_path_2 / "1")
    run(["a,b"], ["a,b"], {"1,2,3,-1"})
    run(["a,b", "1,4"], ["a,b", "1,8"], {"1,2,8,1", "1,4,8,1", "1,4,4,1"})


@pytest.mark.parametrize(