Hive ExprNodeEvaluator 及其子类分析

ExprNodeevaluator 有两个重要的方法，一个是用 rowInspector 对一行进行解析，得到 result。

public abstract class ExprNodeevaluator {

  
  public abstract void evaluate(Object row, ObjectInspector rowInspector, InspectableObject result) throws HiveException;

  
  public abstract ObjectInspector evaluateInspector(ObjectInspector rowInspector) throws HiveException;

}

ExprNodeColumnevaluator 因为历史原因，支持类似a.b.c 的多层字段。

public class ExprNodeColumnevaluator extends ExprNodeevaluator {

  protected exprNodeColumnDesc expr;
  transient StructObjectInspector cachedRowInspector;
  transient String[] fieldNames;
  transient StructField[] fields;
  transient ObjectInspector[] fieldsObjectInspector;
  
  public ExprNodeColumnevaluator(exprNodeColumnDesc expr) {
    this.expr = expr;
  }

  public void evaluate(Object row, ObjectInspector rowInspector,
      InspectableObject result) throws HiveException {
    
    assert(result != null);
    // If this is the first row, or the dynamic structure of this row 
    // is different from the previous row 
    if (fields == null || cachedRowInspector != rowInspector) {
      evaluateInspector(rowInspector);
    }
    result.o = cachedRowInspector.getStructFieldData(row, fields[0]);
    for(int i=1; i 
ExprNodeNullevaluator 
调用这两个方法，总是抛出异常。
 
public class ExprNodeNullevaluator extends ExprNodeevaluator {

  protected exprNodeNullDesc expr;

  public ExprNodeNullevaluator(exprNodeNullDesc expr) {
    this.expr = expr;
  }

  public void evaluate(Object row, ObjectInspector rowInspector,
      InspectableObject result) throws HiveException {
    throw new HiveException("Hive 2 Internal exception: should not reach here.");
  }

  public ObjectInspector evaluateInspector(ObjectInspector rowInspector)
      throws HiveException {
    throw new HiveException("Hive 2 Internal exception: should not reach here.");
  }

}
 
ExprNodeFuncevaluator 
public class ExprNodeFuncevaluator extends ExprNodeevaluator {

  private static final Log LOG = LogFactory.getLog(ExprNodeFuncevaluator.class.getName());
  
  protected exprNodeFuncDesc expr;
  transient ExprNodeevaluator[] paramevaluators;
  transient InspectableObject[] paramInspectableObjects;
  transient Object[] paramValues;
  transient UDF udf;
  transient Method udfMethod;
  transient ObjectInspector outputObjectInspector;
  
  public ExprNodeFuncevaluator(exprNodeFuncDesc expr) {
    this.expr = expr;
    assert(expr != null);
    Class c = expr.getUDFClass();
    udfMethod = expr.getUDFMethod();
    LOG.debug(c.toString());
    LOG.debug(udfMethod.toString());
    udf = (UDF)ReflectionUtils.newInstance(expr.getUDFClass(), null);
    int paramNumber = expr.getChildren().size();
    paramevaluators = new ExprNodeevaluator[paramNumber];
    paramInspectableObjects  = new InspectableObject[paramNumber];
    for(int i=0; i 
ExprNodeIndexevaluator 
ExprNodeIndexevaluator 可以查询 list 或者 map 的值。
 
public class ExprNodeIndexevaluator extends ExprNodeevaluator {

  protected exprNodeIndexDesc expr;
  transient ExprNodeevaluator mainevaluator;
  transient InspectableObject mainInspectableObject = new InspectableObject();
  transient ExprNodeevaluator indexevaluator;
  transient InspectableObject indexInspectableObject = new InspectableObject();
  
  public ExprNodeIndexevaluator(exprNodeIndexDesc expr) {
    this.expr = expr;
    mainevaluator = ExprNodeevaluatorFactory.get(expr.getDesc());
    indexevaluator = ExprNodeevaluatorFactory.get(expr.getIndex());
  }

  public void evaluate(Object row, ObjectInspector rowInspector,
      InspectableObject result) throws HiveException {
    
    assert(result != null);
    mainevaluator.evaluate(row, rowInspector, mainInspectableObject);
    indexevaluator.evaluate(row, rowInspector, indexInspectableObject);

    if (mainInspectableObject.oi.getCategory() == Category.LIST) {
      int index = ((Number)indexInspectableObject.o).intValue();
    
      ListObjectInspector loi = (ListObjectInspector)mainInspectableObject.oi;
      result.oi = loi.getListElementObjectInspector();
      result.o = loi.getListElement(mainInspectableObject.o, index);
    }
    else if (mainInspectableObject.oi.getCategory() == Category.MAP) {
      MapObjectInspector moi = (MapObjectInspector)mainInspectableObject.oi;
      result.oi = moi.getMapValueObjectInspector();
      result.o = moi.getMapValueElement(mainInspectableObject.o, indexInspectableObject.o);
    }
    else {
      // Should never happen because we checked this in SemanticAnalyzer.getXpathOrFuncExprNodeDesc
      throw new RuntimeException("Hive 2 Internal error: cannot evaluate index expression on "
          + mainInspectableObject.oi.getTypeName());
    }
  }

  public ObjectInspector evaluateInspector(ObjectInspector rowInspector)
      throws HiveException {
    ObjectInspector mainInspector = mainevaluator.evaluateInspector(rowInspector);
    if (mainInspector.getCategory() == Category.LIST) {
      return ((ListObjectInspector)mainInspector).getListElementObjectInspector();
    } else if (mainInspector.getCategory() == Category.MAP) {
      return ((MapObjectInspector)mainInspector).getMapValueObjectInspector();
    } else {
      // Should never happen because we checked this in SemanticAnalyzer.getXpathOrFuncExprNodeDesc
      throw new RuntimeException("Hive 2 Internal error: cannot evaluate index expression on "
          + mainInspector.getTypeName());
    }
  }

}
 
ExprNodeevaluatorFactory 
ExprNodeevaluatorFactory 提供了根据 exprNodeDesc 创建 ExprNodeevaluator 对象的静态方法。
 
public class ExprNodeevaluatorFactory {
  
  public ExprNodeevaluatorFactory() {}

  public static ExprNodeevaluator get(exprNodeDesc desc) {
    // Constant node
    if (desc instanceof exprNodeConstantDesc) {
      return new ExprNodeConstantevaluator((exprNodeConstantDesc)desc);
    }
    // Column-reference node, e.g. a column in the input row
    if (desc instanceof exprNodeColumnDesc) {
      return new ExprNodeColumnevaluator((exprNodeColumnDesc)desc);
    }
    // Function node, e.g. an operator or a UDF node
    if (desc instanceof exprNodeFuncDesc) {
      return new ExprNodeFuncevaluator((exprNodeFuncDesc)desc);
    }
    // Field node, e.g. get a.myfield1 from a
    if (desc instanceof exprNodeFieldDesc) {
      return new ExprNodeFieldevaluator((exprNodeFieldDesc)desc);
    }
    // Index node, e.g. get a[index] from a
    if (desc instanceof exprNodeIndexDesc) {
      return new ExprNodeIndexevaluator((exprNodeIndexDesc)desc);
    }
    // Null node, a constant node with value NULL and no type information 
    if (desc instanceof exprNodeNullDesc) {
      return new ExprNodeNullevaluator((exprNodeNullDesc)desc);
    }

    throw new RuntimeException("Cannot find ExprNodeevaluator for the exprNodeDesc = " + desc);
  }
}