hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From xu...@apache.org
Subject svn commit: r1639547 - in /hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql: exec/ exec/spark/ optimizer/spark/ plan/
Date Thu, 13 Nov 2014 23:04:33 GMT
Author: xuefu
Date: Thu Nov 13 23:04:33 2014
New Revision: 1639547

URL: http://svn.apache.org/r1639547
Log:
HIVE-8810: Make HashTableSinkOperator works for Spark Branch [Spark Branch] (Jimmy via Xuefu)

Added:
    hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/SparkHashTableSinkOperator.java
    hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/plan/SparkHashTableSinkDesc.java
Modified:
    hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/HashTableSinkOperator.java
    hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/OperatorFactory.java
    hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/HashTableLoader.java
    hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/optimizer/spark/SparkReduceSinkMapJoinProc.java
    hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/plan/MapredLocalWork.java

Modified: hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/HashTableSinkOperator.java
URL: http://svn.apache.org/viewvc/hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/HashTableSinkOperator.java?rev=1639547&r1=1639546&r2=1639547&view=diff
==============================================================================
--- hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/HashTableSinkOperator.java
(original)
+++ hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/HashTableSinkOperator.java
Thu Nov 13 23:04:33 2014
@@ -31,13 +31,12 @@ import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
 import org.apache.hadoop.hive.ql.exec.mapjoin.MapJoinMemoryExhaustionHandler;
 import org.apache.hadoop.hive.ql.exec.persistence.HashMapWrapper;
-import org.apache.hadoop.hive.ql.exec.persistence.MapJoinEagerRowContainer;
 import org.apache.hadoop.hive.ql.exec.persistence.MapJoinKeyObject;
 import org.apache.hadoop.hive.ql.exec.persistence.MapJoinObjectSerDeContext;
 import org.apache.hadoop.hive.ql.exec.persistence.MapJoinPersistableTableContainer;
+import org.apache.hadoop.hive.ql.exec.persistence.MapJoinEagerRowContainer;
 import org.apache.hadoop.hive.ql.exec.persistence.MapJoinRowContainer;
 import org.apache.hadoop.hive.ql.exec.persistence.MapJoinTableContainer;
 import org.apache.hadoop.hive.ql.exec.persistence.MapJoinTableContainerSerDe;
@@ -50,10 +49,11 @@ import org.apache.hadoop.hive.serde2.Ser
 import org.apache.hadoop.hive.serde2.SerDeException;
 import org.apache.hadoop.hive.serde2.SerDeUtils;
 import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils.ObjectInspectorCopyOption;
 import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
 import org.apache.hadoop.util.ReflectionUtils;
 
-@SuppressWarnings({"rawtypes", "deprecation"})
 public class HashTableSinkOperator extends TerminalOperator<HashTableSinkDesc> implements
     Serializable {
   private static final long serialVersionUID = 1L;
@@ -91,14 +91,10 @@ public class HashTableSinkOperator exten
   private transient List<ObjectInspector>[] joinFilterObjectInspectors;
 
   private transient Byte[] order; // order in which the results should
-  private Configuration hconf;
+  protected Configuration hconf;
 
-  // Used as a differentiator for different files
-  // in case multiple files are created for one operator.
-  private int fileIndex = 0;
-
-  private transient MapJoinPersistableTableContainer[] mapJoinTables;
-  private transient MapJoinTableContainerSerDe[] mapJoinTableSerdes;
+  protected transient MapJoinPersistableTableContainer[] mapJoinTables;
+  protected transient MapJoinTableContainerSerDe[] mapJoinTableSerdes;
 
   private static final Object[] EMPTY_OBJECT_ARRAY = new Object[0];
   private static final MapJoinEagerRowContainer EMPTY_ROW_CONTAINER = new MapJoinEagerRowContainer();
@@ -107,7 +103,7 @@ public class HashTableSinkOperator exten
   }
   
   private long rowNumber = 0;
-  private transient LogHelper console;
+  protected transient LogHelper console;
   private long hashTableScale;
   private MapJoinMemoryExhaustionHandler memoryExhaustionHandler;
   
@@ -201,24 +197,25 @@ public class HashTableSinkOperator exten
     return mapJoinTables;
   }
 
