http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/DistinctCountBigDecimalAggregatorObjectSet.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/DistinctCountBigDecimalAggregatorObjectSet.java b/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/DistinctCountBigDecimalAggregatorObjectSet.java deleted file mode 100644 index 2f44bca..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/DistinctCountBigDecimalAggregatorObjectSet.java +++ /dev/null @@ -1,86 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.distinct; - -import java.math.BigDecimal; -import java.util.HashSet; -import java.util.Set; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.query.aggregator.MeasureAggregator; - -public class DistinctCountBigDecimalAggregatorObjectSet - extends AbstractDistinctCountAggregatorObjectSet { - - private static final long serialVersionUID = 6313463368629960186L; - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - valueSetForObj.add(dataChunk.getMeasureDataHolder().getReadableBigDecimalValueByIndex(index)); - } - } - - private void agg(Set<Object> set2) { - valueSetForObj.addAll(set2); - } - - /** - * merge the valueset so that we get the count of unique values - */ - @Override public void merge(MeasureAggregator aggregator) { - DistinctCountBigDecimalAggregatorObjectSet distinctCountBigDecimalAggregatorObjectSet = - (DistinctCountBigDecimalAggregatorObjectSet) aggregator; - agg(distinctCountBigDecimalAggregatorObjectSet.valueSetForObj); - } - - @Override public MeasureAggregator getCopy() { - DistinctCountBigDecimalAggregatorObjectSet aggregator = - new DistinctCountBigDecimalAggregatorObjectSet(); - aggregator.valueSetForObj = new HashSet<Object>(valueSetForObj); - return aggregator; - } - - @Override public int compareTo(MeasureAggregator measureAggr) { - BigDecimal valueSetForObjSize = getBigDecimalValue(); - BigDecimal otherVal = measureAggr.getBigDecimalValue(); - return valueSetForObjSize.compareTo(otherVal); - } - - @Override public boolean equals(Object obj) { - if (!(obj instanceof DistinctCountBigDecimalAggregatorObjectSet)) { - return false; - } - DistinctCountBigDecimalAggregatorObjectSet o = (DistinctCountBigDecimalAggregatorObjectSet) obj; - return getBigDecimalValue().equals(o.getBigDecimalValue()); - } - - @Override public int hashCode() { - return getBigDecimalValue().hashCode(); - } - - @Override public MeasureAggregator get() { - return this; - } - - @Override public MeasureAggregator getNew() { - return new DistinctCountBigDecimalAggregatorObjectSet(); - } - -}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/DistinctCountLongAggregatorObjectSet.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/DistinctCountLongAggregatorObjectSet.java b/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/DistinctCountLongAggregatorObjectSet.java deleted file mode 100644 index c4f7216..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/DistinctCountLongAggregatorObjectSet.java +++ /dev/null @@ -1,89 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.distinct; - -import java.util.HashSet; -import java.util.Set; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.query.aggregator.MeasureAggregator; - -public class DistinctCountLongAggregatorObjectSet extends AbstractDistinctCountAggregatorObjectSet { - - private static final long serialVersionUID = 6313463368629960186L; - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - valueSetForObj.add(dataChunk.getMeasureDataHolder().getReadableLongValueByIndex(index)); - } - } - - private void agg(Set<Object> set2) { - valueSetForObj.addAll(set2); - } - - /** - * merge the valueset so that we get the count of unique values - */ - @Override public void merge(MeasureAggregator aggregator) { - DistinctCountLongAggregatorObjectSet distinctCountAggregator = - (DistinctCountLongAggregatorObjectSet) aggregator; - agg(distinctCountAggregator.valueSetForObj); - } - - @Override public MeasureAggregator getCopy() { - DistinctCountLongAggregatorObjectSet aggregator = new DistinctCountLongAggregatorObjectSet(); - aggregator.valueSetForObj = new HashSet<Object>(valueSetForObj); - return aggregator; - } - - @Override public int compareTo(MeasureAggregator measureAggr) { - long valueSetForObjSize = getLongValue(); - long otherVal = measureAggr.getLongValue(); - if (valueSetForObjSize > otherVal) { - return 1; - } - if (valueSetForObjSize < otherVal) { - return -1; - } - return 0; - } - - @Override public boolean equals(Object obj) { - if (!(obj instanceof DistinctCountLongAggregatorObjectSet)) { - return false; - } - DistinctCountLongAggregatorObjectSet o = (DistinctCountLongAggregatorObjectSet) obj; - return getLongValue().equals(o.getLongValue()); - } - - @Override public int hashCode() { - return getLongValue().hashCode(); - } - - @Override public MeasureAggregator get() { - return this; - } - - @Override public MeasureAggregator getNew() { - return new DistinctCountLongAggregatorObjectSet(); - } - -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/DistinctStringCountAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/DistinctStringCountAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/DistinctStringCountAggregator.java deleted file mode 100644 index e3d4623..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/DistinctStringCountAggregator.java +++ /dev/null @@ -1,165 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.distinct; - -import java.io.DataInput; -import java.io.DataOutput; -import java.io.IOException; -import java.math.BigDecimal; -import java.nio.ByteBuffer; -import java.nio.charset.Charset; -import java.util.HashSet; -import java.util.Set; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.core.constants.CarbonCommonConstants; -import org.carbondata.query.aggregator.MeasureAggregator; - -public class DistinctStringCountAggregator implements MeasureAggregator { - private static final long serialVersionUID = 6313463368629960186L; - - private Set<String> valueSetForStr; - - public DistinctStringCountAggregator() { - this.valueSetForStr = new HashSet<String>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); - } - - public void agg(double newVal) { - } - - public void agg(String newVal) { - this.valueSetForStr.add(newVal); - } - - private void agg(Set<String> set2) { - this.valueSetForStr.addAll(set2); - } - - public void merge(MeasureAggregator aggregator) { - DistinctStringCountAggregator distinctCountAggregator = - (DistinctStringCountAggregator) aggregator; - agg(distinctCountAggregator.valueSetForStr); - } - - public Double getDoubleValue() { - return (double) this.valueSetForStr.size(); - } - - public Long getLongValue() { - return (long) this.valueSetForStr.size(); - } - - public BigDecimal getBigDecimalValue() { - return new BigDecimal(this.valueSetForStr.size()); - } - - public Object getValueObject() { - return Integer.valueOf(this.valueSetForStr.size()); - } - - public void setNewValue(Object newValue) { - } - - public boolean isFirstTime() { - return false; - } - - public void writeData(DataOutput output) throws IOException { - int length = this.valueSetForStr.size() * 8; - ByteBuffer byteBuffer = ByteBuffer.allocate(length + 4); - byteBuffer.putInt(length); - for (String val : this.valueSetForStr) { - byte[] b = val.getBytes(Charset.defaultCharset()); - byteBuffer.putInt(b.length); - byteBuffer.put(b); - } - byteBuffer.flip(); - output.write(byteBuffer.array()); - } - - public void readData(DataInput inPut) throws IOException { - int length = inPut.readInt(); - length /= 8; - this.valueSetForStr = new HashSet<String>(length + 1, 1.0F); - for (int i = 0; i < length; i++) { - byte[] b = new byte[inPut.readInt()]; - inPut.readFully(b); - this.valueSetForStr.add(new String(b, Charset.defaultCharset())); - } - } - - public MeasureAggregator getCopy() { - DistinctStringCountAggregator aggregator = new DistinctStringCountAggregator(); - aggregator.valueSetForStr = new HashSet<String>(this.valueSetForStr); - return aggregator; - } - - public int compareTo(MeasureAggregator o) { - double val = getDoubleValue(); - double otherVal = o.getDoubleValue(); - if (val > otherVal) { - return 1; - } - if (val < otherVal) { - return -1; - } - return 0; - } - - @Override public boolean equals(Object obj) { - if(!(obj instanceof DistinctStringCountAggregator)) { - return false; - } - DistinctStringCountAggregator o = (DistinctStringCountAggregator) obj; - return getDoubleValue().equals(o.getDoubleValue()); - } - - @Override public int hashCode() { - return getDoubleValue().hashCode(); - } - - @Override public void agg(Object newVal) { - this.valueSetForStr.add((String) newVal); - } - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - } - - @Override public byte[] getByteArray() { - return null; - } - - @Override public MeasureAggregator get() { - return this; - } - - public String toString() { - return valueSetForStr.size() + ""; - } - - @Override public void merge(byte[] value) { - } - - @Override public MeasureAggregator getNew() { - // TODO Auto-generated method stub - return new DistinctStringCountAggregator(); - } - -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/SumDistinctBigDecimalAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/SumDistinctBigDecimalAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/SumDistinctBigDecimalAggregator.java deleted file mode 100644 index 6a59ec9..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/SumDistinctBigDecimalAggregator.java +++ /dev/null @@ -1,233 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.distinct; - -import java.io.DataInput; -import java.io.DataOutput; -import java.io.IOException; -import java.math.BigDecimal; -import java.nio.ByteBuffer; -import java.nio.charset.Charset; -import java.util.HashSet; -import java.util.Iterator; -import java.util.Set; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.core.constants.CarbonCommonConstants; -import org.carbondata.core.util.DataTypeUtil; -import org.carbondata.query.aggregator.MeasureAggregator; -import org.carbondata.query.aggregator.impl.AbstractMeasureAggregatorBasic; - -/** - * The sum distinct aggregator - * Ex: - * ID NAME Sales - * 1 a 200 - * 2 a 100 - * 3 a 200 - * select sum(distinct sales) # would result 300 - */ -public class SumDistinctBigDecimalAggregator extends AbstractMeasureAggregatorBasic { - - /** - * - */ - private static final long serialVersionUID = 6313463368629960155L; - - /** - * For Spark CARBON to avoid heavy object transfer it better to flatten the - * Aggregators. There is no aggregation expected after setting this value. - */ - private BigDecimal computedFixedValue; - - /** - * - */ - private Set<BigDecimal> valueSet; - - public SumDistinctBigDecimalAggregator() { - valueSet = new HashSet<BigDecimal>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); - } - - /** - * Distinct Aggregate function which update the Distinct set - * - * @param newVal new value - */ - @Override public void agg(Object newVal) { - valueSet.add( - newVal instanceof BigDecimal ? (BigDecimal) newVal : new BigDecimal(newVal.toString())); - firstTime = false; - } - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - valueSet.add(dataChunk.getMeasureDataHolder().getReadableBigDecimalValueByIndex(index)); - firstTime = false; - } - } - - /** - * Below method will be used to get the value byte array - */ - @Override public byte[] getByteArray() { - Iterator<BigDecimal> iterator = valueSet.iterator(); - ByteBuffer buffer = - ByteBuffer.allocate(valueSet.size() * CarbonCommonConstants.DOUBLE_SIZE_IN_BYTE); - while (iterator.hasNext()) { - byte[] bytes = DataTypeUtil.bigDecimalToByte(iterator.next()); - buffer.putInt(bytes.length); - buffer.put(bytes); - } - buffer.rewind(); - return buffer.array(); - } - - private void agg(Set<BigDecimal> set2) { - valueSet.addAll(set2); - } - - /** - * merge the valueset so that we get the count of unique values - */ - @Override public void merge(MeasureAggregator aggregator) { - SumDistinctBigDecimalAggregator distinctAggregator = - (SumDistinctBigDecimalAggregator) aggregator; - if (!aggregator.isFirstTime()) { - agg(distinctAggregator.valueSet); - firstTime = false; - } - } - - @Override public BigDecimal getBigDecimalValue() { - if (computedFixedValue == null) { - BigDecimal result = new BigDecimal(0); - for (BigDecimal aValue : valueSet) { - result = result.add(aValue); - } - return result; - } - return computedFixedValue; - } - - @Override public Object getValueObject() { - return getBigDecimalValue(); - } - - @Override public void setNewValue(Object newValue) { - computedFixedValue = (BigDecimal) newValue; - valueSet = null; - } - - @Override public boolean isFirstTime() { - return firstTime; - } - - @Override public void writeData(DataOutput dataOutput) throws IOException { - if (computedFixedValue != null) { - byte[] bytes = DataTypeUtil.bigDecimalToByte(computedFixedValue); - ByteBuffer byteBuffer = ByteBuffer.allocate(4 + 4 + bytes.length); - byteBuffer.putInt(-1); - byteBuffer.putInt(bytes.length); - byteBuffer.put(bytes); - byteBuffer.flip(); - dataOutput.write(byteBuffer.array()); - } else { - int length = valueSet.size() * 8 + valueSet.size() * 4; - ByteBuffer byteBuffer = ByteBuffer.allocate(length + 4 + 1); - byteBuffer.putInt(length); - for (BigDecimal val : valueSet) { - byte[] bytes = - val.toString().getBytes(Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)); - byteBuffer.putInt(-1); - byteBuffer.putInt(bytes.length); - byteBuffer.put(bytes); - } - byteBuffer.flip(); - dataOutput.write(byteBuffer.array()); - } - } - - @Override public void readData(DataInput inPut) throws IOException { - int length = inPut.readInt(); - - if (length == -1) { - computedFixedValue = new BigDecimal(inPut.readUTF()); - valueSet = null; - } else { - length = length / 8; - valueSet = new HashSet<BigDecimal>(length + 1, 1.0f); - for (int i = 0; i < length; i++) { - valueSet.add(new BigDecimal(inPut.readUTF())); - } - } - } - - @Override public void merge(byte[] value) { - if (0 == value.length) { - return; - } - ByteBuffer buffer = ByteBuffer.wrap(value); - buffer.rewind(); - while (buffer.hasRemaining()) { - byte[] valueByte = new byte[buffer.getInt()]; - buffer.get(valueByte); - BigDecimal valueBigDecimal = DataTypeUtil.byteToBigDecimal(valueByte); - agg(valueBigDecimal); - } - } - - public String toString() { - if (computedFixedValue == null) { - return valueSet.size() + ""; - } - return computedFixedValue + ""; - } - - @Override public MeasureAggregator getCopy() { - SumDistinctBigDecimalAggregator aggregator = new SumDistinctBigDecimalAggregator(); - aggregator.valueSet = new HashSet<BigDecimal>(valueSet); - return aggregator; - } - - @Override public int compareTo(MeasureAggregator msr) { - BigDecimal msrValObj = getBigDecimalValue(); - BigDecimal otherVal = msr.getBigDecimalValue(); - - return msrValObj.compareTo(otherVal); - } - - @Override public boolean equals(Object obj) { - if (!(obj instanceof SumDistinctBigDecimalAggregator)) { - return false; - } - SumDistinctBigDecimalAggregator o = (SumDistinctBigDecimalAggregator) obj; - return getBigDecimalValue().equals(o.getBigDecimalValue()); - } - - @Override public int hashCode() { - return getBigDecimalValue().hashCode(); - } - - @Override public MeasureAggregator getNew() { - // TODO Auto-generated method stub - return new SumDistinctBigDecimalAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/SumDistinctDoubleAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/SumDistinctDoubleAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/SumDistinctDoubleAggregator.java deleted file mode 100644 index 0229f24..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/SumDistinctDoubleAggregator.java +++ /dev/null @@ -1,231 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.distinct; - -import java.io.DataInput; -import java.io.DataOutput; -import java.io.IOException; -import java.nio.ByteBuffer; -import java.util.HashSet; -import java.util.Iterator; -import java.util.Set; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.core.constants.CarbonCommonConstants; -import org.carbondata.query.aggregator.MeasureAggregator; -import org.carbondata.query.aggregator.impl.AbstractMeasureAggregatorBasic; - -/** - * The sum distinct aggregator - * Ex: - * ID NAME Sales - * 1 a 200 - * 2 a 100 - * 3 a 200 - * select sum(distinct sales) # would result 300 - */ - -public class SumDistinctDoubleAggregator extends AbstractMeasureAggregatorBasic { - - /** - * - */ - private static final long serialVersionUID = 6313463368629960155L; - - /** - * For Spark CARBON to avoid heavy object transfer it better to flatten the - * Aggregators. There is no aggregation expected after setting this value. - */ - private Double computedFixedValue; - - /** - * - */ - private Set<Double> valueSet; - - public SumDistinctDoubleAggregator() { - valueSet = new HashSet<Double>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); - } - - /** - * just need to add the unique values to agg set - */ - @Override public void agg(double newVal) { - valueSet.add(newVal); - firstTime = false; - } - - /** - * Distinct Aggregate function which update the Distinct set - * - * @param newVal new value - */ - @Override public void agg(Object newVal) { - valueSet.add(newVal instanceof Double ? (Double) newVal : new Double(newVal.toString())); - firstTime = false; - } - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - valueSet.add(dataChunk.getMeasureDataHolder().getReadableDoubleValueByIndex(index)); - firstTime = false; - } - } - - /** - * Below method will be used to get the value byte array - */ - @Override public byte[] getByteArray() { - Iterator<Double> iterator = valueSet.iterator(); - ByteBuffer buffer = - ByteBuffer.allocate(valueSet.size() * CarbonCommonConstants.DOUBLE_SIZE_IN_BYTE); - while (iterator.hasNext()) { - buffer.putDouble(iterator.next()); - } - buffer.rewind(); - return buffer.array(); - } - - private void agg(Set<Double> set2) { - valueSet.addAll(set2); - } - - /** - * merge the valueset so that we get the count of unique values - */ - @Override public void merge(MeasureAggregator aggregator) { - SumDistinctDoubleAggregator distinctAggregator = (SumDistinctDoubleAggregator) aggregator; - if (!aggregator.isFirstTime()) { - agg(distinctAggregator.valueSet); - firstTime = false; - } - } - - @Override public Double getDoubleValue() { - if (computedFixedValue == null) { - double result = 0; - for (Double aValue : valueSet) { - result += aValue; - } - return result; - } - return computedFixedValue; - } - - @Override public Object getValueObject() { - return getDoubleValue(); - } - - @Override public void setNewValue(Object newValue) { - computedFixedValue = (Double) newValue; - valueSet = null; - } - - @Override public boolean isFirstTime() { - return firstTime; - } - - @Override public void writeData(DataOutput dataOutput) throws IOException { - if (computedFixedValue != null) { - ByteBuffer byteBuffer = ByteBuffer.allocate(4 + 8); - byteBuffer.putInt(-1); - byteBuffer.putDouble(computedFixedValue); - byteBuffer.flip(); - dataOutput.write(byteBuffer.array()); - } else { - int length = valueSet.size() * 8; - ByteBuffer byteBuffer = ByteBuffer.allocate(length + 4 + 1); - byteBuffer.putInt(length); - for (double val : valueSet) { - byteBuffer.putDouble(val); - } - byteBuffer.flip(); - dataOutput.write(byteBuffer.array()); - } - } - - @Override public void readData(DataInput inPut) throws IOException { - int length = inPut.readInt(); - - if (length == -1) { - computedFixedValue = inPut.readDouble(); - valueSet = null; - } else { - length = length / 8; - valueSet = new HashSet<Double>(length + 1, 1.0f); - for (int i = 0; i < length; i++) { - valueSet.add(inPut.readDouble()); - } - } - } - - @Override public void merge(byte[] value) { - if (0 == value.length) { - return; - } - ByteBuffer buffer = ByteBuffer.wrap(value); - buffer.rewind(); - while (buffer.hasRemaining()) { - agg(buffer.getDouble()); - } - } - - public String toString() { - if (computedFixedValue == null) { - return valueSet.size() + ""; - } - return computedFixedValue + ""; - } - - @Override public MeasureAggregator getCopy() { - SumDistinctDoubleAggregator aggregator = new SumDistinctDoubleAggregator(); - aggregator.valueSet = new HashSet<Double>(valueSet); - return aggregator; - } - - @Override public int compareTo(MeasureAggregator msr) { - double msrValObj = getDoubleValue(); - double otherVal = msr.getDoubleValue(); - if (msrValObj > otherVal) { - return 1; - } - if (msrValObj < otherVal) { - return -1; - } - return 0; - } - - @Override public boolean equals(Object obj) { - if (!(obj instanceof SumDistinctDoubleAggregator)) { - return false; - } - SumDistinctDoubleAggregator o = (SumDistinctDoubleAggregator) obj; - return getDoubleValue().equals(o.getDoubleValue()); - } - - @Override public int hashCode() { - return getDoubleValue().hashCode(); - } - - @Override public MeasureAggregator getNew() { - // TODO Auto-generated method stub - return new SumDistinctDoubleAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/SumDistinctLongAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/SumDistinctLongAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/SumDistinctLongAggregator.java deleted file mode 100644 index d57c34b..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/distinct/SumDistinctLongAggregator.java +++ /dev/null @@ -1,221 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.distinct; - -import java.io.DataInput; -import java.io.DataOutput; -import java.io.IOException; -import java.nio.ByteBuffer; -import java.util.HashSet; -import java.util.Iterator; -import java.util.Set; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.core.constants.CarbonCommonConstants; -import org.carbondata.query.aggregator.MeasureAggregator; -import org.carbondata.query.aggregator.impl.AbstractMeasureAggregatorBasic; - -/** - * The sum distinct aggregator - * Ex: - * ID NAME Sales - * 1 a 200 - * 2 a 100 - * 3 a 200 - * select sum(distinct sales) # would result 300 - */ - -public class SumDistinctLongAggregator extends AbstractMeasureAggregatorBasic { - - private static final long serialVersionUID = 6313463368629960155L; - - /** - * For Spark CARBON to avoid heavy object transfer it better to flatten the - * Aggregators. There is no aggregation expected after setting this value. - */ - private Long computedFixedValue; - - /** - * - */ - private Set<Long> valueSet; - - public SumDistinctLongAggregator() { - valueSet = new HashSet<Long>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); - } - - /** - * Distinct Aggregate function which update the Distinct set - * - * @param newVal new value - */ - @Override public void agg(Object newVal) { - valueSet.add(newVal instanceof Long ? (Long) newVal : Long.valueOf(newVal.toString())); - firstTime = false; - } - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - valueSet.add(dataChunk.getMeasureDataHolder().getReadableLongValueByIndex(index)); - firstTime = false; - } - } - - /** - * Below method will be used to get the value byte array - */ - @Override public byte[] getByteArray() { - Iterator<Long> iterator = valueSet.iterator(); - ByteBuffer buffer = - ByteBuffer.allocate(valueSet.size() * CarbonCommonConstants.DOUBLE_SIZE_IN_BYTE); - while (iterator.hasNext()) { - buffer.putLong(iterator.next()); - } - buffer.rewind(); - return buffer.array(); - } - - private void agg(Set<Long> set2) { - valueSet.addAll(set2); - } - - /** - * merge the valueset so that we get the count of unique values - */ - @Override public void merge(MeasureAggregator aggregator) { - SumDistinctLongAggregator distinctAggregator = (SumDistinctLongAggregator) aggregator; - if (!aggregator.isFirstTime()) { - agg(distinctAggregator.valueSet); - firstTime = false; - } - } - - @Override public Long getLongValue() { - if (computedFixedValue == null) { - long result = 0; - for (Long aValue : valueSet) { - result += aValue; - } - return result; - } - return computedFixedValue; - } - - @Override public Object getValueObject() { - return getLongValue(); - } - - @Override public void setNewValue(Object newValue) { - computedFixedValue = (Long) newValue; - valueSet = null; - } - - @Override public boolean isFirstTime() { - return firstTime; - } - - @Override public void writeData(DataOutput dataOutput) throws IOException { - if (computedFixedValue != null) { - ByteBuffer byteBuffer = ByteBuffer.allocate(4 + 8); - byteBuffer.putInt(-1); - byteBuffer.putLong(computedFixedValue); - byteBuffer.flip(); - dataOutput.write(byteBuffer.array()); - } else { - int length = valueSet.size() * 8; - ByteBuffer byteBuffer = ByteBuffer.allocate(length + 4 + 1); - byteBuffer.putInt(length); - for (long val : valueSet) { - byteBuffer.putLong(val); - } - byteBuffer.flip(); - dataOutput.write(byteBuffer.array()); - } - } - - @Override public void readData(DataInput inPut) throws IOException { - int length = inPut.readInt(); - - if (length == -1) { - computedFixedValue = inPut.readLong(); - valueSet = null; - } else { - length = length / 8; - valueSet = new HashSet<Long>(length + 1, 1.0f); - for (int i = 0; i < length; i++) { - valueSet.add(inPut.readLong()); - } - } - - } - - @Override public void merge(byte[] value) { - if (0 == value.length) { - return; - } - ByteBuffer buffer = ByteBuffer.wrap(value); - buffer.rewind(); - while (buffer.hasRemaining()) { - agg(buffer.getLong()); - } - } - - public String toString() { - if (computedFixedValue == null) { - return valueSet.size() + ""; - } - return computedFixedValue + ""; - } - - @Override public MeasureAggregator getCopy() { - SumDistinctLongAggregator aggregator = new SumDistinctLongAggregator(); - aggregator.valueSet = new HashSet<Long>(valueSet); - return aggregator; - } - - @Override public int compareTo(MeasureAggregator msr) { - long msrValObj = getLongValue(); - long otherVal = msr.getLongValue(); - if (msrValObj > otherVal) { - return 1; - } - if (msrValObj < otherVal) { - return -1; - } - return 0; - } - - @Override public boolean equals(Object obj) { - if (!(obj instanceof SumDistinctLongAggregator)) { - return false; - } - SumDistinctLongAggregator o = (SumDistinctLongAggregator) obj; - return getLongValue().equals(o.getLongValue()); - } - - @Override public int hashCode() { - return getLongValue().hashCode(); - } - - @Override public MeasureAggregator getNew() { - // TODO Auto-generated method stub - return new SumDistinctLongAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/AbstractMeasureAggregatorDummy.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/AbstractMeasureAggregatorDummy.java b/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/AbstractMeasureAggregatorDummy.java deleted file mode 100644 index f423085..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/AbstractMeasureAggregatorDummy.java +++ /dev/null @@ -1,70 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.dummy; - -import java.io.DataInput; -import java.io.DataOutput; -import java.io.IOException; - -import org.carbondata.query.aggregator.MeasureAggregator; -import org.carbondata.query.aggregator.impl.AbstractMeasureAggregatorBasic; - -/** - * AbstractMeasureAggregatorDummy - * Used for custom Carbon Aggregator dummy - */ -public abstract class AbstractMeasureAggregatorDummy extends AbstractMeasureAggregatorBasic { - private static final long serialVersionUID = 1L; - - @Override public int compareTo(MeasureAggregator o) { - if (equals(o)) { - return 0; - } - return -1; - } - - @Override public boolean equals(Object arg0) { - return super.equals(arg0); - } - - @Override public int hashCode() { - return super.hashCode(); - } - - @Override public byte[] getByteArray() { - return null; - } - - @Override public void merge(MeasureAggregator aggregator) { - } - - @Override public MeasureAggregator getCopy() { - return null; - } - - @Override public void writeData(DataOutput output) throws IOException { - } - - @Override public void readData(DataInput inPut) throws IOException { - } - - @Override public void merge(byte[] value) { - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/DummyBigDecimalAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/DummyBigDecimalAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/DummyBigDecimalAggregator.java deleted file mode 100644 index 8a33fe1..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/DummyBigDecimalAggregator.java +++ /dev/null @@ -1,63 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.dummy; - -import java.math.BigDecimal; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.query.aggregator.MeasureAggregator; - -public class DummyBigDecimalAggregator extends AbstractMeasureAggregatorDummy { - private static final long serialVersionUID = 1L; - - /** - * aggregate value - */ - private BigDecimal aggVal; - - @Override public void agg(Object newVal) { - aggVal = (BigDecimal) newVal; - firstTime = false; - } - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - aggVal = dataChunk.getMeasureDataHolder().getReadableBigDecimalValueByIndex(index); - firstTime = false; - } - } - - @Override public BigDecimal getBigDecimalValue() { - return aggVal; - } - - @Override public Object getValueObject() { - return aggVal; - } - - @Override public void setNewValue(Object newValue) { - aggVal = (BigDecimal) newValue; - } - - @Override public MeasureAggregator getNew() { - // TODO Auto-generated method stub - return new DummyBigDecimalAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/DummyDoubleAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/DummyDoubleAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/DummyDoubleAggregator.java deleted file mode 100644 index cd4fe56..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/DummyDoubleAggregator.java +++ /dev/null @@ -1,64 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.dummy; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.query.aggregator.MeasureAggregator; - -public class DummyDoubleAggregator extends AbstractMeasureAggregatorDummy { - private static final long serialVersionUID = 1L; - - /** - * aggregate value - */ - private double aggVal; - - @Override public void agg(double newVal) { - aggVal = newVal; - firstTime = false; - } - - @Override public void agg(Object newVal) { - aggVal = (Double) newVal; - firstTime = false; - } - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - aggVal = dataChunk.getMeasureDataHolder().getReadableDoubleValueByIndex(index); - firstTime = false; - } - } - - @Override public Double getDoubleValue() { - return aggVal; - } - - @Override public Object getValueObject() { - return aggVal; - } - - @Override public void setNewValue(Object newValue) { - aggVal = (Double) newValue; - } - - @Override public MeasureAggregator getNew() { - return new DummyDoubleAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/DummyLongAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/DummyLongAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/DummyLongAggregator.java deleted file mode 100644 index 4131895..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/dummy/DummyLongAggregator.java +++ /dev/null @@ -1,59 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.dummy; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.query.aggregator.MeasureAggregator; - -public class DummyLongAggregator extends AbstractMeasureAggregatorDummy { - private static final long serialVersionUID = 1L; - - /** - * aggregate value - */ - private long aggVal; - - @Override public void agg(Object newVal) { - aggVal = (Long) newVal; - firstTime = false; - } - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - aggVal = dataChunk.getMeasureDataHolder().getReadableLongValueByIndex(index); - firstTime = false; - } - } - - @Override public Long getLongValue() { - return aggVal; - } - - @Override public Object getValueObject() { - return aggVal; - } - - @Override public void setNewValue(Object newValue) { - aggVal = (Long) newValue; - } - - @Override public MeasureAggregator getNew() { - return new DummyLongAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/max/AbstractMaxAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/max/AbstractMaxAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/max/AbstractMaxAggregator.java deleted file mode 100644 index 27b1876..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/max/AbstractMaxAggregator.java +++ /dev/null @@ -1,61 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.max; - -import java.io.ByteArrayInputStream; -import java.io.ObjectInput; -import java.io.ObjectInputStream; - -import org.carbondata.common.logging.LogService; -import org.carbondata.common.logging.LogServiceFactory; -import org.carbondata.core.util.CarbonUtil; -import org.carbondata.query.aggregator.impl.AbstractMeasureAggregatorMaxMin; - -public abstract class AbstractMaxAggregator extends AbstractMeasureAggregatorMaxMin { - - private static final LogService LOGGER = - LogServiceFactory.getLogService(AbstractMaxAggregator.class.getName()); - - protected void internalAgg(Object value) { - if (value instanceof Comparable) { - @SuppressWarnings("unchecked") Comparable<Object> newValue = ((Comparable<Object>) value); - aggVal = (aggVal == null || aggVal.compareTo(newValue) < 0) ? newValue : aggVal; - } - } - - @Override public void merge(byte[] value) { - if (0 == value.length) { - return; - } - ByteArrayInputStream bytesInputStream = null; - ObjectInput in = null; - try { - bytesInputStream = new ByteArrayInputStream(value); - in = new ObjectInputStream(bytesInputStream); - Object newVal = (Comparable<Object>) in.readObject(); - internalAgg(newVal); - firstTime = false; - } catch (Exception e) { - LOGGER.error(e, "Problem while merging byte array in maxAggregator: " + e.getMessage()); - } finally { - CarbonUtil.closeStreams(bytesInputStream); - } - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/max/MaxAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/max/MaxAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/max/MaxAggregator.java deleted file mode 100644 index d0d8af0..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/max/MaxAggregator.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.max; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.query.aggregator.MeasureAggregator; - -/** - * Class Description : - * It will return max of values - */ -public class MaxAggregator extends AbstractMaxAggregator { - - /** - * serialVersionUID - */ - private static final long serialVersionUID = -5850218739083899419L; - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - internalAgg(dataChunk.getMeasureDataHolder().getReadableDoubleValueByIndex(index)); - firstTime = false; - } - } - - /** - * Merge the value, it will update the max aggregate value if aggregator - * passed as an argument will have value greater than aggVal - * - * @param aggregator MaxAggregator - */ - @Override public void merge(MeasureAggregator aggregator) { - MaxAggregator maxAggregator = (MaxAggregator) aggregator; - if (!aggregator.isFirstTime()) { - agg(maxAggregator.aggVal); - firstTime = false; - } - - } - - @Override public MeasureAggregator getCopy() { - MaxAggregator aggregator = new MaxAggregator(); - aggregator.aggVal = aggVal; - aggregator.firstTime = firstTime; - return aggregator; - } - - @Override public MeasureAggregator getNew() { - return new MaxAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/max/MaxBigDecimalAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/max/MaxBigDecimalAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/max/MaxBigDecimalAggregator.java deleted file mode 100644 index c4149c6..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/max/MaxBigDecimalAggregator.java +++ /dev/null @@ -1,67 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.max; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.query.aggregator.MeasureAggregator; - -/** - * Class Description : - * It will return max of values - */ -public class MaxBigDecimalAggregator extends AbstractMaxAggregator { - - /** - * serialVersionUID - */ - private static final long serialVersionUID = -5850218739083899419L; - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - internalAgg(dataChunk.getMeasureDataHolder().getReadableBigDecimalValueByIndex(index)); - firstTime = false; - } - } - - /** - * Merge the value, it will update the max aggregate value if aggregator - * passed as an argument will have value greater than aggVal - * - * @param aggregator MaxAggregator - */ - @Override public void merge(MeasureAggregator aggregator) { - MaxAggregator maxAggregator = (MaxAggregator) aggregator; - if (!aggregator.isFirstTime()) { - agg(maxAggregator.aggVal); - firstTime = false; - } - } - - @Override public MeasureAggregator getCopy() { - MaxAggregator aggregator = new MaxAggregator(); - aggregator.aggVal = aggVal; - aggregator.firstTime = firstTime; - return aggregator; - } - - @Override public MeasureAggregator getNew() { - return new MaxBigDecimalAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/max/MaxLongAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/max/MaxLongAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/max/MaxLongAggregator.java deleted file mode 100644 index 55b3be5..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/max/MaxLongAggregator.java +++ /dev/null @@ -1,67 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.max; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.query.aggregator.MeasureAggregator; - -/** - * Class Description : - * It will return max of values - */ -public class MaxLongAggregator extends AbstractMaxAggregator { - - /** - * serialVersionUID - */ - private static final long serialVersionUID = -5850218739083899419L; - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - internalAgg(dataChunk.getMeasureDataHolder().getReadableLongValueByIndex(index)); - firstTime = false; - } - } - - /** - * Merge the value, it will update the max aggregate value if aggregator - * passed as an argument will have value greater than aggVal - * - * @param aggregator MaxAggregator - */ - @Override public void merge(MeasureAggregator aggregator) { - MaxAggregator maxAggregator = (MaxAggregator) aggregator; - if (!aggregator.isFirstTime()) { - agg(maxAggregator.aggVal); - firstTime = false; - } - } - - @Override public MeasureAggregator getCopy() { - MaxAggregator aggregator = new MaxAggregator(); - aggregator.aggVal = aggVal; - aggregator.firstTime = firstTime; - return aggregator; - } - - @Override public MeasureAggregator getNew() { - return new MaxLongAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/min/AbstractMinAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/min/AbstractMinAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/min/AbstractMinAggregator.java deleted file mode 100644 index 77aa163..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/min/AbstractMinAggregator.java +++ /dev/null @@ -1,61 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.min; - -import java.io.ByteArrayInputStream; -import java.io.ObjectInput; -import java.io.ObjectInputStream; - -import org.carbondata.common.logging.LogService; -import org.carbondata.common.logging.LogServiceFactory; -import org.carbondata.core.util.CarbonUtil; -import org.carbondata.query.aggregator.impl.AbstractMeasureAggregatorMaxMin; - -public abstract class AbstractMinAggregator extends AbstractMeasureAggregatorMaxMin { - - private static final LogService LOGGER = - LogServiceFactory.getLogService(AbstractMinAggregator.class.getName()); - - protected void internalAgg(Object value) { - if (value instanceof Comparable) { - @SuppressWarnings("unchecked") Comparable<Object> newValue = ((Comparable<Object>) value); - aggVal = (aggVal == null || aggVal.compareTo(newValue) > 0) ? newValue : aggVal; - } - } - - @Override public void merge(byte[] value) { - if (0 == value.length) { - return; - } - ByteArrayInputStream bis = null; - ObjectInput objectInput = null; - try { - bis = new ByteArrayInputStream(value); - objectInput = new ObjectInputStream(bis); - Object newVal = (Comparable<Object>) objectInput.readObject(); - internalAgg(newVal); - firstTime = false; - } catch (Exception e) { - LOGGER.error(e, "Problem while merging byte array in minAggregator: " + e.getMessage()); - } finally { - CarbonUtil.closeStreams(bis); - } - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/min/MinAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/min/MinAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/min/MinAggregator.java deleted file mode 100644 index 4bb3d73..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/min/MinAggregator.java +++ /dev/null @@ -1,66 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.min; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.query.aggregator.MeasureAggregator; - -/** - * Class Description : It will return min of values - */ -public class MinAggregator extends AbstractMinAggregator { - - /** - * serialVersionUID - */ - private static final long serialVersionUID = -8077547753784906280L; - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - internalAgg(dataChunk.getMeasureDataHolder().getReadableDoubleValueByIndex(index)); - firstTime = false; - } - } - - /** - * Merge the value, it will update the min aggregate value if aggregator - * passed as an argument will have value less than aggVal - * - * @param aggregator MinAggregator - */ - @Override public void merge(MeasureAggregator aggregator) { - MinAggregator minAggregator = (MinAggregator) aggregator; - if (!aggregator.isFirstTime()) { - agg(minAggregator.aggVal); - firstTime = false; - } - } - - @Override public MeasureAggregator getCopy() { - MinAggregator aggregator = new MinAggregator(); - aggregator.aggVal = aggVal; - aggregator.firstTime = firstTime; - return aggregator; - } - - @Override public MeasureAggregator getNew() { - return new MinAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/min/MinBigDecimalAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/min/MinBigDecimalAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/min/MinBigDecimalAggregator.java deleted file mode 100644 index 7347dc1..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/min/MinBigDecimalAggregator.java +++ /dev/null @@ -1,66 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.min; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.query.aggregator.MeasureAggregator; - -/** - * Class Description : It will return min of values - */ -public class MinBigDecimalAggregator extends AbstractMinAggregator { - - /** - * serialVersionUID - */ - private static final long serialVersionUID = -8077547753784906280L; - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - internalAgg(dataChunk.getMeasureDataHolder().getReadableBigDecimalValueByIndex(index)); - firstTime = false; - } - } - - /** - * Merge the value, it will update the min aggregate value if aggregator - * passed as an argument will have value less than aggVal - * - * @param aggregator MinAggregator - */ - @Override public void merge(MeasureAggregator aggregator) { - MinAggregator minAggregator = (MinAggregator) aggregator; - if (!aggregator.isFirstTime()) { - agg(minAggregator.aggVal); - firstTime = false; - } - } - - @Override public MeasureAggregator getCopy() { - MinAggregator aggregator = new MinAggregator(); - aggregator.aggVal = aggVal; - aggregator.firstTime = firstTime; - return aggregator; - } - - @Override public MeasureAggregator getNew() { - return new MinBigDecimalAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/min/MinLongAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/min/MinLongAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/min/MinLongAggregator.java deleted file mode 100644 index af614c9..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/min/MinLongAggregator.java +++ /dev/null @@ -1,66 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.min; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.query.aggregator.MeasureAggregator; - -/** - * Class Description : It will return min of values - */ -public class MinLongAggregator extends AbstractMinAggregator { - - /** - * serialVersionUID - */ - private static final long serialVersionUID = -8077547753784906280L; - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - internalAgg(dataChunk.getMeasureDataHolder().getReadableLongValueByIndex(index)); - firstTime = false; - } - } - - /** - * Merge the value, it will update the min aggregate value if aggregator - * passed as an argument will have value less than aggVal - * - * @param aggregator MinAggregator - */ - @Override public void merge(MeasureAggregator aggregator) { - MinAggregator minAggregator = (MinAggregator) aggregator; - if (!aggregator.isFirstTime()) { - agg(minAggregator.aggVal); - firstTime = false; - } - } - - @Override public MeasureAggregator getCopy() { - MinAggregator aggregator = new MinAggregator(); - aggregator.aggVal = aggVal; - aggregator.firstTime = firstTime; - return aggregator; - } - - @Override public MeasureAggregator getNew() { - return new MinLongAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/sum/SumBigDecimalAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/sum/SumBigDecimalAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/sum/SumBigDecimalAggregator.java deleted file mode 100644 index b901878..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/sum/SumBigDecimalAggregator.java +++ /dev/null @@ -1,179 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.sum; - -import java.io.DataInput; -import java.io.DataOutput; -import java.io.IOException; -import java.math.BigDecimal; -import java.nio.ByteBuffer; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.core.util.DataTypeUtil; -import org.carbondata.query.aggregator.MeasureAggregator; -import org.carbondata.query.aggregator.impl.AbstractMeasureAggregatorBasic; - -public class SumBigDecimalAggregator extends AbstractMeasureAggregatorBasic { - - /** - * serialVersionUID - */ - private static final long serialVersionUID = 623750056131364540L; - - /** - * aggregate value - */ - private BigDecimal aggVal; - - public SumBigDecimalAggregator() { - aggVal = new BigDecimal(0); - firstTime = false; - } - - /** - * This method will update the aggVal it will add new value to aggVal - * - * @param newVal new value - */ - @Override public void agg(Object newVal) { - if (firstTime) { - aggVal = (BigDecimal) newVal; - firstTime = false; - } else { - aggVal = aggVal.add((BigDecimal) newVal); - } - } - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - BigDecimal value = dataChunk.getMeasureDataHolder().getReadableBigDecimalValueByIndex(index); - aggVal = aggVal.add(value); - firstTime = false; - } - } - - /** - * Below method will be used to get the value byte array - */ - @Override public byte[] getByteArray() { - if (firstTime) { - return new byte[0]; - } - byte[] bytes = DataTypeUtil.bigDecimalToByte(aggVal); - ByteBuffer allocate = ByteBuffer.allocate(4 + bytes.length); - - allocate.putInt(bytes.length); - allocate.put(bytes); - allocate.rewind(); - return allocate.array(); - } - - /** - * This method will return aggVal - * - * @return sum value - */ - @Override public BigDecimal getBigDecimalValue() { - return aggVal; - } - - /* Merge the value, it will update the sum aggregate value it will add new - * value to aggVal - * - * @param aggregator - * SumAggregator - * - */ - @Override public void merge(MeasureAggregator aggregator) { - if (!aggregator.isFirstTime()) { - agg(aggregator.getBigDecimalValue()); - } - } - - /** - * This method return the sum value as an object - * - * @return sum value as an object - */ - @Override public Object getValueObject() { - return aggVal; - } - - @Override public void setNewValue(Object newValue) { - aggVal = (BigDecimal) newValue; - } - - @Override public void readData(DataInput inPut) throws IOException { - firstTime = inPut.readBoolean(); - aggVal = new BigDecimal(inPut.readUTF()); - } - - @Override public void writeData(DataOutput output) throws IOException { - output.writeBoolean(firstTime); - output.writeUTF(aggVal.toString()); - - } - - @Override public MeasureAggregator getCopy() { - SumBigDecimalAggregator aggr = new SumBigDecimalAggregator(); - aggr.aggVal = aggVal; - aggr.firstTime = firstTime; - return aggr; - } - - @Override public void merge(byte[] value) { - if (0 == value.length) { - return; - } - - ByteBuffer buffer = ByteBuffer.wrap(value); - byte[] valueByte = new byte[buffer.getInt()]; - buffer.get(valueByte); - BigDecimal valueBigDecimal = DataTypeUtil.byteToBigDecimal(valueByte); - aggVal = aggVal.add(valueBigDecimal); - firstTime = false; - } - - public String toString() { - return aggVal + ""; - } - - @Override public int compareTo(MeasureAggregator o) { - BigDecimal value = getBigDecimalValue(); - BigDecimal otherVal = o.getBigDecimalValue(); - return value.compareTo(otherVal); - } - - @Override public boolean equals(Object obj) { - if (!(obj instanceof SumBigDecimalAggregator)) { - return false; - } - SumBigDecimalAggregator o = (SumBigDecimalAggregator) obj; - return getBigDecimalValue().equals(o.getBigDecimalValue()); - } - - @Override public int hashCode() { - return getBigDecimalValue().hashCode(); - } - - @Override public MeasureAggregator getNew() { - return new SumBigDecimalAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/sum/SumDoubleAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/sum/SumDoubleAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/sum/SumDoubleAggregator.java deleted file mode 100644 index 777318d..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/sum/SumDoubleAggregator.java +++ /dev/null @@ -1,179 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.sum; - -import java.io.DataInput; -import java.io.DataOutput; -import java.io.IOException; -import java.nio.ByteBuffer; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.core.constants.CarbonCommonConstants; -import org.carbondata.query.aggregator.MeasureAggregator; -import org.carbondata.query.aggregator.impl.AbstractMeasureAggregatorBasic; - -public class SumDoubleAggregator extends AbstractMeasureAggregatorBasic { - - /** - * serialVersionUID - */ - private static final long serialVersionUID = 623750056131364540L; - - /** - * aggregate value - */ - private double aggVal; - - /** - * This method will update the aggVal it will add new value to aggVal - * - * @param newVal new value - */ - @Override public void agg(double newVal) { - aggVal += newVal; - firstTime = false; - } - - /** - * This method will update the aggVal it will add new value to aggVal - * - * @param newVal new value - */ - @Override public void agg(Object newVal) { - aggVal += ((Number) newVal).doubleValue(); - firstTime = false; - } - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - aggVal+= dataChunk.getMeasureDataHolder().getReadableDoubleValueByIndex(index); - firstTime = false; - } - } - - /** - * Below method will be used to get the value byte array - */ - @Override public byte[] getByteArray() { - if (firstTime) { - return new byte[0]; - } - ByteBuffer buffer = ByteBuffer.allocate(CarbonCommonConstants.DOUBLE_SIZE_IN_BYTE); - buffer.putDouble(aggVal); - return buffer.array(); - } - - /** - * This method will return aggVal - * - * @return sum value - */ - - @Override public Double getDoubleValue() { - return aggVal; - } - - /* Merge the value, it will update the sum aggregate value it will add new - * value to aggVal - * - * @param aggregator SumAggregator - * - */ - @Override public void merge(MeasureAggregator aggregator) { - if (!aggregator.isFirstTime()) { - agg(aggregator.getDoubleValue()); - } - } - - /** - * This method return the sum value as an object - * - * @return sum value as an object - */ - @Override public Object getValueObject() { - return aggVal; - } - - @Override public void setNewValue(Object newValue) { - aggVal = (Double) newValue; - } - - @Override public boolean isFirstTime() { - return firstTime; - } - - @Override public void readData(DataInput inPut) throws IOException { - firstTime = inPut.readBoolean(); - aggVal = inPut.readDouble(); - } - - @Override public void writeData(DataOutput output) throws IOException { - output.writeBoolean(firstTime); - output.writeDouble(aggVal); - - } - - @Override public MeasureAggregator getCopy() { - SumDoubleAggregator aggr = new SumDoubleAggregator(); - aggr.aggVal = aggVal; - aggr.firstTime = firstTime; - return aggr; - } - - @Override public void merge(byte[] value) { - if (0 == value.length) { - return; - } - aggVal += ByteBuffer.wrap(value).getDouble(); - firstTime = false; - } - - public String toString() { - return aggVal + ""; - } - - @Override public int compareTo(MeasureAggregator o) { - double value = getDoubleValue(); - double otherVal = o.getDoubleValue(); - if (value > otherVal) { - return 1; - } - if (value < otherVal) { - return -1; - } - return 0; - } - - @Override public boolean equals(Object obj) { - if(!(obj instanceof SumDoubleAggregator)) { - return false; - } - SumDoubleAggregator o = (SumDoubleAggregator) obj; - return getDoubleValue().equals(o.getDoubleValue()); - } - - @Override public int hashCode() { - return getDoubleValue().hashCode(); - } - - @Override public MeasureAggregator getNew() { - return new SumDoubleAggregator(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/core/src/main/java/org/carbondata/query/aggregator/impl/sum/SumLongAggregator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/carbondata/query/aggregator/impl/sum/SumLongAggregator.java b/core/src/main/java/org/carbondata/query/aggregator/impl/sum/SumLongAggregator.java deleted file mode 100644 index 7c245d9..0000000 --- a/core/src/main/java/org/carbondata/query/aggregator/impl/sum/SumLongAggregator.java +++ /dev/null @@ -1,165 +0,0 @@ -/* - * 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.carbondata.query.aggregator.impl.sum; - -import java.io.DataInput; -import java.io.DataOutput; -import java.io.IOException; -import java.nio.ByteBuffer; - -import org.carbondata.core.carbon.datastore.chunk.MeasureColumnDataChunk; -import org.carbondata.core.constants.CarbonCommonConstants; -import org.carbondata.query.aggregator.MeasureAggregator; -import org.carbondata.query.aggregator.impl.AbstractMeasureAggregatorBasic; - -public class SumLongAggregator extends AbstractMeasureAggregatorBasic { - - /** - * serialVersionUID - */ - private static final long serialVersionUID = 623750056131364540L; - - /** - * aggregate value - */ - private long aggVal; - - /** - * This method will update the aggVal it will add new value to aggVal - * - * @param newVal new value - */ - @Override public void agg(Object newVal) { - aggVal += (long) newVal; - firstTime = false; - } - - @Override public void agg(MeasureColumnDataChunk dataChunk, int index) { - if (!dataChunk.getNullValueIndexHolder().getBitSet().get(index)) { - aggVal+= dataChunk.getMeasureDataHolder().getReadableLongValueByIndex(index); - firstTime = false; - } - } - - /** - * Below method will be used to get the value byte array - */ - @Override public byte[] getByteArray() { - if (firstTime) { - return new byte[0]; - } - ByteBuffer buffer = ByteBuffer.allocate(CarbonCommonConstants.LONG_SIZE_IN_BYTE); - buffer.putLong(aggVal); - return buffer.array(); - } - - /** - * This method will return aggVal - * - * @return sum value - */ - @Override public Long getLongValue() { - return aggVal; - } - - /* Merge the value, it will update the sum aggregate value it will add new - * value to aggVal - * - * @param aggregator SumAggregator - * - */ - @Override public void merge(MeasureAggregator aggregator) { - if (!aggregator.isFirstTime()) { - agg(aggregator.getLongValue()); - } - } - - /** - * This method return the sum value as an object - * - * @return sum long value as an object - */ - @Override public Object getValueObject() { - return aggVal; - } - - @Override public void setNewValue(Object newValue) { - aggVal = (long) newValue; - } - - @Override public void readData(DataInput inPut) throws IOException { - firstTime = inPut.readBoolean(); - aggVal = inPut.readLong(); - } - - @Override public void writeData(DataOutput output) throws IOException { - output.writeBoolean(firstTime); - output.writeLong(aggVal); - - } - - @Override public MeasureAggregator getCopy() { - SumLongAggregator aggr = new SumLongAggregator(); - aggr.aggVal = aggVal; - aggr.firstTime = firstTime; - return aggr; - } - - @Override public void merge(byte[] value) { - if (0 == value.length) { - return; - } - aggVal += ByteBuffer.wrap(value).getLong(); - firstTime = false; - } - - public String toString() { - return aggVal + ""; - } - - @Override public int compareTo(MeasureAggregator o) { - Long value = getLongValue(); - Long otherVal = o.getLongValue(); - if (value > otherVal) { - return 1; - } - if (value < otherVal) { - return -1; - } - return 0; - } - - @Override public boolean equals(Object obj) { - if(!(obj instanceof SumLongAggregator)) { - return false; - } - SumLongAggregator o = (SumLongAggregator) obj; - return getLongValue().equals(o.getLongValue()); - } - - @Override public int hashCode() { - return getLongValue().hashCode(); - } - - @Override public MeasureAggregator getNew() { - // TODO Auto-generated method stub - return new SumLongAggregator(); - } -}
