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/02 08:11:38 UTC
[1/7] phoenix git commit: PHOENIX-4759 During restart RS that hosts
SYSTEM.CATALOG table may get stuck [Forced Update!]
Repository: phoenix
Updated Branches:
refs/heads/4.x-cdh5.12 fc75ea703 -> 6b399f890 (forced update)
PHOENIX-4759 During restart RS that hosts SYSTEM.CATALOG table may get stuck
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/5cc7575e
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/5cc7575e
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/5cc7575e
Branch: refs/heads/4.x-cdh5.12
Commit: 5cc7575ef4a996930f1adaef3a50fd447c87ae7b
Parents: 58082f8
Author: Sergey Soldatov <ss...@apache.org>
Authored: Thu May 31 12:07:29 2018 -0700
Committer: ss77892 <ss...@apache.org>
Committed: Thu May 31 12:59:24 2018 -0700
----------------------------------------------------------------------
.../phoenix/coprocessor/MetaDataEndpointImpl.java | 6 +++---
.../phoenix/coprocessor/MetaDataProtocol.java | 16 +++++++++++++++-
.../apache/phoenix/exception/SQLExceptionCode.java | 6 +++---
.../index/write/ParallelWriterIndexCommitter.java | 4 ++--
.../TrackingParallelWriterIndexCommitter.java | 4 ++--
.../phoenix/index/PhoenixTransactionalIndexer.java | 4 ++--
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 17 +----------------
.../phoenix/query/ConnectionQueryServicesImpl.java | 4 ++--
.../org/apache/phoenix/schema/MetaDataClient.java | 4 ++--
.../java/org/apache/phoenix/util/ScanUtil.java | 3 +--
10 files changed, 33 insertions(+), 35 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cc7575e/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 ff62c92..5e2e4df 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
@@ -946,12 +946,12 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements Coprocesso
.getValueArray()[indexStateKv.getValueOffset()]);
// If client is not yet up to 4.12, then translate PENDING_ACTIVE to ACTIVE (as would have been
// the value in those versions) since the client won't have this index state in its enum.
- if (indexState == PIndexState.PENDING_ACTIVE && clientVersion < PhoenixDatabaseMetaData.MIN_PENDING_ACTIVE_INDEX) {
+ if (indexState == PIndexState.PENDING_ACTIVE && clientVersion < MetaDataProtocol.MIN_PENDING_ACTIVE_INDEX) {
indexState = PIndexState.ACTIVE;
}
// If client is not yet up to 4.14, then translate PENDING_DISABLE to DISABLE
// since the client won't have this index state in its enum.
- if (indexState == PIndexState.PENDING_DISABLE && clientVersion < PhoenixDatabaseMetaData.MIN_PENDING_DISABLE_INDEX) {
+ if (indexState == PIndexState.PENDING_DISABLE && clientVersion < MetaDataProtocol.MIN_PENDING_DISABLE_INDEX) {
// note: for older clients, we have to rely on the rebuilder to transition PENDING_DISABLE -> DISABLE
indexState = PIndexState.DISABLE;
}
@@ -3687,7 +3687,7 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements Coprocesso
GetVersionResponse.Builder builder = GetVersionResponse.newBuilder();
Configuration config = env.getConfiguration();
if (isTablesMappingEnabled
- && PhoenixDatabaseMetaData.MIN_NAMESPACE_MAPPED_PHOENIX_VERSION > request.getClientVersion()) {
+ && MetaDataProtocol.MIN_NAMESPACE_MAPPED_PHOENIX_VERSION > request.getClientVersion()) {
logger.error("Old client is not compatible when" + " system tables are upgraded to map to namespace");
ProtobufUtil.setControllerException(controller,
ServerUtil.createIOException(
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cc7575e/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java
index 62b701d..883f96d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java
@@ -93,7 +93,21 @@ public abstract class MetaDataProtocol extends MetaDataService {
public static final long MIN_SYSTEM_TABLE_TIMESTAMP_4_14_0 = MIN_TABLE_TIMESTAMP + 28;
// MIN_SYSTEM_TABLE_TIMESTAMP needs to be set to the max of all the MIN_SYSTEM_TABLE_TIMESTAMP_* constants
public static final long MIN_SYSTEM_TABLE_TIMESTAMP = MIN_SYSTEM_TABLE_TIMESTAMP_4_14_0;
-
+ // Version below which we should disallow usage of mutable secondary indexing.
+ public static final int MUTABLE_SI_VERSION_THRESHOLD = VersionUtil.encodeVersion("0", "94", "10");
+ public static final int MAX_LOCAL_SI_VERSION_DISALLOW = VersionUtil.encodeVersion("0", "98", "8");
+ public static final int MIN_LOCAL_SI_VERSION_DISALLOW = VersionUtil.encodeVersion("0", "98", "6");
+ public static final int MIN_RENEW_LEASE_VERSION = VersionUtil.encodeVersion("1", "1", "3");
+ public static final int MIN_NAMESPACE_MAPPED_PHOENIX_VERSION = VersionUtil.encodeVersion("4", "8", "0");
+ public static final int MIN_PENDING_ACTIVE_INDEX = VersionUtil.encodeVersion("4", "12", "0");
+ public static final int MIN_CLIENT_RETRY_INDEX_WRITES = VersionUtil.encodeVersion("4", "14", "0");
+ public static final int MIN_TX_CLIENT_SIDE_MAINTENANCE = VersionUtil.encodeVersion("4", "14", "0");
+ public static final int MIN_PENDING_DISABLE_INDEX = VersionUtil.encodeVersion("4", "14", "0");
+ // Version below which we should turn off essential column family.
+ public static final int ESSENTIAL_FAMILY_VERSION_THRESHOLD = VersionUtil.encodeVersion("0", "94", "7");
+ /** Version below which we fall back on the generic KeyValueBuilder */
+ public static final int CLIENT_KEY_VALUE_BUILDER_THRESHOLD = VersionUtil.encodeVersion("0", "94", "14");
+
// ALWAYS update this map whenever rolling out a new release (major, minor or patch release).
// Key is the SYSTEM.CATALOG timestamp for the version and value is the version string.
private static final NavigableMap<Long, String> TIMESTAMP_VERSION_MAP = new TreeMap<>();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cc7575e/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
index e9e209b..d6a70f2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
@@ -21,8 +21,8 @@ import java.sql.SQLException;
import java.sql.SQLTimeoutException;
import java.util.Map;
+import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.hbase.index.util.IndexManagementUtil;
-import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.schema.AmbiguousColumnException;
@@ -243,7 +243,7 @@ public enum SQLExceptionCode {
SALT_ONLY_ON_CREATE_TABLE(1024, "42Y82", "Salt bucket number may only be specified when creating a table."),
SET_UNSUPPORTED_PROP_ON_ALTER_TABLE(1025, "42Y83", "Unsupported property set in ALTER TABLE command."),
CANNOT_ADD_NOT_NULLABLE_COLUMN(1038, "42Y84", "Only nullable columns may be added for a pre-existing table."),
- NO_MUTABLE_INDEXES(1026, "42Y85", "Mutable secondary indexes are only supported for HBase version " + MetaDataUtil.decodeHBaseVersionAsString(PhoenixDatabaseMetaData.MUTABLE_SI_VERSION_THRESHOLD) + " and above."),
+ NO_MUTABLE_INDEXES(1026, "42Y85", "Mutable secondary indexes are only supported for HBase version " + MetaDataUtil.decodeHBaseVersionAsString(MetaDataProtocol.MUTABLE_SI_VERSION_THRESHOLD) + " and above."),
INVALID_INDEX_STATE_TRANSITION(1028, "42Y87", "Invalid index state transition."),
INVALID_MUTABLE_INDEX_CONFIG(1029, "42Y88", "Mutable secondary indexes must have the "
+ IndexManagementUtil.WAL_EDIT_CODEC_CLASS_KEY + " property set to "
@@ -266,7 +266,7 @@ public enum SQLExceptionCode {
CANNOT_SET_TABLE_PROPERTY_ADD_COLUMN(1053, "43A10", "Table level property cannot be set when adding a column."),
NO_LOCAL_INDEXES(1054, "43A11", "Local secondary indexes are not supported for HBase versions " +
- MetaDataUtil.decodeHBaseVersionAsString(PhoenixDatabaseMetaData.MIN_LOCAL_SI_VERSION_DISALLOW) + " through " + MetaDataUtil.decodeHBaseVersionAsString(PhoenixDatabaseMetaData.MAX_LOCAL_SI_VERSION_DISALLOW) + " inclusive."),
+ MetaDataUtil.decodeHBaseVersionAsString(MetaDataProtocol.MIN_LOCAL_SI_VERSION_DISALLOW) + " through " + MetaDataUtil.decodeHBaseVersionAsString(MetaDataProtocol.MAX_LOCAL_SI_VERSION_DISALLOW) + " inclusive."),
UNALLOWED_LOCAL_INDEXES(1055, "43A12", "Local secondary indexes are configured to not be allowed."),
DESC_VARBINARY_NOT_SUPPORTED(1056, "43A13", "Descending VARBINARY columns not supported."),
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cc7575e/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/ParallelWriterIndexCommitter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/ParallelWriterIndexCommitter.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/ParallelWriterIndexCommitter.java
index e06efcc..290e1be 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/ParallelWriterIndexCommitter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/ParallelWriterIndexCommitter.java
@@ -25,6 +25,7 @@ import org.apache.hadoop.hbase.Stoppable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.hbase.index.exception.SingleIndexWriteFailureException;
import org.apache.phoenix.hbase.index.parallel.EarlyExitFailure;
import org.apache.phoenix.hbase.index.parallel.QuickFailingTaskRunner;
@@ -36,7 +37,6 @@ import org.apache.phoenix.hbase.index.table.HTableFactory;
import org.apache.phoenix.hbase.index.table.HTableInterfaceReference;
import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
import org.apache.phoenix.index.PhoenixIndexFailurePolicy;
-import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.util.IndexUtil;
import com.google.common.collect.Multimap;
@@ -166,7 +166,7 @@ public class ParallelWriterIndexCommitter implements IndexCommitter {
}
}
// if the client can retry index writes, then we don't need to retry here
- HTableFactory factory = clientVersion < PhoenixDatabaseMetaData.MIN_CLIENT_RETRY_INDEX_WRITES ? retryingFactory : noRetriesfactory;
+ HTableFactory factory = clientVersion < MetaDataProtocol.MIN_CLIENT_RETRY_INDEX_WRITES ? retryingFactory : noRetriesfactory;
table = factory.getTable(tableReference.get());
throwFailureIfDone();
table.batch(mutations);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cc7575e/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java
index 4ba1155..4fa2d0b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.hbase.Stoppable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.hbase.index.CapturingAbortable;
import org.apache.phoenix.hbase.index.exception.MultiIndexWriteFailureException;
import org.apache.phoenix.hbase.index.exception.SingleIndexWriteFailureException;
@@ -41,7 +42,6 @@ import org.apache.phoenix.hbase.index.table.HTableFactory;
import org.apache.phoenix.hbase.index.table.HTableInterfaceReference;
import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
import org.apache.phoenix.index.PhoenixIndexFailurePolicy;
-import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.util.IndexUtil;
import com.google.common.collect.Multimap;
@@ -179,7 +179,7 @@ public class TrackingParallelWriterIndexCommitter implements IndexCommitter {
LOG.trace("Writing index update:" + mutations + " to table: " + tableReference);
}
// if the client can retry index writes, then we don't need to retry here
- HTableFactory factory = clientVersion < PhoenixDatabaseMetaData.MIN_CLIENT_RETRY_INDEX_WRITES ? retryingFactory : noRetriesFactory;
+ HTableFactory factory = clientVersion < MetaDataProtocol.MIN_CLIENT_RETRY_INDEX_WRITES ? retryingFactory : noRetriesFactory;
table = factory.getTable(tableReference.get());
throwFailureIfDone();
table.batch(mutations);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cc7575e/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixTransactionalIndexer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixTransactionalIndexer.java b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixTransactionalIndexer.java
index 0893bf6..189102d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixTransactionalIndexer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixTransactionalIndexer.java
@@ -45,11 +45,11 @@ import org.apache.htrace.Span;
import org.apache.htrace.Trace;
import org.apache.htrace.TraceScope;
import org.apache.phoenix.coprocessor.DelegateRegionCoprocessorEnvironment;
+import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.execute.PhoenixTxIndexMutationGenerator;
import org.apache.phoenix.hbase.index.write.IndexWriter;
import org.apache.phoenix.hbase.index.write.LeaveIndexActiveFailurePolicy;
import org.apache.phoenix.hbase.index.write.ParallelWriterIndexCommitter;
-import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.trace.TracingUtils;
import org.apache.phoenix.trace.util.NullSpan;
import org.apache.phoenix.transaction.PhoenixTransactionContext;
@@ -141,7 +141,7 @@ public class PhoenixTransactionalIndexer extends BaseRegionObserver {
}
PhoenixIndexMetaData indexMetaData = new PhoenixIndexMetaDataBuilder(c.getEnvironment()).getIndexMetaData(miniBatchOp);
- if ( indexMetaData.getClientVersion() >= PhoenixDatabaseMetaData.MIN_TX_CLIENT_SIDE_MAINTENANCE
+ if ( indexMetaData.getClientVersion() >= MetaDataProtocol.MIN_TX_CLIENT_SIDE_MAINTENANCE
&& !indexMetaData.hasLocalIndexes()) { // Still generate index updates server side for local indexes
super.preBatchMutate(c, miniBatchOp);
return;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cc7575e/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 2e12ca2..5bcd286 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
@@ -325,22 +325,7 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
private final PhoenixConnection connection;
private final ResultSet emptyResultSet;
- public static final int MAX_LOCAL_SI_VERSION_DISALLOW = VersionUtil.encodeVersion("0", "98", "8");
- public static final int MIN_LOCAL_SI_VERSION_DISALLOW = VersionUtil.encodeVersion("0", "98", "6");
- public static final int MIN_RENEW_LEASE_VERSION = VersionUtil.encodeVersion("1", "1", "3");
- public static final int MIN_NAMESPACE_MAPPED_PHOENIX_VERSION = VersionUtil.encodeVersion("4", "8", "0");
- public static final int MIN_PENDING_ACTIVE_INDEX = VersionUtil.encodeVersion("4", "12", "0");
- public static final int MIN_PENDING_DISABLE_INDEX = VersionUtil.encodeVersion("4", "14", "0");
- public static final int MIN_CLIENT_RETRY_INDEX_WRITES = VersionUtil.encodeVersion("4", "14", "0");
- public static final int MIN_TX_CLIENT_SIDE_MAINTENANCE = VersionUtil.encodeVersion("4", "14", "0");
-
- // Version below which we should turn off essential column family.
- public static final int ESSENTIAL_FAMILY_VERSION_THRESHOLD = VersionUtil.encodeVersion("0", "94", "7");
- // Version below which we should disallow usage of mutable secondary indexing.
- public static final int MUTABLE_SI_VERSION_THRESHOLD = VersionUtil.encodeVersion("0", "94", "10");
- /** Version below which we fall back on the generic KeyValueBuilder */
- public static final int CLIENT_KEY_VALUE_BUILDER_THRESHOLD = VersionUtil.encodeVersion("0", "94", "14");
-
+
public static final String IMMUTABLE_STORAGE_SCHEME = "IMMUTABLE_STORAGE_SCHEME";
public static final byte[] STORAGE_SCHEME_BYTES = Bytes.toBytes(IMMUTABLE_STORAGE_SCHEME);
public static final String ENCODING_SCHEME = "ENCODING_SCHEME";
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cc7575e/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 39bdeb2..4290d23 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
@@ -336,14 +336,14 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
@Override
public boolean isSupported(ConnectionQueryServices services) {
int hbaseVersion = services.getLowestClusterHBaseVersion();
- return hbaseVersion < PhoenixDatabaseMetaData.MIN_LOCAL_SI_VERSION_DISALLOW || hbaseVersion > PhoenixDatabaseMetaData.MAX_LOCAL_SI_VERSION_DISALLOW;
+ return hbaseVersion < MetaDataProtocol.MIN_LOCAL_SI_VERSION_DISALLOW || hbaseVersion > MetaDataProtocol.MAX_LOCAL_SI_VERSION_DISALLOW;
}
},
Feature.RENEW_LEASE, new FeatureSupported() {
@Override
public boolean isSupported(ConnectionQueryServices services) {
int hbaseVersion = services.getLowestClusterHBaseVersion();
- return hbaseVersion >= PhoenixDatabaseMetaData.MIN_RENEW_LEASE_VERSION;
+ return hbaseVersion >= MetaDataProtocol.MIN_RENEW_LEASE_VERSION;
}
});
private QueryLoggerDisruptor queryDisruptor;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cc7575e/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 4bdc5ff..9979ae7 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
@@ -1517,7 +1517,7 @@ public class MetaDataClient {
}
}
if (!dataTable.isImmutableRows()) {
- if (hbaseVersion < PhoenixDatabaseMetaData.MUTABLE_SI_VERSION_THRESHOLD) {
+ if (hbaseVersion < MetaDataProtocol.MUTABLE_SI_VERSION_THRESHOLD) {
throw new SQLExceptionInfo.Builder(SQLExceptionCode.NO_MUTABLE_INDEXES).setTableName(indexTableName.getTableName()).build().buildException();
}
if (!connection.getQueryServices().hasIndexWALCodec() && !dataTable.isTransactional()) {
@@ -3422,7 +3422,7 @@ public class MetaDataClient {
// have existing indexes.
if (Boolean.FALSE.equals(metaPropertiesEvaluated.getIsImmutableRows()) && !table.getIndexes().isEmpty()) {
int hbaseVersion = connection.getQueryServices().getLowestClusterHBaseVersion();
- if (hbaseVersion < PhoenixDatabaseMetaData.MUTABLE_SI_VERSION_THRESHOLD) {
+ if (hbaseVersion < MetaDataProtocol.MUTABLE_SI_VERSION_THRESHOLD) {
throw new SQLExceptionInfo.Builder(SQLExceptionCode.NO_MUTABLE_INDEXES)
.setSchemaName(schemaName).setTableName(tableName).build().buildException();
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cc7575e/phoenix-core/src/main/java/org/apache/phoenix/util/ScanUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ScanUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ScanUtil.java
index 996e1dc..62ecebd 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/ScanUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ScanUtil.java
@@ -60,7 +60,6 @@ import org.apache.phoenix.filter.MultiEncodedCQKeyValueComparisonFilter;
import org.apache.phoenix.filter.SkipScanFilter;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.hbase.index.util.VersionUtil;
-import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.query.KeyRange;
import org.apache.phoenix.query.KeyRange.Bound;
import org.apache.phoenix.query.QueryConstants;
@@ -895,7 +894,7 @@ public class ScanUtil {
* the server side. To make sure HBase doesn't cancel the leases and close the open
* scanners, we need to periodically renew leases. To look at the earliest HBase version
* that supports renewing leases, see
- * {@link PhoenixDatabaseMetaData#MIN_RENEW_LEASE_VERSION}
+ * {@link MetaDataProtocol#MIN_RENEW_LEASE_VERSION}
*/
public static boolean isPacingScannersPossible(StatementContext context) {
return context.getConnection().getQueryServices().isRenewingLeasesEnabled();
[2/7] phoenix git commit: PHOENIX-4762 Performance regression with
transactional immutable indexes
Posted by pb...@apache.org.
http://git-wip-us.apache.org/repos/asf/phoenix/blob/edb6dcf5/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 f3ad42f..751945a 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
@@ -108,6 +108,11 @@ public interface PhoenixTransactionContext {
public HTableInterface getTransactionalTable(HTableInterface htable, boolean isImmutable) {
return null;
}
+
+ @Override
+ public HTableInterface getTransactionalTableWriter(HTableInterface htable, PTable table) {
+ return null;
+ }
};
/**
*
@@ -225,4 +230,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);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/edb6dcf5/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 9c6af63..94d6765 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
@@ -469,5 +469,21 @@ public class TephraTransactionContext implements PhoenixTransactionContext {
this.addTransactionAware(transactionAwareHTable);
return transactionAwareHTable;
}
+
+ @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);
+ // 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.startTx(getTransaction());
+ } else {
+ // 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);
+ }
+ return transactionAwareHTable;
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/edb6dcf5/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 7e280f4..78a68d2 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
@@ -42,7 +42,6 @@ import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
@@ -116,7 +115,6 @@ import org.apache.phoenix.schema.types.PDecimal;
import org.apache.phoenix.schema.types.PLong;
import org.apache.phoenix.schema.types.PVarbinary;
import org.apache.phoenix.schema.types.PVarchar;
-import org.apache.phoenix.transaction.PhoenixTransactionContext;
import com.google.common.collect.Lists;
@@ -269,32 +267,6 @@ public class IndexUtil {
.getLength()) == 0);
}
- public static List<Delete> generateDeleteIndexData(final PTable table, PTable index,
- List<Delete> dataMutations, ImmutableBytesWritable ptr, final KeyValueBuilder kvBuilder, PhoenixConnection connection)
- throws SQLException {
- try {
- IndexMaintainer maintainer = index.getIndexMaintainer(table, connection);
- List<Delete> indexMutations = Lists.newArrayListWithExpectedSize(dataMutations.size());
- for (final Mutation dataMutation : dataMutations) {
- long ts = MetaDataUtil.getClientTimeStamp(dataMutation);
- ptr.set(dataMutation.getRow());
- byte[] regionStartKey = null;
- byte[] regionEndkey = null;
- if(maintainer.isLocalIndex()) {
- HRegionLocation tableRegionLocation = connection.getQueryServices().getTableRegionLocation(table.getPhysicalName().getBytes(), dataMutation.getRow());
- regionStartKey = tableRegionLocation.getRegionInfo().getStartKey();
- regionEndkey = tableRegionLocation.getRegionInfo().getEndKey();
- }
- Delete delete = maintainer.buildDeleteMutation(kvBuilder, null, ptr, Collections.<KeyValue>emptyList(), ts, regionStartKey, regionEndkey);
- delete.setAttribute(PhoenixTransactionContext.TX_ROLLBACK_ATTRIBUTE_KEY, dataMutation.getAttribute(PhoenixTransactionContext.TX_ROLLBACK_ATTRIBUTE_KEY));
- indexMutations.add(delete);
- }
- return indexMutations;
- } catch (IOException e) {
- throw new SQLException(e);
- }
- }
-
public static List<Mutation> generateIndexData(final PTable table, PTable index,
final MultiRowMutationState multiRowMutationState, List<Mutation> dataMutations, final KeyValueBuilder kvBuilder, PhoenixConnection connection)
throws SQLException {
[6/7] 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/9f23cfec
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/9f23cfec
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/9f23cfec
Branch: refs/heads/4.x-cdh5.12
Commit: 9f23cfec1301017286ba9c563536d4f051a1441a
Parents: f6caad5
Author: Pedro Boado <pb...@apache.org>
Authored: Sat Mar 10 17:54:04 2018 +0000
Committer: Pedro Boado <pb...@apache.org>
Committed: Sat Jun 2 09:11:09 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/9f23cfec/phoenix-assembly/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-assembly/pom.xml b/phoenix-assembly/pom.xml
index 7d98c25..14225ee 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-assembly</artifactId>
<name>Phoenix Assembly</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/phoenix-client/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-client/pom.xml b/phoenix-client/pom.xml
index fe5836b..ffad100 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-client</artifactId>
<name>Phoenix Client</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/phoenix-core/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-core/pom.xml b/phoenix-core/pom.xml
index 8b09c67..7d5b784 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-core</artifactId>
<name>Phoenix Core</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/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/9f23cfec/phoenix-flume/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-flume/pom.xml b/phoenix-flume/pom.xml
index e400589..8a78010 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-flume</artifactId>
<name>Phoenix - Flume</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/phoenix-hive/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-hive/pom.xml b/phoenix-hive/pom.xml
index 4687887..804ba5f 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-hive</artifactId>
<name>Phoenix - Hive</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/phoenix-kafka/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-kafka/pom.xml b/phoenix-kafka/pom.xml
index 158026e..08a8bfd 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-kafka</artifactId>
<name>Phoenix - Kafka</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/phoenix-load-balancer/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-load-balancer/pom.xml b/phoenix-load-balancer/pom.xml
index b9b4672..cdb4c1b 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-load-balancer</artifactId>
<name>Phoenix Load Balancer</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/phoenix-parcel/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-parcel/pom.xml b/phoenix-parcel/pom.xml
index 5bfa9b5..4661e0e 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-parcel</artifactId>
<name>Phoenix Parcels for CDH</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/phoenix-pherf/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-pherf/pom.xml b/phoenix-pherf/pom.xml
index 04df091..0634a01 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-pherf</artifactId>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/phoenix-pig/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-pig/pom.xml b/phoenix-pig/pom.xml
index a88cece..cd4f6bc 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-pig</artifactId>
<name>Phoenix - Pig</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/phoenix-queryserver-client/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-queryserver-client/pom.xml b/phoenix-queryserver-client/pom.xml
index 424ce82..86c56b9 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-queryserver-client</artifactId>
<name>Phoenix Query Server Client</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/phoenix-queryserver/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-queryserver/pom.xml b/phoenix-queryserver/pom.xml
index 66e0735..e6c32cb 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-queryserver</artifactId>
<name>Phoenix Query Server</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/phoenix-server/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-server/pom.xml b/phoenix-server/pom.xml
index 98719fb..d971596 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-server</artifactId>
<name>Phoenix Server</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/phoenix-spark/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-spark/pom.xml b/phoenix-spark/pom.xml
index e6f5d95..b3b6e1c 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-spark</artifactId>
<name>Phoenix - Spark</name>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/phoenix-tracing-webapp/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-tracing-webapp/pom.xml b/phoenix-tracing-webapp/pom.xml
index 4f79b4e..5226f84 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-SNAPSHOT</version>
</parent>
<artifactId>phoenix-tracing-webapp</artifactId>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9f23cfec/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index bea1199..4d852c9 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-SNAPSHOT</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>
[5/7] phoenix git commit: PHOENIX-4768 Re-enable
testCompactUpdatesStats and testCompactUpdatesStatsWithMinStatsUpdateFreq of
StatsCollectorIT
Posted by pb...@apache.org.
PHOENIX-4768 Re-enable testCompactUpdatesStats and testCompactUpdatesStatsWithMinStatsUpdateFreq of StatsCollectorIT
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/f6caad56
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/f6caad56
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/f6caad56
Branch: refs/heads/4.x-cdh5.12
Commit: f6caad56dabd1576116d44625f1c4a150ae26368
Parents: 5b45e8f
Author: Ankit Singhal <an...@gmail.com>
Authored: Fri Jun 1 22:43:01 2018 +0100
Committer: Pedro Boado <pb...@apache.org>
Committed: Sat Jun 2 09:09:53 2018 +0100
----------------------------------------------------------------------
.../org/apache/phoenix/schema/stats/StatsCollectorIT.java | 7 ++-----
1 file changed, 2 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f6caad56/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/StatsCollectorIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/StatsCollectorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/StatsCollectorIT.java
index 09d28f8..5436311 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/StatsCollectorIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/StatsCollectorIT.java
@@ -68,7 +68,6 @@ import org.apache.phoenix.util.SchemaUtil;
import org.apache.phoenix.util.TestUtil;
import org.junit.Before;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -404,13 +403,11 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
}
@Test
- @Ignore //TODO remove this once https://issues.apache.org/jira/browse/TEPHRA-208 is fixed
public void testCompactUpdatesStats() throws Exception {
testCompactUpdatesStats(0, fullTableName);
}
@Test
- @Ignore //TODO remove this once https://issues.apache.org/jira/browse/TEPHRA-208 is fixed
public void testCompactUpdatesStatsWithMinStatsUpdateFreq() throws Exception {
testCompactUpdatesStats(QueryServicesOptions.DEFAULT_STATS_UPDATE_FREQ_MS, fullTableName);
}
@@ -463,7 +460,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
Scan scan = new Scan();
scan.setRaw(true);
PhoenixConnection phxConn = conn.unwrap(PhoenixConnection.class);
- try (HTableInterface htable = phxConn.getQueryServices().getTable(Bytes.toBytes(tableName))) {
+ try (HTableInterface htable = phxConn.getQueryServices().getTable(Bytes.toBytes(physicalTableName))) {
ResultScanner scanner = htable.getScanner(scan);
Result result;
while ((result = scanner.next())!=null) {
@@ -476,7 +473,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
scan = new Scan();
scan.setRaw(true);
phxConn = conn.unwrap(PhoenixConnection.class);
- try (HTableInterface htable = phxConn.getQueryServices().getTable(Bytes.toBytes(tableName))) {
+ try (HTableInterface htable = phxConn.getQueryServices().getTable(Bytes.toBytes(physicalTableName))) {
ResultScanner scanner = htable.getScanner(scan);
Result result;
while ((result = scanner.next())!=null) {
[7/7] phoenix git commit: Set version to v4.14.0-cdh5.12 for release
Posted by pb...@apache.org.
Set version to v4.14.0-cdh5.12 for release
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/6b399f89
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/6b399f89
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/6b399f89
Branch: refs/heads/4.x-cdh5.12
Commit: 6b399f890f191c0cb5533e0552e0f315f45919d9
Parents: 9f23cfe
Author: Pedro Boado <pb...@apache.org>
Authored: Sat May 26 08:44:25 2018 +0100
Committer: Pedro Boado <pb...@apache.org>
Committed: Sat Jun 2 09:11:09 2018 +0100
----------------------------------------------------------------------
phoenix-assembly/pom.xml | 2 +-
phoenix-client/pom.xml | 2 +-
phoenix-core/pom.xml | 2 +-
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 | 2 +-
16 files changed, 16 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/6b399f89/phoenix-assembly/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-assembly/pom.xml b/phoenix-assembly/pom.xml
index 14225ee..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.12.2-SNAPSHOT</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/6b399f89/phoenix-client/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-client/pom.xml b/phoenix-client/pom.xml
index ffad100..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.12.2-SNAPSHOT</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/6b399f89/phoenix-core/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-core/pom.xml b/phoenix-core/pom.xml
index 7d5b784..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.12.2-SNAPSHOT</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/6b399f89/phoenix-flume/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-flume/pom.xml b/phoenix-flume/pom.xml
index 8a78010..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.12.2-SNAPSHOT</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/6b399f89/phoenix-hive/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-hive/pom.xml b/phoenix-hive/pom.xml
index 804ba5f..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.12.2-SNAPSHOT</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/6b399f89/phoenix-kafka/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-kafka/pom.xml b/phoenix-kafka/pom.xml
index 08a8bfd..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.12.2-SNAPSHOT</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/6b399f89/phoenix-load-balancer/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-load-balancer/pom.xml b/phoenix-load-balancer/pom.xml
index cdb4c1b..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.12.2-SNAPSHOT</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/6b399f89/phoenix-parcel/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-parcel/pom.xml b/phoenix-parcel/pom.xml
index 4661e0e..de72394 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.12.2-SNAPSHOT</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/6b399f89/phoenix-pherf/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-pherf/pom.xml b/phoenix-pherf/pom.xml
index 0634a01..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.12.2-SNAPSHOT</version>
+ <version>4.14.0-cdh5.12.2</version>
</parent>
<artifactId>phoenix-pherf</artifactId>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/6b399f89/phoenix-pig/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-pig/pom.xml b/phoenix-pig/pom.xml
index cd4f6bc..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.12.2-SNAPSHOT</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/6b399f89/phoenix-queryserver-client/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-queryserver-client/pom.xml b/phoenix-queryserver-client/pom.xml
index 86c56b9..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.12.2-SNAPSHOT</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/6b399f89/phoenix-queryserver/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-queryserver/pom.xml b/phoenix-queryserver/pom.xml
index e6c32cb..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.12.2-SNAPSHOT</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/6b399f89/phoenix-server/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-server/pom.xml b/phoenix-server/pom.xml
index d971596..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.12.2-SNAPSHOT</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/6b399f89/phoenix-spark/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-spark/pom.xml b/phoenix-spark/pom.xml
index b3b6e1c..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.12.2-SNAPSHOT</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/6b399f89/phoenix-tracing-webapp/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-tracing-webapp/pom.xml b/phoenix-tracing-webapp/pom.xml
index 5226f84..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.12.2-SNAPSHOT</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/6b399f89/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 4d852c9..df71d03 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.12.2-SNAPSHOT</version>
+ <version>4.14.0-cdh5.12.2</version>
<packaging>pom</packaging>
<name>Apache Phoenix</name>
<description>A SQL layer over HBase</description>
[3/7] phoenix git commit: PHOENIX-4762 Performance regression with
transactional immutable indexes
Posted by pb...@apache.org.
PHOENIX-4762 Performance regression with transactional immutable indexes
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/edb6dcf5
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/edb6dcf5
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/edb6dcf5
Branch: refs/heads/4.x-cdh5.12
Commit: edb6dcf5f976d91971941166746191a1b49be211
Parents: 5cc7575
Author: James Taylor <jt...@salesforce.com>
Authored: Fri Jun 1 09:03:21 2018 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Fri Jun 1 10:30:40 2018 -0700
----------------------------------------------------------------------
.../apache/phoenix/execute/MutationState.java | 701 ++++++++++---------
.../PhoenixTxIndexMutationGenerator.java | 2 +-
.../transaction/OmidTransactionContext.java | 6 +
.../transaction/PhoenixTransactionContext.java | 6 +
.../transaction/TephraTransactionContext.java | 16 +
.../java/org/apache/phoenix/util/IndexUtil.java | 28 -
6 files changed, 385 insertions(+), 374 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/edb6dcf5/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 b9a6c9e..2e795b1 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,10 +38,9 @@ import javax.annotation.Nonnull;
import javax.annotation.concurrent.Immutable;
import org.apache.hadoop.hbase.HConstants;
-import org.apache.phoenix.util.KeyValueUtil;
import org.apache.hadoop.hbase.client.Delete;
-import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.HTableInterface;
+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;
@@ -83,7 +82,6 @@ import org.apache.phoenix.schema.PMetaData;
import org.apache.phoenix.schema.PName;
import org.apache.phoenix.schema.PRow;
import org.apache.phoenix.schema.PTable;
-import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.PTableRef;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.RowKeySchema;
@@ -98,6 +96,7 @@ 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;
@@ -116,9 +115,7 @@ import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
/**
- *
* Tracks the uncommitted state
- *
*/
public class MutationState implements SQLCloseable {
private static final Logger logger = LoggerFactory.getLogger(MutationState.class);
@@ -150,30 +147,34 @@ public class MutationState implements SQLCloseable {
this(maxSize, maxSizeBytes, connection, false, null);
}
- public MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection, PhoenixTransactionContext txContext) {
+ public MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection,
+ PhoenixTransactionContext txContext) {
this(maxSize, maxSizeBytes, connection, false, txContext);
}
public MutationState(MutationState mutationState) {
- this(mutationState.maxSize, mutationState.maxSizeBytes, mutationState.connection, true, mutationState.getPhoenixTransactionContext());
+ this(mutationState.maxSize, mutationState.maxSizeBytes, mutationState.connection, true, mutationState
+ .getPhoenixTransactionContext());
}
public MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection, long sizeOffset) {
this(maxSize, maxSizeBytes, connection, false, null, sizeOffset);
}
- private MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection, boolean subTask, PhoenixTransactionContext txContext) {
+ private MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection, boolean subTask,
+ PhoenixTransactionContext txContext) {
this(maxSize, maxSizeBytes, connection, subTask, txContext, 0);
}
- private MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection, boolean subTask, PhoenixTransactionContext txContext, long sizeOffset) {
- this(maxSize, maxSizeBytes, connection, Maps.<TableRef, MultiRowMutationState>newHashMapWithExpectedSize(5), subTask, txContext);
+ private MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection, boolean subTask,
+ PhoenixTransactionContext txContext, long sizeOffset) {
+ this(maxSize, maxSizeBytes, connection, Maps.<TableRef, MultiRowMutationState> newHashMapWithExpectedSize(5),
+ subTask, txContext);
this.sizeOffset = sizeOffset;
}
MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection,
- Map<TableRef, MultiRowMutationState> mutations,
- boolean subTask, PhoenixTransactionContext txContext) {
+ Map<TableRef, MultiRowMutationState> mutations, boolean subTask, PhoenixTransactionContext txContext) {
this.maxSize = maxSize;
this.maxSizeBytes = maxSizeBytes;
this.connection = connection;
@@ -193,7 +194,8 @@ public class MutationState implements SQLCloseable {
}
}
- public MutationState(TableRef table, MultiRowMutationState mutations, long sizeOffset, long maxSize, long maxSizeBytes, PhoenixConnection connection) throws SQLException {
+ public MutationState(TableRef table, MultiRowMutationState mutations, long sizeOffset, long maxSize,
+ long maxSizeBytes, PhoenixConnection connection) throws SQLException {
this(maxSize, maxSizeBytes, connection, false, null, sizeOffset);
if (!mutations.isEmpty()) {
this.mutations.put(table, mutations);
@@ -202,7 +204,7 @@ public class MutationState implements SQLCloseable {
this.estimatedSize = KeyValueUtil.getEstimatedRowMutationSize(this.mutations);
throwIfTooBig();
}
-
+
public long getEstimatedSize() {
return estimatedSize;
}
@@ -220,13 +222,12 @@ public class MutationState implements SQLCloseable {
}
/**
- * Commit a write fence when creating an index so that we can detect
- * when a data table transaction is started before the create index
- * but completes after it. In this case, we need to rerun the data
- * table transaction after the index creation so that the index rows
- * are generated. See TEPHRA-157
- * for more information.
- * @param dataTable the data table upon which an index is being added
+ * Commit a write fence when creating an index so that we can detect when a data table transaction is started before
+ * the create index but completes after it. In this case, we need to rerun the data table transaction after the
+ * index creation so that the index rows are generated. See TEPHRA-157 for more information.
+ *
+ * @param dataTable
+ * the data table upon which an index is being added
* @throws SQLException
*/
public void commitDDLFence(PTable dataTable) throws SQLException {
@@ -242,19 +243,17 @@ public class MutationState implements SQLCloseable {
}
}
}
-
+
public boolean checkpointIfNeccessary(MutationPlan plan) throws SQLException {
- if (! phoenixTransactionContext.isTransactionRunning() || plan.getTargetRef() == null || plan.getTargetRef().getTable() == null || !plan.getTargetRef().getTable().isTransactional()) {
- return false;
- }
+ if (!phoenixTransactionContext.isTransactionRunning() || plan.getTargetRef() == null
+ || plan.getTargetRef().getTable() == null || !plan.getTargetRef().getTable().isTransactional()) { return false; }
Set<TableRef> sources = plan.getSourceRefs();
- if (sources.isEmpty()) {
- return false;
- }
+ if (sources.isEmpty()) { return false; }
// For a DELETE statement, we're always querying the table being deleted from. This isn't
// a problem, but it potentially could be if there are other references to the same table
// nested in the DELETE statement (as a sub query or join, for example).
- TableRef ignoreForExcludeCurrent = plan.getOperation() == Operation.DELETE && sources.size() == 1 ? plan.getTargetRef() : null;
+ TableRef ignoreForExcludeCurrent = plan.getOperation() == Operation.DELETE && sources.size() == 1 ? plan
+ .getTargetRef() : null;
boolean excludeCurrent = false;
String targetPhysicalName = plan.getTargetRef().getTable().getPhysicalName().getString();
for (TableRef source : sources) {
@@ -279,7 +278,8 @@ public class MutationState implements SQLCloseable {
// external transaction context, it's possible that data was already written at the
// current transaction timestamp, so we always checkpoint in that case is we're
// reading and writing to the same table.
- if (source.getTable().isTransactional() && (isExternalTxContext || uncommittedPhysicalNames.contains(sourcePhysicalName))) {
+ if (source.getTable().isTransactional()
+ && (isExternalTxContext || uncommittedPhysicalNames.contains(sourcePhysicalName))) {
hasUncommittedData = true;
break;
}
@@ -331,30 +331,20 @@ public class MutationState implements SQLCloseable {
}
public boolean startTransaction(Provider provider) throws SQLException {
- if (provider == null) {
- return false;
- }
- if (!connection.getQueryServices().getProps().getBoolean(
- QueryServices.TRANSACTIONS_ENABLED,
- QueryServicesOptions.DEFAULT_TRANSACTIONS_ENABLED)) {
- throw new SQLExceptionInfo.Builder(
- SQLExceptionCode.CANNOT_START_TXN_IF_TXN_DISABLED)
- .build().buildException();
- }
- if (connection.getSCN() != null) {
- throw new SQLExceptionInfo.Builder(
- SQLExceptionCode.CANNOT_START_TRANSACTION_WITH_SCN_SET)
- .build().buildException();
- }
+ if (provider == null) { return false; }
+ if (!connection.getQueryServices().getProps()
+ .getBoolean(QueryServices.TRANSACTIONS_ENABLED, QueryServicesOptions.DEFAULT_TRANSACTIONS_ENABLED)) { throw new SQLExceptionInfo.Builder(
+ SQLExceptionCode.CANNOT_START_TXN_IF_TXN_DISABLED).build().buildException(); }
+ if (connection.getSCN() != null) { throw new SQLExceptionInfo.Builder(
+ SQLExceptionCode.CANNOT_START_TRANSACTION_WITH_SCN_SET).build().buildException(); }
if (phoenixTransactionContext == PhoenixTransactionContext.NULL_CONTEXT) {
phoenixTransactionContext = provider.getTransactionProvider().getTransactionContext(connection);
} else {
- if (provider != phoenixTransactionContext.getProvider()) {
- throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_MIX_TXN_PROVIDERS)
- .setMessage(phoenixTransactionContext.getProvider().name() + " and " + provider.name())
- .build().buildException();
- }
+ if (provider != phoenixTransactionContext.getProvider()) { throw new SQLExceptionInfo.Builder(
+ SQLExceptionCode.CANNOT_MIX_TXN_PROVIDERS)
+ .setMessage(phoenixTransactionContext.getProvider().name() + " and " + provider.name()).build()
+ .buildException(); }
}
if (!isTransactionStarted()) {
// Clear any transactional state in case transaction was ended outside
@@ -371,28 +361,28 @@ public class MutationState implements SQLCloseable {
}
public static MutationState emptyMutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection) {
- MutationState state = new MutationState(maxSize, maxSizeBytes, connection, Collections.<TableRef, MultiRowMutationState>emptyMap(), false, null);
+ MutationState state = new MutationState(maxSize, maxSizeBytes, connection,
+ Collections.<TableRef, MultiRowMutationState> emptyMap(), false, null);
state.sizeOffset = 0;
return state;
}
-
+
private void throwIfTooBig() throws SQLException {
if (numRows > maxSize) {
resetState();
- throw new SQLExceptionInfo.Builder(SQLExceptionCode.MAX_MUTATION_SIZE_EXCEEDED).build()
- .buildException();
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.MAX_MUTATION_SIZE_EXCEEDED).build().buildException();
}
if (estimatedSize > maxSizeBytes) {
resetState();
- throw new SQLExceptionInfo.Builder(SQLExceptionCode.MAX_MUTATION_SIZE_BYTES_EXCEEDED)
- .build().buildException();
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.MAX_MUTATION_SIZE_BYTES_EXCEEDED).build()
+ .buildException();
}
}
-
+
public long getUpdateCount() {
return sizeOffset + numRows;
}
-
+
private void joinMutationState(TableRef tableRef, MultiRowMutationState srcRows,
Map<TableRef, MultiRowMutationState> dstMutations) {
PTable table = tableRef.getTable();
@@ -401,21 +391,22 @@ public class MutationState implements SQLCloseable {
MultiRowMutationState existingRows = dstMutations.put(tableRef, srcRows);
if (existingRows != null) { // Rows for that table already exist
// Loop through new rows and replace existing with new
- for (Map.Entry<ImmutableBytesPtr,RowMutationState> rowEntry : srcRows.entrySet()) {
+ for (Map.Entry<ImmutableBytesPtr, RowMutationState> rowEntry : srcRows.entrySet()) {
// Replace existing row with new row
RowMutationState existingRowMutationState = existingRows.put(rowEntry.getKey(), rowEntry.getValue());
if (existingRowMutationState != null) {
- Map<PColumn,byte[]> existingValues = existingRowMutationState.getColumnValues();
+ Map<PColumn, byte[]> existingValues = existingRowMutationState.getColumnValues();
if (existingValues != PRow.DELETE_MARKER) {
- Map<PColumn,byte[]> newRow = rowEntry.getValue().getColumnValues();
- // if new row is PRow.DELETE_MARKER, it means delete, and we don't need to merge it with existing row.
+ Map<PColumn, byte[]> newRow = rowEntry.getValue().getColumnValues();
+ // if new row is PRow.DELETE_MARKER, it means delete, and we don't need to merge it with
+ // existing row.
if (newRow != PRow.DELETE_MARKER) {
// decrement estimated size by the size of the old row
- estimatedSize-=existingRowMutationState.calculateEstimatedSize();
+ estimatedSize -= existingRowMutationState.calculateEstimatedSize();
// Merge existing column values with new column values
existingRowMutationState.join(rowEntry.getValue());
// increment estimated size by the size of the new row
- estimatedSize+=existingRowMutationState.calculateEstimatedSize();
+ estimatedSize += existingRowMutationState.calculateEstimatedSize();
// Now that the existing row has been merged with the new row, replace it back
// again (since it was merged with the new one above).
existingRows.put(rowEntry.getKey(), existingRowMutationState);
@@ -440,12 +431,12 @@ public class MutationState implements SQLCloseable {
numRows += srcRows.size();
// if we added all the rows from newMutationState we can just increment the
// estimatedSize by newMutationState.estimatedSize
- estimatedSize += srcRows.estimatedSize;
+ estimatedSize += srcRows.estimatedSize;
}
}
}
-
- private void joinMutationState(Map<TableRef, MultiRowMutationState> srcMutations,
+
+ private void joinMutationState(Map<TableRef, MultiRowMutationState> srcMutations,
Map<TableRef, MultiRowMutationState> dstMutations) {
// Merge newMutation with this one, keeping state from newMutation for any overlaps
for (Map.Entry<TableRef, MultiRowMutationState> entry : srcMutations.entrySet()) {
@@ -455,11 +446,13 @@ public class MutationState implements SQLCloseable {
joinMutationState(tableRef, srcRows, dstMutations);
}
}
+
/**
* Combine a newer mutation with this one, where in the event of overlaps, the newer one will take precedence.
* Combine any metrics collected for the newer mutation.
*
- * @param newMutationState the newer mutation state
+ * @param newMutationState
+ * the newer mutation state
*/
public void join(MutationState newMutationState) throws SQLException {
if (this == newMutationState) { // Doesn't make sense
@@ -485,17 +478,16 @@ public class MutationState implements SQLCloseable {
throwIfTooBig();
}
-
private static ImmutableBytesPtr getNewRowKeyWithRowTimestamp(ImmutableBytesPtr ptr, long rowTimestamp, PTable table) {
RowKeySchema schema = table.getRowKeySchema();
int rowTimestampColPos = table.getRowTimestampColPos();
- Field rowTimestampField = schema.getField(rowTimestampColPos);
+ Field rowTimestampField = schema.getField(rowTimestampColPos);
byte[] rowTimestampBytes = PLong.INSTANCE.toBytes(rowTimestamp, rowTimestampField.getSortOrder());
int oldOffset = ptr.getOffset();
int oldLength = ptr.getLength();
// Move the pointer to the start byte of the row timestamp pk
schema.position(ptr, 0, rowTimestampColPos);
- byte[] b = ptr.get();
+ byte[] b = ptr.get();
int newOffset = ptr.getOffset();
int length = ptr.getLength();
for (int i = newOffset; i < newOffset + length; i++) {
@@ -506,24 +498,29 @@ public class MutationState implements SQLCloseable {
ptr.set(ptr.get(), oldOffset, oldLength);
return ptr;
}
-
- private Iterator<Pair<PName,List<Mutation>>> addRowMutations(final TableRef tableRef, final MultiRowMutationState values,
- final long mutationTimestamp, final long serverTimestamp, boolean includeAllIndexes, final boolean sendAll) {
+
+ 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 Iterator<PTable> indexIterator = // Only maintain tables with immutable rows through this client-side mechanism
- includeAllIndexes ?
- IndexMaintainer.maintainedIndexes(table.getIndexes().iterator()) :
- (table.isImmutableRows() || table.isTransactional()) ?
- IndexMaintainer.maintainedGlobalIndexes(table.getIndexes().iterator()) :
- Collections.<PTable>emptyIterator();
- final List<PTable> indexList = Lists.newArrayList(indexIterator);
+ final List<PTable> indexList = includeAllIndexes ? Lists.newArrayList(IndexMaintainer.maintainedIndexes(table
+ .getIndexes().iterator())) : getClientMaintainedIndexes(table);
final Iterator<PTable> indexes = indexList.iterator();
final List<Mutation> mutationList = Lists.newArrayListWithExpectedSize(values.size());
- final List<Mutation> mutationsPertainingToIndex = indexes.hasNext() ? Lists.<Mutation>newArrayListWithExpectedSize(values.size()) : null;
- generateMutations(tableRef, mutationTimestamp, serverTimestamp, values, mutationList, mutationsPertainingToIndex);
- return new Iterator<Pair<PName,List<Mutation>>>() {
+ final List<Mutation> mutationsPertainingToIndex = indexes.hasNext() ? Lists
+ .<Mutation> newArrayListWithExpectedSize(values.size()) : null;
+ generateMutations(tableRef, mutationTimestamp, serverTimestamp, values, mutationList,
+ mutationsPertainingToIndex);
+ return new Iterator<Pair<PName, List<Mutation>>>() {
boolean isFirst = true;
- Map<byte[],List<Mutation>> indexMutationsMap = null;
+ Map<byte[], List<Mutation>> indexMutationsMap = null;
@Override
public boolean hasNext() {
@@ -534,19 +531,22 @@ public class MutationState implements SQLCloseable {
public Pair<PName, List<Mutation>> next() {
if (isFirst) {
isFirst = false;
- return new Pair<PName,List<Mutation>>(table.getPhysicalName(), mutationList);
+ return new Pair<PName, List<Mutation>>(table.getPhysicalName(), mutationList);
}
PTable index = indexes.next();
-
+
List<Mutation> indexMutations = null;
try {
if (!mutationsPertainingToIndex.isEmpty()) {
if (table.isTransactional()) {
if (indexMutationsMap == null) {
- PhoenixTxIndexMutationGenerator generator = newTxIndexMutationGenerator(table, indexList, mutationsPertainingToIndex.get(0).getAttributesMap());
- try (HTableInterface htable = connection.getQueryServices().getTable(table.getPhysicalName().getBytes())) {
- Collection<Pair<Mutation, byte[]>> allMutations = generator.getIndexUpdates(htable, mutationsPertainingToIndex.iterator());
+ PhoenixTxIndexMutationGenerator generator = newTxIndexMutationGenerator(table,
+ indexList, mutationsPertainingToIndex.get(0).getAttributesMap());
+ try (HTableInterface htable = connection.getQueryServices().getTable(
+ table.getPhysicalName().getBytes())) {
+ Collection<Pair<Mutation, byte[]>> allMutations = generator.getIndexUpdates(htable,
+ mutationsPertainingToIndex.iterator());
indexMutationsMap = Maps.newTreeMap(Bytes.BYTES_COMPARATOR);
for (Pair<Mutation, byte[]> mutation : allMutations) {
List<Mutation> mutations = indexMutationsMap.get(mutation.getSecond());
@@ -559,20 +559,21 @@ public class MutationState implements SQLCloseable {
}
}
indexMutations = indexMutationsMap.get(index.getPhysicalName().getBytes());
- } else {
- indexMutations =
- IndexUtil.generateIndexData(table, index, values, mutationsPertainingToIndex,
- connection.getKeyValueBuilder(), connection);
+ } else {
+ indexMutations = IndexUtil.generateIndexData(table, index, values,
+ mutationsPertainingToIndex, connection.getKeyValueBuilder(), connection);
}
}
- // we may also have to include delete mutations for immutable tables if we are not processing all the tables in the mutations map
+ // we may also have to include delete mutations for immutable tables if we are not processing all
+ // the tables in the mutations map
if (!sendAll) {
TableRef key = new TableRef(index);
MultiRowMutationState multiRowMutationState = mutations.remove(key);
- if (multiRowMutationState!=null) {
+ if (multiRowMutationState != null) {
final List<Mutation> deleteMutations = Lists.newArrayList();
- generateMutations(tableRef, mutationTimestamp, serverTimestamp, multiRowMutationState, deleteMutations, null);
+ generateMutations(tableRef, mutationTimestamp, serverTimestamp, multiRowMutationState,
+ deleteMutations, null);
if (indexMutations == null) {
indexMutations = deleteMutations;
} else {
@@ -583,18 +584,20 @@ public class MutationState implements SQLCloseable {
} catch (SQLException | IOException e) {
throw new IllegalDataException(e);
}
- return new Pair<PName,List<Mutation>>(index.getPhysicalName(),indexMutations == null ? Collections.<Mutation>emptyList() : indexMutations);
+ return new Pair<PName, List<Mutation>>(index.getPhysicalName(),
+ indexMutations == null ? Collections.<Mutation> emptyList() : indexMutations);
}
@Override
public void remove() {
throw new UnsupportedOperationException();
}
-
+
};
}
- private PhoenixTxIndexMutationGenerator newTxIndexMutationGenerator(PTable table, List<PTable> indexes, Map<String,byte[]> attributes) {
+ 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);
@@ -603,8 +606,7 @@ public class MutationState implements SQLCloseable {
IndexMetaDataCache indexMetaDataCache = new IndexMetaDataCache() {
@Override
- public void close() throws IOException {
- }
+ public void close() throws IOException {}
@Override
public List<IndexMaintainer> getIndexMaintainers() {
@@ -613,30 +615,30 @@ public class MutationState implements SQLCloseable {
@Override
public PhoenixTransactionContext getTransactionContext() {
- return phoenixTransactionContext;
+ 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());
+ 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) {
+
+ private void generateMutations(final TableRef tableRef, final long mutationTimestamp, final long serverTimestamp,
+ final MultiRowMutationState values, final List<Mutation> mutationList,
+ final List<Mutation> mutationsPertainingToIndex) {
final PTable table = tableRef.getTable();
boolean tableWithRowTimestampCol = table.getRowTimestampColPos() != -1;
- Iterator<Map.Entry<ImmutableBytesPtr, RowMutationState>> iterator =
- values.entrySet().iterator();
+ Iterator<Map.Entry<ImmutableBytesPtr, RowMutationState>> iterator = values.entrySet().iterator();
long timestampToUse = mutationTimestamp;
MultiRowMutationState modifiedValues = new MultiRowMutationState(16);
while (iterator.hasNext()) {
@@ -650,11 +652,11 @@ public class MutationState implements SQLCloseable {
if (rowTsColInfo.useServerTimestamp()) {
// regenerate the key with this timestamp.
key = getNewRowKeyWithRowTimestamp(key, serverTimestamp, table);
- // since we are about to modify the byte[] stored in key (which changes its hashcode)
- // we need to remove the entry from the values map and add a new entry with the modified byte[]
- modifiedValues.put(key, state);
- iterator.remove();
- timestampToUse = serverTimestamp;
+ // since we are about to modify the byte[] stored in key (which changes its hashcode)
+ // we need to remove the entry from the values map and add a new entry with the modified byte[]
+ modifiedValues.put(key, state);
+ iterator.remove();
+ timestampToUse = serverTimestamp;
} else {
if (rowTsColInfo.getTimestamp() != null) {
timestampToUse = rowTsColInfo.getTimestamp();
@@ -669,15 +671,14 @@ public class MutationState implements SQLCloseable {
// The DeleteCompiler already generates the deletes for indexes, so no need to do it again
rowMutationsPertainingToIndex = Collections.emptyList();
} else {
- for (Map.Entry<PColumn, byte[]> valueEntry : rowEntry.getValue().getColumnValues()
- .entrySet()) {
+ for (Map.Entry<PColumn, byte[]> valueEntry : rowEntry.getValue().getColumnValues().entrySet()) {
row.setValue(valueEntry.getKey(), valueEntry.getValue());
}
rowMutations = row.toRowMutations();
// Pass through ON DUPLICATE KEY info through mutations
// In the case of the same clause being used on many statements, this will be
// inefficient because we're transmitting the same information for each mutation.
- // TODO: use our ServerCache
+ // TODO: use our ServerCache
for (Mutation mutation : rowMutations) {
if (onDupKeyBytes != null) {
mutation.setAttribute(PhoenixIndexBuilder.ATOMIC_OP_ATTRIB, onDupKeyBytes);
@@ -686,61 +687,61 @@ public class MutationState implements SQLCloseable {
rowMutationsPertainingToIndex = rowMutations;
}
mutationList.addAll(rowMutations);
- if (mutationsPertainingToIndex != null) mutationsPertainingToIndex
- .addAll(rowMutationsPertainingToIndex);
+ if (mutationsPertainingToIndex != null) mutationsPertainingToIndex.addAll(rowMutationsPertainingToIndex);
}
values.putAll(modifiedValues);
}
-
+
/**
* Get the unsorted list of HBase mutations for the tables with uncommitted data.
+ *
* @return list of HBase mutations for uncommitted data.
*/
- public Iterator<Pair<byte[],List<Mutation>>> toMutations(Long timestamp) {
+ public Iterator<Pair<byte[], List<Mutation>>> toMutations(Long timestamp) {
return toMutations(false, timestamp);
}
-
- public Iterator<Pair<byte[],List<Mutation>>> toMutations() {
+
+ public Iterator<Pair<byte[], List<Mutation>>> toMutations() {
return toMutations(false, null);
}
-
- public Iterator<Pair<byte[],List<Mutation>>> toMutations(final boolean includeMutableIndexes) {
+
+ public Iterator<Pair<byte[], List<Mutation>>> toMutations(final boolean includeMutableIndexes) {
return toMutations(includeMutableIndexes, null);
}
-
- public Iterator<Pair<byte[],List<Mutation>>> toMutations(final boolean includeMutableIndexes, final Long tableTimestamp) {
+
+ public Iterator<Pair<byte[], List<Mutation>>> toMutations(final boolean includeMutableIndexes,
+ final Long tableTimestamp) {
final Iterator<Map.Entry<TableRef, MultiRowMutationState>> iterator = this.mutations.entrySet().iterator();
- if (!iterator.hasNext()) {
- return Collections.emptyIterator();
- }
+ if (!iterator.hasNext()) { return Collections.emptyIterator(); }
Long scn = connection.getSCN();
final long serverTimestamp = getTableTimestamp(tableTimestamp, scn);
final long mutationTimestamp = getMutationTimestamp(scn);
- return new Iterator<Pair<byte[],List<Mutation>>>() {
+ return new Iterator<Pair<byte[], List<Mutation>>>() {
private Map.Entry<TableRef, MultiRowMutationState> current = iterator.next();
- private Iterator<Pair<byte[],List<Mutation>>> innerIterator = init();
-
- private Iterator<Pair<byte[],List<Mutation>>> init() {
- final Iterator<Pair<PName, List<Mutation>>> mutationIterator = addRowMutations(current.getKey(), current.getValue(), mutationTimestamp, serverTimestamp, includeMutableIndexes, true);
- return new Iterator<Pair<byte[],List<Mutation>>>() {
+ private Iterator<Pair<byte[], List<Mutation>>> innerIterator = init();
+
+ private Iterator<Pair<byte[], List<Mutation>>> init() {
+ final Iterator<Pair<PName, List<Mutation>>> mutationIterator = addRowMutations(current.getKey(),
+ current.getValue(), mutationTimestamp, serverTimestamp, includeMutableIndexes, true);
+ return new Iterator<Pair<byte[], List<Mutation>>>() {
@Override
public boolean hasNext() {
return mutationIterator.hasNext();
}
@Override
- public Pair<byte[], List<Mutation>> next() {
+ public Pair<byte[], List<Mutation>> next() {
Pair<PName, List<Mutation>> pair = mutationIterator.next();
return new Pair<byte[], List<Mutation>>(pair.getFirst().getBytes(), pair.getSecond());
}
-
+
@Override
public void remove() {
mutationIterator.remove();
}
};
}
-
+
@Override
public boolean hasNext() {
return innerIterator.hasNext() || iterator.hasNext();
@@ -750,7 +751,7 @@ public class MutationState implements SQLCloseable {
public Pair<byte[], List<Mutation>> next() {
if (!innerIterator.hasNext()) {
current = iterator.next();
- innerIterator=init();
+ innerIterator = init();
}
return innerIterator.next();
}
@@ -759,24 +760,26 @@ public class MutationState implements SQLCloseable {
public void remove() {
throw new UnsupportedOperationException();
}
-
+
};
}
public static long getTableTimestamp(final Long tableTimestamp, Long scn) {
- return (tableTimestamp!=null && tableTimestamp!=QueryConstants.UNSET_TIMESTAMP) ? tableTimestamp : (scn == null ? HConstants.LATEST_TIMESTAMP : scn);
+ return (tableTimestamp != null && tableTimestamp != QueryConstants.UNSET_TIMESTAMP) ? tableTimestamp
+ : (scn == null ? HConstants.LATEST_TIMESTAMP : scn);
}
-
+
public static long getMutationTimestamp(final Long scn) {
return scn == null ? HConstants.LATEST_TIMESTAMP : scn;
}
/**
- * Validates that the meta data is valid against the server meta data if we haven't yet done so.
- * Otherwise, for every UPSERT VALUES call, we'd need to hit the server to see if the meta data
- * has changed.
+ * Validates that the meta data is valid against the server meta data if we haven't yet done so. Otherwise, for
+ * every UPSERT VALUES call, we'd need to hit the server to see if the meta data has changed.
+ *
* @return the server time to use for the upsert
- * @throws SQLException if the table or any columns no longer exist
+ * @throws SQLException
+ * if the table or any columns no longer exist
*/
private long[] validateAll() throws SQLException {
int i = 0;
@@ -787,19 +790,20 @@ public class MutationState implements SQLCloseable {
}
return timeStamps;
}
-
- private long validateAndGetServerTimestamp(TableRef tableRef, MultiRowMutationState rowKeyToColumnMap) throws SQLException {
+
+ 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,
// so no need to do it again here.
PTable table = tableRef.getTable();
- MetaDataMutationResult result = client.updateCache(table.getSchemaName().getString(), table.getTableName().getString());
+ MetaDataMutationResult result = client.updateCache(table.getSchemaName().getString(), table.getTableName()
+ .getString());
PTable resolvedTable = result.getTable();
- if (resolvedTable == null) {
- throw new TableNotFoundException(table.getSchemaName().getString(), table.getTableName().getString());
- }
+ if (resolvedTable == null) { throw new TableNotFoundException(table.getSchemaName().getString(), table
+ .getTableName().getString()); }
// Always update tableRef table as the one we've cached may be out of date since when we executed
// the UPSERT VALUES call and updated in the cache before this.
tableRef.setTable(resolvedTable);
@@ -807,39 +811,38 @@ public class MutationState implements SQLCloseable {
for (PTable idxTtable : indexes) {
// If index is still active, but has a non zero INDEX_DISABLE_TIMESTAMP value, then infer that
// our failure mode is block writes on index failure.
- if ((idxTtable.getIndexState() == PIndexState.ACTIVE || idxTtable.getIndexState() == PIndexState.PENDING_ACTIVE) && idxTtable.getIndexDisableTimestamp() > 0) {
- throw new SQLExceptionInfo.Builder(SQLExceptionCode.INDEX_FAILURE_BLOCK_WRITE)
- .setSchemaName(table.getSchemaName().getString())
- .setTableName(table.getTableName().getString()).build().buildException();
- }
- }
+ if ((idxTtable.getIndexState() == PIndexState.ACTIVE || idxTtable.getIndexState() == PIndexState.PENDING_ACTIVE)
+ && idxTtable.getIndexDisableTimestamp() > 0) { throw new SQLExceptionInfo.Builder(
+ SQLExceptionCode.INDEX_FAILURE_BLOCK_WRITE).setSchemaName(table.getSchemaName().getString())
+ .setTableName(table.getTableName().getString()).build().buildException(); }
+ }
long timestamp = result.getMutationTime();
if (timestamp != QueryConstants.UNSET_TIMESTAMP) {
serverTimeStamp = timestamp;
if (result.wasUpdated()) {
List<PColumn> columns = Lists.newArrayListWithExpectedSize(table.getColumns().size());
- for (Map.Entry<ImmutableBytesPtr,RowMutationState> rowEntry : rowKeyToColumnMap.entrySet()) {
+ for (Map.Entry<ImmutableBytesPtr, RowMutationState> rowEntry : rowKeyToColumnMap.entrySet()) {
RowMutationState valueEntry = rowEntry.getValue();
if (valueEntry != null) {
Map<PColumn, byte[]> colValues = valueEntry.getColumnValues();
if (colValues != PRow.DELETE_MARKER) {
for (PColumn column : colValues.keySet()) {
- if (!column.isDynamic())
- columns.add(column);
+ if (!column.isDynamic()) columns.add(column);
}
}
}
}
for (PColumn column : columns) {
if (column != null) {
- resolvedTable.getColumnFamily(column.getFamilyName().getString()).getPColumnForColumnName(column.getName().getString());
+ resolvedTable.getColumnFamily(column.getFamilyName().getString()).getPColumnForColumnName(
+ column.getName().getString());
}
}
}
}
return serverTimeStamp == QueryConstants.UNSET_TIMESTAMP ? HConstants.LATEST_TIMESTAMP : serverTimeStamp;
}
-
+
private static long calculateMutationSize(List<Mutation> mutations) {
long byteSize = 0;
if (GlobalClientMetrics.isMetricsEnabled()) {
@@ -850,24 +853,6 @@ public class MutationState implements SQLCloseable {
GLOBAL_MUTATION_BYTES.update(byteSize);
return byteSize;
}
-
- private boolean hasKeyValueColumn(PTable table, PTable index) {
- IndexMaintainer maintainer = index.getIndexMaintainer(table, connection);
- return !maintainer.getAllColumns().isEmpty();
- }
-
- private void divideImmutableIndexes(Iterator<PTable> enabledImmutableIndexes, PTable table, List<PTable> rowKeyIndexes, List<PTable> keyValueIndexes) {
- while (enabledImmutableIndexes.hasNext()) {
- PTable index = enabledImmutableIndexes.next();
- if (index.getIndexType() != IndexType.LOCAL) {
- if (hasKeyValueColumn(table, index)) {
- keyValueIndexes.add(index);
- } else {
- rowKeyIndexes.add(index);
- }
- }
- }
- }
public long getBatchSizeBytes() {
return batchSizeBytes;
@@ -879,56 +864,52 @@ public class MutationState implements SQLCloseable {
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.
+ * 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();
- List<PTable> indexes = table.getIndexes();
- Iterator<PTable> enabledIndexes = IndexMaintainer.maintainedIndexes(indexes.iterator());
- if (enabledIndexes.hasNext()) {
- List<PTable> keyValueIndexes = Collections.emptyList();
- ImmutableBytesWritable indexMetaDataPtr = new ImmutableBytesWritable();
- boolean attachMetaData = table.getIndexMaintainers(indexMetaDataPtr, connection);
- if (table.isImmutableRows()) {
- List<PTable> rowKeyIndexes = Lists.newArrayListWithExpectedSize(indexes.size());
- keyValueIndexes = Lists.newArrayListWithExpectedSize(indexes.size());
- divideImmutableIndexes(enabledIndexes, table, rowKeyIndexes, keyValueIndexes);
- // Generate index deletes for immutable indexes that only reference row key
- // columns and submit directly here.
- ImmutableBytesWritable ptr = new ImmutableBytesWritable();
- for (PTable index : rowKeyIndexes) {
- List<Delete> indexDeletes = IndexUtil.generateDeleteIndexData(table, index, deletes, ptr, connection.getKeyValueBuilder(), connection);
- HTableInterface hindex = connection.getQueryServices().getTable(index.getPhysicalName().getBytes());
- hindex.delete(indexDeletes);
+ 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);
}
}
-
- // If we have mutable indexes, local immutable indexes, or global immutable indexes
- // that reference key value columns, setup index meta data and attach here. In this
- // case updates to the indexes will be generated on the server side.
- // An alternative would be to let Tephra track the row keys for the immutable index
- // by adding it as a transaction participant (soon we can prevent any conflict
- // detection from occurring) with the downside being the additional memory required.
- if (!keyValueIndexes.isEmpty()) {
- attachMetaData = true;
- IndexMaintainer.serializeAdditional(table, indexMetaDataPtr, keyValueIndexes, connection);
- }
- if (attachMetaData) {
- cache = setMetaDataOnMutations(tableRef, deletes, indexMetaDataPtr);
- }
}
+
delegate.delete(deletes);
} catch (SQLException e) {
throw new IOException(e);
@@ -939,13 +920,15 @@ public class MutationState implements SQLCloseable {
}
}
}
-
+
private static class TableInfo {
-
+
private final boolean isDataTable;
- @Nonnull private final PName hTableName;
- @Nonnull private final TableRef origTableRef;
-
+ @Nonnull
+ private final PName hTableName;
+ @Nonnull
+ private final TableRef origTableRef;
+
public TableInfo(boolean isDataTable, PName hTableName, TableRef origTableRef) {
super();
checkNotNull(hTableName);
@@ -981,14 +964,14 @@ public class MutationState implements SQLCloseable {
if (this == obj) return true;
if (obj == null) return false;
if (getClass() != obj.getClass()) return false;
- TableInfo other = (TableInfo) obj;
+ TableInfo other = (TableInfo)obj;
if (!hTableName.equals(other.hTableName)) return false;
if (isDataTable != other.isDataTable) return false;
return true;
}
}
-
+
@SuppressWarnings("deprecation")
private void send(Iterator<TableRef> tableRefIterator) throws SQLException {
int i = 0;
@@ -1001,7 +984,7 @@ public class MutationState implements SQLCloseable {
}
MultiRowMutationState multiRowMutationState;
- Map<TableInfo,List<Mutation>> physicalTableMutationMap = Maps.newLinkedHashMap();
+ Map<TableInfo, List<Mutation>> physicalTableMutationMap = Maps.newLinkedHashMap();
// add tracing for this operation
try (TraceScope trace = Tracing.startNewSpan(connection, "Committing mutations to tables")) {
Span span = trace.getSpan();
@@ -1014,21 +997,22 @@ public class MutationState implements SQLCloseable {
continue;
}
// Validate as we go if transactional since we can undo if a problem occurs (which is unlikely)
- long serverTimestamp = serverTimeStamps == null ? validateAndGetServerTimestamp(tableRef, multiRowMutationState) : serverTimeStamps[i++];
+ long serverTimestamp = serverTimeStamps == null ? validateAndGetServerTimestamp(tableRef,
+ multiRowMutationState) : serverTimeStamps[i++];
Long scn = connection.getSCN();
long mutationTimestamp = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
final PTable table = tableRef.getTable();
- Iterator<Pair<PName,List<Mutation>>> mutationsIterator = addRowMutations(tableRef, multiRowMutationState, mutationTimestamp, serverTimestamp, false, sendAll);
+ Iterator<Pair<PName, List<Mutation>>> mutationsIterator = addRowMutations(tableRef,
+ multiRowMutationState, mutationTimestamp, serverTimestamp, false, sendAll);
// build map from physical table to mutation list
boolean isDataTable = true;
while (mutationsIterator.hasNext()) {
- Pair<PName,List<Mutation>> pair = mutationsIterator.next();
+ Pair<PName, List<Mutation>> pair = mutationsIterator.next();
PName hTableName = pair.getFirst();
List<Mutation> mutationList = pair.getSecond();
TableInfo tableInfo = new TableInfo(isDataTable, hTableName, tableRef);
List<Mutation> oldMutationList = physicalTableMutationMap.put(tableInfo, mutationList);
- if (oldMutationList!=null)
- mutationList.addAll(0, oldMutationList);
+ if (oldMutationList != null) mutationList.addAll(0, oldMutationList);
isDataTable = false;
}
// For transactions, track the statement indexes as we send data
@@ -1049,23 +1033,25 @@ public class MutationState implements SQLCloseable {
}
}
long serverTimestamp = HConstants.LATEST_TIMESTAMP;
- Iterator<Entry<TableInfo, List<Mutation>>> mutationsIterator = physicalTableMutationMap.entrySet().iterator();
+ Iterator<Entry<TableInfo, List<Mutation>>> mutationsIterator = physicalTableMutationMap.entrySet()
+ .iterator();
while (mutationsIterator.hasNext()) {
Entry<TableInfo, List<Mutation>> pair = mutationsIterator.next();
TableInfo tableInfo = pair.getKey();
byte[] htableName = tableInfo.getHTableName().getBytes();
List<Mutation> mutationList = pair.getValue();
-
- //create a span per target table
- //TODO maybe we can be smarter about the table name to string here?
- Span child = Tracing.child(span,"Writing mutation batch for table: "+Bytes.toString(htableName));
+
+ // create a span per target table
+ // TODO maybe we can be smarter about the table name to string here?
+ Span child = Tracing.child(span, "Writing mutation batch for table: " + Bytes.toString(htableName));
int retryCount = 0;
boolean shouldRetry = false;
long numMutations = 0;
long mutationSizeBytes = 0;
long mutationCommitTime = 0;
- long numFailedMutations = 0;;
+ long numFailedMutations = 0;
+ ;
long startTime = 0;
boolean shouldRetryIndexedMutation = false;
IndexWriteException iwe = null;
@@ -1073,11 +1059,12 @@ 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() ? setMetaDataOnMutations(origTableRef,
+ 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.
- shouldRetry = cache!=null;
+ shouldRetry = cache != null;
SQLException sqlE = null;
HTableInterface hTable = connection.getQueryServices().getTable(htableName);
try {
@@ -1093,16 +1080,17 @@ public class MutationState implements SQLCloseable {
hTable = new MetaDataAwareHTable(hTable, origTableRef);
}
- hTable = phoenixTransactionContext.getTransactionalTable(hTable, table.isImmutableRows());
+ hTable = phoenixTransactionContext.getTransactionalTableWriter(hTable, table);
}
-
+
numMutations = mutationList.size();
GLOBAL_MUTATION_BATCH_SIZE.update(numMutations);
mutationSizeBytes = calculateMutationSize(mutationList);
-
+
startTime = System.currentTimeMillis();
child.addTimelineAnnotation("Attempt " + retryCount);
- List<List<Mutation>> mutationBatchList = getMutationBatchList(batchSize, batchSizeBytes, mutationList);
+ List<List<Mutation>> mutationBatchList = getMutationBatchList(batchSize, batchSizeBytes,
+ mutationList);
for (final List<Mutation> mutationBatch : mutationBatchList) {
if (shouldRetryIndexedMutation) {
// if there was an index write failure, retry the mutation in a loop
@@ -1116,14 +1104,16 @@ public class MutationState implements SQLCloseable {
Thread.currentThread().interrupt();
throw new IOException(e);
}
- }}, iwe,
- connection, connection.getQueryServices().getProps());
+ }
+ }, iwe, connection, connection.getQueryServices().getProps());
} else {
hTable.batch(mutationBatch);
}
batchCount++;
- if (logger.isDebugEnabled()) logger.debug("Sent batch of " + mutationBatch.size() + " for " + Bytes.toString(htableName));
+ if (logger.isDebugEnabled())
+ logger.debug("Sent batch of " + mutationBatch.size() + " for "
+ + Bytes.toString(htableName));
}
child.stop();
child.stop();
@@ -1131,7 +1121,7 @@ public class MutationState implements SQLCloseable {
mutationCommitTime = System.currentTimeMillis() - startTime;
GLOBAL_MUTATION_COMMIT_TIME.update(mutationCommitTime);
numFailedMutations = 0;
-
+
// Remove batches as we process them
mutations.remove(origTableRef);
if (tableInfo.isDataTable()) {
@@ -1140,22 +1130,28 @@ public class MutationState implements SQLCloseable {
estimatedSize = KeyValueUtil.getEstimatedRowMutationSize(mutations);
}
} catch (Exception e) {
- mutationCommitTime = System.currentTimeMillis() - startTime;
+ mutationCommitTime = System.currentTimeMillis() - startTime;
serverTimestamp = ServerUtil.parseServerTimestamp(e);
SQLException inferredE = ServerUtil.parseServerExceptionOrNull(e);
if (inferredE != null) {
- if (shouldRetry && retryCount == 0 && inferredE.getErrorCode() == SQLExceptionCode.INDEX_METADATA_NOT_FOUND.getErrorCode()) {
- // Swallow this exception once, as it's possible that we split after sending the index metadata
- // and one of the region servers doesn't have it. This will cause it to have it the next go around.
+ if (shouldRetry
+ && retryCount == 0
+ && inferredE.getErrorCode() == SQLExceptionCode.INDEX_METADATA_NOT_FOUND
+ .getErrorCode()) {
+ // Swallow this exception once, as it's possible that we split after sending the index
+ // metadata
+ // and one of the region servers doesn't have it. This will cause it to have it the next
+ // go around.
// If it fails again, we don't retry.
- String msg = "Swallowing exception and retrying after clearing meta cache on connection. " + inferredE;
+ String msg = "Swallowing exception and retrying after clearing meta cache on connection. "
+ + inferredE;
logger.warn(LogUtil.addCustomAnnotations(msg, connection));
connection.getQueryServices().clearTableRegionCache(htableName);
// add a new child span as this one failed
child.addTimelineAnnotation(msg);
child.stop();
- child = Tracing.child(span,"Failed batch, attempting retry");
+ child = Tracing.child(span, "Failed batch, attempting retry");
continue;
} else if (inferredE.getErrorCode() == SQLExceptionCode.INDEX_WRITE_FAILURE.getErrorCode()) {
@@ -1164,7 +1160,8 @@ public class MutationState implements SQLCloseable {
// For an index write failure, the data table write succeeded,
// so when we retry we need to set REPLAY_WRITES
for (Mutation m : mutationList) {
- m.setAttribute(BaseScannerRegionObserver.REPLAY_WRITES, BaseScannerRegionObserver.REPLAY_ONLY_INDEX_WRITES);
+ m.setAttribute(BaseScannerRegionObserver.REPLAY_WRITES,
+ BaseScannerRegionObserver.REPLAY_ONLY_INDEX_WRITES);
KeyValueUtil.setTimestamp(m, serverTimestamp);
}
shouldRetry = true;
@@ -1177,29 +1174,26 @@ public class MutationState implements SQLCloseable {
// Throw to client an exception that indicates the statements that
// were not committed successfully.
int[] uncommittedStatementIndexes = getUncommittedStatementIndexes();
- sqlE = new CommitException(e, uncommittedStatementIndexes, serverTimestamp);
- numFailedMutations = uncommittedStatementIndexes.length;
- GLOBAL_MUTATION_BATCH_FAILED_COUNT.update(numFailedMutations);
+ sqlE = new CommitException(e, uncommittedStatementIndexes, serverTimestamp);
+ numFailedMutations = uncommittedStatementIndexes.length;
+ GLOBAL_MUTATION_BATCH_FAILED_COUNT.update(numFailedMutations);
} finally {
- MutationMetric mutationsMetric = new MutationMetric(numMutations, mutationSizeBytes, mutationCommitTime, numFailedMutations);
+ MutationMetric mutationsMetric = new MutationMetric(numMutations, mutationSizeBytes,
+ mutationCommitTime, numFailedMutations);
mutationMetricQueue.addMetricsForTable(Bytes.toString(htableName), mutationsMetric);
try {
- if (cache!=null)
- cache.close();
+ if (cache != null) cache.close();
} finally {
try {
hTable.close();
- }
- catch (IOException e) {
+ } catch (IOException e) {
if (sqlE != null) {
sqlE.setNextException(ServerUtil.parseServerException(e));
} else {
sqlE = ServerUtil.parseServerException(e);
}
- }
- if (sqlE != null) {
- throw sqlE;
}
+ if (sqlE != null) { throw sqlE; }
}
}
} while (shouldRetry && retryCount++ < 1);
@@ -1209,10 +1203,13 @@ public class MutationState implements SQLCloseable {
/**
* Split the list of mutations into multiple lists that don't exceed row and byte thresholds
- * @param allMutationList List of HBase mutations
+ *
+ * @param allMutationList
+ * List of HBase mutations
* @return List of lists of mutations
*/
- public static List<List<Mutation>> getMutationBatchList(long batchSize, long batchSizeBytes, List<Mutation> allMutationList) {
+ public static List<List<Mutation>> getMutationBatchList(long batchSize, long batchSizeBytes,
+ List<Mutation> allMutationList) {
List<List<Mutation>> mutationBatchList = Lists.newArrayList();
List<Mutation> currentList = Lists.newArrayList();
long currentBatchSizeBytes = 0L;
@@ -1243,12 +1240,10 @@ public class MutationState implements SQLCloseable {
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);
+ 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();
}
@@ -1261,7 +1256,8 @@ public class MutationState implements SQLCloseable {
}
boolean hasIndexMetaData = indexMetaDataPtr.getLength() > 0;
if (hasIndexMetaData) {
- if (IndexMetaDataCacheClient.useIndexMetadataCache(connection, mutations, indexMetaDataPtr.getLength() + txState.length)) {
+ if (IndexMetaDataCacheClient.useIndexMetadataCache(connection, mutations, indexMetaDataPtr.getLength()
+ + txState.length)) {
IndexMetaDataCacheClient client = new IndexMetaDataCacheClient(connection, tableRef);
cache = client.addIndexMetadataCache(mutations, indexMetaDataPtr, txState);
uuidValue = cache.getId();
@@ -1269,9 +1265,7 @@ public class MutationState implements SQLCloseable {
attribValue = ByteUtil.copyKeyBytesIfNecessary(indexMetaDataPtr);
uuidValue = ServerCacheClient.generateId();
}
- } else if (txState.length == 0) {
- return null;
- }
+ } 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) {
@@ -1281,7 +1275,8 @@ public class MutationState implements SQLCloseable {
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));
+ mutation.setAttribute(BaseScannerRegionObserver.CLIENT_VERSION,
+ Bytes.toBytes(MetaDataProtocol.PHOENIX_VERSION));
if (txState.length > 0) {
mutation.setAttribute(BaseScannerRegionObserver.TX_STATE, txState);
}
@@ -1291,23 +1286,23 @@ public class MutationState implements SQLCloseable {
}
return cache;
}
-
+
private void addUncommittedStatementIndexes(Collection<RowMutationState> rowMutations) {
for (RowMutationState rowMutationState : rowMutations) {
- uncommittedStatementIndexes = joinSortedIntArrays(uncommittedStatementIndexes, rowMutationState.getStatementIndexes());
+ uncommittedStatementIndexes = joinSortedIntArrays(uncommittedStatementIndexes,
+ rowMutationState.getStatementIndexes());
}
}
-
+
private int[] getUncommittedStatementIndexes() {
for (MultiRowMutationState rowMutationMap : mutations.values()) {
addUncommittedStatementIndexes(rowMutationMap.values());
}
return uncommittedStatementIndexes;
}
-
+
@Override
- public void close() throws SQLException {
- }
+ public void close() throws SQLException {}
private void resetState() {
numRows = 0;
@@ -1335,26 +1330,29 @@ public class MutationState implements SQLCloseable {
Map<TableRef, MultiRowMutationState> txMutations = Collections.emptyMap();
int retryCount = 0;
do {
- boolean sendSuccessful=false;
+ boolean sendSuccessful = false;
boolean retryCommit = false;
SQLException sqlE = null;
try {
send();
txMutations = this.txMutations;
- sendSuccessful=true;
+ sendSuccessful = true;
} catch (SQLException e) {
sqlE = e;
} finally {
try {
- boolean finishSuccessful=false;
+ boolean finishSuccessful = false;
try {
if (sendSuccessful) {
phoenixTransactionContext.commit();
finishSuccessful = true;
}
} catch (SQLException e) {
- if (logger.isInfoEnabled()) logger.info(e.getClass().getName() + " at timestamp " + getInitialWritePointer() + " with retry count of " + retryCount);
- retryCommit = (e.getErrorCode() == SQLExceptionCode.TRANSACTION_CONFLICT_EXCEPTION.getErrorCode() && retryCount < MAX_COMMIT_RETRIES);
+ if (logger.isInfoEnabled())
+ logger.info(e.getClass().getName() + " at timestamp " + getInitialWritePointer()
+ + " with retry count of " + retryCount);
+ retryCommit = (e.getErrorCode() == SQLExceptionCode.TRANSACTION_CONFLICT_EXCEPTION
+ .getErrorCode() && retryCount < MAX_COMMIT_RETRIES);
if (sqlE == null) {
sqlE = e;
} else {
@@ -1401,9 +1399,7 @@ public class MutationState implements SQLCloseable {
}
}
}
- if (sqlE != null && !retryCommit) {
- throw sqlE;
- }
+ if (sqlE != null && !retryCommit) { throw sqlE; }
}
}
}
@@ -1418,11 +1414,12 @@ public class MutationState implements SQLCloseable {
/**
* Determines whether indexes were added to mutated tables while the transaction was in progress.
+ *
* @return true if indexes were added and false otherwise.
- * @throws SQLException
+ * @throws SQLException
*/
private boolean shouldResubmitTransaction(Set<TableRef> txTableRefs) throws SQLException {
- if (logger.isInfoEnabled()) logger.info("Checking for index updates as of " + getInitialWritePointer());
+ if (logger.isInfoEnabled()) logger.info("Checking for index updates as of " + getInitialWritePointer());
MetaDataClient client = new MetaDataClient(connection);
PMetaData cache = connection.getMetaDataCache();
boolean addedAnyIndexes = false;
@@ -1432,15 +1429,16 @@ public class MutationState implements SQLCloseable {
List<PTable> oldIndexes;
PTableRef ptableRef = cache.getTableRef(dataTable.getKey());
oldIndexes = ptableRef.getTable().getIndexes();
- // Always check at server for metadata change, as it's possible that the table is configured to not check for metadata changes
+ // Always check at server for metadata change, as it's possible that the table is configured to not check
+ // for metadata changes
// but in this case, the tx manager is telling us it's likely that there has been a change.
- MetaDataMutationResult result = client.updateCache(dataTable.getTenantId(), dataTable.getSchemaName().getString(), dataTable.getTableName().getString(), true);
+ MetaDataMutationResult result = client.updateCache(dataTable.getTenantId(), dataTable.getSchemaName()
+ .getString(), dataTable.getTableName().getString(), true);
long timestamp = TransactionUtil.getResolvedTime(connection, result);
tableRef.setTimeStamp(timestamp);
PTable updatedDataTable = result.getTable();
- if (updatedDataTable == null) {
- throw new TableNotFoundException(dataTable.getSchemaName().getString(), dataTable.getTableName().getString());
- }
+ if (updatedDataTable == null) { throw new TableNotFoundException(dataTable.getSchemaName().getString(),
+ dataTable.getTableName().getString()); }
allImmutableTables &= updatedDataTable.isImmutableRows();
tableRef.setTable(updatedDataTable);
if (!addedAnyIndexes) {
@@ -1448,10 +1446,14 @@ public class MutationState implements SQLCloseable {
// that an index was dropped and recreated with the same name but different
// indexed/covered columns.
addedAnyIndexes = (!oldIndexes.equals(updatedDataTable.getIndexes()));
- if (logger.isInfoEnabled()) logger.info((addedAnyIndexes ? "Updates " : "No updates ") + "as of " + timestamp + " to " + updatedDataTable.getName().getString() + " with indexes " + updatedDataTable.getIndexes());
+ if (logger.isInfoEnabled())
+ logger.info((addedAnyIndexes ? "Updates " : "No updates ") + "as of " + timestamp + " to "
+ + updatedDataTable.getName().getString() + " with indexes " + updatedDataTable.getIndexes());
}
}
- if (logger.isInfoEnabled()) logger.info((addedAnyIndexes ? "Updates " : "No updates ") + "to indexes as of " + getInitialWritePointer() + " over " + (allImmutableTables ? " all immutable tables" : " some mutable tables"));
+ if (logger.isInfoEnabled())
+ logger.info((addedAnyIndexes ? "Updates " : "No updates ") + "to indexes as of " + getInitialWritePointer()
+ + " over " + (allImmutableTables ? " all immutable tables" : " some mutable tables"));
// If all tables are immutable, we know the conflict we got was due to our DDL/DML fence.
// If any indexes were added, then the conflict might be due to DDL/DML fence.
return allImmutableTables || addedAnyIndexes;
@@ -1459,16 +1461,18 @@ public class MutationState implements SQLCloseable {
/**
* Send to HBase any uncommitted data for transactional tables.
+ *
* @return true if any data was sent and false otherwise.
* @throws SQLException
*/
public boolean sendUncommitted() throws SQLException {
return sendUncommitted(mutations.keySet().iterator());
}
+
/**
- * Support read-your-own-write semantics by sending uncommitted data to HBase prior to running a
- * query. In this way, they are visible to subsequent reads but are not actually committed until
- * commit is called.
+ * Support read-your-own-write semantics by sending uncommitted data to HBase prior to running a query. In this way,
+ * they are visible to subsequent reads but are not actually committed until commit is called.
+ *
* @param tableRefs
* @return true if any data was sent and false otherwise.
* @throws SQLException
@@ -1481,7 +1485,7 @@ public class MutationState implements SQLCloseable {
phoenixTransactionContext.setVisibilityLevel(PhoenixVisibilityLevel.SNAPSHOT);
}
- Iterator<TableRef> filteredTableRefs = Iterators.filter(tableRefs, new Predicate<TableRef>(){
+ Iterator<TableRef> filteredTableRefs = Iterators.filter(tableRefs, new Predicate<TableRef>() {
@Override
public boolean apply(TableRef tableRef) {
return tableRef.getTable().isTransactional();
@@ -1497,113 +1501,120 @@ public class MutationState implements SQLCloseable {
if (tableRef.getTable().isTransactional()) {
startTransaction(tableRef.getTable().getTransactionProvider());
}
- strippedAliases.add(new TableRef(null, tableRef.getTable(), tableRef.getTimeStamp(), tableRef.getLowerBoundTimeStamp(), tableRef.hasDynamicCols()));
+ strippedAliases.add(new TableRef(null, tableRef.getTable(), tableRef.getTimeStamp(), tableRef
+ .getLowerBoundTimeStamp(), tableRef.hasDynamicCols()));
}
send(strippedAliases.iterator());
return true;
}
return false;
}
-
+
public void send() throws SQLException {
send(null);
}
-
+
public static int[] joinSortedIntArrays(int[] a, int[] b) {
int[] result = new int[a.length + b.length];
int i = 0, j = 0, k = 0, current;
while (i < a.length && j < b.length) {
current = a[i] < b[j] ? a[i++] : b[j++];
- for ( ; i < a.length && a[i] == current; i++);
- for ( ; j < b.length && b[j] == current; j++);
+ for (; i < a.length && a[i] == current; i++)
+ ;
+ for (; j < b.length && b[j] == current; j++)
+ ;
result[k++] = current;
}
while (i < a.length) {
- for (current = a[i++] ; i < a.length && a[i] == current; i++);
+ for (current = a[i++]; i < a.length && a[i] == current; i++)
+ ;
result[k++] = current;
}
while (j < b.length) {
- for (current = b[j++] ; j < b.length && b[j] == current; j++);
+ for (current = b[j++]; j < b.length && b[j] == current; j++)
+ ;
result[k++] = current;
}
return Arrays.copyOf(result, k);
}
-
+
@Immutable
public static class RowTimestampColInfo {
private final boolean useServerTimestamp;
- private final Long rowTimestamp;
-
+ private final Long rowTimestamp;
+
public static final RowTimestampColInfo NULL_ROWTIMESTAMP_INFO = new RowTimestampColInfo(false, null);
public RowTimestampColInfo(boolean autoGenerate, Long value) {
this.useServerTimestamp = autoGenerate;
this.rowTimestamp = value;
}
-
+
public boolean useServerTimestamp() {
return useServerTimestamp;
}
-
+
public Long getTimestamp() {
return rowTimestamp;
}
}
-
+
public static class MultiRowMutationState {
- private Map<ImmutableBytesPtr,RowMutationState> rowKeyToRowMutationState;
+ private Map<ImmutableBytesPtr, RowMutationState> rowKeyToRowMutationState;
private long estimatedSize;
-
+
public MultiRowMutationState(int size) {
this.rowKeyToRowMutationState = Maps.newHashMapWithExpectedSize(size);
this.estimatedSize = 0;
}
-
- public RowMutationState put(ImmutableBytesPtr ptr, RowMutationState rowMutationState) {
+
+ public RowMutationState put(ImmutableBytesPtr ptr, RowMutationState rowMutationState) {
estimatedSize += rowMutationState.calculateEstimatedSize();
return rowKeyToRowMutationState.put(ptr, rowMutationState);
}
-
+
public void putAll(MultiRowMutationState other) {
estimatedSize += other.estimatedSize;
rowKeyToRowMutationState.putAll(other.rowKeyToRowMutationState);
}
-
+
public boolean isEmpty() {
return rowKeyToRowMutationState.isEmpty();
}
-
+
public int size() {
return rowKeyToRowMutationState.size();
}
-
+
public Set<Entry<ImmutableBytesPtr, RowMutationState>> entrySet() {
return rowKeyToRowMutationState.entrySet();
}
-
- public void clear(){
+
+ public void clear() {
rowKeyToRowMutationState.clear();
estimatedSize = 0;
}
-
+
public Collection<RowMutationState> values() {
return rowKeyToRowMutationState.values();
}
}
-
+
public static class RowMutationState {
- @Nonnull private Map<PColumn,byte[]> columnValues;
+ @Nonnull
+ private Map<PColumn, byte[]> columnValues;
private int[] statementIndexes;
- @Nonnull private final RowTimestampColInfo rowTsColInfo;
+ @Nonnull
+ private final RowTimestampColInfo rowTsColInfo;
private byte[] onDupKeyBytes;
private long colValuesSize;
-
- public RowMutationState(@Nonnull Map<PColumn,byte[]> columnValues, long colValuesSize, int statementIndex, @Nonnull RowTimestampColInfo rowTsColInfo,
- byte[] onDupKeyBytes) {
+
+ public RowMutationState(@Nonnull Map<PColumn, byte[]> columnValues, long colValuesSize, int statementIndex,
+ @Nonnull RowTimestampColInfo rowTsColInfo, byte[] onDupKeyBytes) {
checkNotNull(columnValues);
checkNotNull(rowTsColInfo);
this.columnValues = columnValues;
- this.statementIndexes = new int[] {statementIndex};
+ this.statementIndexes = new int[] { statementIndex };
this.rowTsColInfo = rowTsColInfo;
this.onDupKeyBytes = onDupKeyBytes;
this.colValuesSize = colValuesSize;
@@ -1631,13 +1642,13 @@ public class MutationState implements SQLCloseable {
// ignore the new values (as that's what the server will do).
if (newRow.onDupKeyBytes == null) {
// increment the column value size by the new row column value size
- colValuesSize+=newRow.colValuesSize;
- for (Map.Entry<PColumn,byte[]> entry : newRow.columnValues.entrySet()) {
+ colValuesSize += newRow.colValuesSize;
+ for (Map.Entry<PColumn, byte[]> entry : newRow.columnValues.entrySet()) {
PColumn col = entry.getKey();
byte[] oldValue = columnValues.put(col, entry.getValue());
- if (oldValue!=null) {
+ if (oldValue != null) {
// decrement column value size by the size of all column values that were replaced
- colValuesSize-=(col.getEstimatedSize() + oldValue.length);
+ colValuesSize -= (col.getEstimatedSize() + oldValue.length);
}
}
}
@@ -1646,14 +1657,14 @@ public class MutationState implements SQLCloseable {
this.onDupKeyBytes = PhoenixIndexBuilder.combineOnDupKey(this.onDupKeyBytes, newRow.onDupKeyBytes);
statementIndexes = joinSortedIntArrays(statementIndexes, newRow.getStatementIndexes());
}
-
+
@Nonnull
RowTimestampColInfo getRowTimestampColInfo() {
return rowTsColInfo;
}
}
-
+
public ReadMetricQueue getReadMetricQueue() {
return readMetricQueue;
}
@@ -1665,5 +1676,5 @@ public class MutationState implements SQLCloseable {
public MutationMetricQueue getMutationMetricQueue() {
return mutationMetricQueue;
}
-
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/edb6dcf5/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 6348d6d..0016fa9 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
@@ -95,7 +95,7 @@ public class PhoenixTxIndexMutationGenerator {
stored.addAll(m);
}
- public Collection<Pair<Mutation, byte[]>> getIndexUpdates(HTableInterface htable, Iterator<Mutation> mutationIterator) throws IOException, SQLException {
+ public Collection<Pair<Mutation, byte[]>> getIndexUpdates(HTableInterface htable, Iterator<? extends Mutation> mutationIterator) throws IOException, SQLException {
if (!mutationIterator.hasNext()) {
return Collections.emptyList();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/edb6dcf5/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 543eda1..ab9e8a6 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
@@ -130,4 +130,10 @@ public class OmidTransactionContext implements PhoenixTransactionContext {
// TODO Auto-generated method stub
return null;
}
+
+ @Override
+ public HTableInterface getTransactionalTableWriter(HTableInterface htable, PTable table) {
+ // TODO Auto-generated method stub
+ return null;
+ }
}
[4/7] phoenix git commit: PHOENIX-4769 Annotate SystemCatalogIT so
that it will run with the test suite
Posted by pb...@apache.org.
PHOENIX-4769 Annotate SystemCatalogIT so that it will run with the test suite
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/5b45e8fd
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/5b45e8fd
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/5b45e8fd
Branch: refs/heads/4.x-cdh5.12
Commit: 5b45e8fdeb1735bbee898b227b9bbd6f37e5ae31
Parents: edb6dcf
Author: Ankit Singhal <an...@gmail.com>
Authored: Fri Jun 1 22:42:50 2018 +0100
Committer: Pedro Boado <pb...@apache.org>
Committed: Sat Jun 2 09:09:42 2018 +0100
----------------------------------------------------------------------
.../apache/phoenix/end2end/SystemCatalogIT.java | 23 +++++++++++++-------
1 file changed, 15 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5b45e8fd/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemCatalogIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemCatalogIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemCatalogIT.java
index a754cb6..7e479c6 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemCatalogIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemCatalogIT.java
@@ -18,7 +18,6 @@
package org.apache.phoenix.end2end;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
import java.sql.Connection;
import java.sql.DriverManager;
@@ -26,14 +25,18 @@ import java.sql.SQLException;
import java.sql.Statement;
import java.util.Properties;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.RegionLocator;
+import org.apache.phoenix.query.BaseTest;
import org.apache.phoenix.util.PhoenixRuntime;
import org.junit.After;
import org.junit.Test;
+import org.junit.experimental.categories.Category;
-public class SystemCatalogIT {
+@Category(NeedsOwnMiniClusterTest.class)
+public class SystemCatalogIT extends BaseTest {
private HBaseTestingUtility testUtil = null;
@After
@@ -57,13 +60,17 @@ public class SystemCatalogIT {
TableName systemCatalog = TableName.valueOf("SYSTEM.CATALOG");
RegionLocator rl = testUtil.getConnection().getRegionLocator(systemCatalog);
assertEquals(rl.getAllRegionLocations().size(), 1);
+ try {
+ // now attempt to split SYSTEM.CATALOG
+ testUtil.getHBaseAdmin().split(systemCatalog);
- // now attempt to split SYSTEM.CATALOG
- testUtil.getHBaseAdmin().split(systemCatalog);
-
- // make sure the split finishes (there's no synchronous splitting before HBase 2.x)
- testUtil.getHBaseAdmin().disableTable(systemCatalog);
- testUtil.getHBaseAdmin().enableTable(systemCatalog);
+ // make sure the split finishes (there's no synchronous splitting before HBase 2.x)
+ testUtil.getHBaseAdmin().disableTable(systemCatalog);
+ testUtil.getHBaseAdmin().enableTable(systemCatalog);
+ } catch (DoNotRetryIOException e) {
+ // table is not splittable
+ assert (e.getMessage().contains("NOT splittable"));
+ }
// test again... Must still be exactly one region.
rl = testUtil.getConnection().getRegionLocator(systemCatalog);