You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hawq.apache.org by nh...@apache.org on 2015/10/13 00:43:22 UTC
[1/4] incubator-hawq git commit: HAWQ-28. JavaDoc fixes for PXF
Repository: incubator-hawq
Updated Branches:
refs/heads/master 445e075ee -> 88f5f5f63
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/ProtocolData.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/ProtocolData.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/ProtocolData.java
index 7ca3f05..5001c40 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/ProtocolData.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/ProtocolData.java
@@ -16,8 +16,8 @@ import com.pivotal.pxf.api.utilities.InputData;
import com.pivotal.pxf.api.utilities.ProfilesConf;
/**
- * Common configuration of all MetaData classes.
- * Provides read-only access to common parameters supplied using system properties.
+ * Common configuration of all MetaData classes. Provides read-only access to
+ * common parameters supplied using system properties.
*/
public class ProtocolData extends InputData {
@@ -35,8 +35,7 @@ public class ProtocolData extends InputData {
protected String token;
/**
- * Constructs a ProtocolData.
- * Parses X-GP-* configuration variables.
+ * Constructs a ProtocolData. Parses X-GP-* configuration variables.
*
* @param paramsMap contains all query-specific parameters from Hawq
*/
@@ -61,10 +60,10 @@ public class ProtocolData extends InputData {
parseTupleDescription();
/*
- * accessor - will throw exception from getPropery() if outputFormat is BINARY
- * and the user did not supply accessor=... or profile=...
- * resolver - will throw exception from getPropery() if outputFormat is BINARY
- * and the user did not supply resolver=... or profile=...
+ * accessor - will throw exception from getPropery() if outputFormat is
+ * BINARY and the user did not supply accessor=... or profile=...
+ * resolver - will throw exception from getPropery() if outputFormat is
+ * BINARY and the user did not supply resolver=... or profile=...
*/
profile = getOptionalProperty("PROFILE");
if (profile != null) {
@@ -94,8 +93,8 @@ public class ProtocolData extends InputData {
}
/**
- * Constructs an InputDataBuilder from a copy.
- * Used to create from an extending class.
+ * Constructs an InputDataBuilder from a copy. Used to create from an
+ * extending class.
*
* @param copy the input data to copy
*/
@@ -128,7 +127,8 @@ public class ProtocolData extends InputData {
}
/**
- * Sets the requested profile plugins from profile file into {@link #requestParametersMap}.
+ * Sets the requested profile plugins from profile file into
+ * {@link #requestParametersMap}.
*/
private void setProfilePlugins() {
Map<String, String> pluginsMap = ProfilesConf.getProfilePluginsMap(profile);
@@ -137,12 +137,13 @@ public class ProtocolData extends InputData {
}
/**
- * Verifies there are no duplicates between parameters declared in the table definition
- * and parameters defined in a profile.
+ * Verifies there are no duplicates between parameters declared in the table
+ * definition and parameters defined in a profile.
*
* The parameters' names are case insensitive.
*/
- private void checkForDuplicates(Map<String, String> plugins, Map<String, String> params) {
+ private void checkForDuplicates(Map<String, String> plugins,
+ Map<String, String> params) {
List<String> duplicates = new ArrayList<>();
for (String key : plugins.keySet()) {
if (params.containsKey(key)) {
@@ -151,17 +152,29 @@ public class ProtocolData extends InputData {
}
if (!duplicates.isEmpty()) {
- throw new IllegalArgumentException("Profile '" + profile + "' already defines: " + String.valueOf(duplicates).replace("X-GP-", ""));
+ throw new IllegalArgumentException("Profile '" + profile
+ + "' already defines: "
+ + String.valueOf(duplicates).replace("X-GP-", ""));
}
}
- /** Returns the request parameters */
+ /**
+ * Returns the request parameters.
+ *
+ * @return map of request parameters
+ */
public Map<String, String> getParametersMap() {
return requestParametersMap;
}
- public void protocolViolation(String property)
- {
+ /**
+ * Throws an exception when the given property value is missing in request.
+ *
+ * @param property missing property name
+ * @throws IllegalArgumentException throws an exception with the property
+ * name in the error message
+ */
+ public void protocolViolation(String property) {
String error = "Internal server error. Property \"" + property
+ "\" has no value in current request";
@@ -170,7 +183,8 @@ public class ProtocolData extends InputData {
}
/**
- * Returns the value to which the specified property is mapped in {@link #requestParametersMap}.
+ * Returns the value to which the specified property is mapped in
+ * {@link #requestParametersMap}.
*
* @param property the lookup property key
* @throws IllegalArgumentException if property key is missing
@@ -186,8 +200,8 @@ public class ProtocolData extends InputData {
}
/**
- * Returns the optional property value.
- * Unlike {@link #getProperty}, it will not fail if the property is not found. It will just return null instead.
+ * Returns the optional property value. Unlike {@link #getProperty}, it will
+ * not fail if the property is not found. It will just return null instead.
*
* @param property the lookup optional property
* @return property value as a String
@@ -201,44 +215,56 @@ public class ProtocolData extends InputData {
*
* @param property the lookup property
* @return property value as an int type
- * @throws NumberFormatException
- * if the value is missing or can't be represented by an Integer
+ * @throws NumberFormatException if the value is missing or can't be
+ * represented by an Integer
*/
private int getIntProperty(String property) {
return Integer.parseInt(getProperty(property));
}
/**
- * Returns a property value as boolean type.
- * A boolean property is defined as an int where 0 means false, and anything else true (like C).
+ * Returns a property value as boolean type. A boolean property is defined
+ * as an int where 0 means false, and anything else true (like C).
*
* @param property the lookup property
* @return property value as boolean
- * @throws NumberFormatException
- * if the value is missing or can't be represented by an Integer
+ * @throws NumberFormatException if the value is missing or can't be
+ * represented by an Integer
*/
private boolean getBoolProperty(String property) {
return getIntProperty(property) != 0;
}
- /** Returns the current outputFormat, either {@link OutputFormat#TEXT} or {@link OutputFormat#BINARY}. */
+ /**
+ * Returns the current output format, either {@link OutputFormat#TEXT} or
+ * {@link OutputFormat#BINARY}.
+ *
+ * @return output format
+ */
public OutputFormat outputFormat() {
return outputFormat;
}
- /** Returns the server name providing the service. */
+ /**
+ * Returns the server name providing the service.
+ *
+ * @return server name
+ */
public String serverName() {
return host;
}
- /** Returns the server port providing the service. */
+ /**
+ * Returns the server port providing the service.
+ *
+ * @return server port
+ */
public int serverPort() {
return port;
}
/**
- * Sets the thread safe parameter.
- * Default value - true.
+ * Sets the thread safe parameter. Default value - true.
*/
private void parseThreadSafe() {
@@ -257,24 +283,27 @@ public class ProtocolData extends InputData {
if (threadSafeStr.equalsIgnoreCase(FALSE_LCASE)) {
return false;
}
- throw new IllegalArgumentException("Illegal boolean value '" + threadSafeStr + "'." +
- " Usage: [TRUE|FALSE]");
+ throw new IllegalArgumentException("Illegal boolean value '"
+ + threadSafeStr + "'." + " Usage: [TRUE|FALSE]");
}
/**
- * Sets the format type based on the input string.
- * Allowed values are: {@link OutputFormat#TEXT}, {@link OutputFormat#BINARY}.
+ * Sets the format type based on the input string. Allowed values are:
+ * "TEXT", "GPDBWritable".
+ *
+ * @param formatString format string
*/
protected void parseFormat(String formatString) {
switch (formatString) {
- case "TEXT":
- outputFormat = OutputFormat.TEXT;
- break;
- case "GPDBWritable":
- outputFormat = OutputFormat.BINARY;
- break;
- default:
- throw new IllegalArgumentException("Wrong value for greenplum.format " + formatString);
+ case "TEXT":
+ outputFormat = OutputFormat.TEXT;
+ break;
+ case "GPDBWritable":
+ outputFormat = OutputFormat.BINARY;
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Wrong value for greenplum.format " + formatString);
}
}
@@ -288,7 +317,8 @@ public class ProtocolData extends InputData {
int columnTypeCode = getIntProperty("ATTR-TYPECODE" + i);
String columnTypeName = getProperty("ATTR-TYPENAME" + i);
- ColumnDescriptor column = new ColumnDescriptor(columnName, columnTypeCode, i, columnTypeName);
+ ColumnDescriptor column = new ColumnDescriptor(columnName,
+ columnTypeCode, i, columnTypeName);
tupleDescription.add(column);
if (columnName.equalsIgnoreCase(ColumnDescriptor.RECORD_KEY_NAME)) {
@@ -299,6 +329,7 @@ public class ProtocolData extends InputData {
/**
* Sets the index of the allocated data fragment
+ *
* @param fragment the allocated data fragment
*/
protected void parseDataFragment(String fragment) {
@@ -313,7 +344,8 @@ public class ProtocolData extends InputData {
}
private void parseFragmentMetadata() {
- fragmentMetadata = parseBase64("FRAGMENT-METADATA", "Fragment metadata information");
+ fragmentMetadata = parseBase64("FRAGMENT-METADATA",
+ "Fragment metadata information");
}
private void parseUserData() {
@@ -326,8 +358,9 @@ public class ProtocolData extends InputData {
return null;
}
if (!Base64.isArrayByteBase64(encoded.getBytes())) {
- throw new IllegalArgumentException(errName + " must be Base64 encoded." +
- "(Bad value: " + encoded + ")");
+ throw new IllegalArgumentException(errName
+ + " must be Base64 encoded." + "(Bad value: " + encoded
+ + ")");
}
byte[] parsed = Base64.decodeBase64(encoded);
LOG.debug("decoded " + key + ": " + new String(parsed));
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/Utilities.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/Utilities.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/Utilities.java
index afcbc0e..220e4b2 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/Utilities.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/Utilities.java
@@ -13,64 +13,73 @@ import com.pivotal.pxf.api.utilities.InputData;
*/
public class Utilities {
private static final Log LOG = LogFactory.getLog(Utilities.class);
-
+
/**
- * Creates an object using the class name.
- * The class name has to be a class located in the webapp's CLASSPATH.
- *
- * @param confClass the class of the metaData used to initialize the instance
+ * Creates an object using the class name. The class name has to be a class
+ * located in the webapp's CLASSPATH.
+ *
+ * @param confClass the class of the metaData used to initialize the
+ * instance
* @param className a class name to be initialized.
* @param metaData input data used to initialize the class
* @return Initialized instance of given className
- * @throws Exception
+ * @throws Exception throws exception if classname was not found in
+ * classpath, didn't have expected constructor or failed to be
+ * instantiated
*/
- public static Object createAnyInstance(Class<?> confClass, String className, InputData metaData) throws Exception {
+ public static Object createAnyInstance(Class<?> confClass,
+ String className, InputData metaData)
+ throws Exception {
Class<?> cls = Class.forName(className);
Constructor<?> con = cls.getConstructor(confClass);
return instantiate(con, metaData);
}
-
+
/**
- * Creates an object using the class name with its default constructor.
- * The class name has to be a class located in the webapp's CLASSPATH.
- *
+ * Creates an object using the class name with its default constructor. The
+ * class name has to be a class located in the webapp's CLASSPATH.
+ *
* @param className a class name to be initialized
* @return initialized instance of given className
- * @throws Exception
+ * @throws Exception throws exception if classname was not found in
+ * classpath, didn't have expected constructor or failed to be
+ * instantiated
*/
public static Object createAnyInstance(String className) throws Exception {
- Class<?> cls = Class.forName(className);
+ Class<?> cls = Class.forName(className);
Constructor<?> con = cls.getConstructor();
return instantiate(con);
}
-
- private static Object instantiate(Constructor<?> con, Object... args) throws Exception {
- try {
- return con.newInstance(args);
- } catch (InvocationTargetException e) {
- /*
- * We are creating resolvers, accessors, fragmenters, etc.
- * using the reflection framework. If for example, a resolver, during
- * its instantiation - in the c'tor, will throw an exception, the
- * Resolver's exception will reach the Reflection layer and there it
- * will be wrapped inside an InvocationTargetException. Here we are
- * above the Reflection layer and we need to unwrap the Resolver's
- * initial exception and throw it instead of the wrapper
- * InvocationTargetException so that our initial Exception text will
- * be displayed in psql instead of the message:
- * "Internal Server Error"
- */
- throw (e.getCause() != null) ? new Exception(e.getCause()) : e;
- }
+
+ private static Object instantiate(Constructor<?> con, Object... args)
+ throws Exception {
+ try {
+ return con.newInstance(args);
+ } catch (InvocationTargetException e) {
+ /*
+ * We are creating resolvers, accessors, fragmenters, etc. using the
+ * reflection framework. If for example, a resolver, during its
+ * instantiation - in the c'tor, will throw an exception, the
+ * Resolver's exception will reach the Reflection layer and there it
+ * will be wrapped inside an InvocationTargetException. Here we are
+ * above the Reflection layer and we need to unwrap the Resolver's
+ * initial exception and throw it instead of the wrapper
+ * InvocationTargetException so that our initial Exception text will
+ * be displayed in psql instead of the message:
+ * "Internal Server Error"
+ */
+ throw (e.getCause() != null) ? new Exception(e.getCause()) : e;
+ }
}
/**
- * Transforms a byte array into a string of octal codes in the form \\xyz\\xyz
+ * Transforms a byte array into a string of octal codes in the form
+ * \\xyz\\xyz
*
- * We double escape each char because it is required in postgres bytea for some bytes.
- * In the minimum all non-printables, backslash, null and single quote.
- * Easier to just escape everything
- * see http://www.postgresql.org/docs/9.0/static/datatype-binary.html
+ * We double escape each char because it is required in postgres bytea for
+ * some bytes. In the minimum all non-printables, backslash, null and single
+ * quote. Easier to just escape everything see
+ * http://www.postgresql.org/docs/9.0/static/datatype-binary.html
*
* Octal codes must be padded to 3 characters (001, 012)
*
@@ -82,7 +91,8 @@ public class Utilities {
return;
}
- sb.ensureCapacity(sb.length() + (bytes.length * 5 /* characters per byte */));
+ sb.ensureCapacity(sb.length()
+ + (bytes.length * 5 /* characters per byte */));
for (int b : bytes) {
sb.append(String.format("\\\\%03o", b & 0xff));
}
[3/4] incubator-hawq git commit: HAWQ-28. JavaDoc fixes for PXF
Posted by nh...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkRecordReader.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkRecordReader.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkRecordReader.java
index 48fb5b5..13a3546 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkRecordReader.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkRecordReader.java
@@ -1,22 +1,20 @@
package com.pivotal.pxf.plugins.hdfs;
-import com.pivotal.pxf.plugins.hdfs.ChunkWritable;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_CLIENT_READ_SHORTCIRCUIT_SKIP_CHECKSUM_KEY;
+import static org.apache.hadoop.mapreduce.lib.input.LineRecordReader.MAX_LINE_LENGTH;
import java.io.IOException;
-import java.io.InputStream;
-import java.lang.IllegalArgumentException;
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.Seekable;
-import org.apache.hadoop.hdfs.DFSInputStream.ReadStatistics;
import org.apache.hadoop.hdfs.DFSInputStream;
+import org.apache.hadoop.hdfs.DFSInputStream.ReadStatistics;
import org.apache.hadoop.io.LongWritable;
-import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.compress.CodecPool;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
@@ -25,241 +23,259 @@ import org.apache.hadoop.io.compress.SplitCompressionInputStream;
import org.apache.hadoop.io.compress.SplittableCompressionCodec;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.RecordReader;
-import org.apache.commons.logging.LogFactory;
-import org.apache.commons.logging.Log;
-
-import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_CLIENT_READ_SHORTCIRCUIT_SKIP_CHECKSUM_KEY;
-import static org.apache.hadoop.mapreduce.lib.input.LineRecordReader.MAX_LINE_LENGTH;
-
/**
- * ChunkRecordReader is designed for fast reading of a file split. The idea is to bring chunks of
- * data instead of single records. The chunks contain many records and the chunk end is not aligned
- * on a record boundary. The size of the chunk is a class hardcoded parameter - CHUNK_SIZE.
- * This behaviour sets this reader apart from the other readers which will fetch one record and
- * stop when reaching a record delimiter.
+ * ChunkRecordReader is designed for fast reading of a file split. The idea is
+ * to bring chunks of data instead of single records. The chunks contain many
+ * records and the chunk end is not aligned on a record boundary. The size of
+ * the chunk is a class hardcoded parameter - CHUNK_SIZE. This behaviour sets
+ * this reader apart from the other readers which will fetch one record and stop
+ * when reaching a record delimiter.
*/
-public class ChunkRecordReader implements RecordReader<LongWritable, ChunkWritable> {
- private static final Log LOG
- = LogFactory.getLog(ChunkRecordReader.class.getName());
+public class ChunkRecordReader implements
+ RecordReader<LongWritable, ChunkWritable> {
+ private static final Log LOG = LogFactory.getLog(ChunkRecordReader.class.getName());
+
+ private CompressionCodecFactory compressionCodecs = null;
+ private long start;
+ private long pos;
+ private long end;
+ private long fileLength;
+ private ChunkReader in;
+ private FSDataInputStream fileIn;
+ private final Seekable filePosition;
+ private int maxLineLength;
+ private CompressionCodec codec;
+ private Decompressor decompressor;
+ private static final int CHUNK_SIZE = 1024 * 1024;
+
+ /**
+ * Translates the FSDataInputStream into a DFSInputStream.
+ */
+ private DFSInputStream getInputStream() {
+ return (DFSInputStream) (fileIn.getWrappedStream());
+ }
+
+ /**
+ * Returns statistics of the input stream's read operation: total bytes
+ * read, bytes read locally, bytes read in short-circuit (directly from file
+ * descriptor).
+ *
+ * @return an instance of ReadStatistics class
+ */
+ public ReadStatistics getReadStatistics() {
+ return getInputStream().getReadStatistics();
+ }
+
+ /**
+ * Constructs a ChunkRecordReader instance.
+ *
+ * @param job the job configuration
+ * @param split contains the file name, begin byte of the split and the
+ * bytes length
+ * @throws IOException if an I/O error occurs when accessing the file or
+ * creating input stream to read from it
+ */
+ public ChunkRecordReader(Configuration job, FileSplit split)
+ throws IOException {
+ maxLineLength = job.getInt(MAX_LINE_LENGTH, Integer.MAX_VALUE);
+ validateLength(maxLineLength);
+ start = split.getStart();
+ end = start + split.getLength();
+ final Path file = split.getPath();
+ compressionCodecs = new CompressionCodecFactory(job);
+ codec = compressionCodecs.getCodec(file);
+
+ // open the file and seek to the start of the split
+ job.setBoolean(DFS_CLIENT_READ_SHORTCIRCUIT_SKIP_CHECKSUM_KEY, true);
+ final FileSystem fs = file.getFileSystem(job);
+ fs.setVerifyChecksum(false);
+ fileIn = fs.open(file, ChunkReader.DEFAULT_BUFFER_SIZE);
+ fileLength = getInputStream().getFileLength();
+ if (isCompressedInput()) {
+ decompressor = CodecPool.getDecompressor(codec);
+ if (codec instanceof SplittableCompressionCodec) {
+ final SplitCompressionInputStream cIn = ((SplittableCompressionCodec) codec).createInputStream(
+ fileIn, decompressor, start, end,
+ SplittableCompressionCodec.READ_MODE.BYBLOCK);
+ in = new ChunkReader(cIn);
+ start = cIn.getAdjustedStart();
+ end = cIn.getAdjustedEnd();
+ filePosition = cIn; // take pos from compressed stream
+ } else {
+ in = new ChunkReader(codec.createInputStream(fileIn,
+ decompressor));
+ filePosition = fileIn;
+ }
+ } else {
+ fileIn.seek(start);
+ in = new ChunkReader(fileIn);
+ filePosition = fileIn;
+ }
+ /*
+ * If this is not the first split, we always throw away first record
+ * because we always (except the last split) read one extra line in
+ * next() method.
+ */
+ if (start != 0) {
+ start += in.readLine(new ChunkWritable(), maxBytesToConsume(start));
+ }
+ this.pos = start;
+ }
+
+ /**
+ * Used by the client of this class to create the 'key' output parameter for
+ * next() method.
+ *
+ * @return an instance of LongWritable
+ */
+ @Override
+ public LongWritable createKey() {
+ return new LongWritable();
+ }
+
+ /**
+ * Used by the client of this class to create the 'value' output parameter
+ * for next() method.
+ *
+ * @return an instance of ChunkWritable
+ */
+ @Override
+ public ChunkWritable createValue() {
+ return new ChunkWritable();
+ }
+
+ /**
+ * Fetches the next data chunk from the file split. The size of the chunk is
+ * a class hardcoded parameter - CHUNK_SIZE. This behaviour sets this reader
+ * apart from the other readers which will fetch one record and stop when
+ * reaching a record delimiter.
+ *
+ * @param key - output parameter. When method returns will contain the key -
+ * the number of the start byte of the chunk
+ * @param value - output parameter. When method returns will contain the
+ * value - the chunk, a byte array inside the ChunkWritable
+ * instance
+ * @return false - when end of split was reached
+ * @throws IOException if an I/O error occurred while reading the next chunk
+ * or line
+ */
+ @Override
+ public synchronized boolean next(LongWritable key, ChunkWritable value)
+ throws IOException {
+ /*
+ * Usually a record is spread between the end of current split and the
+ * beginning of next split. So when reading the last record in the split
+ * we usually need to cross over to the next split. This tricky logic is
+ * implemented in ChunkReader.readLine(). In order not to rewrite this
+ * logic we will read the lust chunk in the split with readLine(). For a
+ * split of 120M, reading the last 1M line by line doesn't have a huge
+ * impact. Applying a factor to the last chunk to make sure we start
+ * before the last record.
+ */
+ float factor = 1.5f;
+ int limit = (int) (factor * CHUNK_SIZE);
+ long curPos = getFilePosition();
+ int newSize = 0;
+
+ while (curPos <= end) {
+ key.set(pos);
+
+ if ((end - curPos) > limit) {
+ newSize = in.readChunk(value, CHUNK_SIZE);
+ } else {
+ newSize = in.readLine(value,
+ Math.max(maxBytesToConsume(pos), maxLineLength));
+ }
+ if (newSize == 0) {
+ break;
+ }
+
+ pos += newSize;
+
+ if (pos == fileLength) { /*
+ * in case text file last character is not
+ * a linefeed
+ */
+ if (value.box[value.box.length - 1] != '\n') {
+ int newLen = value.box.length + 1;
+ byte[] tmp = new byte[newLen];
+ System.arraycopy(value.box, 0, tmp, 0, newLen - 1);
+ tmp[newLen - 1] = '\n';
+ value.box = tmp;
+ }
+ }
+
+ return true;
+ }
+ /*
+ * if we got here, either newSize was 0 or curPos is bigger than end
+ */
- private CompressionCodecFactory compressionCodecs = null;
- private long start;
- private long pos;
- private long end;
- private long fileLength;
- private ChunkReader in;
- private FSDataInputStream fileIn;
- private final Seekable filePosition;
- private int maxLineLength;
- private CompressionCodec codec;
- private Decompressor decompressor;
- private static final int CHUNK_SIZE = 1024 * 1024;
+ return false;
+ }
- /*
- * Translate the FSDataInputStream into a DFSInputStream
- */
- private DFSInputStream getInputStream() {
- return (DFSInputStream)(fileIn.getWrappedStream());
- }
-
- /**
- * Returns statistics of the input stream's read operation: total bytes read,
- * bytes read localy, bytes read in short-circuit (directly from file descriptor)
- * @return an instance of ReadStatistics class
- */
- public ReadStatistics getReadStatistics() {
- return getInputStream().getReadStatistics();
- }
+ /**
+ * Gets the progress within the split.
+ */
+ @Override
+ public synchronized float getProgress() throws IOException {
+ if (start == end) {
+ return 0.0f;
+ } else {
+ return Math.min(1.0f, (getFilePosition() - start)
+ / (float) (end - start));
+ }
+ }
- /**
- * Constructs a ChunkRecordReader instance
- * @param job the job configuration
- * @param split contains the file name, begin byte of the split and the bytes length
- */
- public ChunkRecordReader(Configuration job, FileSplit split) throws IOException {
- maxLineLength = job.getInt(MAX_LINE_LENGTH, Integer.MAX_VALUE);
- validateLength(maxLineLength);
- start = split.getStart();
- end = start + split.getLength();
- final Path file = split.getPath();
- compressionCodecs = new CompressionCodecFactory(job);
- codec = compressionCodecs.getCodec(file);
+ /**
+ * Returns the position of the unread tail of the file
+ *
+ * @return pos - start byte of the unread tail of the file
+ */
+ @Override
+ public synchronized long getPos() throws IOException {
+ return pos;
+ }
- // open the file and seek to the start of the split
- job.setBoolean(DFS_CLIENT_READ_SHORTCIRCUIT_SKIP_CHECKSUM_KEY, true);
- final FileSystem fs = file.getFileSystem(job);
- fs.setVerifyChecksum(false);
- fileIn = fs.open(file, ChunkReader.DEFAULT_BUFFER_SIZE);
- fileLength = getInputStream().getFileLength();
- if (isCompressedInput()) {
- decompressor = CodecPool.getDecompressor(codec);
- if (codec instanceof SplittableCompressionCodec) {
- final SplitCompressionInputStream cIn =
- ((SplittableCompressionCodec)codec).createInputStream(
- fileIn, decompressor, start, end,
- SplittableCompressionCodec.READ_MODE.BYBLOCK);
- in = new ChunkReader(cIn);
- start = cIn.getAdjustedStart();
- end = cIn.getAdjustedEnd();
- filePosition = cIn; // take pos from compressed stream
- } else {
- in = new ChunkReader(codec.createInputStream(fileIn, decompressor));
- filePosition = fileIn;
- }
- } else {
- fileIn.seek(start);
- in = new ChunkReader(fileIn);
- filePosition = fileIn;
- }
- /*
- * If this is not the first split, we always throw away first record
- * because we always (except the last split) read one extra line in
- * next() method.
- */
- if (start != 0) {
- start += in.readLine(new ChunkWritable(), maxBytesToConsume(start));
- }
- this.pos = start;
- }
-
- /**
- * Used by the client of this class to create the 'key' output parameter for next() method
- * @return an instance of LongWritable
- */
- @Override
- public LongWritable createKey() {
- return new LongWritable();
- }
-
- /**
- * Used by the client of this class to create the 'value' output parameter for next() method
- * @return an instance of ChunkWritable
- */
- @Override
- public ChunkWritable createValue() {
- return new ChunkWritable();
- }
-
- /**
- * Fetch the next data chunk from the file split. The size of the chunk is a class hardcoded
- * parameter - CHUNK_SIZE. This behaviour sets this reader apart from the other readers which
- * will fetch one record and stop when reaching a record delimiter.
- * @param key - output parameter. When method returns will contain the key - the number
- * of the start byte of the chunk
- * @param value - output parameter. When method returns will contain the value - the chunk,
- * a byte array inside the ChunkWritable instance
- * @return false - when end of split was reached
- */
- @Override
- public synchronized boolean next(LongWritable key, ChunkWritable value)
- throws IOException {
- /*
- * Usualy a record is spread between the end of current split and the beginning
- * of next split. So when reading the last record in the split we usually need to
- * cross over to the next split. This tricky logic is implemented in
- * ChunkReader.readLine().
- * In order not to rewrite this logic we will read the lust chunk in the split
- * with readLine(). For a split of 120M, reading the last 1M line by line doesn't
- * have a huge impact. Applying a factor to the last chunk to make sure we start
- * before the last record.
- */
- float factor = 1.5f;
- int limit = (int)(factor*CHUNK_SIZE);
- long curPos = getFilePosition();
- int newSize = 0;
-
- while (curPos <= end) {
- key.set(pos);
-
- if ((end - curPos) > limit) {
- newSize = in.readChunk(value, CHUNK_SIZE);
- }
- else {
- newSize = in.readLine(value, Math.max(maxBytesToConsume(pos), maxLineLength));
- }
- if (newSize == 0) {
- break;
- }
-
- pos += newSize;
-
- if (pos == fileLength) { /* in case text file last character is not a linefeed*/
- if (value.box[value.box.length - 1] != (int)'\n') {
- int newLen = value.box.length + 1;
- byte [] tmp = new byte [newLen];
- System.arraycopy(value.box, 0, tmp, 0, newLen - 1);
- tmp[newLen - 1] = '\n';
- value.box = tmp;
- }
- }
-
- return true;
- }
- /*
- * if we got here, either newSize was 0 or curPos is bigger than end
- */
-
- return false;
- }
-
- /**
- * Get the progress within the split
- */
- @Override
- public synchronized float getProgress() throws IOException {
- if (start == end) {
- return 0.0f;
- } else {
- return Math.min(1.0f, (getFilePosition() - start) / (float)(end - start));
- }
- }
-
- /**
- * @return pos - start byte of the unread tail of the file
- */
- public synchronized long getPos() throws IOException {
- return pos;
- }
-
- /**
- * Close the input stream
- */
- @Override
- public synchronized void close() throws IOException {
- try {
- if (in != null) {
- in.close();
- }
- } finally {
- if (decompressor != null) {
- CodecPool.returnDecompressor(decompressor);
- }
- }
- }
+ /**
+ * Closes the input stream.
+ */
+ @Override
+ public synchronized void close() throws IOException {
+ try {
+ if (in != null) {
+ in.close();
+ }
+ } finally {
+ if (decompressor != null) {
+ CodecPool.returnDecompressor(decompressor);
+ }
+ }
+ }
- private void validateLength(int maxLineLength) {
- if (maxLineLength <= 0)
- throw new IllegalArgumentException("maxLineLength must be a positive value");
- }
+ private void validateLength(int maxLineLength) {
+ if (maxLineLength <= 0)
+ throw new IllegalArgumentException(
+ "maxLineLength must be a positive value");
+ }
- private boolean isCompressedInput() {
- return (codec != null);
- }
+ private boolean isCompressedInput() {
+ return (codec != null);
+ }
- private int maxBytesToConsume(long pos) {
- return isCompressedInput()
- ? Integer.MAX_VALUE
- : (int) Math.min(Integer.MAX_VALUE, end - pos);
- }
+ private int maxBytesToConsume(long pos) {
+ return isCompressedInput() ? Integer.MAX_VALUE : (int) Math.min(
+ Integer.MAX_VALUE, end - pos);
+ }
- private long getFilePosition() throws IOException {
- long retVal;
- if (isCompressedInput() && null != filePosition) {
- retVal = filePosition.getPos();
- } else {
- retVal = pos;
- }
- return retVal;
- }
+ private long getFilePosition() throws IOException {
+ long retVal;
+ if (isCompressedInput() && null != filePosition) {
+ retVal = filePosition.getPos();
+ } else {
+ retVal = pos;
+ }
+ return retVal;
+ }
} // class ChunkRecordReader
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkWritable.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkWritable.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkWritable.java
index b19b3fd..a0a8b17 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkWritable.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkWritable.java
@@ -2,7 +2,6 @@ package com.pivotal.pxf.plugins.hdfs;
import java.io.DataOutput;
import java.io.DataInput;
-import java.io.IOException;
import java.lang.UnsupportedOperationException;
import org.apache.hadoop.io.Writable;
@@ -13,29 +12,28 @@ import org.apache.hadoop.io.Writable;
*/
public class ChunkWritable implements Writable {
public byte [] box;
-
+
/**
- * Serialize the fields of this object to <code>out</code>.
+ * Serializes the fields of this object to <code>out</code>.
*
* @param out <code>DataOutput</code> to serialize this object into.
- * @throws IOException
+ * @throws UnsupportedOperationException this function is not supported
*/
@Override
public void write(DataOutput out) {
throw new UnsupportedOperationException("ChunkWritable.write() is not implemented");
}
-
+
/**
- * Deserialize the fields of this object from <code>in</code>.
+ * Deserializes the fields of this object from <code>in</code>.
* <p>For efficiency, implementations should attempt to re-use storage in the
* existing object where possible.</p>
*
* @param in <code>DataInput</code> to deserialize this object from.
- * @throws IOException
+ * @throws UnsupportedOperationException this function is not supported
*/
@Override
public void readFields(DataInput in) {
throw new UnsupportedOperationException("ChunkWritable.readFields() is not implemented");
}
-
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsAnalyzer.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsAnalyzer.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsAnalyzer.java
index f18ecef..0f8f908 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsAnalyzer.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsAnalyzer.java
@@ -20,7 +20,6 @@ import org.apache.hadoop.mapred.JobConf;
import java.io.IOException;
import java.util.ArrayList;
-
/**
* Analyzer class for HDFS data resources
*
@@ -33,10 +32,10 @@ public class HdfsAnalyzer extends Analyzer {
private Log Log;
/**
- * Constructs an HdfsAnalyzer object
- *
+ * Constructs an HdfsAnalyzer object.
+ *
* @param inputData all input parameters coming from the client
- * @throws IOException
+ * @throws IOException if HDFS file system cannot be retrieved
*/
public HdfsAnalyzer(InputData inputData) throws IOException {
super(inputData);
@@ -49,11 +48,13 @@ public class HdfsAnalyzer extends Analyzer {
/**
* Collects a number of basic statistics based on an estimate. Statistics
* are: number of records, number of hdfs blocks and hdfs block size.
- *
- * @param datapath path is a data source URI that can appear as a file
+ *
+ * @param datapath path is a data source URI that can appear as a file
* name, a directory name or a wildcard pattern
- * @return statistics in json format
- * @throws Exception
+ * @return statistics in JSON format
+ * @throws Exception if path is wrong, its metadata cannot be retrieved
+ * from file system, or if scanning the first block
+ * using the accessor failed
*/
@Override
public AnalyzerStats getEstimatedStats(String datapath) throws Exception {
@@ -89,8 +90,8 @@ public class HdfsAnalyzer extends Analyzer {
return stats;
}
- /*
- * Calculate the number of tuples in a split (block)
+ /**
+ * Calculates the number of tuples in a split (block).
* Reads one block from HDFS. Exception during reading will
* filter upwards and handled in AnalyzerResource
*/
@@ -129,7 +130,7 @@ public class HdfsAnalyzer extends Analyzer {
PxfInputFormat.setInputPaths(jobConf, path);
InputSplit[] splits = fformat.getSplits(jobConf, 1);
ArrayList<InputSplit> result = new ArrayList<InputSplit>();
-
+
// remove empty splits
if (splits != null) {
for (InputSplit split : splits) {
@@ -138,7 +139,7 @@ public class HdfsAnalyzer extends Analyzer {
}
}
}
-
- return result;
+
+ return result;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsAtomicDataAccessor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsAtomicDataAccessor.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsAtomicDataAccessor.java
index da02590..4a43c5f 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsAtomicDataAccessor.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsAtomicDataAccessor.java
@@ -15,19 +15,18 @@ import java.io.InputStream;
import java.net.URI;
/**
- * Base class for enforcing the complete access of a file in one accessor.
+ * Base class for enforcing the complete access of a file in one accessor.
* Since we are not accessing the file using the splittable API, but instead are
* using the "simple" stream API, it means that we cannot fetch different parts
* (splits) of the file in different segments. Instead each file access brings
- * the complete file. And, if several segments would access the same file, then
+ * the complete file. And, if several segments would access the same file, then
* each one will return the whole file and we will observe in the query result,
- * each record appearing number_of_segments times. To avoid this we will only
- * have one segment (segment 0) working for this case - enforced with
- * isWorkingSegment() method. Naturally this is the less recommended working
+ * each record appearing number_of_segments times. To avoid this we will only
+ * have one segment (segment 0) working for this case - enforced with
+ * isWorkingSegment() method. Naturally this is the less recommended working
* mode since we are not making use of segment parallelism. HDFS accessors for
* a specific file type should inherit from this class only if the file they are
* reading does not support splitting: a protocol-buffer file, regular file, ...
- *
*/
public abstract class HdfsAtomicDataAccessor extends Plugin implements ReadAccessor {
private Configuration conf = null;
@@ -36,10 +35,10 @@ public abstract class HdfsAtomicDataAccessor extends Plugin implements ReadAcces
/**
* Constructs a HdfsAtomicDataAccessor object.
+ *
* @param input all input parameters coming from the client
- * @throws Exception
*/
- public HdfsAtomicDataAccessor(InputData input) throws Exception {
+ public HdfsAtomicDataAccessor(InputData input) {
// 0. Hold the configuration data
super(input);
@@ -52,10 +51,11 @@ public abstract class HdfsAtomicDataAccessor extends Plugin implements ReadAcces
/**
* Opens the file using the non-splittable API for HADOOP HDFS file access
* This means that instead of using a FileInputFormat for access, we use a
- * Java stream
- *
+ * Java stream.
+ *
* @return true for successful file open, false otherwise
*/
+ @Override
public boolean openForRead() throws Exception {
if (!isWorkingSegment()) {
return false;
@@ -70,8 +70,10 @@ public abstract class HdfsAtomicDataAccessor extends Plugin implements ReadAcces
/**
* Fetches one record from the file.
- * @return a {@link OneRow} record as a Java object. returns null if none.
+ *
+ * @return a {@link OneRow} record as a Java object. Returns null if none.
*/
+ @Override
public OneRow readNextObject() throws IOException {
if (!isWorkingSegment()) {
return null;
@@ -83,6 +85,7 @@ public abstract class HdfsAtomicDataAccessor extends Plugin implements ReadAcces
/**
* Closes the access stream when finished reading the file
*/
+ @Override
public void closeForRead() throws Exception {
if (!isWorkingSegment()) {
return;
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsDataFragmenter.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsDataFragmenter.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsDataFragmenter.java
index ab4aa67..1bf5aab 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsDataFragmenter.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsDataFragmenter.java
@@ -15,43 +15,43 @@ import java.io.IOException;
import java.util.List;
/**
- * Fragmenter class for HDFS data resources
+ * Fragmenter class for HDFS data resources.
*
- * Given an HDFS data source (a file, directory, or wild card pattern)
- * divide the data into fragments and return a list of them along with
- * a list of host:port locations for each.
+ * Given an HDFS data source (a file, directory, or wild card pattern) divide
+ * the data into fragments and return a list of them along with a list of
+ * host:port locations for each.
*/
public class HdfsDataFragmenter extends Fragmenter {
private JobConf jobConf;
/**
- * Constructs an HdfsDataFragmenter object
+ * Constructs an HdfsDataFragmenter object.
+ *
* @param md all input parameters coming from the client
- * @throws IOException
*/
- public HdfsDataFragmenter(InputData md) throws IOException {
+ public HdfsDataFragmenter(InputData md) {
super(md);
jobConf = new JobConf(new Configuration(), HdfsDataFragmenter.class);
}
- /*
- * path is a data source URI that can appear as a file
- * name, a directory name or a wildcard returns the data
- * fragments in json format
+ /**
+ * Gets the fragments for a data source URI that can appear as a file name,
+ * a directory name or a wildcard. Returns the data fragments in JSON
+ * format.
*/
@Override
public List<Fragment> getFragments() throws Exception {
- String absoluteDataPath = HdfsUtilities.absoluteDataPath(inputData.getDataSource());
+ String absoluteDataPath = HdfsUtilities.absoluteDataPath(inputData.getDataSource());
InputSplit[] splits = getSplits(new Path(absoluteDataPath));
- for (InputSplit split : splits != null ? splits : new InputSplit[]{}) {
+ for (InputSplit split : splits != null ? splits : new InputSplit[] {}) {
FileSplit fsp = (FileSplit) split;
- /*
- * HD-2547: If the file is empty, an empty split is returned:
- * no locations and no length.
- */
+ /*
+ * HD-2547: If the file is empty, an empty split is returned: no
+ * locations and no length.
+ */
if (fsp.getLength() <= 0) {
continue;
}
@@ -59,10 +59,10 @@ public class HdfsDataFragmenter extends Fragmenter {
String filepath = fsp.getPath().toUri().getPath();
String[] hosts = fsp.getLocations();
- /*
- * metadata information includes: file split's
- * start, length and hosts (locations).
- */
+ /*
+ * metadata information includes: file split's start, length and
+ * hosts (locations).
+ */
byte[] fragmentMetadata = HdfsUtilities.prepareFragmentMetadata(fsp);
Fragment fragment = new Fragment(filepath, hosts, fragmentMetadata);
fragments.add(fragment);
@@ -76,5 +76,4 @@ public class HdfsDataFragmenter extends Fragmenter {
PxfInputFormat.setInputPaths(jobConf, path);
return format.getSplits(jobConf, 1);
}
-
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsSplittableDataAccessor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsSplittableDataAccessor.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsSplittableDataAccessor.java
index 8f49d48..744342d 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsSplittableDataAccessor.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/HdfsSplittableDataAccessor.java
@@ -16,10 +16,11 @@ import java.util.ListIterator;
* Accessor for accessing a splittable HDFS data sources. HDFS will divide the
* file into splits based on an internal decision (by default, the block size is
* also the split size).
- *
+ *
* Accessors that require such base functionality should extend this class.
*/
-public abstract class HdfsSplittableDataAccessor extends Plugin implements ReadAccessor {
+public abstract class HdfsSplittableDataAccessor extends Plugin implements
+ ReadAccessor {
protected Configuration conf = null;
protected RecordReader<Object, Object> reader = null;
protected InputFormat<?, ?> inputFormat = null;
@@ -29,12 +30,12 @@ public abstract class HdfsSplittableDataAccessor extends Plugin implements ReadA
/**
* Constructs an HdfsSplittableDataAccessor
- *
+ *
* @param input all input parameters coming from the client request
* @param inFormat the HDFS {@link InputFormat} the caller wants to use
- * @throws Exception
*/
- public HdfsSplittableDataAccessor(InputData input, InputFormat<?, ?> inFormat) throws Exception {
+ public HdfsSplittableDataAccessor(InputData input,
+ InputFormat<?, ?> inFormat) {
super(input);
inputFormat = inFormat;
@@ -46,11 +47,12 @@ public abstract class HdfsSplittableDataAccessor extends Plugin implements ReadA
}
/**
- * Fetches the requested fragment (file split) for the current client
+ * Fetches the requested fragment (file split) for the current client
* request, and sets a record reader for the job.
- *
+ *
* @return true if succeeded, false if no more splits to be read
*/
+ @Override
public boolean openForRead() throws Exception {
LinkedList<InputSplit> requestSplits = new LinkedList<InputSplit>();
FileSplit fileSplit = HdfsUtilities.parseFragmentMetadata(inputData);
@@ -62,23 +64,28 @@ public abstract class HdfsSplittableDataAccessor extends Plugin implements ReadA
}
/**
- * Specialized accessors will override this method and implement their own
+ * Specialized accessors will override this method and implement their own
* recordReader. For example, a plain delimited text accessor may want to
- * return a LineRecordReader.
- *
+ * return a LineRecordReader.
+ *
* @param jobConf the hadoop jobconf to use for the selected InputFormat
* @param split the input split to be read by the accessor
- * @return a recordreader to be used for reading the data records of the split
- * @throws IOException
+ * @return a recordreader to be used for reading the data records of the
+ * split
+ * @throws IOException if recordreader could not be created
*/
- abstract protected Object getReader(JobConf jobConf, InputSplit split) throws IOException;
+ abstract protected Object getReader(JobConf jobConf, InputSplit split)
+ throws IOException;
- /*
- * getNextSplit
- * Sets the current split and initializes a RecordReader who feeds from the split
+ /**
+ * Sets the current split and initializes a RecordReader who feeds from the
+ * split
+ *
+ * @return true if there is a split to read
+ * @throws IOException if record reader could not be created
*/
@SuppressWarnings(value = "unchecked")
- protected boolean getNextSplit() throws IOException {
+ protected boolean getNextSplit() throws IOException {
if (!iter.hasNext()) {
return false;
}
@@ -90,36 +97,40 @@ public abstract class HdfsSplittableDataAccessor extends Plugin implements ReadA
return true;
}
- /*
- * readNextObject
- * Fetches one record from the file. The record is returned as a Java object.
+ /**
+ * Fetches one record from the file. The record is returned as a Java
+ * object.
*/
@Override
public OneRow readNextObject() throws IOException {
-
- if (!reader.next(key, data)) { // if there is one more record in the current split
- if (getNextSplit()) {// the current split is exhausted. try to move to the next split.
- if (!reader.next(key, data)) {// read the first record of the new split
- return null; // make sure we return nulls
+ // if there is one more record in the current split
+ if (!reader.next(key, data)) {
+ // the current split is exhausted. try to move to the next split
+ if (getNextSplit()) {
+ // read the first record of the new split
+ if (!reader.next(key, data)) {
+ // make sure we return nulls
+ return null;
}
} else {
- return null; // make sure we return nulls
+ // make sure we return nulls
+ return null;
}
}
- /*
- * if neither condition was met, it means we already read all the
- * records in all the splits, and in this call record variable was not
- * set, so we return null and thus we are signaling end of records
- * sequence
- */
+ /*
+ * if neither condition was met, it means we already read all the
+ * records in all the splits, and in this call record variable was not
+ * set, so we return null and thus we are signaling end of records
+ * sequence
+ */
return new OneRow(key, data);
}
- /*
- * closeForRead
+ /**
* When user finished reading the file, it closes the RecordReader
*/
+ @Override
public void closeForRead() throws Exception {
if (reader != null) {
reader.close();
@@ -129,7 +140,7 @@ public abstract class HdfsSplittableDataAccessor extends Plugin implements ReadA
@Override
public boolean isThreadSafe() {
return HdfsUtilities.isThreadSafe(inputData.getDataSource(),
- inputData.getUserProperty("COMPRESSION_CODEC"));
+ inputData.getUserProperty("COMPRESSION_CODEC"));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/LineBreakAccessor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/LineBreakAccessor.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/LineBreakAccessor.java
index 28b125c..32002a1 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/LineBreakAccessor.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/LineBreakAccessor.java
@@ -16,39 +16,38 @@ import org.apache.hadoop.mapred.*;
import java.io.DataOutputStream;
import java.io.IOException;
-
/**
- * A PXF Accessor for reading delimited plain text records
+ * A PXF Accessor for reading delimited plain text records.
*/
-public class LineBreakAccessor extends HdfsSplittableDataAccessor implements WriteAccessor {
+public class LineBreakAccessor extends HdfsSplittableDataAccessor implements
+ WriteAccessor {
private DataOutputStream dos;
private FSDataOutputStream fsdos;
private Configuration conf;
private FileSystem fs;
private Path file;
- private Log Log;
+ private static Log Log = LogFactory.getLog(LineBreakAccessor.class);
/**
- * Constructs a LineReaderAccessor
- *
+ * Constructs a LineReaderAccessor.
+ *
* @param input all input parameters coming from the client request
- * @throws Exception
*/
- public LineBreakAccessor(InputData input) throws Exception {
+ public LineBreakAccessor(InputData input) {
super(input, new TextInputFormat());
((TextInputFormat) inputFormat).configure(jobConf);
-
- Log = LogFactory.getLog(LineBreakAccessor.class);
}
@Override
- protected Object getReader(JobConf jobConf, InputSplit split) throws IOException {
+ protected Object getReader(JobConf jobConf, InputSplit split)
+ throws IOException {
return new ChunkRecordReader(jobConf, (FileSplit) split);
}
-
- /*
- * opens file for write
+
+ /**
+ * Opens file for write.
*/
+ @Override
public boolean openForWrite() throws Exception {
String fileName = inputData.getDataSource();
@@ -68,7 +67,8 @@ public class LineBreakAccessor extends HdfsSplittableDataAccessor implements Wri
file = new Path(fileName);
if (fs.exists(file)) {
- throw new IOException("file " + file.toString() + " already exists, can't write data");
+ throw new IOException("file " + file.toString()
+ + " already exists, can't write data");
}
org.apache.hadoop.fs.Path parent = file.getParent();
if (!fs.exists(parent)) {
@@ -83,10 +83,11 @@ public class LineBreakAccessor extends HdfsSplittableDataAccessor implements Wri
}
/*
- * Create output stream from given file.
- * If compression codec is provided, wrap it around stream.
+ * Creates output stream from given file. If compression codec is provided,
+ * wrap it around stream.
*/
- private void createOutputStream(Path file, CompressionCodec codec) throws IOException {
+ private void createOutputStream(Path file, CompressionCodec codec)
+ throws IOException {
fsdos = fs.create(file, false);
if (codec != null) {
dos = new DataOutputStream(codec.createOutputStream(fsdos));
@@ -96,28 +97,30 @@ public class LineBreakAccessor extends HdfsSplittableDataAccessor implements Wri
}
- /*
- * write row into stream
+ /**
+ * Writes row into stream.
*/
+ @Override
public boolean writeNextObject(OneRow onerow) throws Exception {
dos.write((byte[]) onerow.getData());
return true;
}
-
- /*
- * close the output stream after done writing
+
+ /**
+ * Closes the output stream after done writing.
*/
+ @Override
public void closeForWrite() throws Exception {
if ((dos != null) && (fsdos != null)) {
Log.debug("Closing writing stream for path " + file);
dos.flush();
/*
* From release 0.21.0 sync() is deprecated in favor of hflush(),
- * which only guarantees that new readers will see all data written
- * to that point, and hsync(), which makes a stronger guarantee that
- * the operating system has flushed the data to disk (like POSIX
- * fsync), although data may still be in the disk cache.
- */
+ * which only guarantees that new readers will see all data written
+ * to that point, and hsync(), which makes a stronger guarantee that
+ * the operating system has flushed the data to disk (like POSIX
+ * fsync), although data may still be in the disk cache.
+ */
fsdos.hsync();
dos.close();
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/QuotedLineBreakAccessor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/QuotedLineBreakAccessor.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/QuotedLineBreakAccessor.java
index 3fc0f5e..13880b8 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/QuotedLineBreakAccessor.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/QuotedLineBreakAccessor.java
@@ -16,12 +16,11 @@ public class QuotedLineBreakAccessor extends HdfsAtomicDataAccessor {
private BufferedReader reader;
/**
- * Constructs a QuotedLineBreakAccessor
- *
+ * Constructs a QuotedLineBreakAccessor.
+ *
* @param input all input parameters coming from the client request
- * @throws Exception
*/
- public QuotedLineBreakAccessor(InputData input) throws Exception {
+ public QuotedLineBreakAccessor(InputData input) {
super(input);
}
@@ -34,8 +33,7 @@ public class QuotedLineBreakAccessor extends HdfsAtomicDataAccessor {
return true;
}
- /*
- * readNextObject
+ /**
* Fetches one record (maybe partial) from the file. The record is returned as a Java object.
*/
@Override
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/SequenceFileAccessor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/SequenceFileAccessor.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/SequenceFileAccessor.java
index d3e7292..5f3f3dd 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/SequenceFileAccessor.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/SequenceFileAccessor.java
@@ -4,6 +4,7 @@ import com.pivotal.pxf.api.OneRow;
import com.pivotal.pxf.api.WriteAccessor;
import com.pivotal.pxf.api.utilities.InputData;
import com.pivotal.pxf.plugins.hdfs.utilities.HdfsUtilities;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -24,7 +25,8 @@ import java.util.EnumSet;
/**
* A PXF Accessor for reading and writing Sequence File records
*/
-public class SequenceFileAccessor extends HdfsSplittableDataAccessor implements WriteAccessor {
+public class SequenceFileAccessor extends HdfsSplittableDataAccessor implements
+ WriteAccessor {
private Configuration conf;
private FileContext fc;
@@ -34,27 +36,24 @@ public class SequenceFileAccessor extends HdfsSplittableDataAccessor implements
private SequenceFile.Writer writer;
private LongWritable defaultKey; // used when recordkey is not defined
- private Log Log;
+ private static Log Log = LogFactory.getLog(SequenceFileAccessor.class);;
/**
- * Constructs a SequenceFileAccessor
- *
+ * Constructs a SequenceFileAccessor.
+ *
* @param input all input parameters coming from the client request
- * @throws Exception
*/
- public SequenceFileAccessor(InputData input) throws Exception {
-
- super(input,
- new SequenceFileInputFormat<Writable, Writable>());
-
- Log = LogFactory.getLog(SequenceFileAccessor.class);
+ public SequenceFileAccessor(InputData input) {
+ super(input, new SequenceFileInputFormat<Writable, Writable>());
}
- /*
- * Override virtual method to create specialized record reader
+ /**
+ * Overrides virtual method to create specialized record reader
*/
- protected Object getReader(JobConf jobConf, InputSplit split) throws IOException {
- return new SequenceFileRecordReader(jobConf, (FileSplit) split);
+ @Override
+ protected Object getReader(JobConf jobConf, InputSplit split)
+ throws IOException {
+ return new SequenceFileRecordReader<Object, Object>(jobConf, (FileSplit) split);
}
@Override
@@ -74,7 +73,8 @@ public class SequenceFileAccessor extends HdfsSplittableDataAccessor implements
defaultKey = new LongWritable(inputData.getSegmentId());
if (fs.exists(file)) {
- throw new IOException("file " + file + " already exists, can't write data");
+ throw new IOException("file " + file
+ + " already exists, can't write data");
}
parent = file.getParent();
if (!fs.exists(parent)) {
@@ -87,10 +87,11 @@ public class SequenceFileAccessor extends HdfsSplittableDataAccessor implements
}
/**
- * Compression: based on compression codec and compression type (default value RECORD).
- * If there is no codec, compression type is ignored, and NONE is used.
+ * Compression: based on compression codec and compression type (default
+ * value RECORD). If there is no codec, compression type is ignored, and
+ * NONE is used.
*
- * @param inputData - container where compression codec and type are held.
+ * @param inputData - container where compression codec and type are held
*/
private void getCompressionCodec(InputData inputData) {
@@ -106,21 +107,23 @@ public class SequenceFileAccessor extends HdfsSplittableDataAccessor implements
try {
compressionType = CompressionType.valueOf(parsedCompressType);
} catch (IllegalArgumentException e) {
- throw new IllegalArgumentException("Illegal value for compression type " +
- "'" + parsedCompressType + "'");
+ throw new IllegalArgumentException(
+ "Illegal value for compression type " + "'"
+ + parsedCompressType + "'");
}
if (compressionType == null) {
- throw new IllegalArgumentException("Compression type must be defined");
+ throw new IllegalArgumentException(
+ "Compression type must be defined");
}
- Log.debug("Compression ON: " +
- "compression codec: " + userCompressCodec +
- ", compression type: " + compressionType);
+ Log.debug("Compression ON: " + "compression codec: "
+ + userCompressCodec + ", compression type: "
+ + compressionType);
}
}
/*
- * Parse compression type for sequence file. If null, default to RECORD.
+ * Parses compression type for sequence file. If null, default to RECORD.
* Allowed values: RECORD, BLOCK.
*/
private String parseCompressionType(String compressType) {
@@ -133,13 +136,15 @@ public class SequenceFileAccessor extends HdfsSplittableDataAccessor implements
}
if (compressType.equalsIgnoreCase(COMPRESSION_TYPE_NONE)) {
- throw new IllegalArgumentException("Illegal compression type 'NONE'. " +
- "For disabling compression remove COMPRESSION_CODEC parameter.");
+ throw new IllegalArgumentException(
+ "Illegal compression type 'NONE'. "
+ + "For disabling compression remove COMPRESSION_CODEC parameter.");
}
- if (!compressType.equalsIgnoreCase(COMPRESSION_TYPE_RECORD) &&
- !compressType.equalsIgnoreCase(COMPRESSION_TYPE_BLOCK)) {
- throw new IllegalArgumentException("Illegal compression type '" + compressType + "'");
+ if (!compressType.equalsIgnoreCase(COMPRESSION_TYPE_RECORD)
+ && !compressType.equalsIgnoreCase(COMPRESSION_TYPE_BLOCK)) {
+ throw new IllegalArgumentException("Illegal compression type '"
+ + compressType + "'");
}
return compressType.toUpperCase();
@@ -165,11 +170,13 @@ public class SequenceFileAccessor extends HdfsSplittableDataAccessor implements
// init writer on first approach here, based on onerow.getData type
// TODO: verify data is serializable.
if (writer == null) {
- Class valueClass = value.getClass();
- Class keyClass = (key == null) ? LongWritable.class : key.getClass();
+ Class<? extends Writable> valueClass = value.getClass();
+ Class<? extends Writable> keyClass = (key == null) ? LongWritable.class
+ : key.getClass();
// create writer - do not allow overwriting existing file
- writer = SequenceFile.createWriter(fc, conf, file, keyClass, valueClass,
- compressionType, codec, new SequenceFile.Metadata(), EnumSet.of(CreateFlag.CREATE));
+ writer = SequenceFile.createWriter(fc, conf, file, keyClass,
+ valueClass, compressionType, codec,
+ new SequenceFile.Metadata(), EnumSet.of(CreateFlag.CREATE));
}
try {
@@ -188,20 +195,21 @@ public class SequenceFileAccessor extends HdfsSplittableDataAccessor implements
writer.sync();
/*
* From release 0.21.0 sync() is deprecated in favor of hflush(),
- * which only guarantees that new readers will see all data written to that point,
- * and hsync(), which makes a stronger guarantee that the operating system has flushed
- * the data to disk (like POSIX fsync), although data may still be in the disk cache.
- */
+ * which only guarantees that new readers will see all data written
+ * to that point, and hsync(), which makes a stronger guarantee that
+ * the operating system has flushed the data to disk (like POSIX
+ * fsync), although data may still be in the disk cache.
+ */
writer.hsync();
writer.close();
}
}
- public CompressionType getCompressionType() {
- return compressionType;
- }
+ public CompressionType getCompressionType() {
+ return compressionType;
+ }
- public CompressionCodec getCodec() {
- return codec;
- }
+ public CompressionCodec getCodec() {
+ return codec;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/StringPassResolver.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/StringPassResolver.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/StringPassResolver.java
index 588792b..7ace4c8 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/StringPassResolver.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/StringPassResolver.java
@@ -15,8 +15,8 @@ import java.util.List;
import static com.pivotal.pxf.api.io.DataType.VARCHAR;
/**
- * StringPassResolver handles "deserialization" and serialization of
- * String records. StringPassResolver implements IReadResolver and
+ * StringPassResolver handles "deserialization" and serialization of
+ * String records. StringPassResolver implements IReadResolver and
* IWriteResolver interfaces. Returns strings as-is.
*/
public class StringPassResolver extends Plugin implements ReadResolver, WriteResolver {
@@ -24,20 +24,19 @@ public class StringPassResolver extends Plugin implements ReadResolver, WriteRes
private OneRow oneRow;
/**
- * Constructs a StringPassResolver
- *
+ * Constructs a StringPassResolver.
+ *
* @param inputData input all input parameters coming from the client request
- * @throws Exception
*/
- public StringPassResolver(InputData inputData) throws Exception {
+ public StringPassResolver(InputData inputData) {
super(inputData);
oneRow = new OneRow();
this.inputData = inputData;
}
- /*
- * getFields returns a list of the fields of one record.
- * Each record field is represented by a OneField item.
+ /**
+ * Returns a list of the fields of one record.
+ * Each record field is represented by a {@link OneField} item.
* OneField item contains two fields: an integer representing the field type and a Java
* Object representing the field value.
*/
@@ -45,7 +44,7 @@ public class StringPassResolver extends Plugin implements ReadResolver, WriteRes
public List<OneField> getFields(OneRow onerow) {
/*
* This call forces a whole text line into a single varchar field and replaces
- * the proper field separation code can be found in previous revisions. The reasons
+ * the proper field separation code can be found in previous revisions. The reasons
* for doing so as this point are:
* 1. performance
* 2. desire to not replicate text parsing logic from the backend into java
@@ -61,8 +60,8 @@ public class StringPassResolver extends Plugin implements ReadResolver, WriteRes
return record;
}
- /*
- * Creates a OneRow object from the singleton list
+ /**
+ * Creates a OneRow object from the singleton list.
*/
@Override
public OneRow setFields(List<OneField> record) throws Exception {
@@ -74,5 +73,3 @@ public class StringPassResolver extends Plugin implements ReadResolver, WriteRes
return oneRow;
}
}
-
-
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/WritableResolver.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/WritableResolver.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/WritableResolver.java
index 2913433..e9df907 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/WritableResolver.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/WritableResolver.java
@@ -23,11 +23,11 @@ import java.util.List;
import static com.pivotal.pxf.api.io.DataType.*;
/**
- * WritableResolver handles serialization and deserialization of records
+ * WritableResolver handles serialization and deserialization of records
* that were serialized using Hadoop's Writable serialization framework.
- *
- * A field named 'recordkey' is treated as a key of the given row, and not as
- * part of the data schema. @See RecordkeyAdapter.
+ *
+ * A field named 'recordkey' is treated as a key of the given row, and not as
+ * part of the data schema. See {@link RecordkeyAdapter}.
*/
public class WritableResolver extends Plugin implements ReadResolver, WriteResolver {
private static final int RECORDKEY_UNDEFINED = -1;
@@ -40,10 +40,11 @@ public class WritableResolver extends Plugin implements ReadResolver, WriteResol
/**
- * Constructs a WritableResolver
- *
+ * Constructs a WritableResolver.
+ *
* @param input all input parameters coming from the client
- * @throws Exception
+ * @throws Exception if schema file is missing, cannot be found in
+ * classpath or fails to instantiate
*/
public WritableResolver(InputData input) throws Exception {
super(input);
@@ -77,7 +78,6 @@ public class WritableResolver extends Plugin implements ReadResolver, WriteResol
" type: " + javaType + ", " +
(isArray(javaType) ? "Array" : "Primitive") + ", " +
(isPrivate ? "Private" : "accessible") + " field");
-
}
}
}
@@ -86,12 +86,6 @@ public class WritableResolver extends Plugin implements ReadResolver, WriteResol
return (javaType.startsWith("[") && !"[B".equals(javaType));
}
- /*
- * getFields returns a list of the fields of one record.
- * Each record field is represented by a OneField item.
- * OneField item contains two fields: an integer representing the field type and a Java
- * Object representing the field value.
- */
@Override
public List<OneField> getFields(OneRow onerow) throws Exception {
userObject = onerow.getData();
@@ -168,8 +162,8 @@ public class WritableResolver extends Plugin implements ReadResolver, WriteResol
}
}
- /*
- * Sets customWritable fields and creates a OneRow object
+ /**
+ * Sets customWritable fields and creates a OneRow object.
*/
@Override
public OneRow setFields(List<OneField> record) throws Exception {
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/utilities/HdfsUtilities.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/utilities/HdfsUtilities.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/utilities/HdfsUtilities.java
index 03fe94c..c7fe103 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/utilities/HdfsUtilities.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/utilities/HdfsUtilities.java
@@ -26,20 +26,20 @@ import java.io.*;
import java.util.List;
/**
- * HdfsUtilities class exposes helper methods for PXF classes
+ * HdfsUtilities class exposes helper methods for PXF classes.
*/
public class HdfsUtilities {
private static Log Log = LogFactory.getLog(HdfsUtilities.class);
private static Configuration config = new Configuration();
- private static CompressionCodecFactory factory =
- new CompressionCodecFactory(config);
+ private static CompressionCodecFactory factory = new CompressionCodecFactory(
+ config);
/**
- * Hdfs data sources are absolute data paths. Method ensures
- * that dataSource begins with '/'
+ * Hdfs data sources are absolute data paths. Method ensures that dataSource
+ * begins with '/'.
*
* @param dataSource The HDFS path to a file or directory of interest.
- * Retrieved from the client request.
+ * Retrieved from the client request.
* @return an absolute data path
*/
public static String absoluteDataPath(String dataSource) {
@@ -49,20 +49,22 @@ public class HdfsUtilities {
/*
* Helper routine to get a compression codec class
*/
- private static Class<? extends CompressionCodec> getCodecClass(
- Configuration conf, String name) {
+ private static Class<? extends CompressionCodec> getCodecClass(Configuration conf,
+ String name) {
Class<? extends CompressionCodec> codecClass;
try {
- codecClass = conf.getClassByName(name).asSubclass(CompressionCodec.class);
+ codecClass = conf.getClassByName(name).asSubclass(
+ CompressionCodec.class);
} catch (ClassNotFoundException e) {
- throw new IllegalArgumentException("Compression codec " + name + " was not found.", e);
+ throw new IllegalArgumentException("Compression codec " + name
+ + " was not found.", e);
}
return codecClass;
}
/**
- * Helper routine to get compression codec through reflection
+ * Helper routine to get compression codec through reflection.
*
* @param conf configuration used for reflection
* @param name codec name
@@ -73,7 +75,7 @@ public class HdfsUtilities {
}
/**
- * Helper routine to get compression codec class by path (file suffix)
+ * Helper routine to get compression codec class by path (file suffix).
*
* @param path path of file to get codec for
* @return matching codec class for the path. null if no codec is needed.
@@ -91,8 +93,8 @@ public class HdfsUtilities {
}
/**
- * Returns true if the needed codec is splittable.
- * If no codec is needed returns true as well.
+ * Returns true if the needed codec is splittable. If no codec is needed
+ * returns true as well.
*
* @param path path of the file to be read
* @return if the codec needed for reading the specified path is splittable.
@@ -110,30 +112,32 @@ public class HdfsUtilities {
/**
* Checks if requests should be handle in a single thread or not.
*
- * @param dataDir hdfs path to the data source
+ * @param dataDir hdfs path to the data source
* @param compCodec the fully qualified name of the compression codec
* @return if the request can be run in multi-threaded mode.
*/
public static boolean isThreadSafe(String dataDir, String compCodec) {
- Class<? extends CompressionCodec> codecClass = (compCodec != null)
- ? HdfsUtilities.getCodecClass(config, compCodec)
- : HdfsUtilities.getCodecClassByPath(dataDir);
+ Class<? extends CompressionCodec> codecClass = (compCodec != null) ? HdfsUtilities.getCodecClass(
+ config, compCodec) : HdfsUtilities.getCodecClassByPath(dataDir);
/* bzip2 codec is not thread safe */
return (codecClass == null || !BZip2Codec.class.isAssignableFrom(codecClass));
}
/**
- * Prepare byte serialization of a file split information
- * (start, length, hosts) using {@link ObjectOutputStream}.
+ * Prepares byte serialization of a file split information (start, length,
+ * hosts) using {@link ObjectOutputStream}.
*
* @param fsp file split to be serialized
* @return byte serialization of fsp
- * @throws IOException
+ * @throws IOException if I/O errors occur while writing to the underlying
+ * stream
*/
- public static byte[] prepareFragmentMetadata(FileSplit fsp) throws IOException {
+ public static byte[] prepareFragmentMetadata(FileSplit fsp)
+ throws IOException {
ByteArrayOutputStream byteArrayStream = new ByteArrayOutputStream();
- ObjectOutputStream objectStream = new ObjectOutputStream(byteArrayStream);
+ ObjectOutputStream objectStream = new ObjectOutputStream(
+ byteArrayStream);
objectStream.writeLong(fsp.getStart());
objectStream.writeLong(fsp.getLength());
objectStream.writeObject(fsp.getLocations());
@@ -142,7 +146,7 @@ public class HdfsUtilities {
}
/**
- * Parse fragment metadata and return matching {@link FileSplit}
+ * Parses fragment metadata and return matching {@link FileSplit}.
*
* @param inputData request input data
* @return FileSplit with fragment metadata
@@ -151,10 +155,12 @@ public class HdfsUtilities {
try {
byte[] serializedLocation = inputData.getFragmentMetadata();
if (serializedLocation == null) {
- throw new IllegalArgumentException("Missing fragment location information");
+ throw new IllegalArgumentException(
+ "Missing fragment location information");
}
- ByteArrayInputStream bytesStream = new ByteArrayInputStream(serializedLocation);
+ ByteArrayInputStream bytesStream = new ByteArrayInputStream(
+ serializedLocation);
ObjectInputStream objectStream = new ObjectInputStream(bytesStream);
long start = objectStream.readLong();
@@ -162,44 +168,56 @@ public class HdfsUtilities {
String[] hosts = (String[]) objectStream.readObject();
- FileSplit fileSplit = new FileSplit(new Path(inputData.getDataSource()),
- start,
- end,
- hosts);
+ FileSplit fileSplit = new FileSplit(new Path(
+ inputData.getDataSource()), start, end, hosts);
- Log.debug("parsed file split: path " + inputData.getDataSource() +
- ", start " + start + ", end " + end +
- ", hosts " + ArrayUtils.toString(hosts));
+ Log.debug("parsed file split: path " + inputData.getDataSource()
+ + ", start " + start + ", end " + end + ", hosts "
+ + ArrayUtils.toString(hosts));
return fileSplit;
} catch (Exception e) {
- throw new RuntimeException("Exception while reading expected fragment metadata", e);
+ throw new RuntimeException(
+ "Exception while reading expected fragment metadata", e);
}
}
/**
- * Accessing the avro file through the "unsplittable" API just to get the schema.
- * The splittable API (AvroInputFormat) which is the one we will be using to fetch
- * the records, does not support getting the avro schema yet.
+ * Accessing the Avro file through the "unsplittable" API just to get the
+ * schema. The splittable API (AvroInputFormat) which is the one we will be
+ * using to fetch the records, does not support getting the Avro schema yet.
*
- * @param conf Hadoop configuration
+ * @param conf Hadoop configuration
* @param dataSource Avro file (i.e fileName.avro) path
* @return the Avro schema
- * @throws IOException
+ * @throws IOException if I/O error occured while accessing Avro schema file
*/
- public static Schema getAvroSchema(Configuration conf, String dataSource) throws IOException {
+ public static Schema getAvroSchema(Configuration conf, String dataSource)
+ throws IOException {
FsInput inStream = new FsInput(new Path(dataSource), conf);
DatumReader<GenericRecord> dummyReader = new GenericDatumReader<>();
- DataFileReader<GenericRecord> dummyFileReader = new DataFileReader<>(inStream, dummyReader);
- return dummyFileReader.getSchema();
+ DataFileReader<GenericRecord> dummyFileReader = new DataFileReader<>(
+ inStream, dummyReader);
+ Schema schema = dummyFileReader.getSchema();
+ dummyFileReader.close();
+ return schema;
}
+ /**
+ * Returns string serialization of list of fields. Fields of binary type
+ * (BYTEA) are converted to octal representation to make sure they will be
+ * relayed properly to the DB.
+ *
+ * @param complexRecord list of fields to be stringified
+ * @param delimiter delimiter between fields
+ * @return string of serialized fields using delimiter
+ */
public static String toString(List<OneField> complexRecord, String delimiter) {
StringBuilder buff = new StringBuilder();
String delim = ""; // first iteration has no delimiter
for (OneField complex : complexRecord) {
- if(complex.type == DataType.BYTEA.getOID()) {
+ if (complex.type == DataType.BYTEA.getOID()) {
/** Serialize byte array as string */
buff.append(delim);
Utilities.byteArrayToOctalString((byte[]) complex.val, buff);
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/utilities/RecordkeyAdapter.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/utilities/RecordkeyAdapter.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/utilities/RecordkeyAdapter.java
index 89ac86e..a88ade0 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/utilities/RecordkeyAdapter.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/utilities/RecordkeyAdapter.java
@@ -8,11 +8,10 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.*;
-import java.io.IOException;
import java.util.List;
/**
- * Adapter used for adding a recordkey field to the records output List<OneField>
+ * Adapter used for adding a recordkey field to the records output {@code List<OneField>}.
*/
public class RecordkeyAdapter {
private Log Log;
@@ -36,7 +35,7 @@ public class RecordkeyAdapter {
private ValConverter converter = null;
/**
- * Constructs a RecordkeyAdapter
+ * Constructs a RecordkeyAdapter.
*/
public RecordkeyAdapter() {
Log = LogFactory.getLog(RecordkeyAdapter.class);
@@ -45,9 +44,9 @@ public class RecordkeyAdapter {
/**
* Adds the recordkey to the end of the passed in recFields list.
* <p>
- * This method also verifies cases in which record keys are not supported
- * by the underlying source type, and therefore "illegally" requested.
- *
+ * This method also verifies cases in which record keys are not supported
+ * by the underlying source type, and therefore "illegally" requested.
+ *
* @param recFields existing list of record (non-key) fields and their values.
* @param input all input parameters coming from the client request
* @param onerow a row object which is used here in order to find out if
@@ -55,12 +54,11 @@ public class RecordkeyAdapter {
* @return 0 if record key not needed, or 1 if record key was appended
* @throws NoSuchFieldException when the given record type does not support
* recordkeys
- * @throws IOException
*/
public int appendRecordkeyField(List<OneField> recFields,
InputData input,
- OneRow onerow) throws NoSuchFieldException, IOException {
-
+ OneRow onerow) throws NoSuchFieldException {
+
/*
* user did not request the recordkey field in the
* "create external table" statement
@@ -107,7 +105,7 @@ public class RecordkeyAdapter {
* key is already a Java primitive we returned it as is If it is an unknown
* type we throw an exception
*/
- private Object extractVal(Object key) throws IOException {
+ private Object extractVal(Object key) {
if (extractor == null) {
extractor = InitializeExtractor(key);
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveAccessor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveAccessor.java b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveAccessor.java
index 865a547..462c7a2 100644
--- a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveAccessor.java
+++ b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveAccessor.java
@@ -15,17 +15,15 @@ import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
-
/**
- * Accessor for Hive tables.
- * The accessor will open and read a split belonging to a Hive table.
- * Opening a split means creating the corresponding InputFormat and RecordReader required to access the
- * split's data. The actual record reading is done in the base class -
- * {@link com.pivotal.pxf.plugins.hdfs.HdfsSplittableDataAccessor}.
- * <p/>
- * HiveAccessor will also enforce Hive partition filtering by filtering-out a split which does not
- * belong to a partition filter. Naturally, the partition filtering will be done only for Hive tables
- * that are partitioned.
+ * Accessor for Hive tables. The accessor will open and read a split belonging
+ * to a Hive table. Opening a split means creating the corresponding InputFormat
+ * and RecordReader required to access the split's data. The actual record
+ * reading is done in the base class -
+ * {@link com.pivotal.pxf.plugins.hdfs.HdfsSplittableDataAccessor}. <br>
+ * HiveAccessor will also enforce Hive partition filtering by filtering-out a
+ * split which does not belong to a partition filter. Naturally, the partition
+ * filtering will be done only for Hive tables that are partitioned.
*/
public class HiveAccessor extends HdfsSplittableDataAccessor {
private static final Log LOG = LogFactory.getLog(HiveAccessor.class);
@@ -46,15 +44,18 @@ public class HiveAccessor extends HdfsSplittableDataAccessor {
protected Boolean filterInFragmenter = false;
/**
- * Constructs a HiveAccessor and creates an InputFormat (derived from {@link org.apache.hadoop.mapred.InputFormat})
- * and the Hive partition fields
+ * Constructs a HiveAccessor and creates an InputFormat (derived from
+ * {@link org.apache.hadoop.mapred.InputFormat}) and the Hive partition
+ * fields
*
* @param input contains the InputFormat class name and the partition fields
+ * @throws Exception if failed to create input format
*/
public HiveAccessor(InputData input) throws Exception {
/*
- * Unfortunately, Java does not allow us to call a function before calling the base constructor,
- * otherwise it would have been: super(input, createInputFormat(input))
+ * Unfortunately, Java does not allow us to call a function before
+ * calling the base constructor, otherwise it would have been:
+ * super(input, createInputFormat(input))
*/
super(input, null);
inputFormat = createInputFormat(input);
@@ -63,20 +64,21 @@ public class HiveAccessor extends HdfsSplittableDataAccessor {
/**
* Constructs a HiveAccessor
*
- * @param input contains the InputFormat class name and the partition fields
+ * @param input contains the InputFormat class name and the partition fields
* @param inputFormat Hive InputFormat
*/
- public HiveAccessor(InputData input, InputFormat<?, ?> inputFormat) throws Exception {
+ public HiveAccessor(InputData input, InputFormat<?, ?> inputFormat) {
super(input, inputFormat);
}
/**
- * openForRead
- * Enables Hive partition filtering
+ * Opens Hive data split for read. Enables Hive partition filtering. <br>
*
- * @return true if there are no partitions or there is no partition filter or
- * partition filter is set and the file currently opened by the accessor belongs
- * to the partition.
+ * @return true if there are no partitions or there is no partition filter
+ * or partition filter is set and the file currently opened by the
+ * accessor belongs to the partition.
+ * @throws Exception if filter could not be built, connection to Hive failed
+ * or resource failed to open
*/
@Override
public boolean openForRead() throws Exception {
@@ -86,27 +88,36 @@ public class HiveAccessor extends HdfsSplittableDataAccessor {
/**
* Creates the RecordReader suitable for this given split.
*
- * @param jobConf configuraton data for the Hadoop framework
- * @param split the split that was allocated for reading to this accessor
+ * @param jobConf configuration data for the Hadoop framework
+ * @param split the split that was allocated for reading to this accessor
+ * @return record reader
+ * @throws IOException if failed to create record reader
*/
@Override
- protected Object getReader(JobConf jobConf, InputSplit split) throws IOException {
+ protected Object getReader(JobConf jobConf, InputSplit split)
+ throws IOException {
return inputFormat.getRecordReader(split, jobConf, Reporter.NULL);
}
/*
- * Parse the user-data supplied by the HiveFragmenter from InputData. Based on the
- * user-data construct the partition fields and the InputFormat for current split
+ * Parses the user-data supplied by the HiveFragmenter from InputData. Based
+ * on the user-data construct the partition fields and the InputFormat for
+ * current split
*/
- private InputFormat<?, ?> createInputFormat(InputData input) throws Exception {
+ private InputFormat<?, ?> createInputFormat(InputData input)
+ throws Exception {
String userData = new String(input.getFragmentUserData());
String[] toks = userData.split(HiveDataFragmenter.HIVE_UD_DELIM);
initPartitionFields(toks[3]);
filterInFragmenter = new Boolean(toks[4]);
- return HiveDataFragmenter.makeInputFormat(toks[0]/* inputFormat name */, jobConf);
+ return HiveDataFragmenter.makeInputFormat(
+ toks[0]/* inputFormat name */, jobConf);
}
- /* The partition fields are initialized one time base on userData provided by the fragmenter */
+ /*
+ * The partition fields are initialized one time base on userData provided
+ * by the fragmenter
+ */
void initPartitionFields(String partitionKeys) {
partitions = new LinkedList<HivePartition>();
if (partitionKeys.equals(HiveDataFragmenter.HIVE_NO_PART_TBL)) {
@@ -140,9 +151,11 @@ public class HiveAccessor extends HdfsSplittableDataAccessor {
boolean returnData = isFiltered(partitions, filter);
if (LOG.isDebugEnabled()) {
- LOG.debug("segmentId: " + inputData.getSegmentId() + " " + inputData.getDataSource() + "--" + filterStr + " returnData: " + returnData);
+ LOG.debug("segmentId: " + inputData.getSegmentId() + " "
+ + inputData.getDataSource() + "--" + filterStr
+ + " returnData: " + returnData);
if (filter instanceof List) {
- for (Object f : (List) filter) {
+ for (Object f : (List<?>) filter) {
printOneBasicFilter(f);
}
} else {
@@ -153,14 +166,16 @@ public class HiveAccessor extends HdfsSplittableDataAccessor {
return returnData;
}
- private boolean isFiltered(List<HivePartition> partitionFields, Object filter) {
+ private boolean isFiltered(List<HivePartition> partitionFields,
+ Object filter) {
if (filter instanceof List) {
/*
- * We are going over each filter in the filters list and test it against all the partition fields
- * since filters are connected only by AND operators, its enough for one filter to fail in order to
+ * We are going over each filter in the filters list and test it
+ * against all the partition fields since filters are connected only
+ * by AND operators, its enough for one filter to fail in order to
* deny this data.
*/
- for (Object f : (List) filter) {
+ for (Object f : (List<?>) filter) {
if (!testOneFilter(partitionFields, f, inputData)) {
return false;
}
@@ -172,21 +187,27 @@ public class HiveAccessor extends HdfsSplittableDataAccessor {
}
/*
- * We are testing one filter against all the partition fields.
- * The filter has the form "fieldA = valueA".
- * The partitions have the form partitionOne=valueOne/partitionTwo=ValueTwo/partitionThree=valueThree
- * 1. For a filter to match one of the partitions, lets say partitionA for example, we need:
- * fieldA = partittionOne and valueA = valueOne. If this condition occurs, we return true.
- * 2. If fieldA does not match any one of the partition fields we also return true, it means we ignore this filter
- * because it is not on a partition field.
- * 3. If fieldA = partittionOne and valueA != valueOne, then we return false.
+ * We are testing one filter against all the partition fields. The filter
+ * has the form "fieldA = valueA". The partitions have the form
+ * partitionOne=valueOne/partitionTwo=ValueTwo/partitionThree=valueThree 1.
+ * For a filter to match one of the partitions, lets say partitionA for
+ * example, we need: fieldA = partittionOne and valueA = valueOne. If this
+ * condition occurs, we return true. 2. If fieldA does not match any one of
+ * the partition fields we also return true, it means we ignore this filter
+ * because it is not on a partition field. 3. If fieldA = partittionOne and
+ * valueA != valueOne, then we return false.
*/
- private boolean testOneFilter(List<HivePartition> partitionFields, Object filter, InputData input) {
+ private boolean testOneFilter(List<HivePartition> partitionFields,
+ Object filter, InputData input) {
// Let's look first at the filter
FilterParser.BasicFilter bFilter = (FilterParser.BasicFilter) filter;
boolean isFilterOperationEqual = (bFilter.getOperation() == FilterParser.Operation.HDOP_EQ);
- if (!isFilterOperationEqual) /* in case this is not an "equality filter" we ignore it here - in partition filtering */ {
+ if (!isFilterOperationEqual) /*
+ * in case this is not an "equality filter"
+ * we ignore it here - in partition
+ * filtering
+ */{
return true;
}
@@ -197,12 +218,18 @@ public class HiveAccessor extends HdfsSplittableDataAccessor {
for (HivePartition partition : partitionFields) {
if (filterColumnName.equals(partition.name)) {
- /* the filter field matches a partition field, but the values do not match */
+ /*
+ * the filter field matches a partition field, but the values do
+ * not match
+ */
return filterValue.equals(partition.val);
}
}
- /* filter field did not match any partition field, so we ignore this filter and hence return true */
+ /*
+ * filter field did not match any partition field, so we ignore this
+ * filter and hence return true
+ */
return true;
}
@@ -211,6 +238,7 @@ public class HiveAccessor extends HdfsSplittableDataAccessor {
boolean isOperationEqual = (bFilter.getOperation() == FilterParser.Operation.HDOP_EQ);
int columnIndex = bFilter.getColumn().index();
String value = bFilter.getConstant().constant().toString();
- LOG.debug("isOperationEqual: " + isOperationEqual + " columnIndex: " + columnIndex + " value: " + value);
+ LOG.debug("isOperationEqual: " + isOperationEqual + " columnIndex: "
+ + columnIndex + " value: " + value);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveDataFragmenter.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveDataFragmenter.java b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveDataFragmenter.java
index 66fb0cb..6ebc62e 100644
--- a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveDataFragmenter.java
+++ b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveDataFragmenter.java
@@ -35,8 +35,8 @@ import com.pivotal.pxf.plugins.hdfs.utilities.HdfsUtilities;
import com.pivotal.pxf.plugins.hive.utilities.HiveUtilities;
/**
- * Fragmenter class for HIVE tables
- * <p/>
+ * Fragmenter class for HIVE tables.
+ * <br>
* Given a Hive table and its partitions divide the data into fragments (here a
* data fragment is actually a HDFS file block) and return a list of them. Each
* data fragment will contain the following information:
@@ -70,7 +70,7 @@ public class HiveDataFragmenter extends Fragmenter {
private Set<String> setPartitions = new TreeSet<String>(
String.CASE_INSENSITIVE_ORDER);
- /*
+ /**
* A Hive table unit - means a subset of the HIVE table, where we can say
* that for all files in this subset, they all have the same InputFormat and
* Serde. For a partitioned table the HiveTableUnit will be one partition
@@ -101,7 +101,7 @@ public class HiveDataFragmenter extends Fragmenter {
}
/**
- * Constructs a HiveDataFragmenter object
+ * Constructs a HiveDataFragmenter object.
*
* @param inputData all input parameters coming from the client
*/
@@ -110,7 +110,7 @@ public class HiveDataFragmenter extends Fragmenter {
}
/**
- * Constructs a HiveDataFragmenter object
+ * Constructs a HiveDataFragmenter object.
*
* @param inputData all input parameters coming from the client
* @param clazz Class for JobConf
@@ -131,11 +131,12 @@ public class HiveDataFragmenter extends Fragmenter {
}
/**
- * Creates the partition InputFormat
+ * Creates the partition InputFormat.
*
* @param inputFormatName input format class name
* @param jobConf configuration data for the Hadoop framework
* @return a {@link org.apache.hadoop.mapred.InputFormat} derived object
+ * @throws Exception if failed to create input format
*/
public static InputFormat<?, ?> makeInputFormat(String inputFormatName,
JobConf jobConf)
[4/4] incubator-hawq git commit: HAWQ-28. JavaDoc fixes for PXF
Posted by nh...@apache.org.
HAWQ-28. JavaDoc fixes for PXF
Project: http://git-wip-us.apache.org/repos/asf/incubator-hawq/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-hawq/commit/88f5f5f6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-hawq/tree/88f5f5f6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-hawq/diff/88f5f5f6
Branch: refs/heads/master
Commit: 88f5f5f63dd7a151eb4eaa9d8dcf5a4ab030945f
Parents: 445e075
Author: Noa Horn <nh...@pivotal.io>
Authored: Mon Oct 12 15:43:06 2015 -0700
Committer: Noa Horn <nh...@pivotal.io>
Committed: Mon Oct 12 15:43:06 2015 -0700
----------------------------------------------------------------------
.../main/java/com/pivotal/pxf/api/Analyzer.java | 6 +-
.../java/com/pivotal/pxf/api/AnalyzerStats.java | 4 +-
.../java/com/pivotal/pxf/api/FilterParser.java | 25 +-
.../java/com/pivotal/pxf/api/Fragmenter.java | 2 +-
.../main/java/com/pivotal/pxf/api/Metadata.java | 2 +-
.../com/pivotal/pxf/api/MetadataFetcher.java | 4 +-
.../java/com/pivotal/pxf/api/ReadAccessor.java | 6 +-
.../java/com/pivotal/pxf/api/ReadResolver.java | 2 +-
.../java/com/pivotal/pxf/api/WriteAccessor.java | 6 +-
.../java/com/pivotal/pxf/api/WriteResolver.java | 2 +-
.../pxf/api/utilities/ColumnDescriptor.java | 13 +-
.../pivotal/pxf/api/utilities/InputData.java | 197 +++++---
.../pxf/plugins/hbase/HBaseFilterBuilder.java | 14 +-
.../pxf/plugins/hbase/HBaseResolver.java | 14 +-
.../hbase/utilities/HBaseColumnDescriptor.java | 12 +-
.../hbase/utilities/HBaseIntegerComparator.java | 16 +-
.../hbase/utilities/HBaseLookupTable.java | 12 +-
.../hbase/utilities/HBaseTupleDescription.java | 12 +-
.../plugins/hbase/utilities/HBaseUtilities.java | 8 +-
.../pxf/plugins/hdfs/AvroFileAccessor.java | 14 +-
.../pivotal/pxf/plugins/hdfs/AvroResolver.java | 212 +++++---
.../pivotal/pxf/plugins/hdfs/ChunkReader.java | 312 ++++++------
.../pxf/plugins/hdfs/ChunkRecordReader.java | 482 ++++++++++---------
.../pivotal/pxf/plugins/hdfs/ChunkWritable.java | 14 +-
.../pivotal/pxf/plugins/hdfs/HdfsAnalyzer.java | 27 +-
.../plugins/hdfs/HdfsAtomicDataAccessor.java | 25 +-
.../pxf/plugins/hdfs/HdfsDataFragmenter.java | 43 +-
.../hdfs/HdfsSplittableDataAccessor.java | 81 ++--
.../pxf/plugins/hdfs/LineBreakAccessor.java | 59 +--
.../plugins/hdfs/QuotedLineBreakAccessor.java | 10 +-
.../pxf/plugins/hdfs/SequenceFileAccessor.java | 98 ++--
.../pxf/plugins/hdfs/StringPassResolver.java | 25 +-
.../pxf/plugins/hdfs/WritableResolver.java | 26 +-
.../plugins/hdfs/utilities/HdfsUtilities.java | 104 ++--
.../hdfs/utilities/RecordkeyAdapter.java | 18 +-
.../pivotal/pxf/plugins/hive/HiveAccessor.java | 124 +++--
.../pxf/plugins/hive/HiveDataFragmenter.java | 13 +-
.../pxf/plugins/hive/HiveFilterBuilder.java | 96 ++--
.../plugins/hive/HiveInputFormatFragmenter.java | 64 +--
.../pxf/plugins/hive/HiveLineBreakAccessor.java | 5 +-
.../pxf/plugins/hive/HiveRCFileAccessor.java | 5 +-
.../pivotal/pxf/plugins/hive/HiveResolver.java | 17 +-
.../plugins/hive/utilities/HiveUtilities.java | 42 +-
.../pxf/service/BridgeOutputBuilder.java | 170 ++++---
.../pivotal/pxf/service/FragmentsResponse.java | 7 +-
.../pxf/service/FragmentsResponseFormatter.java | 8 +-
.../pxf/service/MetadataResponseFormatter.java | 4 +
.../pivotal/pxf/service/io/BufferWritable.java | 58 +--
.../pivotal/pxf/service/io/GPDBWritable.java | 74 +--
.../java/com/pivotal/pxf/service/io/Text.java | 148 +++---
.../com/pivotal/pxf/service/io/Writable.java | 5 +-
.../pxf/service/rest/InvalidPathResource.java | 2 +-
.../pxf/service/rest/MetadataResource.java | 40 +-
.../pivotal/pxf/service/rest/RestResource.java | 3 +-
.../service/utilities/CustomWebappLoader.java | 54 +--
.../pxf/service/utilities/ProtocolData.java | 129 +++--
.../pxf/service/utilities/Utilities.java | 86 ++--
57 files changed, 1764 insertions(+), 1297 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Analyzer.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Analyzer.java b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Analyzer.java
index c18a0da..673aed4 100644
--- a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Analyzer.java
+++ b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Analyzer.java
@@ -7,8 +7,8 @@ import com.pivotal.pxf.api.utilities.Plugin;
* Abstract class that defines getting statistics for ANALYZE.
* {@link #getEstimatedStats} returns statistics for a given path
* (block size, number of blocks, number of tuples).
- * Used when calling ANALYZE on a PXF external table, to get
- * table's statistics that are used by the optimizer to plan queries.
+ * Used when calling ANALYZE on a PXF external table, to get
+ * table's statistics that are used by the optimizer to plan queries.
*/
public abstract class Analyzer extends Plugin {
/**
@@ -28,7 +28,7 @@ public abstract class Analyzer extends Plugin {
*
* @param data the data source name (e.g, file, dir, wildcard, table name).
* @return AnalyzerStats the data statistics in json format.
- * @throws Exception
+ * @throws Exception if fails to get stats
*/
public AnalyzerStats getEstimatedStats(String data) throws Exception {
/* Return default values */
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/AnalyzerStats.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/AnalyzerStats.java b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/AnalyzerStats.java
index d831c1e..9f9206c 100644
--- a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/AnalyzerStats.java
+++ b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/AnalyzerStats.java
@@ -44,9 +44,9 @@ public class AnalyzerStats {
*
* @param stats the data to be serialized
* @return the result in json format
- * @throws IOException
+ * @throws IOException if converting to JSON format failed
*/
- public static String dataToJSON(AnalyzerStats stats) throws IOException {
+ public static String dataToJSON(AnalyzerStats stats) throws IOException {
ObjectMapper mapper = new ObjectMapper();
// mapper serializes all members of the class by default
return "{\"PXFDataSourceStats\":" + mapper.writeValueAsString(stats) + "}";
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/FilterParser.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/FilterParser.java b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/FilterParser.java
index 9489601..a3803c5 100644
--- a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/FilterParser.java
+++ b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/FilterParser.java
@@ -8,19 +8,21 @@ import java.util.Stack;
* The parser code which goes over a filter string and pushes operands onto a stack.
* Once an operation is read, the evaluate function is called for the {@link FilterBuilder}
* interface with two pop-ed operands.
- *
+ * <br>
* A string of filters looks like this:
- * a2c5o1a1c"abc"o2o7
- * which means column#2 < 5 AND column#1 > "abc"
- *
+ * <code>a2c5o1a1c"abc"o2o7</code>
+ * which means {@code column#2 < 5 AND column#1 > "abc"}
+ * <br>
* It is a RPN serialized representation of a filters tree in GPDB where
- * a means an attribute (column)
- * c means a constant (either string or numeric)
- * o means operator
+ * <ul>
+ * <li> a means an attribute (column)</li>
+ * <li>c means a constant (either string or numeric)</li>
+ * <li>o means operator</li>
+ * </ul>
*
* Assuming all operators are binary, RPN representation allows it to be read left to right easily.
- *
- * FilterParser only knows about columns and constants. The rest is up to the {@link FilterBuilder} implementer. *
+ * <br>
+ * FilterParser only knows about columns and constants. The rest is up to the {@link FilterBuilder} implementer.
* FilterParser makes sure column objects are always on the left of the expression (when relevant).
*/
public class FilterParser {
@@ -55,7 +57,7 @@ public class FilterParser {
* @param left the left operand
* @param right the right operand
* @return the built filter
- * @throws Exception
+ * @throws Exception if building the filter failed
*/
public Object build(Operation operation, Object left, Object right) throws Exception;
}
@@ -125,6 +127,7 @@ public class FilterParser {
/**
* Thrown when a filter's parsing exception occurs.
*/
+ @SuppressWarnings("serial")
class FilterStringSyntaxException extends Exception {
FilterStringSyntaxException(String desc) {
super(desc + " (filter string: '" + filterString + "')");
@@ -146,7 +149,7 @@ public class FilterParser {
*
* @param filter the filter to parse
* @return the parsed filter
- * @throws Exception
+ * @throws Exception if the filter string had wrong syntax
*/
public Object parse(String filter) throws Exception {
index = 0;
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Fragmenter.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Fragmenter.java b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Fragmenter.java
index dc7f9fb..831a7a6 100644
--- a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Fragmenter.java
+++ b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Fragmenter.java
@@ -27,7 +27,7 @@ public abstract class Fragmenter extends Plugin {
* Used to get fragments of data that could be read in parallel from the different segments.
*
* @return list of data fragments
- * @throws Exception
+ * @throws Exception if fragment list could not be retrieved
*/
public abstract List<Fragment> getFragments() throws Exception;
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Metadata.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Metadata.java b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Metadata.java
index bf02c30..5a31360 100644
--- a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Metadata.java
+++ b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/Metadata.java
@@ -37,7 +37,7 @@ public class Metadata {
}
/**
- * Returns full table name in the form <db_name>.<table_name>
+ * Returns full table name in the form db_name.table_name
*/
@Override
public String toString() {
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/MetadataFetcher.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/MetadataFetcher.java b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/MetadataFetcher.java
index cc3ed30..5ed9b24 100644
--- a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/MetadataFetcher.java
+++ b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/MetadataFetcher.java
@@ -16,10 +16,10 @@ public abstract class MetadataFetcher {
/**
* Gets a metadata of a given table
- *
+ *
* @param tableName table name
* @return metadata of given table
- * @throws Exception
+ * @throws Exception if metadata information could not be retrieved
*/
public abstract Metadata getTableMetadata(String tableName) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/ReadAccessor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/ReadAccessor.java b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/ReadAccessor.java
index 6aa42a3..85469fb 100644
--- a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/ReadAccessor.java
+++ b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/ReadAccessor.java
@@ -8,7 +8,7 @@ public interface ReadAccessor {
* Opens the resource for reading.
*
* @return true if the resource is successfully opened
- * @throws Exception
+ * @throws Exception if opening the resource failed
*/
boolean openForRead() throws Exception;
@@ -16,14 +16,14 @@ public interface ReadAccessor {
* Reads the next object.
*
* @return the object which was read
- * @throws Exception
+ * @throws Exception if reading from the resource failed
*/
OneRow readNextObject() throws Exception;
/**
* Closes the resource.
*
- * @throws Exception
+ * @throws Exception if closing the resource failed
*/
void closeForRead() throws Exception;
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/ReadResolver.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/ReadResolver.java b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/ReadResolver.java
index 109ec09..2fe850d 100644
--- a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/ReadResolver.java
+++ b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/ReadResolver.java
@@ -12,7 +12,7 @@ public interface ReadResolver {
*
* @param row the row to get the fields from
* @return the {@link OneField} list of one row.
- * @throws Exception
+ * @throws Exception if decomposing the row into fields failed
*/
List<OneField> getFields(OneRow row) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/WriteAccessor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/WriteAccessor.java b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/WriteAccessor.java
index da49e50..85b2535 100644
--- a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/WriteAccessor.java
+++ b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/WriteAccessor.java
@@ -8,7 +8,7 @@ public interface WriteAccessor {
* Opens the resource for write.
*
* @return true if the resource is successfully opened
- * @throws Exception
+ * @throws Exception if opening the resource failed
*/
boolean openForWrite() throws Exception;
@@ -17,14 +17,14 @@ public interface WriteAccessor {
*
* @param onerow the object to be written
* @return true if the write succeeded
- * @throws Exception
+ * @throws Exception writing to the resource failed
*/
boolean writeNextObject(OneRow onerow) throws Exception;
/**
* Closes the resource for write.
*
- * @throws Exception
+ * @throws Exception if closing the resource failed
*/
void closeForWrite() throws Exception;
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/WriteResolver.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/WriteResolver.java b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/WriteResolver.java
index 63dc692..1749572 100644
--- a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/WriteResolver.java
+++ b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/WriteResolver.java
@@ -13,7 +13,7 @@ public interface WriteResolver {
*
* @param record list of {@link OneField}
* @return the constructed {@link OneRow}
- * @throws Exception
+ * @throws Exception if constructing a row from the fields failed
*/
OneRow setFields(List<OneField> record) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/utilities/ColumnDescriptor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/utilities/ColumnDescriptor.java b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/utilities/ColumnDescriptor.java
index 4e02896..70b2502 100644
--- a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/utilities/ColumnDescriptor.java
+++ b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/utilities/ColumnDescriptor.java
@@ -5,13 +5,14 @@ package com.pivotal.pxf.api.utilities;
* Currently it means a name, a type id (HAWQ/GPDB OID), a type name and column index.
*/
public class ColumnDescriptor {
-
+
int gpdbColumnTypeCode;
String gpdbColumnName;
String gpdbColumnTypeName;
int gpdbColumnIndex;
+
/**
- * Reserved word for a table record key.
+ * Reserved word for a table record key.
* A field with this name will be treated as record key.
*/
public static final String RECORD_KEY_NAME = "recordkey";
@@ -59,11 +60,15 @@ public class ColumnDescriptor {
return gpdbColumnTypeName;
}
- /** Returns <tt>true</tt> if {@link #gpdbColumnName} is a {@link #RECORD_KEY_NAME}. */
+ /**
+ * Returns <tt>true</tt> if {@link #gpdbColumnName} is a {@link #RECORD_KEY_NAME}.
+ *
+ * @return whether column is a record key column
+ */
public boolean isKeyColumn() {
return RECORD_KEY_NAME.equalsIgnoreCase(gpdbColumnName);
}
-
+
@Override
public String toString() {
return "ColumnDescriptor [gpdbColumnTypeCode=" + gpdbColumnTypeCode
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/utilities/InputData.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/utilities/InputData.java b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/utilities/InputData.java
index c8133c7..f7c2f78 100644
--- a/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/utilities/InputData.java
+++ b/pxf/pxf-api/src/main/java/com/pivotal/pxf/api/utilities/InputData.java
@@ -10,7 +10,7 @@ import java.util.*;
* coming from client applications, such as Hawq.
*/
public class InputData {
-
+
public static final int INVALID_SPLIT_IDX = -1;
private static final Log LOG = LogFactory.getLog(InputData.class);
@@ -32,22 +32,22 @@ public class InputData {
protected int dataFragment; /* should be deprecated */
/**
- * When false the bridge has to run in synchronized mode.
- * default value - true.
+ * When false the bridge has to run in synchronized mode. default value -
+ * true.
*/
protected boolean threadSafe;
- /**
- * The name of the recordkey column. It can appear in any location in the
- * columns list. By specifying the recordkey column, the user declares that
- * he is interested to receive for every record retrieved also the the
- * recordkey in the database. The recordkey is present in HBase table (it is
- * called rowkey), and in sequence files. When the HDFS storage element
- * queried will not have a recordkey and the user will still specify it in
- * the "create external table" statement, then the values for this field
- * will be null. This field will always be the first field in the tuple
- * returned.
- */
+ /**
+ * The name of the recordkey column. It can appear in any location in the
+ * columns list. By specifying the recordkey column, the user declares that
+ * he is interested to receive for every record retrieved also the the
+ * recordkey in the database. The recordkey is present in HBase table (it is
+ * called rowkey), and in sequence files. When the HDFS storage element
+ * queried will not have a recordkey and the user will still specify it in
+ * the "create external table" statement, then the values for this field
+ * will be null. This field will always be the first field in the tuple
+ * returned.
+ */
protected ColumnDescriptor recordkeyColumn;
/**
@@ -55,10 +55,10 @@ public class InputData {
*/
public InputData() {
}
-
+
/**
- * Constructs an InputData from a copy.
- * Used to create from an extending class.
+ * Constructs an InputData from a copy. Used to create from an extending
+ * class.
*
* @param copy the input data to copy
*/
@@ -89,122 +89,183 @@ public class InputData {
* @param userProp the lookup user property
* @return property value as a String
*/
- public String getUserProperty(String userProp) {
+ public String getUserProperty(String userProp) {
return requestParametersMap.get("X-GP-" + userProp.toUpperCase());
}
/**
- * set the byte serialization of a fragment meta data
+ * Sets the byte serialization of a fragment meta data.
+ *
* @param location start, len, and location of the fragment
*/
public void setFragmentMetadata(byte[] location) {
this.fragmentMetadata = location;
}
- /** the byte serialization of a data fragment */
+ /**
+ * The byte serialization of a data fragment.
+ *
+ * @return serialized fragment metadata
+ */
public byte[] getFragmentMetadata() {
return fragmentMetadata;
}
- /**
- * Gets any custom user data that may have been passed from the
- * fragmenter. Will mostly be used by the accessor or resolver.
+ /**
+ * Gets any custom user data that may have been passed from the fragmenter.
+ * Will mostly be used by the accessor or resolver.
+ *
+ * @return fragment user data
*/
public byte[] getFragmentUserData() {
return userData;
}
-
- /**
- * Sets any custom user data that needs to be shared across plugins.
- * Will mostly be set by the fragmenter.
+
+ /**
+ * Sets any custom user data that needs to be shared across plugins. Will
+ * mostly be set by the fragmenter.
+ *
+ * @param userData user data
*/
public void setFragmentUserData(byte[] userData) {
this.userData = userData;
}
- /** Returns the number of segments in GP. */
+ /**
+ * Returns the number of segments in HAWQ.
+ *
+ * @return number of segments
+ */
public int getTotalSegments() {
return totalSegments;
}
- /** Returns the current segment ID. */
+ /**
+ * Returns the current segment ID in HAWQ.
+ *
+ * @return current segment ID
+ */
public int getSegmentId() {
return segmentId;
}
- /** Returns true if there is a filter string to parse. */
+ /**
+ * Returns true if there is a filter string to parse.
+ *
+ * @return whether there is a filter string
+ */
public boolean hasFilter() {
return filterStringValid;
}
- /** Returns the filter string, <tt>null</tt> if #hasFilter is <tt>false</tt> */
+ /**
+ * Returns the filter string, <tt>null</tt> if #hasFilter is <tt>false</tt>.
+ *
+ * @return the filter string or null
+ */
public String getFilterString() {
return filterString;
}
- /** Returns tuple description. */
+ /**
+ * Returns tuple description.
+ *
+ * @return tuple description
+ */
public ArrayList<ColumnDescriptor> getTupleDescription() {
return tupleDescription;
}
- /** Returns the number of columns in tuple description. */
+ /**
+ * Returns the number of columns in tuple description.
+ *
+ * @return number of columns
+ */
public int getColumns() {
return tupleDescription.size();
}
- /** Returns column index from tuple description. */
+ /**
+ * Returns column index from tuple description.
+ *
+ * @param index index of column
+ * @return column by index
+ */
public ColumnDescriptor getColumn(int index) {
return tupleDescription.get(index);
}
- /**
- * Returns the column descriptor of the recordkey column. If the recordkey
- * column was not specified by the user in the create table statement will
- * return null.
- */
+ /**
+ * Returns the column descriptor of the recordkey column. If the recordkey
+ * column was not specified by the user in the create table statement will
+ * return null.
+ *
+ * @return column of record key or null
+ */
public ColumnDescriptor getRecordkeyColumn() {
return recordkeyColumn;
}
- /** Returns the data source of the required resource (i.e a file path or a table name). */
+ /**
+ * Returns the data source of the required resource (i.e a file path or a
+ * table name).
+ *
+ * @return data source
+ */
public String getDataSource() {
return dataSource;
}
- /** Sets the data source for the required resource */
+ /**
+ * Sets the data source for the required resource.
+ *
+ * @param dataSource data source to be set
+ */
public void setDataSource(String dataSource) {
this.dataSource = dataSource;
}
- /** Returns the ClassName for the java class that was defined as Accessor */
+ /**
+ * Returns the ClassName for the java class that was defined as Accessor.
+ *
+ * @return class name for Accessor
+ */
public String getAccessor() {
return accessor;
}
- /** Returns the ClassName for the java class that was defined as Resolver */
+ /**
+ * Returns the ClassName for the java class that was defined as Resolver.
+ *
+ * @return class name for Resolver
+ */
public String getResolver() {
return resolver;
}
- /**
- * Returns the ClassName for the java class that was defined as Fragmenter
- * or null if no fragmenter was defined
- */
+ /**
+ * Returns the ClassName for the java class that was defined as Fragmenter
+ * or null if no fragmenter was defined.
+ *
+ * @return class name for Fragmenter or null
+ */
public String getFragmenter() {
- return fragmenter;
+ return fragmenter;
}
- /**
- * Returns the ClassName for the java class that was defined as Analyzer or
- * null if no analyzer was defined
- */
+ /**
+ * Returns the ClassName for the java class that was defined as Analyzer or
+ * null if no analyzer was defined.
+ *
+ * @return class name for Analyzer or null
+ */
public String getAnalyzer() {
- return analyzer;
+ return analyzer;
}
/**
- * Returns the contents of pxf_remote_service_login set in Hawq.
- * Should the user set it to an empty string this function will return null.
+ * Returns the contents of pxf_remote_service_login set in Hawq. Should the
+ * user set it to an empty string this function will return null.
*
* @return remote login details if set, null otherwise
*/
@@ -213,8 +274,8 @@ public class InputData {
}
/**
- * Returns the contents of pxf_remote_service_secret set in Hawq.
- * Should the user set it to an empty string this function will return null.
+ * Returns the contents of pxf_remote_service_secret set in Hawq. Should the
+ * user set it to an empty string this function will return null.
*
* @return remote password if set, null otherwise
*/
@@ -222,16 +283,24 @@ public class InputData {
return remoteSecret;
}
+ /**
+ * Returns whether this request is thread safe.
+ * If it is not, request will be handled consequentially and not in parallel.
+ *
+ * @return whether the request is thread safe
+ */
public boolean isThreadSafe() {
return threadSafe;
}
- /**
- * Returns a data fragment index. plan to deprecate it in favor of using
- * getFragmentMetadata().
- */
- public int getDataFragment() {
- return dataFragment;
- }
+ /**
+ * Returns a data fragment index. plan to deprecate it in favor of using
+ * getFragmentMetadata().
+ *
+ * @return data fragment index
+ */
+ public int getDataFragment() {
+ return dataFragment;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/HBaseFilterBuilder.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/HBaseFilterBuilder.java b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/HBaseFilterBuilder.java
index 5bce949..82b920b 100644
--- a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/HBaseFilterBuilder.java
+++ b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/HBaseFilterBuilder.java
@@ -45,6 +45,10 @@ public class HBaseFilterBuilder implements FilterParser.FilterBuilder {
/**
* Translates a filterString into a HBase {@link Filter} object.
+ *
+ * @param filterString filter string
+ * @return filter object
+ * @throws Exception if parsing failed
*/
public Filter getFilterObject(String filterString) throws Exception {
FilterParser parser = new FilterParser(this);
@@ -59,9 +63,11 @@ public class HBaseFilterBuilder implements FilterParser.FilterBuilder {
/**
* Returns the startKey for scanning the HBase table.
- * If the user specified a {@code > / >=} operation
+ * If the user specified a {@code > / >=} operation
* on a textual row key column, this value will be returned.
* Otherwise, the start of table.
+ *
+ * @return start key for scanning HBase table
*/
public byte[] startKey() {
return startKey;
@@ -72,6 +78,8 @@ public class HBaseFilterBuilder implements FilterParser.FilterBuilder {
* If the user specified a {@code < / <=} operation
* on a textual row key column, this value will be returned.
* Otherwise, the end of table.
+ *
+ * @return end key for scanning HBase table
*/
public byte[] endKey() {
return endKey;
@@ -87,7 +95,7 @@ public class HBaseFilterBuilder implements FilterParser.FilterBuilder {
* The only supported operation is {@code AND}. </li>
* </ol>
* <p>
- * This function is called by {@link FilterParser},
+ * This function is called by {@link FilterParser},
* each time the parser comes across an operator.
*/
@Override
@@ -138,7 +146,7 @@ public class HBaseFilterBuilder implements FilterParser.FilterBuilder {
constant.constant());
/**
- * If row key is of type TEXT, allow filter in start/stop row key API in
+ * If row key is of type TEXT, allow filter in start/stop row key API in
* HBaseAccessor/Scan object.
*/
if (textualRowKey(hbaseColumn)) {
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/HBaseResolver.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/HBaseResolver.java b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/HBaseResolver.java
index 3815dc4..0b4e364 100644
--- a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/HBaseResolver.java
+++ b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/HBaseResolver.java
@@ -17,7 +17,7 @@ import java.util.List;
/**
* Record resolver for HBase.
- *
+ *
* The class is responsible to convert rows from HBase scans (returned as {@link Result} objects)
* into a List of {@link OneField} objects.
* That also includes the conversion process of each HBase column's value into its HAWQ assigned type.
@@ -26,9 +26,11 @@ import java.util.List;
*/
public class HBaseResolver extends Plugin implements ReadResolver {
private HBaseTupleDescription tupleDescription;
-
+
/**
* Constructs a resolver and initializes the table's tuple description.
+ *
+ * @param input query information, contains HBase table name and filter
*/
public HBaseResolver(InputData input) {
super(input);
@@ -37,9 +39,9 @@ public class HBaseResolver extends Plugin implements ReadResolver {
/**
* Splits an HBase {@link Result} object into a list of {@link OneField},
- * based on the table's tuple description.
+ * based on the table's tuple description.
* Each field is converted from HBase bytes into its column description type.
- *
+ *
* @return list of fields
*/
@Override
@@ -70,7 +72,7 @@ public class HBaseResolver extends Plugin implements ReadResolver {
/**
* Converts given byte array value to the matching java object, according to
* the given type code.
- *
+ *
* @param typeCode ColumnDescriptor type id
* @param typeName type name. Used for error messages
* @param val value to be converted
@@ -127,7 +129,7 @@ public class HBaseResolver extends Plugin implements ReadResolver {
/**
* Returns the value of a column from a Result object.
- *
+ *
* @param result HBase table row
* @param column HBase column to be retrieved
* @return HBase column value
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseColumnDescriptor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseColumnDescriptor.java b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseColumnDescriptor.java
index 3057d72..cf5a897 100644
--- a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseColumnDescriptor.java
+++ b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseColumnDescriptor.java
@@ -14,6 +14,8 @@ public class HBaseColumnDescriptor extends ColumnDescriptor {
/**
* Constructs a column descriptor using the given copy's column name.
+ *
+ * @param copy column descriptor to be copied
*/
public HBaseColumnDescriptor(ColumnDescriptor copy) {
this(copy, copy.columnName().getBytes());
@@ -28,9 +30,9 @@ public class HBaseColumnDescriptor extends ColumnDescriptor {
* <li>recordkey - Row key column (case insensitive).</li>
* </ol>
* <p>
- * For recordkey, no HBase name is created.
- *
- * @param copy column descriptor
+ * For recordkey, no HBase name is created.
+ *
+ * @param copy column descriptor
* @param newColumnName HBase column name - can be different than the given column descriptor name.
*/
public HBaseColumnDescriptor(ColumnDescriptor copy, byte[] newColumnName) {
@@ -49,6 +51,8 @@ public class HBaseColumnDescriptor extends ColumnDescriptor {
/**
* Returns the family column name.
* (E.g. "cf1:q2" will return "cf1")
+ *
+ * @return family column name
*/
public byte[] columnFamilyBytes() {
return columnFamily;
@@ -57,6 +61,8 @@ public class HBaseColumnDescriptor extends ColumnDescriptor {
/**
* Returns the qualifier column name.
* (E.g. "cf1:q2" will return "q2")
+ *
+ * @return qualifier column name
*/
public byte[] qualifierBytes() {
return qualifier;
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseIntegerComparator.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseIntegerComparator.java b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseIntegerComparator.java
index 6a9cc4f..a935534 100644
--- a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseIntegerComparator.java
+++ b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseIntegerComparator.java
@@ -14,7 +14,7 @@ import com.google.protobuf.InvalidProtocolBufferException;
* <p>
* To use with HBase it must reside in the classpath of every region server.
* <p>
- * It converts a value into {@link Long} before comparing.
+ * It converts a value into {@link Long} before comparing.
* The filter is good for any integer numeric comparison i.e. integer, bigint, smallint.
* <p>
* according to HBase 0.96 requirements, this must serialized using Protocol Buffers
@@ -25,7 +25,7 @@ import com.google.protobuf.InvalidProtocolBufferException;
public class HBaseIntegerComparator extends ByteArrayComparable {
private Long val;
-
+
public HBaseIntegerComparator(Long inVal) {
super(Bytes.toBytes(inVal));
this.val = inVal;
@@ -33,6 +33,10 @@ public class HBaseIntegerComparator extends ByteArrayComparable {
/**
* The comparison function. Currently uses {@link Long#parseLong(String)}.
+ *
+ * @return 0 if equal;
+ * a value less than 0 if row value is less than filter value;
+ * and a value greater than 0 if the row value is greater than the filter value.
*/
@Override
public int compareTo(byte[] value, int offset, int length) {
@@ -42,7 +46,7 @@ public class HBaseIntegerComparator extends ByteArrayComparable {
if (length == 0)
return 1; // empty line, can't compare.
- /**
+ /**
* TODO optimize by parsing the bytes directly.
* Maybe we can even determine if it is an int or a string encoded.
*/
@@ -53,6 +57,8 @@ public class HBaseIntegerComparator extends ByteArrayComparable {
/**
* Returns the comparator serialized using Protocol Buffers.
+ *
+ * @return serialized comparator
*/
@Override
public byte[] toByteArray() {
@@ -64,12 +70,12 @@ public class HBaseIntegerComparator extends ByteArrayComparable {
/**
* Hides ("overrides") a static method in {@link ByteArrayComparable}.
* This method will be called in deserialization.
- *
+ *
* @param pbBytes
* A pb serialized instance
* @return An instance of {@link HBaseIntegerComparator} made from
* <code>bytes</code>
- * @throws DeserializationException
+ * @throws DeserializationException if deserialization of bytes to Protocol Buffers failed
* @see #toByteArray
*/
public static ByteArrayComparable parseFrom(final byte[] pbBytes)
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseLookupTable.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseLookupTable.java b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseLookupTable.java
index b522479..459e97c 100644
--- a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseLookupTable.java
+++ b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseLookupTable.java
@@ -19,8 +19,9 @@ import java.util.Map;
* HBaseLookupTable will load a table's lookup information
* from HBase pxflookup table if exists.<br>
* This table holds mappings between HAWQ column names (key) and HBase column names (value).<br>
- * E.g. for an HBase table "hbase_table", mappings between HAWQ column names and HBase column name
- * "hawq1" -> "cf1:hbase1" and "hawq2" -> "cf1:hbase2" will be:<br>
+ * E.g. for an HBase table "hbase_table", mappings between HAWQ column names and HBase column names,
+ * when <code>"hawq1"</code> is mapped to <code>"cf1:hbase1"</code> and
+ * <code>"hawq2"</code> is mapped to <code>"cf1:hbase2"</code>, will be:<br>
* <pre>
* ROW COLUMN+CELL
* hbase_table column=mapping:hawq1, value=cf1:hbase1
@@ -47,6 +48,7 @@ public class HBaseLookupTable implements Closeable {
* Constructs a connector to HBase lookup table.
* Requires calling {@link #close()} to close {@link HBaseAdmin} instance.
*
+ * @param conf HBase configuration
* @throws IOException when initializing HBaseAdmin fails
*/
public HBaseLookupTable(Configuration conf) throws Exception {
@@ -93,6 +95,8 @@ public class HBaseLookupTable implements Closeable {
/**
* Returns true if {@link #LOOKUPTABLENAME} is available and enabled.
+ *
+ * @return whether lookup table is valid
*/
private boolean lookupTableValid() throws IOException {
return (HBaseUtilities.isTableAvailable(admin, LOOKUPTABLENAME) &&
@@ -101,6 +105,8 @@ public class HBaseLookupTable implements Closeable {
/**
* Returns true if {@link #LOOKUPTABLENAME} has {@value #LOOKUPCOLUMNFAMILY} family.
+ *
+ * @return whether lookup has expected column family name
*/
private boolean lookupHasCorrectStructure() throws IOException {
HTableDescriptor htd = admin.getTableDescriptor(TableName.valueOf(LOOKUPTABLENAME));
@@ -109,6 +115,8 @@ public class HBaseLookupTable implements Closeable {
/**
* Loads table name mappings from {@link #LOOKUPTABLENAME} lookup table.
+ *
+ * @param tableName table name
*/
private void loadTableMappings(String tableName) throws IOException {
openLookupTable();
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseTupleDescription.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseTupleDescription.java b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseTupleDescription.java
index e91ff8c..74f3ec3 100644
--- a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseTupleDescription.java
+++ b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseTupleDescription.java
@@ -23,7 +23,7 @@ public class HBaseTupleDescription {
/**
* Constructs tuple description of the HBase table.
- *
+ *
* @param conf data containing table tuple description
*/
public HBaseTupleDescription(InputData conf) {
@@ -33,6 +33,8 @@ public class HBaseTupleDescription {
/**
* Returns the number of fields.
+ *
+ * @return number of fields
*/
public int columns() {
return tupleDescription.size();
@@ -40,7 +42,7 @@ public class HBaseTupleDescription {
/**
* Returns the column description of index column.
- *
+ *
* @param index column index to be returned
* @return column description
*/
@@ -57,7 +59,7 @@ public class HBaseTupleDescription {
/**
* Loads user information from fragmenter.
* The data contains optional table mappings from the lookup table,
- * between field names in HAWQ table and in the HBase table.
+ * between field names in HAWQ table and in the HBase table.
*/
@SuppressWarnings("unchecked")
private void loadUserData() {
@@ -87,7 +89,7 @@ public class HBaseTupleDescription {
/**
* Returns the {@link #HBaseColumnDescriptor} for given column.
* If the column has a lookup table mapping, the HBase column name is used.
- *
+ *
* @param column HAWQ column description
* @return matching HBase column description
*/
@@ -108,7 +110,7 @@ public class HBaseTupleDescription {
/**
* Returns the HBase name mapping for the given column name.
- *
+ *
* @param column HAWQ column description
* @return HBase name for the column
*/
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseUtilities.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseUtilities.java b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseUtilities.java
index 3edf07d..b338c02 100644
--- a/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseUtilities.java
+++ b/pxf/pxf-hbase/src/main/java/com/pivotal/pxf/plugins/hbase/utilities/HBaseUtilities.java
@@ -31,7 +31,7 @@ public class HBaseUtilities {
* @param hbaseAdmin HBase admin, must be initialized
* @param tableName table name
* @return true if table exists
- * @throws IOException
+ * @throws IOException if a remote or network exception occurs when connecting to HBase
*/
public static boolean isTableAvailable(Admin hbaseAdmin, String tableName) throws IOException {
TableName name = TableName.valueOf(tableName);
@@ -42,9 +42,9 @@ public class HBaseUtilities {
/**
* Closes HBase admin and connection if they are open.
*
- * @param hbaseAdmin
- * @param hbaseConnection
- * @throws IOException
+ * @param hbaseAdmin HBase admin
+ * @param hbaseConnection HBase connection
+ * @throws IOException if an I/O error occurs when connecting to HBase
*/
public static void closeConnection(Admin hbaseAdmin, Connection hbaseConnection) throws IOException {
if (hbaseAdmin != null) {
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/AvroFileAccessor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/AvroFileAccessor.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/AvroFileAccessor.java
index a496698..1a667f4 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/AvroFileAccessor.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/AvroFileAccessor.java
@@ -2,6 +2,7 @@ package com.pivotal.pxf.plugins.hdfs;
import com.pivotal.pxf.api.OneRow;
import com.pivotal.pxf.api.utilities.InputData;
+
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.mapred.*;
@@ -21,10 +22,11 @@ public class AvroFileAccessor extends HdfsSplittableDataAccessor {
private AvroWrapper<GenericRecord> avroWrapper = null;
/**
- * Constructs a AvroFileAccessor that creates the job configuration and
+ * Constructs a AvroFileAccessor that creates the job configuration and
* accesses the avro file to fetch the avro schema
- *
+ *
* @param input all input parameters coming from the client
+ * @throws Exception if getting the avro schema fails
*/
public AvroFileAccessor(InputData input) throws Exception {
// 1. Call the base class
@@ -44,15 +46,15 @@ public class AvroFileAccessor extends HdfsSplittableDataAccessor {
@Override
protected Object getReader(JobConf jobConf, InputSplit split) throws IOException {
- return new AvroRecordReader(jobConf, (FileSplit) split);
+ return new AvroRecordReader<Object>(jobConf, (FileSplit) split);
}
/**
* readNextObject
- * The AVRO accessor is currently the only specialized accessor that
- * overrides this method. This happens, because the special
+ * The AVRO accessor is currently the only specialized accessor that
+ * overrides this method. This happens, because the special
* AvroRecordReader.next() semantics (use of the AvroWrapper), so it
- * cannot use the RecordReader's default implementation in
+ * cannot use the RecordReader's default implementation in
* SplittableFileAccessor
*/
@Override
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/AvroResolver.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/AvroResolver.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/AvroResolver.java
index 3560404..8b86479 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/AvroResolver.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/AvroResolver.java
@@ -9,6 +9,7 @@ import com.pivotal.pxf.api.utilities.Plugin;
import com.pivotal.pxf.plugins.hdfs.utilities.DataSchemaException;
import com.pivotal.pxf.plugins.hdfs.utilities.HdfsUtilities;
import com.pivotal.pxf.plugins.hdfs.utilities.RecordkeyAdapter;
+
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
@@ -38,7 +39,8 @@ import static com.pivotal.pxf.plugins.hdfs.utilities.HdfsUtilities.getAvroSchema
public class AvroResolver extends Plugin implements ReadResolver {
private GenericRecord avroRecord = null;
private DatumReader<GenericRecord> reader = null;
- private BinaryDecoder decoder = null; // member kept to enable reuse, and thus avoid repeated allocation
+ // member kept to enable reuse, and thus avoid repeated allocation
+ private BinaryDecoder decoder = null;
private List<Schema.Field> fields = null;
private RecordkeyAdapter recordkeyAdapter = new RecordkeyAdapter();
private static final String MAPKEY_DELIM = ":";
@@ -49,19 +51,19 @@ public class AvroResolver extends Plugin implements ReadResolver {
private String recordkeyDelim;
/**
- * Constructs an AvroResolver. Initializes Avro data structure: the avro
- * record - fields information and the avro record reader. All Avro data is
+ * Constructs an AvroResolver. Initializes Avro data structure: the Avro
+ * record - fields information and the Avro record reader. All Avro data is
* build from the Avro schema, which is based on the *.avsc file that was
* passed by the user
*
* @param input all input parameters coming from the client
- * @throws IOException
+ * @throws IOException if Avro schema could not be retrieved or parsed
*/
public AvroResolver(InputData input) throws IOException {
super(input);
- Schema schema = isAvroFile()
- ? getAvroSchema(new Configuration(), input.getDataSource())
+ Schema schema = isAvroFile() ? getAvroSchema(new Configuration(),
+ input.getDataSource())
: (new Schema.Parser()).parse(openExternalSchema());
reader = new GenericDatumReader<>(schema);
@@ -75,56 +77,66 @@ public class AvroResolver extends Plugin implements ReadResolver {
: input.getUserProperty("RECORDKEY_DELIM");
}
- /*
- * getFields returns a list of the fields of one record. Each record field
- * is represented by a OneField item. OneField item contains two fields:
- * an integer representing the field type and a Java Object representing
- * the field value.
+ /**
+ * Returns a list of the fields of one record. Each record field is
+ * represented by a OneField item. OneField item contains two fields: an
+ * integer representing the field type and a Java Object representing the
+ * field value.
*/
@Override
public List<OneField> getFields(OneRow row) throws Exception {
avroRecord = makeAvroRecord(row.getData(), avroRecord);
List<OneField> record = new LinkedList<OneField>();
- int recordkeyIndex = (inputData.getRecordkeyColumn() == null) ? -1 :
- inputData.getRecordkeyColumn().columnIndex();
+ int recordkeyIndex = (inputData.getRecordkeyColumn() == null) ? -1
+ : inputData.getRecordkeyColumn().columnIndex();
int currentIndex = 0;
for (Schema.Field field : fields) {
/*
* Add the record key if exists
- */
+ */
if (currentIndex == recordkeyIndex) {
- currentIndex += recordkeyAdapter.appendRecordkeyField(record, inputData, row);
+ currentIndex += recordkeyAdapter.appendRecordkeyField(record,
+ inputData, row);
}
- currentIndex += populateRecord(record, avroRecord.get(field.name()), field.schema());
+ currentIndex += populateRecord(record,
+ avroRecord.get(field.name()), field.schema());
}
return record;
}
- /*
- * Test if the Avro records are residing inside an AVRO file.
- * If the Avro records are not residing inside an AVRO file, then
- * they may reside inside a sequence file, regular file, ...
+ /**
+ * Tests if the Avro records are residing inside an AVRO file. If the Avro
+ * records are not residing inside an AVRO file, then they may reside inside
+ * a sequence file, regular file, ...
+ *
+ * @return whether the resource is an Avro file
*/
boolean isAvroFile() {
return inputData.getAccessor().toLowerCase().contains("avro");
}
- /*
+ /**
* The record can arrive from one out of two different sources: a sequence
* file or an AVRO file. If it comes from an AVRO file, then it was already
- * obtained as a GenericRecord when when it was fetched from the file with
- * the AvroRecorReader so in this case a cast is enough. On the other hand,
- * if the source is a sequence file, then the input parameter obj hides a
- * bytes [] buffer which is in fact one Avro record serialized. Here, we
- * build the Avro record from the flat buffer, using the AVRO API. Then
- * (for both cases) in the remaining functions we build a List<OneField>
- * record from the Avro record
+ * obtained as a {@link GenericRecord} when when it was fetched from the
+ * file with the {@link AvroRecorReader} so in this case a cast is enough.
+ * On the other hand, if the source is a sequence file, then the input
+ * parameter obj hides a bytes [] buffer which is in fact one Avro record
+ * serialized. Here, we build the Avro record from the flat buffer, using
+ * the AVRO API. Then (for both cases) in the remaining functions we build a
+ * {@code List<OneField>} record from the Avro record.
+ *
+ * @param obj object holding an Avro record
+ * @param reuseRecord Avro record to be reused to create new record from obj
+ * @return Avro record
+ * @throws IOException if creating the Avro record from byte array failed
*/
- GenericRecord makeAvroRecord(Object obj, GenericRecord reuseRecord) throws IOException {
+ GenericRecord makeAvroRecord(Object obj, GenericRecord reuseRecord)
+ throws IOException {
if (isAvroFile()) {
return (GenericRecord) obj;
} else {
@@ -134,12 +146,18 @@ public class AvroResolver extends Plugin implements ReadResolver {
}
}
- /*
+ /**
* For a given field in the Avro record we extract its value and insert it
- * into the output List<OneField> record. An Avro field can be a primitive
- * type or an array type.
+ * into the output {@code List<OneField>} record. An Avro field can be a
+ * primitive type or an array type.
+ *
+ * @param record list of fields to be populated
+ * @param fieldValue field value
+ * @param fieldSchema field schema
+ * @return the number of populated fields
*/
- int populateRecord(List<OneField> record, Object fieldValue, Schema fieldSchema) throws IllegalAccessException {
+ int populateRecord(List<OneField> record, Object fieldValue,
+ Schema fieldSchema) {
Schema.Type fieldType = fieldSchema.getType();
int ret = 0;
@@ -149,28 +167,38 @@ public class AvroResolver extends Plugin implements ReadResolver {
case ARRAY:
List<OneField> listRecord = new LinkedList<>();
ret = setArrayField(listRecord, fieldValue, fieldSchema);
- addOneFieldToRecord(record, TEXT, String.format("[%s]", HdfsUtilities.toString(listRecord, collectionDelim)));
+ addOneFieldToRecord(record, TEXT, String.format("[%s]",
+ HdfsUtilities.toString(listRecord, collectionDelim)));
break;
case MAP:
List<OneField> mapRecord = new LinkedList<>();
ret = setMapField(mapRecord, fieldValue, fieldSchema);
- addOneFieldToRecord(record, TEXT, String.format("{%s}", HdfsUtilities.toString(mapRecord, collectionDelim)));
+ addOneFieldToRecord(record, TEXT, String.format("{%s}",
+ HdfsUtilities.toString(mapRecord, collectionDelim)));
break;
case RECORD:
List<OneField> recRecord = new LinkedList<>();
ret = setRecordField(recRecord, fieldValue, fieldSchema);
- addOneFieldToRecord(record, TEXT, String.format("{%s}", HdfsUtilities.toString(recRecord, collectionDelim)));
+ addOneFieldToRecord(record, TEXT, String.format("{%s}",
+ HdfsUtilities.toString(recRecord, collectionDelim)));
break;
case UNION:
- /* When an Avro field is actually a union, we resolve the type of the union
- * element, and delegate the record update via recursion
- */
- int unionIndex = GenericData.get().resolveUnion(fieldSchema, fieldValue);
- /** Retrieve index of the non null data type from the type array if value is null */
+ /*
+ * When an Avro field is actually a union, we resolve the type
+ * of the union element, and delegate the record update via
+ * recursion
+ */
+ int unionIndex = GenericData.get().resolveUnion(fieldSchema,
+ fieldValue);
+ /**
+ * Retrieve index of the non null data type from the type array
+ * if value is null
+ */
if (fieldValue == null) {
unionIndex ^= 1;
}
- ret = populateRecord(record, fieldValue, fieldSchema.getTypes().get(unionIndex));
+ ret = populateRecord(record, fieldValue,
+ fieldSchema.getTypes().get(unionIndex));
break;
case ENUM:
value = (fieldValue != null) ? fieldValue : null;
@@ -185,7 +213,8 @@ public class AvroResolver extends Plugin implements ReadResolver {
ret = addOneFieldToRecord(record, FLOAT8, value);
break;
case STRING:
- value = (fieldValue != null) ? String.format("%s", fieldValue) : null;
+ value = (fieldValue != null) ? String.format("%s", fieldValue)
+ : null;
ret = addOneFieldToRecord(record, TEXT, value);
break;
case FLOAT:
@@ -214,59 +243,77 @@ public class AvroResolver extends Plugin implements ReadResolver {
return ret;
}
- /*
+ /**
* When an Avro field is actually a record, we iterate through each field
* for each entry, the field name and value are added to a local record
- * List<OneField> complexRecord with the necessary delimiter
- * we create an object of type OneField and insert it into the output List<OneField> record
- *
+ * {@code List<OneField>} complexRecord with the necessary delimiter we
+ * create an object of type OneField and insert it into the output
+ * {@code List<OneField>} record.
+ *
+ * @param record list of fields to be populated
+ * @param value field value
+ * @param recSchema record schema
+ * @return number of populated fields
*/
- int setRecordField(List<OneField> record, Object value, Schema recSchema) throws IllegalAccessException {
+ int setRecordField(List<OneField> record, Object value, Schema recSchema) {
- GenericRecord rec = ((GenericData.Record)value);
+ GenericRecord rec = ((GenericData.Record) value);
Schema fieldKeySchema = Schema.create(Schema.Type.STRING);
int currentIndex = 0;
- for(Schema.Field field: recSchema.getFields()) {
+ for (Schema.Field field : recSchema.getFields()) {
Schema fieldSchema = field.schema();
- Schema.Type fieldType = fieldSchema.getType();
Object fieldValue = rec.get(field.name());
List<OneField> complexRecord = new LinkedList<>();
populateRecord(complexRecord, field.name(), fieldKeySchema);
populateRecord(complexRecord, fieldValue, fieldSchema);
- addOneFieldToRecord(record, TEXT, HdfsUtilities.toString(complexRecord, recordkeyDelim));
+ addOneFieldToRecord(record, TEXT,
+ HdfsUtilities.toString(complexRecord, recordkeyDelim));
currentIndex++;
}
return currentIndex;
}
- /*
- * When an Avro field is actually a map, we resolve the type of the map value
- * For each entry, the field name and value are added to a local record
- * we create an object of type OneField and insert it into the output List<OneField> record
+ /**
+ * When an Avro field is actually a map, we resolve the type of the map
+ * value For each entry, the field name and value are added to a local
+ * record we create an object of type OneField and insert it into the output
+ * {@code List<OneField>} record.
+ *
* Unchecked warning is suppressed to enable us to cast fieldValue to a Map.
* (since the value schema has been identified to me of type map)
+ *
+ * @param record list of fields to be populated
+ * @param fieldValue field value
+ * @param mapSchema map schema
+ * @return number of populated fields
*/
@SuppressWarnings("unchecked")
- int setMapField(List<OneField> record, Object fieldValue, Schema mapSchema) throws IllegalAccessException {
+ int setMapField(List<OneField> record, Object fieldValue, Schema mapSchema) {
Schema keySchema = Schema.create(Schema.Type.STRING);
Schema valueSchema = mapSchema.getValueType();
- Map<String, ?> avroMap = ((Map<String, ?>)fieldValue);
+ Map<String, ?> avroMap = ((Map<String, ?>) fieldValue);
for (Map.Entry<String, ?> entry : avroMap.entrySet()) {
List<OneField> complexRecord = new LinkedList<>();
populateRecord(complexRecord, entry.getKey(), keySchema);
populateRecord(complexRecord, entry.getValue(), valueSchema);
- addOneFieldToRecord(record, TEXT, HdfsUtilities.toString(complexRecord, mapkeyDelim));
+ addOneFieldToRecord(record, TEXT,
+ HdfsUtilities.toString(complexRecord, mapkeyDelim));
}
return avroMap.size();
}
- /*
+ /**
* When an Avro field is actually an array, we resolve the type of the array
* element, and for each element in the Avro array, we recursively invoke
- * the population of List<OneField> record
+ * the population of {@code List<OneField>} record.
*
+ * @param record list of fields to be populated
+ * @param fieldValue field value
+ * @param arraySchema array schema
+ * @return number of populated fields
*/
- int setArrayField(List<OneField> record, Object fieldValue, Schema arraySchema) throws IllegalAccessException {
+ int setArrayField(List<OneField> record, Object fieldValue,
+ Schema arraySchema) {
Schema typeSchema = arraySchema.getElementType();
GenericData.Array<?> array = (GenericData.Array<?>) fieldValue;
int length = array.size();
@@ -276,21 +323,30 @@ public class AvroResolver extends Plugin implements ReadResolver {
return length;
}
- /*
- * Creates the OneField object and adds it to the output List<OneField> record.
- * Strings and byte arrays are held inside special types in the Avro record so we
- * transfer them to standard types in order to enable their insertion in the
- * GPDBWritable instance
+ /**
+ * Creates the {@link OneField} object and adds it to the output {@code List<OneField>}
+ * record. Strings and byte arrays are held inside special types in the Avro
+ * record so we transfer them to standard types in order to enable their
+ * insertion in the GPDBWritable instance.
+ *
+ * @param record list of fields to be populated
+ * @param gpdbWritableType field type
+ * @param val field value
+ * @return 1 (number of populated fields)
*/
- int addOneFieldToRecord(List<OneField> record, DataType gpdbWritableType, Object val) {
+ int addOneFieldToRecord(List<OneField> record, DataType gpdbWritableType,
+ Object val) {
OneField oneField = new OneField();
oneField.type = gpdbWritableType.getOID();
switch (gpdbWritableType) {
case BYTEA:
- if(val instanceof ByteBuffer) {
+ if (val instanceof ByteBuffer) {
oneField.val = ((ByteBuffer) val).array();
} else {
- /** Entry point when the underlying bytearray is from a Fixed data */
+ /**
+ * Entry point when the underlying bytearray is from a Fixed
+ * data
+ */
oneField.val = ((GenericData.Fixed) val).bytes();
}
break;
@@ -303,13 +359,23 @@ public class AvroResolver extends Plugin implements ReadResolver {
return 1;
}
- InputStream openExternalSchema() throws IOException {
+ /**
+ * Opens Avro schema based on DATA-SCHEMA parameter.
+ *
+ * @return InputStream of schema file
+ * @throws DataSchemaException if schema file could not be opened
+ */
+ InputStream openExternalSchema() {
String schemaName = inputData.getUserProperty("DATA-SCHEMA");
- /** Testing that the schema name was supplied by the user - schema is an optional properly. */
+ /**
+ * Testing that the schema name was supplied by the user - schema is an
+ * optional properly.
+ */
if (schemaName == null) {
- throw new DataSchemaException(SCHEMA_NOT_INDICATED, this.getClass().getName());
+ throw new DataSchemaException(SCHEMA_NOT_INDICATED,
+ this.getClass().getName());
}
/** Testing that the schema resource exists. */
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkReader.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkReader.java b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkReader.java
index 897085c..1873d7d 100644
--- a/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkReader.java
+++ b/pxf/pxf-hdfs/src/main/java/com/pivotal/pxf/plugins/hdfs/ChunkReader.java
@@ -6,163 +6,171 @@ import java.io.InputStream;
import java.util.LinkedList;
import java.util.List;
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Writable;
import com.pivotal.pxf.plugins.hdfs.ChunkWritable;
/**
- * A class that provides a line reader from an input stream.
- * Lines are terminated by '\n' (LF)
- * EOF also terminates an otherwise unterminated line.
+ * A class that provides a line reader from an input stream. Lines are
+ * terminated by '\n' (LF) EOF also terminates an otherwise unterminated line.
*/
public class ChunkReader implements Closeable {
- public static final int DEFAULT_BUFFER_SIZE = 64*1024;
- private int bufferSize = DEFAULT_BUFFER_SIZE;
- private InputStream in;
- private byte[] buffer;
- // the number of bytes of real data in the buffer
- private int bufferLength = 0;
- // the current position in the buffer
- private int bufferPosn = 0;
- private static final byte LF = '\n';
-
- /**
- * Constructs a ChunkReader instance
- * @param in input stream
- */
- public ChunkReader(InputStream in) throws IOException {
- this.in = in;
- this.buffer = new byte[this.bufferSize];
- }
-
-
- /**
- * Close the underlying stream.
- */
- @Override
- public void close() throws IOException {
- in.close();
- }
-
- /*
- * Internal class used for holding part of a chunk brought by one read()
- * operation on the input stream. We collect seveal such nodes in a list
- * by doing several read operation until we reach the chunk size - maxBytesToConsume
- */
- private class Node {
- /* part of a chunk brought in a single inputstream.read() operation */
- public byte [] slice;
- /* the size of the slice */
- public int len;
- }
-
+ public static final int DEFAULT_BUFFER_SIZE = 64 * 1024;
+ private int bufferSize = DEFAULT_BUFFER_SIZE;
+ private InputStream in;
+ private byte[] buffer;
+ // the number of bytes of real data in the buffer
+ private int bufferLength = 0;
+ // the current position in the buffer
+ private int bufferPosn = 0;
+ private static final byte LF = '\n';
+
+ /**
+ * Constructs a ChunkReader instance
+ *
+ * @param in input stream
+ */
+ public ChunkReader(InputStream in) {
+ this.in = in;
+ this.buffer = new byte[this.bufferSize];
+ }
+
+ /**
+ * Closes the underlying stream.
+ */
+ @Override
+ public void close() throws IOException {
+ in.close();
+ }
+
+ /*
+ * Internal class used for holding part of a chunk brought by one read()
+ * operation on the input stream. We collect several such nodes in a list by
+ * doing several read operation until we reach the chunk size -
+ * maxBytesToConsume
+ */
+ private class Node {
+ /* part of a chunk brought in a single inputstream.read() operation */
+ public byte[] slice;
+ /* the size of the slice */
+ public int len;
+ }
+
+ /**
+ * Reads data in chunks of DEFAULT_CHUNK_SIZE, until we reach
+ * maxBytesToConsume.
+ *
+ * @param str - output parameter, will contain the read chunk byte array
+ * @param maxBytesToConsume - requested chunk size
+ * @return actual chunk size
+ * @throws IOException if the first byte cannot be read for any reason
+ * other than the end of the file, if the input stream has been closed,
+ * or if some other I/O error occurs.
+ */
+ public int readChunk(Writable str, int maxBytesToConsume) throws IOException
+ {
+ ChunkWritable cw = (ChunkWritable) str;
+ List<Node> list = new LinkedList<Node>();
+
+ long bytesConsumed = 0;
+
+ do {
+ if (bufferLength > 0) {
+ int remaining = bufferLength - bufferPosn;
+ Node nd = new Node();
+ nd.slice = new byte[remaining];
+ nd.len = remaining;
+ System.arraycopy(buffer, bufferPosn, nd.slice, 0, nd.len);
+ list.add(nd);
+ bytesConsumed += nd.len;
+ } else {
+ Node nd = new Node();
+ nd.slice = new byte[buffer.length];
+ nd.len = in.read(nd.slice);
+ if (nd.len <= 0) {
+ break; // EOF
+ }
+ bytesConsumed += nd.len;
+ list.add(nd);
+ }
+
+ bufferLength = bufferPosn = 0;
+
+ } while (bytesConsumed < maxBytesToConsume);
+
+ if (list.size() > 0) {
+ cw.box = new byte[(int) bytesConsumed];
+ int pos = 0;
+ for (int i = 0; i < list.size(); i++) {
+ Node n = list.get(i);
+ System.arraycopy(n.slice, 0, cw.box, pos, n.len);
+ pos += n.len;
+ }
+ }
+
+ return (int) bytesConsumed;
+ }
+
/**
- * Read data in chunks of DEFAULT_CHUNK_SIZE, until we reach maxBytesToConsume
- * @param str - output parameter, will contain the read chunk byte array
- * @param maxBytesToConsume - requested chunk size
- * @return actual chunk size
- */
- public int readChunk(Writable str, int maxBytesToConsume) throws IOException {
- ChunkWritable cw = (ChunkWritable)str;
- List<Node> list = new LinkedList<Node>();
-
- long bytesConsumed = 0;
-
- do {
- if (bufferLength > 0) {
- int remaining = bufferLength - bufferPosn;
- Node nd = new Node();
- nd.slice = new byte[remaining];
- nd.len = remaining;
- System.arraycopy(buffer, bufferPosn, nd.slice, 0, nd.len);
- list.add(nd);
- bytesConsumed += nd.len;
- } else {
- Node nd = new Node();
- nd.slice = new byte[buffer.length];
- nd.len = in.read(nd.slice);
- if (nd.len <= 0) {
- break; // EOF
- }
- bytesConsumed += nd.len;
- list.add(nd);
- }
-
- bufferLength = bufferPosn = 0;
-
- } while (bytesConsumed < maxBytesToConsume);
-
- if (list.size() > 0) {
- cw.box = new byte[(int)bytesConsumed];
- int pos = 0;
- for (int i = 0; i < list.size(); i++) {
- Node n = list.get(i);
- System.arraycopy(n.slice, 0, cw.box, pos, n.len);
- pos += n.len;
- }
- }
-
- return (int)bytesConsumed;
- }
-
- /**
- * Read a line terminated by LF.
- * @param str - output parameter, will contain the read record
- * @param maxBytesToConsume - the line mustn't exceed this value
- * @return length of the line read
- */
- public int readLine(Writable str, int maxBytesToConsume) throws IOException {
- ChunkWritable cw = (ChunkWritable)str;
- List<Node> list = new LinkedList<Node>();
-
- boolean newLine = false; //length of terminating newline
- long bytesConsumed = 0;
-
- do
- {
- int startPosn = bufferPosn; //starting from where we left off the last time
- if (bufferPosn >= bufferLength) {
- startPosn = bufferPosn = 0;
-
- bufferLength = in.read(buffer);
- if (bufferLength <= 0) {
- break; // EOF
- }
- }
-
- for (; bufferPosn < bufferLength; ++bufferPosn) { //search for newline
- if (buffer[bufferPosn] == LF) {
- newLine = true;
- ++bufferPosn; // at next invocation proceed from following byte
- break;
- }
- }
-
- int readLength = bufferPosn - startPosn;
- bytesConsumed += readLength;
-
- if (readLength > 0) {
- Node nd = new Node();
- nd.slice = new byte[readLength];
- nd.len = readLength;
- System.arraycopy(buffer, startPosn, nd.slice, 0, nd.len);
- list.add(nd);
- }
- } while (!newLine && bytesConsumed < maxBytesToConsume);
-
- if (list.size() > 0) {
- cw.box = new byte[(int)bytesConsumed];
- int pos = 0;
- for (int i = 0; i < list.size(); i++)
- {
- Node n = list.get(i);
- System.arraycopy(n.slice, 0, cw.box, pos, n.len);
- pos += n.len;
- }
- }
-
- return (int)bytesConsumed;
- }
+ * Reads a line terminated by LF.
+ *
+ * @param str - output parameter, will contain the read record
+ * @param maxBytesToConsume - the line mustn't exceed this value
+ * @return length of the line read
+ * @throws IOException if the first byte cannot be read for any reason
+ * other than the end of the file, if the input stream has been closed,
+ * or if some other I/O error occurs.
+ */
+ public int readLine(Writable str, int maxBytesToConsume) throws IOException {
+ ChunkWritable cw = (ChunkWritable) str;
+ List<Node> list = new LinkedList<Node>();
+
+ boolean newLine = false; // length of terminating newline
+ long bytesConsumed = 0;
+
+ do {
+ int startPosn = bufferPosn; // starting from where we left off the
+ // last time
+ if (bufferPosn >= bufferLength) {
+ startPosn = bufferPosn = 0;
+
+ bufferLength = in.read(buffer);
+ if (bufferLength <= 0) {
+ break; // EOF
+ }
+ }
+
+ for (; bufferPosn < bufferLength; ++bufferPosn) { // search for
+ // newline
+ if (buffer[bufferPosn] == LF) {
+ newLine = true;
+ ++bufferPosn; // at next invocation proceed from following
+ // byte
+ break;
+ }
+ }
+
+ int readLength = bufferPosn - startPosn;
+ bytesConsumed += readLength;
+
+ if (readLength > 0) {
+ Node nd = new Node();
+ nd.slice = new byte[readLength];
+ nd.len = readLength;
+ System.arraycopy(buffer, startPosn, nd.slice, 0, nd.len);
+ list.add(nd);
+ }
+ } while (!newLine && bytesConsumed < maxBytesToConsume);
+
+ if (list.size() > 0) {
+ cw.box = new byte[(int) bytesConsumed];
+ int pos = 0;
+ for (int i = 0; i < list.size(); i++) {
+ Node n = list.get(i);
+ System.arraycopy(n.slice, 0, cw.box, pos, n.len);
+ pos += n.len;
+ }
+ }
+
+ return (int) bytesConsumed;
+ }
}
[2/4] incubator-hawq git commit: HAWQ-28. JavaDoc fixes for PXF
Posted by nh...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveFilterBuilder.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveFilterBuilder.java b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveFilterBuilder.java
index abdc82c..1fe3a64 100644
--- a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveFilterBuilder.java
+++ b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveFilterBuilder.java
@@ -7,60 +7,75 @@ import java.util.LinkedList;
import java.util.List;
/**
- * Uses the filter parser code to build a filter object,
- * either simple - a single {@link com.pivotal.pxf.api.FilterParser.BasicFilter}
- * object or a compound - a {@link java.util.List} of
+ * Uses the filter parser code to build a filter object, either simple - a
+ * single {@link com.pivotal.pxf.api.FilterParser.BasicFilter} object or a
+ * compound - a {@link java.util.List} of
* {@link com.pivotal.pxf.api.FilterParser.BasicFilter} objects.
- * {@link com.pivotal.pxf.plugins.hive.HiveAccessor} will use the filter
- * for partition filtering
+ * {@link com.pivotal.pxf.plugins.hive.HiveAccessor} will use the filter for
+ * partition filtering.
*/
public class HiveFilterBuilder implements FilterParser.FilterBuilder {
private InputData inputData;
- /**
- * Constructs a HiveFilterBuilder object
- * @param input
- */
+ /**
+ * Constructs a HiveFilterBuilder object.
+ *
+ * @param input input data containing filter string
+ */
public HiveFilterBuilder(InputData input) {
inputData = input;
}
/**
- * Translates a filterString into a FilterParser.BasicFilter or a list of such filters
- * @param filterString the string representation of the filter
- * @return a single {@link com.pivotal.pxf.api.FilterParser.BasicFilter} object or
- * a {@link java.util.List} of {@link com.pivotal.pxf.api.FilterParser.BasicFilter} objects.
+ * Translates a filterString into a {@link com.pivotal.pxf.api.FilterParser.BasicFilter} or a
+ * list of such filters.
+ *
+ * @param filterString the string representation of the filter
+ * @return a single {@link com.pivotal.pxf.api.FilterParser.BasicFilter}
+ * object or a {@link java.util.List} of
+ * {@link com.pivotal.pxf.api.FilterParser.BasicFilter} objects.
+ * @throws Exception if parsing the filter failed or filter is not a basic
+ * filter or list of basic filters
*/
public Object getFilterObject(String filterString) throws Exception {
FilterParser parser = new FilterParser(this);
Object result = parser.parse(filterString);
- if (!(result instanceof FilterParser.BasicFilter) && !(result instanceof List)) {
- throw new Exception("String " + filterString + " resolved to no filter");
+ if (!(result instanceof FilterParser.BasicFilter)
+ && !(result instanceof List)) {
+ throw new Exception("String " + filterString
+ + " resolved to no filter");
}
return result;
}
- @SuppressWarnings("unchecked")
- public Object build(FilterParser.Operation opId,
- Object leftOperand,
+ @Override
+ @SuppressWarnings("unchecked")
+ public Object build(FilterParser.Operation opId, Object leftOperand,
Object rightOperand) throws Exception {
- if (leftOperand instanceof FilterParser.BasicFilter || leftOperand instanceof List) {
- if (opId != FilterParser.Operation.HDOP_AND ||
- !(rightOperand instanceof FilterParser.BasicFilter)) {
- throw new Exception("Only AND is allowed between compound expressions");
+ if (leftOperand instanceof FilterParser.BasicFilter
+ || leftOperand instanceof List) {
+ if (opId != FilterParser.Operation.HDOP_AND
+ || !(rightOperand instanceof FilterParser.BasicFilter)) {
+ throw new Exception(
+ "Only AND is allowed between compound expressions");
}
if (leftOperand instanceof List) {
- return handleCompoundOperations((List<FilterParser.BasicFilter>) leftOperand, (FilterParser.BasicFilter) rightOperand);
+ return handleCompoundOperations(
+ (List<FilterParser.BasicFilter>) leftOperand,
+ (FilterParser.BasicFilter) rightOperand);
} else {
- return handleCompoundOperations((FilterParser.BasicFilter) leftOperand, (FilterParser.BasicFilter) rightOperand);
+ return handleCompoundOperations(
+ (FilterParser.BasicFilter) leftOperand,
+ (FilterParser.BasicFilter) rightOperand);
}
}
if (!(rightOperand instanceof FilterParser.Constant)) {
- throw new Exception("expressions of column-op-column are not supported");
+ throw new Exception(
+ "expressions of column-op-column are not supported");
}
// Assume column is on the left
@@ -70,8 +85,8 @@ public class HiveFilterBuilder implements FilterParser.FilterBuilder {
}
/*
- * Handles simple column-operator-constant expressions
- * Creates a special filter in the case the column is the row key column
+ * Handles simple column-operator-constant expressions Creates a special
+ * filter in the case the column is the row key column
*/
private FilterParser.BasicFilter handleSimpleOperations(FilterParser.Operation opId,
FilterParser.ColumnIndex column,
@@ -79,24 +94,31 @@ public class HiveFilterBuilder implements FilterParser.FilterBuilder {
return new FilterParser.BasicFilter(opId, column, constant);
}
- /*
- * Handle AND of already calculated expressions
- * Currently only AND, in the future OR can be added
+ /**
+ * Handles AND of already calculated expressions. Currently only AND, in the
+ * future OR can be added
*
* Four cases here:
- * 1) both are simple filters
- * 2) left is a FilterList and right is a filter
- * 3) left is a filter and right is a FilterList
- * 4) both are FilterLists
- *
+ * <ol>
+ * <li>both are simple filters</li>
+ * <li>left is a FilterList and right is a filter</li>
+ * <li>left is a filter and right is a FilterList</li>
+ * <li>both are FilterLists</li>
+ * </ol>
* Currently, 1, 2 can occur, since no parenthesis are used
+ *
+ * @param left left hand filter
+ * @param right right hand filter
+ * @return list of filters constructing the filter tree
*/
- private List<FilterParser.BasicFilter> handleCompoundOperations(List<FilterParser.BasicFilter> left, FilterParser.BasicFilter right) {
+ private List<FilterParser.BasicFilter> handleCompoundOperations(List<FilterParser.BasicFilter> left,
+ FilterParser.BasicFilter right) {
left.add(right);
return left;
}
- private List<FilterParser.BasicFilter> handleCompoundOperations(FilterParser.BasicFilter left, FilterParser.BasicFilter right) {
+ private List<FilterParser.BasicFilter> handleCompoundOperations(FilterParser.BasicFilter left,
+ FilterParser.BasicFilter right) {
List<FilterParser.BasicFilter> result = new LinkedList<FilterParser.BasicFilter>();
result.add(left);
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveInputFormatFragmenter.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveInputFormatFragmenter.java b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveInputFormatFragmenter.java
index eefcac6..5c51e93 100644
--- a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveInputFormatFragmenter.java
+++ b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveInputFormatFragmenter.java
@@ -15,17 +15,20 @@ import java.util.List;
/**
* Specialized Hive fragmenter for RC and Text files tables. Unlike the
- * HiveDataFragmenter, this class does not send the serde properties to the
- * accessor/resolvers. This is done to avoid memory explosion in Hawq. For RC
- * use together with HiveRCFileAccessor/HiveColumnarSerdeResolver. For Text use
- * together with HiveLineBreakAccessor/HiveStringPassResolver.
- * <p/>
+ * {@link HiveDataFragmenter}, this class does not send the serde properties to
+ * the accessor/resolvers. This is done to avoid memory explosion in Hawq. For
+ * RC use together with {@link HiveRCFileAccessor}/
+ * {@link HiveColumnarSerdeResolver}. For Text use together with
+ * {@link HiveLineBreakAccessor}/{@link HiveStringPassResolver}. <br>
* Given a Hive table and its partitions, divide the data into fragments (here a
* data fragment is actually a HDFS file block) and return a list of them. Each
- * data fragment will contain the following information: a. sourceName: full
- * HDFS path to the data file that this data fragment is part of b. hosts: a
- * list of the datanode machines that hold a replica of this block c. userData:
- * inputformat name, serde names and partition keys
+ * data fragment will contain the following information:
+ * <ol>
+ * <li>sourceName: full HDFS path to the data file that this data fragment is
+ * part of</li>
+ * <li>hosts: a list of the datanode machines that hold a replica of this block</li>
+ * <li>userData: inputformat name, serde names and partition keys</li>
+ * </ol>
*/
public class HiveInputFormatFragmenter extends HiveDataFragmenter {
private static final Log LOG = LogFactory.getLog(HiveInputFormatFragmenter.class);
@@ -40,13 +43,13 @@ public class HiveInputFormatFragmenter extends HiveDataFragmenter {
public static final int TOK_KEYS = 1;
public static final int TOK_FILTER_DONE = 2;
- /* defines the Hive input formats currently supported in pxf */
+ /** Defines the Hive input formats currently supported in pxf */
public enum PXF_HIVE_INPUT_FORMATS {
RC_FILE_INPUT_FORMAT,
TEXT_FILE_INPUT_FORMAT
}
- /* defines the Hive serializers (serde classes) currently supported in pxf */
+ /** Defines the Hive serializers (serde classes) currently supported in pxf */
public enum PXF_HIVE_SERDES {
COLUMNAR_SERDE,
LAZY_BINARY_COLUMNAR_SERDE,
@@ -54,7 +57,7 @@ public class HiveInputFormatFragmenter extends HiveDataFragmenter {
}
/**
- * Constructs a HiveInputFormatFragmenter
+ * Constructs a HiveInputFormatFragmenter.
*
* @param inputData all input parameters coming from the client
*/
@@ -63,10 +66,17 @@ public class HiveInputFormatFragmenter extends HiveDataFragmenter {
}
/**
- * Extracts the user data
+ * Extracts the user data:
+ * serde, partition keys and whether filter was included in fragmenter
+ *
+ * @param input input data from client
+ * @param supportedSerdes supported serde names
+ * @return parsed tokens
+ * @throws UserDataException if user data contains unsupported serde
+ * or wrong number of tokens
*/
static public String[] parseToks(InputData input, String... supportedSerdes)
- throws Exception {
+ throws UserDataException {
String userData = new String(input.getFragmentUserData());
String[] toks = userData.split(HIVE_UD_DELIM);
if (supportedSerdes.length > 0
@@ -143,23 +153,24 @@ public class HiveInputFormatFragmenter extends HiveDataFragmenter {
}
/**
- * Converts HAWQ type to hive type. The supported mappings are:<br/>
- * BOOLEAN -> boolean<br/>
- * SMALLINT -> smallint (tinyint is converted to smallint)<br/>
- * BIGINT -> bigint<br/>
- * TIMESTAMP, TIME -> timestamp<br/>
- * NUMERIC -> decimal<br/>
- * BYTEA -> binary<br/>
- * INTERGER -> int<br/>
- * TEXT -> string<br/>
- * REAL -> float<br/>
- * FLOAT8 -> double
- * <p/>
+ * Converts HAWQ type to hive type. The supported mappings are:<ul>
+ * <li>{@code BOOLEAN -> boolean}</li>
+ * <li>{@code SMALLINT -> smallint (tinyint is converted to smallint)}</li>
+ * <li>{@code BIGINT -> bigint}</li>
+ * <li>{@code TIMESTAMP, TIME -> timestamp}</li>
+ * <li>{@code NUMERIC -> decimal}</li>
+ * <li>{@code BYTEA -> binary}</li>
+ * <li>{@code INTERGER -> int}</li>
+ * <li>{@code TEXT -> string}</li>
+ * <li>{@code REAL -> float}</li>
+ * <li>{@code FLOAT8 -> double}</li>
+ * </ul>
* All other types (both in HAWQ and in HIVE) are not supported.
*
* @param type HAWQ data type
* @param name field name
* @return Hive type
+ * @throws UnsupportedTypeException if type is not supported
*/
public static String toHiveType(DataType type, String name) {
switch (type) {
@@ -245,7 +256,6 @@ public class HiveInputFormatFragmenter extends HiveDataFragmenter {
assertFileType(inputFormatName, partData);
String userData = assertSerde(serdeName, partData) + HIVE_UD_DELIM
+ partitionKeys + HIVE_UD_DELIM + filterInFragmenter;
- ;
return userData.getBytes();
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveLineBreakAccessor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveLineBreakAccessor.java b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveLineBreakAccessor.java
index ad6e70e..b293123 100644
--- a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveLineBreakAccessor.java
+++ b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveLineBreakAccessor.java
@@ -10,12 +10,15 @@ import static com.pivotal.pxf.plugins.hive.HiveInputFormatFragmenter.PXF_HIVE_SE
/**
* Specialization of HiveAccessor for a Hive table stored as Text files.
- * Use together with HiveInputFormatFragmenter/HiveStringPassResolver
+ * Use together with {@link HiveInputFormatFragmenter}/{@link HiveStringPassResolver}.
*/
public class HiveLineBreakAccessor extends HiveAccessor {
/**
* Constructs a HiveLineBreakAccessor.
+ *
+ * @param input input containing user data
+ * @throws Exception if user data was wrong
*/
public HiveLineBreakAccessor(InputData input) throws Exception {
super(input, new TextInputFormat());
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveRCFileAccessor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveRCFileAccessor.java b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveRCFileAccessor.java
index 66acd64..6e64296 100644
--- a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveRCFileAccessor.java
+++ b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveRCFileAccessor.java
@@ -15,12 +15,15 @@ import static com.pivotal.pxf.plugins.hive.HiveInputFormatFragmenter.PXF_HIVE_SE
/**
* Specialization of HiveAccessor for a Hive table that stores only RC files.
* This class replaces the generic HiveAccessor for a case where a table is stored entirely as RC files.
- * Use together with HiveInputFormatFragmenter/HiveColumnarSerdeResolver
+ * Use together with {@link HiveInputFormatFragmenter}/{@link HiveColumnarSerdeResolver}
*/
public class HiveRCFileAccessor extends HiveAccessor {
/**
* Constructs a HiveRCFileAccessor.
+ *
+ * @param input input containing user data
+ * @throws Exception if user data was wrong
*/
public HiveRCFileAccessor(InputData input) throws Exception {
super(input, new RCFileInputFormat());
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveResolver.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveResolver.java b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveResolver.java
index 3096e68..85a04d3 100644
--- a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveResolver.java
+++ b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/HiveResolver.java
@@ -71,6 +71,8 @@ public class HiveResolver extends Plugin implements ReadResolver {
*
* @param input contains the Serde class name, the serde properties string
* and the partition keys
+ * @throws Exception if user data was wrong or serde failed to be
+ * instantiated
*/
public HiveResolver(InputData input) throws Exception {
super(input);
@@ -100,7 +102,7 @@ public class HiveResolver extends Plugin implements ReadResolver {
return record;
}
- /* parse user data string (arrived from fragmenter) */
+ /* Parses user data string (arrived from fragmenter). */
void parseUserData(InputData input) throws Exception {
final int EXPECTED_NUM_OF_TOKS = 5;
@@ -122,7 +124,7 @@ public class HiveResolver extends Plugin implements ReadResolver {
: input.getUserProperty("MAPKEY_DELIM");
}
- /* Get and init the deserializer for the records of this Hive data fragment */
+ /* Gets and init the deserializer for the records of this Hive data fragment. */
void initSerde(InputData inputData) throws Exception {
Properties serdeProperties;
@@ -138,7 +140,7 @@ public class HiveResolver extends Plugin implements ReadResolver {
/*
* The partition fields are initialized one time base on userData provided
- * by the fragmenter
+ * by the fragmenter.
*/
void initPartitionFields() {
partitionFields = new LinkedList<>();
@@ -234,7 +236,7 @@ public class HiveResolver extends Plugin implements ReadResolver {
/*
* The partition fields are initialized one time based on userData provided
- * by the fragmenter
+ * by the fragmenter.
*/
int initPartitionFields(StringBuilder parts) {
if (partitionKeys.equals(HiveDataFragmenter.HIVE_NO_PART_TBL)) {
@@ -287,8 +289,7 @@ public class HiveResolver extends Plugin implements ReadResolver {
parts.append(HiveDecimal.create(val).bigDecimalValue());
break;
case serdeConstants.BINARY_TYPE_NAME:
- Utilities.byteArrayToOctalString(val.getBytes(),
- parts);
+ Utilities.byteArrayToOctalString(val.getBytes(), parts);
break;
default:
throw new UnsupportedTypeException(
@@ -301,7 +302,7 @@ public class HiveResolver extends Plugin implements ReadResolver {
/**
* Returns true if the partition value is Hive's default partition name
- * (defined in hive.exec.default.partition.name)
+ * (defined in hive.exec.default.partition.name).
*
* @param partitionType partition field type
* @param partitionValue partition value
@@ -559,7 +560,7 @@ public class HiveResolver extends Plugin implements ReadResolver {
}
/*
- * Get the delimiter character from the URL, verify and store it. Must be a
+ * Gets the delimiter character from the URL, verify and store it. Must be a
* single ascii character (same restriction as Hawq's). If a hex
* representation was passed, convert it to its char.
*/
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/utilities/HiveUtilities.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/utilities/HiveUtilities.java b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/utilities/HiveUtilities.java
index 13de547..700a88f 100644
--- a/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/utilities/HiveUtilities.java
+++ b/pxf/pxf-hive/src/main/java/com/pivotal/pxf/plugins/hive/utilities/HiveUtilities.java
@@ -64,25 +64,27 @@ public class HiveUtilities {
* Checks if hive type is supported, and if so
* return its matching HAWQ type.
* Unsupported types will result in an exception.
- *
- * The supported mappings are:
- * tinyint -> int2
- * smallint -> int2
- * int -> int4
- * bigint -> int8
- * boolean -> bool
- * float -> float4
- * double -> float8
- * string -> text
- * binary -> bytea
- * timestamp -> timestamp
- * date -> date
- * decimal(precision, scale) -> numeric(precision, scale)
- * varchar(size) -> varchar(size)
- * char(size) -> bpchar(size)
+ * <br>
+ * The supported mappings are:<ul>
+ * <li>{@code tinyint -> int2}</li>
+ * <li>{@code smallint -> int2}</li>
+ * <li>{@code int -> int4}</li>
+ * <li>{@code bigint -> int8}</li>
+ * <li>{@code boolean -> bool}</li>
+ * <li>{@code float -> float4}</li>
+ * <li>{@code double -> float8}</li>
+ * <li>{@code string -> text}</li>
+ * <li>{@code binary -> bytea}</li>
+ * <li>{@code timestamp -> timestamp}</li>
+ * <li>{@code date -> date}</li>
+ * <li>{@code decimal(precision, scale) -> numeric(precision, scale)}</li>
+ * <li>{@code varchar(size) -> varchar(size)}</li>
+ * <li>{@code char(size) -> bpchar(size)}</li>
+ * </ul>
*
* @param hiveColumn hive column schema
* @return field with mapped HAWQ type and modifiers
+ * @throws UnsupportedTypeException if the column type is not supported
*/
public static Metadata.Field mapHiveType(FieldSchema hiveColumn) throws UnsupportedTypeException {
String fieldName = hiveColumn.getName();
@@ -156,8 +158,11 @@ public class HiveUtilities {
/**
* Verifies modifiers are null or integers.
+ * Modifier is a value assigned to a type,
+ * e.g. size of a varchar - varchar(size).
*
* @param modifiers type modifiers to be verified
+ * @return whether modifiers are null or integers
*/
private static boolean verifyModifers(String[] modifiers) {
if (modifiers == null) {
@@ -175,7 +180,10 @@ public class HiveUtilities {
* Extracts the db_name and table_name from the qualifiedName.
* qualifiedName is the Hive table name that the user enters in the CREATE EXTERNAL TABLE statement
* or when querying HCatalog table.
- * It can be either <table_name> or <db_name.table_name>.
+ * It can be either <code>table_name</code> or <code>db_name.table_name</code>.
+ *
+ * @param qualifiedName Hive table name
+ * @return {@link com.pivotal.pxf.api.Metadata.Table} object holding the full table name
*/
public static Metadata.Table parseTableQualifiedName(String qualifiedName) {
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/BridgeOutputBuilder.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/BridgeOutputBuilder.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/BridgeOutputBuilder.java
index 634890c..dd8111b 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/BridgeOutputBuilder.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/BridgeOutputBuilder.java
@@ -18,12 +18,13 @@ import java.util.List;
import static com.pivotal.pxf.api.io.DataType.TEXT;
-/*
- * Class creates the output record that is piped by the java process to the GPDB backend
- * Actually, the output record is serialized and the obtained byte string is piped to the GPDB
- * segment. The output record will implement Writable, and the mission of BridgeOutputBuilder
- * will be to translate a list of OneField objects (obtained from the Resolver) into
- * an output record.
+/**
+ * Class creates the output record that is piped by the java process to the HAWQ
+ * backend. Actually, the output record is serialized and the obtained byte
+ * string is piped to the HAWQ segment. The output record will implement
+ * Writable, and the mission of BridgeOutputBuilder will be to translate a list
+ * of {@link OneField} objects (obtained from the Resolver) into an output
+ * record.
*/
public class BridgeOutputBuilder {
private ProtocolData inputData;
@@ -33,24 +34,29 @@ public class BridgeOutputBuilder {
private String[] colNames;
/**
- * Constructs a BridgeOutputBuilder
+ * Constructs a BridgeOutputBuilder.
+ *
+ * @param input input data, like requested output format and schema
+ * information
*/
public BridgeOutputBuilder(ProtocolData input) {
inputData = input;
makeErrorRecord();
}
- /*
- * We need a separate GPDBWritable record to represent the error record. Just setting
- * the errorFlag on the "output" GPDBWritable variable is not good enough, since the GPDBWritable is built
- * only after the first record is read from the file. And if we encounter an error while fetching
- * the first record from the file, then the output member will be null. The reason we cannot count on
- * the schema to build the GPDBWritable output variable before reading the first record, is
- * because the schema does not account for arrays - we cannot know from the schema the length of
- * an array. We find out only after fetching the first record.
+ /**
+ * We need a separate GPDBWritable record to represent the error record.
+ * Just setting the errorFlag on the "output" GPDBWritable variable is not
+ * good enough, since the GPDBWritable is built only after the first record
+ * is read from the file. And if we encounter an error while fetching the
+ * first record from the file, then the output member will be null. The
+ * reason we cannot count on the schema to build the GPDBWritable output
+ * variable before reading the first record, is because the schema does not
+ * account for arrays - we cannot know from the schema the length of an
+ * array. We find out only after fetching the first record.
*/
void makeErrorRecord() {
- int[] errSchema = {TEXT.getOID()};
+ int[] errSchema = { TEXT.getOID() };
if (inputData.outputFormat() != OutputFormat.BINARY) {
return;
@@ -60,8 +66,13 @@ public class BridgeOutputBuilder {
errorRecord.setError(true);
}
- /*
- * Returns the error record
+ /**
+ * Returns the error record. If the output format is not binary, error
+ * records are not supported, and the given exception will be thrown
+ *
+ * @param ex exception to be stored in record
+ * @return error record
+ * @throws Exception if the output format is not binary
*/
public Writable getErrorOutput(Exception ex) throws Exception {
if (inputData.outputFormat() == OutputFormat.BINARY) {
@@ -72,10 +83,15 @@ public class BridgeOutputBuilder {
}
}
- /*
+ /**
* Translates recFields (obtained from the Resolver) into an output record.
- */
- public Writable makeOutput(List<OneField> recFields) throws BadRecordException {
+ *
+ * @param recFields record fields to be serialized
+ * @return Writable object with serialized row
+ * @throws BadRecordException if building the output record failed
+ */
+ public Writable makeOutput(List<OneField> recFields)
+ throws BadRecordException {
if (output == null && inputData.outputFormat() == OutputFormat.BINARY) {
makeGPDBWritableOutput();
}
@@ -85,9 +101,11 @@ public class BridgeOutputBuilder {
return output;
}
- /*
- * Creates the GPDBWritable object. The object is created one time
- * and is refilled from recFields for each record sent
+ /**
+ * Creates the GPDBWritable object. The object is created one time and is
+ * refilled from recFields for each record sent
+ *
+ * @return empty GPDBWritable object with set columns
*/
GPDBWritable makeGPDBWritableOutput() {
int num_actual_fields = inputData.getColumns();
@@ -104,8 +122,11 @@ public class BridgeOutputBuilder {
return (GPDBWritable) output;
}
- /*
- * Fills the output record based on the fields in recFields
+ /**
+ * Fills the output record based on the fields in recFields.
+ *
+ * @param recFields record fields
+ * @throws BadRecordException if building the output record failed
*/
void fillOutputRecord(List<OneField> recFields) throws BadRecordException {
if (inputData.outputFormat() == OutputFormat.BINARY) {
@@ -115,56 +136,80 @@ public class BridgeOutputBuilder {
}
}
- /*
- * Fills a GPDBWritable object based on recFields
- * The input record recFields must correspond to schema.
- * If the record has more or less fields than the schema we throw an exception.
- * We require that the type of field[i] in recFields corresponds to the type
- * of field[i] in the schema.
+ /**
+ * Fills a GPDBWritable object based on recFields. The input record
+ * recFields must correspond to schema. If the record has more or less
+ * fields than the schema we throw an exception. We require that the type of
+ * field[i] in recFields corresponds to the type of field[i] in the schema.
+ *
+ * @param recFields record fields
+ * @throws BadRecordException if building the output record failed
*/
void fillGPDBWritable(List<OneField> recFields) throws BadRecordException {
int size = recFields.size();
- if (size == 0) { // size 0 means the resolver couldn't deserialize any of the record fields
+ if (size == 0) { // size 0 means the resolver couldn't deserialize any
+ // of the record fields
throw new BadRecordException("No fields in record");
} else if (size != schema.length) {
- throw new BadRecordException("Record has " + size + " fields but the schema size is " + schema.length);
+ throw new BadRecordException("Record has " + size
+ + " fields but the schema size is " + schema.length);
}
for (int i = 0; i < size; i++) {
OneField current = recFields.get(i);
if (!isTypeInSchema(current.type, schema[i])) {
- throw new BadRecordException("For field " + colNames[i] + " schema requires type " + DataType.get(schema[i]).toString() +
- " but input record has type " + DataType.get(current.type).toString());
+ throw new BadRecordException("For field " + colNames[i]
+ + " schema requires type "
+ + DataType.get(schema[i]).toString()
+ + " but input record has type "
+ + DataType.get(current.type).toString());
}
fillOneGPDBWritableField(current, i);
}
}
- /* Tests if data type is a string type */
+ /**
+ * Tests if data type is a string type. String type is a type that can be
+ * serialized as string, such as varchar, bpchar, text, numeric, timestamp,
+ * date.
+ *
+ * @param type data type
+ * @return whether data type is string type
+ */
boolean isStringType(DataType type) {
- return Arrays.asList(DataType.VARCHAR, DataType.BPCHAR, DataType.TEXT, DataType.NUMERIC, DataType.TIMESTAMP, DataType.DATE)
- .contains(type);
+ return Arrays.asList(DataType.VARCHAR, DataType.BPCHAR, DataType.TEXT,
+ DataType.NUMERIC, DataType.TIMESTAMP, DataType.DATE).contains(
+ type);
}
- /* Tests if record field type and schema type correspond */
+ /**
+ * Tests if record field type and schema type correspond.
+ *
+ * @param recType record type code
+ * @param schemaType schema type code
+ * @return whether record type and schema type match
+ */
boolean isTypeInSchema(int recType, int schemaType) {
DataType dtRec = DataType.get(recType);
DataType dtSchema = DataType.get(schemaType);
- return (dtSchema == DataType.UNSUPPORTED_TYPE || dtRec == dtSchema ||
- (isStringType(dtRec) && isStringType(dtSchema)));
+ return (dtSchema == DataType.UNSUPPORTED_TYPE || dtRec == dtSchema || (isStringType(dtRec) && isStringType(dtSchema)));
}
- /*
- * Fills a Text object based on recFields
+ /**
+ * Fills a Text object based on recFields.
+ *
+ * @param recFields record fields
+ * @throws BadRecordException if text formatted record has more than one field
*/
void fillText(List<OneField> recFields) throws BadRecordException {
/*
* For the TEXT case there must be only one record in the list
- */
+ */
if (recFields.size() != 1) {
- throw new BadRecordException("BridgeOutputBuilder must receive one field when handling the TEXT format");
+ throw new BadRecordException(
+ "BridgeOutputBuilder must receive one field when handling the TEXT format");
}
OneField fld = recFields.get(0);
@@ -178,32 +223,37 @@ public class BridgeOutputBuilder {
}
}
- /*
- * Fills one GPDBWritable field
+ /**
+ * Fills one GPDBWritable field.
+ *
+ * @param oneField field
+ * @param colIdx column index
+ * @throws BadRecordException if field type is not supported or doesn't match the schema
*/
- void fillOneGPDBWritableField(OneField oneField, int i) throws BadRecordException {
+ void fillOneGPDBWritableField(OneField oneField, int colIdx)
+ throws BadRecordException {
int type = oneField.type;
Object val = oneField.val;
GPDBWritable GPDBoutput = (GPDBWritable) output;
try {
switch (DataType.get(type)) {
case INTEGER:
- GPDBoutput.setInt(i, (Integer) val);
+ GPDBoutput.setInt(colIdx, (Integer) val);
break;
case FLOAT8:
- GPDBoutput.setDouble(i, (Double) val);
+ GPDBoutput.setDouble(colIdx, (Double) val);
break;
case REAL:
- GPDBoutput.setFloat(i, (Float) val);
+ GPDBoutput.setFloat(colIdx, (Float) val);
break;
case BIGINT:
- GPDBoutput.setLong(i, (Long) val);
+ GPDBoutput.setLong(colIdx, (Long) val);
break;
case SMALLINT:
- GPDBoutput.setShort(i, (Short) val);
+ GPDBoutput.setShort(colIdx, (Short) val);
break;
case BOOLEAN:
- GPDBoutput.setBoolean(i, (Boolean) val);
+ GPDBoutput.setBoolean(colIdx, (Boolean) val);
break;
case BYTEA:
byte[] bts = null;
@@ -214,7 +264,7 @@ public class BridgeOutputBuilder {
bts[j] = Array.getByte(val, j);
}
}
- GPDBoutput.setBytes(i, bts);
+ GPDBoutput.setBytes(colIdx, bts);
break;
case VARCHAR:
case BPCHAR:
@@ -223,11 +273,13 @@ public class BridgeOutputBuilder {
case NUMERIC:
case TIMESTAMP:
case DATE:
- GPDBoutput.setString(i, ObjectUtils.toString(val, null));
+ GPDBoutput.setString(colIdx, ObjectUtils.toString(val, null));
break;
default:
- String valClassName = (val != null) ? val.getClass().getSimpleName() : null;
- throw new UnsupportedOperationException(valClassName + " is not supported for HAWQ conversion");
+ String valClassName = (val != null) ? val.getClass().getSimpleName()
+ : null;
+ throw new UnsupportedOperationException(valClassName
+ + " is not supported for HAWQ conversion");
}
} catch (TypeMismatchException e) {
throw new BadRecordException(e);
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/FragmentsResponse.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/FragmentsResponse.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/FragmentsResponse.java
index dd8d2cf..87e918b 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/FragmentsResponse.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/FragmentsResponse.java
@@ -26,6 +26,11 @@ public class FragmentsResponse implements StreamingOutput {
private List<Fragment> fragments;
+ /**
+ * Constructs fragments response out of a list of fragments
+ *
+ * @param fragments fragment list
+ */
public FragmentsResponse(List<Fragment> fragments) {
this.fragments = fragments;
}
@@ -34,7 +39,7 @@ public class FragmentsResponse implements StreamingOutput {
* Serializes a fragments list in JSON,
* To be used as the result string for HAWQ.
* An example result is as follows:
- * {"PXFFragments":[{"replicas":["sdw1.corp.emc.com","sdw3.corp.emc.com","sdw8.corp.emc.com"],"sourceName":"text2.csv", "index":"0", "metadata":<base64 metadata for fragment>, "userData":"<data_specific_to_third_party_fragmenter>"},{"replicas":["sdw2.corp.emc.com","sdw4.corp.emc.com","sdw5.corp.emc.com"],"sourceName":"text_data.csv","index":"0","metadata":<base64 metadata for fragment>,"userData":"<data_specific_to_third_party_fragmenter>"}]}
+ * {@code {"PXFFragments":[{"replicas":["sdw1.corp.emc.com","sdw3.corp.emc.com","sdw8.corp.emc.com"],"sourceName":"text2.csv", "index":"0", "metadata":<base64 metadata for fragment>, "userData":"<data_specific_to_third_party_fragmenter>"},{"replicas":["sdw2.corp.emc.com","sdw4.corp.emc.com","sdw5.corp.emc.com"],"sourceName":"text_data.csv","index":"0","metadata":<base64 metadata for fragment>,"userData":"<data_specific_to_third_party_fragmenter>"}]}}
*/
@Override
public void write(OutputStream output) throws IOException,
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/FragmentsResponseFormatter.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/FragmentsResponseFormatter.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/FragmentsResponseFormatter.java
index a09e105..5145929 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/FragmentsResponseFormatter.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/FragmentsResponseFormatter.java
@@ -25,9 +25,9 @@ public class FragmentsResponseFormatter {
* @param fragments list of fragments
* @param data data (e.g. path) related to the fragments
* @return FragmentsResponse with given fragments
- * @throws IOException
+ * @throws UnknownHostException if converting host names to IP fails
*/
- public static FragmentsResponse formatResponse(List<Fragment> fragments, String data) throws IOException {
+ public static FragmentsResponse formatResponse(List<Fragment> fragments, String data) throws UnknownHostException {
/* print the raw fragment list to log when in debug level */
if (LOG.isDebugEnabled()) {
LOG.debug("Fragments before conversion to IP list:");
@@ -69,7 +69,9 @@ public class FragmentsResponseFormatter {
}
/**
- * Converts hosts to their matching IP addresses
+ * Converts hosts to their matching IP addresses.
+ *
+ * @throws UnknownHostException if converting host name to IP fails
*/
private static void convertHostsToIPs(List<Fragment> fragments) throws UnknownHostException {
/* host converted to IP map. Used to limit network calls. */
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/MetadataResponseFormatter.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/MetadataResponseFormatter.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/MetadataResponseFormatter.java
index 4e1323f..1bb0cfd 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/MetadataResponseFormatter.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/MetadataResponseFormatter.java
@@ -18,6 +18,10 @@ public class MetadataResponseFormatter {
/**
* Converts {@link Metadata} to JSON String format.
+ *
+ * @param metadata metadata to convert
+ * @return JSON formatted response
+ * @throws IOException if converting the data to JSON fails
*/
public static String formatResponseString(Metadata metadata) throws IOException {
/* print the metadata before serialization */
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/BufferWritable.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/BufferWritable.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/BufferWritable.java
index 2be23d3..3115ca7 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/BufferWritable.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/BufferWritable.java
@@ -5,49 +5,53 @@ import java.io.DataOutput;
import java.io.IOException;
import java.lang.UnsupportedOperationException;
-
/**
- * A serializable object for transporting a byte array through the Bridge framework
+ * A serializable object for transporting a byte array through the Bridge
+ * framework
*/
public class BufferWritable implements Writable {
-
- byte [] buf = null;
-
- /**
- * Constructs a BufferWritable.
- * Copies the buffer reference and not the actual bytes. This class
- * is used when we intend to transport a buffer through the Bridge
- * framework without copying the data each time the buffer is passed
- * between the Bridge objects.
- */
- public BufferWritable(byte [] inBuf) {
- buf = inBuf;
- }
+
+ byte[] buf = null;
+
+ /**
+ * Constructs a BufferWritable. Copies the buffer reference and not the
+ * actual bytes. This class is used when we intend to transport a buffer
+ * through the Bridge framework without copying the data each time the
+ * buffer is passed between the Bridge objects.
+ *
+ * @param inBuf buffer
+ */
+ public BufferWritable(byte[] inBuf) {
+ buf = inBuf;
+ }
/**
* Serializes the fields of this object to <code>out</code>.
*
* @param out <code>DataOutput</code> to serialize this object into.
- * @throws IOException
+ * @throws IOException if the buffer was not set
*/
- @Override
+ @Override
public void write(DataOutput out) throws IOException {
- if (buf == null)
- throw new IOException("BufferWritable was not set");
- out.write(buf);
+ if (buf == null)
+ throw new IOException("BufferWritable was not set");
+ out.write(buf);
}
/**
* Deserializes the fields of this object from <code>in</code>.
- * <p>For efficiency, implementations should attempt to re-use storage in the
- * existing object where possible.</p>
+ * <p>
+ * For efficiency, implementations should attempt to re-use storage in the
+ * existing object where possible.
+ * </p>
*
- * @param in <code>DataInput</code> to deserialize this object from.
- * @throws IOException
+ * @param in <code>DataInput</code> to deserialize this object from
+ * @throws UnsupportedOperationException this function is not supported
*/
- @Override
+ @Override
public void readFields(DataInput in) {
- throw new UnsupportedOperationException("BufferWritable.readFields() is not implemented");
- }
+ throw new UnsupportedOperationException(
+ "BufferWritable.readFields() is not implemented");
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/GPDBWritable.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/GPDBWritable.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/GPDBWritable.java
index 9e22a85..074f97e 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/GPDBWritable.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/GPDBWritable.java
@@ -19,7 +19,7 @@ public class GPDBWritable implements Writable {
* GPDBWritable is using the following serialization form:
* Total Length | Version | Error Flag | # of columns | Col type |...| Col type | Null Bit array | Col val...
* 4 byte | 2 byte | 1 byte | 2 byte | 1 byte |...| 1 byte | ceil(# of columns/8) byte | Fixed or Var length
- *
+ *
* For fixed length type, we know the length.
* In the col val, we align pad according to the alignment requirement of the type.
* For var length type, the alignment is always 4 byte.
@@ -150,11 +150,11 @@ public class GPDBWritable implements Writable {
public void readFields(DataInput in) throws IOException {
/*
* extract pkt len.
- *
- * GPSQL-1107:
+ *
+ * GPSQL-1107:
* The DataInput might already be empty (EOF), but we can't check it beforehand.
* If that's the case, pktlen is updated to -1, to mark that the object is still empty.
- * (can be checked with isEmpty()).
+ * (can be checked with isEmpty()).
*/
pktlen = readPktLen(in);
if (isEmpty()) {
@@ -185,7 +185,7 @@ public class GPDBWritable implements Writable {
colType = new int[colCnt];
DBType[] coldbtype = new DBType[colCnt];
for (int i = 0; i < colCnt; i++) {
- int enumType = (int) (in.readByte());
+ int enumType = (in.readByte());
curOffset += 1;
if (enumType == DBType.BIGINT.ordinal()) {
colType[i] = BIGINT.getOID();
@@ -362,7 +362,7 @@ public class GPDBWritable implements Writable {
} else {
nullBits[i] = false;
- /*
+ /*
* For fixed length type, we get the fixed length.
* For var len binary format, the length is in the col value.
* For text format, we must convert encoding first.
@@ -378,7 +378,7 @@ public class GPDBWritable implements Writable {
/* calculate and add the type alignment padding */
padLength[i] = roundUpAlignment(datlen, coldbtype.getAlignment()) - datlen;
datlen += padLength[i];
-
+
/* for variable length type, we add a 4 byte length header */
if (coldbtype.isVarLength()) {
datlen += 4;
@@ -392,22 +392,22 @@ public class GPDBWritable implements Writable {
*/
int endpadding = roundUpAlignment(datlen, 8) - datlen;
datlen += endpadding;
-
+
/* Construct the packet header */
out.writeInt(datlen);
out.writeShort(VERSION);
out.writeByte(errorFlag);
out.writeShort(numCol);
-
+
/* Write col type */
for (int i = 0; i < numCol; i++) {
out.writeByte(enumType[i]);
}
-
+
/* Nullness */
byte[] nullBytes = boolArrayToByteArray(nullBits);
out.write(nullBytes);
-
+
/* Column Value */
for (int i = 0; i < numCol; i++) {
if (!nullBits[i]) {
@@ -442,7 +442,7 @@ public class GPDBWritable implements Writable {
out.writeInt(colLength[i]);
out.write((byte[]) colValue[i]);
break;
-
+
/* For text format, add 4byte length header. string is already '\0' terminated */
default: {
out.writeInt(colLength[i]);
@@ -513,7 +513,7 @@ public class GPDBWritable implements Writable {
*/
/**
- * Set the column value of the record
+ * Sets the column value of the record.
*
* @param colIdx the column index
* @param val the value
@@ -526,7 +526,7 @@ public class GPDBWritable implements Writable {
}
/**
- * Set the column value of the record
+ * Sets the column value of the record.
*
* @param colIdx the column index
* @param val the value
@@ -539,7 +539,7 @@ public class GPDBWritable implements Writable {
}
/**
- * Set the column value of the record
+ * Sets the column value of the record.
*
* @param colIdx the column index
* @param val the value
@@ -552,7 +552,7 @@ public class GPDBWritable implements Writable {
}
/**
- * Set the column value of the record
+ * Sets the column value of the record.
*
* @param colIdx the column index
* @param val the value
@@ -569,7 +569,7 @@ public class GPDBWritable implements Writable {
}
/**
- * Set the column value of the record
+ * Sets the column value of the record.
*
* @param colIdx the column index
* @param val the value
@@ -582,7 +582,7 @@ public class GPDBWritable implements Writable {
}
/**
- * Set the column value of the record
+ * Sets the column value of the record.
*
* @param colIdx the column index
* @param val the value
@@ -595,7 +595,7 @@ public class GPDBWritable implements Writable {
}
/**
- * Set the column value of the record
+ * Sets the column value of the record.
*
* @param colIdx the column index
* @param val the value
@@ -608,7 +608,7 @@ public class GPDBWritable implements Writable {
}
/**
- * Set the column value of the record
+ * Sets the column value of the record.
*
* @param colIdx the column index
* @param val the value
@@ -621,9 +621,10 @@ public class GPDBWritable implements Writable {
}
/**
- * Get the column value of the record
+ * Gets the column value of the record.
*
* @param colIdx the column index
+ * @return column value
* @throws TypeMismatchException the column type does not match
*/
public Long getLong(int colIdx)
@@ -633,9 +634,10 @@ public class GPDBWritable implements Writable {
}
/**
- * Get the column value of the record
+ * Gets the column value of the record.
*
* @param colIdx the column index
+ * @return column value
* @throws TypeMismatchException the column type does not match
*/
public Boolean getBoolean(int colIdx)
@@ -645,9 +647,10 @@ public class GPDBWritable implements Writable {
}
/**
- * Get the column value of the record
+ * Gets the column value of the record.
*
* @param colIdx the column index
+ * @return column value
* @throws TypeMismatchException the column type does not match
*/
public byte[] getBytes(int colIdx)
@@ -657,9 +660,10 @@ public class GPDBWritable implements Writable {
}
/**
- * Get the column value of the record
+ * Gets the column value of the record.
*
* @param colIdx the column index
+ * @return column value
* @throws TypeMismatchException the column type does not match
*/
public String getString(int colIdx)
@@ -669,9 +673,10 @@ public class GPDBWritable implements Writable {
}
/**
- * Get the column value of the record
+ * Gets the column value of the record.
*
* @param colIdx the column index
+ * @return column value
* @throws TypeMismatchException the column type does not match
*/
public Float getFloat(int colIdx)
@@ -681,9 +686,10 @@ public class GPDBWritable implements Writable {
}
/**
- * Get the column value of the record
+ * Gets the column value of the record.
*
* @param colIdx the column index
+ * @return column value
* @throws TypeMismatchException the column type does not match
*/
public Double getDouble(int colIdx)
@@ -693,9 +699,10 @@ public class GPDBWritable implements Writable {
}
/**
- * Get the column value of the record
+ * Gets the column value of the record.
*
* @param colIdx the column index
+ * @return column value
* @throws TypeMismatchException the column type does not match
*/
public Integer getInt(int colIdx)
@@ -705,9 +712,10 @@ public class GPDBWritable implements Writable {
}
/**
- * Get the column value of the record
+ * Gets the column value of the record.
*
* @param colIdx the column index
+ * @return column value
* @throws TypeMismatchException the column type does not match
*/
public Short getShort(int colIdx)
@@ -717,7 +725,7 @@ public class GPDBWritable implements Writable {
}
/**
- * Set the error field
+ * Sets the error field.
*
* @param errorVal the error value
*/
@@ -726,8 +734,9 @@ public class GPDBWritable implements Writable {
}
/**
- * Return a string representation of the object
+ * Returns a string representation of the object.
*/
+ @Override
public String toString() {
if (colType == null) {
return null;
@@ -798,6 +807,9 @@ public class GPDBWritable implements Writable {
* Helper to get the type name.
* If a given oid is not in the commonly used list, we
* would expect a TEXT for it (for the error message).
+ *
+ * @param oid type OID
+ * @return type name
*/
public static String getTypeName(int oid) {
switch (DataType.get(oid)) {
@@ -852,6 +864,8 @@ public class GPDBWritable implements Writable {
* Returns if the writable object is empty,
* based on the pkt len as read from stream.
* -1 means nothing was read (eof).
+ *
+ * @return whether the writable object is empty
*/
public boolean isEmpty() {
return pktlen == EOF;
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/Text.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/Text.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/Text.java
index 4a5af6c..2ce28ba 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/Text.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/Text.java
@@ -13,9 +13,9 @@ import java.nio.charset.*;
import java.util.Arrays;
/**
- * This class stores text using standard UTF8 encoding.
- * It provides methods to serialize, deserialize.
- * The type of length is integer and is serialized using zero-compressed format.
+ * This class stores text using standard UTF8 encoding. It provides methods to
+ * serialize, deserialize. The type of length is integer and is serialized using
+ * zero-compressed format.
*/
public class Text implements Writable {
@@ -28,22 +28,22 @@ public class Text implements Writable {
private static final int EOF = -1;
private static final byte[] EMPTY_BYTES = new byte[0];
- private static ThreadLocal<CharsetEncoder> ENCODER_FACTORY =
- new ThreadLocal<CharsetEncoder>() {
- protected CharsetEncoder initialValue() {
- return Charset.forName("UTF-8").newEncoder().
- onMalformedInput(CodingErrorAction.REPORT).
- onUnmappableCharacter(CodingErrorAction.REPORT);
- }
- };
- private static ThreadLocal<CharsetDecoder> DECODER_FACTORY =
- new ThreadLocal<CharsetDecoder>() {
- protected CharsetDecoder initialValue() {
- return Charset.forName("UTF-8").newDecoder().
- onMalformedInput(CodingErrorAction.REPORT).
- onUnmappableCharacter(CodingErrorAction.REPORT);
- }
- };
+ private static ThreadLocal<CharsetEncoder> ENCODER_FACTORY = new ThreadLocal<CharsetEncoder>() {
+ @Override
+ protected CharsetEncoder initialValue() {
+ return Charset.forName("UTF-8").newEncoder().onMalformedInput(
+ CodingErrorAction.REPORT).onUnmappableCharacter(
+ CodingErrorAction.REPORT);
+ }
+ };
+ private static ThreadLocal<CharsetDecoder> DECODER_FACTORY = new ThreadLocal<CharsetDecoder>() {
+ @Override
+ protected CharsetDecoder initialValue() {
+ return Charset.forName("UTF-8").newDecoder().onMalformedInput(
+ CodingErrorAction.REPORT).onUnmappableCharacter(
+ CodingErrorAction.REPORT);
+ }
+ };
private byte[] bytes;
private int length;
@@ -54,6 +54,8 @@ public class Text implements Writable {
/**
* Construct from a string.
+ *
+ * @param string input string
*/
public Text(String string) {
set(string);
@@ -61,6 +63,8 @@ public class Text implements Writable {
/**
* Construct from another text.
+ *
+ * @param utf8 text to copy
*/
public Text(Text utf8) {
set(utf8);
@@ -68,6 +72,8 @@ public class Text implements Writable {
/**
* Construct from a byte array.
+ *
+ * @param utf8 input byte array
*/
public Text(byte[] utf8) {
set(utf8);
@@ -107,13 +113,22 @@ public class Text implements Writable {
}
/**
- * Converts the provided byte array to a String using the
- * UTF-8 encoding. If <code>replace</code> is true, then
- * malformed input is replaced with the
- * substitution character, which is U+FFFD. Otherwise the
- * method throws a MalformedInputException.
+ * Converts the provided byte array to a String using the UTF-8 encoding. If
+ * <code>replace</code> is true, then malformed input is replaced with the
+ * substitution character, which is U+FFFD. Otherwise the method throws a
+ * MalformedInputException.
+ *
+ * @param utf8 UTF-8 encoded byte array
+ * @param start start point
+ * @param length length of array
+ * @param replace whether to replace malformed input with substitution
+ * character
+ * @return decoded string
+ * @throws MalformedInputException if a malformed input is used
+ * @throws CharacterCodingException if the conversion failed
*/
- public static String decode(byte[] utf8, int start, int length, boolean replace)
+ public static String decode(byte[] utf8, int start, int length,
+ boolean replace)
throws CharacterCodingException {
return decode(ByteBuffer.wrap(utf8, start, length), replace);
}
@@ -122,8 +137,7 @@ public class Text implements Writable {
throws CharacterCodingException {
CharsetDecoder decoder = DECODER_FACTORY.get();
if (replace) {
- decoder.onMalformedInput(
- java.nio.charset.CodingErrorAction.REPLACE);
+ decoder.onMalformedInput(java.nio.charset.CodingErrorAction.REPLACE);
decoder.onUnmappableCharacter(CodingErrorAction.REPLACE);
}
String str = decoder.decode(utf8).toString();
@@ -136,28 +150,31 @@ public class Text implements Writable {
}
/**
- * Converts the provided String to bytes using the
- * UTF-8 encoding. If the input is malformed,
- * invalid chars are replaced by a default value.
+ * Converts the provided String to bytes using the UTF-8 encoding. If the
+ * input is malformed, invalid chars are replaced by a default value.
*
- * @return ByteBuffer: bytes stores at ByteBuffer.array()
- * and length is ByteBuffer.limit()
+ * @param string string to encode
+ * @return ByteBuffer: bytes stores at ByteBuffer.array() and length is
+ * ByteBuffer.limit()
+ * @throws CharacterCodingException if conversion failed
*/
-
public static ByteBuffer encode(String string)
throws CharacterCodingException {
return encode(string, true);
}
/**
- * Converts the provided String to bytes using the
- * UTF-8 encoding. If <code>replace</code> is true, then
- * malformed input is replaced with the
- * substitution character, which is U+FFFD. Otherwise the
- * method throws a MalformedInputException.
+ * Converts the provided String to bytes using the UTF-8 encoding. If
+ * <code>replace</code> is true, then malformed input is replaced with the
+ * substitution character, which is U+FFFD. Otherwise the method throws a
+ * MalformedInputException.
*
- * @return ByteBuffer: bytes stores at ByteBuffer.array()
- * and length is ByteBuffer.limit()
+ * @param string string to encode
+ * @param replace whether to replace malformed input with substitution character
+ * @return ByteBuffer: bytes stores at ByteBuffer.array() and length is
+ * ByteBuffer.limit()
+ * @throws MalformedInputException if a malformed input is used
+ * @throws CharacterCodingException if the conversion failed
*/
public static ByteBuffer encode(String string, boolean replace)
throws CharacterCodingException {
@@ -166,8 +183,7 @@ public class Text implements Writable {
encoder.onMalformedInput(CodingErrorAction.REPLACE);
encoder.onUnmappableCharacter(CodingErrorAction.REPLACE);
}
- ByteBuffer bytes =
- encoder.encode(CharBuffer.wrap(string.toCharArray()));
+ ByteBuffer bytes = encoder.encode(CharBuffer.wrap(string.toCharArray()));
if (replace) {
encoder.onMalformedInput(CodingErrorAction.REPORT);
encoder.onUnmappableCharacter(CodingErrorAction.REPORT);
@@ -178,6 +194,8 @@ public class Text implements Writable {
/**
* Returns the raw bytes; however, only data up to {@link #getLength()} is
* valid.
+ *
+ * @return raw bytes of byte array
*/
public byte[] getBytes() {
return bytes;
@@ -185,13 +203,17 @@ public class Text implements Writable {
/**
* Returns the number of bytes in the byte array
+ *
+ * @return number of bytes in byte array
*/
public int getLength() {
return length;
}
/**
- * Set to contain the contents of a string.
+ * Sets to contain the contents of a string.
+ *
+ * @param string input string
*/
public void set(String string) {
try {
@@ -199,30 +221,35 @@ public class Text implements Writable {
bytes = bb.array();
length = bb.limit();
} catch (CharacterCodingException e) {
- throw new RuntimeException("Should not have happened " + e.toString());
+ throw new RuntimeException("Should not have happened "
+ + e.toString());
}
}
/**
- * Set to a utf8 byte array
+ * Sets to a UTF-8 byte array.
+ *
+ * @param utf8 input UTF-8 byte array
*/
public void set(byte[] utf8) {
set(utf8, 0, utf8.length);
}
/**
- * copy a text.
+ * Copies a text.
+ *
+ * @param other text object to copy.
*/
public void set(Text other) {
set(other.getBytes(), 0, other.getLength());
}
/**
- * Set the Text to range of bytes
+ * Sets the Text to range of bytes.
*
- * @param utf8 the data to copy from
+ * @param utf8 the data to copy from
* @param start the first position of the new string
- * @param len the number of bytes of the new string
+ * @param len the number of bytes of the new string
*/
public void set(byte[] utf8, int start, int len) {
setCapacity(len, false);
@@ -231,11 +258,11 @@ public class Text implements Writable {
}
/**
- * Append a range of bytes to the end of the given text
+ * Appends a range of bytes to the end of the given text.
*
- * @param utf8 the data to copy from
+ * @param utf8 the data to copy from
* @param start the first position to append from utf8
- * @param len the number of bytes to append
+ * @param len the number of bytes to append
*/
public void append(byte[] utf8, int start, int len) {
setCapacity(length + len, true);
@@ -244,7 +271,7 @@ public class Text implements Writable {
}
/**
- * Clear the string to empty.
+ * Clears the string to empty.
*/
public void clear() {
length = 0;
@@ -252,12 +279,13 @@ public class Text implements Writable {
/*
* Sets the capacity of this Text object to <em>at least</em>
- * <code>len</code> bytes. If the current buffer is longer,
- * then the capacity and existing content of the buffer are
- * unchanged. If <code>len</code> is larger
- * than the current capacity, the Text object's capacity is
- * increased to match.
+ * <code>len</code> bytes. If the current buffer is longer, then the
+ * capacity and existing content of the buffer are unchanged. If
+ * <code>len</code> is larger than the current capacity, the Text object's
+ * capacity is increased to match.
+ *
* @param len the number of bytes we need
+ *
* @param keepData should the old data be kept
*/
private void setCapacity(int len, boolean keepData) {
@@ -275,11 +303,13 @@ public class Text implements Writable {
*
* @see java.lang.Object#toString()
*/
+ @Override
public String toString() {
try {
return decode(bytes, 0, length);
} catch (CharacterCodingException e) {
- throw new RuntimeException("Should not have happened " + e.toString());
+ throw new RuntimeException("Should not have happened "
+ + e.toString());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/Writable.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/Writable.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/Writable.java
index 8a550cf..3bb3d50 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/Writable.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/io/Writable.java
@@ -4,7 +4,6 @@ import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
-
/**
* A serializable object which implements a simple, efficient, serialization
* protocol, based on {@link DataInput} and {@link DataOutput}.
@@ -15,7 +14,7 @@ public interface Writable {
* Serialize the fields of this object to <code>out</code>.
*
* @param out <code>DataOutput</code> to serialize this object into.
- * @throws IOException
+ * @throws IOException if I/O error occurs
*/
void write(DataOutput out) throws IOException;
@@ -25,7 +24,7 @@ public interface Writable {
* existing object where possible.</p>
*
* @param in <code>DataInput</code> to deserialize this object from.
- * @throws IOException
+ * @throws IOException if I/O error occurs
*/
void readFields(DataInput in) throws IOException;
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/InvalidPathResource.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/InvalidPathResource.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/InvalidPathResource.java
index 50c2936..8a32b01 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/InvalidPathResource.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/InvalidPathResource.java
@@ -25,7 +25,7 @@ class Version {
* as it is used to verify proper load of the PXF webapp.
*
* For each path, the version is compared to the current version PXF_VERSION.
- * The expected format of a path is "http://<host>:<port>/pxf/<version>/<rest of path>
+ * The expected format of a path is "{@code http://<host>:<port>/pxf/<version>/<rest of path>}"
*
* The returned value is always a Server Error code (500).
* If the version is different than the current version, an appropriate error is returned with version details.
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/MetadataResource.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/MetadataResource.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/MetadataResource.java
index 368a7a5..0e5252f 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/MetadataResource.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/MetadataResource.java
@@ -24,9 +24,10 @@ import com.pivotal.pxf.service.MetadataResponseFormatter;
/**
* Class enhances the API of the WEBHDFS REST server.
* Returns the metadata of a given hcatalog table.
- * Example for querying API FRAGMENTER from a web client
- * curl -i "http://localhost:51200/pxf/v13/Metadata/getTableMetadata?table=t1"
- * /pxf/ is made part of the path when there is a webapp by that name in tcServer.
+ * <br>
+ * Example for querying API FRAGMENTER from a web client:<br>
+ * <code>curl -i "http://localhost:51200/pxf/v13/Metadata/getTableMetadata?table=t1"</code><br>
+ * /pxf/ is made part of the path when there is a webapp by that name in tomcat.
*/
@Path("/" + Version.PXF_PROTOCOL_VERSION + "/Metadata/")
public class MetadataResource extends RestResource {
@@ -40,14 +41,21 @@ public class MetadataResource extends RestResource {
* This function queries the HiveMetaStore to get the given table's metadata:
* Table name, field names, field types.
* The types are converted from HCatalog types to HAWQ types.
- * Supported HCatalog types:
- * TINYINT, SMALLINT, INT, BIGINT, BOOLEAN, FLOAT, DOUBLE,
- * STRING, BINARY, TIMESTAMP, DATE, DECIMAL, VARCHAR, CHAR.
+ * Supported HCatalog types:
+ * TINYINT, SMALLINT, INT, BIGINT, BOOLEAN, FLOAT, DOUBLE,
+ * STRING, BINARY, TIMESTAMP, DATE, DECIMAL, VARCHAR, CHAR.
+ * <br>
* Unsupported types result in an error.
- *
- * Response Examples:
- * For a table default.t1 with 2 fields (a int, b float) will be returned as:
- * {"PXFMetadata":[{"table":{"dbName":"default","tableName":"t1"},"fields":[{"name":"a","type":"int"},{"name":"b","type":"float"}]}]}
+ * <br>
+ * Response Examples:<br>
+ * For a table <code>default.t1</code> with 2 fields (a int, b float) will be returned as:
+ * <code>{"PXFMetadata":[{"table":{"dbName":"default","tableName":"t1"},"fields":[{"name":"a","type":"int"},{"name":"b","type":"float"}]}]}</code>
+ *
+ * @param servletContext servlet context
+ * @param headers http headers
+ * @param table HCatalog table name
+ * @return JSON formatted response with metadata for given table
+ * @throws Exception if connection to Hcatalog failed, table didn't exist or its type or fields are not supported
*/
@GET
@Path("getTableMetadata")
@@ -58,18 +66,18 @@ public class MetadataResource extends RestResource {
Log.debug("getTableMetadata started");
String jsonOutput;
try {
- // 1. start MetadataFetcher
- MetadataFetcher metadataFetcher =
+ // 1. start MetadataFetcher
+ MetadataFetcher metadataFetcher =
MetadataFetcherFactory.create("com.pivotal.pxf.plugins.hive.HiveMetadataFetcher"); //TODO: nhorn - 09-03-15 - pass as param
-
+
// 2. get Metadata
Metadata metadata = metadataFetcher.getTableMetadata(table);
-
+
// 3. serialize to JSON
jsonOutput = MetadataResponseFormatter.formatResponseString(metadata);
-
+
Log.debug("getTableMetadata output: " + jsonOutput);
-
+
} catch (ClientAbortException e) {
Log.error("Remote connection closed by HAWQ", e);
throw e;
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/RestResource.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/RestResource.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/RestResource.java
index fb62927..a77c967 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/RestResource.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/rest/RestResource.java
@@ -21,10 +21,11 @@ public abstract class RestResource {
/**
* Converts the request headers multivalued map to a case-insensitive regular map
* by taking only first values and storing them in a CASE_INSENSITIVE_ORDER TreeMap.
+ * All values are converted from ISO_8859_1 (ISO-LATIN-1) to UTF_8.
*
* @param requestHeaders request headers multi map.
* @return a regular case-insensitive map.
- * @throws UnsupportedEncodingException
+ * @throws UnsupportedEncodingException if the named charsets ISO_8859_1 and UTF_8 are not supported
*/
public Map<String, String> convertToCaseInsensitiveMap(MultivaluedMap<String, String> requestHeaders)
throws UnsupportedEncodingException {
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/88f5f5f6/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/CustomWebappLoader.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/CustomWebappLoader.java b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/CustomWebappLoader.java
index 3c62cfc..cc78026 100644
--- a/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/CustomWebappLoader.java
+++ b/pxf/pxf-service/src/main/java/com/pivotal/pxf/service/utilities/CustomWebappLoader.java
@@ -18,22 +18,22 @@ import org.apache.juli.logging.LogFactory;
/**
* A WebappLoader that allows a customized classpath to be added through configuration in context xml.
* Any additional classpath entry will be added to the default webapp classpath.
- * <p/>
+ *
* <pre>
- * <Context>
+ * <Context>
* <Loader className="com.pivotal.pxf.service.utilities.CustomWebappLoader"
- * classpathFiles="/somedir/classpathFile1;/somedir/classpathFile2"/>
- * </Context>
+ * classpathFiles="/somedir/classpathFile1;/somedir/classpathFile2"/>
+ * </Context>
* </pre>
*/
public class CustomWebappLoader extends WebappLoader {
-
+
/**
- * Because this class belongs in tcServer itself, logs go into tcServer's log facility that is separate
+ * Because this class belongs in tcServer itself, logs go into tcServer's log facility that is separate
* from the web app's log facility.
- *
+ *
* Logs are directed to catalina.log file. By default only INFO or higher messages are logged.
- * To change log level, add the following line to {catalina.base}/conf/logging.properties
+ * To change log level, add the following line to {catalina.base}/conf/logging.properties
* <code>com.pivotal.pxf.level = FINE/INFO/WARNING</code> (FINE = debug).
*/
private static final Log LOG = LogFactory.getLog(CustomWebappLoader.class);
@@ -46,9 +46,9 @@ public class CustomWebappLoader extends WebappLoader {
* anotherdir/somejar.jar
* anotherone/hadoop*.jar
* anotherone/pxf*[0-9].jar
- * Unix wildcard convention can be used to match a number of files
+ * Unix wildcard convention can be used to match a number of files
* (e.g. <code>*</code>, <code>[0-9]</code>, <code>?</code>), but not a number of directories.
- *
+ *
* The files specified under classpathFiles must exist - if they can't be read an exception will be thrown.
*/
private String classpathFiles;
@@ -101,35 +101,35 @@ public class CustomWebappLoader extends WebappLoader {
addRepositories(classpathFiles, true);
addRepositories(secondaryClasspathFiles, false);
-
+
super.startInternal();
}
private void addRepositories(String classpathFiles, boolean throwException) throws LifecycleException {
-
+
for (String classpathFile : classpathFiles.split(";")) {
-
+
String classpath = readClasspathFile(classpathFile, throwException);
if (classpath == null) {
- continue;
+ continue;
}
-
+
ArrayList<String> classpathEntries = trimEntries(classpath.split("[\\r\\n]+"));
LOG.info("Classpath file " + classpathFile + " has " + classpathEntries.size() + " entries");
-
+
for (String entry : classpathEntries) {
LOG.debug("Trying to load entry " + entry);
int repositoriesCount = 0;
Path pathEntry = Paths.get(entry);
/*
- * For each entry, we look at the parent directory and try to match each of the files
+ * For each entry, we look at the parent directory and try to match each of the files
* inside it to the file name or pattern in the file name (the last part of the path).
* E.g., for path '/some/path/with/pattern*', the parent directory will be '/some/path/with/'
- * and the file name will be 'pattern*'. Each file under that directory matching
- * this pattern will be added to the class loader repository.
+ * and the file name will be 'pattern*'. Each file under that directory matching
+ * this pattern will be added to the class loader repository.
*/
try (DirectoryStream<Path> repositories = Files.newDirectoryStream(pathEntry.getParent(),
- pathEntry.getFileName().toString())) {
+ pathEntry.getFileName().toString())) {
for (Path repository : repositories) {
if (addPathToRepository(repository, entry)) {
repositoriesCount++;
@@ -137,7 +137,7 @@ public class CustomWebappLoader extends WebappLoader {
}
} catch (IOException e) {
LOG.warn("Failed to load entry " + entry + ": " + e);
- }
+ }
if (repositoriesCount == 0) {
LOG.warn("Entry " + entry + " doesn't match any files");
}
@@ -166,17 +166,17 @@ public class CustomWebappLoader extends WebappLoader {
* @return valid entries
*/
private ArrayList<String> trimEntries(String[] classpathEntries) {
-
+
ArrayList<String> trimmed = new ArrayList<String>();
int line = 0;
for (String entry : classpathEntries) {
-
+
line++;
if (entry == null) {
LOG.debug("Skipping entry #" + line + " (null)");
continue;
}
-
+
entry = entry.trim();
if (entry.isEmpty() || entry.startsWith("#")) {
LOG.debug("Skipping entry #" + line + " (" + entry + ")");
@@ -186,9 +186,9 @@ public class CustomWebappLoader extends WebappLoader {
}
return trimmed;
}
-
+
private boolean addPathToRepository(Path path, String entry) {
-
+
try {
URI pathUri = path.toUri();
String pathUriStr = pathUri.toString();
@@ -205,7 +205,7 @@ public class CustomWebappLoader extends WebappLoader {
return false;
}
-
+
}