Apache
/
lucene
mirror of https://github.com/apache/lucene.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

SOLR-9729: JDBCStream improvements

This commit is contained in:
Kevin Risden 2016-11-04 10:27:32 -05:00
parent ace423e958
commit c20d1298d3
3 changed files with 38 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
solr/CHANGES.txt
View File

@ -162,6 +162,8 @@ Bug Fixes
* SOLR-9284: The HDFS BlockDirectoryCache should not let it's keysToRelease or names maps grow indefinitely.
(Mark Miller, Michael Sun)
* SOLR-9729: JDBCStream improvements (Kevin Risden)
Other Changes
----------------------

58
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/JDBCStream.java
View File

@ -67,7 +67,7 @@ public class JDBCStream extends TupleStream implements Expressible {
// These are java types that we can directly support as an Object instance. Other supported
// types will require some level of conversion (short -> long, etc...)
// We'll use a static constructor to load this set.
private static HashSet<String> directSupportedTypes = new HashSet<String>();
private static final HashSet<String> directSupportedTypes = new HashSet<>();
static {
directSupportedTypes.add(String.class.getName());
directSupportedTypes.add(Double.class.getName());
@ -85,8 +85,8 @@ public class JDBCStream extends TupleStream implements Expressible {
private Connection connection;
private Properties connectionProperties;
private Statement statement;
private ResultSet resultSet;
private ResultSetValueSelector[] valueSelectors;
protected ResultSet resultSet;
protected transient StreamContext streamContext;
public JDBCStream(String connectionUrl, String sqlQuery, StreamComparator definedSort) throws IOException {
@ -107,7 +107,7 @@ public class JDBCStream extends TupleStream implements Expressible {
// Validate there are no unknown parameters - zkHost and alias are namedParameter so we don't need to count it twice
if(expression.getParameters().size() != namedParams.size()){
throw new IOException(String.format(Locale.ROOT,"invalid expression %s - unknown operands found",expression));
throw new IOException(String.format(Locale.ROOT,"invalid expression %s - unknown operands found", expression));
}
// All named params we don't care about will be passed to the driver on connection
@ -124,7 +124,7 @@ public class JDBCStream extends TupleStream implements Expressible {
connectionUrl = ((StreamExpressionValue)connectionUrlExpression.getParameter()).getValue();
}
if(null == connectionUrl){
throw new IOException(String.format(Locale.ROOT,"invalid expression %s - connection not found"));
throw new IOException(String.format(Locale.ROOT,"invalid expression %s - connection not found", connectionUrlExpression));
}
// sql, required
@ -133,16 +133,16 @@ public class JDBCStream extends TupleStream implements Expressible {
sqlQuery = ((StreamExpressionValue)sqlQueryExpression.getParameter()).getValue();
}
if(null == sqlQuery){
throw new IOException(String.format(Locale.ROOT,"invalid expression %s - sql not found"));
throw new IOException(String.format(Locale.ROOT,"invalid expression %s - sql not found", sqlQueryExpression));
}
// definedSort, required
StreamComparator definedSort = null;
if(null != sqlQueryExpression && sqlQueryExpression.getParameter() instanceof StreamExpressionValue){
if(null != definedSortExpression && definedSortExpression.getParameter() instanceof StreamExpressionValue){
definedSort = factory.constructComparator(((StreamExpressionValue)definedSortExpression.getParameter()).getValue(), FieldComparator.class);
}
if(null == definedSort){
throw new IOException(String.format(Locale.ROOT,"invalid expression %s - sort not found"));
throw new IOException(String.format(Locale.ROOT,"invalid expression %s - sort not found", definedSortExpression));
}
// driverClass, optional
@ -155,7 +155,7 @@ public class JDBCStream extends TupleStream implements Expressible {
init(connectionUrl, sqlQuery, definedSort, connectionProperties, driverClass);
}
private void init(String connectionUrl, String sqlQuery, StreamComparator definedSort, Properties connectionProperties, String driverClassName) throws IOException {
private void init(String connectionUrl, String sqlQuery, StreamComparator definedSort, Properties connectionProperties, String driverClassName) {
this.connectionUrl = connectionUrl;
this.sqlQuery = sqlQuery;
this.definedSort = definedSort;
@ -188,7 +188,9 @@ public class JDBCStream extends TupleStream implements Expressible {
throw new SQLException("DriverManager.getDriver(url) returned null");
}
} catch(SQLException e){
throw new IOException(String.format(Locale.ROOT, "Failed to determine JDBC driver from connection url '%s'. Usually this means the driver is not loaded - you can have JDBCStream try to load it by providing the 'driverClassName' value", connectionUrl), e);
throw new IOException(String.format(Locale.ROOT,
"Failed to determine JDBC driver from connection url '%s'. Usually this means the driver is not loaded - " +
"you can have JDBCStream try to load it by providing the 'driverClassName' value", connectionUrl), e);
}
try {
@ -200,20 +202,23 @@ public class JDBCStream extends TupleStream implements Expressible {
try{
statement = connection.createStatement();
} catch (SQLException e) {
throw new IOException(String.format(Locale.ROOT, "Failed to create a statement from JDBC connection '%s'", connectionUrl), e);
throw new IOException(String.format(Locale.ROOT, "Failed to create a statement from JDBC connection '%s'",
connectionUrl), e);
}
try{
resultSet = statement.executeQuery(sqlQuery);
} catch (SQLException e) {
throw new IOException(String.format(Locale.ROOT, "Failed to execute sqlQuery '%s' against JDBC connection '%s'", sqlQuery, connectionUrl), e);
throw new IOException(String.format(Locale.ROOT, "Failed to execute sqlQuery '%s' against JDBC connection '%s'",
sqlQuery, connectionUrl), e);
}
try{
// using the metadata, build selectors for each column
valueSelectors = constructValueSelectors(resultSet.getMetaData());
} catch (SQLException e) {
throw new IOException(String.format(Locale.ROOT, "Failed to generate value selectors for sqlQuery '%s' against JDBC connection '%s'", sqlQuery, connectionUrl), e);
throw new IOException(String.format(Locale.ROOT,
"Failed to generate value selectors for sqlQuery '%s' against JDBC connection '%s'", sqlQuery, connectionUrl), e);
}
}
@ -222,8 +227,9 @@ public class JDBCStream extends TupleStream implements Expressible {
for(int columnIdx = 0; columnIdx < metadata.getColumnCount(); ++columnIdx){
final int columnNumber = columnIdx + 1; // cause it starts at 1
final String columnName = metadata.getColumnName(columnNumber);
final int columnNumber = columnIdx + 1; // cause it starts at 1
// Use getColumnLabel instead of getColumnName to make sure fields renamed with AS as picked up properly
final String columnName = metadata.getColumnLabel(columnNumber);
String className = metadata.getColumnClassName(columnNumber);
String typeName = metadata.getColumnTypeName(columnNumber);
@ -238,8 +244,7 @@ public class JDBCStream extends TupleStream implements Expressible {
return columnName;
}
};
}
else if(Short.class.getName() == className){
} else if(Short.class.getName().equals(className)) {
valueSelectors[columnIdx] = new ResultSetValueSelector() {
public Object selectValue(ResultSet resultSet) throws SQLException {
Short obj = resultSet.getShort(columnNumber);
@ -250,8 +255,7 @@ public class JDBCStream extends TupleStream implements Expressible {
return columnName;
}
};
}
else if(Integer.class.getName() == className){
} else if(Integer.class.getName().equals(className)) {
valueSelectors[columnIdx] = new ResultSetValueSelector() {
public Object selectValue(ResultSet resultSet) throws SQLException {
Integer obj = resultSet.getInt(columnNumber);
@ -262,8 +266,7 @@ public class JDBCStream extends TupleStream implements Expressible {
return columnName;
}
};
}
else if(Float.class.getName() == className){
} else if(Float.class.getName().equals(className)) {
valueSelectors[columnIdx] = new ResultSetValueSelector() {
public Object selectValue(ResultSet resultSet) throws SQLException {
Float obj = resultSet.getFloat(columnNumber);
@ -274,9 +277,10 @@ public class JDBCStream extends TupleStream implements Expressible {
return columnName;
}
};
}
else{
throw new SQLException(String.format(Locale.ROOT, "Unable to determine the valueSelector for column '%s' (col #%d) of java class '%s' and type '%s'", columnName, columnNumber, className, typeName));
} else {
throw new SQLException(String.format(Locale.ROOT,
"Unable to determine the valueSelector for column '%s' (col #%d) of java class '%s' and type '%s'",
columnName, columnNumber, className, typeName));
}
}
@ -305,7 +309,7 @@ public class JDBCStream extends TupleStream implements Expressible {
public Tuple read() throws IOException {
try{
Map<Object,Object> fields = new HashMap<Object,Object>();
Map<Object,Object> fields = new HashMap<>();
if(resultSet.next()){
// we have a record
for(ResultSetValueSelector selector : valueSelectors){
@ -391,7 +395,7 @@ public class JDBCStream extends TupleStream implements Expressible {
@Override
public List<TupleStream> children() {
return new ArrayList<TupleStream>();
return new ArrayList<>();
}
@Override
@ -404,6 +408,6 @@ public class JDBCStream extends TupleStream implements Expressible {
}
interface ResultSetValueSelector {
public String getColumnName();
public Object selectValue(ResultSet resultSet) throws SQLException;
String getColumnName();
Object selectValue(ResultSet resultSet) throws SQLException;
}

16
solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/JDBCStreamTest.java
View File

@ -351,11 +351,8 @@ public class JDBCStreamTest extends SolrCloudTestCase {
TupleStream stream;
List<Tuple> tuples;
// Basic test
// the test here is the setting of the property get_column_name=true. In hsqldb if this value is set to true then the use of an
// as clause in a select will have no effect. As such even though we have PEOPLE.ID as PERSONID we will still expect the column
// name to come out as ID and not PERSONID
expression =
// Basic test for no alias
expression =
"innerJoin("
+ " select("
+ " search(" + COLLECTIONORALIAS + ", fl=\"personId_i,rating_f\", q=\"rating_f:*\", sort=\"personId_i asc\"),"
@ -363,7 +360,7 @@ public class JDBCStreamTest extends SolrCloudTestCase {
+ " rating_f as rating"
+ " ),"
+ " select("
+ " jdbc(connection=\"jdbc:hsqldb:mem:.\", sql=\"select PEOPLE.ID as PERSONID, PEOPLE.NAME, COUNTRIES.COUNTRY_NAME from PEOPLE inner join COUNTRIES on PEOPLE.COUNTRY_CODE = COUNTRIES.CODE order by PEOPLE.ID\", sort=\"ID asc\", get_column_name=true),"
+ " jdbc(connection=\"jdbc:hsqldb:mem:.\", sql=\"select PEOPLE.ID, PEOPLE.NAME, COUNTRIES.COUNTRY_NAME from PEOPLE inner join COUNTRIES on PEOPLE.COUNTRY_CODE = COUNTRIES.CODE order by PEOPLE.ID\", sort=\"ID asc\"),"
+ " ID as personId,"
+ " NAME as personName,"
+ " COUNTRY_NAME as country"
@ -380,10 +377,7 @@ public class JDBCStreamTest extends SolrCloudTestCase {
assertOrderOf(tuples, "personName", "Emma","Grace","Hailey","Isabella","Lily","Madison","Mia","Natalie","Olivia","Samantha");
assertOrderOf(tuples, "country", "Netherlands","United States","Netherlands","Netherlands","Netherlands","United States","United States","Netherlands","Netherlands","United States");
// Basic test
// the test here is the setting of the property get_column_name=false. In hsqldb if this value is set to false then the use of an
// as clause in a select will have effect. As such we have PEOPLE.ID as PERSONID we will still expect the column name to come out
// PERSONID and not ID
// Basic test for alias
expression =
"innerJoin("
+ " select("
@ -392,7 +386,7 @@ public class JDBCStreamTest extends SolrCloudTestCase {
+ " rating_f as rating"
+ " ),"
+ " select("
+ " jdbc(connection=\"jdbc:hsqldb:mem:.\", sql=\"select PEOPLE.ID as PERSONID, PEOPLE.NAME, COUNTRIES.COUNTRY_NAME from PEOPLE inner join COUNTRIES on PEOPLE.COUNTRY_CODE = COUNTRIES.CODE order by PEOPLE.ID\", sort=\"PERSONID asc\", get_column_name=false),"
+ " jdbc(connection=\"jdbc:hsqldb:mem:.\", sql=\"select PEOPLE.ID as PERSONID, PEOPLE.NAME, COUNTRIES.COUNTRY_NAME from PEOPLE inner join COUNTRIES on PEOPLE.COUNTRY_CODE = COUNTRIES.CODE order by PEOPLE.ID\", sort=\"PERSONID asc\"),"
+ " PERSONID as personId,"
+ " NAME as personName,"
+ " COUNTRY_NAME as country"