[大数据]【软件工程实践】Hive研究-Blog7

【软件工程实践】Hive研究-Blog7

2021SC@SDUSC

研究内容介绍

本人负责的是负责的是将查询块QB转换成逻辑查询计划（OP Tree）
如下的代码出自apaceh-hive-3.1.2-src/ql/src/java/org/apache/hadoop/hive/ql/plan中，也就是我的分析目标代码。本周的研究计划是继续解析PlanMapper.java文件源码。由于在Blog6中已经完成了对内部类CompositeMap的全部解析，因此在本次Blog中我们就完成对PlanMapper.java文件的剩余内容解析。

PlanMapper.java文件代码解析

我们附上整个java文件代码

/*
 * 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.mapper;

import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Set;

import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.optimizer.signature.OpTreeSignature;
import org.apache.hadoop.hive.ql.optimizer.signature.OpTreeSignatureFactory;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;

/**
 * Enables to connect related objects to eachother.
 *
 * Most importantly it aids to connect Operators to OperatorStats and probably RelNodes.
 */
public class PlanMapper {

  Set<EquivGroup> groups = new HashSet<>();
  private Map<Object, EquivGroup> objectMap = new CompositeMap<>(OpTreeSignature.class);

  /**
   * Specialized class which can compare by identity or value; based on the key type.
   */
  private static class CompositeMap<K, V> implements Map<K, V> {

    Map<K, V> comparedMap = new HashMap<>();
    Map<K, V> identityMap = new IdentityHashMap<>();
    final Set<Class<?>> typeCompared;

    CompositeMap(Class<?>... comparedTypes) {
      for (Class<?> class1 : comparedTypes) {
        if (!Modifier.isFinal(class1.getModifiers())) {
          throw new RuntimeException(class1 + " is not final...for this to reliably work; it should be");
        }
      }
      typeCompared = Sets.newHashSet(comparedTypes);
    }

    @Override
    public int size() {
      return comparedMap.size() + identityMap.size();
    }

    @Override
    public boolean isEmpty() {
      return comparedMap.isEmpty() && identityMap.isEmpty();
    }

    @Override
    public boolean containsKey(Object key) {
      return comparedMap.containsKey(key) || identityMap.containsKey(key);
    }

    @Override
    public boolean containsValue(Object value) {
      return comparedMap.containsValue(value) || identityMap.containsValue(value);
    }

    @Override
    public V get(Object key) {
      V v0 = comparedMap.get(key);
      if (v0 != null) {
        return v0;
      }
      return identityMap.get(key);
    }

    @Override
    public V put(K key, V value) {
      if (shouldCompare(key.getClass())) {
        return comparedMap.put(key, value);
      } else {
        return identityMap.put(key, value);
      }
    }

    @Override
    public V remove(Object key) {
      if (shouldCompare(key.getClass())) {
        return comparedMap.remove(key);
      } else {
        return identityMap.remove(key);
      }
    }

    private boolean shouldCompare(Class<?> key) {
      return typeCompared.contains(key);
    }

    @Override
    public void putAll(Map<? extends K, ? extends V> m) {
      for (Entry<? extends K, ? extends V> e : m.entrySet()) {
        put(e.getKey(), e.getValue());
      }
    }

    @Override
    public void clear() {
      comparedMap.clear();
      identityMap.clear();
    }

    @Override
    public Set<K> keySet() {
      return Sets.union(comparedMap.keySet(), identityMap.keySet());
    }

    @Override
    public Collection<V> values() {
      throw new UnsupportedOperationException("This method is not supported");
    }

    @Override
    public Set<Entry<K, V>> entrySet() {
      return Sets.union(comparedMap.entrySet(), identityMap.entrySet());
    }

  }

