| /** |
| * 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.gora.accumulo.store; |
| |
| import java.io.ByteArrayOutputStream; |
| import java.io.IOException; |
| import java.net.InetAddress; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.concurrent.TimeUnit; |
| |
| import javax.xml.parsers.DocumentBuilder; |
| import javax.xml.parsers.DocumentBuilderFactory; |
| |
| import org.apache.accumulo.core.Constants; |
| import org.apache.accumulo.core.client.AccumuloException; |
| import org.apache.accumulo.core.client.AccumuloSecurityException; |
| import org.apache.accumulo.core.client.BatchWriter; |
| import org.apache.accumulo.core.client.BatchWriterConfig; |
| import org.apache.accumulo.core.client.Connector; |
| import org.apache.accumulo.core.client.IsolatedScanner; |
| import org.apache.accumulo.core.client.IteratorSetting; |
| import org.apache.accumulo.core.client.MutationsRejectedException; |
| import org.apache.accumulo.core.client.RowIterator; |
| import org.apache.accumulo.core.client.Scanner; |
| import org.apache.accumulo.core.client.TableDeletedException; |
| import org.apache.accumulo.core.client.TableExistsException; |
| import org.apache.accumulo.core.client.TableNotFoundException; |
| import org.apache.accumulo.core.client.TableOfflineException; |
| import org.apache.accumulo.core.client.ZooKeeperInstance; |
| import org.apache.accumulo.core.client.impl.Tables; |
| import org.apache.accumulo.core.client.impl.TabletLocator; |
| import org.apache.accumulo.core.client.mock.MockConnector; |
| import org.apache.accumulo.core.client.mock.MockInstance; |
| import org.apache.accumulo.core.client.mock.MockTabletLocator; |
| import org.apache.accumulo.core.client.security.tokens.AuthenticationToken; |
| import org.apache.accumulo.core.client.security.tokens.PasswordToken; |
| import org.apache.accumulo.core.data.ByteSequence; |
| import org.apache.accumulo.core.data.Key; |
| import org.apache.accumulo.core.data.KeyExtent; |
| import org.apache.accumulo.core.data.Mutation; |
| import org.apache.accumulo.core.data.Range; |
| import org.apache.accumulo.core.data.Value; |
| import org.apache.accumulo.core.iterators.SortedKeyIterator; |
| import org.apache.accumulo.core.iterators.user.TimestampFilter; |
| import org.apache.accumulo.core.master.state.tables.TableState; |
| import org.apache.accumulo.core.security.ColumnVisibility; |
| import org.apache.accumulo.core.security.thrift.TCredentials; |
| import org.apache.accumulo.core.util.Pair; |
| import org.apache.accumulo.core.util.TextUtil; |
| import org.apache.accumulo.core.util.UtilWaitThread; |
| import org.apache.avro.Schema; |
| import org.apache.avro.Schema.Field; |
| import org.apache.avro.Schema.Type; |
| import org.apache.avro.generic.GenericData; |
| import org.apache.avro.io.BinaryDecoder; |
| import org.apache.avro.io.Decoder; |
| import org.apache.avro.io.DecoderFactory; |
| import org.apache.avro.io.EncoderFactory; |
| import org.apache.avro.specific.SpecificDatumReader; |
| import org.apache.avro.specific.SpecificDatumWriter; |
| import org.apache.avro.util.Utf8; |
| import org.apache.gora.accumulo.encoders.BinaryEncoder; |
| import org.apache.gora.accumulo.encoders.Encoder; |
| import org.apache.gora.accumulo.query.AccumuloQuery; |
| import org.apache.gora.accumulo.query.AccumuloResult; |
| import org.apache.gora.persistency.impl.DirtyListWrapper; |
| import org.apache.gora.persistency.impl.DirtyMapWrapper; |
| import org.apache.gora.persistency.impl.PersistentBase; |
| import org.apache.gora.query.PartitionQuery; |
| import org.apache.gora.query.Query; |
| import org.apache.gora.query.Result; |
| import org.apache.gora.query.impl.PartitionQueryImpl; |
| import org.apache.gora.store.DataStoreFactory; |
| import org.apache.gora.store.impl.DataStoreBase; |
| import org.apache.gora.util.AvroUtils; |
| import org.apache.gora.util.GoraException; |
| import org.apache.hadoop.io.Text; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.w3c.dom.Document; |
| import org.w3c.dom.Element; |
| import org.w3c.dom.NodeList; |
| |
| /** |
| * |
| */ |
| public class AccumuloStore<K,T extends PersistentBase> extends DataStoreBase<K,T> { |
| |
| protected static final String MOCK_PROPERTY = "accumulo.mock"; |
| protected static final String INSTANCE_NAME_PROPERTY = "accumulo.instance"; |
| protected static final String ZOOKEEPERS_NAME_PROPERTY = "accumulo.zookeepers"; |
| protected static final String USERNAME_PROPERTY = "accumulo.user"; |
| protected static final String PASSWORD_PROPERTY = "accumulo.password"; |
| protected static final String DEFAULT_MAPPING_FILE = "gora-accumulo-mapping.xml"; |
| |
| private Connector conn; |
| private BatchWriter batchWriter; |
| private AccumuloMapping mapping; |
| private TCredentials credentials; |
| private Encoder encoder; |
| |
| public static final Logger LOG = LoggerFactory.getLogger(AccumuloStore.class); |
| |
| public Object fromBytes(Schema schema, byte data[]) throws GoraException { |
| Schema fromSchema = null; |
| if (schema.getType() == Type.UNION) { |
| try { |
| Decoder decoder = DecoderFactory.get().binaryDecoder(data, null); |
| int unionIndex = decoder.readIndex(); |
| List<Schema> possibleTypes = schema.getTypes(); |
| fromSchema = possibleTypes.get(unionIndex); |
| Schema effectiveSchema = possibleTypes.get(unionIndex); |
| if (effectiveSchema.getType() == Type.NULL) { |
| decoder.readNull(); |
| return null; |
| } else { |
| data = decoder.readBytes(null).array(); |
| } |
| } catch (IOException e) { |
| e.printStackTrace(); |
| throw new GoraException("Error decoding union type: ", e); |
| } |
| } else { |
| fromSchema = schema; |
| } |
| return fromBytes(encoder, fromSchema, data); |
| } |
| |
| public static Object fromBytes(Encoder encoder, Schema schema, byte data[]) { |
| switch (schema.getType()) { |
| case BOOLEAN: |
| return encoder.decodeBoolean(data); |
| case DOUBLE: |
| return encoder.decodeDouble(data); |
| case FLOAT: |
| return encoder.decodeFloat(data); |
| case INT: |
| return encoder.decodeInt(data); |
| case LONG: |
| return encoder.decodeLong(data); |
| case STRING: |
| return new Utf8(data); |
| case BYTES: |
| return ByteBuffer.wrap(data); |
| case ENUM: |
| return AvroUtils.getEnumValue(schema, encoder.decodeInt(data)); |
| case ARRAY: |
| break; |
| case FIXED: |
| break; |
| case MAP: |
| break; |
| case NULL: |
| break; |
| case RECORD: |
| break; |
| case UNION: |
| break; |
| default: |
| break; |
| } |
| throw new IllegalArgumentException("Unknown type " + schema.getType()); |
| |
| } |
| |
| public K fromBytes(Class<K> clazz, byte[] val) { |
| return fromBytes(encoder, clazz, val); |
| } |
| |
| @SuppressWarnings("unchecked") |
| public static <K> K fromBytes(Encoder encoder, Class<K> clazz, byte[] val) { |
| try { |
| if (clazz.equals(Byte.TYPE) || clazz.equals(Byte.class)) { |
| return (K) Byte.valueOf(encoder.decodeByte(val)); |
| } else if (clazz.equals(Boolean.TYPE) || clazz.equals(Boolean.class)) { |
| return (K) Boolean.valueOf(encoder.decodeBoolean(val)); |
| } else if (clazz.equals(Short.TYPE) || clazz.equals(Short.class)) { |
| return (K) Short.valueOf(encoder.decodeShort(val)); |
| } else if (clazz.equals(Integer.TYPE) || clazz.equals(Integer.class)) { |
| return (K) Integer.valueOf(encoder.decodeInt(val)); |
| } else if (clazz.equals(Long.TYPE) || clazz.equals(Long.class)) { |
| return (K) Long.valueOf(encoder.decodeLong(val)); |
| } else if (clazz.equals(Float.TYPE) || clazz.equals(Float.class)) { |
| return (K) Float.valueOf(encoder.decodeFloat(val)); |
| } else if (clazz.equals(Double.TYPE) || clazz.equals(Double.class)) { |
| return (K) Double.valueOf(encoder.decodeDouble(val)); |
| } else if (clazz.equals(String.class)) { |
| return (K) new String(val, "UTF-8"); |
| } else if (clazz.equals(Utf8.class)) { |
| return (K) new Utf8(val); |
| } |
| |
| throw new IllegalArgumentException("Unknown type " + clazz.getName()); |
| } catch (IOException ioe) { |
| throw new RuntimeException(ioe); |
| } |
| } |
| |
| private static byte[] copyIfNeeded(byte b[], int offset, int len) { |
| if (len != b.length || offset != 0) { |
| byte copy[] = new byte[len]; |
| System.arraycopy(b, offset, copy, 0, copy.length); |
| b = copy; |
| } |
| return b; |
| } |
| |
| public byte[] toBytes(Schema toSchema, Object o) { |
| if (toSchema != null && toSchema.getType() == Type.UNION) { |
| ByteArrayOutputStream baos = new ByteArrayOutputStream(); |
| org.apache.avro.io.BinaryEncoder avroEncoder = EncoderFactory.get().binaryEncoder(baos, null); |
| int unionIndex = 0; |
| try { |
| if (o == null) { |
| unionIndex = firstNullSchemaTypeIndex(toSchema); |
| avroEncoder.writeIndex(unionIndex); |
| avroEncoder.writeNull(); |
| } else { |
| unionIndex = firstNotNullSchemaTypeIndex(toSchema); |
| avroEncoder.writeIndex(unionIndex); |
| avroEncoder.writeBytes(toBytes(o)); |
| } |
| avroEncoder.flush(); |
| return baos.toByteArray(); |
| } catch (IOException e) { |
| e.printStackTrace(); |
| return toBytes(o); |
| } |
| } else { |
| return toBytes(o); |
| } |
| } |
| |
| private int firstNullSchemaTypeIndex(Schema toSchema) { |
| List<Schema> possibleTypes = toSchema.getTypes(); |
| int unionIndex = 0; |
| for (int i = 0; i < possibleTypes.size(); i++ ) { |
| Type pType = possibleTypes.get(i).getType(); |
| if (pType == Type.NULL) { // FIXME HUGE kludge to pass tests |
| unionIndex = i; break; |
| } |
| } |
| return unionIndex; |
| } |
| |
| private int firstNotNullSchemaTypeIndex(Schema toSchema) { |
| List<Schema> possibleTypes = toSchema.getTypes(); |
| int unionIndex = 0; |
| for (int i = 0; i < possibleTypes.size(); i++ ) { |
| Type pType = possibleTypes.get(i).getType(); |
| if (pType != Type.NULL) { // FIXME HUGE kludge to pass tests |
| unionIndex = i; break; |
| } |
| } |
| return unionIndex; |
| } |
| |
| public byte[] toBytes(Object o) { |
| return toBytes(encoder, o); |
| } |
| |
| public static byte[] toBytes(Encoder encoder, Object o) { |
| |
| try { |
| if (o instanceof String) { |
| return ((String) o).getBytes("UTF-8"); |
| } else if (o instanceof Utf8) { |
| return copyIfNeeded(((Utf8) o).getBytes(), 0, ((Utf8) o).getByteLength()); |
| } else if (o instanceof ByteBuffer) { |
| return copyIfNeeded(((ByteBuffer) o).array(), ((ByteBuffer) o).arrayOffset() + ((ByteBuffer) o).position(), ((ByteBuffer) o).remaining()); |
| } else if (o instanceof Long) { |
| return encoder.encodeLong((Long) o); |
| } else if (o instanceof Integer) { |
| return encoder.encodeInt((Integer) o); |
| } else if (o instanceof Short) { |
| return encoder.encodeShort((Short) o); |
| } else if (o instanceof Byte) { |
| return encoder.encodeByte((Byte) o); |
| } else if (o instanceof Boolean) { |
| return encoder.encodeBoolean((Boolean) o); |
| } else if (o instanceof Float) { |
| return encoder.encodeFloat((Float) o); |
| } else if (o instanceof Double) { |
| return encoder.encodeDouble((Double) o); |
| } else if (o instanceof Enum) { |
| return encoder.encodeInt(((Enum<?>) o).ordinal()); |
| } |
| } catch (IOException ioe) { |
| throw new RuntimeException(ioe); |
| } |
| |
| throw new IllegalArgumentException("Uknown type " + o.getClass().getName()); |
| } |
| |
| private BatchWriter getBatchWriter() throws IOException { |
| if (batchWriter == null) |
| try { |
| BatchWriterConfig batchWriterConfig = new BatchWriterConfig(); |
| batchWriterConfig.setMaxMemory(10000000); |
| batchWriterConfig.setMaxLatency(60000l, TimeUnit.MILLISECONDS); |
| batchWriterConfig.setMaxWriteThreads(4); |
| batchWriter = conn.createBatchWriter(mapping.tableName, batchWriterConfig); |
| } catch (TableNotFoundException e) { |
| throw new IOException(e); |
| } |
| return batchWriter; |
| } |
| |
| @Override |
| public void initialize(Class<K> keyClass, Class<T> persistentClass, Properties properties) { |
| try{ |
| super.initialize(keyClass, persistentClass, properties); |
| |
| String mock = DataStoreFactory.findProperty(properties, this, MOCK_PROPERTY, null); |
| String mappingFile = DataStoreFactory.getMappingFile(properties, this, DEFAULT_MAPPING_FILE); |
| String user = DataStoreFactory.findProperty(properties, this, USERNAME_PROPERTY, null); |
| String password = DataStoreFactory.findProperty(properties, this, PASSWORD_PROPERTY, null); |
| |
| mapping = readMapping(mappingFile); |
| |
| if (mapping.encoder == null || mapping.encoder.equals("")) { |
| encoder = new BinaryEncoder(); |
| } else { |
| try { |
| encoder = (Encoder) getClass().getClassLoader().loadClass(mapping.encoder).newInstance(); |
| } catch (InstantiationException e) { |
| throw new IOException(e); |
| } catch (IllegalAccessException e) { |
| throw new IOException(e); |
| } catch (ClassNotFoundException e) { |
| throw new IOException(e); |
| } |
| } |
| |
| try { |
| AuthenticationToken token = new PasswordToken(password); |
| if (mock == null || !mock.equals("true")) { |
| String instance = DataStoreFactory.findProperty(properties, this, INSTANCE_NAME_PROPERTY, null); |
| String zookeepers = DataStoreFactory.findProperty(properties, this, ZOOKEEPERS_NAME_PROPERTY, null); |
| credentials = new TCredentials(user, |
| "org.apache.accumulo.core.client.security.tokens.PasswordToken", |
| ByteBuffer.wrap(password.getBytes()), instance); |
| conn = new ZooKeeperInstance(instance, zookeepers).getConnector(user, token); |
| } else { |
| conn = new MockInstance().getConnector(user, new PasswordToken(password)); |
| credentials = new TCredentials(user, |
| "org.apache.accumulo.core.client.security.tokens.PasswordToken", |
| ByteBuffer.wrap(password.getBytes()), conn.getInstance().getInstanceID()); |
| } |
| |
| if (autoCreateSchema) |
| createSchema(); |
| } catch (AccumuloException e) { |
| throw new IOException(e); |
| } catch (AccumuloSecurityException e) { |
| throw new IOException(e); |
| } |
| }catch(IOException e){ |
| LOG.error(e.getMessage(), e); |
| } |
| } |
| |
| protected AccumuloMapping readMapping(String filename) throws IOException { |
| try { |
| |
| AccumuloMapping mapping = new AccumuloMapping(); |
| |
| DocumentBuilder db = DocumentBuilderFactory.newInstance().newDocumentBuilder(); |
| Document dom = db.parse(getClass().getClassLoader().getResourceAsStream(filename)); |
| |
| Element root = dom.getDocumentElement(); |
| |
| NodeList nl = root.getElementsByTagName("class"); |
| for (int i = 0; i < nl.getLength(); i++) { |
| |
| Element classElement = (Element) nl.item(i); |
| if (classElement.getAttribute("keyClass").equals(keyClass.getCanonicalName()) |
| && classElement.getAttribute("name").equals(persistentClass.getCanonicalName())) { |
| |
| mapping.tableName = getSchemaName(classElement.getAttribute("table"), persistentClass); |
| mapping.encoder = classElement.getAttribute("encoder"); |
| |
| NodeList fields = classElement.getElementsByTagName("field"); |
| for (int j = 0; j < fields.getLength(); j++) { |
| Element fieldElement = (Element) fields.item(j); |
| |
| String name = fieldElement.getAttribute("name"); |
| String family = fieldElement.getAttribute("family"); |
| String qualifier = fieldElement.getAttribute("qualifier"); |
| if (qualifier.equals("")) |
| qualifier = null; |
| |
| Pair<Text,Text> col = new Pair<Text,Text>(new Text(family), qualifier == null ? null : new Text(qualifier)); |
| mapping.fieldMap.put(name, col); |
| mapping.columnMap.put(col, name); |
| } |
| } |
| |
| } |
| |
| if (mapping.tableName == null) { |
| throw new GoraException("Please define the accumulo 'table' name mapping in " + filename + " for " + persistentClass.getCanonicalName()); |
| } |
| |
| nl = root.getElementsByTagName("table"); |
| for (int i = 0; i < nl.getLength(); i++) { |
| Element tableElement = (Element) nl.item(i); |
| if (tableElement.getAttribute("name").equals(mapping.tableName)) { |
| NodeList configs = tableElement.getElementsByTagName("config"); |
| for (int j = 0; j < configs.getLength(); j++) { |
| Element configElement = (Element) configs.item(j); |
| String key = configElement.getAttribute("key"); |
| String val = configElement.getAttribute("value"); |
| mapping.tableConfig.put(key, val); |
| } |
| } |
| } |
| |
| return mapping; |
| } catch (Exception ex) { |
| throw new IOException("Unable to read " + filename, ex); |
| } |
| |
| } |
| |
| @Override |
| public String getSchemaName() { |
| return mapping.tableName; |
| } |
| |
| @Override |
| public void createSchema() { |
| try { |
| conn.tableOperations().create(mapping.tableName); |
| Set<Entry<String,String>> es = mapping.tableConfig.entrySet(); |
| for (Entry<String,String> entry : es) { |
| conn.tableOperations().setProperty(mapping.tableName, entry.getKey(), entry.getValue()); |
| } |
| |
| } catch (AccumuloException e) { |
| LOG.error(e.getMessage(), e); |
| } catch (AccumuloSecurityException e) { |
| LOG.error(e.getMessage(), e); |
| } catch (TableExistsException e) { |
| LOG.error(e.getMessage(), e); |
| } |
| } |
| |
| @Override |
| public void deleteSchema() { |
| try { |
| if (batchWriter != null) |
| batchWriter.close(); |
| batchWriter = null; |
| conn.tableOperations().delete(mapping.tableName); |
| } catch (AccumuloException e) { |
| LOG.error(e.getMessage(), e); |
| } catch (AccumuloSecurityException e) { |
| LOG.error(e.getMessage(), e); |
| } catch (TableNotFoundException e) { |
| LOG.error(e.getMessage(), e); |
| } |
| } |
| |
| @Override |
| public boolean schemaExists() { |
| return conn.tableOperations().exists(mapping.tableName); |
| } |
| |
| public ByteSequence populate(Iterator<Entry<Key,Value>> iter, T persistent) throws IOException { |
| ByteSequence row = null; |
| |
| Map<Utf8, Object> currentMap = null; |
| List currentArray = null; |
| Text currentFam = null; |
| int currentPos = 0; |
| Schema currentSchema = null; |
| Field currentField = null; |
| |
| BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(new byte[0], null); |
| |
| while (iter.hasNext()) { |
| Entry<Key,Value> entry = iter.next(); |
| |
| if (row == null) { |
| row = entry.getKey().getRowData(); |
| } |
| byte[] val = entry.getValue().get(); |
| |
| Field field = fieldMap.get(getFieldName(entry)); |
| |
| if (currentMap != null) { |
| if (currentFam.equals(entry.getKey().getColumnFamily())) { |
| currentMap.put(new Utf8(entry.getKey().getColumnQualifierData().toArray()), |
| fromBytes(currentSchema, entry.getValue().get())); |
| continue; |
| } else { |
| persistent.put(currentPos, currentMap); |
| currentMap = null; |
| } |
| } else if (currentArray != null) { |
| if (currentFam.equals(entry.getKey().getColumnFamily())) { |
| currentArray.add(fromBytes(currentSchema, entry.getValue().get())); |
| continue; |
| } else { |
| persistent.put(currentPos, new GenericData.Array<T>(currentField.schema(), currentArray)); |
| currentArray = null; |
| } |
| } |
| |
| switch (field.schema().getType()) { |
| case MAP: // first entry only. Next are handled above on the next loop |
| currentMap = new DirtyMapWrapper<Utf8, Object>(new HashMap<Utf8, Object>()); |
| currentPos = field.pos(); |
| currentFam = entry.getKey().getColumnFamily(); |
| currentSchema = field.schema().getValueType(); |
| |
| currentMap.put(new Utf8(entry.getKey().getColumnQualifierData().toArray()), |
| fromBytes(currentSchema, entry.getValue().get())); |
| break; |
| case ARRAY: |
| currentArray = new DirtyListWrapper<Object>(new ArrayList<Object>()); |
| currentPos = field.pos(); |
| currentFam = entry.getKey().getColumnFamily(); |
| currentSchema = field.schema().getElementType(); |
| currentField = field; |
| |
| currentArray.add(fromBytes(currentSchema, entry.getValue().get())); |
| |
| break; |
| case UNION:// default value of null acts like union with null |
| Schema effectiveSchema = field.schema().getTypes() |
| .get(firstNotNullSchemaTypeIndex(field.schema())); |
| // map and array were coded without union index so need to be read the same way |
| if (effectiveSchema.getType() == Type.ARRAY) { |
| currentArray = new DirtyListWrapper<Object>(new ArrayList<Object>()); |
| currentPos = field.pos(); |
| currentFam = entry.getKey().getColumnFamily(); |
| currentSchema = field.schema().getElementType(); |
| currentField = field; |
| |
| currentArray.add(fromBytes(currentSchema, entry.getValue().get())); |
| break; |
| } |
| else if (effectiveSchema.getType() == Type.MAP) { |
| currentMap = new DirtyMapWrapper<Utf8, Object>(new HashMap<Utf8, Object>()); |
| currentPos = field.pos(); |
| currentFam = entry.getKey().getColumnFamily(); |
| currentSchema = effectiveSchema.getValueType(); |
| |
| currentMap.put(new Utf8(entry.getKey().getColumnQualifierData().toArray()), |
| fromBytes(currentSchema, entry.getValue().get())); |
| break; |
| } |
| // continue like a regular top-level union |
| case RECORD: |
| SpecificDatumReader<?> reader = new SpecificDatumReader<Schema>(field.schema()); |
| persistent.put(field.pos(), reader.read(null, DecoderFactory.get().binaryDecoder(val, decoder))); |
| break; |
| default: |
| persistent.put(field.pos(), fromBytes(field.schema(), entry.getValue().get())); |
| } |
| } |
| |
| if (currentMap != null) { |
| persistent.put(currentPos, currentMap); |
| } else if (currentArray != null) { |
| persistent.put(currentPos, new GenericData.Array<T>(currentField.schema(), currentArray)); |
| } |
| |
| persistent.clearDirty(); |
| |
| return row; |
| } |
| |
| /** |
| * Retrieve field name from entry. |
| * @param entry The Key-Value entry |
| * @return String The field name |
| */ |
| private String getFieldName(Entry<Key, Value> entry) { |
| String fieldName = mapping.columnMap.get(new Pair<Text,Text>(entry.getKey().getColumnFamily(), |
| entry.getKey().getColumnQualifier())); |
| if (fieldName == null) { |
| fieldName = mapping.columnMap.get(new Pair<Text,Text>(entry.getKey().getColumnFamily(), null)); |
| } |
| return fieldName; |
| } |
| |
| private void setFetchColumns(Scanner scanner, String fields[]) { |
| fields = getFieldsToQuery(fields); |
| for (String field : fields) { |
| Pair<Text,Text> col = mapping.fieldMap.get(field); |
| if (col != null) { |
| if (col.getSecond() == null) { |
| scanner.fetchColumnFamily(col.getFirst()); |
| } else { |
| scanner.fetchColumn(col.getFirst(), col.getSecond()); |
| } |
| } else { |
| LOG.error("Mapping not found for field: " + field); |
| } |
| } |
| } |
| |
| @Override |
| public T get(K key, String[] fields) { |
| try { |
| // TODO make isolated scanner optional? |
| Scanner scanner = new IsolatedScanner(conn.createScanner(mapping.tableName, Constants.NO_AUTHS)); |
| Range rowRange = new Range(new Text(toBytes(key))); |
| |
| scanner.setRange(rowRange); |
| setFetchColumns(scanner, fields); |
| |
| T persistent = newPersistent(); |
| ByteSequence row = populate(scanner.iterator(), persistent); |
| if (row == null) |
| return null; |
| return persistent; |
| } catch (TableNotFoundException e) { |
| LOG.error(e.getMessage(), e); |
| return null; |
| } catch (IOException e) { |
| LOG.error(e.getMessage(), e); |
| return null; |
| } |
| } |
| |
| @Override |
| public void put(K key, T val) { |
| |
| try{ |
| Mutation m = new Mutation(new Text(toBytes(key))); |
| |
| Schema schema = val.getSchema(); |
| List<Field> fields = schema.getFields(); |
| int count = 0; |
| |
| for (int i = 1; i < fields.size(); i++) { |
| if (!val.isDirty(i)) { |
| continue; |
| } |
| Field field = fields.get(i); |
| |
| Object o = val.get(field.pos()); |
| |
| Pair<Text,Text> col = mapping.fieldMap.get(field.name()); |
| |
| if (col == null) { |
| throw new GoraException("Please define the gora to accumulo mapping for field " + field.name()); |
| } |
| |
| switch (field.schema().getType()) { |
| case MAP: |
| count = putMap(m, count, field.schema().getValueType(), o, col); |
| break; |
| case ARRAY: |
| count = putArray(m, count, o, col); |
| break; |
| case UNION: // default value of null acts like union with null |
| Schema effectiveSchema = field.schema().getTypes() |
| .get(firstNotNullSchemaTypeIndex(field.schema())); |
| // map and array need to compute qualifier |
| if (effectiveSchema.getType() == Type.ARRAY) { |
| count = putArray(m, count, o, col); |
| break; |
| } |
| else if (effectiveSchema.getType() == Type.MAP) { |
| count = putMap(m, count, effectiveSchema.getValueType(), o, col); |
| break; |
| } |
| // continue like a regular top-level union |
| case RECORD: |
| SpecificDatumWriter<Object> writer = new SpecificDatumWriter<Object>(field.schema()); |
| ByteArrayOutputStream os = new ByteArrayOutputStream(); |
| org.apache.avro.io.BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(os, null); |
| writer.write(o, encoder); |
| encoder.flush(); |
| m.put(col.getFirst(), col.getSecond(), new Value(os.toByteArray())); |
| count++; |
| break; |
| default: |
| m.put(col.getFirst(), col.getSecond(), new Value(toBytes(o))); |
| count++; |
| } |
| |
| } |
| |
| if (count > 0) |
| try { |
| getBatchWriter().addMutation(m); |
| } catch (MutationsRejectedException e) { |
| LOG.error(e.getMessage(), e); |
| } |
| } catch (IOException e) { |
| LOG.error(e.getMessage(), e); |
| } |
| } |
| |
| private int putMap(Mutation m, int count, Schema valueType, Object o, Pair<Text, Text> col) throws GoraException { |
| |
| // First of all we delete map field on accumulo store |
| Text rowKey = new Text(m.getRow()); |
| Query<K, T> query = newQuery(); |
| query.setFields(col.getFirst().toString()); |
| query.setStartKey((K)rowKey.toString()); |
| query.setEndKey((K)rowKey.toString()); |
| deleteByQuery(query); |
| flush(); |
| if (o == null){ |
| return 0; |
| } |
| |
| Set<?> es = ((Map<?, ?>)o).entrySet(); |
| for (Object entry : es) { |
| Object mapKey = ((Entry<?, ?>) entry).getKey(); |
| Object mapVal = ((Entry<?, ?>) entry).getValue(); |
| if ((o instanceof DirtyMapWrapper && ((DirtyMapWrapper<?, ?>)o).isDirty()) |
| || !(o instanceof DirtyMapWrapper)) { //mapVal instanceof Dirtyable && ((Dirtyable)mapVal).isDirty()) { |
| m.put(col.getFirst(), new Text(toBytes(mapKey)), new Value(toBytes(valueType, mapVal))); |
| count++; |
| } |
| // TODO map value deletion |
| } |
| return count; |
| } |
| |
| private int putArray(Mutation m, int count, Object o, Pair<Text, Text> col) { |
| |
| // First of all we delete array field on accumulo store |
| Text rowKey = new Text(m.getRow()); |
| Query<K, T> query = newQuery(); |
| query.setFields(col.getFirst().toString()); |
| query.setStartKey((K)rowKey.toString()); |
| query.setEndKey((K)rowKey.toString()); |
| deleteByQuery(query); |
| flush(); |
| if (o == null){ |
| return 0; |
| } |
| |
| List<?> array = (List<?>) o; // both GenericArray and DirtyListWrapper |
| int j = 0; |
| for (Object item : array) { |
| m.put(col.getFirst(), new Text(toBytes(j++)), new Value(toBytes(item))); |
| count++; |
| } |
| return count; |
| } |
| |
| @Override |
| public boolean delete(K key) { |
| Query<K,T> q = newQuery(); |
| q.setKey(key); |
| return deleteByQuery(q) > 0; |
| } |
| |
| @Override |
| public long deleteByQuery(Query<K,T> query) { |
| try { |
| Scanner scanner = createScanner(query); |
| // add iterator that drops values on the server side |
| scanner.addScanIterator(new IteratorSetting(Integer.MAX_VALUE, SortedKeyIterator.class)); |
| RowIterator iterator = new RowIterator(scanner.iterator()); |
| |
| long count = 0; |
| |
| while (iterator.hasNext()) { |
| Iterator<Entry<Key,Value>> row = iterator.next(); |
| Mutation m = null; |
| while (row.hasNext()) { |
| Entry<Key,Value> entry = row.next(); |
| Key key = entry.getKey(); |
| if (m == null) |
| m = new Mutation(key.getRow()); |
| // TODO optimize to avoid continually creating column vis? prob does not matter for empty |
| m.putDelete(key.getColumnFamily(), key.getColumnQualifier(), new ColumnVisibility(key.getColumnVisibility()), key.getTimestamp()); |
| } |
| getBatchWriter().addMutation(m); |
| count++; |
| } |
| |
| return count; |
| } catch (TableNotFoundException e) { |
| // TODO return 0? |
| LOG.error(e.getMessage(), e); |
| return 0; |
| } catch (MutationsRejectedException e) { |
| LOG.error(e.getMessage(), e); |
| return 0; |
| } catch (IOException e){ |
| LOG.error(e.getMessage(), e); |
| return 0; |
| } |
| } |
| |
| private Range createRange(Query<K,T> query) { |
| Text startRow = null; |
| Text endRow = null; |
| |
| if (query.getStartKey() != null) |
| startRow = new Text(toBytes(query.getStartKey())); |
| |
| if (query.getEndKey() != null) |
| endRow = new Text(toBytes(query.getEndKey())); |
| |
| return new Range(startRow, true, endRow, true); |
| |
| } |
| |
| private Scanner createScanner(Query<K,T> query) throws TableNotFoundException { |
| // TODO make isolated scanner optional? |
| Scanner scanner = new IsolatedScanner(conn.createScanner(mapping.tableName, Constants.NO_AUTHS)); |
| setFetchColumns(scanner, query.getFields()); |
| |
| scanner.setRange(createRange(query)); |
| |
| if (query.getStartTime() != -1 || query.getEndTime() != -1) { |
| IteratorSetting is = new IteratorSetting(30, TimestampFilter.class); |
| if (query.getStartTime() != -1) |
| TimestampFilter.setStart(is, query.getStartTime(), true); |
| if (query.getEndTime() != -1) |
| TimestampFilter.setEnd(is, query.getEndTime(), true); |
| |
| scanner.addScanIterator(is); |
| } |
| |
| return scanner; |
| } |
| |
| @Override |
| public Result<K,T> execute(Query<K,T> query) { |
| try { |
| Scanner scanner = createScanner(query); |
| return new AccumuloResult<K,T>(this, query, scanner); |
| } catch (TableNotFoundException e) { |
| // TODO return empty result? |
| LOG.error(e.getMessage(), e); |
| return null; |
| } |
| } |
| |
| @Override |
| public Query<K,T> newQuery() { |
| return new AccumuloQuery<K,T>(this); |
| } |
| |
| Text pad(Text key, int bytes) { |
| if (key.getLength() < bytes) |
| key = new Text(key); |
| |
| while (key.getLength() < bytes) { |
| key.append(new byte[] {0}, 0, 1); |
| } |
| |
| return key; |
| } |
| |
| @Override |
| public List<PartitionQuery<K,T>> getPartitions(Query<K,T> query) throws IOException { |
| try { |
| TabletLocator tl; |
| if (conn instanceof MockConnector) |
| tl = new MockTabletLocator(); |
| else |
| tl = TabletLocator.getInstance(conn.getInstance(), new Text(Tables.getTableId(conn.getInstance(), mapping.tableName))); |
| |
| Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new HashMap<String,Map<KeyExtent,List<Range>>>(); |
| |
| tl.invalidateCache(); |
| while (tl.binRanges(Collections.singletonList(createRange(query)), binnedRanges, credentials).size() > 0) { |
| // TODO log? |
| if (!Tables.exists(conn.getInstance(), Tables.getTableId(conn.getInstance(), mapping.tableName))) |
| throw new TableDeletedException(Tables.getTableId(conn.getInstance(), mapping.tableName)); |
| else if (Tables.getTableState(conn.getInstance(), Tables.getTableId(conn.getInstance(), mapping.tableName)) == TableState.OFFLINE) |
| throw new TableOfflineException(conn.getInstance(), Tables.getTableId(conn.getInstance(), mapping.tableName)); |
| UtilWaitThread.sleep(100); |
| tl.invalidateCache(); |
| } |
| |
| List<PartitionQuery<K,T>> ret = new ArrayList<PartitionQuery<K,T>>(); |
| |
| Text startRow = null; |
| Text endRow = null; |
| if (query.getStartKey() != null) |
| startRow = new Text(toBytes(query.getStartKey())); |
| if (query.getEndKey() != null) |
| endRow = new Text(toBytes(query.getEndKey())); |
| |
| //hadoop expects hostnames, accumulo keeps track of IPs... so need to convert |
| HashMap<String,String> hostNameCache = new HashMap<String,String>(); |
| |
| for (Entry<String,Map<KeyExtent,List<Range>>> entry : binnedRanges.entrySet()) { |
| String ip = entry.getKey().split(":", 2)[0]; |
| String location = hostNameCache.get(ip); |
| if (location == null) { |
| InetAddress inetAddress = InetAddress.getByName(ip); |
| location = inetAddress.getHostName(); |
| hostNameCache.put(ip, location); |
| } |
| |
| Map<KeyExtent,List<Range>> tablets = entry.getValue(); |
| for (KeyExtent ke : tablets.keySet()) { |
| |
| K startKey = null; |
| if (startRow == null || !ke.contains(startRow)) { |
| if (ke.getPrevEndRow() != null) { |
| startKey = followingKey(encoder, getKeyClass(), TextUtil.getBytes(ke.getPrevEndRow())); |
| } |
| } else { |
| startKey = fromBytes(getKeyClass(), TextUtil.getBytes(startRow)); |
| } |
| |
| K endKey = null; |
| if (endRow == null || !ke.contains(endRow)) { |
| if (ke.getEndRow() != null) |
| endKey = lastPossibleKey(encoder, getKeyClass(), TextUtil.getBytes(ke.getEndRow())); |
| } else { |
| endKey = fromBytes(getKeyClass(), TextUtil.getBytes(endRow)); |
| } |
| |
| PartitionQueryImpl<K, T> pqi = new PartitionQueryImpl<K,T>(query, startKey, endKey, new String[] {location}); |
| pqi.setConf(getConf()); |
| ret.add(pqi); |
| } |
| } |
| |
| return ret; |
| } catch (TableNotFoundException e) { |
| throw new IOException(e); |
| } catch (AccumuloException e) { |
| throw new IOException(e); |
| } catch (AccumuloSecurityException e) { |
| throw new IOException(e); |
| } |
| |
| } |
| |
| static <K> K lastPossibleKey(Encoder encoder, Class<K> clazz, byte[] er) { |
| |
| if (clazz.equals(Byte.TYPE) || clazz.equals(Byte.class)) { |
| throw new UnsupportedOperationException(); |
| } else if (clazz.equals(Boolean.TYPE) || clazz.equals(Boolean.class)) { |
| throw new UnsupportedOperationException(); |
| } else if (clazz.equals(Short.TYPE) || clazz.equals(Short.class)) { |
| return fromBytes(encoder, clazz, encoder.lastPossibleKey(2, er)); |
| } else if (clazz.equals(Integer.TYPE) || clazz.equals(Integer.class)) { |
| return fromBytes(encoder, clazz, encoder.lastPossibleKey(4, er)); |
| } else if (clazz.equals(Long.TYPE) || clazz.equals(Long.class)) { |
| return fromBytes(encoder, clazz, encoder.lastPossibleKey(8, er)); |
| } else if (clazz.equals(Float.TYPE) || clazz.equals(Float.class)) { |
| return fromBytes(encoder, clazz, encoder.lastPossibleKey(4, er)); |
| } else if (clazz.equals(Double.TYPE) || clazz.equals(Double.class)) { |
| return fromBytes(encoder, clazz, encoder.lastPossibleKey(8, er)); |
| } else if (clazz.equals(String.class)) { |
| throw new UnsupportedOperationException(); |
| } else if (clazz.equals(Utf8.class)) { |
| return fromBytes(encoder, clazz, er); |
| } |
| |
| throw new IllegalArgumentException("Unknown type " + clazz.getName()); |
| } |
| |
| |
| |
| /** |
| * @param keyClass |
| * @param bytes |
| * @return |
| */ |
| @SuppressWarnings("unchecked") |
| static <K> K followingKey(Encoder encoder, Class<K> clazz, byte[] per) { |
| |
| if (clazz.equals(Byte.TYPE) || clazz.equals(Byte.class)) { |
| return (K) Byte.valueOf(encoder.followingKey(1, per)[0]); |
| } else if (clazz.equals(Boolean.TYPE) || clazz.equals(Boolean.class)) { |
| throw new UnsupportedOperationException(); |
| } else if (clazz.equals(Short.TYPE) || clazz.equals(Short.class)) { |
| return fromBytes(encoder, clazz, encoder.followingKey(2, per)); |
| } else if (clazz.equals(Integer.TYPE) || clazz.equals(Integer.class)) { |
| return fromBytes(encoder, clazz, encoder.followingKey(4, per)); |
| } else if (clazz.equals(Long.TYPE) || clazz.equals(Long.class)) { |
| return fromBytes(encoder, clazz, encoder.followingKey(8, per)); |
| } else if (clazz.equals(Float.TYPE) || clazz.equals(Float.class)) { |
| return fromBytes(encoder, clazz, encoder.followingKey(4, per)); |
| } else if (clazz.equals(Double.TYPE) || clazz.equals(Double.class)) { |
| return fromBytes(encoder, clazz, encoder.followingKey(8, per)); |
| } else if (clazz.equals(String.class)) { |
| throw new UnsupportedOperationException(); |
| } else if (clazz.equals(Utf8.class)) { |
| return fromBytes(encoder, clazz, Arrays.copyOf(per, per.length + 1)); |
| } |
| |
| throw new IllegalArgumentException("Unknown type " + clazz.getName()); |
| } |
| |
| @Override |
| public void flush() { |
| try { |
| if (batchWriter != null) { |
| batchWriter.flush(); |
| } |
| } catch (MutationsRejectedException e) { |
| LOG.error(e.getMessage(), e); |
| } |
| } |
| |
| @Override |
| public void close() { |
| try { |
| if (batchWriter != null) { |
| batchWriter.close(); |
| batchWriter = null; |
| } |
| } catch (MutationsRejectedException e) { |
| LOG.error(e.getMessage(), e); |
| } |
| } |
| } |
