| /* |
| * 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.vxquery.metadata; |
| |
| import java.io.DataInputStream; |
| import java.io.File; |
| import java.io.IOException; |
| import java.nio.ByteBuffer; |
| import java.util.Map; |
| import java.util.logging.Logger; |
| |
| import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException; |
| import org.apache.hyracks.api.client.NodeControllerInfo; |
| import org.apache.hyracks.api.comm.IFrame; |
| import org.apache.hyracks.api.comm.IFrameFieldAppender; |
| import org.apache.hyracks.api.comm.VSizeFrame; |
| import org.apache.hyracks.api.context.IHyracksTaskContext; |
| import org.apache.hyracks.api.dataflow.IOperatorNodePushable; |
| import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider; |
| import org.apache.hyracks.api.dataflow.value.RecordDescriptor; |
| import org.apache.hyracks.api.exceptions.HyracksDataException; |
| import org.apache.hyracks.api.job.IOperatorDescriptorRegistry; |
| import org.apache.hyracks.data.std.api.IPointable; |
| import org.apache.hyracks.data.std.primitive.UTF8StringPointable; |
| import org.apache.hyracks.data.std.util.ArrayBackedValueStorage; |
| import org.apache.hyracks.dataflow.common.comm.io.FrameFixedFieldTupleAppender; |
| import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor; |
| import org.apache.hyracks.dataflow.common.comm.util.ByteBufferInputStream; |
| import org.apache.hyracks.dataflow.common.comm.util.FrameUtils; |
| import org.apache.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor; |
| import org.apache.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperatorNodePushable; |
| import org.apache.vxquery.datamodel.accessors.TaggedValuePointable; |
| import org.apache.vxquery.datamodel.builders.sequence.SequenceBuilder; |
| import org.apache.vxquery.datamodel.values.XDMConstants; |
| import org.apache.vxquery.exceptions.SystemException; |
| import org.apache.vxquery.functions.BuiltinFunctions; |
| import org.apache.vxquery.runtime.functions.index.IndexConstructorUtil; |
| import org.apache.vxquery.runtime.functions.index.VXQueryIndexReader; |
| import org.apache.vxquery.runtime.functions.index.indexCentralizer.IndexCentralizerUtil; |
| import org.apache.vxquery.runtime.functions.index.updateIndex.IndexUpdater; |
| import org.apache.vxquery.xmlparser.ITreeNodeIdProvider; |
| import org.apache.vxquery.xmlparser.TreeNodeIdProvider; |
| |
| public class VXQueryIndexingOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor { |
| protected static final Logger LOGGER = Logger.getLogger(VXQueryCollectionOperatorDescriptor.class.getName()); |
| private static final long serialVersionUID = 1L; |
| private short dataSourceId; |
| private short totalDataSources; |
| private String[] collectionPartitions; |
| private String elementPath; |
| private final String functionCall; |
| |
| public VXQueryIndexingOperatorDescriptor(IOperatorDescriptorRegistry spec, VXQueryIndexingDataSource ds, |
| RecordDescriptor rDesc, String hdfsConf, Map<String, NodeControllerInfo> nodeControllerInfos) { |
| super(spec, 1, 1); |
| this.functionCall = ds.getFunctionCall(); |
| collectionPartitions = ds.getPartitions(); |
| dataSourceId = (short) ds.getDataSourceId(); |
| totalDataSources = (short) ds.getTotalDataSources(); |
| recordDescriptors[0] = rDesc; |
| this.elementPath = ds.getElementPath(); |
| } |
| |
| @Override |
| public IOperatorNodePushable createPushRuntime(IHyracksTaskContext ctx, |
| IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) throws HyracksDataException { |
| final FrameTupleAccessor fta = new FrameTupleAccessor( |
| recordDescProvider.getInputRecordDescriptor(getActivityId(), 0)); |
| final int fieldOutputCount = recordDescProvider.getOutputRecordDescriptor(getActivityId(), 0).getFieldCount(); |
| final IFrame frame = new VSizeFrame(ctx); |
| final IFrameFieldAppender appender = new FrameFixedFieldTupleAppender(fieldOutputCount); |
| final short partitionId = (short) ctx.getTaskAttemptId().getTaskId().getPartition(); |
| final ITreeNodeIdProvider nodeIdProvider = new TreeNodeIdProvider(partitionId, dataSourceId, totalDataSources); |
| final String nodeId = ctx.getJobletContext().getApplicationContext().getNodeId(); |
| final String collectionName = collectionPartitions[partition % collectionPartitions.length]; |
| String collectionModifiedName = collectionName.replace("${nodeId}", nodeId); |
| IndexCentralizerUtil indexCentralizerUtil = new IndexCentralizerUtil( |
| ctx.getIOManager().getIODevices().get(0).getMount()); |
| indexCentralizerUtil.readIndexDirectory(); |
| |
| return new AbstractUnaryInputUnaryOutputOperatorNodePushable() { |
| @Override |
| public void open() throws HyracksDataException { |
| appender.reset(frame, true); |
| writer.open(); |
| } |
| |
| @Override |
| public void nextFrame(ByteBuffer buffer) throws HyracksDataException { |
| fta.reset(buffer); |
| |
| IPointable result = new TaggedValuePointable(); |
| |
| final UTF8StringPointable stringp = (UTF8StringPointable) UTF8StringPointable.FACTORY.createPointable(); |
| final TaggedValuePointable nodep = (TaggedValuePointable) TaggedValuePointable.FACTORY |
| .createPointable(); |
| |
| final ByteBufferInputStream bbis = new ByteBufferInputStream(); |
| final DataInputStream di = new DataInputStream(bbis); |
| final SequenceBuilder sb = new SequenceBuilder(); |
| final ArrayBackedValueStorage abvs = new ArrayBackedValueStorage(); |
| final ArrayBackedValueStorage abvsFileNode = new ArrayBackedValueStorage(); |
| |
| String indexModifiedName; |
| if (collectionModifiedName.contains("hdfs://")) { |
| throw new HyracksDataException("Indexing support for HDFS not yet implemented."); |
| } else { |
| |
| if (functionCall.equals( |
| BuiltinFunctions.FN_BUILD_INDEX_ON_COLLECTION_1.getFunctionIdentifier().getName())) { |
| indexModifiedName = indexCentralizerUtil.putIndexForCollection(collectionModifiedName); |
| File collectionDirectory = new File(collectionModifiedName); |
| |
| //check if directory is in the local file system |
| if (collectionDirectory.exists() && collectionDirectory.isDirectory()) { |
| IndexConstructorUtil indexConstructorUtil = new IndexConstructorUtil(); |
| try { |
| indexConstructorUtil.evaluate(collectionModifiedName, indexModifiedName, result, |
| stringp, bbis, di, sb, abvs, nodeIdProvider, abvsFileNode, nodep, false, |
| nodeId); |
| XDMConstants.setTrue(result); |
| FrameUtils.appendFieldToWriter(writer, appender, result.getByteArray(), |
| result.getStartOffset(), result.getLength()); |
| } catch (SystemException e) { |
| throw new HyracksDataException("Could not create index for collection: " |
| + collectionName + " in dir: " + indexModifiedName + " " + e.getMessage()); |
| } |
| } else { |
| throw new HyracksDataException("Cannot find Collection Directory (" + nodeId + ":" |
| + collectionDirectory.getAbsolutePath() + ")"); |
| } |
| } else if (functionCall |
| .equals(BuiltinFunctions.FN_UPDATE_INDEX_1.getFunctionIdentifier().getName())) { |
| indexModifiedName = indexCentralizerUtil.getIndexForCollection(collectionModifiedName); |
| IndexUpdater updater = new IndexUpdater(indexModifiedName, result, stringp, bbis, di, sb, abvs, |
| nodeIdProvider, abvsFileNode, nodep, nodeId); |
| try { |
| updater.setup(); |
| updater.updateIndex(); |
| updater.updateMetadataFile(); |
| updater.exit(); |
| XDMConstants.setTrue(result); |
| FrameUtils.appendFieldToWriter(writer, appender, result.getByteArray(), |
| result.getStartOffset(), result.getLength()); |
| } catch (IOException e) { |
| throw new HyracksDataException( |
| "Could not update index in " + indexModifiedName + " " + e.getMessage()); |
| } |
| } else if (functionCall |
| .equals(BuiltinFunctions.FN_DELETE_INDEX_1.getFunctionIdentifier().getName())) { |
| indexModifiedName = indexCentralizerUtil.getIndexForCollection(collectionModifiedName); |
| IndexUpdater updater = new IndexUpdater(indexModifiedName, result, stringp, bbis, di, sb, abvs, |
| nodeIdProvider, abvsFileNode, nodep, nodeId); |
| indexCentralizerUtil.deleteEntryForCollection(collectionModifiedName); |
| try { |
| updater.setup(); |
| updater.deleteAllIndexes(); |
| XDMConstants.setTrue(result); |
| FrameUtils.appendFieldToWriter(writer, appender, result.getByteArray(), |
| result.getStartOffset(), result.getLength()); |
| } catch (IOException e) { |
| throw new HyracksDataException( |
| "Could not delete index in " + indexModifiedName + " " + e.getMessage()); |
| } |
| |
| } else if (functionCall |
| .equals(BuiltinFunctions.FN_COLLECTION_FROM_INDEX_2.getFunctionIdentifier().getName())) { |
| indexModifiedName = indexCentralizerUtil.getIndexForCollection(collectionModifiedName); |
| VXQueryIndexReader indexReader = new VXQueryIndexReader(ctx, indexModifiedName, elementPath); |
| try { |
| indexReader.init(); |
| while (indexReader.step(result)) { |
| FrameUtils.appendFieldToWriter(writer, appender, result.getByteArray(), |
| result.getStartOffset(), result.getLength()); |
| } |
| } catch (AlgebricksException e) { |
| throw new HyracksDataException("Could not read index."); |
| } |
| |
| } else { |
| throw new HyracksDataException("Unsupported function call (" + functionCall + ")"); |
| } |
| } |
| } |
| |
| @Override |
| public void fail() throws HyracksDataException { |
| writer.fail(); |
| } |
| |
| @Override |
| public void close() throws HyracksDataException { |
| // Check if needed? |
| if (appender.getTupleCount() > 0) { |
| appender.flush(writer); |
| } |
| writer.close(); |
| indexCentralizerUtil.writeIndexDirectory(); |
| } |
| }; |
| } |
| } |
