public static VectorPTFEvaluatorBase getEvaluator()

in ql/src/java/org/apache/hadoop/hive/ql/plan/VectorPTFDesc.java [172:420]
233 lines of code
63 McCabe index (conditional complexity)

  public static VectorPTFEvaluatorBase getEvaluator(SupportedFunctionType functionType,
      boolean isDistinct, WindowFrameDef windowFrameDef, Type[] columnVectorTypes,
      VectorExpression[] inputVectorExpressions, int outputColumnNum) {

    final boolean isRowEndCurrent = (windowFrameDef.getWindowType() == WindowType.ROWS
        && windowFrameDef.getEnd().isCurrentRow());
    /*
     * we should only stream when the window start is unbounded and the end row is the current,
     * because that's the way how streaming evaluation works: calculate from the very-first row then
     * create result for the current row on the fly, so with other words: currently we cannot force
     * a boundary on a streaming evaluator
     */
    final boolean canStream = windowFrameDef.getStart().isUnbounded() && isRowEndCurrent;

    // most of the evaluators will use only first argument
    VectorExpression inputVectorExpression = inputVectorExpressions[0];
    Type columnVectorType = columnVectorTypes[0];

    VectorPTFEvaluatorBase evaluator;
    switch (functionType) {
    case ROW_NUMBER:
      evaluator =
          new VectorPTFEvaluatorRowNumber(windowFrameDef, inputVectorExpression, outputColumnNum);
      break;
    case RANK:
      evaluator = new VectorPTFEvaluatorRank(windowFrameDef, outputColumnNum);
      break;
    case DENSE_RANK:
      evaluator = new VectorPTFEvaluatorDenseRank(windowFrameDef, outputColumnNum);
      break;
    case MIN:
      switch (columnVectorType) {
      case LONG:
        evaluator = !canStream ?
            new VectorPTFEvaluatorLongMin(
                windowFrameDef, inputVectorExpression, outputColumnNum) :
            new VectorPTFEvaluatorStreamingLongMin(
                windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      case DOUBLE:
        evaluator = !canStream ?
            new VectorPTFEvaluatorDoubleMin(
                windowFrameDef, inputVectorExpression, outputColumnNum) :
            new VectorPTFEvaluatorStreamingDoubleMin(
                windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      case DECIMAL:
        evaluator = !canStream ?
            new VectorPTFEvaluatorDecimalMin(
                windowFrameDef, inputVectorExpression, outputColumnNum) :
            new VectorPTFEvaluatorStreamingDecimalMin(
                windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      default:
        throw new RuntimeException("Unexpected column vector type " + columnVectorType + " for " + functionType);
      }
      break;
    case MAX:
      switch (columnVectorType) {
      case LONG:
        evaluator = !canStream ?
            new VectorPTFEvaluatorLongMax(
                windowFrameDef, inputVectorExpression, outputColumnNum) :
            new VectorPTFEvaluatorStreamingLongMax(
                windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      case DOUBLE:
        evaluator = !canStream ?
            new VectorPTFEvaluatorDoubleMax(
                windowFrameDef, inputVectorExpression, outputColumnNum) :
            new VectorPTFEvaluatorStreamingDoubleMax(
                windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      case DECIMAL:
        evaluator = !canStream ?
            new VectorPTFEvaluatorDecimalMax(
                windowFrameDef, inputVectorExpression, outputColumnNum) :
            new VectorPTFEvaluatorStreamingDecimalMax(
                windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      default:
        throw new RuntimeException("Unexpected column vector type " + columnVectorType + " for " + functionType);
      }
      break;
    case SUM:
      switch (columnVectorType) {
      case LONG:
        evaluator = !canStream ?
            new VectorPTFEvaluatorLongSum(
                windowFrameDef, inputVectorExpression, outputColumnNum) :
            new VectorPTFEvaluatorStreamingLongSum(
                windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      case DOUBLE:
        evaluator = !canStream ?
            new VectorPTFEvaluatorDoubleSum(
                windowFrameDef, inputVectorExpression, outputColumnNum) :
            new VectorPTFEvaluatorStreamingDoubleSum(
                windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      case DECIMAL:
        evaluator = !canStream ?
            new VectorPTFEvaluatorDecimalSum(
                windowFrameDef, inputVectorExpression, outputColumnNum) :
            new VectorPTFEvaluatorStreamingDecimalSum(
                windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      default:
        throw new RuntimeException("Unexpected column vector type " + columnVectorType + " for " + functionType);
      }
      break;
    case AVG:
      switch (columnVectorType) {
      case LONG:
        evaluator = !canStream ?
            new VectorPTFEvaluatorLongAvg(
                windowFrameDef, inputVectorExpression, outputColumnNum) :
            new VectorPTFEvaluatorStreamingLongAvg(
                windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      case DOUBLE:
        evaluator = !canStream ?
            new VectorPTFEvaluatorDoubleAvg(
                windowFrameDef, inputVectorExpression, outputColumnNum) :
            new VectorPTFEvaluatorStreamingDoubleAvg(
                windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      case DECIMAL:
        evaluator = !canStream ?
            new VectorPTFEvaluatorDecimalAvg(
                windowFrameDef, inputVectorExpression, outputColumnNum) :
            new VectorPTFEvaluatorStreamingDecimalAvg(
                windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      default:
        throw new RuntimeException("Unexpected column vector type " + columnVectorType + " for " + functionType);
      }
      break;
    case FIRST_VALUE:
      switch (columnVectorType) {
      case LONG:
        evaluator = new VectorPTFEvaluatorLongFirstValue(windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      case DOUBLE:
        evaluator = new VectorPTFEvaluatorDoubleFirstValue(windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      case DECIMAL:
        evaluator = new VectorPTFEvaluatorDecimalFirstValue(windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      default:
        throw new RuntimeException("Unexpected column vector type " + columnVectorType + " for " + functionType);
      }
      break;
    case LAST_VALUE:
      switch (columnVectorType) {
      case LONG:
        evaluator = new VectorPTFEvaluatorLongLastValue(windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      case DOUBLE:
        evaluator = new VectorPTFEvaluatorDoubleLastValue(windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      case DECIMAL:
        evaluator = new VectorPTFEvaluatorDecimalLastValue(windowFrameDef, inputVectorExpression, outputColumnNum);
        break;
      default:
        throw new RuntimeException("Unexpected column vector type " + columnVectorType + " for " + functionType);
      }
      break;
    case COUNT:
      if (inputVectorExpression == null) {
        evaluator = new VectorPTFEvaluatorCountStar(windowFrameDef, inputVectorExpression, outputColumnNum);
      } else {
        if (isDistinct) {
          switch (columnVectorType) {
          case BYTES:
            evaluator = new VectorPTFEvaluatorBytesCountDistinct(windowFrameDef,
                inputVectorExpression, outputColumnNum);
            break;
          case DECIMAL_64: //Decimal64ColumnVector is a LongColumnVector
          case LONG:
            evaluator = new VectorPTFEvaluatorLongCountDistinct(windowFrameDef,
                inputVectorExpression, outputColumnNum);
            break;
          case DOUBLE:
            evaluator = new VectorPTFEvaluatorDoubleCountDistinct(windowFrameDef,
                inputVectorExpression, outputColumnNum);
            break;
          case DECIMAL:
            evaluator = new VectorPTFEvaluatorDecimalCountDistinct(windowFrameDef,
                inputVectorExpression, outputColumnNum);
            break;
          case TIMESTAMP:
            evaluator = new VectorPTFEvaluatorTimestampCountDistinct(windowFrameDef,
                inputVectorExpression, outputColumnNum);
            break;
          default:
            throw new RuntimeException(
                "Unexpected column type for ptf count distinct: " + columnVectorType);
          }
        } else {
          evaluator =
              new VectorPTFEvaluatorCount(windowFrameDef, inputVectorExpression, outputColumnNum);
        }
      }
      break;
    case LAG:
      // lag(column, constant, ...)
      int amt = inputVectorExpressions.length > 1
        ? (int) ((ConstantVectorExpression) inputVectorExpressions[1]).getLongValue() : 1;

      // lag(column, constant, constant/column)
      VectorExpression defaultValueExpression =
          inputVectorExpressions.length > 2 ? inputVectorExpressions[2] : null;
      switch (columnVectorType) {
      case LONG:
      case DOUBLE:
      case DECIMAL:
        evaluator = new VectorPTFEvaluatorLag(windowFrameDef, inputVectorExpression,
            outputColumnNum, columnVectorType, amt, defaultValueExpression);
        break;
      default:
        throw new RuntimeException(
            "Unexpected column vector type " + columnVectorType + " for " + functionType);
      }
      break;
    case LEAD:
      // lead(column, constant, ...)
      amt = inputVectorExpressions.length > 1
        ? (int) ((ConstantVectorExpression) inputVectorExpressions[1]).getLongValue() : 1;

      // lead(column, constant, constant/column)
      defaultValueExpression = inputVectorExpressions.length > 2 ? inputVectorExpressions[2] : null;
      switch (columnVectorType) {
      case LONG:
      case DOUBLE:
      case DECIMAL:
        evaluator = new VectorPTFEvaluatorLead(windowFrameDef, inputVectorExpression,
            outputColumnNum, columnVectorType, amt, defaultValueExpression);
        break;
      default:
        throw new RuntimeException(
            "Unexpected column vector type " + columnVectorType + " for " + functionType);
      }
      break;
    default:
      throw new RuntimeException("Unexpected function type " + functionType);
    }
    return evaluator;
  }