-//  private static List<ObjectInspector>[] getStandardObjectInspectors(
-//      List<ObjectInspector>[] aliasToObjectInspectors, int maxTag) {
-//    @SuppressWarnings("unchecked")
-//    List<ObjectInspector>[] result = new List[maxTag];
-//    for (byte alias = 0; alias < aliasToObjectInspectors.length; alias++) {
-//      List<ObjectInspector> oiList = aliasToObjectInspectors[alias];
-//      if (oiList == null) {
-//        continue;
-//      }
-//      ArrayList<ObjectInspector> fieldOIList = new ArrayList<ObjectInspector>(oiList.size());
-//      for (int i = 0; i < oiList.size(); i++) {
-//        fieldOIList.add(ObjectInspectorUtils.getStandardObjectInspector(oiList.get(i),
-//            ObjectInspectorCopyOption.WRITABLE));
-//      }
-//      result[alias] = fieldOIList;
-//    }
-//    return result;
-//  }
+  private static List<ObjectInspector>[] getStandardObjectInspectors(
+      List<ObjectInspector>[] aliasToObjectInspectors, int maxTag) {
+    @SuppressWarnings("unchecked")
+    List<ObjectInspector>[] result = new List[maxTag];
+    for (byte alias = 0; alias < aliasToObjectInspectors.length; alias++) {
+      List<ObjectInspector> oiList = aliasToObjectInspectors[alias];
+      if (oiList == null) {
+        continue;
+      }
+      ArrayList<ObjectInspector> fieldOIList = new ArrayList<ObjectInspector>(oiList.size());
+      for (int i = 0; i < oiList.size(); i++) {
+        fieldOIList.add(ObjectInspectorUtils.getStandardObjectInspector(oiList.get(i),
+            ObjectInspectorCopyOption.WRITABLE));
+      }
+      result[alias] = fieldOIList;
+    }
+    return result;
+
+  }
 
   /*
    * This operator only process small tables Read the key/value pairs Load them into hashtable
@@ -298,22 +295,11 @@ public class HashTableSinkOperator exten
       // get the tmp URI path; it will be a hdfs path if not local mode
       String dumpFilePrefix = conf.getDumpFilePrefix();
       Path path = Utilities.generatePath(tmpURI, dumpFilePrefix, tag, fileName);
-      FileSystem fs = path.getFileSystem(hconf);
-      short replication = fs.getDefaultReplication(path);
-
-      // For Spark, path is a folder. Let's create it now.
-      if (HiveConf.getVar(hconf, ConfVars.HIVE_EXECUTION_ENGINE).equals("spark")) {
-        fs.mkdirs(path);  // Create the folder and its parents if not there
-        path = new Path(path, getOperatorId() + "-" + (fileIndex++));
-        // TODO find out numOfPartitions for the big table
-        int numOfPartitions = 10;
-        replication = (short)Math.min(10, numOfPartitions);
-      }
       console.printInfo(Utilities.now() + "\tDump the side-table for tag: " + tag +
           " with group count: " + tableContainer.size() + " into file: " + path);
       // get the hashtable file and path
-      ObjectOutputStream out = new ObjectOutputStream(
-        new BufferedOutputStream(fs.create(path, replication), 4096));
+      FileSystem fs = path.getFileSystem(hconf);
+      ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(fs.create(path),
4096));
       try {
         mapJoinTableSerdes[tag].persist(out, tableContainer);
       } finally {

Modified: hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/OperatorFactory.java
URL: http://svn.apache.org/viewvc/hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/OperatorFactory.java?rev=1639547&r1=1639546&r2=1639547&view=diff
==============================================================================
--- hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/OperatorFactory.java (original)
+++ hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/OperatorFactory.java Thu
Nov 13 23:04:33 2014
@@ -18,6 +18,10 @@
 
 package org.apache.hadoop.hive.ql.exec;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
 import org.apache.hadoop.hive.ql.exec.vector.VectorAppMasterEventOperator;
 import org.apache.hadoop.hive.ql.exec.vector.VectorExtractOperator;
 import org.apache.hadoop.hive.ql.exec.vector.VectorFileSinkOperator;
@@ -58,14 +62,11 @@ import org.apache.hadoop.hive.ql.plan.Re
 import org.apache.hadoop.hive.ql.plan.SMBJoinDesc;
 import org.apache.hadoop.hive.ql.plan.ScriptDesc;
 import org.apache.hadoop.hive.ql.plan.SelectDesc;
+import org.apache.hadoop.hive.ql.plan.SparkHashTableSinkDesc;
 import org.apache.hadoop.hive.ql.plan.TableScanDesc;
 import org.apache.hadoop.hive.ql.plan.UDTFDesc;
 import org.apache.hadoop.hive.ql.plan.UnionDesc;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
 /**
  * OperatorFactory.
  *
@@ -102,6 +103,8 @@ public final class OperatorFactory {
         HashTableDummyOperator.class));
     opvec.add(new OpTuple<HashTableSinkDesc>(HashTableSinkDesc.class,
         HashTableSinkOperator.class));
+    opvec.add(new OpTuple<SparkHashTableSinkDesc>(SparkHashTableSinkDesc.class,
+        SparkHashTableSinkOperator.class));
     opvec.add(new OpTuple<DummyStoreDesc>(DummyStoreDesc.class,
         DummyStoreOperator.class));
     opvec.add(new OpTuple<DemuxDesc>(DemuxDesc.class,

Added: hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/SparkHashTableSinkOperator.java
URL: http://svn.apache.org/viewvc/hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/SparkHashTableSinkOperator.java?rev=1639547&view=auto
==============================================================================
--- hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/SparkHashTableSinkOperator.java
(added)
+++ hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/SparkHashTableSinkOperator.java
Thu Nov 13 23:04:33 2014
@@ -0,0 +1,159 @@
+/**
+ * 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.hadoop.hive.ql.exec;
+
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.io.OutputStream;
+import java.io.Serializable;
+
+import org.apache.commons.io.FileExistsException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.exec.persistence.MapJoinPersistableTableContainer;
+import org.apache.hadoop.hive.ql.exec.persistence.MapJoinTableContainerSerDe;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.SparkHashTableSinkDesc;
+import org.apache.hadoop.hive.ql.plan.api.OperatorType;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+
+public class SparkHashTableSinkOperator
+    extends TerminalOperator<SparkHashTableSinkDesc> implements Serializable {
+  private static final long serialVersionUID = 1L;
+  protected static final Log LOG = LogFactory.getLog(SparkHashTableSinkOperator.class.getName());
+
+  private HashTableSinkOperator htsOperator;
+
+  // The position of this table
+  private byte tag;
+
+  public SparkHashTableSinkOperator() {
+    htsOperator = new HashTableSinkOperator();
+  }
+
+  @Override
+  protected void initializeOp(Configuration hconf) throws HiveException {
+    ObjectInspector[] inputOIs = new ObjectInspector[conf.getTagLength()];
+    inputOIs[tag] = inputObjInspectors[0];
+    htsOperator.setConf(conf);
+    htsOperator.initialize(hconf, inputOIs);
+  }
+
+  @Override
+  public void processOp(Object row, int tag) throws HiveException {
+    // Ignore the tag passed in, which should be 0, not what we want
+    htsOperator.processOp(row, this.tag);
+  }
+
+  @Override
+  public void closeOp(boolean abort) throws HiveException {
+    try {
+      MapJoinPersistableTableContainer[] mapJoinTables = htsOperator.mapJoinTables;
+      if (mapJoinTables == null || mapJoinTables.length < tag
+          || mapJoinTables[tag] == null) {
+        LOG.debug("mapJoinTable is null");
+      } else {
+        flushToFile(mapJoinTables[tag], tag);
+      }
+      super.closeOp(abort);
+    } catch (HiveException e) {
+      throw e;
+    } catch (Exception e) {
+      throw new HiveException(e);
+    }
+  }
+
+  protected void flushToFile(MapJoinPersistableTableContainer tableContainer,
+      byte tag) throws IOException, HiveException {
+    // get tmp file URI
+    Path tmpURI = getExecContext().getLocalWork().getTmpHDFSPath();
+    LOG.info("Temp URI for side table: " + tmpURI);
+    // get current input file name
+    String bigBucketFileName = getExecContext().getCurrentBigBucketFile();
+    String fileName = getExecContext().getLocalWork().getBucketFileName(bigBucketFileName);
+    // get the tmp URI path; it will be a hdfs path if not local mode
+    String dumpFilePrefix = conf.getDumpFilePrefix();
+    Path path = Utilities.generatePath(tmpURI, dumpFilePrefix, tag, fileName);
+    FileSystem fs = path.getFileSystem(htsOperator.getConfiguration());
+    short replication = fs.getDefaultReplication(path);
+
+    fs.mkdirs(path);  // Create the folder and its parents if not there
+    while (true) {
+      path = new Path(path, getOperatorId()
+        + "-" + Math.abs(Utilities.randGen.nextInt()));
+      try {
+        // This will guarantee file name uniqueness.
+        // TODO: can we use the task id, which should be unique
+        if (fs.createNewFile(path)) {
+          break;
+        }
+      } catch (FileExistsException e) {
+        // No problem, use a new name
+      }
+      // TODO find out numOfPartitions for the big table
+      int numOfPartitions = replication;
+      replication = (short)Math.min(10, numOfPartitions);
+    }
+    htsOperator.console.printInfo(Utilities.now() + "\tDump the side-table for tag: " + tag
+
+      " with group count: " + tableContainer.size() + " into file: " + path);
+    // get the hashtable file and path
+    // get the hashtable file and path
+    OutputStream os = null;
+    ObjectOutputStream out = null;
+    try {
+      os = fs.create(path, replication);
+      out = new ObjectOutputStream(new BufferedOutputStream(os, 4096));
+      MapJoinTableContainerSerDe mapJoinTableSerde = htsOperator.mapJoinTableSerdes[tag];
+      mapJoinTableSerde.persist(out, tableContainer);
+    } finally {
+      if (out != null) {
+        out.close();
+      } else if (os != null) {
+        os.close();
+      }
+    }
+    tableContainer.clear();
+    FileStatus status = fs.getFileStatus(path);
+    htsOperator.console.printInfo(Utilities.now() + "\tUploaded 1 File to: " + path +
+      " (" + status.getLen() + " bytes)");
+  }
+
+  public void setTag(byte tag) {
+    this.tag = tag;
+  }
+
+  /**
+   * Implements the getName function for the Node Interface.
+   *
+   * @return the name of the operator
+   */
+  @Override
+  public String getName() {
+    return HashTableSinkOperator.getOperatorName();
+  }
+
+  @Override
+  public OperatorType getType() {
+    return OperatorType.HASHTABLESINK;
+  }
+}

