http://git-wip-us.apache.org/repos/asf/eagle/blob/6e919c2e/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/RowKeyLogReader.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/RowKeyLogReader.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/RowKeyLogReader.java index 1c16dc8..1f29121 100755 --- a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/RowKeyLogReader.java +++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/RowKeyLogReader.java @@ -31,12 +31,12 @@ import org.apache.hadoop.hbase.client.Result; import org.apache.eagle.log.entity.meta.EntityDefinition; public class RowKeyLogReader extends IndexLogReader { - private final EntityDefinition ed; - private final List<byte[]> rowkeys; + private final EntityDefinition ed; + private final List<byte[]> rowkeys; private final byte[][] qualifiers; private HTableInterface tbl; - private boolean isOpen = false; - private Result[] entityResult; + private boolean isOpen = false; + private Result[] entityResult; private int getIndex = -1; public RowKeyLogReader(EntityDefinition ed, byte[] rowkey) { @@ -46,38 +46,39 @@ public class RowKeyLogReader extends IndexLogReader { this.qualifiers = null; } - public RowKeyLogReader(EntityDefinition ed, byte[] rowkey,byte[][] qualifiers) { - this.ed = ed; - this.rowkeys = new ArrayList<>(); + public RowKeyLogReader(EntityDefinition ed, byte[] rowkey, byte[][] qualifiers) { + this.ed = ed; + this.rowkeys = new ArrayList<>(); this.rowkeys.add(rowkey); this.qualifiers = qualifiers; - } + } - public RowKeyLogReader(EntityDefinition ed, List<byte[]> rowkeys,byte[][] qualifiers) { - this.ed = ed; - this.rowkeys = rowkeys; + public RowKeyLogReader(EntityDefinition ed, List<byte[]> rowkeys, byte[][] qualifiers) { + this.ed = ed; + this.rowkeys = rowkeys; this.qualifiers = qualifiers; - } + } - @Override - public void open() throws IOException { - if (isOpen) - return; // silently return - try { - tbl = EagleConfigFactory.load().getHTable(ed.getTable()); - } catch (RuntimeException ex) { - throw new IOException(ex); - } - final byte[] family = ed.getColumnFamily().getBytes(); + @Override + public void open() throws IOException { + if (isOpen) { + return; // silently return + } + try { + tbl = EagleConfigFactory.load().getHTable(ed.getTable()); + } catch (RuntimeException ex) { + throw new IOException(ex); + } + final byte[] family = ed.getColumnFamily().getBytes(); List<Get> gets = new ArrayList<>(this.rowkeys.size()); - for(byte[] rowkey:rowkeys) { + for (byte[] rowkey : rowkeys) { Get get = new Get(rowkey); get.addFamily(family); - if(qualifiers != null) { - for(byte[] qualifier: qualifiers){ - get.addColumn(family,qualifier); + if (qualifiers != null) { + for (byte[] qualifier : qualifiers) { + get.addColumn(family, qualifier); } } @@ -85,23 +86,23 @@ public class RowKeyLogReader extends IndexLogReader { } entityResult = tbl.get(gets); - isOpen = true; - } + isOpen = true; + } - @Override - public void close() throws IOException { - if(tbl != null){ - new HTableFactory().releaseHTableInterface(tbl); - } - } + @Override + public void close() throws IOException { + if (tbl != null) { + new HTableFactory().releaseHTableInterface(tbl); + } + } - @Override - public InternalLog read() throws IOException { - if(entityResult == null || entityResult.length == 0 || this.getIndex >= entityResult.length - 1){ + @Override + public InternalLog read() throws IOException { + if (entityResult == null || entityResult.length == 0 || this.getIndex >= entityResult.length - 1) { return null; } - getIndex ++; - InternalLog t = HBaseInternalLogHelper.parse(ed, entityResult[getIndex], this.qualifiers); - return t; - } -} \ No newline at end of file + getIndex++; + InternalLog t = HBaseInternalLogHelper.parse(ed, entityResult[getIndex], this.qualifiers); + return t; + } +}
http://git-wip-us.apache.org/repos/asf/eagle/blob/6e919c2e/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexLogReader.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexLogReader.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexLogReader.java index 8ff3448..855beaf 100755 --- a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexLogReader.java +++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexLogReader.java @@ -33,32 +33,34 @@ import org.apache.hadoop.hbase.filter.Filter; public class UniqueIndexLogReader extends IndexLogReader { - private final IndexDefinition indexDef; - private final List<byte[]> indexRowkeys; - private final byte[][] qualifiers; - private final Filter filter; - private HTableInterface tbl; - private boolean isOpen = false; - private Result[] entityResults; - private int index = -1; + private final IndexDefinition indexDef; + private final List<byte[]> indexRowkeys; + private final byte[][] qualifiers; + private final Filter filter; + private HTableInterface tbl; + private boolean isOpen = false; + private Result[] entityResults; + private int index = -1; - public UniqueIndexLogReader(IndexDefinition indexDef, List<byte[]> indexRowkeys, byte[][] qualifiers, Filter filter) { - this.indexDef = indexDef; - this.indexRowkeys = indexRowkeys; - this.qualifiers = qualifiers; - this.filter = filter; - } + public UniqueIndexLogReader(IndexDefinition indexDef, List<byte[]> indexRowkeys, byte[][] qualifiers, + Filter filter) { + this.indexDef = indexDef; + this.indexRowkeys = indexRowkeys; + this.qualifiers = qualifiers; + this.filter = filter; + } - @Override - public void open() throws IOException { - if (isOpen) - return; // silently return - try { - tbl = EagleConfigFactory.load().getHTable(indexDef.getEntityDefinition().getTable()); - } catch (RuntimeException ex) { - throw new IOException(ex); - } - final byte[] family = indexDef.getEntityDefinition().getColumnFamily().getBytes(); + @Override + public void open() throws IOException { + if (isOpen) { + return; // silently return + } + try { + tbl = EagleConfigFactory.load().getHTable(indexDef.getEntityDefinition().getTable()); + } catch (RuntimeException ex) { + throw new IOException(ex); + } + final byte[] family = indexDef.getEntityDefinition().getColumnFamily().getBytes(); final List<Get> indexGets = new ArrayList<>(); for (byte[] rowkey : indexRowkeys) { Get get = new Get(rowkey); @@ -69,56 +71,56 @@ public class UniqueIndexLogReader extends IndexLogReader { final Result[] indexResults = tbl.get(indexGets); indexGets.clear(); for (Result indexResult : indexResults) { - final NavigableMap<byte[], byte[]> map = indexResult.getFamilyMap(family); - if (map == null) { - continue; - } - for (byte[] entityRowkey : map.keySet()) { + final NavigableMap<byte[], byte[]> map = indexResult.getFamilyMap(family); + if (map == null) { + continue; + } + for (byte[] entityRowkey : map.keySet()) { Get get = new Get(entityRowkey); if (filter != null) { - get.setFilter(filter); + get.setFilter(filter); } - if(qualifiers == null) { - // filter all qualifiers if output qualifiers are null - get.addFamily(family); - }else { - for (int i = 0; i < qualifiers.length; ++i) { - // Return the specified qualifiers - get.addColumn(family, qualifiers[i]); - } - } - workaroundHBASE2198(get, filter,qualifiers); - indexGets.add(get); - } + if (qualifiers == null) { + // filter all qualifiers if output qualifiers are null + get.addFamily(family); + } else { + for (int i = 0; i < qualifiers.length; ++i) { + // Return the specified qualifiers + get.addColumn(family, qualifiers[i]); + } + } + workaroundHBASE2198(get, filter, qualifiers); + indexGets.add(get); + } } entityResults = tbl.get(indexGets); - isOpen = true; - } + isOpen = true; + } - @Override - public void close() throws IOException { - if(tbl != null){ - new HTableFactory().releaseHTableInterface(tbl); - } - } + @Override + public void close() throws IOException { + if (tbl != null) { + new HTableFactory().releaseHTableInterface(tbl); + } + } - @Override - public InternalLog read() throws IOException { - if (entityResults == null) { - throw new IllegalArgumentException("entityResults haven't been initialized before reading"); - } - InternalLog t = null; - while (entityResults.length > ++index) { - Result r = entityResults[index]; - if (r != null) { - if (r.getRow() == null) { - continue; - } - t = HBaseInternalLogHelper.parse(indexDef.getEntityDefinition(), r, qualifiers); - break; - } - } - return t; - } + @Override + public InternalLog read() throws IOException { + if (entityResults == null) { + throw new IllegalArgumentException("entityResults haven't been initialized before reading"); + } + InternalLog t = null; + while (entityResults.length > ++index) { + Result r = entityResults[index]; + if (r != null) { + if (r.getRow() == null) { + continue; + } + t = HBaseInternalLogHelper.parse(indexDef.getEntityDefinition(), r, qualifiers); + break; + } + } + return t; + } } http://git-wip-us.apache.org/repos/asf/eagle/blob/6e919c2e/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexStreamReader.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexStreamReader.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexStreamReader.java index 0391d57..cf94c11 100755 --- a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexStreamReader.java +++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexStreamReader.java @@ -27,26 +27,29 @@ import java.util.ArrayList; import java.util.List; public class UniqueIndexStreamReader extends IndexStreamReader { - public UniqueIndexStreamReader(IndexDefinition indexDef, SearchCondition condition) { - super(indexDef, condition, new ArrayList<byte[]>()); - final IndexType type = indexDef.canGoThroughIndex(condition.getQueryExpression(), indexRowkeys); - if (!IndexType.UNIQUE_INDEX.equals(type)) { - throw new IllegalArgumentException("This query can't go through index: " + condition.getQueryExpression()); - } - } + public UniqueIndexStreamReader(IndexDefinition indexDef, SearchCondition condition) { + super(indexDef, condition, new ArrayList<byte[]>()); + final IndexType type = indexDef.canGoThroughIndex(condition.getQueryExpression(), indexRowkeys); + if (!IndexType.UNIQUE_INDEX.equals(type)) { + throw new IllegalArgumentException("This query can't go through index: " + + condition.getQueryExpression()); + } + } - public UniqueIndexStreamReader(IndexDefinition indexDef, SearchCondition condition, List<byte[]> indexRowkeys) { - super(indexDef, condition, indexRowkeys); - } + public UniqueIndexStreamReader(IndexDefinition indexDef, SearchCondition condition, + List<byte[]> indexRowkeys) { + super(indexDef, condition, indexRowkeys); + } - @Override - protected LogReader createIndexReader() { - final EntityDefinition entityDef = indexDef.getEntityDefinition(); -// final - byte[][] outputQualifiers = null; - if(!condition.isOutputAll()) { - outputQualifiers = HBaseInternalLogHelper.getOutputQualifiers(entityDef, condition.getOutputFields()); - } - return new UniqueIndexLogReader(indexDef, indexRowkeys, outputQualifiers, condition.getFilter()); - } + @Override + protected LogReader createIndexReader() { + final EntityDefinition entityDef = indexDef.getEntityDefinition(); + // final + byte[][] outputQualifiers = null; + if (!condition.isOutputAll()) { + outputQualifiers = HBaseInternalLogHelper.getOutputQualifiers(entityDef, + condition.getOutputFields()); + } + return new UniqueIndexLogReader(indexDef, indexRowkeys, outputQualifiers, condition.getFilter()); + } } http://git-wip-us.apache.org/repos/asf/eagle/blob/6e919c2e/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/BooleanSerDeser.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/BooleanSerDeser.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/BooleanSerDeser.java index cf40e31..2e0882b 100755 --- a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/BooleanSerDeser.java +++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/BooleanSerDeser.java @@ -21,34 +21,35 @@ package org.apache.eagle.log.entity.meta; */ public class BooleanSerDeser implements EntitySerDeser<Boolean> { - public BooleanSerDeser(){} + public BooleanSerDeser() { + } - @Override - public Boolean deserialize(byte[] bytes){ - if(bytes != null && bytes.length > 0){ - if(bytes[0] == 0){ - return false; - }else if(bytes[0] == 1){ - return true; - } - } - return null; - } + @Override + public Boolean deserialize(byte[] bytes) { + if (bytes != null && bytes.length > 0) { + if (bytes[0] == 0) { + return false; + } else if (bytes[0] == 1) { + return true; + } + } + return null; + } - @Override - public byte[] serialize(Boolean obj){ - if(obj != null){ - if(obj){ - return new byte[]{1}; - }else{ - return new byte[]{0}; - } - } - return null; - } + @Override + public byte[] serialize(Boolean obj) { + if (obj != null) { + if (obj) { + return new byte[] { 1 }; + } else { + return new byte[] { 0 }; + } + } + return null; + } - @Override - public Class<Boolean> type() { - return Boolean.class; - } + @Override + public Class<Boolean> type() { + return Boolean.class; + } } http://git-wip-us.apache.org/repos/asf/eagle/blob/6e919c2e/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Column.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Column.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Column.java index b64e528..73978e9 100644 --- a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Column.java +++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Column.java @@ -21,8 +21,10 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; -@Target({ElementType.FIELD}) +@Target({ + ElementType.FIELD +}) @Retention(RetentionPolicy.RUNTIME) public @interface Column { - String value() default ""; + String value() default ""; } http://git-wip-us.apache.org/repos/asf/eagle/blob/6e919c2e/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/ColumnFamily.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/ColumnFamily.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/ColumnFamily.java index 6e3e9c6..cb03e65 100644 --- a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/ColumnFamily.java +++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/ColumnFamily.java @@ -21,8 +21,10 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; -@Target({ElementType.TYPE}) +@Target({ + ElementType.TYPE +}) @Retention(RetentionPolicy.RUNTIME) public @interface ColumnFamily { - String value() default "f"; + String value() default "f"; } http://git-wip-us.apache.org/repos/asf/eagle/blob/6e919c2e/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DefaultJavaObjctSerDeser.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DefaultJavaObjctSerDeser.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DefaultJavaObjctSerDeser.java index 24385a9..36fc63f 100644 --- a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DefaultJavaObjctSerDeser.java +++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DefaultJavaObjctSerDeser.java @@ -22,11 +22,11 @@ package org.apache.eagle.log.entity.meta; import org.apache.eagle.common.SerializableUtils; public class DefaultJavaObjctSerDeser implements EntitySerDeser<Object> { - public final static EntitySerDeser<Object> INSTANCE = new DefaultJavaObjctSerDeser(); + public static final EntitySerDeser<Object> INSTANCE = new DefaultJavaObjctSerDeser(); @Override public Object deserialize(byte[] bytes) { - return SerializableUtils.deserializeFromByteArray(bytes,"Deserialize from java object bytes"); + return SerializableUtils.deserializeFromByteArray(bytes, "Deserialize from java object bytes"); } @Override http://git-wip-us.apache.org/repos/asf/eagle/blob/6e919c2e/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Double2DArraySerDeser.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Double2DArraySerDeser.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Double2DArraySerDeser.java index 27b011c..b3af8a6 100644 --- a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Double2DArraySerDeser.java +++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Double2DArraySerDeser.java @@ -25,22 +25,23 @@ import java.io.IOException; * @since 7/22/15 */ public class Double2DArraySerDeser implements EntitySerDeser<double[][]> { - private final int SIZE = 8; + private static final int SIZE = 8; + @Override - public double[][] deserialize(byte[] bytes){ -// if((bytes.length-4) % SIZE != 0) -// return null; + public double[][] deserialize(byte[] bytes) { + // if((bytes.length-4) % SIZE != 0) + // return null; int offset = 0; // get size of int array int rowSize = ByteUtil.bytesToInt(bytes, offset); offset += 4; double[][] data = new double[rowSize][]; - for(int i=0; i<rowSize; i++) { + for (int i = 0; i < rowSize; i++) { int colSize = ByteUtil.bytesToInt(bytes, offset); offset += 4; double[] values = null; - if (colSize >= 0){ + if (colSize >= 0) { values = new double[colSize]; for (int j = 0; j < colSize; j++) { values[j] = ByteUtil.bytesToDouble(bytes, offset); @@ -54,27 +55,28 @@ public class Double2DArraySerDeser implements EntitySerDeser<double[][]> { } /** - * * @param obj * @return */ @Override - public byte[] serialize(double[][] obj){ - if(obj == null) return null; + public byte[] serialize(double[][] obj) { + if (obj == null) { + return null; + } ByteArrayOutputStream data = new ByteArrayOutputStream(); int size = obj.length; byte[] sizeBytes = ByteUtil.intToBytes(size); - data.write(sizeBytes,0,sizeBytes.length); + data.write(sizeBytes, 0, sizeBytes.length); - try{ - for(double[] o:obj){ - if(o!=null){ + try { + for (double[] o : obj) { + if (o != null) { data.write(ByteUtil.intToBytes(o.length)); - for(double d:o){ - data.write(ByteUtil.doubleToBytes(d),0,SIZE); + for (double d : o) { + data.write(ByteUtil.doubleToBytes(d), 0, SIZE); } - }else{ - data.write(ByteUtil.intToBytes(-1),0,4); + } else { + data.write(ByteUtil.intToBytes(-1), 0, 4); } } } catch (IOException e) { @@ -94,4 +96,4 @@ public class Double2DArraySerDeser implements EntitySerDeser<double[][]> { public Class<double[][]> type() { return double[][].class; } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/eagle/blob/6e919c2e/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleArraySerDeser.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleArraySerDeser.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleArraySerDeser.java index d87e31c..a8eb965 100755 --- a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleArraySerDeser.java +++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleArraySerDeser.java @@ -18,51 +18,54 @@ package org.apache.eagle.log.entity.meta; import org.apache.eagle.common.ByteUtil; -public class DoubleArraySerDeser implements EntitySerDeser<double[]>{ +public class DoubleArraySerDeser implements EntitySerDeser<double[]> { - public DoubleArraySerDeser(){} + private static final int SIZE = 8; - private final int SIZE = 8; - @Override - public double[] deserialize(byte[] bytes){ - if((bytes.length-4) % SIZE != 0) - return null; - int offset = 0; - // get size of int array - int size = ByteUtil.bytesToInt(bytes, offset); - offset += 4; - double[] values = new double[size]; - for(int i=0; i<size; i++){ - values[i] = ByteUtil.bytesToDouble(bytes, offset); - offset += SIZE; - } - return values; - } - - /** - * - * @param obj - * @return - */ - @Override - public byte[] serialize(double[] obj){ - if(obj == null) - return null; - int size = obj.length; - byte[] array = new byte[4 + SIZE*size]; - byte[] first = ByteUtil.intToBytes(size); - int offset = 0; - System.arraycopy(first, 0, array, offset, first.length); - offset += first.length; - for(int i=0; i<size; i++){ - System.arraycopy(ByteUtil.doubleToBytes(obj[i]), 0, array, offset, SIZE); - offset += SIZE; - } - return array; - } + public DoubleArraySerDeser() { + } - @Override - public Class<double[]> type() { - return double[].class; - } + @Override + public double[] deserialize(byte[] bytes) { + if ((bytes.length - 4) % SIZE != 0) { + return null; + } + int offset = 0; + // get size of int array + int size = ByteUtil.bytesToInt(bytes, offset); + offset += 4; + double[] values = new double[size]; + for (int i = 0; i < size; i++) { + values[i] = ByteUtil.bytesToDouble(bytes, offset); + offset += SIZE; + } + return values; + } + + /** + * @param obj + * @return + */ + @Override + public byte[] serialize(double[] obj) { + if (obj == null) { + return null; + } + int size = obj.length; + byte[] array = new byte[4 + SIZE * size]; + byte[] first = ByteUtil.intToBytes(size); + int offset = 0; + System.arraycopy(first, 0, array, offset, first.length); + offset += first.length; + for (int i = 0; i < size; i++) { + System.arraycopy(ByteUtil.doubleToBytes(obj[i]), 0, array, offset, SIZE); + offset += SIZE; + } + return array; + } + + @Override + public Class<double[]> type() { + return double[].class; + } } http://git-wip-us.apache.org/repos/asf/eagle/blob/6e919c2e/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleSerDeser.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleSerDeser.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleSerDeser.java index 330a99d..bf58d39 100755 --- a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleSerDeser.java +++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleSerDeser.java @@ -18,24 +18,26 @@ package org.apache.eagle.log.entity.meta; import org.apache.eagle.common.ByteUtil; -public class DoubleSerDeser implements EntitySerDeser<Double>{ +public class DoubleSerDeser implements EntitySerDeser<Double> { - @Override - public Double deserialize(byte[] bytes){ - if(bytes.length < 8) - return null; - return ByteUtil.bytesToDouble(bytes); - } - - @Override - public byte[] serialize(Double obj){ - if(obj == null) - return null; - return ByteUtil.doubleToBytes(obj); - } + @Override + public Double deserialize(byte[] bytes) { + if (bytes.length < 8) { + return null; + } + return ByteUtil.bytesToDouble(bytes); + } - @Override - public Class<Double> type(){ - return Double.class; - } + @Override + public byte[] serialize(Double obj) { + if (obj == null) { + return null; + } + return ByteUtil.doubleToBytes(obj); + } + + @Override + public Class<Double> type() { + return Double.class; + } } http://git-wip-us.apache.org/repos/asf/eagle/blob/6e919c2e/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/EntityDefinition.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/EntityDefinition.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/EntityDefinition.java index d2d9eef..c1f4a24 100755 --- a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/EntityDefinition.java +++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/EntityDefinition.java @@ -32,295 +32,333 @@ import java.util.HashMap; import java.util.Map; /** - * - * This object should be regarded as read-only metadata for an entity as it will be shared across all entity object - * with the same entity name, so don't try to set different values for any of the fields, - * otherwise it's not thread safe + * This object should be regarded as read-only metadata for an entity as it will be shared across all entity + * object with the same entity name, so don't try to set different values for any of the fields, otherwise + * it's not thread safe */ -public class EntityDefinition implements Writable{ - private final static Logger LOG = LoggerFactory.getLogger(EntityDefinition.class); - - private Class<? extends TaggedLogAPIEntity> entityClass; - private String table; - private String columnFamily; - // TODO prefix be within search/get condition instead of entity definition. Topology entity should have pre-defined prefix. - private String prefix; - private String service; - private String serviceCreationPath; - private String serviceDeletionPath; - private String[] partitions; - private Map<String, Qualifier> displayNameMap = new HashMap<String, Qualifier>(); - private Map<String, Qualifier> qualifierNameMap = new HashMap<String, Qualifier>(); - private Map<String, Method> qualifierGetterMap = new HashMap<String, Method>(); - private String[] tags; - private boolean isTimeSeries; - private MetricDefinition metricDefinition; - private IndexDefinition[] indexes; - - - public EntityDefinition(){} - - public MetricDefinition getMetricDefinition() { - return metricDefinition; - } - public void setMetricDefinition(MetricDefinition metricDefinition) { - this.metricDefinition = metricDefinition; - } - public boolean isTimeSeries() { - return isTimeSeries; - } - public void setTimeSeries(boolean isTimeSeries) { - this.isTimeSeries = isTimeSeries; - } - public String getColumnFamily() { - return columnFamily; - } - public void setColumnFamily(String columnFamily) { - this.columnFamily = columnFamily; - } - public Class<? extends TaggedLogAPIEntity> getEntityClass() { - return entityClass; - } - public void setEntityClass(Class<? extends TaggedLogAPIEntity> entityClass) { - this.entityClass = entityClass; - } - public String getTable() { - return table; - } - public void setTable(String table) { - this.table = table; - } - public Map<String, Qualifier> getDisplayNameMap() { - return displayNameMap; - } - public void setDisplayNameMap(Map<String, Qualifier> displayNameMap) { - this.displayNameMap = displayNameMap; - } - public Map<String, Qualifier> getQualifierNameMap() { - return qualifierNameMap; - } - public void setQualifierNameMap(Map<String, Qualifier> qualifierNameMap) { - this.qualifierNameMap = qualifierNameMap; - } - public String getPrefix() { - return prefix; - } - public void setPrefix(String prefix) { - this.prefix = prefix; - } - public String getService() { - return service; - } - public void setService(String service) { - this.service = service; - } - public String getServiceCreationPath() { - return serviceCreationPath; - } - public void setServiceCreationPath(String serviceCreationPath) { - this.serviceCreationPath = serviceCreationPath; - } - public String getServiceDeletionPath() { - return serviceDeletionPath; - } - public void setServiceDeletionPath(String serviceDeletionPath) { - this.serviceDeletionPath = serviceDeletionPath; - } - public String[] getPartitions() { - return partitions; - } - public void setPartitions(String[] partitions) { - this.partitions = partitions; - } - public IndexDefinition[] getIndexes() { - return indexes; - } - public void setIndexes(IndexDefinition[] indexes) { - this.indexes = indexes; - } - public Map<String, Method> getQualifierGetterMap() { - return qualifierGetterMap; - } - public void setQualifierGetterMap(Map<String, Method> qualifierGetterMap) { - this.qualifierGetterMap = qualifierGetterMap; - } - public String[] getTags(){ - return tags; - } - public void setTags(String[] tags){ - this.tags = tags; - } - -// public Map<String,String> getQualifierDisplayNameMap(){ -// Map<String,String> qualifierDisplayNameMap = new HashMap<String, String>(); -// for(Map.Entry<String,Qualifier> entry: qualifierNameMap.entrySet()){ -// qualifierDisplayNameMap.put(entry.getKey(),entry.getValue().getDisplayName()); -// } -// return qualifierDisplayNameMap; -// } - - /** - * a filed is a tag when this field is neither in qualifierNameMap nor in displayNameMap - * @param field - * @return - */ - public boolean isTag(String field){ - return (qualifierNameMap.get(field) == null && displayNameMap.get(field) == null); -// return (qualifierNameMap.get(field) == null); - } - - /** - * Check if the specified field is a partition tag field - */ - public boolean isPartitionTag(String field) { - if (partitions == null || (!isTag(field))) { - return false; - } - for (String partition : partitions) { - if (partition.equals(field)) { - return true; - } - } - return false; - - } - - public Object getValue(TaggedLogAPIEntity entity, String field) throws IllegalAccessException, IllegalArgumentException, InvocationTargetException { - if (!entityClass.equals(entity.getClass())) { - if ((entityClass.equals(GenericMetricEntity.class) && entity.getClass().equals(GenericMetricShadowEntity.class))) { - GenericMetricShadowEntity e = (GenericMetricShadowEntity)entity; - return e.getValue(); - } else { - throw new IllegalArgumentException("Invalid entity type: " + entity.getClass().getSimpleName()); - } - } - final Method m = qualifierGetterMap.get(field); - if (m == null) { - // The field is a tag - if (entity.getTags() != null) { - return entity.getTags().get(field); - } - } - if (m != null) { - return m.invoke(entity); - } - return null; - } - - - @Override - public void write(DataOutput out) throws IOException { - out.writeUTF(entityClass.getName()); - out.writeUTF(table); - out.writeUTF(columnFamily); - out.writeUTF(prefix); - out.writeUTF(service); - - int partitionsLen = 0; - if(partitions != null) partitionsLen =partitions.length; - out.writeInt(partitionsLen); - for (int i = 0; i < partitionsLen; i++) { - out.writeUTF(partitions[i]); - } - - int displayNameMapSize = displayNameMap.size(); - out.writeInt(displayNameMapSize); - for(Map.Entry<String,Qualifier> entry: displayNameMap.entrySet()){ - out.writeUTF(entry.getKey()); - entry.getValue().write(out); - } - - int qualifierNameMapSize = qualifierNameMap.size(); - out.writeInt(qualifierNameMapSize); - for(Map.Entry<String,Qualifier> entry: qualifierNameMap.entrySet()){ - out.writeUTF(entry.getKey()); - entry.getValue().write(out); - } - - // TODO: write qualifierGetterMap - out.writeBoolean(isTimeSeries); - - boolean hasMetricDefinition = metricDefinition != null; - out.writeBoolean(hasMetricDefinition); - if(hasMetricDefinition) { - // write MetricDefinition - metricDefinition.write(out); - } - - // TODO: write indexes - } - - - public void setEntityDefinition(EntityDefinition ed){ - this.entityClass = ed.getEntityClass(); - this.table = ed.getTable(); - this.columnFamily = ed.getColumnFamily(); - this.prefix = ed.getPrefix(); - this.service = ed.getService(); - this.partitions = ed.getPartitions(); - this.displayNameMap = ed.getDisplayNameMap(); - this.qualifierGetterMap = ed.getQualifierGetterMap(); +public class EntityDefinition implements Writable { + private static final Logger LOG = LoggerFactory.getLogger(EntityDefinition.class); + + private Class<? extends TaggedLogAPIEntity> entityClass; + private String table; + private String columnFamily; + // TODO prefix be within search/get condition instead of entity definition. Topology entity should have + // pre-defined prefix. + private String prefix; + private String service; + private String serviceCreationPath; + private String serviceDeletionPath; + private String[] partitions; + private Map<String, Qualifier> displayNameMap = new HashMap<String, Qualifier>(); + private Map<String, Qualifier> qualifierNameMap = new HashMap<String, Qualifier>(); + private Map<String, Method> qualifierGetterMap = new HashMap<String, Method>(); + private String[] tags; + private boolean isTimeSeries; + private MetricDefinition metricDefinition; + private IndexDefinition[] indexes; + + public EntityDefinition() { + } + + public MetricDefinition getMetricDefinition() { + return metricDefinition; + } + + public void setMetricDefinition(MetricDefinition metricDefinition) { + this.metricDefinition = metricDefinition; + } + + public boolean isTimeSeries() { + return isTimeSeries; + } + + public void setTimeSeries(boolean isTimeSeries) { + this.isTimeSeries = isTimeSeries; + } + + public String getColumnFamily() { + return columnFamily; + } + + public void setColumnFamily(String columnFamily) { + this.columnFamily = columnFamily; + } + + public Class<? extends TaggedLogAPIEntity> getEntityClass() { + return entityClass; + } + + public void setEntityClass(Class<? extends TaggedLogAPIEntity> entityClass) { + this.entityClass = entityClass; + } + + public String getTable() { + return table; + } + + public void setTable(String table) { + this.table = table; + } + + public Map<String, Qualifier> getDisplayNameMap() { + return displayNameMap; + } + + public void setDisplayNameMap(Map<String, Qualifier> displayNameMap) { + this.displayNameMap = displayNameMap; + } + + public Map<String, Qualifier> getQualifierNameMap() { + return qualifierNameMap; + } + + public void setQualifierNameMap(Map<String, Qualifier> qualifierNameMap) { + this.qualifierNameMap = qualifierNameMap; + } + + public String getPrefix() { + return prefix; + } + + public void setPrefix(String prefix) { + this.prefix = prefix; + } + + public String getService() { + return service; + } + + public void setService(String service) { + this.service = service; + } + + public String getServiceCreationPath() { + return serviceCreationPath; + } + + public void setServiceCreationPath(String serviceCreationPath) { + this.serviceCreationPath = serviceCreationPath; + } + + public String getServiceDeletionPath() { + return serviceDeletionPath; + } + + public void setServiceDeletionPath(String serviceDeletionPath) { + this.serviceDeletionPath = serviceDeletionPath; + } + + public String[] getPartitions() { + return partitions; + } + + public void setPartitions(String[] partitions) { + this.partitions = partitions; + } + + public IndexDefinition[] getIndexes() { + return indexes; + } + + public void setIndexes(IndexDefinition[] indexes) { + this.indexes = indexes; + } + + public Map<String, Method> getQualifierGetterMap() { + return qualifierGetterMap; + } + + public void setQualifierGetterMap(Map<String, Method> qualifierGetterMap) { + this.qualifierGetterMap = qualifierGetterMap; + } + + public String[] getTags() { + return tags; + } + + public void setTags(String[] tags) { + this.tags = tags; + } + + // public Map<String,String> getQualifierDisplayNameMap(){ + // Map<String,String> qualifierDisplayNameMap = new HashMap<String, String>(); + // for(Map.Entry<String,Qualifier> entry: qualifierNameMap.entrySet()){ + // qualifierDisplayNameMap.put(entry.getKey(),entry.getValue().getDisplayName()); + // } + // return qualifierDisplayNameMap; + // } + + /** + * a filed is a tag when this field is neither in qualifierNameMap nor in displayNameMap + * + * @param field + * @return + */ + public boolean isTag(String field) { + return (qualifierNameMap.get(field) == null && displayNameMap.get(field) == null); + // return (qualifierNameMap.get(field) == null); + } + + /** + * Check if the specified field is a partition tag field + */ + public boolean isPartitionTag(String field) { + if (partitions == null || (!isTag(field))) { + return false; + } + for (String partition : partitions) { + if (partition.equals(field)) { + return true; + } + } + return false; + + } + + public Object getValue(TaggedLogAPIEntity entity, String field) + throws IllegalAccessException, IllegalArgumentException, InvocationTargetException { + if (!entityClass.equals(entity.getClass())) { + if ((entityClass.equals(GenericMetricEntity.class) + && entity.getClass().equals(GenericMetricShadowEntity.class))) { + GenericMetricShadowEntity e = (GenericMetricShadowEntity)entity; + return e.getValue(); + } else { + throw new IllegalArgumentException("Invalid entity type: " + + entity.getClass().getSimpleName()); + } + } + final Method m = qualifierGetterMap.get(field); + if (m == null) { + // The field is a tag + if (entity.getTags() != null) { + return entity.getTags().get(field); + } + } + if (m != null) { + return m.invoke(entity); + } + return null; + } + + @Override + public void write(DataOutput out) throws IOException { + out.writeUTF(entityClass.getName()); + out.writeUTF(table); + out.writeUTF(columnFamily); + out.writeUTF(prefix); + out.writeUTF(service); + + int partitionsLen = 0; + if (partitions != null) { + partitionsLen = partitions.length; + } + out.writeInt(partitionsLen); + for (int i = 0; i < partitionsLen; i++) { + out.writeUTF(partitions[i]); + } + + int displayNameMapSize = displayNameMap.size(); + out.writeInt(displayNameMapSize); + for (Map.Entry<String, Qualifier> entry : displayNameMap.entrySet()) { + out.writeUTF(entry.getKey()); + entry.getValue().write(out); + } + + int qualifierNameMapSize = qualifierNameMap.size(); + out.writeInt(qualifierNameMapSize); + for (Map.Entry<String, Qualifier> entry : qualifierNameMap.entrySet()) { + out.writeUTF(entry.getKey()); + entry.getValue().write(out); + } + + // TODO: write qualifierGetterMap + out.writeBoolean(isTimeSeries); + + boolean hasMetricDefinition = metricDefinition != null; + out.writeBoolean(hasMetricDefinition); + if (hasMetricDefinition) { + // write MetricDefinition + metricDefinition.write(out); + } + + // TODO: write indexes + } + + public void setEntityDefinition(EntityDefinition ed) { + this.entityClass = ed.getEntityClass(); + this.table = ed.getTable(); + this.columnFamily = ed.getColumnFamily(); + this.prefix = ed.getPrefix(); + this.service = ed.getService(); + this.partitions = ed.getPartitions(); + this.displayNameMap = ed.getDisplayNameMap(); + this.qualifierGetterMap = ed.getQualifierGetterMap(); this.qualifierNameMap = ed.getQualifierNameMap(); - this.isTimeSeries = ed.isTimeSeries(); - this.metricDefinition = ed.metricDefinition; - this.indexes = ed.getIndexes(); - } - - ////////////////////////////////////////////// - // TODO: Cache object for reading in region side - ////////////////////////////////////////////// - // private final static Map<String,EntityDefinition> _classEntityDefinitionCache = new HashMap<String, EntityDefinition>(); - - @Override - public void readFields(DataInput in) throws IOException { - String entityClassName = in.readUTF(); -// EntityDefinition _cached = _classEntityDefinitionCache.get(entityClassName); -// if(_cached !=null){ -// setEntityDefinition(_cached); -// LOG.info("Got cached definition for entity: "+entityClassName); -// return; -// } - if(LOG.isDebugEnabled()) LOG.debug("Reading EntityDefinition entity: "+entityClassName); - try { - entityClass = (Class<? extends TaggedLogAPIEntity>) Class.forName(entityClassName); - } catch (Exception e) { - // ignore - } - table = in.readUTF(); - columnFamily = in.readUTF(); - prefix = in.readUTF(); - service = in.readUTF(); - - int partitionsLen = in.readInt(); - partitions = new String[partitionsLen]; - for (int i = 0; i < partitionsLen; i++) { - partitions[i] = in.readUTF(); - } - int displayNameMapSize = in.readInt(); - for(int i=0;i<displayNameMapSize;i++){ - String key = in.readUTF(); - Qualifier value = new Qualifier(); - value.readFields(in); - displayNameMap.put(key,value); - } - int qualifierNameMapSize = in.readInt(); - for(int i=0;i<qualifierNameMapSize;i++){ - String key = in.readUTF(); - Qualifier value = new Qualifier(); - value.readFields(in); - qualifierNameMap.put(key,value); - } - // TODO: readFields qualifierGetterMap - isTimeSeries = in.readBoolean(); - - // readFields MetricDefinition - boolean hasMetricDefinition = in.readBoolean(); - if(hasMetricDefinition) { - if(LOG.isDebugEnabled()) LOG.debug("reading metricDefinition"); - metricDefinition = new MetricDefinition(); - metricDefinition.readFields(in); - } - // TODO: readFields indexes -// _classEntityDefinitionCache.put(entityClassName,this); - } -} \ No newline at end of file + this.isTimeSeries = ed.isTimeSeries(); + this.metricDefinition = ed.metricDefinition; + this.indexes = ed.getIndexes(); + } + + ////////////////////////////////////////////// + // TODO: Cache object for reading in region side + ////////////////////////////////////////////// + // private static final Map<String,EntityDefinition> _classEntityDefinitionCache = new HashMap<String, + ////////////////////////////////////////////// EntityDefinition>(); + + @Override + public void readFields(DataInput in) throws IOException { + String entityClassName = in.readUTF(); + // EntityDefinition _cached = _classEntityDefinitionCache.get(entityClassName); + // if(_cached !=null){ + // setEntityDefinition(_cached); + // LOG.info("Got cached definition for entity: "+entityClassName); + // return; + // } + if (LOG.isDebugEnabled()) { + LOG.debug("Reading EntityDefinition entity: " + entityClassName); + } + try { + entityClass = (Class<? extends TaggedLogAPIEntity>)Class.forName(entityClassName); + } catch (Exception e) { + // ignore + } + table = in.readUTF(); + columnFamily = in.readUTF(); + prefix = in.readUTF(); + service = in.readUTF(); + + int partitionsLen = in.readInt(); + partitions = new String[partitionsLen]; + for (int i = 0; i < partitionsLen; i++) { + partitions[i] = in.readUTF(); + } + int displayNameMapSize = in.readInt(); + for (int i = 0; i < displayNameMapSize; i++) { + String key = in.readUTF(); + Qualifier value = new Qualifier(); + value.readFields(in); + displayNameMap.put(key, value); + } + int qualifierNameMapSize = in.readInt(); + for (int i = 0; i < qualifierNameMapSize; i++) { + String key = in.readUTF(); + Qualifier value = new Qualifier(); + value.readFields(in); + qualifierNameMap.put(key, value); + } + // TODO: readFields qualifierGetterMap + isTimeSeries = in.readBoolean(); + + // readFields MetricDefinition + boolean hasMetricDefinition = in.readBoolean(); + if (hasMetricDefinition) { + if (LOG.isDebugEnabled()) { + LOG.debug("reading metricDefinition"); + } + metricDefinition = new MetricDefinition(); + metricDefinition.readFields(in); + } + // TODO: readFields indexes + // _classEntityDefinitionCache.put(entityClassName,this); + } +}
