You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by pb...@apache.org on 2018/06/28 00:02:56 UTC
[01/11] phoenix git commit: PHOENIX-4773 Move HTable rollback wrapper
into Tephra TAL method [Forced Update!]
Repository: phoenix
Updated Branches:
refs/heads/4.x-cdh5.12 44939d29e -> d32eac94d (forced update)
PHOENIX-4773 Move HTable rollback wrapper into Tephra TAL method
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/3aaff768
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/3aaff768
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/3aaff768
Branch: refs/heads/4.x-cdh5.12
Commit: 3aaff7682fe8609d9c6a165ef778dd7a13076a7c
Parents: 0a06e59
Author: James Taylor <ja...@apache.org>
Authored: Mon Jun 4 20:27:36 2018 -0700
Committer: James Taylor <ja...@apache.org>
Committed: Tue Jun 5 21:55:16 2018 -0700
----------------------------------------------------------------------
.../apache/phoenix/cache/ServerCacheClient.java | 21 +-
.../apache/phoenix/execute/HashJoinPlan.java | 7 +-
.../apache/phoenix/execute/MutationState.java | 190 ++-----------------
.../PhoenixTxIndexMutationGenerator.java | 42 ++++
.../phoenix/index/IndexMetaDataCacheClient.java | 67 ++++++-
.../apache/phoenix/join/HashCacheClient.java | 5 +-
.../transaction/OmidTransactionContext.java | 3 +-
.../transaction/PhoenixTransactionContext.java | 5 +-
.../transaction/TephraTransactionContext.java | 91 ++++++++-
.../java/org/apache/phoenix/util/IndexUtil.java | 8 +
10 files changed, 230 insertions(+), 209 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3aaff768/phoenix-core/src/main/java/org/apache/phoenix/cache/ServerCacheClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/cache/ServerCacheClient.java b/phoenix-core/src/main/java/org/apache/phoenix/cache/ServerCacheClient.java
index 68de747..5e284bd 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/cache/ServerCacheClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/cache/ServerCacheClient.java
@@ -70,7 +70,6 @@ import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTable.IndexType;
-import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.util.Closeables;
import org.apache.phoenix.util.SQLCloseable;
import org.apache.phoenix.util.SQLCloseables;
@@ -90,7 +89,7 @@ public class ServerCacheClient {
private static final Random RANDOM = new Random();
public static final String HASH_JOIN_SERVER_CACHE_RESEND_PER_SERVER = "hash.join.server.cache.resend.per.server";
private final PhoenixConnection connection;
- private final Map<Integer, TableRef> cacheUsingTableRefMap = new ConcurrentHashMap<Integer, TableRef>();
+ private final Map<Integer, PTable> cacheUsingTableMap = new ConcurrentHashMap<Integer, PTable>();
/**
* Construct client used to create a serialized cached snapshot of a table and send it to each region server
@@ -220,12 +219,12 @@ public class ServerCacheClient {
}
public ServerCache addServerCache(ScanRanges keyRanges, final ImmutableBytesWritable cachePtr, final byte[] txState,
- final ServerCacheFactory cacheFactory, final TableRef cacheUsingTableRef) throws SQLException {
- return addServerCache(keyRanges, cachePtr, txState, cacheFactory, cacheUsingTableRef, false);
+ final ServerCacheFactory cacheFactory, final PTable cacheUsingTable) throws SQLException {
+ return addServerCache(keyRanges, cachePtr, txState, cacheFactory, cacheUsingTable, false);
}
public ServerCache addServerCache(ScanRanges keyRanges, final ImmutableBytesWritable cachePtr, final byte[] txState,
- final ServerCacheFactory cacheFactory, final TableRef cacheUsingTableRef, boolean storeCacheOnClient)
+ final ServerCacheFactory cacheFactory, final PTable cacheUsingTable, boolean storeCacheOnClient)
throws SQLException {
ConnectionQueryServices services = connection.getQueryServices();
List<Closeable> closeables = new ArrayList<Closeable>();
@@ -241,7 +240,6 @@ public class ServerCacheClient {
ExecutorService executor = services.getExecutor();
List<Future<Boolean>> futures = Collections.emptyList();
try {
- final PTable cacheUsingTable = cacheUsingTableRef.getTable();
List<HRegionLocation> locations = services.getAllTableRegions(cacheUsingTable.getPhysicalName().getBytes());
int nRegions = locations.size();
// Size these based on worst case
@@ -258,7 +256,7 @@ public class ServerCacheClient {
servers.add(entry);
if (LOG.isDebugEnabled()) {LOG.debug(addCustomAnnotations("Adding cache entry to be sent for " + entry, connection));}
final byte[] key = getKeyInRegion(entry.getRegionInfo().getStartKey());
- final HTableInterface htable = services.getTable(cacheUsingTableRef.getTable().getPhysicalName().getBytes());
+ final HTableInterface htable = services.getTable(cacheUsingTable.getPhysicalName().getBytes());
closeables.add(htable);
futures.add(executor.submit(new JobCallable<Boolean>() {
@@ -294,7 +292,7 @@ public class ServerCacheClient {
future.get(timeoutMs, TimeUnit.MILLISECONDS);
}
- cacheUsingTableRefMap.put(Bytes.mapKey(cacheId), cacheUsingTableRef);
+ cacheUsingTableMap.put(Bytes.mapKey(cacheId), cacheUsingTable);
success = true;
} catch (SQLException e) {
firstException = e;
@@ -337,9 +335,8 @@ public class ServerCacheClient {
try {
ConnectionQueryServices services = connection.getQueryServices();
Throwable lastThrowable = null;
- TableRef cacheUsingTableRef = cacheUsingTableRefMap.get(Bytes.mapKey(cacheId));
- final PTable cacheUsingTable = cacheUsingTableRef.getTable();
- byte[] tableName = cacheUsingTableRef.getTable().getPhysicalName().getBytes();
+ final PTable cacheUsingTable = cacheUsingTableMap.get(Bytes.mapKey(cacheId));
+ byte[] tableName = cacheUsingTable.getPhysicalName().getBytes();
iterateOverTable = services.getTable(tableName);
List<HRegionLocation> locations = services.getAllTableRegions(tableName);
@@ -403,7 +400,7 @@ public class ServerCacheClient {
lastThrowable);
}
} finally {
- cacheUsingTableRefMap.remove(cacheId);
+ cacheUsingTableMap.remove(cacheId);
Closeables.closeQuietly(iterateOverTable);
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3aaff768/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
index 5b433b3..bfe089d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
@@ -66,7 +66,10 @@ import org.apache.phoenix.join.HashCacheClient;
import org.apache.phoenix.join.HashJoinInfo;
import org.apache.phoenix.monitoring.TaskExecutionMetricsHolder;
import org.apache.phoenix.optimize.Cost;
-import org.apache.phoenix.parse.*;
+import org.apache.phoenix.parse.FilterableStatement;
+import org.apache.phoenix.parse.ParseNode;
+import org.apache.phoenix.parse.SQLParser;
+import org.apache.phoenix.parse.SelectStatement;
import org.apache.phoenix.query.ConnectionQueryServices;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
@@ -494,7 +497,7 @@ public class HashJoinPlan extends DelegateQueryPlan {
cache =
parent.hashClient.addHashCache(ranges, iterator,
plan.getEstimatedSize(), hashExpressions, singleValueOnly,
- parent.delegate.getTableRef(), keyRangeRhsExpression,
+ parent.delegate.getTableRef().getTable(), keyRangeRhsExpression,
keyRangeRhsValues);
long endTime = System.currentTimeMillis();
boolean isSet = parent.firstJobEndTime.compareAndSet(0, endTime);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3aaff768/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
index 2e795b1..c29d6b5 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
@@ -38,7 +38,6 @@ import javax.annotation.Nonnull;
import javax.annotation.concurrent.Immutable;
import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
@@ -51,7 +50,6 @@ import org.apache.phoenix.cache.ServerCacheClient;
import org.apache.phoenix.cache.ServerCacheClient.ServerCache;
import org.apache.phoenix.compile.MutationPlan;
import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
-import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.coprocessor.MetaDataProtocol.MetaDataMutationResult;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
@@ -60,10 +58,8 @@ import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.index.IndexMetaDataCacheClient;
import org.apache.phoenix.index.PhoenixIndexBuilder;
-import org.apache.phoenix.index.PhoenixIndexCodec;
import org.apache.phoenix.index.PhoenixIndexFailurePolicy;
import org.apache.phoenix.index.PhoenixIndexFailurePolicy.MutateCommand;
-import org.apache.phoenix.index.PhoenixIndexMetaData;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixStatement.Operation;
import org.apache.phoenix.monitoring.GlobalClientMetrics;
@@ -94,14 +90,10 @@ import org.apache.phoenix.transaction.PhoenixTransactionContext;
import org.apache.phoenix.transaction.PhoenixTransactionContext.PhoenixVisibilityLevel;
import org.apache.phoenix.transaction.TransactionFactory;
import org.apache.phoenix.transaction.TransactionFactory.Provider;
-import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.IndexUtil;
import org.apache.phoenix.util.KeyValueUtil;
import org.apache.phoenix.util.LogUtil;
-import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.SQLCloseable;
-import org.apache.phoenix.util.SQLCloseables;
-import org.apache.phoenix.util.ScanUtil;
import org.apache.phoenix.util.ServerUtil;
import org.apache.phoenix.util.SizedUtil;
import org.apache.phoenix.util.TransactionUtil;
@@ -499,19 +491,13 @@ public class MutationState implements SQLCloseable {
return ptr;
}
- private static List<PTable> getClientMaintainedIndexes(PTable table) {
- Iterator<PTable> indexIterator = // Only maintain tables with immutable rows through this client-side mechanism
- (table.isImmutableRows() || table.isTransactional()) ? IndexMaintainer.maintainedGlobalIndexes(table
- .getIndexes().iterator()) : Collections.<PTable> emptyIterator();
- return Lists.newArrayList(indexIterator);
- }
-
private Iterator<Pair<PName, List<Mutation>>> addRowMutations(final TableRef tableRef,
final MultiRowMutationState values, final long mutationTimestamp, final long serverTimestamp,
boolean includeAllIndexes, final boolean sendAll) {
final PTable table = tableRef.getTable();
- final List<PTable> indexList = includeAllIndexes ? Lists.newArrayList(IndexMaintainer.maintainedIndexes(table
- .getIndexes().iterator())) : getClientMaintainedIndexes(table);
+ final List<PTable> indexList = includeAllIndexes ?
+ Lists.newArrayList(IndexMaintainer.maintainedIndexes(table.getIndexes().iterator())) :
+ IndexUtil.getClientMaintainedIndexes(table);
final Iterator<PTable> indexes = indexList.iterator();
final List<Mutation> mutationList = Lists.newArrayListWithExpectedSize(values.size());
final List<Mutation> mutationsPertainingToIndex = indexes.hasNext() ? Lists
@@ -541,7 +527,7 @@ public class MutationState implements SQLCloseable {
if (!mutationsPertainingToIndex.isEmpty()) {
if (table.isTransactional()) {
if (indexMutationsMap == null) {
- PhoenixTxIndexMutationGenerator generator = newTxIndexMutationGenerator(table,
+ PhoenixTxIndexMutationGenerator generator = PhoenixTxIndexMutationGenerator.newGenerator(connection, table,
indexList, mutationsPertainingToIndex.get(0).getAttributesMap());
try (HTableInterface htable = connection.getQueryServices().getTable(
table.getPhysicalName().getBytes())) {
@@ -596,43 +582,6 @@ public class MutationState implements SQLCloseable {
};
}
- private PhoenixTxIndexMutationGenerator newTxIndexMutationGenerator(PTable table, List<PTable> indexes,
- Map<String, byte[]> attributes) {
- final List<IndexMaintainer> indexMaintainers = Lists.newArrayListWithExpectedSize(indexes.size());
- for (PTable index : indexes) {
- IndexMaintainer maintainer = index.getIndexMaintainer(table, connection);
- indexMaintainers.add(maintainer);
- }
- IndexMetaDataCache indexMetaDataCache = new IndexMetaDataCache() {
-
- @Override
- public void close() throws IOException {}
-
- @Override
- public List<IndexMaintainer> getIndexMaintainers() {
- return indexMaintainers;
- }
-
- @Override
- public PhoenixTransactionContext getTransactionContext() {
- return phoenixTransactionContext.newTransactionContext(phoenixTransactionContext, true);
- }
-
- @Override
- public int getClientVersion() {
- return MetaDataProtocol.PHOENIX_VERSION;
- }
-
- };
- try {
- PhoenixIndexMetaData indexMetaData = new PhoenixIndexMetaData(indexMetaDataCache, attributes);
- return new PhoenixTxIndexMutationGenerator(connection.getQueryServices().getConfiguration(), indexMetaData,
- table.getPhysicalName().getBytes());
- } catch (IOException e) {
- throw new RuntimeException(e); // Impossible
- }
- }
-
private void generateMutations(final TableRef tableRef, final long mutationTimestamp, final long serverTimestamp,
final MultiRowMutationState values, final List<Mutation> mutationList,
final List<Mutation> mutationsPertainingToIndex) {
@@ -793,7 +742,6 @@ public class MutationState implements SQLCloseable {
private long validateAndGetServerTimestamp(TableRef tableRef, MultiRowMutationState rowKeyToColumnMap)
throws SQLException {
- Long scn = connection.getSCN();
MetaDataClient client = new MetaDataClient(connection);
long serverTimeStamp = tableRef.getTimeStamp();
// If we're auto committing, we've already validated the schema when we got the ColumnResolver,
@@ -862,65 +810,6 @@ public class MutationState implements SQLCloseable {
return batchCount;
}
- private class MetaDataAwareHTable extends DelegateHTable {
- private final TableRef tableRef;
-
- private MetaDataAwareHTable(HTableInterface delegate, TableRef tableRef) {
- super(delegate);
- this.tableRef = tableRef;
- }
-
- /**
- * Called by Tephra when a transaction is aborted. We have this wrapper so that we get an opportunity to attach
- * our index meta data to the mutations such that we can also undo the index mutations.
- */
- @Override
- public void delete(List<Delete> deletes) throws IOException {
- ServerCache cache = null;
- try {
- if (deletes.isEmpty()) { return; }
- // Attach meta data for server maintained indexes
- PTable table = tableRef.getTable();
- ImmutableBytesWritable indexMetaDataPtr = new ImmutableBytesWritable();
- if (table.getIndexMaintainers(indexMetaDataPtr, connection)) {
- cache = setMetaDataOnMutations(tableRef, deletes, indexMetaDataPtr);
- }
-
- // Send deletes for client maintained indexes
- List<PTable> indexes = getClientMaintainedIndexes(table);
- if (!indexes.isEmpty()) {
- PhoenixTxIndexMutationGenerator generator = newTxIndexMutationGenerator(table, indexes, deletes.get(0)
- .getAttributesMap());
- Collection<Pair<Mutation, byte[]>> indexUpdates = generator.getIndexUpdates(delegate,
- deletes.iterator());
- for (PTable index : indexes) {
- byte[] physicalName = index.getPhysicalName().getBytes();
- try (HTableInterface hindex = connection.getQueryServices().getTable(physicalName)) {
- List<Mutation> indexDeletes = Lists.newArrayListWithExpectedSize(deletes.size());
- for (Pair<Mutation, byte[]> mutationPair : indexUpdates) {
- if (Bytes.equals(mutationPair.getSecond(), physicalName)) {
- indexDeletes.add(mutationPair.getFirst());
- }
- hindex.batch(indexDeletes);
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new IOException(e);
- }
- }
- }
-
- delegate.delete(deletes);
- } catch (SQLException e) {
- throw new IOException(e);
- } finally {
- if (cache != null) {
- SQLCloseables.closeAllQuietly(Collections.singletonList(cache));
- }
- }
- }
- }
-
private static class TableInfo {
private final boolean isDataTable;
@@ -1059,8 +948,9 @@ public class MutationState implements SQLCloseable {
TableRef origTableRef = tableInfo.getOrigTableRef();
PTable table = origTableRef.getTable();
table.getIndexMaintainers(indexMetaDataPtr, connection);
- final ServerCache cache = tableInfo.isDataTable() ? setMetaDataOnMutations(origTableRef,
- mutationList, indexMetaDataPtr) : null;
+ final ServerCache cache = tableInfo.isDataTable() ?
+ IndexMetaDataCacheClient.setMetaDataOnMutations(connection, table,
+ mutationList, indexMetaDataPtr) : null;
// If we haven't retried yet, retry for this case only, as it's possible that
// a split will occur after we send the index metadata cache to all known
// region servers.
@@ -1070,19 +960,12 @@ public class MutationState implements SQLCloseable {
try {
if (table.isTransactional()) {
// Track tables to which we've sent uncommitted data
- uncommittedPhysicalNames.add(table.getPhysicalName().getString());
- phoenixTransactionContext.markDMLFence(table);
-
- // If we have indexes, wrap the HTable in a delegate HTable that
- // will attach the necessary index meta data in the event of a
- // rollback
- if (!table.getIndexes().isEmpty()) {
- hTable = new MetaDataAwareHTable(hTable, origTableRef);
+ if (tableInfo.isDataTable()) {
+ uncommittedPhysicalNames.add(table.getPhysicalName().getString());
+ phoenixTransactionContext.markDMLFence(table);
}
-
- hTable = phoenixTransactionContext.getTransactionalTableWriter(hTable, table);
+ hTable = phoenixTransactionContext.getTransactionalTableWriter(connection, table, hTable, !tableInfo.isDataTable());
}
-
numMutations = mutationList.size();
GLOBAL_MUTATION_BATCH_SIZE.update(numMutations);
mutationSizeBytes = calculateMutationSize(mutationList);
@@ -1236,57 +1119,6 @@ public class MutationState implements SQLCloseable {
return phoenixTransactionContext.encodeTransaction();
}
- private ServerCache setMetaDataOnMutations(TableRef tableRef, List<? extends Mutation> mutations,
- ImmutableBytesWritable indexMetaDataPtr) throws SQLException {
- PTable table = tableRef.getTable();
- final byte[] tenantIdBytes;
- if (table.isMultiTenant()) {
- tenantIdBytes = connection.getTenantId() == null ? null : ScanUtil.getTenantIdBytes(
- table.getRowKeySchema(), table.getBucketNum() != null, connection.getTenantId(),
- table.getViewIndexId() != null);
- } else {
- tenantIdBytes = connection.getTenantId() == null ? null : connection.getTenantId().getBytes();
- }
- ServerCache cache = null;
- byte[] attribValue = null;
- byte[] uuidValue = null;
- byte[] txState = ByteUtil.EMPTY_BYTE_ARRAY;
- if (table.isTransactional()) {
- txState = encodeTransaction();
- }
- boolean hasIndexMetaData = indexMetaDataPtr.getLength() > 0;
- if (hasIndexMetaData) {
- if (IndexMetaDataCacheClient.useIndexMetadataCache(connection, mutations, indexMetaDataPtr.getLength()
- + txState.length)) {
- IndexMetaDataCacheClient client = new IndexMetaDataCacheClient(connection, tableRef);
- cache = client.addIndexMetadataCache(mutations, indexMetaDataPtr, txState);
- uuidValue = cache.getId();
- } else {
- attribValue = ByteUtil.copyKeyBytesIfNecessary(indexMetaDataPtr);
- uuidValue = ServerCacheClient.generateId();
- }
- } else if (txState.length == 0) { return null; }
- // Either set the UUID to be able to access the index metadata from the cache
- // or set the index metadata directly on the Mutation
- for (Mutation mutation : mutations) {
- if (connection.getTenantId() != null) {
- mutation.setAttribute(PhoenixRuntime.TENANT_ID_ATTRIB, tenantIdBytes);
- }
- mutation.setAttribute(PhoenixIndexCodec.INDEX_UUID, uuidValue);
- if (attribValue != null) {
- mutation.setAttribute(PhoenixIndexCodec.INDEX_PROTO_MD, attribValue);
- mutation.setAttribute(BaseScannerRegionObserver.CLIENT_VERSION,
- Bytes.toBytes(MetaDataProtocol.PHOENIX_VERSION));
- if (txState.length > 0) {
- mutation.setAttribute(BaseScannerRegionObserver.TX_STATE, txState);
- }
- } else if (!hasIndexMetaData && txState.length > 0) {
- mutation.setAttribute(BaseScannerRegionObserver.TX_STATE, txState);
- }
- }
- return cache;
- }
-
private void addUncommittedStatementIndexes(Collection<RowMutationState> rowMutations) {
for (RowMutationState rowMutationState : rowMutations) {
uncommittedStatementIndexes = joinSortedIntArrays(uncommittedStatementIndexes,
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3aaff768/phoenix-core/src/main/java/org/apache/phoenix/execute/PhoenixTxIndexMutationGenerator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/PhoenixTxIndexMutationGenerator.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/PhoenixTxIndexMutationGenerator.java
index 0016fa9..a7b5687 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/PhoenixTxIndexMutationGenerator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/PhoenixTxIndexMutationGenerator.java
@@ -45,7 +45,9 @@ import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.cache.IndexMetaDataCache;
import org.apache.phoenix.compile.ScanRanges;
+import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.filter.SkipScanFilter;
import org.apache.phoenix.hbase.index.MultiMutation;
import org.apache.phoenix.hbase.index.ValueGetter;
@@ -59,7 +61,9 @@ import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.index.PhoenixIndexCodec;
import org.apache.phoenix.index.PhoenixIndexMetaData;
+import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.query.KeyRange;
+import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.types.PVarbinary;
import org.apache.phoenix.transaction.PhoenixTransactionContext;
import org.apache.phoenix.transaction.PhoenixTransactionContext.PhoenixVisibilityLevel;
@@ -440,4 +444,42 @@ public class PhoenixTxIndexMutationGenerator {
return pair;
}
}
+
+ public static PhoenixTxIndexMutationGenerator newGenerator(final PhoenixConnection connection, PTable table, List<PTable> indexes,
+ Map<String, byte[]> attributes) {
+ final List<IndexMaintainer> indexMaintainers = Lists.newArrayListWithExpectedSize(indexes.size());
+ for (PTable index : indexes) {
+ IndexMaintainer maintainer = index.getIndexMaintainer(table, connection);
+ indexMaintainers.add(maintainer);
+ }
+ IndexMetaDataCache indexMetaDataCache = new IndexMetaDataCache() {
+
+ @Override
+ public void close() throws IOException {}
+
+ @Override
+ public List<IndexMaintainer> getIndexMaintainers() {
+ return indexMaintainers;
+ }
+
+ @Override
+ public PhoenixTransactionContext getTransactionContext() {
+ PhoenixTransactionContext context = connection.getMutationState().getPhoenixTransactionContext();
+ return context.newTransactionContext(context, true);
+ }
+
+ @Override
+ public int getClientVersion() {
+ return MetaDataProtocol.PHOENIX_VERSION;
+ }
+
+ };
+ try {
+ PhoenixIndexMetaData indexMetaData = new PhoenixIndexMetaData(indexMetaDataCache, attributes);
+ return new PhoenixTxIndexMutationGenerator(connection.getQueryServices().getConfiguration(), indexMetaData,
+ table.getPhysicalName().getBytes());
+ } catch (IOException e) {
+ throw new RuntimeException(e); // Impossible
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3aaff768/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMetaDataCacheClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMetaDataCacheClient.java b/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMetaDataCacheClient.java
index fcabdfd..bd308c0 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMetaDataCacheClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMetaDataCacheClient.java
@@ -24,20 +24,25 @@ import java.util.List;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.cache.ServerCacheClient;
import org.apache.phoenix.cache.ServerCacheClient.ServerCache;
import org.apache.phoenix.compile.ScanRanges;
+import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
+import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.join.MaxServerCacheSizeExceededException;
import org.apache.phoenix.query.QueryServicesOptions;
-import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.util.ByteUtil;
+import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.ReadOnlyProps;
import org.apache.phoenix.util.ScanUtil;
public class IndexMetaDataCacheClient {
private final ServerCacheClient serverCache;
- private TableRef cacheUsingTableRef;
+ private PTable cacheUsingTable;
/**
* Construct client used to send index metadata to each region server
@@ -45,9 +50,9 @@ public class IndexMetaDataCacheClient {
* @param connection the client connection
* @param cacheUsingTableRef table ref to table that will use the cache during its scan
*/
- public IndexMetaDataCacheClient(PhoenixConnection connection, TableRef cacheUsingTableRef) {
+ public IndexMetaDataCacheClient(PhoenixConnection connection, PTable cacheUsingTable) {
serverCache = new ServerCacheClient(connection);
- this.cacheUsingTableRef = cacheUsingTableRef;
+ this.cacheUsingTable = cacheUsingTable;
}
/**
@@ -75,7 +80,7 @@ public class IndexMetaDataCacheClient {
/**
* Serialize and compress hashCacheTable
*/
- return serverCache.addServerCache(ScanUtil.newScanRanges(mutations), ptr, txState, new IndexMetaDataCacheFactory(), cacheUsingTableRef);
+ return serverCache.addServerCache(ScanUtil.newScanRanges(mutations), ptr, txState, new IndexMetaDataCacheFactory(), cacheUsingTable);
}
@@ -91,7 +96,55 @@ public class IndexMetaDataCacheClient {
/**
* Serialize and compress hashCacheTable
*/
- return serverCache.addServerCache(ranges, ptr, txState, new IndexMetaDataCacheFactory(), cacheUsingTableRef);
+ return serverCache.addServerCache(ranges, ptr, txState, new IndexMetaDataCacheFactory(), cacheUsingTable);
+ }
+
+ public static ServerCache setMetaDataOnMutations(PhoenixConnection connection, PTable table, List<? extends Mutation> mutations,
+ ImmutableBytesWritable indexMetaDataPtr) throws SQLException {
+ final byte[] tenantIdBytes;
+ if (table.isMultiTenant()) {
+ tenantIdBytes = connection.getTenantId() == null ? null : ScanUtil.getTenantIdBytes(
+ table.getRowKeySchema(), table.getBucketNum() != null, connection.getTenantId(),
+ table.getViewIndexId() != null);
+ } else {
+ tenantIdBytes = connection.getTenantId() == null ? null : connection.getTenantId().getBytes();
+ }
+ ServerCache cache = null;
+ byte[] attribValue = null;
+ byte[] uuidValue = null;
+ byte[] txState = ByteUtil.EMPTY_BYTE_ARRAY;
+ if (table.isTransactional()) {
+ txState = connection.getMutationState().encodeTransaction();
+ }
+ boolean hasIndexMetaData = indexMetaDataPtr.getLength() > 0;
+ if (hasIndexMetaData) {
+ if (useIndexMetadataCache(connection, mutations, indexMetaDataPtr.getLength() + txState.length)) {
+ IndexMetaDataCacheClient client = new IndexMetaDataCacheClient(connection, table);
+ cache = client.addIndexMetadataCache(mutations, indexMetaDataPtr, txState);
+ uuidValue = cache.getId();
+ } else {
+ attribValue = ByteUtil.copyKeyBytesIfNecessary(indexMetaDataPtr);
+ uuidValue = ServerCacheClient.generateId();
+ }
+ } else if (txState.length == 0) { return null; }
+ // Either set the UUID to be able to access the index metadata from the cache
+ // or set the index metadata directly on the Mutation
+ for (Mutation mutation : mutations) {
+ if (connection.getTenantId() != null) {
+ mutation.setAttribute(PhoenixRuntime.TENANT_ID_ATTRIB, tenantIdBytes);
+ }
+ mutation.setAttribute(PhoenixIndexCodec.INDEX_UUID, uuidValue);
+ if (attribValue != null) {
+ mutation.setAttribute(PhoenixIndexCodec.INDEX_PROTO_MD, attribValue);
+ mutation.setAttribute(BaseScannerRegionObserver.CLIENT_VERSION,
+ Bytes.toBytes(MetaDataProtocol.PHOENIX_VERSION));
+ if (txState.length > 0) {
+ mutation.setAttribute(BaseScannerRegionObserver.TX_STATE, txState);
+ }
+ } else if (!hasIndexMetaData && txState.length > 0) {
+ mutation.setAttribute(BaseScannerRegionObserver.TX_STATE, txState);
+ }
+ }
+ return cache;
}
-
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3aaff768/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheClient.java b/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheClient.java
index 2ec509c..83ac32d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheClient.java
@@ -37,7 +37,6 @@ import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.PTable;
-import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.util.ByteUtil;
@@ -77,13 +76,13 @@ public class HashCacheClient {
* @throws MaxServerCacheSizeExceededException if size of hash cache exceeds max allowed
* size
*/
- public ServerCache addHashCache(ScanRanges keyRanges, ResultIterator iterator, long estimatedSize, List<Expression> onExpressions, boolean singleValueOnly, TableRef cacheUsingTableRef, Expression keyRangeRhsExpression, List<Expression> keyRangeRhsValues) throws SQLException {
+ public ServerCache addHashCache(ScanRanges keyRanges, ResultIterator iterator, long estimatedSize, List<Expression> onExpressions, boolean singleValueOnly, PTable cacheUsingTable, Expression keyRangeRhsExpression, List<Expression> keyRangeRhsValues) throws SQLException {
/**
* Serialize and compress hashCacheTable
*/
ImmutableBytesWritable ptr = new ImmutableBytesWritable();
serialize(ptr, iterator, estimatedSize, onExpressions, singleValueOnly, keyRangeRhsExpression, keyRangeRhsValues);
- ServerCache cache = serverCache.addServerCache(keyRanges, ptr, ByteUtil.EMPTY_BYTE_ARRAY, new HashCacheFactory(), cacheUsingTableRef, true);
+ ServerCache cache = serverCache.addServerCache(keyRanges, ptr, ByteUtil.EMPTY_BYTE_ARRAY, new HashCacheFactory(), cacheUsingTable, true);
return cache;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3aaff768/phoenix-core/src/main/java/org/apache/phoenix/transaction/OmidTransactionContext.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/transaction/OmidTransactionContext.java b/phoenix-core/src/main/java/org/apache/phoenix/transaction/OmidTransactionContext.java
index ab9e8a6..d235d4b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/transaction/OmidTransactionContext.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/transaction/OmidTransactionContext.java
@@ -20,6 +20,7 @@ package org.apache.phoenix.transaction;
import java.sql.SQLException;
import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.transaction.TransactionFactory.Provider;
@@ -132,7 +133,7 @@ public class OmidTransactionContext implements PhoenixTransactionContext {
}
@Override
- public HTableInterface getTransactionalTableWriter(HTableInterface htable, PTable table) {
+ public HTableInterface getTransactionalTableWriter(PhoenixConnection connection, PTable table, HTableInterface htable, boolean isIndex) {
// TODO Auto-generated method stub
return null;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3aaff768/phoenix-core/src/main/java/org/apache/phoenix/transaction/PhoenixTransactionContext.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/transaction/PhoenixTransactionContext.java b/phoenix-core/src/main/java/org/apache/phoenix/transaction/PhoenixTransactionContext.java
index 751945a..dfa35be 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/transaction/PhoenixTransactionContext.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/transaction/PhoenixTransactionContext.java
@@ -21,6 +21,7 @@ import java.sql.SQLException;
import java.util.concurrent.TimeoutException;
import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.transaction.TransactionFactory.Provider;
@@ -110,7 +111,7 @@ public interface PhoenixTransactionContext {
}
@Override
- public HTableInterface getTransactionalTableWriter(HTableInterface htable, PTable table) {
+ public HTableInterface getTransactionalTableWriter(PhoenixConnection connection, PTable table, HTableInterface htable, boolean isIndex) {
return null;
}
};
@@ -230,5 +231,5 @@ public interface PhoenixTransactionContext {
public PhoenixTransactionContext newTransactionContext(PhoenixTransactionContext contex, boolean subTask);
public HTableInterface getTransactionalTable(HTableInterface htable, boolean isImmutable);
- public HTableInterface getTransactionalTableWriter(HTableInterface htable, PTable table);
+ public HTableInterface getTransactionalTableWriter(PhoenixConnection connection, PTable table, HTableInterface htable, boolean isIndex) throws SQLException;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3aaff768/phoenix-core/src/main/java/org/apache/phoenix/transaction/TephraTransactionContext.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/transaction/TephraTransactionContext.java b/phoenix-core/src/main/java/org/apache/phoenix/transaction/TephraTransactionContext.java
index 94d6765..35b82c0 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/transaction/TephraTransactionContext.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/transaction/TephraTransactionContext.java
@@ -20,22 +20,33 @@ package org.apache.phoenix.transaction;
import java.io.IOException;
import java.sql.SQLException;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.cache.ServerCacheClient.ServerCache;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.execute.DelegateHTable;
+import org.apache.phoenix.execute.PhoenixTxIndexMutationGenerator;
+import org.apache.phoenix.index.IndexMetaDataCacheClient;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.transaction.TephraTransactionProvider.TephraTransactionClient;
+import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.util.SQLCloseables;
import org.apache.tephra.Transaction;
import org.apache.tephra.Transaction.VisibilityLevel;
import org.apache.tephra.TransactionAware;
@@ -471,14 +482,19 @@ public class TephraTransactionContext implements PhoenixTransactionContext {
}
@Override
- public HTableInterface getTransactionalTableWriter(HTableInterface htable, PTable table) {
- boolean isIndex = table.getType() == PTableType.INDEX;
- TransactionAwareHTableDelegate transactionAwareHTable = new TransactionAwareHTableDelegate(htable, table.isImmutableRows() || isIndex ? TxConstants.ConflictDetection.NONE : TxConstants.ConflictDetection.ROW);
+ public HTableInterface getTransactionalTableWriter(PhoenixConnection connection, PTable table, HTableInterface htable, boolean isIndex) throws SQLException {
+ // If we have indexes, wrap the HTable in a delegate HTable that
+ // will attach the necessary index meta data in the event of a
+ // rollback
+ TransactionAwareHTableDelegate transactionAwareHTable;
// Don't add immutable indexes (those are the only ones that would participate
// during a commit), as we don't need conflict detection for these.
if (isIndex) {
+ transactionAwareHTable = new TransactionAwareHTableDelegate(htable, TxConstants.ConflictDetection.NONE);
transactionAwareHTable.startTx(getTransaction());
} else {
+ htable = new RollbackHookHTableWrapper(htable, table, connection);
+ transactionAwareHTable = new TransactionAwareHTableDelegate(htable, table.isImmutableRows() ? TxConstants.ConflictDetection.NONE : TxConstants.ConflictDetection.ROW);
// Even for immutable, we need to do this so that an abort has the state
// necessary to generate the rows to delete.
this.addTransactionAware(transactionAwareHTable);
@@ -486,4 +502,73 @@ public class TephraTransactionContext implements PhoenixTransactionContext {
return transactionAwareHTable;
}
+ /**
+ *
+ * Wraps Tephra data table HTables to catch when a rollback occurs so
+ * that index Delete mutations can be generated and applied (as
+ * opposed to storing them in the Tephra change set). This technique
+ * allows the Tephra API to be used directly with HBase APIs and
+ * Phoenix APIs since we can detect the rollback as a callback
+ * when the Tephra rollback is called.
+ *
+ */
+ private static class RollbackHookHTableWrapper extends DelegateHTable {
+ private final PTable table;
+ private final PhoenixConnection connection;
+
+ private RollbackHookHTableWrapper(HTableInterface delegate, PTable table, PhoenixConnection connection) {
+ super(delegate);
+ this.table = table;
+ this.connection = connection;
+ }
+
+ /**
+ * Called by Tephra when a transaction is aborted. We have this wrapper so that we get an opportunity to attach
+ * our index meta data to the mutations such that we can also undo the index mutations.
+ */
+ @Override
+ public void delete(List<Delete> deletes) throws IOException {
+ ServerCache cache = null;
+ try {
+ if (deletes.isEmpty()) { return; }
+ // Attach meta data for server maintained indexes
+ ImmutableBytesWritable indexMetaDataPtr = new ImmutableBytesWritable();
+ if (table.getIndexMaintainers(indexMetaDataPtr, connection)) {
+ cache = IndexMetaDataCacheClient.setMetaDataOnMutations(connection, table, deletes, indexMetaDataPtr);
+ }
+
+ // Send deletes for client maintained indexes
+ List<PTable> indexes = IndexUtil.getClientMaintainedIndexes(table);
+ if (!indexes.isEmpty()) {
+ PhoenixTxIndexMutationGenerator generator = PhoenixTxIndexMutationGenerator.newGenerator(connection, table, indexes, deletes.get(0)
+ .getAttributesMap());
+ Collection<Pair<Mutation, byte[]>> indexUpdates = generator.getIndexUpdates(delegate,
+ deletes.iterator());
+ for (PTable index : indexes) {
+ byte[] physicalName = index.getPhysicalName().getBytes();
+ try (HTableInterface hindex = connection.getQueryServices().getTable(physicalName)) {
+ List<Mutation> indexDeletes = Lists.newArrayListWithExpectedSize(deletes.size());
+ for (Pair<Mutation, byte[]> mutationPair : indexUpdates) {
+ if (Bytes.equals(mutationPair.getSecond(), physicalName)) {
+ indexDeletes.add(mutationPair.getFirst());
+ }
+ hindex.batch(indexDeletes);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ }
+ }
+ }
+
+ delegate.delete(deletes);
+ } catch (SQLException e) {
+ throw new IOException(e);
+ } finally {
+ if (cache != null) {
+ SQLCloseables.closeAllQuietly(Collections.singletonList(cache));
+ }
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3aaff768/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
index 78a68d2..3fe5438 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
@@ -32,6 +32,7 @@ import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
@@ -816,5 +817,12 @@ public class IndexUtil {
.setTableName(indexName).build().buildException();
}
}
+
+ public static List<PTable> getClientMaintainedIndexes(PTable table) {
+ Iterator<PTable> indexIterator = // Only maintain tables with immutable rows through this client-side mechanism
+ (table.isImmutableRows() || table.isTransactional()) ? IndexMaintainer.maintainedGlobalIndexes(table
+ .getIndexes().iterator()) : Collections.<PTable> emptyIterator();
+ return Lists.newArrayList(indexIterator);
+ }
}
[07/11] phoenix git commit: PHOENIX-4777 Fix rat:check failures
Posted by pb...@apache.org.
PHOENIX-4777 Fix rat:check failures
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/fdebb15a
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/fdebb15a
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/fdebb15a
Branch: refs/heads/4.x-cdh5.12
Commit: fdebb15a64765f8cf36efefed1f224d38728fa80
Parents: f95f860
Author: Pedro Boado <pb...@apache.org>
Authored: Tue Jun 12 00:08:02 2018 +0100
Committer: Pedro Boado <pb...@apache.org>
Committed: Tue Jun 12 00:08:02 2018 +0100
----------------------------------------------------------------------
pom.xml | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/fdebb15a/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index e55b080..b5c2a5b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -231,6 +231,7 @@
<plugin>
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
+ <version>0.10</version>
<!-- Avoid defining exclusions in pluginManagement as they are global.
We already inherit some from the ASF parent pom. -->
</plugin>
[11/11] phoenix git commit: Changes for CDH 5.12.x
Posted by pb...@apache.org.
Changes for CDH 5.12.x
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/d32eac94
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/d32eac94
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/d32eac94
Branch: refs/heads/4.x-cdh5.12
Commit: d32eac94d683ba2b0c0cf2639d0e36f21e29f3b6
Parents: a162c74
Author: Pedro Boado <pb...@apache.org>
Authored: Sat Mar 10 17:54:04 2018 +0000
Committer: Pedro Boado <pb...@apache.org>
Committed: Thu Jun 28 01:02:43 2018 +0100
----------------------------------------------------------------------
phoenix-assembly/pom.xml | 2 +-
phoenix-client/pom.xml | 2 +-
phoenix-core/pom.xml | 2 +-
.../hadoop/hbase/ipc/PhoenixRpcScheduler.java | 34 ++++++++++++++++++--
phoenix-flume/pom.xml | 2 +-
phoenix-hive/pom.xml | 2 +-
phoenix-kafka/pom.xml | 2 +-
phoenix-load-balancer/pom.xml | 2 +-
phoenix-parcel/pom.xml | 2 +-
phoenix-pherf/pom.xml | 2 +-
phoenix-pig/pom.xml | 2 +-
phoenix-queryserver-client/pom.xml | 2 +-
phoenix-queryserver/pom.xml | 2 +-
phoenix-server/pom.xml | 2 +-
phoenix-spark/pom.xml | 2 +-
phoenix-tracing-webapp/pom.xml | 2 +-
pom.xml | 4 +--
17 files changed, 49 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-assembly/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-assembly/pom.xml b/phoenix-assembly/pom.xml
index 7d98c25..47b593f 100644
--- a/phoenix-assembly/pom.xml
+++ b/phoenix-assembly/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-assembly</artifactId>
<name>Phoenix Assembly</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-client/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-client/pom.xml b/phoenix-client/pom.xml
index fe5836b..5736f9b 100644
--- a/phoenix-client/pom.xml
+++ b/phoenix-client/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-client</artifactId>
<name>Phoenix Client</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-core/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-core/pom.xml b/phoenix-core/pom.xml
index 8b09c67..ad429bf 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.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-core</artifactId>
<name>Phoenix Core</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-core/src/main/java/org/apache/hadoop/hbase/ipc/PhoenixRpcScheduler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/hadoop/hbase/ipc/PhoenixRpcScheduler.java b/phoenix-core/src/main/java/org/apache/hadoop/hbase/ipc/PhoenixRpcScheduler.java
index 4fdddf5..d1f05f8 100644
--- a/phoenix-core/src/main/java/org/apache/hadoop/hbase/ipc/PhoenixRpcScheduler.java
+++ b/phoenix-core/src/main/java/org/apache/hadoop/hbase/ipc/PhoenixRpcScheduler.java
@@ -124,6 +124,36 @@ public class PhoenixRpcScheduler extends RpcScheduler {
public void setMetadataExecutorForTesting(RpcExecutor executor) {
this.metadataCallExecutor = executor;
}
-
-
+
+ @Override
+ public int getReadQueueLength() {
+ return delegate.getReadQueueLength();
+ }
+
+ @Override
+ public int getWriteQueueLength() {
+ return delegate.getWriteQueueLength();
+ }
+
+ @Override
+ public int getScanQueueLength() {
+ return delegate.getScanQueueLength();
+ }
+
+ @Override
+ public int getActiveReadRpcHandlerCount() {
+ return delegate.getActiveReadRpcHandlerCount();
+ }
+
+ @Override
+ public int getActiveWriteRpcHandlerCount() {
+ return delegate.getActiveWriteRpcHandlerCount();
+ }
+
+ @Override
+ public int getActiveScanRpcHandlerCount() {
+ return delegate.getActiveScanRpcHandlerCount();
+ }
+
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-flume/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-flume/pom.xml b/phoenix-flume/pom.xml
index e400589..bb2654c 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.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-flume</artifactId>
<name>Phoenix - Flume</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-hive/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-hive/pom.xml b/phoenix-hive/pom.xml
index 4687887..235f919 100644
--- a/phoenix-hive/pom.xml
+++ b/phoenix-hive/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-hive</artifactId>
<name>Phoenix - Hive</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-kafka/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-kafka/pom.xml b/phoenix-kafka/pom.xml
index 158026e..0dcfaf9 100644
--- a/phoenix-kafka/pom.xml
+++ b/phoenix-kafka/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-kafka</artifactId>
<name>Phoenix - Kafka</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-load-balancer/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-load-balancer/pom.xml b/phoenix-load-balancer/pom.xml
index b9b4672..2f69a7e 100644
--- a/phoenix-load-balancer/pom.xml
+++ b/phoenix-load-balancer/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-load-balancer</artifactId>
<name>Phoenix Load Balancer</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-parcel/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-parcel/pom.xml b/phoenix-parcel/pom.xml
index 6f5777f..9835b8a 100644
--- a/phoenix-parcel/pom.xml
+++ b/phoenix-parcel/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-parcel</artifactId>
<name>Phoenix Parcels for CDH</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-pherf/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-pherf/pom.xml b/phoenix-pherf/pom.xml
index 04df091..e67e547 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.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-pherf</artifactId>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-pig/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-pig/pom.xml b/phoenix-pig/pom.xml
index a88cece..4645396 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.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-pig</artifactId>
<name>Phoenix - Pig</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-queryserver-client/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-queryserver-client/pom.xml b/phoenix-queryserver-client/pom.xml
index 424ce82..dfd399d 100644
--- a/phoenix-queryserver-client/pom.xml
+++ b/phoenix-queryserver-client/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-queryserver-client</artifactId>
<name>Phoenix Query Server Client</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-queryserver/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-queryserver/pom.xml b/phoenix-queryserver/pom.xml
index 66e0735..095dd88 100644
--- a/phoenix-queryserver/pom.xml
+++ b/phoenix-queryserver/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-queryserver</artifactId>
<name>Phoenix Query Server</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-server/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-server/pom.xml b/phoenix-server/pom.xml
index 98719fb..34c436b 100644
--- a/phoenix-server/pom.xml
+++ b/phoenix-server/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
- <version>4.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-server</artifactId>
<name>Phoenix Server</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-spark/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-spark/pom.xml b/phoenix-spark/pom.xml
index e6f5d95..46f66d8 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.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-spark</artifactId>
<name>Phoenix - Spark</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/phoenix-tracing-webapp/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-tracing-webapp/pom.xml b/phoenix-tracing-webapp/pom.xml
index 4f79b4e..7294740 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.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-tracing-webapp</artifactId>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d32eac94/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index b5c2a5b..02e3f6e 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.14.0-cdh5.11.2</version>
+ <version>4.14.0-cdh5.12.2</version>
<packaging>pom</packaging>
<name>Apache Phoenix</name>
<description>A SQL layer over HBase</description>
@@ -86,7 +86,7 @@
<parent>
<groupId>com.cloudera.cdh</groupId>
<artifactId>cdh-root</artifactId>
- <version>5.11.2</version>
+ <version>5.12.2</version>
</parent>
<scm>
[09/11] phoenix git commit: PHOENIX-4786 Reduce log level to debug
when logging new aggregate row key found and added results for scan ordered
queries(Rajeshbabu)
Posted by pb...@apache.org.
PHOENIX-4786 Reduce log level to debug when logging new aggregate row key found and added results for scan ordered queries(Rajeshbabu)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/5edcd6b0
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/5edcd6b0
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/5edcd6b0
Branch: refs/heads/4.x-cdh5.12
Commit: 5edcd6b0de8a8534740343c3bc72c3f68190c2b6
Parents: 35e09ce
Author: Rajeshbabu Chintaguntla <ra...@apache.org>
Authored: Fri Jun 15 23:40:25 2018 +0100
Committer: Pedro Boado <pb...@apache.org>
Committed: Thu Jun 28 00:57:28 2018 +0100
----------------------------------------------------------------------
.../phoenix/coprocessor/GroupedAggregateRegionObserver.java | 8 --------
1 file changed, 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5edcd6b0/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GroupedAggregateRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GroupedAggregateRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GroupedAggregateRegionObserver.java
index 86ab275..aefe916 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GroupedAggregateRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GroupedAggregateRegionObserver.java
@@ -534,14 +534,6 @@ public class GroupedAggregateRegionObserver extends BaseScannerRegionObserver {
currentKey.getLength(), SINGLE_COLUMN_FAMILY, SINGLE_COLUMN,
AGG_TIMESTAMP, value, 0, value.length);
results.add(keyValue);
- if (logger.isInfoEnabled()) {
- logger.info(LogUtil.addCustomAnnotations("Adding new aggregate row: "
- + keyValue
- + ",for current key "
- + Bytes.toStringBinary(currentKey.get(), currentKey.getOffset(),
- currentKey.getLength()) + ", aggregated values: "
- + Arrays.asList(rowAggregators), ScanUtil.getCustomAnnotations(scan)));
- }
// If we're at an aggregation boundary, reset the
// aggregators and
// aggregate with the current result (which is not a part of
[05/11] phoenix git commit: PHOENIX-4544 Update statistics
inconsistent behavior
Posted by pb...@apache.org.
PHOENIX-4544 Update statistics inconsistent behavior
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/c5a0e00f
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/c5a0e00f
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/c5a0e00f
Branch: refs/heads/4.x-cdh5.12
Commit: c5a0e00ffde98b8a0c9288e67fa67d864efe81d0
Parents: 91ffc04
Author: Ankit Singhal <an...@gmail.com>
Authored: Thu Jun 7 19:23:56 2018 +0100
Committer: Pedro Boado <pb...@apache.org>
Committed: Sat Jun 9 19:42:37 2018 +0100
----------------------------------------------------------------------
.../StatisticsCollectionRunTrackerIT.java | 32 ++++++++------
.../UngroupedAggregateRegionObserver.java | 4 +-
.../apache/phoenix/schema/MetaDataClient.java | 10 ++++-
.../stats/StatisticsCollectionRunTracker.java | 45 +++++++++++++++++---
.../java/org/apache/phoenix/util/ByteUtil.java | 16 ++++++-
5 files changed, 85 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c5a0e00f/phoenix-core/src/it/java/org/apache/phoenix/coprocessor/StatisticsCollectionRunTrackerIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/coprocessor/StatisticsCollectionRunTrackerIT.java b/phoenix-core/src/it/java/org/apache/phoenix/coprocessor/StatisticsCollectionRunTrackerIT.java
index cf475f9..a643383 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/coprocessor/StatisticsCollectionRunTrackerIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/coprocessor/StatisticsCollectionRunTrackerIT.java
@@ -25,6 +25,8 @@ import static org.junit.Assert.assertTrue;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
+import java.util.Arrays;
+import java.util.HashSet;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
@@ -61,12 +63,15 @@ public class StatisticsCollectionRunTrackerIT extends ParallelStatsEnabledIT {
StatisticsCollectionRunTracker tracker =
StatisticsCollectionRunTracker.getInstance(new Configuration());
// assert that the region wasn't added to the tracker
- assertTrue(tracker.addUpdateStatsCommandRegion(regionInfo));
+ assertTrue(tracker.addUpdateStatsCommandRegion(regionInfo, new HashSet<byte[]>(Arrays.asList(Bytes.toBytes("0")))));
+ assertTrue(tracker.addUpdateStatsCommandRegion(regionInfo, new HashSet<byte[]>(Arrays.asList(Bytes.toBytes("L#0")))));
// assert that removing the region from the tracker works
- assertTrue(tracker.removeUpdateStatsCommandRegion(regionInfo));
+ assertTrue(tracker.removeUpdateStatsCommandRegion(regionInfo, new HashSet<byte[]>(Arrays.asList(Bytes.toBytes("0")))));
+ assertTrue(tracker.removeUpdateStatsCommandRegion(regionInfo, new HashSet<byte[]>(Arrays.asList(Bytes.toBytes("L#0")))));
runUpdateStats(tableName);
// assert that after update stats is complete, tracker isn't tracking the region any more
- assertFalse(tracker.removeUpdateStatsCommandRegion(regionInfo));
+ assertFalse(tracker.removeUpdateStatsCommandRegion(regionInfo, new HashSet<byte[]>(Arrays.asList(Bytes.toBytes("0")))));
+ assertFalse(tracker.removeUpdateStatsCommandRegion(regionInfo, new HashSet<byte[]>(Arrays.asList(Bytes.toBytes("L#0")))));;
}
@Test
@@ -103,25 +108,27 @@ public class StatisticsCollectionRunTrackerIT extends ParallelStatsEnabledIT {
HRegionInfo regionInfo = createTableAndGetRegion(tableName);
// simulate stats collection via major compaction by marking the region as compacting in the tracker
markRegionAsCompacting(regionInfo);
- Assert.assertEquals("Row count didn't match", COMPACTION_UPDATE_STATS_ROW_COUNT, runUpdateStats(tableName));
- StatisticsCollectionRunTracker tracker =
+ // there will be no update for local index and a table , so checking 2 * COMPACTION_UPDATE_STATS_ROW_COUNT
+ Assert.assertEquals("Row count didn't match", COMPACTION_UPDATE_STATS_ROW_COUNT * 2, runUpdateStats(tableName));
StatisticsCollectionRunTracker.getInstance(new Configuration());
// assert that the tracker state was cleared.
- assertFalse(tracker.removeUpdateStatsCommandRegion(regionInfo));
+ HashSet<byte[]> familyMap = new HashSet<byte[]>(Arrays.asList(Bytes.toBytes("0")));
+ assertFalse(tracker.removeUpdateStatsCommandRegion(regionInfo, familyMap));
}
@Test
public void testUpdateStatsPreventsAnotherUpdateStatsFromRunning() throws Exception {
String tableName = fullTableName;
HRegionInfo regionInfo = createTableAndGetRegion(tableName);
- markRunningUpdateStats(regionInfo);
- Assert.assertEquals("Row count didn't match", CONCURRENT_UPDATE_STATS_ROW_COUNT,
- runUpdateStats(tableName));
+ HashSet<byte[]> familyMap = new HashSet<byte[]>(Arrays.asList(Bytes.toBytes("0")));
+ markRunningUpdateStats(regionInfo,familyMap);
+ //there will be no update for a table but local index should succeed, so checking 2 * COMPACTION_UPDATE_STATS_ROW_COUNT
+ assertTrue("Local index stats are not updated!", CONCURRENT_UPDATE_STATS_ROW_COUNT < runUpdateStats(tableName));
// assert that running the concurrent and race-losing update stats didn't clear the region
// from the tracker. If the method returned true it means the tracker was still tracking
// the region. Slightly counter-intuitive, yes.
- assertTrue(tracker.removeUpdateStatsCommandRegion(regionInfo));
+ assertTrue(tracker.removeUpdateStatsCommandRegion(regionInfo,familyMap));
}
private void markRegionAsCompacting(HRegionInfo regionInfo) {
@@ -130,10 +137,10 @@ public class StatisticsCollectionRunTrackerIT extends ParallelStatsEnabledIT {
tracker.addCompactingRegion(regionInfo);
}
- private void markRunningUpdateStats(HRegionInfo regionInfo) {
+ private void markRunningUpdateStats(HRegionInfo regionInfo, HashSet<byte[]> familyMap) {
StatisticsCollectionRunTracker tracker =
StatisticsCollectionRunTracker.getInstance(new Configuration());
- tracker.addUpdateStatsCommandRegion(regionInfo);
+ tracker.addUpdateStatsCommandRegion(regionInfo, familyMap);
}
private HRegionInfo createTableAndGetRegion(String tableName) throws Exception {
@@ -141,6 +148,7 @@ public class StatisticsCollectionRunTrackerIT extends ParallelStatsEnabledIT {
String ddl = "CREATE TABLE " + tableName + " (PK1 VARCHAR PRIMARY KEY, KV1 VARCHAR)";
try (Connection conn = DriverManager.getConnection(getUrl())) {
conn.createStatement().execute(ddl);
+ conn.createStatement().execute("CREATE LOCAL INDEX " + tableName + "_IDX ON " + tableName + "(KV1)");
PhoenixConnection phxConn = conn.unwrap(PhoenixConnection.class);
try (HBaseAdmin admin = phxConn.getQueryServices().getAdmin()) {
List<HRegionInfo> tableRegions = admin.getTableRegions(tableNameBytes);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c5a0e00f/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
index 14213f4..c325d70 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
@@ -1129,7 +1129,7 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
long rowCount = 0; // in case of async, we report 0 as number of rows updated
StatisticsCollectionRunTracker statsRunTracker =
StatisticsCollectionRunTracker.getInstance(config);
- boolean runUpdateStats = statsRunTracker.addUpdateStatsCommandRegion(region.getRegionInfo());
+ boolean runUpdateStats = statsRunTracker.addUpdateStatsCommandRegion(region.getRegionInfo(),scan.getFamilyMap().keySet());
if (runUpdateStats) {
if (!async) {
rowCount = callable.call();
@@ -1248,7 +1248,7 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
}
} finally {
try {
- StatisticsCollectionRunTracker.getInstance(config).removeUpdateStatsCommandRegion(region.getRegionInfo());
+ StatisticsCollectionRunTracker.getInstance(config).removeUpdateStatsCommandRegion(region.getRegionInfo(), scan.getFamilyMap().keySet());
statsCollector.close();
} finally {
try {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c5a0e00f/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 9979ae7..21391f3 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -1168,7 +1168,15 @@ public class MetaDataClient {
// If the table is a view, then we will end up calling update stats
// here for all the view indexes on it. We take care of local indexes later.
if (index.getIndexType() != IndexType.LOCAL) {
- rowCount += updateStatisticsInternal(table.getPhysicalName(), index, updateStatisticsStmt.getProps(), true);
+ if (index.getIndexType() != IndexType.LOCAL) {
+ if (table.getType() != PTableType.VIEW) {
+ rowCount += updateStatisticsInternal(index.getPhysicalName(), index,
+ updateStatisticsStmt.getProps(), true);
+ } else {
+ rowCount += updateStatisticsInternal(table.getPhysicalName(), index,
+ updateStatisticsStmt.getProps(), true);
+ }
+ }
}
}
/*
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c5a0e00f/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectionRunTracker.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectionRunTracker.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectionRunTracker.java
index 560fc0a..c312b9e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectionRunTracker.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectionRunTracker.java
@@ -29,6 +29,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.util.ByteUtil;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -38,8 +39,8 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder;
*/
public class StatisticsCollectionRunTracker {
private static volatile StatisticsCollectionRunTracker INSTANCE;
- private final Set<HRegionInfo> updateStatsRegions = Collections
- .newSetFromMap(new ConcurrentHashMap<HRegionInfo, Boolean>());
+ private final Set<ColumnFamilyRegionInfo> updateStatsRegions = Collections
+ .newSetFromMap(new ConcurrentHashMap<ColumnFamilyRegionInfo, Boolean>());
private final Set<HRegionInfo> compactingRegions = Collections
.newSetFromMap(new ConcurrentHashMap<HRegionInfo, Boolean>());
private final ExecutorService executor;
@@ -103,16 +104,16 @@ public class StatisticsCollectionRunTracker {
* @param regionInfo for the region to run UPDATE STATISTICS command on.
* @return true if UPDATE STATISTICS wasn't already running on the region, false otherwise.
*/
- public boolean addUpdateStatsCommandRegion(HRegionInfo regionInfo) {
- return updateStatsRegions.add(regionInfo);
+ public boolean addUpdateStatsCommandRegion(HRegionInfo regionInfo, Set<byte[]> familySet) {
+ return updateStatsRegions.add(new ColumnFamilyRegionInfo(regionInfo,familySet));
}
/**
* @param regionInfo for the region to mark as not running UPDATE STATISTICS command on.
* @return true if UPDATE STATISTICS was running on the region, false otherwise.
*/
- public boolean removeUpdateStatsCommandRegion(HRegionInfo regionInfo) {
- return updateStatsRegions.remove(regionInfo);
+ public boolean removeUpdateStatsCommandRegion(HRegionInfo regionInfo, Set<byte[]> familySet) {
+ return updateStatsRegions.remove(new ColumnFamilyRegionInfo(regionInfo,familySet));
}
/**
@@ -124,4 +125,36 @@ public class StatisticsCollectionRunTracker {
return executor.submit(c);
}
+ class ColumnFamilyRegionInfo {
+ private HRegionInfo regionInfo;
+ private Set<byte[]> familySet;
+
+ public ColumnFamilyRegionInfo(HRegionInfo regionInfo, Set<byte[]> familySet) {
+ this.regionInfo = regionInfo;
+ this.familySet = familySet;
+ }
+
+ public HRegionInfo getRegionInfo() {
+ return regionInfo;
+ }
+
+ public Set<byte[]> getFamilySet() {
+ return familySet;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) { return true; }
+ if (!(obj instanceof ColumnFamilyRegionInfo)) { return false; }
+
+ ColumnFamilyRegionInfo c = (ColumnFamilyRegionInfo)obj;
+ return c.getRegionInfo().equals(this.regionInfo) && ByteUtil.match(this.familySet, c.getFamilySet());
+ }
+
+ @Override
+ public int hashCode() {
+ return this.getRegionInfo().hashCode();
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c5a0e00f/phoenix-core/src/main/java/org/apache/phoenix/util/ByteUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ByteUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ByteUtil.java
index d11f3a2..5a2b624 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/ByteUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ByteUtil.java
@@ -24,8 +24,10 @@ import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Comparator;
import java.util.List;
+import java.util.Set;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
@@ -579,7 +581,7 @@ public class ByteUtil {
}
}
- public static boolean contains(List<byte[]> keys, byte[] key) {
+ public static boolean contains(Collection<byte[]> keys, byte[] key) {
for (byte[] k : keys) {
if (Arrays.equals(k, key)) { return true; }
}
@@ -592,4 +594,16 @@ public class ByteUtil {
}
return false;
}
+
+ public static boolean match(Set<byte[]> keys, Set<byte[]> keys2) {
+ if (keys == keys2) return true;
+ if (keys == null || keys2 == null) return false;
+
+ int size = keys.size();
+ if (keys2.size() != size) return false;
+ for (byte[] k : keys) {
+ if (!contains(keys2, k)) { return false; }
+ }
+ return true;
+ }
}
[03/11] phoenix git commit: PHOENIX-4758 Validate that
HADOOP_CONF_DIR is not set for HiveMRIT
Posted by pb...@apache.org.
PHOENIX-4758 Validate that HADOOP_CONF_DIR is not set for HiveMRIT
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/6916b8d5
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/6916b8d5
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/6916b8d5
Branch: refs/heads/4.x-cdh5.12
Commit: 6916b8d54ed8156ec6309663512c7fe606ad6be2
Parents: be882e9
Author: Josh Elser <el...@apache.org>
Authored: Tue May 29 19:14:04 2018 +0100
Committer: Pedro Boado <pb...@apache.org>
Committed: Sat Jun 9 19:42:06 2018 +0100
----------------------------------------------------------------------
.../src/it/java/org/apache/phoenix/hive/HiveMapReduceIT.java | 8 ++++++++
1 file changed, 8 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/6916b8d5/phoenix-hive/src/it/java/org/apache/phoenix/hive/HiveMapReduceIT.java
----------------------------------------------------------------------
diff --git a/phoenix-hive/src/it/java/org/apache/phoenix/hive/HiveMapReduceIT.java b/phoenix-hive/src/it/java/org/apache/phoenix/hive/HiveMapReduceIT.java
index 644ff24..4bc5a7d 100644
--- a/phoenix-hive/src/it/java/org/apache/phoenix/hive/HiveMapReduceIT.java
+++ b/phoenix-hive/src/it/java/org/apache/phoenix/hive/HiveMapReduceIT.java
@@ -18,6 +18,10 @@
package org.apache.phoenix.hive;
+import static org.junit.Assert.fail;
+
+import java.util.Map;
+
import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
import org.junit.BeforeClass;
import org.junit.experimental.categories.Category;
@@ -28,6 +32,10 @@ public class HiveMapReduceIT extends HivePhoenixStoreIT {
@BeforeClass
public static void setUpBeforeClass() throws Exception {
+ final String hadoopConfDir = System.getenv("HADOOP_CONF_DIR");
+ if (hadoopConfDir != null && hadoopConfDir.length() != 0) {
+ fail("HADOOP_CONF_DIR is non-empty in the current shell environment which will very likely cause this test to fail.");
+ }
setup(HiveTestUtil.MiniClusterType.mr);
}
}
[04/11] phoenix git commit: PHOENIX-4772 phoenix.sequence.saltBuckets
is not honoured
Posted by pb...@apache.org.
PHOENIX-4772 phoenix.sequence.saltBuckets is not honoured
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/91ffc048
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/91ffc048
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/91ffc048
Branch: refs/heads/4.x-cdh5.12
Commit: 91ffc048cda58cd986ea0ae74d857568801d15a2
Parents: 6916b8d
Author: Ankit Singhal <an...@gmail.com>
Authored: Thu Jun 7 19:23:44 2018 +0100
Committer: Pedro Boado <pb...@apache.org>
Committed: Sat Jun 9 19:42:27 2018 +0100
----------------------------------------------------------------------
.../org/apache/phoenix/end2end/SequenceIT.java | 57 +---------
.../phoenix/end2end/SequencePointInTimeIT.java | 112 +++++++++++++++++++
.../query/ConnectionQueryServicesImpl.java | 13 ++-
.../query/ConnectionlessQueryServicesImpl.java | 7 +-
.../apache/phoenix/query/QueryConstants.java | 2 +
.../org/apache/phoenix/schema/Sequence.java | 6 +-
6 files changed, 139 insertions(+), 58 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/91ffc048/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequenceIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequenceIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequenceIT.java
index 4cc9628..b76cc4e 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequenceIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequenceIT.java
@@ -18,6 +18,8 @@
package org.apache.phoenix.end2end;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE_SEQUENCE;
import static org.apache.phoenix.query.QueryServicesTestImpl.DEFAULT_SEQUENCE_CACHE_SIZE;
import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
import static org.junit.Assert.assertEquals;
@@ -38,6 +40,7 @@ import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.jdbc.PhoenixStatement;
import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesTestImpl;
import org.apache.phoenix.schema.SchemaNotFoundException;
import org.apache.phoenix.schema.SequenceAlreadyExistsException;
import org.apache.phoenix.schema.SequenceNotFoundException;
@@ -202,6 +205,8 @@ public class SequenceIT extends ParallelStatsDisabledIT {
String schemaName = getSchemaName(sequenceName);
conn.createStatement().execute("CREATE SEQUENCE " + sequenceName + " START WITH 2 INCREMENT BY 4");
+ int bucketNum = PhoenixRuntime.getTableNoCache(conn, SYSTEM_CATALOG_SCHEMA + "." + TYPE_SEQUENCE).getBucketNum();
+ assertEquals("Salt bucket for SYSTEM.SEQUENCE should be test default",bucketNum , QueryServicesTestImpl.DEFAULT_SEQUENCE_TABLE_SALT_BUCKETS);
String query = "SELECT sequence_schema, sequence_name, current_value, increment_by FROM \"SYSTEM\".\"SEQUENCE\" WHERE sequence_name='" + sequenceNameWithoutSchema + "'";
ResultSet rs = conn.prepareStatement(query).executeQuery();
assertTrue(rs.next());
@@ -1406,56 +1411,4 @@ public class SequenceIT extends ParallelStatsDisabledIT {
return tableName.substring(tableName.indexOf(".") + 1, tableName.length());
}
- @Test
- public void testPointInTimeSequence() throws Exception {
- String seqName = generateSequenceNameWithSchema();
- Properties scnProps = PropertiesUtil.deepCopy(TEST_PROPERTIES);
- scnProps.put(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(EnvironmentEdgeManager.currentTimeMillis()));
- Connection beforeSeqConn = DriverManager.getConnection(getUrl(), scnProps);
-
- ResultSet rs;
- Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
- Connection conn = DriverManager.getConnection(getUrl(), props);
- conn.createStatement().execute("CREATE SEQUENCE " + seqName + "");
-
- try {
- beforeSeqConn.createStatement().executeQuery("SELECT next value for " + seqName);
- fail();
- } catch (SequenceNotFoundException e) {
- beforeSeqConn.close();
- }
-
- scnProps.put(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(EnvironmentEdgeManager.currentTimeMillis()));
- Connection afterSeqConn = DriverManager.getConnection(getUrl(), scnProps);
-
- rs = conn.createStatement().executeQuery("SELECT next value for " + seqName);
- assertTrue(rs.next());
- assertEquals(1, rs.getInt(1));
- rs = conn.createStatement().executeQuery("SELECT next value for " + seqName);
- assertTrue(rs.next());
- assertEquals(2, rs.getInt(1));
-
- conn.createStatement().execute("DROP SEQUENCE " + seqName + "");
-
- rs = afterSeqConn.createStatement().executeQuery("SELECT next value for " + seqName);
- assertTrue(rs.next());
- assertEquals(3, rs.getInt(1));
-
- try {
- rs = conn.createStatement().executeQuery("SELECT next value for " + seqName);
- fail();
- } catch (SequenceNotFoundException e) { // expected
- }
-
- conn.createStatement().execute("CREATE SEQUENCE " + seqName);
- rs = conn.createStatement().executeQuery("SELECT next value for " + seqName);
- assertTrue(rs.next());
- assertEquals(1, rs.getInt(1));
-
- rs = afterSeqConn.createStatement().executeQuery("SELECT next value for " + seqName);
- assertTrue(rs.next());
- assertEquals(4, rs.getInt(1));
- afterSeqConn.close();
- }
-
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/91ffc048/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequencePointInTimeIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequencePointInTimeIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequencePointInTimeIT.java
new file mode 100644
index 0000000..3608901
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequencePointInTimeIT.java
@@ -0,0 +1,112 @@
+/*
+ * 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.end2end;
+
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+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.DriverManager;
+import java.sql.ResultSet;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.schema.SequenceNotFoundException;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.SchemaUtil;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.collect.Maps;
+
+
+public class SequencePointInTimeIT extends BaseUniqueNamesOwnClusterIT {
+ private static final String SCHEMA_NAME = "S";
+
+ private static String generateSequenceNameWithSchema() {
+ return SchemaUtil.getTableName(SCHEMA_NAME, generateUniqueSequenceName());
+ }
+ @BeforeClass
+ public static void doSetup() throws Exception {
+ Map<String,String> props = Maps.newHashMapWithExpectedSize(5);
+ // Must update config before starting server
+ props.put(QueryServices.DEFAULT_SYSTEM_KEEP_DELETED_CELLS_ATTRIB, Boolean.TRUE.toString());
+ props.put(QueryServices.DEFAULT_SYSTEM_MAX_VERSIONS_ATTRIB, "5");
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+ }
+
+ @Test
+ public void testPointInTimeSequence() throws Exception {
+ String seqName = generateSequenceNameWithSchema();
+ Properties scnProps = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ scnProps.put(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(EnvironmentEdgeManager.currentTimeMillis()));
+ Connection beforeSeqConn = DriverManager.getConnection(getUrl(), scnProps);
+
+ ResultSet rs;
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ Connection conn = DriverManager.getConnection(getUrl(), props);
+ conn.createStatement().execute("CREATE SEQUENCE " + seqName + "");
+
+ try {
+ beforeSeqConn.createStatement().executeQuery("SELECT next value for " + seqName);
+ fail();
+ } catch (SequenceNotFoundException e) {
+ beforeSeqConn.close();
+ }
+
+ scnProps.put(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(EnvironmentEdgeManager.currentTimeMillis()));
+ Connection afterSeqConn = DriverManager.getConnection(getUrl(), scnProps);
+
+ rs = conn.createStatement().executeQuery("SELECT next value for " + seqName);
+ assertTrue(rs.next());
+ assertEquals(1, rs.getInt(1));
+ rs = conn.createStatement().executeQuery("SELECT next value for " + seqName);
+ assertTrue(rs.next());
+ assertEquals(2, rs.getInt(1));
+
+ conn.createStatement().execute("DROP SEQUENCE " + seqName + "");
+
+ rs = afterSeqConn.createStatement().executeQuery("SELECT next value for " + seqName);
+ assertTrue(rs.next());
+ assertEquals(3, rs.getInt(1));
+
+ try {
+ rs = conn.createStatement().executeQuery("SELECT next value for " + seqName);
+ fail();
+ } catch (SequenceNotFoundException e) { // expected
+ }
+
+ conn.createStatement().execute("CREATE SEQUENCE " + seqName);
+ rs = conn.createStatement().executeQuery("SELECT next value for " + seqName);
+ assertTrue(rs.next());
+ assertEquals(1, rs.getInt(1));
+
+ rs = afterSeqConn.createStatement().executeQuery("SELECT next value for " + seqName);
+ assertTrue(rs.next());
+ assertEquals(4, rs.getInt(1));
+ afterSeqConn.close();
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/91ffc048/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 4290d23..bfaffbd 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
@@ -2464,6 +2464,11 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
return setSystemDDLProperties(QueryConstants.CREATE_TABLE_METADATA);
}
+ protected String getSystemSequenceTableDDL(int nSaltBuckets) {
+ String schema = String.format(setSystemDDLProperties(QueryConstants.CREATE_SEQUENCE_METADATA));
+ return Sequence.getCreateTableStatement(schema, nSaltBuckets);
+ }
+
// Available for testing
protected String getFunctionTableDDL() {
return setSystemDDLProperties(QueryConstants.CREATE_FUNCTION_METADATA);
@@ -2679,7 +2684,11 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
private void createOtherSystemTables(PhoenixConnection metaConnection, HBaseAdmin hbaseAdmin) throws SQLException, IOException {
try {
- metaConnection.createStatement().execute(QueryConstants.CREATE_SEQUENCE_METADATA);
+
+ nSequenceSaltBuckets = ConnectionQueryServicesImpl.this.props.getInt(
+ QueryServices.SEQUENCE_SALT_BUCKETS_ATTRIB,
+ QueryServicesOptions.DEFAULT_SEQUENCE_TABLE_SALT_BUCKETS);
+ metaConnection.createStatement().execute(getSystemSequenceTableDDL(nSequenceSaltBuckets));
} catch (TableAlreadyExistsException e) {
nSequenceSaltBuckets = getSaltBuckets(e);
}
@@ -3054,7 +3063,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
QueryServices.SEQUENCE_SALT_BUCKETS_ATTRIB,
QueryServicesOptions.DEFAULT_SEQUENCE_TABLE_SALT_BUCKETS);
try {
- String createSequenceTable = Sequence.getCreateTableStatement(nSaltBuckets);
+ String createSequenceTable = getSystemSequenceTableDDL(nSaltBuckets);
metaConnection.createStatement().executeUpdate(createSequenceTable);
nSequenceSaltBuckets = nSaltBuckets;
} catch (NewerTableAlreadyExistsException e) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/91ffc048/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
index cbc73ba..caec670 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
@@ -159,6 +159,11 @@ public class ConnectionlessQueryServicesImpl extends DelegateQueryServices imple
return setSystemDDLProperties(QueryConstants.CREATE_TABLE_METADATA);
}
+ protected String getSystemSequenceTableDDL(int nSaltBuckets) {
+ String schema = String.format(setSystemDDLProperties(QueryConstants.CREATE_SEQUENCE_METADATA));
+ return Sequence.getCreateTableStatement(schema, nSaltBuckets);
+ }
+
protected String getFunctionTableDDL() {
return setSystemDDLProperties(QueryConstants.CREATE_FUNCTION_METADATA);
}
@@ -343,7 +348,7 @@ public class ConnectionlessQueryServicesImpl extends DelegateQueryServices imple
}
try {
int nSaltBuckets = getSequenceSaltBuckets();
- String createTableStatement = Sequence.getCreateTableStatement(nSaltBuckets);
+ String createTableStatement = getSystemSequenceTableDDL(nSaltBuckets);
metaConnection.createStatement().executeUpdate(createTableStatement);
} catch (NewerTableAlreadyExistsException ignore) {
// Ignore, as this will happen if the SYSTEM.SEQUENCE already exists at this fixed timestamp.
http://git-wip-us.apache.org/repos/asf/phoenix/blob/91ffc048/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
index 52abed0..b31175a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
@@ -279,6 +279,8 @@ public interface QueryConstants {
CYCLE_FLAG + " BOOLEAN, \n" +
LIMIT_REACHED_FLAG + " BOOLEAN \n" +
" CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (" + TENANT_ID + "," + SEQUENCE_SCHEMA + "," + SEQUENCE_NAME + "))\n" +
+ HConstants.VERSIONS + "=%s,\n" +
+ HColumnDescriptor.KEEP_DELETED_CELLS + "=%s,\n"+
PhoenixDatabaseMetaData.TRANSACTIONAL + "=" + Boolean.FALSE;
public static final String CREATE_SYSTEM_SCHEMA = "CREATE SCHEMA " + SYSTEM_CATALOG_SCHEMA;
public static final String UPGRADE_TABLE_SNAPSHOT_PREFIX = "_UPGRADING_TABLE_";
http://git-wip-us.apache.org/repos/asf/phoenix/blob/91ffc048/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 9598ace..e3704dc 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
@@ -626,10 +626,10 @@ public class Sequence {
.build().buildException();
}
- public static String getCreateTableStatement(int nSaltBuckets) {
+ public static String getCreateTableStatement(String schema, int nSaltBuckets) {
if (nSaltBuckets <= 0) {
- return QueryConstants.CREATE_SEQUENCE_METADATA;
+ return schema;
}
- return QueryConstants.CREATE_SEQUENCE_METADATA + "," + PhoenixDatabaseMetaData.SALT_BUCKETS + "=" + nSaltBuckets;
+ return schema + "," + PhoenixDatabaseMetaData.SALT_BUCKETS + "=" + nSaltBuckets;
}
}
[06/11] phoenix git commit: PHOENIX-4776 Remove creation of .md5
files from dev/make_rc.sh
Posted by pb...@apache.org.
PHOENIX-4776 Remove creation of .md5 files from dev/make_rc.sh
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/f95f860e
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/f95f860e
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/f95f860e
Branch: refs/heads/4.x-cdh5.12
Commit: f95f860e4745f6334f2b1c11ca3ca100d56dba44
Parents: c5a0e00
Author: Pedro Boado <pb...@apache.org>
Authored: Mon Jun 11 23:48:34 2018 +0100
Committer: Pedro Boado <pb...@apache.org>
Committed: Mon Jun 11 23:48:34 2018 +0100
----------------------------------------------------------------------
dev/make_rc.sh | 2 --
1 file changed, 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f95f860e/dev/make_rc.sh
----------------------------------------------------------------------
diff --git a/dev/make_rc.sh b/dev/make_rc.sh
index 9bff7d1..963a9d8 100755
--- a/dev/make_rc.sh
+++ b/dev/make_rc.sh
@@ -123,13 +123,11 @@ function_sign() {
# if on MAC OS
if [[ "$OSTYPE" == "darwin"* ]]; then
gpg2 --armor --output $file.asc --detach-sig $file;
- openssl md5 $file > $file.md5;
openssl dgst -sha512 $file > $file.sha512;
openssl dgst -sha256 $file >> $file.sha256;
# all other OS
else
gpg --armor --output $file.asc --detach-sig $file;
- md5sum -b $file > $file.md5;
sha512sum -b $file > $file.sha512;
sha256sum -b $file >> $file.sha256;
fi
[02/11] phoenix git commit: PHOENIX-4774 Disable doclint in 1.8+ JDKs
Posted by pb...@apache.org.
PHOENIX-4774 Disable doclint in 1.8+ JDKs
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/be882e91
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/be882e91
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/be882e91
Branch: refs/heads/4.x-cdh5.12
Commit: be882e911a063d4dbb38b6dc4b248fc335e555be
Parents: 3aaff76
Author: Alex Araujo <al...@gmail.com>
Authored: Tue Jun 5 11:20:17 2018 -0700
Committer: Vincent Poon <vi...@apache.org>
Committed: Wed Jun 6 11:21:11 2018 -0700
----------------------------------------------------------------------
pom.xml | 13 +++++++++++++
1 file changed, 13 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/be882e91/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index bea1199..e55b080 100644
--- a/pom.xml
+++ b/pom.xml
@@ -451,6 +451,9 @@
<!-- TODO turn back on javadocs - disabled now for testing -->
<!-- <goal>jar</goal> -->
</goals>
+ <configuration>
+ <additionalparam>${javadoc.opts}</additionalparam>
+ </configuration>
</execution>
</executions>
</plugin>
@@ -1004,6 +1007,16 @@
</dependencyManagement>
<profiles>
+ <!-- disable doclint with 1.8+ JDKs-->
+ <profile>
+ <id>java8-doclint-disabled</id>
+ <activation>
+ <jdk>[1.8,)</jdk>
+ </activation>
+ <properties>
+ <javadoc.opts>-Xdoclint:none</javadoc.opts>
+ </properties>
+ </profile>
<!-- this profile should be activated for release builds -->
<profile>
<id>release</id>
[08/11] phoenix git commit: PHOENIX-4789 Exception when setting TTL
on Tephra transactional table
Posted by pb...@apache.org.
PHOENIX-4789 Exception when setting TTL on Tephra transactional table
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/35e09ce4
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/35e09ce4
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/35e09ce4
Branch: refs/heads/4.x-cdh5.12
Commit: 35e09ce48c6f9f1d3964a50302f486c8da2356db
Parents: fdebb15
Author: James Taylor <ja...@apache.org>
Authored: Mon Jun 18 15:00:02 2018 +0200
Committer: James Taylor <ja...@apache.org>
Committed: Mon Jun 18 15:12:45 2018 +0200
----------------------------------------------------------------------
.../org/apache/phoenix/tx/TransactionIT.java | 30 ++++++++++++++++++++
.../query/ConnectionQueryServicesImpl.java | 3 ++
2 files changed, 33 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/35e09ce4/phoenix-core/src/it/java/org/apache/phoenix/tx/TransactionIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/tx/TransactionIT.java b/phoenix-core/src/it/java/org/apache/phoenix/tx/TransactionIT.java
index c0ec6b8..12c3b7a 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/tx/TransactionIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/tx/TransactionIT.java
@@ -25,6 +25,7 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import java.io.IOException;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.DriverManager;
@@ -38,6 +39,9 @@ import java.util.Properties;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.TableNotFoundException;
+import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.end2end.ParallelStatsDisabledIT;
import org.apache.phoenix.exception.SQLExceptionCode;
@@ -54,6 +58,7 @@ import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.StringUtil;
import org.apache.phoenix.util.TestUtil;
+import org.apache.tephra.TxConstants;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -370,4 +375,29 @@ public class TransactionIT extends ParallelStatsDisabledIT {
conn.close();
}
}
+
+ private static void assertTTL(Admin admin, String tableName, int ttl) throws TableNotFoundException, IOException {
+ HTableDescriptor tableDesc = admin.getTableDescriptor(TableName.valueOf(tableName));
+ for (HColumnDescriptor colDesc : tableDesc.getFamilies()) {
+ assertEquals(ttl,Integer.parseInt(colDesc.getValue(TxConstants.PROPERTY_TTL)));
+ assertEquals(HColumnDescriptor.DEFAULT_TTL,colDesc.getTimeToLive());
+ }
+ }
+
+ @Test
+ public void testSetTTL() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ TransactionFactory.Provider txProvider = TransactionFactory.Provider.valueOf(this.txProvider);
+ try (Connection conn = DriverManager.getConnection(getUrl(), props); Admin admin = conn.unwrap(PhoenixConnection.class).getQueryServices().getAdmin()) {
+ String tableName = generateUniqueName();
+ conn.createStatement().execute("CREATE TABLE " + tableName +
+ "(K VARCHAR PRIMARY KEY) TRANSACTIONAL=true,TRANSACTION_PROVIDER='" + txProvider + "',TTL=100");
+ assertTTL(admin, tableName, 100);
+ tableName = generateUniqueName();
+ conn.createStatement().execute("CREATE TABLE " + tableName +
+ "(K VARCHAR PRIMARY KEY) TRANSACTIONAL=true,TRANSACTION_PROVIDER='" + txProvider + "'");
+ conn.createStatement().execute("ALTER TABLE " + tableName + " SET TTL=" + 200);
+ assertTTL(admin, tableName, 200);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/35e09ce4/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 bfaffbd..f8ca7a4 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
@@ -2234,6 +2234,8 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
Map<String, Object> props = entry.getValue();
if (props == null) {
props = new HashMap<String, Object>();
+ } else {
+ props = new HashMap<String, Object>(props);
}
props.put(PhoenixTransactionContext.PROPERTY_TTL, ttl);
// Remove HBase TTL if we're not transitioning an existing table to become transactional
@@ -2241,6 +2243,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
if (!willBeTransactional && !Boolean.valueOf(newTableDescriptor.getValue(PhoenixTransactionContext.READ_NON_TX_DATA))) {
props.remove(TTL);
}
+ entry.setValue(props);
}
}
}
[10/11] phoenix git commit: PHOENIX-4785 Unable to write to table if
index is made active during retry
Posted by pb...@apache.org.
PHOENIX-4785 Unable to write to table if index is made active during retry
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/a162c74c
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/a162c74c
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/a162c74c
Branch: refs/heads/4.x-cdh5.12
Commit: a162c74cdf6939edec0b41e1081b08cbf0258b89
Parents: 5edcd6b
Author: Ankit Singhal <an...@gmail.com>
Authored: Fri Jun 22 00:13:50 2018 +0100
Committer: Pedro Boado <pb...@apache.org>
Committed: Thu Jun 28 00:57:44 2018 +0100
----------------------------------------------------------------------
.../end2end/index/MutableIndexFailureIT.java | 128 ++++++++++++++-----
.../MutableIndexFailureWithNamespaceIT.java | 80 ++++++++++++
.../coprocessor/MetaDataEndpointImpl.java | 30 +++++
.../index/PhoenixIndexFailurePolicy.java | 71 +++++++++-
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 2 +-
5 files changed, 276 insertions(+), 35 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/a162c74c/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java
index 715e37f..aac20ac 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java
@@ -28,10 +28,16 @@ import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.DoNotRetryIOException;
@@ -104,10 +110,10 @@ public class MutableIndexFailureIT extends BaseTest {
private final boolean throwIndexWriteFailure;
private String schema = generateUniqueName();
private List<CommitException> exceptions = Lists.newArrayList();
- private static RegionCoprocessorEnvironment indexRebuildTaskRegionEnvironment;
- private static final int forwardOverlapMs = 1000;
- private static final int disableTimestampThresholdMs = 10000;
- private static final int numRpcRetries = 2;
+ protected static RegionCoprocessorEnvironment indexRebuildTaskRegionEnvironment;
+ protected static final int forwardOverlapMs = 1000;
+ protected static final int disableTimestampThresholdMs = 10000;
+ protected static final int numRpcRetries = 2;
public MutableIndexFailureIT(boolean transactional, boolean localIndex, boolean isNamespaceMapped, Boolean disableIndexOnWriteFailure, boolean failRebuildTask, Boolean throwIndexWriteFailure) {
this.transactional = transactional;
@@ -127,6 +133,23 @@ public class MutableIndexFailureIT extends BaseTest {
@BeforeClass
public static void doSetup() throws Exception {
+ Map<String, String> serverProps = getServerProps();
+ Map<String, String> clientProps = Maps.newHashMapWithExpectedSize(2);
+ clientProps.put(HConstants.HBASE_CLIENT_RETRIES_NUMBER, "2");
+ NUM_SLAVES_BASE = 4;
+ setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()), new ReadOnlyProps(clientProps.entrySet().iterator()));
+ indexRebuildTaskRegionEnvironment =
+ (RegionCoprocessorEnvironment) getUtility()
+ .getRSForFirstRegionInTable(
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG_HBASE_TABLE_NAME)
+ .getOnlineRegions(PhoenixDatabaseMetaData.SYSTEM_CATALOG_HBASE_TABLE_NAME)
+ .get(0).getCoprocessorHost()
+ .findCoprocessorEnvironment(MetaDataRegionObserver.class.getName());
+ MetaDataRegionObserver.initRebuildIndexConnectionProps(
+ indexRebuildTaskRegionEnvironment.getConfiguration());
+ }
+
+ protected static Map<String,String> getServerProps(){
Map<String, String> serverProps = Maps.newHashMapWithExpectedSize(10);
serverProps.put("hbase.coprocessor.region.classes", FailingRegionObserver.class.getName());
serverProps.put(HConstants.HBASE_RPC_TIMEOUT_KEY, "10000");
@@ -142,19 +165,7 @@ public class MutableIndexFailureIT extends BaseTest {
* because we want to control it's execution ourselves
*/
serverProps.put(QueryServices.INDEX_REBUILD_TASK_INITIAL_DELAY, Long.toString(Long.MAX_VALUE));
- Map<String, String> clientProps = Maps.newHashMapWithExpectedSize(2);
- clientProps.put(HConstants.HBASE_CLIENT_RETRIES_NUMBER, "2");
- NUM_SLAVES_BASE = 4;
- setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()), new ReadOnlyProps(clientProps.entrySet().iterator()));
- indexRebuildTaskRegionEnvironment =
- (RegionCoprocessorEnvironment) getUtility()
- .getRSForFirstRegionInTable(
- PhoenixDatabaseMetaData.SYSTEM_CATALOG_HBASE_TABLE_NAME)
- .getOnlineRegions(PhoenixDatabaseMetaData.SYSTEM_CATALOG_HBASE_TABLE_NAME)
- .get(0).getCoprocessorHost()
- .findCoprocessorEnvironment(MetaDataRegionObserver.class.getName());
- MetaDataRegionObserver.initRebuildIndexConnectionProps(
- indexRebuildTaskRegionEnvironment.getConfiguration());
+ return serverProps;
}
@Parameters(name = "MutableIndexFailureIT_transactional={0},localIndex={1},isNamespaceMapped={2},disableIndexOnWriteFailure={3},failRebuildTask={4},throwIndexWriteFailure={5}") // name is used by failsafe as file name in reports
@@ -162,16 +173,10 @@ public class MutableIndexFailureIT extends BaseTest {
return Arrays.asList(new Object[][] {
// note - can't disableIndexOnWriteFailure without throwIndexWriteFailure, PHOENIX-4130
{ false, false, false, false, false, false},
- { false, false, true, true, false, null},
- { false, false, true, true, false, true},
{ false, false, false, true, false, null},
{ true, false, false, true, false, null},
- { true, false, true, true, false, null},
- { false, true, true, true, false, null},
{ false, true, false, null, false, null},
{ true, true, false, true, false, null},
- { true, true, true, null, false, null},
-
{ false, false, false, false, false, null},
{ false, true, false, false, false, null},
{ false, false, false, false, false, null},
@@ -180,9 +185,7 @@ public class MutableIndexFailureIT extends BaseTest {
{ false, true, false, true, false, null},
{ false, true, false, true, false, null},
{ false, false, false, true, true, null},
- { false, false, true, true, true, null},
{ false, false, false, false, true, false},
- { false, false, true, false, true, false},
}
);
}
@@ -258,6 +261,9 @@ public class MutableIndexFailureIT extends BaseTest {
rs = conn.createStatement().executeQuery(query);
assertFalse(rs.next());
+ initializeTable(conn, fullTableName);
+ addRowsInTableDuringRetry(fullTableName);
+
// Verify the metadata for index is correct.
rs = conn.getMetaData().getTables(null, StringUtil.escapeLike(schema), null,
new String[] { PTableType.INDEX.toString() });
@@ -270,8 +276,9 @@ public class MutableIndexFailureIT extends BaseTest {
assertTrue(rs.next());
assertEquals(thirdIndexName, rs.getString(3));
assertEquals(PIndexState.ACTIVE.toString(), rs.getString("INDEX_STATE"));
- initializeTable(conn, fullTableName);
-
+ // we should be able to write to ACTIVE index even in case of disable index on failure policy
+ addRowToTable(conn, fullTableName);
+
query = "SELECT /*+ NO_INDEX */ k,v1 FROM " + fullTableName;
rs = conn.createStatement().executeQuery("EXPLAIN " + query);
String expectedPlan = "CLIENT PARALLEL 2-WAY FULL SCAN OVER "
@@ -402,16 +409,67 @@ public class MutableIndexFailureIT extends BaseTest {
stmt.setString(2, "x");
stmt.setString(3, "1");
stmt.execute();
- stmt.setString(1, "b");
- stmt.setString(2, "y");
- stmt.setString(3, "2");
- stmt.execute();
+ conn.commit();
+ }
+
+ private void addRowToTable(Connection conn, String tableName) throws SQLException {
+ PreparedStatement stmt = conn.prepareStatement("UPSERT INTO " + tableName + " VALUES(?,?,?)");
stmt.setString(1, "c");
stmt.setString(2, "z");
stmt.setString(3, "3");
stmt.execute();
conn.commit();
+ }
+ private void addRowsInTableDuringRetry(final String tableName)
+ throws SQLException, InterruptedException, ExecutionException {
+ int threads=10;
+ boolean wasFailWrite = FailingRegionObserver.FAIL_WRITE;
+ boolean wasToggleFailWriteForRetry = FailingRegionObserver.TOGGLE_FAIL_WRITE_FOR_RETRY;
+ try {
+ Callable callable = new Callable() {
+
+ @Override
+ public Boolean call() {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ props.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, String.valueOf(isNamespaceMapped));
+ try (Connection conn = driver.connect(url, props)) {
+ // In case of disable index on failure policy, INDEX will be in PENDING_DISABLE on first retry
+ // but will
+ // become active if retry is successfull
+ PreparedStatement stmt = conn.prepareStatement("UPSERT INTO " + tableName + " VALUES(?,?,?)");
+ stmt.setString(1, "b");
+ stmt.setString(2, "y");
+ stmt.setString(3, "2");
+ stmt.execute();
+ if (!leaveIndexActiveOnFailure && !transactional) {
+ FailingRegionObserver.FAIL_WRITE = true;
+ FailingRegionObserver.TOGGLE_FAIL_WRITE_FOR_RETRY = true;
+ }
+ conn.commit();
+ } catch (SQLException e) {
+ return false;
+ }
+ return true;
+ }
+ };
+ ExecutorService executor = Executors.newFixedThreadPool(threads);
+ List<Future<Boolean>> futures = new ArrayList<Future<Boolean>>();
+ for (int i = 0; i < threads; i++) {
+ futures.add(executor.submit(callable));
+ }
+ for (Future<Boolean> future : futures) {
+ Boolean isSuccess = future.get();
+ // transactions can have conflict so ignoring the check for them
+ if (!transactional) {
+ assertTrue(isSuccess);
+ }
+ }
+ executor.shutdown();
+ } finally {
+ FailingRegionObserver.FAIL_WRITE = wasFailWrite;
+ FailingRegionObserver.TOGGLE_FAIL_WRITE_FOR_RETRY = wasToggleFailWriteForRetry;
+ }
}
private void validateDataWithIndex(Connection conn, String fullTableName, String fullIndexName, boolean localIndex) throws SQLException {
@@ -510,6 +568,7 @@ public class MutableIndexFailureIT extends BaseTest {
}
public static class FailingRegionObserver extends SimpleRegionObserver {
+ public static boolean TOGGLE_FAIL_WRITE_FOR_RETRY = false;
public static volatile boolean FAIL_WRITE = false;
public static final String FAIL_INDEX_NAME = "FAIL_IDX";
public static final String FAIL_TABLE_NAME = "FAIL_TABLE";
@@ -520,6 +579,9 @@ public class MutableIndexFailureIT extends BaseTest {
if (c.getEnvironment().getRegionInfo().getTable().getNameAsString().endsWith("A_" + FAIL_INDEX_NAME)
&& FAIL_WRITE) {
throwException = true;
+ if (TOGGLE_FAIL_WRITE_FOR_RETRY) {
+ FAIL_WRITE = !FAIL_WRITE;
+ }
} else {
// When local index updates are atomic with data updates, testing a write failure to a local
// index won't make sense.
@@ -542,7 +604,9 @@ public class MutableIndexFailureIT extends BaseTest {
}
}
if (throwException) {
- dropIndex(c);
+ if (!TOGGLE_FAIL_WRITE_FOR_RETRY) {
+ dropIndex(c);
+ }
throw new DoNotRetryIOException();
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/a162c74c/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureWithNamespaceIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureWithNamespaceIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureWithNamespaceIT.java
new file mode 100644
index 0000000..5ed9e1f
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureWithNamespaceIT.java
@@ -0,0 +1,80 @@
+/*
+ * 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.end2end.index;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.phoenix.coprocessor.MetaDataRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.SchemaUtil;
+import org.junit.BeforeClass;
+import org.junit.runners.Parameterized.Parameters;
+
+import com.google.common.collect.Maps;
+
+/*
+ * This class is to ensure gets its own cluster with Namespace Enabled
+ */
+public class MutableIndexFailureWithNamespaceIT extends MutableIndexFailureIT {
+
+ public MutableIndexFailureWithNamespaceIT(boolean transactional, boolean localIndex, boolean isNamespaceMapped,
+ Boolean disableIndexOnWriteFailure, boolean failRebuildTask, Boolean throwIndexWriteFailure) {
+ super(transactional, localIndex, isNamespaceMapped, disableIndexOnWriteFailure, failRebuildTask,
+ throwIndexWriteFailure);
+ }
+
+ @BeforeClass
+ public static void doSetup() throws Exception {
+ Map<String, String> serverProps = getServerProps();
+ serverProps.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, Boolean.TRUE.toString());
+ Map<String, String> clientProps = Maps.newHashMapWithExpectedSize(2);
+ clientProps.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, Boolean.TRUE.toString());
+ clientProps.put(HConstants.HBASE_CLIENT_RETRIES_NUMBER, "2");
+ NUM_SLAVES_BASE = 4;
+ setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()), new ReadOnlyProps(clientProps.entrySet().iterator()));
+ TableName systemTable = SchemaUtil.getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES,
+ true);
+ indexRebuildTaskRegionEnvironment = (RegionCoprocessorEnvironment)getUtility()
+ .getRSForFirstRegionInTable(systemTable).getOnlineRegions(systemTable).get(0).getCoprocessorHost()
+ .findCoprocessorEnvironment(MetaDataRegionObserver.class.getName());
+ MetaDataRegionObserver.initRebuildIndexConnectionProps(indexRebuildTaskRegionEnvironment.getConfiguration());
+ }
+
+ @Parameters(name = "MutableIndexFailureIT_transactional={0},localIndex={1},isNamespaceMapped={2},disableIndexOnWriteFailure={3},failRebuildTask={4},throwIndexWriteFailure={5}") // name is used by failsafe as file name in reports
+ public static List<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ // note - can't disableIndexOnWriteFailure without throwIndexWriteFailure, PHOENIX-4130
+ { false, false, true, true, false, null},
+ { false, false, true, true, false, true},
+ { true, false, true, true, false, null},
+ { false, true, true, true, false, null},
+ { true, true, true, null, false, null},
+ { false, false, true, true, true, null},
+ { false, false, true, false, true, false},
+ }
+ );
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/a162c74c/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
index 5e2e4df..ae2fa66 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
@@ -3871,7 +3871,37 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements Coprocesso
return;
}
}
+ if (newState == PIndexState.PENDING_DISABLE && currentState != PIndexState.PENDING_DISABLE) {
+ // reset count for first PENDING_DISABLE
+ newKVs.add(KeyValueUtil.newKeyValue(key, TABLE_FAMILY_BYTES,
+ PhoenixDatabaseMetaData.PENDING_DISABLE_COUNT_BYTES, timeStamp, Bytes.toBytes(0L)));
+ }
+ if (currentState == PIndexState.PENDING_DISABLE) {
+ if (newState == PIndexState.ACTIVE) {
+ //before making index ACTIVE check if all clients succeed otherwise keep it PENDING_DISABLE
+ byte[] count = region
+ .get(new Get(key).addColumn(TABLE_FAMILY_BYTES,
+ PhoenixDatabaseMetaData.PENDING_DISABLE_COUNT_BYTES))
+ .getValue(TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.PENDING_DISABLE_COUNT_BYTES);
+ if (count != null && Bytes.toLong(count) != 0) {
+ newState = PIndexState.PENDING_DISABLE;
+ newKVs.remove(disableTimeStampKVIndex);
+ newKVs.set(indexStateKVIndex, KeyValueUtil.newKeyValue(key, TABLE_FAMILY_BYTES,
+ INDEX_STATE_BYTES, timeStamp, Bytes.toBytes(newState.getSerializedValue())));
+ }
+ } else if (newState == PIndexState.DISABLE) {
+ //reset the counter for pending disable when transitioning from PENDING_DISABLE to DISABLE
+ newKVs.add(KeyValueUtil.newKeyValue(key, TABLE_FAMILY_BYTES,
+ PhoenixDatabaseMetaData.PENDING_DISABLE_COUNT_BYTES, timeStamp, Bytes.toBytes(0L)));
+ }
+ }
+
+ if(newState == PIndexState.ACTIVE||newState == PIndexState.PENDING_ACTIVE||newState == PIndexState.DISABLE){
+ newKVs.add(KeyValueUtil.newKeyValue(key, TABLE_FAMILY_BYTES,
+ PhoenixDatabaseMetaData.PENDING_DISABLE_COUNT_BYTES, timeStamp, Bytes.toBytes(0L)));
+ }
+
if (currentState == PIndexState.BUILDING && newState != PIndexState.ACTIVE) {
timeStamp = currentStateKV.getTimestamp();
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/a162c74c/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java
index 55192e7..e7f5ac2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java
@@ -17,6 +17,8 @@
*/
package org.apache.phoenix.index;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES;
+
import java.io.IOException;
import java.security.PrivilegedExceptionAction;
import java.sql.SQLException;
@@ -40,6 +42,7 @@ import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.Stoppable;
import org.apache.hadoop.hbase.client.ConnectionUtils;
import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
@@ -441,6 +444,7 @@ public class PhoenixIndexFailurePolicy extends DelegateIndexFailurePolicy {
public static void doBatchWithRetries(MutateCommand mutateCommand,
IndexWriteException iwe, PhoenixConnection connection, ReadOnlyProps config)
throws IOException {
+ incrementPendingDisableCounter(iwe, connection);
int maxTries = config.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
long pause = config.getLong(HConstants.HBASE_CLIENT_PAUSE,
@@ -475,6 +479,57 @@ public class PhoenixIndexFailurePolicy extends DelegateIndexFailurePolicy {
throw new DoNotRetryIOException(iwe); // send failure back to client
}
+ private static void incrementPendingDisableCounter(IndexWriteException indexWriteException,PhoenixConnection conn) {
+ try {
+ Set<String> indexesToUpdate = new HashSet<>();
+ if (indexWriteException instanceof MultiIndexWriteFailureException) {
+ MultiIndexWriteFailureException indexException =
+ (MultiIndexWriteFailureException) indexWriteException;
+ List<HTableInterfaceReference> failedIndexes = indexException.getFailedTables();
+ if (indexException.isDisableIndexOnFailure() && failedIndexes != null) {
+ for (HTableInterfaceReference failedIndex : failedIndexes) {
+ String failedIndexTable = failedIndex.getTableName();
+ if (!indexesToUpdate.contains(failedIndexTable)) {
+ incrementCounterForIndex(conn,failedIndexTable);
+ indexesToUpdate.add(failedIndexTable);
+ }
+ }
+ }
+ } else if (indexWriteException instanceof SingleIndexWriteFailureException) {
+ SingleIndexWriteFailureException indexException =
+ (SingleIndexWriteFailureException) indexWriteException;
+ String failedIndex = indexException.getTableName();
+ if (indexException.isDisableIndexOnFailure() && failedIndex != null) {
+ incrementCounterForIndex(conn,failedIndex);
+ }
+ }
+ } catch (Exception handleE) {
+ LOG.warn("Error while trying to handle index write exception", indexWriteException);
+ }
+ }
+
+ private static void incrementCounterForIndex(PhoenixConnection conn, String failedIndexTable) throws IOException {
+ incrementCounterForIndex(conn, failedIndexTable, 1);
+ }
+
+ private static void decrementCounterForIndex(PhoenixConnection conn, String failedIndexTable) throws IOException {
+ incrementCounterForIndex(conn, failedIndexTable, -1);
+ }
+
+ private static void incrementCounterForIndex(PhoenixConnection conn, String failedIndexTable,long amount) throws IOException {
+ byte[] indexTableKey = SchemaUtil.getTableKeyFromFullName(failedIndexTable);
+ Increment incr = new Increment(indexTableKey);
+ incr.addColumn(TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.PENDING_DISABLE_COUNT_BYTES, amount);
+ try {
+ conn.getQueryServices()
+ .getTable(SchemaUtil.getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME,
+ conn.getQueryServices().getProps()).getName())
+ .increment(incr);
+ } catch (SQLException e) {
+ throw new IOException(e);
+ }
+ }
+
private static boolean canRetryMore(int numRetry, int maxRetries, long canRetryUntil) {
// If there is a single try we must not take into account the time.
return numRetry < maxRetries
@@ -497,13 +552,25 @@ public class PhoenixIndexFailurePolicy extends DelegateIndexFailurePolicy {
}
private static void updateIndex(String indexFullName, PhoenixConnection conn,
- PIndexState indexState) throws SQLException {
+ PIndexState indexState) throws SQLException, IOException {
+ //Decrement the counter because we will be here when client give retry after getting failed or succeed
+ decrementCounterForIndex(conn,indexFullName);
+ Long indexDisableTimestamp = null;
if (PIndexState.DISABLE.equals(indexState)) {
LOG.info("Disabling index after hitting max number of index write retries: "
+ indexFullName);
+ IndexUtil.updateIndexState(conn, indexFullName, indexState, indexDisableTimestamp);
} else if (PIndexState.ACTIVE.equals(indexState)) {
LOG.debug("Resetting index to active after subsequent success " + indexFullName);
+ //At server disabled timestamp will be reset only if there is no other client is in PENDING_DISABLE state
+ indexDisableTimestamp = 0L;
+ try {
+ IndexUtil.updateIndexState(conn, indexFullName, indexState, indexDisableTimestamp);
+ } catch (SQLException e) {
+ // It's possible that some other client had made the Index DISABLED already , so we can ignore unallowed
+ // transition(DISABLED->ACTIVE)
+ if (e.getErrorCode() != SQLExceptionCode.INVALID_INDEX_STATE_TRANSITION.getErrorCode()) { throw e; }
+ }
}
- IndexUtil.updateIndexState(conn, indexFullName, indexState, null);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/a162c74c/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 5bcd286..8dd4a88 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
@@ -49,7 +49,6 @@ import org.apache.phoenix.expression.function.SQLViewTypeFunction;
import org.apache.phoenix.expression.function.SqlTypeNameFunction;
import org.apache.phoenix.expression.function.TransactionProviderNameFunction;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
-import org.apache.phoenix.hbase.index.util.VersionUtil;
import org.apache.phoenix.iterate.DelegateResultIterator;
import org.apache.phoenix.iterate.MaterializedResultIterator;
import org.apache.phoenix.iterate.ResultIterator;
@@ -214,6 +213,7 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
public static final String TABLE_FAMILY = QueryConstants.DEFAULT_COLUMN_FAMILY;
public static final byte[] TABLE_FAMILY_BYTES = QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES;
+ public static final byte[] PENDING_DISABLE_COUNT_BYTES = Bytes.toBytes("PENDING_DISABLE_COUNT");
public static final String TYPE_SEQUENCE = "SEQUENCE";
public static final String SYSTEM_FUNCTION_TABLE = "FUNCTION";