| /* |
| * 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.declared; |
| |
| import java.util.ArrayList; |
| import java.util.List; |
| |
| import org.apache.asterix.active.EntityId; |
| import org.apache.asterix.common.exceptions.CompilationException; |
| import org.apache.asterix.common.exceptions.ErrorCode; |
| import org.apache.asterix.external.feed.management.FeedConnectionId; |
| import org.apache.asterix.external.operators.FeedCollectOperatorDescriptor; |
| import org.apache.asterix.external.util.FeedUtils.FeedRuntimeType; |
| import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider; |
| import org.apache.asterix.metadata.entities.Feed; |
| import org.apache.asterix.metadata.entities.FeedConnection; |
| import org.apache.asterix.metadata.entities.FeedPolicyEntity; |
| import org.apache.asterix.metadata.feeds.BuiltinFeedPolicies; |
| import org.apache.asterix.om.types.ARecordType; |
| import org.apache.asterix.om.types.IAType; |
| import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint; |
| import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint; |
| import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException; |
| import org.apache.hyracks.algebricks.common.utils.Pair; |
| import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable; |
| import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment; |
| import org.apache.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression; |
| import org.apache.hyracks.algebricks.core.algebra.metadata.IDataSource; |
| import org.apache.hyracks.algebricks.core.algebra.metadata.IProjectionFiltrationInfo; |
| import org.apache.hyracks.algebricks.core.algebra.operators.logical.IOperatorSchema; |
| import org.apache.hyracks.algebricks.core.algebra.properties.INodeDomain; |
| import org.apache.hyracks.algebricks.core.jobgen.impl.JobGenContext; |
| import org.apache.hyracks.api.dataflow.IOperatorDescriptor; |
| import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer; |
| import org.apache.hyracks.api.dataflow.value.RecordDescriptor; |
| import org.apache.hyracks.api.job.JobSpecification; |
| import org.apache.hyracks.storage.am.common.api.ITupleFilterFactory; |
| |
| public class FeedDataSource extends DataSource implements IMutationDataSource { |
| |
| private final Feed feed; |
| private final EntityId sourceFeedId; |
| private final FeedRuntimeType location; |
| private final String targetDataset; |
| private final String[] locations; |
| private final INodeDomain computationNodeDomain; |
| private final List<IAType> pkTypes; |
| private final List<ScalarFunctionCallExpression> keyAccessExpression; |
| private final FeedConnection feedConnection; |
| |
| public FeedDataSource(Feed feed, DataSourceId id, String targetDataset, IAType itemType, IAType metaType, |
| List<IAType> pkTypes, List<ScalarFunctionCallExpression> keyAccessExpression, EntityId sourceFeedId, |
| FeedRuntimeType location, String[] locations, INodeDomain domain, FeedConnection feedConnection) |
| throws AlgebricksException { |
| super(id, itemType, metaType, Type.FEED, domain); |
| this.feed = feed; |
| this.targetDataset = targetDataset; |
| this.sourceFeedId = sourceFeedId; |
| this.location = location; |
| this.locations = locations; |
| this.pkTypes = pkTypes; |
| this.keyAccessExpression = keyAccessExpression; |
| this.computationNodeDomain = domain; |
| this.feedConnection = feedConnection; |
| initFeedDataSource(); |
| } |
| |
| public Feed getFeed() { |
| return feed; |
| } |
| |
| @Override |
| public IAType[] getSchemaTypes() { |
| return schemaTypes; |
| } |
| |
| public String getTargetDataset() { |
| return targetDataset; |
| } |
| |
| public EntityId getSourceFeedId() { |
| return sourceFeedId; |
| } |
| |
| public FeedRuntimeType getLocation() { |
| return location; |
| } |
| |
| public String[] getLocations() { |
| return locations; |
| } |
| |
| private void initFeedDataSource() { |
| int i = 0; |
| // record + meta (if exists) + PKs (if exists) |
| schemaTypes = new IAType[(1 + (metaItemType != null ? 1 : 0) + (pkTypes != null ? pkTypes.size() : 0))]; |
| schemaTypes[i++] = itemType; |
| if (metaItemType != null) { |
| schemaTypes[i++] = metaItemType; |
| } |
| if (pkTypes != null) { |
| for (IAType type : pkTypes) { |
| schemaTypes[i++] = type; |
| } |
| } |
| } |
| |
| public List<IAType> getPkTypes() { |
| return pkTypes; |
| } |
| |
| @Override |
| public List<ScalarFunctionCallExpression> getKeyAccessExpression() { |
| return keyAccessExpression; |
| } |
| |
| @Override |
| public LogicalVariable getMetaVariable(List<LogicalVariable> dataScanVariables) { |
| return metaItemType == null ? null : dataScanVariables.get(1); |
| } |
| |
| @Override |
| public LogicalVariable getDataRecordVariable(List<LogicalVariable> dataScanVariables) { |
| return dataScanVariables.get(0); |
| } |
| |
| @Override |
| public boolean isChange() { |
| return pkTypes != null; |
| } |
| |
| @Override |
| public List<LogicalVariable> getPkVars(List<LogicalVariable> allVars) { |
| if (pkTypes == null) { |
| return null; |
| } |
| if (metaItemType != null) { |
| return allVars.subList(2, allVars.size()); |
| } else { |
| return allVars.subList(1, allVars.size()); |
| } |
| } |
| |
| @Override |
| public Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> buildDatasourceScanRuntime( |
| MetadataProvider metadataProvider, IDataSource<DataSourceId> dataSource, |
| List<LogicalVariable> scanVariables, List<LogicalVariable> projectVariables, boolean projectPushed, |
| List<LogicalVariable> minFilterVars, List<LogicalVariable> maxFilterVars, |
| ITupleFilterFactory tupleFilterFactory, long outputLimit, IOperatorSchema opSchema, |
| IVariableTypeEnvironment typeEnv, JobGenContext context, JobSpecification jobSpec, Object implConfig, |
| IProjectionFiltrationInfo projectionFiltrationInfo) throws AlgebricksException { |
| try { |
| if (tupleFilterFactory != null || outputLimit >= 0) { |
| throw CompilationException.create(ErrorCode.COMPILATION_ILLEGAL_STATE, |
| "Tuple filter and limit are not supported by FeedDataSource"); |
| } |
| ARecordType feedOutputType = (ARecordType) itemType; |
| ISerializerDeserializer payloadSerde = |
| metadataProvider.getDataFormat().getSerdeProvider().getSerializerDeserializer(feedOutputType); |
| ArrayList<ISerializerDeserializer> serdes = new ArrayList<>(); |
| serdes.add(payloadSerde); |
| if (metaItemType != null) { |
| serdes.add(SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(metaItemType)); |
| } |
| if (pkTypes != null) { |
| for (IAType type : pkTypes) { |
| serdes.add(SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(type)); |
| } |
| } |
| RecordDescriptor feedDesc = |
| new RecordDescriptor(serdes.toArray(new ISerializerDeserializer[serdes.size()])); |
| FeedPolicyEntity feedPolicy = |
| (FeedPolicyEntity) getProperties().get(BuiltinFeedPolicies.CONFIG_FEED_POLICY_KEY); |
| if (feedPolicy == null) { |
| throw new AlgebricksException("Feed not configured with a policy"); |
| } |
| feedPolicy.getProperties().put(BuiltinFeedPolicies.CONFIG_FEED_POLICY_KEY, feedPolicy.getPolicyName()); |
| FeedConnectionId feedConnectionId = new FeedConnectionId(getId().getDatabaseName(), |
| getId().getDataverseName(), getId().getDatasourceName(), getTargetDataset()); |
| FeedCollectOperatorDescriptor feedCollector = new FeedCollectOperatorDescriptor(jobSpec, feedConnectionId, |
| feedOutputType, feedDesc, feedPolicy.getProperties(), getLocation()); |
| |
| return new Pair<>(feedCollector, new AlgebricksAbsolutePartitionConstraint(getLocations())); |
| |
| } catch (Exception e) { |
| throw new AlgebricksException(e); |
| } |
| } |
| |
| @Override |
| public boolean isScanAccessPathALeaf() { |
| return true; |
| } |
| |
| public FeedConnection getFeedConnection() { |
| return feedConnection; |
| } |
| |
| public INodeDomain getComputationNodeDomain() { |
| return computationNodeDomain; |
| } |
| } |
