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.utils.IdentifierUtil.dataset;
  import static org.apache.asterix.om.utils.ProjectionFiltrationTypeUtil.EMPTY_TYPE;
  
  import java.io.DataOutput;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.UUID;
  import java.util.stream.Collectors;
  
  import org.apache.asterix.builders.IARecordBuilder;
  import org.apache.asterix.builders.RecordBuilder;
  import org.apache.asterix.common.cluster.PartitioningProperties;
  import org.apache.asterix.common.config.DatasetConfig;
  import org.apache.asterix.common.config.DatasetConfig.DatasetType;
  import org.apache.asterix.common.context.CorrelatedPrefixMergePolicyFactory;
  import org.apache.asterix.common.context.DatasetLifecycleManager;
  import org.apache.asterix.common.context.DatasetResource;
  import org.apache.asterix.common.context.IStorageComponentProvider;
  import org.apache.asterix.common.context.ITransactionSubsystemProvider;
  import org.apache.asterix.common.context.TransactionSubsystemProvider;
  import org.apache.asterix.common.dataflow.DatasetLocalResource;
  import org.apache.asterix.common.dataflow.ICcApplicationContext;
  import org.apache.asterix.common.exceptions.ACIDException;
  import org.apache.asterix.common.exceptions.AsterixException;
  import org.apache.asterix.common.exceptions.ErrorCode;
  import org.apache.asterix.common.metadata.DataverseName;
  import org.apache.asterix.common.metadata.MetadataConstants;
  import org.apache.asterix.common.metadata.MetadataUtil;
  import org.apache.asterix.common.transactions.IRecoveryManager;
  import org.apache.asterix.common.utils.JobUtils;
  import org.apache.asterix.external.util.ExternalDataUtils;
  import org.apache.asterix.formats.base.IDataFormat;
  import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
  import org.apache.asterix.formats.nontagged.TypeTraitProvider;
  import org.apache.asterix.metadata.IDatasetDetails;
  import org.apache.asterix.metadata.MetadataManager;
  import org.apache.asterix.metadata.MetadataTransactionContext;
  import org.apache.asterix.metadata.declared.MetadataProvider;
  import org.apache.asterix.metadata.entities.CompactionPolicy;
  import org.apache.asterix.metadata.entities.Dataset;
  import org.apache.asterix.metadata.entities.Dataverse;
  import org.apache.asterix.metadata.entities.ExternalDatasetDetails;
  import org.apache.asterix.metadata.entities.Index;
  import org.apache.asterix.metadata.entities.InternalDatasetDetails;
  import org.apache.asterix.metadata.entities.NodeGroup;
  import org.apache.asterix.om.base.AMutableString;
  import org.apache.asterix.om.base.AString;
  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.types.visitor.SimpleStringBuilderForIATypeVisitor;
  import org.apache.asterix.om.utils.ProjectionFiltrationTypeUtil;
  import org.apache.asterix.runtime.operators.LSMPrimaryUpsertOperatorDescriptor;
  import org.apache.asterix.runtime.utils.RuntimeUtils;
  import org.apache.asterix.transaction.management.opcallbacks.PrimaryIndexInstantSearchOperationCallbackFactory;
  import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
  import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
  import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraintHelper;
  import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
  import org.apache.hyracks.algebricks.common.utils.Pair;
  import org.apache.hyracks.algebricks.data.IBinaryComparatorFactoryProvider;
  import org.apache.hyracks.api.client.IHyracksClientConnection;
  import org.apache.hyracks.api.dataflow.IOperatorDescriptor;
  import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
  import org.apache.hyracks.api.dataflow.value.IBinaryHashFunctionFactory;
  import org.apache.hyracks.api.dataflow.value.IMissingWriterFactory;
  import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
  import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
  import org.apache.hyracks.api.dataflow.value.ITypeTraits;
  import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
  import org.apache.hyracks.api.exceptions.HyracksDataException;
  import org.apache.hyracks.api.io.FileSplit;
  import org.apache.hyracks.api.job.JobSpecification;
  import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
  import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
  import org.apache.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
  import org.apache.hyracks.dataflow.common.data.partition.FieldHashPartitionerFactory;
  import org.apache.hyracks.dataflow.std.file.IFileSplitProvider;
  import org.apache.hyracks.dataflow.std.misc.ConstantTupleSourceOperatorDescriptor;
  import org.apache.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
  import org.apache.hyracks.storage.am.common.api.IModificationOperationCallbackFactory;
  import org.apache.hyracks.storage.am.common.api.ISearchOperationCallbackFactory;
  import org.apache.hyracks.storage.am.common.build.IndexBuilderFactory;
  import org.apache.hyracks.storage.am.common.dataflow.IIndexDataflowHelperFactory;
  import org.apache.hyracks.storage.am.common.dataflow.IndexCreateOperatorDescriptor;
  import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelper;
  import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory;
  import org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorDescriptor;
  import org.apache.hyracks.storage.am.common.impls.DefaultTupleProjectorFactory;
  import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
  import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
  import org.apache.hyracks.storage.am.lsm.common.api.ILSMTupleFilterCallbackFactory;
  import org.apache.hyracks.storage.am.lsm.common.dataflow.LSMTreeIndexCompactOperatorDescriptor;
  import org.apache.hyracks.storage.common.IResourceFactory;
  import org.apache.hyracks.storage.common.IStorageManager;
  import org.apache.hyracks.storage.common.LocalResource;
  import org.apache.hyracks.storage.common.projection.ITupleProjectorFactory;
  import org.apache.hyracks.util.LogRedactionUtil;
  import org.apache.logging.log4j.LogManager;
  import org.apache.logging.log4j.Logger;
  
  public class DatasetUtil {
      private static final Logger LOGGER = LogManager.getLogger();
      /*
       * Dataset related operations
       */
      public static final byte OP_UPSERT = 0x03;
  
      private DatasetUtil() {
      }
  
      public static Integer getFilterSourceIndicator(Dataset dataset) {
          return ((InternalDatasetDetails) dataset.getDatasetDetails()).getFilterSourceIndicator();
      }
  
      public static List<String> getFilterField(Dataset dataset) {
          return ((InternalDatasetDetails) dataset.getDatasetDetails()).getFilterField();
      }
  
      public static IBinaryComparatorFactory[] computeFilterBinaryComparatorFactories(Dataset dataset,
              ARecordType recordType, ARecordType metaType, IBinaryComparatorFactoryProvider comparatorFactoryProvider)
              throws AlgebricksException {
          if (dataset.getDatasetType() == DatasetType.EXTERNAL) {
              return null;
          }
          Integer filterFieldSourceIndicator = getFilterSourceIndicator(dataset);
          List<String> filterField = getFilterField(dataset);
          if (filterField == null) {
              return null;
          }
          IBinaryComparatorFactory[] bcfs = new IBinaryComparatorFactory[1];
          ARecordType itemType = filterFieldSourceIndicator == 0 ? recordType : metaType;
          IAType type = itemType.getSubFieldType(filterField);
          bcfs[0] = comparatorFactoryProvider.getBinaryComparatorFactory(type, true);
          return bcfs;
      }
  
      public static ITypeTraits[] computeFilterTypeTraits(Dataset dataset, ARecordType recordType, ARecordType metaType)
              throws AlgebricksException {
          if (dataset.getDatasetType() == DatasetType.EXTERNAL) {
              return null;
          }
          Integer filterFieldSourceIndicator = getFilterSourceIndicator(dataset);
          List<String> filterField = getFilterField(dataset);
          if (filterField == null) {
              return null;
          }
          ARecordType itemType = filterFieldSourceIndicator == 0 ? recordType : metaType;
          ITypeTraits[] typeTraits = new ITypeTraits[1];
          IAType type = itemType.getSubFieldType(filterField);
          typeTraits[0] = TypeTraitProvider.INSTANCE.getTypeTrait(type);
          return typeTraits;
      }
  
      public static int[] createFilterFields(Dataset dataset) {
          if (dataset.getDatasetType() == DatasetType.EXTERNAL) {
              return null;
          }
  
          List<String> filterField = getFilterField(dataset);
          if (filterField == null) {
              return null;
          }
          List<List<String>> partitioningKeys = dataset.getPrimaryKeys();
          int numKeys = partitioningKeys.size();
  
          int[] filterFields = new int[1];
          int valueFields = dataset.hasMetaPart() ? 2 : 1;
          filterFields[0] = numKeys + valueFields;
          return filterFields;
      }
  
      public static int[] createBTreeFieldsWhenThereisAFilter(Dataset dataset) {
          if (dataset.getDatasetType() == DatasetType.EXTERNAL) {
              return null;
          }
  
          List<String> filterField = getFilterField(dataset);
          if (filterField == null) {
              return null;
          }
  
          List<List<String>> partitioningKeys = dataset.getPrimaryKeys();
          int valueFields = dataset.hasMetaPart() ? 2 : 1;
          int[] btreeFields = new int[partitioningKeys.size() + valueFields];
          for (int i = 0; i < btreeFields.length; ++i) {
              btreeFields[i] = i;
          }
          return btreeFields;
      }
  
      /**
       * Returns the primary key source indicators of the {@code dataset} or {@code null} if the dataset does not have
       * primary key source indicators (e.g. external datasets)
       */
      public static List<Integer> getKeySourceIndicators(Dataset dataset) {
          IDatasetDetails datasetDetails = dataset.getDatasetDetails();
          if (datasetDetails.getDatasetType() == DatasetType.INTERNAL) {
              return ((InternalDatasetDetails) datasetDetails).getKeySourceIndicator();
          }
          return null;
      }
  
      public static int getPositionOfPartitioningKeyField(Dataset dataset, List<String> fieldExpr,
              boolean fieldFromMeta) {
          List<Integer> keySourceIndicator = null;
          IDatasetDetails datasetDetails = dataset.getDatasetDetails();
          if (datasetDetails.getDatasetType() == DatasetType.INTERNAL) {
              keySourceIndicator = ((InternalDatasetDetails) datasetDetails).getKeySourceIndicator();
          }
          List<List<String>> partitioningKeys = dataset.getPrimaryKeys();
          for (int i = 0; i < partitioningKeys.size(); i++) {
              List<String> partitioningKey = partitioningKeys.get(i);
              if (partitioningKey.equals(fieldExpr) && keySourceMatches(keySourceIndicator, i, fieldFromMeta)) {
                  return i;
              }
          }
          return -1;
      }
  
      /**
       * Once it's determined that a field name is a key (by just comparing the names), this method checks whether the
       * field is actually a key by making sure the field is coming from the right record (data record or meta record),
       * e.g. if the field name happens to be equal to the key name but the field is coming from the data record while
       * the key is coming from the meta record.
       *
       * @param keySourceIndicator indicates where the key is coming from, 1 from meta record, 0 from data record
       * @param keyIndex           the key index we're checking the field against
       * @param fieldFromMeta      whether the field is coming from the meta record or the data record
       * @return true if the key source matches the field source. Otherwise, false.
       */
      private static boolean keySourceMatches(List<Integer> keySourceIndicator, int keyIndex, boolean fieldFromMeta) {
          if (keySourceIndicator != null) {
              return (fieldFromMeta && keySourceIndicator.get(keyIndex) == 1)
                      || (!fieldFromMeta && keySourceIndicator.get(keyIndex) == 0);
          }
          return true;
      }
  
      public static Pair<ILSMMergePolicyFactory, Map<String, String>> getMergePolicyFactory(Dataset dataset,
              MetadataTransactionContext mdTxnCtx) throws AlgebricksException {
          String policyName = dataset.getCompactionPolicy();
          CompactionPolicy compactionPolicy = MetadataManager.INSTANCE.getCompactionPolicy(mdTxnCtx,
                  MetadataConstants.SYSTEM_DATABASE, MetadataConstants.METADATA_DATAVERSE_NAME, policyName);
          String compactionPolicyFactoryClassName = compactionPolicy.getClassName();
          ILSMMergePolicyFactory mergePolicyFactory;
          Map<String, String> properties = dataset.getCompactionPolicyProperties();
          try {
              mergePolicyFactory = (ILSMMergePolicyFactory) Class.forName(compactionPolicyFactoryClassName).newInstance();
              if (mergePolicyFactory.getName().compareTo(CorrelatedPrefixMergePolicyFactory.NAME) == 0) {
                  properties.put(CorrelatedPrefixMergePolicyFactory.KEY_DATASET_ID,
                          Integer.toString(dataset.getDatasetId()));
              }
          } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {
              throw new AlgebricksException(e);
          }
          return new Pair<>(mergePolicyFactory, properties);
      }
  
      public static void writePropertyTypeRecord(String name, String value, DataOutput out, ARecordType recordType)
              throws HyracksDataException {
          IARecordBuilder propertyRecordBuilder = new RecordBuilder();
          ArrayBackedValueStorage fieldValue = new ArrayBackedValueStorage();
          propertyRecordBuilder.reset(recordType);
          AMutableString aString = new AMutableString("");
          ISerializerDeserializer<AString> stringSerde =
                  SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.ASTRING);
  
          // write field 0
          fieldValue.reset();
          aString.setValue(name);
          stringSerde.serialize(aString, fieldValue.getDataOutput());
          propertyRecordBuilder.addField(0, fieldValue);
  
          // write field 1
          fieldValue.reset();
          aString.setValue(value);
          stringSerde.serialize(aString, fieldValue.getDataOutput());
          propertyRecordBuilder.addField(1, fieldValue);
  
          propertyRecordBuilder.write(out, true);
      }
  
      public static ARecordType getMetaType(MetadataProvider metadataProvider, Dataset dataset)
              throws AlgebricksException {
          if (dataset.hasMetaPart()) {
              return (ARecordType) metadataProvider.findType(dataset.getMetaItemTypeDatabaseName(),
                      dataset.getMetaItemTypeDataverseName(), dataset.getMetaItemTypeName());
          }
          return null;
      }
  
      public static JobSpecification dropDatasetJobSpec(Dataset dataset, MetadataProvider metadataProvider)
              throws AlgebricksException, ACIDException {
          return dropDatasetJobSpec(dataset, metadataProvider, Collections.emptySet());
      }
  
      public static JobSpecification dropDatasetJobSpec(Dataset dataset, MetadataProvider metadataProvider,
              Set<IndexDropOperatorDescriptor.DropOption> options) throws AlgebricksException, ACIDException {
          LOGGER.info("DROP DATASET: " + dataset);
          if (dataset.getDatasetType() == DatasetType.EXTERNAL) {
              return RuntimeUtils.createJobSpecification(metadataProvider.getApplicationContext());
          }
          JobSpecification specPrimary = RuntimeUtils.createJobSpecification(metadataProvider.getApplicationContext());
          PartitioningProperties partitioningProperties = metadataProvider.getPartitioningProperties(dataset);
          IIndexDataflowHelperFactory indexHelperFactory =
                  new IndexDataflowHelperFactory(metadataProvider.getStorageComponentProvider().getStorageManager(),
                          partitioningProperties.getSplitsProvider());
          IndexDropOperatorDescriptor primaryBtreeDrop = new IndexDropOperatorDescriptor(specPrimary, indexHelperFactory,
                  options, partitioningProperties.getComputeStorageMap());
          AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(specPrimary, primaryBtreeDrop,
                  partitioningProperties.getConstraints());
          specPrimary.addRoot(primaryBtreeDrop);
          return specPrimary;
      }
  
      public static JobSpecification createDatasetJobSpec(Dataset dataset, MetadataProvider metadataProvider)
              throws AlgebricksException {
          Index index = IndexUtil.getPrimaryIndex(dataset);
          ARecordType itemType = (ARecordType) metadataProvider.findType(dataset);
          // get meta item type
          ARecordType metaItemType = null;
          if (dataset.hasMetaPart()) {
              metaItemType = (ARecordType) metadataProvider.findMetaType(dataset);
          }
          itemType = (ARecordType) metadataProvider.findTypeForDatasetWithoutType(itemType, dataset);
  
          JobSpecification spec = RuntimeUtils.createJobSpecification(metadataProvider.getApplicationContext());
          PartitioningProperties partitioningProperties = metadataProvider.getPartitioningProperties(dataset);
          IFileSplitProvider splitsProvider = partitioningProperties.getSplitsProvider();
          FileSplit[] fs = splitsProvider.getFileSplits();
          StringBuilder sb = new StringBuilder();
          for (FileSplit f : fs) {
              sb.append(f).append(" ");
          }
          LOGGER.info("CREATING File Splits: {}", sb);
          Pair<ILSMMergePolicyFactory, Map<String, String>> compactionInfo =
                  DatasetUtil.getMergePolicyFactory(dataset, metadataProvider.getMetadataTxnContext());
          // prepare a LocalResourceMetadata which will be stored in NC's local resource
          // repository
          int[][] computeStorageMap = partitioningProperties.getComputeStorageMap();
          IndexBuilderFactory[][] indexBuilderFactories = getIndexBuilderFactories(dataset, metadataProvider, index,
                  itemType, metaItemType, splitsProvider, compactionInfo.first, compactionInfo.second, computeStorageMap);
          IndexCreateOperatorDescriptor indexCreateOp =
                  new IndexCreateOperatorDescriptor(spec, indexBuilderFactories, computeStorageMap);
          AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, indexCreateOp,
                  partitioningProperties.getConstraints());
          spec.addRoot(indexCreateOp);
          return spec;
      }
  
      public static IndexBuilderFactory[][] getIndexBuilderFactories(Dataset dataset, MetadataProvider metadataProvider,
              Index index, ARecordType itemType, ARecordType metaItemType, IFileSplitProvider fileSplitProvider,
              ILSMMergePolicyFactory mergePolicyFactory, Map<String, String> mergePolicyProperties,
              int[][] computeStorageMap) throws AlgebricksException {
          IndexBuilderFactory[][] indexBuilderFactories = new IndexBuilderFactory[computeStorageMap.length][];
          for (int i = 0; i < computeStorageMap.length; i++) {
              int len = computeStorageMap[i].length;
              indexBuilderFactories[i] = new IndexBuilderFactory[len];
              for (int k = 0; k < len; k++) {
                  IResourceFactory resourceFactory = dataset.getResourceFactory(metadataProvider, index, itemType,
                          metaItemType, mergePolicyFactory, mergePolicyProperties);
                  IndexBuilderFactory indexBuilderFactory =
                          new IndexBuilderFactory(metadataProvider.getStorageComponentProvider().getStorageManager(),
                                  fileSplitProvider, resourceFactory, true);
                  indexBuilderFactories[i][k] = indexBuilderFactory;
              }
          }
          return indexBuilderFactories;
      }
  
      public static JobSpecification compactDatasetJobSpec(Dataverse dataverse, String datasetName,
              MetadataProvider metadataProvider) throws AlgebricksException {
          DataverseName dataverseName = dataverse.getDataverseName();
          Dataset dataset = metadataProvider.findDataset(dataverse.getDatabaseName(), dataverseName, datasetName);
          if (dataset == null) {
              throw new AsterixException(ErrorCode.UNKNOWN_DATASET_IN_DATAVERSE, datasetName, MetadataUtil
                      .dataverseName(dataverse.getDatabaseName(), dataverseName, metadataProvider.isUsingDatabase()));
          }
          JobSpecification spec = RuntimeUtils.createJobSpecification(metadataProvider.getApplicationContext());
          PartitioningProperties partitioningProperties = metadataProvider.getPartitioningProperties(dataset);
          IIndexDataflowHelperFactory indexHelperFactory =
                  new IndexDataflowHelperFactory(metadataProvider.getStorageComponentProvider().getStorageManager(),
                          partitioningProperties.getSplitsProvider());
          LSMTreeIndexCompactOperatorDescriptor compactOp = new LSMTreeIndexCompactOperatorDescriptor(spec,
                  indexHelperFactory, partitioningProperties.getComputeStorageMap());
          AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, compactOp,
                  partitioningProperties.getConstraints());
          spec.addRoot(compactOp);
          return spec;
      }
  
      /**
       * Creates a primary index scan operator for a given dataset.
       *
       * @param spec,             the job specification.
       * @param metadataProvider, the metadata provider.
       * @param dataset,          the dataset to scan.
       * @return a primary index scan operator.
       * @throws AlgebricksException
       */
      public static IOperatorDescriptor createPrimaryIndexScanOp(JobSpecification spec, MetadataProvider metadataProvider,
              Dataset dataset) throws AlgebricksException {
          return createPrimaryIndexScanOp(spec, metadataProvider, dataset, DefaultTupleProjectorFactory.INSTANCE);
      }
  
      public static IOperatorDescriptor createPrimaryIndexScanOp(JobSpecification spec, MetadataProvider metadataProvider,
              Dataset dataset, ITupleProjectorFactory projectorFactory) throws AlgebricksException {
          PartitioningProperties partitioningProperties = metadataProvider.getPartitioningProperties(dataset);
          IFileSplitProvider primaryFileSplitProvider = partitioningProperties.getSplitsProvider();
          AlgebricksPartitionConstraint primaryPartitionConstraint = partitioningProperties.getConstraints();
          // -Infinity
          int[] lowKeyFields = null;
          // +Infinity
          int[] highKeyFields = null;
          ITransactionSubsystemProvider txnSubsystemProvider = TransactionSubsystemProvider.INSTANCE;
          ISearchOperationCallbackFactory searchCallbackFactory = new PrimaryIndexInstantSearchOperationCallbackFactory(
                  dataset.getDatasetId(), dataset.getPrimaryBloomFilterFields(), txnSubsystemProvider,
                  IRecoveryManager.ResourceType.LSM_BTREE);
          IndexDataflowHelperFactory indexHelperFactory = new IndexDataflowHelperFactory(
                  metadataProvider.getStorageComponentProvider().getStorageManager(), primaryFileSplitProvider);
          BTreeSearchOperatorDescriptor primarySearchOp = new BTreeSearchOperatorDescriptor(spec,
                  dataset.getPrimaryRecordDescriptor(metadataProvider), lowKeyFields, highKeyFields, true, true,
                  indexHelperFactory, false, false, null, searchCallbackFactory, null, null, false, null, null, -1, false,
                  null, null, projectorFactory, null, partitioningProperties.getComputeStorageMap());
          AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, primarySearchOp,
                  primaryPartitionConstraint);
          return primarySearchOp;
      }
  
      /**
       * Creates a primary index upsert operator for a given dataset.
       *
       * @param spec,                 the job specification.
       * @param metadataProvider,     the metadata provider.
       * @param dataset,              the dataset to upsert.
       * @param inputRecordDesc,the   record descriptor for an input tuple.
       * @param fieldPermutation,     the field permutation according to the input.
       * @param missingWriterFactory, the factory for customizing missing value serialization.
       * @return a primary index scan operator and its location constraints.
       * @throws AlgebricksException
       */
      public static Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> createPrimaryIndexUpsertOp(
              JobSpecification spec, MetadataProvider metadataProvider, Dataset dataset, RecordDescriptor inputRecordDesc,
              int[] fieldPermutation, IMissingWriterFactory missingWriterFactory) throws AlgebricksException {
          int numKeys = dataset.getPrimaryKeys().size();
          int numFilterFields = DatasetUtil.getFilterField(dataset) == null ? 0 : 1;
          ARecordType itemType = (ARecordType) metadataProvider.findType(dataset);
          ARecordType metaItemType = (ARecordType) metadataProvider.findMetaType(dataset);
          itemType = (ARecordType) metadataProvider.findTypeForDatasetWithoutType(itemType, dataset);
  
          Index primaryIndex = metadataProvider.getIndex(dataset.getDatabaseName(), dataset.getDataverseName(),
                  dataset.getDatasetName(), dataset.getDatasetName());
          PartitioningProperties partitioningProperties = metadataProvider.getPartitioningProperties(dataset);
  
          // prepare callback
          int[] primaryKeyFields = new int[numKeys];
          int[] pkFields = new int[numKeys];
          for (int i = 0; i < numKeys; i++) {
              primaryKeyFields[i] = i;
              pkFields[i] = fieldPermutation[i];
          }
          boolean hasSecondaries = metadataProvider
                  .getDatasetIndexes(dataset.getDatabaseName(), dataset.getDataverseName(), dataset.getDatasetName())
                  .size() > 1;
          IStorageComponentProvider storageComponentProvider = metadataProvider.getStorageComponentProvider();
          IModificationOperationCallbackFactory modificationCallbackFactory = dataset.getModificationCallbackFactory(
                  storageComponentProvider, primaryIndex, IndexOperation.UPSERT, primaryKeyFields);
          ISearchOperationCallbackFactory searchCallbackFactory = dataset.getSearchCallbackFactory(
                  storageComponentProvider, primaryIndex, IndexOperation.UPSERT, primaryKeyFields);
          IIndexDataflowHelperFactory idfh = new IndexDataflowHelperFactory(storageComponentProvider.getStorageManager(),
                  partitioningProperties.getSplitsProvider());
          LSMPrimaryUpsertOperatorDescriptor op;
          ITypeTraits[] outputTypeTraits = new ITypeTraits[inputRecordDesc.getFieldCount() + 1
                  + (dataset.hasMetaPart() ? 2 : 1) + numFilterFields];
          ISerializerDeserializer<?>[] outputSerDes = new ISerializerDeserializer[inputRecordDesc.getFieldCount() + 1
                  + (dataset.hasMetaPart() ? 2 : 1) + numFilterFields];
          IDataFormat dataFormat = metadataProvider.getDataFormat();
  
          int f = 0;
          // add the upsert operation var
          outputSerDes[f] = dataFormat.getSerdeProvider().getSerializerDeserializer(BuiltinType.AINT8);
          outputTypeTraits[f] = dataFormat.getTypeTraitProvider().getTypeTrait(BuiltinType.AINT8);
          f++;
          // add the previous record
          outputSerDes[f] = dataFormat.getSerdeProvider().getSerializerDeserializer(itemType);
          outputTypeTraits[f] = dataFormat.getTypeTraitProvider().getTypeTrait(itemType);
          f++;
          // add the previous meta second
          if (dataset.hasMetaPart()) {
              outputSerDes[f] = dataFormat.getSerdeProvider().getSerializerDeserializer(metaItemType);
              outputTypeTraits[f] = dataFormat.getTypeTraitProvider().getTypeTrait(metaItemType);
              f++;
          }
          // add the previous filter third
          int fieldIdx = -1;
          Integer filterSourceIndicator = null;
          ARecordType filterItemType = null;
          if (numFilterFields > 0) {
              filterSourceIndicator = DatasetUtil.getFilterSourceIndicator(dataset);
              String filterField = DatasetUtil.getFilterField(dataset).get(0);
              filterItemType = filterSourceIndicator == 0 ? itemType : metaItemType;
              String[] fieldNames = filterItemType.getFieldNames();
              int i = 0;
              for (; i < fieldNames.length; i++) {
                  if (fieldNames[i].equals(filterField)) {
                      break;
                  }
              }
              fieldIdx = i;
              outputTypeTraits[f] =
                      dataFormat.getTypeTraitProvider().getTypeTrait(filterItemType.getFieldTypes()[fieldIdx]);
              outputSerDes[f] =
                      dataFormat.getSerdeProvider().getSerializerDeserializer(filterItemType.getFieldTypes()[fieldIdx]);
              f++;
          }
          for (int j = 0; j < inputRecordDesc.getFieldCount(); j++) {
              outputTypeTraits[j + f] = inputRecordDesc.getTypeTraits()[j];
              outputSerDes[j + f] = inputRecordDesc.getFields()[j];
          }
          RecordDescriptor outputRecordDesc = new RecordDescriptor(outputSerDes, outputTypeTraits);
  
          ILSMTupleFilterCallbackFactory tupleFilterCallbackFactory = dataset.getTupleFilterCallbackFactory();
  
          // This allows to project only the indexed fields instead of the entirety of the record
          ARecordType requestedType = getPrevRecordType(metadataProvider, dataset, itemType);
          ITupleProjectorFactory projectorFactory = IndexUtil.createUpsertTupleProjectorFactory(
                  dataset.getDatasetFormatInfo(), requestedType, itemType, metaItemType, numKeys);
          IBinaryHashFunctionFactory[] pkHashFunFactories = dataset.getPrimaryHashFunctionFactories(metadataProvider);
          ITuplePartitionerFactory tuplePartitionerFactory = new FieldHashPartitionerFactory(pkFields, pkHashFunFactories,
                  partitioningProperties.getNumberOfPartitions());
          op = new LSMPrimaryUpsertOperatorDescriptor(spec, outputRecordDesc, fieldPermutation, idfh,
                  missingWriterFactory, modificationCallbackFactory, searchCallbackFactory,
                  dataset.getFrameOpCallbackFactory(metadataProvider), numKeys, filterSourceIndicator, filterItemType,
                  fieldIdx, hasSecondaries, projectorFactory, tuplePartitionerFactory,
                  partitioningProperties.getComputeStorageMap(), tupleFilterCallbackFactory);
          return new Pair<>(op, partitioningProperties.getConstraints());
      }
  
      /**
       * Returns a type that contains indexed fields for columnar datasets.
       * The type is used retrieve the previous record with only the indexed fields -- minimizing the
       * I/O cost for point lookups.
       *
       * @param metadataProvider metadata provider
       * @param dataset          the dataset to upsert to
       * @param itemType         dataset type
       * @return a type with the requested fields
       */
      private static ARecordType getPrevRecordType(MetadataProvider metadataProvider, Dataset dataset,
              ARecordType itemType) throws AlgebricksException {
          if (dataset.getDatasetFormatInfo().getFormat() == DatasetConfig.DatasetFormat.ROW) {
              return itemType;
          }
  
          // Column
          List<Index> secondaryIndexes = metadataProvider.getDatasetIndexes(dataset.getDatabaseName(),
                  dataset.getDataverseName(), dataset.getDatasetName());
          List<ARecordType> indexPaths = new ArrayList<>();
  
          for (Index index : secondaryIndexes) {
              if (!index.isSecondaryIndex() || index.isPrimaryKeyIndex() || index.isSampleIndex()) {
                  continue;
              }
  
              if (index.getIndexType() == DatasetConfig.IndexType.BTREE) {
                  Index.ValueIndexDetails indexDetails = (Index.ValueIndexDetails) index.getIndexDetails();
                  indexPaths.add(indexDetails.getIndexExpectedType());
              } else if (index.getIndexType() == DatasetConfig.IndexType.ARRAY) {
                  Index.ArrayIndexDetails indexDetails = (Index.ArrayIndexDetails) index.getIndexDetails();
                  indexPaths.add(indexDetails.getIndexExpectedType());
              }
          }
  
          ARecordType result = indexPaths.isEmpty() ? EMPTY_TYPE : ProjectionFiltrationTypeUtil.merge(indexPaths);
  
          if (LOGGER.isInfoEnabled() && result != EMPTY_TYPE) {
              SimpleStringBuilderForIATypeVisitor schemaPrinter = new SimpleStringBuilderForIATypeVisitor();
              StringBuilder builder = new StringBuilder();
              result.accept(schemaPrinter, builder);
              LOGGER.info("Upsert previous tuple schema: {}", LogRedactionUtil.userData(builder.toString()));
          }
  
          return result;
      }
  
      /**
       * Creates a dummy key provider operator for the primary index scan.
       *
       * @param spec,             the job specification.
       * @param dataset,          the dataset to scan.
       * @param metadataProvider, the metadata provider.
       * @return a dummy key provider operator.
       * @throws AlgebricksException
       */
      public static IOperatorDescriptor createDummyKeyProviderOp(JobSpecification spec, Dataset dataset,
              MetadataProvider metadataProvider) throws AlgebricksException {
          PartitioningProperties partitioningProperties = metadataProvider.getPartitioningProperties(dataset);
          AlgebricksPartitionConstraint primaryPartitionConstraint = partitioningProperties.getConstraints();
          IOperatorDescriptor dummyKeyProviderOp = createDummyKeyProviderOp(spec);
          AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, dummyKeyProviderOp,
                  primaryPartitionConstraint);
          return dummyKeyProviderOp;
      }
  
      public static IOperatorDescriptor createCorrelatedDummyKeyProviderOp(JobSpecification spec,
              AlgebricksPartitionConstraint apc) throws AlgebricksException {
          IOperatorDescriptor dummyKeyProviderOp = createDummyKeyProviderOp(spec);
          AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, dummyKeyProviderOp, apc);
          return dummyKeyProviderOp;
      }
  
      private static IOperatorDescriptor createDummyKeyProviderOp(JobSpecification spec) throws AlgebricksException {
          // Build dummy tuple containing one field with a dummy value inside.
          ArrayTupleBuilder tb = new ArrayTupleBuilder(1);
          DataOutput dos = tb.getDataOutput();
          tb.reset();
          try {
              // Serialize dummy value into a field.
              IntegerSerializerDeserializer.INSTANCE.serialize(0, dos);
          } catch (HyracksDataException e) {
              throw new AsterixException(e);
          }
          // Add dummy field.
          tb.addFieldEndOffset();
          ISerializerDeserializer[] keyRecDescSers = { IntegerSerializerDeserializer.INSTANCE };
          RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
          ConstantTupleSourceOperatorDescriptor keyProviderOp = new ConstantTupleSourceOperatorDescriptor(spec,
                  keyRecDesc, tb.getFieldEndOffsets(), tb.getByteArray(), tb.getSize());
          return keyProviderOp;
      }
  
      public static String getFullyQualifiedDisplayName(Dataset dataset) {
          return getFullyQualifiedDisplayName(dataset.getDataverseName(), dataset.getDatasetName());
      }
  
      public static String getFullyQualifiedDisplayName(DataverseName dataverseName, String datasetName) {
          return MetadataUtil.getFullyQualifiedDisplayName(dataverseName, datasetName);
      }
  
      public static String getFullyQualifiedDisplayName(String databaseName, DataverseName dataverseName,
              String datasetName) {
          return MetadataUtil.getFullyQualifiedDisplayName(databaseName, dataverseName, datasetName);
      }
  
      /***
       * Creates a node group that is associated with a new dataset.
       *
       * @param databaseName,
       *            the database name of the dataset.
       * @param dataverseName,
       *            the dataverse name of the dataset.
       * @param datasetName,
       *            the name of the dataset.
       * @param ncNames,
       *            the set of node names.
       * @param metadataProvider,
       *            the metadata provider.
       * @return the name of the created node group.
       * @throws Exception
       */
      public static String createNodeGroupForNewDataset(String databaseName, DataverseName dataverseName,
              String datasetName, Set<String> ncNames, MetadataProvider metadataProvider) throws Exception {
          return createNodeGroupForNewDataset(databaseName, dataverseName, datasetName, 0L, ncNames, metadataProvider);
      }
  
      /***
       * Creates a node group that is associated with a new dataset.
       *
       * @param databaseName,
       *            the database name of the dataset.
       * @param dataverseName,
       *            the dataverse name of the dataset.
       * @param datasetName,
       *            the name of the dataset.
       * @param rebalanceCount
       *            , the rebalance count of the dataset.
       * @param ncNames,
       *            the set of node names.
       * @param metadataProvider,
       *            the metadata provider.
       * @return the name of the created node group.
       * @throws Exception
       */
      public static String createNodeGroupForNewDataset(String databaseName, DataverseName dataverseName,
              String datasetName, long rebalanceCount, Set<String> ncNames, MetadataProvider metadataProvider)
              throws Exception {
          ICcApplicationContext appCtx = metadataProvider.getApplicationContext();
          boolean useDb = metadataProvider.getNamespaceResolver().isUsingDatabase();
          String nodeGroup = (useDb ? databaseName + "." : "") + dataverseName.getCanonicalForm() + "." + datasetName
                  + (rebalanceCount == 0L ? "" : "_" + rebalanceCount);
          MetadataTransactionContext mdTxnCtx = metadataProvider.getMetadataTxnContext();
          appCtx.getMetadataLockManager().acquireNodeGroupWriteLock(metadataProvider.getLocks(), nodeGroup);
          NodeGroup ng = MetadataManager.INSTANCE.getNodegroup(mdTxnCtx, nodeGroup);
          if (ng != null) {
              nodeGroup = nodeGroup + "_" + UUID.randomUUID();
              appCtx.getMetadataLockManager().acquireNodeGroupWriteLock(metadataProvider.getLocks(), nodeGroup);
          }
          MetadataManager.INSTANCE.addNodegroup(mdTxnCtx, NodeGroup.createOrdered(nodeGroup, ncNames));
          return nodeGroup;
      }
  
      public static String getDatasetTypeDisplayName(DatasetType datasetType) {
          return datasetType == DatasetType.VIEW ? "view" : dataset();
      }
  
      public static boolean isNotView(Dataset dataset) {
          return dataset.getDatasetType() != DatasetType.VIEW;
      }
  
      public static boolean isFieldAccessPushdownSupported(Dataset dataset) {
          DatasetType datasetType = dataset.getDatasetType();
          if (datasetType == DatasetType.INTERNAL) {
              return dataset.getDatasetFormatInfo().getFormat() == DatasetConfig.DatasetFormat.COLUMN;
          } else if (datasetType == DatasetType.EXTERNAL) {
              ExternalDatasetDetails edd = (ExternalDatasetDetails) dataset.getDatasetDetails();
              return ExternalDataUtils.supportsPushdown(edd.getProperties());
          }
          return false;
      }
  
      public static boolean isFilterPushdownSupported(Dataset dataset) {
          if (dataset.getDatasetType() == DatasetType.INTERNAL) {
              return dataset.getDatasetFormatInfo().getFormat() == DatasetConfig.DatasetFormat.COLUMN;
          }
          // External dataset support filter pushdown
          return true;
      }
  
      public static boolean isRangeFilterPushdownSupported(Dataset dataset) {
          return dataset.getDatasetType() == DatasetType.INTERNAL
                  && dataset.getDatasetFormatInfo().getFormat() == DatasetConfig.DatasetFormat.COLUMN;
      }
  
      public static void truncate(MetadataProvider metadataProvider, Dataset ds) throws Exception {
          if (ds.getDatasetType() == DatasetType.INTERNAL) {
              IHyracksClientConnection hcc;
              Map<String, List<DatasetPartitions>> nc2Resources = getNodeResources(metadataProvider, ds);
              AlgebricksAbsolutePartitionConstraint nodeSet =
                      new AlgebricksAbsolutePartitionConstraint(nc2Resources.keySet().toArray(new String[0]));
              JobSpecification job = new JobSpecification();
              IOperatorDescriptor truncateOp = new TruncateOperatorDescriptor(job, nc2Resources);
              AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(job, truncateOp, nodeSet);
              hcc = metadataProvider.getApplicationContext().getHcc();
              JobUtils.runJobIfActive(hcc, job, true);
          } else {
              // check should have been done by caller
              throw new IllegalArgumentException(
                      "Cannot truncate a non-internal dataset " + ds.getDatasetFullyQualifiedName());
          }
      }
  
      public static DatasetPartitions getPartitionsAndDataflowHelperFactory(Dataset dataset,
              List<DatasetPartitions> ncResources, IIndexDataflowHelperFactory primary,
              List<IIndexDataflowHelperFactory> secondaries) {
          DatasetPartitions partitionsAndDataflowHelperFactory;
          if (ncResources.isEmpty()) {
              partitionsAndDataflowHelperFactory =
                      new DatasetPartitions(dataset, new ArrayList<>(), primary, secondaries);
              ncResources.add(partitionsAndDataflowHelperFactory);
          } else {
              DatasetPartitions last = ncResources.get(ncResources.size() - 1);
              if (last.getPrimaryIndexDataflowHelperFactory() == primary) {
                  partitionsAndDataflowHelperFactory = last;
              } else {
                  partitionsAndDataflowHelperFactory =
                          new DatasetPartitions(dataset, new ArrayList<>(), primary, secondaries);
                  ncResources.add(partitionsAndDataflowHelperFactory);
              }
          }
          return partitionsAndDataflowHelperFactory;
      }
  
      public static Map<String, List<DatasetPartitions>> getNodeResources(MetadataProvider metadataProvider,
              Dataset dataset) throws AlgebricksException {
          Map<String, List<DatasetPartitions>> nc2Resources = new HashMap<>();
          IStorageManager storageManager = metadataProvider.getStorageComponentProvider().getStorageManager();
  
          // get secondary indexes
          List<Index> secondaryIndexes =
                  metadataProvider
                          .getDatasetIndexes(dataset.getDatabaseName(), dataset.getDataverseName(),
                                  dataset.getDatasetName())
                          .stream().filter(Index::isSecondaryIndex).collect(Collectors.toList());
          PartitioningProperties partitioningProperties = metadataProvider.getPartitioningProperties(dataset);
          IIndexDataflowHelperFactory primary =
                  new IndexDataflowHelperFactory(storageManager, partitioningProperties.getSplitsProvider());
          List<IIndexDataflowHelperFactory> secondaries =
                  secondaryIndexes.isEmpty() ? Collections.emptyList() : new ArrayList<>();
          for (Index index : secondaryIndexes) {
              PartitioningProperties idxPartitioningProperties =
                      metadataProvider.getPartitioningProperties(dataset, index.getIndexName());
              secondaries
                      .add(new IndexDataflowHelperFactory(storageManager, idxPartitioningProperties.getSplitsProvider()));
          }
          populateNcResources(dataset, partitioningProperties, nc2Resources, primary, secondaries);
          return nc2Resources;
      }
  
      private static void populateNcResources(Dataset dataset, PartitioningProperties partitioningProperties,
              Map<String, List<DatasetPartitions>> nc2Resources, IIndexDataflowHelperFactory primary,
              List<IIndexDataflowHelperFactory> secondaries) {
          AlgebricksAbsolutePartitionConstraint computeLocations =
                  (AlgebricksAbsolutePartitionConstraint) partitioningProperties.getConstraints();
          int[][] computeStorageMap = partitioningProperties.getComputeStorageMap();
          for (int j = 0; j < computeLocations.getLocations().length; j++) {
              String loc = computeLocations.getLocations()[j];
              DatasetPartitions partitionsAndDataflowHelperFactories = getPartitionsAndDataflowHelperFactory(dataset,
                      nc2Resources.computeIfAbsent(loc, key -> new ArrayList<>()), primary, secondaries);
              List<Integer> dsPartitions = partitionsAndDataflowHelperFactories.getPartitions();
              int[] computeStoragePartitions = computeStorageMap[j];
              for (int storagePartition : computeStoragePartitions) {
                  dsPartitions.add(storagePartition);
              }
          }
      }
  
      public static Map<String, List<DatasetPartitions>> getNodeResourcesWithoutSecondaries(
              MetadataProvider metadataProvider, Dataset dataset) throws AlgebricksException {
          Map<String, List<DatasetPartitions>> nc2Resources = new HashMap<>();
          IStorageManager storageManager = metadataProvider.getStorageComponentProvider().getStorageManager();
  
          PartitioningProperties partitioningProperties = metadataProvider.getPartitioningProperties(dataset);
          IIndexDataflowHelperFactory primary =
                  new IndexDataflowHelperFactory(storageManager, partitioningProperties.getSplitsProvider());
  
          populateNcResources(dataset, partitioningProperties, nc2Resources, primary, Collections.emptyList());
          return nc2Resources;
      }
  
      public static DatasetResource getDatasetResource(DatasetLifecycleManager dslMgr, Integer partition,
              IndexDataflowHelper indexHelper) throws HyracksDataException {
          LocalResource lr = indexHelper.getResource();
          DatasetLocalResource dslr = (DatasetLocalResource) lr.getResource();
          int datasetId = dslr.getDatasetId();
          DatasetResource dsr = dslMgr.getDatasetLifecycle(datasetId);
          // Ensure that no active operations exists
          int numActiveOperations =
                  dslMgr.getOperationTracker(datasetId, partition, lr.getPath()).getNumActiveOperations();
          if (numActiveOperations > 0) {
              throw new IllegalStateException("Can't truncate the collection " + dsr.getDatasetInfo().getDatasetID()
                      + " because the number of active operations = " + numActiveOperations + " in the partition "
                      + partition);
          }
          return dsr;
      }
  
      public static boolean isDeltaTable(Dataset dataset) {
          return dataset.getDatasetType() == DatasetType.EXTERNAL && ExternalDataUtils
                  .isDeltaTable(((ExternalDatasetDetails) dataset.getDatasetDetails()).getProperties());
      }
  
      public static boolean isParquetFormat(Dataset dataset) {
          return dataset.getDatasetType() == DatasetType.EXTERNAL && ExternalDataUtils
                  .isParquetFormat(((ExternalDatasetDetails) dataset.getDatasetDetails()).getProperties());
      }
  }
  