You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ma...@apache.org on 2016/03/29 21:50:34 UTC
[01/30] phoenix git commit: PHOENIX-2718 Prevent stats creation for
stats table
Repository: phoenix
Updated Branches:
refs/heads/calcite 86b9deddf -> 6ec421c12
PHOENIX-2718 Prevent stats creation for stats table
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/4f743230
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/4f743230
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/4f743230
Branch: refs/heads/calcite
Commit: 4f743230bcf91630cff0ae1bb0e86bc9680787ce
Parents: 29ca38b
Author: James Taylor <jt...@salesforce.com>
Authored: Sat Feb 27 00:15:22 2016 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Sat Feb 27 00:15:22 2016 -0800
----------------------------------------------------------------------
.../phoenix/end2end/PhoenixRuntimeIT.java | 4 +-
.../phoenix/end2end/index/LocalIndexIT.java | 2 +-
.../phoenix/end2end/index/ViewIndexIT.java | 2 +-
.../ipc/controller/MetadataRpcController.java | 2 +-
.../coprocessor/SequenceRegionObserver.java | 12 ++---
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 17 ++++----
.../query/ConnectionQueryServicesImpl.java | 20 ++++-----
.../org/apache/phoenix/schema/Sequence.java | 46 ++++++++++----------
.../stats/StatisticsCollectorFactory.java | 3 +-
.../org/apache/phoenix/util/MetaDataUtil.java | 2 +-
.../org/apache/phoenix/util/SchemaUtil.java | 4 +-
.../org/apache/phoenix/util/UpgradeUtil.java | 18 ++++----
.../java/org/apache/phoenix/query/BaseTest.java | 2 +-
.../apache/phoenix/util/PhoenixRuntimeTest.java | 2 +-
14 files changed, 69 insertions(+), 67 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixRuntimeIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixRuntimeIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixRuntimeIT.java
index 234ea1c..e748825 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixRuntimeIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixRuntimeIT.java
@@ -107,8 +107,8 @@ public class PhoenixRuntimeIT extends BaseHBaseManagedTimeIT {
props.setProperty(PhoenixRuntime.TENANT_ID_ATTRIB, "t1");
Connection tsconn = DriverManager.getConnection(getUrl(), props);
tsconn.createStatement().execute("CREATE SEQUENCE s1");
- Expression e1 = PhoenixRuntime.getTenantIdExpression(tsconn, PhoenixDatabaseMetaData.SEQUENCE_FULLNAME);
- HTableInterface htable1 = tsconn.unwrap(PhoenixConnection.class).getQueryServices().getTable(PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_BYTES);
+ Expression e1 = PhoenixRuntime.getTenantIdExpression(tsconn, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME);
+ HTableInterface htable1 = tsconn.unwrap(PhoenixConnection.class).getQueryServices().getTable(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES);
assertTenantIds(e1, htable1, new FirstKeyOnlyFilter(), new String[] {"", "t1"} );
tsconn.createStatement().execute("CREATE VIEW A.BAR(V1 VARCHAR) AS SELECT * FROM " + tableName);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/LocalIndexIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/LocalIndexIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/LocalIndexIT.java
index a79d3d1..d0d56b8 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/LocalIndexIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/LocalIndexIT.java
@@ -184,7 +184,7 @@ public class LocalIndexIT extends BaseHBaseManagedTimeIT {
ResultSet rs = conn2.createStatement().executeQuery("SELECT "
+ PhoenixDatabaseMetaData.SEQUENCE_SCHEMA + ","
+ PhoenixDatabaseMetaData.SEQUENCE_NAME
- + " FROM " + PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_ESCAPED);
+ + " FROM " + PhoenixDatabaseMetaData.SYSTEM_SEQUENCE);
assertFalse("View index sequences should be deleted.", rs.next());
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/ViewIndexIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/ViewIndexIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/ViewIndexIT.java
index fe8ea77..79b776c 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/ViewIndexIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/ViewIndexIT.java
@@ -83,7 +83,7 @@ public class ViewIndexIT extends BaseHBaseManagedTimeIT {
ResultSet rs = conn2.createStatement().executeQuery("SELECT "
+ PhoenixDatabaseMetaData.SEQUENCE_SCHEMA + ","
+ PhoenixDatabaseMetaData.SEQUENCE_NAME
- + " FROM " + PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_ESCAPED);
+ + " FROM " + PhoenixDatabaseMetaData.SYSTEM_SEQUENCE);
assertFalse("View index sequences should be deleted.", rs.next());
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/main/java/org/apache/hadoop/hbase/ipc/controller/MetadataRpcController.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/hadoop/hbase/ipc/controller/MetadataRpcController.java b/phoenix-core/src/main/java/org/apache/hadoop/hbase/ipc/controller/MetadataRpcController.java
index 2d51d08..ab463e9 100644
--- a/phoenix-core/src/main/java/org/apache/hadoop/hbase/ipc/controller/MetadataRpcController.java
+++ b/phoenix-core/src/main/java/org/apache/hadoop/hbase/ipc/controller/MetadataRpcController.java
@@ -35,7 +35,7 @@ class MetadataRpcController extends DelegatingPayloadCarryingRpcController {
private static final List<String> SYSTEM_TABLE_NAMES = new ImmutableList.Builder<String>()
.add(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME)
.add(PhoenixDatabaseMetaData.SYSTEM_STATS_NAME)
- .add(PhoenixDatabaseMetaData.SEQUENCE_FULLNAME)
+ .add(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME)
.add(PhoenixDatabaseMetaData.SYSTEM_FUNCTION_NAME).build();
public MetadataRpcController(PayloadCarryingRpcController delegate,
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
index bb6c4b5..49962e0 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
@@ -85,7 +85,7 @@ public class SequenceRegionObserver extends BaseRegionObserver {
PInteger.INSTANCE.getCodec().encodeInt(errorCode, errorCodeBuf, 0);
return Result.create(Collections.singletonList(
(Cell)KeyValueUtil.newKeyValue(row,
- PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES,
+ PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES,
QueryConstants.EMPTY_COLUMN_BYTES, timestamp, errorCodeBuf)));
}
@@ -154,7 +154,7 @@ public class SequenceRegionObserver extends BaseRegionObserver {
long timestamp = currentValueKV.getTimestamp();
Put put = new Put(row, timestamp);
- int numIncrementKVs = increment.getFamilyCellMap().get(PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES).size();
+ int numIncrementKVs = increment.getFamilyCellMap().get(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES).size();
// creates the list of KeyValues used for the Result that will be returned
List<Cell> cells = Sequence.getCells(result, numIncrementKVs);
@@ -298,7 +298,7 @@ public class SequenceRegionObserver extends BaseRegionObserver {
KeyValue createKeyValue(byte[] key, byte[] cqBytes, long value, long timestamp) {
byte[] valueBuffer = new byte[PLong.INSTANCE.getByteSize()];
PLong.INSTANCE.getCodec().encodeLong(value, valueBuffer, 0);
- return KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, cqBytes, timestamp, valueBuffer);
+ return KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, cqBytes, timestamp, valueBuffer);
}
/**
@@ -312,7 +312,7 @@ public class SequenceRegionObserver extends BaseRegionObserver {
*/
private KeyValue createKeyValue(byte[] key, byte[] cqBytes, boolean value, long timestamp) throws IOException {
// create new key value for put
- return KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, cqBytes,
+ return KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, cqBytes,
timestamp, value ? PDataType.TRUE_BYTES : PDataType.FALSE_BYTES);
}
@@ -399,7 +399,7 @@ public class SequenceRegionObserver extends BaseRegionObserver {
// Timestamp should match exactly, or we may have the wrong sequence
if (expectedValue != value || currentValueKV.getTimestamp() != clientTimestamp) {
return Result.create(Collections.singletonList(
- (Cell)KeyValueUtil.newKeyValue(row, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES,
+ (Cell)KeyValueUtil.newKeyValue(row, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES,
QueryConstants.EMPTY_COLUMN_BYTES, currentValueKV.getTimestamp(), ByteUtil.EMPTY_BYTE_ARRAY)));
}
m = new Put(row, currentValueKV.getTimestamp());
@@ -427,7 +427,7 @@ public class SequenceRegionObserver extends BaseRegionObserver {
// the client cares about is the timestamp, which is the timestamp of
// when the mutation was actually performed (useful in the case of .
return Result.create(Collections.singletonList(
- (Cell)KeyValueUtil.newKeyValue(row, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, serverTimestamp, SUCCESS_VALUE)));
+ (Cell)KeyValueUtil.newKeyValue(row, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, serverTimestamp, SUCCESS_VALUE)));
} finally {
region.releaseRowLocks(locks);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index fabd949..26d57fa 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -104,6 +104,15 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
public static final byte[] SYSTEM_STATS_TABLE_BYTES = Bytes.toBytes(SYSTEM_STATS_TABLE);
public static final String SYSTEM_CATALOG_ALIAS = "\"SYSTEM.TABLE\"";
+ public static final byte[] SYSTEM_SEQUENCE_FAMILY_BYTES = QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES;
+ public static final String SYSTEM_SEQUENCE_SCHEMA = SYSTEM_CATALOG_SCHEMA;
+ public static final byte[] SYSTEM_SEQUENCE_SCHEMA_BYTES = Bytes.toBytes(SYSTEM_SEQUENCE_SCHEMA);
+ public static final String SYSTEM_SEQUENCE_TABLE = "SEQUENCE";
+ public static final byte[] SYSTEM_SEQUENCE_TABLE_BYTES = Bytes.toBytes(SYSTEM_SEQUENCE_TABLE);
+ public static final String SYSTEM_SEQUENCE = SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_SEQUENCE_TABLE + "\"";
+ public static final String SYSTEM_SEQUENCE_NAME = SchemaUtil.getTableName(SYSTEM_SEQUENCE_SCHEMA, SYSTEM_SEQUENCE_TABLE);
+ public static final byte[] SYSTEM_SEQUENCE_NAME_BYTES = Bytes.toBytes(SYSTEM_SEQUENCE_NAME);
+
public static final String TABLE_NAME = "TABLE_NAME";
public static final byte[] TABLE_NAME_BYTES = Bytes.toBytes(TABLE_NAME);
public static final String TABLE_TYPE = "TABLE_TYPE";
@@ -223,14 +232,6 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
public static final String NUM_ARGS = "NUM_ARGS";
public static final byte[] NUM_ARGS_BYTES = Bytes.toBytes(NUM_ARGS);
- public static final byte[] SEQUENCE_FAMILY_BYTES = QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES;
- public static final String SEQUENCE_SCHEMA_NAME = SYSTEM_CATALOG_SCHEMA;
- public static final byte[] SEQUENCE_SCHEMA_NAME_BYTES = Bytes.toBytes(SEQUENCE_SCHEMA_NAME);
- public static final String SEQUENCE_TABLE_NAME = TYPE_SEQUENCE;
- public static final byte[] SEQUENCE_TABLE_NAME_BYTES = Bytes.toBytes(SEQUENCE_TABLE_NAME);
- public static final String SEQUENCE_FULLNAME_ESCAPED = SYSTEM_CATALOG_SCHEMA + ".\"" + TYPE_SEQUENCE + "\"";
- public static final String SEQUENCE_FULLNAME = SchemaUtil.getTableName(SEQUENCE_SCHEMA_NAME, SEQUENCE_TABLE_NAME);
- public static final byte[] SEQUENCE_FULLNAME_BYTES = Bytes.toBytes(SEQUENCE_FULLNAME);
public static final String SEQUENCE_SCHEMA = "SEQUENCE_SCHEMA";
public static final String SEQUENCE_NAME = "SEQUENCE_NAME";
public static final String CURRENT_VALUE = "CURRENT_VALUE";
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index 3327861..39118a3 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@ -2417,14 +2417,14 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_2_1) {
if (UpgradeUtil.upgradeSequenceTable(metaConnection, nSaltBuckets, e.getTable())) {
metaConnection.removeTable(null,
- PhoenixDatabaseMetaData.SEQUENCE_SCHEMA_NAME,
- PhoenixDatabaseMetaData.SEQUENCE_TABLE_NAME,
+ PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_SCHEMA,
+ PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_TABLE,
MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP);
clearTableFromCache(ByteUtil.EMPTY_BYTE_ARRAY,
- PhoenixDatabaseMetaData.SEQUENCE_SCHEMA_NAME_BYTES,
- PhoenixDatabaseMetaData.SEQUENCE_TABLE_NAME_BYTES,
+ PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_SCHEMA_BYTES,
+ PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_TABLE_BYTES,
MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP);
- clearTableRegionCache(PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_BYTES);
+ clearTableRegionCache(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES);
}
nSequenceSaltBuckets = nSaltBuckets;
} else {
@@ -2773,7 +2773,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
// Now that we have the lock we need, create the sequence
Append append = sequence.createSequence(startWith, incrementBy, cacheSize, timestamp, minValue, maxValue, cycle);
HTableInterface htable =
- this.getTable(PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_BYTES);
+ this.getTable(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES);
htable.setAutoFlush(true);
try {
Result result = htable.append(append);
@@ -2800,7 +2800,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
sequence.getLock().lock();
// Now that we have the lock we need, create the sequence
Append append = sequence.dropSequence(timestamp);
- HTableInterface htable = this.getTable(PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_BYTES);
+ HTableInterface htable = this.getTable(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES);
try {
Result result = htable.append(append);
return sequence.dropSequence(result);
@@ -2896,7 +2896,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
if (toIncrementList.isEmpty()) {
return;
}
- HTableInterface hTable = this.getTable(PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_BYTES);
+ HTableInterface hTable = this.getTable(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES);
Object[] resultObjects = null;
SQLException sqlE = null;
try {
@@ -3019,7 +3019,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
if (toReturnList.isEmpty()) {
return;
}
- HTableInterface hTable = this.getTable(PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_BYTES);
+ HTableInterface hTable = this.getTable(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES);
Object[] resultObjects = null;
SQLException sqlE = null;
try {
@@ -3072,7 +3072,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
if (mutations.isEmpty()) {
return;
}
- HTableInterface hTable = this.getTable(PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_BYTES);
+ HTableInterface hTable = this.getTable(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES);
SQLException sqlE = null;
try {
hTable.batch(mutations);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java
index a2041f7..9c1a74b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java
@@ -25,7 +25,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INCREMENT_BY_BYTES
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MAX_VALUE_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MIN_VALUE_BYTES;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES;
import java.io.IOException;
import java.sql.SQLException;
@@ -66,13 +66,13 @@ public class Sequence {
public enum MetaOp {CREATE_SEQUENCE, DROP_SEQUENCE, RETURN_SEQUENCE};
// create empty Sequence key values used while created a sequence row
- private static final KeyValue CURRENT_VALUE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, CURRENT_VALUE_BYTES);
- private static final KeyValue INCREMENT_BY_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, INCREMENT_BY_BYTES);
- private static final KeyValue CACHE_SIZE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, CACHE_SIZE_BYTES);
- private static final KeyValue MIN_VALUE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, MIN_VALUE_BYTES);
- private static final KeyValue MAX_VALUE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, MAX_VALUE_BYTES);
- private static final KeyValue CYCLE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, CYCLE_FLAG_BYTES);
- private static final KeyValue LIMIT_REACHED_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, LIMIT_REACHED_FLAG_BYTES);
+ private static final KeyValue CURRENT_VALUE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, CURRENT_VALUE_BYTES);
+ private static final KeyValue INCREMENT_BY_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, INCREMENT_BY_BYTES);
+ private static final KeyValue CACHE_SIZE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, CACHE_SIZE_BYTES);
+ private static final KeyValue MIN_VALUE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, MIN_VALUE_BYTES);
+ private static final KeyValue MAX_VALUE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, MAX_VALUE_BYTES);
+ private static final KeyValue CYCLE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, CYCLE_FLAG_BYTES);
+ private static final KeyValue LIMIT_REACHED_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, LIMIT_REACHED_FLAG_BYTES);
private static final List<KeyValue> SEQUENCE_KV_COLUMNS = Arrays.<KeyValue>asList(
CURRENT_VALUE_KV,
INCREMENT_BY_KV,
@@ -300,9 +300,9 @@ public class Sequence {
append.setAttribute(SequenceRegionObserver.OPERATION_ATTRIB, opBuf);
append.setAttribute(SequenceRegionObserver.CURRENT_VALUE_ATTRIB, PLong.INSTANCE.toBytes(value.nextValue));
Map<byte[], List<Cell>> familyMap = append.getFamilyCellMap();
- familyMap.put(PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, Arrays.<Cell>asList(
- (Cell)KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CURRENT_VALUE_BYTES, value.timestamp, PLong.INSTANCE.toBytes(value.currentValue)),
- (Cell)KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG_BYTES, value.timestamp, PBoolean.INSTANCE.toBytes(value.limitReached))
+ familyMap.put(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, Arrays.<Cell>asList(
+ (Cell)KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CURRENT_VALUE_BYTES, value.timestamp, PLong.INSTANCE.toBytes(value.currentValue)),
+ (Cell)KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG_BYTES, value.timestamp, PBoolean.INSTANCE.toBytes(value.limitReached))
));
return append;
}
@@ -550,16 +550,16 @@ public class Sequence {
}
Map<byte[], List<Cell>> familyMap = append.getFamilyCellMap();
byte[] startWithBuf = PLong.INSTANCE.toBytes(startWith);
- familyMap.put(PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, Arrays.<Cell>asList(
- KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, timestamp, ByteUtil.EMPTY_BYTE_ARRAY),
- KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CURRENT_VALUE_BYTES, timestamp, startWithBuf),
- KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.START_WITH_BYTES, timestamp, startWithBuf),
- KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.INCREMENT_BY_BYTES, timestamp, PLong.INSTANCE.toBytes(incrementBy)),
- KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CACHE_SIZE_BYTES, timestamp, PLong.INSTANCE.toBytes(cacheSize)),
- KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.MIN_VALUE_BYTES, timestamp, PLong.INSTANCE.toBytes(minValue)),
- KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.MAX_VALUE_BYTES, timestamp, PLong.INSTANCE.toBytes(maxValue)),
- KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CYCLE_FLAG_BYTES, timestamp, PBoolean.INSTANCE.toBytes(cycle)),
- KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG_BYTES, timestamp, PDataType.FALSE_BYTES)
+ familyMap.put(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, Arrays.<Cell>asList(
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, timestamp, ByteUtil.EMPTY_BYTE_ARRAY),
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CURRENT_VALUE_BYTES, timestamp, startWithBuf),
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.START_WITH_BYTES, timestamp, startWithBuf),
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.INCREMENT_BY_BYTES, timestamp, PLong.INSTANCE.toBytes(incrementBy)),
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CACHE_SIZE_BYTES, timestamp, PLong.INSTANCE.toBytes(cacheSize)),
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.MIN_VALUE_BYTES, timestamp, PLong.INSTANCE.toBytes(minValue)),
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.MAX_VALUE_BYTES, timestamp, PLong.INSTANCE.toBytes(maxValue)),
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CYCLE_FLAG_BYTES, timestamp, PBoolean.INSTANCE.toBytes(cycle)),
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG_BYTES, timestamp, PDataType.FALSE_BYTES)
));
return append;
}
@@ -588,8 +588,8 @@ public class Sequence {
append.setAttribute(SequenceRegionObserver.MAX_TIMERANGE_ATTRIB, Bytes.toBytes(timestamp));
}
Map<byte[], List<Cell>> familyMap = append.getFamilyCellMap();
- familyMap.put(PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, Arrays.<Cell>asList(
- (Cell)KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, timestamp, ByteUtil.EMPTY_BYTE_ARRAY)));
+ familyMap.put(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, Arrays.<Cell>asList(
+ (Cell)KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, timestamp, ByteUtil.EMPTY_BYTE_ARRAY)));
return append;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectorFactory.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectorFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectorFactory.java
index 964bb28..dcffa41 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectorFactory.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectorFactory.java
@@ -61,7 +61,8 @@ public class StatisticsCollectorFactory {
static {
DISABLE_STATS.add(TableName.valueOf(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME));
DISABLE_STATS.add(TableName.valueOf(PhoenixDatabaseMetaData.SYSTEM_FUNCTION_NAME));
- DISABLE_STATS.add(TableName.valueOf(PhoenixDatabaseMetaData.SEQUENCE_FULLNAME));
+ DISABLE_STATS.add(TableName.valueOf(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME));
+ DISABLE_STATS.add(TableName.valueOf(PhoenixDatabaseMetaData.SYSTEM_STATS_NAME));
}
/**
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
index 6d5903d..5127f5b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
@@ -375,7 +375,7 @@ public class MetaDataUtil {
public static void deleteViewIndexSequences(PhoenixConnection connection, PName name) throws SQLException {
String schemaName = getViewIndexSchemaName(name);
- connection.createStatement().executeUpdate("DELETE FROM " + PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_ESCAPED +
+ connection.createStatement().executeUpdate("DELETE FROM " + PhoenixDatabaseMetaData.SYSTEM_SEQUENCE +
" WHERE " + PhoenixDatabaseMetaData.TENANT_ID + " IS NULL AND " +
PhoenixDatabaseMetaData.SEQUENCE_SCHEMA + " = '" + schemaName + "'");
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
index de92046..523f684 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
@@ -426,11 +426,11 @@ public class SchemaUtil {
}
public static boolean isSequenceTable(byte[] tableName) {
- return Bytes.compareTo(tableName, PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_BYTES) == 0;
+ return Bytes.compareTo(tableName, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES) == 0;
}
public static boolean isSequenceTable(PTable table) {
- return PhoenixDatabaseMetaData.SEQUENCE_FULLNAME.equals(table.getName().getString());
+ return PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME.equals(table.getName().getString());
}
public static boolean isMetaTable(PTable table) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java
index ac4ff7a..ba33a2c 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java
@@ -128,17 +128,17 @@ public class UpgradeUtil {
}
private static byte[] getSequenceSnapshotName() {
- return Bytes.toBytes("_BAK_" + PhoenixDatabaseMetaData.SEQUENCE_FULLNAME);
+ return Bytes.toBytes("_BAK_" + PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME);
}
private static void createSequenceSnapshot(HBaseAdmin admin, PhoenixConnection conn) throws SQLException {
byte[] tableName = getSequenceSnapshotName();
- HColumnDescriptor columnDesc = new HColumnDescriptor(PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES);
+ HColumnDescriptor columnDesc = new HColumnDescriptor(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES);
HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
desc.addFamily(columnDesc);
try {
admin.createTable(desc);
- copyTable(conn, PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_BYTES, tableName);
+ copyTable(conn, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES, tableName);
} catch (IOException e) {
throw ServerUtil.parseServerException(e);
}
@@ -146,7 +146,7 @@ public class UpgradeUtil {
private static void restoreSequenceSnapshot(HBaseAdmin admin, PhoenixConnection conn) throws SQLException {
byte[] tableName = getSequenceSnapshotName();
- copyTable(conn, tableName, PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_BYTES);
+ copyTable(conn, tableName, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES);
}
private static void deleteSequenceSnapshot(HBaseAdmin admin) throws SQLException {
@@ -237,11 +237,11 @@ public class UpgradeUtil {
return;
}
logger.warn("Pre-splitting SYSTEM.SEQUENCE table " + nSaltBuckets + "-ways. This may take some time - please do not close window.");
- HTableDescriptor desc = admin.getTableDescriptor(PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_BYTES);
+ HTableDescriptor desc = admin.getTableDescriptor(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES);
createSequenceSnapshot(admin, conn);
snapshotCreated = true;
- admin.disableTable(PhoenixDatabaseMetaData.SEQUENCE_FULLNAME);
- admin.deleteTable(PhoenixDatabaseMetaData.SEQUENCE_FULLNAME);
+ admin.disableTable(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME);
+ admin.deleteTable(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME);
byte[][] splitPoints = SaltingUtil.getSalteByteSplitPoints(nSaltBuckets);
admin.createTable(desc, splitPoints);
restoreSequenceSnapshot(admin, conn);
@@ -272,7 +272,7 @@ public class UpgradeUtil {
public static boolean upgradeSequenceTable(PhoenixConnection conn, int nSaltBuckets, PTable oldTable) throws SQLException {
logger.info("Upgrading SYSTEM.SEQUENCE table");
- byte[] seqTableKey = SchemaUtil.getTableKey(null, PhoenixDatabaseMetaData.SEQUENCE_SCHEMA_NAME, PhoenixDatabaseMetaData.SEQUENCE_TABLE_NAME);
+ byte[] seqTableKey = SchemaUtil.getTableKey(null, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_SCHEMA, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_TABLE);
HTableInterface sysTable = conn.getQueryServices().getTable(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES);
try {
logger.info("Setting SALT_BUCKETS property of SYSTEM.SEQUENCE to " + SaltingUtil.MAX_BUCKET_NUM);
@@ -326,7 +326,7 @@ public class UpgradeUtil {
Scan scan = new Scan();
scan.setRaw(true);
scan.setMaxVersions(MetaDataProtocol.DEFAULT_MAX_META_DATA_VERSIONS);
- HTableInterface seqTable = conn.getQueryServices().getTable(PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_BYTES);
+ HTableInterface seqTable = conn.getQueryServices().getTable(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES);
try {
boolean committed = false;
logger.info("Adding salt byte to all SYSTEM.SEQUENCE rows");
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java b/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
index 1192059..8b9b74f 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
@@ -943,7 +943,7 @@ public abstract class BaseTest {
+ PhoenixDatabaseMetaData.TENANT_ID + ","
+ PhoenixDatabaseMetaData.SEQUENCE_SCHEMA + ","
+ PhoenixDatabaseMetaData.SEQUENCE_NAME
- + " FROM " + PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_ESCAPED);
+ + " FROM " + PhoenixDatabaseMetaData.SYSTEM_SEQUENCE);
String lastTenantId = null;
Connection conn = globalConn;
while (rs.next()) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4f743230/phoenix-core/src/test/java/org/apache/phoenix/util/PhoenixRuntimeTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/util/PhoenixRuntimeTest.java b/phoenix-core/src/test/java/org/apache/phoenix/util/PhoenixRuntimeTest.java
index b6465c3..895e29e 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/util/PhoenixRuntimeTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/util/PhoenixRuntimeTest.java
@@ -171,7 +171,7 @@ public class PhoenixRuntimeTest extends BaseConnectionlessQueryTest {
Expression e2 = PhoenixRuntime.getTenantIdExpression(conn, PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME);
assertNotNull(e2);
- Expression e3 = PhoenixRuntime.getTenantIdExpression(conn, PhoenixDatabaseMetaData.SEQUENCE_FULLNAME);
+ Expression e3 = PhoenixRuntime.getTenantIdExpression(conn, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME);
assertNotNull(e3);
conn.createStatement().execute("CREATE TABLE FOO (k VARCHAR PRIMARY KEY)");
[06/30] phoenix git commit: PHOENIX-2731 Ensure upgrade from 4.7.0
RCx to 4.7.0 RC5 works
Posted by ma...@apache.org.
PHOENIX-2731 Ensure upgrade from 4.7.0 RCx to 4.7.0 RC5 works
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/f9c8fe8d
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/f9c8fe8d
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/f9c8fe8d
Branch: refs/heads/calcite
Commit: f9c8fe8df68bfd3b60b2a9d4ab83ff74b1fbd3de
Parents: 1c06c5d
Author: James Taylor <jt...@salesforce.com>
Authored: Wed Mar 2 10:28:34 2016 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Wed Mar 2 10:40:30 2016 -0800
----------------------------------------------------------------------
.../java/org/apache/phoenix/coprocessor/MetaDataProtocol.java | 2 +-
.../org/apache/phoenix/query/ConnectionQueryServicesImpl.java | 7 +++++--
2 files changed, 6 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f9c8fe8d/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java
index fb9d228..7a7cf31 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java
@@ -73,7 +73,7 @@ public abstract class MetaDataProtocol extends MetaDataService {
public static final long MIN_SYSTEM_TABLE_TIMESTAMP_4_3_0 = MIN_TABLE_TIMESTAMP + 7;
public static final long MIN_SYSTEM_TABLE_TIMESTAMP_4_5_0 = MIN_TABLE_TIMESTAMP + 8;
public static final long MIN_SYSTEM_TABLE_TIMESTAMP_4_6_0 = MIN_TABLE_TIMESTAMP + 9;
- public static final long MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0 = MIN_TABLE_TIMESTAMP + 14;
+ public static final long MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0 = MIN_TABLE_TIMESTAMP + 15;
// MIN_SYSTEM_TABLE_TIMESTAMP needs to be set to the max of all the MIN_SYSTEM_TABLE_TIMESTAMP_* constants
public static final long MIN_SYSTEM_TABLE_TIMESTAMP = MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0;
// TODO: pare this down to minimum, as we don't need duplicates for both table and column errors, nor should we need
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f9c8fe8d/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index 39118a3..a473cde 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@ -2436,13 +2436,16 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
metaConnection.createStatement().executeUpdate(
QueryConstants.CREATE_STATS_TABLE_METADATA);
} catch (NewerTableAlreadyExistsException ignore) {
- } catch(TableAlreadyExistsException ignore) {
- metaConnection = addColumnsIfNotExists(
+ } catch(TableAlreadyExistsException e) {
+ long currentServerSideTableTimeStamp = e.getTable().getTimeStamp();
+ if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_3_0) {
+ metaConnection = addColumnsIfNotExists(
metaConnection,
PhoenixDatabaseMetaData.SYSTEM_STATS_NAME,
MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP,
PhoenixDatabaseMetaData.GUIDE_POSTS_ROW_COUNT + " "
+ PLong.INSTANCE.getSqlTypeName());
+ }
}
try {
metaConnection.createStatement().executeUpdate(
[28/30] phoenix git commit: PHOENIX-2803 Initializing ToCharFunction
with the List constructor fails
Posted by ma...@apache.org.
PHOENIX-2803 Initializing ToCharFunction with the List<Expression> constructor fails
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/b9880503
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/b9880503
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/b9880503
Branch: refs/heads/calcite
Commit: b98805039516bfa706f1fa78ed0850d802df5dc8
Parents: d07fed5
Author: Thomas D'Silva <td...@salesforce.com>
Authored: Mon Mar 28 12:01:12 2016 -0700
Committer: Thomas D'Silva <td...@salesforce.com>
Committed: Mon Mar 28 14:23:51 2016 -0700
----------------------------------------------------------------------
.../apache/phoenix/end2end/ToCharFunctionIT.java | 15 +++++++++++++++
.../apache/phoenix/end2end/ToDateFunctionIT.java | 15 +++++++++++++++
.../expression/function/ToCharFunction.java | 13 +++++++++----
.../expression/function/ToDateFunction.java | 17 +++++++++++------
.../expression/function/ToNumberFunction.java | 14 ++++++++++----
.../expression/function/ToTimeFunction.java | 13 +++++++++----
.../expression/function/ToTimestampFunction.java | 13 +++++++++----
.../function/BuiltinFunctionConstructorTest.java | 11 ++++++++---
8 files changed, 86 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b9880503/phoenix-core/src/it/java/org/apache/phoenix/end2end/ToCharFunctionIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ToCharFunctionIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ToCharFunctionIT.java
index 52b5207..1cccf07 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ToCharFunctionIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ToCharFunctionIT.java
@@ -28,6 +28,7 @@ import java.sql.Date;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
+import java.sql.SQLException;
import java.sql.Time;
import java.sql.Timestamp;
import java.text.DateFormat;
@@ -240,4 +241,18 @@ public class ToCharFunctionIT extends BaseClientManagedTimeIT {
result.setTimeZone(TimeZone.getTimeZone("GMT"));
return result;
}
+
+ @Test
+ public void testToCharWithCloneMethod() throws SQLException {
+ Connection conn = DriverManager.getConnection(getUrl());
+ String ddl = "create table t (k varchar primary key, v integer[])";
+ conn.createStatement().execute(ddl);
+ conn.createStatement().execute("UPSERT INTO T VALUES('x',ARRAY[1234])");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery("select to_char(v[1],'000') from t");
+ assertTrue(rs.next());
+ assertEquals("Unexpected value for date ", String.valueOf(1234), rs.getString(1));
+ assertFalse(rs.next());
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b9880503/phoenix-core/src/it/java/org/apache/phoenix/end2end/ToDateFunctionIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ToDateFunctionIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ToDateFunctionIT.java
index dbe47ca..acb29df 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ToDateFunctionIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ToDateFunctionIT.java
@@ -251,4 +251,19 @@ public class ToDateFunctionIT extends BaseHBaseManagedTimeIT {
assertEquals("Did not get value that was inserted!!", dateString2, obtainedString);
assertFalse("No more rows expected!!", rs.next());
}
+
+ @Test
+ public void testToDateWithCloneMethod() throws SQLException {
+ Connection conn = DriverManager.getConnection(getUrl());
+ String ddl = "create table t (k varchar primary key, v varchar[])";
+ conn.createStatement().execute(ddl);
+ String dateStr = "2100-01-01";
+ conn.createStatement().execute("UPSERT INTO T VALUES('x',ARRAY['"+dateStr+"'])");
+ conn.commit();
+ ResultSet rs = conn.createStatement().executeQuery("select to_date(v[1], 'yyyy-MM-dd', 'local') from t");
+
+ assertTrue(rs.next());
+ assertEquals("Unexpected value for date ", Date.valueOf(dateStr), rs.getDate(1));
+ assertFalse(rs.next());
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b9880503/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToCharFunction.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToCharFunction.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToCharFunction.java
index a14a0cb..d284e25 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToCharFunction.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToCharFunction.java
@@ -57,10 +57,6 @@ public class ToCharFunction extends ScalarFunction {
public ToCharFunction() {
}
- public ToCharFunction(List<Expression> children) throws SQLException {
- this(children, null, null, null);
- }
-
public ToCharFunction(List<Expression> children, FunctionArgumentType type, String formatString, Format formatter) throws SQLException {
super(children.subList(0, 1));
Preconditions.checkNotNull(formatString);
@@ -72,6 +68,15 @@ public class ToCharFunction extends ScalarFunction {
}
@Override
+ public ToCharFunction clone(List<Expression> children) {
+ try {
+ return new ToCharFunction(children, type, formatString, formatter);
+ } catch (Exception e) {
+ throw new RuntimeException(e); // Impossible, since it was originally constructed this way
+ }
+ }
+
+ @Override
public int hashCode() {
final int prime = 31;
int result = 1;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b9880503/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToDateFunction.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToDateFunction.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToDateFunction.java
index f41d959..17f6847 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToDateFunction.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToDateFunction.java
@@ -53,21 +53,26 @@ import org.apache.phoenix.util.DateUtil;
public class ToDateFunction extends ScalarFunction {
public static final String NAME = "TO_DATE";
private DateUtil.DateTimeParser dateParser;
- private String dateFormat;
- private String timeZoneId;
+ protected String dateFormat;
+ protected String timeZoneId;
public ToDateFunction() {
}
- public ToDateFunction(List<Expression> children) throws SQLException {
- this(children, null, null);
- }
-
public ToDateFunction(List<Expression> children, String dateFormat, String timeZoneId) throws SQLException {
super(children);
init(dateFormat, timeZoneId);
}
+ @Override
+ public ToDateFunction clone(List<Expression> children) {
+ try {
+ return new ToDateFunction(children, dateFormat, timeZoneId);
+ } catch (Exception e) {
+ throw new RuntimeException(e); // Impossible, since it was originally constructed this way
+ }
+ }
+
private void init(String dateFormat, String timeZoneId) {
this.dateFormat = dateFormat;
this.dateParser = DateUtil.getDateTimeParser(dateFormat, getDataType(), timeZoneId);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b9880503/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToNumberFunction.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToNumberFunction.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToNumberFunction.java
index e7659a1..c43062f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToNumberFunction.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToNumberFunction.java
@@ -30,6 +30,7 @@ import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.io.WritableUtils;
import com.google.common.base.Preconditions;
+
import org.apache.phoenix.expression.Expression;
import org.apache.phoenix.parse.FunctionParseNode.Argument;
import org.apache.phoenix.parse.FunctionParseNode.BuiltInFunction;
@@ -60,10 +61,6 @@ public class ToNumberFunction extends ScalarFunction {
public ToNumberFunction() {}
- public ToNumberFunction(List<Expression> children) throws SQLException {
- this(children, null, null, null);
- }
-
public ToNumberFunction(List<Expression> children, FunctionArgumentType type, String formatString, Format formatter) throws SQLException {
super(children.subList(0, 1));
Preconditions.checkNotNull(type);
@@ -71,6 +68,15 @@ public class ToNumberFunction extends ScalarFunction {
this.formatString = formatString;
this.format = formatter;
}
+
+ @Override
+ public ToNumberFunction clone(List<Expression> children) {
+ try {
+ return new ToNumberFunction(children, type, formatString, format);
+ } catch (Exception e) {
+ throw new RuntimeException(e); // Impossible, since it was originally constructed this way
+ }
+ }
@Override
public boolean evaluate(Tuple tuple, ImmutableBytesWritable ptr) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b9880503/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToTimeFunction.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToTimeFunction.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToTimeFunction.java
index 2fa0553..588cb92 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToTimeFunction.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToTimeFunction.java
@@ -47,13 +47,18 @@ public class ToTimeFunction extends ToDateFunction {
public ToTimeFunction() {
}
- public ToTimeFunction(List<Expression> children) throws SQLException {
- this(children, null, null);
- }
-
public ToTimeFunction(List<Expression> children, String dateFormat, String timeZoneId) throws SQLException {
super(children, dateFormat, timeZoneId);
}
+
+ @Override
+ public ToTimeFunction clone(List<Expression> children) {
+ try {
+ return new ToTimeFunction(children, dateFormat, timeZoneId);
+ } catch (Exception e) {
+ throw new RuntimeException(e); // Impossible, since it was originally constructed this way
+ }
+ }
@Override
public PDataType getDataType() {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b9880503/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToTimestampFunction.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToTimestampFunction.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToTimestampFunction.java
index 5a4828f..6c14595 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToTimestampFunction.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ToTimestampFunction.java
@@ -47,13 +47,18 @@ public class ToTimestampFunction extends ToDateFunction {
public ToTimestampFunction() {
}
- public ToTimestampFunction(List<Expression> children) throws SQLException {
- this(children, null, null);
- }
-
public ToTimestampFunction(List<Expression> children, String dateFormat, String timeZoneId) throws SQLException {
super(children, dateFormat, timeZoneId);
}
+
+ @Override
+ public ToTimestampFunction clone(List<Expression> children) {
+ try {
+ return new ToTimestampFunction(children, dateFormat, timeZoneId);
+ } catch (Exception e) {
+ throw new RuntimeException(e); // Impossible, since it was originally constructed this way
+ }
+ }
@Override
public PDataType getDataType() {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b9880503/phoenix-core/src/test/java/org/apache/phoenix/expression/function/BuiltinFunctionConstructorTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/expression/function/BuiltinFunctionConstructorTest.java b/phoenix-core/src/test/java/org/apache/phoenix/expression/function/BuiltinFunctionConstructorTest.java
index 5258283..69f39fb 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/expression/function/BuiltinFunctionConstructorTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/expression/function/BuiltinFunctionConstructorTest.java
@@ -18,6 +18,7 @@
package org.apache.phoenix.expression.function;
import java.lang.reflect.Constructor;
+import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
@@ -35,9 +36,13 @@ public class BuiltinFunctionConstructorTest {
for(int i = 0; i < types.length; i++) {
try {
if((ScalarFunction.class.isAssignableFrom(types[i].getExpressionClass())) && (types[i].getExpressionClass() != UDFExpression.class)) {
- Constructor cons = types[i].getExpressionClass().getDeclaredConstructor(List.class);
- cons.setAccessible(true);
- cons.newInstance(children);
+ Method cloneMethod = types[i].getExpressionClass().getMethod("clone", List.class);
+ // ScalarFunctions that implement clone(List<Expression>) don't need to implement a constructor that takes a List<Expression>
+ if (cloneMethod==null) {
+ Constructor cons = types[i].getExpressionClass().getDeclaredConstructor(List.class);
+ cons.setAccessible(true);
+ cons.newInstance(children);
+ }
}
} catch (NoSuchMethodException e) {
throw new RuntimeException(e);
[08/30] phoenix git commit: PHOENIX-2728 Remove usage of usage of
com.google.common.io.Closeables.closeQuietly(Closeable)
Posted by ma...@apache.org.
PHOENIX-2728 Remove usage of usage of com.google.common.io.Closeables.closeQuietly(Closeable)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/42e2b85e
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/42e2b85e
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/42e2b85e
Branch: refs/heads/calcite
Commit: 42e2b85e391893ba033272bc42e9a0ebad1caca1
Parents: db4e5cd
Author: James Taylor <jt...@salesforce.com>
Authored: Tue Mar 1 22:24:46 2016 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Wed Mar 2 10:40:30 2016 -0800
----------------------------------------------------------------------
.../main/java/org/apache/phoenix/iterate/BaseResultIterators.java | 2 +-
.../main/java/org/apache/phoenix/schema/stats/PTableStatsImpl.java | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/42e2b85e/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
index a48de13..9f3b3e6 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
@@ -82,6 +82,7 @@ import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.schema.stats.GuidePostsInfo;
import org.apache.phoenix.schema.stats.PTableStats;
import org.apache.phoenix.util.ByteUtil;
+import org.apache.phoenix.util.Closeables;
import org.apache.phoenix.util.LogUtil;
import org.apache.phoenix.util.PrefixByteCodec;
import org.apache.phoenix.util.PrefixByteDecoder;
@@ -95,7 +96,6 @@ import org.slf4j.LoggerFactory;
import com.google.common.base.Function;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
-import com.google.common.io.Closeables;
/**
http://git-wip-us.apache.org/repos/asf/phoenix/blob/42e2b85e/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/PTableStatsImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/PTableStatsImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/PTableStatsImpl.java
index 42e7bed..5eb02fd 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/PTableStatsImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/PTableStatsImpl.java
@@ -28,11 +28,11 @@ import java.util.TreeMap;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.coprocessor.MetaDataProtocol;
+import org.apache.phoenix.util.Closeables;
import org.apache.phoenix.util.PrefixByteCodec;
import org.apache.phoenix.util.PrefixByteDecoder;
import org.apache.phoenix.util.SizedUtil;
-import com.google.common.io.Closeables;
import com.sun.istack.NotNull;
/**
[19/30] phoenix git commit: PHOENIX-2756 FilteredKeyValueScanner
should not implement KeyValueScanner (addendum)
Posted by ma...@apache.org.
PHOENIX-2756 FilteredKeyValueScanner should not implement KeyValueScanner (addendum)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/c19c3215
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/c19c3215
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/c19c3215
Branch: refs/heads/calcite
Commit: c19c32157a2ef4b65e4ef13ff1ba8c4ef9ef4597
Parents: 4377e33
Author: Thomas D'Silva <td...@salesforce.com>
Authored: Mon Mar 14 14:47:30 2016 -0700
Committer: Thomas D'Silva <td...@salesforce.com>
Committed: Mon Mar 14 14:57:40 2016 -0700
----------------------------------------------------------------------
.../org/apache/phoenix/hbase/index/scanner/ScannerBuilder.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c19c3215/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ScannerBuilder.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ScannerBuilder.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ScannerBuilder.java
index b4282ab..330cb0e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ScannerBuilder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ScannerBuilder.java
@@ -144,13 +144,13 @@ public class ScannerBuilder {
if (peek != null) {
int compare = KeyValue.COMPARATOR.compare(peek, next);
if (compare < 0) {
- return kvScanner.reseek(KeyValueUtil.ensureKeyValue(next));
+ return kvScanner.reseek(next);
} else if (compare == 0) {
// we are already at the given key!
return true;
}
}
- return kvScanner.seek(KeyValueUtil.ensureKeyValue(next));
+ return kvScanner.seek(next);
}
@Override
[03/30] phoenix git commit: PHOENIX-2720 Minor additions to
PhoenixRuntime utils
Posted by ma...@apache.org.
PHOENIX-2720 Minor additions to PhoenixRuntime utils
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/58cf5fa2
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/58cf5fa2
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/58cf5fa2
Branch: refs/heads/calcite
Commit: 58cf5fa2d2e147780258a1e88c79898fc199d61e
Parents: c083bc8
Author: James Taylor <jt...@salesforce.com>
Authored: Sun Feb 28 12:06:40 2016 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Sun Feb 28 12:06:40 2016 -0800
----------------------------------------------------------------------
.../phoenix/end2end/PhoenixRuntimeIT.java | 9 ++++++
.../org/apache/phoenix/util/PhoenixRuntime.java | 31 ++++++++++++++++++++
.../apache/phoenix/util/PhoenixRuntimeTest.java | 11 +++++++
3 files changed, 51 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58cf5fa2/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixRuntimeIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixRuntimeIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixRuntimeIT.java
index e748825..3cab7fa 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixRuntimeIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixRuntimeIT.java
@@ -133,6 +133,15 @@ public class PhoenixRuntimeIT extends BaseHBaseManagedTimeIT {
Expression e6 = PhoenixRuntime.getTenantIdExpression(conn, "I2");
HTableInterface htable6 = conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes("I2"));
assertTenantIds(e6, htable6, new FirstKeyOnlyFilter(), new String[] {"t1", "t2"} );
+
+ tableName = "BAR_" + (isSalted ? "SALTED" : "UNSALTED");
+ conn.createStatement().execute("CREATE TABLE " + tableName + " (k1 VARCHAR NOT NULL, k2 VARCHAR, CONSTRAINT PK PRIMARY KEY(K1,K2)) " + (isSalted ? "SALT_BUCKETS=3" : ""));
+ conn.createStatement().execute("UPSERT INTO " + tableName + " VALUES('t1','x')");
+ conn.createStatement().execute("UPSERT INTO " + tableName + " VALUES('t2','y')");
+ Expression e7 = PhoenixRuntime.getFirstPKColumnExpression(conn, tableName);
+ HTableInterface htable7 = conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(tableName));
+ assertTenantIds(e7, htable7, new FirstKeyOnlyFilter(), new String[] {"t1", "t2"} );
+
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58cf5fa2/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
index 05ba6d2..73b2f7a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
@@ -89,6 +89,8 @@ import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
+import co.cask.tephra.util.TxUtils;
+
/**
*
* Collection of non JDBC compliant utility methods
@@ -982,6 +984,25 @@ public class PhoenixRuntime {
if (!SchemaUtil.isMetaTable(table) && !SchemaUtil.isSequenceTable(table) && !table.isMultiTenant()) {
return null;
}
+ return getFirstPKColumnExpression(table);
+ }
+
+ /**
+ * Get expression that may be used to evaluate to the value of the first
+ * column of a given row in a Phoenix table.
+ * @param conn open Phoenix connection
+ * @param fullTableName full table name
+ * @return An expression that may be evaluated for a row in the provided table.
+ * @throws SQLException if the table name is not found, a TableNotFoundException
+ * is thrown. If a local index is supplied a SQLFeatureNotSupportedException
+ * is thrown.
+ */
+ public static Expression getFirstPKColumnExpression(Connection conn, String fullTableName) throws SQLException {
+ PTable table = getTable(conn, fullTableName);
+ return getFirstPKColumnExpression(table);
+ }
+
+ private static Expression getFirstPKColumnExpression(PTable table) throws SQLException {
if (table.getIndexType() == IndexType.LOCAL) {
/*
* With some hackery, we could deduce the tenant ID from a multi-tenant local index,
@@ -1156,4 +1177,14 @@ public class PhoenixRuntime {
pConn.clearMetrics();
}
+ /**
+ * Use this utility function to ensure that a timestamp is in milliseconds across transactional and
+ * non transactional tables. This expects that the Cell timestamp is based either on wall clock
+ * time or transaction manager nanos wall clock time.
+ * @param tsOfCell Cell time stamp to be converted.
+ * @return wall clock time in milliseconds (i.e. Epoch time) of a given Cell time stamp.
+ */
+ public static long getWallClockTimeFromCellTimeStamp(long tsOfCell) {
+ return TxUtils.isPreExistingVersion(tsOfCell) ? tsOfCell : TransactionUtil.convertToMilliseconds(tsOfCell);
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58cf5fa2/phoenix-core/src/test/java/org/apache/phoenix/util/PhoenixRuntimeTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/util/PhoenixRuntimeTest.java b/phoenix-core/src/test/java/org/apache/phoenix/util/PhoenixRuntimeTest.java
index 895e29e..fe81120 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/util/PhoenixRuntimeTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/util/PhoenixRuntimeTest.java
@@ -40,6 +40,7 @@ import org.apache.phoenix.compile.QueryPlan;
import org.apache.phoenix.expression.Expression;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.query.BaseConnectionlessQueryTest;
+import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.TableNotFoundException;
import org.apache.phoenix.schema.types.PDataType;
@@ -288,4 +289,14 @@ public class PhoenixRuntimeTest extends BaseConnectionlessQueryTest {
return "\"" + string + "\"";
}
+ @Test
+ public void testGetWallClockTimeFromCellTimeStamp() {
+ long ts = System.currentTimeMillis();
+ assertEquals(ts, PhoenixRuntime.getWallClockTimeFromCellTimeStamp(ts));
+ long nanoTs = TransactionUtil.convertToNanoseconds(ts);
+ assertEquals(ts, PhoenixRuntime.getWallClockTimeFromCellTimeStamp(nanoTs));
+ long skewedTs = ts + QueryConstants.MILLIS_IN_DAY; // skew of a day
+ // Even with a day of skew, we won't consider the ts a nanos timestamp
+ assertEquals(skewedTs, PhoenixRuntime.getWallClockTimeFromCellTimeStamp(skewedTs));
+ }
}
[10/30] phoenix git commit: PHOENIX-2733 Minor cleanup for
improvements to CSV Bulk Loader performance (Sergey Soldatov)
Posted by ma...@apache.org.
PHOENIX-2733 Minor cleanup for improvements to CSV Bulk Loader performance (Sergey Soldatov)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/952a01cc
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/952a01cc
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/952a01cc
Branch: refs/heads/calcite
Commit: 952a01cce6d7d314ca26e89a398e75f38690cdad
Parents: 1d03e02
Author: James Taylor <jt...@salesforce.com>
Authored: Wed Mar 2 13:54:05 2016 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Wed Mar 2 13:54:24 2016 -0800
----------------------------------------------------------------------
.../phoenix/mapreduce/AbstractBulkLoadTool.java | 12 +-
.../phoenix/mapreduce/CsvToKeyValueMapper.java | 2 +-
.../mapreduce/FormatToBytesWritableMapper.java | 385 +++++++++++++++++++
.../mapreduce/FormatToKeyValueMapper.java | 378 ------------------
.../mapreduce/FormatToKeyValueReducer.java | 28 +-
.../phoenix/mapreduce/JsonToKeyValueMapper.java | 2 +-
.../bulkload/TargetTableRefFunctions.java | 2 +-
.../util/PhoenixConfigurationUtil.java | 4 +-
.../FormatToBytesWritableMapperTest.java | 102 +++++
.../mapreduce/FormatToKeyValueMapperTest.java | 102 -----
10 files changed, 516 insertions(+), 501 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/952a01cc/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
index ab2848f..ff73530 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
@@ -209,10 +209,10 @@ public abstract class AbstractBulkLoadTool extends Configured implements Tool {
List<ColumnInfo> importColumns = buildImportColumns(conn, cmdLine, qualifiedTableName);
Preconditions.checkNotNull(importColumns);
Preconditions.checkArgument(!importColumns.isEmpty(), "Column info list is empty");
- FormatToKeyValueMapper.configureColumnInfoList(conf, importColumns);
+ FormatToBytesWritableMapper.configureColumnInfoList(conf, importColumns);
boolean ignoreInvalidRows = cmdLine.hasOption(IGNORE_ERRORS_OPT.getOpt());
- conf.setBoolean(FormatToKeyValueMapper.IGNORE_INVALID_ROW_CONFKEY, ignoreInvalidRows);
- conf.set(FormatToKeyValueMapper.TABLE_NAME_CONFKEY, qualifiedTableName);
+ conf.setBoolean(FormatToBytesWritableMapper.IGNORE_INVALID_ROW_CONFKEY, ignoreInvalidRows);
+ conf.set(FormatToBytesWritableMapper.TABLE_NAME_CONFKEY, qualifiedTableName);
// give subclasses their hook
configureOptions(cmdLine, importColumns, conf);
@@ -277,10 +277,10 @@ public abstract class AbstractBulkLoadTool extends Configured implements Tool {
MultiHfileOutputFormat.configureIncrementalLoad(job, tablesToBeLoaded);
final String tableNamesAsJson = TargetTableRefFunctions.NAMES_TO_JSON.apply(tablesToBeLoaded);
- final String logicalNamesAsJson = TargetTableRefFunctions.LOGICAN_NAMES_TO_JSON.apply(tablesToBeLoaded);
+ final String logicalNamesAsJson = TargetTableRefFunctions.LOGICAL_NAMES_TO_JSON.apply(tablesToBeLoaded);
- job.getConfiguration().set(FormatToKeyValueMapper.TABLE_NAMES_CONFKEY,tableNamesAsJson);
- job.getConfiguration().set(FormatToKeyValueMapper.LOGICAL_NAMES_CONFKEY,logicalNamesAsJson);
+ job.getConfiguration().set(FormatToBytesWritableMapper.TABLE_NAMES_CONFKEY,tableNamesAsJson);
+ job.getConfiguration().set(FormatToBytesWritableMapper.LOGICAL_NAMES_CONFKEY,logicalNamesAsJson);
// give subclasses their hook
setupJob(job);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/952a01cc/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapper.java
index 2cb1ac7..e5a6e5b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapper.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapper.java
@@ -41,7 +41,7 @@ import com.google.common.collect.Iterables;
* extracting the created KeyValues and rolling back the statement execution before it is
* committed to HBase.
*/
-public class CsvToKeyValueMapper extends FormatToKeyValueMapper<CSVRecord> {
+public class CsvToKeyValueMapper extends FormatToBytesWritableMapper<CSVRecord> {
/** Configuration key for the field delimiter for input csv records */
public static final String FIELD_DELIMITER_CONFKEY = "phoenix.mapreduce.import.fielddelimiter";
http://git-wip-us.apache.org/repos/asf/phoenix/blob/952a01cc/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java
new file mode 100644
index 0000000..2c9b6d9
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java
@@ -0,0 +1,385 @@
+/*
+ * 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.mapreduce;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.mapreduce.bulkload.TableRowkeyPair;
+import org.apache.phoenix.mapreduce.bulkload.TargetTableRefFunctions;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.util.ColumnInfo;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.UpsertExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
+import com.google.common.base.Splitter;
+import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+
+/**
+ * Base class for converting some input source format into {@link ImmutableBytesWritable}s that
+ * contains packed in a single byte array values for all columns.
+ * Assumes input format is text-based, with one row per line. Depends on an online cluster
+ * to retrieve {@link ColumnInfo} from the target table.
+ */
+public abstract class FormatToBytesWritableMapper<RECORD> extends Mapper<LongWritable, Text, TableRowkeyPair,
+ ImmutableBytesWritable> {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(FormatToBytesWritableMapper.class);
+
+ protected static final String COUNTER_GROUP_NAME = "Phoenix MapReduce Import";
+
+ /** Configuration key for the name of the output table */
+ public static final String TABLE_NAME_CONFKEY = "phoenix.mapreduce.import.tablename";
+
+ /** Configuration key for the columns to be imported */
+ public static final String COLUMN_INFO_CONFKEY = "phoenix.mapreduce.import.columninfos";
+
+ /** Configuration key for the flag to ignore invalid rows */
+ public static final String IGNORE_INVALID_ROW_CONFKEY = "phoenix.mapreduce.import.ignoreinvalidrow";
+
+ /** Configuration key for the table names */
+ public static final String TABLE_NAMES_CONFKEY = "phoenix.mapreduce.import.tablenames";
+
+ /** Configuration key for the table logical names */
+ public static final String LOGICAL_NAMES_CONFKEY = "phoenix.mapreduce.import.logicalnames";
+
+ /**
+ * Parses a single input line, returning a {@code T}.
+ */
+ public interface LineParser<T> {
+ T parse(String input) throws IOException;
+ }
+
+ protected PhoenixConnection conn;
+ protected UpsertExecutor<RECORD, ?> upsertExecutor;
+ protected ImportPreUpsertKeyValueProcessor preUpdateProcessor;
+ protected List<String> tableNames;
+ protected List<String> logicalNames;
+ protected MapperUpsertListener<RECORD> upsertListener;
+
+ /*
+ lookup table for column index. Index in the List matches to the index in tableNames List
+ */
+ protected List<Map<byte[], Map<byte[], Integer>>> columnIndexes;
+
+ protected abstract UpsertExecutor<RECORD,?> buildUpsertExecutor(Configuration conf);
+ protected abstract LineParser<RECORD> getLineParser();
+
+ @Override
+ protected void setup(Context context) throws IOException, InterruptedException {
+
+ Configuration conf = context.getConfiguration();
+
+ // pass client configuration into driver
+ Properties clientInfos = new Properties();
+ for (Map.Entry<String, String> entry : conf) {
+ clientInfos.setProperty(entry.getKey(), entry.getValue());
+ }
+
+ try {
+ conn = (PhoenixConnection) QueryUtil.getConnection(clientInfos, conf);
+
+ final String tableNamesConf = conf.get(TABLE_NAMES_CONFKEY);
+ final String logicalNamesConf = conf.get(LOGICAL_NAMES_CONFKEY);
+ tableNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(tableNamesConf);
+ logicalNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(logicalNamesConf);
+
+ columnIndexes = initColumnIndexes();
+ } catch (SQLException | ClassNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+
+ upsertListener = new MapperUpsertListener<RECORD>(
+ context, conf.getBoolean(IGNORE_INVALID_ROW_CONFKEY, true));
+ upsertExecutor = buildUpsertExecutor(conf);
+ preUpdateProcessor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
+ }
+
+ @SuppressWarnings("deprecation")
+ @Override
+ protected void map(LongWritable key, Text value, Context context) throws IOException,
+ InterruptedException {
+ if (conn == null) {
+ throw new RuntimeException("Connection not initialized.");
+ }
+ try {
+ RECORD record = null;
+ try {
+ record = getLineParser().parse(value.toString());
+ } catch (IOException e) {
+ context.getCounter(COUNTER_GROUP_NAME, "Parser errors").increment(1L);
+ return;
+ }
+
+ if (record == null) {
+ context.getCounter(COUNTER_GROUP_NAME, "Empty records").increment(1L);
+ return;
+ }
+ upsertExecutor.execute(ImmutableList.<RECORD>of(record));
+ Map<Integer, List<KeyValue>> map = new HashMap<>();
+ Iterator<Pair<byte[], List<KeyValue>>> uncommittedDataIterator
+ = PhoenixRuntime.getUncommittedDataIterator(conn, true);
+ while (uncommittedDataIterator.hasNext()) {
+ Pair<byte[], List<KeyValue>> kvPair = uncommittedDataIterator.next();
+ List<KeyValue> keyValueList = kvPair.getSecond();
+ keyValueList = preUpdateProcessor.preUpsert(kvPair.getFirst(), keyValueList);
+ byte[] first = kvPair.getFirst();
+ // Create a list of KV for each table
+ for (int i = 0; i < tableNames.size(); i++) {
+ if (Bytes.compareTo(Bytes.toBytes(tableNames.get(i)), first) == 0) {
+ if (!map.containsKey(i)) {
+ map.put(i, new ArrayList<KeyValue>());
+ }
+ List<KeyValue> list = map.get(i);
+ for (KeyValue kv : keyValueList) {
+ list.add(kv);
+ }
+ break;
+ }
+ }
+ }
+ for (Map.Entry<Integer, List<KeyValue>> rowEntry : map.entrySet()) {
+ int tableIndex = rowEntry.getKey();
+ List<KeyValue> lkv = rowEntry.getValue();
+ // All KV values combines to a single byte array
+ writeAggregatedRow(context, tableIndex, lkv);
+ }
+ conn.rollback();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private List<Map<byte[], Map<byte[], Integer>>> initColumnIndexes() throws SQLException {
+ List<Map<byte[], Map<byte[], Integer>>> tableMap = new ArrayList<>();
+ int tableIndex;
+ for (tableIndex = 0; tableIndex < tableNames.size(); tableIndex++) {
+ PTable table = PhoenixRuntime.getTable(conn, logicalNames.get(tableIndex));
+ Map<byte[], Map<byte[], Integer>> columnMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
+ List<PColumn> cls = table.getColumns();
+ for (int i = 0; i < cls.size(); i++) {
+ PColumn c = cls.get(i);
+ if (c.getFamilyName() == null) continue; // Skip PK column
+ byte[] family = c.getFamilyName().getBytes();
+ byte[] name = c.getName().getBytes();
+ if (!columnMap.containsKey(family)) {
+ columnMap.put(family, new TreeMap<byte[], Integer>(Bytes.BYTES_COMPARATOR));
+ }
+ Map<byte[], Integer> qualifier = columnMap.get(family);
+ qualifier.put(name, i);
+ }
+ tableMap.add(columnMap);
+ }
+ return tableMap;
+ }
+
+ /**
+ * Find the column index which will replace the column name in
+ * the aggregated array and will be restored in Reducer
+ *
+ * @param tableIndex Table index in tableNames list
+ * @param cell KeyValue for the column
+ * @return column index for the specified cell or -1 if was not found
+ */
+ private int findIndex(int tableIndex, Cell cell) {
+ Map<byte[], Map<byte[], Integer>> columnMap = columnIndexes.get(tableIndex);
+ Map<byte[], Integer> qualifiers = columnMap.get(Bytes.copy(cell.getFamilyArray(),
+ cell.getFamilyOffset(), cell.getFamilyLength()));
+ if (qualifiers != null) {
+ Integer result = qualifiers.get(Bytes.copy(cell.getQualifierArray(),
+ cell.getQualifierOffset(), cell.getQualifierLength()));
+ if (result != null) {
+ return result;
+ }
+ }
+ return -1;
+ }
+
+ /**
+ * Collect all column values for the same rowKey
+ *
+ * @param context Current mapper context
+ * @param tableIndex Table index in tableNames list
+ * @param lkv List of KV values that will be combined in a single ImmutableBytesWritable
+ * @throws IOException
+ * @throws InterruptedException
+ */
+
+ private void writeAggregatedRow(Context context, int tableIndex, List<KeyValue> lkv)
+ throws IOException, InterruptedException {
+ ByteArrayOutputStream bos = new ByteArrayOutputStream(1024);
+ DataOutputStream outputStream = new DataOutputStream(bos);
+ ImmutableBytesWritable outputKey = new ImmutableBytesWritable();
+ if (!lkv.isEmpty()) {
+ // All Key Values for the same row are supposed to be the same, so init rowKey only once
+ Cell first = lkv.get(0);
+ outputKey.set(first.getRowArray(), first.getRowOffset(), first.getRowLength());
+ for (KeyValue cell : lkv) {
+ if (isEmptyCell(cell)) {
+ continue;
+ }
+ int i = findIndex(tableIndex, cell);
+ if (i == -1) {
+ throw new IOException("No column found for KeyValue");
+ }
+ WritableUtils.writeVInt(outputStream, i);
+ outputStream.writeByte(cell.getTypeByte());
+ WritableUtils.writeVInt(outputStream, cell.getValueLength());
+ outputStream.write(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
+ }
+ }
+ ImmutableBytesWritable aggregatedArray = new ImmutableBytesWritable(bos.toByteArray());
+ outputStream.close();
+ context.write(new TableRowkeyPair(tableNames.get(tableIndex), outputKey), aggregatedArray);
+ }
+
+ protected boolean isEmptyCell(KeyValue cell) {
+ if (Bytes.compareTo(cell.getQualifierArray(), cell.getQualifierOffset(),
+ cell.getQualifierLength(), QueryConstants.EMPTY_COLUMN_BYTES, 0,
+ QueryConstants.EMPTY_COLUMN_BYTES.length) != 0)
+ return false;
+ else
+ return true;
+ }
+
+
+ @Override
+ protected void cleanup(Context context) throws IOException, InterruptedException {
+ try {
+ if (conn != null) {
+ conn.close();
+ }
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Write the list of to-import columns to a job configuration.
+ *
+ * @param conf configuration to be written to
+ * @param columnInfoList list of ColumnInfo objects to be configured for import
+ */
+ @VisibleForTesting
+ static void configureColumnInfoList(Configuration conf, List<ColumnInfo> columnInfoList) {
+ conf.set(COLUMN_INFO_CONFKEY, Joiner.on("|").useForNull("").join(columnInfoList));
+ }
+
+ /**
+ * Build the list of ColumnInfos for the import based on information in the configuration.
+ */
+ @VisibleForTesting
+ static List<ColumnInfo> buildColumnInfoList(Configuration conf) {
+
+ return Lists.newArrayList(
+ Iterables.transform(
+ Splitter.on("|").split(conf.get(COLUMN_INFO_CONFKEY)),
+ new Function<String, ColumnInfo>() {
+ @Nullable
+ @Override
+ public ColumnInfo apply(@Nullable String input) {
+ if (input == null || input.isEmpty()) {
+ // An empty string represents a null that was passed in to
+ // the configuration, which corresponds to an input column
+ // which is to be skipped
+ return null;
+ }
+ return ColumnInfo.fromString(input);
+ }
+ }));
+ }
+
+ /**
+ * Listener that logs successful upserts and errors to job counters.
+ */
+ @VisibleForTesting
+ static class MapperUpsertListener<T> implements UpsertExecutor.UpsertListener<T> {
+
+ private final Mapper<LongWritable, Text,
+ TableRowkeyPair, ImmutableBytesWritable>.Context context;
+ private final boolean ignoreRecordErrors;
+
+ private MapperUpsertListener(
+ Mapper<LongWritable, Text, TableRowkeyPair, ImmutableBytesWritable>.Context context,
+ boolean ignoreRecordErrors) {
+ this.context = context;
+ this.ignoreRecordErrors = ignoreRecordErrors;
+ }
+
+ @Override
+ public void upsertDone(long upsertCount) {
+ context.getCounter(COUNTER_GROUP_NAME, "Upserts Done").increment(1L);
+ }
+
+ @Override
+ public void errorOnRecord(T record, Throwable throwable) {
+ LOG.error("Error on record " + record, throwable);
+ context.getCounter(COUNTER_GROUP_NAME, "Errors on records").increment(1L);
+ if (!ignoreRecordErrors) {
+ throw Throwables.propagate(throwable);
+ }
+ }
+ }
+
+ /**
+ * A default implementation of {@code ImportPreUpsertKeyValueProcessor} that is used if no
+ * specific class is configured. This implementation simply passes through the KeyValue
+ * list that is passed in.
+ */
+ public static class DefaultImportPreUpsertKeyValueProcessor implements
+ ImportPreUpsertKeyValueProcessor {
+
+ @Override
+ public List<KeyValue> preUpsert(byte[] rowKey, List<KeyValue> keyValues) {
+ return keyValues;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/952a01cc/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
deleted file mode 100644
index 8dbb4aa..0000000
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
+++ /dev/null
@@ -1,378 +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.apache.phoenix.mapreduce;
-
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.sql.SQLException;
-import java.util.*;
-import javax.annotation.Nullable;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.Pair;
-import org.apache.hadoop.io.LongWritable;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.WritableUtils;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.phoenix.jdbc.PhoenixConnection;
-import org.apache.phoenix.mapreduce.bulkload.TableRowkeyPair;
-import org.apache.phoenix.mapreduce.bulkload.TargetTableRefFunctions;
-import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
-import org.apache.phoenix.query.QueryConstants;
-import org.apache.phoenix.schema.PColumn;
-import org.apache.phoenix.schema.PTable;
-import org.apache.phoenix.util.*;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Function;
-import com.google.common.base.Joiner;
-import com.google.common.base.Splitter;
-import com.google.common.base.Throwables;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-
-/**
- * Base class for converting some input source format into {@link KeyValue}s of a target
- * schema. Assumes input format is text-based, with one row per line. Depends on an online cluster
- * to retrieve {@link ColumnInfo} from the target table.
- */
-public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable, Text, TableRowkeyPair,
- ImmutableBytesWritable> {
-
- protected static final Logger LOG = LoggerFactory.getLogger(FormatToKeyValueMapper.class);
-
- protected static final String COUNTER_GROUP_NAME = "Phoenix MapReduce Import";
-
- /** Configuration key for the name of the output table */
- public static final String TABLE_NAME_CONFKEY = "phoenix.mapreduce.import.tablename";
-
- /** Configuration key for the name of the output index table */
- public static final String INDEX_TABLE_NAME_CONFKEY = "phoenix.mapreduce.import.indextablename";
-
- /** Configuration key for the columns to be imported */
- public static final String COLUMN_INFO_CONFKEY = "phoenix.mapreduce.import.columninfos";
-
- /** Configuration key for the flag to ignore invalid rows */
- public static final String IGNORE_INVALID_ROW_CONFKEY = "phoenix.mapreduce.import.ignoreinvalidrow";
-
- /** Configuration key for the table names */
- public static final String TABLE_NAMES_CONFKEY = "phoenix.mapreduce.import.tablenames";
- public static final String LOGICAL_NAMES_CONFKEY = "phoenix.mapreduce.import.logicalnames";
-
- /** Configuration key for the table configurations */
- public static final String TABLE_CONFIG_CONFKEY = "phoenix.mapreduce.import.table.config";
-
- /**
- * Parses a single input line, returning a {@code T}.
- */
- public interface LineParser<T> {
- T parse(String input) throws IOException;
- }
-
- protected PhoenixConnection conn;
- protected UpsertExecutor<RECORD, ?> upsertExecutor;
- protected ImportPreUpsertKeyValueProcessor preUpdateProcessor;
- protected List<String> tableNames;
- protected List<String> logicalNames;
- protected MapperUpsertListener<RECORD> upsertListener;
-
- /*
- lookup table for column index. Index in the List matches to the index in tableNames List
- */
- protected List<Map<byte[], Map<byte[], Integer>>> columnIndexes;
-
- protected abstract UpsertExecutor<RECORD,?> buildUpsertExecutor(Configuration conf);
- protected abstract LineParser<RECORD> getLineParser();
-
- @Override
- protected void setup(Context context) throws IOException, InterruptedException {
-
- Configuration conf = context.getConfiguration();
-
- // pass client configuration into driver
- Properties clientInfos = new Properties();
- for (Map.Entry<String, String> entry : conf) {
- clientInfos.setProperty(entry.getKey(), entry.getValue());
- }
-
- try {
- conn = (PhoenixConnection) QueryUtil.getConnection(clientInfos, conf);
-
- final String tableNamesConf = conf.get(TABLE_NAMES_CONFKEY);
- final String logicalNamesConf = conf.get(LOGICAL_NAMES_CONFKEY);
- tableNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(tableNamesConf);
- logicalNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(logicalNamesConf);
-
- columnIndexes = initColumnIndexes();
- } catch (SQLException | ClassNotFoundException e) {
- throw new RuntimeException(e);
- }
-
- upsertListener = new MapperUpsertListener<RECORD>(
- context, conf.getBoolean(IGNORE_INVALID_ROW_CONFKEY, true));
- upsertExecutor = buildUpsertExecutor(conf);
- preUpdateProcessor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
- }
-
- @SuppressWarnings("deprecation")
- @Override
- protected void map(LongWritable key, Text value, Context context) throws IOException,
- InterruptedException {
- if (conn == null) {
- throw new RuntimeException("Connection not initialized.");
- }
- try {
- RECORD record = null;
- try {
- record = getLineParser().parse(value.toString());
- } catch (IOException e) {
- context.getCounter(COUNTER_GROUP_NAME, "Parser errors").increment(1L);
- return;
- }
-
- if (record == null) {
- context.getCounter(COUNTER_GROUP_NAME, "Empty records").increment(1L);
- return;
- }
- upsertExecutor.execute(ImmutableList.<RECORD>of(record));
- Map<Integer, List<KeyValue>> map = new HashMap<>();
- Iterator<Pair<byte[], List<KeyValue>>> uncommittedDataIterator
- = PhoenixRuntime.getUncommittedDataIterator(conn, true);
- while (uncommittedDataIterator.hasNext()) {
- Pair<byte[], List<KeyValue>> kvPair = uncommittedDataIterator.next();
- List<KeyValue> keyValueList = kvPair.getSecond();
- keyValueList = preUpdateProcessor.preUpsert(kvPair.getFirst(), keyValueList);
- byte[] first = kvPair.getFirst();
- // Create a list of KV for each table
- for (int i = 0; i < tableNames.size(); i++) {
- if (Bytes.compareTo(Bytes.toBytes(tableNames.get(i)), first) == 0) {
- if (!map.containsKey(i)) {
- map.put(i, new ArrayList<KeyValue>());
- }
- List<KeyValue> list = map.get(i);
- for (KeyValue kv : keyValueList) {
- list.add(kv);
- }
- break;
- }
- }
- }
- for(Map.Entry<Integer, List<KeyValue>> rowEntry : map.entrySet()) {
- int tableIndex = rowEntry.getKey();
- List<KeyValue> lkv = rowEntry.getValue();
- // All KV values combines to a single byte array
- writeAggregatedRow(context, tableIndex, lkv);
- }
- conn.rollback();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- private List<Map<byte[],Map<byte[], Integer>>> initColumnIndexes() throws SQLException {
- List<Map<byte[], Map<byte[], Integer>>> tableMap = new ArrayList<>();
- int tableIndex;
- for (tableIndex = 0; tableIndex < tableNames.size(); tableIndex++) {
- PTable table = PhoenixRuntime.getTable(conn, logicalNames.get(tableIndex));
- Map<byte[], Map<byte[], Integer>> columnMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
- List<PColumn> cls = table.getColumns();
- for(int i = 0; i < cls.size(); i++) {
- PColumn c = cls.get(i);
- if(c.getFamilyName() == null) continue; // Skip PK column
- byte[] family = c.getFamilyName().getBytes();
- byte[] name = c.getName().getBytes();
- if(!columnMap.containsKey(family)) {
- columnMap.put(family, new TreeMap<byte[], Integer>(Bytes.BYTES_COMPARATOR));
- }
- Map<byte[], Integer> qualifier = columnMap.get(family);
- qualifier.put(name, i);
- }
- tableMap.add(columnMap);
- }
- return tableMap;
- }
-
- /**
- * Find the column index which will replace the column name in
- * the aggregated array and will be restored in Reducer
- *
- * @param tableIndex Table index in tableNames list
- * @param cell KeyValue for the column
- * @return column index for the specified cell or -1 if was not found
- */
- private int findIndex(int tableIndex, Cell cell) {
- Map<byte[], Map<byte[], Integer>> columnMap = columnIndexes.get(tableIndex);
- Map<byte[], Integer> qualifiers = columnMap.get(Bytes.copy(cell.getFamilyArray(),
- cell.getFamilyOffset(), cell.getFamilyLength()));
- if(qualifiers!= null) {
- Integer result = qualifiers.get(Bytes.copy(cell.getQualifierArray(),
- cell.getQualifierOffset(), cell.getQualifierLength()));
- if(result!=null) {
- return result;
- }
- }
- return -1;
- }
-
- /**
- * Collect all column values for the same rowKey
- *
- * @param context Current mapper context
- * @param tableIndex Table index in tableNames list
- * @param lkv List of KV values that will be combined in a single ImmutableBytesWritable
- * @throws IOException
- * @throws InterruptedException
- */
-
- private void writeAggregatedRow(Context context, int tableIndex, List<KeyValue> lkv)
- throws IOException, InterruptedException {
- TrustedByteArrayOutputStream bos = new TrustedByteArrayOutputStream(1024);
- DataOutputStream outputStream = new DataOutputStream(bos);
- ImmutableBytesWritable outputKey = new ImmutableBytesWritable();
- if (!lkv.isEmpty()) {
- // All Key Values for the same row are supposed to be the same, so init rowKey only once
- Cell first = lkv.get(0);
- outputKey.set(first.getRowArray(), first.getRowOffset(), first.getRowLength());
- for (KeyValue cell : lkv) {
- if(isEmptyCell(cell)) {
- continue;
- }
- int i = findIndex(tableIndex, cell);
- if (i == -1) {
- throw new IOException("No column found for KeyValue");
- }
- WritableUtils.writeVInt(outputStream, i);
- outputStream.writeByte(cell.getTypeByte());
- WritableUtils.writeVInt(outputStream, cell.getValueLength());
- outputStream.write(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
- }
- }
- ImmutableBytesWritable aggregatedArray = new ImmutableBytesWritable(bos.toByteArray());
- outputStream.close();
- context.write(new TableRowkeyPair(tableNames.get(tableIndex), outputKey), aggregatedArray);
- }
-
- protected boolean isEmptyCell(KeyValue cell) {
- if (Bytes.compareTo(cell.getQualifierArray(), cell.getQualifierOffset(),
- cell.getQualifierLength(), QueryConstants.EMPTY_COLUMN_BYTES, 0,
- QueryConstants.EMPTY_COLUMN_BYTES.length) != 0)
- return false;
- else
- return true;
- }
-
-
- @Override
- protected void cleanup(Context context) throws IOException, InterruptedException {
- try {
- if (conn != null) {
- conn.close();
- }
- } catch (SQLException e) {
- throw new RuntimeException(e);
- }
- }
-
- /**
- * Write the list of to-import columns to a job configuration.
- *
- * @param conf configuration to be written to
- * @param columnInfoList list of ColumnInfo objects to be configured for import
- */
- @VisibleForTesting
- static void configureColumnInfoList(Configuration conf, List<ColumnInfo> columnInfoList) {
- conf.set(COLUMN_INFO_CONFKEY, Joiner.on("|").useForNull("").join(columnInfoList));
- }
-
- /**
- * Build the list of ColumnInfos for the import based on information in the configuration.
- */
- @VisibleForTesting
- static List<ColumnInfo> buildColumnInfoList(Configuration conf) {
-
- return Lists.newArrayList(
- Iterables.transform(
- Splitter.on("|").split(conf.get(COLUMN_INFO_CONFKEY)),
- new Function<String, ColumnInfo>() {
- @Nullable
- @Override
- public ColumnInfo apply(@Nullable String input) {
- if (input == null || input.isEmpty()) {
- // An empty string represents a null that was passed in to
- // the configuration, which corresponds to an input column
- // which is to be skipped
- return null;
- }
- return ColumnInfo.fromString(input);
- }
- }));
- }
-
- /**
- * Listener that logs successful upserts and errors to job counters.
- */
- @VisibleForTesting
- static class MapperUpsertListener<T> implements UpsertExecutor.UpsertListener<T> {
-
- private final Mapper<LongWritable, Text,
- TableRowkeyPair, ImmutableBytesWritable>.Context context;
- private final boolean ignoreRecordErrors;
-
- private MapperUpsertListener(
- Mapper<LongWritable, Text, TableRowkeyPair, ImmutableBytesWritable>.Context context,
- boolean ignoreRecordErrors) {
- this.context = context;
- this.ignoreRecordErrors = ignoreRecordErrors;
- }
-
- @Override
- public void upsertDone(long upsertCount) {
- context.getCounter(COUNTER_GROUP_NAME, "Upserts Done").increment(1L);
- }
-
- @Override
- public void errorOnRecord(T record, Throwable throwable) {
- LOG.error("Error on record " + record, throwable);
- context.getCounter(COUNTER_GROUP_NAME, "Errors on records").increment(1L);
- if (!ignoreRecordErrors) {
- throw Throwables.propagate(throwable);
- }
- }
- }
-
- /**
- * A default implementation of {@code ImportPreUpsertKeyValueProcessor} that is used if no
- * specific class is configured. This implementation simply passes through the KeyValue
- * list that is passed in.
- */
- public static class DefaultImportPreUpsertKeyValueProcessor implements
- ImportPreUpsertKeyValueProcessor {
-
- @Override
- public List<KeyValue> preUpsert(byte[] rowKey, List<KeyValue> keyValues) {
- return keyValues;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/952a01cc/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
index fb61855..e906431 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
@@ -21,7 +21,11 @@ import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.sql.SQLException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.TreeSet;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.KeyValue;
@@ -38,7 +42,11 @@ import org.apache.phoenix.mapreduce.bulkload.TargetTableRefFunctions;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PTable;
-import org.apache.phoenix.util.*;
+import org.apache.phoenix.util.ByteUtil;
+import org.apache.phoenix.util.Closeables;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.SchemaUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,7 +55,7 @@ import org.slf4j.LoggerFactory;
* Performs similar functionality to {@link KeyValueSortReducer}
*/
public class FormatToKeyValueReducer
- extends Reducer<TableRowkeyPair,ImmutableBytesWritable,TableRowkeyPair,KeyValue> {
+ extends Reducer<TableRowkeyPair, ImmutableBytesWritable, TableRowkeyPair, KeyValue> {
protected static final Logger LOG = LoggerFactory.getLogger(FormatToKeyValueReducer.class);
@@ -72,8 +80,8 @@ public class FormatToKeyValueReducer
try {
PhoenixConnection conn = (PhoenixConnection) QueryUtil.getConnection(clientInfos, conf);
builder = conn.getKeyValueBuilder();
- final String tableNamesConf = conf.get(FormatToKeyValueMapper.TABLE_NAMES_CONFKEY);
- final String logicalNamesConf = conf.get(FormatToKeyValueMapper.LOGICAL_NAMES_CONFKEY);
+ final String tableNamesConf = conf.get(FormatToBytesWritableMapper.TABLE_NAMES_CONFKEY);
+ final String logicalNamesConf = conf.get(FormatToBytesWritableMapper.LOGICAL_NAMES_CONFKEY);
tableNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(tableNamesConf);
logicalNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(logicalNamesConf);
@@ -91,9 +99,9 @@ public class FormatToKeyValueReducer
emptyFamilyName.add(SchemaUtil.getEmptyColumnFamilyPtr(table));
List<PColumn> cls = table.getColumns();
List<Pair<byte[], byte[]>> list = new ArrayList(cls.size());
- for(int i = 0; i < cls.size(); i++) {
+ for (int i = 0; i < cls.size(); i++) {
PColumn c = cls.get(i);
- if(c.getFamilyName() == null) {
+ if (c.getFamilyName() == null) {
list.add(null); // Skip PK column
continue;
}
@@ -108,14 +116,14 @@ public class FormatToKeyValueReducer
@Override
protected void reduce(TableRowkeyPair key, Iterable<ImmutableBytesWritable> values,
- Reducer<TableRowkeyPair, ImmutableBytesWritable, TableRowkeyPair, KeyValue>.Context context)
- throws IOException, InterruptedException {
+ Reducer<TableRowkeyPair, ImmutableBytesWritable, TableRowkeyPair, KeyValue>.Context context)
+ throws IOException, InterruptedException {
TreeSet<KeyValue> map = new TreeSet<KeyValue>(KeyValue.COMPARATOR);
int tableIndex = tableNames.indexOf(key.getTableName());
List<Pair<byte[], byte[]>> columns = columnIndexes.get(tableIndex);
for (ImmutableBytesWritable aggregatedArray : values) {
DataInputStream input = new DataInputStream(new ByteArrayInputStream(aggregatedArray.get()));
- while (input.available()!= 0) {
+ while (input.available() != 0) {
int index = WritableUtils.readVInt(input);
Pair<byte[], byte[]> pair = columns.get(index);
byte type = input.readByte();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/952a01cc/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/JsonToKeyValueMapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/JsonToKeyValueMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/JsonToKeyValueMapper.java
index 5173a0e..2a51b2c 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/JsonToKeyValueMapper.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/JsonToKeyValueMapper.java
@@ -37,7 +37,7 @@ import com.google.common.base.Preconditions;
* extracting the created KeyValues and rolling back the statement execution before it is
* committed to HBase.
*/
-public class JsonToKeyValueMapper extends FormatToKeyValueMapper<Map<?, ?>> {
+public class JsonToKeyValueMapper extends FormatToBytesWritableMapper<Map<?, ?>> {
private LineParser<Map<?, ?>> lineParser;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/952a01cc/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
index e02065f..58725c2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
@@ -78,7 +78,7 @@ public class TargetTableRefFunctions {
}
};
- public static Function<List<TargetTableRef>,String> LOGICAN_NAMES_TO_JSON = new Function<List<TargetTableRef>,String>() {
+ public static Function<List<TargetTableRef>,String> LOGICAL_NAMES_TO_JSON = new Function<List<TargetTableRef>,String>() {
@Override
public String apply(List<TargetTableRef> input) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/952a01cc/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
index 280daa2..b1879d1 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
@@ -36,7 +36,7 @@ import org.apache.hadoop.mapreduce.lib.db.DBInputFormat.NullDBWritable;
import org.apache.hadoop.mapreduce.lib.db.DBWritable;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.phoenix.jdbc.PhoenixConnection;
-import org.apache.phoenix.mapreduce.FormatToKeyValueMapper;
+import org.apache.phoenix.mapreduce.FormatToBytesWritableMapper;
import org.apache.phoenix.mapreduce.ImportPreUpsertKeyValueProcessor;
import org.apache.phoenix.mapreduce.PhoenixInputFormat;
import org.apache.phoenix.util.ColumnInfo;
@@ -418,7 +418,7 @@ public final class PhoenixConfigurationUtil {
Class<? extends ImportPreUpsertKeyValueProcessor> processorClass = null;
try {
processorClass = conf.getClass(
- UPSERT_HOOK_CLASS_CONFKEY, FormatToKeyValueMapper.DefaultImportPreUpsertKeyValueProcessor.class,
+ UPSERT_HOOK_CLASS_CONFKEY, FormatToBytesWritableMapper.DefaultImportPreUpsertKeyValueProcessor.class,
ImportPreUpsertKeyValueProcessor.class);
} catch (Exception e) {
throw new IllegalStateException("Couldn't load upsert hook class", e);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/952a01cc/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapperTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapperTest.java b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapperTest.java
new file mode 100644
index 0000000..6424976
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapperTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.mapreduce;
+
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
+import org.apache.phoenix.schema.types.PInteger;
+import org.apache.phoenix.schema.types.PIntegerArray;
+import org.apache.phoenix.schema.types.PUnsignedInt;
+import org.apache.phoenix.util.ColumnInfo;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+import static org.junit.Assert.assertEquals;
+
+public class FormatToBytesWritableMapperTest {
+
+ @Test
+ public void testBuildColumnInfoList() {
+ List<ColumnInfo> columnInfoList = ImmutableList.of(
+ new ColumnInfo("idCol", PInteger.INSTANCE.getSqlType()),
+ new ColumnInfo("unsignedIntCol", PUnsignedInt.INSTANCE.getSqlType()),
+ new ColumnInfo("stringArrayCol", PIntegerArray.INSTANCE.getSqlType()));
+
+ Configuration conf = new Configuration();
+ FormatToBytesWritableMapper.configureColumnInfoList(conf, columnInfoList);
+ List<ColumnInfo> fromConfig = FormatToBytesWritableMapper.buildColumnInfoList(conf);
+
+ assertEquals(columnInfoList, fromConfig);
+ }
+
+ @Test
+ public void testBuildColumnInfoList_ContainingNulls() {
+ // A null value in the column info list means "skip that column in the input"
+ List<ColumnInfo> columnInfoListWithNull = Lists.newArrayList(
+ new ColumnInfo("idCol", PInteger.INSTANCE.getSqlType()),
+ null,
+ new ColumnInfo("unsignedIntCol", PUnsignedInt.INSTANCE.getSqlType()),
+ new ColumnInfo("stringArrayCol", PIntegerArray.INSTANCE.getSqlType()));
+
+ Configuration conf = new Configuration();
+ FormatToBytesWritableMapper.configureColumnInfoList(conf, columnInfoListWithNull);
+ List<ColumnInfo> fromConfig = FormatToBytesWritableMapper.buildColumnInfoList(conf);
+
+ assertEquals(columnInfoListWithNull, fromConfig);
+ }
+
+ @Test
+ public void testLoadPreUpdateProcessor() {
+ Configuration conf = new Configuration();
+ conf.setClass(PhoenixConfigurationUtil.UPSERT_HOOK_CLASS_CONFKEY, MockUpsertProcessor.class,
+ ImportPreUpsertKeyValueProcessor.class);
+
+ ImportPreUpsertKeyValueProcessor processor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
+ assertEquals(MockUpsertProcessor.class, processor.getClass());
+ }
+
+ @Test
+ public void testLoadPreUpdateProcessor_NotConfigured() {
+
+ Configuration conf = new Configuration();
+ ImportPreUpsertKeyValueProcessor processor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
+
+ assertEquals(FormatToBytesWritableMapper.DefaultImportPreUpsertKeyValueProcessor.class,
+ processor.getClass());
+ }
+
+ @Test(expected=IllegalStateException.class)
+ public void testLoadPreUpdateProcessor_ClassNotFound() {
+ Configuration conf = new Configuration();
+ conf.set(PhoenixConfigurationUtil.UPSERT_HOOK_CLASS_CONFKEY, "MyUndefinedClass");
+
+ PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
+ }
+
+ static class MockUpsertProcessor implements ImportPreUpsertKeyValueProcessor {
+ @Override
+ public List<KeyValue> preUpsert(byte[] rowKey, List<KeyValue> keyValues) {
+ throw new UnsupportedOperationException("Not yet implemented");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/952a01cc/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapperTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapperTest.java b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapperTest.java
deleted file mode 100644
index 3455616..0000000
--- a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapperTest.java
+++ /dev/null
@@ -1,102 +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.apache.phoenix.mapreduce;
-
-import java.util.List;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
-import org.apache.phoenix.schema.types.PInteger;
-import org.apache.phoenix.schema.types.PIntegerArray;
-import org.apache.phoenix.schema.types.PUnsignedInt;
-import org.apache.phoenix.util.ColumnInfo;
-import org.junit.Test;
-
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Lists;
-
-import static org.junit.Assert.assertEquals;
-
-public class FormatToKeyValueMapperTest {
-
- @Test
- public void testBuildColumnInfoList() {
- List<ColumnInfo> columnInfoList = ImmutableList.of(
- new ColumnInfo("idCol", PInteger.INSTANCE.getSqlType()),
- new ColumnInfo("unsignedIntCol", PUnsignedInt.INSTANCE.getSqlType()),
- new ColumnInfo("stringArrayCol", PIntegerArray.INSTANCE.getSqlType()));
-
- Configuration conf = new Configuration();
- FormatToKeyValueMapper.configureColumnInfoList(conf, columnInfoList);
- List<ColumnInfo> fromConfig = FormatToKeyValueMapper.buildColumnInfoList(conf);
-
- assertEquals(columnInfoList, fromConfig);
- }
-
- @Test
- public void testBuildColumnInfoList_ContainingNulls() {
- // A null value in the column info list means "skip that column in the input"
- List<ColumnInfo> columnInfoListWithNull = Lists.newArrayList(
- new ColumnInfo("idCol", PInteger.INSTANCE.getSqlType()),
- null,
- new ColumnInfo("unsignedIntCol", PUnsignedInt.INSTANCE.getSqlType()),
- new ColumnInfo("stringArrayCol", PIntegerArray.INSTANCE.getSqlType()));
-
- Configuration conf = new Configuration();
- FormatToKeyValueMapper.configureColumnInfoList(conf, columnInfoListWithNull);
- List<ColumnInfo> fromConfig = FormatToKeyValueMapper.buildColumnInfoList(conf);
-
- assertEquals(columnInfoListWithNull, fromConfig);
- }
-
- @Test
- public void testLoadPreUpdateProcessor() {
- Configuration conf = new Configuration();
- conf.setClass(PhoenixConfigurationUtil.UPSERT_HOOK_CLASS_CONFKEY, MockUpsertProcessor.class,
- ImportPreUpsertKeyValueProcessor.class);
-
- ImportPreUpsertKeyValueProcessor processor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
- assertEquals(MockUpsertProcessor.class, processor.getClass());
- }
-
- @Test
- public void testLoadPreUpdateProcessor_NotConfigured() {
-
- Configuration conf = new Configuration();
- ImportPreUpsertKeyValueProcessor processor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
-
- assertEquals(FormatToKeyValueMapper.DefaultImportPreUpsertKeyValueProcessor.class,
- processor.getClass());
- }
-
- @Test(expected=IllegalStateException.class)
- public void testLoadPreUpdateProcessor_ClassNotFound() {
- Configuration conf = new Configuration();
- conf.set(PhoenixConfigurationUtil.UPSERT_HOOK_CLASS_CONFKEY, "MyUndefinedClass");
-
- PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
- }
-
- static class MockUpsertProcessor implements ImportPreUpsertKeyValueProcessor {
- @Override
- public List<KeyValue> preUpsert(byte[] rowKey, List<KeyValue> keyValues) {
- throw new UnsupportedOperationException("Not yet implemented");
- }
- }
-}
[18/30] phoenix git commit: PHOENIX-2756 - FilteredKeyValueScanner
should not implement KeyValueScanner
Posted by ma...@apache.org.
PHOENIX-2756 - FilteredKeyValueScanner should not implement KeyValueScanner
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/4377e339
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/4377e339
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/4377e339
Branch: refs/heads/calcite
Commit: 4377e3395d7b8107b2b0b3dbe4dff295d355cd0e
Parents: cc98e46
Author: rahul gidwani <rg...@salesforce.com>
Authored: Thu Mar 10 14:45:26 2016 -0800
Committer: Thomas D'Silva <td...@salesforce.com>
Committed: Mon Mar 14 11:41:31 2016 -0700
----------------------------------------------------------------------
.../index/scanner/FilteredKeyValueScanner.java | 56 +-------------------
1 file changed, 1 insertion(+), 55 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4377e339/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/FilteredKeyValueScanner.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/FilteredKeyValueScanner.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/FilteredKeyValueScanner.java
index 435a1c0..a1cbd61 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/FilteredKeyValueScanner.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/FilteredKeyValueScanner.java
@@ -19,12 +19,10 @@
package org.apache.phoenix.hbase.index.scanner;
import java.io.IOException;
-import java.util.SortedSet;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValueUtil;
-import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.Filter.ReturnCode;
import org.apache.hadoop.hbase.regionserver.KeyValueScanner;
@@ -35,7 +33,7 @@ import org.apache.phoenix.hbase.index.covered.KeyValueStore;
* here because we are only concerned with a single MemStore for the index; we don't need to worry about multiple column
* families or minimizing seeking through file - we just want to iterate the kvs quickly, in-memory.
*/
-public class FilteredKeyValueScanner implements KeyValueScanner {
+public class FilteredKeyValueScanner implements Scanner {
private KeyValueScanner delegate;
private Filter filter;
@@ -102,67 +100,15 @@ public class FilteredKeyValueScanner implements KeyValueScanner {
}
}
- @Override
public boolean reseek(Cell key) throws IOException {
this.delegate.reseek(key);
return this.seekToNextUnfilteredKeyValue();
}
- @Override
- public boolean requestSeek(Cell kv, boolean forward, boolean useBloom) throws IOException {
- return this.reseek(kv);
- }
-
- @Override
- public boolean isFileScanner() {
- return false;
- }
-
- @Override
- public long getSequenceID() {
- return this.delegate.getSequenceID();
- }
-
- @Override
- public boolean shouldUseScanner(Scan scan, SortedSet<byte[]> columns, long oldestUnexpiredTS) {
- throw new UnsupportedOperationException(this.getClass().getName()
- + " doesn't support checking to see if it should use a scanner!");
- }
-
- @Override
- public boolean realSeekDone() {
- return this.delegate.realSeekDone();
- }
-
- @Override
- public void enforceSeek() throws IOException {
- this.delegate.enforceSeek();
- }
@Override
public void close() {
this.delegate.close();
}
- @Override
- public boolean backwardSeek(Cell arg0) throws IOException {
- return this.delegate.backwardSeek(arg0);
- }
-
- @Override
- public boolean seekToLastRow() throws IOException {
- return this.delegate.seekToLastRow();
- }
-
- @Override
- public boolean seekToPreviousRow(Cell arg0) throws IOException {
- return this.delegate.seekToPreviousRow(arg0);
- }
-
- // Added for compatibility with HBASE-13109
- // Once we drop support for older versions, add an @override annotation here
- // and figure out how to get the next indexed key
- public Cell getNextIndexedKey() {
- return null; // indicate that we cannot use the optimization
- }
}
[24/30] phoenix git commit: PHOENIX-2774 MemStoreScanner and
KeyValueStore should not be aware of KeyValueScanner (Churro Morales)
Posted by ma...@apache.org.
PHOENIX-2774 MemStoreScanner and KeyValueStore should not be aware of KeyValueScanner (Churro Morales)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/bea110c8
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/bea110c8
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/bea110c8
Branch: refs/heads/calcite
Commit: bea110c8e70e26514aa658fb2b02d38d215ae6b4
Parents: cd8e86c
Author: James Taylor <jt...@salesforce.com>
Authored: Mon Mar 28 08:21:16 2016 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Mon Mar 28 08:21:16 2016 -0700
----------------------------------------------------------------------
.../hbase/index/covered/KeyValueStore.java | 4 +-
.../hbase/index/covered/LocalTableState.java | 27 ------------
.../hbase/index/covered/data/IndexMemStore.java | 43 +++-----------------
.../index/scanner/FilteredKeyValueScanner.java | 11 +++--
.../hbase/index/scanner/ReseekableScanner.java | 28 +++++++++++++
.../hbase/index/scanner/ScannerBuilder.java | 2 +-
.../index/covered/data/TestIndexMemStore.java | 6 +--
7 files changed, 44 insertions(+), 77 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/bea110c8/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/KeyValueStore.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/KeyValueStore.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/KeyValueStore.java
index 39f9062..30d2904 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/KeyValueStore.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/KeyValueStore.java
@@ -18,7 +18,7 @@
package org.apache.phoenix.hbase.index.covered;
import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.regionserver.KeyValueScanner;
+import org.apache.phoenix.hbase.index.scanner.ReseekableScanner;
/**
* Store a collection of KeyValues in memory.
@@ -27,7 +27,7 @@ public interface KeyValueStore {
public void add(KeyValue kv, boolean overwrite);
- public KeyValueScanner getScanner();
+ public ReseekableScanner getScanner();
public void rollback(KeyValue kv);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/bea110c8/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/LocalTableState.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/LocalTableState.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/LocalTableState.java
index 2739cc2..3a7a7eb 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/LocalTableState.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/LocalTableState.java
@@ -194,33 +194,6 @@ public class LocalTableState implements TableState {
return this.update.getRow();
}
- public Result getCurrentRowState() {
- KeyValueScanner scanner = this.memstore.getScanner();
- List<Cell> kvs = new ArrayList<Cell>();
- while (scanner.peek() != null) {
- try {
- kvs.add(scanner.next());
- } catch (IOException e) {
- // this should never happen - something has gone terribly arwy if it has
- throw new RuntimeException("Local MemStore threw IOException!");
- }
- }
- return Result.create(kvs);
- }
-
- /**
- * Helper to add a {@link Mutation} to the values stored for the current row
- *
- * @param pendingUpdate
- * update to apply
- */
- public void addUpdateForTesting(Mutation pendingUpdate) {
- for (Map.Entry<byte[], List<Cell>> e : pendingUpdate.getFamilyCellMap().entrySet()) {
- List<KeyValue> edits = KeyValueUtil.ensureKeyValues(e.getValue());
- addUpdate(edits);
- }
- }
-
/**
* @param hints
*/
http://git-wip-us.apache.org/repos/asf/phoenix/blob/bea110c8/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/IndexMemStore.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/IndexMemStore.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/IndexMemStore.java
index 7ae54ad..c7c5e3a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/IndexMemStore.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/IndexMemStore.java
@@ -17,7 +17,6 @@
*/
package org.apache.phoenix.hbase.index.covered.data;
-import java.io.IOException;
import java.util.Comparator;
import java.util.Iterator;
import java.util.SortedSet;
@@ -30,13 +29,12 @@ import org.apache.hadoop.hbase.KeyValue.KVComparator;
import org.apache.hadoop.hbase.KeyValueUtil;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.regionserver.IndexKeyValueSkipListSet;
-import org.apache.hadoop.hbase.regionserver.KeyValueScanner;
import org.apache.hadoop.hbase.regionserver.MemStore;
-import org.apache.hadoop.hbase.regionserver.NonLazyKeyValueScanner;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.hbase.index.covered.KeyValueStore;
import org.apache.phoenix.hbase.index.covered.LocalTableState;
+import org.apache.phoenix.hbase.index.scanner.ReseekableScanner;
/**
* Like the HBase {@link MemStore}, but without all that extra work around maintaining snapshots and
@@ -55,7 +53,7 @@ import org.apache.phoenix.hbase.index.covered.LocalTableState;
* <li>ignoring memstore timestamps in favor of deciding when we want to overwrite keys based on how
* we obtain them</li>
* <li>ignoring time range updates (so
- * {@link KeyValueScanner#shouldUseScanner(Scan, SortedSet, long)} isn't supported from
+ * {@link ReseekableScanner#shouldUseScanner(Scan, SortedSet, long)} isn't supported from
* {@link #getScanner()}).</li>
* </ol>
* <p>
@@ -158,19 +156,19 @@ public class IndexMemStore implements KeyValueStore {
}
@Override
- public KeyValueScanner getScanner() {
+ public ReseekableScanner getScanner() {
return new MemStoreScanner();
}
/*
- * MemStoreScanner implements the KeyValueScanner. It lets the caller scan the contents of a
+ * MemStoreScanner implements the ReseekableScanner. It lets the caller scan the contents of a
* memstore -- both current map and snapshot. This behaves as if it were a real scanner but does
* not maintain position.
*/
// This class is adapted from org.apache.hadoop.hbase.MemStore.MemStoreScanner, HBase 0.94.12
// It does basically the same thing as the MemStoreScanner, but it only keeps track of a single
// set, rather than a primary and a secondary set of KeyValues.
- protected class MemStoreScanner extends NonLazyKeyValueScanner {
+ protected class MemStoreScanner implements ReseekableScanner {
// Next row information for the set
private KeyValue nextRow = null;
@@ -276,7 +274,6 @@ public class IndexMemStore implements KeyValueStore {
@Override
public synchronized Cell peek() {
- // DebugPrint.println(" MS@" + hashCode() + " peek = " + getLowest());
return nextRow;
}
@@ -300,35 +297,5 @@ public class IndexMemStore implements KeyValueStore {
this.kvsetIt = null;
this.kvsetItRow = null;
}
-
- /**
- * MemStoreScanner returns max value as sequence id because it will always have the latest data
- * among all files.
- */
- @Override
- public long getSequenceID() {
- return Long.MAX_VALUE;
- }
-
- @Override
- public boolean shouldUseScanner(Scan scan, SortedSet<byte[]> columns, long oldestUnexpiredTS) {
- throw new UnsupportedOperationException(this.getClass().getName()
- + " doesn't support checking to see if it should use a scanner!");
- }
-
- @Override
- public boolean backwardSeek(Cell arg0) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean seekToLastRow() throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean seekToPreviousRow(Cell arg0) throws IOException {
- throw new UnsupportedOperationException();
- }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/bea110c8/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/FilteredKeyValueScanner.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/FilteredKeyValueScanner.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/FilteredKeyValueScanner.java
index a1cbd61..072b624 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/FilteredKeyValueScanner.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/FilteredKeyValueScanner.java
@@ -25,7 +25,6 @@ import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValueUtil;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.Filter.ReturnCode;
-import org.apache.hadoop.hbase.regionserver.KeyValueScanner;
import org.apache.phoenix.hbase.index.covered.KeyValueStore;
/**
@@ -33,22 +32,22 @@ import org.apache.phoenix.hbase.index.covered.KeyValueStore;
* here because we are only concerned with a single MemStore for the index; we don't need to worry about multiple column
* families or minimizing seeking through file - we just want to iterate the kvs quickly, in-memory.
*/
-public class FilteredKeyValueScanner implements Scanner {
+public class FilteredKeyValueScanner implements ReseekableScanner {
- private KeyValueScanner delegate;
+ private ReseekableScanner delegate;
private Filter filter;
public FilteredKeyValueScanner(Filter filter, KeyValueStore store) {
this(filter, store.getScanner());
}
- private FilteredKeyValueScanner(Filter filter, KeyValueScanner delegate) {
+ private FilteredKeyValueScanner(Filter filter, ReseekableScanner delegate) {
this.delegate = delegate;
this.filter = filter;
}
@Override
- public Cell peek() {
+ public Cell peek() throws IOException {
return delegate.peek();
}
@@ -107,7 +106,7 @@ public class FilteredKeyValueScanner implements Scanner {
@Override
- public void close() {
+ public void close() throws IOException {
this.delegate.close();
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/bea110c8/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ReseekableScanner.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ReseekableScanner.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ReseekableScanner.java
new file mode 100644
index 0000000..cc82fb6
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ReseekableScanner.java
@@ -0,0 +1,28 @@
+/**
+ *
+ * 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.hbase.index.scanner;
+
+import org.apache.hadoop.hbase.Cell;
+
+import java.io.IOException;
+
+public interface ReseekableScanner extends Scanner {
+
+ boolean reseek(Cell key) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/bea110c8/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ScannerBuilder.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ScannerBuilder.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ScannerBuilder.java
index 330cb0e..f8d0cf1 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ScannerBuilder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/scanner/ScannerBuilder.java
@@ -159,7 +159,7 @@ public class ScannerBuilder {
}
@Override
- public void close() {
+ public void close() throws IOException {
kvScanner.close();
}
};
http://git-wip-us.apache.org/repos/asf/phoenix/blob/bea110c8/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/data/TestIndexMemStore.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/data/TestIndexMemStore.java b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/data/TestIndexMemStore.java
index 42e0b03..400757d 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/data/TestIndexMemStore.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/data/TestIndexMemStore.java
@@ -22,8 +22,8 @@ import static org.junit.Assert.assertTrue;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValue.Type;
-import org.apache.hadoop.hbase.regionserver.KeyValueScanner;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.hbase.index.scanner.ReseekableScanner;
import org.junit.Test;
public class TestIndexMemStore {
@@ -45,7 +45,7 @@ public class TestIndexMemStore {
store.add(kv, true);
// adding the exact same kv shouldn't change anything stored if not overwritting
store.add(kv2, false);
- KeyValueScanner scanner = store.getScanner();
+ ReseekableScanner scanner = store.getScanner();
KeyValue first = KeyValue.createFirstOnRow(row);
scanner.seek(first);
assertTrue("Overwrote kv when specifically not!", kv == scanner.next());
@@ -79,7 +79,7 @@ public class TestIndexMemStore {
store.add(d, true);
// null qualifiers should always sort before the non-null cases
- KeyValueScanner scanner = store.getScanner();
+ ReseekableScanner scanner = store.getScanner();
KeyValue first = KeyValue.createFirstOnRow(row);
assertTrue("Didn't have any data in the scanner", scanner.seek(first));
assertTrue("Didn't get delete family first (no qualifier == sort first)", df == scanner.next());
[21/30] phoenix git commit: PHOENIX-2768 Add test for case sensitive
table with index hint
Posted by ma...@apache.org.
PHOENIX-2768 Add test for case sensitive table with index hint
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/d8b45bcb
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/d8b45bcb
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/d8b45bcb
Branch: refs/heads/calcite
Commit: d8b45bcb3d923124455788d27718aafd3c0158c1
Parents: c19c321
Author: James Taylor <jt...@salesforce.com>
Authored: Mon Mar 14 20:33:03 2016 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Tue Mar 15 15:56:03 2016 -0700
----------------------------------------------------------------------
.../phoenix/compile/QueryOptimizerTest.java | 28 ++++++++++++++++++++
1 file changed, 28 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d8b45bcb/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryOptimizerTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryOptimizerTest.java b/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryOptimizerTest.java
index e528d3b..c3345ba 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryOptimizerTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryOptimizerTest.java
@@ -273,6 +273,34 @@ public class QueryOptimizerTest extends BaseConnectionlessQueryTest {
PhoenixStatement stmt = conn.createStatement().unwrap(PhoenixStatement.class);
QueryPlan plan = stmt.optimizeQuery("SELECT /*+ INDEX(t idx1) */ k FROM t WHERE v1 = 'foo' AND v2 = 'bar'");
assertEquals("IDX1", plan.getTableRef().getTable().getTableName().getString());
+ plan = stmt.optimizeQuery("SELECT k FROM t WHERE v1 = 'foo' AND v2 = 'bar'");
+ assertEquals("IDX2", plan.getTableRef().getTable().getTableName().getString());
+ }
+
+ @Test
+ public void testChooseIndexFromCaseSensitiveHint() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+ conn.createStatement().execute("CREATE TABLE \"t\" (k INTEGER NOT NULL PRIMARY KEY, v1 VARCHAR, v2 VARCHAR) IMMUTABLE_ROWS=true");
+ conn.createStatement().execute("CREATE INDEX idx1 ON \"t\"(v1) INCLUDE(v2)");
+ conn.createStatement().execute("CREATE INDEX idx2 ON \"t\"(v1,v2)");
+ PhoenixStatement stmt = conn.createStatement().unwrap(PhoenixStatement.class);
+ QueryPlan plan = stmt.optimizeQuery("SELECT /*+ INDEX(\"t\" idx1) */ k FROM \"t\" WHERE v1 = 'foo' AND v2 = 'bar'");
+ assertEquals("IDX1", plan.getTableRef().getTable().getTableName().getString());
+ plan = stmt.optimizeQuery("SELECT k FROM \"t\" WHERE v1 = 'foo' AND v2 = 'bar'");
+ assertEquals("IDX2", plan.getTableRef().getTable().getTableName().getString());
+ }
+
+ @Test
+ public void testChooseIndexFromCaseSensitiveHint2() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+ conn.createStatement().execute("CREATE TABLE \"t\" (k INTEGER NOT NULL PRIMARY KEY, v1 VARCHAR, v2 VARCHAR) IMMUTABLE_ROWS=true");
+ conn.createStatement().execute("CREATE INDEX \"idx1\" ON \"t\"(v1) INCLUDE(v2)");
+ conn.createStatement().execute("CREATE INDEX \"idx2\" ON \"t\"(v1,v2)");
+ PhoenixStatement stmt = conn.createStatement().unwrap(PhoenixStatement.class);
+ QueryPlan plan = stmt.optimizeQuery("SELECT /*+ INDEX(\"t\" \"idx1\") */ k FROM \"t\" WHERE v1 = 'foo' AND v2 = 'bar'");
+ assertEquals("idx1", plan.getTableRef().getTable().getTableName().getString());
+ plan = stmt.optimizeQuery("SELECT k FROM \"t\" WHERE v1 = 'foo' AND v2 = 'bar'");
+ assertEquals("idx2", plan.getTableRef().getTable().getTableName().getString());
}
[15/30] phoenix git commit: Set version to 4.8.0-HBase-1.1-SNAPSHOT
after release
Posted by ma...@apache.org.
Set version to 4.8.0-HBase-1.1-SNAPSHOT after release
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/5977e1f7
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/5977e1f7
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/5977e1f7
Branch: refs/heads/calcite
Commit: 5977e1f7f7c6ba61e806e85ef1e61f8ef030d95d
Parents: d22ec4b
Author: Mujtaba <mu...@apache.org>
Authored: Tue Mar 8 16:12:11 2016 -0800
Committer: Mujtaba <mu...@apache.org>
Committed: Tue Mar 8 16:12:11 2016 -0800
----------------------------------------------------------------------
phoenix-assembly/pom.xml | 2 +-
phoenix-core/pom.xml | 2 +-
phoenix-flume/pom.xml | 2 +-
phoenix-pherf/pom.xml | 2 +-
phoenix-pig/pom.xml | 2 +-
phoenix-server-client/pom.xml | 2 +-
phoenix-server/pom.xml | 2 +-
phoenix-spark/pom.xml | 2 +-
phoenix-tracing-webapp/pom.xml | 2 +-
pom.xml | 2 +-
10 files changed, 10 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5977e1f7/phoenix-assembly/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-assembly/pom.xml b/phoenix-assembly/pom.xml
index a598e61..3e3e097 100644
--- a/phoenix-assembly/pom.xml
+++ b/phoenix-assembly/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.7.0-HBase-1.1</version>
+ <version>4.8.0-HBase-1.1-SNAPSHOT</version>
</parent>
<artifactId>phoenix-assembly</artifactId>
<name>Phoenix Assembly</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5977e1f7/phoenix-core/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-core/pom.xml b/phoenix-core/pom.xml
index 2b6d287..30a4875 100644
--- a/phoenix-core/pom.xml
+++ b/phoenix-core/pom.xml
@@ -4,7 +4,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.7.0-HBase-1.1</version>
+ <version>4.8.0-HBase-1.1-SNAPSHOT</version>
</parent>
<artifactId>phoenix-core</artifactId>
<name>Phoenix Core</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5977e1f7/phoenix-flume/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-flume/pom.xml b/phoenix-flume/pom.xml
index 15a921b..75bc76d 100644
--- a/phoenix-flume/pom.xml
+++ b/phoenix-flume/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.7.0-HBase-1.1</version>
+ <version>4.8.0-HBase-1.1-SNAPSHOT</version>
</parent>
<artifactId>phoenix-flume</artifactId>
<name>Phoenix - Flume</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5977e1f7/phoenix-pherf/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-pherf/pom.xml b/phoenix-pherf/pom.xml
index b94828a..3daf001 100644
--- a/phoenix-pherf/pom.xml
+++ b/phoenix-pherf/pom.xml
@@ -15,7 +15,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.7.0-HBase-1.1</version>
+ <version>4.8.0-HBase-1.1-SNAPSHOT</version>
</parent>
<artifactId>phoenix-pherf</artifactId>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5977e1f7/phoenix-pig/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-pig/pom.xml b/phoenix-pig/pom.xml
index fcaa489..c5f3cf0 100644
--- a/phoenix-pig/pom.xml
+++ b/phoenix-pig/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.7.0-HBase-1.1</version>
+ <version>4.8.0-HBase-1.1-SNAPSHOT</version>
</parent>
<artifactId>phoenix-pig</artifactId>
<name>Phoenix - Pig</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5977e1f7/phoenix-server-client/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-server-client/pom.xml b/phoenix-server-client/pom.xml
index 8bff95c..2e7a3b9 100644
--- a/phoenix-server-client/pom.xml
+++ b/phoenix-server-client/pom.xml
@@ -4,7 +4,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.7.0-HBase-1.1</version>
+ <version>4.8.0-HBase-1.1-SNAPSHOT</version>
</parent>
<artifactId>phoenix-server-client</artifactId>
<name>Phoenix Query Server Client</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5977e1f7/phoenix-server/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-server/pom.xml b/phoenix-server/pom.xml
index 36bf726..ad298dc 100644
--- a/phoenix-server/pom.xml
+++ b/phoenix-server/pom.xml
@@ -4,7 +4,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.7.0-HBase-1.1</version>
+ <version>4.8.0-HBase-1.1-SNAPSHOT</version>
</parent>
<artifactId>phoenix-server</artifactId>
<name>Phoenix Query Server</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5977e1f7/phoenix-spark/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-spark/pom.xml b/phoenix-spark/pom.xml
index 79a029a..219798b 100644
--- a/phoenix-spark/pom.xml
+++ b/phoenix-spark/pom.xml
@@ -28,7 +28,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.7.0-HBase-1.1</version>
+ <version>4.8.0-HBase-1.1-SNAPSHOT</version>
</parent>
<artifactId>phoenix-spark</artifactId>
<name>Phoenix - Spark</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5977e1f7/phoenix-tracing-webapp/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-tracing-webapp/pom.xml b/phoenix-tracing-webapp/pom.xml
index 63edaf4..438e6b4 100755
--- a/phoenix-tracing-webapp/pom.xml
+++ b/phoenix-tracing-webapp/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.7.0-HBase-1.1</version>
+ <version>4.8.0-HBase-1.1-SNAPSHOT</version>
</parent>
<artifactId>phoenix-tracing-webapp</artifactId>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5977e1f7/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 20aaf87..4021a0a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -3,7 +3,7 @@
<modelVersion>4.0.0</modelVersion>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.7.0-HBase-1.1</version>
+ <version>4.8.0-HBase-1.1-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Apache Phoenix</name>
<description>A SQL layer over HBase</description>
[17/30] phoenix git commit: PHOENIX-2763 rowcount stats not correct
after major compaction
Posted by ma...@apache.org.
PHOENIX-2763 rowcount stats not correct after major compaction
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/cc98e469
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/cc98e469
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/cc98e469
Branch: refs/heads/calcite
Commit: cc98e469d7bdb41c822fbf4f395d0c796f12be74
Parents: 5d33fba
Author: Ankit Singhal <an...@gmail.com>
Authored: Mon Mar 14 20:57:16 2016 +0530
Committer: Ankit Singhal <an...@gmail.com>
Committed: Mon Mar 14 20:57:16 2016 +0530
----------------------------------------------------------------------
.../org/apache/phoenix/end2end/StatsCollectorIT.java | 13 ++++++++++---
.../schema/stats/DefaultStatisticsCollector.java | 6 +++++-
2 files changed, 15 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc98e469/phoenix-core/src/it/java/org/apache/phoenix/end2end/StatsCollectorIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/StatsCollectorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/StatsCollectorIT.java
index bc575fd..2284ee2 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/StatsCollectorIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/StatsCollectorIT.java
@@ -39,6 +39,7 @@ import java.util.Properties;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.query.ConnectionQueryServices;
import org.apache.phoenix.query.KeyRange;
import org.apache.phoenix.query.QueryServices;
@@ -366,11 +367,13 @@ public class StatsCollectorIT extends StatsCollectorAbstractIT {
props.setProperty(QueryServices.MIN_STATS_UPDATE_FREQ_MS_ATTRIB, minStatsUpdateFreq.toString());
}
conn = DriverManager.getConnection(getUrl(), props);
- conn.createStatement().execute("CREATE TABLE " + tableName + "(k CHAR(1) PRIMARY KEY, v INTEGER) " + HColumnDescriptor.KEEP_DELETED_CELLS + "=" + Boolean.FALSE);
- stmt = conn.prepareStatement("UPSERT INTO " + tableName + " VALUES(?,?)");
+ conn.createStatement().execute("CREATE TABLE " + tableName + "(k CHAR(1) PRIMARY KEY, v INTEGER, w INTEGER) "
+ + HColumnDescriptor.KEEP_DELETED_CELLS + "=" + Boolean.FALSE);
+ stmt = conn.prepareStatement("UPSERT INTO " + tableName + " VALUES(?,?,?)");
for (int i = 0; i < nRows; i++) {
stmt.setString(1, Character.toString((char) ('a' + i)));
stmt.setInt(2, i);
+ stmt.setInt(3, i);
stmt.executeUpdate();
}
conn.commit();
@@ -392,7 +395,7 @@ public class StatsCollectorIT extends StatsCollectorAbstractIT {
List<KeyRange>keyRanges = getAllSplits(conn, tableName);
assertEquals(nRows+1, keyRanges.size());
- int nDeletedRows = conn.createStatement().executeUpdate("DELETE FROM " + tableName + " WHERE V < 5");
+ int nDeletedRows = conn.createStatement().executeUpdate("DELETE FROM " + tableName + " WHERE V < " + nRows / 2);
conn.commit();
assertEquals(5, nDeletedRows);
@@ -411,6 +414,10 @@ public class StatsCollectorIT extends StatsCollectorAbstractIT {
keyRanges = getAllSplits(conn, tableName);
}
assertEquals(nRows/2+1, keyRanges.size());
+ ResultSet rs = conn.createStatement().executeQuery("SELECT SUM(GUIDE_POSTS_ROW_COUNT) FROM "
+ + PhoenixDatabaseMetaData.SYSTEM_STATS_NAME + " WHERE PHYSICAL_NAME='" + tableName + "'");
+ rs.next();
+ assertEquals(nRows - nDeletedRows, rs.getLong(1));
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc98e469/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java
index b81d206..0db204d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java
@@ -166,6 +166,7 @@ class DefaultStatisticsCollector implements StatisticsCollector {
@Override
public void collectStatistics(final List<Cell> results) {
Map<ImmutableBytesPtr, Boolean> famMap = Maps.newHashMap();
+ boolean incrementRow = true;
for (Cell cell : results) {
KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
maxTimeStamp = Math.max(maxTimeStamp, kv.getTimestamp());
@@ -185,7 +186,10 @@ class DefaultStatisticsCollector implements StatisticsCollector {
}
} else {
gps = cachedGps;
- cachedGps.getSecond().incrementRowCount();
+ if (incrementRow) {
+ cachedGps.getSecond().incrementRowCount();
+ incrementRow = false;
+ }
}
int kvLength = kv.getLength();
long byteCount = gps.getFirst() + kvLength;
[11/30] phoenix git commit: PHOENIX-2100 Indicate in explain plan
when round robin iterator is being used
Posted by ma...@apache.org.
PHOENIX-2100 Indicate in explain plan when round robin iterator is being used
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/6e7acaa4
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/6e7acaa4
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/6e7acaa4
Branch: refs/heads/calcite
Commit: 6e7acaa4564e837dad3f54f79ac7a9b1e670d93a
Parents: 952a01c
Author: James Taylor <jt...@salesforce.com>
Authored: Wed Mar 2 15:27:05 2016 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Wed Mar 2 15:58:27 2016 -0800
----------------------------------------------------------------------
.../src/it/java/org/apache/phoenix/end2end/IndexToolIT.java | 6 +++---
.../it/java/org/apache/phoenix/end2end/MutableIndexToolIT.java | 2 +-
2 files changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/6e7acaa4/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
index fe95470..ef22094 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
@@ -144,7 +144,7 @@ public class IndexToolIT extends BaseOwnClusterHBaseManagedTimeIT {
String actualExplainPlan = QueryUtil.getExplainPlan(rs);
//assert we are pulling from data table.
- assertEquals(String.format("CLIENT 1-CHUNK PARALLEL 1-WAY FULL SCAN OVER %s", fullTableName), actualExplainPlan);
+ assertEquals(String.format("CLIENT 1-CHUNK PARALLEL 1-WAY ROUND ROBIN FULL SCAN OVER %s", fullTableName), actualExplainPlan);
rs = stmt1.executeQuery(selectSql);
assertTrue(rs.next());
@@ -205,10 +205,10 @@ public class IndexToolIT extends BaseOwnClusterHBaseManagedTimeIT {
String expectedExplainPlan = "";
if(isLocal) {
final String localIndexName = MetaDataUtil.getLocalIndexTableName(SchemaUtil.getTableName(schemaName, dataTable));
- expectedExplainPlan = String.format("CLIENT 1-CHUNK PARALLEL 1-WAY RANGE SCAN OVER %s [-32768]"
+ expectedExplainPlan = String.format("CLIENT 1-CHUNK PARALLEL 1-WAY ROUND ROBIN RANGE SCAN OVER %s [-32768]"
+ "\n SERVER FILTER BY FIRST KEY ONLY", localIndexName);
} else {
- expectedExplainPlan = String.format("CLIENT 1-CHUNK PARALLEL 1-WAY FULL SCAN OVER %s"
+ expectedExplainPlan = String.format("CLIENT 1-CHUNK PARALLEL 1-WAY ROUND ROBIN FULL SCAN OVER %s"
+ "\n SERVER FILTER BY FIRST KEY ONLY",SchemaUtil.getTableName(schemaName, indxTable));
}
assertEquals(expectedExplainPlan,actualExplainPlan);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/6e7acaa4/phoenix-core/src/it/java/org/apache/phoenix/end2end/MutableIndexToolIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MutableIndexToolIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MutableIndexToolIT.java
index 8125007..c335ff8 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MutableIndexToolIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MutableIndexToolIT.java
@@ -91,7 +91,7 @@ public class MutableIndexToolIT extends BaseOwnClusterHBaseManagedTimeIT {
String actualExplainPlan = QueryUtil.getExplainPlan(rs);
//assert we are pulling from data table.
- assertEquals(String.format("CLIENT 1-CHUNK PARALLEL 1-WAY FULL SCAN OVER %s",dataTable),actualExplainPlan);
+ assertEquals(String.format("CLIENT 1-CHUNK PARALLEL 1-WAY ROUND ROBIN FULL SCAN OVER %s",dataTable),actualExplainPlan);
rs = stmt1.executeQuery(selectSql);
assertTrue(rs.next());
[26/30] phoenix git commit: PHOENIX-2760 Upgrade phoenix-server and
phoenix-server-client to Avatica 1.7.1 (Josh Elser)
Posted by ma...@apache.org.
PHOENIX-2760 Upgrade phoenix-server and phoenix-server-client to Avatica 1.7.1 (Josh Elser)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/64fdf840
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/64fdf840
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/64fdf840
Branch: refs/heads/calcite
Commit: 64fdf840be576db1484d55c4079c81b03d365b4c
Parents: d4d973c
Author: James Taylor <jt...@salesforce.com>
Authored: Mon Mar 28 08:28:13 2016 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Mon Mar 28 08:28:13 2016 -0700
----------------------------------------------------------------------
phoenix-server-client/pom.xml | 4 ++--
phoenix-server/pom.xml | 12 ++++++------
phoenix-server/src/build/query-server-runnable.xml | 2 +-
pom.xml | 14 +++++++-------
4 files changed, 16 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/64fdf840/phoenix-server-client/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-server-client/pom.xml b/phoenix-server-client/pom.xml
index 2e7a3b9..1270c46 100644
--- a/phoenix-server-client/pom.xml
+++ b/phoenix-server-client/pom.xml
@@ -54,8 +54,8 @@
<dependencies>
<dependency>
- <groupId>org.apache.calcite</groupId>
- <artifactId>calcite-avatica</artifactId>
+ <groupId>org.apache.calcite.avatica</groupId>
+ <artifactId>avatica</artifactId>
</dependency>
<dependency>
<!-- a dependency for the thin-client uberjar -->
http://git-wip-us.apache.org/repos/asf/phoenix/blob/64fdf840/phoenix-server/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-server/pom.xml b/phoenix-server/pom.xml
index ad298dc..b5f6e1f 100644
--- a/phoenix-server/pom.xml
+++ b/phoenix-server/pom.xml
@@ -66,14 +66,14 @@
<artifactId>phoenix-server-client</artifactId>
</dependency>
<dependency>
- <groupId>org.apache.calcite</groupId>
- <artifactId>calcite-avatica</artifactId>
- <version>${calcite.version}</version>
+ <groupId>org.apache.calcite.avatica</groupId>
+ <artifactId>avatica</artifactId>
+ <version>${avatica.version}</version>
</dependency>
<dependency>
- <groupId>org.apache.calcite</groupId>
- <artifactId>calcite-avatica-server</artifactId>
- <version>${calcite.version}</version>
+ <groupId>org.apache.calcite.avatica</groupId>
+ <artifactId>avatica-server</artifactId>
+ <version>${avatica.version}</version>
</dependency>
<dependency>
<groupId>org.apache.phoenix</groupId>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/64fdf840/phoenix-server/src/build/query-server-runnable.xml
----------------------------------------------------------------------
diff --git a/phoenix-server/src/build/query-server-runnable.xml b/phoenix-server/src/build/query-server-runnable.xml
index d4259bf..db22142 100644
--- a/phoenix-server/src/build/query-server-runnable.xml
+++ b/phoenix-server/src/build/query-server-runnable.xml
@@ -45,7 +45,7 @@
<includes>
<include>org.apache.phoenix:phoenix-server</include>
<include>org.apache.phoenix:phoenix-server-client</include>
- <include>org.apache.calcite:*</include>
+ <include>org.apache.calcite.avatica:*</include>
</includes>
</dependencySet>
</dependencySets>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/64fdf840/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 0add065..389dcc0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -91,7 +91,7 @@
<!-- Do not change jodatime.version until HBASE-15199 is fixed -->
<jodatime.version>1.6</jodatime.version>
<joni.version>2.1.2</joni.version>
- <calcite.version>1.6.0</calcite.version>
+ <avatica.version>1.7.1</avatica.version>
<jettyVersion>8.1.7.v20120910</jettyVersion>
<tephra.version>0.7.0</tephra.version>
<spark.version>1.5.2</spark.version>
@@ -610,14 +610,14 @@
</exclusions>
</dependency>
<dependency>
- <groupId>org.apache.calcite</groupId>
- <artifactId>calcite-avatica</artifactId>
- <version>${calcite.version}</version>
+ <groupId>org.apache.calcite.avatica</groupId>
+ <artifactId>avatica</artifactId>
+ <version>${avatica.version}</version>
</dependency>
<dependency>
- <groupId>org.apache.calcite</groupId>
- <artifactId>calcite-avatica-server</artifactId>
- <version>${calcite.version}</version>
+ <groupId>org.apache.calcite.avatica</groupId>
+ <artifactId>avatica-server</artifactId>
+ <version>${avatica.version}</version>
</dependency>
<!-- Transaction dependencies -->
[09/30] phoenix git commit: PHOENIX-2100 Indicate in explain plan
when round robin iterator is being used
Posted by ma...@apache.org.
PHOENIX-2100 Indicate in explain plan when round robin iterator is being used
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/1d03e02f
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/1d03e02f
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/1d03e02f
Branch: refs/heads/calcite
Commit: 1d03e02f06dac97dafe6254bd8d9b75efcccd9a7
Parents: f9c8fe8
Author: Samarth <sa...@salesforce.com>
Authored: Wed Mar 2 13:33:27 2016 -0800
Committer: Samarth <sa...@salesforce.com>
Committed: Wed Mar 2 13:33:27 2016 -0800
----------------------------------------------------------------------
.../phoenix/iterate/BaseResultIterators.java | 7 ++++++
.../org/apache/phoenix/query/QueryPlanTest.java | 25 ++++++++++++++++++++
2 files changed, 32 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/1d03e02f/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
index 9f3b3e6..c36a72b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
@@ -882,6 +882,13 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
}
}
buf.append(getName()).append(" ").append(size()).append("-WAY ");
+ try {
+ if (plan.useRoundRobinIterator()) {
+ buf.append("ROUND ROBIN ");
+ }
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
explain(buf.toString(),planSteps);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/1d03e02f/phoenix-core/src/test/java/org/apache/phoenix/query/QueryPlanTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/query/QueryPlanTest.java b/phoenix-core/src/test/java/org/apache/phoenix/query/QueryPlanTest.java
index 4ba4d16..b00ccc0 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/query/QueryPlanTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/query/QueryPlanTest.java
@@ -252,5 +252,30 @@ public class QueryPlanTest extends BaseConnectionlessQueryTest {
conn.close();
}
}
+
+ @Test
+ public void testUseOfRoundRobinIteratorSurfaced() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(new Properties());
+ props.put(QueryServices.FORCE_ROW_KEY_ORDER_ATTRIB, Boolean.toString(false));
+ Connection conn = DriverManager.getConnection(getUrl(), props);
+ String tableName = "testUseOfRoundRobinIteratorSurfaced".toUpperCase();
+ try {
+ conn.createStatement().execute("CREATE TABLE " + tableName + "(\n" +
+ " a VARCHAR NOT NULL,\n" +
+ " b TIMESTAMP NOT NULL,\n" +
+ " c VARCHAR,\n" +
+ " CONSTRAINT pk PRIMARY KEY (a, b DESC, c)\n" +
+ " ) IMMUTABLE_ROWS=true\n" +
+ " ,SALT_BUCKETS=20");
+ String query = "select * from " + tableName + " where a = 'a' and b >= timestamp '2016-01-28 00:00:00' and b < timestamp '2016-01-29 00:00:00'";
+ ResultSet rs = conn.createStatement().executeQuery("EXPLAIN " + query);
+ String queryPlan = QueryUtil.getExplainPlan(rs);
+ assertEquals(
+ "CLIENT PARALLEL 20-WAY ROUND ROBIN RANGE SCAN OVER " + tableName + " [0,'a',~'2016-01-28 23:59:59.999'] - [0,'a',~'2016-01-28 00:00:00.000']\n" +
+ " SERVER FILTER BY FIRST KEY ONLY", queryPlan);
+ } finally {
+ conn.close();
+ }
+ }
}
[20/30] phoenix git commit: PHOENIX-2773 Remove unnecessary repos in
poms
Posted by ma...@apache.org.
PHOENIX-2773 Remove unnecessary repos in poms
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/e7bcfe4f
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/e7bcfe4f
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/e7bcfe4f
Branch: refs/heads/calcite
Commit: e7bcfe4f2b889575f7a20e0e2d9f76175282108c
Parents: 6eadadd
Author: James Taylor <jt...@salesforce.com>
Authored: Tue Mar 15 10:24:37 2016 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Tue Mar 15 15:56:03 2016 -0700
----------------------------------------------------------------------
pom.xml | 19 -------------------
1 file changed, 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/e7bcfe4f/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 4c87161..0add065 100644
--- a/pom.xml
+++ b/pom.xml
@@ -39,25 +39,6 @@
<id>apache release</id>
<url>https://repository.apache.org/content/repositories/releases/</url>
</repository>
- <repository>
- <id>conjars.org</id>
- <url>http://conjars.org/repo</url>
- </repository>
- <repository>
- <id>apache snapshot</id>
- <url>https://repository.apache.org/content/repositories/snapshots/</url>
- <snapshots>
- <enabled>true</enabled>
- </snapshots>
- </repository>
- <repository>
- <id>sonatype-nexus-snapshots</id>
- <name>Sonatype Nexus Snapshots</name>
- <url>https://oss.sonatype.org/content/repositories/snapshots</url>
- <snapshots>
- <enabled>true</enabled>
- </snapshots>
- </repository>
</repositories>
<parent>
[02/30] phoenix git commit: Revert "PHOENIX-1973 Improve
CsvBulkLoadTool performance by moving keyvalue construction from map phase to
reduce phase(Sergey Soldatov)"
Posted by ma...@apache.org.
Revert "PHOENIX-1973 Improve CsvBulkLoadTool performance by moving keyvalue construction from map phase to reduce phase(Sergey Soldatov)"
This reverts commit e797b36c2ce42e9b9fd6b37fd8b9f79f79d6f18f.
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/c083bc82
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/c083bc82
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/c083bc82
Branch: refs/heads/calcite
Commit: c083bc825e2f5fd99f7343a784ab5700dbcff6df
Parents: 4f74323
Author: James Taylor <jt...@salesforce.com>
Authored: Sat Feb 27 09:57:01 2016 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Sat Feb 27 09:57:01 2016 -0800
----------------------------------------------------------------------
.../phoenix/mapreduce/AbstractBulkLoadTool.java | 6 +-
.../mapreduce/FormatToKeyValueMapper.java | 164 +++----------------
.../mapreduce/FormatToKeyValueReducer.java | 127 ++------------
.../bulkload/TargetTableRefFunctions.java | 22 +--
4 files changed, 38 insertions(+), 281 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c083bc82/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
index ab2848f..39ee4b1 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
@@ -41,7 +41,6 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
@@ -269,7 +268,7 @@ public abstract class AbstractBulkLoadTool extends Configured implements Tool {
job.setInputFormatClass(TextInputFormat.class);
job.setMapOutputKeyClass(TableRowkeyPair.class);
- job.setMapOutputValueClass(ImmutableBytesWritable.class);
+ job.setMapOutputValueClass(KeyValue.class);
job.setOutputKeyClass(TableRowkeyPair.class);
job.setOutputValueClass(KeyValue.class);
job.setReducerClass(FormatToKeyValueReducer.class);
@@ -277,10 +276,7 @@ public abstract class AbstractBulkLoadTool extends Configured implements Tool {
MultiHfileOutputFormat.configureIncrementalLoad(job, tablesToBeLoaded);
final String tableNamesAsJson = TargetTableRefFunctions.NAMES_TO_JSON.apply(tablesToBeLoaded);
- final String logicalNamesAsJson = TargetTableRefFunctions.LOGICAN_NAMES_TO_JSON.apply(tablesToBeLoaded);
-
job.getConfiguration().set(FormatToKeyValueMapper.TABLE_NAMES_CONFKEY,tableNamesAsJson);
- job.getConfiguration().set(FormatToKeyValueMapper.LOGICAL_NAMES_CONFKEY,logicalNamesAsJson);
// give subclasses their hook
setupJob(job);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c083bc82/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
index 95b099e..7e115e5 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
@@ -17,30 +17,30 @@
*/
package org.apache.phoenix.mapreduce;
-import java.io.DataOutputStream;
import java.io.IOException;
import java.sql.SQLException;
-import java.util.*;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
import javax.annotation.Nullable;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.mapreduce.bulkload.TableRowkeyPair;
import org.apache.phoenix.mapreduce.bulkload.TargetTableRefFunctions;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
-import org.apache.phoenix.query.QueryConstants;
-import org.apache.phoenix.schema.PColumn;
-import org.apache.phoenix.schema.PTable;
-import org.apache.phoenix.util.*;
+import org.apache.phoenix.util.ColumnInfo;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.UpsertExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -59,7 +59,7 @@ import com.google.common.collect.Lists;
* to retrieve {@link ColumnInfo} from the target table.
*/
public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable, Text, TableRowkeyPair,
- ImmutableBytesWritable> {
+ KeyValue> {
protected static final Logger LOG = LoggerFactory.getLogger(FormatToKeyValueMapper.class);
@@ -79,7 +79,6 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
/** Configuration key for the table names */
public static final String TABLE_NAMES_CONFKEY = "phoenix.mapreduce.import.tablenames";
- public static final String LOGICAL_NAMES_CONFKEY = "phoenix.mapreduce.import.logicalnames";
/** Configuration key for the table configurations */
public static final String TABLE_CONFIG_CONFKEY = "phoenix.mapreduce.import.table.config";
@@ -95,14 +94,8 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
protected UpsertExecutor<RECORD, ?> upsertExecutor;
protected ImportPreUpsertKeyValueProcessor preUpdateProcessor;
protected List<String> tableNames;
- protected List<String> logicalNames;
protected MapperUpsertListener<RECORD> upsertListener;
- /*
- lookup table for column index. Index in the List matches to the index in tableNames List
- */
- protected List<Map<byte[], Map<byte[], Integer>>> columnIndexes;
-
protected abstract UpsertExecutor<RECORD,?> buildUpsertExecutor(Configuration conf);
protected abstract LineParser<RECORD> getLineParser();
@@ -119,17 +112,13 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
try {
conn = (PhoenixConnection) QueryUtil.getConnection(clientInfos, conf);
-
- final String tableNamesConf = conf.get(TABLE_NAMES_CONFKEY);
- final String logicalNamesConf = conf.get(LOGICAL_NAMES_CONFKEY);
- tableNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(tableNamesConf);
- logicalNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(logicalNamesConf);
-
- columnIndexes = initColumnIndexes();
} catch (SQLException | ClassNotFoundException e) {
throw new RuntimeException(e);
}
+ final String tableNamesConf = conf.get(TABLE_NAMES_CONFKEY);
+ tableNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(tableNamesConf);
+
upsertListener = new MapperUpsertListener<RECORD>(
context, conf.getBoolean(IGNORE_INVALID_ROW_CONFKEY, true));
upsertExecutor = buildUpsertExecutor(conf);
@@ -138,8 +127,7 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
@SuppressWarnings("deprecation")
@Override
- protected void map(LongWritable key, Text value, Context context) throws IOException,
- InterruptedException {
+ protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
if (conn == null) {
throw new RuntimeException("Connection not initialized.");
}
@@ -157,7 +145,7 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
return;
}
upsertExecutor.execute(ImmutableList.<RECORD>of(record));
- Map<Integer, List<KeyValue>> map = new HashMap<>();
+
Iterator<Pair<byte[], List<KeyValue>>> uncommittedDataIterator
= PhoenixRuntime.getUncommittedDataIterator(conn, true);
while (uncommittedDataIterator.hasNext()) {
@@ -165,125 +153,24 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
List<KeyValue> keyValueList = kvPair.getSecond();
keyValueList = preUpdateProcessor.preUpsert(kvPair.getFirst(), keyValueList);
byte[] first = kvPair.getFirst();
- // Create a list of KV for each table
- for (int i = 0; i < tableNames.size(); i++) {
- if (Bytes.compareTo(Bytes.toBytes(tableNames.get(i)), first) == 0) {
- if (!map.containsKey(i)) {
- map.put(i, new ArrayList<KeyValue>());
- }
- List<KeyValue> list = map.get(i);
- for (KeyValue kv : keyValueList) {
- list.add(kv);
- }
- break;
+ for (String tableName : tableNames) {
+ if (Bytes.compareTo(Bytes.toBytes(tableName), first) != 0) {
+ // skip edits for other tables
+ continue;
+ }
+ for (KeyValue kv : keyValueList) {
+ ImmutableBytesWritable outputKey = new ImmutableBytesWritable();
+ outputKey.set(kv.getRowArray(), kv.getRowOffset(), kv.getRowLength());
+ context.write(new TableRowkeyPair(tableName, outputKey), kv);
}
}
}
- for(Map.Entry<Integer, List<KeyValue>> rowEntry : map.entrySet()) {
- int tableIndex = rowEntry.getKey();
- List<KeyValue> lkv = rowEntry.getValue();
- // All KV values combines to a single byte array
- writeAggregatedRow(context, tableIndex, lkv);
- }
conn.rollback();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
- private List<Map<byte[],Map<byte[], Integer>>> initColumnIndexes() throws SQLException {
- List<Map<byte[], Map<byte[], Integer>>> tableMap = new ArrayList<>();
- int tableIndex;
- for (tableIndex = 0; tableIndex < tableNames.size(); tableIndex++) {
- PTable table = PhoenixRuntime.getTable(conn, logicalNames.get(tableIndex));
- Map<byte[], Map<byte[], Integer>> columnMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
- List<PColumn> cls = table.getColumns();
- for(int i = 0; i < cls.size(); i++) {
- PColumn c = cls.get(i);
- if(c.getFamilyName() == null) continue; // Skip PK column
- byte[] family = c.getFamilyName().getBytes();
- byte[] name = c.getName().getBytes();
- if(!columnMap.containsKey(family)) {
- columnMap.put(family, new TreeMap<byte[], Integer>(Bytes.BYTES_COMPARATOR));
- }
- Map<byte[], Integer> qualifier = columnMap.get(family);
- qualifier.put(name, i);
- }
- tableMap.add(columnMap);
- }
- return tableMap;
- }
-
- /**
- * Find the column index which will replace the column name in
- * the aggregated array and will be restored in Reducer
- *
- * @param tableIndex Table index in tableNames list
- * @param cell KeyValue for the column
- * @return column index for the specified cell or -1 if was not found
- */
- private int findIndex(int tableIndex, Cell cell) {
- Map<byte[], Map<byte[], Integer>> columnMap = columnIndexes.get(tableIndex);
- Map<byte[], Integer> qualifiers = columnMap.get(Bytes.copy(cell.getFamilyArray(),
- cell.getFamilyOffset(), cell.getFamilyLength()));
- if(qualifiers!= null) {
- Integer result = qualifiers.get(Bytes.copy(cell.getQualifierArray(),
- cell.getQualifierOffset(), cell.getQualifierLength()));
- if(result!=null) {
- return result;
- }
- }
- return -1;
- }
-
- /**
- * Collect all column values for the same rowKey
- *
- * @param context Current mapper context
- * @param tableIndex Table index in tableNames list
- * @param lkv List of KV values that will be combined in a single ImmutableBytesWritable
- * @throws IOException
- * @throws InterruptedException
- */
-
- private void writeAggregatedRow(Context context, int tableIndex, List<KeyValue> lkv)
- throws IOException, InterruptedException {
- TrustedByteArrayOutputStream bos = new TrustedByteArrayOutputStream(1024);
- DataOutputStream outputStream = new DataOutputStream(bos);
- ImmutableBytesWritable outputKey = new ImmutableBytesWritable();
- if (!lkv.isEmpty()) {
- // All Key Values for the same row are supposed to be the same, so init rowKey only once
- Cell first = lkv.get(0);
- outputKey.set(first.getRowArray(), first.getRowOffset(), first.getRowLength());
- for (KeyValue cell : lkv) {
- if(isEmptyCell(cell)) {
- continue;
- }
- int i = findIndex(tableIndex, cell);
- if (i == -1) {
- throw new IOException("No column found for KeyValue");
- }
- WritableUtils.writeVInt(outputStream, i);
- outputStream.writeByte(cell.getTypeByte());
- WritableUtils.writeVInt(outputStream, cell.getValueLength());
- outputStream.write(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
- }
- }
- ImmutableBytesWritable aggregatedArray = new ImmutableBytesWritable(bos.toByteArray());
- outputStream.close();
- context.write(new TableRowkeyPair(Integer.toString(tableIndex), outputKey), aggregatedArray);
- }
-
- protected boolean isEmptyCell(KeyValue cell) {
- if (Bytes.compareTo(cell.getQualifierArray(), cell.getQualifierOffset(),
- cell.getQualifierLength(), QueryConstants.EMPTY_COLUMN_BYTES, 0,
- QueryConstants.EMPTY_COLUMN_BYTES.length) != 0)
- return false;
- else
- return true;
- }
-
-
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
try {
@@ -336,12 +223,11 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
@VisibleForTesting
static class MapperUpsertListener<T> implements UpsertExecutor.UpsertListener<T> {
- private final Mapper<LongWritable, Text,
- TableRowkeyPair, ImmutableBytesWritable>.Context context;
+ private final Mapper<LongWritable, Text, TableRowkeyPair, KeyValue>.Context context;
private final boolean ignoreRecordErrors;
private MapperUpsertListener(
- Mapper<LongWritable, Text, TableRowkeyPair, ImmutableBytesWritable>.Context context,
+ Mapper<LongWritable, Text, TableRowkeyPair, KeyValue>.Context context,
boolean ignoreRecordErrors) {
this.context = context;
this.ignoreRecordErrors = ignoreRecordErrors;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c083bc82/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
index 0f90e45..5d00656 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
@@ -17,143 +17,36 @@
*/
package org.apache.phoenix.mapreduce;
-import java.io.ByteArrayInputStream;
-import java.io.DataInputStream;
import java.io.IOException;
-import java.sql.SQLException;
-import java.util.*;
+import java.util.TreeSet;
-import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.KeyValueSortReducer;
-import org.apache.hadoop.hbase.util.Pair;
-import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
-import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
-import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.mapreduce.bulkload.TableRowkeyPair;
-import org.apache.phoenix.mapreduce.bulkload.TargetTableRefFunctions;
-import org.apache.phoenix.query.QueryConstants;
-import org.apache.phoenix.schema.PColumn;
-import org.apache.phoenix.schema.PTable;
-import org.apache.phoenix.util.*;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* Reducer class for the bulkload jobs.
* Performs similar functionality to {@link KeyValueSortReducer}
*/
public class FormatToKeyValueReducer
- extends Reducer<TableRowkeyPair,ImmutableBytesWritable,TableRowkeyPair,KeyValue> {
-
- protected static final Logger LOG = LoggerFactory.getLogger(FormatToKeyValueReducer.class);
-
-
- protected List<String> tableNames;
- protected List<String> logicalNames;
- protected KeyValueBuilder builder;
- List<List<Pair<byte[], byte[]>>> columnIndexes;
- List<ImmutableBytesPtr> emptyFamilyName;
-
-
- @Override
- protected void setup(Context context) throws IOException, InterruptedException {
- Configuration conf = context.getConfiguration();
-
- // pass client configuration into driver
- Properties clientInfos = new Properties();
- for (Map.Entry<String, String> entry : conf) {
- clientInfos.setProperty(entry.getKey(), entry.getValue());
- }
-
- try {
- PhoenixConnection conn = (PhoenixConnection) QueryUtil.getConnection(clientInfos, conf);
- builder = conn.getKeyValueBuilder();
- final String tableNamesConf = conf.get(FormatToKeyValueMapper.TABLE_NAMES_CONFKEY);
- final String logicalNamesConf = conf.get(FormatToKeyValueMapper.LOGICAL_NAMES_CONFKEY);
- tableNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(tableNamesConf);
- logicalNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(logicalNamesConf);
-
- columnIndexes = new ArrayList<>(tableNames.size());
- emptyFamilyName = new ArrayList<>();
- initColumnsMap(conn);
- } catch (SQLException | ClassNotFoundException e) {
- throw new RuntimeException(e);
- }
- }
-
- private void initColumnsMap(PhoenixConnection conn) throws SQLException {
- for (String tableName : logicalNames) {
- PTable table = PhoenixRuntime.getTable(conn, tableName);
- emptyFamilyName.add(SchemaUtil.getEmptyColumnFamilyPtr(table));
- List<PColumn> cls = table.getColumns();
- List<Pair<byte[], byte[]>> list = new ArrayList(cls.size());
- for(int i = 0; i < cls.size(); i++) {
- PColumn c = cls.get(i);
- if(c.getFamilyName() == null) {
- list.add(null); // Skip PK column
- continue;
- }
- byte[] family = c.getFamilyName().getBytes();
- byte[] name = c.getName().getBytes();
- list.add(new Pair(family, name));
- }
- columnIndexes.add(list);
- }
-
- }
+ extends Reducer<TableRowkeyPair,KeyValue,TableRowkeyPair,KeyValue> {
@Override
- protected void reduce(TableRowkeyPair key, Iterable<ImmutableBytesWritable> values,
- Reducer<TableRowkeyPair, ImmutableBytesWritable, TableRowkeyPair, KeyValue>.Context context)
+ protected void reduce(TableRowkeyPair key, Iterable<KeyValue> values,
+ Reducer<TableRowkeyPair, KeyValue, TableRowkeyPair, KeyValue>.Context context)
throws IOException, InterruptedException {
TreeSet<KeyValue> map = new TreeSet<KeyValue>(KeyValue.COMPARATOR);
- int tableIndex = Integer.parseInt(key.getTableName());
- key.setTableName(tableNames.get(tableIndex));
- List<Pair<byte[], byte[]>> columns = columnIndexes.get(tableIndex);
- for (ImmutableBytesWritable aggregatedArray : values) {
- DataInputStream input = new DataInputStream(new ByteArrayInputStream(aggregatedArray.get()));
- while (input.available()!= 0) {
- int index = WritableUtils.readVInt(input);
- Pair<byte[], byte[]> pair = columns.get(index);
- byte type = input.readByte();
- ImmutableBytesWritable value = null;
- int len = WritableUtils.readVInt(input);
- if (len > 0) {
- byte[] array = new byte[len];
- input.read(array);
- value = new ImmutableBytesWritable(array);
- }
- KeyValue kv;
- KeyValue.Type kvType = KeyValue.Type.codeToType(type);
- switch (kvType) {
- case Put: // not null value
- kv = builder.buildPut(key.getRowkey(),
- new ImmutableBytesWritable(pair.getFirst()),
- new ImmutableBytesWritable(pair.getSecond()), value);
- break;
- case DeleteColumn: // null value
- kv = builder.buildDeleteColumns(key.getRowkey(),
- new ImmutableBytesWritable(pair.getFirst()),
- new ImmutableBytesWritable(pair.getSecond()));
- break;
- default:
- throw new IOException("Unsupported KeyValue type " + kvType);
- }
- map.add(kv);
+ for (KeyValue kv: values) {
+ try {
+ map.add(kv.clone());
+ } catch (CloneNotSupportedException e) {
+ throw new java.io.IOException(e);
}
- KeyValue empty = builder.buildPut(key.getRowkey(),
- emptyFamilyName.get(tableIndex),
- QueryConstants.EMPTY_COLUMN_BYTES_PTR, ByteUtil.EMPTY_BYTE_ARRAY_PTR);
- map.add(empty);
- Closeables.closeQuietly(input);
}
context.setStatus("Read " + map.getClass());
int index = 0;
- for (KeyValue kv : map) {
+ for (KeyValue kv: map) {
context.write(key, kv);
if (++index % 100 == 0) context.setStatus("Wrote " + index);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c083bc82/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
index e02065f..d786842 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
@@ -78,25 +78,7 @@ public class TargetTableRefFunctions {
}
};
- public static Function<List<TargetTableRef>,String> LOGICAN_NAMES_TO_JSON = new Function<List<TargetTableRef>,String>() {
-
- @Override
- public String apply(List<TargetTableRef> input) {
- try {
- List<String> tableNames = Lists.newArrayListWithCapacity(input.size());
- for(TargetTableRef table : input) {
- tableNames.add(table.getLogicalName());
- }
- ObjectMapper mapper = new ObjectMapper();
- return mapper.writeValueAsString(tableNames);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
-
- }
- };
-
- public static Function<String,List<String>> NAMES_FROM_JSON = new Function<String,List<String>>() {
+ public static Function<String,List<String>> NAMES_FROM_JSON = new Function<String,List<String>>() {
@SuppressWarnings("unchecked")
@Override
@@ -110,4 +92,4 @@ public class TargetTableRefFunctions {
}
};
-}
+ }
[04/30] phoenix git commit: PHOENIX-2730 Fix min hbase versions for
scanner lease renewal
Posted by ma...@apache.org.
PHOENIX-2730 Fix min hbase versions for scanner lease renewal
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/f97b4af8
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/f97b4af8
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/f97b4af8
Branch: refs/heads/calcite
Commit: f97b4af8982d2e9c1dbe56a14eb3a9cc90b84a98
Parents: 58cf5fa
Author: Samarth <sa...@salesforce.com>
Authored: Wed Mar 2 08:51:23 2016 -0800
Committer: Samarth <sa...@salesforce.com>
Committed: Wed Mar 2 08:51:23 2016 -0800
----------------------------------------------------------------------
.../main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f97b4af8/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index 26d57fa..67814e8 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -295,7 +295,7 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
private final ResultSet emptyResultSet;
public static final int MAX_LOCAL_SI_VERSION_DISALLOW = VersionUtil.encodeVersion("0", "98", "8");
public static final int MIN_LOCAL_SI_VERSION_DISALLOW = VersionUtil.encodeVersion("0", "98", "6");
- public static final int MIN_RENEW_LEASE_VERSION = VersionUtil.encodeVersion("1", "1", "4");
+ public static final int MIN_RENEW_LEASE_VERSION = VersionUtil.encodeVersion("1", "1", "3");
// Version below which we should turn off essential column family.
public static final int ESSENTIAL_FAMILY_VERSION_THRESHOLD = VersionUtil.encodeVersion("0", "94", "7");
[05/30] phoenix git commit: PHOENIX-2732 Update HBase versions in pom
Posted by ma...@apache.org.
PHOENIX-2732 Update HBase versions in pom
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/db4e5cda
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/db4e5cda
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/db4e5cda
Branch: refs/heads/calcite
Commit: db4e5cda53988da05f6ec28666cf664bd5dd43f4
Parents: f97b4af
Author: Samarth <sa...@salesforce.com>
Authored: Wed Mar 2 09:55:24 2016 -0800
Committer: Samarth <sa...@salesforce.com>
Committed: Wed Mar 2 09:55:24 2016 -0800
----------------------------------------------------------------------
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/db4e5cda/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index b27c3b9..d9e5ff9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -80,7 +80,7 @@
<top.dir>${project.basedir}</top.dir>
<!-- Hadoop Versions -->
- <hbase.version>1.1.2</hbase.version>
+ <hbase.version>1.1.3</hbase.version>
<hadoop-two.version>2.5.1</hadoop-two.version>
<!-- Dependency versions -->
[22/30] phoenix git commit: PHOENIX-2772 Upgrade pom to sqlline 1.1.19
Posted by ma...@apache.org.
PHOENIX-2772 Upgrade pom to sqlline 1.1.19
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/6eadadd2
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/6eadadd2
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/6eadadd2
Branch: refs/heads/calcite
Commit: 6eadadd2a4faa640fec4b141c357f7d4e2180dcc
Parents: d8b45bc
Author: James Taylor <jt...@salesforce.com>
Authored: Tue Mar 15 09:43:59 2016 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Tue Mar 15 15:56:03 2016 -0700
----------------------------------------------------------------------
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/6eadadd2/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 4021a0a..4c87161 100644
--- a/pom.xml
+++ b/pom.xml
@@ -97,7 +97,7 @@
<commons-lang.version>2.5</commons-lang.version>
<commons-logging.version>1.2</commons-logging.version>
<commons-csv.version>1.0</commons-csv.version>
- <sqlline.version>1.1.8</sqlline.version>
+ <sqlline.version>1.1.9</sqlline.version>
<guava.version>13.0.1</guava.version>
<flume.version>1.4.0</flume.version>
<findbugs.version>1.3.2</findbugs.version>
[23/30] phoenix git commit: PHOENIX-2746 Delete on the table with
immutable rows may fail with INVALID_FILTER_ON_IMMUTABLE_ROWS error
code.(Rajeshbabu)
Posted by ma...@apache.org.
PHOENIX-2746 Delete on the table with immutable rows may fail with INVALID_FILTER_ON_IMMUTABLE_ROWS error code.(Rajeshbabu)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/cd8e86ca
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/cd8e86ca
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/cd8e86ca
Branch: refs/heads/calcite
Commit: cd8e86ca7170876a30771fcc16c027f8dc8dd386
Parents: e7bcfe4
Author: Rajeshbabu Chintaguntla <ra...@apache.org>
Authored: Fri Mar 18 01:28:53 2016 +0530
Committer: Rajeshbabu Chintaguntla <ra...@apache.org>
Committed: Fri Mar 18 01:28:53 2016 +0530
----------------------------------------------------------------------
.../org/apache/phoenix/end2end/DeleteIT.java | 12 +++++++++
.../apache/phoenix/compile/DeleteCompiler.java | 28 +++++++++++---------
.../apache/phoenix/optimize/QueryOptimizer.java | 14 ++++++----
.../org/apache/phoenix/schema/PTableImpl.java | 2 +-
4 files changed, 37 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cd8e86ca/phoenix-core/src/it/java/org/apache/phoenix/end2end/DeleteIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/DeleteIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/DeleteIT.java
index 6b4eead..10152e3 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/DeleteIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/DeleteIT.java
@@ -19,6 +19,7 @@ package org.apache.phoenix.end2end;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import java.sql.Connection;
import java.sql.Date;
@@ -370,6 +371,7 @@ public class DeleteIT extends BaseHBaseManagedTimeIT {
"STATS.ACTIVE_VISITOR INTEGER " +
"CONSTRAINT PK PRIMARY KEY (HOST, DOMAIN, FEATURE, DATE)) IMMUTABLE_ROWS=true");
stm.execute("CREATE " + (localIndex ? "LOCAL" : "") + " INDEX web_stats_idx ON web_stats (DATE, FEATURE)");
+ stm.execute("CREATE " + (localIndex ? "LOCAL" : "") + " INDEX web_stats_idx2 ON web_stats (DATE, FEATURE, USAGE.DB)");
stm.close();
Date date = new Date(0);
@@ -406,6 +408,16 @@ public class DeleteIT extends BaseHBaseManagedTimeIT {
assertTrue(rs.next());
assertEquals(0, rs.getLong(1));
+ stm.execute("DROP INDEX web_stats_idx ON web_stats");
+ stm.execute("DROP INDEX web_stats_idx2 ON web_stats");
+
+ stm.execute("CREATE " + (localIndex ? "LOCAL" : "") + " INDEX web_stats_idx ON web_stats (USAGE.DB)");
+ stm.execute("CREATE " + (localIndex ? "LOCAL" : "") + " INDEX web_stats_idx2 ON web_stats (USAGE.DB, DATE)");
+ try{
+ psInsert = con.prepareStatement("DELETE FROM web_stats WHERE USAGE.DB=2");
+ } catch(Exception e) {
+ fail("There should not be any exception while deleting row");
+ }
} finally {
try {
con.close();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cd8e86ca/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java
index 8e9e1de..24a2add 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java
@@ -22,6 +22,7 @@ import java.sql.ParameterMetaData;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -233,18 +234,18 @@ public class DeleteCompiler {
}
- private Set<PTable> getNonDisabledImmutableIndexes(TableRef tableRef) {
+ private Map<PTableKey, PTable> getNonDisabledImmutableIndexes(TableRef tableRef) {
PTable table = tableRef.getTable();
if (table.isImmutableRows() && !table.getIndexes().isEmpty()) {
- Set<PTable> nonDisabledIndexes = Sets.newHashSetWithExpectedSize(table.getIndexes().size());
+ Map<PTableKey, PTable> nonDisabledIndexes = new HashMap<PTableKey, PTable>(table.getIndexes().size());
for (PTable index : table.getIndexes()) {
if (index.getIndexState() != PIndexState.DISABLE) {
- nonDisabledIndexes.add(index);
+ nonDisabledIndexes.put(index.getKey(), index);
}
}
return nonDisabledIndexes;
}
- return Collections.emptySet();
+ return Collections.emptyMap();
}
private class MultiDeleteMutationPlan implements MutationPlan {
@@ -312,8 +313,8 @@ public class DeleteCompiler {
boolean runOnServer = false;
SelectStatement select = null;
ColumnResolver resolverToBe = null;
- Set<PTable> immutableIndex = Collections.emptySet();
- DeletingParallelIteratorFactory parallelIteratorFactory = null;
+ Map<PTableKey, PTable> immutableIndex = Collections.emptyMap();
+ DeletingParallelIteratorFactory parallelIteratorFactory;
QueryPlan dataPlanToBe = null;
while (true) {
try {
@@ -406,7 +407,7 @@ public class DeleteCompiler {
PTable table = plan.getTableRef().getTable();
if (table.getType() == PTableType.INDEX) { // index plans
tableRefs[i++] = plan.getTableRef();
- immutableIndex.remove(table);
+ immutableIndex.remove(table.getKey());
} else { // data plan
/*
* If we have immutable indexes that we need to maintain, don't execute the data plan
@@ -586,12 +587,7 @@ public class DeleteCompiler {
});
} else {
final boolean deleteFromImmutableIndexToo = hasImmutableIndexes && !plan.getTableRef().equals(tableRef);
- if (parallelIteratorFactory != null) {
- parallelIteratorFactory.setRowProjector(plan.getProjector());
- parallelIteratorFactory.setTargetTableRef(tableRef);
- parallelIteratorFactory.setSourceTableRef(plan.getTableRef());
- parallelIteratorFactory.setIndexTargetTableRef(deleteFromImmutableIndexToo ? plan.getTableRef() : null);
- }
+ final DeletingParallelIteratorFactory parallelIteratorFactory2 = parallelIteratorFactory;
mutationPlans.add( new MutationPlan() {
@Override
public ParameterMetaData getParameterMetaData() {
@@ -625,6 +621,12 @@ public class DeleteCompiler {
if (!hasLimit) {
Tuple tuple;
long totalRowCount = 0;
+ if (parallelIteratorFactory2 != null) {
+ parallelIteratorFactory2.setRowProjector(plan.getProjector());
+ parallelIteratorFactory2.setTargetTableRef(tableRef);
+ parallelIteratorFactory2.setSourceTableRef(plan.getTableRef());
+ parallelIteratorFactory2.setIndexTargetTableRef(deleteFromImmutableIndexToo ? plan.getTableRef() : null);
+ }
while ((tuple=iterator.next()) != null) {// Runs query
Cell kv = tuple.getValue(0);
totalRowCount += PLong.INSTANCE.getCodec().decodeLong(kv.getValueArray(), kv.getValueOffset(), SortOrder.getDefault());
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cd8e86ca/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java b/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java
index adc3c2d..40591c2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java
@@ -153,7 +153,7 @@ public class QueryOptimizer {
}
}
- return hintedPlan == null ? orderPlansBestToWorst(select, plans) : plans;
+ return hintedPlan == null ? orderPlansBestToWorst(select, plans, stopAtBestPlan) : plans;
}
private static QueryPlan getHintedQueryPlan(PhoenixStatement statement, SelectStatement select, List<PTable> indexes, List<? extends PDatum> targetColumns, ParallelIteratorFactory parallelIteratorFactory, List<QueryPlan> plans) throws SQLException {
@@ -319,7 +319,7 @@ public class QueryOptimizer {
* @param plans the list of candidate plans
* @return list of plans ordered from best to worst.
*/
- private List<QueryPlan> orderPlansBestToWorst(SelectStatement select, List<QueryPlan> plans) {
+ private List<QueryPlan> orderPlansBestToWorst(SelectStatement select, List<QueryPlan> plans, boolean stopAtBestPlan) {
final QueryPlan dataPlan = plans.get(0);
if (plans.size() == 1) {
return plans;
@@ -330,10 +330,14 @@ public class QueryOptimizer {
* keys), then favor those first.
*/
List<QueryPlan> candidates = Lists.newArrayListWithExpectedSize(plans.size());
- for (QueryPlan plan : plans) {
- if (plan.getContext().getScanRanges().isPointLookup()) {
- candidates.add(plan);
+ if (stopAtBestPlan) { // If we're stopping at the best plan, only consider point lookups if there are any
+ for (QueryPlan plan : plans) {
+ if (plan.getContext().getScanRanges().isPointLookup()) {
+ candidates.add(plan);
+ }
}
+ } else {
+ candidates.addAll(plans);
}
/**
* If we have a plan(s) that removes the order by, choose from among these,
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cd8e86ca/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
index bb90c16..8165451 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
@@ -86,7 +86,7 @@ import com.sun.istack.NotNull;
* storing data in a single column (ColumnLayout.SINGLE) or in
* multiple columns (ColumnLayout.MULTI).
*
- *
+ * TODO add hashCode and equal methods to check equality of two PTableImpl objects.
* @since 0.1
*/
public class PTableImpl implements PTable {
[12/30] phoenix git commit: PHOENIX-2575 Prevent LocalIndexIT from
hanging indefinitely(Rajeshbabu)
Posted by ma...@apache.org.
PHOENIX-2575 Prevent LocalIndexIT from hanging indefinitely(Rajeshbabu)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/57c17ec9
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/57c17ec9
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/57c17ec9
Branch: refs/heads/calcite
Commit: 57c17ec9299d46c1e3ec44936a73747bbececf9b
Parents: 6e7acaa
Author: Rajeshbabu Chintaguntla <ra...@apache.org>
Authored: Fri Mar 4 00:53:32 2016 +0530
Committer: Rajeshbabu Chintaguntla <ra...@apache.org>
Committed: Fri Mar 4 00:53:32 2016 +0530
----------------------------------------------------------------------
.../hadoop/hbase/regionserver/IndexHalfStoreFileReader.java | 8 --------
1 file changed, 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/57c17ec9/phoenix-core/src/main/java/org/apache/hadoop/hbase/regionserver/IndexHalfStoreFileReader.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/hadoop/hbase/regionserver/IndexHalfStoreFileReader.java b/phoenix-core/src/main/java/org/apache/hadoop/hbase/regionserver/IndexHalfStoreFileReader.java
index 9befc8c..cbc4ed6 100644
--- a/phoenix-core/src/main/java/org/apache/hadoop/hbase/regionserver/IndexHalfStoreFileReader.java
+++ b/phoenix-core/src/main/java/org/apache/hadoop/hbase/regionserver/IndexHalfStoreFileReader.java
@@ -485,14 +485,6 @@ public class IndexHalfStoreFileReader extends StoreFile.Reader {
}
@Override
- public byte[] getLastKey() {
- // This method won't get used for the index region. There is no need to call
- // getClosestRowBefore on the index table. Also this is a split region. Can not be further
- // split
- throw new UnsupportedOperationException("Method is not implemented!");
- }
-
- @Override
public byte[] midkey() throws IOException {
// Returns null to indicate file is not splitable.
return null;
[07/30] phoenix git commit: PHOENIX-1973 Improve CsvBulkLoadTool
performance by moving keyvalue construction from map phase to reduce phase
(Sergey Soldatov)
Posted by ma...@apache.org.
PHOENIX-1973 Improve CsvBulkLoadTool performance by moving keyvalue construction from map phase to reduce phase (Sergey Soldatov)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/1c06c5db
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/1c06c5db
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/1c06c5db
Branch: refs/heads/calcite
Commit: 1c06c5dbd1d04b3dc67e75eb505df8d62f1a09e4
Parents: 42e2b85
Author: James Taylor <jt...@salesforce.com>
Authored: Wed Mar 2 00:38:05 2016 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Wed Mar 2 10:40:30 2016 -0800
----------------------------------------------------------------------
.../phoenix/mapreduce/AbstractBulkLoadTool.java | 6 +-
.../mapreduce/FormatToKeyValueMapper.java | 164 ++++++++++++++++---
.../mapreduce/FormatToKeyValueReducer.java | 126 ++++++++++++--
.../bulkload/TargetTableRefFunctions.java | 22 ++-
4 files changed, 280 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/1c06c5db/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
index 39ee4b1..ab2848f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
@@ -41,6 +41,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
@@ -268,7 +269,7 @@ public abstract class AbstractBulkLoadTool extends Configured implements Tool {
job.setInputFormatClass(TextInputFormat.class);
job.setMapOutputKeyClass(TableRowkeyPair.class);
- job.setMapOutputValueClass(KeyValue.class);
+ job.setMapOutputValueClass(ImmutableBytesWritable.class);
job.setOutputKeyClass(TableRowkeyPair.class);
job.setOutputValueClass(KeyValue.class);
job.setReducerClass(FormatToKeyValueReducer.class);
@@ -276,7 +277,10 @@ public abstract class AbstractBulkLoadTool extends Configured implements Tool {
MultiHfileOutputFormat.configureIncrementalLoad(job, tablesToBeLoaded);
final String tableNamesAsJson = TargetTableRefFunctions.NAMES_TO_JSON.apply(tablesToBeLoaded);
+ final String logicalNamesAsJson = TargetTableRefFunctions.LOGICAN_NAMES_TO_JSON.apply(tablesToBeLoaded);
+
job.getConfiguration().set(FormatToKeyValueMapper.TABLE_NAMES_CONFKEY,tableNamesAsJson);
+ job.getConfiguration().set(FormatToKeyValueMapper.LOGICAL_NAMES_CONFKEY,logicalNamesAsJson);
// give subclasses their hook
setupJob(job);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/1c06c5db/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
index 7e115e5..8dbb4aa 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
@@ -17,30 +17,30 @@
*/
package org.apache.phoenix.mapreduce;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.sql.SQLException;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
+import java.util.*;
import javax.annotation.Nullable;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.mapreduce.bulkload.TableRowkeyPair;
import org.apache.phoenix.mapreduce.bulkload.TargetTableRefFunctions;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
-import org.apache.phoenix.util.ColumnInfo;
-import org.apache.phoenix.util.PhoenixRuntime;
-import org.apache.phoenix.util.QueryUtil;
-import org.apache.phoenix.util.UpsertExecutor;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -59,7 +59,7 @@ import com.google.common.collect.Lists;
* to retrieve {@link ColumnInfo} from the target table.
*/
public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable, Text, TableRowkeyPair,
- KeyValue> {
+ ImmutableBytesWritable> {
protected static final Logger LOG = LoggerFactory.getLogger(FormatToKeyValueMapper.class);
@@ -79,6 +79,7 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
/** Configuration key for the table names */
public static final String TABLE_NAMES_CONFKEY = "phoenix.mapreduce.import.tablenames";
+ public static final String LOGICAL_NAMES_CONFKEY = "phoenix.mapreduce.import.logicalnames";
/** Configuration key for the table configurations */
public static final String TABLE_CONFIG_CONFKEY = "phoenix.mapreduce.import.table.config";
@@ -94,8 +95,14 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
protected UpsertExecutor<RECORD, ?> upsertExecutor;
protected ImportPreUpsertKeyValueProcessor preUpdateProcessor;
protected List<String> tableNames;
+ protected List<String> logicalNames;
protected MapperUpsertListener<RECORD> upsertListener;
+ /*
+ lookup table for column index. Index in the List matches to the index in tableNames List
+ */
+ protected List<Map<byte[], Map<byte[], Integer>>> columnIndexes;
+
protected abstract UpsertExecutor<RECORD,?> buildUpsertExecutor(Configuration conf);
protected abstract LineParser<RECORD> getLineParser();
@@ -112,13 +119,17 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
try {
conn = (PhoenixConnection) QueryUtil.getConnection(clientInfos, conf);
+
+ final String tableNamesConf = conf.get(TABLE_NAMES_CONFKEY);
+ final String logicalNamesConf = conf.get(LOGICAL_NAMES_CONFKEY);
+ tableNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(tableNamesConf);
+ logicalNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(logicalNamesConf);
+
+ columnIndexes = initColumnIndexes();
} catch (SQLException | ClassNotFoundException e) {
throw new RuntimeException(e);
}
- final String tableNamesConf = conf.get(TABLE_NAMES_CONFKEY);
- tableNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(tableNamesConf);
-
upsertListener = new MapperUpsertListener<RECORD>(
context, conf.getBoolean(IGNORE_INVALID_ROW_CONFKEY, true));
upsertExecutor = buildUpsertExecutor(conf);
@@ -127,7 +138,8 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
@SuppressWarnings("deprecation")
@Override
- protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
+ protected void map(LongWritable key, Text value, Context context) throws IOException,
+ InterruptedException {
if (conn == null) {
throw new RuntimeException("Connection not initialized.");
}
@@ -145,7 +157,7 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
return;
}
upsertExecutor.execute(ImmutableList.<RECORD>of(record));
-
+ Map<Integer, List<KeyValue>> map = new HashMap<>();
Iterator<Pair<byte[], List<KeyValue>>> uncommittedDataIterator
= PhoenixRuntime.getUncommittedDataIterator(conn, true);
while (uncommittedDataIterator.hasNext()) {
@@ -153,24 +165,125 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
List<KeyValue> keyValueList = kvPair.getSecond();
keyValueList = preUpdateProcessor.preUpsert(kvPair.getFirst(), keyValueList);
byte[] first = kvPair.getFirst();
- for (String tableName : tableNames) {
- if (Bytes.compareTo(Bytes.toBytes(tableName), first) != 0) {
- // skip edits for other tables
- continue;
- }
- for (KeyValue kv : keyValueList) {
- ImmutableBytesWritable outputKey = new ImmutableBytesWritable();
- outputKey.set(kv.getRowArray(), kv.getRowOffset(), kv.getRowLength());
- context.write(new TableRowkeyPair(tableName, outputKey), kv);
+ // Create a list of KV for each table
+ for (int i = 0; i < tableNames.size(); i++) {
+ if (Bytes.compareTo(Bytes.toBytes(tableNames.get(i)), first) == 0) {
+ if (!map.containsKey(i)) {
+ map.put(i, new ArrayList<KeyValue>());
+ }
+ List<KeyValue> list = map.get(i);
+ for (KeyValue kv : keyValueList) {
+ list.add(kv);
+ }
+ break;
}
}
}
+ for(Map.Entry<Integer, List<KeyValue>> rowEntry : map.entrySet()) {
+ int tableIndex = rowEntry.getKey();
+ List<KeyValue> lkv = rowEntry.getValue();
+ // All KV values combines to a single byte array
+ writeAggregatedRow(context, tableIndex, lkv);
+ }
conn.rollback();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
+ private List<Map<byte[],Map<byte[], Integer>>> initColumnIndexes() throws SQLException {
+ List<Map<byte[], Map<byte[], Integer>>> tableMap = new ArrayList<>();
+ int tableIndex;
+ for (tableIndex = 0; tableIndex < tableNames.size(); tableIndex++) {
+ PTable table = PhoenixRuntime.getTable(conn, logicalNames.get(tableIndex));
+ Map<byte[], Map<byte[], Integer>> columnMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
+ List<PColumn> cls = table.getColumns();
+ for(int i = 0; i < cls.size(); i++) {
+ PColumn c = cls.get(i);
+ if(c.getFamilyName() == null) continue; // Skip PK column
+ byte[] family = c.getFamilyName().getBytes();
+ byte[] name = c.getName().getBytes();
+ if(!columnMap.containsKey(family)) {
+ columnMap.put(family, new TreeMap<byte[], Integer>(Bytes.BYTES_COMPARATOR));
+ }
+ Map<byte[], Integer> qualifier = columnMap.get(family);
+ qualifier.put(name, i);
+ }
+ tableMap.add(columnMap);
+ }
+ return tableMap;
+ }
+
+ /**
+ * Find the column index which will replace the column name in
+ * the aggregated array and will be restored in Reducer
+ *
+ * @param tableIndex Table index in tableNames list
+ * @param cell KeyValue for the column
+ * @return column index for the specified cell or -1 if was not found
+ */
+ private int findIndex(int tableIndex, Cell cell) {
+ Map<byte[], Map<byte[], Integer>> columnMap = columnIndexes.get(tableIndex);
+ Map<byte[], Integer> qualifiers = columnMap.get(Bytes.copy(cell.getFamilyArray(),
+ cell.getFamilyOffset(), cell.getFamilyLength()));
+ if(qualifiers!= null) {
+ Integer result = qualifiers.get(Bytes.copy(cell.getQualifierArray(),
+ cell.getQualifierOffset(), cell.getQualifierLength()));
+ if(result!=null) {
+ return result;
+ }
+ }
+ return -1;
+ }
+
+ /**
+ * Collect all column values for the same rowKey
+ *
+ * @param context Current mapper context
+ * @param tableIndex Table index in tableNames list
+ * @param lkv List of KV values that will be combined in a single ImmutableBytesWritable
+ * @throws IOException
+ * @throws InterruptedException
+ */
+
+ private void writeAggregatedRow(Context context, int tableIndex, List<KeyValue> lkv)
+ throws IOException, InterruptedException {
+ TrustedByteArrayOutputStream bos = new TrustedByteArrayOutputStream(1024);
+ DataOutputStream outputStream = new DataOutputStream(bos);
+ ImmutableBytesWritable outputKey = new ImmutableBytesWritable();
+ if (!lkv.isEmpty()) {
+ // All Key Values for the same row are supposed to be the same, so init rowKey only once
+ Cell first = lkv.get(0);
+ outputKey.set(first.getRowArray(), first.getRowOffset(), first.getRowLength());
+ for (KeyValue cell : lkv) {
+ if(isEmptyCell(cell)) {
+ continue;
+ }
+ int i = findIndex(tableIndex, cell);
+ if (i == -1) {
+ throw new IOException("No column found for KeyValue");
+ }
+ WritableUtils.writeVInt(outputStream, i);
+ outputStream.writeByte(cell.getTypeByte());
+ WritableUtils.writeVInt(outputStream, cell.getValueLength());
+ outputStream.write(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
+ }
+ }
+ ImmutableBytesWritable aggregatedArray = new ImmutableBytesWritable(bos.toByteArray());
+ outputStream.close();
+ context.write(new TableRowkeyPair(tableNames.get(tableIndex), outputKey), aggregatedArray);
+ }
+
+ protected boolean isEmptyCell(KeyValue cell) {
+ if (Bytes.compareTo(cell.getQualifierArray(), cell.getQualifierOffset(),
+ cell.getQualifierLength(), QueryConstants.EMPTY_COLUMN_BYTES, 0,
+ QueryConstants.EMPTY_COLUMN_BYTES.length) != 0)
+ return false;
+ else
+ return true;
+ }
+
+
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
try {
@@ -223,11 +336,12 @@ public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable
@VisibleForTesting
static class MapperUpsertListener<T> implements UpsertExecutor.UpsertListener<T> {
- private final Mapper<LongWritable, Text, TableRowkeyPair, KeyValue>.Context context;
+ private final Mapper<LongWritable, Text,
+ TableRowkeyPair, ImmutableBytesWritable>.Context context;
private final boolean ignoreRecordErrors;
private MapperUpsertListener(
- Mapper<LongWritable, Text, TableRowkeyPair, KeyValue>.Context context,
+ Mapper<LongWritable, Text, TableRowkeyPair, ImmutableBytesWritable>.Context context,
boolean ignoreRecordErrors) {
this.context = context;
this.ignoreRecordErrors = ignoreRecordErrors;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/1c06c5db/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
index 5d00656..fb61855 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
@@ -17,36 +17,142 @@
*/
package org.apache.phoenix.mapreduce;
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
import java.io.IOException;
-import java.util.TreeSet;
+import java.sql.SQLException;
+import java.util.*;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.KeyValueSortReducer;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
+import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.mapreduce.bulkload.TableRowkeyPair;
+import org.apache.phoenix.mapreduce.bulkload.TargetTableRefFunctions;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.util.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Reducer class for the bulkload jobs.
* Performs similar functionality to {@link KeyValueSortReducer}
*/
public class FormatToKeyValueReducer
- extends Reducer<TableRowkeyPair,KeyValue,TableRowkeyPair,KeyValue> {
+ extends Reducer<TableRowkeyPair,ImmutableBytesWritable,TableRowkeyPair,KeyValue> {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(FormatToKeyValueReducer.class);
+
+
+ protected List<String> tableNames;
+ protected List<String> logicalNames;
+ protected KeyValueBuilder builder;
+ List<List<Pair<byte[], byte[]>>> columnIndexes;
+ List<ImmutableBytesPtr> emptyFamilyName;
+
+
+ @Override
+ protected void setup(Context context) throws IOException, InterruptedException {
+ Configuration conf = context.getConfiguration();
+
+ // pass client configuration into driver
+ Properties clientInfos = new Properties();
+ for (Map.Entry<String, String> entry : conf) {
+ clientInfos.setProperty(entry.getKey(), entry.getValue());
+ }
+
+ try {
+ PhoenixConnection conn = (PhoenixConnection) QueryUtil.getConnection(clientInfos, conf);
+ builder = conn.getKeyValueBuilder();
+ final String tableNamesConf = conf.get(FormatToKeyValueMapper.TABLE_NAMES_CONFKEY);
+ final String logicalNamesConf = conf.get(FormatToKeyValueMapper.LOGICAL_NAMES_CONFKEY);
+ tableNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(tableNamesConf);
+ logicalNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(logicalNamesConf);
+
+ columnIndexes = new ArrayList<>(tableNames.size());
+ emptyFamilyName = new ArrayList<>();
+ initColumnsMap(conn);
+ } catch (SQLException | ClassNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void initColumnsMap(PhoenixConnection conn) throws SQLException {
+ for (String tableName : logicalNames) {
+ PTable table = PhoenixRuntime.getTable(conn, tableName);
+ emptyFamilyName.add(SchemaUtil.getEmptyColumnFamilyPtr(table));
+ List<PColumn> cls = table.getColumns();
+ List<Pair<byte[], byte[]>> list = new ArrayList(cls.size());
+ for(int i = 0; i < cls.size(); i++) {
+ PColumn c = cls.get(i);
+ if(c.getFamilyName() == null) {
+ list.add(null); // Skip PK column
+ continue;
+ }
+ byte[] family = c.getFamilyName().getBytes();
+ byte[] name = c.getName().getBytes();
+ list.add(new Pair(family, name));
+ }
+ columnIndexes.add(list);
+ }
+
+ }
@Override
- protected void reduce(TableRowkeyPair key, Iterable<KeyValue> values,
- Reducer<TableRowkeyPair, KeyValue, TableRowkeyPair, KeyValue>.Context context)
+ protected void reduce(TableRowkeyPair key, Iterable<ImmutableBytesWritable> values,
+ Reducer<TableRowkeyPair, ImmutableBytesWritable, TableRowkeyPair, KeyValue>.Context context)
throws IOException, InterruptedException {
TreeSet<KeyValue> map = new TreeSet<KeyValue>(KeyValue.COMPARATOR);
- for (KeyValue kv: values) {
- try {
- map.add(kv.clone());
- } catch (CloneNotSupportedException e) {
- throw new java.io.IOException(e);
+ int tableIndex = tableNames.indexOf(key.getTableName());
+ List<Pair<byte[], byte[]>> columns = columnIndexes.get(tableIndex);
+ for (ImmutableBytesWritable aggregatedArray : values) {
+ DataInputStream input = new DataInputStream(new ByteArrayInputStream(aggregatedArray.get()));
+ while (input.available()!= 0) {
+ int index = WritableUtils.readVInt(input);
+ Pair<byte[], byte[]> pair = columns.get(index);
+ byte type = input.readByte();
+ ImmutableBytesWritable value = null;
+ int len = WritableUtils.readVInt(input);
+ if (len > 0) {
+ byte[] array = new byte[len];
+ input.read(array);
+ value = new ImmutableBytesWritable(array);
+ }
+ KeyValue kv;
+ KeyValue.Type kvType = KeyValue.Type.codeToType(type);
+ switch (kvType) {
+ case Put: // not null value
+ kv = builder.buildPut(key.getRowkey(),
+ new ImmutableBytesWritable(pair.getFirst()),
+ new ImmutableBytesWritable(pair.getSecond()), value);
+ break;
+ case DeleteColumn: // null value
+ kv = builder.buildDeleteColumns(key.getRowkey(),
+ new ImmutableBytesWritable(pair.getFirst()),
+ new ImmutableBytesWritable(pair.getSecond()));
+ break;
+ default:
+ throw new IOException("Unsupported KeyValue type " + kvType);
+ }
+ map.add(kv);
}
+ KeyValue empty = builder.buildPut(key.getRowkey(),
+ emptyFamilyName.get(tableIndex),
+ QueryConstants.EMPTY_COLUMN_BYTES_PTR, ByteUtil.EMPTY_BYTE_ARRAY_PTR);
+ map.add(empty);
+ Closeables.closeQuietly(input);
}
context.setStatus("Read " + map.getClass());
int index = 0;
- for (KeyValue kv: map) {
+ for (KeyValue kv : map) {
context.write(key, kv);
if (++index % 100 == 0) context.setStatus("Wrote " + index);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/1c06c5db/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
index d786842..e02065f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
@@ -78,7 +78,25 @@ public class TargetTableRefFunctions {
}
};
- public static Function<String,List<String>> NAMES_FROM_JSON = new Function<String,List<String>>() {
+ public static Function<List<TargetTableRef>,String> LOGICAN_NAMES_TO_JSON = new Function<List<TargetTableRef>,String>() {
+
+ @Override
+ public String apply(List<TargetTableRef> input) {
+ try {
+ List<String> tableNames = Lists.newArrayListWithCapacity(input.size());
+ for(TargetTableRef table : input) {
+ tableNames.add(table.getLogicalName());
+ }
+ ObjectMapper mapper = new ObjectMapper();
+ return mapper.writeValueAsString(tableNames);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ }
+ };
+
+ public static Function<String,List<String>> NAMES_FROM_JSON = new Function<String,List<String>>() {
@SuppressWarnings("unchecked")
@Override
@@ -92,4 +110,4 @@ public class TargetTableRefFunctions {
}
};
- }
+}
[16/30] phoenix git commit: PHOENIX-2748 Disable auto-commit during
bulk load
Posted by ma...@apache.org.
PHOENIX-2748 Disable auto-commit during bulk load
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/5d33fba6
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/5d33fba6
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/5d33fba6
Branch: refs/heads/calcite
Commit: 5d33fba633bdc5b2cbc31fa1736f56dce1e17ac6
Parents: 5977e1f
Author: Gabriel Reid <ga...@ngdata.com>
Authored: Fri Mar 11 15:13:58 2016 +0100
Committer: Gabriel Reid <ga...@ngdata.com>
Committed: Fri Mar 11 20:46:04 2016 +0100
----------------------------------------------------------------------
.../org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java | 3 +++
1 file changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5d33fba6/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java
index 2c9b6d9..1211b83 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java
@@ -126,6 +126,9 @@ public abstract class FormatToBytesWritableMapper<RECORD> extends Mapper<LongWri
try {
conn = (PhoenixConnection) QueryUtil.getConnection(clientInfos, conf);
+ // We are dependent on rolling back before performing commits, so we need to be sure
+ // that auto-commit is not turned on
+ conn.setAutoCommit(false);
final String tableNamesConf = conf.get(TABLE_NAMES_CONFKEY);
final String logicalNamesConf = conf.get(LOGICAL_NAMES_CONFKEY);
[14/30] phoenix git commit: PHOENIX-2749 Upgrade Apache Commons
Collections to v3.2.2 (Justine Tunney)
Posted by ma...@apache.org.
PHOENIX-2749 Upgrade Apache Commons Collections to v3.2.2 (Justine Tunney)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/d22ec4ba
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/d22ec4ba
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/d22ec4ba
Branch: refs/heads/calcite
Commit: d22ec4ba1ce9446bbd3d50a2c88dde18b6d63a24
Parents: 9815e16
Author: James Taylor <jt...@salesforce.com>
Authored: Sat Mar 5 16:47:08 2016 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Sat Mar 5 16:47:08 2016 -0800
----------------------------------------------------------------------
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d22ec4ba/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 9cb83ff..20aaf87 100644
--- a/pom.xml
+++ b/pom.xml
@@ -106,7 +106,7 @@
<netty.version>4.0.23.Final</netty.version>
<commons-codec.version>1.7</commons-codec.version>
<htrace.version>3.1.0-incubating</htrace.version>
- <collections.version>3.2.1</collections.version>
+ <collections.version>3.2.2</collections.version>
<!-- Do not change jodatime.version until HBASE-15199 is fixed -->
<jodatime.version>1.6</jodatime.version>
<joni.version>2.1.2</joni.version>
[25/30] phoenix git commit: PHOENIX-2734 Literal expressions for
UNSIGNED_DATE/UNSIGNED_TIME/etc (Sergey Soldatov)
Posted by ma...@apache.org.
PHOENIX-2734 Literal expressions for UNSIGNED_DATE/UNSIGNED_TIME/etc (Sergey Soldatov)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/d4d973c7
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/d4d973c7
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/d4d973c7
Branch: refs/heads/calcite
Commit: d4d973c7b9fd5bb6bd206c9082c4c6d53e54c59b
Parents: bea110c
Author: James Taylor <jt...@salesforce.com>
Authored: Mon Mar 28 08:22:37 2016 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Mon Mar 28 08:22:37 2016 -0700
----------------------------------------------------------------------
.../org/apache/phoenix/end2end/DateTimeIT.java | 20 ++++++++++++++++++++
.../phoenix/expression/LiteralExpression.java | 12 ++++--------
2 files changed, 24 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d4d973c7/phoenix-core/src/it/java/org/apache/phoenix/end2end/DateTimeIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/DateTimeIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/DateTimeIT.java
index e87d8d4..af8f0c1 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/DateTimeIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/DateTimeIT.java
@@ -513,6 +513,26 @@ public class DateTimeIT extends BaseHBaseManagedTimeIT {
}
@Test
+ public void testUnsignedTimeDateWithLiteral() throws Exception {
+ String ddl =
+ "CREATE TABLE IF NOT EXISTS UT (k1 INTEGER NOT NULL," +
+ "unsignedDates UNSIGNED_DATE, unsignedTimestamps UNSIGNED_TIMESTAMP, unsignedTimes UNSIGNED_TIME CONSTRAINT pk PRIMARY KEY (k1))";
+ conn.createStatement().execute(ddl);
+ String dml = "UPSERT INTO UT VALUES (1, " +
+ "'2010-06-20 12:00:00', '2012-07-28 12:00:00', '2015-12-25 12:00:00')";
+ conn.createStatement().execute(dml);
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery("SELECT k1, unsignedDates, " +
+ "unsignedTimestamps, unsignedTimes FROM UT where k1 = 1");
+ assertTrue(rs.next());
+ assertEquals(DateUtil.parseDate("2010-06-20 12:00:00"), rs.getDate(2));
+ assertEquals(DateUtil.parseTimestamp("2012-07-28 12:00:00"), rs.getTimestamp(3));
+ assertEquals(DateUtil.parseTime("2015-12-25 12:00:00"), rs.getTime(4));
+ assertFalse(rs.next());
+ }
+
+ @Test
public void testSecondFuncAgainstColumns() throws Exception {
String ddl =
"CREATE TABLE IF NOT EXISTS T1 (k1 INTEGER NOT NULL, dates DATE, timestamps TIMESTAMP, times TIME, " +
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d4d973c7/phoenix-core/src/main/java/org/apache/phoenix/expression/LiteralExpression.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/LiteralExpression.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/LiteralExpression.java
index ad1c7c0..90882a2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/LiteralExpression.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/LiteralExpression.java
@@ -25,6 +25,7 @@ import java.sql.SQLException;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.io.WritableUtils;
import org.apache.phoenix.expression.visitor.ExpressionVisitor;
+import org.apache.phoenix.schema.IllegalDataException;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.TypeMismatchException;
import org.apache.phoenix.schema.tuple.Tuple;
@@ -182,16 +183,11 @@ public class LiteralExpression extends BaseTerminalExpression {
return getBooleanLiteralExpression((Boolean)value, determinism);
}
PDataType actualType = PDataType.fromLiteral(value);
- // For array we should check individual element in it?
- // It would be costly though!!!!!
- // UpsertStatement can try to cast varchar to date type but PVarchar can't CoercibleTo Date or Timestamp
- // otherwise TO_NUMBER like functions will fail
- if (!actualType.isCoercibleTo(type, value) &&
- (!actualType.equals(PVarchar.INSTANCE) ||
- !(type.equals(PDate.INSTANCE) || type.equals(PTimestamp.INSTANCE) || type.equals(PTime.INSTANCE)))) {
+ try {
+ value = type.toObject(value, actualType);
+ } catch (IllegalDataException e) {
throw TypeMismatchException.newException(type, actualType, value.toString());
}
- value = type.toObject(value, actualType);
byte[] b = type.isArrayType() ? ((PArrayDataType)type).toBytes(value, PArrayDataType.arrayBaseType(type), sortOrder, rowKeyOrderOptimizable) :
type.toBytes(value, sortOrder);
if (type == PVarchar.INSTANCE || type == PChar.INSTANCE) {
[13/30] phoenix git commit: PHOENIX-2465 upgrade ANTLR version to
correct generation error with Java 8 (Gabor Liptak)
Posted by ma...@apache.org.
PHOENIX-2465 upgrade ANTLR version to correct generation error with Java 8 (Gabor Liptak)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/9815e16e
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/9815e16e
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/9815e16e
Branch: refs/heads/calcite
Commit: 9815e16e2e067aa9372d212d4fe9337b5473e5a1
Parents: 57c17ec
Author: James Taylor <jt...@salesforce.com>
Authored: Sat Mar 5 16:39:29 2016 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Sat Mar 5 16:39:29 2016 -0800
----------------------------------------------------------------------
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9815e16e/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index d9e5ff9..9cb83ff 100644
--- a/pom.xml
+++ b/pom.xml
@@ -88,7 +88,7 @@
<hadoop.version>2.5.1</hadoop.version>
<pig.version>0.13.0</pig.version>
<jackson.version>1.9.2</jackson.version>
- <antlr.version>3.5</antlr.version>
+ <antlr.version>3.5.2</antlr.version>
<log4j.version>1.2.17</log4j.version>
<slf4j.version>1.6.4</slf4j.version>
<protobuf-java.version>2.5.0</protobuf-java.version>
[29/30] phoenix git commit: Merge remote-tracking branch
'origin/master' into calcite
Posted by ma...@apache.org.
Merge remote-tracking branch 'origin/master' into calcite
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/c55f74de
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/c55f74de
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/c55f74de
Branch: refs/heads/calcite
Commit: c55f74dedd73142e0c585742a2c1e913ad43fac5
Parents: 86b9ded b988050
Author: maryannxue <ma...@gmail.com>
Authored: Tue Mar 29 14:51:43 2016 -0400
Committer: maryannxue <ma...@gmail.com>
Committed: Tue Mar 29 14:51:43 2016 -0400
----------------------------------------------------------------------
phoenix-assembly/pom.xml | 2 +-
phoenix-core/pom.xml | 2 +-
.../org/apache/phoenix/end2end/DateTimeIT.java | 20 +
.../org/apache/phoenix/end2end/DeleteIT.java | 12 +
.../org/apache/phoenix/end2end/IndexToolIT.java | 6 +-
.../phoenix/end2end/MutableIndexToolIT.java | 2 +-
.../phoenix/end2end/PhoenixRuntimeIT.java | 13 +-
.../phoenix/end2end/StatsCollectorIT.java | 13 +-
.../phoenix/end2end/ToCharFunctionIT.java | 15 +
.../phoenix/end2end/ToDateFunctionIT.java | 15 +
.../phoenix/end2end/index/LocalIndexIT.java | 2 +-
.../phoenix/end2end/index/ViewIndexIT.java | 2 +-
.../ipc/controller/MetadataRpcController.java | 2 +-
.../regionserver/IndexHalfStoreFileReader.java | 8 -
.../apache/phoenix/compile/DeleteCompiler.java | 28 +-
.../phoenix/coprocessor/MetaDataProtocol.java | 2 +-
.../coprocessor/SequenceRegionObserver.java | 12 +-
.../phoenix/expression/LiteralExpression.java | 12 +-
.../expression/function/ToCharFunction.java | 13 +-
.../expression/function/ToDateFunction.java | 17 +-
.../expression/function/ToNumberFunction.java | 14 +-
.../expression/function/ToTimeFunction.java | 13 +-
.../function/ToTimestampFunction.java | 13 +-
.../hbase/index/covered/KeyValueStore.java | 4 +-
.../hbase/index/covered/LocalTableState.java | 27 --
.../hbase/index/covered/data/IndexMemStore.java | 48 +--
.../index/scanner/FilteredKeyValueScanner.java | 65 +---
.../hbase/index/scanner/ReseekableScanner.java | 28 ++
.../hbase/index/scanner/ScannerBuilder.java | 6 +-
.../phoenix/iterate/BaseResultIterators.java | 9 +-
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 19 +-
.../phoenix/mapreduce/AbstractBulkLoadTool.java | 12 +-
.../phoenix/mapreduce/CsvToKeyValueMapper.java | 2 +-
.../mapreduce/FormatToBytesWritableMapper.java | 388 +++++++++++++++++++
.../mapreduce/FormatToKeyValueMapper.java | 378 ------------------
.../mapreduce/FormatToKeyValueReducer.java | 31 +-
.../phoenix/mapreduce/JsonToKeyValueMapper.java | 2 +-
.../bulkload/TargetTableRefFunctions.java | 2 +-
.../util/PhoenixConfigurationUtil.java | 4 +-
.../apache/phoenix/optimize/QueryOptimizer.java | 14 +-
.../query/ConnectionQueryServicesImpl.java | 27 +-
.../org/apache/phoenix/schema/PTableImpl.java | 2 +-
.../org/apache/phoenix/schema/Sequence.java | 46 +--
.../stats/DefaultStatisticsCollector.java | 6 +-
.../phoenix/schema/stats/PTableStatsImpl.java | 2 +-
.../stats/StatisticsCollectorFactory.java | 3 +-
.../org/apache/phoenix/util/MetaDataUtil.java | 2 +-
.../org/apache/phoenix/util/PhoenixRuntime.java | 31 ++
.../org/apache/phoenix/util/SchemaUtil.java | 4 +-
.../org/apache/phoenix/util/UpgradeUtil.java | 18 +-
.../phoenix/compile/QueryOptimizerTest.java | 28 ++
.../BuiltinFunctionConstructorTest.java | 11 +-
.../index/covered/data/TestIndexMemStore.java | 6 +-
.../FormatToBytesWritableMapperTest.java | 102 +++++
.../mapreduce/FormatToKeyValueMapperTest.java | 102 -----
.../java/org/apache/phoenix/query/BaseTest.java | 2 +-
.../org/apache/phoenix/query/QueryPlanTest.java | 25 ++
.../apache/phoenix/util/PhoenixRuntimeTest.java | 13 +-
phoenix-flume/pom.xml | 2 +-
phoenix-pherf/pom.xml | 2 +-
phoenix-pig/pom.xml | 2 +-
phoenix-server-client/pom.xml | 6 +-
phoenix-server/pom.xml | 14 +-
.../src/build/query-server-runnable.xml | 2 +-
phoenix-spark/pom.xml | 2 +-
phoenix-tracing-webapp/pom.xml | 2 +-
pom.xml | 43 +-
67 files changed, 952 insertions(+), 830 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c55f74de/phoenix-core/pom.xml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c55f74de/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
index 22c0197,c36a72b..2a0d3b3
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
@@@ -1006,8 -882,23 +1006,15 @@@ public abstract class BaseResultIterato
}
}
buf.append(getName()).append(" ").append(size()).append("-WAY ");
+ try {
+ if (plan.useRoundRobinIterator()) {
+ buf.append("ROUND ROBIN ");
+ }
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
explain(buf.toString(),planSteps);
}
-
- public Long getEstimatedRowCount() {
- return this.estimatedRows;
- }
-
- public Long getEstimatedByteCount() {
- return this.estimatedSize;
- }
@Override
public String toString() {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c55f74de/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c55f74de/pom.xml
----------------------------------------------------------------------
[30/30] phoenix git commit: Sync with master
Posted by ma...@apache.org.
Sync with master
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/6ec421c1
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/6ec421c1
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/6ec421c1
Branch: refs/heads/calcite
Commit: 6ec421c120eb382865ac90c2f129d446cd231a83
Parents: c55f74d
Author: maryannxue <ma...@gmail.com>
Authored: Tue Mar 29 15:50:21 2016 -0400
Committer: maryannxue <ma...@gmail.com>
Committed: Tue Mar 29 15:50:21 2016 -0400
----------------------------------------------------------------------
phoenix-core/pom.xml | 4 ++--
.../src/it/java/org/apache/phoenix/calcite/CalciteIT.java | 2 +-
.../src/main/java/org/apache/phoenix/calcite/PhoenixSchema.java | 2 +-
pom.xml | 1 +
4 files changed, 5 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/6ec421c1/phoenix-core/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-core/pom.xml b/phoenix-core/pom.xml
index 8db8a51..58e0760 100644
--- a/phoenix-core/pom.xml
+++ b/phoenix-core/pom.xml
@@ -451,8 +451,8 @@
<artifactId>commons-csv</artifactId>
</dependency>
<dependency>
- <groupId>org.apache.calcite</groupId>
- <artifactId>calcite-avatica</artifactId>
+ <groupId>org.apache.calcite.avatica</groupId>
+ <artifactId>avatica</artifactId>
</dependency>
<dependency>
<groupId>org.apache.calcite</groupId>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/6ec421c1/phoenix-core/src/it/java/org/apache/phoenix/calcite/CalciteIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/calcite/CalciteIT.java b/phoenix-core/src/it/java/org/apache/phoenix/calcite/CalciteIT.java
index fc08252..d583a9e 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/calcite/CalciteIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/calcite/CalciteIT.java
@@ -1071,7 +1071,7 @@ public class CalciteIT extends BaseCalciteIT {
" PhoenixTableScan(table=[[phoenix, Join, ItemTable]])\n" +
" PhoenixServerProject(item_id=[$2])\n" +
" PhoenixTableScan(table=[[phoenix, Join, OrderTable]])\n";
- start(correlProps).sql(q5).explainIs(p5Correlate).resultIs(0, r5).close();
+ //start(correlProps).sql(q5).explainIs(p5Correlate).resultIs(0, r5).close();
//TODO long-running query, disable for now.
//start(decorrelProps).sql(q5).explainIs(p5Decorrelated).resultIs(false, r5).close();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/6ec421c1/phoenix-core/src/main/java/org/apache/phoenix/calcite/PhoenixSchema.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/calcite/PhoenixSchema.java b/phoenix-core/src/main/java/org/apache/phoenix/calcite/PhoenixSchema.java
index b8401c6..5e26017 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/calcite/PhoenixSchema.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/calcite/PhoenixSchema.java
@@ -282,7 +282,7 @@ public class PhoenixSchema implements Schema {
try {
// FIXME: Do this the same way as resolving a table after PHOENIX-2489.
String tenantId = pc.getTenantId() == null ? null : pc.getTenantId().getString();
- String q = "select 1 from " + PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_ESCAPED
+ String q = "select 1 from " + PhoenixDatabaseMetaData.SYSTEM_SEQUENCE
+ " where " + PhoenixDatabaseMetaData.SEQUENCE_SCHEMA
+ (schemaName == null ? " is null" : " = '" + schemaName + "'")
+ " and " + PhoenixDatabaseMetaData.SEQUENCE_NAME
http://git-wip-us.apache.org/repos/asf/phoenix/blob/6ec421c1/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 052f15f..b882e0e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -92,6 +92,7 @@
<jodatime.version>1.6</jodatime.version>
<joni.version>2.1.2</joni.version>
<avatica.version>1.7.1</avatica.version>
+ <calcite.version>1.6.0</calcite.version>
<jettyVersion>8.1.7.v20120910</jettyVersion>
<tephra.version>0.7.0</tephra.version>
<spark.version>1.5.2</spark.version>
[27/30] phoenix git commit: PHOENIX-2774 MemStoreScanner and
KeyValueStore should not be aware of KeyValueScanner (Churro Morales)
Posted by ma...@apache.org.
PHOENIX-2774 MemStoreScanner and KeyValueStore should not be aware of KeyValueScanner (Churro Morales)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/d07fed5e
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/d07fed5e
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/d07fed5e
Branch: refs/heads/calcite
Commit: d07fed5e997cfcbf48e09516716a919d5e3315cb
Parents: 64fdf84
Author: James Taylor <jt...@salesforce.com>
Authored: Mon Mar 28 10:39:15 2016 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Mon Mar 28 10:39:15 2016 -0700
----------------------------------------------------------------------
.../apache/phoenix/hbase/index/covered/data/IndexMemStore.java | 5 +----
1 file changed, 1 insertion(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d07fed5e/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/IndexMemStore.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/IndexMemStore.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/IndexMemStore.java
index c7c5e3a..5b68fa2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/IndexMemStore.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/IndexMemStore.java
@@ -31,7 +31,6 @@ import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.regionserver.IndexKeyValueSkipListSet;
import org.apache.hadoop.hbase.regionserver.MemStore;
import org.apache.hadoop.hbase.util.Bytes;
-
import org.apache.phoenix.hbase.index.covered.KeyValueStore;
import org.apache.phoenix.hbase.index.covered.LocalTableState;
import org.apache.phoenix.hbase.index.scanner.ReseekableScanner;
@@ -251,9 +250,7 @@ public class IndexMemStore implements KeyValueStore {
* Unfortunately the Java API does not offer a method to get it. So we remember the last keys
* we iterated to and restore the reseeked set to at least that point.
*/
-
- KeyValue kv = KeyValueUtil.ensureKeyValue(key);
- kvsetIt = kvsetAtCreation.tailSet(getHighest(kv, kvsetItRow)).iterator();
+ kvsetIt = kvsetAtCreation.tailSet(getHighest(KeyValueUtil.ensureKeyValue(key), kvsetItRow)).iterator();
return seekInSubLists();
}