You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ch...@apache.org on 2014/05/19 23:35:05 UTC
svn commit: r1596061 - in /pig/trunk: CHANGES.txt
contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/DBStorage.java
contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestDBStorage.java
Author: cheolsoo
Date: Mon May 19 21:35:04 2014
New Revision: 1596061
URL: http://svn.apache.org/r1596061
Log:
PIG-3936: DBStorage fails on storing nulls for non varchar columns (jeremykarn via cheolsoo)
Modified:
pig/trunk/CHANGES.txt
pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/DBStorage.java
pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestDBStorage.java
Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1596061&r1=1596060&r2=1596061&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Mon May 19 21:35:04 2014
@@ -133,6 +133,8 @@ PIG-3882: Multiquery off mode execution
BUG FIXES
+PIG-3936: DBStorage fails on storing nulls for non varchar columns (jeremykarn via cheolsoo)
+
PIG-3945: Ant not sending hadoopversion to piggybank sub-ant (mrflip via cheolsoo)
PIG-3942: Util.buildPp() is incompatible with Non-MR execution engine (cheolsoo)
Modified: pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/DBStorage.java
URL: http://svn.apache.org/viewvc/pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/DBStorage.java?rev=1596061&r1=1596060&r2=1596061&view=diff
==============================================================================
--- pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/DBStorage.java (original)
+++ pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/DBStorage.java Mon May 19 21:35:04 2014
@@ -17,7 +17,13 @@
*/
package org.apache.pig.piggybank.storage;
-import org.joda.time.DateTime;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.Date;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.Properties;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -28,14 +34,16 @@ import org.apache.hadoop.mapreduce.Outpu
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.pig.ResourceSchema;
+import org.apache.pig.ResourceSchema.ResourceFieldSchema;
import org.apache.pig.StoreFunc;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
-
-import java.io.IOException;
-import java.sql.*;
+import org.apache.pig.impl.util.UDFContext;
+import org.apache.pig.impl.util.Utils;
+import org.joda.time.DateTime;
public class DBStorage extends StoreFunc {
private final Log log = LogFactory.getLog(getClass());
@@ -49,6 +57,11 @@ public class DBStorage extends StoreFunc
private int count = 0;
private String insertQuery;
+ //We want to store the schema if possible so that we can try to deal with nulls.
+ protected ResourceSchema schema = null;
+ private String udfcSignature = null;
+ private static final String SCHEMA_SIGNATURE = "pig.dbstorage.schema";
+
public DBStorage(String driver, String jdbcURL, String insertQuery) {
this(driver, jdbcURL, null, null, insertQuery, "100");
}
@@ -85,10 +98,15 @@ public class DBStorage extends StoreFunc
for (int i = 0; i < size; i++) {
try {
Object field = tuple.get(i);
-
switch (DataType.findType(field)) {
case DataType.NULL:
- ps.setNull(sqlPos, java.sql.Types.VARCHAR);
+ //Default to varchar
+ int nullSqlType = java.sql.Types.VARCHAR;
+ if (schema != null) {
+ ResourceFieldSchema fs = schema.getFields()[i];
+ nullSqlType = sqlDataTypeFromPigDataType(fs.getType());
+ }
+ ps.setNull(sqlPos, nullSqlType);
sqlPos++;
break;
@@ -164,9 +182,7 @@ public class DBStorage extends StoreFunc
}
} catch (SQLException e) {
try {
- log
- .error("Unable to insert record:" + tuple.toDelimitedString("\t"),
- e);
+ log.error("Unable to insert record:" + tuple.toDelimitedString("\t"), e);
} catch (ExecException ee) {
// do nothing
}
@@ -180,6 +196,30 @@ public class DBStorage extends StoreFunc
}
}
+ protected int sqlDataTypeFromPigDataType(byte pigDataType) {
+ switch(pigDataType) {
+ case DataType.INTEGER:
+ return java.sql.Types.INTEGER;
+ case DataType.LONG:
+ return java.sql.Types.BIGINT;
+ case DataType.FLOAT:
+ return java.sql.Types.FLOAT;
+ case DataType.DOUBLE:
+ return java.sql.Types.DOUBLE;
+ case DataType.BOOLEAN:
+ return java.sql.Types.BOOLEAN;
+ case DataType.DATETIME:
+ return java.sql.Types.DATE;
+ case DataType.BYTEARRAY:
+ case DataType.CHARARRAY:
+ case DataType.BYTE:
+ return java.sql.Types.VARCHAR;
+ default:
+ log.warn("Can not find SQL data type for " + pigDataType + " returning VARCHAR");
+ return java.sql.Types.VARCHAR;
+ }
+ }
+
class MyDBOutputFormat extends OutputFormat<NullWritable, NullWritable> {
@Override
@@ -250,18 +290,18 @@ public class DBStorage extends StoreFunc
@Override
public RecordWriter<NullWritable, NullWritable> getRecordWriter(
- TaskAttemptContext context) throws IOException, InterruptedException {
+ TaskAttemptContext context) throws IOException, InterruptedException {
// We don't use a record writer to write to database
- return new RecordWriter<NullWritable, NullWritable>() {
- @Override
- public void close(TaskAttemptContext context) {
- // Noop
- }
- @Override
- public void write(NullWritable k, NullWritable v) {
- // Noop
- }
- };
+ return new RecordWriter<NullWritable, NullWritable>() {
+ @Override
+ public void close(TaskAttemptContext context) {
+ // Noop
+ }
+ @Override
+ public void write(NullWritable k, NullWritable v) {
+ // Noop
+ }
+ };
}
}
@@ -298,10 +338,39 @@ public class DBStorage extends StoreFunc
throw new IOException("JDBC Error", e);
}
count = 0;
+
+ // Try to get the schema from the UDFContext object.
+ UDFContext udfc = UDFContext.getUDFContext();
+ Properties p =
+ udfc.getUDFProperties(this.getClass(), new String[]{udfcSignature});
+ String strSchema = p.getProperty(SCHEMA_SIGNATURE);
+ if (strSchema != null) {
+ // Parse the schema from the string stored in the properties object.
+ schema = new ResourceSchema(Utils.getSchemaFromString(strSchema));
+ }
}
@Override
public void setStoreLocation(String location, Job job) throws IOException {
// IGNORE since we are writing records to DB.
}
+
+ @Override
+ public void setStoreFuncUDFContextSignature(String signature) {
+ // store the signature so we can use it later
+ udfcSignature = signature;
+ }
+
+ @Override
+ public void checkSchema(ResourceSchema s) throws IOException {
+ // We won't really check the schema here, we'll store it in our
+ // UDFContext properties object so we have it when we need it on the
+ // backend
+
+ UDFContext udfc = UDFContext.getUDFContext();
+ Properties p =
+ udfc.getUDFProperties(this.getClass(), new String[]{udfcSignature});
+ p.setProperty(SCHEMA_SIGNATURE, s.toString());
+ }
+
}
Modified: pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestDBStorage.java
URL: http://svn.apache.org/viewvc/pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestDBStorage.java?rev=1596061&r1=1596060&r2=1596061&view=diff
==============================================================================
--- pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestDBStorage.java (original)
+++ pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestDBStorage.java Mon May 19 21:35:04 2014
@@ -46,156 +46,173 @@ import junit.framework.TestCase;
public class TestDBStorage extends TestCase {
- private PigServer pigServer;
- private MiniCluster cluster;
- private Server dbServer;
- private String driver = "org.hsqldb.jdbcDriver";
- // private String url = "jdbc:hsqldb:mem:.";
+ private PigServer pigServer;
+ private MiniCluster cluster;
+ private Server dbServer;
+ private String driver = "org.hsqldb.jdbcDriver";
+ // private String url = "jdbc:hsqldb:mem:.";
private String TMP_DIR;
private String dblocation;
private String url;
private String dbUrl = "jdbc:hsqldb:hsql://localhost/" + "batchtest";
- private String user = "sa";
- private String password = "";
+ private String user = "sa";
+ private String password = "";
- private static final String INPUT_FILE = "datafile.txt";
+ private static final String INPUT_FILE = "datafile.txt";
- public TestDBStorage() throws ExecException, IOException {
- // Initialise Pig server
- cluster = MiniCluster.buildCluster();
- pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
- pigServer.getPigContext().getProperties()
- .setProperty("mapred.map.max.attempts", "1");
- pigServer.getPigContext().getProperties()
- .setProperty("mapred.reduce.max.attempts", "1");
- System.out.println("Pig server initialized successfully");
+ public TestDBStorage() throws ExecException, IOException {
+ // Initialise Pig server
+ cluster = MiniCluster.buildCluster();
+ pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer.getPigContext().getProperties()
+ .setProperty("mapred.map.max.attempts", "1");
+ pigServer.getPigContext().getProperties()
+ .setProperty("mapred.reduce.max.attempts", "1");
+ System.out.println("Pig server initialized successfully");
TMP_DIR = System.getProperty("user.dir") + "/build/test/";
dblocation = TMP_DIR + "batchtest";
url = "jdbc:hsqldb:file:" + dblocation
+ ";hsqldb.default_table_type=cached;hsqldb.cache_rows=100";
- // Initialise DBServer
- dbServer = new Server();
- dbServer.setDatabaseName(0, "batchtest");
- // dbServer.setDatabasePath(0, "mem:test;sql.enforce_strict_size=true");
- dbServer.setDatabasePath(0,
+ // Initialise DBServer
+ dbServer = new Server();
+ dbServer.setDatabaseName(0, "batchtest");
+ // dbServer.setDatabasePath(0, "mem:test;sql.enforce_strict_size=true");
+ dbServer.setDatabasePath(0,
"file:" + TMP_DIR + "batchtest;sql.enforce_strict_size=true");
- dbServer.setLogWriter(null);
- dbServer.setErrWriter(null);
- dbServer.start();
- System.out.println("Database URL: " + dbUrl);
- try {
- Class.forName(driver);
- } catch (Exception e) {
- e.printStackTrace();
- System.out.println(this + ".setUp() error: " + e.getMessage());
- }
- System.out.println("Database server started on port: " + dbServer.getPort());
- }
-
- private void createFile() throws IOException {
- PrintWriter w = new PrintWriter(new FileWriter(INPUT_FILE));
- w = new PrintWriter(new FileWriter(INPUT_FILE));
- w.println("100\tapple\t1.0\t2008-01-01");
- w.println("100\torange\t2.0\t2008-02-01");
- w.println("100\tbanana\t1.1\t2008-03-01");
- w.close();
- Util.copyFromLocalToCluster(cluster, INPUT_FILE, INPUT_FILE);
- }
-
- private void createTable() throws IOException {
- Connection con = null;
- String sql = "create table ttt (id integer, name varchar(32), ratio double, dt date)";
- try {
- con = DriverManager.getConnection(url, user, password);
- } catch (SQLException sqe) {
- throw new IOException("Unable to obtain a connection to the database",
- sqe);
- }
- try {
- Statement st = con.createStatement();
- st.executeUpdate(sql);
- st.close();
- con.commit();
- con.close();
- } catch (SQLException sqe) {
- throw new IOException("Cannot create table", sqe);
- }
- }
-
- @Before
- public void setUp() throws IOException {
- createFile();
- createTable();
- }
-
- @After
- public void tearDown() throws IOException {
- new File(INPUT_FILE).delete();
- Util.deleteFile(cluster, INPUT_FILE);
- pigServer.shutdown();
- dbServer.stop();
- cluster.shutDown();
-
- File[] dbFiles = new File(TMP_DIR).listFiles(new FilenameFilter() {
- @Override
- public boolean accept(File dir, String name) {
- if (name.startsWith("batchtest")) {
- return true;
- } else {
- return false;
- }
- }
- });
- if (dbFiles != null) {
- for (File file : dbFiles) {
- file.delete();
- }
- }
- }
-
- public void testWriteToDB() throws IOException {
- String insertQuery = "insert into ttt (id, name, ratio, dt) values (?,?,?,?)";
- pigServer.setBatchOn();
- String dbStore = "org.apache.pig.piggybank.storage.DBStorage('" + driver
- + "', '" + Util.encodeEscape(url) + "', '" + insertQuery + "');";
- pigServer.registerQuery("A = LOAD '" + INPUT_FILE
- + "' as (id:int, fruit:chararray, ratio:double, dt : datetime);");
- pigServer.registerQuery("STORE A INTO 'dummy' USING " + dbStore);
- ExecJob job = pigServer.executeBatch().get(0);
- try {
- while(!job.hasCompleted()) Thread.sleep(1000);
- } catch(InterruptedException ie) {// ignore
- }
-
- assertNotSame("Failed: " + job.getException(), job.getStatus(),
- ExecJob.JOB_STATUS.FAILED);
-
- Connection con = null;
- String selectQuery = "select id, name, ratio, dt from ttt order by name";
- try {
- con = DriverManager.getConnection(url, user, password);
- } catch (SQLException sqe) {
- throw new IOException(
- "Unable to obtain database connection for data verification", sqe);
- }
- try {
- PreparedStatement ps = con.prepareStatement(selectQuery);
- ResultSet rs = ps.executeQuery();
-
- int expId = 100;
- String[] expNames = { "apple", "banana", "orange" };
- double[] expRatios = { 1.0, 1.1, 2.0 };
+ dbServer.setLogWriter(null);
+ dbServer.setErrWriter(null);
+ dbServer.start();
+ System.out.println("Database URL: " + dbUrl);
+ try {
+ Class.forName(driver);
+ } catch (Exception e) {
+ e.printStackTrace();
+ System.out.println(this + ".setUp() error: " + e.getMessage());
+ }
+ System.out.println("Database server started on port: " + dbServer.getPort());
+ }
+
+ private void createFile() throws IOException {
+ PrintWriter w = new PrintWriter(new FileWriter(INPUT_FILE));
+ w = new PrintWriter(new FileWriter(INPUT_FILE));
+ w.println("100\tapple\t1.0\t2008-01-01");
+ w.println("100\torange\t2.0\t2008-02-01");
+ w.println("100\tbanana\t1.1\t2008-03-01");
+ w.println("\t\t\t");
+ w.close();
+ Util.copyFromLocalToCluster(cluster, INPUT_FILE, INPUT_FILE);
+ }
+
+ private void createTable() throws IOException {
+ Connection con = null;
+ String sql = "create table ttt (id integer, name varchar(32), ratio double, dt date)";
+ try {
+ con = DriverManager.getConnection(url, user, password);
+ } catch (SQLException sqe) {
+ throw new IOException("Unable to obtain a connection to the database",
+ sqe);
+ }
+ try {
+ Statement st = con.createStatement();
+ st.executeUpdate(sql);
+ st.close();
+ con.commit();
+ con.close();
+ } catch (SQLException sqe) {
+ throw new IOException("Cannot create table", sqe);
+ }
+ }
+
+ @Before
+ public void setUp() throws IOException {
+ createFile();
+ createTable();
+ }
+
+ @After
+ public void tearDown() throws IOException {
+ new File(INPUT_FILE).delete();
+ Util.deleteFile(cluster, INPUT_FILE);
+ pigServer.shutdown();
+ dbServer.stop();
+ cluster.shutDown();
+
+ File[] dbFiles = new File(TMP_DIR).listFiles(new FilenameFilter() {
+ @Override
+ public boolean accept(File dir, String name) {
+ if (name.startsWith("batchtest")) {
+ return true;
+ } else {
+ return false;
+ }
+ }
+ });
+ if (dbFiles != null) {
+ for (File file : dbFiles) {
+ file.delete();
+ }
+ }
+ }
+
+ public void testWriteToDB() throws IOException {
+ String insertQuery = "insert into ttt (id, name, ratio, dt) values (?,?,?,?)";
+ pigServer.setBatchOn();
+ String dbStore = "org.apache.pig.piggybank.storage.DBStorage('" + driver
+ + "', '" + Util.encodeEscape(url) + "', '" + insertQuery + "');";
+ pigServer.registerQuery("A = LOAD '" + INPUT_FILE
+ + "' as (id:int, fruit:chararray, ratio:double, dt : datetime);");
+ pigServer.registerQuery("STORE A INTO 'dummy' USING " + dbStore);
+ ExecJob job = pigServer.executeBatch().get(0);
+ try {
+ while(!job.hasCompleted()) Thread.sleep(1000);
+ } catch(InterruptedException ie) {// ignore
+ }
+
+ assertNotSame("Failed: " + job.getException(), job.getStatus(),
+ ExecJob.JOB_STATUS.FAILED);
+
+ Connection con = null;
+ String selectQuery = "select id, name, ratio, dt from ttt order by name";
+ try {
+ con = DriverManager.getConnection(url, user, password);
+ } catch (SQLException sqe) {
+ throw new IOException(
+ "Unable to obtain database connection for data verification", sqe);
+ }
+ try {
+ PreparedStatement ps = con.prepareStatement(selectQuery);
+ ResultSet rs = ps.executeQuery();
+
+ int expId = 100;
+ String[] expNames = { "apple", "banana", "orange" };
+ double[] expRatios = { 1.0, 1.1, 2.0 };
Date [] expDates = {new Date(2008,01,01),new Date(2008,02,01),new Date(2008,03,01)};
- for (int i = 0; i < 3 && rs.next(); i++) {
- assertEquals("Id mismatch", expId, rs.getInt(1));
- assertEquals("Name mismatch", expNames[i], rs.getString(2));
- assertEquals("Ratio mismatch", expRatios[i], rs.getDouble(3), 0.0001);
- assertEquals("Date mismatch", expDates[i], rs.getDate(4));
- }
-
- } catch (SQLException sqe) {
- throw new IOException(
- "Unable to read data from database for verification", sqe);
- }
- }
+ for (int i = 0; i < 4 && rs.next(); i++) {
+ //Need to check for nulls explicitly.
+ if ( i == 0) {
+ //Id
+ rs.getInt(1);
+ assertTrue(rs.wasNull());
+ //Name
+ rs.getString(2);
+ assertTrue(rs.wasNull());
+ //Ratio
+ rs.getDouble(3);
+ assertTrue(rs.wasNull());
+ //Date
+ rs.getDate(4);
+ assertTrue(rs.wasNull());
+ } else {
+ assertEquals("Id mismatch", expId, rs.getInt(1));
+ assertEquals("Name mismatch", expNames[i-1], rs.getString(2));
+ assertEquals("Ratio mismatch", expRatios[i-1], rs.getDouble(3), 0.0001);
+ assertEquals("Date mismatch", expDates[i-1], rs.getDate(4));
+ }
+ }
+
+ } catch (SQLException sqe) {
+ throw new IOException(
+ "Unable to read data from database for verification", sqe);
+ }
+ }
}