JdbcSource to partition queries for potential performance improvements

- Spark can partition data on a **JDBC** data frame by by specifying the following binding parameters which are all **longs**: **lowerBound**, **upperBound**, **numPartitions** and and **partition key** column
- To take advantage of this in **JdbcSource**, the data has to be divided into multiple partitions (in multiple Threads). In turn these binding parameters are used to alter the original query for each partition, e.g. add a predicate to restrict query on each partition - for **Oracle** the partition key could be **rownum** which is available on every table, e.g. for a population consisting of 100 rows with a specification of: **lowerBound=1**, **upperBound=100** and **numPartitions=5** and **query=select col1, col2 from table where blah** would result in the following partitions queries on each partition:

## Part 1:
```sql
select * (select col1, col2 from table where blah)
where rownum between 1 and 20
```
## Part 2:
```sql
select * (select col1, col2 from table where blah)
where rownum between 21 and 40
```
## Part 3:
```sql
select * (select col1, col2 from table where blah)
where rownum between 41 and 60
```
## Part 4:
```sql
select * (select col1, col2 from table where blah)
where rownum between 61 and 80
```
## Part 5:
```sql
select * (select col1, col2 from table where blah)
where rownum >= 81
```
- For **partition 5** just return the remainder of rows.

- Note it may not be necessary to create N connections for each partition - simply return N **JDBC** result sets - one for each partition - investigating this...

# Proposal
- **withPartition(lowerBound, upperBound, partitionColumn)**
```scala
.withPartition(1, 100, rownum)
```
- **rownum** is the partition key which is internal to **Oracle** however you can't use this across the board with a function like **withPartition(1,100)**, e.g. **SQLServer** doesn't have **rownum**, however it can be achieved using a the windowing function **ROW_NUMBER()**:
```sql
where RowNumber = ROW_NUMBER() OVER (ORDER BY CustomerID ASC)
```
- where **CustomerID** is the primary key and **ROW_NUMBER()** is the **SQLServer** windowing function.
- One idea is to provide a upper bound function like so:
```scala
withPartition(lowerBound:Long, upperBoundFn(query:String) => Long, partitionColumn:String)
```
- the **upperBoundFn()** could do anything you like such as execute a separate query or just supply the count.

# Example of returning N JDBC result sets

- http://stackoverflow.com/questions/9696572/queries-returning-multiple-result-sets

```java

public static void executeProcedure(Connection con) {
try {
CallableStatement stmt = con.prepareCall(...);
..... //Set call parameters, if you have IN,OUT, or IN/OUT parameters

boolean results = stmt.execute();
int rsCount = 0;

//Loop through the available result sets.
while (results) {
ResultSet rs = stmt.getResultSet();
//Retrieve data from the result set.
while (rs.next()) {
....// using rs.getxxx() method to retieve data
}
rs.close();

//Check for next result set
results = stmt.getMoreResults();
}
stmt.close();
}
catch (Exception e) {
e.printStackTrace();
}
}
```

The JdbcSource class likely handles JDBC connections and query execution. Look for methods that build queries and manage result sets. The proposal involves modifying the query generation to add partition predicates and potentially handling multiple result sets. Start by examining the existing JDBC integration and Spark partitioning parameters. Testing would require setting up a database and verifying the partitioned queries work correctly across different database systems like Oracle and SQL Server.

Scritto dal modello di indicizzazione a partire dal testo della issue.