You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ga...@apache.org on 2008/02/27 01:52:08 UTC
svn commit: r631443 [2/10] - in /incubator/pig/branches/types: ./
lib-src/bzip2/org/apache/tools/bzip2r/ lib-src/shock/org/apache/pig/shock/
lib/ scripts/ src/org/apache/pig/ src/org/apache/pig/backend/
src/org/apache/pig/backend/datastorage/ src/org/a...
Modified: incubator/pig/branches/types/src/org/apache/pig/PigServer.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/PigServer.java?rev=631443&r1=631442&r2=631443&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/PigServer.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/PigServer.java Tue Feb 26 16:51:49 2008
@@ -19,46 +19,40 @@
import java.io.File;
import java.io.IOException;
-import java.io.InputStream;
import java.io.PrintStream;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+import java.util.Enumeration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
-import java.util.Enumeration;
-import java.util.Vector;
-import java.net.URL;
-import java.net.URI;
-import java.net.URISyntaxException;
+import java.util.Properties;
-import org.apache.hadoop.dfs.DistributedFileSystem;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.FileStatus;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.pig.backend.datastorage.ContainerDescriptor;
+import org.apache.pig.backend.datastorage.DataStorage;
+import org.apache.pig.backend.datastorage.DataStorageException;
+import org.apache.pig.backend.datastorage.ElementDescriptor;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.executionengine.ExecJob;
+import org.apache.pig.backend.executionengine.ExecPhysicalPlan;
+import org.apache.pig.backend.executionengine.ExecJob.JOB_STATUS;
+import org.apache.pig.backend.hadoop.executionengine.mapreduceExec.MapReduceLauncher;
import org.apache.pig.builtin.PigStorage;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
-import org.apache.pig.impl.io.BufferedPositionedInputStream;
import org.apache.pig.impl.io.FileLocalizer;
-import org.apache.pig.impl.io.FileSpec;
-import org.apache.pig.impl.logicalLayer.LOLoad;
-import org.apache.pig.impl.logicalLayer.LORead;
-import org.apache.pig.impl.logicalLayer.LOStore;
import org.apache.pig.impl.logicalLayer.LogicalOperator;
import org.apache.pig.impl.logicalLayer.LogicalPlan;
-import org.apache.pig.impl.logicalLayer.LOVisitor;
-import org.apache.pig.impl.logicalLayer.LOPrinter;
import org.apache.pig.impl.logicalLayer.LogicalPlanBuilder;
+import org.apache.pig.impl.logicalLayer.OperatorKey;
import org.apache.pig.impl.logicalLayer.parser.ParseException;
+import org.apache.pig.impl.logicalLayer.parser.QueryParser;
import org.apache.pig.impl.logicalLayer.schema.TupleSchema;
-import org.apache.pig.impl.mapreduceExec.MapReduceLauncher;
-import org.apache.pig.impl.physicalLayer.IntermedResult;
-import org.apache.pig.impl.physicalLayer.POMapreduce;
-import org.apache.pig.impl.physicalLayer.POStore;
-import org.apache.pig.impl.physicalLayer.PhysicalPlan;
-import org.apache.pig.impl.physicalLayer.POVisitor;
-import org.apache.pig.impl.physicalLayer.POPrinter;
-import org.apache.pig.impl.util.PigLogger;
-
+import org.apache.pig.impl.util.WrappedIOException;
/**
@@ -70,6 +64,8 @@
*/
public class PigServer {
+ private final Log log = LogFactory.getLog(getClass());
+
/**
* The type of query execution
*/
@@ -99,41 +95,45 @@
throw new IOException("Unrecognized exec type: " + str);
}
- /**
- * a table mapping intermediate results to physical plans that read them (the state
- * of how much has been read is maintained in PORead
- */
- Map<IntermedResult, PhysicalPlan> physicalPlans = new HashMap<IntermedResult, PhysicalPlan>();
- /**
- * a table mapping ID's to intermediate results
- */
- Map<String, IntermedResult> queryResults = new HashMap<String, IntermedResult>();
+
+ Map<String, LogicalPlan> aliases = new HashMap<String, LogicalPlan>();
+ Map<OperatorKey, LogicalOperator> opTable = new HashMap<OperatorKey, LogicalOperator>();
+
PigContext pigContext;
- public PigServer(String execType) throws IOException {
+ private String scope = constructScope();
+
+ private String constructScope() {
+ // scope servers for now as a session id
+ // scope = user_id + "-" + time_stamp;
+
+ String user = System.getProperty("user.name", "DEFAULT_USER_ID");
+ String date = (new Date()).toString();
+
+ return user + "-" + date;
+ }
+
+ public PigServer(String execType) throws ExecException, IOException {
this(parseExecType(execType));
}
- public PigServer() {
- this(ExecType.MAPREDUCE);
+ public PigServer() throws ExecException {
+ this(ExecType.MAPREDUCE);
}
- public PigServer(ExecType execType) {
+ public PigServer(ExecType execType) throws ExecException {
this.pigContext = new PigContext(execType);
pigContext.connect();
}
- public PigServer(PigContext context) {
+ public PigServer(PigContext context) throws ExecException {
this.pigContext = context;
pigContext.connect();
}
-
-
-
-
+
public PigContext getPigContext(){
- return pigContext;
+ return pigContext;
}
public void debugOn() {
@@ -154,7 +154,7 @@
* It should have the form: classname('arg1', 'arg2', ...)
*/
public void registerFunction(String function, String functionSpec) {
- pigContext.registerFunction(function, functionSpec);
+ pigContext.registerFunction(function, functionSpec);
}
private URL locateJarFromResources(String jarName) throws IOException {
@@ -162,7 +162,7 @@
URL resourceLocation = null;
if (urls.hasMoreElements()) {
- resourceLocation = urls.nextElement();
+ resourceLocation = urls.nextElement();
}
if (pigContext.debug && urls.hasMoreElements()) {
@@ -170,10 +170,10 @@
+ jarName + ": " + resourceLocation;
while (urls.hasMoreElements()) {
- logMessage += (logMessage + urls.nextElement() + "; ");
+ logMessage += (logMessage + urls.nextElement() + "; ");
}
- PigLogger.getLogger().debug(logMessage);
+ log.debug(logMessage);
}
return resourceLocation;
@@ -194,7 +194,7 @@
// first try to locate jar via system resources
// if this fails, try by using "name" as File (this preserves
// compatibility with case when user passes absolute path or path
- // relative to current working directory.)
+ // relative to current working directory.)
if (name != null) {
URL resource = locateJarFromResources(name);
@@ -208,7 +208,7 @@
resource = f.toURI().toURL();
}
- pigContext.addJar(resource);
+ pigContext.addJar(resource);
}
}
@@ -222,7 +222,7 @@
* @throws IOException
*/
public void registerQuery(String query) throws IOException {
- // Bugzilla Bug 1006706 -- ignore empty queries
+ // Bugzilla Bug 1006706 -- ignore empty queries
//=============================================
if(query != null) {
query = query.trim();
@@ -234,147 +234,62 @@
// parse the query into a logical plan
LogicalPlan lp = null;
try {
- lp = (new LogicalPlanBuilder(pigContext).parse(query, queryResults));
- // System.out.println(lp.toString());
+ lp = (new LogicalPlanBuilder(pigContext).parse(scope, query, aliases, opTable));
} catch (ParseException e) {
throw (IOException) new IOException(e.getMessage()).initCause(e);
}
- if (lp.getAlias() != null)
- queryResults.put(lp.getAlias(), new IntermedResult(lp, pigContext));
+
+ if (lp.getAlias() != null) {
+ aliases.put(lp.getAlias(), lp);
+ }
}
public void dumpSchema(String alias) throws IOException{
- IntermedResult ir = queryResults.get(alias);
- if (ir == null)
- throw new IOException("Invalid alias - " + alias);
+ LogicalPlan lp = aliases.get(alias);
+ if (lp == null)
+ throw new IOException("Invalid alias - " + alias);
- TupleSchema schema = ir.lp.getRoot().outputSchema();
+ TupleSchema schema = lp.getOpTable().get(lp.getRoot()).outputSchema();
- System.out.println(schema.toString());
+ System.out.println(schema.toString());
}
public void setJobName(String name){
pigContext.setJobName(name);
}
-
- public void newRelation(String id) {
- queryResults.put(id, new IntermedResult());
- }
-
- public void insertTuple(String id, Tuple t) throws IOException {
- if (!queryResults.containsKey(id)) throw new IOException("Attempt to insert tuple into nonexistent relation.");
- queryResults.get(id).add(t);
- }
-
-
- public Iterator<Tuple> openIterator(String id, boolean continueFromLast) throws IOException{
- if (!continueFromLast)
- return openIterator(id);
-
- if (pigContext.getExecType()!= ExecType.LOCAL){
- System.err.println("Streaming execution not supported in non-local mode.");
- System.exit(-1);
- }
-
- if (!queryResults.containsKey(id))
- throw new IOException("Invalid alias: " + id);
-
- IntermedResult readFrom = (IntermedResult) queryResults.get(id);
-
- if (readFrom.getOutputType()==LogicalOperator.FIXED){
- //Its not a continuous plan, just let it go through the non-continuous channel
- return openIterator(id);
- }
-
- PhysicalPlan pp = null;
-
- if (!physicalPlans.containsKey(readFrom)){
-
- //Not trying to do sharing, so won't compile and exec it as in the other cases
- //First check if some other operator tried to execute this intermediate result
- if (readFrom.executed() || readFrom.compiled()){
- //Compile it again, so that POSlaves are not added in between query execution
- readFrom.compile(queryResults);
- }
- LogicalPlan lp = new LogicalPlan(new LORead(readFrom), pigContext);
- pp = new PhysicalPlan(lp,queryResults);
- physicalPlans.put(readFrom, pp);
- }else{
- pp = physicalPlans.get(readFrom);
- }
-
- return pp.exec(continueFromLast).iterator();
-
- }
/**
* Forces execution of query (and all queries from which it reads), in order to materialize
* result
*/
public Iterator<Tuple> openIterator(String id) throws IOException {
- if (!queryResults.containsKey(id))
+ if (!aliases.containsKey(id))
throw new IOException("Invalid alias: " + id);
- IntermedResult readFrom = (IntermedResult) queryResults.get(id);
+ // TODO: front-end could actually remember what logical plans have been
+ // already submitted to the back-end for compilation and
+ // execution.
+
+ LogicalPlan readFrom = (LogicalPlan) aliases.get(id);
- readFrom.compile(queryResults);
- readFrom.exec();
- if (pigContext.getExecType() == ExecType.LOCAL)
- return readFrom.read().iterator();
- final LoadFunc p;
-
- try{
- p = (LoadFunc)PigContext.instantiateFuncFromSpec(readFrom.outputFileSpec.getFuncSpec());
- }catch (Exception e){
- IOException ioe = new IOException(e.getMessage());
- ioe.setStackTrace(e.getStackTrace());
- throw ioe;
- }
- InputStream is = FileLocalizer.open(readFrom.outputFileSpec.getFileName(), pigContext);
- p.bindTo(readFrom.outputFileSpec.getFileName(), new BufferedPositionedInputStream(is), 0, Long.MAX_VALUE);
- return new Iterator<Tuple>() {
- Tuple t;
- boolean atEnd;
-
- public boolean hasNext() {
- if (atEnd)
- return false;
- try {
- if (t == null)
- t = p.getNext();
- if (t == null)
- atEnd = true;
- } catch (Exception e) {
- e.printStackTrace();
- t = null;
- atEnd = true;
- }
- return !atEnd;
- }
+ try {
+ ExecPhysicalPlan pp =
+ pigContext.getExecutionEngine().compile(readFrom, null);
+
+ ExecJob job = pigContext.getExecutionEngine().execute(pp);
- public Tuple next() {
- Tuple next = t;
- if (next != null) {
- t = null;
- return next;
- }
- try {
- next = p.getNext();
- } catch (Exception e) {
- e.printStackTrace();
- }
- if (next == null)
- atEnd = true;
- return next;
+ // invocation of "execute" is synchronous!
+ if (job.getStatus() == JOB_STATUS.COMPLETED) {
+ return job.getResults();
}
-
- public void remove() {
- throw new RuntimeException("Removal not supported");
+ else {
+ throw new IOException("Job terminated with anomalous status " + job.getStatus().toString());
}
-
- };
-
- }
+ }
+ catch (ExecException e) {
+ throw WrappedIOException.wrap("Unable to open iterator for alias: " + id, e);
+ }
+ }
/**
* Store an alias into a file
@@ -382,145 +297,45 @@
* @param filename: The file to which to store to
* @throws IOException
*/
-
- public void store(String id, String filename) throws IOException {
- store(id, filename, PigStorage.class.getName() + "()"); // SFPig is the default store function
- }
-
- /**
- * Continuous case of store: store the updates to particular alias into the given file
- * @param id
- * @param filename
- * @throws IOException
- */
-
- public void update(String id, String filename, boolean append) throws IOException {
- update(id, filename, PigStorage.class.getName() + "()", append);
- }
-
- /**
- * forces execution of query (and all queries from which it reads), in order to store result in file
- */
- public void store(String id, String filename, String func) throws IOException{
- if (!queryResults.containsKey(id))
+
+ public void store(String id, String filename) throws IOException {
+ store(id, filename, PigStorage.class.getName() + "()"); // SFPig is the default store function
+ }
+
+ /**
+ * forces execution of query (and all queries from which it reads), in order to store result in file
+ */
+ public void store(String id, String filename, String func) throws IOException{
+ if (!aliases.containsKey(id))
throw new IOException("Invalid alias: " + id);
if (FileLocalizer.fileExists(filename, pigContext))
- throw new IOException("Output file " + filename + " already exists. Can't overwrite.");
+ throw new IOException("Output file " + filename + " already exists. Can't overwrite.");
- IntermedResult readFrom = (IntermedResult) queryResults.get(id);
+ LogicalPlan readFrom = aliases.get(id);
store(readFrom,filename,func);
- }
+ }
- public void store(IntermedResult readFrom, String filename, String func) throws IOException{
- if (func == null)
- func = PigStorage.class.getName();
-
- filename = removeQuotes(filename);
- FileSpec fileSpec = new FileSpec(filename,func);
-
- if (pigContext.getExecType() == ExecType.LOCAL){
- if (!readFrom.executed()){
- readFrom.compile(queryResults);
- readFrom.exec();
- }
- // generate a simple logical plan to store the results
- LogicalPlan lp = new LogicalPlan(new LOStore(new LORead(readFrom), new FileSpec(filename, func),false), pigContext);
+ public void store(LogicalPlan readFrom, String filename, String func) throws IOException {
+ LogicalPlan storePlan = QueryParser.generateStorePlan(readFrom.getOpTable(),
+ scope,
+ readFrom,
+ filename,
+ func,
+ pigContext);
- // temporary code to test the ExampleGenerator:
- //ExGen.PrintExamples(lp, ExGen.GenerateExamples(lp, pigContext));
-
- // then execute it to write to disk
- (new PhysicalPlan(lp, queryResults)).exec(false);
- return;
- }else{
- if (!readFrom.executed()) {
- readFrom.compile(queryResults);
- if (pigContext.getExecType() == ExecType.MAPREDUCE){
- POMapreduce pom = (POMapreduce)readFrom.pp.root;
- pom.outputFileSpec = fileSpec;
- }
- readFrom.setPermanentFileSpec(fileSpec);
- readFrom.exec();
- }else{
- if (readFrom.outputFileSpec == null){
- readFrom.toDFSFile(fileSpec, pigContext);
- }else{
- String src = readFrom.outputFileSpec.getFileName();
- String dst = filename;
- boolean islocal = false;
- if (dst.startsWith(FileLocalizer.LOCAL_PREFIX)) {
- dst = dst.substring(FileLocalizer.LOCAL_PREFIX.length());
- islocal = true;
- }
- if (readFrom.outputFileSpec.getFuncSpec().equals(func)){
- if (readFrom.isTemporary()) {
- if (islocal) {
- pigContext.copy(src, dst, islocal);
- } else {
- pigContext.rename(src, dst);
- }
- readFrom.setPermanentFileSpec(new FileSpec(filename,func));
- }else{
- pigContext.copy(src, dst, islocal);
- }
- }else{
- try{
- store(new IntermedResult(new LogicalPlan(new LOLoad(readFrom.outputFileSpec),pigContext), pigContext), filename, func);
- } catch (ParseException e) {
- throw (IOException) new IOException(e.getMessage()).initCause(e);
- }
- }
- }
- }
+ try {
+ ExecPhysicalPlan pp =
+ pigContext.getExecutionEngine().compile(storePlan, null);
+
+ pigContext.getExecutionEngine().execute(pp);
+ }
+ catch (ExecException e) {
+ throw WrappedIOException.wrap("Unable to store alias " + readFrom.getAlias(), e);
}
}
- /**
- * Continuous version of the store function above
- *
- */
- public void update(String id, String filename, String func, boolean append) throws IOException{
-
-
- if (pigContext.getExecType()!=ExecType.LOCAL){
- System.err.println("Streaming execution not supported in non-local mode.");
- System.exit(-1);
- }
-
- filename = removeQuotes(filename);
-
- if (!queryResults.containsKey(id))
- throw new IOException("Invalid alias: " + id);
-
- IntermedResult readFrom = (IntermedResult) queryResults.get(id);
-
- if (readFrom.getOutputType()==LogicalOperator.FIXED){
- //Its not a continuous plan, just let it go through the non-continuous channel
- store(id,filename,func);
- return;
- }
-
- PhysicalPlan pp = null;
-
- if (!physicalPlans.containsKey(readFrom)){
-
- //Not trying to do sharing, so won't compile and exec it as in the other cases
- //First check if some other operator tried to execute this intermediate result
- if (readFrom.executed() || readFrom.compiled()){
- //Compile it again, so that POSlaves are not added in between query execution
- readFrom.compile(queryResults);
- }
- LogicalPlan lp = new LogicalPlan(new LORead(readFrom), pigContext);
- pp = new PhysicalPlan(lp,queryResults);
- physicalPlans.put(readFrom, pp);
- }else{
- pp = physicalPlans.get(readFrom);
- }
- new PhysicalPlan(new POStore(pp.root,new FileSpec(filename,func),append,pigContext)).exec(true);
- }
-
/**
* Provide information on how a pig query will be executed. For now
* this information is very developer focussed, and probably not very
@@ -529,25 +344,31 @@
* @param stream PrintStream to write explanation to.
* @throws IOException if the requested alias cannot be found.
*/
- public void explain(
- String alias,
- PrintStream stream) throws IOException {
+ public void explain(String alias,
+ PrintStream stream) throws IOException {
stream.println("Logical Plan:");
- IntermedResult ir = queryResults.get(alias);
- if (ir == null) {
- PigLogger.getLogger().error("Invalid alias: " + alias);
+ LogicalPlan lp = aliases.get(alias);
+ if (lp == null) {
+ log.error("Invalid alias: " + alias);
+ stream.println("Invalid alias: " + alias);
throw new IOException("Invalid alias: " + alias);
}
- LOVisitor lprinter = new LOPrinter(stream);
- ir.lp.getRoot().visit(lprinter);
-
+ lp.explain(stream);
+
stream.println("-----------------------------------------------");
stream.println("Physical Plan:");
- // have to first compile the plan
- ir.compile(queryResults);
- POVisitor pprinter = new POPrinter(stream);
- ir.pp.root.visit(pprinter);
+ try {
+ ExecPhysicalPlan pp =
+ pigContext.getExecutionEngine().compile(lp, null);
+
+ pp.explain(stream);
+ }
+ catch (ExecException e) {
+ log.error("Failed to compile to physical plan: " + alias);
+ stream.println("Failed to compile the logical plan for " + alias + " into a physical plan");
+ throw WrappedIOException.wrap("Failed to compile to phyiscal plan: " + alias, e);
+ }
}
/**
@@ -562,9 +383,23 @@
public long capacity() throws IOException {
if (pigContext.getExecType() == ExecType.LOCAL) {
throw new IOException("capacity only supported for non-local execution");
- } else {
- DistributedFileSystem dfs = (DistributedFileSystem) pigContext.getDfs();
- return dfs.getRawCapacity() - dfs.getRawUsed();
+ }
+ else {
+ DataStorage dds = pigContext.getDfs();
+
+ Map<String, Object> stats = dds.getStatistics();
+
+ String rawCapacityStr = (String) stats.get(DataStorage.RAW_CAPACITY_KEY);
+ String rawUsedStr = (String) stats.get(DataStorage.RAW_USED_KEY);
+
+ if ((rawCapacityStr == null) || (rawUsedStr == null)) {
+ throw new IOException("Failed to retrieve capacity stats");
+ }
+
+ long rawCapacityBytes = new Long(rawCapacityStr).longValue();
+ long rawUsedBytes = new Long(rawUsedStr).longValue();
+
+ return rawCapacityBytes - rawUsedBytes;
}
}
@@ -575,76 +410,66 @@
* @throws IOException
*/
public long fileSize(String filename) throws IOException {
- FileSystem dfs = pigContext.getDfs();
- Path p = new Path(filename);
- long len = dfs.getFileStatus(p).getLen();
- long replication = dfs.getDefaultReplication(); // did not work, for some reason: dfs.getReplication(p);
- return len * replication;
+ DataStorage dfs = pigContext.getDfs();
+ ElementDescriptor elem = dfs.asElement(filename);
+ Map<String, Object> elemProps = elem.getStatistics();
+ String length = (String) elemProps.get(ElementDescriptor.LENGTH_KEY);
+
+ Properties dfsProps = dfs.getConfiguration();
+ String replication = dfsProps.getProperty(DataStorage.DEFAULT_REPLICATION_FACTOR_KEY);
+
+ return (new Long(length)).longValue() * (new Integer(replication)).intValue();
}
public boolean existsFile(String filename) throws IOException {
- return pigContext.getDfs().exists(new Path(filename));
+ ElementDescriptor elem = pigContext.getDfs().asElement(filename);
+ return elem.exists();
}
public boolean deleteFile(String filename) throws IOException {
- return pigContext.getDfs().delete(new Path(filename));
+ ElementDescriptor elem = pigContext.getDfs().asElement(filename);
+ elem.delete();
+ return true;
}
public boolean renameFile(String source, String target) throws IOException {
- return pigContext.getDfs().rename(new Path(source), new Path(target));
+ pigContext.rename(source, target);
+ return true;
}
public boolean mkdirs(String dirs) throws IOException {
- return pigContext.getDfs().mkdirs(new Path(dirs));
+ ContainerDescriptor container = pigContext.getDfs().asContainer(dirs);
+ container.create();
+ return true;
}
public String[] listPaths(String dir) throws IOException {
- FileStatus stats[] = pigContext.getDfs().listStatus(new Path(dir));
- String strPaths[] = new String[stats.length];
- for (int i = 0; i < stats.length; i++) {
- strPaths[i] = stats[i].getPath().toString();
+ Collection<String> allPaths = new ArrayList<String>();
+ ContainerDescriptor container = pigContext.getDfs().asContainer(dir);
+ Iterator<ElementDescriptor> iter = container.iterator();
+
+ while (iter.hasNext()) {
+ ElementDescriptor elem = iter.next();
+ allPaths.add(elem.toString());
}
- return strPaths;
+
+ return (String[])(allPaths.toArray());
}
public long totalHadoopTimeSpent() {
return MapReduceLauncher.totalHadoopTimeSpent;
}
-
-// public static class PStat {
-// // These are the members of the unix fstat and stat structure
-// // we currently don't implement all of them, but try to provide
-// // a mirror for the function and values for HDFS files
-// // ====================================================
-// // struct stat {
-// // dev_t st_dev; /* device inode resides on */
-// // ino_t st_ino; /* inode's number */
-// // mode_t st_mode; /* inode protection mode */
-// // nlink_t st_nlink; /* number or hard links to the file */
-// // uid_t st_uid; /* user-id of owner */
-// // gid_t st_gid; /* group-id of owner */
-// // dev_t st_rdev; /* device type, for special file inode */
-// // struct timespec st_atimespec; /* time of last access */
-// // struct timespec st_mtimespec; /* time of last data modification */
-// // struct timespec st_ctimespec; /* time of last file status change */
-// // off_t st_size; /* file size, in bytes */
-// // quad_t st_blocks; /* blocks allocated for file */
-// // u_long st_blksize;/* optimal file sys I/O ops blocksize */
-// // u_long st_flags; /* user defined flags for file */
-// // u_long st_gen; /* file generation number */
-// // };
-// public long st_size;
-// public long st_blocks;
-// }
-
- private String removeQuotes(String str) {
- if (str.startsWith("\'") && str.endsWith("\'"))
- return str.substring(1, str.length() - 1);
- else
- return str;
- }
-
- public Map<String, IntermedResult> getQueryResults() {
- return queryResults;
- }
+
+ public Map<String, LogicalPlan> getAliases() {
+ return this.aliases;
+ }
+
+ public void shutdown() {
+ // clean-up activities
+ // TODO: reclaim scope to free up resources. Currently
+ // this is not implemented and throws an exception
+ // hence, for now, we won't call it.
+ //
+ // pigContext.getExecutionEngine().reclaimScope(this.scope);
+ }
}
Modified: incubator/pig/branches/types/src/org/apache/pig/StandAloneParser.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/StandAloneParser.java?rev=631443&r1=631442&r2=631443&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/StandAloneParser.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/StandAloneParser.java Tue Feb 26 16:51:49 2008
@@ -17,18 +17,24 @@
*/
package org.apache.pig;
-import java.io.*;
-import java.util.*;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.Iterator;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.pig.PigServer.ExecType;
-import org.apache.pig.impl.physicalLayer.IntermedResult;
-
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.impl.logicalLayer.LogicalPlan;
public class StandAloneParser {
+ private static final Log log = LogFactory.getLog(StandAloneParser.class);
+
static PigServer pig;
- public static void main(String args[]) throws IOException {
+ public static void main(String args[]) throws IOException, ExecException {
BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
pig = new PigServer(ExecType.LOCAL);
@@ -40,7 +46,7 @@
try {
line = in.readLine();
} catch (IOException e) {
- e.printStackTrace();
+ log.error(e);
return;
}
@@ -54,20 +60,20 @@
}
private static void tryParse(String query) {
- if (query.trim().equals(""))
- return;
+ if (query.trim().equals(""))
+ return;
try{
- pig.registerQuery(query);
- System.out.print("Current aliases: ");
- for (Iterator<String> it = pig.getQueryResults().keySet().iterator(); it.hasNext(); ) {
+ pig.registerQuery(query);
+ System.out.print("Current aliases: ");
+ for (Iterator<String> it = pig.getAliases().keySet().iterator(); it.hasNext(); ) {
String alias = it.next();
- IntermedResult ir = pig.getQueryResults().get(alias);
- System.out.print(alias + "->" + ir.lp.getRoot().outputSchema());
+ LogicalPlan lp = pig.getAliases().get(alias);
+ System.out.print(alias + "->" + lp.getOpTable().get(lp.getRoot()).outputSchema());
if (it.hasNext()) System.out.print(", \n");
else System.out.print("\n");
}
} catch (IOException e) {
- System.err.println(e);
+ log.error(e);
}
}
}
Modified: incubator/pig/branches/types/src/org/apache/pig/StoreFunc.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/StoreFunc.java?rev=631443&r1=631442&r2=631443&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/StoreFunc.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/StoreFunc.java Tue Feb 26 16:51:49 2008
@@ -32,14 +32,14 @@
*/
public interface StoreFunc {
-
- /**
- * Specifies the OutputStream to write to. This will be called before
- * store(Tuple) is invoked.
- *
- * @param os The stream to write tuples to.
- * @throws IOException
- */
+
+ /**
+ * Specifies the OutputStream to write to. This will be called before
+ * store(Tuple) is invoked.
+ *
+ * @param os The stream to write tuples to.
+ * @throws IOException
+ */
public abstract void bindTo(OutputStream os) throws IOException;
/**
@@ -51,7 +51,7 @@
*/
public abstract void putNext(Tuple f) throws IOException;
- /**
+ /**
* Do any kind of post processing because the last tuple has been
* stored. DO NOT CLOSE THE STREAM in this method. The stream will be
* closed later outside of this function.
Added: incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/ContainerDescriptor.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/ContainerDescriptor.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/ContainerDescriptor.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/ContainerDescriptor.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.datastorage;
+
+public interface ContainerDescriptor
+ extends ElementDescriptor,
+ Iterable<ElementDescriptor> {
+}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/DataStorage.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/DataStorage.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/DataStorage.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/DataStorage.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.datastorage;
+
+import java.util.Properties;
+import java.util.Map;
+import java.io.IOException;
+
+/**
+ * DataStorage provides an abstraction of a generic container.
+ * Special instances of it can be a file system.
+ *
+ */
+
+public interface DataStorage {
+
+ public static final String DEFAULT_REPLICATION_FACTOR_KEY = "pig.default.replication.factor";
+ public static final String USED_BYTES_KEY = "pig.used.bytes";
+ public static final String RAW_CAPACITY_KEY = "pig.raw.capacity.bytes"; // replication is disregarded
+ public static final String RAW_USED_KEY = "pig.raw.used.capacity.bytes"; // replication is disregarded
+
+ //
+ // TODO: more keys
+ //
+
+ /**
+ * Place holder for possible initialization activities.
+ */
+ public void init();
+
+ /**
+ * Clean-up and releasing of resources.
+ */
+ public void close() throws IOException;
+
+ /**
+ * Provides configuration information about the storage itself.
+ * For instance global data-replication policies if any, default
+ * values, ... Some of such values could be overridden at a finer
+ * granularity (e.g. on a specific object in the Data Storage)
+ *
+ * @return - configuration information
+ */
+ public Properties getConfiguration();
+
+ /**
+ * Provides a way to change configuration parameters
+ * at the Data Storage level. For instance, change the
+ * data replication policy.
+ *
+ * @param newConfiguration - the new configuration settings
+ * @throws when configuration conflicts are detected
+ *
+ */
+ public void updateConfiguration(Properties newConfiguration)
+ throws DataStorageException;
+
+ /**
+ * Provides statistics on the Storage: capacity values, how much
+ * storage is in use...
+ * @return statistics on the Data Storage
+ */
+ public Map<String, Object> getStatistics() throws IOException;
+
+ /**
+ * Creates an entity handle for an object (no containment
+ * relation) from a String
+ *
+ * @param name of the object
+ * @return an object descriptor
+ * @throws DataStorageException if name does not conform to naming
+ * convention enforced by the Data Storage.
+ */
+ public ElementDescriptor asElement(String name)
+ throws DataStorageException;
+
+ public ElementDescriptor asElement(ElementDescriptor element)
+ throws DataStorageException;
+
+ public ElementDescriptor asElement(String parent,
+ String child)
+ throws DataStorageException;
+
+ public ElementDescriptor asElement(ContainerDescriptor parent,
+ String child)
+ throws DataStorageException;
+
+ public ElementDescriptor asElement(ContainerDescriptor parent,
+ ElementDescriptor child)
+ throws DataStorageException;
+
+ public boolean isContainer(String name) throws DataStorageException;
+
+ /**
+ * Created an entity handle for a container.
+ *
+ * @param name of the container
+ * @return a container descriptor
+ * @throws DataStorageException if name does not conform to naming
+ * convention enforced by the Data Storage.
+ */
+ public ContainerDescriptor asContainer(String name)
+ throws DataStorageException;
+
+ public ContainerDescriptor asContainer(ContainerDescriptor container)
+ throws DataStorageException;
+
+ public ContainerDescriptor asContainer(String parent,
+ String child)
+ throws DataStorageException;
+
+ public ContainerDescriptor asContainer(ContainerDescriptor parent,
+ String child)
+ throws DataStorageException;
+
+ public ContainerDescriptor asContainer(ContainerDescriptor parent,
+ ContainerDescriptor child)
+ throws DataStorageException;
+
+ public ElementDescriptor[] asCollection(String pattern)
+ throws DataStorageException;
+
+ public void setActiveContainer(ContainerDescriptor container);
+
+ public ContainerDescriptor getActiveContainer();
+}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/DataStorageException.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/DataStorageException.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/DataStorageException.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/DataStorageException.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.datastorage;
+
+import java.io.IOException;
+
+public class DataStorageException extends IOException {
+
+ static final long serialVersionUID = 1;
+
+ public DataStorageException(String message, Throwable cause) {
+ super(message);
+ initCause(cause);
+ }
+
+ public DataStorageException() {
+ this(null, null);
+ }
+
+ public DataStorageException(String message) {
+ this(message, null);
+ }
+
+ public DataStorageException(Throwable cause) {
+ this(null, cause);
+ }
+}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/ElementDescriptor.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/ElementDescriptor.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/ElementDescriptor.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/ElementDescriptor.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.datastorage;
+
+import java.io.OutputStream;
+import java.io.InputStream;
+import java.io.IOException;
+
+import java.util.Properties;
+import java.util.Map;
+
+/**
+ * DataStorageElementDescriptor provides methods necessary to manage an
+ * element in a DataStorage.
+ *
+ */
+
+public interface ElementDescriptor extends
+ Comparable<ElementDescriptor> {
+
+ public static final String BLOCK_SIZE_KEY = "pig.path.block.size";
+ public static final String BLOCK_REPLICATION_KEY = "pig.path.block.replication";
+ public static final String LENGTH_KEY = "pig.path.length";
+ public static final String MODIFICATION_TIME_KEY = "pig.path.modification.time";
+
+ //
+ // TODO: more keys
+ //
+
+ public DataStorage getDataStorage();
+
+ /**
+ * Opens a stream onto which an entity can be written to.
+ *
+ * @param configuration information at the object level
+ * @return stream where to write
+ * @throws DataStorageException
+ */
+ public OutputStream create(Properties configuration)
+ throws IOException;
+
+ public OutputStream create()
+ throws IOException;
+
+ /**
+ * Copy entity from an existing one, possibly residing in a
+ * different Data Storage.
+ *
+ * @param dstName name of entity to create
+ * @param dstConfiguration configuration for the new entity
+ * @param removeSrc if src entity needs to be removed after copying it
+ * @throws DataStorageException for instance, configuration
+ * information for new entity is not compatible with
+ * configuration information at the Data
+ * Storage level, user does not have privileges to read from
+ * source entity or write to destination storage...
+ */
+ public void copy(ElementDescriptor dstName,
+ Properties dstConfiguration,
+ boolean removeSrc)
+ throws IOException;
+
+ public void copy(ElementDescriptor dstName,
+ boolean removeSrc)
+ throws IOException;
+
+ /**
+ * Open for read a given entity
+ *
+ * @param configuration
+ * @return entity to read from
+ * @throws DataStorageExecption e.g. entity does not exist...
+ */
+ public InputStream open(Properties configuration) throws IOException;
+
+ public InputStream open() throws IOException;
+
+ /**
+ * Open an element in the Data Storage with support for random access
+ * (seek operations).
+ *
+ * @param configuration
+ * @return a seekable input stream
+ * @throws DataStorageException
+ */
+ public SeekableInputStream sopen(Properties configuration)
+ throws IOException;
+
+ public SeekableInputStream sopen() throws IOException;
+ /**
+ * Checks whether the entity exists or not
+ *
+ * @param name of entity
+ * @return true if entity exists, false otherwise.
+ */
+ public boolean exists() throws IOException;
+
+ /**
+ * Changes the name of an entity in the Data Storage
+ *
+ * @param newName new name of entity
+ * @throws DataStorageException
+ */
+ public void rename(ElementDescriptor newName)
+ throws IOException;
+
+ /**
+ * Remove entity from the Data Storage.
+ *
+ * @throws DataStorageException
+ */
+ public void delete() throws IOException;
+
+ /**
+ * Retrieve configuration information for entity
+ * @return configuration
+ */
+ public Properties getConfiguration() throws IOException;
+
+ /**
+ * Update configuration information for this entity
+ *
+ * @param newConfig configuration
+ * @throws DataStorageException
+ */
+ public void updateConfiguration(Properties newConfig)
+ throws IOException;
+
+ /**
+ * List entity statistics
+ * @return DataStorageProperties
+ */
+ public Map<String, Object> getStatistics() throws IOException;
+}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/ImmutableOutputStream.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/ImmutableOutputStream.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/ImmutableOutputStream.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/ImmutableOutputStream.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.datastorage;
+
+import java.io.OutputStream;
+import java.io.IOException;
+
+public class ImmutableOutputStream extends OutputStream {
+
+ private String destination;
+
+ public ImmutableOutputStream(String destination) {
+ super();
+ this.destination = destination;
+ }
+
+ public void write(int b) throws IOException {
+ throw new IOException("Write not supported on " + destination);
+ }
+}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/SeekableInputStream.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/SeekableInputStream.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/SeekableInputStream.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/datastorage/SeekableInputStream.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.datastorage;
+
+import java.io.InputStream;
+import java.io.IOException;
+
+/**
+ * Unix-like API for an input stream that supports random access.
+ *
+ */
+public abstract class SeekableInputStream extends InputStream {
+
+ public enum FLAGS {
+ SEEK_SET,
+ SEEK_CUR,
+ SEEK_END,
+ }
+
+ /**
+ * Seeks to a given offset as specified by whence flags.
+ * If whence is SEEK_SET, offset is added to beginning of stream
+ * If whence is SEEK_CUR, offset is added to current position inside stream
+ * If whence is SEEK_END, offset is added to end of file position
+ *
+ * @param offset
+ * @param whence
+ * @throws IOException
+ */
+ public abstract void seek(long offset, FLAGS whence) throws IOException;
+
+ /**
+ * Returns current offset
+ *
+ * @return offset
+ * @throws IOException
+ */
+ public abstract long tell() throws IOException;
+}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecException.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecException.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecException.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecException.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.executionengine;
+
+public class ExecException extends Exception {
+ static final long serialVersionUID = 1;
+
+ public ExecException (String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public ExecException() {
+ this(null, null);
+ }
+
+ public ExecException(String message) {
+ this(message, null);
+ }
+
+ public ExecException(Throwable cause) {
+ this(null, cause);
+ }
+}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecJob.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecJob.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecJob.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecJob.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.executionengine;
+
+import java.util.Iterator;
+import java.util.Properties;
+import java.util.Map;
+import java.io.OutputStream;
+
+import org.apache.pig.data.Tuple;
+
+/**
+ * Abstraction on a job that the execution engine runs. It allows the front-end to
+ * retrieve information on job status and manage a running job.
+ *
+ */
+
+public interface ExecJob {
+
+ public enum JOB_STATUS {
+ QUEUED,
+ RUNNING,
+ SUSPENDED,
+ TERMINATED,
+ FAILED,
+ COMPLETED,
+ }
+
+ public static final String PROGRESS_KEY = "job.progress";
+
+ public JOB_STATUS getStatus();
+
+ /**
+ * true is the physical plan has executed successfully and results are ready
+ * to be retrieved
+ *
+ * @return
+ * @throws ExecException
+ */
+ public boolean hasCompleted() throws ExecException;
+
+ /**
+ * if query has executed successfully we want to retrieve the results
+ * via iterating over them.
+ *
+ * @return
+ * @throws ExecException
+ */
+ public Iterator<Tuple> getResults() throws ExecException;
+
+ /**
+ * Get configuration information
+ *
+ * @return
+ */
+ public Properties getContiguration();
+
+ /**
+ * Can be information about the state (not submitted, e.g. the execute method
+ * has not been called yet; not running, e.g. execute has been issued,
+ * but job is waiting; running...; completed; aborted...; progress information
+ *
+ * @return
+ */
+ public Map<String, Object> getStatistics();
+
+ /**
+ * hook for asynchronous notification of job completion pushed from the back-end
+ */
+ public void completionNotification(Object cookie);
+
+ /**
+ * Kills current job.
+ *
+ * @throws ExecException
+ */
+ public void kill() throws ExecException;
+
+ /**
+ * Collecting various forms of outputs
+ */
+ public void getLogs(OutputStream log) throws ExecException;
+
+ public void getSTDOut(OutputStream out) throws ExecException;
+
+ public void getSTDError(OutputStream error) throws ExecException;
+}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecLogicalPlan.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecLogicalPlan.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecLogicalPlan.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecLogicalPlan.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.executionengine;
+
+import java.io.OutputStream;
+import java.io.Serializable;
+import java.util.Map;
+
+import org.apache.pig.impl.logicalLayer.LogicalOperator;
+import org.apache.pig.impl.logicalLayer.OperatorKey;
+
+/**
+ * A logical plan has a root of the operator tree.
+ * A table of operators collects the set of operators making up a logical tree.
+ * It is possible to navigate a logical tree by looking up the input operator
+ * keys of the logical operator in the operator table.
+ *
+ */
+public interface ExecLogicalPlan extends Serializable {
+
+ public OperatorKey getRoot();
+
+ public Map<OperatorKey, LogicalOperator> getOpTable();
+
+ public void explain(OutputStream out);
+}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecPhysicalOperator.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecPhysicalOperator.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecPhysicalOperator.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecPhysicalOperator.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.executionengine;
+
+public interface ExecPhysicalOperator {
+
+ public String getScope();
+
+ public long getId();
+}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecPhysicalPlan.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecPhysicalPlan.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecPhysicalPlan.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecPhysicalPlan.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.executionengine;
+
+import java.util.Map;
+import java.util.Properties;
+import java.io.OutputStream;
+import java.io.Serializable;
+
+import org.apache.pig.impl.logicalLayer.OperatorKey;
+
+/**
+ *
+ *
+ */
+
+public interface ExecPhysicalPlan extends Serializable {
+
+ /*
+ * As the fron end is not really supposed to do much with the physical representation
+ * the handle to the physical plan can just be an GUId in the back-end state, which
+ * brings up the problem is persistency...
+ */
+
+ /**
+ * A job may have properties, like a priority, degree of parallelism...
+ * Some of such properties may be inherited from the ExecutionEngine
+ * configuration, other may have been set specifically for this job.
+ * For instance, a job scheduler may attribute low priority to
+ * jobs automatically started for maintenance purpose.
+ *
+ * @return set of properties
+ */
+ public Properties getConfiguration();
+
+ /**
+ * Some properties of the job may be changed, say the priority...
+ *
+ * @param configuration
+ * @throws some changes may not be allowed, for instance the some
+ * job-level properties cannot override Execution-Engine-level properties
+ * or maybe some properties can only be changes only in certain states of the
+ * job, say, once the job is started, parallelism level may not be changed...
+ */
+ public void updateConfiguration(Properties configuration)
+ throws ExecException;
+
+ /**
+ * To provide an "explanation" about how the physical plan has been constructed
+ *
+ */
+ public void explain(OutputStream out);
+
+ public Map<OperatorKey, ExecPhysicalOperator> getOpTable();
+
+ public OperatorKey getRoot();
+}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecScopedLogicalOperator.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecScopedLogicalOperator.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecScopedLogicalOperator.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecScopedLogicalOperator.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.executionengine;
+
+import java.io.Serializable;
+
+import org.apache.pig.impl.logicalLayer.OperatorKey;
+
+public interface ExecScopedLogicalOperator extends Serializable {
+
+ public String getScope();
+
+ public long getId();
+
+ public OperatorKey getOperatorKey();
+}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecutionEngine.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecutionEngine.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecutionEngine.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/executionengine/ExecutionEngine.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.executionengine;
+
+import java.util.Collection;
+import java.util.Properties;
+import java.util.Map;
+
+import org.apache.pig.backend.datastorage.DataStorage;
+
+/**
+ * This is the main interface that various execution engines
+ * need to support and it is also the main interface that Pig
+ * will need to use to submit jobs for execution, retrieve information
+ * about their progress and possibly terminate them.
+ *
+ */
+
+/**
+ *
+ * TODO: provide a manner to generate/collect logging information for DBG purposes
+ *
+ * TODO: add keys for properties/statistics
+ *
+ */
+public interface ExecutionEngine {
+ /**
+ * Place holder for possible initialization activities.
+ */
+ public void init() throws ExecException;
+
+ /**
+ * Clean-up and releasing of resources.
+ */
+ public void close() throws ExecException;
+
+ public DataStorage getDataStorage();
+
+ /**
+ * Provides configuration information about the execution engine itself.
+ *
+ * @return - information about the configuration used to connect to execution engine
+ */
+ public Properties getConfiguration() throws ExecException;
+
+ /**
+ * Provides a way to dynamically change configuration parameters
+ * at the Execution Engine level.
+ *
+ * @param newConfiguration - the new configuration settings
+ * @throws when configuration conflicts are detected
+ *
+ */
+ public void updateConfiguration(Properties newConfiguration)
+ throws ExecException;
+
+ /**
+ * Provides statistics on the Execution Engine: number of nodes,
+ * node failure rates, average load, average job wait time...
+ * @return ExecutionEngineProperties
+ */
+ public Map<String, Object> getStatistics() throws ExecException;
+
+ /**
+ * Compiles a logical plan into a physical plan, given a set of configuration
+ * properties that apply at the plan-level. For instance desired degree of
+ * parallelism for this plan, which could be different from the "default"
+ * one set at the execution engine level.
+ *
+ * @param logical plan
+ * @param properties
+ * @return physical plan
+ */
+ public ExecPhysicalPlan compile(ExecLogicalPlan plan,
+ Properties properties)
+ throws ExecException;
+
+ public ExecPhysicalPlan compile(ExecLogicalPlan[] plans,
+ Properties properties)
+ throws ExecException;
+
+ /**
+ * Execute the physical plan in blocking mode.
+ *
+ * @throws
+ */
+ public ExecJob execute(ExecPhysicalPlan plan) throws ExecException;
+
+ /**
+ * Execute the physical plan in non-blocking mode
+ *
+ * @throws ExecException
+ */
+ public ExecJob submit(ExecPhysicalPlan plan) throws ExecException;
+
+ /**
+ * Return currently running jobs (can be useful for admin purposes)
+ *
+ * @return
+ * @throws ExecException
+ */
+ public Collection<ExecJob> runningJobs(Properties properties) throws ExecException;
+
+ /**
+ * List scopes that are active in the back-end
+ *
+ * @return
+ * @throws ExecException
+ */
+ public Collection<String> activeScopes() throws ExecException;
+
+ /**
+ * A mechanism to communicate to the back-end that a set of logical plans go out of scope
+ *
+ * @param scope
+ */
+ public void reclaimScope(String scope) throws ExecException;
+}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/datastorage/HConfiguration.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/datastorage/HConfiguration.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/datastorage/HConfiguration.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/datastorage/HConfiguration.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.hadoop.datastorage;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Enumeration;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapred.JobConf;
+
+import java.util.Properties;
+
+
+public class HConfiguration extends Properties {
+
+ private static final long serialVersionUID = 1L;
+
+ public HConfiguration() {
+ }
+
+ //
+ // TODO: this implementation has a problem: it does not
+ // respect final attributes
+ //
+ public HConfiguration(Configuration other) {
+ if (other != null) {
+ Iterator<Map.Entry<String, String>> iter = other.iterator();
+
+ while (iter.hasNext()) {
+ Map.Entry<String, String> entry = iter.next();
+
+ put(entry.getKey(), entry.getValue());
+ }
+ }
+ }
+
+
+ public Configuration getConfiguration() {
+ Configuration config = new Configuration();
+
+ Enumeration<Object> iter = keys();
+
+ while (iter.hasMoreElements()) {
+ String key = (String) iter.nextElement();
+ String val = getProperty(key);
+
+ config.set(key, val);
+ }
+
+ return config;
+ }
+}
+
+
+
Added: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/datastorage/HDataStorage.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/datastorage/HDataStorage.java?rev=631443&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/datastorage/HDataStorage.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/datastorage/HDataStorage.java Tue Feb 26 16:51:49 2008
@@ -0,0 +1,216 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.hadoop.datastorage;
+
+import java.net.URI;
+import java.io.IOException;
+import java.util.Properties;
+import java.util.Enumeration;
+import java.util.Map;
+import java.util.HashMap;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.dfs.DistributedFileSystem;
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.pig.backend.datastorage.*;
+
+
+public class HDataStorage implements DataStorage {
+
+
+ private FileSystem fs;
+
+ public HDataStorage(URI uri, Configuration conf) throws IOException {
+ this(uri, new HConfiguration(conf));
+ }
+
+ public HDataStorage(URI uri, HConfiguration conf) throws IOException {
+ fs = FileSystem.get(uri, conf.getConfiguration());
+ }
+
+ public HDataStorage(Configuration conf) throws IOException {
+ this(new HConfiguration(conf));
+ }
+
+ public HDataStorage(HConfiguration conf) throws IOException {
+ fs = FileSystem.get(conf.getConfiguration());
+ }
+
+ public void init() { }
+
+ public void close() throws IOException {
+ fs.close();
+ }
+
+ public Properties getConfiguration() {
+ Properties props = new HConfiguration(fs.getConf());
+
+ short defaultReplication = fs.getDefaultReplication();
+ props.setProperty(DEFAULT_REPLICATION_FACTOR_KEY,
+ (new Short(defaultReplication)).toString());
+
+ return props;
+ }
+
+ public void updateConfiguration(Properties newConfiguration)
+ throws DataStorageException {
+ if (newConfiguration == null) {
+ return;
+ }
+
+ Enumeration<Object> newKeys = newConfiguration.keys();
+
+ while (newKeys.hasMoreElements()) {
+ String key = (String) newKeys.nextElement();
+ String value = null;
+
+ value = newConfiguration.getProperty(key);
+
+ fs.getConf().set(key,value);
+ }
+ }
+
+ public Map<String, Object> getStatistics() throws IOException {
+ Map<String, Object> stats = new HashMap<String, Object>();
+
+ long usedBytes = fs.getUsed();
+ stats.put(USED_BYTES_KEY , new Long(usedBytes).toString());
+
+ if (fs instanceof DistributedFileSystem) {
+ DistributedFileSystem dfs = (DistributedFileSystem) fs;
+
+ long rawCapacityBytes = dfs.getRawCapacity();
+ stats.put(RAW_CAPACITY_KEY, new Long(rawCapacityBytes).toString());
+
+ long rawUsedBytes = dfs.getRawUsed();
+ stats.put(RAW_USED_KEY, new Long(rawUsedBytes).toString());
+ }
+
+ return stats;
+ }
+
+ public ElementDescriptor asElement(String name)
+ throws DataStorageException {
+ if (this.isContainer(name)) {
+ return new HDirectory(this, name);
+ }
+ else {
+ return new HFile(this, name);
+ }
+ }
+
+ public ElementDescriptor asElement(ElementDescriptor element)
+ throws DataStorageException {
+ return asElement(element.toString());
+ }
+
+ public ElementDescriptor asElement(String parent,
+ String child)
+ throws DataStorageException {
+ return asElement((new Path(parent, child)).toString());
+ }
+
+ public ElementDescriptor asElement(ContainerDescriptor parent,
+ String child)
+ throws DataStorageException {
+ return asElement(parent.toString(), child);
+ }
+
+ public ElementDescriptor asElement(ContainerDescriptor parent,
+ ElementDescriptor child)
+ throws DataStorageException {
+ return asElement(parent.toString(), child.toString());
+ }
+
+ public ContainerDescriptor asContainer(String name)
+ throws DataStorageException {
+ return new HDirectory(this, name);
+ }
+
+ public ContainerDescriptor asContainer(ContainerDescriptor container)
+ throws DataStorageException {
+ return new HDirectory(this, container.toString());
+ }
+
+ public ContainerDescriptor asContainer(String parent,
+ String child)
+ throws DataStorageException {
+ return new HDirectory(this, parent, child);
+ }
+
+ public ContainerDescriptor asContainer(ContainerDescriptor parent,
+ String child)
+ throws DataStorageException {
+ return new HDirectory(this, parent.toString(), child);
+ }
+
+ public ContainerDescriptor asContainer(ContainerDescriptor parent,
+ ContainerDescriptor child)
+ throws DataStorageException {
+ return new HDirectory(this, parent.toString(), child.toString());
+ }
+
+ public void setActiveContainer(ContainerDescriptor container) {
+ fs.setWorkingDirectory(new Path(container.toString()));
+ }
+
+ public ContainerDescriptor getActiveContainer() {
+ return new HDirectory(this, fs.getWorkingDirectory());
+ }
+
+ public boolean isContainer(String name) throws DataStorageException {
+ boolean isContainer = false;
+ Path path = new Path(name);
+
+ try {
+ if ((this.fs.exists(path)) && (! this.fs.isFile(path))) {
+ isContainer = true;
+ }
+ }
+ catch (IOException e) {
+ throw new DataStorageException("Unable to check name " + name, e);
+ }
+
+ return isContainer;
+ }
+
+ public HPath[] asCollection(String pattern) throws DataStorageException {
+ try {
+ Path[] paths = this.fs.globPaths(new Path(pattern));
+
+ HPath[] hpaths = new HPath[ paths.length ];
+
+ for (int i = 0; i < paths.length; ++i) {
+ hpaths[ i ] = ((HPath)this.asElement(paths[ i ].toString()));
+ }
+
+
+ return hpaths;
+ }
+ catch (IOException e) {
+ throw new DataStorageException("Failed to obtain glob for " + pattern, e);
+ }
+ }
+
+ public FileSystem getHFS() {
+ return fs;
+ }
+}