You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ra...@apache.org on 2014/09/21 19:33:31 UTC
[1/2] git commit: Phoenix 1044 : Changes to PhoenixRecordReader to
fix scan.
Repository: phoenix
Updated Branches:
refs/heads/3.0 48a2482b4 -> 8eaf08e09
Phoenix 1044 : Changes to PhoenixRecordReader to fix scan.
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/b9da2dc6
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/b9da2dc6
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/b9da2dc6
Branch: refs/heads/3.0
Commit: b9da2dc623799d386218a59119fc97ea1feb24b9
Parents: 48a2482
Author: mravi <ma...@gmail.com>
Authored: Sun Sep 21 10:32:33 2014 -0700
Committer: mravi <ma...@gmail.com>
Committed: Sun Sep 21 10:32:33 2014 -0700
----------------------------------------------------------------------
.../phoenix/pig/PhoenixHBaseLoaderIT.java | 56 ++++++++++++++++++++
.../phoenix/pig/hadoop/PhoenixInputFormat.java | 21 +++-----
.../phoenix/pig/hadoop/PhoenixRecordReader.java | 11 ++--
3 files changed, 67 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b9da2dc6/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixHBaseLoaderIT.java
----------------------------------------------------------------------
diff --git a/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixHBaseLoaderIT.java b/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixHBaseLoaderIT.java
index 8daea9c..6017065 100644
--- a/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixHBaseLoaderIT.java
+++ b/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixHBaseLoaderIT.java
@@ -307,6 +307,62 @@ public class PhoenixHBaseLoaderIT {
}
/**
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testForNonPKSQLQuery() throws Exception {
+
+ //create the table
+ String ddl = "CREATE TABLE " + TABLE_FULL_NAME
+ + " ( ID VARCHAR PRIMARY KEY, FOO VARCHAR, BAR INTEGER, BAZ UNSIGNED_INT)";
+
+ conn.createStatement().execute(ddl);
+
+ //upsert data.
+ final String dml = "UPSERT INTO " + TABLE_FULL_NAME + " VALUES(?,?,?,?) ";
+ PreparedStatement stmt = conn.prepareStatement(dml);
+ stmt.setString(1, "a");
+ stmt.setString(2, "a");
+ stmt.setInt(3,-1);
+ stmt.setInt(4,1);
+ stmt.execute();
+ stmt.setString(1, "b");
+ stmt.setString(2, "b");
+ stmt.setInt(3,-2);
+ stmt.setInt(4,2);
+ stmt.execute();
+
+ conn.commit();
+
+ //sql query
+ final String sqlQuery = String.format(" SELECT FOO, BAZ FROM %s WHERE BAR = -1 " , TABLE_FULL_NAME);
+
+ pigServer.registerQuery(String.format(
+ "A = load 'hbase://query/%s' using org.apache.phoenix.pig.PhoenixHBaseLoader('%s');", sqlQuery,
+ zkQuorum));
+
+ final Iterator<Tuple> iterator = pigServer.openIterator("A");
+ int recordsRead = 0;
+ while (iterator.hasNext()) {
+ final Tuple tuple = iterator.next();
+ assertEquals("a", tuple.get(0));
+ assertEquals(1, tuple.get(1));
+ recordsRead++;
+ }
+ assertEquals(1, recordsRead);
+
+ //test the schema. Test for PHOENIX-1123
+ Schema schema = pigServer.dumpSchema("A");
+ List<FieldSchema> fields = schema.getFields();
+ assertEquals(2, fields.size());
+ assertTrue(fields.get(0).alias.equalsIgnoreCase("FOO"));
+ assertTrue(fields.get(0).type == DataType.CHARARRAY);
+ assertTrue(fields.get(1).alias.equalsIgnoreCase("BAZ"));
+ assertTrue(fields.get(1).type == DataType.INTEGER);
+ }
+
+ /**
* @throws Exception
*/
@Test
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b9da2dc6/phoenix-pig/src/main/java/org/apache/phoenix/pig/hadoop/PhoenixInputFormat.java
----------------------------------------------------------------------
diff --git a/phoenix-pig/src/main/java/org/apache/phoenix/pig/hadoop/PhoenixInputFormat.java b/phoenix-pig/src/main/java/org/apache/phoenix/pig/hadoop/PhoenixInputFormat.java
index ebb9023..4326876 100644
--- a/phoenix-pig/src/main/java/org/apache/phoenix/pig/hadoop/PhoenixInputFormat.java
+++ b/phoenix-pig/src/main/java/org/apache/phoenix/pig/hadoop/PhoenixInputFormat.java
@@ -37,7 +37,6 @@ import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.phoenix.compile.QueryPlan;
import org.apache.phoenix.compile.StatementContext;
-import org.apache.phoenix.iterate.ResultIterator;
import org.apache.phoenix.jdbc.PhoenixStatement;
import org.apache.phoenix.pig.PhoenixPigConfiguration;
import org.apache.phoenix.query.KeyRange;
@@ -46,7 +45,6 @@ import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.util.ScanUtil;
import com.google.common.base.Preconditions;
-import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
/**
@@ -81,19 +79,11 @@ public final class PhoenixInputFormat extends InputFormat<NullWritable, PhoenixR
@Override
- public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException {
- List<InputSplit> splits = null;
- try{
- setConf(context.getConfiguration());
- final QueryPlan queryPlan = getQueryPlan(context);
- @SuppressWarnings("unused")
- final ResultIterator iterator = queryPlan.iterator();
- final List<KeyRange> allSplits = queryPlan.getSplits();
- splits = generateSplits(queryPlan,allSplits);
- } catch(SQLException sqlE) {
- LOG.error(String.format(" Error [%s] in getSplits of PhoenixInputFormat ", sqlE.getMessage()));
- Throwables.propagate(sqlE);
- }
+ public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException {
+ setConf(context.getConfiguration());
+ final QueryPlan queryPlan = getQueryPlan(context);
+ final List<KeyRange> allSplits = queryPlan.getSplits();
+ final List<InputSplit> splits = generateSplits(queryPlan,allSplits);
return splits;
}
@@ -157,6 +147,7 @@ public final class PhoenixInputFormat extends InputFormat<NullWritable, PhoenixR
final Statement statement = connection.createStatement();
final PhoenixStatement pstmt = statement.unwrap(PhoenixStatement.class);
this.queryPlan = pstmt.compileQuery(selectStatement);
+ this.queryPlan.iterator();
} catch(Exception exception) {
LOG.error(String.format("Failed to get the query plan with error [%s]",exception.getMessage()));
throw new RuntimeException(exception);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b9da2dc6/phoenix-pig/src/main/java/org/apache/phoenix/pig/hadoop/PhoenixRecordReader.java
----------------------------------------------------------------------
diff --git a/phoenix-pig/src/main/java/org/apache/phoenix/pig/hadoop/PhoenixRecordReader.java b/phoenix-pig/src/main/java/org/apache/phoenix/pig/hadoop/PhoenixRecordReader.java
index 6d30c3f..a445ce6 100644
--- a/phoenix-pig/src/main/java/org/apache/phoenix/pig/hadoop/PhoenixRecordReader.java
+++ b/phoenix-pig/src/main/java/org/apache/phoenix/pig/hadoop/PhoenixRecordReader.java
@@ -36,6 +36,7 @@ import org.apache.phoenix.iterate.TableResultIterator;
import org.apache.phoenix.jdbc.PhoenixResultSet;
import org.apache.phoenix.pig.PhoenixPigConfiguration;
import org.apache.phoenix.query.KeyRange;
+import org.apache.phoenix.util.ScanUtil;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
@@ -96,13 +97,11 @@ public final class PhoenixRecordReader extends RecordReader<NullWritable,Phoenix
final KeyRange keyRange = pSplit.getKeyRange();
final Scan splitScan = queryPlan.getContext().getScan();
final Scan scan = new Scan(splitScan);
- scan.setStartRow(keyRange.getLowerRange());
- scan.setStopRow(keyRange.getUpperRange());
- try {
- //this.resultIterator = queryPlan.iterator();
- TableResultIterator tableResultIterator = new TableResultIterator(queryPlan.getContext(), queryPlan.getTableRef(),scan);
+ ScanUtil.intersectScanRange(scan, keyRange.getLowerRange(), keyRange.getUpperRange(), queryPlan.getContext().getScanRanges().useSkipScanFilter());
+ try {
+ TableResultIterator tableResultIterator = new TableResultIterator(queryPlan.getContext(), queryPlan.getTableRef(),scan);
if(queryPlan.getContext().getSequenceManager().getSequenceCount() > 0) {
- this.resultIterator = new SequenceResultIterator(tableResultIterator, queryPlan.getContext().getSequenceManager());
+ this.resultIterator = new SequenceResultIterator(tableResultIterator, queryPlan.getContext().getSequenceManager());
} else {
this.resultIterator = tableResultIterator;
}
[2/2] git commit: Phoenix 1050 : Support for DataByteArray
Posted by ra...@apache.org.
Phoenix 1050 : Support for DataByteArray
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/8eaf08e0
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/8eaf08e0
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/8eaf08e0
Branch: refs/heads/3.0
Commit: 8eaf08e0999a7407d9f33fd0771935c01e4edcd5
Parents: b9da2dc
Author: mravi <ma...@gmail.com>
Authored: Sun Sep 21 10:33:01 2014 -0700
Committer: mravi <ma...@gmail.com>
Committed: Sun Sep 21 10:33:01 2014 -0700
----------------------------------------------------------------------
.../phoenix/pig/PhoenixHBaseStorerIT.java | 115 +++++++++++++++++++
.../org/apache/phoenix/pig/util/TypeUtil.java | 48 ++++----
2 files changed, 139 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/8eaf08e0/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixHBaseStorerIT.java
----------------------------------------------------------------------
diff --git a/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixHBaseStorerIT.java b/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixHBaseStorerIT.java
index 1d82362..e0021d9 100644
--- a/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixHBaseStorerIT.java
+++ b/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixHBaseStorerIT.java
@@ -23,6 +23,7 @@ import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR;
import static org.apache.phoenix.util.TestUtil.LOCALHOST;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertArrayEquals;
import java.sql.Connection;
import java.sql.DriverManager;
@@ -30,6 +31,7 @@ import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Collection;
+import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.end2end.BaseHBaseManagedTimeIT;
import org.apache.phoenix.end2end.HBaseManagedTimeTest;
import org.apache.pig.ExecType;
@@ -37,8 +39,10 @@ import org.apache.pig.PigServer;
import org.apache.pig.backend.executionengine.ExecJob.JOB_STATUS;
import org.apache.pig.builtin.mock.Storage;
import org.apache.pig.builtin.mock.Storage.Data;
+import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
+import org.joda.time.DateTime;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
@@ -182,4 +186,115 @@ public class PhoenixHBaseStorerIT extends BaseHBaseManagedTimeIT {
assertEquals(0, rs.getInt(3));
}
}
+
+ /**
+ * Test storage of DataByteArray columns to Phoenix
+ * Maps the DataByteArray with the target PhoenixDataType and persists in HBase.
+ * @throws Exception
+ */
+ @Test
+ public void testStoreWithBinaryDataTypes() throws Exception {
+
+ final String tableName = "TABLE3";
+ final Statement stmt = conn.createStatement();
+
+ stmt.execute("CREATE TABLE " + tableName +
+ " (col1 BIGINT NOT NULL, col2 INTEGER , col3 FLOAT, col4 DOUBLE , col5 TINYINT , " +
+ " col6 BOOLEAN , col7 VARBINARY CONSTRAINT my_pk PRIMARY KEY (col1))");
+
+ final Data data = Storage.resetData(pigServer);
+ final Collection<Tuple> list = Lists.newArrayList();
+
+ int rows = 10;
+ for (int i = 1; i <= rows; i++) {
+ Tuple t = tupleFactory.newTuple();
+ t.append(i);
+ t.append(new DataByteArray(Bytes.toBytes(i * 5)));
+ t.append(new DataByteArray(Bytes.toBytes(i * 10.0F)));
+ t.append(new DataByteArray(Bytes.toBytes(i * 15.0D)));
+ t.append(new DataByteArray(Bytes.toBytes(i)));
+ t.append(new DataByteArray(Bytes.toBytes( i % 2 == 0)));
+ t.append(new DataByteArray(Bytes.toBytes(i)));
+ list.add(t);
+ }
+ data.set("in", "col1:int,col2:bytearray,col3:bytearray,col4:bytearray,col5:bytearray,col6:bytearray,col7:bytearray ", list);
+
+ pigServer.setBatchOn();
+ pigServer.registerQuery("A = LOAD 'in' USING mock.Storage();");
+
+ pigServer.registerQuery("Store A into 'hbase://" + tableName
+ + "' using " + PhoenixHBaseStorage.class.getName() + "('"
+ + zkQuorum + "', '-batchSize 1000');");
+
+ if (pigServer.executeBatch().get(0).getStatus() != JOB_STATUS.COMPLETED) {
+ throw new RuntimeException("Job failed", pigServer.executeBatch()
+ .get(0).getException());
+ }
+
+ final ResultSet rs = stmt
+ .executeQuery(String.format("SELECT col1 , col2 , col3 , col4 , col5 , col6, col7 FROM %s ORDER BY col1" , tableName));
+
+ int count = 0;
+ for (int i = 1; i <= rows; i++) {
+ assertTrue(rs.next());
+ assertEquals(i, rs.getInt(1));
+ assertEquals(i * 5, rs.getInt(2));
+ assertEquals(i * 10.0F, rs.getFloat(3),0.0);
+ assertEquals(i * 15.0D, rs.getInt(4),0.0);
+ assertEquals(i,rs.getInt(5));
+ assertEquals(i % 2 == 0, rs.getBoolean(6));
+ assertArrayEquals(Bytes.toBytes(i), rs.getBytes(7));
+ count++;
+ }
+ assertEquals(rows, count);
+ }
+
+ @Test
+ public void testStoreWithDateTime() throws Exception {
+
+ final String tableName = "TABLE4";
+ final Statement stmt = conn.createStatement();
+
+ stmt.execute("CREATE TABLE " + tableName +
+ " (col1 BIGINT NOT NULL, col2 DATE , col3 TIME, " +
+ " col4 TIMESTAMP CONSTRAINT my_pk PRIMARY KEY (col1))");
+
+ long now = System.currentTimeMillis();
+ final DateTime dt = new DateTime(now);
+
+ final Data data = Storage.resetData(pigServer);
+ final Collection<Tuple> list = Lists.newArrayList();
+ Tuple t = tupleFactory.newTuple();
+
+ t.append(1);
+ t.append(dt);
+ t.append(dt);
+ t.append(dt);
+
+ list.add(t);
+
+ data.set("in", "col1:int,col2:datetime,col3:datetime,col4:datetime", list);
+
+ pigServer.setBatchOn();
+ pigServer.registerQuery("A = LOAD 'in' USING mock.Storage();");
+
+ pigServer.registerQuery("Store A into 'hbase://" + tableName
+ + "' using " + PhoenixHBaseStorage.class.getName() + "('"
+ + zkQuorum + "', '-batchSize 1000');");
+
+ if (pigServer.executeBatch().get(0).getStatus() != JOB_STATUS.COMPLETED) {
+ throw new RuntimeException("Job failed", pigServer.executeBatch()
+ .get(0).getException());
+ }
+
+ final ResultSet rs = stmt
+ .executeQuery(String.format("SELECT col1 , col2 , col3 , col4 FROM %s " , tableName));
+
+ assertTrue(rs.next());
+ assertEquals(1, rs.getInt(1));
+ assertEquals(now, rs.getDate(2).getTime());
+ assertEquals(now, rs.getTime(3).getTime());
+ assertEquals(now, rs.getTimestamp(4).getTime());
+
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/8eaf08e0/phoenix-pig/src/main/java/org/apache/phoenix/pig/util/TypeUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-pig/src/main/java/org/apache/phoenix/pig/util/TypeUtil.java b/phoenix-pig/src/main/java/org/apache/phoenix/pig/util/TypeUtil.java
index f3cacfd..1cdd66d 100644
--- a/phoenix-pig/src/main/java/org/apache/phoenix/pig/util/TypeUtil.java
+++ b/phoenix-pig/src/main/java/org/apache/phoenix/pig/util/TypeUtil.java
@@ -31,6 +31,7 @@ import org.apache.phoenix.pig.hadoop.PhoenixRecord;
import org.apache.phoenix.schema.PDataType;
import org.apache.pig.PigException;
import org.apache.pig.ResourceSchema.ResourceFieldSchema;
+import org.apache.pig.backend.hadoop.hbase.HBaseBinaryConverter;
import org.apache.pig.builtin.Utf8StorageConverter;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.DataType;
@@ -45,7 +46,7 @@ import com.google.common.collect.ImmutableMap.Builder;
public final class TypeUtil {
private static final Log LOG = LogFactory.getLog(TypeUtil.class);
- private static final Utf8StorageConverter utf8Converter = new Utf8StorageConverter();
+ private static final HBaseBinaryConverter binaryConverter = new HBaseBinaryConverter ();
private static final ImmutableMap<PDataType,Byte> phoenixTypeToPigDataType = init();
private TypeUtil(){
@@ -97,7 +98,6 @@ public final class TypeUtil {
if (obj == null) {
return null;
}
-
PDataType sqlType;
switch (type) {
@@ -108,6 +108,7 @@ public final class TypeUtil {
sqlType = PDataType.VARCHAR;
break;
case DataType.DOUBLE:
+ case DataType.BIGDECIMAL:
sqlType = PDataType.DOUBLE;
break;
case DataType.FLOAT:
@@ -117,6 +118,7 @@ public final class TypeUtil {
sqlType = PDataType.INTEGER;
break;
case DataType.LONG:
+ case DataType.BIGINTEGER:
sqlType = PDataType.LONG;
break;
case DataType.BOOLEAN:
@@ -125,6 +127,9 @@ public final class TypeUtil {
case DataType.DATETIME:
sqlType = PDataType.DATE;
break;
+ case DataType.BYTE:
+ sqlType = PDataType.TINYINT;
+ break;
default:
throw new RuntimeException("Unknown type " + obj.getClass().getName()
+ " passed to PhoenixHBaseStorage");
@@ -150,16 +155,17 @@ public final class TypeUtil {
if(inferredPType == null) {
return null;
}
-
- if(inferredPType == PDataType.VARBINARY && targetPhoenixType != PDataType.VARBINARY) {
+
+ if(inferredPType == PDataType.VARBINARY) {
try {
o = castBytes(o, targetPhoenixType);
- inferredPType = getType(o, DataType.findType(o));
+ if(targetPhoenixType != PDataType.VARBINARY && targetPhoenixType != PDataType.BINARY) {
+ inferredPType = getType(o, DataType.findType(o));
+ }
} catch (IOException e) {
throw new RuntimeException("Error while casting bytes for object " +o);
}
}
-
if(inferredPType == PDataType.DATE) {
int inferredSqlType = targetPhoenixType.getSqlType();
@@ -192,42 +198,36 @@ public final class TypeUtil {
* @return Object
* @throws IOException
*/
- public static Object castBytes(Object o, PDataType targetPhoenixType) throws IOException {
+ private static Object castBytes(Object o, PDataType targetPhoenixType) throws IOException {
byte[] bytes = ((DataByteArray)o).get();
switch(targetPhoenixType) {
case CHAR:
case VARCHAR:
- return utf8Converter.bytesToCharArray(bytes);
+ return binaryConverter.bytesToCharArray(bytes);
case UNSIGNED_SMALLINT:
case SMALLINT:
- return utf8Converter.bytesToInteger(bytes).shortValue();
+ return binaryConverter.bytesToInteger(bytes).shortValue();
case UNSIGNED_TINYINT:
case TINYINT:
- return utf8Converter.bytesToInteger(bytes).byteValue();
+ return binaryConverter.bytesToInteger(bytes).byteValue();
case UNSIGNED_INT:
case INTEGER:
- return utf8Converter.bytesToInteger(bytes);
+ return binaryConverter.bytesToInteger(bytes);
case BOOLEAN:
- return utf8Converter.bytesToBoolean(bytes);
- case DECIMAL:
- return utf8Converter.bytesToBigDecimal(bytes);
+ return binaryConverter.bytesToBoolean(bytes);
case FLOAT:
case UNSIGNED_FLOAT:
- return utf8Converter.bytesToFloat(bytes);
+ return binaryConverter.bytesToFloat(bytes);
case DOUBLE:
case UNSIGNED_DOUBLE:
- return utf8Converter.bytesToDouble(bytes);
+ return binaryConverter.bytesToDouble(bytes);
case UNSIGNED_LONG:
case LONG:
- return utf8Converter.bytesToLong(bytes);
- case TIME:
- case TIMESTAMP:
- case DATE:
- case UNSIGNED_TIME:
- case UNSIGNED_TIMESTAMP:
- case UNSIGNED_DATE:
- return utf8Converter.bytesToDateTime(bytes);
+ return binaryConverter.bytesToLong(bytes);
+ case VARBINARY :
+ case BINARY:
+ return bytes;
default:
return o;
}