public static SerializableFunction<Row, Mutation> beamRowToMutationFn(
      Mutation.Op operation, String table) {
    return (row -> {
      switch (operation) {
        case INSERT:
          return MutationUtils.createMutationFromBeamRows(Mutation.newInsertBuilder(table), row);
        case DELETE:
          return Mutation.delete(table, MutationUtils.createKeyFromBeamRow(row));
        case UPDATE:
          return MutationUtils.createMutationFromBeamRows(Mutation.newUpdateBuilder(table), row);
        case REPLACE:
          return MutationUtils.createMutationFromBeamRows(Mutation.newReplaceBuilder(table), row);
        case INSERT_OR_UPDATE:
          return MutationUtils.createMutationFromBeamRows(
              Mutation.newInsertOrUpdateBuilder(table), row);
        default:
          throw new IllegalArgumentException(
              String.format("Unknown mutation operation type: %s", operation));
      }
    });
  }
@Test
  public void testCreateInsertMutationFromRow() {
    Mutation expectedMutation = createMutation(Mutation.Op.INSERT);
    Mutation mutation = beamRowToMutationFn(Mutation.Op.INSERT, TABLE).apply(WRITE_ROW);
    assertEquals(expectedMutation, mutation);
  }