You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ja...@apache.org on 2014/01/27 23:15:23 UTC
[07/51] [partial] Initial commit of master branch from github
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/schema/ValueBitSet.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/ValueBitSet.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/ValueBitSet.java
new file mode 100644
index 0000000..2b1937b
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/ValueBitSet.java
@@ -0,0 +1,200 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.schema;
+
+import java.util.Arrays;
+
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
+
+import org.apache.phoenix.util.SizedUtil;
+
+
+/**
+ *
+ * Class to track whether or not a value is null.
+ * The value is a zero-based position in the schema provided.
+ *
+ * @author jtaylor
+ * @since 0.1
+ *
+ */
+public class ValueBitSet {
+ public final static ValueBitSet EMPTY_VALUE_BITSET = new ValueBitSet();
+ private static final int BITS_PER_LONG = 64;
+ private static final int BITS_PER_SHORT = 16;
+ private final long[] bits;
+ private final ValueSchema schema;
+
+ private int maxSetBit = -1;
+
+ public static ValueBitSet newInstance(ValueSchema schema) {
+ if (schema.getFieldCount() == schema.getMinNullable()) {
+ return EMPTY_VALUE_BITSET;
+ }
+ return new ValueBitSet(schema);
+ }
+
+ private ValueBitSet() {
+ schema = null;
+ bits = new long[0];
+ }
+
+ private ValueBitSet(ValueSchema schema) {
+ this.schema = schema;
+ bits = new long[Math.max(1,(schema.getFieldCount() - schema.getMinNullable() + BITS_PER_LONG -1) / BITS_PER_LONG)];
+ }
+
+ public int getMaxSetBit() {
+ return maxSetBit;
+ }
+
+ private boolean isVarLength() {
+ return schema == null ? false : schema.getFieldCount() - schema.getMinNullable() > BITS_PER_SHORT;
+ }
+
+ public int getNullCount(int nBit, int nFields) {
+ if (schema == null) {
+ return 0;
+ }
+ int count = 0;
+ int index = nBit/BITS_PER_LONG;
+ // Shift right based on the bit index, because we aren't interested in the bits before this.
+ int shiftRight = nBit % BITS_PER_LONG;
+ int bitsToLeft = BITS_PER_LONG - shiftRight;
+ // Shift left based on the number of fields we're interested in counting.
+ int shiftLeft = Math.max(0, (BITS_PER_LONG - nFields));
+ // Mask off the bits of interest by shifting the bitset.
+ count += Math.min(nFields, bitsToLeft) - (Long.bitCount((bits[index] >>> shiftRight) << shiftLeft));
+ // Subtract from the number of fields the total number of possible fields we looked at
+ nFields -= bitsToLeft;
+ if (nFields > 0) {
+ // If more fields to count, then walk through the successive long bits
+ while (nFields > BITS_PER_LONG) {
+ count += BITS_PER_LONG - Long.bitCount(bits[++index]);
+ nFields -= BITS_PER_LONG;
+ }
+ // Count the final remaining fields
+ if (nFields > 0) {
+ count += nFields - Long.bitCount(bits[++index] << (BITS_PER_LONG - nFields));
+ }
+ }
+ return count;
+ }
+
+ /**
+ * Serialize the value bit set into a byte array. The byte array
+ * is expected to have enough room (use {@link #getEstimatedLength()}
+ * to ensure enough room exists.
+ * @param b the byte array into which to put the serialized bit set
+ * @param offset the offset into the byte array
+ * @return the incremented offset
+ */
+ public int toBytes(byte[] b, int offset) {
+ if (schema == null) {
+ return offset;
+ }
+ // If the total number of possible values is bigger than 16 bits (the
+ // size of a short), then serialize the long array followed by the
+ // array length.
+ if (isVarLength()) {
+ short nLongs = (short)((maxSetBit + BITS_PER_LONG - 1) / BITS_PER_LONG);
+ for (int i = 0; i < nLongs; i++) {
+ offset = Bytes.putLong(b, offset, bits[i]);
+ }
+ offset = Bytes.putShort(b, offset, nLongs);
+ } else {
+ // Else if the number of values is less than or equal to 16,
+ // serialize the bits directly into a short.
+ offset = Bytes.putShort(b, offset, (short)bits[0]);
+ }
+ return offset;
+ }
+
+ public void clear() {
+ Arrays.fill(bits, 0);
+ maxSetBit = -1;
+ }
+
+ public boolean get(int nBit) {
+ int lIndex = nBit / BITS_PER_LONG;
+ int bIndex = nBit % BITS_PER_LONG;
+ return (bits[lIndex] & (1L << bIndex)) != 0;
+ }
+
+ public void set(int nBit) {
+ int lIndex = nBit / BITS_PER_LONG;
+ int bIndex = nBit % BITS_PER_LONG;
+ bits[lIndex] |= (1L << bIndex);
+ maxSetBit = Math.max(maxSetBit, nBit);
+ }
+
+ public void or(ImmutableBytesWritable ptr) {
+ if (schema == null) {
+ return;
+ }
+ if (isVarLength()) {
+ int offset = ptr.getOffset() + ptr.getLength() - Bytes.SIZEOF_SHORT;
+ short nLongs = Bytes.toShort(ptr.get(), offset);
+ offset -= nLongs * Bytes.SIZEOF_LONG;
+ for (int i = 0; i < nLongs; i++) {
+ bits[i] |= Bytes.toLong(ptr.get(), offset);
+ offset += Bytes.SIZEOF_LONG;
+ }
+ maxSetBit = Math.max(maxSetBit, nLongs * Bytes.SIZEOF_LONG - 1);
+ } else {
+ long l = Bytes.toShort(ptr.get(), ptr.getOffset() + ptr.getLength() - Bytes.SIZEOF_SHORT);
+ bits[0] |= l;
+ maxSetBit = Math.max(maxSetBit, BITS_PER_SHORT - 1);
+ }
+
+ }
+
+ /**
+ * @return Max serialization size
+ */
+ public int getEstimatedLength() {
+ if (schema == null) {
+ return 0;
+ }
+ return Bytes.SIZEOF_SHORT + (isVarLength() ? (maxSetBit + BITS_PER_LONG - 1) / BITS_PER_LONG * Bytes.SIZEOF_LONG : 0);
+ }
+
+ public static int getSize(int nBits) {
+ return SizedUtil.OBJECT_SIZE + SizedUtil.POINTER_SIZE + SizedUtil.ARRAY_SIZE + SizedUtil.INT_SIZE + (nBits + BITS_PER_LONG - 1) / BITS_PER_LONG * Bytes.SIZEOF_LONG;
+ }
+
+ /**
+ * @return Size of object in memory
+ */
+ public int getSize() {
+ if (schema == null) {
+ return 0;
+ }
+ return SizedUtil.OBJECT_SIZE + SizedUtil.POINTER_SIZE + SizedUtil.ARRAY_SIZE + SizedUtil.LONG_SIZE * bits.length + SizedUtil.INT_SIZE;
+ }
+
+ public void or(ValueBitSet isSet) {
+ for (int i = 0; i < bits.length; i++) {
+ bits[i] |= isSet.bits[i];
+ }
+ maxSetBit = Math.max(maxSetBit, isSet.maxSetBit);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/schema/ValueRangeExcpetion.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/ValueRangeExcpetion.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/ValueRangeExcpetion.java
new file mode 100644
index 0000000..aff2ef7
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/ValueRangeExcpetion.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.schema;
+
+import java.sql.SQLException;
+
+import org.apache.phoenix.exception.SQLExceptionCode;
+import org.apache.phoenix.exception.SQLExceptionInfo;
+
+/**
+ * Exception thrown when we try to use use an argument that has the wrong type.
+ *
+ * @author anoopsjohn
+ * @since 1.1.2
+ */
+public class ValueRangeExcpetion extends SQLException{
+ private static final long serialVersionUID = 1L;
+ private static SQLExceptionCode code = SQLExceptionCode.VALUE_OUTSIDE_RANGE;
+
+ public ValueRangeExcpetion(Object minValue, Object maxValue, Object actualValue, String location){
+ super(new SQLExceptionInfo.Builder(code).setMessage("expected: [" + minValue + " , " + maxValue + "] but was: " + actualValue + " at " + location).build().toString(), code.getSQLState(), code.getErrorCode());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/schema/ValueSchema.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/ValueSchema.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/ValueSchema.java
new file mode 100644
index 0000000..41daf2a
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/ValueSchema.java
@@ -0,0 +1,337 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.schema;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableUtils;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+/**
+ *
+ * Simple flat schema over a byte array where fields may be any of {@link PDataType}.
+ * Optimized for positional access by index.
+ *
+ * @author jtaylor
+ * @since 0.1
+ */
+public abstract class ValueSchema implements Writable {
+ public static final int ESTIMATED_VARIABLE_LENGTH_SIZE = 10;
+ private int[] fieldIndexByPosition;
+ private List<Field> fields;
+ private int estimatedLength;
+ private boolean isFixedLength;
+ private boolean isMaxLength;
+ private int minNullable;
+
+ public ValueSchema() {
+ }
+
+ protected ValueSchema(int minNullable, List<Field> fields) {
+ init(minNullable, fields);
+ }
+
+ private void init(int minNullable, List<Field> fields) {
+ this.minNullable = minNullable;
+ this.fields = ImmutableList.copyOf(fields);
+ int estimatedLength = 0;
+ boolean isMaxLength = true, isFixedLength = true;
+ int positions = 0;
+ for (Field field : fields) {
+ int fieldEstLength = 0;
+ PDataType type = field.getDataType();
+ Integer byteSize = type.getByteSize();
+ if (type.isFixedWidth()) {
+ fieldEstLength += field.getByteSize();
+ } else {
+ isFixedLength = false;
+ // Account for vint for length if not fixed
+ if (byteSize == null) {
+ isMaxLength = false;
+ fieldEstLength += ESTIMATED_VARIABLE_LENGTH_SIZE;
+ } else {
+ fieldEstLength += WritableUtils.getVIntSize(byteSize);
+ fieldEstLength = byteSize;
+ }
+ }
+ positions += field.getCount();
+ estimatedLength += fieldEstLength * field.getCount();
+ }
+ fieldIndexByPosition = new int[positions];
+ for (int i = 0, j= 0; i < fields.size(); i++) {
+ Field field = fields.get(i);
+ Arrays.fill(fieldIndexByPosition, j, j + field.getCount(), i);
+ j += field.getCount();
+ }
+ this.isFixedLength = isFixedLength;
+ this.isMaxLength = isMaxLength;
+ this.estimatedLength = estimatedLength;
+ }
+
+ public int getFieldCount() {
+ return fieldIndexByPosition.length;
+ }
+
+ public List<Field> getFields() {
+ return fields;
+ }
+
+ /**
+ * @return true if all types are fixed width
+ */
+ public boolean isFixedLength() {
+ return isFixedLength;
+ }
+
+ /**
+ * @return true if {@link #getEstimatedValueLength()} returns the maximum length
+ * of a serialized value for this schema
+ */
+ public boolean isMaxLength() {
+ return isMaxLength;
+ }
+
+ /**
+ * @return estimated size in bytes of a serialized value for this schema
+ */
+ public int getEstimatedValueLength() {
+ return estimatedLength;
+ }
+
+ /**
+ * Non-nullable fields packed to the left so that we do not need to store trailing nulls.
+ * Knowing the minimum position of a nullable field enables this.
+ * @return the minimum position of a nullable field
+ */
+ public int getMinNullable() {
+ return minNullable;
+ }
+
+ public static final class Field implements Writable {
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + byteSize;
+ result = prime * result + type.hashCode();
+ result = prime * result + ((columnModifier == null) ? 0 : columnModifier.hashCode());
+ result = prime * result + (isNullable ? 1231 : 1237);
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) return true;
+ if (obj == null) return false;
+ if (getClass() != obj.getClass()) return false;
+ Field other = (Field)obj;
+ if (byteSize != other.byteSize) return false;
+ if (columnModifier != other.columnModifier) return false;
+ if (isNullable != other.isNullable) return false;
+ if (type != other.type) return false;
+ return true;
+ }
+
+ private int count;
+ private PDataType type;
+ private int byteSize = 0;
+ private boolean isNullable;
+ private ColumnModifier columnModifier;
+
+ public Field() {
+ }
+
+ private Field(PDatum datum, boolean isNullable, int count, ColumnModifier columnModifier) {
+ this.type = datum.getDataType();
+ this.columnModifier = columnModifier;
+ this.count = count;
+ this.isNullable = isNullable;
+ if (this.type.isFixedWidth() && this.type.getByteSize() == null) {
+ this.byteSize = datum.getByteSize();
+ }
+ }
+
+ private Field(Field field, int count) {
+ this.type = field.getDataType();
+ this.byteSize = field.byteSize;
+ this.count = count;
+ }
+
+ public final ColumnModifier getColumnModifier() {
+ return columnModifier;
+ }
+
+ public final PDataType getDataType() {
+ return type;
+ }
+
+ public final boolean isNullable() {
+ return isNullable;
+ }
+
+ public final int getByteSize() {
+ return type.getByteSize() == null ? byteSize : type.getByteSize();
+ }
+
+ public final int getCount() {
+ return count;
+ }
+
+ @Override
+ public void readFields(DataInput input) throws IOException {
+ // Encode isNullable in sign bit of type ordinal (offset by 1, since ordinal could be 0)
+ int typeOrdinal = WritableUtils.readVInt(input);
+ if (typeOrdinal < 0) {
+ typeOrdinal *= -1;
+ this.isNullable = true;
+ }
+ this.type = PDataType.values()[typeOrdinal-1];
+ this.count = WritableUtils.readVInt(input);
+ if (this.count < 0) {
+ this.count *= -1;
+ this.columnModifier = ColumnModifier.SORT_DESC;
+ }
+ if (this.type.isFixedWidth() && this.type.getByteSize() == null) {
+ this.byteSize = WritableUtils.readVInt(input);
+ }
+ }
+
+ @Override
+ public void write(DataOutput output) throws IOException {
+ WritableUtils.writeVInt(output, (type.ordinal() + 1) * (this.isNullable ? -1 : 1));
+ WritableUtils.writeVInt(output, count * (columnModifier == null ? 1 : -1));
+ if (type.isFixedWidth() && type.getByteSize() == null) {
+ WritableUtils.writeVInt(output, byteSize);
+ }
+ }
+ }
+
+ public abstract static class ValueSchemaBuilder {
+ protected List<Field> fields = new ArrayList<Field>();
+ protected int nFields = Integer.MAX_VALUE;
+ protected final int minNullable;
+
+ public ValueSchemaBuilder(int minNullable) {
+ this.minNullable = minNullable;
+ }
+
+ protected List<Field> buildFields() {
+ List<Field> condensedFields = new ArrayList<Field>(fields.size());
+ for (int i = 0; i < Math.min(nFields,fields.size()); ) {
+ Field field = fields.get(i);
+ int count = 1;
+ while ( ++i < fields.size() && field.equals(fields.get(i))) {
+ count++;
+ }
+ condensedFields.add(count == 1 ? field : new Field(field,count));
+ }
+ return condensedFields;
+ }
+
+ abstract public ValueSchema build();
+
+ public ValueSchemaBuilder setMaxFields(int nFields) {
+ this.nFields = nFields;
+ return this;
+ }
+
+ protected ValueSchemaBuilder addField(PDatum datum, boolean isNullable, ColumnModifier columnModifier) {
+ fields.add(new Field(datum, isNullable, 1, columnModifier));
+ return this;
+ }
+ }
+
+ public int getEstimatedByteSize() {
+ int size = 0;
+ size += WritableUtils.getVIntSize(minNullable);
+ size += WritableUtils.getVIntSize(fields.size());
+ size += fields.size() * 3;
+ return size;
+ }
+
+ public void serialize(DataOutput output) throws IOException {
+ WritableUtils.writeVInt(output, minNullable);
+ WritableUtils.writeVInt(output, fields.size());
+ for (int i = 0; i < fields.size(); i++) {
+ fields.get(i).write(output);
+ }
+ }
+
+ public Field getField(int position) {
+ return fields.get(fieldIndexByPosition[position]);
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + minNullable;
+ for (Field field : fields) {
+ result = prime * result + field.hashCode();
+ }
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) return true;
+ if (obj == null) return false;
+ if (getClass() != obj.getClass()) return false;
+ ValueSchema other = (ValueSchema)obj;
+ if (minNullable != other.minNullable) return false;
+ if (fields.size() != other.fields.size()) return false;
+ for (int i = 0; i < fields.size(); i++) {
+ if (!fields.get(i).equals(other.fields.get(i)))
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ int minNullable = WritableUtils.readVInt(in);
+ int nFields = WritableUtils.readVInt(in);
+ List<Field> fields = Lists.newArrayListWithExpectedSize(nFields);
+ for (int i = 0; i < nFields; i++) {
+ Field field = new Field();
+ field.readFields(in);
+ fields.add(field);
+ }
+ init(minNullable, fields);
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ WritableUtils.writeVInt(out, minNullable);
+ WritableUtils.writeVInt(out, fields.size());
+ for (int i = 0; i < fields.size(); i++) {
+ fields.get(i).write(out);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/schema/stat/PTableStats.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stat/PTableStats.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stat/PTableStats.java
new file mode 100644
index 0000000..5e38a27
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stat/PTableStats.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.schema.stat;
+
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.HRegionInfo;
+
+
+/**
+ * Interface for Phoenix table statistics. Statistics is collected on the server
+ * side and can be used for various purpose like splitting region for scanning, etc.
+ *
+ * The table is defined on the client side, but it is populated on the server side. The client
+ * should not populate any data to the statistics object.
+ */
+public interface PTableStats {
+
+ /**
+ * Given the region info, returns an array of bytes that is the current estimate of key
+ * distribution inside that region. The keys should split that region into equal chunks.
+ *
+ * @param region
+ * @return array of keys
+ */
+ byte[][] getRegionGuidePosts(HRegionInfo region);
+
+ void write(DataOutput output) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/schema/stat/PTableStatsImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stat/PTableStatsImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stat/PTableStatsImpl.java
new file mode 100644
index 0000000..27bb665
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stat/PTableStatsImpl.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.schema.stat;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.WritableUtils;
+
+import com.google.common.collect.ImmutableMap;
+
+
+/**
+ * Implementation for PTableStats.
+ */
+public class PTableStatsImpl implements PTableStats {
+
+ // The map for guide posts should be immutable. We only take the current snapshot from outside
+ // method call and store it.
+ private Map<String, byte[][]> regionGuidePosts;
+
+ public PTableStatsImpl() { }
+
+ public PTableStatsImpl(Map<String, byte[][]> stats) {
+ regionGuidePosts = ImmutableMap.copyOf(stats);
+ }
+
+ @Override
+ public byte[][] getRegionGuidePosts(HRegionInfo region) {
+ return regionGuidePosts.get(region.getRegionNameAsString());
+ }
+
+ @Override
+ public void write(DataOutput output) throws IOException {
+ if (regionGuidePosts == null) {
+ WritableUtils.writeVInt(output, 0);
+ return;
+ }
+ WritableUtils.writeVInt(output, regionGuidePosts.size());
+ for (Entry<String, byte[][]> entry : regionGuidePosts.entrySet()) {
+ WritableUtils.writeString(output, entry.getKey());
+ byte[][] value = entry.getValue();
+ WritableUtils.writeVInt(output, value.length);
+ for (int i=0; i<value.length; i++) {
+ Bytes.writeByteArray(output, value[i]);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/MultiKeyValueTuple.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/MultiKeyValueTuple.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/MultiKeyValueTuple.java
new file mode 100644
index 0000000..ed69560
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/MultiKeyValueTuple.java
@@ -0,0 +1,85 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.schema.tuple;
+
+import java.util.List;
+
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.phoenix.util.KeyValueUtil;
+
+
+public class MultiKeyValueTuple implements Tuple {
+ private List<KeyValue> values;
+
+ public MultiKeyValueTuple(List<KeyValue> values) {
+ setKeyValues(values);
+ }
+
+ public MultiKeyValueTuple() {
+ }
+
+ public void setKeyValues(List<KeyValue> values) {
+ this.values = ImmutableList.copyOf(values);
+ }
+
+ @Override
+ public void getKey(ImmutableBytesWritable ptr) {
+ KeyValue value = values.get(0);
+ ptr.set(value.getBuffer(), value.getRowOffset(), value.getRowLength());
+ }
+
+ @Override
+ public boolean isImmutable() {
+ return true;
+ }
+
+ @Override
+ public KeyValue getValue(byte[] family, byte[] qualifier) {
+ return KeyValueUtil.getColumnLatest(values, family, qualifier);
+ }
+
+ @Override
+ public String toString() {
+ return values.toString();
+ }
+
+ @Override
+ public int size() {
+ return values.size();
+ }
+
+ @Override
+ public KeyValue getValue(int index) {
+ return values.get(index);
+ }
+
+ @Override
+ public boolean getValue(byte[] family, byte[] qualifier,
+ ImmutableBytesWritable ptr) {
+ KeyValue kv = getValue(family, qualifier);
+ if (kv == null)
+ return false;
+ ptr.set(kv.getBuffer(), kv.getValueOffset(), kv.getValueLength());
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/ResultTuple.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/ResultTuple.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/ResultTuple.java
new file mode 100644
index 0000000..c99a990
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/ResultTuple.java
@@ -0,0 +1,104 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.schema.tuple;
+
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
+
+import org.apache.phoenix.util.ResultUtil;
+
+
+public class ResultTuple implements Tuple {
+ private Result result;
+
+ public ResultTuple(Result result) {
+ this.result = result;
+ }
+
+ public ResultTuple() {
+ }
+
+ public Result getResult() {
+ return this.result;
+ }
+
+ public void setResult(Result result) {
+ this.result = result;
+ }
+
+ @Override
+ public void getKey(ImmutableBytesWritable ptr) {
+ ResultUtil.getKey(result, ptr);
+ }
+
+ @Override
+ public boolean isImmutable() {
+ return true;
+ }
+
+ @Override
+ public KeyValue getValue(byte[] family, byte[] qualifier) {
+ return result.getColumnLatest(family, qualifier);
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("keyvalues=");
+ if(this.result.isEmpty()) {
+ sb.append("NONE");
+ return sb.toString();
+ }
+ sb.append("{");
+ boolean moreThanOne = false;
+ for(KeyValue kv : this.result.list()) {
+ if(moreThanOne) {
+ sb.append(", \n");
+ } else {
+ moreThanOne = true;
+ }
+ sb.append(kv.toString()+"/value="+Bytes.toString(kv.getValue()));
+ }
+ sb.append("}\n");
+ return sb.toString();
+ }
+
+ @Override
+ public int size() {
+ return result.size();
+ }
+
+ @Override
+ public KeyValue getValue(int index) {
+ return result.raw()[index];
+ }
+
+ @Override
+ public boolean getValue(byte[] family, byte[] qualifier,
+ ImmutableBytesWritable ptr) {
+ KeyValue kv = getValue(family, qualifier);
+ if (kv == null)
+ return false;
+ ptr.set(kv.getBuffer(), kv.getValueOffset(), kv.getValueLength());
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/SingleKeyValueTuple.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/SingleKeyValueTuple.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/SingleKeyValueTuple.java
new file mode 100644
index 0000000..5d76604
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/SingleKeyValueTuple.java
@@ -0,0 +1,111 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.schema.tuple;
+
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
+
+
+public class SingleKeyValueTuple implements Tuple {
+ private static final byte[] UNITIALIZED_KEY_BUFFER = new byte[0];
+ private KeyValue keyValue;
+ private ImmutableBytesWritable keyPtr = new ImmutableBytesWritable(UNITIALIZED_KEY_BUFFER);
+
+ public SingleKeyValueTuple() {
+ }
+
+ public SingleKeyValueTuple(KeyValue keyValue) {
+ if (keyValue == null) {
+ throw new NullPointerException();
+ }
+ setKeyValue(keyValue);
+ }
+
+ public boolean hasKey() {
+ return keyPtr.get() != UNITIALIZED_KEY_BUFFER;
+ }
+
+ public void reset() {
+ this.keyValue = null;
+ keyPtr.set(UNITIALIZED_KEY_BUFFER);
+ }
+
+ public void setKeyValue(KeyValue keyValue) {
+ if (keyValue == null) {
+ throw new IllegalArgumentException();
+ }
+ this.keyValue = keyValue;
+ setKey(keyValue);
+ }
+
+ public void setKey(ImmutableBytesWritable ptr) {
+ keyPtr.set(ptr.get(), ptr.getOffset(), ptr.getLength());
+ }
+
+ public void setKey(KeyValue keyValue) {
+ if (keyValue == null) {
+ throw new IllegalArgumentException();
+ }
+ keyPtr.set(keyValue.getBuffer(), keyValue.getRowOffset(), keyValue.getRowLength());
+ }
+
+ @Override
+ public void getKey(ImmutableBytesWritable ptr) {
+ ptr.set(keyPtr.get(), keyPtr.getOffset(), keyPtr.getLength());
+ }
+
+ @Override
+ public KeyValue getValue(byte[] cf, byte[] cq) {
+ return keyValue;
+ }
+
+ @Override
+ public boolean isImmutable() {
+ return true;
+ }
+
+ @Override
+ public String toString() {
+ return "SingleKeyValueTuple[" + keyValue == null ? keyPtr.get() == UNITIALIZED_KEY_BUFFER ? "null" : Bytes.toStringBinary(keyPtr.get(),keyPtr.getOffset(),keyPtr.getLength()) : keyValue.toString() + "]";
+ }
+
+ @Override
+ public int size() {
+ return keyValue == null ? 0 : 1;
+ }
+
+ @Override
+ public KeyValue getValue(int index) {
+ if (index != 0 || keyValue == null) {
+ throw new IndexOutOfBoundsException(Integer.toString(index));
+ }
+ return keyValue;
+ }
+
+ @Override
+ public boolean getValue(byte[] family, byte[] qualifier,
+ ImmutableBytesWritable ptr) {
+ if (keyValue == null)
+ return false;
+ ptr.set(keyValue.getBuffer(), keyValue.getValueOffset(), keyValue.getValueLength());
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/Tuple.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/Tuple.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/Tuple.java
new file mode 100644
index 0000000..ffa3f72
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/Tuple.java
@@ -0,0 +1,84 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.schema.tuple;
+
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+
+/**
+ *
+ * Interface representing an ordered list of KeyValues returned as the
+ * result of a query. Each tuple represents a row (i.e. all its KeyValues
+ * will have the same key), and each KeyValue represents a column value.
+ *
+ * @author jtaylor
+ * @since 0.1
+ */
+public interface Tuple {
+ /**
+ * @return Number of KeyValues contained by the Tuple.
+ */
+ public int size();
+
+ /**
+ * Determines whether or not the Tuple is immutable (the typical case)
+ * or will potentially have additional KeyValues added to it (the case
+ * during filter evaluation when we see one KeyValue at a time).
+ * @return true if Tuple is immutable and false otherwise.
+ */
+ public boolean isImmutable();
+
+ /**
+ * Get the row key for the Tuple
+ * @param ptr the bytes pointer that will be updated to point to
+ * the key buffer.
+ */
+ public void getKey(ImmutableBytesWritable ptr);
+
+ /**
+ * Get the KeyValue at the given index.
+ * @param index the zero-based KeyValue index between 0 and {@link #size()} exclusive
+ * @return the KeyValue at the given index
+ * @throws IndexOutOfBoundsException if an invalid index is used
+ */
+ public KeyValue getValue(int index);
+
+ /***
+ * Get the KeyValue contained by the Tuple with the given family and
+ * qualifier name.
+ * @param family the column family of the KeyValue being retrieved
+ * @param qualifier the column qualify of the KeyValue being retrieved
+ * @return the KeyValue with the given family and qualifier name or
+ * null if not found.
+ */
+ public KeyValue getValue(byte [] family, byte [] qualifier);
+
+ /***
+ * Get the value byte array of the KeyValue contained by the Tuple with
+ * the given family and qualifier name.
+ * @param family the column family of the KeyValue being retrieved
+ * @param qualifier the column qualify of the KeyValue being retrieved
+ * @param ptr the bytes pointer that will be updated to point to the
+ * value buffer.
+ * @return true if the KeyValue with the given family and qualifier name
+ * exists; otherwise false.
+ */
+ public boolean getValue(byte [] family, byte [] qualifier, ImmutableBytesWritable ptr);
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/util/BigDecimalUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/BigDecimalUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/BigDecimalUtil.java
new file mode 100644
index 0000000..4b49f0f
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/BigDecimalUtil.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.util;
+
+import org.apache.hadoop.hbase.util.Pair;
+
+/**
+ *
+ * @author anoopsjohn
+ * @since 1.2.1
+ */
+public class BigDecimalUtil {
+
+ /**
+ * Calculates the precision and scale for BigDecimal arithmetic operation results. It uses the algorithm mentioned
+ * <a href="http://db.apache.org/derby/docs/10.0/manuals/reference/sqlj124.html#HDRSII-SQLJ-36146">here</a>
+ * @param lp precision of the left operand
+ * @param ls scale of the left operand
+ * @param rp precision of the right operand
+ * @param rs scale of the right operand
+ * @param op The operation type
+ * @return {@link Pair} comprising of the precision and scale.
+ */
+ public static Pair<Integer, Integer> getResultPrecisionScale(int lp, int ls, int rp, int rs, Operation op) {
+ int resultPrec = 0, resultScale = 0;
+ switch (op) {
+ case MULTIPLY:
+ resultPrec = lp + rp;
+ resultScale = ls + rs;
+ break;
+ case DIVIDE:
+ resultPrec = lp - ls + rp + Math.max(ls + rp - rs + 1, 4);
+ resultScale = 31 - lp + ls - rs;
+ break;
+ case ADD:
+ resultPrec = 2 * (lp - ls) + ls; // Is this correct? The page says addition -> 2 * (p - s) + s.
+ resultScale = Math.max(ls, rs);
+ break;
+ case AVG:
+ resultPrec = Math.max(lp - ls, rp - rs) + 1 + Math.max(ls, rs);
+ resultScale = Math.max(Math.max(ls, rs), 4);
+ break;
+ case OTHERS:
+ resultPrec = Math.max(lp - ls, rp - rs) + 1 + Math.max(ls, rs);
+ resultScale = Math.max(ls, rs);
+ }
+ return new Pair<Integer, Integer>(resultPrec, resultScale);
+ }
+
+ public static enum Operation {
+ MULTIPLY, DIVIDE, ADD, AVG, OTHERS;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/util/BitSet.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/BitSet.java b/phoenix-core/src/main/java/org/apache/phoenix/util/BitSet.java
new file mode 100644
index 0000000..7cdbe6a
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/BitSet.java
@@ -0,0 +1,106 @@
+package org.apache.phoenix.util;
+
+import java.io.*;
+
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ *
+ * BitSet that can be initialized with primitive types, which
+ * is only available in Java 7 or above.
+ *
+ * @author jtaylor
+ * @since 2.1.0
+ */
+public class BitSet {
+ public static final int BITS_PER_LONG = 64;
+ public static final int BITS_PER_INT = 32;
+ public static final int BITS_PER_SHORT = 16;
+ public static final int BITS_PER_BYTE = 8;
+ private final long[] bits;
+
+ public static int getByteSize(int capacity) {
+ if (capacity <= BitSet.BITS_PER_BYTE) {
+ return Bytes.SIZEOF_BYTE;
+ } else if (capacity <= BitSet.BITS_PER_SHORT) {
+ return Bytes.SIZEOF_SHORT;
+ } else if (capacity <= BitSet.BITS_PER_INT) {
+ return Bytes.SIZEOF_INT;
+ } else if (capacity <= BitSet.BITS_PER_LONG) {
+ return Bytes.SIZEOF_LONG;
+ } else {
+ int nLongs = (capacity-1) / BitSet.BITS_PER_LONG + 1;
+ return nLongs * Bytes.SIZEOF_LONG;
+ }
+ }
+
+ public static BitSet read(DataInput input, int capacity) throws IOException {
+ if (capacity <= BitSet.BITS_PER_BYTE) {
+ return fromPrimitive(input.readByte());
+ } else if (capacity <= BitSet.BITS_PER_SHORT) {
+ return fromPrimitive(input.readShort());
+ } else if (capacity <= BitSet.BITS_PER_INT) {
+ return fromPrimitive(input.readInt());
+ } else if (capacity <= BitSet.BITS_PER_LONG) {
+ return fromPrimitive(input.readLong());
+ } else {
+ int nLongs = (capacity-1) / BitSet.BITS_PER_LONG + 1;
+ return fromArray(ByteUtil.readFixedLengthLongArray(input, nLongs));
+ }
+ }
+
+ public static void write(DataOutput output, BitSet bitSet, int capacity) throws IOException {
+ if (capacity <= BitSet.BITS_PER_BYTE) {
+ output.writeByte((byte)bitSet.bits[0]);
+ } else if (capacity <= BitSet.BITS_PER_SHORT) {
+ output.writeShort((short)bitSet.bits[0]);
+ } else if (capacity <= BitSet.BITS_PER_INT) {
+ output.writeInt((int)bitSet.bits[0]);
+ } else if (capacity <= BitSet.BITS_PER_LONG) {
+ output.writeLong(bitSet.bits[0]);
+ } else {
+ ByteUtil.writeFixedLengthLongArray(output, bitSet.bits);
+ }
+ }
+
+ public static BitSet fromPrimitive(byte bits) {
+ return new BitSet(new long[] { bits });
+ }
+
+ public static BitSet fromPrimitive(short bits) {
+ return new BitSet(new long[] { bits });
+ }
+
+ public static BitSet fromPrimitive(int bits) {
+ return new BitSet(new long[] { bits });
+ }
+
+ public static BitSet fromPrimitive(long bits) {
+ return new BitSet(new long[] { bits });
+ }
+
+ public static BitSet fromArray(long[] bits) {
+ return new BitSet(bits);
+ }
+
+ public static BitSet withCapacity(int maxBits) {
+ int size = Math.max(1,(maxBits + BITS_PER_LONG -1) / BITS_PER_LONG);
+ return new BitSet(new long[size]);
+ }
+
+ public BitSet(long[] bits) {
+ this.bits = bits;
+ }
+
+ public boolean get(int nBit) {
+ int lIndex = nBit / BITS_PER_LONG;
+ int bIndex = nBit % BITS_PER_LONG;
+ return (bits[lIndex] & (1L << bIndex)) != 0;
+ }
+
+ public void set(int nBit) {
+ int lIndex = nBit / BITS_PER_LONG;
+ int bIndex = nBit % BITS_PER_LONG;
+ bits[lIndex] |= (1L << bIndex);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/util/ByteUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ByteUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ByteUtil.java
new file mode 100644
index 0000000..93c4df2
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ByteUtil.java
@@ -0,0 +1,561 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.util;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Comparator;
+
+import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.WritableUtils;
+
+import org.apache.hadoop.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.query.KeyRange;
+import org.apache.phoenix.schema.ColumnModifier;
+import org.apache.phoenix.schema.PDataType;
+
+
+/**
+ *
+ * Byte utilities
+ *
+ * @author jtaylor
+ * @since 0.1
+ */
+public class ByteUtil {
+ public static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
+ public static final ImmutableBytesPtr EMPTY_BYTE_ARRAY_PTR = new ImmutableBytesPtr(
+ EMPTY_BYTE_ARRAY);
+
+ public static final Comparator<ImmutableBytesPtr> BYTES_PTR_COMPARATOR = new Comparator<ImmutableBytesPtr>() {
+
+ @Override
+ public int compare(ImmutableBytesPtr o1, ImmutableBytesPtr o2) {
+ return Bytes.compareTo(o1.get(), o1.getOffset(), o1.getLength(), o2.get(), o2.getOffset(), o2.getLength());
+ }
+
+ };
+
+ /**
+ * Serialize an array of byte arrays into a single byte array. Used
+ * to pass through a set of bytes arrays as an attribute of a Scan.
+ * Use {@link #toByteArrays(byte[], int)} to convert the serialized
+ * byte array back to the array of byte arrays.
+ * @param byteArrays the array of byte arrays to serialize
+ * @return the byte array
+ */
+ public static byte[] toBytes(byte[][] byteArrays) {
+ int size = 0;
+ for (byte[] b : byteArrays) {
+ if (b == null) {
+ size++;
+ } else {
+ size += b.length;
+ size += WritableUtils.getVIntSize(b.length);
+ }
+ }
+ TrustedByteArrayOutputStream bytesOut = new TrustedByteArrayOutputStream(size);
+ DataOutputStream out = new DataOutputStream(bytesOut);
+ try {
+ for (byte[] b : byteArrays) {
+ if (b == null) {
+ WritableUtils.writeVInt(out, 0);
+ } else {
+ WritableUtils.writeVInt(out, b.length);
+ out.write(b);
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e); // not possible
+ } finally {
+ try {
+ out.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e); // not possible
+ }
+ }
+ return bytesOut.getBuffer();
+ }
+
+ /**
+ * Deserialize a byte array into a set of byte arrays. Used in
+ * coprocessor to reconstruct byte arrays from attribute value
+ * passed through the Scan.
+ * @param b byte array containing serialized byte arrays (created by {@link #toBytes(byte[][])}).
+ * @param length number of byte arrays that were serialized
+ * @return array of now deserialized byte arrays
+ * @throws IllegalStateException if there are more than length number of byte arrays that were serialized
+ */
+ public static byte[][] toByteArrays(byte[] b, int length) {
+ return toByteArrays(b, 0, length);
+ }
+
+ public static byte[][] toByteArrays(byte[] b, int offset, int length) {
+ ByteArrayInputStream bytesIn = new ByteArrayInputStream(b, offset, b.length - offset);
+ DataInputStream in = new DataInputStream(bytesIn);
+ byte[][] byteArrays = new byte[length][];
+ try {
+ for (int i = 0; i < length; i++) {
+ int bLength = WritableUtils.readVInt(in);
+ if (bLength == 0) {
+ byteArrays[i] = null;
+ } else {
+ byteArrays[i] = new byte[bLength];
+ int rLength = in.read(byteArrays[i], 0, bLength);
+ assert (rLength == bLength); // For find bugs
+ }
+ }
+ if (in.read() != -1) {
+ throw new IllegalStateException("Expected only " + length + " byte arrays, but found more");
+ }
+ return byteArrays;
+ } catch (IOException e) {
+ throw new RuntimeException(e); // not possible
+ } finally {
+ try {
+ in.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e); // not possible
+ }
+ }
+ }
+
+ public static byte[] serializeVIntArray(int[] intArray) {
+ return serializeVIntArray(intArray,intArray.length);
+ }
+
+ public static byte[] serializeVIntArray(int[] intArray, int encodedLength) {
+ int size = WritableUtils.getVIntSize(encodedLength);
+ for (int i = 0; i < intArray.length; i++) {
+ size += WritableUtils.getVIntSize(intArray[i]);
+ }
+ int offset = 0;
+ byte[] out = new byte[size];
+ offset += ByteUtil.vintToBytes(out, offset, size);
+ for (int i = 0; i < intArray.length; i++) {
+ offset += ByteUtil.vintToBytes(out, offset, intArray[i]);
+ }
+ return out;
+ }
+
+ public static void serializeVIntArray(DataOutput output, int[] intArray) throws IOException {
+ serializeVIntArray(output, intArray, intArray.length);
+ }
+
+ /**
+ * Allows additional stuff to be encoded in length
+ * @param output
+ * @param intArray
+ * @param encodedLength
+ * @throws IOException
+ */
+ public static void serializeVIntArray(DataOutput output, int[] intArray, int encodedLength) throws IOException {
+ WritableUtils.writeVInt(output, encodedLength);
+ for (int i = 0; i < intArray.length; i++) {
+ WritableUtils.writeVInt(output, intArray[i]);
+ }
+ }
+
+ public static long[] readFixedLengthLongArray(DataInput input, int length) throws IOException {
+ long[] longArray = new long[length];
+ for (int i = 0; i < length; i++) {
+ longArray[i] = input.readLong();
+ }
+ return longArray;
+ }
+
+ public static void writeFixedLengthLongArray(DataOutput output, long[] longArray) throws IOException {
+ for (int i = 0; i < longArray.length; i++) {
+ output.writeLong(longArray[i]);
+ }
+ }
+
+ /**
+ * Deserialize a byte array into a int array.
+ * @param b byte array storing serialized vints
+ * @return int array
+ */
+ public static int[] deserializeVIntArray(byte[] b) {
+ ByteArrayInputStream bytesIn = new ByteArrayInputStream(b);
+ DataInputStream in = new DataInputStream(bytesIn);
+ try {
+ int length = WritableUtils.readVInt(in);
+ return deserializeVIntArray(in, length);
+ } catch (IOException e) {
+ throw new RuntimeException(e); // not possible
+ } finally {
+ try {
+ in.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e); // not possible
+ }
+ }
+ }
+
+ public static int[] deserializeVIntArray(DataInput in) throws IOException {
+ return deserializeVIntArray(in, WritableUtils.readVInt(in));
+ }
+
+ public static int[] deserializeVIntArray(DataInput in, int length) throws IOException {
+ int i = 0;
+ int[] intArray = new int[length];
+ while (i < length) {
+ intArray[i++] = WritableUtils.readVInt(in);
+ }
+ return intArray;
+ }
+
+ /**
+ * Deserialize a byte array into a int array.
+ * @param b byte array storing serialized vints
+ * @param length number of serialized vints
+ * @return int array
+ */
+ public static int[] deserializeVIntArray(byte[] b, int length) {
+ ByteArrayInputStream bytesIn = new ByteArrayInputStream(b);
+ DataInputStream in = new DataInputStream(bytesIn);
+ try {
+ return deserializeVIntArray(in,length);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Concatenate together one or more byte arrays
+ * @param first first byte array
+ * @param rest rest of byte arrays
+ * @return newly allocated byte array that is a concatenation of all the byte arrays passed in
+ */
+ public static byte[] concat(byte[] first, byte[]... rest) {
+ int totalLength = first.length;
+ for (byte[] array : rest) {
+ totalLength += array.length;
+ }
+ byte[] result = Arrays.copyOf(first, totalLength);
+ int offset = first.length;
+ for (byte[] array : rest) {
+ System.arraycopy(array, 0, result, offset, array.length);
+ offset += array.length;
+ }
+ return result;
+ }
+
+ public static <T> T[] concat(T[] first, T[]... rest) {
+ int totalLength = first.length;
+ for (T[] array : rest) {
+ totalLength += array.length;
+ }
+ T[] result = Arrays.copyOf(first, totalLength);
+ int offset = first.length;
+ for (T[] array : rest) {
+ System.arraycopy(array, 0, result, offset, array.length);
+ offset += array.length;
+ }
+ return result;
+ }
+
+ public static byte[] concat(ColumnModifier columnModifier, ImmutableBytesWritable... writables) {
+ int totalLength = 0;
+ for (ImmutableBytesWritable writable : writables) {
+ totalLength += writable.getLength();
+ }
+ byte[] result = new byte[totalLength];
+ int offset = 0;
+ for (ImmutableBytesWritable array : writables) {
+ byte[] bytes = array.get();
+ if (columnModifier != null) {
+ bytes = columnModifier.apply(bytes, array.getOffset(), new byte[array.getLength()], 0, array.getLength());
+ }
+ System.arraycopy(bytes, array.getOffset(), result, offset, array.getLength());
+ offset += array.getLength();
+ }
+ return result;
+ }
+
+ public static int vintFromBytes(byte[] buffer, int offset) {
+ try {
+ return (int)Bytes.readVLong(buffer, offset);
+ } catch (IOException e) { // Impossible
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Decode a vint from the buffer pointed at to by ptr and
+ * increment the offset of the ptr by the length of the
+ * vint.
+ * @param ptr a pointer to a byte array buffer
+ * @return the decoded vint value as an int
+ */
+ public static int vintFromBytes(ImmutableBytesWritable ptr) {
+ return (int) vlongFromBytes(ptr);
+ }
+
+ /**
+ * Decode a vint from the buffer pointed at to by ptr and
+ * increment the offset of the ptr by the length of the
+ * vint.
+ * @param ptr a pointer to a byte array buffer
+ * @return the decoded vint value as a long
+ */
+ public static long vlongFromBytes(ImmutableBytesWritable ptr) {
+ final byte [] buffer = ptr.get();
+ final int offset = ptr.getOffset();
+ byte firstByte = buffer[offset];
+ int len = WritableUtils.decodeVIntSize(firstByte);
+ if (len == 1) {
+ ptr.set(buffer, offset+1, ptr.getLength());
+ return firstByte;
+ }
+ long i = 0;
+ for (int idx = 0; idx < len-1; idx++) {
+ byte b = buffer[offset + 1 + idx];
+ i = i << 8;
+ i = i | (b & 0xFF);
+ }
+ ptr.set(buffer, offset+len, ptr.getLength());
+ return (WritableUtils.isNegativeVInt(firstByte) ? ~i : i);
+ }
+
+
+ /**
+ * Put long as variable length encoded number at the offset in the result byte array
+ * @param vint Integer to make a vint of.
+ * @param result buffer to put vint into
+ * @return Vint length in bytes of vint
+ */
+ public static int vintToBytes(byte[] result, int offset, final long vint) {
+ long i = vint;
+ if (i >= -112 && i <= 127) {
+ result[offset] = (byte) i;
+ return 1;
+ }
+
+ int len = -112;
+ if (i < 0) {
+ i ^= -1L; // take one's complement'
+ len = -120;
+ }
+
+ long tmp = i;
+ while (tmp != 0) {
+ tmp = tmp >> 8;
+ len--;
+ }
+
+ result[offset++] = (byte) len;
+
+ len = (len < -120) ? -(len + 120) : -(len + 112);
+
+ for (int idx = len; idx != 0; idx--) {
+ int shiftbits = (idx - 1) * 8;
+ long mask = 0xFFL << shiftbits;
+ result[offset++] = (byte)((i & mask) >> shiftbits);
+ }
+ return len + 1;
+ }
+
+ /**
+ * Increment the key to the next key
+ * @param key the key to increment
+ * @return a new byte array with the next key or null
+ * if the key could not be incremented because it's
+ * already at its max value.
+ */
+ public static byte[] nextKey(byte[] key) {
+ byte[] nextStartRow = new byte[key.length];
+ System.arraycopy(key, 0, nextStartRow, 0, key.length);
+ if (!nextKey(nextStartRow, nextStartRow.length)) {
+ return null;
+ }
+ return nextStartRow;
+ }
+
+ /**
+ * Increment the key in-place to the next key
+ * @param key the key to increment
+ * @param length the length of the key
+ * @return true if the key can be incremented and
+ * false otherwise if the key is at its max
+ * value.
+ */
+ public static boolean nextKey(byte[] key, int length) {
+ return nextKey(key, 0, length);
+ }
+
+ public static boolean nextKey(byte[] key, int offset, int length) {
+ if (length == 0) {
+ return false;
+ }
+ int i = offset + length - 1;
+ while (key[i] == -1) {
+ key[i] = 0;
+ i--;
+ if (i < offset) {
+ // Change bytes back to the way they were
+ do {
+ key[++i] = -1;
+ } while (i < offset + length - 1);
+ return false;
+ }
+ }
+ key[i] = (byte)(key[i] + 1);
+ return true;
+ }
+
+ public static byte[] previousKey(byte[] key) {
+ byte[] previousKey = new byte[key.length];
+ System.arraycopy(key, 0, previousKey, 0, key.length);
+ if (!previousKey(previousKey, previousKey.length)) {
+ return null;
+ }
+ return previousKey;
+ }
+
+ public static boolean previousKey(byte[] key, int length) {
+ return previousKey(key, 0, length);
+ }
+
+ public static boolean previousKey(byte[] key, int offset, int length) {
+ if (length == 0) {
+ return false;
+ }
+ int i = offset + length - 1;
+ while (key[i] == 0) {
+ key[i] = -1;
+ i--;
+ if (i < offset) {
+ // Change bytes back to the way they were
+ do {
+ key[++i] = 0;
+ } while (i < offset + length - 1);
+ return false;
+ }
+ }
+ key[i] = (byte)(key[i] - 1);
+ return true;
+ }
+
+ /**
+ * Expand the key to length bytes using the fillByte to fill the
+ * bytes beyond the current key length.
+ */
+ public static byte[] fillKey(byte[] key, int length) {
+ if(key.length > length) {
+ throw new IllegalStateException();
+ }
+ if (key.length == length) {
+ return key;
+ }
+ byte[] newBound = new byte[length];
+ System.arraycopy(key, 0, newBound, 0, key.length);
+ return newBound;
+ }
+
+ /**
+ * Get the size in bytes of the UTF-8 encoded CharSequence
+ * @param sequence the CharSequence
+ */
+ public static int getSize(CharSequence sequence) {
+ int count = 0;
+ for (int i = 0, len = sequence.length(); i < len; i++) {
+ char ch = sequence.charAt(i);
+ if (ch <= 0x7F) {
+ count++;
+ } else if (ch <= 0x7FF) {
+ count += 2;
+ } else if (Character.isHighSurrogate(ch)) {
+ count += 4;
+ ++i;
+ } else {
+ count += 3;
+ }
+ }
+ return count;
+ }
+
+ public static boolean isInclusive(CompareOp op) {
+ switch (op) {
+ case LESS:
+ case GREATER:
+ return false;
+ case EQUAL:
+ case NOT_EQUAL:
+ case LESS_OR_EQUAL:
+ case GREATER_OR_EQUAL:
+ return true;
+ default:
+ throw new RuntimeException("Unknown Compare op " + op.name());
+ }
+ }
+ public static boolean compare(CompareOp op, int compareResult) {
+ switch (op) {
+ case LESS:
+ return compareResult < 0;
+ case LESS_OR_EQUAL:
+ return compareResult <= 0;
+ case EQUAL:
+ return compareResult == 0;
+ case NOT_EQUAL:
+ return compareResult != 0;
+ case GREATER_OR_EQUAL:
+ return compareResult >= 0;
+ case GREATER:
+ return compareResult > 0;
+ default:
+ throw new RuntimeException("Unknown Compare op " + op.name());
+ }
+ }
+
+ /**
+ * Given an ImmutableBytesWritable, returns the payload part of the argument as an byte array.
+ */
+ public static byte[] copyKeyBytesIfNecessary(ImmutableBytesWritable ptr) {
+ if (ptr.getOffset() == 0 && ptr.getLength() == ptr.get().length) {
+ return ptr.get();
+ }
+ return ptr.copyBytes();
+ }
+
+ public static KeyRange getKeyRange(byte[] key, CompareOp op, PDataType type) {
+ switch (op) {
+ case EQUAL:
+ return type.getKeyRange(key, true, key, true);
+ case GREATER:
+ return type.getKeyRange(key, false, KeyRange.UNBOUND, false);
+ case GREATER_OR_EQUAL:
+ return type.getKeyRange(key, true, KeyRange.UNBOUND, false);
+ case LESS:
+ return type.getKeyRange(KeyRange.UNBOUND, false, key, false);
+ case LESS_OR_EQUAL:
+ return type.getKeyRange(KeyRange.UNBOUND, false, key, true);
+ default:
+ throw new IllegalArgumentException("Unknown operator " + op);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/util/CSVLoader.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/CSVLoader.java b/phoenix-core/src/main/java/org/apache/phoenix/util/CSVLoader.java
new file mode 100644
index 0000000..508818d
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/CSVLoader.java
@@ -0,0 +1,250 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.util;
+
+import au.com.bytecode.opencsv.CSVReader;
+import com.google.common.collect.Maps;
+import org.apache.phoenix.exception.SQLExceptionCode;
+import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.PDataType;
+
+import java.io.FileReader;
+import java.sql.DatabaseMetaData;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/***
+ * Upserts CSV data using Phoenix JDBC connection
+ *
+ * @author mchohan
+ *
+ */
+public class CSVLoader {
+
+ private final PhoenixConnection conn;
+ private final String tableName;
+ private final List<String> columns;
+ private final boolean isStrict;
+ private final List<String> delimiter;
+ private final Map<String,Character> ctrlTable = new HashMap<String,Character>() {
+ { put("1",'\u0001');
+ put("2",'\u0002');
+ put("3",'\u0003');
+ put("4",'\u0004');
+ put("5",'\u0005');
+ put("6",'\u0006');
+ put("7",'\u0007');
+ put("8",'\u0008');
+ put("9",'\u0009');}};
+
+ private int unfoundColumnCount;
+
+ public CSVLoader(PhoenixConnection conn, String tableName, List<String> columns, boolean isStrict,List<String> delimiter) {
+ this.conn = conn;
+ this.tableName = tableName;
+ this.columns = columns;
+ this.isStrict = isStrict;
+ this.delimiter = delimiter;
+ }
+
+ public CSVLoader(PhoenixConnection conn, String tableName, List<String> columns, boolean isStrict) {
+ this(conn,tableName,columns,isStrict,null);
+ }
+
+
+ /**
+ * Upserts data from CSV file. Data is batched up based on connection batch
+ * size. Column PDataType is read from metadata and is used to convert
+ * column value to correct type before upsert. Note: Column Names are
+ * expected as first line of CSV file.
+ *
+ * @param fileName
+ * @throws Exception
+ */
+ public void upsert(String fileName) throws Exception {
+ List<String> delimiter = this.delimiter;
+ CSVReader reader;
+ if ((delimiter != null) && (delimiter.size() == 3)) {
+ reader = new CSVReader(new FileReader(fileName),
+ getCSVCustomField(this.delimiter.get(0)),
+ getCSVCustomField(this.delimiter.get(1)),
+ getCSVCustomField(this.delimiter.get(2)));
+ } else {
+ reader = new CSVReader(new FileReader(fileName));
+ }
+ upsert(reader);
+ }
+
+
+ public char getCSVCustomField(String field) {
+ if(this.ctrlTable.containsKey(field)) {
+ return this.ctrlTable.get(field);
+ } else {
+ return field.charAt(0);
+ }
+ }
+
+ /**
+ * Upserts data from CSV file. Data is batched up based on connection batch
+ * size. Column PDataType is read from metadata and is used to convert
+ * column value to correct type before upsert. Note: Column Names are
+ * expected as first line of CSV file.
+ *
+ * @param reader CSVReader instance
+ * @throws Exception
+ */
+ public void upsert(CSVReader reader) throws Exception {
+ List<String> columns = this.columns;
+ if (columns != null && columns.isEmpty()) {
+ columns = Arrays.asList(reader.readNext());
+ }
+ ColumnInfo[] columnInfo = generateColumnInfo(columns);
+ PreparedStatement stmt = null;
+ PreparedStatement[] stmtCache = null;
+ if (columns == null) {
+ stmtCache = new PreparedStatement[columnInfo.length];
+ } else {
+ String upsertStatement = QueryUtil.constructUpsertStatement(columnInfo, tableName, columnInfo.length - unfoundColumnCount);
+ stmt = conn.prepareStatement(upsertStatement);
+ }
+ String[] nextLine;
+ int rowCount = 0;
+ int upsertBatchSize = conn.getMutateBatchSize();
+ boolean wasAutoCommit = conn.getAutoCommit();
+ try {
+ conn.setAutoCommit(false);
+ Object upsertValue = null;
+ long start = System.currentTimeMillis();
+
+ // Upsert data based on SqlType of each column
+ while ((nextLine = reader.readNext()) != null) {
+ if (columns == null) {
+ stmt = stmtCache[nextLine.length-1];
+ if (stmt == null) {
+ String upsertStatement = QueryUtil.constructUpsertStatement(columnInfo, tableName, nextLine.length);
+ stmt = conn.prepareStatement(upsertStatement);
+ stmtCache[nextLine.length-1] = stmt;
+ }
+ }
+ for (int index = 0; index < columnInfo.length; index++) {
+ if (columnInfo[index] == null) {
+ continue;
+ }
+ String line = nextLine[index];
+ Integer info = columnInfo[index].getSqlType();
+ upsertValue = convertTypeSpecificValue(line, info);
+ if (upsertValue != null) {
+ stmt.setObject(index + 1, upsertValue, columnInfo[index].getSqlType());
+ } else {
+ stmt.setNull(index + 1, columnInfo[index].getSqlType());
+ }
+ }
+ stmt.execute();
+
+ // Commit when batch size is reached
+ if (++rowCount % upsertBatchSize == 0) {
+ conn.commit();
+ System.out.println("Rows upserted: " + rowCount);
+ }
+ }
+ conn.commit();
+ double elapsedDuration = ((System.currentTimeMillis() - start) / 1000.0);
+ System.out.println("CSV Upsert complete. " + rowCount + " rows upserted");
+ System.out.println("Time: " + elapsedDuration + " sec(s)\n");
+ } finally {
+ if(stmt != null) {
+ stmt.close();
+ }
+ if (wasAutoCommit) conn.setAutoCommit(true);
+ }
+ }
+
+ /**
+ * Gets CSV string input converted to correct type
+ */
+ private Object convertTypeSpecificValue(String s, Integer sqlType) throws Exception {
+ return PDataType.fromTypeId(sqlType).toObject(s);
+ }
+
+ /**
+ * Get array of ColumnInfos that contain Column Name and its associated
+ * PDataType
+ *
+ * @param columns
+ * @return
+ * @throws SQLException
+ */
+ private ColumnInfo[] generateColumnInfo(List<String> columns)
+ throws SQLException {
+ Map<String,Integer> columnNameToTypeMap = Maps.newLinkedHashMap();
+ DatabaseMetaData dbmd = conn.getMetaData();
+ // TODO: escape wildcard characters here because we don't want that behavior here
+ String escapedTableName = StringUtil.escapeLike(tableName);
+ String[] schemaAndTable = escapedTableName.split("\\.");
+ ResultSet rs = null;
+ try {
+ rs = dbmd.getColumns(null, (schemaAndTable.length == 1 ? "" : schemaAndTable[0]),
+ (schemaAndTable.length == 1 ? escapedTableName : schemaAndTable[1]),
+ null);
+ while (rs.next()) {
+ columnNameToTypeMap.put(rs.getString(QueryUtil.COLUMN_NAME_POSITION), rs.getInt(QueryUtil.DATA_TYPE_POSITION));
+ }
+ } finally {
+ if(rs != null) {
+ rs.close();
+ }
+ }
+ ColumnInfo[] columnType;
+ if (columns == null) {
+ int i = 0;
+ columnType = new ColumnInfo[columnNameToTypeMap.size()];
+ for (Map.Entry<String, Integer> entry : columnNameToTypeMap.entrySet()) {
+ columnType[i++] = new ColumnInfo(entry.getKey(),entry.getValue());
+ }
+ } else {
+ // Leave "null" as indication to skip b/c it doesn't exist
+ columnType = new ColumnInfo[columns.size()];
+ for (int i = 0; i < columns.size(); i++) {
+ String columnName = SchemaUtil.normalizeIdentifier(columns.get(i).trim());
+ Integer sqlType = columnNameToTypeMap.get(columnName);
+ if (sqlType == null) {
+ if (isStrict) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.COLUMN_NOT_FOUND)
+ .setColumnName(columnName).setTableName(tableName).build().buildException();
+ }
+ unfoundColumnCount++;
+ } else {
+ columnType[i] = new ColumnInfo(columnName, sqlType);
+ }
+ }
+ if (unfoundColumnCount == columns.size()) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.COLUMN_NOT_FOUND)
+ .setColumnName(Arrays.toString(columns.toArray(new String[0]))).setTableName(tableName).build().buildException();
+ }
+ }
+ return columnType;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/util/Closeables.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/Closeables.java b/phoenix-core/src/main/java/org/apache/phoenix/util/Closeables.java
new file mode 100644
index 0000000..b09ebe0
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/Closeables.java
@@ -0,0 +1,124 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.util;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.*;
+
+import com.google.common.collect.Iterables;
+
+
+/**
+ * Utilities for operating on {@link Closeable}s.
+ *
+ */
+public class Closeables {
+ /** Not constructed */
+ private Closeables() { }
+
+ /**
+ * Allows you to close as many of the {@link Closeable}s as possible.
+ *
+ * If any of the close's fail with an IOException, those exception(s) will
+ * be thrown after attempting to close all of the inputs.
+ */
+ public static void closeAll(Iterable<? extends Closeable> iterable) throws IOException {
+ IOException ex = closeAllQuietly(iterable);
+ if (ex != null) throw ex;
+ }
+
+ public static IOException closeAllQuietly(Iterable<? extends Closeable> iterable) {
+ if (iterable == null) return null;
+
+ LinkedList<IOException> exceptions = null;
+ for (Closeable closeable : iterable) {
+ try {
+ closeable.close();
+ } catch (IOException x) {
+ if (exceptions == null) exceptions = new LinkedList<IOException>();
+ exceptions.add(x);
+ }
+ }
+
+ IOException ex = MultipleCausesIOException.fromIOExceptions(exceptions);
+ return ex;
+ }
+
+ static private class MultipleCausesIOException extends IOException {
+ private static final long serialVersionUID = 1L;
+
+ static IOException fromIOExceptions(Collection<? extends IOException> exceptions) {
+ if (exceptions == null || exceptions.isEmpty()) return null;
+ if (exceptions.size() == 1) return Iterables.getOnlyElement(exceptions);
+
+ return new MultipleCausesIOException(exceptions);
+ }
+
+ private final Collection<? extends IOException> exceptions;
+ private boolean hasSetStackTrace;
+
+ /**
+ * Use the {@link #fromIOExceptions(Collection) factory}.
+ */
+ private MultipleCausesIOException(Collection<? extends IOException> exceptions) {
+ this.exceptions = exceptions;
+ }
+
+ @Override
+ public String getMessage() {
+ StringBuilder sb = new StringBuilder(this.exceptions.size() * 50);
+ int exceptionNum = 0;
+ for (IOException ex : this.exceptions) {
+ sb.append("Cause Number " + exceptionNum + ": " + ex.getMessage() + "\n");
+ exceptionNum++;
+ }
+ return sb.toString();
+ }
+
+ @Override
+ public StackTraceElement[] getStackTrace() {
+ if (!this.hasSetStackTrace) {
+ ArrayList<StackTraceElement> frames = new ArrayList<StackTraceElement>(this.exceptions.size() * 20);
+
+ int exceptionNum = 0;
+ for (IOException exception : this.exceptions) {
+ StackTraceElement header = new StackTraceElement(MultipleCausesIOException.class.getName(),
+ "Exception Number " + exceptionNum,
+ "<no file>",
+ 0);
+
+ frames.add(header);
+ for (StackTraceElement ste : exception.getStackTrace()) {
+ frames.add(ste);
+ }
+ exceptionNum++;
+ }
+
+ setStackTrace(frames.toArray(new StackTraceElement[frames.size()]));
+ this.hasSetStackTrace = true;
+ }
+
+ return super.getStackTrace();
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java
new file mode 100644
index 0000000..9ca05e7
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java
@@ -0,0 +1,22 @@
+package org.apache.phoenix.util;
+
+/**
+ * ColumnInfo used to store Column Name and its associated PDataType
+ */
+public class ColumnInfo {
+ private String columnName;
+ private Integer sqlType;
+
+ public ColumnInfo(String columnName, Integer sqlType) {
+ this.columnName = columnName;
+ this.sqlType = sqlType;
+ }
+
+ public String getColumnName() {
+ return columnName;
+ }
+
+ public Integer getSqlType() {
+ return sqlType;
+ }
+}