Modified: hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/HashTableLoader.java
URL: http://svn.apache.org/viewvc/hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/HashTableLoader.java?rev=1639547&r1=1639546&r2=1639547&view=diff
==============================================================================
--- hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/HashTableLoader.java
(original)
+++ hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/HashTableLoader.java
Thu Nov 13 23:04:33 2014
@@ -76,7 +76,7 @@ public class HashTableLoader implements 
       }
       // All HashTables share the same base dir,
       // which is passed in as the tmp path
-      Path baseDir = localWork.getTmpPath();
+      Path baseDir = localWork.getTmpHDFSPath();
       if (baseDir == null) {
         return;
       }

Modified: hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/optimizer/spark/SparkReduceSinkMapJoinProc.java
URL: http://svn.apache.org/viewvc/hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/optimizer/spark/SparkReduceSinkMapJoinProc.java?rev=1639547&r1=1639546&r2=1639547&view=diff
==============================================================================
--- hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/optimizer/spark/SparkReduceSinkMapJoinProc.java
(original)
+++ hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/optimizer/spark/SparkReduceSinkMapJoinProc.java
Thu Nov 13 23:04:33 2014
@@ -26,15 +26,13 @@ import java.util.Stack;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.ql.exec.HashTableDummyOperator;
-import org.apache.hadoop.hive.ql.exec.HashTableSinkOperator;
 import org.apache.hadoop.hive.ql.exec.MapJoinOperator;
 import org.apache.hadoop.hive.ql.exec.Operator;
 import org.apache.hadoop.hive.ql.exec.OperatorFactory;
 import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
 import org.apache.hadoop.hive.ql.exec.RowSchema;