  /**
   * A set of objects which are representing the same thing.
   *
   * A Group may contain different kind of things which are connected by their purpose;
   * For example currently a group may contain the following objects:
   * <ul>
   *   <li> Operator(s) - which are doing the actual work;
   *   there might be more than one, since an optimization may replace an operator with a new one
   *   <li> Signature - to enable inter-plan look up of the same data
   *   <li> OperatorStats - collected runtime information
   * </ul>
   */
  public class EquivGroup {
    Set<Object> members = new HashSet<>();

    public void add(Object o) {
      if (members.contains(o)) {
        return;
      }
      members.add(o);
      objectMap.put(o, this);
    }

    @SuppressWarnings("unchecked")
    public <T> List<T> getAll(Class<T> clazz) {
      List<T> ret = new ArrayList<>();
      for (Object m : members) {
        if (clazz.isInstance(m)) {
          ret.add((T) m);
        }
      }
      return ret;
    }
  }

  /**
   * States that the two objects are representing the same.
   *
   * For example if during an optimization Operator_A is replaced by a specialized Operator_A1;
   * then those two can be linked.
   */
  public void link(Object o1, Object o2) {

    Set<Object> keySet = Collections.newSetFromMap(new IdentityHashMap<Object, Boolean>());
    keySet.add(o1);
    keySet.add(o2);
    keySet.add(getKeyFor(o1));
    keySet.add(getKeyFor(o2));

    Set<EquivGroup> mGroups = Collections.newSetFromMap(new IdentityHashMap<EquivGroup, Boolean>());

    for (Object object : keySet) {
      EquivGroup group = objectMap.get(object);
      if (group != null) {
        mGroups.add(group);
      }
    }
    if (mGroups.size() > 1) {
      throw new RuntimeException("equivalence mapping violation");
    }
    EquivGroup targetGroup = mGroups.isEmpty() ? new EquivGroup() : mGroups.iterator().next();
    groups.add(targetGroup);
    targetGroup.add(o1);
    targetGroup.add(o2);

  }

  private OpTreeSignatureFactory signatureCache = OpTreeSignatureFactory.newCache();

  private Object getKeyFor(Object o) {
    if (o instanceof Operator) {
      Operator<?> operator = (Operator<?>) o;
      return signatureCache.getSignature(operator);
    }
    return o;
  }

  public <T> List<T> getAll(Class<T> clazz) {
    List<T> ret = new ArrayList<>();
    for (EquivGroup g : groups) {
      ret.addAll(g.getAll(clazz));
    }
    return ret;
  }

  public void runMapper(GroupTransformer mapper) {
    for (EquivGroup equivGroup : groups) {
      mapper.map(equivGroup);
    }
  }

  public <T> List<T> lookupAll(Class<T> clazz, Object key) {
    EquivGroup group = objectMap.get(key);
    if (group == null) {
      throw new NoSuchElementException(Objects.toString(key));
    }
    return group.getAll(clazz);
  }

  public <T> T lookup(Class<T> clazz, Object key) {
    List<T> all = lookupAll(clazz, key);
    if (all.size() != 1) {
      // FIXME: use a different exception type?
      throw new IllegalArgumentException("Expected match count is 1; but got:" + all);
    }
    return all.get(0);
  }

  @VisibleForTesting
  public Iterator<EquivGroup> iterateGroups() {
    return groups.iterator();

  }

  public OpTreeSignature getSignatureOf(Operator<?> op) {
    OpTreeSignature sig = signatureCache.getSignature(op);
    return sig;
  }

}

我们开始解析剩余代码。

方法add

    public void add(Object o) {
      if (members.contains(o)) {
        return;
      }
      members.add(o);
      objectMap.put(o, this);
    }

