asterixdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From amo...@apache.org
Subject [19/25] asterixdb git commit: Separate index build from index access
Date Thu, 11 May 2017 23:43:34 GMT
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/735532e4/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/InvertedIndexResourceFactoryProvider.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/InvertedIndexResourceFactoryProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/InvertedIndexResourceFactoryProvider.java
new file mode 100644
index 0000000..362305e
--- /dev/null
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/InvertedIndexResourceFactoryProvider.java
@@ -0,0 +1,248 @@
+/*
+ * 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 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.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.IAType;
+import org.apache.asterix.om.utils.NonTaggedFormatUtil;
+import org.apache.asterix.om.utils.RecordUtil;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.common.utils.Pair;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.ITypeTraits;
+import org.apache.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
+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.invertedindex.dataflow.LSMInvertedIndexLocalResourceFactory;
+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();
+        List<List<String>> secondaryKeys = index.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 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);
+        IMetadataPageManagerFactory metadataPageManagerFactory =
+                storageComponentProvider.getMetadataPageManagerFactory();
+        AsterixVirtualBufferCacheProvider vbcProvider = new AsterixVirtualBufferCacheProvider(dataset.getDatasetId());
+        ILSMIOOperationSchedulerProvider ioSchedulerProvider =
+                storageComponentProvider.getIoOperationSchedulerProvider();
+        boolean durable = !dataset.isTemp();
+        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);
+        return new LSMInvertedIndexLocalResourceFactory(storageManager, typeTraits, cmpFactories, filterTypeTraits,
+                filterCmpFactories, secondaryFilterFields, opTrackerFactory, ioOpCallbackFactory,
+                metadataPageManagerFactory, vbcProvider, ioSchedulerProvider, mergePolicyFactory, mergePolicyProperties,
+                durable, tokenTypeTraits, tokenCmpFactories, tokenizerFactory, isPartitioned, invertedIndexFields,
+                secondaryFilterFieldsForNonBulkLoadOps, invertedIndexFieldsForNonBulkLoadOps,
+                bloomFilterFalsePositiveRate);
+    }
+
+    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();
+        int numSecondaryKeys = index.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, RecordUtil.toFullyQualifiedName(dataset.getDataverseName(), dataset.getDatasetName()));
+        }
+        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 = index.getKeyFieldSourceIndicators();
+        if (keySourceIndicators == null || keySourceIndicators.get(0) == 0) {
+            sourceType = recordType;
+        } else {
+            sourceType = metaType;
+        }
+        Pair<IAType, Boolean> keyTypePair = Index.getNonNullableOpenFieldType(index.getKeyFieldTypes().get(0),
+                index.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();
+        int numSecondaryKeys = index.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, RecordUtil.toFullyQualifiedName(dataset.getDataverseName(), dataset.getDatasetName()));
+        }
+        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 = index.getKeyFieldSourceIndicators();
+        ARecordType sourceType;
+        if (keySourceIndicators == null || keySourceIndicators.get(0) == 0) {
+            sourceType = recordType;
+        } else {
+            sourceType = metaType;
+        }
+        Pair<IAType, Boolean> keyTypePair = Index.getNonNullableOpenFieldType(index.getKeyFieldTypes().get(0),
+                index.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] = PointableBinaryComparatorFactory.of(ShortPointable.FACTORY);
+        }
+        return tokenComparatorFactories;
+    }
+
+    private static IBinaryTokenizerFactory getTokenizerFactory(Dataset dataset, Index index, ARecordType recordType,
+            ARecordType metaType) throws AlgebricksException {
+        int numPrimaryKeys = dataset.getPrimaryKeys().size();
+        int numSecondaryKeys = index.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, RecordUtil.toFullyQualifiedName(dataset.getDataverseName(), dataset.getDatasetName()));
+        }
+        if (numSecondaryKeys > 1) {
+            throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_INDEX_NUM_OF_FIELD, numSecondaryKeys,
+                    indexType, 1);
+        }
+        ARecordType sourceType;
+        List<Integer> keySourceIndicators = index.getKeyFieldSourceIndicators();
+        if (keySourceIndicators == null || keySourceIndicators.get(0) == 0) {
+            sourceType = recordType;
+        } else {
+            sourceType = metaType;
+        }
+        Pair<IAType, Boolean> keyTypePair = Index.getNonNullableOpenFieldType(index.getKeyFieldTypes().get(0),
+                index.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,
+                index.getGramLength());
+    }
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/735532e4/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/RTreeDataflowHelperFactoryProvider.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/RTreeDataflowHelperFactoryProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/RTreeDataflowHelperFactoryProvider.java
deleted file mode 100644
index 45034de..0000000
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/RTreeDataflowHelperFactoryProvider.java
+++ /dev/null
@@ -1,172 +0,0 @@
-/*
- * 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 java.util.List;
-import java.util.Map;
-
-import org.apache.asterix.common.config.DatasetConfig.DatasetType;
-import org.apache.asterix.common.context.AsterixVirtualBufferCacheProvider;
-import org.apache.asterix.common.context.IStorageComponentProvider;
-import org.apache.asterix.common.exceptions.CompilationException;
-import org.apache.asterix.common.exceptions.ErrorCode;
-import org.apache.asterix.external.indexing.IndexingConstants;
-import org.apache.asterix.metadata.api.IIndexDataflowHelperFactoryProvider;
-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.entities.InternalDatasetDetails;
-import org.apache.asterix.metadata.lock.ExternalDatasetsRegistry;
-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.NonTaggedFormatUtil;
-import org.apache.commons.lang.StringUtils;
-import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
-import org.apache.hyracks.api.dataflow.value.ITypeTraits;
-import org.apache.hyracks.storage.am.common.api.IPrimitiveValueProviderFactory;
-import org.apache.hyracks.storage.am.common.dataflow.IIndexDataflowHelperFactory;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
-import org.apache.hyracks.storage.am.lsm.rtree.dataflow.ExternalRTreeDataflowHelperFactory;
-import org.apache.hyracks.storage.am.lsm.rtree.dataflow.LSMRTreeWithAntiMatterTuplesDataflowHelperFactory;
-import org.apache.hyracks.storage.am.rtree.frames.RTreePolicyType;
-
-public class RTreeDataflowHelperFactoryProvider implements IIndexDataflowHelperFactoryProvider {
-
-    public static final RTreeDataflowHelperFactoryProvider INSTANCE = new RTreeDataflowHelperFactoryProvider();
-
-    private RTreeDataflowHelperFactoryProvider() {
-    }
-
-    protected RTreePolicyType rTreePolicyType() {
-        return RTreePolicyType.RTREE;
-    }
-
-    @Override
-    public IIndexDataflowHelperFactory getIndexDataflowHelperFactory(MetadataProvider mdProvider, Dataset dataset,
-            Index index, ARecordType recordType, ARecordType metaType, ILSMMergePolicyFactory mergePolicyFactory,
-            Map<String, String> mergePolicyProperties, ITypeTraits[] filterTypeTraits,
-            IBinaryComparatorFactory[] filterCmpFactories) throws AlgebricksException {
-        if (index.getKeyFieldNames().size() != 1) {
-            throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_INDEX_NUM_OF_FIELD,
-                    index.getKeyFieldNames().size(), index.getIndexType(), 1);
-        }
-        IAType spatialType = Index.getNonNullableOpenFieldType(index.getKeyFieldTypes().get(0),
-                index.getKeyFieldNames().get(0), recordType).first;
-        if (spatialType == null) {
-            throw new CompilationException(ErrorCode.COMPILATION_FIELD_NOT_FOUND,
-                    StringUtils.join(index.getKeyFieldNames().get(0), '.'));
-        }
-        List<List<String>> primaryKeyFields = DatasetUtil.getPartitioningKeys(dataset);
-        int numPrimaryKeys = primaryKeyFields.size();
-        ITypeTraits[] primaryTypeTraits = null;
-        IBinaryComparatorFactory[] primaryComparatorFactories = null;
-        IStorageComponentProvider storageComponentProvider = mdProvider.getStorageComponentProvider();
-        if (dataset.getDatasetType() == DatasetType.INTERNAL) {
-            primaryTypeTraits = new ITypeTraits[numPrimaryKeys + 1 + (dataset.hasMetaPart() ? 1 : 0)];
-            primaryComparatorFactories = new IBinaryComparatorFactory[numPrimaryKeys];
-            List<Integer> indicators = null;
-            if (dataset.hasMetaPart()) {
-                indicators = ((InternalDatasetDetails) dataset.getDatasetDetails()).getKeySourceIndicator();
-            }
-            for (int i = 0; i < numPrimaryKeys; i++) {
-                IAType keyType = (indicators == null || indicators.get(i) == 0)
-                        ? recordType.getSubFieldType(primaryKeyFields.get(i))
-                        : metaType.getSubFieldType(primaryKeyFields.get(i));
-                primaryComparatorFactories[i] = storageComponentProvider.getComparatorFactoryProvider()
-                        .getBinaryComparatorFactory(keyType, true);
-                primaryTypeTraits[i] = storageComponentProvider.getTypeTraitProvider().getTypeTrait(keyType);
-            }
-            primaryTypeTraits[numPrimaryKeys] =
-                    storageComponentProvider.getTypeTraitProvider().getTypeTrait(recordType);
-            if (dataset.hasMetaPart()) {
-                primaryTypeTraits[numPrimaryKeys + 1] =
-                        storageComponentProvider.getTypeTraitProvider().getTypeTrait(recordType);
-            }
-        }
-        boolean isPointMBR =
-                spatialType.getTypeTag() == ATypeTag.POINT || spatialType.getTypeTag() == ATypeTag.POINT3D;
-        int numDimensions = NonTaggedFormatUtil.getNumDimensions(spatialType.getTypeTag());
-        int numNestedSecondaryKeyFields = numDimensions * 2;
-        IBinaryComparatorFactory[] secondaryComparatorFactories =
-                new IBinaryComparatorFactory[numNestedSecondaryKeyFields];
-        IPrimitiveValueProviderFactory[] valueProviderFactories =
-                new IPrimitiveValueProviderFactory[numNestedSecondaryKeyFields];
-        ITypeTraits[] secondaryTypeTraits = new ITypeTraits[numNestedSecondaryKeyFields + numPrimaryKeys];
-        IAType nestedKeyType = NonTaggedFormatUtil.getNestedSpatialType(spatialType.getTypeTag());
-        ATypeTag keyType = nestedKeyType.getTypeTag();
-        for (int i = 0; i < numNestedSecondaryKeyFields; i++) {
-            secondaryComparatorFactories[i] = storageComponentProvider.getComparatorFactoryProvider()
-                    .getBinaryComparatorFactory(nestedKeyType, true);
-            secondaryTypeTraits[i] = storageComponentProvider.getTypeTraitProvider().getTypeTrait(nestedKeyType);
-            valueProviderFactories[i] = storageComponentProvider.getPrimitiveValueProviderFactory();
-
-        }
-        for (int i = 0; i < numPrimaryKeys; i++) {
-            secondaryTypeTraits[numNestedSecondaryKeyFields + i] = (dataset.getDatasetType() == DatasetType.INTERNAL)
-                    ? primaryTypeTraits[i] : IndexingConstants.getTypeTraits(i);
-        }
-        int[] rtreeFields = null;
-        if (filterTypeTraits != null && filterTypeTraits.length > 0) {
-            rtreeFields = new int[numNestedSecondaryKeyFields + numPrimaryKeys];
-            for (int i = 0; i < rtreeFields.length; i++) {
-                rtreeFields[i] = i;
-            }
-        }
-        if (dataset.getDatasetType() == DatasetType.INTERNAL) {
-            int[] secondaryFilterFields = (filterTypeTraits != null && filterTypeTraits.length > 0)
-                    ? new int[] { numNestedSecondaryKeyFields + numPrimaryKeys } : null;
-            IBinaryComparatorFactory[] btreeCompFactories = getComparatorFactoriesForDeletedKeyBTree(
-                    secondaryTypeTraits, primaryComparatorFactories, secondaryComparatorFactories);
-            return new LSMRTreeWithAntiMatterTuplesDataflowHelperFactory(valueProviderFactories, rTreePolicyType(),
-                    btreeCompFactories, new AsterixVirtualBufferCacheProvider(dataset.getDatasetId()),
-                    mergePolicyFactory, mergePolicyProperties, dataset.getIndexOperationTrackerFactory(index),
-                    storageComponentProvider.getIoOperationSchedulerProvider(),
-                    dataset.getIoOperationCallbackFactory(index),
-                    MetadataProvider.proposeLinearizer(keyType, secondaryComparatorFactories.length), rtreeFields,
-                    filterTypeTraits, filterCmpFactories, secondaryFilterFields, !dataset.getDatasetDetails().isTemp(),
-                    isPointMBR);
-        } else {
-            return new ExternalRTreeDataflowHelperFactory(valueProviderFactories, rTreePolicyType(),
-                    ExternalIndexingOperations.getBuddyBtreeComparatorFactories(), mergePolicyFactory,
-                    mergePolicyProperties, dataset.getIndexOperationTrackerFactory(index),
-                    storageComponentProvider.getIoOperationSchedulerProvider(),
-                    dataset.getIoOperationCallbackFactory(index),
-                    MetadataProvider.proposeLinearizer(keyType, secondaryComparatorFactories.length),
-                    mdProvider.getStorageProperties().getBloomFilterFalsePositiveRate(),
-                    new int[] { numNestedSecondaryKeyFields },
-                    ExternalDatasetsRegistry.INSTANCE.getDatasetVersion(dataset), true, isPointMBR);
-        }
-    }
-
-    private static IBinaryComparatorFactory[] getComparatorFactoriesForDeletedKeyBTree(
-            ITypeTraits[] secondaryTypeTraits, IBinaryComparatorFactory[] primaryComparatorFactories,
-            IBinaryComparatorFactory[] secondaryComparatorFactories) {
-        IBinaryComparatorFactory[] btreeCompFactories = new IBinaryComparatorFactory[secondaryTypeTraits.length];
-        int i = 0;
-        for (; i < secondaryComparatorFactories.length; i++) {
-            btreeCompFactories[i] = secondaryComparatorFactories[i];
-        }
-        for (int j = 0; i < secondaryTypeTraits.length; i++, j++) {
-            btreeCompFactories[i] = primaryComparatorFactories[j];
-        }
-        return btreeCompFactories;
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/735532e4/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/RTreeResourceFactoryProvider.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/RTreeResourceFactoryProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/RTreeResourceFactoryProvider.java
new file mode 100644
index 0000000..425f8a1
--- /dev/null
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/RTreeResourceFactoryProvider.java
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.List;
+import java.util.Map;
+
+import org.apache.asterix.common.config.DatasetConfig.DatasetType;
+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.external.indexing.IndexingConstants;
+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.metadata.entities.InternalDatasetDetails;
+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.NonTaggedFormatUtil;
+import org.apache.commons.lang.StringUtils;
+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.algebricks.data.ITypeTraitProvider;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.ILinearizeComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.ITypeTraits;
+import org.apache.hyracks.storage.am.common.api.IMetadataPageManagerFactory;
+import org.apache.hyracks.storage.am.common.api.IPrimitiveValueProviderFactory;
+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.rtree.dataflow.ExternalRTreeLocalResourceFactory;
+import org.apache.hyracks.storage.am.lsm.rtree.dataflow.LSMRTreeWithAntiMatterLocalResourceFactory;
+import org.apache.hyracks.storage.am.rtree.frames.RTreePolicyType;
+import org.apache.hyracks.storage.common.IResourceFactory;
+import org.apache.hyracks.storage.common.IStorageManager;
+
+public class RTreeResourceFactoryProvider implements IResourceFactoryProvider {
+
+    private static final RTreePolicyType rTreePolicyType = RTreePolicyType.RTREE;
+    public static final RTreeResourceFactoryProvider INSTANCE = new RTreeResourceFactoryProvider();
+
+    private RTreeResourceFactoryProvider() {
+    }
+
+    @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 {
+        if (index.getKeyFieldNames().size() != 1) {
+            throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_INDEX_NUM_OF_FIELD,
+                    index.getKeyFieldNames().size(), index.getIndexType(), 1);
+        }
+        IAType spatialType = Index.getNonNullableOpenFieldType(index.getKeyFieldTypes().get(0),
+                index.getKeyFieldNames().get(0), recordType).first;
+        if (spatialType == null) {
+            throw new CompilationException(ErrorCode.COMPILATION_FIELD_NOT_FOUND,
+                    StringUtils.join(index.getKeyFieldNames().get(0), '.'));
+        }
+        List<List<String>> primaryKeyFields = dataset.getPrimaryKeys();
+        int numPrimaryKeys = primaryKeyFields.size();
+        ITypeTraits[] primaryTypeTraits = null;
+        IBinaryComparatorFactory[] primaryComparatorFactories = null;
+        IStorageComponentProvider storageComponentProvider = mdProvider.getStorageComponentProvider();
+        if (dataset.getDatasetType() == DatasetType.INTERNAL) {
+            primaryTypeTraits = new ITypeTraits[numPrimaryKeys + 1 + (dataset.hasMetaPart() ? 1 : 0)];
+            primaryComparatorFactories = new IBinaryComparatorFactory[numPrimaryKeys];
+            List<Integer> indicators = null;
+            if (dataset.hasMetaPart()) {
+                indicators = ((InternalDatasetDetails) dataset.getDatasetDetails()).getKeySourceIndicator();
+            }
+            for (int i = 0; i < numPrimaryKeys; i++) {
+                IAType keyType = (indicators == null || indicators.get(i) == 0)
+                        ? recordType.getSubFieldType(primaryKeyFields.get(i))
+                        : metaType.getSubFieldType(primaryKeyFields.get(i));
+                primaryComparatorFactories[i] = storageComponentProvider.getComparatorFactoryProvider()
+                        .getBinaryComparatorFactory(keyType, true);
+                primaryTypeTraits[i] = storageComponentProvider.getTypeTraitProvider().getTypeTrait(keyType);
+            }
+            primaryTypeTraits[numPrimaryKeys] =
+                    storageComponentProvider.getTypeTraitProvider().getTypeTrait(recordType);
+            if (dataset.hasMetaPart()) {
+                primaryTypeTraits[numPrimaryKeys + 1] =
+                        storageComponentProvider.getTypeTraitProvider().getTypeTrait(recordType);
+            }
+        }
+        boolean isPointMBR = spatialType.getTypeTag() == ATypeTag.POINT || spatialType.getTypeTag() == ATypeTag.POINT3D;
+        int numDimensions = NonTaggedFormatUtil.getNumDimensions(spatialType.getTypeTag());
+        int numNestedSecondaryKeyFields = numDimensions * 2;
+        IBinaryComparatorFactory[] secondaryComparatorFactories =
+                new IBinaryComparatorFactory[numNestedSecondaryKeyFields];
+        IPrimitiveValueProviderFactory[] valueProviderFactories =
+                new IPrimitiveValueProviderFactory[numNestedSecondaryKeyFields];
+        ITypeTraits[] secondaryTypeTraits = new ITypeTraits[numNestedSecondaryKeyFields + numPrimaryKeys];
+        IAType nestedKeyType = NonTaggedFormatUtil.getNestedSpatialType(spatialType.getTypeTag());
+        ATypeTag keyType = nestedKeyType.getTypeTag();
+        for (int i = 0; i < numNestedSecondaryKeyFields; i++) {
+            secondaryComparatorFactories[i] = storageComponentProvider.getComparatorFactoryProvider()
+                    .getBinaryComparatorFactory(nestedKeyType, true);
+            secondaryTypeTraits[i] = storageComponentProvider.getTypeTraitProvider().getTypeTrait(nestedKeyType);
+            valueProviderFactories[i] = storageComponentProvider.getPrimitiveValueProviderFactory();
+
+        }
+        for (int i = 0; i < numPrimaryKeys; i++) {
+            secondaryTypeTraits[numNestedSecondaryKeyFields + i] = (dataset.getDatasetType() == DatasetType.INTERNAL)
+                    ? primaryTypeTraits[i] : IndexingConstants.getTypeTraits(i);
+        }
+        int[] rtreeFields = null;
+        if (filterTypeTraits != null && filterTypeTraits.length > 0) {
+            rtreeFields = new int[numNestedSecondaryKeyFields + numPrimaryKeys];
+            for (int i = 0; i < rtreeFields.length; i++) {
+                rtreeFields[i] = i;
+            }
+        }
+        IStorageManager storageManager = storageComponentProvider.getStorageManager();
+        ILSMOperationTrackerFactory opTrackerFactory = dataset.getIndexOperationTrackerFactory(index);
+        ILSMIOOperationCallbackFactory ioOpCallbackFactory = dataset.getIoOperationCallbackFactory(index);
+        IMetadataPageManagerFactory metadataPageManagerFactory =
+                storageComponentProvider.getMetadataPageManagerFactory();
+        ILSMIOOperationSchedulerProvider ioSchedulerProvider =
+                storageComponentProvider.getIoOperationSchedulerProvider();
+        boolean durable = !dataset.isTemp();
+        ILinearizeComparatorFactory linearizeCmpFactory =
+                MetadataProvider.proposeLinearizer(keyType, secondaryComparatorFactories.length);
+        ITypeTraits[] typeTraits = getTypeTraits(mdProvider, dataset, index, recordType, metaType);
+        IBinaryComparatorFactory[] rtreeCmpFactories = getCmpFactories(mdProvider, index, recordType, metaType);
+        int[] secondaryFilterFields = (filterTypeTraits != null && filterTypeTraits.length > 0)
+                ? new int[] { numNestedSecondaryKeyFields + numPrimaryKeys } : null;
+        IBinaryComparatorFactory[] btreeCompFactories =
+                dataset.getDatasetType() == DatasetType.EXTERNAL ? IndexingConstants.getBuddyBtreeComparatorFactories()
+                        : getComparatorFactoriesForDeletedKeyBTree(secondaryTypeTraits, primaryComparatorFactories,
+                                secondaryComparatorFactories);
+        if (dataset.getDatasetType() == DatasetType.INTERNAL) {
+            AsterixVirtualBufferCacheProvider vbcProvider =
+                    new AsterixVirtualBufferCacheProvider(dataset.getDatasetId());
+            return new LSMRTreeWithAntiMatterLocalResourceFactory(storageManager, typeTraits, rtreeCmpFactories,
+                    filterTypeTraits, filterCmpFactories, secondaryFilterFields, opTrackerFactory, ioOpCallbackFactory,
+                    metadataPageManagerFactory, vbcProvider, ioSchedulerProvider, mergePolicyFactory,
+                    mergePolicyProperties, durable, valueProviderFactories, rTreePolicyType, linearizeCmpFactory,
+                    rtreeFields, isPointMBR, btreeCompFactories);
+        } else {
+            return new ExternalRTreeLocalResourceFactory(storageManager, typeTraits, rtreeCmpFactories,
+                    filterTypeTraits, filterCmpFactories, secondaryFilterFields, opTrackerFactory, ioOpCallbackFactory,
+                    metadataPageManagerFactory, ioSchedulerProvider, mergePolicyFactory, mergePolicyProperties, durable,
+                    btreeCompFactories, valueProviderFactories, rTreePolicyType, linearizeCmpFactory, rtreeFields,
+                    new int[] { numNestedSecondaryKeyFields }, isPointMBR,
+                    mdProvider.getStorageProperties().getBloomFilterFalsePositiveRate());
+        }
+    }
+
+    private static IBinaryComparatorFactory[] getComparatorFactoriesForDeletedKeyBTree(
+            ITypeTraits[] secondaryTypeTraits, IBinaryComparatorFactory[] primaryComparatorFactories,
+            IBinaryComparatorFactory[] secondaryComparatorFactories) {
+        IBinaryComparatorFactory[] btreeCompFactories = new IBinaryComparatorFactory[secondaryTypeTraits.length];
+        int i = 0;
+        for (; i < secondaryComparatorFactories.length; i++) {
+            btreeCompFactories[i] = secondaryComparatorFactories[i];
+        }
+        for (int j = 0; i < secondaryTypeTraits.length; i++, j++) {
+            btreeCompFactories[i] = primaryComparatorFactories[j];
+        }
+        return btreeCompFactories;
+    }
+
+    private static ITypeTraits[] getTypeTraits(MetadataProvider metadataProvider, Dataset dataset, Index index,
+            ARecordType recordType, ARecordType metaType) throws AlgebricksException {
+        ITypeTraitProvider ttProvider = metadataProvider.getStorageComponentProvider().getTypeTraitProvider();
+        List<List<String>> secondaryKeyFields = index.getKeyFieldNames();
+        int numSecondaryKeys = secondaryKeyFields.size();
+        int numPrimaryKeys = dataset.getPrimaryKeys().size();
+        ITypeTraits[] primaryTypeTraits = dataset.getPrimaryTypeTraits(metadataProvider, recordType, metaType);
+        if (numSecondaryKeys != 1) {
+            throw new AsterixException("Cannot use " + numSecondaryKeys + " fields as a key for the R-tree index. "
+                    + "There can be only one field as a key for the R-tree index.");
+        }
+        ARecordType sourceType;
+        List<Integer> keySourceIndicators = index.getKeyFieldSourceIndicators();
+        if (keySourceIndicators == null || keySourceIndicators.get(0) == 0) {
+            sourceType = recordType;
+        } else {
+            sourceType = metaType;
+        }
+        Pair<IAType, Boolean> spatialTypePair = Index.getNonNullableOpenFieldType(index.getKeyFieldTypes().get(0),
+                secondaryKeyFields.get(0), sourceType);
+        IAType spatialType = spatialTypePair.first;
+        if (spatialType == null) {
+            throw new AsterixException("Could not find field " + secondaryKeyFields.get(0) + " in the schema.");
+        }
+        int numDimensions = NonTaggedFormatUtil.getNumDimensions(spatialType.getTypeTag());
+        int numNestedSecondaryKeyFields = numDimensions * 2;
+        ITypeTraits[] secondaryTypeTraits = new ITypeTraits[numNestedSecondaryKeyFields + numPrimaryKeys];
+        IAType nestedKeyType = NonTaggedFormatUtil.getNestedSpatialType(spatialType.getTypeTag());
+        for (int i = 0; i < numNestedSecondaryKeyFields; i++) {
+            secondaryTypeTraits[i] = ttProvider.getTypeTrait(nestedKeyType);
+        }
+        for (int i = 0; i < numPrimaryKeys; i++) {
+            secondaryTypeTraits[numNestedSecondaryKeyFields + i] = primaryTypeTraits[i];
+        }
+        return secondaryTypeTraits;
+    }
+
+    private static IBinaryComparatorFactory[] getCmpFactories(MetadataProvider metadataProvider, Index index,
+            ARecordType recordType, ARecordType metaType) throws AlgebricksException {
+        IBinaryComparatorFactoryProvider cmpFactoryProvider =
+                metadataProvider.getStorageComponentProvider().getComparatorFactoryProvider();
+        List<List<String>> secondaryKeyFields = index.getKeyFieldNames();
+        int numSecondaryKeys = secondaryKeyFields.size();
+        if (numSecondaryKeys != 1) {
+            throw new AsterixException("Cannot use " + numSecondaryKeys + " fields as a key for the R-tree index. "
+                    + "There can be only one field as a key for the R-tree index.");
+        }
+        List<Integer> keySourceIndicators = index.getKeyFieldSourceIndicators();
+        ARecordType sourceType;
+        if (keySourceIndicators == null || keySourceIndicators.get(0) == 0) {
+            sourceType = recordType;
+        } else {
+            sourceType = metaType;
+        }
+        Pair<IAType, Boolean> spatialTypePair = Index.getNonNullableOpenFieldType(index.getKeyFieldTypes().get(0),
+                secondaryKeyFields.get(0), sourceType);
+        IAType spatialType = spatialTypePair.first;
+        if (spatialType == null) {
+            throw new AsterixException("Could not find field " + secondaryKeyFields.get(0) + " in the schema.");
+        }
+        IAType nestedKeyType = NonTaggedFormatUtil.getNestedSpatialType(spatialType.getTypeTag());
+        int numDimensions = NonTaggedFormatUtil.getNumDimensions(spatialType.getTypeTag());
+        int numNestedSecondaryKeyFields = numDimensions * 2;
+        IBinaryComparatorFactory[] secondaryComparatorFactories =
+                new IBinaryComparatorFactory[numNestedSecondaryKeyFields];
+        for (int i = 0; i < numNestedSecondaryKeyFields; i++) {
+            secondaryComparatorFactories[i] = cmpFactoryProvider.getBinaryComparatorFactory(nestedKeyType, true);
+        }
+        return secondaryComparatorFactories;
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/735532e4/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryBTreeOperationsHelper.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryBTreeOperationsHelper.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryBTreeOperationsHelper.java
index faea7fd..1f7914d 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryBTreeOperationsHelper.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryBTreeOperationsHelper.java
@@ -22,7 +22,6 @@ import java.util.List;
 
 import org.apache.asterix.common.config.DatasetConfig.DatasetType;
 import org.apache.asterix.common.config.GlobalConfig;
-import org.apache.asterix.common.context.IStorageComponentProvider;
 import org.apache.asterix.external.indexing.IndexingConstants;
 import org.apache.asterix.external.operators.ExternalScanOperatorDescriptor;
 import org.apache.asterix.metadata.declared.MetadataProvider;
@@ -31,10 +30,6 @@ import org.apache.asterix.metadata.entities.Index;
 import org.apache.asterix.om.types.ARecordType;
 import org.apache.asterix.om.types.IAType;
 import org.apache.asterix.runtime.utils.RuntimeUtils;
-import org.apache.asterix.transaction.management.resource.ExternalBTreeWithBuddyLocalResourceMetadataFactory;
-import org.apache.asterix.transaction.management.resource.LSMBTreeLocalResourceMetadataFactory;
-import org.apache.asterix.transaction.management.resource.PersistentLocalResourceFactoryProvider;
-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.core.jobgen.impl.ConnectorPolicyAssignmentPolicy;
@@ -53,16 +48,13 @@ import org.apache.hyracks.api.dataflow.value.ITypeTraits;
 import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
 import org.apache.hyracks.api.job.JobSpecification;
 import org.apache.hyracks.dataflow.std.base.AbstractOperatorDescriptor;
+import org.apache.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
 import org.apache.hyracks.dataflow.std.connectors.OneToOneConnectorDescriptor;
 import org.apache.hyracks.dataflow.std.sort.ExternalSortOperatorDescriptor;
 import org.apache.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
-import org.apache.hyracks.storage.am.common.dataflow.AbstractTreeIndexOperatorDescriptor;
 import org.apache.hyracks.storage.am.common.dataflow.IIndexDataflowHelperFactory;
+import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory;
 import org.apache.hyracks.storage.am.common.dataflow.TreeIndexBulkLoadOperatorDescriptor;
-import org.apache.hyracks.storage.am.common.dataflow.TreeIndexCreateOperatorDescriptor;
-import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
-import org.apache.hyracks.storage.common.file.ILocalResourceFactoryProvider;
-import org.apache.hyracks.storage.common.file.LocalResource;
 
 public class SecondaryBTreeOperationsHelper extends SecondaryTreeIndexOperationsHelper {
 
@@ -73,57 +65,12 @@ public class SecondaryBTreeOperationsHelper extends SecondaryTreeIndexOperations
     }
 
     @Override
-    public JobSpecification buildCreationJobSpec() throws AlgebricksException {
-        JobSpecification spec = RuntimeUtils.createJobSpecification(metadataProvider.getApplicationContext());
-        ILocalResourceFactoryProvider localResourceFactoryProvider;
-        IIndexDataflowHelperFactory indexDataflowHelperFactory = dataset.getIndexDataflowHelperFactory(metadataProvider,
-                index, itemType, metaType, mergePolicyFactory, mergePolicyFactoryProperties);
-        IStorageComponentProvider storageComponentProvider = metadataProvider.getStorageComponentProvider();
-        if (dataset.getDatasetType() == DatasetType.INTERNAL) {
-            //prepare a LocalResourceMetadata which will be stored in NC's local resource repository
-            LSMBTreeLocalResourceMetadataFactory localResourceMetadata = new LSMBTreeLocalResourceMetadataFactory(
-                    secondaryTypeTraits, secondaryComparatorFactories, secondaryBloomFilterKeyFields, false,
-                    dataset.getDatasetId(), mergePolicyFactory, mergePolicyFactoryProperties, filterTypeTraits,
-                    filterCmpFactories, secondaryBTreeFields, secondaryFilterFields,
-                    dataset.getIndexOperationTrackerFactory(index), dataset.getIoOperationCallbackFactory(index),
-                    storageComponentProvider.getMetadataPageManagerFactory());
-            localResourceFactoryProvider =
-                    new PersistentLocalResourceFactoryProvider(localResourceMetadata, LocalResource.LSMBTreeResource);
-        } else {
-            // External dataset local resource and dataflow helper
-            int[] buddyBreeFields = new int[] { index.getKeyFieldNames().size() };
-            ExternalBTreeWithBuddyLocalResourceMetadataFactory localResourceMetadata =
-                    new ExternalBTreeWithBuddyLocalResourceMetadataFactory(dataset.getDatasetId(),
-                            secondaryComparatorFactories, secondaryTypeTraits, mergePolicyFactory,
-                            mergePolicyFactoryProperties, buddyBreeFields,
-                            dataset.getIndexOperationTrackerFactory(index),
-                            dataset.getIoOperationCallbackFactory(index),
-                            storageComponentProvider.getMetadataPageManagerFactory());
-            localResourceFactoryProvider = new PersistentLocalResourceFactoryProvider(localResourceMetadata,
-                    LocalResource.ExternalBTreeWithBuddyResource);
-        }
-        TreeIndexCreateOperatorDescriptor secondaryIndexCreateOp =
-                new TreeIndexCreateOperatorDescriptor(spec, storageComponentProvider.getStorageManager(),
-                        storageComponentProvider.getIndexLifecycleManagerProvider(), secondaryFileSplitProvider,
-                        secondaryTypeTraits, secondaryComparatorFactories, secondaryBloomFilterKeyFields,
-                        indexDataflowHelperFactory,
-                        localResourceFactoryProvider, dataset.getModificationCallbackFactory(storageComponentProvider,
-                                index, null, IndexOperation.CREATE, null),
-                        storageComponentProvider.getMetadataPageManagerFactory());
-        AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, secondaryIndexCreateOp,
-                secondaryPartitionConstraint);
-        spec.addRoot(secondaryIndexCreateOp);
-        spec.setConnectorPolicyAssignmentPolicy(new ConnectorPolicyAssignmentPolicy());
-        return spec;
-    }
-
-    @Override
     public JobSpecification buildLoadingJobSpec() throws AlgebricksException {
         JobSpecification spec = RuntimeUtils.createJobSpecification(metadataProvider.getApplicationContext());
         boolean isEnforcingKeyTypes = index.isEnforcingKeyFileds();
         int[] fieldPermutation = createFieldPermutationForBulkLoadOp(index.getKeyFieldNames().size());
-        IIndexDataflowHelperFactory dataflowHelperFactory = dataset.getIndexDataflowHelperFactory(metadataProvider,
-                index, itemType, metaType, mergePolicyFactory, mergePolicyFactoryProperties);
+        IIndexDataflowHelperFactory dataflowHelperFactory = new IndexDataflowHelperFactory(
+                metadataProvider.getStorageComponentProvider().getStorageManager(), secondaryFileSplitProvider);
         if (dataset.getDatasetType() == DatasetType.EXTERNAL) {
             /*
              * In case of external data,
@@ -152,23 +99,22 @@ public class SecondaryBTreeOperationsHelper extends SecondaryTreeIndexOperations
             // Sort by secondary keys.
             ExternalSortOperatorDescriptor sortOp = createSortOp(spec, secondaryComparatorFactories, secondaryRecDesc);
             // Create secondary BTree bulk load op.
-            AbstractTreeIndexOperatorDescriptor secondaryBulkLoadOp;
+            AbstractSingleActivityOperatorDescriptor secondaryBulkLoadOp;
             IOperatorDescriptor root;
             if (externalFiles != null) {
                 // Transaction load
                 secondaryBulkLoadOp = createExternalIndexBulkModifyOp(spec, fieldPermutation, dataflowHelperFactory,
                         GlobalConfig.DEFAULT_TREE_FILL_FACTOR);
-                root = secondaryBulkLoadOp;
             } else {
                 // Initial load
-                secondaryBulkLoadOp = createTreeIndexBulkLoadOp(spec, fieldPermutation, dataflowHelperFactory,
+                secondaryBulkLoadOp = createExternalIndexBulkLoadOp(spec, fieldPermutation, dataflowHelperFactory,
                         GlobalConfig.DEFAULT_TREE_FILL_FACTOR);
-                AlgebricksMetaOperatorDescriptor metaOp = new AlgebricksMetaOperatorDescriptor(spec, 1, 0,
-                        new IPushRuntimeFactory[] { new SinkRuntimeFactory() },
-                        new RecordDescriptor[] { secondaryRecDesc });
-                spec.connect(new OneToOneConnectorDescriptor(spec), secondaryBulkLoadOp, 0, metaOp, 0);
-                root = metaOp;
             }
+            AlgebricksMetaOperatorDescriptor metaOp = new AlgebricksMetaOperatorDescriptor(spec, 1, 0,
+                    new IPushRuntimeFactory[] { new SinkRuntimeFactory() },
+                    new RecordDescriptor[] { secondaryRecDesc });
+            spec.connect(new OneToOneConnectorDescriptor(spec), secondaryBulkLoadOp, 0, metaOp, 0);
+            root = metaOp;
             spec.connect(new OneToOneConnectorDescriptor(spec), sourceOp, 0, asterixAssignOp, 0);
             if (anySecondaryKeyIsNullable || isEnforcingKeyTypes) {
                 spec.connect(new OneToOneConnectorDescriptor(spec), asterixAssignOp, 0, selectOp, 0);

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/735532e4/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryIndexOperationsHelper.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryIndexOperationsHelper.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryIndexOperationsHelper.java
index e5b5b9f..fa519e1 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryIndexOperationsHelper.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryIndexOperationsHelper.java
@@ -34,17 +34,18 @@ import org.apache.asterix.common.transactions.IRecoveryManager.ResourceType;
 import org.apache.asterix.common.transactions.JobId;
 import org.apache.asterix.external.indexing.ExternalFile;
 import org.apache.asterix.external.indexing.IndexingConstants;
+import org.apache.asterix.external.operators.ExternalIndexBulkLoadOperatorDescriptor;
 import org.apache.asterix.external.operators.ExternalIndexBulkModifyOperatorDescriptor;
 import org.apache.asterix.external.operators.ExternalScanOperatorDescriptor;
 import org.apache.asterix.formats.nontagged.BinaryBooleanInspector;
 import org.apache.asterix.formats.nontagged.BinaryComparatorFactoryProvider;
 import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
 import org.apache.asterix.formats.nontagged.TypeTraitProvider;
-import org.apache.asterix.metadata.MetadataManager;
 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.entities.InternalDatasetDetails;
+import org.apache.asterix.metadata.lock.ExternalDatasetsRegistry;
 import org.apache.asterix.om.types.ARecordType;
 import org.apache.asterix.om.types.IAType;
 import org.apache.asterix.runtime.evaluators.functions.AndDescriptor;
@@ -52,7 +53,6 @@ import org.apache.asterix.runtime.evaluators.functions.CastTypeDescriptor;
 import org.apache.asterix.runtime.evaluators.functions.IsUnknownDescriptor;
 import org.apache.asterix.runtime.evaluators.functions.NotDescriptor;
 import org.apache.asterix.runtime.job.listener.JobEventListenerFactory;
-import org.apache.asterix.runtime.utils.RuntimeComponentsProvider;
 import org.apache.asterix.transaction.management.opcallbacks.PrimaryIndexInstantSearchOperationCallbackFactory;
 import org.apache.asterix.transaction.management.service.transaction.JobIdFactory;
 import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
@@ -83,6 +83,7 @@ import org.apache.hyracks.dataflow.std.sort.ExternalSortOperatorDescriptor;
 import org.apache.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
 import org.apache.hyracks.storage.am.common.api.ISearchOperationCallbackFactory;
 import org.apache.hyracks.storage.am.common.dataflow.IIndexDataflowHelperFactory;
+import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory;
 import org.apache.hyracks.storage.am.common.dataflow.TreeIndexBulkLoadOperatorDescriptor;
 import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallbackFactory;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
@@ -117,7 +118,7 @@ public abstract class SecondaryIndexOperationsHelper {
     protected RecordDescriptor secondaryRecDesc;
     protected IScalarEvaluatorFactory[] secondaryFieldAccessEvalFactories;
     protected ILSMMergePolicyFactory mergePolicyFactory;
-    protected Map<String, String> mergePolicyFactoryProperties;
+    protected Map<String, String> mergePolicyProperties;
     protected RecordDescriptor enforcedRecDesc;
     protected int numFilterFields;
     protected List<String> filterFieldName;
@@ -132,8 +133,8 @@ public abstract class SecondaryIndexOperationsHelper {
 
     // Prevent public construction. Should be created via createIndexCreator().
     protected SecondaryIndexOperationsHelper(Dataset dataset, Index index, PhysicalOptimizationConfig physOptConf,
-            MetadataProvider metadataProvider, ARecordType recType,
-            ARecordType metaType, ARecordType enforcedType, ARecordType enforcedMetaType) {
+            MetadataProvider metadataProvider, ARecordType recType, ARecordType metaType, ARecordType enforcedType,
+            ARecordType enforcedMetaType) {
         this.dataset = dataset;
         this.index = index;
         this.physOptConf = physOptConf;
@@ -150,14 +151,12 @@ public abstract class SecondaryIndexOperationsHelper {
         SecondaryIndexOperationsHelper indexOperationsHelper;
         switch (index.getIndexType()) {
             case BTREE:
-                indexOperationsHelper =
-                        new SecondaryBTreeOperationsHelper(dataset, index, physOptConf,
-                                metadataProvider, recType, metaType, enforcedType, enforcedMetaType);
+                indexOperationsHelper = new SecondaryBTreeOperationsHelper(dataset, index, physOptConf,
+                        metadataProvider, recType, metaType, enforcedType, enforcedMetaType);
                 break;
             case RTREE:
-                indexOperationsHelper =
-                        new SecondaryRTreeOperationsHelper(dataset, index, physOptConf,
-                                metadataProvider, recType, metaType, enforcedType, enforcedMetaType);
+                indexOperationsHelper = new SecondaryRTreeOperationsHelper(dataset, index, physOptConf,
+                        metadataProvider, recType, metaType, enforcedType, enforcedMetaType);
                 break;
             case SINGLE_PARTITION_WORD_INVIX:
             case SINGLE_PARTITION_NGRAM_INVIX:
@@ -189,7 +188,7 @@ public abstract class SecondaryIndexOperationsHelper {
                 metadataProvider.getSplitProviderAndConstraints(dataset, index.getIndexName());
         secondaryFileSplitProvider = secondarySplitsAndConstraint.first;
         secondaryPartitionConstraint = secondarySplitsAndConstraint.second;
-        numPrimaryKeys = DatasetUtil.getPartitioningKeys(dataset).size();
+        numPrimaryKeys = dataset.getPrimaryKeys().size();
         if (dataset.getDatasetType() == DatasetType.INTERNAL) {
             filterFieldName = DatasetUtil.getFilterField(dataset);
             if (filterFieldName != null) {
@@ -208,7 +207,7 @@ public abstract class SecondaryIndexOperationsHelper {
         Pair<ILSMMergePolicyFactory, Map<String, String>> compactionInfo =
                 DatasetUtil.getMergePolicyFactory(dataset, metadataProvider.getMetadataTxnContext());
         mergePolicyFactory = compactionInfo.first;
-        mergePolicyFactoryProperties = compactionInfo.second;
+        mergePolicyProperties = compactionInfo.second;
         if (numFilterFields > 0) {
             setFilterTypeTraitsAndComparators();
         }
@@ -238,7 +237,7 @@ public abstract class SecondaryIndexOperationsHelper {
     protected abstract int getNumSecondaryKeys();
 
     protected void setPrimaryRecDescAndComparators() throws AlgebricksException {
-        List<List<String>> partitioningKeys = DatasetUtil.getPartitioningKeys(dataset);
+        List<List<String>> partitioningKeys = dataset.getPrimaryKeys();
         ISerializerDeserializer[] primaryRecFields =
                 new ISerializerDeserializer[numPrimaryKeys + 1 + (dataset.hasMetaPart() ? 1 : 0)];
         ITypeTraits[] primaryTypeTraits = new ITypeTraits[numPrimaryKeys + 1 + (dataset.hasMetaPart() ? 1 : 0)];
@@ -292,8 +291,7 @@ public abstract class SecondaryIndexOperationsHelper {
         return keyProviderOp;
     }
 
-    protected BTreeSearchOperatorDescriptor createPrimaryIndexScanOp(JobSpecification spec)
-            throws AlgebricksException {
+    protected BTreeSearchOperatorDescriptor createPrimaryIndexScanOp(JobSpecification spec) throws AlgebricksException {
         // -Infinity
         int[] lowKeyFields = null;
         // +Infinity
@@ -304,22 +302,15 @@ public abstract class SecondaryIndexOperationsHelper {
         boolean isWriteTransaction = metadataProvider.isWriteTransaction();
         IJobletEventListenerFactory jobEventListenerFactory = new JobEventListenerFactory(jobId, isWriteTransaction);
         spec.setJobletEventListenerFactory(jobEventListenerFactory);
-        Index primaryIndex = MetadataManager.INSTANCE.getIndex(metadataProvider.getMetadataTxnContext(),
-                dataset.getDataverseName(), dataset.getDatasetName(), dataset.getDatasetName());
-
         boolean temp = dataset.getDatasetDetails().isTemp();
         ISearchOperationCallbackFactory searchCallbackFactory = temp ? NoOpOperationCallbackFactory.INSTANCE
                 : new PrimaryIndexInstantSearchOperationCallbackFactory(jobId, dataset.getDatasetId(),
                         primaryBloomFilterKeyFields, txnSubsystemProvider, ResourceType.LSM_BTREE);
-        BTreeSearchOperatorDescriptor primarySearchOp = new BTreeSearchOperatorDescriptor(spec, primaryRecDesc,
-                RuntimeComponentsProvider.RUNTIME_PROVIDER, RuntimeComponentsProvider.RUNTIME_PROVIDER,
-                primaryFileSplitProvider, primaryRecDesc.getTypeTraits(), primaryComparatorFactories,
-                primaryBloomFilterKeyFields, lowKeyFields, highKeyFields, true, true,
-                dataset.getIndexDataflowHelperFactory(metadataProvider, primaryIndex, itemType, metaType,
-                        mergePolicyFactory, mergePolicyFactoryProperties),
-                false, false, null, searchCallbackFactory, null, null,
-                metadataProvider.getStorageComponentProvider().getMetadataPageManagerFactory());
-
+        IndexDataflowHelperFactory indexHelperFactory = new IndexDataflowHelperFactory(
+                metadataProvider.getStorageComponentProvider().getStorageManager(), primaryFileSplitProvider);
+        BTreeSearchOperatorDescriptor primarySearchOp =
+                new BTreeSearchOperatorDescriptor(spec, primaryRecDesc, lowKeyFields, highKeyFields, true, true,
+                        indexHelperFactory, false, false, null, searchCallbackFactory, null, null, false);
         AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, primarySearchOp,
                 primaryPartitionConstraint);
         return primarySearchOp;
@@ -401,11 +392,18 @@ public abstract class SecondaryIndexOperationsHelper {
             int[] fieldPermutation, IIndexDataflowHelperFactory dataflowHelperFactory, float fillFactor)
             throws AlgebricksException {
         TreeIndexBulkLoadOperatorDescriptor treeIndexBulkLoadOp = new TreeIndexBulkLoadOperatorDescriptor(spec,
-                secondaryRecDesc, RuntimeComponentsProvider.RUNTIME_PROVIDER,
-                RuntimeComponentsProvider.RUNTIME_PROVIDER, secondaryFileSplitProvider,
-                secondaryRecDesc.getTypeTraits(), secondaryComparatorFactories, secondaryBloomFilterKeyFields,
-                fieldPermutation, fillFactor, false, numElementsHint, false, dataflowHelperFactory,
-                metadataProvider.getStorageComponentProvider().getMetadataPageManagerFactory());
+                secondaryRecDesc, fieldPermutation, fillFactor, false, numElementsHint, false, dataflowHelperFactory);
+        AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, treeIndexBulkLoadOp,
+                secondaryPartitionConstraint);
+        return treeIndexBulkLoadOp;
+    }
+
+    protected TreeIndexBulkLoadOperatorDescriptor createExternalIndexBulkLoadOp(JobSpecification spec,
+            int[] fieldPermutation, IIndexDataflowHelperFactory dataflowHelperFactory, float fillFactor)
+            throws AlgebricksException {
+        ExternalIndexBulkLoadOperatorDescriptor treeIndexBulkLoadOp = new ExternalIndexBulkLoadOperatorDescriptor(spec,
+                secondaryRecDesc, fieldPermutation, fillFactor, false, numElementsHint, false, dataflowHelperFactory,
+                ExternalDatasetsRegistry.INSTANCE.getAndLockDatasetVersion(dataset, metadataProvider));
         AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, treeIndexBulkLoadOp,
                 secondaryPartitionConstraint);
         return treeIndexBulkLoadOp;
@@ -517,11 +515,7 @@ public abstract class SecondaryIndexOperationsHelper {
             }
         }
         ExternalIndexBulkModifyOperatorDescriptor treeIndexBulkLoadOp = new ExternalIndexBulkModifyOperatorDescriptor(
-                spec, RuntimeComponentsProvider.RUNTIME_PROVIDER, RuntimeComponentsProvider.RUNTIME_PROVIDER,
-                secondaryFileSplitProvider, secondaryTypeTraits, secondaryComparatorFactories,
-                secondaryBloomFilterKeyFields, dataflowHelperFactory, NoOpOperationCallbackFactory.INSTANCE,
-                deletedFiles, fieldPermutation, fillFactor, numElementsHint,
-                metadataProvider.getStorageComponentProvider().getMetadataPageManagerFactory());
+                spec, dataflowHelperFactory, deletedFiles, fieldPermutation, fillFactor, false, numElementsHint, false);
         AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, treeIndexBulkLoadOp,
                 secondaryPartitionConstraint);
         return treeIndexBulkLoadOp;

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/735532e4/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryInvertedIndexOperationsHelper.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryInvertedIndexOperationsHelper.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryInvertedIndexOperationsHelper.java
index 985f8cc..1bb1377 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryInvertedIndexOperationsHelper.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryInvertedIndexOperationsHelper.java
@@ -18,14 +18,10 @@
  */
 package org.apache.asterix.metadata.utils;
 
