| /* |
| * Copyright 2009-2010 by The Regents of the University of California |
| * Licensed 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 from |
| * |
| * 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 edu.uci.ics.asterix.metadata.declared; |
| |
| import java.io.File; |
| import java.util.List; |
| |
| import edu.uci.ics.asterix.common.config.DatasetConfig.DatasetType; |
| import edu.uci.ics.asterix.common.config.GlobalConfig; |
| import edu.uci.ics.asterix.common.dataflow.IAsterixApplicationContextInfo; |
| import edu.uci.ics.asterix.common.parse.IParseFileSplitsDecl; |
| import edu.uci.ics.asterix.dataflow.data.nontagged.valueproviders.AqlPrimitiveValueProviderFactory; |
| import edu.uci.ics.asterix.external.data.adapter.api.IDatasourceAdapter; |
| import edu.uci.ics.asterix.external.data.adapter.api.IDatasourceReadAdapter; |
| import edu.uci.ics.asterix.external.data.operator.ExternalDataScanOperatorDescriptor; |
| import edu.uci.ics.asterix.feed.comm.IFeedMessage; |
| import edu.uci.ics.asterix.feed.mgmt.FeedId; |
| import edu.uci.ics.asterix.feed.operator.FeedIntakeOperatorDescriptor; |
| import edu.uci.ics.asterix.feed.operator.FeedMessageOperatorDescriptor; |
| import edu.uci.ics.asterix.formats.base.IDataFormat; |
| import edu.uci.ics.asterix.formats.nontagged.AqlBinaryComparatorFactoryProvider; |
| import edu.uci.ics.asterix.formats.nontagged.AqlSerializerDeserializerProvider; |
| import edu.uci.ics.asterix.formats.nontagged.AqlTypeTraitProvider; |
| import edu.uci.ics.asterix.metadata.declared.AqlCompiledIndexDecl.IndexKind; |
| import edu.uci.ics.asterix.metadata.utils.DatasetUtils; |
| import edu.uci.ics.asterix.om.functions.AsterixBuiltinFunctions; |
| import edu.uci.ics.asterix.om.types.ARecordType; |
| import edu.uci.ics.asterix.om.types.ATypeTag; |
| import edu.uci.ics.asterix.om.types.IAType; |
| import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil; |
| import edu.uci.ics.asterix.runtime.base.AsterixTupleFilterFactory; |
| import edu.uci.ics.asterix.runtime.transaction.TreeIndexInsertUpdateDeleteOperatorDescriptor; |
| import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint; |
| import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint; |
| import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException; |
| import edu.uci.ics.hyracks.algebricks.common.utils.Pair; |
| import edu.uci.ics.hyracks.algebricks.common.utils.Triple; |
| import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression; |
| import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable; |
| import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.ILogicalExpressionJobGen; |
| import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment; |
| import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression; |
| import edu.uci.ics.hyracks.algebricks.core.algebra.functions.FunctionIdentifier; |
| import edu.uci.ics.hyracks.algebricks.core.algebra.functions.IFunctionInfo; |
| import edu.uci.ics.hyracks.algebricks.core.algebra.metadata.IDataSink; |
| import edu.uci.ics.hyracks.algebricks.core.algebra.metadata.IDataSource; |
| import edu.uci.ics.hyracks.algebricks.core.algebra.metadata.IDataSourceIndex; |
| import edu.uci.ics.hyracks.algebricks.core.algebra.metadata.IMetadataProvider; |
| import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.IOperatorSchema; |
| import edu.uci.ics.hyracks.algebricks.core.jobgen.impl.JobGenContext; |
| import edu.uci.ics.hyracks.algebricks.data.IPrinterFactory; |
| import edu.uci.ics.hyracks.algebricks.runtime.base.IEvaluatorFactory; |
| import edu.uci.ics.hyracks.algebricks.runtime.base.IPushRuntimeFactory; |
| import edu.uci.ics.hyracks.algebricks.runtime.operators.std.SinkWriterRuntimeFactory; |
| import edu.uci.ics.hyracks.api.dataflow.IOperatorDescriptor; |
| import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory; |
| import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer; |
| import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits; |
| import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor; |
| import edu.uci.ics.hyracks.api.job.JobSpecification; |
| import edu.uci.ics.hyracks.dataflow.std.file.ConstantFileSplitProvider; |
| import edu.uci.ics.hyracks.dataflow.std.file.FileScanOperatorDescriptor; |
| import edu.uci.ics.hyracks.dataflow.std.file.FileSplit; |
| import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider; |
| import edu.uci.ics.hyracks.dataflow.std.file.ITupleParserFactory; |
| import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeDataflowHelperFactory; |
| import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor; |
| import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory; |
| import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory; |
| import edu.uci.ics.hyracks.storage.am.common.api.IPrimitiveValueProviderFactory; |
| import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory; |
| import edu.uci.ics.hyracks.storage.am.common.dataflow.TreeIndexBulkLoadOperatorDescriptor; |
| import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallbackProvider; |
| import edu.uci.ics.hyracks.storage.am.common.ophelpers.IndexOp; |
| import edu.uci.ics.hyracks.storage.am.common.tuples.TypeAwareTupleWriterFactory; |
| import edu.uci.ics.hyracks.storage.am.rtree.dataflow.RTreeDataflowHelperFactory; |
| import edu.uci.ics.hyracks.storage.am.rtree.dataflow.RTreeSearchOperatorDescriptor; |
| |
| public class AqlMetadataProvider implements IMetadataProvider<AqlSourceId, String> { |
| private final long txnId; |
| private boolean isWriteTransaction; |
| private final AqlCompiledMetadataDeclarations metadata; |
| |
| public AqlMetadataProvider(long txnId, boolean isWriteTransaction, AqlCompiledMetadataDeclarations metadata) { |
| this.txnId = txnId; |
| this.isWriteTransaction = isWriteTransaction; |
| this.metadata = metadata; |
| } |
| |
| @Override |
| public AqlDataSource findDataSource(AqlSourceId id) throws AlgebricksException { |
| AqlSourceId aqlId = (AqlSourceId) id; |
| return lookupSourceInMetadata(metadata, aqlId); |
| } |
| |
| public AqlCompiledMetadataDeclarations getMetadataDeclarations() { |
| return metadata; |
| } |
| |
| public boolean isWriteTransaction() { |
| return isWriteTransaction; |
| } |
| |
| @Override |
| public Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> getScannerRuntime( |
| IDataSource<AqlSourceId> dataSource, List<LogicalVariable> scanVariables, |
| List<LogicalVariable> projectVariables, boolean projectPushed, JobGenContext context, |
| JobSpecification jobSpec) throws AlgebricksException { |
| AqlCompiledDatasetDecl adecl = metadata.findDataset(dataSource.getId().getDatasetName()); |
| if (adecl == null) { |
| throw new AlgebricksException("Unknown dataset " + dataSource.getId().getDatasetName()); |
| } |
| switch (adecl.getDatasetType()) { |
| case FEED: |
| if (dataSource instanceof ExternalFeedDataSource) { |
| return buildExternalDatasetScan(jobSpec, adecl, dataSource); |
| } else { |
| return buildInternalDatasetScan(jobSpec, adecl, dataSource, context); |
| } |
| |
| case INTERNAL: { |
| return buildInternalDatasetScan(jobSpec, adecl, dataSource, context); |
| } |
| |
| case EXTERNAL: { |
| return buildExternalDatasetScan(jobSpec, adecl, dataSource); |
| } |
| |
| default: { |
| throw new IllegalArgumentException(); |
| } |
| } |
| } |
| |
| private Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> buildInternalDatasetScan(JobSpecification jobSpec, |
| AqlCompiledDatasetDecl acedl, IDataSource<AqlSourceId> dataSource, JobGenContext context) |
| throws AlgebricksException { |
| AqlSourceId asid = dataSource.getId(); |
| String datasetName = asid.getDatasetName(); |
| String indexName = DatasetUtils.getPrimaryIndex(acedl).getIndexName(); |
| |
| try { |
| return buildBtreeRuntime(metadata, context, jobSpec, datasetName, acedl, indexName, null, null, true, true); |
| } catch (AlgebricksException e) { |
| throw new AlgebricksException(e); |
| } |
| } |
| |
| private Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> buildExternalDatasetScan(JobSpecification jobSpec, |
| AqlCompiledDatasetDecl acedl, IDataSource<AqlSourceId> dataSource) throws AlgebricksException { |
| String itemTypeName = acedl.getItemTypeName(); |
| IAType itemType; |
| try { |
| itemType = metadata.findType(itemTypeName); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| |
| if (dataSource instanceof ExternalFeedDataSource) { |
| AqlCompiledFeedDatasetDetails acfdd = (AqlCompiledFeedDatasetDetails) ((ExternalFeedDataSource) dataSource) |
| .getCompiledDatasetDecl().getAqlCompiledDatasetDetails(); |
| |
| return buildFeedIntakeRuntime(jobSpec, metadata.getDataverseName(), acedl.getName(), itemType, acfdd, |
| metadata.getFormat()); |
| } else { |
| return buildExternalDataScannerRuntime(jobSpec, itemType, |
| (AqlCompiledExternalDatasetDetails) acedl.getAqlCompiledDatasetDetails(), metadata.getFormat()); |
| } |
| } |
| |
| @SuppressWarnings("rawtypes") |
| public static Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> buildExternalDataScannerRuntime( |
| JobSpecification jobSpec, IAType itemType, AqlCompiledExternalDatasetDetails decl, IDataFormat format) |
| throws AlgebricksException { |
| if (itemType.getTypeTag() != ATypeTag.RECORD) { |
| throw new AlgebricksException("Can only scan datasets of records."); |
| } |
| |
| IDatasourceReadAdapter adapter; |
| try { |
| adapter = (IDatasourceReadAdapter) Class.forName(decl.getAdapter()).newInstance(); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| throw new AlgebricksException("unable to load the adapter class " + e); |
| } |
| |
| if (!(adapter.getAdapterType().equals(IDatasourceAdapter.AdapterType.READ) || adapter.getAdapterType().equals( |
| IDatasourceAdapter.AdapterType.READ_WRITE))) { |
| throw new AlgebricksException("external dataset adapter does not support read operation"); |
| } |
| ARecordType rt = (ARecordType) itemType; |
| |
| try { |
| adapter.configure(decl.getProperties(), itemType); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| throw new AlgebricksException("unable to configure the datasource adapter " + e); |
| } |
| |
| ISerializerDeserializer payloadSerde = format.getSerdeProvider().getSerializerDeserializer(itemType); |
| RecordDescriptor scannerDesc = new RecordDescriptor(new ISerializerDeserializer[] { payloadSerde }); |
| |
| ExternalDataScanOperatorDescriptor dataScanner = new ExternalDataScanOperatorDescriptor(jobSpec, |
| decl.getAdapter(), decl.getProperties(), rt, scannerDesc); |
| dataScanner.setDatasourceAdapter(adapter); |
| AlgebricksPartitionConstraint constraint = adapter.getPartitionConstraint(); |
| return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(dataScanner, constraint); |
| } |
| |
| @SuppressWarnings("rawtypes") |
| public static Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> buildScannerRuntime( |
| JobSpecification jobSpec, IAType itemType, IParseFileSplitsDecl decl, IDataFormat format) |
| throws AlgebricksException { |
| if (itemType.getTypeTag() != ATypeTag.RECORD) { |
| throw new AlgebricksException("Can only scan datasets of records."); |
| } |
| ARecordType rt = (ARecordType) itemType; |
| ITupleParserFactory tupleParser = format.createTupleParser(rt, decl); |
| FileSplit[] splits = decl.getSplits(); |
| IFileSplitProvider scannerSplitProvider = new ConstantFileSplitProvider(splits); |
| ISerializerDeserializer payloadSerde = format.getSerdeProvider().getSerializerDeserializer(itemType); |
| RecordDescriptor scannerDesc = new RecordDescriptor(new ISerializerDeserializer[] { payloadSerde }); |
| IOperatorDescriptor scanner = new FileScanOperatorDescriptor(jobSpec, scannerSplitProvider, tupleParser, |
| scannerDesc); |
| String[] locs = new String[splits.length]; |
| for (int i = 0; i < splits.length; i++) { |
| locs[i] = splits[i].getNodeName(); |
| } |
| AlgebricksPartitionConstraint apc = new AlgebricksAbsolutePartitionConstraint(locs); |
| return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(scanner, apc); |
| } |
| |
| @SuppressWarnings("rawtypes") |
| public static Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> buildFeedIntakeRuntime( |
| JobSpecification jobSpec, String dataverse, String dataset, IAType itemType, |
| AqlCompiledFeedDatasetDetails decl, IDataFormat format) throws AlgebricksException { |
| if (itemType.getTypeTag() != ATypeTag.RECORD) { |
| throw new AlgebricksException("Can only consume records."); |
| } |
| IDatasourceAdapter adapter; |
| try { |
| adapter = (IDatasourceAdapter) Class.forName(decl.getAdapter()).newInstance(); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| throw new AlgebricksException("unable to load the adapter class " + e); |
| } |
| |
| ARecordType rt = (ARecordType) itemType; |
| try { |
| adapter.configure(decl.getProperties(), itemType); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| throw new AlgebricksException("unable to configure the datasource adapter " + e); |
| } |
| |
| ISerializerDeserializer payloadSerde = format.getSerdeProvider().getSerializerDeserializer(itemType); |
| RecordDescriptor feedDesc = new RecordDescriptor(new ISerializerDeserializer[] { payloadSerde }); |
| |
| FeedIntakeOperatorDescriptor feedIngestor = new FeedIntakeOperatorDescriptor(jobSpec, new FeedId(dataverse, |
| dataset), decl.getAdapter(), decl.getProperties(), rt, feedDesc); |
| |
| AlgebricksPartitionConstraint constraint = adapter.getPartitionConstraint(); |
| return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(feedIngestor, constraint); |
| } |
| |
| @SuppressWarnings("rawtypes") |
| public static Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> buildFeedMessengerRuntime( |
| JobSpecification jobSpec, AqlCompiledMetadataDeclarations metadata, AqlCompiledFeedDatasetDetails decl, |
| String dataverse, String dataset, List<IFeedMessage> feedMessages) throws AlgebricksException { |
| |
| Pair<IFileSplitProvider, AlgebricksPartitionConstraint> spPc; |
| try { |
| spPc = metadata.splitProviderAndPartitionConstraintsForInternalOrFeedDataset(dataset, dataset); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| |
| FeedMessageOperatorDescriptor feedMessenger = new FeedMessageOperatorDescriptor(jobSpec, dataverse, dataset, |
| feedMessages); |
| |
| return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(feedMessenger, spPc.second); |
| } |
| |
| @SuppressWarnings("rawtypes") |
| public static Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> buildBtreeRuntime( |
| AqlCompiledMetadataDeclarations metadata, JobGenContext context, JobSpecification jobSpec, |
| String datasetName, AqlCompiledDatasetDecl ddecl, String indexName, int[] lowKeyFields, |
| int[] highKeyFields, boolean lowKeyInclusive, boolean highKeyInclusive) throws AlgebricksException { |
| String itemTypeName = ddecl.getItemTypeName(); |
| IAType itemType; |
| try { |
| itemType = metadata.findType(itemTypeName); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| |
| boolean isSecondary = true; |
| AqlCompiledIndexDecl primIdxDecl = DatasetUtils.getPrimaryIndex(ddecl); |
| |
| if (primIdxDecl != null) { |
| isSecondary = !indexName.equals(primIdxDecl.getIndexName()); |
| } |
| |
| int numPrimaryKeys = DatasetUtils.getPartitioningFunctions(ddecl).size(); |
| ISerializerDeserializer[] recordFields; |
| IBinaryComparatorFactory[] comparatorFactories; |
| ITypeTraits[] typeTraits; |
| int numSecondaryKeys = 0; |
| int i = 0; |
| if (isSecondary) { |
| AqlCompiledIndexDecl cid = DatasetUtils.findSecondaryIndexByName(ddecl, indexName); |
| if (cid == null) { |
| throw new AlgebricksException("Code generation error: no index " + indexName + " for dataset " |
| + datasetName); |
| } |
| List<String> secondaryKeyFields = cid.getFieldExprs(); |
| numSecondaryKeys = secondaryKeyFields.size(); |
| int numKeys = numSecondaryKeys + numPrimaryKeys; |
| recordFields = new ISerializerDeserializer[numKeys]; |
| typeTraits = new ITypeTraits[numKeys]; |
| // comparatorFactories = new |
| // IBinaryComparatorFactory[numSecondaryKeys]; |
| comparatorFactories = new IBinaryComparatorFactory[numKeys]; |
| if (itemType.getTypeTag() != ATypeTag.RECORD) { |
| throw new AlgebricksException("Only record types can be indexed."); |
| } |
| ARecordType recType = (ARecordType) itemType; |
| for (i = 0; i < numSecondaryKeys; i++) { |
| Pair<IAType, Boolean> keyTypePair = AqlCompiledIndexDecl.getNonNullableKeyFieldType( |
| secondaryKeyFields.get(i), recType); |
| IAType keyType = keyTypePair.first; |
| ISerializerDeserializer keySerde = metadata.getFormat().getSerdeProvider() |
| .getSerializerDeserializer(keyType); |
| recordFields[i] = keySerde; |
| comparatorFactories[i] = AqlBinaryComparatorFactoryProvider.INSTANCE.getBinaryComparatorFactory( |
| keyType, true); |
| typeTraits[i] = AqlTypeTraitProvider.INSTANCE.getTypeTrait(keyType); |
| } |
| } else { |
| recordFields = new ISerializerDeserializer[numPrimaryKeys + 1]; |
| comparatorFactories = new IBinaryComparatorFactory[numPrimaryKeys]; |
| typeTraits = new ITypeTraits[numPrimaryKeys + 1]; |
| ISerializerDeserializer payloadSerde = metadata.getFormat().getSerdeProvider() |
| .getSerializerDeserializer(itemType); |
| recordFields[numPrimaryKeys] = payloadSerde; |
| typeTraits[numPrimaryKeys] = AqlTypeTraitProvider.INSTANCE.getTypeTrait(itemType); |
| } |
| |
| for (Triple<IEvaluatorFactory, ScalarFunctionCallExpression, IAType> evalFactoryAndType : DatasetUtils |
| .getPartitioningFunctions(ddecl)) { |
| IAType keyType = evalFactoryAndType.third; |
| ISerializerDeserializer keySerde = metadata.getFormat().getSerdeProvider() |
| .getSerializerDeserializer(keyType); |
| recordFields[i] = keySerde; |
| // if (!isSecondary) { |
| comparatorFactories[i] = AqlBinaryComparatorFactoryProvider.INSTANCE.getBinaryComparatorFactory(keyType, |
| true); |
| // } |
| typeTraits[i] = AqlTypeTraitProvider.INSTANCE.getTypeTrait(keyType); |
| ++i; |
| } |
| |
| IAsterixApplicationContextInfo appContext = (IAsterixApplicationContextInfo) context.getAppContext(); |
| RecordDescriptor recDesc = new RecordDescriptor(recordFields); |
| |
| Pair<IFileSplitProvider, AlgebricksPartitionConstraint> spPc; |
| try { |
| spPc = metadata.splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, indexName); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| |
| BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(jobSpec, recDesc, |
| appContext.getStorageManagerInterface(), appContext.getIndexRegistryProvider(), spPc.first, typeTraits, |
| comparatorFactories, lowKeyFields, highKeyFields, lowKeyInclusive, highKeyInclusive, |
| new BTreeDataflowHelperFactory(), NoOpOperationCallbackProvider.INSTANCE); |
| return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(btreeSearchOp, spPc.second); |
| } |
| |
| @SuppressWarnings("rawtypes") |
| public static Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> buildRtreeRuntime( |
| AqlCompiledMetadataDeclarations metadata, JobGenContext context, JobSpecification jobSpec, |
| String datasetName, AqlCompiledDatasetDecl ddecl, String indexName, int[] keyFields) |
| throws AlgebricksException { |
| String itemTypeName = ddecl.getItemTypeName(); |
| IAType itemType; |
| try { |
| itemType = metadata.findType(itemTypeName); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| |
| boolean isSecondary = true; |
| AqlCompiledIndexDecl primIdxDecl = DatasetUtils.getPrimaryIndex(ddecl); |
| if (primIdxDecl != null) { |
| isSecondary = !indexName.equals(primIdxDecl.getIndexName()); |
| } |
| |
| int numPrimaryKeys = DatasetUtils.getPartitioningFunctions(ddecl).size(); |
| ISerializerDeserializer[] recordFields; |
| IBinaryComparatorFactory[] comparatorFactories; |
| ITypeTraits[] typeTraits; |
| IPrimitiveValueProviderFactory[] valueProviderFactories; |
| int numSecondaryKeys = 0; |
| int numNestedSecondaryKeyFields = 0; |
| int i = 0; |
| if (isSecondary) { |
| AqlCompiledIndexDecl cid = DatasetUtils.findSecondaryIndexByName(ddecl, indexName); |
| if (cid == null) { |
| throw new AlgebricksException("Code generation error: no index " + indexName + " for dataset " |
| + datasetName); |
| } |
| List<String> secondaryKeyFields = cid.getFieldExprs(); |
| numSecondaryKeys = secondaryKeyFields.size(); |
| |
| if (numSecondaryKeys != 1) { |
| throw new AlgebricksException( |
| "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."); |
| } |
| |
| if (itemType.getTypeTag() != ATypeTag.RECORD) { |
| throw new AlgebricksException("Only record types can be indexed."); |
| } |
| ARecordType recType = (ARecordType) itemType; |
| |
| Pair<IAType, Boolean> keyTypePair = AqlCompiledIndexDecl.getNonNullableKeyFieldType( |
| secondaryKeyFields.get(0), recType); |
| IAType keyType = keyTypePair.first; |
| if (keyType == null) { |
| throw new AlgebricksException("Could not find field " + secondaryKeyFields.get(0) + " in the schema."); |
| } |
| |
| int dimension = NonTaggedFormatUtil.getNumDimensions(keyType.getTypeTag()); |
| numNestedSecondaryKeyFields = dimension * 2; |
| |
| int numFields = numNestedSecondaryKeyFields + numPrimaryKeys; |
| recordFields = new ISerializerDeserializer[numFields]; |
| typeTraits = new ITypeTraits[numFields]; |
| comparatorFactories = new IBinaryComparatorFactory[numNestedSecondaryKeyFields]; |
| valueProviderFactories = new IPrimitiveValueProviderFactory[numNestedSecondaryKeyFields]; |
| |
| IAType nestedKeyType = NonTaggedFormatUtil.getNestedSpatialType(keyType.getTypeTag()); |
| for (i = 0; i < numNestedSecondaryKeyFields; i++) { |
| ISerializerDeserializer keySerde = AqlSerializerDeserializerProvider.INSTANCE |
| .getSerializerDeserializer(nestedKeyType); |
| recordFields[i] = keySerde; |
| comparatorFactories[i] = AqlBinaryComparatorFactoryProvider.INSTANCE.getBinaryComparatorFactory( |
| nestedKeyType, true); |
| typeTraits[i] = AqlTypeTraitProvider.INSTANCE.getTypeTrait(nestedKeyType); |
| valueProviderFactories[i] = AqlPrimitiveValueProviderFactory.INSTANCE; |
| } |
| } else { |
| throw new AlgebricksException("R-tree can only be used as a secondary index"); |
| } |
| |
| for (Triple<IEvaluatorFactory, ScalarFunctionCallExpression, IAType> evalFactoryAndType : DatasetUtils |
| .getPartitioningFunctions(ddecl)) { |
| IAType keyType = evalFactoryAndType.third; |
| ISerializerDeserializer keySerde = AqlSerializerDeserializerProvider.INSTANCE |
| .getSerializerDeserializer(keyType); |
| recordFields[i] = keySerde; |
| typeTraits[i] = AqlTypeTraitProvider.INSTANCE.getTypeTrait(keyType); |
| ++i; |
| } |
| |
| IAsterixApplicationContextInfo appContext = (IAsterixApplicationContextInfo) context.getAppContext(); |
| RecordDescriptor recDesc = new RecordDescriptor(recordFields); |
| |
| Pair<IFileSplitProvider, AlgebricksPartitionConstraint> spPc; |
| try { |
| spPc = metadata.splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, indexName); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| |
| RTreeSearchOperatorDescriptor rtreeSearchOp = new RTreeSearchOperatorDescriptor(jobSpec, recDesc, |
| appContext.getStorageManagerInterface(), appContext.getIndexRegistryProvider(), spPc.first, typeTraits, |
| comparatorFactories, keyFields, new RTreeDataflowHelperFactory(valueProviderFactories), |
| NoOpOperationCallbackProvider.INSTANCE); |
| |
| return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(rtreeSearchOp, spPc.second); |
| } |
| |
| @Override |
| public Pair<IPushRuntimeFactory, AlgebricksPartitionConstraint> getWriteFileRuntime(IDataSink sink, |
| int[] printColumns, IPrinterFactory[] printerFactories, RecordDescriptor inputDesc) { |
| FileSplitDataSink fsds = (FileSplitDataSink) sink; |
| FileSplitSinkId fssi = (FileSplitSinkId) fsds.getId(); |
| FileSplit fs = fssi.getFileSplit(); |
| File outFile = fs.getLocalFile().getFile(); |
| String nodeId = fs.getNodeName(); |
| |
| SinkWriterRuntimeFactory runtime = new SinkWriterRuntimeFactory(printColumns, printerFactories, outFile, |
| metadata.getWriterFactory(), inputDesc); |
| AlgebricksPartitionConstraint apc = new AlgebricksAbsolutePartitionConstraint(new String[] { nodeId }); |
| return new Pair<IPushRuntimeFactory, AlgebricksPartitionConstraint>(runtime, apc); |
| } |
| |
| @Override |
| public IDataSourceIndex<String, AqlSourceId> findDataSourceIndex(String indexId, AqlSourceId dataSourceId) |
| throws AlgebricksException { |
| AqlDataSource ads = findDataSource(dataSourceId); |
| AqlCompiledDatasetDecl adecl = ads.getCompiledDatasetDecl(); |
| if (adecl.getDatasetType() == DatasetType.EXTERNAL) { |
| throw new AlgebricksException("No index for external dataset " + dataSourceId); |
| } |
| |
| String idxName = (String) indexId; |
| AqlCompiledIndexDecl acid = DatasetUtils.findSecondaryIndexByName(adecl, idxName); |
| AqlSourceId asid = (AqlSourceId) dataSourceId; |
| if (acid != null) { |
| return new AqlIndex(acid, metadata, asid.getDatasetName()); |
| } else { |
| AqlCompiledIndexDecl primIdx = DatasetUtils.getPrimaryIndex(adecl); |
| if (primIdx.getIndexName().equals(indexId)) { |
| return new AqlIndex(primIdx, metadata, asid.getDatasetName()); |
| } else { |
| return null; |
| } |
| } |
| } |
| |
| public static AqlDataSource lookupSourceInMetadata(AqlCompiledMetadataDeclarations metadata, AqlSourceId aqlId) |
| throws AlgebricksException { |
| if (!aqlId.getDataverseName().equals(metadata.getDataverseName())) { |
| return null; |
| } |
| AqlCompiledDatasetDecl acdd = metadata.findDataset(aqlId.getDatasetName()); |
| if (acdd == null) { |
| throw new AlgebricksException("Datasource with id " + aqlId + " was not found."); |
| } |
| String tName = acdd.getItemTypeName(); |
| IAType itemType; |
| try { |
| itemType = metadata.findType(tName); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| return new AqlDataSource(aqlId, acdd, itemType); |
| } |
| |
| @Override |
| public boolean scannerOperatorIsLeaf(IDataSource<AqlSourceId> dataSource) { |
| AqlSourceId asid = dataSource.getId(); |
| String datasetName = asid.getDatasetName(); |
| AqlCompiledDatasetDecl adecl = metadata.findDataset(datasetName); |
| if (adecl == null) { |
| throw new IllegalArgumentException("Unknown dataset " + datasetName); |
| } |
| return adecl.getDatasetType() == DatasetType.EXTERNAL; |
| } |
| |
| @Override |
| public Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> getWriteResultRuntime( |
| IDataSource<AqlSourceId> dataSource, IOperatorSchema propagatedSchema, List<LogicalVariable> keys, |
| LogicalVariable payload, JobGenContext context, JobSpecification spec) throws AlgebricksException { |
| String datasetName = dataSource.getId().getDatasetName(); |
| int numKeys = keys.size(); |
| // move key fields to front |
| int[] fieldPermutation = new int[numKeys + 1]; |
| // System.arraycopy(keys, 0, fieldPermutation, 0, numKeys); |
| int i = 0; |
| for (LogicalVariable varKey : keys) { |
| int idx = propagatedSchema.findVariable(varKey); |
| fieldPermutation[i] = idx; |
| i++; |
| } |
| fieldPermutation[numKeys] = propagatedSchema.findVariable(payload); |
| |
| AqlCompiledDatasetDecl compiledDatasetDecl = metadata.findDataset(datasetName); |
| if (compiledDatasetDecl == null) { |
| throw new AlgebricksException("Unknown dataset " + datasetName); |
| } |
| String indexName = DatasetUtils.getPrimaryIndex(compiledDatasetDecl).getIndexName(); |
| |
| ITypeTraits[] typeTraits = DatasetUtils.computeTupleTypeTraits(compiledDatasetDecl, metadata); |
| |
| IAsterixApplicationContextInfo appContext = (IAsterixApplicationContextInfo) context.getAppContext(); |
| |
| IBinaryComparatorFactory[] comparatorFactories = DatasetUtils.computeKeysBinaryComparatorFactories( |
| compiledDatasetDecl, context.getBinaryComparatorFactoryProvider()); |
| |
| Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint; |
| try { |
| splitsAndConstraint = metadata.splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, |
| indexName); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| |
| TreeIndexBulkLoadOperatorDescriptor btreeBulkLoad = new TreeIndexBulkLoadOperatorDescriptor(spec, |
| appContext.getStorageManagerInterface(), appContext.getIndexRegistryProvider(), |
| splitsAndConstraint.first, typeTraits, comparatorFactories, fieldPermutation, |
| GlobalConfig.DEFAULT_BTREE_FILL_FACTOR, new BTreeDataflowHelperFactory(), |
| NoOpOperationCallbackProvider.INSTANCE); |
| return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(btreeBulkLoad, splitsAndConstraint.second); |
| } |
| |
| @Override |
| public Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> getInsertRuntime( |
| IDataSource<AqlSourceId> dataSource, IOperatorSchema propagatedSchema, List<LogicalVariable> keys, |
| LogicalVariable payload, RecordDescriptor recordDesc, JobGenContext context, JobSpecification spec) |
| throws AlgebricksException { |
| String datasetName = dataSource.getId().getDatasetName(); |
| int numKeys = keys.size(); |
| // move key fields to front |
| int[] fieldPermutation = new int[numKeys + 1]; |
| // System.arraycopy(keys, 0, fieldPermutation, 0, numKeys); |
| int i = 0; |
| for (LogicalVariable varKey : keys) { |
| int idx = propagatedSchema.findVariable(varKey); |
| fieldPermutation[i] = idx; |
| i++; |
| } |
| fieldPermutation[numKeys] = propagatedSchema.findVariable(payload); |
| |
| AqlCompiledDatasetDecl compiledDatasetDecl = metadata.findDataset(datasetName); |
| if (compiledDatasetDecl == null) { |
| throw new AlgebricksException("Unknown dataset " + datasetName); |
| } |
| String indexName = DatasetUtils.getPrimaryIndex(compiledDatasetDecl).getIndexName(); |
| |
| ITypeTraits[] typeTraits = DatasetUtils.computeTupleTypeTraits(compiledDatasetDecl, metadata); |
| |
| IAsterixApplicationContextInfo appContext = (IAsterixApplicationContextInfo) context.getAppContext(); |
| |
| IBinaryComparatorFactory[] comparatorFactories = DatasetUtils.computeKeysBinaryComparatorFactories( |
| compiledDatasetDecl, context.getBinaryComparatorFactoryProvider()); |
| |
| Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint; |
| try { |
| splitsAndConstraint = metadata.splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, |
| indexName); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| |
| TreeIndexInsertUpdateDeleteOperatorDescriptor btreeBulkLoad = new TreeIndexInsertUpdateDeleteOperatorDescriptor( |
| spec, recordDesc, appContext.getStorageManagerInterface(), appContext.getIndexRegistryProvider(), |
| splitsAndConstraint.first, typeTraits, comparatorFactories, fieldPermutation, IndexOp.INSERT, |
| new BTreeDataflowHelperFactory(), null, NoOpOperationCallbackProvider.INSTANCE, txnId); |
| return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(btreeBulkLoad, splitsAndConstraint.second); |
| } |
| |
| @Override |
| public Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> getDeleteRuntime( |
| IDataSource<AqlSourceId> dataSource, IOperatorSchema propagatedSchema, List<LogicalVariable> keys, |
| LogicalVariable payload, RecordDescriptor recordDesc, JobGenContext context, JobSpecification spec) |
| throws AlgebricksException { |
| String datasetName = dataSource.getId().getDatasetName(); |
| int numKeys = keys.size(); |
| // move key fields to front |
| int[] fieldPermutation = new int[numKeys + 1]; |
| // System.arraycopy(keys, 0, fieldPermutation, 0, numKeys); |
| int i = 0; |
| for (LogicalVariable varKey : keys) { |
| int idx = propagatedSchema.findVariable(varKey); |
| fieldPermutation[i] = idx; |
| i++; |
| } |
| fieldPermutation[numKeys] = propagatedSchema.findVariable(payload); |
| |
| AqlCompiledDatasetDecl compiledDatasetDecl = metadata.findDataset(datasetName); |
| if (compiledDatasetDecl == null) { |
| throw new AlgebricksException("Unknown dataset " + datasetName); |
| } |
| String indexName = DatasetUtils.getPrimaryIndex(compiledDatasetDecl).getIndexName(); |
| |
| ITypeTraits[] typeTraits = DatasetUtils.computeTupleTypeTraits(compiledDatasetDecl, metadata); |
| |
| IAsterixApplicationContextInfo appContext = (IAsterixApplicationContextInfo) context.getAppContext(); |
| |
| IBinaryComparatorFactory[] comparatorFactories = DatasetUtils.computeKeysBinaryComparatorFactories( |
| compiledDatasetDecl, context.getBinaryComparatorFactoryProvider()); |
| |
| Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint; |
| try { |
| splitsAndConstraint = metadata.splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, |
| indexName); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| |
| TreeIndexInsertUpdateDeleteOperatorDescriptor btreeBulkLoad = new TreeIndexInsertUpdateDeleteOperatorDescriptor( |
| spec, recordDesc, appContext.getStorageManagerInterface(), appContext.getIndexRegistryProvider(), |
| splitsAndConstraint.first, typeTraits, comparatorFactories, fieldPermutation, IndexOp.DELETE, |
| new BTreeDataflowHelperFactory(), null, NoOpOperationCallbackProvider.INSTANCE, txnId); |
| return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(btreeBulkLoad, splitsAndConstraint.second); |
| } |
| |
| @Override |
| public Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> getIndexInsertRuntime( |
| IDataSourceIndex<String, AqlSourceId> dataSourceIndex, IOperatorSchema propagatedSchema, |
| IOperatorSchema[] inputSchemas, IVariableTypeEnvironment typeEnv, List<LogicalVariable> primaryKeys, |
| List<LogicalVariable> secondaryKeys, ILogicalExpression filterExpr, RecordDescriptor recordDesc, |
| JobGenContext context, JobSpecification spec) throws AlgebricksException { |
| String indexName = dataSourceIndex.getId(); |
| String datasetName = dataSourceIndex.getDataSource().getId().getDatasetName(); |
| AqlCompiledDatasetDecl compiledDatasetDecl = metadata.findDataset(datasetName); |
| if (compiledDatasetDecl == null) { |
| throw new AlgebricksException("Unknown dataset " + datasetName); |
| } |
| AqlCompiledIndexDecl cid = DatasetUtils.findSecondaryIndexByName(compiledDatasetDecl, indexName); |
| AsterixTupleFilterFactory filterFactory = createTupleFilterFactory(inputSchemas, typeEnv, filterExpr, context); |
| if (cid.getKind() == IndexKind.BTREE) { |
| return getBTreeDmlRuntime(datasetName, indexName, propagatedSchema, primaryKeys, secondaryKeys, |
| filterFactory, recordDesc, context, spec, IndexOp.INSERT); |
| } else { |
| return getRTreeDmlRuntime(datasetName, indexName, propagatedSchema, primaryKeys, secondaryKeys, |
| filterFactory, recordDesc, context, spec, IndexOp.INSERT); |
| } |
| } |
| |
| @Override |
| public Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> getIndexDeleteRuntime( |
| IDataSourceIndex<String, AqlSourceId> dataSourceIndex, IOperatorSchema propagatedSchema, |
| IOperatorSchema[] inputSchemas, IVariableTypeEnvironment typeEnv, List<LogicalVariable> primaryKeys, |
| List<LogicalVariable> secondaryKeys, ILogicalExpression filterExpr, RecordDescriptor recordDesc, |
| JobGenContext context, JobSpecification spec) throws AlgebricksException { |
| String indexName = dataSourceIndex.getId(); |
| String datasetName = dataSourceIndex.getDataSource().getId().getDatasetName(); |
| AqlCompiledDatasetDecl compiledDatasetDecl = metadata.findDataset(datasetName); |
| if (compiledDatasetDecl == null) { |
| throw new AlgebricksException("Unknown dataset " + datasetName); |
| } |
| AqlCompiledIndexDecl cid = DatasetUtils.findSecondaryIndexByName(compiledDatasetDecl, indexName); |
| AsterixTupleFilterFactory filterFactory = createTupleFilterFactory(inputSchemas, typeEnv, filterExpr, context); |
| if (cid.getKind() == IndexKind.BTREE) { |
| return getBTreeDmlRuntime(datasetName, indexName, propagatedSchema, primaryKeys, secondaryKeys, |
| filterFactory, recordDesc, context, spec, IndexOp.DELETE); |
| } else { |
| return getRTreeDmlRuntime(datasetName, indexName, propagatedSchema, primaryKeys, secondaryKeys, |
| filterFactory, recordDesc, context, spec, IndexOp.DELETE); |
| } |
| } |
| |
| private AsterixTupleFilterFactory createTupleFilterFactory(IOperatorSchema[] inputSchemas, |
| IVariableTypeEnvironment typeEnv, ILogicalExpression filterExpr, JobGenContext context) |
| throws AlgebricksException { |
| // No filtering condition. |
| if (filterExpr == null) { |
| return null; |
| } |
| ILogicalExpressionJobGen exprJobGen = context.getExpressionJobGen(); |
| IEvaluatorFactory filterEvalFactory = exprJobGen.createEvaluatorFactory(filterExpr, typeEnv, inputSchemas, |
| context); |
| return new AsterixTupleFilterFactory(filterEvalFactory, context.getBinaryBooleanInspector()); |
| } |
| |
| private Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> getBTreeDmlRuntime(String datasetName, |
| String indexName, IOperatorSchema propagatedSchema, List<LogicalVariable> primaryKeys, |
| List<LogicalVariable> secondaryKeys, AsterixTupleFilterFactory filterFactory, RecordDescriptor recordDesc, |
| JobGenContext context, JobSpecification spec, IndexOp indexOp) throws AlgebricksException { |
| int numKeys = primaryKeys.size() + secondaryKeys.size(); |
| // generate field permutations |
| int[] fieldPermutation = new int[numKeys]; |
| int i = 0; |
| for (LogicalVariable varKey : secondaryKeys) { |
| int idx = propagatedSchema.findVariable(varKey); |
| fieldPermutation[i] = idx; |
| i++; |
| } |
| for (LogicalVariable varKey : primaryKeys) { |
| int idx = propagatedSchema.findVariable(varKey); |
| fieldPermutation[i] = idx; |
| i++; |
| } |
| |
| // dataset |
| AqlCompiledDatasetDecl compiledDatasetDecl = metadata.findDataset(datasetName); |
| if (compiledDatasetDecl == null) { |
| throw new AlgebricksException("Unknown dataset " + datasetName); |
| } |
| String itemTypeName = compiledDatasetDecl.getItemTypeName(); |
| IAType itemType; |
| try { |
| itemType = metadata.findType(itemTypeName); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| if (itemType.getTypeTag() != ATypeTag.RECORD) { |
| throw new AlgebricksException("Only record types can be indexed."); |
| } |
| ARecordType recType = (ARecordType) itemType; |
| |
| // index parameters |
| AqlCompiledIndexDecl cid = DatasetUtils.findSecondaryIndexByName(compiledDatasetDecl, indexName); |
| List<String> secondaryKeyExprs = cid.getFieldExprs(); |
| ITypeTraits[] typeTraits = new ITypeTraits[numKeys]; |
| IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[numKeys]; |
| for (i = 0; i < secondaryKeys.size(); ++i) { |
| IAType keyType = AqlCompiledIndexDecl.keyFieldType(secondaryKeyExprs.get(i).toString(), recType); |
| comparatorFactories[i] = AqlBinaryComparatorFactoryProvider.INSTANCE.getBinaryComparatorFactory(keyType, |
| true); |
| typeTraits[i] = AqlTypeTraitProvider.INSTANCE.getTypeTrait(keyType); |
| } |
| for (Triple<IEvaluatorFactory, ScalarFunctionCallExpression, IAType> evalFactoryAndType : DatasetUtils |
| .getPartitioningFunctions(compiledDatasetDecl)) { |
| IAType keyType = evalFactoryAndType.third; |
| comparatorFactories[i] = AqlBinaryComparatorFactoryProvider.INSTANCE.getBinaryComparatorFactory(keyType, |
| true); |
| typeTraits[i] = AqlTypeTraitProvider.INSTANCE.getTypeTrait(keyType); |
| ++i; |
| } |
| |
| IAsterixApplicationContextInfo appContext = (IAsterixApplicationContextInfo) context.getAppContext(); |
| Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint; |
| try { |
| splitsAndConstraint = metadata.splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, |
| indexName); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| TreeIndexInsertUpdateDeleteOperatorDescriptor btreeBulkLoad = new TreeIndexInsertUpdateDeleteOperatorDescriptor( |
| spec, recordDesc, appContext.getStorageManagerInterface(), appContext.getIndexRegistryProvider(), |
| splitsAndConstraint.first, typeTraits, comparatorFactories, fieldPermutation, indexOp, |
| new BTreeDataflowHelperFactory(), filterFactory, NoOpOperationCallbackProvider.INSTANCE, txnId); |
| return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(btreeBulkLoad, splitsAndConstraint.second); |
| } |
| |
| private Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> getRTreeDmlRuntime(String datasetName, |
| String indexName, IOperatorSchema propagatedSchema, List<LogicalVariable> primaryKeys, |
| List<LogicalVariable> secondaryKeys, AsterixTupleFilterFactory filterFactory, RecordDescriptor recordDesc, |
| JobGenContext context, JobSpecification spec, IndexOp indexOp) throws AlgebricksException { |
| AqlCompiledDatasetDecl compiledDatasetDecl = metadata.findDataset(datasetName); |
| String itemTypeName = compiledDatasetDecl.getItemTypeName(); |
| IAType itemType; |
| try { |
| itemType = metadata.findType(itemTypeName); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| if (itemType.getTypeTag() != ATypeTag.RECORD) { |
| throw new AlgebricksException("Only record types can be indexed."); |
| } |
| ARecordType recType = (ARecordType) itemType; |
| AqlCompiledIndexDecl cid = DatasetUtils.findSecondaryIndexByName(compiledDatasetDecl, indexName); |
| List<String> secondaryKeyExprs = cid.getFieldExprs(); |
| Pair<IAType, Boolean> keyPairType = AqlCompiledIndexDecl.getNonNullableKeyFieldType(secondaryKeyExprs.get(0), |
| recType); |
| IAType spatialType = keyPairType.first; |
| int dimension = NonTaggedFormatUtil.getNumDimensions(spatialType.getTypeTag()); |
| int numSecondaryKeys = dimension * 2; |
| int numPrimaryKeys = primaryKeys.size(); |
| int numKeys = numSecondaryKeys + numPrimaryKeys; |
| ITypeTraits[] typeTraits = new ITypeTraits[numKeys]; |
| IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[numKeys]; |
| int[] fieldPermutation = new int[numKeys]; |
| int i = 0; |
| |
| for (LogicalVariable varKey : secondaryKeys) { |
| int idx = propagatedSchema.findVariable(varKey); |
| fieldPermutation[i] = idx; |
| i++; |
| } |
| for (LogicalVariable varKey : primaryKeys) { |
| int idx = propagatedSchema.findVariable(varKey); |
| fieldPermutation[i] = idx; |
| i++; |
| } |
| IAType nestedKeyType = NonTaggedFormatUtil.getNestedSpatialType(spatialType.getTypeTag()); |
| IPrimitiveValueProviderFactory[] valueProviderFactories = new IPrimitiveValueProviderFactory[numSecondaryKeys]; |
| for (i = 0; i < numSecondaryKeys; i++) { |
| ISerializerDeserializer keySerde = AqlSerializerDeserializerProvider.INSTANCE |
| .getSerializerDeserializer(nestedKeyType); |
| comparatorFactories[i] = AqlBinaryComparatorFactoryProvider.INSTANCE.getBinaryComparatorFactory( |
| nestedKeyType, true); |
| typeTraits[i] = AqlTypeTraitProvider.INSTANCE.getTypeTrait(nestedKeyType); |
| valueProviderFactories[i] = AqlPrimitiveValueProviderFactory.INSTANCE; |
| } |
| for (Triple<IEvaluatorFactory, ScalarFunctionCallExpression, IAType> evalFactoryAndType : DatasetUtils |
| .getPartitioningFunctions(compiledDatasetDecl)) { |
| IAType keyType = evalFactoryAndType.third; |
| comparatorFactories[i] = AqlBinaryComparatorFactoryProvider.INSTANCE.getBinaryComparatorFactory(keyType, |
| true); |
| typeTraits[i] = AqlTypeTraitProvider.INSTANCE.getTypeTrait(keyType); |
| ++i; |
| } |
| |
| IAsterixApplicationContextInfo appContext = (IAsterixApplicationContextInfo) context.getAppContext(); |
| Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint; |
| try { |
| splitsAndConstraint = metadata.splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, |
| indexName); |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| TreeIndexInsertUpdateDeleteOperatorDescriptor rtreeUpdate = new TreeIndexInsertUpdateDeleteOperatorDescriptor( |
| spec, recordDesc, appContext.getStorageManagerInterface(), appContext.getIndexRegistryProvider(), |
| splitsAndConstraint.first, typeTraits, comparatorFactories, fieldPermutation, indexOp, |
| new RTreeDataflowHelperFactory(valueProviderFactories), filterFactory, |
| NoOpOperationCallbackProvider.INSTANCE, txnId); |
| return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(rtreeUpdate, splitsAndConstraint.second); |
| } |
| |
| public long getTxnId() { |
| return txnId; |
| } |
| |
| public static ITreeIndexFrameFactory createBTreeNSMInteriorFrameFactory(ITypeTraits[] typeTraits) { |
| return new BTreeNSMInteriorFrameFactory(new TypeAwareTupleWriterFactory(typeTraits)); |
| } |
| |
| public static ITreeIndexFrameFactory createBTreeNSMLeafFrameFactory(ITypeTraits[] typeTraits) { |
| return new BTreeNSMLeafFrameFactory(new TypeAwareTupleWriterFactory(typeTraits)); |
| } |
| |
| @Override |
| public IFunctionInfo lookupFunction(FunctionIdentifier fid) { |
| return AsterixBuiltinFunctions.lookupFunction(fid); |
| } |
| } |