-import org.apache.hadoop.hive.ql.exec.TableScanOperator;
+import org.apache.hadoop.hive.ql.exec.SparkHashTableSinkOperator;
 import org.apache.hadoop.hive.ql.lib.Node;
 import org.apache.hadoop.hive.ql.lib.NodeProcessor;
 import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
@@ -43,12 +41,12 @@ import org.apache.hadoop.hive.ql.parse.s
 import org.apache.hadoop.hive.ql.plan.BaseWork;
 import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
 import org.apache.hadoop.hive.ql.plan.HashTableDummyDesc;
-import org.apache.hadoop.hive.ql.plan.HashTableSinkDesc;
 import org.apache.hadoop.hive.ql.plan.MapJoinDesc;
 import org.apache.hadoop.hive.ql.plan.OperatorDesc;
 import org.apache.hadoop.hive.ql.plan.PlanUtils;
 import org.apache.hadoop.hive.ql.plan.ReduceSinkDesc;
 import org.apache.hadoop.hive.ql.plan.SparkEdgeProperty;
+import org.apache.hadoop.hive.ql.plan.SparkHashTableSinkDesc;
 import org.apache.hadoop.hive.ql.plan.SparkWork;
 import org.apache.hadoop.hive.ql.plan.TableDesc;
 
@@ -234,9 +232,9 @@ public class SparkReduceSinkMapJoinProc 
     //replace ReduceSinkOp with HashTableSinkOp for the RSops which are parents of MJop
     MapJoinDesc mjDesc = mapJoinOp.getConf();
 
