content: /*
   * 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.asterix.optimizer.rules.am;
  
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Queue;
  
  import org.apache.asterix.common.config.DatasetConfig.DatasetType;
  import org.apache.asterix.common.exceptions.CompilationException;
  import org.apache.asterix.common.exceptions.ErrorCode;
  import org.apache.asterix.common.metadata.DataverseName;
  import org.apache.asterix.metadata.declared.DataSource;
  import org.apache.asterix.metadata.declared.DatasetDataSource;
  import org.apache.asterix.metadata.declared.MetadataProvider;
  import org.apache.asterix.metadata.entities.Dataset;
  import org.apache.asterix.metadata.entities.Index;
  import org.apache.asterix.metadata.utils.ArrayIndexUtil;
  import org.apache.asterix.metadata.utils.DatasetUtil;
  import org.apache.asterix.metadata.utils.KeyFieldTypeUtil;
  import org.apache.asterix.om.functions.BuiltinFunctions;
  import org.apache.asterix.om.types.ARecordType;
  import org.apache.asterix.om.types.ATypeTag;
  import org.apache.asterix.om.types.IAType;
  import org.apache.asterix.om.utils.ConstantExpressionUtil;
  import org.apache.commons.lang3.mutable.Mutable;
  import org.apache.commons.lang3.mutable.MutableObject;
  import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
  import org.apache.hyracks.algebricks.common.utils.Pair;
  import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
  import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
  import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
  import org.apache.hyracks.algebricks.core.algebra.base.LogicalExpressionTag;
  import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
  import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
  import org.apache.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
  import org.apache.hyracks.algebricks.core.algebra.expressions.AbstractLogicalExpression;
  import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
  import org.apache.hyracks.algebricks.core.algebra.expressions.VariableReferenceExpression;
  import org.apache.hyracks.algebricks.core.algebra.functions.AlgebricksBuiltinFunctions;
  import org.apache.hyracks.algebricks.core.algebra.functions.AlgebricksBuiltinFunctions.ComparisonKind;
  import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
  import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
  import org.apache.hyracks.algebricks.core.algebra.operators.logical.AssignOperator;
  import org.apache.hyracks.algebricks.core.algebra.operators.logical.DataSourceScanOperator;
  import org.apache.hyracks.algebricks.core.algebra.operators.logical.IntersectOperator;
  import org.apache.hyracks.algebricks.core.algebra.operators.logical.SelectOperator;
  import org.apache.hyracks.algebricks.core.algebra.operators.logical.UnnestMapOperator;
  import org.apache.hyracks.algebricks.core.algebra.util.OperatorPropertiesUtil;
  import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
  import org.apache.hyracks.api.exceptions.SourceLocation;
  
  public class IntroduceLSMComponentFilterRule implements IAlgebraicRewriteRule {
  
      protected IVariableTypeEnvironment typeEnvironment = null;
  
      @Override
      public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
              throws AlgebricksException {
          return false;
      }
  
      @Override
      public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
              throws AlgebricksException {
          if (!checkIfRuleIsApplicable(opRef, context)) {
              return false;
          }
  
          AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
  
          Dataset dataset = getDataset(op, context);
          Integer filterSourceIndicator = null;
          List<String> filterFieldName = null;
          ARecordType itemType = null;
          MetadataProvider mp = (MetadataProvider) context.getMetadataProvider();
          if (dataset != null && dataset.getDatasetType() == DatasetType.INTERNAL) {
              filterSourceIndicator = DatasetUtil.getFilterSourceIndicator(dataset);
              filterFieldName = DatasetUtil.getFilterField(dataset);
              IAType dataType = mp.findType(dataset.getItemTypeDatabaseName(), dataset.getItemTypeDataverseName(),
                      dataset.getItemTypeName());
              dataType = mp.findTypeForDatasetWithoutType(dataType, dataset);
              IAType metaItemType = mp.findType(dataset.getMetaItemTypeDatabaseName(),
                      dataset.getMetaItemTypeDataverseName(), dataset.getMetaItemTypeName());
              IAType filterSourceType =
                      filterSourceIndicator == null || filterSourceIndicator == 0 ? dataType : metaItemType;
              filterSourceType = mp.findTypeForDatasetWithoutType(filterSourceType, dataset);
  
              if (filterSourceType.getTypeTag() == ATypeTag.OBJECT) {
                  itemType = (ARecordType) filterSourceType;
              }
          }
          if (filterFieldName == null || itemType == null) {
              return false;
          }
  
          IAType filterType = itemType.getSubFieldType(filterFieldName);
  
          typeEnvironment = context.getOutputTypeEnvironment(op);
          ILogicalExpression condExpr = ((SelectOperator) op).getCondition().getValue();
          AccessMethodAnalysisContext analysisCtx = analyzeCondition(condExpr, context, typeEnvironment);
  
          List<IOptimizableFuncExpr> optFuncExprs = new ArrayList<>();
  
          if (!analysisCtx.getMatchedFuncExprs().isEmpty()) {
              List<Index> datasetIndexes = mp.getDatasetIndexes(dataset.getDatabaseName(), dataset.getDataverseName(),
                      dataset.getDatasetName());
  
              for (int i = 0; i < analysisCtx.getMatchedFuncExprs().size(); i++) {
                  IOptimizableFuncExpr optFuncExpr = analysisCtx.getMatchedFuncExpr(i);
                  boolean found = findMatchedExprFieldName(optFuncExpr, op, dataset, itemType, datasetIndexes, context,
                          filterSourceIndicator);
                  // the field name source should be consistent with the filter source indicator
                  if (found && optFuncExpr.getFieldName(0).equals(filterFieldName)
                          && optFuncExpr.getFieldSource(0) == filterSourceIndicator) {
                      optFuncExprs.add(optFuncExpr);
                  }
              }
          }
  
          if (optFuncExprs.isEmpty()) {
              assignFilterFromSecondaryUnnestMap(op, dataset, context, filterType);
          } else {
              assignFilterFromQuery(optFuncExprs, op, dataset, context, filterType);
          }
  
          OperatorPropertiesUtil.typeOpRec(opRef, context);
          context.addToDontApplySet(this, op);
          return true;
      }
  
      private AssignOperator createAssignOperator(List<IOptimizableFuncExpr> optFuncExprs,
              List<LogicalVariable> minFilterVars, List<LogicalVariable> maxFilterVars, IOptimizationContext context,
              SourceLocation sourceLoc) {
          List<LogicalVariable> assignKeyVarList = new ArrayList<>();
          List<Mutable<ILogicalExpression>> assignKeyExprList = new ArrayList<>();
  
          for (IOptimizableFuncExpr optFuncExpr : optFuncExprs) {
              ComparisonKind ck =
                      AlgebricksBuiltinFunctions.getComparisonType(optFuncExpr.getFuncExpr().getFunctionIdentifier());
              ILogicalExpression searchKeyExpr = optFuncExpr.getConstantExpr(0);
              LogicalVariable var = context.newVar();
              assignKeyExprList.add(new MutableObject<>(searchKeyExpr));
              assignKeyVarList.add(var);
              if (ck == ComparisonKind.GE || ck == ComparisonKind.GT) {
                  minFilterVars.add(var);
              } else if (ck == ComparisonKind.LE || ck == ComparisonKind.LT) {
                  maxFilterVars.add(var);
              } else if (ck == ComparisonKind.EQ) {
                  minFilterVars.add(var);
                  maxFilterVars.add(var);
              }
          }
          AssignOperator assignOp = new AssignOperator(assignKeyVarList, assignKeyExprList);
          assignOp.setSourceLocation(sourceLoc);
          return assignOp;
      }
  
      private void assignFilterFromQuery(List<IOptimizableFuncExpr> optFuncExprs, AbstractLogicalOperator op,
              Dataset dataset, IOptimizationContext context, IAType filterType) throws AlgebricksException {
  
          List<UnnestMapOperator> primaryUnnestMapOps = new ArrayList<>();
          boolean hasSecondaryIndexMap = false;
          Queue<Mutable<ILogicalOperator>> queue = new LinkedList<>(op.getInputs());
          while (!queue.isEmpty()) {
              AbstractLogicalOperator descendantOp = (AbstractLogicalOperator) queue.poll().getValue();
              if (descendantOp == null) {
                  continue;
              }
              if (descendantOp.getOperatorTag() == LogicalOperatorTag.DATASOURCESCAN) {
                  DataSourceScanOperator dataSourceScanOp = (DataSourceScanOperator) descendantOp;
                  DataSource ds = (DataSource) dataSourceScanOp.getDataSource();
                  if (dataset.getDatasetName().compareTo(((DatasetDataSource) ds).getDataset().getDatasetName()) == 0) {
                      List<LogicalVariable> minFilterVars = new ArrayList<>();
                      List<LogicalVariable> maxFilterVars = new ArrayList<>();
  
                      AssignOperator assignOp = createAssignOperator(optFuncExprs, minFilterVars, maxFilterVars, context,
                              dataSourceScanOp.getSourceLocation());
  
                      dataSourceScanOp.setMinFilterVars(minFilterVars);
                      dataSourceScanOp.setMaxFilterVars(maxFilterVars);
  
                      List<Mutable<ILogicalExpression>> additionalFilteringExpressions = new ArrayList<>();
                      for (LogicalVariable var : assignOp.getVariables()) {
                          VariableReferenceExpression varRef = new VariableReferenceExpression(var);
                          varRef.setSourceLocation(assignOp.getSourceLocation());
                          additionalFilteringExpressions.add(new MutableObject<ILogicalExpression>(varRef));
                      }
  
                      dataSourceScanOp.setAdditionalFilteringExpressions(additionalFilteringExpressions);
  
                      assignOp.getInputs().add(new MutableObject<>(dataSourceScanOp.getInputs().get(0).getValue()));
                      dataSourceScanOp.getInputs().get(0).setValue(assignOp);
                  }
              } else if (descendantOp.getOperatorTag() == LogicalOperatorTag.UNNEST_MAP) {
                  UnnestMapOperator unnestMapOp = (UnnestMapOperator) descendantOp;
                  ILogicalExpression unnestExpr = unnestMapOp.getExpressionRef().getValue();
                  if (unnestExpr.getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL) {
                      AbstractFunctionCallExpression f = (AbstractFunctionCallExpression) unnestExpr;
                      FunctionIdentifier fid = f.getFunctionIdentifier();
                      if (!fid.equals(BuiltinFunctions.INDEX_SEARCH)) {
                          throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_STATE,
                                  unnestMapOp.getSourceLocation(),
                                  "Illegal function found, expected an " + "index-search.");
                      }
                      AccessMethodJobGenParams jobGenParams = new AccessMethodJobGenParams();
                      jobGenParams.readFromFuncArgs(f.getArguments());
                      if (dataset.getDatasetName().compareTo(jobGenParams.datasetName) == 0) {
                          List<LogicalVariable> minFilterVars = new ArrayList<>();
                          List<LogicalVariable> maxFilterVars = new ArrayList<>();
  
                          AssignOperator assignOp = createAssignOperator(optFuncExprs, minFilterVars, maxFilterVars,
                                  context, unnestMapOp.getSourceLocation());
  
                          unnestMapOp.setMinFilterVars(minFilterVars);
                          unnestMapOp.setMaxFilterVars(maxFilterVars);
  
                          List<Mutable<ILogicalExpression>> additionalFilteringExpressions = new ArrayList<>();
                          for (LogicalVariable var : assignOp.getVariables()) {
                              VariableReferenceExpression varRef = new VariableReferenceExpression(var);
                              varRef.setSourceLocation(assignOp.getSourceLocation());
                              additionalFilteringExpressions.add(new MutableObject<ILogicalExpression>(varRef));
                          }
                          unnestMapOp.setAdditionalFilteringExpressions(additionalFilteringExpressions);
                          assignOp.getInputs().add(new MutableObject<>(unnestMapOp.getInputs().get(0).getValue()));
                          unnestMapOp.getInputs().get(0).setValue(assignOp);
  
                          if (jobGenParams.isPrimaryIndex) {
                              primaryUnnestMapOps.add(unnestMapOp);
                          } else {
                              hasSecondaryIndexMap = true;
                          }
                      }
                  }
              }
              queue.addAll(descendantOp.getInputs());
          }
          if (hasSecondaryIndexMap && !primaryUnnestMapOps.isEmpty()) {
              propagateFilterToPrimaryIndex(primaryUnnestMapOps, filterType, context, false);
          }
      }
  
      private void propagateFilterToPrimaryIndex(List<UnnestMapOperator> primaryUnnestMapOps, IAType filterType,
              IOptimizationContext context, boolean isIndexOnlyPlan) throws AlgebricksException {
          for (UnnestMapOperator primaryOp : primaryUnnestMapOps) {
              Mutable<ILogicalOperator> assignOrOrderOrIntersect = primaryOp.getInputs().get(0);
              Mutable<ILogicalOperator> intersectOrSort = assignOrOrderOrIntersect;
  
              if (assignOrOrderOrIntersect.getValue().getOperatorTag() == LogicalOperatorTag.ASSIGN) {
                  intersectOrSort = assignOrOrderOrIntersect.getValue().getInputs().get(0);
              }
  
              switch (intersectOrSort.getValue().getOperatorTag()) {
                  case INTERSECT:
                      IntersectOperator intersect = (IntersectOperator) (intersectOrSort.getValue());
                      List<List<LogicalVariable>> filterVars = new ArrayList<>(intersect.getInputs().size());
                      for (Mutable<ILogicalOperator> mutableOp : intersect.getInputs()) {
                          ILogicalOperator child = mutableOp.getValue();
                          while (!child.getOperatorTag().equals(LogicalOperatorTag.UNNEST_MAP)) {
                              child = child.getInputs().get(0).getValue();
                          }
                          UnnestMapOperator unnestMap = (UnnestMapOperator) child;
                          propagateFilterInSecondaryUnnsetMap(unnestMap, filterType, context);
  
                          List<LogicalVariable> extraVars = Arrays.asList(unnestMap.getPropagateIndexMinFilterVar(),
                                  unnestMap.getPropagateIndexMaxFilterVar());
                          filterVars.add(extraVars);
                      }
                      if (!filterVars.isEmpty()) {
                          int outputFilterVarsCount = filterVars.get(0).size();
                          List<LogicalVariable> outputFilterVars = new ArrayList<>(outputFilterVarsCount);
                          for (int i = 0; i < outputFilterVarsCount; i++) {
                              outputFilterVars.add(context.newVar());
                          }
                          IntersectOperator intersectWithFilter =
                                  createIntersectWithFilter(outputFilterVars, filterVars, intersect);
                          intersectOrSort.setValue(intersectWithFilter);
                          context.computeAndSetTypeEnvironmentForOperator(intersectWithFilter);
                          setPrimaryFilterVar(primaryOp, outputFilterVars.get(0), outputFilterVars.get(1), context);
                      }
                      break;
                  case ORDER:
                  case DISTINCT:
                      ILogicalOperator child = intersectOrSort.getValue().getInputs().get(0).getValue();
                      if (child.getOperatorTag().equals(LogicalOperatorTag.UNNEST_MAP)) {
                          UnnestMapOperator secondaryMap = (UnnestMapOperator) child;
  
                          // If we are already propagating our index filter, do not repeat this action.
                          if (!secondaryMap.propagateIndexFilter()) {
                              propagateFilterInSecondaryUnnsetMap(secondaryMap, filterType, context);
                              setPrimaryFilterVar(primaryOp, secondaryMap.getPropagateIndexMinFilterVar(),
                                      secondaryMap.getPropagateIndexMaxFilterVar(), context);
                          }
                      }
                      break;
  
                  default:
                      throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_STATE,
                              intersectOrSort.getValue().getOperatorTag().toString());
              }
          }
      }
  
      private IntersectOperator createIntersectWithFilter(List<LogicalVariable> outputFilterVars,
              List<List<LogicalVariable>> filterVars, IntersectOperator intersect) throws AlgebricksException {
          int nInput = intersect.getNumInput();
          List<List<LogicalVariable>> inputCompareVars = new ArrayList<>(nInput);
          for (int i = 0; i < nInput; i++) {
              inputCompareVars.add(new ArrayList<>(intersect.getInputCompareVariables(i)));
          }
          IntersectOperator intersectWithFilter = new IntersectOperator(intersect.getOutputCompareVariables(),
                  outputFilterVars, inputCompareVars, filterVars, intersect.getPartitionsMap());
          intersectWithFilter.setSourceLocation(intersect.getSourceLocation());
          intersectWithFilter.getInputs().addAll(intersect.getInputs());
          return intersectWithFilter;
      }
  
      private void propagateFilterInSecondaryUnnsetMap(UnnestMapOperator secondaryUnnest, IAType filterType,
              IOptimizationContext context) throws AlgebricksException {
  
          LogicalVariable minIndexFilterVar = context.newVar();
          LogicalVariable maxIndexFilterVar = context.newVar();
          secondaryUnnest.markPropagageIndexFilter();
          secondaryUnnest.getVariables().add(minIndexFilterVar);
          secondaryUnnest.getVariableTypes().add(filterType);
          secondaryUnnest.getVariables().add(maxIndexFilterVar);
          secondaryUnnest.getVariableTypes().add(filterType);
  
          context.computeAndSetTypeEnvironmentForOperator(secondaryUnnest);
      }
  
      private void setPrimaryFilterVar(UnnestMapOperator primaryOp, LogicalVariable minFilterVar,
              LogicalVariable maxFilterVar, IOptimizationContext context) throws AlgebricksException {
          primaryOp.setMinFilterVars(Collections.singletonList(minFilterVar));
          primaryOp.setMaxFilterVars(Collections.singletonList(maxFilterVar));
  
          VariableReferenceExpression minFilterVarRef = new VariableReferenceExpression(minFilterVar);
          minFilterVarRef.setSourceLocation(primaryOp.getSourceLocation());
          VariableReferenceExpression maxFilterVarRef = new VariableReferenceExpression(maxFilterVar);
          maxFilterVarRef.setSourceLocation(primaryOp.getSourceLocation());
          List<Mutable<ILogicalExpression>> indexFilterExpression =
                  Arrays.asList(new MutableObject<>(minFilterVarRef), new MutableObject<>(maxFilterVarRef));
  
          primaryOp.setAdditionalFilteringExpressions(indexFilterExpression);
          context.computeAndSetTypeEnvironmentForOperator(primaryOp);
      }
  
      private void assignFilterFromSecondaryUnnestMap(AbstractLogicalOperator op, Dataset dataset,
              IOptimizationContext context, IAType filterType) throws AlgebricksException {
          List<UnnestMapOperator> primaryUnnestMapOps = new ArrayList<>();
          boolean hasSecondaryIndexMap = false;
          boolean isIndexOnlyPlan = false;
          Queue<Mutable<ILogicalOperator>> queue = new LinkedList<>(op.getInputs());
          while (!queue.isEmpty()) {
              ILogicalOperator descendantOp = queue.poll().getValue();
              if (descendantOp.getOperatorTag() == LogicalOperatorTag.UNNEST_MAP) {
                  UnnestMapOperator unnestMapOp = (UnnestMapOperator) descendantOp;
                  ILogicalExpression unnestExpr = unnestMapOp.getExpressionRef().getValue();
                  if (unnestExpr.getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL) {
                      AbstractFunctionCallExpression f = (AbstractFunctionCallExpression) unnestExpr;
                      FunctionIdentifier fid = f.getFunctionIdentifier();
                      if (!fid.equals(BuiltinFunctions.INDEX_SEARCH)) {
                          throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_STATE, f.getSourceLocation(),
                                  fid.getName());
                      }
                      AccessMethodJobGenParams jobGenParams = new AccessMethodJobGenParams();
                      jobGenParams.readFromFuncArgs(f.getArguments());
                      if (dataset.getDatasetName().compareTo(jobGenParams.datasetName) == 0) {
                          if (jobGenParams.isPrimaryIndex) {
                              primaryUnnestMapOps.add(unnestMapOp);
                          } else {
                              hasSecondaryIndexMap = true;
                              isIndexOnlyPlan = unnestMapOp.getGenerateCallBackProceedResultVar();
                          }
                      }
                  }
              }
              queue.addAll(descendantOp.getInputs());
          }
          if (hasSecondaryIndexMap && !primaryUnnestMapOps.isEmpty()) {
              propagateFilterToPrimaryIndex(primaryUnnestMapOps, filterType, context, isIndexOnlyPlan);
          }
      }
  
      private Dataset getDataset(AbstractLogicalOperator op, IOptimizationContext context) throws AlgebricksException {
          AbstractLogicalOperator descendantOp = (AbstractLogicalOperator) op.getInputs().get(0).getValue();
          while (descendantOp != null) {
              if (descendantOp.getOperatorTag() == LogicalOperatorTag.DATASOURCESCAN) {
                  DataSourceScanOperator dataSourceScanOp = (DataSourceScanOperator) descendantOp;
                  DataSource ds = (DataSource) dataSourceScanOp.getDataSource();
                  if (ds.getDatasourceType() != DataSource.Type.INTERNAL_DATASET) {
                      return null;
                  }
                  return ((DatasetDataSource) ds).getDataset();
              } else if (descendantOp.getOperatorTag() == LogicalOperatorTag.UNNEST_MAP) {
                  UnnestMapOperator unnestMapOp = (UnnestMapOperator) descendantOp;
                  ILogicalExpression unnestExpr = unnestMapOp.getExpressionRef().getValue();
                  if (unnestExpr.getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL) {
                      AbstractFunctionCallExpression f = (AbstractFunctionCallExpression) unnestExpr;
                      FunctionIdentifier fid = f.getFunctionIdentifier();
                      DataverseName dataverseName;
                      String datasetName;
                      String database;
                      if (BuiltinFunctions.EXTERNAL_LOOKUP.equals(fid)) {
                          dataverseName = DataverseName
                                  .createFromCanonicalForm(AccessMethodUtils.getStringConstant(f.getArguments().get(0)));
                          datasetName = AccessMethodUtils.getStringConstant(f.getArguments().get(1));
                          database = AccessMethodUtils.getStringConstant(f.getArguments().get(2));
                      } else if (fid.equals(BuiltinFunctions.INDEX_SEARCH)) {
                          AccessMethodJobGenParams jobGenParams = new AccessMethodJobGenParams();
                          jobGenParams.readFromFuncArgs(f.getArguments());
                          dataverseName = jobGenParams.dataverseName;
                          datasetName = jobGenParams.datasetName;
                          database = jobGenParams.databaseName;
                      } else {
                          throw new CompilationException(ErrorCode.COMPILATION_ERROR, f.getSourceLocation(),
                                  "Unexpected function for Unnest Map: " + fid);
                      }
                      return ((MetadataProvider) context.getMetadataProvider()).findDataset(database, dataverseName,
                              datasetName);
                  }
              }
              if (descendantOp.getInputs().isEmpty()) {
                  break;
              }
              descendantOp = (AbstractLogicalOperator) descendantOp.getInputs().get(0).getValue();
          }
          return null;
      }
  
      private boolean checkIfRuleIsApplicable(Mutable<ILogicalOperator> opRef, IOptimizationContext context) {
          // First check that the operator is a select and its condition is a function call.
          AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
          if (context.checkIfInDontApplySet(this, op)) {
              return false;
          }
          if (op.getOperatorTag() != LogicalOperatorTag.SELECT) {
              return false;
          }
  
          ILogicalExpression condExpr = ((SelectOperator) op).getCondition().getValue();
          if (condExpr.getExpressionTag() != LogicalExpressionTag.FUNCTION_CALL) {
              return false;
          }
          return true;
      }
  
      private AccessMethodAnalysisContext analyzeCondition(ILogicalExpression cond, IOptimizationContext context,
              IVariableTypeEnvironment typeEnvironment) throws AlgebricksException {
          AccessMethodAnalysisContext analysisCtx = new AccessMethodAnalysisContext();
          AbstractFunctionCallExpression funcExpr = (AbstractFunctionCallExpression) cond;
          FunctionIdentifier funcIdent = funcExpr.getFunctionIdentifier();
          if (funcIdent != AlgebricksBuiltinFunctions.OR) {
              analyzeFunctionExpr(funcExpr, analysisCtx, context, typeEnvironment);
              for (Mutable<ILogicalExpression> arg : funcExpr.getArguments()) {
                  ILogicalExpression argExpr = arg.getValue();
                  if (argExpr.getExpressionTag() != LogicalExpressionTag.FUNCTION_CALL) {
                      continue;
                  }
                  analyzeFunctionExpr((AbstractFunctionCallExpression) argExpr, analysisCtx, context, typeEnvironment);
              }
          }
          return analysisCtx;
      }
  
      private void analyzeFunctionExpr(AbstractFunctionCallExpression funcExpr, AccessMethodAnalysisContext analysisCtx,
              IOptimizationContext context, IVariableTypeEnvironment typeEnvironment) throws AlgebricksException {
          FunctionIdentifier funcIdent = funcExpr.getFunctionIdentifier();
          if (funcIdent == AlgebricksBuiltinFunctions.LE || funcIdent == AlgebricksBuiltinFunctions.GE
                  || funcIdent == AlgebricksBuiltinFunctions.LT || funcIdent == AlgebricksBuiltinFunctions.GT
                  || funcIdent == AlgebricksBuiltinFunctions.EQ) {
              AccessMethodUtils.analyzeFuncExprArgsForOneConstAndVarAndUpdateAnalysisCtx(funcExpr, analysisCtx, context,
                      typeEnvironment, false);
          }
      }
  
      private boolean findMatchedExprFieldName(IOptimizableFuncExpr optFuncExpr, AbstractLogicalOperator op,
              Dataset dataset, ARecordType filterSourceType, List<Index> datasetIndexes, IOptimizationContext context,
              Integer filterSourceIndicator) throws AlgebricksException {
          AbstractLogicalOperator descendantOp = (AbstractLogicalOperator) op.getInputs().get(0).getValue();
          while (descendantOp != null) {
              if (descendantOp.getOperatorTag() == LogicalOperatorTag.ASSIGN) {
                  AssignOperator assignOp = (AssignOperator) descendantOp;
                  List<LogicalVariable> varList = assignOp.getVariables();
                  for (int varIndex = 0; varIndex < varList.size(); varIndex++) {
                      LogicalVariable var = varList.get(varIndex);
                      int funcVarIndex = optFuncExpr.findLogicalVar(var);
                      if (funcVarIndex == -1) {
                          continue;
                      }
                      Pair<ARecordType, List<String>> fieldNamePairs =
                              getFieldNameFromSubAssignTree(optFuncExpr, descendantOp, varIndex, filterSourceType,
                                      filterSourceIndicator, dataset.getPrimaryKeys().size());
                      if (fieldNamePairs == null) {
                          return false;
                      }
                      List<String> fieldName = fieldNamePairs.second;
                      // Since we validated the filter source in getFieldNameFromSubAssignTree, we can safely set the
                      // fieldSource to be filterSourceIndicator
                      optFuncExpr.setFieldName(funcVarIndex, fieldName, filterSourceIndicator);
                      return true;
                  }
              } else if (descendantOp.getOperatorTag() == LogicalOperatorTag.DATASOURCESCAN) {
                  DataSourceScanOperator scanOp = (DataSourceScanOperator) descendantOp;
                  List<LogicalVariable> varList = scanOp.getVariables();
                  for (int varIndex = 0; varIndex < varList.size(); varIndex++) {
                      LogicalVariable var = varList.get(varIndex);
                      int funcVarIndex = optFuncExpr.findLogicalVar(var);
                      if (funcVarIndex == -1) {
                          continue;
                      }
                      // The variable value is one of the partitioning fields.
                      List<String> fieldName = dataset.getPrimaryKeys().get(varIndex);
                      if (fieldName == null) {
                          return false;
                      }
                      List<Integer> keySourceIndicators = DatasetUtil.getKeySourceIndicators(dataset);
                      int keySource = getKeySource(keySourceIndicators, varIndex);
                      optFuncExpr.setFieldName(funcVarIndex, fieldName, keySource);
                      return true;
                  }
              } else if (descendantOp.getOperatorTag() == LogicalOperatorTag.UNNEST_MAP) {
                  UnnestMapOperator unnestMapOp = (UnnestMapOperator) descendantOp;
                  List<LogicalVariable> varList = unnestMapOp.getVariables();
                  for (int varIndex = 0; varIndex < varList.size(); varIndex++) {
                      LogicalVariable var = varList.get(varIndex);
                      int funcVarIndex = optFuncExpr.findLogicalVar(var);
                      if (funcVarIndex == -1) {
                          continue;
                      }
  
                      String indexName;
                      Index index = null;
                      ILogicalExpression unnestExpr = unnestMapOp.getExpressionRef().getValue();
                      if (unnestExpr.getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL) {
                          AbstractFunctionCallExpression f = (AbstractFunctionCallExpression) unnestExpr;
                          FunctionIdentifier fid = f.getFunctionIdentifier();
                          if (!fid.equals(BuiltinFunctions.INDEX_SEARCH)) {
                              throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_STATE,
                                      unnestMapOp.getSourceLocation(),
                                      "Illegal function found, expected an " + "index-search.");
                          }
                          AccessMethodJobGenParams jobGenParams = new AccessMethodJobGenParams();
                          jobGenParams.readFromFuncArgs(f.getArguments());
                          indexName = jobGenParams.indexName;
                          for (Index idx : datasetIndexes) {
                              if (idx.getIndexName().compareTo(indexName) == 0) {
                                  index = idx;
                                  break;
                              }
                          }
                      }
                      if (index == null) {
                          throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_STATE,
                                  unnestMapOp.getSourceLocation(),
                                  "Could not find the corresponding index for an" + " index search.");
                      }
  
                      IAType metaItemType = ((MetadataProvider) context.getMetadataProvider()).findType(
                              dataset.getMetaItemTypeDatabaseName(), dataset.getMetaItemTypeDataverseName(),
                              dataset.getMetaItemTypeName());
                      IAType recordItemType = ((MetadataProvider) context.getMetadataProvider()).findType(
                              dataset.getItemTypeDatabaseName(), dataset.getItemTypeDataverseName(),
                              dataset.getItemTypeName());
                      recordItemType = ((MetadataProvider) context.getMetadataProvider())
                              .findTypeForDatasetWithoutType(recordItemType, dataset);
                      ARecordType recordType = (ARecordType) recordItemType;
                      ARecordType metaRecType = (ARecordType) metaItemType;
                      int numSecondaryKeys = KeyFieldTypeUtil.getNumSecondaryKeys(index, recordType, metaRecType);
                      List<String> fieldName;
                      int keySource;
                      if (varIndex >= numSecondaryKeys) {
                          int idx = varIndex - numSecondaryKeys;
                          fieldName = dataset.getPrimaryKeys().get(idx);
                          keySource = getKeySource(DatasetUtil.getKeySourceIndicators(dataset), idx);
                      } else {
                          List<List<String>> keyFieldNames;
                          List<Integer> keySources;
                          switch (Index.IndexCategory.of(index.getIndexType())) {
                              case ARRAY:
                                  Index.ArrayIndexDetails arrayIndexDetails =
                                          (Index.ArrayIndexDetails) index.getIndexDetails();
                                  keyFieldNames = new ArrayList<>();
                                  keySources = new ArrayList<>();
                                  for (Index.ArrayIndexElement e : arrayIndexDetails.getElementList()) {
                                      for (List<String> project : e.getProjectList()) {
                                          keyFieldNames.add(
                                                  ArrayIndexUtil.getFlattenedKeyFieldNames(e.getUnnestList(), project));
                                          keySources.add(e.getSourceIndicator());
                                      }
                                  }
                                  break;
                              case VALUE:
                                  Index.ValueIndexDetails valueIndexDetails =
                                          (Index.ValueIndexDetails) index.getIndexDetails();
                                  keyFieldNames = valueIndexDetails.getKeyFieldNames();
                                  keySources = valueIndexDetails.getKeyFieldSourceIndicators();
                                  break;
                              case TEXT:
                                  Index.TextIndexDetails textIndexDetails =
                                          (Index.TextIndexDetails) index.getIndexDetails();
                                  keyFieldNames = textIndexDetails.getKeyFieldNames();
                                  keySources = textIndexDetails.getKeyFieldSourceIndicators();
                                  break;
                              default:
                                  throw new CompilationException(ErrorCode.COMPILATION_UNKNOWN_INDEX_TYPE,
                                          String.valueOf(index.getIndexType()));
                          }
                          fieldName = keyFieldNames.get(varIndex);
                          keySource = getKeySource(keySources, varIndex);
                      }
                      if (fieldName == null) {
                          return false;
                      }
                      optFuncExpr.setFieldName(funcVarIndex, fieldName, keySource);
                      return true;
                  }
              }
  
              if (descendantOp.getInputs().isEmpty()) {
                  break;
              }
              descendantOp = (AbstractLogicalOperator) descendantOp.getInputs().get(0).getValue();
          }
          return false;
      }
  
      private static int getKeySource(List<Integer> keySourceIndicators, int keyIdx) {
          return keySourceIndicators == null ? 0 : keySourceIndicators.get(keyIdx);
      }
  
      private Pair<ARecordType, List<String>> getFieldNameFromSubAssignTree(IOptimizableFuncExpr optFuncExpr,
              AbstractLogicalOperator op, int varIndex, ARecordType filterSourceType, Integer filterSourceIndicator,
              int numOfPKeys) {
          AbstractLogicalExpression expr = null;
          if (op.getOperatorTag() == LogicalOperatorTag.ASSIGN) {
              AssignOperator assignOp = (AssignOperator) op;
              expr = (AbstractLogicalExpression) assignOp.getExpressions().get(varIndex).getValue();
          }
          if (expr == null || expr.getExpressionTag() != LogicalExpressionTag.FUNCTION_CALL) {
              return null;
          }
          AbstractFunctionCallExpression funcExpr = (AbstractFunctionCallExpression) expr;
          FunctionIdentifier funcIdent = funcExpr.getFunctionIdentifier();
          if (funcIdent == BuiltinFunctions.FIELD_ACCESS_BY_NAME || funcIdent == BuiltinFunctions.FIELD_ACCESS_BY_INDEX) {
  
              //get the variable from here. Figure out which input it came from. Go to that input!!!
              ArrayList<LogicalVariable> usedVars = new ArrayList<>();
              expr.getUsedVariables(usedVars);
              LogicalVariable usedVar = usedVars.get(0);
              List<String> returnList = new ArrayList<>();
  
              //Find the input that it came from
              for (int varCheck = 0; varCheck < op.getInputs().size(); varCheck++) {
                  AbstractLogicalOperator nestedOp = (AbstractLogicalOperator) op.getInputs().get(varCheck).getValue();
                  if (nestedOp.getOperatorTag() != LogicalOperatorTag.ASSIGN) {
                      if (nestedOp.getOperatorTag() != LogicalOperatorTag.DATASOURCESCAN) {
                          return null;
                      }
                      List<LogicalVariable> scannedVars = ((DataSourceScanOperator) nestedOp).getScanVariables();
                      if (scannedVars.indexOf(usedVar) != filterSourceIndicator + numOfPKeys) {
                          return null;
                      }
                  } else {
                      int nestedAssignVar = ((AssignOperator) nestedOp).getVariables().indexOf(usedVar);
                      if (nestedAssignVar == -1) {
                          continue;
                      }
                      //get the nested info from the lower input
                      Pair<ARecordType, List<String>> lowerInfo = getFieldNameFromSubAssignTree(optFuncExpr,
                              (AbstractLogicalOperator) op.getInputs().get(varCheck).getValue(), nestedAssignVar,
                              filterSourceType, filterSourceIndicator, numOfPKeys);
                      if (lowerInfo != null) {
                          // propagate filterSourceType in case the filter value comes from a nested attribute.
                          filterSourceType = lowerInfo.first;
                          returnList = lowerInfo.second;
                      }
                  }
              }
  
              if (funcIdent == BuiltinFunctions.FIELD_ACCESS_BY_NAME) {
                  String fieldName = ConstantExpressionUtil.getStringArgument(funcExpr, 1);
                  if (fieldName == null) {
                      return null;
                  }
                  returnList.add(fieldName);
                  return new Pair<>(filterSourceType, returnList);
              } else if (funcIdent == BuiltinFunctions.FIELD_ACCESS_BY_INDEX) {
                  Integer fieldIndex = ConstantExpressionUtil.getIntArgument(funcExpr, 1);
                  if (fieldIndex == null) {
                      return null;
                  }
                  returnList.add(filterSourceType.getFieldNames()[fieldIndex]);
                  IAType subType = filterSourceType.getFieldTypes()[fieldIndex];
                  if (subType.getTypeTag() == ATypeTag.OBJECT) {
                      filterSourceType = (ARecordType) subType;
                  }
                  return new Pair<>(filterSourceType, returnList);
              }
  
          }
  
          ILogicalExpression argExpr = funcExpr.getArguments().get(0).getValue();
          if (argExpr.getExpressionTag() != LogicalExpressionTag.VARIABLE) {
              return null;
          }
  
          return null;
      }
  }
  