public Index createIndex(String indexName, IndexType indexType, String origIndexedExpression,
      String origFromClause, String imports, ExecutionContext externalContext,
      PartitionedIndex prIndex, boolean loadEntries)
      throws IndexNameConflictException, IndexExistsException, IndexInvalidException {

    if (QueryMonitor.isLowMemory()) {
      throw new IndexInvalidException(
          "Index creation canceled due to low memory");
    }

    boolean oldReadSerialized = cache.getPdxReadSerializedOverride();
    cache.setPdxReadSerializedOverride(true);

    TXStateProxy tx = null;
    if (!cache.isClient()) {
      tx = ((TXManagerImpl) cache.getCacheTransactionManager()).pauseTransaction();
    }

    try {
      String projectionAttributes = "*"; // for now this is the only option

      if (getIndex(indexName) != null) {
        throw new IndexNameConflictException(
            String.format("Index named ' %s ' already exists.",
                indexName));
      }

      IndexCreationHelper helper = null;
      boolean isCompactOrHash = false;
      // Hash index not supported for overflow but we "thought" we were so let's maintain backwards
      // compatibility
      // and create a regular compact range index instead. This is due to having to reload entries
      // from overflow just
      // to recalculate the index key for the entry for comparisons during query.
      if (indexType == IndexType.HASH && isOverFlowRegion()) {
        indexType = IndexType.FUNCTIONAL;
      }
      if (indexType != IndexType.PRIMARY_KEY) {
        helper = new FunctionalIndexCreationHelper(origFromClause, origIndexedExpression,
            projectionAttributes, imports, (InternalCache) region.getCache(), externalContext,
            this);
        // Asif: For now support Map index as non compact .expand later
        // The limitation for compact range index also apply to hash index for now
        isCompactOrHash = shouldCreateCompactIndex((FunctionalIndexCreationHelper) helper);
      } else if (indexType == IndexType.PRIMARY_KEY) {
        helper = new PrimaryKeyIndexCreationHelper(origFromClause, origIndexedExpression,
            projectionAttributes, (InternalCache) region.getCache(), externalContext, this);
      } else {
        throw new AssertionError("Don't know how to set helper for " + indexType);
      }
      if (!isCompactOrHash && indexType != IndexType.PRIMARY_KEY) {

        if (indexType == IndexType.HASH) {
          if (!isIndexMaintenanceTypeSynchronous()) {
            throw new UnsupportedOperationException(
                "Hash index is currently not supported for regions with Asynchronous index maintenance.");
          }
          throw new UnsupportedOperationException(
              "Hash Index is not supported with from clause having multiple iterators(collections).");
        }
        // Overflow is not supported with range index.
        if (isOverFlowRegion()) {
          throw new UnsupportedOperationException(
              String.format(
                  "The specified index conditions are not supported for regions which overflow to disk. The region involved is %s",
                  region.getFullPath()));
        }
        // OffHeap is not supported with range index.
        if (isOffHeap()) {
          if (!isIndexMaintenanceTypeSynchronous()) {
            throw new UnsupportedOperationException(
                String.format(
                    "Asynchronous index maintenance is currently not supported for off-heap regions. The off-heap region is %s",
                    region.getFullPath()));
          }
          throw new UnsupportedOperationException(
              String.format(
                  "From clauses having multiple iterators(collections) are not supported for off-heap regions. The off-heap region is %s",
                  region.getFullPath()));
        }
      }

      if (logger.isDebugEnabled()) {
        logger.debug("Started creating index with indexName: {} On region: {}", indexName,
            region.getFullPath());
      }

      if (IndexManager.testHook != null) {
        if (logger.isDebugEnabled()) {
          logger.debug("IndexManager TestHook is set.");
        }

        if (((LocalRegion) region).isInitialized()) {
          testHook.hook(1);
        } else {
          testHook.hook(0);
        }
      }

      IndexTask indexTask = new IndexTask(cache, indexName, indexType, origFromClause,
          origIndexedExpression, helper, isCompactOrHash, prIndex, loadEntries);
      FutureTask<Index> indexFutureTask = new FutureTask<>(indexTask);
      Object oldIndex = indexes.putIfAbsent(indexTask, indexFutureTask);

      Index index = null;

      boolean interrupted = false;
      try {
        if (oldIndex == null) {
          // Initialize index.
          indexFutureTask.run();
          // Set the index.
          index = indexFutureTask.get();
        } else {
          // Index with same name or characteristic already exists.
          // Check if index creation is complete.
          if (!(oldIndex instanceof Index)) {
            // Some other thread is creating the same index.
            // Wait for index to be initialized from other thread.
            ((Future) oldIndex).get();
          }

          // The Index is successfully created, throw appropriate error message
          // from this thread.
          if (getIndex(indexName) != null) {
            throw new IndexNameConflictException(
                String.format("Index named ' %s ' already exists.",
                    indexName));
          } else {
            throw new IndexExistsException(
                "Similar Index Exists");
          }
        }
      } catch (InterruptedException ignored) {
        interrupted = true;
      } catch (ExecutionException ee) {
        Throwable c = ee.getCause();
        if (c instanceof IndexNameConflictException) {
          throw (IndexNameConflictException) c;
        } else if (c instanceof IndexExistsException) {
          throw (IndexExistsException) c;
        } else if (c instanceof IMQException) {
          throw new IndexInvalidException(c.getMessage());
        }
        throw new IndexInvalidException(ee);

      } finally {
        // If the index is not successfully created, remove IndexTask from
        // the map.
        if (oldIndex == null && index == null) {
          Object ind = indexes.get(indexTask);
          if (ind != null && !(ind instanceof Index)) {
            indexes.remove(indexTask);
          }
        }
        if (interrupted) {
          Thread.currentThread().interrupt();
        }
      }
      assert (index != null);
      if (logger.isDebugEnabled()) {
        logger.debug("Completed creating index with indexName: {} On region: {}", indexName,
            region.getFullPath());
      }
      return index;

    } finally {
      cache.setPdxReadSerializedOverride(oldReadSerialized);
      ((TXManagerImpl) cache.getCacheTransactionManager()).unpauseTransaction(tx);

    }
  }
Create an index that can be used when executing queries.

@param indexName the name of this index, used for statistics collection
@param indexType the type of index
@param origIndexedExpression the expression to index on, a function dependent on region entries
       individually.
@param origFromClause expression that evaluates to the collection(s) that will be queried over,
       must contain one and only one region path.
@return the newly created Index
geode-core/src/main/java/org/apache/geode/cache/query/internal/index/IndexManager.java