members.contains()这是什么方法？那么我们首先来看members是一个什么变量。我们看类EquivGroup内部的全局变量定义，找到了关于members变量的定义：
Set<Object> members = new HashSet<>();
那么我们就需要寻找Set是否有一个叫contains的方法了。当然，我们不需要急着去网上搜寻，我们先浏览下文看看这是否是一个内部定义的方法。浏览后发现并没有包含该方法，我们从网上搜寻后得到如下信息：Java 集合类中的 Set.contains() 方法判断 Set 集合是否包含指定的对象。该方法返回值为 boolean 类型，如果 Set 集合包含指定的对象，则返回 true，否则返回 false。其调用语法为：contains(Object o)，其中o是要进行查询的对象，可以为任意的类型。我们不妨看一个例子来理解该方法：

public static void main(String[] args){
    Set set = new HashSet();
    set.add(new Date());
    set.add("apple");
    set.add(new Socket());
    boolean contains = set.contains("apple");
    if(contains){
    System.out.println("Set集合包含字符串apple");
    }else{
    System.out.println("Set集合不包含字符串apple");
  }
}

输出

Set集合包含字符串apple

而源码中的members.contains(o)是作为判断语句的。我们看看，如果返回的是true的话，就进入到分支语句中来了，也就是结束该方法，返回。因为是true，说明传入的变量o已经在集合members里面了，我们无需再添加，因此什么也不做直接返回是最好的做法。如果返回的是false，我们继续往下看。这个members.add(o)显然是Set类自带的方法，我们猜测是将o加入到集合中来。我们在网上搜寻资料来验证我们的猜想：add() 方法用于给集合添加元素，如果添加的元素在集合中已存在，则不执行任何操作。我们的猜测是正确的。

我们接着往后看，这个objectMap是什么东西？我们猜测这是一个全局变量，于是我们从文件开头往下寻找，找到了关于这个变量的定义：
private Map<Object, EquivGroup> objectMap = new CompositeMap<>(OpTreeSignature.class);
那么objectMap.put（o,this）方法是什么？这就是Map类自带的put方法，需要传入一个key和一个value。而key显而易见是o，那么this是什么？我们查阅资料得知：把this作为参数传递。当你要把自己作为参数传递给别的对象时，也可以用this。例如下面的例子：

public class A{

public A(){
    new B(this).print();
}

public void print(){
      System.out.println("From A!");
 }

public static void main(String[] args) {
      new A();
   }
}
 
public class B{
   A a;
   public B(A a){
      this.a = a;
   }
 
   public void print(){
      a.print();
      System.out.println("From B!");
   }
}

输出

From A!
	From B!

现在清楚明了了，this就指的是EquivGroup对象，里面包含的members变量已经在上语句中添加了o变量进入到自己的集合中来了，因此可以直接把这个EquivGroup对象装入objectMap中来。

方法getAll

    @SuppressWarnings("unchecked")
    public <T> List<T> getAll(Class<T> clazz) {
      List<T> ret = new ArrayList<>();
      for (Object m : members) {
        if (clazz.isInstance(m)) {
          ret.add((T) m);
        }
      }
      return ret;
    }

我们先来看开头的@SuppressWarnings注解是什么意思。经过查阅资料我们得知：java.lang.SuppressWarnings是J2SE 5.0中标准的Annotation之一。可以标注在类、字段、方法、参数、构造方法，以及局部变量上。作用：告诉编译器忽略指定的警告，不用在编译完成后出现警告信息。

在本文中，·@SuppressWarnings(“unchecked”)告诉编译器忽略 unchecked 警告信息，如使用List，ArrayList等未进行参数化产生的警告信息。

我们先来看一下Class.isInstance是什么一个方法。首先，我们先要知道什么是instanceof关键字：instanceof 关键字用于判断某个实例是否是某个类的实例化对象，形如：
String.class instanceof Class 和 "test" instanceof String
那么，知道了instance是啥，isInstance()方法也就知道了：isInstance是Class类中的方法，也是用于判断某个实例是否是某个类的实例化对象，但是指向则相反。但这样就带来一个疑问，为什么我们需要这个方法呢，官方文档如此解释：