-    HashTableSinkDesc hashTableSinkDesc = new HashTableSinkDesc(mjDesc);
-    HashTableSinkOperator hashTableSinkOp = (HashTableSinkOperator) OperatorFactory
-        .get(hashTableSinkDesc);
+    SparkHashTableSinkDesc hashTableSinkDesc = new SparkHashTableSinkDesc(mjDesc);
+    SparkHashTableSinkOperator hashTableSinkOp =
+      (SparkHashTableSinkOperator) OperatorFactory.get(hashTableSinkDesc);
 
     //get all parents of reduce sink
     List<Operator<? extends OperatorDesc>> RSparentOps = parentRS.getParentOperators();
@@ -244,6 +242,7 @@ public class SparkReduceSinkMapJoinProc 
       parent.replaceChild(parentRS, hashTableSinkOp);
     }
     hashTableSinkOp.setParentOperators(RSparentOps);
+    hashTableSinkOp.setTag((byte)pos);
     return true;
   }
 }

Modified: hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/plan/MapredLocalWork.java
URL: http://svn.apache.org/viewvc/hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/plan/MapredLocalWork.java?rev=1639547&r1=1639546&r2=1639547&view=diff
==============================================================================
--- hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/plan/MapredLocalWork.java (original)
+++ hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/plan/MapredLocalWork.java Thu
Nov 13 23:04:33 2014
@@ -45,6 +45,8 @@ public class MapredLocalWork implements 
   private BucketMapJoinContext bucketMapjoinContext;
   private Path tmpPath;
   private String stageID;
+  // Temp HDFS path for Spark HashTable sink and loader
+  private Path tmpHDFSPath;
 
   private List<Operator<? extends OperatorDesc>> dummyParentOp;
   private Map<MapJoinOperator, List<Operator<? extends OperatorDesc>>>
directFetchOp;
@@ -168,6 +170,14 @@ public class MapredLocalWork implements 
     return tmpPath;
   }
 
+  public void setTmpHDFSPath(Path tmpPath) {
+    this.tmpHDFSPath = tmpPath;
+  }
+
+  public Path getTmpHDFSPath() {
+    return tmpHDFSPath;
+  }
+
   public String getBucketFileName(String bigFileName) {
     if (!inputFileChangeSensitive || bigFileName == null || bigFileName.isEmpty()) {
       return "-";

Added: hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/plan/SparkHashTableSinkDesc.java
URL: http://svn.apache.org/viewvc/hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/plan/SparkHashTableSinkDesc.java?rev=1639547&view=auto
==============================================================================
--- hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/plan/SparkHashTableSinkDesc.java
(added)
+++ hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/plan/SparkHashTableSinkDesc.java
Thu Nov 13 23:04:33 2014
@@ -0,0 +1,35 @@
+/**
+ * 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.hadoop.hive.ql.plan;
+
+/**
+ * Map Join operator Descriptor implementation.
+ *
+ */
+@Explain(displayName = "Spark HashTable Sink Operator")
+public class SparkHashTableSinkDesc extends HashTableSinkDesc {
+  private static final long serialVersionUID = 1L;
+
+  public SparkHashTableSinkDesc() {
+  }
+
+  public SparkHashTableSinkDesc(MapJoinDesc clone) {
+    super(clone);
+  }
+}



Mime
View raw message