You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by am...@apache.org on 2017/05/11 23:43:34 UTC
[19/25] asterixdb git commit: Separate index build from index access
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);
}
}