You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ja...@apache.org on 2014/04/21 18:28:24 UTC
[3/3] git commit: PHOENIX-11 Pig Loader (RaviMagham)
PHOENIX-11 Pig Loader (RaviMagham)
Project: http://git-wip-us.apache.org/repos/asf/incubator-phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-phoenix/commit/a9b8eb9b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-phoenix/tree/a9b8eb9b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-phoenix/diff/a9b8eb9b
Branch: refs/heads/4.0
Commit: a9b8eb9b5129bcc8a274a18b6c18cd554f50c935
Parents: b076655
Author: James Taylor <ja...@apache.org>
Authored: Mon Apr 21 09:28:25 2014 -0700
Committer: James Taylor <ja...@apache.org>
Committed: Mon Apr 21 09:28:25 2014 -0700
----------------------------------------------------------------------
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 2 +-
.../apache/phoenix/query/QueryConstants.java | 3 +-
.../org/apache/phoenix/util/ColumnInfo.java | 13 +
.../org/apache/phoenix/util/PhoenixRuntime.java | 113 ++++-
.../java/org/apache/phoenix/util/QueryUtil.java | 42 +-
.../org/apache/phoenix/util/SchemaUtil.java | 30 +-
.../apache/phoenix/schema/SchemaUtilTest.java | 10 +-
.../org/apache/phoenix/util/QueryUtilTest.java | 16 +-
phoenix-pig/pom.xml | 155 +++++++
.../phoenix/pig/PhoenixHBaseLoaderIT.java | 440 +++++++++++++++++++
.../phoenix/pig/PhoenixHBaseStorerIT.java | 195 ++++++++
.../phoenix/pig/PhoenixPigConfigurationIT.java | 107 +++++
.../apache/phoenix/pig/PhoenixHBaseLoader.java | 253 +++++++++++
.../apache/phoenix/pig/PhoenixHBaseStorage.java | 53 ++-
.../phoenix/pig/PhoenixPigConfiguration.java | 270 +++++++++---
.../java/org/apache/phoenix/pig/TypeUtil.java | 189 --------
.../phoenix/pig/hadoop/PhoenixInputFormat.java | 167 +++++++
.../phoenix/pig/hadoop/PhoenixInputSplit.java | 104 +++++
.../phoenix/pig/hadoop/PhoenixRecord.java | 53 ++-
.../phoenix/pig/hadoop/PhoenixRecordReader.java | 133 ++++++
.../util/ColumnInfoToStringEncoderDecoder.java | 69 +++
.../phoenix/pig/util/PhoenixPigSchemaUtil.java | 70 +++
.../pig/util/QuerySchemaParserFunction.java | 121 +++++
.../pig/util/TableSchemaParserFunction.java | 54 +++
.../org/apache/phoenix/pig/util/TypeUtil.java | 312 +++++++++++++
.../phoenix/pig/PhoenixHBaseStorageTest.java | 136 ------
.../pig/PhoenixPigConfigurationTest.java | 49 +++
.../ColumnInfoToStringEncoderDecoderTest.java | 61 +++
.../pig/util/PhoenixPigSchemaUtilTest.java | 84 ++++
.../pig/util/QuerySchemaParserFunctionTest.java | 109 +++++
.../pig/util/TableSchemaParserFunctionTest.java | 56 +++
31 files changed, 3033 insertions(+), 436 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index 7b2f6da..5f7e017 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -488,7 +488,7 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData, org.apache.pho
@Override
public String getIdentifierQuoteString() throws SQLException {
- return "\"";
+ return SchemaUtil.ESCAPE_CHARACTER;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
index 06bccce..7a5d4f6 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
@@ -38,6 +38,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INCREMENT_BY;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_STATE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_AUTOINCREMENT;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_NULLABLE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_VIEW_REFERENCED;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.KEY_SEQ;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LINK_TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MULTI_TENANT;
@@ -70,7 +71,6 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE_SEQUENCE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_CONSTANT;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_INDEX_ID;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_VIEW_REFERENCED;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_STATEMENT;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_TYPE;
@@ -96,6 +96,7 @@ import org.apache.phoenix.util.ByteUtil;
*/
public interface QueryConstants {
public static final String NAME_SEPARATOR = ".";
+ public static final String NAME_SEPARATOR_REGEX = "\\" + NAME_SEPARATOR;
public final static byte[] NAME_SEPARATOR_BYTES = Bytes.toBytes(NAME_SEPARATOR);
public static final byte NAME_SEPARATOR_BYTE = NAME_SEPARATOR_BYTES[0];
public static final String NULL_SCHEMA_NAME = "";
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java
index b02db20..79eaeb0 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java
@@ -12,6 +12,7 @@ package org.apache.phoenix.util;
import java.util.List;
+import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.PDataType;
import com.google.common.base.Preconditions;
@@ -47,6 +48,18 @@ public class ColumnInfo {
public PDataType getPDataType() {
return PDataType.fromTypeId(sqlType);
}
+
+ /**
+ * Returns the column name without the associated Column Family.
+ * @return
+ */
+ public String getDisplayName() {
+ int index = columnName.indexOf(QueryConstants.NAME_SEPARATOR);
+ if (index < 0) {
+ return columnName;
+ }
+ return columnName.substring(index+1);
+ }
@Override
public String toString() {
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
index 27c0c2a..1ed353b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
@@ -29,7 +29,9 @@ import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;
+import java.util.Set;
import java.util.StringTokenizer;
+import java.util.TreeSet;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValue;
@@ -40,8 +42,11 @@ import org.apache.phoenix.coprocessor.MetaDataProtocol.MetaDataMutationResult;
import org.apache.phoenix.coprocessor.MetaDataProtocol.MutationCode;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.AmbiguousColumnException;
+import org.apache.phoenix.schema.ColumnNotFoundException;
import org.apache.phoenix.schema.MetaDataClient;
import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PColumnFamily;
import org.apache.phoenix.schema.PDataType;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableKey;
@@ -325,7 +330,7 @@ public class PhoenixRuntime {
};
}
- private static PTable getTable(Connection conn, String name) throws SQLException {
+ public static PTable getTable(Connection conn, String name) throws SQLException {
PTable table = null;
PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
try {
@@ -343,6 +348,112 @@ public class PhoenixRuntime {
}
/**
+ * Get list of ColumnInfos that contain Column Name and its associated
+ * PDataType for an import. The supplied list of columns can be null -- if it is non-null,
+ * it represents a user-supplied list of columns to be imported.
+ *
+ * @param conn Phoenix connection from which metadata will be read
+ * @param tableName Phoenix table name whose columns are to be checked. Can include a schema
+ * name
+ * @param columns user-supplied list of import columns, can be null
+ */
+ public static List<ColumnInfo> generateColumnInfo(Connection conn,
+ String tableName, List<String> columns)
+ throws SQLException {
+
+ PTable table = PhoenixRuntime.getTable(conn, tableName);
+ List<ColumnInfo> columnInfoList = Lists.newArrayList();
+ Set<String> unresolvedColumnNames = new TreeSet<String>();
+ if (columns == null) {
+ // use all columns in the table
+ for(PColumn pColumn : table.getColumns()) {
+ int sqlType = pColumn.getDataType().getResultSetSqlType();
+ columnInfoList.add(new ColumnInfo(pColumn.toString(), sqlType));
+ }
+ } else {
+ // Leave "null" as indication to skip b/c it doesn't exist
+ for (int i = 0; i < columns.size(); i++) {
+ String columnName = columns.get(i);
+ try {
+ ColumnInfo columnInfo = PhoenixRuntime.getColumnInfo(table, columnName);
+ columnInfoList.add(columnInfo);
+ } catch (ColumnNotFoundException cnfe) {
+ unresolvedColumnNames.add(columnName.trim());
+ } catch (AmbiguousColumnException ace) {
+ unresolvedColumnNames.add(columnName.trim());
+ }
+ }
+ }
+ // if there exists columns that cannot be resolved, error out.
+ if (unresolvedColumnNames.size()>0) {
+ StringBuilder exceptionMessage = new StringBuilder();
+ boolean first = true;
+ exceptionMessage.append("Unable to resolve these column names:\n");
+ for (String col : unresolvedColumnNames) {
+ if (first) first = false;
+ else exceptionMessage.append(",");
+ exceptionMessage.append(col);
+ }
+ exceptionMessage.append("\nAvailable columns with column families:\n");
+ first = true;
+ for (PColumn pColumn : table.getColumns()) {
+ if (first) first = false;
+ else exceptionMessage.append(",");
+ exceptionMessage.append(pColumn.toString());
+ }
+ throw new SQLException(exceptionMessage.toString());
+ }
+ return columnInfoList;
+ }
+
+ /**
+ * Returns the column info for the given column for the given table.
+ *
+ * @param table
+ * @param columnName User-specified column name. May be family-qualified or bare.
+ * @return columnInfo associated with the column in the table
+ * @throws SQLException if parameters are null or if column is not found or if column is ambiguous.
+ */
+ public static ColumnInfo getColumnInfo(PTable table, String columnName) throws SQLException {
+ if (table==null) {
+ throw new SQLException("Table must not be null.");
+ }
+ if (columnName==null) {
+ throw new SQLException("columnName must not be null.");
+ }
+ columnName = columnName.trim().toUpperCase();
+ PColumn pColumn = null;
+ if (columnName.contains(QueryConstants.NAME_SEPARATOR)) {
+ String[] tokens = columnName.split(QueryConstants.NAME_SEPARATOR_REGEX);
+ if (tokens.length!=2) {
+ throw new SQLException(String.format("Unable to process column %s, expected family-qualified name.",columnName));
+ }
+ String familyName = tokens[0];
+ String familyColumn = tokens[1];
+ PColumnFamily family = table.getColumnFamily(familyName);
+ pColumn = family.getColumn(familyColumn);
+ } else {
+ pColumn = table.getColumn(columnName);
+ }
+ return getColumnInfo(pColumn);
+ }
+
+ /**
+ * Constructs a column info for the supplied pColumn
+ * @param pColumn
+ * @return columnInfo
+ * @throws SQLException if the parameter is null.
+ */
+ public static ColumnInfo getColumnInfo(PColumn pColumn) throws SQLException {
+ if (pColumn==null) {
+ throw new SQLException("pColumn must not be null.");
+ }
+ int sqlType = pColumn.getDataType().getResultSetSqlType();
+ ColumnInfo columnInfo = new ColumnInfo(pColumn.toString(),sqlType);
+ return columnInfo;
+ }
+
+ /**
* Encode the primary key values from the table as a byte array. The values must
* be in the same order as the primary key constraint. If the connection and
* table are both tenant-specific, the tenant ID column must not be present in
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
index 384d845..f25e409 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
@@ -18,6 +18,8 @@
package org.apache.phoenix.util;
+import static org.apache.phoenix.util.SchemaUtil.getEscapedFullColumnName;
+
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -27,10 +29,11 @@ import javax.annotation.Nullable;
import com.google.common.base.Function;
import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
-public class QueryUtil {
+public final class QueryUtil {
/**
* Column family name index within ResultSet resulting from {@link DatabaseMetaData#getColumns(String, String, String, String)}
@@ -54,6 +57,12 @@ public class QueryUtil {
public static final int DATA_TYPE_NAME_POSITION = 6;
/**
+ * Private constructor
+ */
+ private QueryUtil() {
+ }
+
+ /**
* Generate an upsert statement based on a list of {@code ColumnInfo}s with parameter markers. The list of
* {@code ColumnInfo}s must contain at least one element.
*
@@ -81,7 +90,7 @@ public class QueryUtil {
@Nullable
@Override
public String apply(@Nullable ColumnInfo columnInfo) {
- return columnInfo.getColumnName();
+ return getEscapedFullColumnName(columnInfo.getColumnName());
}
})),
Joiner.on(", ").join(parameterList));
@@ -110,6 +119,35 @@ public class QueryUtil {
}
return String.format("UPSERT INTO %s VALUES (%s)", tableName, Joiner.on(", ").join(parameterList));
}
+
+ /**
+ *
+ * @param fullTableName name of the table for which the select statement needs to be created.
+ * @param columnInfos list of columns to be projected in the select statement.
+ * @return Select Query
+ */
+ public static String constructSelectStatement(String fullTableName, List<ColumnInfo> columnInfos) {
+ Preconditions.checkNotNull(fullTableName,"Table name cannot be null");
+ if(columnInfos == null || columnInfos.isEmpty()) {
+ throw new IllegalArgumentException("At least one column must be provided");
+ }
+ // escape the table name to ensure it is case sensitive.
+ final String escapedFullTableName = SchemaUtil.getEscapedFullTableName(fullTableName);
+ StringBuilder sb = new StringBuilder();
+ sb.append("SELECT ");
+ for (ColumnInfo cinfo : columnInfos) {
+ if (cinfo != null) {
+ String fullColumnName = getEscapedFullColumnName(cinfo.getColumnName());
+ sb.append(fullColumnName);
+ sb.append(",");
+ }
+ }
+ // Remove the trailing comma
+ sb.setLength(sb.length() - 1);
+ sb.append(" FROM ");
+ sb.append(escapedFullTableName);
+ return sb.toString();
+ }
public static String getUrl(String server) {
return PhoenixRuntime.JDBC_PROTOCOL + PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR + server;
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
index 2fd1fed..ea7683f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
@@ -52,6 +52,8 @@ import org.apache.phoenix.schema.SaltingUtil;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.ValueSchema.Field;
+import com.google.common.base.Preconditions;
+
/**
*
* Static class for various schema-related utilities
@@ -61,7 +63,7 @@ import org.apache.phoenix.schema.ValueSchema.Field;
*/
public class SchemaUtil {
private static final int VAR_LENGTH_ESTIMATE = 10;
-
+ public static final String ESCAPE_CHARACTER = "\"";
public static final DataBlockEncoding DEFAULT_DATA_BLOCK_ENCODING = DataBlockEncoding.FAST_DIFF;
public static final PDatum VAR_BINARY_DATUM = new PDatum() {
@@ -553,4 +555,30 @@ public class SchemaUtil {
// TODO: when PColumn has getPKPosition, use that instead
return table.getPKColumns().indexOf(column);
}
+
+ public static String getEscapedFullColumnName(String fullColumnName) {
+ int index = fullColumnName.indexOf(QueryConstants.NAME_SEPARATOR);
+ if (index < 0) {
+ return getEscapedArgument(fullColumnName);
+ }
+ String columnFamily = fullColumnName.substring(0,index);
+ String columnName = fullColumnName.substring(index+1);
+ return getEscapedArgument(columnFamily) + QueryConstants.NAME_SEPARATOR + getEscapedArgument(columnName) ;
+ }
+
+ public static String getEscapedFullTableName(String fullTableName) {
+ final String schemaName = getSchemaNameFromFullName(fullTableName);
+ final String tableName = getTableNameFromFullName(fullTableName);
+ return getEscapedTableName(schemaName, tableName);
+ }
+
+ /**
+ * Escapes the given argument with {@value #ESCAPE_CHARACTER}
+ * @param argument any non null value.
+ * @return
+ */
+ public static String getEscapedArgument(String argument) {
+ Preconditions.checkNotNull(argument,"Argument passed cannot be null");
+ return ESCAPE_CHARACTER + argument + ESCAPE_CHARACTER;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/phoenix-core/src/test/java/org/apache/phoenix/schema/SchemaUtilTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/schema/SchemaUtilTest.java b/phoenix-core/src/test/java/org/apache/phoenix/schema/SchemaUtilTest.java
index 36c6a8d..0d465c6 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/schema/SchemaUtilTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/schema/SchemaUtilTest.java
@@ -19,10 +19,9 @@ package org.apache.phoenix.schema;
import static org.junit.Assert.assertEquals;
-import org.junit.Test;
-
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.util.SchemaUtil;
+import org.junit.Test;
public class SchemaUtilTest {
@@ -54,4 +53,11 @@ public class SchemaUtilTest {
columnDisplayName = SchemaUtil.getMetaDataEntityName(null, null, null, "columnName");
assertEquals(columnDisplayName, "columnName");
}
+
+ @Test
+ public void testEscapingColumnName() {
+ assertEquals("\"ID\"", SchemaUtil.getEscapedFullColumnName("ID"));
+ assertEquals("\"0\".\"NAME\"", SchemaUtil.getEscapedFullColumnName("0.NAME"));
+ assertEquals("\"CF1\".\"LOCATION\"", SchemaUtil.getEscapedFullColumnName("CF1.LOCATION"));
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/phoenix-core/src/test/java/org/apache/phoenix/util/QueryUtilTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/util/QueryUtilTest.java b/phoenix-core/src/test/java/org/apache/phoenix/util/QueryUtilTest.java
index 39bde72..0ac2bbc 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/util/QueryUtilTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/util/QueryUtilTest.java
@@ -17,12 +17,13 @@
*/
package org.apache.phoenix.util;
-import com.google.common.collect.ImmutableList;
-import org.junit.Test;
+import static org.junit.Assert.assertEquals;
import java.sql.Types;
-import static org.junit.Assert.assertEquals;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableList;
public class QueryUtilTest {
@@ -32,7 +33,7 @@ public class QueryUtilTest {
@Test
public void testConstructUpsertStatement_ColumnInfos() {
assertEquals(
- "UPSERT INTO MYTAB (ID, NAME) VALUES (?, ?)",
+ "UPSERT INTO MYTAB (\"ID\", \"NAME\") VALUES (?, ?)",
QueryUtil.constructUpsertStatement("MYTAB", ImmutableList.of(ID_COLUMN, NAME_COLUMN)));
}
@@ -53,4 +54,11 @@ public class QueryUtilTest {
public void testConstructGenericUpsertStatement_NoColumns() {
QueryUtil.constructGenericUpsertStatement("MYTAB", 0);
}
+
+ @Test
+ public void testConstructSelectStatement() {
+ assertEquals(
+ "SELECT \"ID\",\"NAME\" FROM \"MYTAB\"",
+ QueryUtil.constructSelectStatement("MYTAB", ImmutableList.of(ID_COLUMN,NAME_COLUMN)));
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/phoenix-pig/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-pig/pom.xml b/phoenix-pig/pom.xml
index edaa1f0..aeb6408 100644
--- a/phoenix-pig/pom.xml
+++ b/phoenix-pig/pom.xml
@@ -37,8 +37,163 @@
<artifactId>phoenix-core</artifactId>
</dependency>
<dependency>
+ <groupId>org.apache.phoenix</groupId>
+ <artifactId>phoenix-core</artifactId>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.apache.pig</groupId>
<artifactId>pig</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ </dependency>
+ <!-- Test Dependencies -->
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ </dependency>
</dependencies>
+
+ <profiles>
+ <!-- Profile for building against Hadoop 1. Active by default. Not used if another
+ Hadoop profile is specified with mvn -Dhadoop.profile=foo -->
+ <profile>
+ <id>hadoop-1</id>
+ <activation>
+ <property>
+ <name>!hadoop.profile</name>
+ </property>
+ </activation>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-testing-util</artifactId>
+ <version>${hbase-hadoop1.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-it</artifactId>
+ <version>${hbase-hadoop1.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <version>${hbase-hadoop1.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-protocol</artifactId>
+ <version>${hbase-hadoop1.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-client</artifactId>
+ <version>${hbase-hadoop1.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-test</artifactId>
+ </dependency>
+ </dependencies>
+ </profile>
+
+ <!-- Profile for building against Hadoop 2. Activate using: mvn -Dhadoop.profile=2 -->
+ <profile>
+ <id>hadoop-2</id>
+ <activation>
+ <property>
+ <name>hadoop.profile</name>
+ <value>2</value>
+ </property>
+ </activation>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-testing-util</artifactId>
+ <version>${hbase-hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-it</artifactId>
+ <version>${hbase-hadoop2.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <version>${hbase-hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-protocol</artifactId>
+ <version>${hbase-hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-client</artifactId>
+ <version>${hbase-hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ <version>${hbase-hadoop2.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ <version>${hbase-hadoop2.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop2-compat</artifactId>
+ <version>${hbase-hadoop2.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop2-compat</artifactId>
+ <version>${hbase-hadoop2.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-annotations</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-minicluster</artifactId>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <version>${maven-dependency-plugin.version}</version>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/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
new file mode 100644
index 0000000..90f1c79
--- /dev/null
+++ b/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixHBaseLoaderIT.java
@@ -0,0 +1,440 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ *distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you maynot use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicablelaw or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.pig;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.phoenix.jdbc.PhoenixDriver;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.TestUtil;
+import org.apache.pig.ExecType;
+import org.apache.pig.PigServer;
+import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
+import org.apache.pig.builtin.mock.Storage;
+import org.apache.pig.builtin.mock.Storage.Data;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.base.Preconditions;
+
+/**
+ *
+ * Test class to run all the integration tests against a virtual map reduce cluster.
+ */
+public class PhoenixHBaseLoaderIT {
+
+ private static final Log LOG = LogFactory.getLog(PhoenixHBaseLoaderIT.class);
+ private static final String SCHEMA_NAME = "T";
+ private static final String TABLE_NAME = "A";
+ private static final String TABLE_FULL_NAME = SchemaUtil.getTableName(SCHEMA_NAME, TABLE_NAME);
+ private static HBaseTestingUtility hbaseTestUtil;
+ private static String zkQuorum;
+ private static Connection conn;
+ private static PigServer pigServer;
+ private static Configuration conf;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ hbaseTestUtil = new HBaseTestingUtility();
+ hbaseTestUtil.getConfiguration().set(QueryServices.DROP_METADATA_ATTRIB, Boolean.toString(true));
+ hbaseTestUtil.startMiniCluster();
+
+ Class.forName(PhoenixDriver.class.getName());
+ zkQuorum = "localhost:" + hbaseTestUtil.getZkCluster().getClientPort();
+ Properties props = TestUtil.TEST_PROPERTIES;
+ props.put(QueryServices.DROP_METADATA_ATTRIB, Boolean.toString(true));
+ conn = DriverManager.getConnection(PhoenixRuntime.JDBC_PROTOCOL +
+ PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR + zkQuorum,props);
+
+ conf = hbaseTestUtil.getConfiguration();
+ }
+
+ @Before
+ public void beforeTest() throws Exception {
+ pigServer = new PigServer(ExecType.LOCAL,
+ ConfigurationUtil.toProperties(conf));
+ }
+
+ /**
+ * Validates the schema returned for a table with Pig data types.
+ * @throws Exception
+ */
+ @Test
+ public void testSchemaForTable() throws Exception {
+ final String ddl = String.format("CREATE TABLE %s "
+ + " (a_string varchar not null, a_binary varbinary not null, a_integer integer, cf1.a_float float"
+ + " CONSTRAINT pk PRIMARY KEY (a_string, a_binary))\n", TABLE_FULL_NAME);
+ conn.createStatement().execute(ddl);
+
+ pigServer.registerQuery(String.format(
+ "A = load 'hbase://table/%s' using org.apache.phoenix.pig.PhoenixHBaseLoader('%s');", TABLE_FULL_NAME,
+ zkQuorum));
+
+ final Schema schema = pigServer.dumpSchema("A");
+ List<FieldSchema> fields = schema.getFields();
+ assertEquals(4, fields.size());
+ assertTrue(fields.get(0).alias.equalsIgnoreCase("a_string"));
+ assertTrue(fields.get(0).type == DataType.CHARARRAY);
+ assertTrue(fields.get(1).alias.equalsIgnoreCase("a_binary"));
+ assertTrue(fields.get(1).type == DataType.BYTEARRAY);
+ assertTrue(fields.get(2).alias.equalsIgnoreCase("a_integer"));
+ assertTrue(fields.get(2).type == DataType.INTEGER);
+ assertTrue(fields.get(3).alias.equalsIgnoreCase("a_float"));
+ assertTrue(fields.get(3).type == DataType.FLOAT);
+ }
+
+ /**
+ * Validates the schema returned when specific columns of a table are given as part of LOAD .
+ * @throws Exception
+ */
+ @Test
+ public void testSchemaForTableWithSpecificColumns() throws Exception {
+
+ //create the table
+ final String ddl = "CREATE TABLE " + TABLE_FULL_NAME
+ + " (ID INTEGER NOT NULL PRIMARY KEY,NAME VARCHAR, AGE INTEGER) ";
+ conn.createStatement().execute(ddl);
+
+
+ final String selectColumns = "ID,NAME";
+ pigServer.registerQuery(String.format(
+ "A = load 'hbase://table/%s/%s' using org.apache.phoenix.pig.PhoenixHBaseLoader('%s');",
+ TABLE_FULL_NAME, selectColumns, zkQuorum));
+
+ Schema schema = pigServer.dumpSchema("A");
+ List<FieldSchema> fields = schema.getFields();
+ assertEquals(2, fields.size());
+ assertTrue(fields.get(0).alias.equalsIgnoreCase("ID"));
+ assertTrue(fields.get(0).type == DataType.INTEGER);
+ assertTrue(fields.get(1).alias.equalsIgnoreCase("NAME"));
+ assertTrue(fields.get(1).type == DataType.CHARARRAY);
+
+ }
+
+ /**
+ * Validates the schema returned when a SQL SELECT query is given as part of LOAD .
+ * @throws Exception
+ */
+ @Test
+ public void testSchemaForQuery() throws Exception {
+
+ //create the table.
+ String ddl = String.format("CREATE TABLE " + TABLE_FULL_NAME +
+ " (A_STRING VARCHAR NOT NULL, A_DECIMAL DECIMAL NOT NULL, CF1.A_INTEGER INTEGER, CF2.A_DOUBLE DOUBLE"
+ + " CONSTRAINT pk PRIMARY KEY (A_STRING, A_DECIMAL))\n", TABLE_FULL_NAME);
+ conn.createStatement().execute(ddl);
+
+
+
+ //sql query for LOAD
+ final String sqlQuery = "SELECT A_STRING,CF1.A_INTEGER,CF2.A_DOUBLE FROM " + TABLE_FULL_NAME;
+ pigServer.registerQuery(String.format(
+ "A = load 'hbase://query/%s' using org.apache.phoenix.pig.PhoenixHBaseLoader('%s');",
+ sqlQuery, zkQuorum));
+
+ //assert the schema.
+ Schema schema = pigServer.dumpSchema("A");
+ List<FieldSchema> fields = schema.getFields();
+ assertEquals(3, fields.size());
+ assertTrue(fields.get(0).alias.equalsIgnoreCase("a_string"));
+ assertTrue(fields.get(0).type == DataType.CHARARRAY);
+ assertTrue(fields.get(1).alias.equalsIgnoreCase("a_integer"));
+ assertTrue(fields.get(1).type == DataType.INTEGER);
+ assertTrue(fields.get(2).alias.equalsIgnoreCase("a_double"));
+ assertTrue(fields.get(2).type == DataType.DOUBLE);
+ }
+
+ /**
+ * Validates the schema when it is given as part of LOAD..AS
+ * @throws Exception
+ */
+ @Test
+ public void testSchemaForTableWithAlias() throws Exception {
+
+ //create the table.
+ String ddl = "CREATE TABLE " + TABLE_FULL_NAME
+ + " (A_STRING VARCHAR NOT NULL, A_DECIMAL DECIMAL NOT NULL, CF1.A_INTEGER INTEGER, CF2.A_DOUBLE DOUBLE"
+ + " CONSTRAINT pk PRIMARY KEY (A_STRING, A_DECIMAL)) \n";
+ conn.createStatement().execute(ddl);
+
+ //select query given as part of LOAD.
+ final String sqlQuery = "SELECT A_STRING,A_DECIMAL,CF1.A_INTEGER,CF2.A_DOUBLE FROM " + TABLE_FULL_NAME;
+
+ LOG.info(String.format("Generated SQL Query [%s]",sqlQuery));
+
+ pigServer.registerQuery(String.format(
+ "raw = load 'hbase://query/%s' using org.apache.phoenix.pig.PhoenixHBaseLoader('%s') AS (a:chararray,b:bigdecimal,c:int,d:double);",
+ sqlQuery, zkQuorum));
+
+ //test the schema.
+ Schema schema = pigServer.dumpSchema("raw");
+ List<FieldSchema> fields = schema.getFields();
+ assertEquals(4, fields.size());
+ assertTrue(fields.get(0).alias.equalsIgnoreCase("a"));
+ assertTrue(fields.get(0).type == DataType.CHARARRAY);
+ assertTrue(fields.get(1).alias.equalsIgnoreCase("b"));
+ assertTrue(fields.get(1).type == DataType.BIGDECIMAL);
+ assertTrue(fields.get(2).alias.equalsIgnoreCase("c"));
+ assertTrue(fields.get(2).type == DataType.INTEGER);
+ assertTrue(fields.get(3).alias.equalsIgnoreCase("d"));
+ assertTrue(fields.get(3).type == DataType.DOUBLE);
+ }
+
+ /**
+ * @throws Exception
+ */
+ @Test
+ public void testDataForTable() throws Exception {
+
+ //create the table
+ String ddl = "CREATE TABLE " + TABLE_FULL_NAME
+ + " (ID INTEGER NOT NULL PRIMARY KEY, NAME VARCHAR, AGE INTEGER) ";
+
+ conn.createStatement().execute(ddl);
+
+ //prepare data with 10 rows having age 25 and the other 30.
+ final String dml = "UPSERT INTO " + TABLE_FULL_NAME + " VALUES(?,?,?)";
+ PreparedStatement stmt = conn.prepareStatement(dml);
+ int rows = 20;
+ for(int i = 0 ; i < rows; i++) {
+ stmt.setInt(1, i);
+ stmt.setString(2, "a"+i);
+ stmt.setInt(3, (i % 2 == 0) ? 25 : 30);
+ stmt.execute();
+ }
+ conn.commit();
+
+ //load data and filter rows whose age is > 25
+ pigServer.registerQuery(String.format(
+ "A = load 'hbase://table/%s' using org.apache.phoenix.pig.PhoenixHBaseLoader('%s');", TABLE_FULL_NAME,
+ zkQuorum));
+ pigServer.registerQuery("B = FILTER A BY AGE > 25;");
+
+ final Iterator<Tuple> iterator = pigServer.openIterator("B");
+ int recordsRead = 0;
+ while (iterator.hasNext()) {
+ final Tuple each = iterator.next();
+ assertEquals(3, each.size());
+ recordsRead++;
+ }
+ assertEquals(rows/2, recordsRead);
+ }
+
+ /**
+ * @throws Exception
+ */
+ @Test
+ public void testDataForSQLQuery() throws Exception {
+
+ //create the table
+ String ddl = "CREATE TABLE " + TABLE_FULL_NAME
+ + " (ID INTEGER NOT NULL PRIMARY KEY, NAME VARCHAR, AGE INTEGER) ";
+
+ conn.createStatement().execute(ddl);
+
+ //prepare data with 10 rows having age 25 and the other 30.
+ final String dml = "UPSERT INTO " + TABLE_FULL_NAME + " VALUES(?,?,?)";
+ PreparedStatement stmt = conn.prepareStatement(dml);
+ int rows = 20;
+ for(int i = 0 ; i < rows; i++) {
+ stmt.setInt(1, i);
+ stmt.setString(2, "a"+i);
+ stmt.setInt(3, (i % 2 == 0) ? 25 : 30);
+ stmt.execute();
+ }
+ conn.commit();
+
+ //sql query
+ final String sqlQuery = " SELECT ID,NAME,AGE FROM " + TABLE_FULL_NAME + " WHERE AGE > 25";
+ //load data and filter rows whose age is > 25
+ 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()) {
+ iterator.next();
+ recordsRead++;
+ }
+ assertEquals(rows/2, recordsRead);
+ }
+
+ /**
+ * @throws Exception
+ */
+ @Test
+ public void testGroupingOfDataForTable() throws Exception {
+
+ //create the table
+ String ddl = "CREATE TABLE " + TABLE_FULL_NAME
+ + " (ID INTEGER NOT NULL PRIMARY KEY, NAME VARCHAR, AGE INTEGER, SAL INTEGER) ";
+
+ conn.createStatement().execute(ddl);
+
+ //prepare data with 10 rows having age 25 and the other 30.
+ final String dml = "UPSERT INTO " + TABLE_FULL_NAME + " VALUES(?,?,?,?)";
+ PreparedStatement stmt = conn.prepareStatement(dml);
+ int rows = 20;
+ int j = 0, k = 0;
+ for(int i = 0 ; i < rows; i++) {
+ stmt.setInt(1, i);
+ stmt.setString(2, "a"+i);
+ if(i % 2 == 0) {
+ stmt.setInt(3, 25);
+ stmt.setInt(4, 10 * 2 * j++);
+ } else {
+ stmt.setInt(3, 30);
+ stmt.setInt(4, 10 * 3 * k++);
+ }
+
+ stmt.execute();
+ }
+ conn.commit();
+
+ //prepare the mock storage with expected output
+ final Data data = Storage.resetData(pigServer);
+ List<Tuple> expectedList = new ArrayList<Tuple>();
+ expectedList.add(Storage.tuple(0,180));
+ expectedList.add(Storage.tuple(0,270));
+
+ //load data and filter rows whose age is > 25
+ pigServer.setBatchOn();
+ pigServer.registerQuery(String.format(
+ "A = load 'hbase://table/%s' using org.apache.phoenix.pig.PhoenixHBaseLoader('%s');", TABLE_FULL_NAME,
+ zkQuorum));
+
+ pigServer.registerQuery("B = GROUP A BY AGE;");
+ pigServer.registerQuery("C = FOREACH B GENERATE MIN(A.SAL),MAX(A.SAL);");
+ pigServer.registerQuery("STORE C INTO 'out' using mock.Storage();");
+ pigServer.executeBatch();
+
+ List<Tuple> actualList = data.get("out");
+ assertEquals(expectedList, actualList);
+ }
+
+ /**
+ * Tests both {@link PhoenixHBaseLoader} and {@link PhoenixHBaseStorage}
+ * @throws Exception
+ */
+ @Test
+ public void testLoadAndStore() throws Exception {
+
+ //create the tables
+ final String sourceTableddl = "CREATE TABLE " + TABLE_FULL_NAME
+ + " (ID INTEGER NOT NULL PRIMARY KEY, NAME VARCHAR, AGE INTEGER, SAL INTEGER) ";
+
+ final String targetTable = "AGGREGATE";
+ final String targetTableddl = "CREATE TABLE " + targetTable
+ + "(AGE INTEGER NOT NULL PRIMARY KEY , MIN_SAL INTEGER , MAX_SAL INTEGER) ";
+
+ conn.createStatement().execute(sourceTableddl);
+ conn.createStatement().execute(targetTableddl);
+
+ //prepare data with 10 rows having age 25 and the other 30.
+ final String dml = "UPSERT INTO " + TABLE_FULL_NAME + " VALUES(?,?,?,?)";
+ PreparedStatement stmt = conn.prepareStatement(dml);
+ int rows = 20;
+ int j = 0, k = 0;
+ for(int i = 0 ; i < rows; i++) {
+ stmt.setInt(1, i);
+ stmt.setString(2, "a"+i);
+ if(i % 2 == 0) {
+ stmt.setInt(3, 25);
+ stmt.setInt(4, 10 * 2 * j++);
+ } else {
+ stmt.setInt(3, 30);
+ stmt.setInt(4, 10 * 3 * k++);
+ }
+
+ stmt.execute();
+ }
+ conn.commit();
+
+
+ //load data and filter rows whose age is > 25
+ pigServer.setBatchOn();
+ pigServer.registerQuery(String.format(
+ "A = load 'hbase://table/%s' using org.apache.phoenix.pig.PhoenixHBaseLoader('%s');", TABLE_FULL_NAME,
+ zkQuorum));
+
+ pigServer.registerQuery("B = GROUP A BY AGE;");
+ pigServer.registerQuery("C = FOREACH B GENERATE group as AGE,MIN(A.SAL),MAX(A.SAL);");
+ pigServer.registerQuery("STORE C INTO 'hbase://" + targetTable
+ + "' using " + PhoenixHBaseStorage.class.getName() + "('"
+ + zkQuorum + "', '-batchSize 1000');");
+ pigServer.executeBatch();
+
+ //validate the data with what is stored.
+ final String selectQuery = "SELECT AGE , MIN_SAL ,MAX_SAL FROM " + targetTable + " ORDER BY AGE";
+ final ResultSet rs = conn.createStatement().executeQuery(selectQuery);
+ rs.next();
+ assertEquals(25, rs.getInt("AGE"));
+ assertEquals(0, rs.getInt("MIN_SAL"));
+ assertEquals(180, rs.getInt("MAX_SAL"));
+ rs.next();
+ assertEquals(30, rs.getInt("AGE"));
+ assertEquals(0, rs.getInt("MIN_SAL"));
+ assertEquals(270, rs.getInt("MAX_SAL"));
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ dropTable(TABLE_FULL_NAME);
+ pigServer.shutdown();
+ }
+
+
+ private void dropTable(String tableFullName) throws SQLException {
+ Preconditions.checkNotNull(conn);
+ conn.createStatement().execute(String.format("DROP TABLE %s",tableFullName));
+ }
+
+ @AfterClass
+ public static void tearDownAfterClass() throws Exception {
+ conn.close();
+ PhoenixDriver.INSTANCE.close();
+ hbaseTestUtil.shutdownMiniCluster();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/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
new file mode 100644
index 0000000..32d0ff9
--- /dev/null
+++ b/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixHBaseStorerIT.java
@@ -0,0 +1,195 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ *distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you maynot use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicablelaw or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.pig;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Collection;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.phoenix.jdbc.PhoenixDriver;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.pig.ExecType;
+import org.apache.pig.PigServer;
+import org.apache.pig.backend.executionengine.ExecJob.JOB_STATUS;
+import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
+import org.apache.pig.builtin.mock.Storage;
+import org.apache.pig.builtin.mock.Storage.Data;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+
+public class PhoenixHBaseStorerIT {
+
+ private static TupleFactory tupleFactory;
+ private static HBaseTestingUtility hbaseTestUtil;
+ private static String zkQuorum;
+ private static Connection conn;
+ private static PigServer pigServer;
+ private static Configuration conf;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ hbaseTestUtil = new HBaseTestingUtility();
+ hbaseTestUtil.startMiniCluster();
+
+ Class.forName(PhoenixDriver.class.getName());
+ zkQuorum = "localhost:" + hbaseTestUtil.getZkCluster().getClientPort();
+ conn = DriverManager.getConnection(PhoenixRuntime.JDBC_PROTOCOL +
+ PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR + zkQuorum);
+ conf = hbaseTestUtil.getConfiguration();
+ // Pig variables
+ tupleFactory = TupleFactory.getInstance();
+ }
+
+ @Before
+ public void beforeTest() throws Exception {
+ pigServer = new PigServer(ExecType.LOCAL,
+ ConfigurationUtil.toProperties(conf));
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ pigServer.shutdown();
+ }
+
+ @AfterClass
+ public static void tearDownAfterClass() throws Exception {
+ conn.close();
+ PhoenixDriver.INSTANCE.close();
+ hbaseTestUtil.shutdownMiniCluster();
+ }
+
+ /**
+ * Basic test - writes data to a Phoenix table and compares the data written
+ * to expected
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testStorer() throws Exception {
+ final String tableName = "TABLE1";
+ final Statement stmt = conn.createStatement();
+
+ stmt.execute("CREATE TABLE " + tableName +
+ " (ID INTEGER NOT NULL PRIMARY KEY, NAME VARCHAR)");
+
+ final Data data = Storage.resetData(pigServer);
+ final Collection<Tuple> list = Lists.newArrayList();
+
+ // Create input dataset
+ int rows = 100;
+ for (int i = 0; i < rows; i++) {
+ Tuple t = tupleFactory.newTuple();
+ t.append(i);
+ t.append("a" + i);
+ list.add(t);
+ }
+ data.set("in", "id:int, name:chararray", 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');");
+
+ // Now run the Pig script
+ if (pigServer.executeBatch().get(0).getStatus() != JOB_STATUS.COMPLETED) {
+ throw new RuntimeException("Job failed", pigServer.executeBatch()
+ .get(0).getException());
+ }
+
+ // Compare data in Phoenix table to the expected
+ final ResultSet rs = stmt
+ .executeQuery("SELECT id, name FROM table1 ORDER BY id");
+
+ for (int i = 0; i < rows; i++) {
+ assertTrue(rs.next());
+ assertEquals(i, rs.getInt(1));
+ assertEquals("a" + i, rs.getString(2));
+ }
+ }
+
+ /**
+ * Basic test - writes specific columns data to a Phoenix table and compares the data written
+ * to expected
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testStorerForSpecificColumns() throws Exception {
+ final String tableName = "TABLE2";
+ final Statement stmt = conn.createStatement();
+
+ stmt.execute("CREATE TABLE " + tableName +
+ " (ID INTEGER NOT NULL PRIMARY KEY, NAME VARCHAR, AGE INTEGER)");
+
+ final Data data = Storage.resetData(pigServer);
+ final Collection<Tuple> list = Lists.newArrayList();
+
+ // Create input dataset
+ int rows = 100;
+ for (int i = 0; i < rows; i++) {
+ Tuple t = tupleFactory.newTuple();
+ t.append(i);
+ t.append("a" + i);
+ t.append(i * 2);
+ list.add(t);
+ }
+ data.set("in", "id:int, name:chararray,age:int", list);
+
+ pigServer.setBatchOn();
+ pigServer.registerQuery("A = LOAD 'in' USING mock.Storage();");
+ pigServer.registerQuery("B = FOREACH A GENERATE id,name;");
+ pigServer.registerQuery("Store B into 'hbase://" + tableName + "/ID,NAME"
+ + "' using " + PhoenixHBaseStorage.class.getName() + "('"
+ + zkQuorum + "', '-batchSize 1000');");
+
+ // Now run the Pig script
+ if (pigServer.executeBatch().get(0).getStatus() != JOB_STATUS.COMPLETED) {
+ throw new RuntimeException("Job failed", pigServer.executeBatch()
+ .get(0).getException());
+ }
+
+ // Compare data in Phoenix table to the expected
+ final ResultSet rs = stmt
+ .executeQuery("SELECT id, name,age FROM " + tableName + "ORDER BY id");
+
+ for (int i = 0; i < rows; i++) {
+ assertTrue(rs.next());
+ assertEquals(i, rs.getInt(1));
+ assertEquals("a" + i, rs.getString(2));
+ assertEquals(0, rs.getInt(3));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixPigConfigurationIT.java
----------------------------------------------------------------------
diff --git a/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixPigConfigurationIT.java b/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixPigConfigurationIT.java
new file mode 100644
index 0000000..e867a4d
--- /dev/null
+++ b/phoenix-pig/src/it/java/org/apache/phoenix/pig/PhoenixPigConfigurationIT.java
@@ -0,0 +1,107 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ *distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you maynot use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicablelaw or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.pig;
+
+import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_TERMINATOR;
+import static org.apache.phoenix.util.PhoenixRuntime.PHOENIX_TEST_DRIVER_URL_PARAM;
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static org.junit.Assert.assertEquals;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.util.Properties;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.phoenix.end2end.BaseHBaseManagedTimeIT;
+import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Test;
+
+public class PhoenixPigConfigurationIT extends BaseHBaseManagedTimeIT {
+ private static final String zkQuorum = TestUtil.LOCALHOST + JDBC_PROTOCOL_TERMINATOR + PHOENIX_TEST_DRIVER_URL_PARAM;
+
+ @Test
+ public void testUpsertStatement() throws Exception {
+ Properties props = new Properties(TEST_PROPERTIES);
+ Connection conn = DriverManager.getConnection(getUrl(), props);
+ conn.setAutoCommit(false);
+ final String tableName = "TEST_TABLE";
+ try {
+ String ddl = "CREATE TABLE "+ tableName +
+ " (a_string varchar not null, a_binary varbinary not null, col1 integer" +
+ " CONSTRAINT pk PRIMARY KEY (a_string, a_binary))\n";
+ createTestTable(getUrl(), ddl);
+ final PhoenixPigConfiguration configuration = newConfiguration (tableName);
+ final String upserStatement = configuration.getUpsertStatement();
+ final String expectedUpsertStatement = "UPSERT INTO " + tableName + " VALUES (?, ?, ?)";
+ assertEquals(expectedUpsertStatement, upserStatement);
+ } finally {
+ conn.close();
+ }
+ }
+
+ @Test
+ public void testSelectStatement() throws Exception {
+ Properties props = new Properties(TEST_PROPERTIES);
+ Connection conn = DriverManager.getConnection(getUrl(), props);
+ conn.setAutoCommit(false);
+ final String tableName = "TEST_TABLE";
+ try {
+ String ddl = "CREATE TABLE "+ tableName +
+ " (a_string varchar not null, a_binary varbinary not null, col1 integer" +
+ " CONSTRAINT pk PRIMARY KEY (a_string, a_binary))\n";
+ createTestTable(getUrl(), ddl);
+ final PhoenixPigConfiguration configuration = newConfiguration (tableName);
+ final String selectStatement = configuration.getSelectStatement();
+ final String expectedSelectStatement = "SELECT \"A_STRING\",\"A_BINARY\",\"0\".\"COL1\" FROM " + SchemaUtil.getEscapedArgument(tableName) ;
+ assertEquals(expectedSelectStatement, selectStatement);
+ } finally {
+ conn.close();
+ }
+ }
+
+ @Test
+ public void testSelectStatementForSpecificColumns() throws Exception {
+ Properties props = new Properties(TEST_PROPERTIES);
+ Connection conn = DriverManager.getConnection(getUrl(), props);
+ conn.setAutoCommit(false);
+ final String tableName = "TEST_TABLE";
+ try {
+ String ddl = "CREATE TABLE "+ tableName +
+ " (a_string varchar not null, a_binary varbinary not null, col1 integer" +
+ " CONSTRAINT pk PRIMARY KEY (a_string, a_binary))\n";
+ createTestTable(getUrl(), ddl);
+ final PhoenixPigConfiguration configuration = newConfiguration (tableName);
+ configuration.setSelectColumns("a_binary");
+ final String selectStatement = configuration.getSelectStatement();
+ final String expectedSelectStatement = "SELECT \"A_BINARY\" FROM " + SchemaUtil.getEscapedArgument(tableName) ;
+ assertEquals(expectedSelectStatement, selectStatement);
+ } finally {
+ conn.close();
+ }
+ }
+
+ private PhoenixPigConfiguration newConfiguration(String tableName) {
+ final Configuration configuration = new Configuration();
+ final PhoenixPigConfiguration phoenixConfiguration = new PhoenixPigConfiguration(configuration);
+ phoenixConfiguration.configure(zkQuorum, tableName.toUpperCase(), 100);
+ return phoenixConfiguration;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/phoenix-pig/src/main/java/org/apache/phoenix/pig/PhoenixHBaseLoader.java
----------------------------------------------------------------------
diff --git a/phoenix-pig/src/main/java/org/apache/phoenix/pig/PhoenixHBaseLoader.java b/phoenix-pig/src/main/java/org/apache/phoenix/pig/PhoenixHBaseLoader.java
new file mode 100644
index 0000000..1f81ad6
--- /dev/null
+++ b/phoenix-pig/src/main/java/org/apache/phoenix/pig/PhoenixHBaseLoader.java
@@ -0,0 +1,253 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ *distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you maynot use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicablelaw or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.pig;
+
+import static org.apache.commons.lang.StringUtils.isEmpty;
+
+import java.io.IOException;
+import java.util.Properties;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.phoenix.pig.hadoop.PhoenixInputFormat;
+import org.apache.phoenix.pig.hadoop.PhoenixRecord;
+import org.apache.phoenix.pig.util.PhoenixPigSchemaUtil;
+import org.apache.phoenix.pig.util.QuerySchemaParserFunction;
+import org.apache.phoenix.pig.util.TableSchemaParserFunction;
+import org.apache.phoenix.pig.util.TypeUtil;
+import org.apache.pig.Expression;
+import org.apache.pig.LoadFunc;
+import org.apache.pig.LoadMetadata;
+import org.apache.pig.PigException;
+import org.apache.pig.ResourceSchema;
+import org.apache.pig.ResourceStatistics;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.util.ObjectSerializer;
+import org.apache.pig.impl.util.UDFContext;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * LoadFunc to load data from HBase using Phoenix .
+ *
+ * Example usage:
+ * a) TABLE
+ * i) A = load 'hbase://table/HIRES' using
+ * org.apache.phoenix.pig.PhoenixHBaseLoader('localhost');
+ *
+ * The above loads the data from a table 'HIRES'
+ *
+ * ii) A = load 'hbase://table/HIRES/id,name' using
+ * org.apache.phoenix.pig.PhoenixHBaseLoader('localhost');
+ *
+ * Here, only id, name are returned from the table HIRES as part of LOAD.
+ *
+ * b) QUERY
+ * i) B = load 'hbase://query/SELECT fname, lname FROM HIRES' using
+ * org.apache.phoenix.pig.PhoenixHBaseLoader('localhost');
+ *
+ * The above loads fname and lname columns from 'HIRES' table.
+ *
+ */
+public final class PhoenixHBaseLoader extends LoadFunc implements LoadMetadata {
+
+ private static final Log LOG = LogFactory.getLog(PhoenixHBaseLoader.class);
+ private static final String PHOENIX_TABLE_NAME_SCHEME = "hbase://table/";
+ private static final String PHOENIX_QUERY_SCHEME = "hbase://query/";
+ private static final String RESOURCE_SCHEMA_SIGNATURE = "phoenix.pig.schema";
+
+ private PhoenixPigConfiguration config;
+ private String tableName;
+ private String selectQuery;
+ private String zkQuorum ;
+ private PhoenixInputFormat inputFormat;
+ private RecordReader<NullWritable, PhoenixRecord> reader;
+ private String contextSignature;
+ private ResourceSchema schema;
+
+ /**
+ * @param zkQuorum
+ */
+ public PhoenixHBaseLoader(String zkQuorum) {
+ super();
+ Preconditions.checkNotNull(zkQuorum);
+ Preconditions.checkState(zkQuorum.length() > 0, "Zookeeper quorum cannot be empty!");
+ this.zkQuorum = zkQuorum;
+ }
+
+ @Override
+ public void setLocation(String location, Job job) throws IOException {
+ final Configuration configuration = job.getConfiguration();
+ //explicitly turning off combining splits.
+ configuration.setBoolean("pig.noSplitCombination", true);
+ this.initializePhoenixPigConfiguration(location, configuration);
+ }
+
+ /**
+ * Initialize PhoenixPigConfiguration if it is null. Called by {@link #setLocation} and {@link #getSchema}
+ * @param location
+ * @param configuration
+ * @throws PigException
+ */
+ private void initializePhoenixPigConfiguration(final String location, final Configuration configuration) throws PigException {
+ if(this.config != null) {
+ return;
+ }
+ this.config = new PhoenixPigConfiguration(configuration);
+ this.config.setServerName(this.zkQuorum);
+ Pair<String,String> pair = null;
+ try {
+ if (location.startsWith(PHOENIX_TABLE_NAME_SCHEME)) {
+ String tableSchema = location.substring(PHOENIX_TABLE_NAME_SCHEME.length());
+ final TableSchemaParserFunction parseFunction = new TableSchemaParserFunction();
+ pair = parseFunction.apply(tableSchema);
+ } else if (location.startsWith(PHOENIX_QUERY_SCHEME)) {
+ this.selectQuery = location.substring(PHOENIX_QUERY_SCHEME.length());
+ final QuerySchemaParserFunction queryParseFunction = new QuerySchemaParserFunction(this.config);
+ pair = queryParseFunction.apply(this.selectQuery);
+ config.setSelectStatement(this.selectQuery);
+ }
+ this.tableName = pair.getFirst();
+ final String selectedColumns = pair.getSecond();
+
+ if(isEmpty(this.tableName) && isEmpty(this.selectQuery)) {
+ printUsage(location);
+ }
+ this.config.setTableName(this.tableName);
+ if(!isEmpty(selectedColumns)) {
+ this.config.setSelectColumns(selectedColumns);
+ }
+ } catch(IllegalArgumentException iae) {
+ printUsage(location);
+ }
+ }
+
+
+ @Override
+ public String relativeToAbsolutePath(String location, Path curDir) throws IOException {
+ return location;
+ }
+
+ @Override
+ public InputFormat getInputFormat() throws IOException {
+ if(inputFormat == null) {
+ inputFormat = new PhoenixInputFormat();
+ }
+ return inputFormat;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void prepareToRead(RecordReader reader, PigSplit split) throws IOException {
+ this.reader = reader;
+ final String resourceSchemaAsStr = getValueFromUDFContext(this.contextSignature,RESOURCE_SCHEMA_SIGNATURE);
+ if (resourceSchemaAsStr == null) {
+ throw new IOException("Could not find schema in UDF context");
+ }
+ schema = (ResourceSchema)ObjectSerializer.deserialize(resourceSchemaAsStr);
+ }
+
+ /*
+ * @see org.apache.pig.LoadFunc#setUDFContextSignature(java.lang.String)
+ */
+ @Override
+ public void setUDFContextSignature(String signature) {
+ this.contextSignature = signature;
+ }
+
+ @Override
+ public Tuple getNext() throws IOException {
+ try {
+ if(!reader.nextKeyValue()) {
+ return null;
+ }
+ final PhoenixRecord phoenixRecord = reader.getCurrentValue();
+ if(phoenixRecord == null) {
+ return null;
+ }
+ final Tuple tuple = TypeUtil.transformToTuple(phoenixRecord,schema.getFields());
+ return tuple;
+ } catch (InterruptedException e) {
+ int errCode = 6018;
+ final String errMsg = "Error while reading input";
+ throw new ExecException(errMsg, errCode,PigException.REMOTE_ENVIRONMENT, e);
+ }
+ }
+
+ private void printUsage(final String location) throws PigException {
+ String locationErrMsg = String.format("The input location in load statement should be of the form " +
+ "%s<table name> or %s<query>. Got [%s] ",PHOENIX_TABLE_NAME_SCHEME,PHOENIX_QUERY_SCHEME,location);
+ LOG.error(locationErrMsg);
+ throw new PigException(locationErrMsg);
+ }
+
+ @Override
+ public ResourceSchema getSchema(String location, Job job) throws IOException {
+ if(schema != null) {
+ return schema;
+ }
+ final Configuration configuration = job.getConfiguration();
+ this.initializePhoenixPigConfiguration(location, configuration);
+ this.schema = PhoenixPigSchemaUtil.getResourceSchema(this.config);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug(String.format("Resource Schema generated for location [%s] is [%s]", location, schema.toString()));
+ }
+ this.storeInUDFContext(this.contextSignature, RESOURCE_SCHEMA_SIGNATURE, ObjectSerializer.serialize(schema));
+ return schema;
+ }
+
+ @Override
+ public ResourceStatistics getStatistics(String location, Job job) throws IOException {
+ // not implemented
+ return null;
+ }
+
+ @Override
+ public String[] getPartitionKeys(String location, Job job) throws IOException {
+ // not implemented
+ return null;
+ }
+
+ @Override
+ public void setPartitionFilter(Expression partitionFilter) throws IOException {
+ // not implemented
+ }
+
+ private void storeInUDFContext(final String signature,final String key,final String value) {
+ final UDFContext udfContext = UDFContext.getUDFContext();
+ final Properties props = udfContext.getUDFProperties(this.getClass(), new String[]{signature});
+ props.put(key, value);
+ }
+
+ private String getValueFromUDFContext(final String signature,final String key) {
+ final UDFContext udfContext = UDFContext.getUDFContext();
+ final Properties props = udfContext.getUDFProperties(this.getClass(), new String[]{signature});
+ return props.getProperty(key);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/a9b8eb9b/phoenix-pig/src/main/java/org/apache/phoenix/pig/PhoenixHBaseStorage.java
----------------------------------------------------------------------
diff --git a/phoenix-pig/src/main/java/org/apache/phoenix/pig/PhoenixHBaseStorage.java b/phoenix-pig/src/main/java/org/apache/phoenix/pig/PhoenixHBaseStorage.java
index 9e237f1..500e403 100644
--- a/phoenix-pig/src/main/java/org/apache/phoenix/pig/PhoenixHBaseStorage.java
+++ b/phoenix-pig/src/main/java/org/apache/phoenix/pig/PhoenixHBaseStorage.java
@@ -18,6 +18,8 @@
package org.apache.phoenix.pig;
import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
import java.util.Properties;
import org.apache.commons.cli.CommandLine;
@@ -31,6 +33,8 @@ import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.phoenix.pig.hadoop.PhoenixOutputFormat;
+import org.apache.phoenix.pig.hadoop.PhoenixRecord;
import org.apache.pig.ResourceSchema;
import org.apache.pig.ResourceSchema.ResourceFieldSchema;
import org.apache.pig.StoreFuncInterface;
@@ -38,9 +42,6 @@ import org.apache.pig.data.Tuple;
import org.apache.pig.impl.util.ObjectSerializer;
import org.apache.pig.impl.util.UDFContext;
-import org.apache.phoenix.pig.hadoop.PhoenixOutputFormat;
-import org.apache.phoenix.pig.hadoop.PhoenixRecord;
-
/**
* StoreFunc that uses Phoenix to store data into HBase.
*
@@ -52,11 +53,21 @@ import org.apache.phoenix.pig.hadoop.PhoenixRecord;
* argument to this StoreFunc is the server, the 2nd argument is the batch size
* for upserts via Phoenix.
*
+ * Alternative usage: A = load 'testdata' as (a:chararray, b:chararray,
+ * e: datetime); STORE A into 'hbase://CORE.ENTITY_HISTORY/ID,F.B,F.E' using
+ * org.apache.bdaas.PhoenixHBaseStorage('localhost','-batchSize 5000');
+ *
+ * The above reads a file 'testdata' and writes the elements ID, F.B, and F.E to HBase.
+ * In this example, ID is the row key, and F is the column family for the data elements.
+ * First argument to this StoreFunc is the server, the 2nd argument is the batch size
+ * for upserts via Phoenix. In this case, less than the full table row is required.
+ * For configuration message, look in the info log file.
+ *
* Note that Pig types must be in sync with the target Phoenix data types. This
* StoreFunc tries best to cast based on input Pig types and target Phoenix data
* types, but it is recommended to supply appropriate schema.
*
- * This is only a STORE implementation. LoadFunc coming soon.
+ *
*
*
*
@@ -65,7 +76,6 @@ import org.apache.phoenix.pig.hadoop.PhoenixRecord;
public class PhoenixHBaseStorage implements StoreFuncInterface {
private PhoenixPigConfiguration config;
- private String tableName;
private RecordWriter<NullWritable, PhoenixRecord> writer;
private String contextSignature = null;
private ResourceSchema schema;
@@ -118,17 +128,28 @@ public class PhoenixHBaseStorage implements StoreFuncInterface {
*/
@Override
public void setStoreLocation(String location, Job job) throws IOException {
- String prefix = "hbase://";
- if (location.startsWith(prefix)) {
- tableName = location.substring(prefix.length());
- }
- config = new PhoenixPigConfiguration(job.getConfiguration());
- config.configure(server, tableName, batchSize);
-
- String serializedSchema = getUDFProperties().getProperty(contextSignature + SCHEMA);
- if (serializedSchema != null) {
- schema = (ResourceSchema) ObjectSerializer.deserialize(serializedSchema);
- }
+ URI locationURI;
+ try {
+ locationURI = new URI(location);
+ if (!"hbase".equals(locationURI.getScheme())) {
+ throw new IOException(String.format("Location must use the hbase protocol, hbase://tableName[/columnList]. Supplied location=%s",location));
+ }
+ String tableName = locationURI.getAuthority();
+ // strip off the leading path token '/'
+ String columns = null;
+ if(!locationURI.getPath().isEmpty()) {
+ columns = locationURI.getPath().substring(1);
+ }
+ config = new PhoenixPigConfiguration(job.getConfiguration());
+ config.configure(server, tableName, batchSize, columns);
+
+ String serializedSchema = getUDFProperties().getProperty(contextSignature + SCHEMA);
+ if (serializedSchema != null) {
+ schema = (ResourceSchema) ObjectSerializer.deserialize(serializedSchema);
+ }
+ } catch (URISyntaxException e) {
+ throw new IOException(String.format("Location must use the hbase protocol, hbase://tableName[/columnList]. Supplied location=%s",location),e);
+ }
}
@SuppressWarnings("unchecked")