Function test_update_rows

python/pathway/tests/test_persistence.py:479–493

(tmp_path, mode)

)
@only_with_license_key("mode", [api.PersistenceMode.OPERATOR_PERSISTING])
def test_update_rows(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.update_rows(t_2)

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

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


@pytest.mark.parametrize(