-import java.util.Map;
-
 import org.apache.asterix.common.config.DatasetConfig.IndexType;
-import org.apache.asterix.common.context.IStorageComponentProvider;
-import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.common.config.GlobalConfig;
 import org.apache.asterix.common.exceptions.CompilationException;
 import org.apache.asterix.common.exceptions.ErrorCode;
-import org.apache.asterix.common.transactions.IResourceFactory;
 import org.apache.asterix.metadata.declared.MetadataProvider;
 import org.apache.asterix.metadata.entities.Dataset;
 import org.apache.asterix.metadata.entities.Index;
@@ -35,9 +31,6 @@ import org.apache.asterix.om.utils.NonTaggedFormatUtil;
 import org.apache.asterix.om.utils.RecordUtil;
 import org.apache.asterix.runtime.formats.FormatUtils;
 import org.apache.asterix.runtime.utils.RuntimeUtils;
-import org.apache.asterix.transaction.management.resource.LSMInvertedIndexLocalResourceMetadataFactory;
-import org.apache.asterix.transaction.management.resource.PersistentLocalResourceFactoryProvider;
-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;
@@ -58,23 +51,16 @@ import org.apache.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
 import org.apache.hyracks.data.std.primitive.ShortPointable;
 import org.apache.hyracks.dataflow.common.data.marshalling.ShortSerializerDeserializer;
 import org.apache.hyracks.dataflow.std.base.AbstractOperatorDescriptor;