Determines if the specified Object is assignment-compatible with the object represented by this Class.
This method is the dynamic equivalent of the Java language instanceof operator.
The method returns true if the specified Object argument is non-null
and can be cast to the reference type represented by this Class object without raising a ClassCastException.
It returns false otherwise.

总而言之，我们只需要记住指向相反这个结论即可。那么这个clazz.isInstance(m)的意思就是判断集合中的m元素是否为clazz的具体实例化，如果是的话就在ret中加入这个元素。而外面的for循环就是遍历整个集合。最后处理完后返回列表ret。

方法link

  public void link(Object o1, Object o2) {

    Set<Object> keySet = Collections.newSetFromMap(new IdentityHashMap<Object, Boolean>());
    keySet.add(o1);
    keySet.add(o2);
    keySet.add(getKeyFor(o1));
    keySet.add(getKeyFor(o2));

    Set<EquivGroup> mGroups = Collections.newSetFromMap(new IdentityHashMap<EquivGroup, Boolean>());

    for (Object object : keySet) {
      EquivGroup group = objectMap.get(object);
      if (group != null) {
        mGroups.add(group);
      }
    }
    if (mGroups.size() > 1) {
      throw new RuntimeException("equivalence mapping violation");
    }
    EquivGroup targetGroup = mGroups.isEmpty() ? new EquivGroup() : mGroups.iterator().next();
    groups.add(targetGroup);
    targetGroup.add(o1);
    targetGroup.add(o2);

  }

我们先来看看方法Collections.newSetFromMap(new IdentityHashMap<Object, Boolean>())是一个什么样的方法。它用于生成对Map进行包装的Set。这个Set和被包装的Map拥有相同的key顺序（遍历Set调用的还是Map的keySet），相同的并发特性（也就是说如果对ConcurrentHashMap进行包装，得到的Set也将线程安全）。本质上来说，这个工厂方法（newSetFromMap）就是提供了一个和Map实现相对应的Set实现。接下来的getkeyfor是一个什么方法？我们经过浏览发现它在下文，故我们先来分析一下这个方法。

方法getKeyFor

  private Object getKeyFor(Object o) {
    if (o instanceof Operator) {
      Operator<?> operator = (Operator<?>) o;
      return signatureCache.getSignature(operator);
    }
    return o;
  }

我们在前面就已经解释了关键词instanceof是什么意思了。那么if语句的判断条件为：o是否为Operator类的具体实例化？如果是则判断为true，不是则判断为false。如果判断为true，则将实例化一个oprator.然后返回语句signatureCache.getSignature(operator);的返回结果。那么这个方法是一个什么方法？我们在网上以及在全文中都未找到。但我们找到了变量signatureCache的定义语句：
private OpTreeSignatureFactory signatureCache = OpTreeSignatureFactory.newCache();
我们从名字上推测：Cache，说明这个类与缓存有关，而缓存又像是数组一般，所以我们推测这是一个缓存区域。在官方的API文档中也证明了这一点：A simple cache backend to prevent repeated signature computations.既然是Signature，也就是签名和署名，说明是唯一标识。因此，我们可以大胆的推测该类是这个样子的：用于管理特定的缓存区域，每一个在缓存块内的对象都有自己的签名，可以通过类中的getSignature方法来获取这个对象或者说是一些属性，可以用于调用这个对象。那么getKeyFor也大概明白了作用了。

我们回到方法link中来，来看for循环的内容。for循环遍历KeySet内的所有元素。然后我们来看一下语句：EquivGroup group = objectMap.get(object);这里的objectMap是在开头定义的一个Map类型的遍历。Map.get方法就是取得所对应的值。然后判断这个值是否为空，如果不为空则加入到mGroups这个集合中来。后面的if语句判断这个集合内的元素是否大于1，如果是的话就会抛出异常并打印对应的语句"equivalence mapping violation"。这是很容易理解的，因为keySet集合是新建立的，里面的元素不应该存在映射关系，如果集合规模大于1了说明先前已经是存在了的，在下文的操作中将会发生预料之外的错误，因此必须抛出异常来结束，以防引发更大的错误。

