fe/src/main/java/com/cloudera/impala/planner/HdfsScanNode.java
/**
   * Populate partitions_ based on all applicable conjuncts and remove
   * conjuncts used for filtering from conjuncts_.
   */
Populate partitions_ based on all applicable conjuncts and remove
conjuncts used for filtering from conjuncts_.
Populate partitions_ based on all applicable conjuncts and remove conjuncts used for filtering from conjuncts_ .
private void prunePartitions(Analyzer analyzer)
      throws InternalException, AuthorizationException {
    List<SlotId> partitionSlots = Lists.newArrayList();
    for (SlotDescriptor slotDesc:
        analyzer.getDescTbl().getTupleDesc(tupleIds_.get(0)).getSlots()) {
      Preconditions.checkState(slotDesc.getColumn() != null);
      if (slotDesc.getColumn().getPosition() < tbl_.getNumClusteringCols()) {
        partitionSlots.add(slotDesc.getId());
      }
    }
    List<HdfsPartitionFilter> partitionFilters = Lists.newArrayList();
    List<Expr> filterConjuncts = Lists.newArrayList();
    for (Expr conjunct: conjuncts_) {
      if (conjunct.isBoundBySlotIds(partitionSlots)) {
        partitionFilters.add(new HdfsPartitionFilter(conjunct, tbl_, analyzer));
        filterConjuncts.add(conjunct);
      }
    }
    conjuncts_.removeAll(filterConjuncts);
    int partitionsCount = tbl_.getPartitions().size();
    HashMap<Long, HdfsPartition> partitionMap =
        new HashMap<Long, HdfsPartition>(partitionsCount);
    HashSet<Long> validPartitionIds = new HashSet<Long>();
    for (HdfsPartition p: tbl_.getPartitions()) {
      if (p.getFileDescriptors().size() == 0) continue;
      Preconditions.checkState(
          p.getPartitionValues().size() == tbl_.getNumClusteringCols());
      partitionMap.put(p.getId(), p);
      validPartitionIds.add(p.getId());
    }
    HashSet<Long> matchingIds = new HashSet<Long>();
    ArrayList<HdfsPartition> partitionBatch = new ArrayList<HdfsPartition>();
    for (HdfsPartitionFilter filter: partitionFilters) {
      for (Long id: validPartitionIds) {
        partitionBatch.add(partitionMap.get(id));
        if (partitionBatch.size() == PARTITION_PRUNING_BATCH_SIZE) {
          matchingIds.addAll(filter.getMatchingPartitionIds(partitionBatch, analyzer));
          partitionBatch.clear();
        }
      }
      if (!partitionBatch.isEmpty()) {
        matchingIds.addAll(filter.getMatchingPartitionIds(partitionBatch, analyzer));
        partitionBatch.clear();
      }
      validPartitionIds.retainAll(matchingIds);
      matchingIds.clear();
    }
    for (Long id: validPartitionIds) {
      partitions_.add(partitionMap.get(id));
    }
  }
private void prunePartitions ( Analyzer analyzer ) throws InternalException , AuthorizationException { List < SlotId > partitionSlots = Lists . newArrayList ( ) ; for ( SlotDescriptor slotDesc : analyzer . getDescTbl ( ) . getTupleDesc ( tupleIds_ . get ( 0 ) ) . getSlots ( ) ) { Preconditions . checkState ( slotDesc . getColumn ( ) != null ) ; if ( slotDesc . getColumn ( ) . getPosition ( ) < tbl_ . getNumClusteringCols ( ) ) { partitionSlots . add ( slotDesc . getId ( ) ) ; } } List < HdfsPartitionFilter > partitionFilters = Lists . newArrayList ( ) ; List < Expr > filterConjuncts = Lists . newArrayList ( ) ; for ( Expr conjunct : conjuncts_ ) { if ( conjunct . isBoundBySlotIds ( partitionSlots ) ) { partitionFilters . add ( new HdfsPartitionFilter ( conjunct , tbl_ , analyzer ) ) ; filterConjuncts . add ( conjunct ) ; } } conjuncts_ . removeAll ( filterConjuncts ) ; int partitionsCount = tbl_ . getPartitions ( ) . size ( ) ; HashMap < Long , HdfsPartition > partitionMap = new HashMap < Long , HdfsPartition > ( partitionsCount ) ; HashSet < Long > validPartitionIds = new HashSet < Long > ( ) ; for ( HdfsPartition p : tbl_ . getPartitions ( ) ) { if ( p . getFileDescriptors ( ) . size ( ) == 0 ) continue ; Preconditions . checkState ( p . getPartitionValues ( ) . size ( ) == tbl_ . getNumClusteringCols ( ) ) ; partitionMap . put ( p . getId ( ) , p ) ; validPartitionIds . add ( p . getId ( ) ) ; } HashSet < Long > matchingIds = new HashSet < Long > ( ) ; ArrayList < HdfsPartition > partitionBatch = new ArrayList < HdfsPartition > ( ) ; for ( HdfsPartitionFilter filter : partitionFilters ) { for ( Long id : validPartitionIds ) { partitionBatch . add ( partitionMap . get ( id ) ) ; if ( partitionBatch . size ( ) == PARTITION_PRUNING_BATCH_SIZE ) { matchingIds . addAll ( filter . getMatchingPartitionIds ( partitionBatch , analyzer ) ) ; partitionBatch . clear ( ) ; } } if ( ! partitionBatch . isEmpty ( ) ) { matchingIds . addAll ( filter . getMatchingPartitionIds ( partitionBatch , analyzer ) ) ; partitionBatch . clear ( ) ; } validPartitionIds . retainAll ( matchingIds ) ; matchingIds . clear ( ) ; } for ( Long id : validPartitionIds ) { partitions_ . add ( partitionMap . get ( id ) ) ; } }
Populate partitions_ based on all applicable conjuncts and remove
 conjuncts used for filtering from conjuncts_.
Populate partitions_ based on all applicable conjuncts and remove conjuncts used for filtering from conjuncts_ .
// loop through all partitions and prune based on applicable conjuncts; // start with creating a collection of partition filters for the applicable conjuncts // filterConjuncts are applied implicitly via partition pruning // Map of partition ids to partition objects // Set of valid partition ids. A partition is considered 'valid' if it // passes all the partition filters. Initially, all the partitions are // considered 'valid'. // TODO: To avoid generating too many objects, the first filter should // iterate directly over the tbl_.getPartitions() and then store the ids that // pass in validPartitionIds. // ignore partitions without data // Set of partition ids that pass a filter. // Batch of partitions // Identify the partitions that pass all filters. // Iterate through the currently valid partitions // Add the partition to the current batch // Batch is full. Evaluate the predicates of this batch in the BE. // Check if there are any unprocessed partitions. // Prune the partitions ids that didn't pass the filter // Populate the list of valid partitions to process