+import org.apache.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
 import org.apache.hyracks.dataflow.std.connectors.OneToOneConnectorDescriptor;
-import org.apache.hyracks.dataflow.std.file.IFileSplitProvider;
 import org.apache.hyracks.dataflow.std.sort.ExternalSortOperatorDescriptor;
 import org.apache.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
 import org.apache.hyracks.storage.am.common.dataflow.IIndexDataflowHelperFactory;
-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.common.dataflow.IndexDataflowHelperFactory;
 import org.apache.hyracks.storage.am.lsm.invertedindex.dataflow.BinaryTokenizerOperatorDescriptor;
-import org.apache.hyracks.storage.am.lsm.invertedindex.dataflow.LSMInvertedIndexBulkLoadOperatorDescriptor;
-import org.apache.hyracks.storage.am.lsm.invertedindex.dataflow.LSMInvertedIndexCompactOperator;
-import org.apache.hyracks.storage.am.lsm.invertedindex.dataflow.LSMInvertedIndexCreateOperatorDescriptor;
-import org.apache.hyracks.storage.am.lsm.invertedindex.dataflow.LSMInvertedIndexDropOperatorDescriptor;
 import org.apache.hyracks.storage.am.lsm.invertedindex.tokenizers.IBinaryTokenizerFactory;