语句

EquivGroup targetGroup = mGroups.isEmpty() ? new EquivGroup() : mGroups.iterator().next();

是一个典型的条件运算符的运用。我们先来看一下"?:"运算符的使用方式：
result = <expression> ? <statement1> : <statement3>;

其中，expression 是一个布尔表达式。当 expression 为真时，执行 statement1， 否则就执行 statement3。此三元运算符要求返回一个结果，因此要实现简单的二分支程序，即可使用该条件运算符。

回到原文，我们来看看判断条件为mGroups.isEmpty()，就是判断这个集合是否为空。如果为空，则将new EquivGroup()的返回值赋予targetGroup。如果不为空，则将mGroups.iterator().next()的结果赋予targetGroup。那么，这个语句是什么意思呢？我们查阅资料得知：使用Collection类的Iterator，可以方便的遍历Vector, ArrayList, LinkedList等集合元素，避免通过get()方法遍历时，针对每一种对象单独进行编码。我们可以使用一个例子来说明：

Collection coll =  new  Vector();  //LinkedList(); //ArrayList();   
coll.add( "Tody" );   
coll.add( "is" );   
coll.add( "Sunday." );   
   
// Output all elements by iterator   
Iterator it = coll.iterator();   
while (it.hasNext()) {   
     System.out.print(it.next() +  " " );   
}  

输出

Tody is Sunday. 

我们再来看看使用使用集合来迭代的

Collection coll =  new  HashSet();   
coll.add( "Tody" );   
coll.add( "is" );   
coll.add( "Sunday." );   
   
// Output all elements by iterator   
Iterator it = coll.iterator();   
while (it.hasNext()) {   
     System.out.print(it.next() +  " " );   
} 

输出

is Sunday. Tody 

由上面两个例子看出，在List和Set对象中，Iterator的next()方法返回的值是不一样的。在List对象中，第一次next()返回的是第一个元素，在Set中，第一次返回的是下一个元素，但Set中，在set的结尾执行hasNext()时，返回true，表示第一个元素，执行next()会把第一个元素返回。回到源码中来，我们拿到的其实是mGroups的第二个元素。

接着往下看，语句groups.add(targetGroup);中的groups是什么变量？我们在文件的开头找到了改变量的定义语句：Set<EquivGroup> groups = new HashSet<>();那么add方法就是往里面添加元素了。后面的add方法也是如此。

总的来说，link方法就是将传入的两个参数进行与自身的集合比对，增加元素以及赋予新的值。

方法getAll

  public <T> List<T> getAll(Class<T> clazz) {
    List<T> ret = new ArrayList<>();
    for (EquivGroup g : groups) {
      ret.addAll(g.getAll(clazz));
    }
    return ret;
  }

我们先来看一下方法addAll是一个什么样的方法：Java 集合类的 List.addAll() 方法用于将指定 collection 中的所有元素添加到列表。我们可以通过一个简单的例子来了解一下：

public static void main(String[] args){
    List<String>list = new ArrayList<String>();
    list.add("保护环境");  //向列表中添加数据
    list.add("爱护地球");  //向列表中添加数据
    list.add("从我做起");  //向列表中添加数据
    list.add(1,"从我做起");  //在第1+1个元素的位置添加数据
    List<String>list_ad = new ArrayList<String>();
    list_ad.add("公益广告");  //将list中的全部元素添加到list_ad中
    System.out.println("是否添加成功:"+list_ad.addAll(list));  //通过循环输出列表中的内容
    for(int i=0;i<list_ad.size();i++){
    System.out.println(i+":"+list_ad.get(i));
  }
}

输出

是否添加成功：true
0：公益广告
1：保护环境
2：从我做起
3：爱护地球
4：从我做起

