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.metadata.utils;
  
  import static org.apache.asterix.common.api.IIdentifierMapper.Modifier.PLURAL;
  import static org.apache.asterix.common.utils.IdentifierUtil.dataset;
  
  import java.util.List;
  import java.util.Map;
  
  import org.apache.asterix.common.config.DatasetConfig.DatasetType;
  import org.apache.asterix.common.config.DatasetConfig.IndexType;
  import org.apache.asterix.common.context.AsterixVirtualBufferCacheProvider;
  import org.apache.asterix.common.context.IStorageComponentProvider;
  import org.apache.asterix.common.exceptions.AsterixException;
  import org.apache.asterix.common.exceptions.CompilationException;
  import org.apache.asterix.common.exceptions.ErrorCode;
  import org.apache.asterix.formats.nontagged.NullIntrospector;
  import org.apache.asterix.metadata.api.IResourceFactoryProvider;
  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.om.types.ARecordType;
  import org.apache.asterix.om.types.BuiltinType;
  import org.apache.asterix.om.types.IAType;
  import org.apache.asterix.om.utils.NonTaggedFormatUtil;
  import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
  import org.apache.hyracks.algebricks.common.utils.Pair;
  import org.apache.hyracks.algebricks.data.ITypeTraitProvider;
  import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
  import org.apache.hyracks.api.dataflow.value.ITypeTraits;
  import org.apache.hyracks.data.std.accessors.ShortBinaryComparatorFactory;
  import org.apache.hyracks.data.std.primitive.ShortPointable;
  import org.apache.hyracks.storage.am.common.api.IMetadataPageManagerFactory;
  import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallbackFactory;
  import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationSchedulerProvider;
  import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
  import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTrackerFactory;
  import org.apache.hyracks.storage.am.lsm.common.api.ILSMPageWriteCallbackFactory;
  import org.apache.hyracks.storage.am.lsm.invertedindex.dataflow.LSMInvertedIndexLocalResourceFactory;
  import org.apache.hyracks.storage.am.lsm.invertedindex.fulltext.IFullTextConfigEvaluatorFactory;
  import org.apache.hyracks.storage.am.lsm.invertedindex.tokenizers.IBinaryTokenizerFactory;
  import org.apache.hyracks.storage.common.IResourceFactory;
  import org.apache.hyracks.storage.common.IStorageManager;
  
  public class InvertedIndexResourceFactoryProvider implements IResourceFactoryProvider {
      public static final InvertedIndexResourceFactoryProvider INSTANCE = new InvertedIndexResourceFactoryProvider();
  
      private InvertedIndexResourceFactoryProvider() {
      }
  
      @Override
      public IResourceFactory getResourceFactory(MetadataProvider mdProvider, Dataset dataset, Index index,
              ARecordType recordType, ARecordType metaType, ILSMMergePolicyFactory mergePolicyFactory,
              Map<String, String> mergePolicyProperties, ITypeTraits[] filterTypeTraits,
              IBinaryComparatorFactory[] filterCmpFactories) throws AlgebricksException {
          // Get basic info
          List<List<String>> primaryKeys = dataset.getPrimaryKeys();
          Index.TextIndexDetails indexDetails = (Index.TextIndexDetails) index.getIndexDetails();
          List<List<String>> secondaryKeys = indexDetails.getKeyFieldNames();
          List<String> filterFieldName = DatasetUtil.getFilterField(dataset);
          int numPrimaryKeys = primaryKeys.size();
          int numSecondaryKeys = secondaryKeys.size();
          // Validate
          if (dataset.getDatasetType() != DatasetType.INTERNAL) {
              throw new CompilationException(ErrorCode.COMPILATION_INDEX_TYPE_NOT_SUPPORTED_FOR_DATASET_TYPE,
                      index.getIndexType().name(), dataset.getDatasetType());
          }
          if (numPrimaryKeys > 1) {
              throw new AsterixException(
                      "Cannot create inverted index on " + dataset(PLURAL) + " with composite primary key.");
          }
          if (numSecondaryKeys > 1) {
              throw new AsterixException("Cannot create composite inverted index on multiple fields.");
          }
          boolean isPartitioned = index.getIndexType() == IndexType.LENGTH_PARTITIONED_WORD_INVIX
                  || index.getIndexType() == IndexType.LENGTH_PARTITIONED_NGRAM_INVIX;
          int numTokenKeyPairFields = (!isPartitioned) ? 1 + numPrimaryKeys : 2 + numPrimaryKeys;
          int[] invertedIndexFields = null;
          int[] secondaryFilterFieldsForNonBulkLoadOps = null;
          int[] invertedIndexFieldsForNonBulkLoadOps = null;
          int[] secondaryFilterFields = null;
          if (filterFieldName != null) {
              invertedIndexFields = new int[numTokenKeyPairFields];
              for (int i = 0; i < invertedIndexFields.length; i++) {
                  invertedIndexFields[i] = i;
              }
              secondaryFilterFieldsForNonBulkLoadOps = new int[filterFieldName.size()];
              secondaryFilterFieldsForNonBulkLoadOps[0] = numSecondaryKeys + numPrimaryKeys;
              invertedIndexFieldsForNonBulkLoadOps = new int[numSecondaryKeys + numPrimaryKeys];
              for (int i = 0; i < invertedIndexFieldsForNonBulkLoadOps.length; i++) {
                  invertedIndexFieldsForNonBulkLoadOps[i] = i;
              }
              secondaryFilterFields = new int[filterFieldName.size()];
              secondaryFilterFields[0] = numTokenKeyPairFields - numPrimaryKeys + numPrimaryKeys;
          }
          IStorageComponentProvider storageComponentProvider = mdProvider.getStorageComponentProvider();
          IStorageManager storageManager = storageComponentProvider.getStorageManager();
          ILSMOperationTrackerFactory opTrackerFactory = dataset.getIndexOperationTrackerFactory(index);
          ILSMIOOperationCallbackFactory ioOpCallbackFactory = dataset.getIoOperationCallbackFactory(index);
          ILSMPageWriteCallbackFactory pageWriteCallbackFactory = dataset.getPageWriteCallbackFactory();
          IMetadataPageManagerFactory metadataPageManagerFactory =
                  storageComponentProvider.getMetadataPageManagerFactory();
          AsterixVirtualBufferCacheProvider vbcProvider = new AsterixVirtualBufferCacheProvider(dataset.getDatasetId());
          ILSMIOOperationSchedulerProvider ioSchedulerProvider =
                  storageComponentProvider.getIoOperationSchedulerProvider();
          double bloomFilterFalsePositiveRate = mdProvider.getStorageProperties().getBloomFilterFalsePositiveRate();
          ITypeTraits[] typeTraits = getInvListTypeTraits(mdProvider, dataset, recordType, metaType);
          IBinaryComparatorFactory[] cmpFactories =
                  getInvListComparatorFactories(mdProvider, dataset, recordType, metaType);
          ITypeTraits[] tokenTypeTraits = getTokenTypeTraits(dataset, index, recordType, metaType);
          IBinaryComparatorFactory[] tokenCmpFactories =
                  getTokenComparatorFactories(dataset, index, recordType, metaType);
          IBinaryTokenizerFactory tokenizerFactory = getTokenizerFactory(dataset, index, recordType, metaType);
          IFullTextConfigEvaluatorFactory fullTextConfigEvaluatorFactory =
                  FullTextUtil.fetchFilterAndCreateConfigEvaluator(mdProvider, index.getDatabaseName(),
                          index.getDataverseName(), indexDetails.getFullTextConfigName());
  
          ITypeTraitProvider typeTraitProvider = mdProvider.getDataFormat().getTypeTraitProvider();
          return new LSMInvertedIndexLocalResourceFactory(storageManager, typeTraits, cmpFactories, filterTypeTraits,
                  filterCmpFactories, secondaryFilterFields, opTrackerFactory, ioOpCallbackFactory,
                  pageWriteCallbackFactory, metadataPageManagerFactory, vbcProvider, ioSchedulerProvider,
                  mergePolicyFactory, mergePolicyProperties, true, tokenTypeTraits, tokenCmpFactories, tokenizerFactory,
                  fullTextConfigEvaluatorFactory, isPartitioned, invertedIndexFields,
                  secondaryFilterFieldsForNonBulkLoadOps, invertedIndexFieldsForNonBulkLoadOps,
                  bloomFilterFalsePositiveRate, typeTraitProvider.getTypeTrait(BuiltinType.ANULL),
                  NullIntrospector.INSTANCE);
      }
  
      // Returns an array of the type traits of the inverted list elements
      // It contains the primary key(s) type traits of the corresponding indexed rows,
      // and those primary keys are the elements in the inverted list.
      private static ITypeTraits[] getInvListTypeTraits(MetadataProvider metadataProvider, Dataset dataset,
              ARecordType recordType, ARecordType metaType) throws AlgebricksException {
          ITypeTraits[] primaryTypeTraits = dataset.getPrimaryTypeTraits(metadataProvider, recordType, metaType);
          ITypeTraits[] typeTraits = new ITypeTraits[primaryTypeTraits.length - 1];
          for (int i = 0; i < typeTraits.length; i++) {
              typeTraits[i] = primaryTypeTraits[i];
          }
          return typeTraits;
      }
  
      private static IBinaryComparatorFactory[] getInvListComparatorFactories(MetadataProvider metadataProvider,
              Dataset dataset, ARecordType recordType, ARecordType metaType) throws AlgebricksException {
          return dataset.getPrimaryComparatorFactories(metadataProvider, recordType, metaType);
      }
  
      private static ITypeTraits[] getTokenTypeTraits(Dataset dataset, Index index, ARecordType recordType,
              ARecordType metaType) throws AlgebricksException {
          int numPrimaryKeys = dataset.getPrimaryKeys().size();
          Index.TextIndexDetails indexDetails = (Index.TextIndexDetails) index.getIndexDetails();
          int numSecondaryKeys = indexDetails.getKeyFieldNames().size();
          IndexType indexType = index.getIndexType();
          // Sanity checks.
          if (numPrimaryKeys > 1) {
              throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_INDEX_FOR_DATASET_WITH_COMPOSITE_PRIMARY_INDEX,
                      indexType, DatasetUtil.getFullyQualifiedDisplayName(dataset));
          }
          if (numSecondaryKeys > 1) {
              throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_INDEX_NUM_OF_FIELD, numSecondaryKeys,
                      indexType, 1);
          }
          boolean isPartitioned = indexType == IndexType.LENGTH_PARTITIONED_WORD_INVIX
                  || indexType == IndexType.LENGTH_PARTITIONED_NGRAM_INVIX;
          ARecordType sourceType;
          List<Integer> keySourceIndicators = indexDetails.getKeyFieldSourceIndicators();
          if (keySourceIndicators == null || keySourceIndicators.get(0) == 0) {
              sourceType = recordType;
          } else {
              sourceType = metaType;
          }
          Pair<IAType, Boolean> keyTypePair = Index.getNonNullableOpenFieldType(index,
                  indexDetails.getKeyFieldTypes().get(0), indexDetails.getKeyFieldNames().get(0), sourceType);
          IAType secondaryKeyType = keyTypePair.first;
          int numTokenFields = (!isPartitioned) ? numSecondaryKeys : numSecondaryKeys + 1;
          ITypeTraits[] tokenTypeTraits = new ITypeTraits[numTokenFields];
          tokenTypeTraits[0] = NonTaggedFormatUtil.getTokenTypeTrait(secondaryKeyType);
          if (isPartitioned) {
              // The partitioning field is hardcoded to be a short *without* an Asterix type tag.
              tokenTypeTraits[1] = ShortPointable.TYPE_TRAITS;
          }
          return tokenTypeTraits;
      }
  
      private static IBinaryComparatorFactory[] getTokenComparatorFactories(Dataset dataset, Index index,
              ARecordType recordType, ARecordType metaType) throws AlgebricksException {
          int numPrimaryKeys = dataset.getPrimaryKeys().size();
          Index.TextIndexDetails indexDetails = (Index.TextIndexDetails) index.getIndexDetails();
          int numSecondaryKeys = indexDetails.getKeyFieldNames().size();
          IndexType indexType = index.getIndexType();
          // Sanity checks.
          if (numPrimaryKeys > 1) {
              throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_INDEX_FOR_DATASET_WITH_COMPOSITE_PRIMARY_INDEX,
                      indexType, DatasetUtil.getFullyQualifiedDisplayName(dataset));
          }
          if (numSecondaryKeys > 1) {
              throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_INDEX_NUM_OF_FIELD, numSecondaryKeys,
                      indexType, 1);
          }
          boolean isPartitioned = indexType == IndexType.LENGTH_PARTITIONED_WORD_INVIX
                  || indexType == IndexType.LENGTH_PARTITIONED_NGRAM_INVIX;
          List<Integer> keySourceIndicators = indexDetails.getKeyFieldSourceIndicators();
          ARecordType sourceType;
          if (keySourceIndicators == null || keySourceIndicators.get(0) == 0) {
              sourceType = recordType;
          } else {
              sourceType = metaType;
          }
          Pair<IAType, Boolean> keyTypePair = Index.getNonNullableOpenFieldType(index,
                  indexDetails.getKeyFieldTypes().get(0), indexDetails.getKeyFieldNames().get(0), sourceType);
          IAType secondaryKeyType = keyTypePair.first;
          // Comparators and type traits for tokens.
          int numTokenFields = (!isPartitioned) ? numSecondaryKeys : numSecondaryKeys + 1;
          IBinaryComparatorFactory[] tokenComparatorFactories = new IBinaryComparatorFactory[numTokenFields];
          tokenComparatorFactories[0] = NonTaggedFormatUtil.getTokenBinaryComparatorFactory(secondaryKeyType);
          if (isPartitioned) {
              // The partitioning field is hardcoded to be a short *without* an Asterix type tag.
              tokenComparatorFactories[1] = ShortBinaryComparatorFactory.INSTANCE;
          }
          return tokenComparatorFactories;
      }
  
      private static IBinaryTokenizerFactory getTokenizerFactory(Dataset dataset, Index index, ARecordType recordType,
              ARecordType metaType) throws AlgebricksException {
          int numPrimaryKeys = dataset.getPrimaryKeys().size();
          Index.TextIndexDetails indexDetails = (Index.TextIndexDetails) index.getIndexDetails();
          int numSecondaryKeys = indexDetails.getKeyFieldNames().size();
          IndexType indexType = index.getIndexType();
          // Sanity checks.
          if (numPrimaryKeys > 1) {
              throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_INDEX_FOR_DATASET_WITH_COMPOSITE_PRIMARY_INDEX,
                      indexType, DatasetUtil.getFullyQualifiedDisplayName(dataset));
          }
          if (numSecondaryKeys > 1) {
              throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_INDEX_NUM_OF_FIELD, numSecondaryKeys,
                      indexType, 1);
          }
          ARecordType sourceType;
          List<Integer> keySourceIndicators = indexDetails.getKeyFieldSourceIndicators();
          if (keySourceIndicators == null || keySourceIndicators.get(0) == 0) {
              sourceType = recordType;
          } else {
              sourceType = metaType;
          }
          Pair<IAType, Boolean> keyTypePair = Index.getNonNullableOpenFieldType(index,
                  indexDetails.getKeyFieldTypes().get(0), indexDetails.getKeyFieldNames().get(0), sourceType);
          IAType secondaryKeyType = keyTypePair.first;
          // Set tokenizer factory.
          // TODO: We might want to expose the hashing option at the AQL level,
          // and add the choice to the index metadata.
          return NonTaggedFormatUtil.getBinaryTokenizerFactory(secondaryKeyType.getTypeTag(), indexType,
                  indexDetails.getGramLength());
      }
  }
  