-import org.apache.hyracks.storage.common.file.ILocalResourceFactoryProvider;
-import org.apache.hyracks.storage.common.file.LocalResource;
 
-public class SecondaryInvertedIndexOperationsHelper extends SecondaryIndexOperationsHelper {
+public class SecondaryInvertedIndexOperationsHelper extends SecondaryTreeIndexOperationsHelper {
 
     private IAType secondaryKeyType;
     private ITypeTraits[] invListsTypeTraits;
@@ -219,36 +205,6 @@ public class SecondaryInvertedIndexOperationsHelper extends SecondaryIndexOperat
     }
 
     @Override
-    public JobSpecification buildCreationJobSpec() throws AlgebricksException {
-        JobSpecification spec = RuntimeUtils.createJobSpecification(metadataProvider.getApplicationContext());
-        IStorageComponentProvider storageComponentProvider = metadataProvider.getStorageComponentProvider();
-        //prepare a LocalResourceMetadata which will be stored in NC's local resource repository
-        IResourceFactory localResourceMetadata = new LSMInvertedIndexLocalResourceMetadataFactory(invListsTypeTraits,
-                primaryComparatorFactories, tokenTypeTraits, tokenComparatorFactories, tokenizerFactory, isPartitioned,
-                dataset.getDatasetId(), mergePolicyFactory, mergePolicyFactoryProperties, filterTypeTraits,
-                filterCmpFactories, invertedIndexFields, secondaryFilterFields, secondaryFilterFieldsForNonBulkLoadOps,
-                invertedIndexFieldsForNonBulkLoadOps, dataset.getIndexOperationTrackerFactory(index),
-                dataset.getIoOperationCallbackFactory(index), storageComponentProvider.getMetadataPageManagerFactory());
-        ILocalResourceFactoryProvider localResourceFactoryProvider = new PersistentLocalResourceFactoryProvider(
-                localResourceMetadata, LocalResource.LSMInvertedIndexResource);
-
-        IIndexDataflowHelperFactory dataflowHelperFactory = createDataflowHelperFactory();
-        LSMInvertedIndexCreateOperatorDescriptor invIndexCreateOp =
-                new LSMInvertedIndexCreateOperatorDescriptor(spec, storageComponentProvider.getStorageManager(),
-                        secondaryFileSplitProvider, storageComponentProvider.getIndexLifecycleManagerProvider(),
-                        tokenTypeTraits, tokenComparatorFactories, invListsTypeTraits, primaryComparatorFactories,
-                        tokenizerFactory, dataflowHelperFactory,
-                        localResourceFactoryProvider, dataset.getModificationCallbackFactory(storageComponentProvider,
-                                index, null, IndexOperation.CREATE, null),
-                        storageComponentProvider.getMetadataPageManagerFactory());
-        AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, invIndexCreateOp,
-                secondaryPartitionConstraint);
-        spec.addRoot(invIndexCreateOp);
-        spec.setConnectorPolicyAssignmentPolicy(new ConnectorPolicyAssignmentPolicy());
-        return spec;
-    }
-
-    @Override
     public JobSpecification buildLoadingJobSpec() throws AlgebricksException {
         JobSpecification spec = RuntimeUtils.createJobSpecification(metadataProvider.getApplicationContext());
 
@@ -282,7 +238,9 @@ public class SecondaryInvertedIndexOperationsHelper extends SecondaryIndexOperat
                 createSortOp(spec, tokenKeyPairComparatorFactories, tokenKeyPairRecDesc);
 
         // Create secondary inverted index bulk load op.
-        LSMInvertedIndexBulkLoadOperatorDescriptor invIndexBulkLoadOp = createInvertedIndexBulkLoadOp(spec);
+        AbstractSingleActivityOperatorDescriptor invIndexBulkLoadOp = createInvertedIndexBulkLoadOp(spec);
+        AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, invIndexBulkLoadOp,
+                secondaryPartitionConstraint);
 
         AlgebricksMetaOperatorDescriptor metaOp = new AlgebricksMetaOperatorDescriptor(spec, 1, 0,
                 new IPushRuntimeFactory[] { new SinkRuntimeFactory() }, new RecordDescriptor[] {});