我们先来看看groups是一个什么东西。我们在上面已经说过了，它是一个集合，里面的元素类型为EquivGroup。那么这个EquivGroup是什么东西呢？我们在Apache官网API接口上找到了关于这个类的详细信息：
A set of objects which are representing the same thing. A Group may contain different kind of things which are connected by their purpose; For example currently a group may contain the following objects:

Operator(s) - which are doing the actual work; there might be more than one, since an optimization may replace an operator with a new one

Signature - to enable inter-plan look up of the same data

OperatorStats - collected runtime information

简单来说，就是一类相同的东西都可以放在这个类中。那么该方法的作用我猜测是传入一个组group，然后对其进行批量操作。我们接着解析。这里调用了一个getAll的方法，我们来看一下这个方法。同样的，我们从官网入手：

官网并没有给出详细的具体方法内容，我们只能进行合理的猜测了：给出了对应的类的实例化对象，然后这个方法会从类实例化对象中获取其自带的变量，然后全部加入到链表ret中来。最后再将链表ret返回。

方法runMapper

  public void runMapper(GroupTransformer mapper) {
    for (EquivGroup equivGroup : groups) {
      mapper.map(equivGroup);
    }
  }

我们来看一下map方法是一个什么方法。我们查阅apache官方文档，找到了如下描述：

在官方的文档里面并没有对该方法的具体实现细节进行描述，故我们在遇到这个方法前先对这个方法的作用进行猜测。首先是看传入的参数类的名字，transformer转换器，而方法又是map映射，故该方法的作用应该是将传入的参数作为一个钥匙，然后找到对应的门取出内容并返回，相当于key-value的键值对模式。

方法lookupAll

  public <T> List<T> lookupAll(Class<T> clazz, Object key) {
    EquivGroup group = objectMap.get(key);
    if (group == null) {
      throw new NoSuchElementException(Objects.toString(key));
    }
    return group.getAll(clazz);
  }

这里的objectMap是在开头定义的一个Map全局变量。Map.get()得到一个值然后赋予变量group。然后我们在if语句判断这个group是否为空，如果为空，就抛出异常，并打印出指定好的语句。然后返回clazz对象内的私有变量，作为一个列表返回。

方法lookup

  public <T> T lookup(Class<T> clazz, Object key) {
    List<T> all = lookupAll(clazz, key);
    if (all.size() != 1) {
      // FIXME: use a different exception type?
      throw new IllegalArgumentException("Expected match count is 1; but got:" + all);
    }
    return all.get(0);
  }

这个方法在开头就调用了在上面解析的方法，得到了一个链表，然后将这个链表赋予为all。进入到if语句的判断中来，如果链表all的大小不等于1，则会抛出异常，并打印指定的语句。至于为什么是不等于1就要抛出异常，我们知道getAll方法返回的链表里面包装的是一个泛型，因此我们大胆的认为返回的应该就只有一个集合，否则是错误的情况，是不能再继续使用的。到方法最后，取出这个集合作为返回结果。

方法iterateGroups

  @VisibleForTesting
  public Iterator<EquivGroup> iterateGroups() {
    return groups.iterator();

  }

我们在上文已经解释了iterator是迭代器的意思。返回的是一个Iterator类的东西，可以用于直接读取内部的值。

方法getSignatureOf

  public OpTreeSignature getSignatureOf(Operator<?> op) {
    OpTreeSignature sig = signatureCache.getSignature(op);
    return sig;
  }

我们来看看类OpTreeSignatureFactory是一个什么样子的类：OpTreeSignatureFactory，翻译成中文为一个简单的缓存后端，以防止重复的签名计算。那么它的方法
显而易见是获取签名的，例如需要一些特权操作。而这个方法是获取这个前面并返回。

小结

通过本周的学习，我深刻的知道了关于迭代器、签名的一些操作和相关知识。希望下一周的学习能够给我继续带来新的发现和认识，让我能够再次学以致用。