@@ -331,71 +289,15 @@ public class SecondaryInvertedIndexOperationsHelper extends SecondaryIndexOperat
         return sortOp;
     }
 
-    private LSMInvertedIndexBulkLoadOperatorDescriptor createInvertedIndexBulkLoadOp(JobSpecification spec)
+    private AbstractSingleActivityOperatorDescriptor createInvertedIndexBulkLoadOp(JobSpecification spec)
             throws AlgebricksException {
         int[] fieldPermutation = new int[numTokenKeyPairFields + numFilterFields];
         for (int i = 0; i < fieldPermutation.length; i++) {
             fieldPermutation[i] = i;
         }
-        IIndexDataflowHelperFactory dataflowHelperFactory = createDataflowHelperFactory();
-        IStorageComponentProvider storageComponentProvider = metadataProvider.getStorageComponentProvider();
-        LSMInvertedIndexBulkLoadOperatorDescriptor invIndexBulkLoadOp = new LSMInvertedIndexBulkLoadOperatorDescriptor(
-                spec, secondaryRecDesc, fieldPermutation, false, numElementsHint, false,
-                storageComponentProvider.getStorageManager(), secondaryFileSplitProvider,
-                storageComponentProvider.getIndexLifecycleManagerProvider(), tokenTypeTraits, tokenComparatorFactories,
-                invListsTypeTraits, primaryComparatorFactories, tokenizerFactory, dataflowHelperFactory,
-                storageComponentProvider.getMetadataPageManagerFactory());
-        AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, invIndexBulkLoadOp,
-                secondaryPartitionConstraint);
-        return invIndexBulkLoadOp;
-    }
-
-    private IIndexDataflowHelperFactory createDataflowHelperFactory() throws AlgebricksException {
-        return dataset.getIndexDataflowHelperFactory(metadataProvider, index, itemType, metaType, mergePolicyFactory,
-                mergePolicyFactoryProperties);
-    }
-
-    @Override
-    public JobSpecification buildCompactJobSpec() throws AsterixException, AlgebricksException {
-        JobSpecification spec = RuntimeUtils.createJobSpecification(metadataProvider.getApplicationContext());
-        IIndexDataflowHelperFactory dataflowHelperFactory = createDataflowHelperFactory();
-        IStorageComponentProvider storageComponentProvider = metadataProvider.getStorageComponentProvider();
-        LSMInvertedIndexCompactOperator compactOp =
-                new LSMInvertedIndexCompactOperator(spec, storageComponentProvider.getStorageManager(),
-                        secondaryFileSplitProvider, storageComponentProvider.getIndexLifecycleManagerProvider(),
-                        tokenTypeTraits, tokenComparatorFactories, invListsTypeTraits, primaryComparatorFactories,
-                        tokenizerFactory, dataflowHelperFactory,
-                        dataset.getModificationCallbackFactory(storageComponentProvider, index, null,
-                                IndexOperation.FULL_MERGE, null),
-                        storageComponentProvider.getMetadataPageManagerFactory());
-        AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, compactOp,
-                secondaryPartitionConstraint);
-
-        spec.addRoot(compactOp);
-        spec.setConnectorPolicyAssignmentPolicy(new ConnectorPolicyAssignmentPolicy());
-        return spec;
-    }
-
-    @Override
-    public JobSpecification buildDropJobSpec() throws AlgebricksException {
-        JobSpecification spec = RuntimeUtils.createJobSpecification(metadataProvider.getApplicationContext());
-        IStorageComponentProvider storageComponentProvider = metadataProvider.getStorageComponentProvider();
-        Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint =
-                metadataProvider.getSplitProviderAndConstraints(dataset, index.getIndexName());
-        Pair<ILSMMergePolicyFactory, Map<String, String>> compactionInfo =
-                DatasetUtil.getMergePolicyFactory(dataset, metadataProvider.getMetadataTxnContext());
-        ARecordType recordType =
-                (ARecordType) metadataProvider.findType(dataset.getItemTypeDataverseName(), dataset.getItemTypeName());
-        ARecordType metaType = DatasetUtil.getMetaType(metadataProvider, dataset);
-        IIndexDataflowHelperFactory dataflowHelperFactory = dataset.getIndexDataflowHelperFactory(metadataProvider,
-                index, recordType, metaType, compactionInfo.first, compactionInfo.second);
-        LSMInvertedIndexDropOperatorDescriptor invIdxDrop =
-                new LSMInvertedIndexDropOperatorDescriptor(spec, storageComponentProvider.getStorageManager(),
-                        storageComponentProvider.getIndexLifecycleManagerProvider(), splitsAndConstraint.first,
-                        dataflowHelperFactory, storageComponentProvider.getMetadataPageManagerFactory());
-        AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, invIdxDrop,
-                splitsAndConstraint.second);
-        spec.addRoot(invIdxDrop);
-        return spec;
+        IIndexDataflowHelperFactory dataflowHelperFactory = new IndexDataflowHelperFactory(
+                metadataProvider.getStorageComponentProvider().getStorageManager(), secondaryFileSplitProvider);
+        return createTreeIndexBulkLoadOp(spec, fieldPermutation, dataflowHelperFactory,
+                GlobalConfig.DEFAULT_TREE_FILL_FACTOR);
     }
 }


Mime
View raw message